Repository navigation
fix(storage-postgres): number stream records after their writes commit - #819
Conversation
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.
| /// 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( |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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()))?; |
There was a problem hiding this comment.
This will fail a GetRecords() call, correct, even if there are numbered records available to read. Is that the behavior we want?
There was a problem hiding this comment.
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 |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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 \ |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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
left a comment
There was a problem hiding this comment.
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
left a comment
There was a problem hiding this comment.
Thanks for addressing the feedback: I think this is good for now.
What
On PostgreSQL, a stream record now gets its sequence number after its write commits.
stream_pending, in its own transaction and with no number. Migration 7 adds the table.FOR NO KEY UPDATE SKIP LOCKED. It moves a batch of the shard's committed pending records tostream_records, in the order the writes stored them. Each record gets the nextstream_seqvalue. A batch holds at most 1,000 records and stops adding records after 1 MiB of record JSON text. Arecord_bytescolumn holds the length of the text of each record.stream_seqand on the id sequence ofstream_pending. The order of a shard's records depends on both. It also makes autovacuum takestream_pendingonce 1,000 of its rows are dead.write_stream_recordandnext_sequence_numberreturn an error on PostgreSQL. Nothing calls them.Why
Closes #808. Refs #660.
A write drew its sequence number with
nextvalinside its own transaction.nextvalassigns 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):
Each write stores its record's data in
stream_pending, and the numbering copies it tostream_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:
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:
close_selected_shardsmust move the pending records of each closed shard on its own connection before it readsMAX(sequence_number).number_shard_batchcannot do it there, because the close holds the shard's row and the call finds the shard busy. The close's data transaction must also be READ COMMITTED. Its tests that readstream_recordsdirectly must callnumber_pending_stream_records_for_testsfirst.latest_sequence_numberfirst, and that call now numbers the shard.created_atis no longer in sequence order on a shard, so the seek can return a record created before the timestamp whose write committed after it.Testing done
New tests in
crates/storage-postgres/tests/stream_commit_order.rs:A new test in
crates/storage-postgres/tests/stream_shard_retention.rsshows 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-mongodbandextenddb-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:
pg_column_size(record_data)in place ofrecord_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 --testsagainstpgvector/pgvector:pg16: 344 tests.cargo test -p extenddb-storage-mongodbwith the fivemigrations_*tests, against a single-nodemongo:7replica set.cargo test -p extenddb-storage-cassandra --test migrations --test migrations_stoppedagainstcassandra:4.1with PasswordAuthenticator.cargo +1.97.0 clippy --all-targets -- -D warningsandcargo fmt --all --check.devtools/schema-dumpafterextenddb init, against the newdocs/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 warningsandcargo fmt --all --check.record_bytesline todocs/schema/postgres.txtby hand. I did not rundevtools/schema-dumpagain.cargo test --workspacepassed except the filesystem tests incrates/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,/,/tmpand/local/homebelong to uid 65534. This change does not touch that crate.Checklist
cargo test --workspace)cargo fmt --check)cargo clippy -- -W clippy::pedantic)Storagetrait, auth model, on-diskformat, 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_THROUGHis 0), and the v1.0 release PR raisesRELEASED_THROUGHto 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 runextenddb migrate --yesonce 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.