feat(outbox): canonical operation-wide publications - #149
Draft
nicolasburtey wants to merge 1 commit into
Draft
nicolasburtey wants to merge 1 commit into
nicolasburtey wants to merge 1 commit into
Conversation
nicolasburtey
changed the title
nicolasburtey
force-pushed
the
poc/atomic-publication-batches
branch
from
September 14, 2026 17:17
02fc288 to
28b1c6d
Compare
Every source operation's publications — across publish calls, persist chunks, post-persist/repost hooks, and re-entrant commit-hook generations — commit as one indivisible publication, sealed by an es-entity finalizer after ordinary hooks finish. A transactional head reserves each operation's contiguous sequence range at its first persist chunk; rollback restores positions together with rows. Delete the gap filler, placeholder compensation, abandonment proofs, contiguous-page cut queries, and the in-transaction notify variant. Bare-transaction publishes are rejected; the frontier is the committed head. Payloads live only in the event table; event and publication consumers share one sequence namespace. load_publication_batches returns complete publications and a PublicationBatches stream selection delivers them whole, checkpointing only after the callback commits. Keep the two schema-dependent derive queries runtime-checked so downstream offline caches stay sufficient. Rebased onto keyed subscribers (0.10.x): singleton subscriber carries the publication runner alongside main's StreamSelection.
nicolasburtey
force-pushed
the
poc/atomic-publication-batches
branch
from
September 14, 2026 17:29
28b1c6d to
a658998
Compare
Member
|
This change serializes all publishes to the outbox at commit time with a global lock. Since almost every operation in our domain writes to the outbox this will cause a complete serialization of all operations - no parallelism at all. The outbox is designed to accept concurrent writes without blocking - serialization happens at the read side to keep writes unblocked. Don't think this is the right approach to fix the 'chunking' issue. |
Member
|
I think we can have a much simpler way of signaling 'chunks' |
bodymindarts
added a commit
that referenced
this pull request
Sep 16, 2026
* feat(out)!: commit-ordered delivery as an opt-in second lane
Add an opt-in delivery lane in which events arrive in commit order, every
source transaction's events are contiguous, and the cursor is a dense,
gap-free `CommitSequence`. The insert-ordered lane (`EventSequence`,
contiguity via gap fill) stays the default and is untouched.
Postgres stamps every event row with its top-level transaction id
(`commit_xid`, a column DEFAULT — no write-side code, no lock, no writer
serialization). A per-process sequencer, fed by the cache loop's contiguity
frontier, appends `(commit_seq, sequence, group_last)` rows to
`persistent_outbox_commit_log` in `(group_max, sequence)` order, where
`group_max` is a transaction's highest insert sequence. One statement per
burst does the whole tick; `FOR UPDATE SKIP LOCKED` on a one-row state table
makes it leaderless, so a loser skips its tick instead of blocking and a
process dying mid-tick leaves committed state untouched.
OutboxEventJobConfig::new(JOB_TYPE) // Insert, default
OutboxEventJobConfig::new(JOB_TYPE).ordering(Ordering::Commit)
This supersedes the approach in #149, which reserved each operation's
contiguous range in a `<events>_batch_head` row whose lock was held to COMMIT.
That serializes every publisher; deriving the order read-side serializes
nobody.
Two defects in the design spec were found and corrected:
* The scan window was anchored on `low_water`, but an open group pins
`low_water` at its lowest member, so the window could never widen to reach
that group's highest member and close it. Every tick recomputed an identical
window: a permanent livelock whenever a group's span exceeded the page size.
`scan_water` is a second watermark that advances past logged rows while
`low_water` stays pinned, carrying the window forward.
* The sequencer's frontier was specified as the cache loop's
`last_broadcast_sequence`, which only advances when the broadcast send
succeeds — and tokio's `send` fails precisely when there are no receivers.
A process hosting only commit-ordered listeners has no insert-lane receiver,
so its frontier was pinned at zero forever. The cursor now advances before
delivery is attempted; nothing an insert-lane subscriber can observe
changes, since with a receiver the send succeeds and without one there is no
subscriber.
Gap-fill placeholders are written with an explicit NULL `commit_xid` rather
than inheriting the filler transaction's id, so "placeholder implies singleton
group" holds.
BREAKING CHANGE, of two independent kinds:
* Rust API. `PersistentOutboxEvent` and `UndecodableEventError` gain
`commit_group`, `commit_sequence` and `commit_boundary`;
`decode_persistent_event` takes a `CommitPosition` instead of loose
arguments; `MailboxTables` gains six methods. Insert-lane execution state is
unchanged on the wire — `commit_sequence` is `skip_serializing_if`, and the
existing `state_without_a_pause_serializes_unchanged` test still asserts
`{"sequence":7}`.
* Migration checksum. The setup migration is amended in place per this repo's
convention, so every already-migrated database must be recreated, and
consumers that copied the SQL by hand must re-copy it. obix exposes no
`migrate!` re-export, so that drift is invisible until a missing relation at
runtime.
Switching an established subscription from the insert lane to the commit lane
redelivers the events of any group straddling its cursor. That is refused
unless the caller also calls `.allow_lane_switch_redelivery()`, so the
duplicate window cannot be opened by flipping one enum value. Switching back
is refused outright.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* ci: allow BRIN in the typo checker
`BRIN` is PostgreSQL's Block Range INdex access method, which the checker
reads as a misspelling of "bring".
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* refactor(out)!: derive commit order by folding the insert stream
Replaces the DB-side windowed sequencer tick with an in-process fold over
the insert-ordered listener, per handoff rev 3.
The tick re-derived contiguity in SQL (holes/low_water/scan_water) over the
events table -- work the cache loop already does for the insert stream. Both
defects found while implementing rev 2 were consequences of that: the scan
window could not widen past an open group (livelock), and the sequencer was
not a broadcast receiver, so the frontier it depended on pinned at zero in a
commit-only process (silent hang). Making the sequencer a listener dissolves
both.
Groups are now placed at first sight rather than by their highest member.
Every outbox row of a transaction is inserted by PersistEvents::pre_commit,
after every read that transaction made, so if T observed A's commit then
min(T) > max(A): ordering by a group's lowest sequence never inverts a
dependency. group_max was the only thing requiring a lookback window.
One statement per group appends it whole and advances (head, cursor) under a
cursor CAS on the locked state row. Every process folds the same stream from
the same state row and computes the same log including commit_seq, so the CAS
is a duplicate suppressor rather than an election: no leader, no lease, and
neither a commit_seq gap nor a duplicated assignment is representable.
The CAS must be a column of the locked row. Postgres re-evaluates it against
the row version a lock wait resolved to; a NOT EXISTS against the log reads
this statement's snapshot, which predates a concurrent winner's commit, and
appends the group twice (541 rows instead of 200 under four concurrent
folds).
The sequencer is now always on, spawned in Outbox::init, so the commit log
never lags a live process. The commit log is partitioned in lock-step with
the events table by the same maintainer, in one transaction under one lock.
Breaking, and only valid on a fresh deploy: the setup migration is amended in
place, commit_xid is NOT NULL with a btree index, the state row is
(head, cursor), PersistentOutboxEvent::commit_group is no longer optional,
and the commit position moves off the event onto the commit lane's item type
CommitOrderedEvent. The insert-to-commit lane switch is removed: the two
cursors count different things, so an established subscription registers a
new job type instead.
Also fixes an undecodable event acknowledged on the commit lane advancing
only the insert cursor, which redelivered it after every restart.
Delivery order changes from group_max to group_min. Both are valid causal
orders; delivery granularity, checkpoint timing and at-least-once semantics
are unchanged.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* test(out): isolate the paging budget from the sequencer's catch-up
`slow_consumer_does_not_inflate_the_rows_read` samples `idx_tup_fetch`,
which is per table and therefore counts every reader of
persistent_outbox_events. The always-on sequencer is now a second reader:
at startup it replays history through its own listener and then looks up
one group per event, which spent most of a budget written for the single
listener under test. Measured 4356 locally and 10418 in CI against a
budget of 8000.
Starting the sequencer already caught up restores what the test measures.
Reads are then 2002-4004 for 2000 events -- one to two passes -- against
the unchanged 4x budget, so the eviction pathology it guards still fails it.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* fix(out): keep the sequencer's published head consistent with the log
The sequencer read its resume point with two statements -- the state row,
then the sequences already logged above that cursor. A peer appending
between them returned a head that did not account for rows the seed then
told the fold to skip, and the skip path returns without appending, so
nothing else in the loop had occasion to correct it. A commit-lane listener
takes that head as `latest_known`, so with `last_returned == latest_known`
it neither backfills nor receives a broadcast, and stays that way until this
process appends a group of its own.
Two changes, either of which closes it; together they make the published
head independent of how the resume point was obtained.
Read both values in one statement, so head and the seed cannot disagree.
Every sequence the fold then sees is either in the seed -- logged as of that
snapshot, and therefore already counted by head -- or absent from it, in
which case the fold attempts an append and a peer that got there first
rejects it, which reconciles from the log.
Reconcile the head against the log once before folding. The seeded
sequences are skipped without appending, so this is the only occasion on
which a head a peer has already moved past is corrected.
The concurrent repro needs groups that straddle the cursor: with singleton
groups the cursor always sits at or above every logged sequence, the seed is
always empty, and the invariant is vacuous. With straddling groups, split
reads fail the snapshot-consistency check in roughly two runs in three; the
reconcile is covered deterministically by building the stale pair directly.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* refactor(out): drop a redundant gap-fill sender clone
The cache loop is the only consumer of the sender once the commit lane no
longer holds one, so `init` can move it instead of cloning. The clone inside
the backfill dispatch stays: each spawned request takes ownership, so it has
to survive the loop.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* refactor(out)!: name the sequencer state columns by what they count
`head` and `cursor` gave no hint that they count different things: one is a
commit_seq, a position in the commit log, and the other an insert sequence,
a position in persistent_outbox_events. Appending a group of three advances
the first by three and sets the second to that group's lowest member.
They become `last_commit_seq` and `logged_through_sequence`, so the suffix
names the space. `logged_through_sequence` also states the stronger
invariant the queries rely on: every payload-bearing event at or below it is
in the log, and the seed is what sits above it because a group straddled it.
`id SMALLINT PRIMARY KEY` documented "exactly one row" without enforcing it
-- nothing stopped id = 2. `singleton BOOLEAN PRIMARY KEY CHECK (singleton)`
admits one row and no other: the check rejects FALSE, the key rejects a
second TRUE. Verified both rejections against the live schema.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* feat(out)!: hand subscribers the event Arc instead of deref'ing it away
The outbox decodes each persistent event once and broadcasts it as an
Arc, but `handle_persistent` / `handle` took `&PersistentOutboxEvent<P>`,
so the runners deref'd that Arc away at the call site. A handler could
therefore not retain an event past the call — every collect/flush
projection had to either pre-derive a smaller item or deep-clone the
payload, and payload enums that are not `Clone` could do neither.
Both trait parameters (and `handle_ephemeral`, for one rule across the
trait) now take `&Arc<…>`. The call sites are unchanged — they already
passed `&event` where `event: Arc<…>` — and since `Arc<T>` derefs to
`T`, no handler body in the suite needed editing: the entire diff
outside the traits is signature lines.
`SubscriptionDef::wake_keys` keeps its plain reference. It classifies
and retains nothing, and the runner's `&Arc` coerces at the call site.
The new test file is generic over a payload enum that deliberately does
not derive `Clone`, with `Batch = Vec<Arc<PersistentOutboxEvent<_>>>` —
a batch that could not be written before this change. It also pins that
`flush` receives the very allocations the invocations were handed, not
copies of them.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* refactor(out)!: name the commit-lane wrapper for what it is, and file the row types with the queries
`CommitOrderedEvent` was not an event — it carries one. Renamed to
`CommitOrderedEnvelope`, which says that the commit sequence and the
boundary flag are wrapped around the event rather than part of it.
`CommitLogRow` and `CommitGroupAppend` move from `out::event` to
`tables`, next to `CommitRestartState`. They are the result shapes of
`load_commit_ordered_page` and `append_commit_group` — query output, not
event vocabulary. `CommitDelivery` stays in `out::event`: it is the
commit lane's counterpart to `PersistentDelivery` and belongs beside it.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* chore(deps): bump job to 0.14.0
Also pulls es-entity 0.12.17 -> 0.12.21 as a transitive update.
No source changes were needed: the major bump did not touch any API obix
uses.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* fix(migrations)!: re-vendor job's schema, which was pinned at 0.13.0
obix vendors job's `20250904065521_job_setup.sql`, and that copy was last
refreshed at the 0.13.0 bump (2e7a14b). job revised its schema twice
since — once during 0.13.1..0.13.5 and again during 0.13.14..0.13.16 —
so the file has been four schema revisions behind while Cargo.toml said
0.13.16. It is byte-identical to job 0.14.0's copy again.
Two differences, both of which had been missing from every obix database:
- `job_events.id` and `job_executions.id` no longer carry a foreign key
to `jobs(id)`.
- The partial pending index leads with `job_type`:
`(execute_at, id)` -> `(job_type, execute_at, id)`. Every consumer
probes per type, so without the leading column each probe filter-scans
the whole pending set instead of its own type's slice.
Requires recreating the database, like the amended obix migration in this
branch: `make clean-deps && make start-deps`. Otherwise sqlx reports
`migration 20250904065521 was previously applied but has been modified`.
Verified against the live database after recreation: the index is
`(job_type, execute_at, id)` and no foreign keys remain on either table.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
First-class operation-wide publications, replacing the pre-allocation event stream where consumers had to reconstruct source-transaction boundaries. Rebased onto main (keyed subscribers, 0.10.x) — the publication model now lives alongside the new singleton/keyed vocabulary, with
StreamSelection::PublicationBatchesas a third delivery kind.Every source operation's publications — across publish calls, persist chunks, post-persist/repost hooks, and re-entrant commit-hook generations — commit as one indivisible unit. An es-entity finalizer seals the boundary only after all ordinary hooks finish. Cala needs no patch (a 0.26.1-based fork just pins this revision).
<events>_batch_headreserves each operation's contiguous position range at its first persist chunk; the row lock serializes writers until commit.<events>_batchesstores only(first_sequence, last_sequence)boundaries, sealed at finalization in the same transaction. Payloads live once, in the event table.gap_fill.rs, contiguous-page cut queries, missing-sequence probes, reactive compensation, the in-transaction notify variant).sqlx::Transactionpublishes are rejected loudly instead of writing unsealed rows.highest_known_persistent_sequence) is the committed head, excluding in-flight operations.load_publication_batches(after, limit)returns complete publications, loading member payloads from the event table; interior cursors, missing prefixes, or missing member rows fail closed.SingletonSubscriber::handle_publication_batchreceives one whole publication per callback; the checkpoint lands only after the callback's transaction commits.Linked drafts
Validation
--all-targets -D warnings), formatting, and a clean-build offline cache regeneration pass.transaction_batch+ 16/16ec_streaming_rollup; Lana relay and Cala-mirror integrations pass against this revision.Deliberate limitations
See
PUBLICATION_BATCHES_POC.mdfor the full contract. Draft for architectural evaluation, not production rollout.