fix(buffer): persist to Redis off the request path, bounded and non-recursive

guard_agent/_buffer_lifecycle.py

@@ -18,6 +18,7 @@ class BufferLifecycleMixin:
     _flush_semaphore: asyncio.Semaphore | None
     _running: bool
     _inflight_flush_tasks: set[asyncio.Task[None]]
+    _inflight_persist_tasks: set[asyncio.Task[None]]
 
     async def start_auto_flush(self) -> None:
         if self._flush_task and not self._flush_task.done():
@@ -41,6 +42,9 @@ async def stop_auto_flush(self) -> None:
         if self._inflight_flush_tasks:
             await asyncio.gather(*self._inflight_flush_tasks, return_exceptions=True)
         self._inflight_flush_tasks.clear()
+        if self._inflight_persist_tasks:
+            await asyncio.gather(*self._inflight_persist_tasks, return_exceptions=True)
+        self._inflight_persist_tasks.clear()
         self._flush_semaphore = None
 
     async def stop(self) -> None:

guard_agent/_buffer_overflow.py

@@ -85,24 +85,28 @@ def _forget_oldest_event_key(self) -> str | None:
         if not self.event_buffer:
             return None
         oldest = self.event_buffer[0]
+        self._pending_event_persists.pop(id(oldest), None)
         return self._event_redis_keys.pop(id(oldest), None)
 
     def _forget_oldest_metric_key(self) -> str | None:
         if not self.metric_buffer:
             return None
         oldest = self.metric_buffer[0]
+        self._pending_metric_persists.pop(id(oldest), None)
         return self._metric_redis_keys.pop(id(oldest), None)
 
     def _forget_newest_event_key(self) -> str | None:
         if not self.event_buffer:
             return None
         newest = self.event_buffer[-1]
+        self._pending_event_persists.pop(id(newest), None)
         return self._event_redis_keys.pop(id(newest), None)
 
     def _forget_newest_metric_key(self) -> str | None:
         if not self.metric_buffer:
             return None
         newest = self.metric_buffer[-1]
+        self._pending_metric_persists.pop(id(newest), None)
         return self._metric_redis_keys.pop(id(newest), None)
 
     def _is_event_buffer_full(self) -> bool:

guard_agent/_buffer_queue.py

@@ -1,14 +1,55 @@
 import asyncio
 import time
+from collections.abc import Callable, Coroutine
+from typing import Any, TypeVar
 
 from guard_agent._buffer_lifecycle import BufferLifecycleMixin
 from guard_agent._buffer_overflow import BLOCK_POLICY_POLL_INTERVAL, BufferOverflowMixin
+from guard_agent._buffer_redis import _PERSISTING
 from guard_agent.models import SecurityEvent, SecurityMetric
 
+T = TypeVar("T")
+
 
 class BufferQueueMixin(BufferOverflowMixin, BufferLifecycleMixin):
     events_flushed: int
     metrics_flushed: int
+    redis_persist_failures: int
+    _persist_overflow_logged: bool
+    _pending_event_persists: dict[int, bool]
+    _pending_metric_persists: dict[int, bool]
+
+    def _spawn_persist(
+        self,
+        persist: Callable[[T], Coroutine[Any, Any, None]],
+        item: T,
+        pending: dict[int, bool],
+    ) -> None:
+        """Run a Redis persist as a tracked background task so the request
+        path never waits on Redis; stop_auto_flush drains it. Nothing is
+        spawned without a Redis handler, or while a persist is in flight in
+        this context (a Redis failure reporting an event must not recurse)."""
+        if not self.redis_handler or _PERSISTING.get():
+            return
+        if len(self._inflight_persist_tasks) >= self.config.buffer_size:
+            self.redis_persist_failures += 1
+            if not self._persist_overflow_logged:
+                self._persist_overflow_logged = True
+                self.logger.warning(
+                    "Redis persist backlog at buffer_size="
+                    f"{self.config.buffer_size}; buffering in memory only "
+                    "until it drains"
+                )
+            return
+        pending[id(item)] = True
+        task = asyncio.create_task(persist(item))
+        self._inflight_persist_tasks.add(task)
+        task.add_done_callback(self._persist_done)
+
+    def _persist_done(self, task: "asyncio.Task[None]") -> None:
+        self._inflight_persist_tasks.discard(task)
+        if not self._inflight_persist_tasks:
+            self._persist_overflow_logged = False
 
     async def add_event(self, event: SecurityEvent) -> None:
         """Add security event to buffer honoring the configured overflow
@@ -23,15 +64,8 @@ async def add_event(self, event: SecurityEvent) -> None:
             while True:
                 must_wait = await self._resolve_event_overflow()
                 if not must_wait:
-                    try:
-                        self.event_buffer.append(event)
-                        self.events_buffered += 1
-                        if self.redis_handler:
-                            key = await self._persist_event_to_redis(event)
-                            if key is not None:
-                                self._event_redis_keys[id(event)] = key
-                    except Exception as e:
-                        self.logger.error(f"Failed to buffer event: {str(e)}")
+                    self.event_buffer.append(event)
+                    self.events_buffered += 1
                     break
                 try:
                     await asyncio.wait_for(
@@ -40,6 +74,10 @@ async def add_event(self, event: SecurityEvent) -> None:
                 except asyncio.TimeoutError:
                     pass
 
+        self._spawn_persist(
+            self._persist_and_track_event, event, self._pending_event_persists
+        )
+
         if (
             len(self.event_buffer)
             >= self.config.buffer_size * self.config.high_watermark_ratio
@@ -58,15 +96,8 @@ async def add_metric(self, metric: SecurityMetric) -> None:
             while True:
                 must_wait = await self._resolve_metric_overflow()
                 if not must_wait:
-                    try:
-                        self.metric_buffer.append(metric)
-                        self.metrics_buffered += 1
-                        if self.redis_handler:
-                            key = await self._persist_metric_to_redis(metric)
-                            if key is not None:
-                                self._metric_redis_keys[id(metric)] = key
-                    except Exception as e:
-                        self.logger.error(f"Failed to buffer metric: {str(e)}")
+                    self.metric_buffer.append(metric)
+                    self.metrics_buffered += 1
                     break
                 try:
                     await asyncio.wait_for(
@@ -75,6 +106,10 @@ async def add_metric(self, metric: SecurityMetric) -> None:
                 except asyncio.TimeoutError:
                     pass
 
+        self._spawn_persist(
+            self._persist_and_track_metric, metric, self._pending_metric_persists
+        )
+
         if (
             len(self.metric_buffer)
             >= self.config.buffer_size * self.config.high_watermark_ratio
@@ -104,6 +139,9 @@ async def flush_events_with_keys(
         condition = self._get_event_condition()
         async with condition:
             events = list(self.event_buffer)
+            for item in events:
+                if id(item) in self._pending_event_persists:
+                    self._pending_event_persists[id(item)] = False
             keys = [self._event_redis_keys.pop(id(event), "") for event in events]
             self.event_buffer.clear()
             self.events_flushed += len(events)
@@ -122,6 +160,9 @@ async def flush_metrics_with_keys(
         condition = self._get_metric_condition()
         async with condition:
             metrics = list(self.metric_buffer)
+            for item in metrics:
+                if id(item) in self._pending_metric_persists:
+                    self._pending_metric_persists[id(item)] = False
             keys = [self._metric_redis_keys.pop(id(metric), "") for metric in metrics]
             self.metric_buffer.clear()
             self.metrics_flushed += len(metrics)
@@ -146,6 +187,8 @@ async def requeue_events_in_memory(
                     if evicted_key:
                         evicted_keys.append(evicted_key)
                 self.event_buffer.appendleft(event)
+                if id(event) in self._pending_event_persists:
+                    self._pending_event_persists[id(event)] = True
                 if key:
                     self._event_redis_keys[id(event)] = key
         return evicted_keys
@@ -166,6 +209,8 @@ async def requeue_metrics_in_memory(
                     if evicted_key:
                         evicted_keys.append(evicted_key)
                 self.metric_buffer.appendleft(metric)
+                if id(metric) in self._pending_metric_persists:
+                    self._pending_metric_persists[id(metric)] = True
                 if key:
                     self._metric_redis_keys[id(metric)] = key
         return evicted_keys
@@ -177,12 +222,14 @@ async def clear_buffer(self) -> None:
         async with event_condition:
             self.event_buffer.clear()
             self._event_redis_keys.clear()
+            self._pending_event_persists.clear()
             event_condition.notify_all()
 
         metric_condition = self._get_metric_condition()
         async with metric_condition:
             self.metric_buffer.clear()
             self._metric_redis_keys.clear()
+            self._pending_metric_persists.clear()
             metric_condition.notify_all()
 
         if self.redis_handler:

guard_agent/_buffer_redis.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextvars
 import logging
 import time
 import uuid
@@ -8,6 +9,10 @@
 from guard_agent.protocols import RedisHandlerProtocol
 from guard_agent.utils import safe_json_deserialize, safe_json_serialize
 
+_PERSISTING: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "guard_agent_redis_persisting", default=False
+)
+
 
 class BufferRedisMixin:
     logger: logging.Logger
@@ -19,6 +24,8 @@ class BufferRedisMixin:
     redis_persist_failures: int
     _event_redis_keys: dict[int, str]
     _metric_redis_keys: dict[int, str]
+    _pending_event_persists: dict[int, bool]
+    _pending_metric_persists: dict[int, bool]
 
     def _get_event_condition(self) -> asyncio.Condition:
         raise NotImplementedError
@@ -83,6 +90,55 @@ async def _persist_metric_to_redis(self, metric: SecurityMetric) -> str | None:
             self.logger.warning(f"Failed to persist metric to Redis: {str(e)}")
             return None
 
+    async def _persist_and_track_event(self, event: SecurityEvent) -> None:
+        """Persist an already-buffered event, outside the buffer lock.
+
+        Skipped while a persist is in flight in this context, so a Redis
+        failure that emits an event cannot recurse into Redis. If the event
+        left the buffer meanwhile (flushed), the new key is deleted instead
+        of tracked, so it neither orphans nor replays after a restart. An
+        event already gone before the task starts is never written.
+        """
+        if not self._pending_event_persists.get(id(event)):
+            self._pending_event_persists.pop(id(event), None)
+            return
+        token = _PERSISTING.set(True)
+        try:
+            key = await self._persist_event_to_redis(event)
+        except Exception as e:
+            self.logger.error(f"Failed to buffer event: {str(e)}")
+            key = None
+        finally:
+            _PERSISTING.reset(token)
+            still_buffered = self._pending_event_persists.pop(id(event), False)
+        if key is None:
+            return
+        if still_buffered:
+            self._event_redis_keys[id(event)] = key
+        else:
+            await self.confirm_event_redis_keys([key])
+
+    async def _persist_and_track_metric(self, metric: SecurityMetric) -> None:
+        """Metric twin of _persist_and_track_event."""
+        if not self._pending_metric_persists.get(id(metric)):
+            self._pending_metric_persists.pop(id(metric), None)
+            return
+        token = _PERSISTING.set(True)
+        try:
+            key = await self._persist_metric_to_redis(metric)
+        except Exception as e:
+            self.logger.error(f"Failed to buffer metric: {str(e)}")
+            key = None
+        finally:
+            _PERSISTING.reset(token)
+            still_buffered = self._pending_metric_persists.pop(id(metric), False)
+        if key is None:
+            return
+        if still_buffered:
+            self._metric_redis_keys[id(metric)] = key
+        else:
+            await self.confirm_metric_redis_keys([key])
+
     async def _load_from_redis(self) -> None:
         """Load persisted events/metrics from Redis on startup; track keys.
 

guard_agent/_transport_lifecycle.py

@@ -74,7 +74,8 @@ def _init_encryption(self) -> None:
             raise
         except Exception as exc:
             raise EncryptionConfigError(
-                "Encryption round-trip failed at startup; refusing plaintext fallback"
+                "Encryption round-trip failed at startup; refusing plaintext "
+                f"fallback (cause: {exc})"
             ) from exc
 
     async def initialize(self) -> None:

guard_agent/buffer.py

@@ -30,6 +30,10 @@ def __init__(
         self._flush_semaphore: asyncio.Semaphore | None = None
         self._running = False
         self._inflight_flush_tasks: set[asyncio.Task[None]] = set()
+        self._inflight_persist_tasks: set[asyncio.Task[None]] = set()
+        self._persist_overflow_logged = False
+        self._pending_event_persists: dict[int, bool] = {}
+        self._pending_metric_persists: dict[int, bool] = {}
 
         self._event_redis_keys: dict[int, str] = {}
         self._metric_redis_keys: dict[int, str] = {}

guard_agent/encryption.py

@@ -69,7 +69,8 @@ def __init__(self, project_key: str) -> None:
             raise EncryptionError("Project key cannot be empty")
 
         try:
-            key_bytes = base64.urlsafe_b64decode(project_key.encode())
+            padded = project_key.strip() + "=" * (-len(project_key.strip()) % 4)
+            key_bytes = base64.urlsafe_b64decode(padded.encode())
 
             if len(key_bytes) != self.KEY_SIZE:
                 raise EncryptionError(

tests/conftest.py

@@ -24,6 +24,7 @@
     django.setup()
 
 from datetime import datetime, timezone
+from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -116,3 +117,28 @@ def security_metric() -> SecurityMetric:
         metric_type="request_count",
         value=1.0,
     )
+
+
+@pytest.fixture(autouse=True)
+def settle_redis_persists(
+    request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Persist is a background task in production; in tests that assert on
+    Redis state right after an add, wait for it. Tests that exercise the
+    background behaviour live in test_buffer_redis_reentrancy.py."""
+    if request.module.__name__.endswith("test_buffer_redis_reentrancy"):
+        return
+    import asyncio
+
+    for name in ("add_event", "add_metric"):
+        original = getattr(EventBuffer, name)
+
+        def make(orig: Any) -> Any:
+            async def wrapper(self: EventBuffer, item: Any) -> None:
+                await orig(self, item)
+                if self._inflight_persist_tasks:
+                    await asyncio.gather(*self._inflight_persist_tasks)
+
+            return wrapper
+
+        monkeypatch.setattr(EventBuffer, name, make(original))