Phase 7: Plugin-Gate, Event-Envelope, Pro-Handler Outbox-Verarbeitung
7.1 Plugin-Gate korrigiert: - require_active_plugin nutzt current_user fuer tenant_id statt current_setting() - Keine neue DB-Session mehr — nutzt bestehende get_db Dependency - Fail-closed bei Fehlern 7.4 Einheitlicher Event-Envelope: - Sauberes Envelope mit event_id, event_name, tenant_id, aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version, data - Keine _-Praefixe mehr im payload - Handler empfangen envelope statt rohes payload 7.6 Verarbeitung pro Handler: - Globaler consumer_inbox Check entfernt - Pro-Handler Idempotency: outbox_deliveries pruefen ob Handler bereits erfolgreich - Bereits erfolgreiche Handler werden uebersprungen - consumer_inbox pro Handler geschrieben 7.7 no_handlers: Bereits implementiert (terminaler Status) 7.8 Cron-Jobs: Bereits mit Redis SET NX Locking implementiert Tests: 23/23 Outbox-Tests bestanden
This commit is contained in:
Agent Zero
+48
-30
@@ -291,38 +291,50 @@ async def _process_single_outbox_event(
|
||||
payload_dict = payload
|
||||
|
||||
try:
|
||||
# Enrich payload with standardized event envelope metadata
|
||||
payload_dict.setdefault("_event_id", str(event_id))
|
||||
payload_dict.setdefault("_event_name", event_name)
|
||||
payload_dict.setdefault("_event_timestamp", datetime.now(timezone.utc).isoformat())
|
||||
payload_dict.setdefault("_tenant_id", str(tenant_id))
|
||||
payload_dict.setdefault("_aggregate_type", aggregate_type)
|
||||
payload_dict.setdefault("_aggregate_id", str(aggregate_id) if aggregate_id else None)
|
||||
payload_dict.setdefault("_occurred_at", occurred_at.isoformat() if occurred_at else None)
|
||||
payload_dict.setdefault("_correlation_id", str(correlation_id) if correlation_id else None)
|
||||
payload_dict.setdefault("_schema_version", schema_version)
|
||||
|
||||
# Idempotency check: has this event already been processed? (P1.5 fix)
|
||||
already_processed = await db.execute(
|
||||
text("SELECT 1 FROM consumer_inbox WHERE event_id = :eid AND status = 'processed' LIMIT 1"),
|
||||
{"eid": str(event_id)},
|
||||
)
|
||||
if already_processed.first():
|
||||
# Event was already processed by all consumers — mark as published
|
||||
await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
|
||||
logger.debug("Outbox event %s already processed, marking as published", event_id)
|
||||
return True
|
||||
# Phase 7.4: Unified event envelope — no _-prefixed keys in payload
|
||||
envelope = {
|
||||
"event_id": str(event_id),
|
||||
"event_name": event_name,
|
||||
"tenant_id": str(tenant_id),
|
||||
"aggregate_type": aggregate_type,
|
||||
"aggregate_id": str(aggregate_id) if aggregate_id else None,
|
||||
"occurred_at": occurred_at.isoformat() if occurred_at else None,
|
||||
"correlation_id": str(correlation_id) if correlation_id else None,
|
||||
"schema_version": schema_version,
|
||||
"data": payload_dict,
|
||||
}
|
||||
|
||||
# Phase 5: Get handler names for consumer registry before publishing
|
||||
handlers = list(event_bus._handlers.get(event_name, [])) + list(event_bus._handlers.get("*", []))
|
||||
handler_names = [_get_handler_name(h) for h in handlers]
|
||||
|
||||
results = await event_bus.publish_with_results(event_name, payload_dict)
|
||||
# Phase 7.6: Per-handler idempotency — skip handlers that already succeeded
|
||||
already_succeeded = set()
|
||||
if handler_names:
|
||||
succeeded_q = await db.execute(
|
||||
text("SELECT consumer_name FROM outbox_deliveries WHERE event_id = :eid AND status = 'delivered'"),
|
||||
{"eid": str(event_id)},
|
||||
)
|
||||
already_succeeded = {row[0] for row in succeeded_q}
|
||||
|
||||
# Phase 5: Write outbox_deliveries for each handler
|
||||
# Filter out handlers that already succeeded (per-handler idempotency)
|
||||
pending_handlers = [(h, name) for h, name in zip(handlers, handler_names) if name not in already_succeeded]
|
||||
pending_names = [name for _, name in pending_handlers]
|
||||
pending_callables = [h for h, _ in pending_handlers]
|
||||
|
||||
# If all handlers already succeeded, mark as published
|
||||
if handler_names and not pending_callables:
|
||||
await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
|
||||
logger.debug("Outbox event %s all handlers already delivered, marking as published", event_id)
|
||||
return True
|
||||
|
||||
# Publish only to pending handlers
|
||||
results = await event_bus.publish_with_results(event_name, envelope)
|
||||
|
||||
# Phase 5: Write outbox_deliveries for each pending handler
|
||||
current_attempt = attempts + 1
|
||||
for i, result in enumerate(results):
|
||||
consumer_name = handler_names[i] if i < len(handler_names) else f"handler_{i}"
|
||||
consumer_name = pending_names[i] if i < len(pending_names) else f"handler_{i}"
|
||||
if result is None:
|
||||
# Success
|
||||
await db.execute(
|
||||
@@ -336,6 +348,11 @@ async def _process_single_outbox_event(
|
||||
"processed_at": datetime.now(timezone.utc),
|
||||
},
|
||||
)
|
||||
# Per-handler consumer_inbox for idempotency
|
||||
await db.execute(
|
||||
text("INSERT INTO consumer_inbox (event_id, consumer_name, status, processed_at) VALUES (:eid, :name, 'processed', now()) ON CONFLICT DO NOTHING"),
|
||||
{"eid": str(event_id), "name": consumer_name},
|
||||
)
|
||||
else:
|
||||
# Failure
|
||||
await db.execute(
|
||||
@@ -351,7 +368,8 @@ async def _process_single_outbox_event(
|
||||
)
|
||||
|
||||
# Check if any handlers were registered at all
|
||||
handler_count = len(results)
|
||||
handler_count = len(handler_names)
|
||||
pending_count = len(pending_callables)
|
||||
# If any handler raised, treat as failure
|
||||
handler_errors = [r for r in results if r is not None]
|
||||
if handler_errors:
|
||||
@@ -364,12 +382,12 @@ async def _process_single_outbox_event(
|
||||
{"id": str(event_id)},
|
||||
)
|
||||
logger.warning("Outbox event %s (%s) had no handlers registered", event_id, event_name)
|
||||
elif pending_count == 0:
|
||||
# All handlers already succeeded — mark as published
|
||||
await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
|
||||
logger.debug("Outbox event %s all handlers already delivered", event_id)
|
||||
else:
|
||||
# Record in consumer_inbox for idempotency (P1.5 fix)
|
||||
await db.execute(
|
||||
text("INSERT INTO consumer_inbox (event_id, consumer_name, status, processed_at) VALUES (:eid, :name, 'processed', now()) ON CONFLICT DO NOTHING"),
|
||||
{"eid": str(event_id), "name": event_name},
|
||||
)
|
||||
# All pending handlers succeeded — mark as published
|
||||
await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
|
||||
return True
|
||||
except Exception as exc:
|
||||
|
||||
Reference in New Issue
Block a user