#12556 Implement adaptive request throttling on HTTP 429 (Too Many Requests) responses

[Binabh]

Description

• Implemented 429 error handling by implementing backoff using RateLimitManager and using it in axios interceptors
• Add possibility to configure how rate limit is handled using localconfig.json file and have documented it
• Add WMS rate-limit waiting in OpenLayers loadFunction and also for cesium
• Also I have added tests to cover that rate-limiting are properly handled

Please check if the PR fulfills these requirements

• The commit message follows our guidelines: https://github.com/geosolutions-it/MapStore2/blob/master/CONTRIBUTING.md
• Tests for the changes have been added (for bug fixes / features)
• Docs have been added / updated (for bug fixes / features)

What kind of change does this PR introduce? (check one with "x", remove the others)

• Feature

Issue

What is the current behavior?
#12556

What is the new behavior?
Mapstore correctly reads Retry-After header in case of 429 response and handles accordingly.

Breaking change

Does this PR introduce a breaking change? (check one with "x", remove the other)

• Yes, and I documented them in migration notes
• No

Other useful information

To test all of the features implemented just having a running Geoserver may not be enough. For testing purpose we can start a simple proxy server in front of Geoserver to apply rate limits. Python proxy server used:

rate_limit_server.py

#!/usr/bin/env python3
"""
Small configurable GeoServer proxy for testing MapStore HTTP 429 throttling.

Run:
  python3 rate_limit_server.py

Then use this as the WMS service URL in MapStore:
  http://localhost:8882/geoserver/wms

Debug endpoints:
  http://localhost:8882/__rate_limit__/state
  http://localhost:8882/__rate_limit__/config
  http://localhost:8882/__rate_limit__/reset
"""

from __future__ import annotations

import email.utils
import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Dict, Optional, Tuple


CONFIG = {
  # The proxy listens on http://localhost:8882 by default.
  "listen_host": "localhost",
  "listen_port": 8882,

  # Incoming /geoserver/... requests are forwarded to this base URL.
  # If your real GeoServer is somewhere else, change only this value.
  "target_base_url": "http://localhost:8082/geoserver",
  "proxy_prefix": "/geoserver",

  # Keep CORS open so Axios/blob fallback can inspect 429 status and headers.
  "cors": {
      "enabled": True,
      "allow_origin": "*",
      "allow_methods": "GET, POST, HEAD, OPTIONS",
      "allow_headers": "Authorization, Content-Type, X-Requested-With",
      "expose_headers": "Retry-After, Content-Type, Content-Length"
  },

  "rate_limit": {
      "enabled": True,

      # Restrict synthetic 429 responses to WMS/OWS-ish paths.
      # Set to ".*" to apply to every proxied request.
      "path_pattern": r"(^|.*/)(ows|wms)$",
      "methods": ["GET", "POST"],

      # Supported bucket values: "global", "origin", "path", "wmsLayer".
      # "wmsLayer" groups by path + LAYERS and ignores tile params.
      "bucket": "wmsLayer",

      # Supported scenarios:
      # - first_n_per_bucket: 429 for the first N requests in each bucket.
      # - every_nth: 429 every Nth request in each bucket.
      # - always: 429 every matching request.
      # - window: allow N requests per time window, then 429 until it resets.
      "scenario": "every_nth",
      "first_n": 1,
      "every_n": 3,
      "window_limit": 4,
      "window_seconds": 10,

      # Supported Retry-After modes:
      # - seconds: Retry-After: <retry_after_seconds>
      # - http-date: Retry-After: <current time + retry_after_seconds>
      # - invalid: Retry-After: not-a-date
      # - none: omit Retry-After to test exponential backoff
      # - fixed: use retry_after_fixed_value as-is
      "retry_after_mode": "seconds",
      "retry_after_seconds": 10,
      "retry_after_fixed_value": "5",

      "response_body": {
          "error": "synthetic-rate-limit",
          "message": "Synthetic HTTP 429 generated by rate_limit_geoserver_proxy.py"
      }
  }
}


HOP_BY_HOP_HEADERS = {
  "connection",
  "keep-alive",
  "proxy-authenticate",
  "proxy-authorization",
  "te",
  "trailers",
  "transfer-encoding",
  "upgrade"
}

STATE = {
  "started_at": time.time(),
  "total_requests": 0,
  "proxied_requests": 0,
  "synthetic_429": 0,
  "buckets": {}
}


def json_bytes(value) -> bytes:
  return json.dumps(value, indent=2, sort_keys=True).encode("utf-8")


def normalize_path(path: str) -> str:
  prefix = CONFIG["proxy_prefix"].rstrip("/")
  if prefix and path.startswith(prefix):
      stripped = path[len(prefix):]
      return stripped or "/"
  return path or "/"


def target_url_for_request(path: str) -> str:
  parsed = urllib.parse.urlsplit(path)
  forward_path = normalize_path(parsed.path)
  target = CONFIG["target_base_url"].rstrip("/") + forward_path
  if parsed.query:
      target += "?" + parsed.query
  return target


def parse_params(query: str, body: bytes, headers) -> Dict[str, list]:
  params = urllib.parse.parse_qs(query, keep_blank_values=True)
  content_type = headers.get("Content-Type", "")
  if body and "application/x-www-form-urlencoded" in content_type:
      body_text = body.decode("utf-8", errors="replace")
      body_params = urllib.parse.parse_qs(body_text, keep_blank_values=True)
      params.update(body_params)
  return params


def first_param(params: Dict[str, list], name: str) -> Optional[str]:
  for key, value in params.items():
      if key.lower() == name.lower() and value:
          return ",".join(value)
  return None


def normalized_layers(params: Dict[str, list]) -> str:
  layers = first_param(params, "layers") or ""
  return ",".join(part.strip() for part in layers.split(",") if part.strip())


def bucket_key(method: str, path: str, body: bytes, headers) -> str:
  cfg = CONFIG["rate_limit"]
  bucket = cfg["bucket"]
  parsed = urllib.parse.urlsplit(path)
  host = headers.get("Host", f"{CONFIG['listen_host']}:{CONFIG['listen_port']}")
  origin = f"http://{host}"

  if bucket == "global":
      return "global"
  if bucket == "origin":
      return origin

  params = parse_params(parsed.query, body, headers)
  clean_path = parsed.path
  if bucket == "wmsLayer":
      layers = normalized_layers(params)
      return f"{clean_path}?LAYERS={layers}" if layers else clean_path

  if bucket == "path":
      return clean_path

  return f"{method}:{clean_path}"


def matching_rate_limit_rule(method: str, path: str) -> bool:
  cfg = CONFIG["rate_limit"]
  if not cfg["enabled"]:
      return False
  if method not in cfg["methods"]:
      return False
  parsed = urllib.parse.urlsplit(path)
  return bool(re.search(cfg["path_pattern"], parsed.path))


def get_bucket_record(key: str) -> Dict[str, float]:
  buckets = STATE["buckets"]
  if key not in buckets:
      buckets[key] = {
          "seen": 0,
          "limited": 0,
          "window_start": time.time(),
          "window_seen": 0
      }
  return buckets[key]


def should_emit_429(method: str, path: str, body: bytes, headers) -> Tuple[bool, Optional[str]]:
  if not matching_rate_limit_rule(method, path):
      return False, None

  cfg = CONFIG["rate_limit"]
  key = bucket_key(method, path, body, headers)
  record = get_bucket_record(key)
  record["seen"] += 1

  scenario = cfg["scenario"]
  emit = False

  if scenario == "always":
      emit = True
  elif scenario == "first_n_per_bucket":
      emit = record["limited"] < cfg["first_n"]
  elif scenario == "every_nth":
      every_n = max(0, int(cfg["every_n"]))
      emit = every_n > 0 and record["seen"] % every_n == 0
  elif scenario == "window":
      now = time.time()
      if now - record["window_start"] > cfg["window_seconds"]:
          record["window_start"] = now
          record["window_seen"] = 0
      record["window_seen"] += 1
      emit = record["window_seen"] > cfg["window_limit"]

  if emit:
      record["limited"] += 1
      STATE["synthetic_429"] += 1
  return emit, key


def retry_after_value() -> Optional[str]:
  cfg = CONFIG["rate_limit"]
  mode = cfg["retry_after_mode"]
  seconds = cfg["retry_after_seconds"]
  if mode == "seconds":
      return str(seconds)
  if mode == "http-date":
      return email.utils.formatdate(time.time() + seconds, usegmt=True)
  if mode == "invalid":
      return "not-a-date"
  if mode == "fixed":
      return str(cfg["retry_after_fixed_value"])
  return None


def filtered_headers(headers, skip_content_length: bool = False) -> Dict[str, str]:
  output = {}
  for key, value in headers.items():
      lower = key.lower()
      if lower in HOP_BY_HOP_HEADERS:
          continue
      if skip_content_length and lower == "content-length":
          continue
      output[key] = value
  return output


class GeoServer429Proxy(BaseHTTPRequestHandler):
  server_version = "GeoServer429Proxy/1.0"

  def log_message(self, fmt: str, *args) -> None:
      sys.stderr.write("[%s] %s\n" % (self.log_date_time_string(), fmt % args))

  def end_headers(self) -> None:
      cors = CONFIG["cors"]
      if cors["enabled"]:
          self.send_header("Access-Control-Allow-Origin", cors["allow_origin"])
          self.send_header("Access-Control-Allow-Methods", cors["allow_methods"])
          self.send_header("Access-Control-Allow-Headers", cors["allow_headers"])
          self.send_header("Access-Control-Expose-Headers", cors["expose_headers"])
      super().end_headers()

  def do_OPTIONS(self) -> None:
      self.send_response(204)
      self.end_headers()

  def do_GET(self) -> None:
      self.handle_request()

  def do_POST(self) -> None:
      self.handle_request()

  def do_HEAD(self) -> None:
      self.handle_request(send_body=False)

  def handle_request(self, send_body: bool = True) -> None:
      if self.path.startswith("/__rate_limit__/"):
          self.handle_debug_endpoint(send_body)
          return

      content_length = int(self.headers.get("Content-Length") or 0)
      body = self.rfile.read(content_length) if content_length else b""
      STATE["total_requests"] += 1

      emit_429, key = should_emit_429(self.command, self.path, body, self.headers)
      if emit_429:
          self.send_synthetic_429(key, send_body)
          return

      self.proxy_to_geoserver(body, send_body)

  def handle_debug_endpoint(self, send_body: bool) -> None:
      if self.path.startswith("/__rate_limit__/reset"):
          STATE["total_requests"] = 0
          STATE["proxied_requests"] = 0
          STATE["synthetic_429"] = 0
          STATE["buckets"] = {}
          payload = {"ok": True, "message": "rate-limit state reset"}
      elif self.path.startswith("/__rate_limit__/config"):
          payload = CONFIG
      else:
          payload = STATE

      data = json_bytes(payload)
      self.send_response(200)
      self.send_header("Content-Type", "application/json")
      self.send_header("Content-Length", str(len(data)))
      self.end_headers()
      if send_body:
          self.wfile.write(data)

  def send_synthetic_429(self, key: Optional[str], send_body: bool) -> None:
      payload = {
          **CONFIG["rate_limit"]["response_body"],
          "bucket": key
      }
      data = json_bytes(payload)
      self.send_response(429)
      self.send_header("Content-Type", "application/json")
      self.send_header("Content-Length", str(len(data)))
      retry_after = retry_after_value()
      if retry_after is not None:
          self.send_header("Retry-After", retry_after)
      self.end_headers()
      if send_body:
          self.wfile.write(data)

  def proxy_to_geoserver(self, body: bytes, send_body: bool) -> None:
      target_url = target_url_for_request(self.path)
      method = self.command
      data = body if method in {"POST", "PUT", "PATCH"} else None
      headers = filtered_headers(self.headers, skip_content_length=True)
      headers["Host"] = urllib.parse.urlsplit(CONFIG["target_base_url"]).netloc
      request = urllib.request.Request(target_url, data=data, headers=headers, method=method)

      try:
          with urllib.request.urlopen(request, timeout=60) as response:
              response_body = response.read()
              STATE["proxied_requests"] += 1
              self.send_response(response.status)
              self.forward_response_headers(response.headers, len(response_body))
              self.end_headers()
              if send_body:
                  self.wfile.write(response_body)
      except urllib.error.HTTPError as error:
          response_body = error.read()
          self.send_response(error.code)
          self.forward_response_headers(error.headers, len(response_body))
          self.end_headers()
          if send_body:
              self.wfile.write(response_body)
      except urllib.error.URLError as error:
          message = {
              "error": "proxy-error",
              "target": target_url,
              "message": str(error)
          }
          data = json_bytes(message)
          self.send_response(502)
          self.send_header("Content-Type", "application/json")
          self.send_header("Content-Length", str(len(data)))
          self.end_headers()
          if send_body:
              self.wfile.write(data)

  def forward_response_headers(self, headers, content_length: int) -> None:
      for key, value in filtered_headers(headers, skip_content_length=True).items():
          self.send_header(key, value)
      self.send_header("Content-Length", str(content_length))


def main() -> None:
  host = CONFIG["listen_host"]
  port = CONFIG["listen_port"]
  server = ThreadingHTTPServer((host, port), GeoServer429Proxy)
  print(f"429 test proxy listening on http://{host}:{port}")
  print(f"Forwarding {CONFIG['proxy_prefix']}/* to {CONFIG['target_base_url']}")
  print("Use Ctrl+C to stop.")
  try:
      server.serve_forever()
  except KeyboardInterrupt:
      print("\nStopping 429 test proxy.")
  finally:
      server.server_close()


if __name__ == "__main__":
  main()