From c1e6c158664d9f23392cc35051b64922f0e347bb Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 09:29:35 +0200 Subject: [PATCH] Partition durable events by tenant entitlement --- src/govoplan_audit/backend/manifest.py | 2 +- src/govoplan_audit/backend/outbox.py | 51 ++++++++++++++--- tests/test_audit_delivery.py | 77 ++++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 8 deletions(-) diff --git a/src/govoplan_audit/backend/manifest.py b/src/govoplan_audit/backend/manifest.py index a2b95d6..a4074c1 100644 --- a/src/govoplan_audit/backend/manifest.py +++ b/src/govoplan_audit/backend/manifest.py @@ -78,7 +78,7 @@ manifest = ModuleManifest( id="audit.recording-retention-and-outbox", title="Operate audit recording and event delivery", summary="Audit owns durable audit records, retention operations, and the transactional platform-event outbox.", - body="Modules record bounded audit facts through the Audit capability. Governed platform events are committed to the outbox with retry and delivery metadata so a failed consumer does not erase the originating transaction. Retention and destructive retirement must preserve the configured evidence and recovery guarantees.", + body="Modules record bounded audit facts through the Audit capability. Governed platform events are committed to the outbox with retry and delivery metadata so a failed consumer does not erase the originating transaction. Worker dispatch is partitioned by tenant entitlement; an unavailable consumer retains its durable delivery and records an operator-required outcome instead of acknowledging the event. Retention and destructive retirement must preserve the configured evidence and recovery guarantees.", documentation_types=("admin",), audience=("auditor", "security_officer", "operator"), related_modules=("policy", "ops"), diff --git a/src/govoplan_audit/backend/outbox.py b/src/govoplan_audit/backend/outbox.py index 2b20e50..d00fce3 100644 --- a/src/govoplan_audit/backend/outbox.py +++ b/src/govoplan_audit/backend/outbox.py @@ -61,15 +61,21 @@ class SqlAuditOutbox: self, session: object, *, + tenant_id: str | None = None, + tenantless_only: bool = False, consumers: Sequence[DurableEventConsumer] = (), observer: EventDispatcher | None = publish_platform_event, dispatcher: EventDispatcher | None = None, limit: int = 100, ) -> dict[str, int]: + if tenant_id is not None and tenantless_only: + raise ValueError( + "Tenant and tenantless event filters are mutually exclusive" + ) db = _session(session) now = datetime.now(timezone.utc) consumers_by_id = _consumer_map(consumers) - rows = ( + query = ( db.query(AuditOutboxEvent) .filter( AuditOutboxEvent.status.in_( @@ -77,7 +83,20 @@ class SqlAuditOutbox: ), or_(AuditOutboxEvent.next_attempt_at.is_(None), AuditOutboxEvent.next_attempt_at <= now), ) - .order_by(AuditOutboxEvent.created_at.asc(), AuditOutboxEvent.id.asc()) + ) + if tenant_id: + query = query.filter( + AuditOutboxEvent.payload["tenant"]["id"].as_string() + == tenant_id + ) + elif tenantless_only: + query = query.filter( + AuditOutboxEvent.payload["tenant"]["id"] + .as_string() + .is_(None) + ) + rows = ( + query.order_by(AuditOutboxEvent.created_at.asc(), AuditOutboxEvent.id.asc()) .with_for_update(skip_locked=True) .limit(max(1, min(int(limit), 500))) .all() @@ -179,18 +198,36 @@ class SqlAuditOutbox: self, session: object, *, + tenant_id: str | None = None, + tenantless_only: bool = False, before: datetime, limit: int = 500, ) -> dict[str, int]: + if tenant_id is not None and tenantless_only: + raise ValueError( + "Tenant and tenantless event filters are mutually exclusive" + ) db = _session(session) + clauses = [ + AuditOutboxEvent.status == "dispatched", + AuditOutboxEvent.dispatched_at.is_not(None), + AuditOutboxEvent.dispatched_at < before, + ] + if tenant_id: + clauses.append( + AuditOutboxEvent.payload["tenant"]["id"].as_string() + == tenant_id + ) + elif tenantless_only: + clauses.append( + AuditOutboxEvent.payload["tenant"]["id"] + .as_string() + .is_(None) + ) ids = tuple( db.scalars( select(AuditOutboxEvent.id) - .where( - AuditOutboxEvent.status == "dispatched", - AuditOutboxEvent.dispatched_at.is_not(None), - AuditOutboxEvent.dispatched_at < before, - ) + .where(*clauses) .order_by( AuditOutboxEvent.dispatched_at, AuditOutboxEvent.id, diff --git a/tests/test_audit_delivery.py b/tests/test_audit_delivery.py index 1e1ef57..8904888 100644 --- a/tests/test_audit_delivery.py +++ b/tests/test_audit_delivery.py @@ -15,6 +15,7 @@ from govoplan_audit.backend.outbox import SqlAuditOutbox from govoplan_core.core.events import ( DurableEventConsumer, EventActorRef, + EventTenantRef, PlatformEvent, ) from govoplan_core.core.institutional import ( @@ -114,6 +115,82 @@ class AuditOutboxTests(unittest.TestCase): delivery.delivery_key, ) + def test_dispatch_partitions_pending_events_by_tenant(self) -> None: + Session = self._database() + outbox = SqlAuditOutbox() + seen: list[str] = [] + + with Session() as session: + first = outbox.enqueue( + session, + PlatformEvent( + type="files.file.created", + module_id="files", + tenant=EventTenantRef(id="tenant-1"), + ), + ) + second = outbox.enqueue( + session, + PlatformEvent( + type="files.file.created", + module_id="files", + tenant=EventTenantRef(id="tenant-2"), + ), + ) + counts = outbox.dispatch_pending( + session, + tenant_id="tenant-1", + consumers=( + DurableEventConsumer( + consumer_id="tests.tenant-filter.v1", + handler=lambda event, _key: seen.append( + event.tenant.id if event.tenant else "system" + ), + ), + ), + observer=None, + ) + + self.assertEqual(1, counts["selected"]) + self.assertEqual(["tenant-1"], seen) + self.assertEqual("dispatched", first.status) + self.assertEqual("pending", second.status) + + def test_dispatch_can_select_only_tenantless_system_events(self) -> None: + Session = self._database() + outbox = SqlAuditOutbox() + seen: list[str] = [] + + with Session() as session: + system = outbox.enqueue( + session, + PlatformEvent(type="system.ready", module_id="core"), + ) + tenant = outbox.enqueue( + session, + PlatformEvent( + type="tenant.ready", + module_id="tenancy", + tenant=EventTenantRef(id="tenant-1"), + ), + ) + counts = outbox.dispatch_pending( + session, + tenantless_only=True, + consumers=( + DurableEventConsumer( + consumer_id="tests.system-filter.v1", + handler=lambda event, _key: seen.append(event.type), + ), + ), + observer=None, + ) + + self.assertEqual(1, counts["selected"]) + self.assertEqual(["system.ready"], seen) + self.assertEqual("dispatched", system.status) + self.assertEqual("pending", tenant.status) + def test_outbox_preserves_institutional_context(self) -> None: Session = self._database() outbox = SqlAuditOutbox()