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

CONTRIBUTING.md

@@ -341,27 +341,28 @@ SQLite, PostgreSQL, Cassandra and MongoDB migrate through this framework.
 
 ### Example: a nullable column on `tables`
 
-Migration 7, `table_note` (7 being the next free id at the time of writing;
-migration 6 is `backup_catalog_index`), adds a nullable `note TEXT` column to
-the catalog's `tables` table on SQLite and PostgreSQL.
+Migration 8, `table_note` (8 being the next free id at the time of writing;
+migrations 6 and 7 are `backup_catalog_index` and `stream_pending`), adds a
+nullable `note TEXT` column to the catalog's `tables` table on SQLite and
+PostgreSQL.
 
 In `crates/storage/src/migrations/registry.rs`, at the end of the
 `migration_ids!` list:
 
 ```rust
     /// Adds a nullable `note` column to the `tables` catalog table.
-    TableNote = 7, "table_note",
+    TableNote = 8, "table_note",
         "add a nullable note column to the tables catalog";
 ```
 
 PostgreSQL has `ADD COLUMN IF NOT EXISTS`, so its body is a file,
-`crates/storage-postgres/migrations/0007_table_note.sql`:
+`crates/storage-postgres/migrations/0008_table_note.sql`:
 
 ```sql
 -- Copyright 2026 ExtendDB contributors
 -- SPDX-License-Identifier: Apache-2.0
 --
--- Migration 7, table_note, PostgreSQL catalog database: a nullable note
+-- Migration 8, table_note, PostgreSQL catalog database: a nullable note
 -- column on tables. IF NOT EXISTS makes a second run change nothing.
 
 ALTER TABLE tables ADD COLUMN IF NOT EXISTS note TEXT;
@@ -376,7 +377,7 @@ In `crates/storage-postgres/src/migrations/mod.rs`, the arm in `migration()`:
             bodies: vec![file_body!(
                 StoreKind::Catalog,
                 "migrations",
-                "0007_table_note.sql"
+                "0008_table_note.sql"
             )],
         },
 ```
@@ -389,19 +390,20 @@ and the id in the transactional arm of `runs_in()`:
         | MigrationId::SortKeyDefinitions
         | MigrationId::DropContinuousBackups
         | MigrationId::BackupCatalogIndex
+        | MigrationId::StreamPending
         | MigrationId::TableNote => RunsIn::Transaction,
 ```
 
 SQLite has no `ADD COLUMN IF NOT EXISTS`, and a file with a plain
 `ADD COLUMN` fails when it runs a second time, so the SQLite body is code
 that checks for the column first.
-`crates/storage-sqlite/src/migrations/m0007_table_note.rs`:
+`crates/storage-sqlite/src/migrations/m0008_table_note.rs`:
 
 ```rust
 // Copyright 2026 ExtendDB contributors
 // SPDX-License-Identifier: Apache-2.0
 
-//! Migration 7, `table_note`, SQLite: a nullable `note` column on `tables`.
+//! Migration 8, `table_note`, SQLite: a nullable `note` column on `tables`.
 //!
 //! SQLite has no `ADD COLUMN IF NOT EXISTS`, so the body reads the table's
 //! columns first and adds the column only when it is missing. A second run
@@ -413,10 +415,10 @@ use extenddb_storage::migrations::{CodeBody, Done, MigrationError};
 use super::SqliteStores;
 
 /// This file, whose SHA-256 the checksum test pins.
-pub(crate) const SOURCE: &str = include_str!("m0007_table_note.rs");
+pub(crate) const SOURCE: &str = include_str!("m0008_table_note.rs");
 
 /// Recorded as this migration's checksum. Changes only with a new migration.
-const FINGERPRINT: &str = "0007_table_note/sqlite/1";
+const FINGERPRINT: &str = "0008_table_note/sqlite/1";
 
 pub(crate) struct TableNote;
 
@@ -455,7 +457,7 @@ beside the others:
 ```rust
 #[rustfmt::skip]
 #[allow(clippy::all, deprecated)]
-mod m0007_table_note;
+mod m0008_table_note;
 ```
 
 and the arm in `migration()`:
@@ -466,8 +468,8 @@ and the arm in `migration()`:
             convergent: true,
             bodies: vec![StoreBody::code(
                 StoreKind::Catalog,
-                &m0007_table_note::TableNote,
-                m0007_table_note::SOURCE,
+                &m0008_table_note::TableNote,
+                m0008_table_note::SOURCE,
             )],
         },
 ```
@@ -479,8 +481,8 @@ rows of `tables`:
 
 ```rust
 #[tokio::test]
-async fn migration_7_adds_the_note_column_to_tables() {
-    let db = temp_db("m7");
+async fn migration_8_adds_the_note_column_to_tables() {
+    let db = temp_db("m8");
     load_dump(&db, DEV_0_1_14).await;
     run(&db).await.expect("migrate");
     let mut text = dump(
@@ -490,12 +492,12 @@ async fn migration_7_adds_the_note_column_to_tables() {
     )
     .await;
     text.push_str(&dump(&db, "SELECT table_name, note FROM tables ORDER BY table_name").await);
-    golden("0007_table_note", &text);
+    golden("0008_table_note", &text);
 }
 ```
 
-`EXTENDDB_BLESS_EXPECTED=1 cargo test -p extenddb-storage-sqlite --lib migration_7`
-writes `crates/storage-sqlite/tests/fixtures/golden/0007_table_note.txt`, one
+`EXTENDDB_BLESS_EXPECTED=1 cargo test -p extenddb-storage-sqlite --lib migration_8`
+writes `crates/storage-sqlite/tests/fixtures/golden/0008_table_note.txt`, one
 tab-separated line per row under each query:
 
 ```
@@ -510,15 +512,15 @@ The PostgreSQL body is a file, so it has no golden test.
 The example declares `Kind::Breaking`. A nullable column can be additive by
 the rule in step 4, but declaring it additive needs the old-binary job, which
 this example does not add. As a breaking migration it makes a 1.x binary
-without migration 7 refuse a catalog that has it, and a rollback restores the
+without migration 8 refuse a catalog that has it, and a rollback restores the
 backup.
 
 The same PR changes the code that reads and writes `tables.note` in each
 backend's catalog store, and commits the `docs/schema/sqlite.txt` and
 `docs/schema/postgres.txt` that the integration jobs regenerate. It changes
 no per-table DDL, because `tables` is a catalog table, and adds no `RELEASED`
-entry. `extenddb migrate --list` then shows migration 7, and on a catalog at
-migration 6 `extenddb migrate` lists migration 7 as pending and exits 3.
+entry. `extenddb migrate --list` then shows migration 8, and on a catalog at
+migration 7 `extenddb migrate` lists migration 8 as pending and exits 3.
 
 ### Running the migration tests
 

crates/storage-cassandra/src/migrations/mod.rs

@@ -150,6 +150,10 @@ pub fn migration(id: MigrationId) -> Migration<CassandraStores> {
                 m0006_backup_catalog_index::SOURCE,
             )],
         },
+        MigrationId::StreamPending => Migration::NotApplicable(
+            "the stream_pending table belongs to the PostgreSQL backend, and Cassandra numbers its \
+             stream records with the server's clock (#660)",
+        ),
     }
 }
 

crates/storage-cassandra/tests/direct/backup_engine.rs

@@ -4164,12 +4164,17 @@ async fn test_a_migration_6_that_fails_in_its_backfill_stays_pending_and_resumes
     let (engine, account_id, _drop_keyspaces) = isolated_engine("mf", true).await;
     let catalog = engine.catalog_keyspace();
     let prefix = prefix_of(&engine);
+    // Migration 6 is pending, and so is every later migration this binary
+    // carries.
     let behind_at_5 = |status: CatalogStatus| match status {
         CatalogStatus::Behind { version, pending } => {
             version == Some(MigrationId::DropContinuousBackups.number())
                 && pending
                     .iter()
-                    .all(|p| p.id == MigrationId::BackupCatalogIndex.number())
+                    .any(|p| p.id == MigrationId::BackupCatalogIndex.number())
+                && pending
+                    .iter()
+                    .all(|p| p.id >= MigrationId::BackupCatalogIndex.number())
         }
         _ => false,
     };

crates/storage-mongodb/src/migrations/mod.rs

@@ -160,6 +160,10 @@ pub fn migration(id: MigrationId) -> Migration<MongoStores> {
                 m0006_backup_catalog_index::SOURCE,
             )],
         },
+        MigrationId::StreamPending => Migration::NotApplicable(
+            "a MongoDB write draws its stream sequence number from its shard's counter inside its \
+             transaction, so the records of a shard commit in number order",
+        ),
     }
 }
 

crates/storage-mongodb/tests/migrations_fresh.rs

@@ -322,6 +322,7 @@ async fn a_fresh_store_gets_every_migration_and_the_expected_definitions() {
             (4, d.target.catalog.clone()),
             (5, d.target.catalog.clone()),
             (6, d.target.catalog.clone()),
+            (7, d.target.catalog.clone()),
         ]
     );
     let not_applicable: Vec<u32> = report
@@ -330,7 +331,7 @@ async fn a_fresh_store_gets_every_migration_and_the_expected_definitions() {
         .filter(|s| s.outcome == Outcome::NotApplicable)
         .map(|s| s.id)
         .collect();
-    assert_eq!(not_applicable, [3, 4], "{report:?}");
+    assert_eq!(not_applicable, [3, 4, 7], "{report:?}");
     assert_eq!(
         d.status().await,
         CatalogStatus::Current {
@@ -348,6 +349,7 @@ async fn a_fresh_store_gets_every_migration_and_the_expected_definitions() {
             (4, Outcome::NotApplicable),
             (5, Outcome::Applied),
             (6, Outcome::Applied),
+            (7, Outcome::NotApplicable),
         ]
     );
     assert!(
@@ -442,8 +444,8 @@ async fn a_second_run_applies_nothing_and_reconverge_keeps_the_rows() {
 
     let reconverged = d.reconverge().await.expect("reconverge");
     // Migration 1's steps in the catalog and the data database, and
-    // migrations 5's and 6's in the catalog: 2 is not convergent, and 3 and 4
-    // do not apply on MongoDB.
+    // migrations 5's and 6's in the catalog: 2 is not convergent, and 3, 4
+    // and 7 do not apply on MongoDB.
     let ran: Vec<(u32, String)> = reconverged
         .applied
         .iter()
@@ -533,19 +535,19 @@ async fn a_changed_checksum_a_newer_catalog_and_an_older_release_are_reported()
         doc! { "_id": id, "name": "later", "kind": kind, "checksum": "c", "outcome": "applied",
         "notes": "", "applied_at": bson::DateTime::now(), "applied_by": "9.9.9" }
     };
-    ledger.insert_one(row(7, "breaking")).await.expect("insert");
+    ledger.insert_one(row(8, "breaking")).await.expect("insert");
     let status = d.status().await;
     assert_eq!(status.label(), "AheadIncompatible");
     assert_eq!(status.exit_code(), exit::AHEAD_INCOMPATIBLE);
     ledger
-        .delete_one(doc! { "_id": 7_i64 })
+        .delete_one(doc! { "_id": 8_i64 })
         .await
         .expect("delete");
-    ledger.insert_one(row(7, "additive")).await.expect("insert");
+    ledger.insert_one(row(8, "additive")).await.expect("insert");
     assert_eq!(d.status().await.label(), "AheadCompatible");
     assert!(d.migrate().await.expect("migrate").applied.is_empty());
     ledger
-        .delete_one(doc! { "_id": 7_i64 })
+        .delete_one(doc! { "_id": 8_i64 })
         .await
         .expect("delete");
 
@@ -626,6 +628,6 @@ async fn a_ledger_document_this_binary_cannot_read_stops_the_status() {
         .expect_err("an error");
     migrations.close().await;
     assert!(error.to_string().contains("\"sideways\""), "{error}");
-    assert_eq!(MigrationId::LATEST.number(), 6);
+    assert_eq!(MigrationId::LATEST.number(), 7);
     d.drop_databases().await;
 }

crates/storage-mongodb/tests/migrations_locks.rs

@@ -53,7 +53,7 @@ async fn two_migrators_started_together_apply_once_and_the_other_finds_nothing()
     applied.sort_unstable();
     assert_eq!(
         applied,
-        [0, 7],
+        [0, 8],
         "one applies every step and the other finds nothing"
     );
     assert_eq!(

crates/storage-mongodb/tests/migrations_released.rs

@@ -161,7 +161,7 @@ async fn every_released_catalog_upgrades_to_the_definitions_of_a_fresh_install()
             panic!("{state} is not behind");
         };
         assert_eq!(version, None, "{state} has no ledger");
-        assert_eq!(pending.len(), 7, "{state}");
+        assert_eq!(pending.len(), 8, "{state}");
         let report = d.migrate().await.expect("migrate");
         assert_eq!(
             (report.from, report.to),

crates/storage-postgres/data_migrations/0007_stream_pending.sql

@@ -0,0 +1,39 @@
+-- Copyright 2026 ExtendDB contributors
+-- SPDX-License-Identifier: Apache-2.0
+--
+-- Migration 7, stream_pending, PostgreSQL data database: the table a write
+-- stores its stream record in, inside the write's own transaction. The record
+-- gets its sequence number only after the write commits, when a numbering
+-- step moves the shard's committed records into stream_records in `id` order
+-- while it holds the shard's row (`number_shard_batch` in stream_engine.rs).
+-- A number drawn inside the write's transaction could commit after a higher
+-- number a reader had already read past (#808).
+--
+-- `id` orders the records of a shard. Two writes to one item take the item's
+-- row lock in turn, so the second inserts its record after the first commits
+-- and gets the higher `id`. IF NOT EXISTS makes a second run change nothing.
+--
+-- Every record is inserted here and deleted by its numbering, so the table
+-- holds few live rows and many dead ones. Autovacuum takes it once 1,000 rows
+-- are dead, whatever its size.
+
+CREATE TABLE IF NOT EXISTS stream_pending (
+    shard_id TEXT NOT NULL REFERENCES stream_shards(shard_id) ON DELETE CASCADE,
+    id BIGSERIAL,
+    table_id TEXT NOT NULL,
+    event_name TEXT NOT NULL,
+    record_data JSONB NOT NULL,
+    -- The byte length of the record's JSON text as the write sent it. A
+    -- numbering batch adds it up to bound the bytes it copies; it is the size
+    -- before TOAST compression, which pg_column_size does not give.
+    record_bytes BIGINT NOT NULL DEFAULT 0,
+    created_at TIMESTAMPTZ NOT NULL,
+    PRIMARY KEY (shard_id, id)
+) WITH (autovacuum_vacuum_scale_factor = 0, autovacuum_vacuum_threshold = 1000);
+
+-- The order of a shard's records needs both sequences to hand out values in
+-- the order of the calls across sessions, which a cache of more than one
+-- value per session breaks. Both have a cache of 1 when they are created;
+-- these statements set it again on a database where it was changed.
+ALTER SEQUENCE stream_seq CACHE 1;
+ALTER SEQUENCE stream_pending_id_seq CACHE 1;