#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()

[offtherailz]

Nice work on this, the shape of the solution looks right to me: the per-bucket state is well isolated in RateLimitManager, and the injected now/scheduler make it properly testable.

Besides the rebase (there are conflicts due to recent work), there are five things I'd like to sort out before merge. The first four are about the defaults rather than the implementation, but together they mean that a single 429 from a remote server can freeze the application with nothing showing up anywhere.

1. The delay coming from Retry-After is never capped. In RateLimitManager.js:239 the Math.min with maxDelay only guards the exponential branch, so a server answering Retry-After: 3600 blocks the bucket for a full hour. On top of that wait() has no timeout and never rejects, so during that hour requests on the blocked bucket neither start nor fail: nothing in the console, no notification, just spinners going forever. Can we cap the delay with maxDelay and send the request anyway once the cap expires? Throttling should be able to delay a request, not to swallow it.

2. maxRetries defaults to null, which means retry forever. Against a server that keeps answering 429 the request never settles, so the user never gets an error either. I'd default it to 2 or 3 and leave null for whoever explicitly wants it.

3. defaultBucket is "origin", so the key is just host:port and every request to that host shares the same backoff. libs/ajax.js is the axios instance the whole frontend goes through, and in a standard deployment MapStore and GeoServer sit on the same origin, so one rate-limited WMS layer ends up throttling login, map save and session refresh too. Could wmsLayer be the default instead, leaving origin for setups that really do have a per-host limit? I'd also keep MapStore's own API calls out of the buckets, since delaying those takes no load off the OGC service that answered 429 in the first place.

   Related to this: msRateLimitBucket and msRateLimitKey are already read from the layer options and take precedence over bucketRules, which is exactly the hook I'd want for per-layer control, but they don't show up anywhere under docs/. Could we document them as a layer parameter, ideally settable on the catalog service too so the layers inherit it? Otherwise I'd rather see them removed, because as they stand nobody configuring MapStore can find them.

4. In the OL layer the whole load function runs inside wait() (WMSLayer.js:73), which leaves the tile in LOADING for the entire backoff window. That queue belongs to the map and is shared by every layer (maxTilesLoading is 16 by default, 8 while panning or zooming), and tilesLoading_ only goes down once a tile reaches LOADED or ERROR. So sixteen waiting tiles and nothing loads any more, including layers pointing at unrelated servers. What about waiting only for short backoffs, a couple of seconds at most, and putting the tile in error state otherwise so it gets requested again later?

5. createSingleTileImageryProvider replaces requestImage and reads provider._image, _hasError and _resource, which are private Cesium fields. I'd rather not have that in a MapStore utility: it will break silently on the next Cesium upgrade, and there is no comment recording which version it was written against. I understand why it's there, since SingleTileImageryProvider.requestImage goes through loadImage(null, resource) and therefore never gets preferBlob, so the 429 status stays invisible. The good news is that the tiled case doesn't need any of it: UrlTemplateImageryProvider calls loadImage(this, resource), and since wmsToCesiumOptions sets a NeverTileDiscardPolicy by default, loadImage already takes the preferBlob: true branch and the retryCallback works through the public API alone. Can we drop the override and accept that single tile WMS in 3D has no 429 handling? If that case matters to someone we can raise it upstream with CesiumJS instead.

One documentation fix while you're in there: local-config.md says wmsLayer ignores tile-specific parameters such as BBOX, WIDTH, HEIGHT, SRS and CRS, but the key is built from origin + pathname + LAYERS, so it actually ignores every parameter. Worth rewording so it matches the code.

The smaller ones I've left inline: the backoff being reset rather than decayed on success, and the retry loop on the native image path.

[Binabh]

@offtherailz thank you for your feedback. I will start working on your suggestions

[Binabh]

@offtherailz I have updated the defaults as you mentioned and changed the behavior for createSingleTileImageryProvider as you suggested. I also have changed the tests to alsign with new defaults.
Also I have tried to make docs more explanatory. Please let me know if any further improvements are needed.
Thank You!

[offtherailz]

I tested this branch against a server that actually rate-limits, and there are two things worth raising before we go further.

• 1. As it stands the feature makes the problem worse. Against a server allowing 1 request per second, 24 tiles produce 142 requests in 1.6 s and 2 tiles end up painted. On top of that the bucket is the origin, and libs/ajax.js is the axios instance the whole frontend goes through, so a rate-limited layer can also stall login, map save and session refresh. This needs fixing whatever we decide about scope.
  @Binabh How did you tested this ?

2. Done properly, requires too much complicated work Expecially because the APIs do not expose a way to work with prioritization of tiles.

So I'd block this PR for the moment. We have to review the plan before to proceed.

[Binabh]

@offtherailz Sorry for that. I tested using the 429 emulation python server that I mentioned in PR description. I will try this on real service and test if its blocking other functionality like login also. For second time I focused my manual verification on cesium single tile load and that default have been properly stored only which is mistake on my side.