Skip to content

fix(storage-postgres): number stream records after their writes commit - #819

Merged
LeeroyHannigan merged 5 commits into
mainfrom
fix/postgres-stream-commit-order
Nov 5, 2026
Merged

LeeroyHannigan merged 5 commits into
mainfrom
fix/postgres-stream-commit-order

Conversation

@LeeroyHannigan

@LeeroyHannigan LeeroyHannigan commented Nov 4, 2026 •

Copy link
Copy Markdown
Collaborator

What

On PostgreSQL, a stream record now gets its sequence number after its write commits.

  • Each write stores its stream record in a new table, stream_pending, in its own transaction and with no number. Migration 7 adds the table.
  • A numbering step locks the shard's row with FOR NO KEY UPDATE SKIP LOCKED. It moves a batch of the shard's committed pending records to stream_records, in the order the writes stored them. Each record gets the next stream_seq value. A batch holds at most 1,000 records and stops adding records after 1 MiB of record JSON text. A record_bytes column holds the length of the text of each record.
  • GetRecords, and GetShardIterator with LATEST, first run the numbering step on their shard for one batch. When another transaction numbers the shard, the read releases its pool connection, pauses and looks again, for 1 s at most. If the numbering fails, the read logs a warning and returns the records numbered before.
  • A read does not return every committed record in three cases. A later read returns the rest, with higher numbers.
    • More records of the shard wait than one batch takes.
    • Another transaction numbers the shard for longer than 1 s.
    • The numbering of the read fails.
  • A worker numbers every shard with pending records every 250 ms. It skips a shard that another transaction numbers.
  • The sweep of orphaned shards keeps the shards of a deleted table while one of its records is pending.
  • Migration 7 sets a cache of 1 on stream_seq and on the id sequence of stream_pending. The order of a shard's records depends on both. It also makes autovacuum take stream_pending once 1,000 of its rows are dead.
  • write_stream_record and next_sequence_number return an error on PostgreSQL. Nothing calls them.
  • Migration 7 does not apply to SQLite, Cassandra or MongoDB. Each backend records its reason.
  • The docs cover the change: the schema file, the upgrade manual, the design guide, the storage extension manual and the troubleshooting guide. The contributing example now adds migration 8, and the sample outputs show migration 7.

Why

Closes #808. Refs #660.

A write drew its sequence number with nextval inside its own transaction. nextval assigns numbers in the order of the calls, but the transactions that drew them can commit in another order. A consumer could read a higher number and then never receive a lower number that committed later. RFC-0003 criterion 5.2 forbids this skip. #808 has the reproduction and an example.

With this change, the numberings of one shard run one after another, because each holds the shard's row until it commits. A numbering draws numbers only for committed records, after every earlier numbering of the shard committed. So a number becomes visible only after every lower number on its shard. Two writes to one item take the item's row lock in turn, so their records keep the order of the writes. A writer locks the shard's row only FOR KEY SHARE, through the foreign key of its insert. That mode does not conflict with the numbering lock, so writes never wait for a numbering.

Database-layer measurements, with pgbench and without the ExtendDB server (details in #808):

Writers Shards written Writes/s on main Writes/s with the fix Records missed on main Records missed with the fix
16 4 7,030 6,852 33 of 105,423 0 of 102,545
64 4 14,031 13,004 353 of 208,778 0 of 193,698
16 1 7,936 7,537 28 of 119,004 0 of 113,032
64 1 14,422 14,357 491 of 215,402 0 of 214,048

Each write stores its record's data in stream_pending, and the numbering copies it to stream_records. In those database-layer runs, WAL per write rose from 701 to 994 bytes at 16 writers, and from 755 to 1,052 bytes at 64 writers.

Server-level runs used release builds of main and of this branch, PostgreSQL 16 on 127.0.0.1, and 64 writers. The writers sent PutItem for 30 s to a table with a NEW_IMAGE stream. One consumer per shard paged GetRecords from TRIM_HORIZON at the same time:

Run Writes acknowledged Records missed Server CPU per write
main, round 1 49,477 185 0.565 ms
main, round 2 22,218 106 0.605 ms
fix, round 1 61,570 0 0.558 ms
fix, round 2 52,386 0 0.554 ms

In those server-level runs, no consumer received a record twice. The writes per second are not comparable across the runs, because of other jobs on the host. Its load average was 5 when the runs started and 21 when they ended. With the fix, the median GetRecords took 57 ms and 63 ms, against 24 ms and 54 ms on main. Part of that difference is the numbering that each read does first. These runs used the branch before 453c8c5 limited the numbering of a read to one batch.

Overlaps with open pull requests:

Testing done

New tests in crates/storage-postgres/tests/stream_commit_order.rs:

  • A write that waits at its commit while a second write commits and a consumer reads: the next read returns the waiting write.
  • Writers on four hot items with a paging consumer: every record once, and each item's records chain from old image to new image.
  • Puts, deletes and two-shard TransactWriteItems beside the worker, with two consumers: the same assertions, and both consumers read the same records in the same order.
  • Read after write, the LATEST iterator, and the worker with a locked shard.
  • Twenty readers that wait on a held shard with a pool of 10: a write still gets a connection at once.
  • Records of about 100 KB: each batch stops at 1 MiB, and one worker pass numbers them all.
  • 30 records of about 100 KB that PostgreSQL stores in under 10 KB each: they take three or more numbering transactions.
  • A backlog of 1,500 records: the first read numbers and returns 1,000 of them. The next read returns the other 500 in order.
  • A numbering that a trigger refuses: the read returns the records numbered before, and LATEST resolves.
  • A read during a numbering under a REPEATABLE READ default.

A new test in crates/storage-postgres/tests/stream_shard_retention.rs shows that a pending record keeps the shards of a deleted table.

On the merge of main at 38992eb (e174492), on 2026-11-05: fmt and clippy pass on extenddb-storage, extenddb-storage-postgres, extenddb-storage-sqlite, extenddb-storage-mongodb and extenddb-storage-cassandra. Their lib tests pass: 228, 75, 257, 166 and 128. CI runs the PostgreSQL integration tests.

I undid one part of the fix at a time, and each time the test that covers that part failed:

  • Main's write path took the place of the new one.
  • GetRecords read the shard without numbering it first.
  • A LATEST iterator resolved without numbering the shard first.
  • The shard lock waited instead of skipping a locked row.
  • A read gave up on a busy shard at once.
  • The first version of the numbering came back. It had no byte limit, and a waiting read held its pool connection.
  • The orphan sweep ignored pending records.
  • A read numbered batches until its shard had no pending records.
  • A numbering failure before a read failed the read.
  • A batch added up pg_column_size(record_data) in place of record_bytes.

With main's write path, the test with hot items failed in 3 of 3 runs. Each run missed 7 to 9 of 1,200 records.

Commands that passed:

  • cargo test -p extenddb-storage-postgres --tests against pgvector/pgvector:pg16: 344 tests.
  • cargo test -p extenddb-storage-mongodb with the five migrations_* tests, against a single-node mongo:7 replica set.
  • cargo test -p extenddb-storage-cassandra --test migrations --test migrations_stopped against cassandra:4.1 with PasswordAuthenticator.
  • cargo +1.97.0 clippy --all-targets -- -D warnings and cargo fmt --all --check.
  • devtools/schema-dump after extenddb init, against the new docs/schema/postgres.txt.

After the review changes, on 1.97.0 against pgvector/pgvector:pg16:

  • cargo test -p extenddb-storage-postgres, unit and integration tests: 347 passed, 3 ignored.
  • cargo clippy -p extenddb-storage-postgres --all-targets -- -D warnings and cargo fmt --all --check.
  • I added the record_bytes line to docs/schema/postgres.txt by hand. I did not run devtools/schema-dump again.

cargo test --workspace passed except the filesystem tests in crates/backup-store/tests/conformance.rs. Those tests refuse a backup root under a directory that neither root nor the extenddb user owns. On this host, /, /tmp and /local/home belong to uid 65534. This change does not touch that crate.

Checklist

  • I have read CONTRIBUTING.md
  • All tests pass (cargo test --workspace)
  • Code is formatted (cargo fmt --check)
  • Clippy is clean (cargo clippy -- -W clippy::pedantic)
  • I have added or updated tests for new functionality
  • I have updated documentation if behavior changed
  • Breaking changes are noted below (if any)
  • If this changes the wire protocol, Storage trait, auth model, on-disk
    format, or public CLI surface, an RFC has been accepted or is linked
    below. Otherwise, an ADR captures the decision (link below).

ADR / RFC: RFC-0003, criterion 5.2 (docs/rfcs/0003-backend-acceptance-criteria.md). #808 lists the designs that were not taken and the reason for each.

Breaking changes

Migration 7 is breaking on PostgreSQL, as migrations 1 to 6 are. A server binary without migration 7 refuses a catalog that has it, with exit status 4. No release carries migrations yet (RELEASED_THROUGH is 0), and the v1.0 release PR raises RELEASED_THROUGH to 7. On SQLite, Cassandra and MongoDB the ledger records migration 7 as not applicable. A deployment that a build of main created at migration 6 must run extenddb migrate --yes once on every backend.


By submitting this pull request, I confirm that my contribution is made under
the terms of the Apache License 2.0 and I agree to the Developer Certificate of
Origin (DCO). See CONTRIBUTING.md for details.

A write drew its stream sequence number with nextval inside its own
transaction. nextval hands out numbers in the order of the calls, and
the transactions that drew them can commit in another order: a write
could commit after another write to the same shard that drew a higher
number. A consumer that read the higher number in
between then asked for the records after it and never received the
first write's record. RFC-0003 criterion 5.2 forbids that skip.

A write now stores its record in a new table, stream_pending, with no
number. A numbering step locks the shard's row FOR NO KEY UPDATE and
moves the shard's committed pending records to stream_records in the
order they were stored. Each record gets the next stream_seq value, in
the sequence_number column and in its SequenceNumber. The numberings of
one shard run one after another, so a number becomes visible only after
every lower number on its shard.

GetRecords and a LATEST shard iterator number their shard before they
read it, so a read still returns every write that committed before it.
Every 250 ms a worker numbers the pending records of every shard and
skips a shard that another transaction is numbering. The records of a
shard that nobody reads are numbered and trimmed as before. A writer
locks the shard row only FOR KEY SHARE, through the foreign key, and
that mode does not conflict with the numbering lock: writes never wait
for a numbering. The numbering transaction is READ COMMITTED whatever
the database default. Under a REPEATABLE READ default its snapshot
would predate its wait for the lock, and moving records that the
earlier numbering had moved would fail with a serialization error.

Migration 6, stream_pending, adds the table to the PostgreSQL data
database. It is breaking, as migrations 1 to 5 are, and it does not
apply to SQLite, Cassandra or MongoDB. The sweep of orphaned shards
keeps the shards of a deleted table while one of its records is still
pending.

Fixes #808
…t a connection

A reader that found its shard being numbered waited on the shard row
with a pool connection held. With as many readers waiting as the pool
had connections, a write to any table waited until the numbering
committed. A numbering's hold on the row also grew with its records'
bytes, because each record is copied to stream_records: 1,000 records
of 100 KB took about 2.4 s.

The shard lock now never waits. A read that finds the shard busy gives
its connection back, pauses for 1 ms to 16 ms, and looks again, for
1 s at most. It still reads every write that committed before it,
because it stops only when no committed record is pending or when its
own numbering took the last of them. A batch stops adding records once
the records before the next one hold 1 MiB, and it always takes one
record, so the statement now reports whether records remain.

The worker finds the shards with pending records through the primary
key, one step per shard, instead of reading the whole table every
250 ms. A failure on one shard no longer stops the pass for the others.
Migration 6 now sets autovacuum to take stream_pending once 1,000 rows
are dead, and it sets a cache of 1 on stream_seq and
stream_pending_id_seq. The order of a shard's records depends on both.

write_stream_record and next_sequence_number, which nothing calls, now
return an error on PostgreSQL, as write_stream_record does on Cassandra
and MongoDB. A number drawn before a record is stored would bring back
the skip.

The docs state the limit of 10,000 records per read for a LATEST
iterator, add a troubleshooting entry for a long transaction on the
data database, and remove two entries for messages that no code logs.
The contributing example now adds migration 7, and the sample outputs
show migration 6.

New tests cover readers waiting on a held shard with a pool of 10, the
batch bound with records of about 100 KB, and puts, deletes and
two-shard TransactWriteItems beside the worker and two consumers.
@LeeroyHannigan LeeroyHannigan added this to the 1.0 milestone Nov 4, 2026
@LeeroyHannigan LeeroyHannigan added the bug Something isn't working label Nov 4, 2026
@LeeroyHannigan LeeroyHannigan added backend:postgresql Related to the PostgreSQL backend specifically. v1 Needed for the 1.0 release labels Nov 4, 2026
@LeeroyHannigan LeeroyHannigan added the release-blocker Must ship for its milestone's release label Nov 4, 2026
@jcshepherd jcshepherd self-assigned this Nov 4, 2026
/// shard, the read waits for it with no connection held, looking again
/// after a pause, for [`NUMBER_WAIT`] at most. With nothing pending this
/// costs one indexed `EXISTS`.
async fn number_before_read(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is on the critical path of GetRecords(), correct? From the performance notes at the top of the file, a stream with a significant backlog of unnumbered records could take > 20s to renumber, depending on the record size: GetRecords() would be very slow or simply timeout. If we know we're in the path of a synchronous call like GetRecords, would it make sense to renumber (say) only min(1000, NUMBER_LIMIT) to limit the time the API call is waiting for any data, or would that create other problems?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it is on the GetRecords path, and the same holds for GetShardIterator with LATEST. In 453c8c5 a read numbers one batch before it reads the shard: 1,000 records, or fewer once they hold 1 MiB of JSON text. The byte cap holds a batch of 100 KB records to about ten. A read can also wait up to 1 s (NUMBER_WAIT) for another transaction that numbers the shard. The worker still numbers up to 10,000 records of each shard every 250 ms.

The cost is the one in your comment on line 205: a read behind a larger backlog does not return every committed record. The rest get higher numbers from a later read or the worker, and a later read returns them. A LATEST iterator taken then can return records whose writes committed before it. The doc comment and the design guide now say so.

New test: a_read_numbers_one_batch_of_a_backlog. With 1,500 records pending, the first read numbers and returns 1,000, and the next read returns the other 500 in order.

.bind(shard_id)
.fetch_one(&self.data_pool)
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will fail a GetRecords() call, correct, even if there are numbered records available to read. Is that the behavior we want?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it did, and that is not the behavior we want. In 453c8c5 a numbering failure before a read is logged as a warning (Stream record numbering before a read of shard <shard> failed: <error>), and GetRecords returns the records numbered before it. GetShardIterator with LATEST does the same. The records left pending get higher numbers from a later read or the worker.

New test: a_read_whose_numbering_fails_returns_the_numbered_records. A trigger on stream_records makes the numbering fail. The read returns the record numbered before, and LATEST resolves to that record's number. docs/troubleshooting.md has an entry for the warning.


/// Number the shard's committed pending records before a read of it, up
/// to [`NUMBER_LIMIT`]. A read then returns every record whose write
/// committed before it started. When another transaction numbers the

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This claim (a read returns every record whose write committed before it) is not quite true: a severelly back-up shard or a numbering failure due to lock contention could both result in some written records not being returned by the read. Similar claim is made in 02-design-guide.md.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed, the claim was too strong. In 453c8c5 the doc comment and the design guide list the three cases in which a read does not return a record whose write committed before it: more records of the shard wait than one batch takes, another transaction numbers the shard for longer than NUMBER_WAIT (1 s), or the read's numbering fails. In each case those records get higher numbers later, and a later read returns them. The troubleshooting entry for the worker's error no longer says that consumers still receive every record. It says when they do.

/// first `$2` it looked at.
const NUMBER_PENDING: &str = "\
WITH first AS ( \
SELECT id, pg_column_size(record_data) AS size FROM stream_pending \

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

On Postgres, pg_column_size() returns the size-as-stored, not the logical data size. E.g., if compression is enabled on Postgres, pg_column_size() value for string columns may be significantly smaller than the uncompressed string length. Is that okay for what you're doing here?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, it was not. The copy into stream_records decompresses and writes each record, so its time follows the uncompressed length. In 453c8c5 stream_pending has a record_bytes column, the length of the record's JSON text as the write sent it, and a batch sums that. The write serializes the record once and binds the text, which the insert casts to jsonb.

New test: a_numbering_batch_counts_compressible_records_at_their_text_length. It writes 30 records of about 100 KB that PostgreSQL stores in under 10 KB each, and checks that they take three or more numbering transactions. Summed with pg_column_size, all 30 went into one.

@jcshepherd jcshepherd left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Main concern is with the large renumbering limit and its impact on GetRecords().

GetRecords and a LATEST shard iterator numbered up to 10,000 pending
records of their shard before they read it. On PostgreSQL 16 a batch of
1,000 records took 26 ms at about 1 KB a record and 2.4 s at 100 KB.
At that rate, 10,000 pending records of 100 KB would hold a read for
about 24 s. A read now numbers one batch: 1,000 records, or fewer once
they hold 1 MiB. The numbering worker still numbers up to 10,000
records of each shard in one pass.

A numbering failure before a read failed the read, even when the shard
held numbered records. The read now logs the failure as a warning and
returns the records numbered before it. The records left pending get
higher numbers from a later read or the worker.

A batch summed pg_column_size(record_data), which is the size after
TOAST compression. The copy into stream_records decompresses and writes
each record, so its time follows the uncompressed length. stream_pending
gets a record_bytes column, the length of the record's JSON text as the
write sent it, and a batch sums that. The write serializes the record
once and sends the text, which the insert casts to jsonb.

The doc comments, the design guide and the troubleshooting guide now
name the three cases in which a read misses a committed record: more
records wait than one batch takes, another transaction numbers the
shard for longer than one second, or the read's numbering fails. A new
troubleshooting entry covers the warning.

New tests in stream_commit_order.rs: a read of a 1,500-record backlog
numbers 1,000 records, and the next read returns the other 500 in
order. A read whose numbering a trigger refuses returns the numbered
records, and LATEST still resolves. 30 records of about 100 KB that are
stored in under 10 KB each take three or more numbering transactions.
Each test fails with its part of this change undone.
crates/storage/src/migrations/registry.rs: main added migration 6,
BackupCatalogIndex (#671). This branch added StreamPending as 6. The
merged list keeps main's 6 and numbers StreamPending 7. The
RELEASED_THROUGH comment names 7 as the id the v1.0 release PR raises
it to.

crates/storage-postgres/src/migrations/mod.rs: main added the arm for
BackupCatalogIndex, whose catalog body is 0006_backup_catalog_index.sql,
and listed it in the transactional arm of runs_in. This branch added
the arm for StreamPending with a data body and listed it in runs_in.
The merged file has both arms and both ids in runs_in. The data body is
renamed 0007_stream_pending.sql, and its header comment names
migration 7.

crates/storage-sqlite/src/migrations/mod.rs,
crates/storage-mongodb/src/migrations/mod.rs,
crates/storage-cassandra/src/migrations/mod.rs: main added the arm that
applies BackupCatalogIndex on each backend. This branch added the arm
that records StreamPending as not applicable with the backend's reason.
Each merged file has both arms.

crates/storage-mongodb/tests/migrations_fresh.rs: main replaced the
version literals with MigrationId::LATEST.number() and added migration
6 to the applied steps and to the applied ledger rows. This branch wrote
6 as the version and added its migration as a not applicable step and
row. The merged file keeps main's LATEST reads, lists migration 7 as an
applied step and a not applicable ledger row beside 6, names 3, 4 and 7
as the ids that do not apply on MongoDB, moves the ledger row for a
later binary from 7 to 8, and asserts LATEST is 7.

crates/storage-mongodb/tests/migrations_guards.rs,
crates/storage-mongodb/tests/migrations_holders.rs,
crates/storage-mongodb/tests/migrations_locks.rs,
crates/storage-mongodb/tests/migrations_released.rs: main reads the
version from MigrationId::LATEST where this branch wrote 6. The merged
files keep main's reads. The step counts in migrations_locks.rs and
migrations_released.rs move from 7 to 8, because migration 1 records
two rows and ids 2 to 7 one each.

CONTRIBUTING.md: both sides moved the worked example from migration 6
to 7, and main noted that 6 is backup_catalog_index. The merged example
is migration 8, notes that 6 and 7 are backup_catalog_index and
stream_pending, and lists StreamPending in the runs_in excerpt.

docs/manuals/07-upgrade-manual.md: main described migration 6 in the
status paragraph and added a section on it. This branch described its
migration in the same paragraph. The merged paragraph says the release
carries migrations 1 to 7 and describes migration 7 after main's
sentence on 6, the breaking range is 1 to 7, and main's section stays.

Both sides moved the sample outputs in docs/getting-started.md and in
the quickstart, Linux install and macOS install manuals from migration
5 to 6. The merged files show 7. The design guide, the comment in
stream_engine.rs on the sequence cache and the comment in
tests/test_cli_vector_catalog_migration.py on the data database bodies
name migration 7.
test_a_migration_6_that_fails_in_its_backfill_stays_pending_and_resumes
expected migration 6 to be the only pending migration of a catalog at
migration 5. This branch adds migration 7, so such a catalog has 6 and
7 pending. The check now asks for migration 6 among the pending
migrations and for no pending migration below it.

@jcshepherd jcshepherd left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for addressing the feedback: I think this is good for now.

@LeeroyHannigan
LeeroyHannigan added this pull request to the merge queue Nov 5, 2026
Merged via the queue into main with commit b12a70e Nov 5, 2026
44 of 46 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

backend:postgresql Related to the PostgreSQL backend specifically. bug Something isn't working release-blocker Must ship for its milestone's release v1 Needed for the 1.0 release

Projects

None yet

Development

Successfully merging this pull request may close these issues.

PostgreSQL: a stream consumer can skip a record that commits after a higher-numbered one

2 participants