diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c6b34b7f3..c0037018b 100755 --- a/CONTRIBUTING.md +++ b/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 diff --git a/crates/storage-cassandra/src/migrations/mod.rs b/crates/storage-cassandra/src/migrations/mod.rs index 3e2b6178c..fc3352c61 100644 --- a/crates/storage-cassandra/src/migrations/mod.rs +++ b/crates/storage-cassandra/src/migrations/mod.rs @@ -150,6 +150,10 @@ pub fn migration(id: MigrationId) -> Migration { 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)", + ), } } diff --git a/crates/storage-cassandra/tests/direct/backup_engine.rs b/crates/storage-cassandra/tests/direct/backup_engine.rs index 6c8c4da0d..15c1381d5 100644 --- a/crates/storage-cassandra/tests/direct/backup_engine.rs +++ b/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, }; diff --git a/crates/storage-mongodb/src/migrations/mod.rs b/crates/storage-mongodb/src/migrations/mod.rs index dfa01dcee..1c6337bc6 100644 --- a/crates/storage-mongodb/src/migrations/mod.rs +++ b/crates/storage-mongodb/src/migrations/mod.rs @@ -160,6 +160,10 @@ pub fn migration(id: MigrationId) -> Migration { 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", + ), } } diff --git a/crates/storage-mongodb/tests/migrations_fresh.rs b/crates/storage-mongodb/tests/migrations_fresh.rs index 10a8a2f33..9b5270035 100644 --- a/crates/storage-mongodb/tests/migrations_fresh.rs +++ b/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 = 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; } diff --git a/crates/storage-mongodb/tests/migrations_locks.rs b/crates/storage-mongodb/tests/migrations_locks.rs index 2978d3921..604cfda33 100644 --- a/crates/storage-mongodb/tests/migrations_locks.rs +++ b/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!( diff --git a/crates/storage-mongodb/tests/migrations_released.rs b/crates/storage-mongodb/tests/migrations_released.rs index 5108b49f4..d85f0342e 100644 --- a/crates/storage-mongodb/tests/migrations_released.rs +++ b/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), diff --git a/crates/storage-postgres/data_migrations/0007_stream_pending.sql b/crates/storage-postgres/data_migrations/0007_stream_pending.sql new file mode 100644 index 000000000..05757be69 --- /dev/null +++ b/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; diff --git a/crates/storage-postgres/src/data/tx_helpers.rs b/crates/storage-postgres/src/data/tx_helpers.rs index d17d0c437..69db42b1d 100755 --- a/crates/storage-postgres/src/data/tx_helpers.rs +++ b/crates/storage-postgres/src/data/tx_helpers.rs @@ -250,8 +250,9 @@ pub(super) async fn delete_item_in_tx( /// Write a stream record within an existing transaction. /// /// Builds the stream record from the old/new items and the `StreamCapture` -/// parameters, assigns a shard, generates a sequence number, and inserts -/// the record — all within the caller's transaction. +/// parameters, assigns a shard, and stores the record in `stream_pending`, +/// all within the caller's transaction. The record gets its sequence number +/// after the transaction commits (`number_shard_batch` in `stream_engine.rs`). /// /// The event type is determined from the old/new items: /// - old=None, new=Some → Insert @@ -337,20 +338,17 @@ pub(super) async fn write_stream_record_in_tx( let idx = (hash as usize) % shards.len(); let shard_id = &shards[idx].0; - // Generate monotonic sequence number within the transaction (CB-21). The - // same statement takes the record's one clock read: clock_timestamp() is - // the database clock when this statement runs, after the write's locking - // reads, while the column default (NOW()) is the transaction start, which - // in a long transaction sits seconds before the record. created_at and + // The record's one clock read: clock_timestamp() is the database clock + // when this statement runs, after the write's locking reads, while the + // column default (NOW()) is the transaction start, which in a long + // transaction sits seconds before the record. created_at and // ApproximateCreationDateTime both derive from this reading, and the // retention sweep compares created_at with the database's NOW(), so the // retention window is measured on the database clock alone. - let (seq_val, created_at): (i64, time::OffsetDateTime) = - sqlx::query_as("SELECT nextval('stream_seq'), clock_timestamp()") - .fetch_one(&mut **tx) - .await - .map_err(db_error)?; - let seq = format!("{seq_val:021}"); + let (created_at,): (time::OffsetDateTime,) = sqlx::query_as("SELECT clock_timestamp()") + .fetch_one(&mut **tx) + .await + .map_err(db_error)?; let approximate_creation_date_time = created_at.unix_timestamp(); let record = StreamRecord { @@ -364,25 +362,42 @@ pub(super) async fn write_stream_record_in_tx( keys, new_image, old_image, - sequence_number: seq, + // Set when the record is numbered, after this write commits. + sequence_number: String::new(), size_bytes: size, stream_view_type: capture.view_type, }, user_identity: capture.user_identity.clone(), }; - let record_json = - serde_json::to_value(&record).map_err(|e| StorageError::Internal(e.to_string()))?; + // The record's JSON text, sent once: the database parses it into the + // jsonb column, and its length in bytes is stored beside it. A numbering + // batch adds up that length (`NUMBER_BATCH_BYTES` in stream_engine.rs). + // It is the record's size before TOAST compresses it, which is what the + // numbering copies and rewrites; `pg_column_size` would give the + // compressed size. + let record_text = + serde_json::to_string(&record).map_err(|e| StorageError::Internal(e.to_string()))?; + let record_bytes = i64::try_from(record_text.len()).unwrap_or(i64::MAX); + // The record waits in stream_pending, with no sequence number, until this + // transaction commits. A reader of the shard, or the numbering worker, + // then moves it to stream_records under the next number + // (`number_shard_batch` in stream_engine.rs). A number drawn here with + // nextval would be handed out at once, in call order, and this write + // could commit after a write that drew a higher number. A reader that had + // already read the higher number would then never return this record + // (#808). sqlx::query( - "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \ - VALUES ($1, $2, $3, $4, $5, $6)", + "INSERT INTO stream_pending \ + (shard_id, table_id, event_name, record_data, record_bytes, created_at) \ + VALUES ($1, $2, $3, $4::jsonb, $5, $6)", ) - .bind(&record.dynamodb.sequence_number) .bind(shard_id) .bind(&key_info.table_id) .bind(format!("{:?}", record.event_name)) - .bind(&record_json) + .bind(&record_text) + .bind(record_bytes) .bind(created_at) .execute(&mut **tx) .await diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 7d0f9d28c..58c87ea13 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -700,6 +700,13 @@ impl ServerRuntimeHooks for PostgresRuntimeHooks { workers::stream_record_cleanup_worker(storage_for_stream, metrics, token).await; }); + // 3b. Stream record numbering worker + let storage_for_numbering = self.engine.clone(); + let token = ctx.shutdown.clone(); + let stream_numbering = tokio::spawn(async move { + workers::stream_record_numbering_worker(storage_for_numbering, token).await; + }); + // 4. Idempotency token cleanup worker let storage_for_token = self.engine.clone(); let metrics = ctx.metrics.clone(); @@ -752,6 +759,7 @@ impl ServerRuntimeHooks for PostgresRuntimeHooks { control_plane, table_size, stream_cleanup, + stream_numbering, idempotency_cleanup, ttl, vector_builds, diff --git a/crates/storage-postgres/src/migrations/mod.rs b/crates/storage-postgres/src/migrations/mod.rs index 1cf595927..bc5df0466 100644 --- a/crates/storage-postgres/src/migrations/mod.rs +++ b/crates/storage-postgres/src/migrations/mod.rs @@ -136,6 +136,15 @@ pub fn migration(id: MigrationId) -> Migration { "0006_backup_catalog_index.sql" )], }, + MigrationId::StreamPending => Migration::Applies { + kind: Kind::Breaking, + convergent: true, + bodies: vec![file_body!( + StoreKind::Data, + "data_migrations", + "0007_stream_pending.sql" + )], + }, } } @@ -158,7 +167,8 @@ pub(crate) const fn runs_in(id: MigrationId) -> RunsIn { | MigrationId::ResourceTagConditionKeys | MigrationId::SortKeyDefinitions | MigrationId::DropContinuousBackups - | MigrationId::BackupCatalogIndex => RunsIn::Transaction, + | MigrationId::BackupCatalogIndex + | MigrationId::StreamPending => RunsIn::Transaction, MigrationId::IndexTableBaseKey => RunsIn::Autocommit, } } diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs index da7fd122d..285d31b32 100755 --- a/crates/storage-postgres/src/stream_engine.rs +++ b/crates/storage-postgres/src/stream_engine.rs @@ -12,6 +12,7 @@ use extenddb_storage::error::StorageError; use extenddb_storage::util::{new_stream_label, parse_stream_arn, shard_in_stream, stream_arn}; use futures::future::BoxFuture; use sqlx::PgPool; +use std::time::Duration; use crate::PostgresEngine; use crate::stream_records_sql::{STREAM_RECORDS_AFTER, STREAM_RECORDS_FROM_START}; @@ -19,7 +20,301 @@ use crate::stream_records_sql::{STREAM_RECORDS_AFTER, STREAM_RECORDS_FROM_START} /// Number of fixed shards per stream (hash-based assignment). const SHARDS_PER_STREAM: u32 = 4; +/// The most pending records one numbering transaction moves. +const NUMBER_BATCH: i64 = 1_000; + +/// The record bytes after which a numbering transaction adds no more records +/// to its batch. A batch always takes its first record, so it ends at most one +/// record past this. The shard's row stays locked until the batch commits, and +/// the time a batch takes grows with its bytes, because each record is copied +/// to `stream_records`: 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. The bytes are the +/// `record_bytes` column, the length of each record's JSON text as the write +/// sent it. The copy decompresses and rewrites each record, so its time +/// follows that length, which TOAST compression does not shrink. +const NUMBER_BATCH_BYTES: i64 = 1 << 20; + +/// How many pending records the numbering worker numbers on one shard in one +/// pass, in batches. A read numbers one batch at most before it reads the +/// shard (`number_before_read`). +const NUMBER_LIMIT: u64 = 10_000; + +/// How long a read waits in all for other transactions that number its shard. +/// The read then reads the shard as it is: the records those transactions +/// number get higher numbers than any the read returns, so the next read +/// returns them. +const NUMBER_WAIT: Duration = Duration::from_secs(1); + +/// The longest pause between two looks at a shard that another transaction +/// numbers. The pauses start at 1 ms and double. +const NUMBER_PAUSE: Duration = Duration::from_millis(16); + +/// Lock a shard's row for a numbering transaction, or no row when another +/// transaction numbers the shard. `$2` is the table the shard must belong to, +/// or NULL for any. +const LOCK_SHARD: &str = "SELECT 1 FROM stream_shards \ + WHERE shard_id = $1 AND ($2::text IS NULL OR table_id = $2) \ + FOR NO KEY UPDATE SKIP LOCKED"; + +/// Whether the shard [`LOCK_SHARD`] names exists, without a lock. +const SHARD_EXISTS: &str = "SELECT EXISTS (SELECT 1 FROM stream_shards \ + WHERE shard_id = $1 AND ($2::text IS NULL OR table_id = $2))"; + +/// The shards that have pending records. Each step finds the next shard id +/// through the primary key of `stream_pending`, so the cost grows with the +/// shards that have pending records, not with the rows of the table. +const PENDING_SHARDS: &str = "\ + WITH RECURSIVE shards AS ( \ + (SELECT shard_id FROM stream_pending ORDER BY shard_id LIMIT 1) \ + UNION ALL \ + SELECT (SELECT p.shard_id FROM stream_pending p WHERE p.shard_id > shards.shard_id \ + ORDER BY p.shard_id LIMIT 1) \ + FROM shards WHERE shards.shard_id IS NOT NULL \ + ) \ + SELECT shard_id FROM shards WHERE shard_id IS NOT NULL"; + +/// Move a batch of shard `$1`'s committed records from `stream_pending` to +/// `stream_records`: the records with the lowest `id`, at most `$2` of them, +/// and no more once the records before one hold `$3` bytes or more. It draws +/// one `stream_seq` value for each record, sorts the values, and gives them +/// out in `id` order, so a record stored later gets a higher number. The +/// number goes in the `sequence_number` column and in the record's +/// `SequenceNumber`. Returns how many records moved and how many of the +/// first `$2` it looked at. +const NUMBER_PENDING: &str = "\ + WITH first AS ( \ + SELECT id, record_bytes AS size FROM stream_pending \ + WHERE shard_id = $1 ORDER BY id LIMIT $2 \ + ), sized AS ( \ + SELECT id, sum(size) OVER (ORDER BY id) - size AS before FROM first \ + ), batch AS MATERIALIZED ( \ + SELECT p.id, p.table_id, p.event_name, p.record_data, p.created_at \ + FROM stream_pending p JOIN sized ON p.shard_id = $1 AND p.id = sized.id \ + WHERE sized.before < $3 \ + ), ranked AS ( \ + SELECT batch.*, row_number() OVER (ORDER BY id) AS rn FROM batch \ + ), numbers AS MATERIALIZED ( \ + SELECT lpad(n::text, 21, '0') AS seq, row_number() OVER (ORDER BY n) AS rn \ + FROM (SELECT nextval('stream_seq') AS n FROM batch) drawn \ + ), moved AS ( \ + DELETE FROM stream_pending p USING batch WHERE p.shard_id = $1 AND p.id = batch.id \ + ), inserted AS ( \ + INSERT INTO stream_records \ + (sequence_number, shard_id, table_id, event_name, record_data, created_at) \ + SELECT numbers.seq, $1, ranked.table_id, ranked.event_name, \ + jsonb_set(ranked.record_data, '{dynamodb,SequenceNumber}', to_jsonb(numbers.seq)), \ + ranked.created_at \ + FROM ranked JOIN numbers USING (rn) \ + RETURNING 1 \ + ) \ + SELECT (SELECT count(*) FROM inserted), (SELECT count(*) FROM first)"; + +/// What one numbering transaction did. +enum Numbered { + /// It moved `moved` records, and `more` is set when the shard had more + /// pending records than the batch took. + Moved { moved: u64, more: bool }, + /// Another transaction holds the shard's row: it numbers the shard. + Busy, + /// The shard is gone, or it belongs to another table than the caller + /// named. + Missing, +} + impl PostgresEngine { + /// Give a batch of a shard's committed records in `stream_pending` their + /// sequence numbers and move them to `stream_records`, in one + /// transaction ([`NUMBER_PENDING`]). + /// + /// A write stores its record in `stream_pending` with no number + /// (`write_stream_record_in_tx`), and the record is numbered here once the + /// write has committed. That is what keeps a reader from passing a record + /// (#808). The transaction holds the shard's row `FOR NO KEY UPDATE` until + /// it commits, so the numberings of one shard run one after another, each + /// after the previous one committed. `stream_seq` keeps no values cached + /// per session (CACHE 1, which migration 7 sets), so it hands out values + /// in the order of the calls across all sessions, and every number drawn + /// here is higher than every number an earlier numbering of the shard + /// committed. The batch holds the records committed when the statement + /// starts, after the lock is granted, and a write that commits later is + /// numbered by a later call, with a higher number still. Two writes to one + /// item take the item's row lock in turn, so the second stores its record + /// after the first commits, with a higher `id`, and it is numbered after + /// the first. + /// + /// The lock does not wait: when another transaction holds the shard's row, + /// this returns [`Numbered::Busy`] and gives its connection back, so a + /// caller that waits for that numbering holds no pool connection while it + /// waits. A writer locks the shard's row only `FOR KEY SHARE`, through the + /// foreign key of its insert, and that mode does not conflict with + /// `FOR NO KEY UPDATE`: writers never wait for a numbering, and a + /// numbering does not wait for writers. + /// + /// The transaction is READ COMMITTED whatever the database's default, so + /// the batch is read with a snapshot taken after the lock is granted. + /// Under REPEATABLE READ the snapshot would date from the start of the + /// lock statement, and a numbering that committed between that start and + /// the lock would leave records in the snapshot that it had already + /// moved. The delete would then fail with a serialization error. + async fn number_shard_batch( + &self, + shard_id: &str, + table_id: Option<&str>, + ) -> Result { + let internal = |e: sqlx::Error| StorageError::Internal(e.to_string()); + let mut tx = self + .data_pool + .begin_with("BEGIN ISOLATION LEVEL READ COMMITTED") + .await + .map_err(internal)?; + let locked: Option = sqlx::query_scalar(LOCK_SHARD) + .bind(shard_id) + .bind(table_id) + .fetch_optional(&mut *tx) + .await + .map_err(internal)?; + if locked.is_none() { + let exists: bool = sqlx::query_scalar(SHARD_EXISTS) + .bind(shard_id) + .bind(table_id) + .fetch_one(&mut *tx) + .await + .map_err(internal)?; + tx.rollback().await.map_err(internal)?; + return Ok(if exists { + Numbered::Busy + } else { + Numbered::Missing + }); + } + let (moved, looked): (i64, i64) = sqlx::query_as(NUMBER_PENDING) + .bind(shard_id) + .bind(NUMBER_BATCH) + .bind(NUMBER_BATCH_BYTES) + .fetch_one(&mut *tx) + .await + .map_err(internal)?; + tx.commit().await.map_err(internal)?; + Ok(Numbered::Moved { + moved: moved.unsigned_abs(), + more: moved < looked || looked == NUMBER_BATCH, + }) + } + + /// Number one batch of the shard's committed pending records before a + /// read of it: [`NUMBER_BATCH`] records, or fewer once they hold + /// [`NUMBER_BATCH_BYTES`]. One batch bounds the time a read spends here + /// to about one batch's copy (26 ms for 1,000 records of about 1 KB on + /// PostgreSQL 16). + /// + /// A read then returns the records whose writes committed before it + /// started, except in three cases: the shard holds more pending records + /// than one batch takes, another transaction numbers the shard for longer + /// than [`NUMBER_WAIT`], or this numbering fails. The records left + /// pending get higher numbers than any the read returns, from the next + /// read or the numbering worker, and a later read returns them. + /// + /// When another transaction numbers the 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( + &self, + shard_id: &str, + table_id: Option<&str>, + ) -> Result<(), StorageError> { + let deadline = tokio::time::Instant::now() + NUMBER_WAIT; + let mut pause = Duration::from_millis(1); + loop { + // A record that a running numbering moves still counts as + // pending here until that numbering commits. + let pending: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM stream_pending WHERE shard_id = $1)", + ) + .bind(shard_id) + .fetch_one(&self.data_pool) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + if !pending { + return Ok(()); + } + match self.number_shard_batch(shard_id, table_id).await? { + Numbered::Moved { .. } | Numbered::Missing => return Ok(()), + Numbered::Busy => { + if tokio::time::Instant::now() + pause > deadline { + return Ok(()); + } + tokio::time::sleep(pause).await; + pause = (pause * 2).min(NUMBER_PAUSE); + } + } + } + } + + /// [`Self::number_before_read`], with a failure logged instead of + /// returned: the read goes on and returns the records numbered so far. + /// The records left pending are numbered by the next read or the + /// numbering worker, with higher numbers than any this read returns. + async fn number_before_read_or_log(&self, shard_id: &str, table_id: Option<&str>) { + if let Err(e) = self.number_before_read(shard_id, table_id).await { + tracing::warn!("Stream record numbering before a read of shard {shard_id} failed: {e}"); + } + } + + /// Number up to [`NUMBER_LIMIT`] of one shard's committed pending + /// records, in batches, and stop at a shard that another transaction + /// numbers. Returns how many records moved. + async fn number_shard(&self, shard_id: &str) -> Result { + let mut numbered = 0; + while numbered < NUMBER_LIMIT { + match self.number_shard_batch(shard_id, None).await? { + Numbered::Moved { moved, more } => { + numbered += moved; + if !more { + break; + } + } + Numbered::Busy | Numbered::Missing => break, + } + } + Ok(numbered) + } + + /// Number the committed pending records of every shard that has some, up + /// to [`NUMBER_LIMIT`] per shard, skipping a shard that another + /// transaction numbers. The numbering worker runs this, so a shard that + /// nobody reads still has its records numbered, and the retention sweep + /// trims them as usual. A shard that fails does not stop the others: the + /// pass numbers every shard it can and then returns the first error. + /// Returns how many records moved. + pub(crate) async fn number_pending_stream_records(&self) -> Result { + let shards: Vec = sqlx::query_scalar(PENDING_SHARDS) + .fetch_all(&self.data_pool) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + let mut total = 0; + let mut first_error = None; + for shard_id in &shards { + match self.number_shard(shard_id).await { + Ok(n) => total += n, + Err(e) => { + first_error.get_or_insert(e); + } + } + } + match first_error { + Some(e) => Err(e), + None => Ok(total), + } + } + + /// [`Self::number_pending_stream_records`], for tests that stand in for + /// the numbering worker. + #[cfg(feature = "test-hooks")] + #[doc(hidden)] + pub async fn number_pending_stream_records_for_tests(&self) -> Result { + self.number_pending_stream_records().await + } + /// Initialize stream shards for a table and set the `stream_label`. /// /// Updates the stream label in the catalog (via the provided catalog @@ -47,10 +342,14 @@ impl PostgresEngine { &self, retention_hours: i64, ) -> Result { + // A record still in stream_pending counts as one of the table's + // records: deleting the shard would take it with it, through the + // foreign key's cascade, before the numbering worker numbers it. let candidates: Vec = sqlx::query_scalar( "SELECT DISTINCT s.table_id FROM stream_shards s \ WHERE s.created_at < NOW() - make_interval(hours => $1::integer) \ - AND NOT EXISTS (SELECT 1 FROM stream_records r WHERE r.table_id = s.table_id)", + AND NOT EXISTS (SELECT 1 FROM stream_records r WHERE r.table_id = s.table_id) \ + AND NOT EXISTS (SELECT 1 FROM stream_pending p WHERE p.table_id = s.table_id)", ) .bind(retention_hours) .fetch_all(&self.data_pool) @@ -77,7 +376,8 @@ impl PostgresEngine { // once none of its records remain. let result = sqlx::query( "DELETE FROM stream_shards s WHERE s.table_id = ANY($1) \ - AND NOT EXISTS (SELECT 1 FROM stream_records r WHERE r.table_id = s.table_id)", + AND NOT EXISTS (SELECT 1 FROM stream_records r WHERE r.table_id = s.table_id) \ + AND NOT EXISTS (SELECT 1 FROM stream_pending p WHERE p.table_id = s.table_id)", ) .bind(&gone) .execute(&self.data_pool) @@ -155,51 +455,22 @@ impl PostgresEngine { impl StreamEngine for PostgresEngine { fn write_stream_record( &self, - account_id: &str, - record: &StreamRecord, - shard_id: &str, - table_name: &str, + _account_id: &str, + _record: &StreamRecord, + _shard_id: &str, + _table_name: &str, ) -> BoxFuture<'_, Result<(), StorageError>> { - let account_id = account_id.to_string(); - let record = record.clone(); - let shard_id = shard_id.to_string(); - let table_name = table_name.to_string(); + // A record stored apart from its write would not wait for the write's + // item lock, and its number would not follow the write's commit + // (#808). Every write path stores its record in its own transaction + // instead (`write_stream_record_in_tx`). Box::pin(async move { - let record_json = - serde_json::to_value(&record).map_err(|e| StorageError::Internal(e.to_string()))?; - - let table_id: String = sqlx::query_scalar( - "SELECT table_id FROM tables WHERE account_id = $1 AND table_name = $2", - ) - .bind(&account_id) - .bind(&table_name) - .fetch_one(&self.pool) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - - // created_at follows the record's own timestamp rather than a - // second clock read (NOW() would be the transaction start time). - // A timestamp outside the range `time` can represent, which a - // SystemTime reading never is, is stamped with the current time, - // as the SQLite engine stamps it. - let created_at = time::OffsetDateTime::from_unix_timestamp( - record.dynamodb.approximate_creation_date_time, - ) - .unwrap_or_else(|_| time::OffsetDateTime::now_utc()); - sqlx::query( - "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \ - VALUES ($1, $2, $3, $4, $5, $6)", - ) - .bind(&record.dynamodb.sequence_number) - .bind(&shard_id) - .bind(&table_id) - .bind(format!("{:?}", record.event_name)) - .bind(&record_json) - .bind(created_at) - .execute(&self.data_pool) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - Ok(()) + Err(StorageError::Internal( + "write_stream_record is not used in the PostgreSQL backend; each write stores \ + its stream record in its own transaction, and the record is numbered after \ + the write commits" + .to_owned(), + )) }) } @@ -243,6 +514,14 @@ impl StreamEngine for PostgresEngine { return Err(extenddb_storage::util::stream_does_not_exist()); }; + // Number a batch of the shard's committed records first, so the + // page holds the writes that committed before this read, within + // the bounds `number_before_read` names. Only the named table's + // shard is numbered. A numbering failure is logged, and the page + // holds the records numbered so far. + self.number_before_read_or_log(&shard_id, Some(&table_id)) + .await; + let rows: Vec<(String, Option)> = if let Some(after) = after_sequence { sqlx::query_as(STREAM_RECORDS_AFTER) @@ -548,12 +827,15 @@ impl StreamEngine for PostgresEngine { } fn next_sequence_number(&self, _shard_id: &str) -> BoxFuture<'_, Result> { + // A number drawn before a record is stored could commit after a higher + // one that a reader has already passed (#808). Records get their + // numbers after their writes commit (`number_shard_batch`). Box::pin(async move { - let (seq_val,): (i64,) = sqlx::query_as("SELECT nextval('stream_seq')") - .fetch_one(&self.data_pool) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - Ok(format!("{seq_val:021}")) + Err(StorageError::Internal( + "next_sequence_number is not used in the PostgreSQL backend; stream records get \ + their sequence numbers after their writes commit" + .to_owned(), + )) }) } @@ -605,6 +887,15 @@ impl StreamEngine for PostgresEngine { ) -> BoxFuture<'_, Result, StorageError>> { let shard_id = shard_id.to_string(); Box::pin(async move { + // A LATEST iterator starts after the records whose writes have + // committed, so a batch of them is numbered first. A write still + // open is numbered later with a higher number, and the iterator + // returns it. When more records of the shard wait for their + // numbers than one batch takes, or the numbering fails (it is + // logged), the rest are numbered later too, above the number + // returned here, and the iterator returns them although their + // writes committed before it was taken. + self.number_before_read_or_log(&shard_id, None).await; let row: Option<(String,)> = sqlx::query_as( "SELECT sequence_number FROM stream_records \ WHERE shard_id = $1 ORDER BY sequence_number DESC LIMIT 1", diff --git a/crates/storage-postgres/src/workers.rs b/crates/storage-postgres/src/workers.rs index ee3296a75..b2fd06283 100644 --- a/crates/storage-postgres/src/workers.rs +++ b/crates/storage-postgres/src/workers.rs @@ -178,6 +178,37 @@ pub(crate) async fn table_size_refresh_worker( } } +/// How often the numbering worker looks for pending stream records. +const STREAM_NUMBERING_INTERVAL: Duration = Duration::from_millis(250); + +/// How long the numbering worker waits after a failed pass, on top of its +/// interval, so an unreachable database is reported about once a second, as +/// the GSI workers report it. +const STREAM_NUMBERING_RETRY: Duration = Duration::from_secs(1); + +/// Numbers the stream records that writes leave in `stream_pending` +/// (`number_pending_stream_records`). A read numbers its own shard before it +/// reads, so this worker is what numbers the records of a shard that nobody +/// reads, which the retention sweep then trims. It skips a shard that a +/// reader or another server is numbering. +pub(crate) async fn stream_record_numbering_worker( + storage: Arc, + token: CancellationToken, +) { + while tick(&token, STREAM_NUMBERING_INTERVAL).await { + match storage.number_pending_stream_records().await { + Ok(0) => {} + Ok(n) => tracing::debug!(records = n, "numbered pending stream records"), + Err(e) => { + tracing::warn!("Stream record numbering failed: {e}"); + if !tick(&token, STREAM_NUMBERING_RETRY).await { + break; + } + } + } + } +} + pub(crate) async fn stream_record_cleanup_worker( storage: Arc, metrics: Arc, diff --git a/crates/storage-postgres/tests/put_create_race.rs b/crates/storage-postgres/tests/put_create_race.rs index 981d29bfb..e62d9a508 100644 --- a/crates/storage-postgres/tests/put_create_race.rs +++ b/crates/storage-postgres/tests/put_create_race.rs @@ -847,6 +847,12 @@ async fn a_lost_create_race_gives_the_lsi_and_the_stream_the_winner() { .expect("read c"); assert_eq!(v, "loser"); + // A write leaves its stream record in stream_pending until a read or the + // numbering worker numbers it (#808). This test stands in for the worker. + s.engine + .number_pending_stream_records_for_tests() + .await + .expect("number the pending stream records"); let records: Vec<(serde_json::Value,)> = sqlx::query_as( "SELECT record_data FROM stream_records WHERE table_id = $1 ORDER BY sequence_number", ) diff --git a/crates/storage-postgres/tests/stream_commit_order.rs b/crates/storage-postgres/tests/stream_commit_order.rs new file mode 100644 index 000000000..72f8acd89 --- /dev/null +++ b/crates/storage-postgres/tests/stream_commit_order.rs @@ -0,0 +1,1317 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! A stream consumer on PostgreSQL reads every committed write once, and the +//! records of one item in the order their writes committed (#808). +//! +//! A write stores its stream record in `stream_pending`, inside the write's +//! own transaction, and the record gets its sequence number only after the +//! write commits. GetRecords and a LATEST iterator number a batch of the +//! shard's committed records before they read it, and a worker numbers the +//! shards that nobody reads. Before this, a write drew its number inside its +//! transaction, and it could commit after a write that drew a higher number. +//! A consumer that had read the higher number then never read the first +//! write's record. +//! +//! The tests hold a write open at its commit, after it stored its record, +//! with a deferred constraint trigger that waits on an advisory lock, the +//! gate. +//! +//! Needs `EXTENDDB_TEST_PG_CONNECTION_STRING` (host-only, e.g. +//! `postgres://user:pass@127.0.0.1:5432`); skips otherwise. Each test builds +//! its own scratch database and drops it when it passes. A failing test +//! leaves its database, named `eddb_sord_*`, behind. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, Item, KeySchemaElement, + KeyType, ReturnValuesOnConditionCheckFailure, ScalarAttributeType, StreamRecord, + StreamSpecification, StreamViewType, TableKeyInfo, +}; +use extenddb_storage::util::stream_arn; +use extenddb_storage::{DataEngine, StreamCapture, StreamEngine, TableEngine, TransactWriteOp}; +use extenddb_storage_postgres::{PostgresConfig, PostgresEngine}; +use sqlx::postgres::PgPoolOptions; +use sqlx::{Connection, PgConnection, PgPool}; + +const ACCOUNT: &str = "123456789012"; +const REGION: &str = "us-east-1"; +const TABLE: &str = "t_stream_order"; +const LABEL: &str = "2026-01-01T00:00:00.000"; + +/// The advisory lock key of the gate. +const GATE: i64 = 0x5374_4f72; + +/// A page size larger than any test writes. +const PAGE: i64 = 1_000; + +struct Scratch { + engine: PostgresEngine, + db: PgPool, + admin: PgPool, + db_name: String, +} + +impl Scratch { + async fn cleanup(self) { + let Scratch { + engine, + db, + admin, + db_name, + } = self; + drop(engine); + db.close().await; + sqlx::query(&format!( + "DROP DATABASE IF EXISTS \"{db_name}\" WITH (FORCE)" + )) + .execute(&admin) + .await + .expect("drop the scratch database"); + admin.close().await; + } + + /// The test table, keyed on `pk` and `sk`, with a NEW_AND_OLD_IMAGES + /// stream whose label is pinned to [`LABEL`]. Returns its key info, its + /// table id and its stream ARN. + async fn stream_table(&self) -> (TableKeyInfo, String, String) { + let key = |name: &str, key_type| { + ( + KeySchemaElement { + attribute_name: name.to_owned(), + key_type, + }, + AttributeDefinition { + attribute_name: name.to_owned(), + attribute_type: ScalarAttributeType::S, + }, + ) + }; + let (key_schema, attribute_definitions) = + [key("pk", KeyType::Hash), key("sk", KeyType::Range)] + .into_iter() + .unzip(); + let desc = self + .engine + .create_table( + ACCOUNT, + CreateTableInput { + table_name: TABLE.to_owned(), + key_schema, + attribute_definitions, + billing_mode: Some(BillingMode::PayPerRequest), + stream_specification: Some(StreamSpecification { + stream_enabled: true, + stream_view_type: Some(StreamViewType::NewAndOldImages), + }), + ..Default::default() + }, + ) + .await + .expect("create the table"); + sqlx::query("UPDATE tables SET stream_label = $1 WHERE table_id = $2") + .bind(LABEL) + .bind(&desc.table_id) + .execute(&self.db) + .await + .expect("pin the stream label"); + let key_info = self + .engine + .table_key_info(ACCOUNT, TABLE) + .await + .expect("read the key info"); + ( + key_info, + desc.table_id, + stream_arn(REGION, ACCOUNT, TABLE, LABEL), + ) + } + + /// The shard the records of partition key `pk` go to. + async fn shard_of(&self, pk: &str) -> String { + self.engine + .assign_shard(ACCOUNT, TABLE, pk) + .await + .expect("assign a shard") + } + + /// Store `item("pk", sk, v)`, with its stream record. + async fn put(&self, key_info: &TableKeyInfo, pk: &str, sk: &str, v: &str) { + self.engine + .put_item( + key_info, + item(pk, sk, v), + false, + None, + &ExpressionMaps::default(), + Some(&capture()), + ) + .await + .expect("store the item"); + } + + /// One GetRecords page of `shard`, after `after` or from the start. + async fn read( + &self, + arn: &str, + shard: &str, + after: Option<&str>, + ) -> (Vec, Option) { + self.engine + .get_stream_records(ACCOUNT, arn, shard, after, PAGE) + .await + .expect("read the shard") + } + + /// How many records of `shard` wait in `stream_pending` for a number. + async fn pending(&self, shard: &str) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM stream_pending WHERE shard_id = $1") + .bind(shard) + .fetch_one(&self.db) + .await + .expect("count the pending records") + } +} + +/// The server's base URL, or `None` to skip. With `EXTENDDB_TEST_PG_REQUIRED` +/// set to any value but `0`, a job that must reach a server fails here +/// instead. +fn base_conn() -> Option { + let conn = std::env::var("EXTENDDB_TEST_PG_CONNECTION_STRING").unwrap_or_default(); + if conn.trim().is_empty() { + let required = + std::env::var("EXTENDDB_TEST_PG_REQUIRED").is_ok_and(|v| !v.is_empty() && v != "0"); + assert!( + !required, + "EXTENDDB_TEST_PG_CONNECTION_STRING is not set, and EXTENDDB_TEST_PG_REQUIRED says \ + this job must reach a PostgreSQL server" + ); + return None; + } + Some(conn.trim_end_matches('/').to_owned()) +} + +/// Every `.sql` file under `migrations/` and `data_migrations/`, each +/// directory in filename order, so a new migration is picked up here without +/// editing this file. +fn shipped_migrations() -> Vec { + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); + let mut out = Vec::new(); + for dir in ["migrations", "data_migrations"] { + let mut files: Vec<_> = std::fs::read_dir(root.join(dir)) + .expect("read the migrations directory") + .map(|e| e.expect("directory entry").path()) + .filter(|p| p.extension().is_some_and(|e| e == "sql")) + .collect(); + files.sort(); + for f in files { + out.push(std::fs::read_to_string(&f).expect("read a migration file")); + } + } + out +} + +fn skip(test: &str) { + eprintln!( + "SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is no PostgreSQL \ + to build a scratch database in." + ); +} + +/// A scratch database with every shipped migration applied, and an engine +/// whose pool holds `pool_size` connections. With `default_isolation`, the +/// engine's sessions start their transactions at that level unless a +/// transaction names its own, as with an operator's +/// `default_transaction_isolation`. +async fn scratch(default_isolation: Option<&str>, pool_size: u32) -> Scratch { + let base = base_conn().expect("caller checks base_conn() first"); + let db_name = format!("eddb_sord_{}", uuid::Uuid::new_v4().simple())[..24].to_owned(); + let admin = PgPoolOptions::new() + .max_connections(1) + .connect(&format!("{base}/postgres")) + .await + .expect("connect to the postgres maintenance database"); + sqlx::query(&format!("CREATE DATABASE \"{db_name}\"")) + .execute(&admin) + .await + .expect("create the scratch database"); + let url = format!("{base}/{db_name}"); + let db = PgPoolOptions::new() + .max_connections(4) + .connect(&url) + .await + .expect("connect to the scratch database"); + // One scratch database stands in for both the catalog and the data + // database, so every shipped migration file from both directories is + // applied to it, in filename order. + for sql in shipped_migrations() { + sqlx::raw_sql(&sql) + .execute(&db) + .await + .expect("apply a shipped migration"); + } + // The catalog schema's `stream_shards` carries a foreign key to `tables` + // that cannot exist in a deployment (the two live in different + // databases) and that the engine's insert order violates. + sqlx::query("ALTER TABLE stream_shards DROP CONSTRAINT IF EXISTS stream_shards_table_id_fkey") + .execute(&db) + .await + .expect("match the two-database layout"); + sqlx::query( + "INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0') \ + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + ) + .execute(&db) + .await + .expect("pin the control-plane delay to zero"); + sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)") + .bind(ACCOUNT) + .bind(format!("acct-{db_name}")) + .execute(&db) + .await + .expect("seed the account row"); + if let Some(level) = default_isolation { + // Sessions opened from here on, the engine's among them, take it. + sqlx::query(&format!( + "ALTER DATABASE \"{db_name}\" SET default_transaction_isolation = '{level}'" + )) + .execute(&admin) + .await + .expect("set the default isolation level"); + } + let engine = PostgresEngine::new( + &PostgresConfig { + connection_string: url, + pool_size, + max_item_size_bytes: 400_000, + }, + REGION, + ) + .await + .expect("open a PostgresEngine on the scratch database"); + Scratch { + engine, + db, + admin, + db_name, + } +} + +fn item(pk: &str, sk: &str, v: &str) -> Item { + BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(pk.to_owned())), + ("sk".to_owned(), AttributeValue::S(sk.to_owned())), + ("v".to_owned(), AttributeValue::S(v.to_owned())), + ]) +} + +fn capture() -> StreamCapture { + StreamCapture { + view_type: StreamViewType::NewAndOldImages, + user_identity: None, + region: Arc::from(REGION), + } +} + +/// The `v` of each record's new image, in the order read. +fn values(records: &[StreamRecord]) -> Vec { + records + .iter() + .map( + |r| match r.dynamodb.new_image.as_ref().and_then(|i| i.get("v")) { + Some(AttributeValue::S(v)) => v.clone(), + other => panic!("a record whose new image has no string v: {other:?}"), + }, + ) + .collect() +} + +fn string_key(r: &StreamRecord, name: &str) -> String { + match r.dynamodb.keys.get(name) { + Some(AttributeValue::S(v)) => v.clone(), + other => panic!("a record without a string {name} key: {other:?}"), + } +} + +/// Open a connection to the scratch database outside its pools. Returns it and +/// its backend pid. +async fn connect(s: &Scratch) -> (PgConnection, i32) { + let base = base_conn().expect("caller checks base_conn() first"); + let mut conn = PgConnection::connect(&format!("{base}/{}", s.db_name)) + .await + .expect("connect to the scratch database"); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut conn) + .await + .expect("read the backend pid"); + (conn, pid) +} + +/// Make a transaction that writes a row of `relation` for which `condition` +/// (an expression on `NEW`) holds wait at its commit while the gate is +/// closed. A deferred constraint trigger runs at commit, after every +/// statement of the transaction, so the write has stored its stream record +/// by then. +async fn install_commit_gate(s: &Scratch, relation: &str, condition: &str) { + sqlx::raw_sql(&format!( + "CREATE FUNCTION hold_at_commit() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN \ + IF {condition} THEN PERFORM pg_advisory_xact_lock_shared({GATE}); END IF; \ + RETURN NULL; \ + END $$; \ + CREATE CONSTRAINT TRIGGER hold_at_commit AFTER INSERT OR UPDATE ON {relation} \ + DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION hold_at_commit();" + )) + .execute(&s.db) + .await + .expect("install the gate trigger"); +} + +/// Close the gate on a session of its own. Returns the session and its pid. +async fn close_gate(s: &Scratch) -> (PgConnection, i32) { + let (mut gate, pid) = connect(s).await; + sqlx::query("SELECT pg_advisory_lock($1)") + .bind(GATE) + .execute(&mut gate) + .await + .expect("close the gate"); + (gate, pid) +} + +async fn open_gate(gate: &mut PgConnection) { + sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(GATE) + .execute(gate) + .await + .expect("open the gate"); +} + +/// The quoted name of the test table's data table. +fn data_table(table_id: &str) -> String { + format!("\"_ddb_{table_id}\"") +} + +/// Wait until a backend waits for a lock of kind `event` that the backend +/// `blocker` holds. Fails after 10 s, so the caller can report what it saw. +async fn wait_until_blocked_by(db: &PgPool, blocker: i32, event: &str) -> Result<(), String> { + for _ in 0..1_000 { + let waiting: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity \ + WHERE datname = current_database() AND wait_event_type = 'Lock' \ + AND wait_event = $2 AND $1 = ANY(pg_blocking_pids(pid))", + ) + .bind(blocker) + .bind(event) + .fetch_one(db) + .await + .expect("read pg_stat_activity"); + if waiting > 0 { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + Err(format!( + "no backend waited on the {event} lock of backend {blocker}" + )) +} + +/// The case #808 names. A write stores its record and then waits at its +/// commit. A second write to the same shard commits, and a consumer reads +/// its record. The first write then commits, and the consumer's next read, +/// after the number it holds, must return the first write's record. With +/// the number drawn inside the write's transaction, the first write's +/// number was below the consumer's, and no read after it returned the +/// record. +#[tokio::test] +async fn a_write_that_commits_after_a_read_is_returned_by_the_next_read() { + let test = "a_write_that_commits_after_a_read_is_returned_by_the_next_read"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch(None, 12).await; + let (key_info, table_id, arn) = s.stream_table().await; + install_commit_gate( + &s, + &data_table(&table_id), + "NEW.item_data->'v'->>'S' = 'held'", + ) + .await; + let shard = s.shard_of("p").await; + let (mut gate, gate_pid) = close_gate(&s).await; + + let held = s.put(&key_info, "p", "1", "held"); + let rest = async { + wait_until_blocked_by(&s.db, gate_pid, "advisory").await?; + s.put(&key_info, "p", "2", "free").await; + let first = s.read(&arn, &shard, None).await; + open_gate(&mut gate).await; + Ok::<_, String>(first) + }; + let ((), rest) = + tokio::time::timeout(Duration::from_secs(30), async { tokio::join!(held, rest) }) + .await + .expect("both writes finished"); + let (first, cursor) = rest.unwrap_or_else(|e| panic!("{e}")); + assert_eq!( + values(&first), + ["free"], + "the first read returns the committed write only" + ); + let cursor = cursor.expect("the first read returns a sequence number"); + + let (next, _) = s.read(&arn, &shard, Some(&cursor)).await; + assert_eq!( + values(&next), + ["held"], + "the read after the consumer's number returns the write that committed late" + ); + assert!( + next[0].dynamodb.sequence_number > cursor, + "the late write's number {} is above the number {cursor} the consumer read", + next[0].dynamodb.sequence_number + ); + let (all, _) = s.read(&arn, &shard, None).await; + assert_eq!( + values(&all), + ["free", "held"], + "from the start, the records are in commit order" + ); + assert_eq!(s.pending(&shard).await, 0, "no record is left pending"); + + drop(gate); + s.cleanup().await; +} + +/// Writers race on four hot items while a consumer pages through every +/// shard. The consumer must read each write's record exactly once, and each +/// item's records must chain: the old image of a record is the new image of +/// the item's record before it. A record out of order breaks the chain. +#[tokio::test] +async fn concurrent_writers_lose_no_record_and_keep_each_items_order() { + let test = "concurrent_writers_lose_no_record_and_keep_each_items_order"; + if base_conn().is_none() { + return skip(test); + } + const WRITERS: usize = 8; + const WRITES: usize = 150; + const ITEMS: [&str; 4] = ["h0", "h1", "h2", "h3"]; + let s = scratch(None, 12).await; + let (key_info, table_id, arn) = s.stream_table().await; + let shards: Vec = + sqlx::query_scalar("SELECT shard_id FROM stream_shards WHERE table_id = $1 ORDER BY 1") + .bind(&table_id) + .fetch_all(&s.db) + .await + .expect("list the shards"); + + let done = AtomicBool::new(false); + let writers = async { + futures::future::join_all((0..WRITERS).map(|w| { + let (s, key_info) = (&s, &key_info); + async move { + for i in 0..WRITES { + let pk = ITEMS[(w + i) % ITEMS.len()]; + s.put(key_info, pk, "s", &format!("w{w}-{i}")).await; + } + } + })) + .await; + done.store(true, Ordering::SeqCst); + }; + let consumer = async { + let mut cursors: Vec> = vec![None; shards.len()]; + let mut read: Vec = Vec::new(); + loop { + // Read once more after the writers finish, to take the rest. + let finished = done.load(Ordering::SeqCst); + for (shard, cursor) in shards.iter().zip(cursors.iter_mut()) { + loop { + let (page, last) = s.read(&arn, shard, cursor.as_deref()).await; + if page.is_empty() { + break; + } + *cursor = last; + read.extend(page); + } + } + if finished { + return read; + } + tokio::time::sleep(Duration::from_millis(2)).await; + } + }; + let ((), read) = tokio::time::timeout(Duration::from_secs(120), async { + tokio::join!(writers, consumer) + }) + .await + .expect("the writers and the consumer finished"); + + let tags = values(&read); + let distinct: BTreeSet<&String> = tags.iter().collect(); + assert_eq!( + distinct.len(), + tags.len(), + "the consumer read a record twice" + ); + assert_eq!( + tags.len(), + WRITERS * WRITES, + "the consumer read {} of the {} writes", + tags.len(), + WRITERS * WRITES + ); + + // One item's records are all on one shard, and a shard's records are + // read in order, so the item's records keep their order in `read`. + let mut last_image: BTreeMap = BTreeMap::new(); + for r in &read { + let pk = string_key(r, "pk"); + assert_eq!( + r.dynamodb.old_image.as_ref(), + last_image.get(&pk), + "item {pk}: the record {} does not follow the item's record before it", + r.dynamodb.sequence_number + ); + let new_image = r.dynamodb.new_image.clone().expect("a put has a new image"); + last_image.insert(pk, new_image); + } + for pk in ITEMS { + let stored: serde_json::Value = sqlx::query_scalar(&format!( + "SELECT item_data FROM {} WHERE pk = $1", + data_table(&table_id) + )) + .bind(pk) + .fetch_one(&s.db) + .await + .expect("read the stored item"); + let stored: Item = serde_json::from_value(stored).expect("an item"); + assert_eq!( + last_image.get(pk), + Some(&stored), + "item {pk}: the last record is the last write" + ); + } + + s.cleanup().await; +} + +/// A read returns every write that committed before it, in every round of +/// a write followed by a read. +#[tokio::test] +async fn a_read_returns_the_write_that_committed_before_it() { + let test = "a_read_returns_the_write_that_committed_before_it"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch(None, 12).await; + let (key_info, _, arn) = s.stream_table().await; + let shard = s.shard_of("r").await; + let mut cursor: Option = None; + for i in 0..20 { + let v = format!("v{i:02}"); + s.put(&key_info, "r", &format!("{i:02}"), &v).await; + let (page, last) = s.read(&arn, &shard, cursor.as_deref()).await; + assert_eq!(values(&page), [v.as_str()], "round {i}"); + let seq = &page[0].dynamodb.sequence_number; + assert_eq!(seq.len(), 21, "round {i}: the number {seq} has 21 digits"); + assert!( + cursor.as_ref().is_none_or(|c| seq > c), + "round {i}: the number {seq} is above the one before, {cursor:?}" + ); + cursor = last; + } + + s.cleanup().await; +} + +/// A LATEST iterator starts after every write that committed before it, so +/// those writes are numbered first. Without that, a shard whose records +/// were all pending had no number to start after, and a LATEST iterator +/// read the shard from the start. A write still open when the iterator is +/// taken is numbered after it commits, and the iterator returns it. +#[tokio::test] +async fn a_latest_iterator_starts_after_the_committed_writes() { + let test = "a_latest_iterator_starts_after_the_committed_writes"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch(None, 12).await; + let (key_info, table_id, arn) = s.stream_table().await; + install_commit_gate( + &s, + &data_table(&table_id), + "NEW.item_data->'v'->>'S' = 'held'", + ) + .await; + let shard = s.shard_of("p").await; + + s.put(&key_info, "p", "1", "a").await; + let after_a = s + .engine + .latest_sequence_number(&shard) + .await + .expect("resolve LATEST") + .expect("LATEST resolves to the number of the committed write"); + let (page, _) = s.read(&arn, &shard, None).await; + assert_eq!(values(&page), ["a"]); + assert_eq!(page[0].dynamodb.sequence_number, after_a); + + s.put(&key_info, "p", "2", "b").await; + let (mut gate, gate_pid) = close_gate(&s).await; + let held = s.put(&key_info, "p", "3", "held"); + let rest = async { + wait_until_blocked_by(&s.db, gate_pid, "advisory").await?; + let after_b = s + .engine + .latest_sequence_number(&shard) + .await + .expect("resolve LATEST"); + open_gate(&mut gate).await; + Ok::<_, String>(after_b) + }; + let ((), rest) = + tokio::time::timeout(Duration::from_secs(30), async { tokio::join!(held, rest) }) + .await + .expect("the held write finished"); + let after_b = rest + .unwrap_or_else(|e| panic!("{e}")) + .expect("LATEST resolves while a write is open"); + + let (page, _) = s.read(&arn, &shard, Some(&after_a)).await; + assert_eq!(values(&page), ["b", "held"]); + assert_eq!( + page[0].dynamodb.sequence_number, after_b, + "LATEST taken during the open write starts after b" + ); + let (page, _) = s.read(&arn, &shard, Some(&after_b)).await; + assert_eq!( + values(&page), + ["held"], + "the write open when LATEST was taken is returned" + ); + + drop(gate); + s.cleanup().await; +} + +/// The numbering worker numbers the shards that nobody reads, and it skips +/// a shard that another transaction is numbering instead of waiting. The +/// number goes in the `sequence_number` column and in the record. +#[tokio::test] +async fn the_worker_numbers_unread_shards_and_skips_a_locked_one() { + let test = "the_worker_numbers_unread_shards_and_skips_a_locked_one"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch(None, 12).await; + let (key_info, _, _) = s.stream_table().await; + // Two partition keys on two shards. + let first = s.shard_of("k0").await; + let mut other = None; + for i in 1..100 { + let pk = format!("k{i}"); + let shard = s.shard_of(&pk).await; + if shard != first { + other = Some((pk, shard)); + break; + } + } + let (other_pk, second) = other.expect("a partition key on another shard"); + s.put(&key_info, "k0", "s", "one").await; + s.put(&key_info, &other_pk, "s", "two").await; + assert_eq!((s.pending(&first).await, s.pending(&second).await), (1, 1)); + + // An outside transaction holds the first shard's row, as a numbering of + // that shard does. + let mut holder = s.db.begin().await.expect("begin the outside transaction"); + sqlx::query("SELECT 1 FROM stream_shards WHERE shard_id = $1 FOR NO KEY UPDATE") + .bind(&first) + .execute(&mut *holder) + .await + .expect("lock the first shard"); + let moved = tokio::time::timeout( + Duration::from_secs(10), + s.engine.number_pending_stream_records_for_tests(), + ) + .await + .expect("the worker does not wait for the locked shard") + .expect("number the pending records"); + assert_eq!(moved, 1, "only the unlocked shard is numbered"); + assert_eq!((s.pending(&first).await, s.pending(&second).await), (1, 0)); + holder.rollback().await.expect("release the first shard"); + + let moved = s + .engine + .number_pending_stream_records_for_tests() + .await + .expect("number the pending records"); + assert_eq!(moved, 1, "the released shard is numbered"); + assert_eq!((s.pending(&first).await, s.pending(&second).await), (0, 0)); + + let rows: Vec<(String, String)> = sqlx::query_as( + "SELECT sequence_number, record_data->'dynamodb'->>'SequenceNumber' FROM stream_records", + ) + .fetch_all(&s.db) + .await + .expect("read the numbered records"); + assert_eq!(rows.len(), 2); + for (column, in_record) in &rows { + assert_eq!(column, in_record, "the record carries its number"); + assert_eq!(column.len(), 21, "{column} has 21 digits"); + } + + s.cleanup().await; +} + +/// Under an operator's REPEATABLE READ default, a read of a shard that +/// another transaction numbers must succeed and return what that numbering +/// commits. The read waits for the numbering with no connection held, and its +/// own numbering transaction is READ COMMITTED whatever the default. +#[tokio::test] +async fn a_read_during_a_numbering_succeeds_under_repeatable_read() { + let test = "a_read_during_a_numbering_succeeds_under_repeatable_read"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch(Some("repeatable read"), 12).await; + let (key_info, _, arn) = s.stream_table().await; + // The first numbering waits at its commit, holding the shard's row. + install_commit_gate( + &s, + "stream_records", + "NEW.record_data->'dynamodb'->'NewImage'->'v'->>'S' = 'held'", + ) + .await; + let shard = s.shard_of("p").await; + s.put(&key_info, "p", "1", "held").await; + let (mut gate, gate_pid) = close_gate(&s).await; + + let numbering = s.engine.number_pending_stream_records_for_tests(); + let read = async { + wait_until_blocked_by(&s.db, gate_pid, "advisory").await?; + let started = std::time::Instant::now(); + let read = s + .engine + .get_stream_records(ACCOUNT, &arn, &shard, None, PAGE) + .await; + Ok::<_, String>((read, started.elapsed())) + }; + let release = async { + let parked = wait_until_blocked_by(&s.db, gate_pid, "advisory").await; + // The read looks at the shard while the numbering holds it. + tokio::time::sleep(Duration::from_millis(200)).await; + open_gate(&mut gate).await; + parked + }; + let (numbered, read, parked) = tokio::time::timeout(Duration::from_secs(30), async { + tokio::join!(numbering, read, release) + }) + .await + .expect("the numbering and the read finished"); + let (read, waited) = read.unwrap_or_else(|e| panic!("{e}")); + if let Err(e) = parked { + panic!("{e}; the read returned {read:?}"); + } + assert_eq!(numbered.expect("the first numbering commits"), 1); + let (records, _) = read.expect("the read during the numbering succeeds"); + assert_eq!(values(&records), ["held"]); + assert!( + waited >= Duration::from_millis(150), + "the read returned after {waited:?}, before the numbering committed" + ); + + drop(gate); + s.cleanup().await; +} + +/// Readers of a shard that another transaction numbers wait for it with no +/// pool connection held. Twice as many readers as the pool has connections +/// wait on a held shard, and a write must still get a connection at once. +/// When each waiting reader held a connection, the write waited until the +/// shard was released. +#[tokio::test] +async fn readers_waiting_for_a_numbering_leave_the_pool_to_writes() { + let test = "readers_waiting_for_a_numbering_leave_the_pool_to_writes"; + if base_conn().is_none() { + return skip(test); + } + const POOL: u32 = 10; + let s = scratch(None, POOL).await; + let (key_info, _, arn) = s.stream_table().await; + let shard = s.shard_of("p").await; + s.put(&key_info, "p", "1", "pending").await; + + // An outside transaction holds the shard's row, as a running numbering + // does. + let mut holder = s.db.begin().await.expect("begin the outside transaction"); + sqlx::query("SELECT 1 FROM stream_shards WHERE shard_id = $1 FOR NO KEY UPDATE") + .bind(&shard) + .execute(&mut *holder) + .await + .expect("hold the shard"); + let readers = futures::future::join_all((0..2 * POOL).map(|_| s.read(&arn, &shard, None))); + let write = async { + // The readers reach the held shard first. + tokio::time::sleep(Duration::from_millis(300)).await; + let started = std::time::Instant::now(); + let put = tokio::time::timeout( + Duration::from_secs(5), + s.put(&key_info, "q", "1", "unrelated"), + ) + .await; + let took = started.elapsed(); + holder.rollback().await.expect("release the shard"); + (put.is_ok(), took) + }; + let (reads, (put_done, took)) = tokio::time::timeout(Duration::from_secs(30), async { + tokio::join!(readers, write) + }) + .await + .expect("the readers and the write finished"); + assert!( + put_done && took < Duration::from_secs(1), + "the write took {took:?} while {} readers waited on the held shard", + 2 * POOL + ); + for (records, _) in &reads { + assert_eq!(values(records), ["pending"], "each reader gets the record"); + } + + s.cleanup().await; +} + +/// A read numbers one batch of the shard's pending records before it reads +/// the shard, so the time GetRecords spends numbering is bounded by one +/// batch. Of a backlog of 1,500 records, the first read numbers and returns +/// the first 1,000, and the rest wait. The next read numbers and returns the +/// rest, after the first read's numbers. When a read numbered up to 10,000 +/// records, the first read numbered all 1,500. +#[tokio::test] +async fn a_read_numbers_one_batch_of_a_backlog() { + let test = "a_read_numbers_one_batch_of_a_backlog"; + if base_conn().is_none() { + return skip(test); + } + const BACKLOG: usize = 1_500; + let s = scratch(None, 12).await; + let (key_info, _, arn) = s.stream_table().await; + let shard = s.shard_of("b").await; + let written: Vec = (0..BACKLOG).map(|i| format!("v{i:04}")).collect(); + for (i, v) in written.iter().enumerate() { + s.put(&key_info, "b", &format!("{i:04}"), v).await; + } + assert_eq!(s.pending(&shard).await, 1_500); + + let (first, cursor) = s.read(&arn, &shard, None).await; + assert_eq!( + values(&first), + &written[..1_000], + "the first read returns the first batch" + ); + assert_eq!( + s.pending(&shard).await, + 500, + "the first read numbers one batch" + ); + + let (rest, _) = s.read(&arn, &shard, cursor.as_deref()).await; + assert_eq!( + values(&rest), + &written[1_000..], + "the next read returns the rest in order" + ); + assert_eq!(s.pending(&shard).await, 0); + assert!( + rest[0].dynamodb.sequence_number > first[999].dynamodb.sequence_number, + "the rest are numbered after the first batch" + ); + + s.cleanup().await; +} + +/// A read whose numbering fails returns the records numbered before it, +/// and logs the failure, where it failed GetRecords before. A LATEST +/// iterator taken then starts after the numbered records. Once the cause is +/// gone, the next read numbers and returns the record that waited. +#[tokio::test] +async fn a_read_whose_numbering_fails_returns_the_numbered_records() { + let test = "a_read_whose_numbering_fails_returns_the_numbered_records"; + if base_conn().is_none() { + return skip(test); + } + let s = scratch(None, 12).await; + let (key_info, _, arn) = s.stream_table().await; + let shard = s.shard_of("f").await; + s.put(&key_info, "f", "1", "a").await; + let (page, after_a) = s.read(&arn, &shard, None).await; + assert_eq!(values(&page), ["a"]); + let after_a = after_a.expect("the read returns the number of a"); + + // A numbering that copies the record whose v is "refused" fails. + sqlx::raw_sql( + "CREATE FUNCTION refuse_numbering() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN \ + IF NEW.record_data->'dynamodb'->'NewImage'->'v'->>'S' = 'refused' THEN \ + RAISE EXCEPTION 'the test refuses this numbering'; \ + END IF; \ + RETURN NEW; \ + END $$; \ + CREATE TRIGGER refuse_numbering BEFORE INSERT ON stream_records \ + FOR EACH ROW EXECUTE FUNCTION refuse_numbering();", + ) + .execute(&s.db) + .await + .expect("install the failing trigger"); + s.put(&key_info, "f", "2", "refused").await; + + let (page, _) = s.read(&arn, &shard, None).await; + assert_eq!( + values(&page), + ["a"], + "the read returns the records numbered before" + ); + assert_eq!(s.pending(&shard).await, 1, "the refused record waits"); + let latest = s + .engine + .latest_sequence_number(&shard) + .await + .expect("LATEST resolves while the numbering fails"); + assert_eq!(latest.as_deref(), Some(after_a.as_str())); + + sqlx::raw_sql("DROP TRIGGER refuse_numbering ON stream_records") + .execute(&s.db) + .await + .expect("remove the failing trigger"); + let (page, _) = s.read(&arn, &shard, Some(&after_a)).await; + assert_eq!( + values(&page), + ["refused"], + "the next read returns the record that waited" + ); + assert_eq!(s.pending(&shard).await, 0); + + s.cleanup().await; +} + +/// About 100 KB of text that compresses poorly, from a fixed seed. +fn noise(len: usize, seed: u64) -> String { + let mut x = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1; + (0..len) + .map(|_| { + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + char::from(b"0123456789abcdef"[(x % 16) as usize]) + }) + .collect() +} + +/// About 100 KB of text that compresses to a small part of its length. +fn compressible(len: usize, seed: u64) -> String { + format!("{seed:04}").repeat(len / 4) +} + +/// Number every pending record of `shard` in one pass of the worker. Returns +/// the `record_bytes` of the records each numbering transaction moved, one +/// list per transaction, in the order the records were stored. +async fn number_in_batches(s: &Scratch, shard: &str) -> Vec> { + let sizes: Vec = sqlx::query_scalar( + "SELECT record_bytes FROM stream_pending WHERE shard_id = $1 ORDER BY id", + ) + .bind(shard) + .fetch_all(&s.db) + .await + .expect("read the sizes of the pending records"); + let moved = s + .engine + .number_pending_stream_records_for_tests() + .await + .expect("number the pending records"); + assert_eq!(moved, sizes.len() as u64, "one pass numbers every record"); + // The records one numbering transaction moved share its xmin, and the + // numbering gives them numbers in the order they were stored. + let xmins: Vec = sqlx::query_scalar( + "SELECT xmin::text FROM stream_records WHERE shard_id = $1 ORDER BY sequence_number", + ) + .bind(shard) + .fetch_all(&s.db) + .await + .expect("read the numbered records"); + assert_eq!(xmins.len(), sizes.len()); + let mut batches: Vec<(String, Vec)> = Vec::new(); + for (xmin, size) in xmins.into_iter().zip(sizes) { + match batches.last_mut() { + Some((x, sizes)) if *x == xmin => sizes.push(size), + _ => batches.push((xmin, vec![size])), + } + } + batches.into_iter().map(|(_, sizes)| sizes).collect() +} + +/// `records` records of about 100 KB took three or more numbering +/// transactions, and in each the records before the last hold less than a +/// mebibyte of JSON text. +fn assert_batches_stop_at_a_mebibyte(batches: &[Vec], records: u64) { + assert!( + batches.len() >= 3, + "{records} records of about 100 KB took {} transaction(s)", + batches.len() + ); + for (i, sizes) in batches.iter().enumerate() { + let before_last: i64 = sizes[..sizes.len() - 1].iter().sum(); + assert!( + before_last < 1 << 20, + "transaction {i} took {} records, {before_last} bytes before its last", + sizes.len() + ); + } +} + +/// A numbering transaction stops adding records to its batch once the +/// records before the next one hold a mebibyte, so the shard's row is held +/// for a bounded copy. Records of about 100 KB go about ten to a +/// transaction, and one pass of the worker still numbers them all. +#[tokio::test] +async fn a_numbering_batch_stops_at_a_mebibyte_of_records() { + let test = "a_numbering_batch_stops_at_a_mebibyte_of_records"; + if base_conn().is_none() { + return skip(test); + } + const RECORDS: u64 = 30; + let s = scratch(None, 12).await; + let (key_info, _, _) = s.stream_table().await; + let shard = s.shard_of("big").await; + for i in 0..RECORDS { + s.put(&key_info, "big", &format!("{i:02}"), &noise(100_000, i)) + .await; + } + let batches = number_in_batches(&s, &shard).await; + assert_batches_stop_at_a_mebibyte(&batches, RECORDS); + + s.cleanup().await; +} + +/// A batch counts each record at the length of its JSON text, not at the +/// size PostgreSQL stores after compression, because the copy decompresses +/// and writes the whole record. Records of about 100 KB that compress to a +/// few kilobytes still go about ten to a transaction. Counted at their +/// stored size, all 30 went into one transaction. +#[tokio::test] +async fn a_numbering_batch_counts_compressible_records_at_their_text_length() { + let test = "a_numbering_batch_counts_compressible_records_at_their_text_length"; + if base_conn().is_none() { + return skip(test); + } + const RECORDS: u64 = 30; + let s = scratch(None, 12).await; + let (key_info, _, _) = s.stream_table().await; + let shard = s.shard_of("big").await; + for i in 0..RECORDS { + s.put( + &key_info, + "big", + &format!("{i:02}"), + &compressible(100_000, i), + ) + .await; + } + let stored: i64 = sqlx::query_scalar( + "SELECT max(pg_column_size(record_data))::bigint FROM stream_pending WHERE shard_id = $1", + ) + .bind(&shard) + .fetch_one(&s.db) + .await + .expect("read the stored sizes"); + assert!( + stored < 10_000, + "a record of about 100 KB is stored in {stored} bytes, so it is compressed" + ); + let batches = number_in_batches(&s, &shard).await; + assert_batches_stop_at_a_mebibyte(&batches, RECORDS); + + s.cleanup().await; +} + +/// Writers put, delete, and put two items on two shards in one +/// TransactWriteItems, on eight hot items, while the numbering worker runs in +/// a loop and two consumers page every shard. Each consumer must read every +/// record once, both must read each shard in the same order, and each item's +/// records must chain: the old image of a record is the new image of the +/// item's record before it, or none after a delete. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn writes_of_every_kind_beside_the_worker_reach_two_consumers_in_order() { + let test = "writes_of_every_kind_beside_the_worker_reach_two_consumers_in_order"; + if base_conn().is_none() { + return skip(test); + } + const WRITERS: usize = 8; + const OPS: usize = 120; + let s = scratch(None, 20).await; + let (key_info, table_id, arn) = s.stream_table().await; + let shards: Vec = + sqlx::query_scalar("SELECT shard_id FROM stream_shards WHERE table_id = $1 ORDER BY 1") + .bind(&table_id) + .fetch_all(&s.db) + .await + .expect("list the shards"); + // Four hot items on one shard and four on another. + let (mut left, mut right) = (Vec::new(), Vec::new()); + for i in 0..200 { + let pk = format!("h{i}"); + let shard = s.shard_of(&pk).await; + if shard == shards[0] && left.len() < 4 { + left.push(pk); + } else if shard == shards[1] && right.len() < 4 { + right.push(pk); + } + } + assert_eq!((left.len(), right.len()), (4, 4), "hot items on two shards"); + + let maps = ExpressionMaps::default(); + let cap = capture(); + let expected = std::sync::atomic::AtomicUsize::new(0); + let done = AtomicBool::new(false); + let writers = async { + futures::future::join_all((0..WRITERS).map(|w| { + let (s, key_info, maps, cap, left, right, expected) = + (&s, &key_info, &maps, &cap, &left, &right, &expected); + async move { + for i in 0..OPS { + let tag = format!("w{w}-{i}"); + let (a, b) = (&left[(w + i) % 4], &right[(w * 3 + i) % 4]); + match i % 10 { + 7 => { + let key = BTreeMap::from([ + ("pk".to_owned(), AttributeValue::S(a.clone())), + ("sk".to_owned(), AttributeValue::S("s".to_owned())), + ]); + let old = s + .engine + .delete_item(key_info, &key, true, None, maps, Some(cap)) + .await + .expect("a delete"); + if old.is_some() { + expected.fetch_add(1, Ordering::SeqCst); + } + } + 8 | 9 => { + let (x, y) = (item(a, "s", &tag), item(b, "s", &format!("{tag}b"))); + let put = |item| TransactWriteOp::Put { + key_info, + item, + condition: None, + maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(cap.clone()), + }; + let ops = [put(&x), put(&y)]; + if s.engine.transact_write_items(&ops, None).await.is_ok() { + expected.fetch_add(2, Ordering::SeqCst); + } + } + _ => { + s.put(key_info, a, "s", &tag).await; + expected.fetch_add(1, Ordering::SeqCst); + } + } + } + } + })) + .await; + done.store(true, Ordering::SeqCst); + }; + let worker = async { + while !done.load(Ordering::SeqCst) { + s.engine + .number_pending_stream_records_for_tests() + .await + .expect("a worker pass"); + tokio::time::sleep(Duration::from_millis(1)).await; + } + }; + let consumer = || async { + let mut cursors: Vec> = vec![None; shards.len()]; + let mut read: Vec> = vec![Vec::new(); shards.len()]; + loop { + let finished = done.load(Ordering::SeqCst); + for ((shard, cursor), got) in shards.iter().zip(cursors.iter_mut()).zip(read.iter_mut()) + { + loop { + let (page, last) = s.read(&arn, shard, cursor.as_deref()).await; + if page.is_empty() { + break; + } + *cursor = last; + got.extend(page); + } + } + if finished { + return read; + } + tokio::time::sleep(Duration::from_millis(2)).await; + } + }; + let ((), (), first, second) = tokio::time::timeout(Duration::from_secs(120), async { + tokio::join!(writers, worker, consumer(), consumer()) + }) + .await + .expect("the writers, the worker and the consumers finished"); + + let expected = expected.load(Ordering::SeqCst); + for (name, read) in [("first", &first), ("second", &second)] { + let all: Vec<&StreamRecord> = read.iter().flatten().collect(); + assert_eq!(all.len(), expected, "the {name} consumer's record count"); + let tags: Vec = all + .iter() + .filter_map(|r| match r.dynamodb.new_image.as_ref()?.get("v") { + Some(AttributeValue::S(v)) => Some(v.clone()), + _ => None, + }) + .collect(); + let distinct: BTreeSet<&String> = tags.iter().collect(); + assert_eq!( + distinct.len(), + tags.len(), + "the {name} consumer read a write twice" + ); + for shard in read { + for pair in shard.windows(2) { + assert!( + pair[0].dynamodb.sequence_number < pair[1].dynamodb.sequence_number, + "the {name} consumer's numbers go back on a shard" + ); + } + } + let mut last_image: BTreeMap> = BTreeMap::new(); + for r in &all { + let pk = string_key(r, "pk"); + let before = last_image.get(&pk).cloned().flatten(); + assert_eq!( + r.dynamodb.old_image, before, + "the {name} consumer: item {pk}'s record {} does not follow its record before", + r.dynamodb.sequence_number + ); + last_image.insert(pk, r.dynamodb.new_image.clone()); + } + } + let numbers = |read: &Vec>| -> Vec> { + read.iter() + .map(|shard| { + shard + .iter() + .map(|r| r.dynamodb.sequence_number.clone()) + .collect() + }) + .collect() + }; + assert_eq!( + numbers(&first), + numbers(&second), + "both consumers read the same records" + ); + + s.cleanup().await; +} diff --git a/crates/storage-postgres/tests/stream_record_clock.rs b/crates/storage-postgres/tests/stream_record_clock.rs index 1ea136154..d783b0960 100644 --- a/crates/storage-postgres/tests/stream_record_clock.rs +++ b/crates/storage-postgres/tests/stream_record_clock.rs @@ -217,12 +217,18 @@ async fn wait_for_lock_waiters(db: &PgPool, own_pid: i32, n: i64) { } /// Stored `(created_at, ApproximateCreationDateTime)` for every record in the -/// scratch database, in sequence order. -async fn stored_times(db: &PgPool) -> Vec<(time::OffsetDateTime, i64)> { +/// scratch database, in sequence order. A write leaves its record in +/// `stream_pending` until a read or the numbering worker numbers it (#808), +/// so the records are numbered first, as the worker would. +async fn stored_times(s: &Scratch) -> Vec<(time::OffsetDateTime, i64)> { + s.engine + .number_pending_stream_records_for_tests() + .await + .expect("number the pending stream records"); let rows: Vec<(time::OffsetDateTime, serde_json::Value)> = sqlx::query_as( "SELECT created_at, record_data FROM stream_records ORDER BY sequence_number", ) - .fetch_all(db) + .fetch_all(&s.db) .await .expect("read the stream records"); rows.into_iter() @@ -328,7 +334,7 @@ async fn created_at_matches_the_record_timestamp_when_the_write_waited() { .await .expect("the TTL-shaped delete succeeded"); - let times = stored_times(&s.db).await; + let times = stored_times(&s).await; assert_eq!(times.len(), 3, "seed, late, and remove records: {times:?}"); for (created_at, approx) in × { assert_eq!( diff --git a/crates/storage-postgres/tests/stream_shard_retention.rs b/crates/storage-postgres/tests/stream_shard_retention.rs index a68833d23..ea8510b36 100644 --- a/crates/storage-postgres/tests/stream_shard_retention.rs +++ b/crates/storage-postgres/tests/stream_shard_retention.rs @@ -276,3 +276,79 @@ async fn shards_of_a_deleted_table_go_once_its_records_are_gone() { s.cleanup().await; } + +/// A record that still waits for its sequence number in `stream_pending` +/// keeps the shards of a deleted table too. Deleting the shards would take +/// the record with them, through the foreign key's cascade, before the +/// numbering worker numbered it (#808). +#[tokio::test] +async fn a_pending_record_keeps_the_shards_of_a_deleted_table() { + if base_conn().is_none() { + eprintln!("SKIP: EXTENDDB_TEST_PG_CONNECTION_STRING is not set"); + return; + } + let s = scratch().await; + let gone = s + .engine + .create_table(ACCOUNT, stream_table("t_gone")) + .await + .expect("create the table to delete"); + let gone_id = gone.table_id.clone(); + s.engine + .delete_table( + ACCOUNT, + DeleteTableInput { + table_name: "t_gone".to_owned(), + }, + ) + .await + .expect("delete the table"); + let shard: String = sqlx::query_scalar( + "SELECT shard_id FROM stream_shards WHERE table_id = $1 ORDER BY shard_id LIMIT 1", + ) + .bind(&gone_id) + .fetch_one(&s.db) + .await + .expect("one shard id"); + // A write's record, stored before the table was deleted and not yet + // numbered. + sqlx::query( + "INSERT INTO stream_pending (shard_id, table_id, event_name, record_data, created_at) \ + VALUES ($1, $2, 'INSERT', '{\"dynamodb\": {\"SequenceNumber\": \"\"}}'::jsonb, NOW())", + ) + .bind(&shard) + .bind(&gone_id) + .execute(&s.db) + .await + .expect("seed a pending record"); + + s.engine + .cleanup_expired_stream_records(0) + .await + .expect("sweep"); + assert_eq!( + shard_count(&s.db, &gone_id).await, + 4, + "shards stay while a record of theirs waits for its number" + ); + + let moved = s + .engine + .number_pending_stream_records_for_tests() + .await + .expect("number the pending record"); + assert_eq!(moved, 1, "the record is numbered"); + // The numbered record is past a retention of zero hours, so the sweep + // trims it, and then the shards. + s.engine + .cleanup_expired_stream_records(0) + .await + .expect("sweep"); + assert_eq!( + shard_count(&s.db, &gone_id).await, + 0, + "the shards go once the record has been numbered and trimmed" + ); + + s.cleanup().await; +} diff --git a/crates/storage-postgres/tests/twi_conflict.rs b/crates/storage-postgres/tests/twi_conflict.rs index 8d5205469..50d1e10fe 100644 --- a/crates/storage-postgres/tests/twi_conflict.rs +++ b/crates/storage-postgres/tests/twi_conflict.rs @@ -865,6 +865,12 @@ async fn a_delete_that_loses_its_reservation_removes_the_winner_everywhere() { .expect("count c"); assert_eq!(left, 0, "c is gone from {t}"); } + // A write leaves its stream record in stream_pending until a read or the + // numbering worker numbers it (#808). This test stands in for the worker. + s.engine + .number_pending_stream_records_for_tests() + .await + .expect("number the pending stream records"); let records: Vec<(serde_json::Value,)> = sqlx::query_as( "SELECT record_data FROM stream_records WHERE table_id = $1 ORDER BY sequence_number", ) diff --git a/crates/storage-sqlite/src/migrations/mod.rs b/crates/storage-sqlite/src/migrations/mod.rs index e784fa82c..94a987d99 100644 --- a/crates/storage-sqlite/src/migrations/mod.rs +++ b/crates/storage-sqlite/src/migrations/mod.rs @@ -118,6 +118,10 @@ pub fn migration(id: MigrationId) -> Migration { m0006_backup_catalog_index::SOURCE, )], }, + MigrationId::StreamPending => Migration::NotApplicable( + "SQLite runs one writer at a time, so its stream records commit in the order of \ + their sequence numbers", + ), } } diff --git a/crates/storage-sqlite/src/migrations/released_tests.rs b/crates/storage-sqlite/src/migrations/released_tests.rs index fe9bb5b09..e453df2ef 100644 --- a/crates/storage-sqlite/src/migrations/released_tests.rs +++ b/crates/storage-sqlite/src/migrations/released_tests.rs @@ -399,11 +399,14 @@ async fn a_catalog_an_older_release_wrote_to_is_reported_and_reconverged() { next_second().await; let report = reconverge(&db).await.expect("reconverge"); - // Every convergent migration ran again; migration 2, which is not - // convergent, is named as skipped. + // Every convergent migration that applies on SQLite ran again; + // migration 2, which is not convergent, is named as skipped. let convergent = MigrationId::ALL .into_iter() - .filter(|id| super::migration(*id).convergent()) + .filter(|id| { + let m = super::migration(*id); + m.convergent() && !m.bodies().is_empty() + }) .count(); assert_eq!(report.applied.len(), convergent, "{trace}: {report:?}"); let skipped = format!("Skipped {}:", MigrationId::ResourceTagConditionKeys); diff --git a/crates/storage-sqlite/src/migrations/tests.rs b/crates/storage-sqlite/src/migrations/tests.rs index 82b3d3b80..934a986ad 100644 --- a/crates/storage-sqlite/src/migrations/tests.rs +++ b/crates/storage-sqlite/src/migrations/tests.rs @@ -264,13 +264,26 @@ async fn a_fresh_database_is_current_after_one_run_and_seeds_nothing() { status(&db).await, CatalogStatus::Current { version: LATEST } ); + // An id that does not apply on SQLite is recorded as not_applicable. + let applies = MigrationId::ALL + .into_iter() + .filter(|id| !migration(*id).bodies().is_empty()) + .count(); assert_eq!( count( &db, "SELECT COUNT(*) FROM extenddb_migrations WHERE outcome = 'applied'" ) .await, - i64::from(LATEST) + i64::try_from(applies).unwrap() + ); + assert_eq!( + count( + &db, + "SELECT COUNT(*) FROM extenddb_migrations WHERE outcome = 'not_applicable'" + ) + .await, + i64::from(LATEST) - i64::try_from(applies).unwrap() ); assert_eq!(version_row(&db).await, Some(LATEST.to_string())); assert_eq!( diff --git a/crates/storage/src/migrations/registry.rs b/crates/storage/src/migrations/registry.rs index 5814e60aa..6f8038c5e 100644 --- a/crates/storage/src/migrations/registry.rs +++ b/crates/storage/src/migrations/registry.rs @@ -73,11 +73,17 @@ migration_ids! { BackupCatalogIndex = 6, "backup_catalog_index", "create the backup catalog index, the restore job fields, the maintenance claim \ record and the catalog id"; + /// Adds the PostgreSQL `stream_pending` table. A write stores its stream + /// record there, and the record gets its sequence number after the write + /// commits, so a reader never passes a number that a later commit fills + /// in (#808). + StreamPending = 7, "stream_pending", + "add stream_pending, which holds a PostgreSQL stream record until its write commits"; } /// Highest id that has shipped in a release. Bodies at or below it are frozen: /// each backend pins their checksums, kinds and convergent flags, and -/// `forget` refuses them. The v1.0 release PR raises it to 5. +/// `forget` refuses them. The v1.0 release PR raises it to 7. pub const RELEASED_THROUGH: u32 = 0; impl MigrationId { diff --git a/docs/getting-started.md b/docs/getting-started.md index c1c27db98..c5e2ea3fb 100755 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -269,7 +269,7 @@ Config: extenddb.toml Catalog database: extenddb_catalog --- Checking catalog migrations... - OK: Current: the catalog is at migration 6 + OK: Current: the catalog is at migration 7 --- Checking catalog connection... OK: Connected to catalog. --- Checking data database... @@ -287,7 +287,7 @@ extenddb runs as a daemon (background process) and logs to syslog. On startup it ```bash ./target/release/extenddb serve --config extenddb.toml -# extenddb 0.1.14 (catalog migration 6) starting on 127.0.0.1:18443 +# extenddb 0.1.14 (catalog migration 7) starting on 127.0.0.1:18443 # storage: postgres (postgresql://extenddb:***@localhost:5432/extenddb_catalog) ``` @@ -1367,7 +1367,7 @@ Each runner requires its tools to be installed. The runner checks prerequisites ```bash ./target/release/extenddb version # extenddb 0.1.14 -# catalog migration 6 (postgres) +# catalog migration 7 (postgres) # commit abc1234 # built 2026-04-17T12:00:00Z ``` diff --git a/docs/manuals/02-design-guide.md b/docs/manuals/02-design-guide.md index 5d7b7ea2f..571e7014b 100755 --- a/docs/manuals/02-design-guide.md +++ b/docs/manuals/02-design-guide.md @@ -40,6 +40,7 @@ On PostgreSQL, the data database connection string is stored in the catalog's `s | `encryption_keys` | AES-256-GCM key used to encrypt access key secrets. | | `stream_shards` | Stream shard metadata (parent shard, sequence range). | | `stream_records` | Stream change records with 24-hour retention. | +| `stream_pending` | Stream records that wait for their sequence number. A write stores its record here, and the record moves to `stream_records` once the write has committed. | ### Data Tables @@ -136,6 +137,8 @@ Credential lookups (access key ID → decrypted secret) go through each server's Stream records are captured atomically with data writes. The engine constructs a `StreamCapture` struct with metadata (stream ARN, view type, shard ID, sequence number, keys). The storage backend persists the stream record in the same PostgreSQL transaction as the data write. +On PostgreSQL the record waits in `stream_pending`, with no sequence number, until the write commits. GetRecords and a `LATEST` shard iterator first number one batch of the committed records of the shard they read, in the order the writes stored them: 1,000 records, or fewer once the records hold 1 MiB of JSON text. Every 250 ms a background worker numbers up to 10,000 committed records of each shard that has some, and skips a shard that another transaction numbers, so the records of a shard that nobody reads are numbered too. A number drawn inside the write's transaction could commit after a higher number that a consumer had already read past, and that consumer would never receive the record. A read returns the records whose writes committed before it started, except in three cases: more records of the shard wait for their numbers than one batch takes, another transaction numbers the shard for longer than one second, or the read's numbering fails, which the server logs as a warning. The records left waiting get higher numbers than any the read returned, from a later read or the worker, and a later read returns them. In the same three cases a `LATEST` iterator can return records whose writes committed before it was taken. The order needs `stream_seq` and `stream_pending_id_seq` to keep a cache of one value per session, which migration 7 sets; a larger cache lets a record take a number below one a consumer has read. + For UpdateItem, the `new_image` is not known until after `apply_update` runs inside the transaction, so the storage backend constructs the full `StreamRecord` after the update. ### Shard Model diff --git a/docs/manuals/04-quickstart-setup-guide.md b/docs/manuals/04-quickstart-setup-guide.md index 08333d9dd..70b3e95e3 100755 --- a/docs/manuals/04-quickstart-setup-guide.md +++ b/docs/manuals/04-quickstart-setup-guide.md @@ -130,7 +130,7 @@ Check the version: ```bash ./target/release/extenddb version # extenddb 0.1.14 -# catalog migration 6 (postgres) +# catalog migration 7 (postgres) # commit abc1234 # built 2026-04-17T12:00:00Z ``` @@ -172,7 +172,7 @@ Config: extenddb.toml Catalog database: extenddb_catalog --- Checking catalog migrations... - OK: Current: the catalog is at migration 6 + OK: Current: the catalog is at migration 7 --- Checking catalog connection... OK: Connected to catalog. --- Checking data database... diff --git a/docs/manuals/07-upgrade-manual.md b/docs/manuals/07-upgrade-manual.md index fb24895ca..c3bc03bfc 100755 --- a/docs/manuals/07-upgrade-manual.md +++ b/docs/manuals/07-upgrade-manual.md @@ -4,7 +4,7 @@ ## Current Status -This release versions the PostgreSQL, SQLite, Cassandra and MongoDB catalogs by a migration ledger and carries catalog migrations 1 to 6, which `extenddb migrate --list` prints. **Every deployment created by a 0.1.x release runs `extenddb migrate --yes` once** before a server of this release starts on it: the server refuses a catalog with pending migrations. Migration 1 brings the catalog of every 0.1.x release to the v1.0 schema, migration 5 drops `continuous_backups`, a table and on MongoDB a collection (see [Catalog 0.0.5](#catalog-005)), and migration 6 adds the backup catalog index and the restore job records ([below](#migration-6-the-backup-catalog-index)). +This release versions the PostgreSQL, SQLite, Cassandra and MongoDB catalogs by a migration ledger and carries catalog migrations 1 to 7, which `extenddb migrate --list` prints. **Every deployment created by a 0.1.x release runs `extenddb migrate --yes` once** before a server of this release starts on it: the server refuses a catalog with pending migrations. Migration 1 brings the catalog of every 0.1.x release to the v1.0 schema, migration 5 drops `continuous_backups`, a table and on MongoDB a collection (see [Catalog 0.0.5](#catalog-005)), and migration 6 adds the backup catalog index and the restore job records ([below](#migration-6-the-backup-catalog-index)). Migration 7 adds the `stream_pending` table to the PostgreSQL data database, where a write stores its stream record until the record gets its sequence number after the write commits; it does not apply to the other backends. A backup that a 0.1.x release took can restore without its global and local secondary indexes: a SQLite backup taken by 0.1.14 records only the table's key schema and attribute definitions, and its restore creates the table with its items and no index. Take a new backup of each table after upgrading to v1.0, and keep the 0.1.x backups for their items only. @@ -39,7 +39,7 @@ Exit status 6 means the migration lock, or a running server, was still held when `verify` also checks the connection to the data database. When that check fails, `verify` exits 1 on a Behind catalog instead of 3, because `migrate --yes` cannot apply the pending migrations while the data database cannot be reached. It still exits 4 or 5 for the statuses that need a person. -A migration is additive when the previous release serves the catalog correctly with it applied, and breaking otherwise. Migrations 1 to 6 are breaking: do not run a 0.1.x server against a migrated catalog. A 0.1.x binary that writes to a migrated catalog (its `migrate --yes`, or a SQLite dev server) leaves it WrittenByOlderRelease. `migrate --yes --reconverge` repairs it by running the convergent migrations again. Migration 2, which rewrites stored policies, is not convergent: the repair skips it and prints a line that says so (see the v0.1.14 notes below). +A migration is additive when the previous release serves the catalog correctly with it applied, and breaking otherwise. Migrations 1 to 7 are breaking: do not run a 0.1.x server against a migrated catalog. A 0.1.x binary that writes to a migrated catalog (its `migrate --yes`, or a SQLite dev server) leaves it WrittenByOlderRelease. `migrate --yes --reconverge` repairs it by running the convergent migrations again. Migration 2, which rewrites stored policies, is not convergent: the repair skips it and prints a line that says so (see the v0.1.14 notes below). ### Migration 6: the backup catalog index diff --git a/docs/manuals/08-install-linux.md b/docs/manuals/08-install-linux.md index 2b868522e..c892006fc 100755 --- a/docs/manuals/08-install-linux.md +++ b/docs/manuals/08-install-linux.md @@ -118,7 +118,7 @@ Expected: ``` === extenddb verify === ... - OK: Current: the catalog is at migration 6 + OK: Current: the catalog is at migration 7 ... === HEALTHY: All checks passed === ``` diff --git a/docs/manuals/09-install-macos.md b/docs/manuals/09-install-macos.md index b1ca6de2d..eb44653fd 100755 --- a/docs/manuals/09-install-macos.md +++ b/docs/manuals/09-install-macos.md @@ -96,7 +96,7 @@ Expected: ``` === extenddb verify === ... - OK: Current: the catalog is at migration 6 + OK: Current: the catalog is at migration 7 ... === HEALTHY: All checks passed === ``` diff --git a/docs/manuals/12-extending-extenddb-storage.md b/docs/manuals/12-extending-extenddb-storage.md index 93db24814..2c8af5c9d 100755 --- a/docs/manuals/12-extending-extenddb-storage.md +++ b/docs/manuals/12-extending-extenddb-storage.md @@ -402,7 +402,7 @@ The PostgreSQL implementation makes backend-specific choices. These are implemen 2. **Transaction isolation**: the PostgreSQL backend runs `transact_write_items` at `READ COMMITTED` and makes it serializable with locks held until commit: `SELECT ... FOR UPDATE` on every item that exists, and a reserved unique key on every item that a ConditionCheck or Delete finds missing. Your backend needs equivalent isolation guarantees. -3. **Sequence generation** — stream sequence numbers use PostgreSQL sequences (`nextval`). Your backend needs a monotonic counter mechanism. +3. **Sequence generation**: stream sequence numbers come from a PostgreSQL sequence (`nextval`), drawn after the write commits. A write stores its record in `stream_pending`, and a numbering step moves a shard's committed records to `stream_records` in the order they were stored, one numbering of a shard at a time; GetRecords and a `LATEST` iterator run one batch of it on their shard before they read. A number drawn inside the write's transaction could commit after a higher number that a consumer had already read past. Your backend needs numbers that become visible on each shard in increasing order: once a consumer has read a number, no record with a lower number may appear on that shard later ([RFC-0003](../rfcs/0003-backend-acceptance-criteria.md), criterion 5.2). 4. **CASCADE deletes** — account deletion cascades through foreign keys. Your backend needs equivalent cascade logic (can be application-level). diff --git a/docs/schema/postgres.txt b/docs/schema/postgres.txt index 144a6d8c4..31ec3693c 100644 --- a/docs/schema/postgres.txt +++ b/docs/schema/postgres.txt @@ -347,6 +347,7 @@ table extenddb_migrations table gsi_pending table idempotency_tokens table schema_history +table stream_pending table stream_records table stream_shards table vector_index_holds @@ -371,6 +372,13 @@ column idempotency_tokens.fingerprint: text NOT NULL column idempotency_tokens.created_at: timestamp with time zone NOT NULL DEFAULT now() column schema_history.filename: text NOT NULL column schema_history.applied_at: timestamp with time zone NOT NULL DEFAULT now() +column stream_pending.shard_id: text NOT NULL +column stream_pending.id: bigint NOT NULL DEFAULT nextval('stream_pending_id_seq'::regclass) +column stream_pending.table_id: text NOT NULL +column stream_pending.event_name: text NOT NULL +column stream_pending.record_data: jsonb NOT NULL +column stream_pending.record_bytes: bigint NOT NULL DEFAULT 0 +column stream_pending.created_at: timestamp with time zone NOT NULL column stream_records.shard_id: text NOT NULL column stream_records.sequence_number: text NOT NULL column stream_records.table_id: text NOT NULL @@ -395,6 +403,7 @@ index idx_stream_records_created: CREATE INDEX idx_stream_records_created ON pub index idx_stream_shards_table: CREATE INDEX idx_stream_shards_table ON public.stream_shards USING btree (table_id) index idx_vector_index_holds_table: CREATE INDEX idx_vector_index_holds_table ON public.vector_index_holds USING btree (table_id) index schema_history_pkey: CREATE UNIQUE INDEX schema_history_pkey ON public.schema_history USING btree (filename) +index stream_pending_pkey: CREATE UNIQUE INDEX stream_pending_pkey ON public.stream_pending USING btree (shard_id, id) index stream_records_pkey: CREATE UNIQUE INDEX stream_records_pkey ON public.stream_records USING btree (shard_id, sequence_number) index stream_shards_pkey: CREATE UNIQUE INDEX stream_shards_pkey ON public.stream_shards USING btree (shard_id) index vector_index_holds_pkey: CREATE UNIQUE INDEX vector_index_holds_pkey ON public.vector_index_holds USING btree (table_id, index_id) @@ -402,9 +411,12 @@ constraint extenddb_migrations.extenddb_migrations_pkey: PRIMARY KEY (id) constraint gsi_pending.gsi_pending_pkey: PRIMARY KEY (id) constraint idempotency_tokens.idempotency_tokens_pkey: PRIMARY KEY (account_id, token) constraint schema_history.schema_history_pkey: PRIMARY KEY (filename) +constraint stream_pending.stream_pending_pkey: PRIMARY KEY (shard_id, id) +constraint stream_pending.stream_pending_shard_id_fkey: FOREIGN KEY (shard_id) REFERENCES stream_shards(shard_id) ON DELETE CASCADE constraint stream_records.stream_records_pkey: PRIMARY KEY (shard_id, sequence_number) constraint stream_records.stream_records_shard_id_fkey: FOREIGN KEY (shard_id) REFERENCES stream_shards(shard_id) ON DELETE CASCADE constraint stream_shards.stream_shards_pkey: PRIMARY KEY (shard_id) constraint vector_index_holds.vector_index_holds_pkey: PRIMARY KEY (table_id, index_id) sequence gsi_pending_id_seq: bigint START 1 INCREMENT 1 +sequence stream_pending_id_seq: bigint START 1 INCREMENT 1 sequence stream_seq: bigint START 1 INCREMENT 1 diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 81fbbc7b3..df803d1cd 100755 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -820,23 +820,29 @@ The error log line names the exact relations for one leaked table and gives this **Fix:** Check PostgreSQL connectivity. Verify the table's stream shards exist in the `stream_shards` table. If the table was created before streams were enabled, the shards may not have been initialized. -### `Stream capture: failed to write record for : ` +### `Stream cleanup worker: ` -**Cause:** A stream record was constructed but could not be persisted to the `stream_records` table. The data write succeeded — only the stream record is missing. +**Cause:** The background worker that deletes stream records older than 24 hours encountered a database error. Expired records will accumulate until the worker succeeds. -**Fix:** Check PostgreSQL connectivity and disk space. If the error mentions a unique constraint violation, two writes to the same shard may have occurred in the same microsecond — retry the operation. +**Fix:** Check PostgreSQL connectivity. The worker retries every hour automatically. -### `Stream capture: failed to get sequence number: ` +### `Stream record numbering failed: ` -**Cause:** extenddb could not generate a sequence number for a stream record. The data write succeeded — only the stream record is missing. +**Cause:** The background worker that gives PostgreSQL stream records their sequence numbers encountered a database error. GetRecords numbers one batch of the records of the shard it reads before it reads them, so a consumer that keeps reading a shard still receives its records, unless the same error stops that numbering too (see the next entry). The records of shards that nobody reads stay in the `stream_pending` table, and the retention sweep does not trim them, until the worker succeeds. -**Fix:** Check PostgreSQL connectivity. +**Fix:** Check PostgreSQL connectivity. After a failure the worker tries again a little over a second later. -### `Stream cleanup worker: ` +### `Stream record numbering before a read of shard failed: ` -**Cause:** The background worker that deletes stream records older than 24 hours encountered a database error. Expired records will accumulate until the worker succeeds. +**Cause:** GetRecords, or GetShardIterator with `LATEST`, could not number the shard's committed stream records before it read the shard, because of a database error. The call went on and answered from the records numbered before. The records left in `stream_pending` get higher sequence numbers from a later read or the background worker, and a later GetRecords call returns them. -**Fix:** Check PostgreSQL connectivity. The worker retries every hour automatically. +**Fix:** Check PostgreSQL connectivity and the error in the message. While the error repeats on every read of the shard and in the worker, consumers of that shard receive no new records. + +### GetRecords is slow on PostgreSQL while a long transaction runs + +**Cause:** Every stream record passes through the `stream_pending` table, and the numbering step deletes it there. A transaction that stays open on the data database, such as the one a backup of a large table reads from, keeps autovacuum from removing those deleted rows. Until it ends, each GetRecords on a busy shard and each numbering read past them. + +**Fix:** Find the long transaction in `pg_stat_activity` (`xact_start`). Once it ends, autovacuum removes the rows; `VACUUM stream_pending` removes them at once. ## Management Console Errors diff --git a/tests/test_cli_vector_catalog_migration.py b/tests/test_cli_vector_catalog_migration.py index f5c4b4ad4..718b07675 100644 --- a/tests/test_cli_vector_catalog_migration.py +++ b/tests/test_cli_vector_catalog_migration.py @@ -259,7 +259,7 @@ def test_init_creates_the_catalog_at_the_current_migration(self, cli_env): assert _ids_in("catalog") == list(range(1, _latest() + 1)) data_ids = _ledger_ids(_data_db(cli_env)) assert data_ids == _ids_in("data") - # Migrations 1 and 4 have bodies in the data database. + # Migrations 1, 4 and 7 have bodies in the data database. assert data_ids[:2] == [1, 4], data_ids # Migration 1 also writes the legacy ledger rows, so a release before