Life of a push

A push request is not a transaction. It is a set of frames handed to a background layer that decides, per partition, when to commit, then commits several partitions together. Under load, one PostgreSQL transaction carries the writes of many concurrent HTTP requests across many partitions. That is where most of the broker’s throughput comes from, and it is the single most important thing to understand about the write path.

Here is the flow a client sees:

const res = await client
.queue('test-queue-v2')
.push([{ data: { message: 'Hello, world!' } }])

And here is what happens behind it.

sequenceDiagram
participant C as client
participant H as push handler
participant F as fusion shard
participant PG as PostgreSQL
C->>H: POST /api/v1/push with an items array
H->>H: mint ids, layer-1 dedup, group by queue and partition
H->>F: one submission per group
F->>F: accumulate, then pick ready not-in-flight partitions
F->>PG: log_push_multi_v1 with N segments
PG->>PG: lock partition rows by ascending id, allocate offsets
PG-->>F: one commit, one fsync, results in input order
F-->>H: verdict routed to every request that fed a segment
H-->>C: 201 with one result per item

The commit boundary belongs to the shard rather than to the request, which is what makes the two fan-ins possible: several partitions ride one transaction, and several still-waiting requests can have contributed frames to any one of them.

1. The handler: mint, dedup, group

POST /api/v1/push takes a body with an items array. An empty array answers 201 with [] immediately.

For each item the handler:

• mints a UUIDv7 message id;
• resolves the transaction id: the client’s transactionId if present, otherwise the minted message id as a string, so every frame has one;
• resolves the partition, defaulting to Default;
• checks layer-1 deduplication: a map keyed on queue \x1f partition \x1f transactionId inside this request. First occurrence wins and becomes a “leader”; a repeat becomes a follower, produces no frame at all, and copies the leader’s final message id when the response is rendered. “Final” matters: if the leader itself turns out to be a duplicate of something already stored, the follower reports the pre-existing id too;
• encrypts the payload when the queue has encryption_enabled (the flag is memoised per queue, so a 1000-item push over three queues does at most three lookups).

Frames are then grouped by (queue, partition). The number of groups becomes the request’s pending counter: the HTTP response resolves when every group it contributed to has committed.

If the broker is in maintenance mode, none of this reaches PostgreSQL: every item is written to the disk spool and reported with status buffered, and the background drain replays it when maintenance ends.

2. Fusion: one accumulating segment per partition

Each group is submitted to the fusion layer, which hashes queue + partition to one of QUEEN_V2_FUSION_SHARDS shards (default 8). A shard owns a disjoint set of partitions, so all of its state is single-task-local and needs no locking.

Inside a shard, frames accumulate into a group keyed tenant \x1f queue \x1f partition. The tenant is part of the key, so two tenants pushing to a queue called orders accumulate separate segments.

Three rules govern dispatch:

Fire-on-idle. On arrival, if the push lane is uncontended (the shard is under its concurrent-bundle cap and a push Vegas permit is immediately available), the ready partitions dispatch right now, with no hold timer. A single low-rate push therefore commits in about one database round trip. If no permit is free, the lane is contended, so the group stays and keeps accumulating.

One flush in flight per partition. While a partition’s flush is in flight, its next segment accumulates and is dispatched only after the previous flush commits. Two flushes therefore never race on UPDATE log_partitions SET last_offset = last_offset + n, and per-partition offset order equals commit order. The completion signal is sent after the commit, not before, which is what makes that guarantee hold rather than merely usually hold.

Cross-partition bundling. Each dispatch collects up to QUEEN_V2_BUNDLE_MAX ready, not-in-flight partitions and commits all of their segments in one log_push_multi_v1 transaction: one commit and one fsync for N partitions. The in-flight gate is the bundle selector: it already guarantees the chosen partitions are disjoint. The bundle size auto-tunes to load. Idle means one ready partition and a bundle of one; busy means many partitions accumulate while permits are scarce, and a freed permit fires a large bundle. Commit rate is thereby decoupled from partition count.

The broker exposes the ratio directly. queen_batch_items_fired_total and queen_fusion_items_per_batch are on /metrics/prometheus, and the fusion log target emits a bundle summary line about every ten seconds with commits, segments and segments_per_commit. Segments far exceeding commits is the direct evidence that fusion is coalescing across partitions.

QUEEN_V2_FUSION_HOLD_MS (default 15) is a liveness backstop tick, not a batching window. It exists so a group held because a permit was freed on another shard eventually gets retried. Lowering it does nothing under load, because arrivals and completions already wake the shard.

3. Building the segment

For each group in the bundle, the flush does layer-2 deduplication: intra-flush, first-wins, comparing the 16-byte xxh3_128 hashes rather than composed strings. This is the same identity SQL will compare, so a broker-local collision behaves exactly like a SQL one. The first frame carrying a hash is the leader and the only one that reaches the segment; later same-hash frames inherit its verdict.

Then it builds two byte arrays per segment:

• the hash blob: 16 × K bytes, one xxh3_128 per frame in frame order, so position k in the blob is packed frame k;
• the payload blob: the packed frames, zstd-compressed.

Push payloads are zero-copy on ingest: a plaintext frame’s payload is a refcounted slice of the HTTP request body, so dropping a flushed bundle decrements refcounts instead of freeing one allocation per message.

4. One transaction, N segments

The bundle is sent as typed parallel arrays (text[], text[], int4[], bytea[], int8[], bytea[], uuid[]) to a prepare_cached statement. No JSONB is parsed on the hot path.

queen.log_push_multi_v1(
    p_queues     TEXT[],
    p_partitions TEXT[],
    p_msg_counts INT[],
    p_hashes     BYTEA[],
    p_verified   INT8[],
    p_blobs      BYTEA[],
    p_tenants    UUID[] DEFAULT NULL
) RETURNS JSONB

Inside, in order:

1. Alignment guard. Any array of a different length raises. A silent misalignment would demux the wrong verdict to the wrong request.
2. Provisioning skip. One cheap existence count over unnest(queues, partitions, tenants). Only if something is missing do the two INSERT ... ON CONFLICT DO NOTHING statements run: one insert into queen.queues (the queue’s identity and configuration are one row; the namespace and task are derived from the dotted name), then log_partitions, whose queue_id references that row. Each insert is ORDER BY-ed on its unique key, because ON CONFLICT DO NOTHING takes speculative locks in row order and two bundles creating overlapping partition sets in different orders would convoy exactly as unordered row locks do.
3. One set-based pre-lock. A single statement locks every partition row in the bundle, ORDER BY p.id FOR UPDATE OF p. Ascending log_partitions.id is the one global lock order, shared with the transaction wire and the retention steps, and it is why no combination of those three can deadlock.
4. A bundle timestamp, taken once after the locks.
5. Per segment, in ascending partition id order, a call to queen.log_push_one_v1 with the partition id, deduplication window and tenant already resolved (the callee must not re-resolve them per segment).
6. A completeness guard. Every input ordinal must produce exactly one result, or the whole bundle raises.

Results come back as a JSON array in input order, so the broker can route each verdict to the requests whose frames landed in that segment.

5. The allocator, under the partition lock

log_push_one_v1 is the single allocator code path: the transaction wire and the streams cycle call it too, so there is exactly one place where an offset is assigned.

It first checks that the hash blob is 16 × msg_count bytes and raises if not. A stride mismatch would mis-address acks silently, so it fails loudly.

Then it splits on the deduplication window:

Window greater than zero. Take the partition row lock first:

SELECT last_offset, txns_start FROM queen.log_partitions WHERE id = v_pid FOR UPDATE;

Stamp clock_timestamp() while holding it. Probe queen.log_txns over (GREATEST(p_verified, txns_start - 1), last_offset], time-capped by the window, joining the stored hashes against the incoming ones. On a match, return duplicate having written nothing: no allocator bump, no inserts, and therefore no rollback and no savepoint. Other segments in the same bundle commit untouched. Only a clean probe proceeds to allocate.

Window zero. Skip the probe entirely and go straight to the single UPDATE ... RETURNING fast path; that UPDATE takes the same row lock and is the whole serialization story.

The allocation itself is one statement:

UPDATE queen.log_partitions SET
    last_offset   = last_offset + p_msg_count,
    last_write_at = CASE WHEN clock_timestamp() - last_write_at > interval '1 second'
                         THEN clock_timestamp() ELSE last_write_at END
WHERE id = v_pid
RETURNING last_offset;

base_offset = last_offset - msg_count + 1. The CASE is the quantization that keeps the update HOT despite last_write_at being indexed.

Finally two inserts: the segment, and the log_txns row (the latter unconditionally, because ack-by-transactionId resolves through it whether or not deduplication is on).

6. Rendering the response

The flush parses the per-segment results and writes each frame’s verdict into its owning request’s result slot.

• queued carries baseOffset and createdAt. Frame k of the segment is at offset baseOffset + k.
• duplicate carries dups: [{i, off}], the incoming frame index and the original occurrence’s absolute offset. The broker then resolves the canonical message id by fetching the covering segment with queen.log_segment_at_v1(pid, off) and unpacking frame off - base in Rust. This is a rare path and uses plain, uncached queries. If retention already deleted that segment, the id is reported as the zero UUID.
• A whole-bundle failure (pool exhaustion, statement timeout, an unparseable result) resolves every participant to error.

Once every group of a request has reported, the handler wakes, renders the per-item array and answers 201.

Two things happen after that:

Discovery. Each pushed (tenant, queue, partition) is marked pending on the hot-list rings, quietly: the local wake is coalesced into a 5 ms tick, so a hot queue costs at most one wake per tick instead of one wake per push (each of which is itself proportional to the number of parked consumers). Peers still get an immediate batched MESSAGE_AVAILABLE. With QUEEN_HOTLIST=0 the push instead wakes parked pops directly with partition hints.

Error spooling. Items that came back error (meaning the transaction that would have stored them failed) are written to the disk spool and relabelled buffered, so a client that only checks the HTTP status code does not silently lose them. The background drain replays them, and deduplication on the preserved transactionId makes the replay idempotent. A spool write failure downgrades the item to failed.

Failure modes worth predicting

A statement timeout also has a server-side consequence: tokio::time::timeout only drops the client future, so the broker sends an out-of-band cancel request on a fresh connection and quarantines the pooled connection instead of recycling it. Otherwise abandoned-but-still-running statements would keep holding their locks.

Why bundling is the interesting part

Throughput on this write path is WAL flush rate × messages per flush. An fsync costs roughly the same regardless of how much it carries, so the lever is messages per commit, and there are two ways to pull it. A client can push larger batches, which makes each segment bigger. Or many partitions can share one commit, which is what fusion does without the client knowing.

That is also why there is no server-side segment size setting. The size a segment reaches is exactly what arrived during the previous flush’s round trip: short round trips give small segments, and load gives large ones, automatically. Setting a threshold would either add latency at low rates or cap the benefit at high ones.