---
title: "Life of a push"
description: "From an HTTP body to committed rows: grouping, two layers of deduplication, and why N partitions cost one commit and one fsync."
---

> Documentation Index
> Fetch the complete documentation index at: https://queenmq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 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:

And here is what happens behind it.

<pre class="mermaid">{`
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
`}</pre>

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.

> **Note**
>
> `QUEEN_V2_FUSION_FRAMES` is still parsed and still printed in the boot configuration block, but
> it is no longer a flush trigger. Segment size is emergent (see below). There is also an
> experiment-only frame floor (`QUEEN_V2_FUSION_MIN_FRAMES`, default 0 = off) that can delay a
> still-small segment to fatten it; the source marks it as not part of any compatibility promise,
> and with the default the dispatch logic is byte-identical to having no floor at all.

## 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.

```sql
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:

```sql
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:

```sql
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).

> **Note**
>
> Because the timestamp is stamped while the partition row lock is held, `created_at` is monotone
> per partition in commit order. Time-based retention and timestamp subscriptions are both a
> single forward primary-key walk as a direct result. The push path pays one `clock_timestamp()`
> call for it.

## 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

| What happens | What the client sees | What is in the database |
| --- | --- | --- |
| Duplicate inside one request body | later items report `duplicate` with the first item's id | one copy |
| Duplicate against a previous push, inside the window | `duplicate` plus the original message id | one copy; nothing was written |
| Duplicate whose original segment was already retained away | `duplicate` with the zero UUID as `messageId` | one copy; the original is gone |
| Statement timeout on the bundle | `error`, then `buffered` if the spool accepted it | nothing: the transaction rolled back |
| Statement timeout, spool write also failed | `failed` | nothing |
| Maintenance mode | `buffered` for every item | nothing until the drain runs |

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.

Source: https://queenmq.com/internals/life-of-a-push/index.mdx