fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(outbox): make the trigger fan-out connection-scoped

Browse Source 
`OutboxEventEmitter` resolved matching triggers and inserted outbox rows
through `Arc<dyn TriggerRepo>` / `Arc<dyn OutboxRepo>`, i.e. against the
pool — each query on whatever connection it happened to get. That makes a
transactional outbox impossible: the data write and the outbox rows can
never share a transaction, so a crash (or an outbox error) between them
silently loses the trigger event.

Take the emitter down to a connection instead of a pool:

  * `outbox_repo::insert_on(exec, row)` and `trigger_repo::list_matching_on`
    / `list_matching_shared_on(exec, ...)` are generic over `PgExecutor`, so
    the same SQL serves a pooled connection or a `&mut *tx`. The repo trait
    methods delegate to them — the SQL keeps exactly one home.
  * `emit_on` / `emit_shared_on` take a `&mut PgConnection` and run the whole
    fan-out on it. The `ServiceEventEmitter` impl acquires one pooled
    connection and calls them, so behaviour is unchanged today; a caller
    holding a transaction can now pass `&mut *tx` and have the outbox rows
    commit with the write.
  * `OutboxEventEmitter::new` takes the `PgPool` directly (it was only ever
    constructed once, in the host wiring).

Also collapses the three copy-pasted per-app match queries (kv/docs/files
differ only in the `kind` discriminator and detail table) and the three
`emit_*` bodies into one `plan()` + one match fn, so the suppression
anti-join, the chain walk, and the empty-ops-means-any-op semantic each
exist once rather than three times.

No behaviour change — pure refactor. It is the seam the transactional
write lands on next.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-14 19:20:41 +02:00
parent 155c5471b7
commit a2360a9464
4 changed files with 351 additions and 505 deletions
Download Patch File Download Diff File Expand all files Collapse all files

473
crates/manager-core/src/outbox_event_emitter.rs
View File

@@ -1,139 +1,135 @@
//! `OutboxEventEmitter` — the real `ServiceEventEmitter` that replaces
//! v1.1.0's `NoopEventEmitter` once the triggers framework lands.
//! `OutboxEventEmitter` — the real `ServiceEventEmitter` behind every stateful
//! service. On each collection mutation (KV / docs / files) it looks up the
//! triggers the event matches and writes one outbox row per match; the
//! dispatcher reads them back out-of-band.
//!
//! On each `emit` (a KV mutation, future doc/file/pubsub event, etc.):
//! 1. Look up matching triggers for the event's (app_id, source, op,
//! collection) tuple via `TriggerRepo::list_matching_*`.
//! 2. For each match, write one outbox row carrying the event payload
//! serialized as a `TriggerEvent`.
//! **The fan-out is connection-scoped, not pool-scoped.** [`emit_on`] and
//! [`emit_shared_on`] take a `&mut PgConnection`, so a caller that already holds
//! a transaction can pass `&mut *tx` and have the outbox rows commit *with* the
//! data write that produced them — the transactional-outbox pattern. That closes
//! the window where a row committed but its trigger never fired because the
//! outbox insert failed (or the process died) immediately afterwards. Services
//! that write through `crate::atomic_write` take that path; the
//! `ServiceEventEmitter` trait impl below is the non-transactional entry point,
//! kept for the callers (and the tests) that only need best-effort emission.
//!
//! Defaults applied at write time so `OutboxRow.payload` carries
//! everything the dispatcher needs to reconstruct the executor
//! invocation without joining back to the trigger row.
//!
//! Non-KV `ServiceEvent` sources are silently dropped in v1.1.1 — the
//! dispatcher only knows how to fire KV triggers this release. Future
//! sources (docs/files/pubsub) add their own dispatch arm.
use std::sync::Arc;
//! Non-collection `ServiceEvent` sources are silently dropped: the SDK calls
//! `events.emit(...)` unconditionally for forward compat, and every source the
//! dispatcher has an arm for is handled here.
use async_trait::async_trait;
use picloud_shared::{
DocsEventOp, EmitError, FileMeta, FilesEventOp, GroupId, KvEventOp, SdkCallCx, ServiceEvent,
ServiceEventEmitter, TriggerEvent,
};
use sqlx::{PgConnection, PgPool};
use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind};
use crate::trigger_repo::TriggerRepo;
use crate::outbox_repo::{insert_on, NewOutboxRow, OutboxSourceKind};
use crate::trigger_repo::{list_matching_on, list_matching_shared_on, KvMatchRow};
pub struct OutboxEventEmitter {
triggers: Arc<dyn TriggerRepo>,
outbox: Arc<dyn OutboxRepo>,
pool: PgPool,
}
impl OutboxEventEmitter {
#[must_use]
pub fn new(triggers: Arc<dyn TriggerRepo>, outbox: Arc<dyn OutboxRepo>) -> Self {
Self { triggers, outbox }
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[async_trait]
impl ServiceEventEmitter for OutboxEventEmitter {
async fn emit(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
match event.source {
"kv" => self.emit_kv(cx, event).await,
"docs" => self.emit_docs(cx, event).await,
"files" => self.emit_files(cx, event).await,
// Future sources land here. For now, silently drop — the
// SDK calls `events.emit(...)` unconditionally for forward
// compat, so swallowing without an error is correct.
_ => Ok(()),
}
let mut conn = self
.pool
.acquire()
.await
.map_err(|e| EmitError::Unavailable(format!("acquire connection: {e}")))?;
emit_on(&mut conn, cx, &event).await
}
#[allow(clippy::too_many_lines)] // one match arm per source kind (kv/docs/files)
async fn emit_shared(
&self,
cx: &SdkCallCx,
owning_group: GroupId,
event: ServiceEvent,
) -> Result<(), EmitError> {
// §11.6: a shared-collection write. Match `shared = true` triggers on
// the OWNING group; each fires under the WRITER app (`cx.app_id`), the
// same "group template runs under the firing app" model.
let source_kind = match event.source {
"kv" => OutboxSourceKind::Kv,
"docs" => OutboxSourceKind::Docs,
"files" => OutboxSourceKind::Files,
_ => return Ok(()),
};
let Some(collection) = event.collection.clone() else {
return Ok(());
};
let (matches, trigger_event) = match event.source {
"kv" => {
let Some(op) = KvEventOp::from_wire(event.op) else {
return Ok(());
};
let m = self
.triggers
.list_matching_shared_kv(owning_group, &collection, op)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
let ev = TriggerEvent::Kv {
let mut conn = self
.pool
.acquire()
.await
.map_err(|e| EmitError::Unavailable(format!("acquire connection: {e}")))?;
emit_shared_on(&mut conn, cx, owning_group, &event).await
}
}
/// Everything the fan-out needs, derived from a `ServiceEvent` with no I/O: the
/// trigger-table coordinates to match on, and the `TriggerEvent` the handler
/// will see as `ctx.event`. `None` for a source or op the dispatcher has no arm
/// for — the caller drops the event quietly.
struct EventPlan {
source_kind: OutboxSourceKind,
/// `triggers.kind` discriminator. A literal — never user data, since it is
/// interpolated into the match SQL.
kind: &'static str,
/// Per-kind detail table. A literal, for the same reason.
detail_table: &'static str,
collection: String,
op_str: &'static str,
trigger_event: TriggerEvent,
}
fn plan(event: &ServiceEvent) -> Option<EventPlan> {
// Collection events always carry a collection; defensively skip if not.
let collection = event.collection.clone()?;
let key = event.key.clone().unwrap_or_default();
match event.source {
"kv" => {
let op = KvEventOp::from_wire(event.op)?;
Some(EventPlan {
source_kind: OutboxSourceKind::Kv,
kind: "kv",
detail_table: "kv_trigger_details",
collection: collection.clone(),
op_str: op.as_str(),
trigger_event: TriggerEvent::Kv {
op,
collection,
key: event.key.clone().unwrap_or_default(),
key,
value: event.payload.clone(),
};
(
m.into_iter()
.map(|x| (x.trigger_id, x.script_id))
.collect::<Vec<_>>(),
ev,
)
}
"docs" => {
let Some(op) = DocsEventOp::from_wire(event.op) else {
return Ok(());
};
let m = self
.triggers
.list_matching_shared_docs(owning_group, &collection, op)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
let ev = TriggerEvent::Docs {
},
})
}
"docs" => {
let op = DocsEventOp::from_wire(event.op)?;
Some(EventPlan {
source_kind: OutboxSourceKind::Docs,
kind: "docs",
detail_table: "docs_trigger_details",
collection: collection.clone(),
op_str: op.as_str(),
trigger_event: TriggerEvent::Docs {
op,
collection,
id: event.key.clone().unwrap_or_default(),
id: key,
data: event.payload.clone(),
prev_data: event.old_payload.clone(),
};
(
m.into_iter()
.map(|x| (x.trigger_id, x.script_id))
.collect::<Vec<_>>(),
ev,
)
}
"files" => {
let Some(op) = FilesEventOp::from_wire(event.op) else {
return Ok(());
};
let Some(meta) = event
.payload
.clone()
.and_then(|v| serde_json::from_value::<FileMeta>(v).ok())
else {
return Ok(());
};
let m = self
.triggers
.list_matching_shared_files(owning_group, &collection, op)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
let ev = TriggerEvent::Files {
},
})
}
"files" => {
let op = FilesEventOp::from_wire(event.op)?;
// The payload is the `FileMeta` JSON the files service emitted —
// never the blob bytes.
let meta: FileMeta = serde_json::from_value(event.payload.clone()?).ok()?;
Some(EventPlan {
source_kind: OutboxSourceKind::Files,
kind: "files",
detail_table: "files_trigger_details",
collection: collection.clone(),
op_str: op.as_str(),
trigger_event: TriggerEvent::Files {
op,
collection,
id: meta.id.to_string(),
@@ -142,206 +138,87 @@ impl ServiceEventEmitter for OutboxEventEmitter {
size: meta.size,
checksum: meta.checksum,
prev: event.old_payload.clone(),
};
(
m.into_iter()
.map(|x| (x.trigger_id, x.script_id))
.collect::<Vec<_>>(),
ev,
)
}
_ => return Ok(()),
};
if matches.is_empty() {
return Ok(());
},
})
}
let payload = serde_json::to_value(&trigger_event)
.map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;
for (trigger_id, script_id) in matches {
self.outbox
.insert(NewOutboxRow {
app_id: cx.app_id,
source_kind,
trigger_id: Some(trigger_id),
script_id: Some(script_id),
reply_to: None,
payload: payload.clone(),
origin_principal: cx.principal.as_ref().map(|p| p.user_id),
trigger_depth: cx.trigger_depth.saturating_add(1),
root_execution_id: Some(cx.root_execution_id),
})
.await
.map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
}
Ok(())
_ => None,
}
}
impl OutboxEventEmitter {
async fn emit_kv(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
let Some(op) = KvEventOp::from_wire(event.op) else {
return Ok(()); // unknown op — drop quietly
};
let Some(collection) = event.collection.clone() else {
return Ok(()); // KV events always carry a collection — defensively skip
};
let key = event.key.clone().unwrap_or_default();
let matches = self
.triggers
.list_matching_kv(cx.app_id, &collection, op)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
if matches.is_empty() {
return Ok(());
}
// Serialize the originating event as a TriggerEvent so the
// dispatcher can hand it to the script as `ctx.event` without
// round-tripping back to the trigger row.
let trigger_event = TriggerEvent::Kv {
op,
collection,
key,
value: event.payload.clone(),
};
let payload = serde_json::to_value(&trigger_event)
.map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;
for m in matches {
self.outbox
.insert(NewOutboxRow {
app_id: cx.app_id,
source_kind: OutboxSourceKind::Kv,
trigger_id: Some(m.trigger_id),
script_id: Some(m.script_id),
reply_to: None,
payload: payload.clone(),
origin_principal: cx.principal.as_ref().map(|p| p.user_id),
trigger_depth: cx.trigger_depth.saturating_add(1),
root_execution_id: Some(cx.root_execution_id),
})
.await
.map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
}
Ok(())
}
/// v1.1.2. Mirrors `emit_kv` — fan out a docs mutation across
/// matching docs triggers + write one outbox row each. The
/// `prev_data` change-data-capture surface is preserved from the
/// `ServiceEvent.old_payload` field (set by `DocsServiceImpl` on
/// update and delete; `None` for create).
async fn emit_docs(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
let Some(op) = DocsEventOp::from_wire(event.op) else {
return Ok(());
};
let Some(collection) = event.collection.clone() else {
return Ok(());
};
let id = event.key.clone().unwrap_or_default();
let matches = self
.triggers
.list_matching_docs(cx.app_id, &collection, op)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
if matches.is_empty() {
return Ok(());
}
let trigger_event = TriggerEvent::Docs {
op,
collection,
id,
data: event.payload.clone(),
prev_data: event.old_payload.clone(),
};
let payload = serde_json::to_value(&trigger_event)
.map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;
for m in matches {
self.outbox
.insert(NewOutboxRow {
app_id: cx.app_id,
source_kind: OutboxSourceKind::Docs,
trigger_id: Some(m.trigger_id),
script_id: Some(m.script_id),
reply_to: None,
payload: payload.clone(),
origin_principal: cx.principal.as_ref().map(|p| p.user_id),
trigger_depth: cx.trigger_depth.saturating_add(1),
root_execution_id: Some(cx.root_execution_id),
})
.await
.map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
}
Ok(())
}
/// v1.1.5. Fan out a files mutation across matching files triggers.
/// The `ServiceEvent.payload` is the file **metadata** (never the
/// blob bytes); `old_payload` is the prior metadata (the deleted
/// row's metadata on delete). The `TriggerEvent::Files` carries the
/// metadata fields explicitly + `prev` for the change-data-capture
/// surface.
async fn emit_files(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
let Some(op) = FilesEventOp::from_wire(event.op) else {
return Ok(());
};
let Some(collection) = event.collection.clone() else {
return Ok(());
};
// The payload is the FileMeta JSON the FilesServiceImpl emitted.
let Some(meta) = event
.payload
.clone()
.and_then(|v| serde_json::from_value::<FileMeta>(v).ok())
else {
return Ok(());
};
let matches = self
.triggers
.list_matching_files(cx.app_id, &collection, op)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
if matches.is_empty() {
return Ok(());
}
let trigger_event = TriggerEvent::Files {
op,
collection,
id: meta.id.to_string(),
name: meta.name,
content_type: meta.content_type,
size: meta.size,
checksum: meta.checksum,
prev: event.old_payload.clone(),
};
let payload = serde_json::to_value(&trigger_event)
.map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;
for m in matches {
self.outbox
.insert(NewOutboxRow {
app_id: cx.app_id,
source_kind: OutboxSourceKind::Files,
trigger_id: Some(m.trigger_id),
script_id: Some(m.script_id),
reply_to: None,
payload: payload.clone(),
origin_principal: cx.principal.as_ref().map(|p| p.user_id),
trigger_depth: cx.trigger_depth.saturating_add(1),
root_execution_id: Some(cx.root_execution_id),
})
.await
.map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
}
Ok(())
}
/// Fan a collection mutation out over the writing app's own + inherited
/// triggers. Pass `&mut *tx` to have the outbox rows commit with the write.
pub(crate) async fn emit_on(
conn: &mut PgConnection,
cx: &SdkCallCx,
event: &ServiceEvent,
) -> Result<(), EmitError> {
let Some(p) = plan(event) else { return Ok(()) };
let matches = list_matching_on(
&mut *conn,
p.kind,
p.detail_table,
cx.app_id,
&p.collection,
p.op_str,
)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
write_outbox_rows(conn, cx, &p, matches).await
}
/// §11.6: fan a SHARED-collection write out over the `shared = true` triggers on
/// the OWNING group. Each fires under the WRITER app (`cx.app_id`) — the same
/// "a group template runs under the firing app" model as an inherited trigger.
pub(crate) async fn emit_shared_on(
conn: &mut PgConnection,
cx: &SdkCallCx,
owning_group: GroupId,
event: &ServiceEvent,
) -> Result<(), EmitError> {
let Some(p) = plan(event) else { return Ok(()) };
let matches = list_matching_shared_on(
&mut *conn,
p.kind,
p.detail_table,
owning_group,
&p.collection,
p.op_str,
)
.await
.map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
write_outbox_rows(conn, cx, &p, matches).await
}
async fn write_outbox_rows(
conn: &mut PgConnection,
cx: &SdkCallCx,
p: &EventPlan,
matches: Vec<KvMatchRow>,
) -> Result<(), EmitError> {
if matches.is_empty() {
return Ok(());
}
// Serialize the originating event once so the dispatcher can hand it to the
// handler as `ctx.event` without joining back to the trigger row.
let payload = serde_json::to_value(&p.trigger_event)
.map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;
for m in matches {
insert_on(
&mut *conn,
NewOutboxRow {
app_id: cx.app_id,
source_kind: p.source_kind,
trigger_id: Some(m.id.into()),
script_id: Some(m.script_id.into()),
reply_to: None,
payload: payload.clone(),
origin_principal: cx.principal.as_ref().map(|pr| pr.user_id),
trigger_depth: cx.trigger_depth.saturating_add(1),
root_execution_id: Some(cx.root_execution_id),
},
)
.await
.map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
}
Ok(())
}