feat(tracing): fail-closed rate gate + breaker feed on ingest (#831)

campus/audit/client/v1/traces.py

@@ -10,13 +10,38 @@
     /audit/v1/traces/search         → Search traces
 """
 
+import contextlib
 from typing import Any, Optional
 
 from campus.common.http.interface import JsonClient, JsonResponse
 
 from ..interface import SLASH, Resource, ResourceCollection, ResourceRoot
 
 
+def _import_ratelimit():
+    """Import the SDK breaker module lazily (it ships in campus_python).
+
+    Resolved via importlib with a string name: the module is optional
+    (a stale campus_python predating #831 lacks it), and opaque
+    resolution keeps type checkers happy against older SDK installs.
+    Returns None when unavailable — callers degrade to no observation.
+    """
+    import importlib
+
+    try:
+        return importlib.import_module("campus_python.audit.ratelimit")
+    except ImportError:
+        return None
+
+
+def _safe_json(response: JsonResponse) -> Any:
+    """Best-effort JSON parse for breaker observation."""
+    try:
+        return response.json()
+    except Exception:
+        return None
+
+
 class Traces(ResourceCollection):
     """Resource for trace spans collection.
 
@@ -80,6 +105,18 @@ def new(self, *spans: dict) -> JsonResponse:
             client.traces.new(span1, span2, span3)
         """
         response = self.client.post(self.make_path(), json={"spans": list(spans)})
+        # Feed the ingest circuit breaker before raising (#831): audit
+        # 429s trip the bucket (Retry-After + tripped key from the
+        # body), 2xx clears expired trips. Observation is guarded — a
+        # breaker failure or a stale SDK can never break ingestion.
+        with contextlib.suppress(Exception):
+            ratelimit_module = _import_ratelimit()
+            if ratelimit_module is not None:
+                ratelimit_module.breaker.observe(
+                    response.status_code,
+                    response.headers,
+                    _safe_json(response),
+                )
         response.raise_for_status()
         return response
 

campus/audit/middleware/__init__.py

@@ -27,6 +27,15 @@ def init_app(app: flask.Flask) -> None:
     """
     from . import tracing
 
+    @app.before_request
+    def audit_rate_gate():
+        """Fail-closed gate for audit ingest rate limiting (#831).
+
+        Registered before start_span so a gated (503) request never
+        opens a span. No-op unless AUDIT_TRACING_FAIL_CLOSED=1.
+        """
+        return tracing.check_rate_gate()
+
     @app.before_request
     def start_span():
         """Start a root span for each incoming request.

campus/audit/middleware/tracing.py

@@ -221,6 +221,75 @@ def is_static_request() -> bool:
     return flask.request.path == "/favicon.ico"
 
 
+def check_rate_gate() -> flask.Response | None:
+    """Fail-closed gate for the audit ingest rate limit (#831).
+
+    When AUDIT_TRACING_FAIL_CLOSED=1 (opt-in per deployment), requests
+    whose identity bucket is tripped in the SDK's ingest circuit
+    breaker (campus_python.audit.ratelimit — fed by this client's 429s)
+    are short-circuited with 503 + Retry-After before any view runs.
+    Off by default: producers run fail-open until flipped, per the
+    ratified rollout (design comment on campus#538).
+
+    The gate is per-identity when identity is already resolvable at
+    request entry (app-level hooks run before blueprint-level ones, so
+    e.g. campus.auth's session resolution has run); it degrades to the
+    producer's learned apikey= fallback bucket (coarse) otherwise.
+
+    Registered BEFORE start_span so a gated request never opens a span
+    (end_span no-ops without span state — no wasted ingest against a
+    bucket that audit is currently rejecting anyway).
+    """
+    from campus.common import env
+
+    if env.get("AUDIT_TRACING_FAIL_CLOSED") != "1":
+        return None
+
+    # importlib with a string name: the module is optional (a stale
+    # campus_python predating #831 lacks it) and opaque resolution
+    # keeps type checkers happy against older SDK installs.
+    import importlib
+
+    try:
+        ratelimit = importlib.import_module("campus_python.audit.ratelimit")
+    except ImportError:
+        logger.error(
+            "AUDIT_TRACING_FAIL_CLOSED=1 but campus_python lacks "
+            "audit.ratelimit (stale SDK); rate gate disabled"
+        )
+        return None
+
+    retry_after = ratelimit.breaker.check_request(
+        client_id=(
+            _identity_id(getattr(flask.g, "current_client", None))
+            or getattr(flask.g, "client_id", None)
+        ),
+        user_id=(
+            _identity_id(getattr(flask.g, "current_user", None))
+            or getattr(flask.g, "user_id", None)
+        ),
+    )
+    if retry_after is None:
+        return None
+
+    logger.warning(
+        "audit ingest breaker tripped; failing request closed "
+        "(retry_after=%ds)",
+        retry_after,
+    )
+    response = flask.jsonify({
+        "error": {
+            "code": "RATE_LIMITED",
+            "message": f"Rate limit exceeded; retry after {retry_after}s",
+            "details": {},
+            "request_id": None,
+        }
+    })
+    response.status_code = 503
+    response.headers["Retry-After"] = str(max(1, int(retry_after)))
+    return response
+
+
 def start_span() -> None:
     """Start a span for the incoming request.
 

tests/unit/audit/test_rate_gate.py

@@ -0,0 +1,178 @@
+"""Unit tests for the audit ingest rate gate + breaker feed (#831).
+
+The producer side of campus.audit's per-identity ingest rate limit
+(campus PR #835): the local AuditClient's traces.new feeds the SDK
+circuit breaker on every ingest response, and the tracing middleware's
+fail-closed gate (AUDIT_TRACING_FAIL_CLOSED=1) 503s matching requests
+at entry.
+"""
+
+import os
+import unittest
+from unittest import mock
+
+import flask
+from campus_python.audit import ratelimit
+
+
+def _set_env(test: unittest.TestCase, name: str, value: str) -> None:
+    """Set an env var for the duration of a test."""
+    os.environ[name] = value
+    test.addCleanup(os.environ.pop, name, None)
+
+
+def _429_body(bucket: str) -> dict:
+    """Error envelope as emitted by campus.audit #835."""
+    return {"error": {"code": "RATE_LIMITED", "details": {"bucket": bucket}}}
+
+
+def _make_app(**identity) -> flask.Flask:
+    """Flask app with the tracing middleware and a stub identity hook.
+
+    The identity stub is registered BEFORE middleware.init_app so the
+    gate (a later before_request hook) observes it — mirroring how
+    app-level hooks registered earlier run first.
+    """
+    app = flask.Flask(__name__)
+    app.config["TESTING"] = True
+
+    @app.before_request
+    def stub_identity():
+        for key, value in identity.items():
+            setattr(flask.g, key, value)
+
+    from campus.audit.middleware import init_app as init_tracing
+
+    init_tracing(app)
+
+    @app.route("/")
+    def index():
+        return "ok"
+
+    return app
+
+
+class TestRateGate(unittest.TestCase):
+    """The fail-closed gate short-circuits tripped buckets at entry."""
+
+    def setUp(self):
+        ratelimit.breaker.reset()
+        self.addCleanup(ratelimit.breaker.reset)
+
+    def test_flag_off_is_fail_open_even_when_tripped(self):
+        _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "")
+        ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1"))
+        client = _make_app().test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 200)
+
+    def test_tripped_identity_bucket_returns_503(self):
+        _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1")
+        ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1"))
+        client = _make_app(user_id="u1").test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 503)
+        self.assertIn("Retry-After", response.headers)
+        self.assertGreaterEqual(int(response.headers["Retry-After"]), 1)
+        data = response.get_json()
+        self.assertEqual(data["error"]["code"], "RATE_LIMITED")
+
+    def test_other_identities_unaffected(self):
+        _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1")
+        ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1"))
+        client = _make_app(user_id="u2").test_client()
+        self.assertEqual(client.get("/").status_code, 200)
+
+    def test_identityless_uses_learned_fallback_bucket(self):
+        _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1")
+        ratelimit.breaker.observe(
+            429, {"Retry-After": "60"}, _429_body("apikey=k-123")
+        )
+        client = _make_app().test_client()
+        self.assertEqual(client.get("/").status_code, 503)
+
+    def test_identityless_before_learning_uses_wildcard(self):
+        _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1")
+        ratelimit.breaker.observe(
+            429, {"Retry-After": "60"}, _429_body(ratelimit.WILDCARD)
+        )
+        client = _make_app().test_client()
+        self.assertEqual(client.get("/").status_code, 503)
+
+    def test_availability_error_does_not_trip_gate(self):
+        _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1")
+        # 503 from audit is an availability error: neutral.
+        ratelimit.breaker.observe(503, None, None)
+        client = _make_app().test_client()
+        self.assertEqual(client.get("/").status_code, 200)
+
+    def test_verifying_state_gate_is_open(self):
+        _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1")
+        ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1"))
+        # Force the cooldown into the past (verifying): gate opens...
+        ratelimit.breaker._tripped["user=u1"]["until"] -= 10_000
+        client = _make_app(user_id="u1").test_client()
+        self.assertEqual(client.get("/").status_code, 200)
+        # ...and a 2xx ingest fully clears the trip.
+        ratelimit.breaker.observe(201, None, None)
+        self.assertEqual(ratelimit.breaker._tripped, {})
+
+
+class TestClientObserveHook(unittest.TestCase):
+    """AuditClient traces.new feeds the SDK breaker (#831)."""
+
+    def setUp(self):
+        ratelimit.breaker.reset()
+        self.addCleanup(ratelimit.breaker.reset)
+
+    def _traces_with_response(self, status: int, body: dict, raises: bool):
+        from campus.audit.client.v1.traces import Traces
+
+        response = mock.Mock()
+        response.status_code = status
+        response.headers = {"Retry-After": "30"}
+        response.json.return_value = body
+        if raises:
+            response.raise_for_status.side_effect = RuntimeError(status)
+        client = mock.Mock()
+        client.post.return_value = response
+        return Traces(client=client, root=mock.Mock()), response
+
+    def test_429_trips_breaker_and_still_raises(self):
+        traces, _ = self._traces_with_response(
+            429, _429_body("client=c1;user=u1"), raises=True
+        )
+        with self.assertRaises(RuntimeError):
+            traces.new({"span_id": "s1"})
+        self.assertIsNotNone(
+            ratelimit.breaker.check_request(client_id="c1", user_id="u1")
+        )
+
+    def test_201_does_not_trip(self):
+        traces, _ = self._traces_with_response(
+            201, {"created": ["s1"]}, raises=False
+        )
+        traces.new({"span_id": "s1"})
+        self.assertEqual(ratelimit.breaker._tripped, {})
+
+    def test_stale_sdk_degrades_silently(self):
+        """A missing campus_python.audit.ratelimit must not break ingest."""
+        traces, response = self._traces_with_response(
+            201, {"created": ["s1"]}, raises=False
+        )
+        import importlib
+
+        real_import_module = importlib.import_module
+
+        def blocked_import(name, *args, **kwargs):
+            if name == "campus_python.audit.ratelimit":
+                raise ImportError("simulated stale SDK")
+            return real_import_module(name, *args, **kwargs)
+
+        with mock.patch("importlib.import_module", side_effect=blocked_import):
+            traces.new({"span_id": "s1"})  # no raise
+        self.assertEqual(response.status_code, 201)
+
+
+if __name__ == "__main__":
+    unittest.main()