From 14f6ef0e37b4c117835d34c2fc559dfd650fc3e8 Mon Sep 17 00:00:00 2001 From: hMB2 <0400eb0da32afbbf16cdd1dee6ffd243587e8f3d6a2823fe7dfa95d16b002b7d@buzz.block.builderlab.xyz> Date: Wed, 7 Oct 2026 02:13:41 -0700 Subject: [PATCH] db: move read-scope schema to 0057; bound its locks and prebuild path 0056 shipped in #7906 and is checksum-frozen, so restore it byte-for-byte and move this branch's columns, index and root-index drop to 0057. 0057 sets lock/statement budgets like 0049, admits a concurrent prebuild of idx_event_mentions_scope_received, and validates its catalog shape. Parity of the altered tables is checked against a migrated database. Signed-off-by: hMB2 <0400eb0da32afbbf16cdd1dee6ffd243587e8f3d6a2823fe7dfa95d16b002b7d@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/migration.rs | 36 ++++++-- crates/buzz-db/src/runtime/tests.rs | 3 + .../personal_read_migration_postgres_tests.rs | 91 +++++++++++++++++++ .../tests/thread_window_postgres_tests.rs | 20 +++- docs/buzz-v1-read-state.md | 56 ++++++++++-- migrations/0056_personal_read_state.sql | 38 +------- migrations/0057_personal_read_scopes.sql | 85 +++++++++++++++++ schema/schema.sql | 2 +- 8 files changed, 276 insertions(+), 55 deletions(-) create mode 100644 crates/buzz-db/src/runtime/tests/personal_read_migration_postgres_tests.rs create mode 100644 migrations/0057_personal_read_scopes.sql diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 26ceba80ca..e834b6d6a1 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,12 +705,17 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 56); + assert_eq!(migrations.len(), 57); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql .as_str() .contains("CREATE TABLE personal_read_accounts")); + assert_eq!(migrations[56].version, 57); + assert!(migrations[56] + .sql + .as_str() + .contains("idx_event_mentions_scope_received")); assert_eq!(migrations[48].version, 49); assert_eq!(migrations[49].version, 50); assert_eq!(migrations[50].version, 51); @@ -2061,11 +2066,14 @@ mod postgres_tests { .sql .as_ref(), ); - for (table, definition) in personal.tables { - assert_eq!( - schema.tables.get(&table), - Some(&definition), - "personal read table {table} differs" + // 0057 adds columns to these tables, so their textual definitions + // differ from schema.sql by design; column and index parity is + // checked against a migrated database in + // admin_schema_parity_between_desired_state_and_migrations. + for table in personal.tables.keys() { + assert!( + schema.tables.contains_key(table), + "schema.sql is missing personal read table {table}" ); } expected_fences.extend(personal.fence_attachments); @@ -2501,14 +2509,18 @@ mod postgres_tests { .await .expect("connect migrated probe database"); MIGRATOR - .run_to(55, &migrated) + .run_to(57, &migrated) .await - .expect("apply migrations 1-55"); + .expect("apply migrations 1-57"); for table in [ "relay_admin_actions", "relay_admin_outbox", "relay_operator_audit", + "personal_read_accounts", + "personal_read_frontiers", + "thread_metadata", + "event_mentions", ] { assert_eq!( columns(&desired, table).await, @@ -2526,6 +2538,14 @@ mod postgres_tests { ); } + // Columns only: idx_channels_id_live already differs between + // schema.sql and the migrations, independent of 0057. + assert_eq!( + columns(&desired, "channels").await, + columns(&migrated, "channels").await, + "column parity mismatch for channels" + ); + desired.close().await; migrated.close().await; for probe_db in [desired_db, migrated_db] { diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 3998e10b3b..dc89132267 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -3967,3 +3967,6 @@ async fn serving_write_guard_insert_indexes_mentions_in_event_transaction() { .await .expect("release serving-write lease")); } + +#[path = "tests/personal_read_migration_postgres_tests.rs"] +mod personal_read_migration_postgres_tests; diff --git a/crates/buzz-db/src/runtime/tests/personal_read_migration_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/personal_read_migration_postgres_tests.rs new file mode 100644 index 0000000000..0d1319f2f3 --- /dev/null +++ b/crates/buzz-db/src/runtime/tests/personal_read_migration_postgres_tests.rs @@ -0,0 +1,91 @@ +use super::*; + +const SCOPE_INDEX: &str = "CREATE INDEX CONCURRENTLY idx_event_mentions_scope_received \ + ON event_mentions (community_id,pubkey_hex,channel_id,root_id,received_at)"; +const SCOPE_COLUMNS: &str = "ALTER TABLE event_mentions \ + ADD COLUMN IF NOT EXISTS received_at TIMESTAMPTZ, ADD COLUMN IF NOT EXISTS root_id BYTEA"; + +async fn regclass_oid(pool: &PgPool, name: &str) -> Option { + sqlx::query_scalar("SELECT to_regclass($1)::oid::bigint") + .bind(name) + .fetch_one(pool) + .await + .unwrap() +} + +async fn migrated_version(pool: &PgPool) -> i64 { + sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations") + .fetch_one(pool) + .await + .unwrap() +} + +/// The documented brownfield path for 0057: prebuild the event_mentions +/// index concurrently and pre-drop the root index. The migration must keep +/// the prebuilt index, reject a wrong or invalid one without recording the +/// version, and bound its budgets like 0049. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_personal_read_prebuild_validation() { + let admin = PgPool::connect(&admin_url().await).await.unwrap(); + let (pool, name) = create_scratch_db_through(&admin, "pr_prebuild", Some(56)).await; + sqlx::raw_sql(SCOPE_COLUMNS).execute(&pool).await.unwrap(); + for setup in [ + "CREATE INDEX idx_event_mentions_scope_received \ + ON event_mentions (community_id,pubkey_hex,channel_id,received_at)", + // Simulate an invalid concurrent-build remnant only in this disposable superuser DB. + "CREATE INDEX idx_event_mentions_scope_received \ + ON event_mentions (community_id,pubkey_hex,channel_id,root_id,received_at); \ + UPDATE pg_index SET indisvalid=false \ + WHERE indexrelid='idx_event_mentions_scope_received'::regclass", + ] { + sqlx::raw_sql(setup).execute(&pool).await.unwrap(); + let error = migration::run_migrations(&pool).await.unwrap_err(); + assert!( + error.to_string().contains("invalid or wrong definition"), + "must reject catalog shape, not merely time out: {error}" + ); + assert_eq!(migrated_version(&pool).await, 56); + sqlx::query("DROP INDEX CONCURRENTLY idx_event_mentions_scope_received") + .execute(&pool) + .await + .unwrap(); + } + + sqlx::query(SCOPE_INDEX).execute(&pool).await.unwrap(); + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS public.idx_thread_metadata_root") + .execute(&pool) + .await + .unwrap(); + let oid = regclass_oid(&pool, "idx_event_mentions_scope_received").await; + assert!(oid.is_some()); + migration::run_migrations(&pool).await.unwrap(); + assert_eq!( + oid, + regclass_oid(&pool, "idx_event_mentions_scope_received").await + ); + assert_eq!(regclass_oid(&pool, "idx_thread_metadata_root").await, None); + assert!(migrated_version(&pool).await >= 57); + drop_scratch_db(&admin, pool, &name).await; +} + +/// Without a prebuild, 0057 builds the index and drops the root index itself. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_personal_read_fresh_build_and_budgets() { + let migration = include_str!("../../../../../migrations/0057_personal_read_scopes.sql"); + assert!(migration.contains("SET LOCAL lock_timeout = '1s'")); + assert!(migration.contains("SET LOCAL statement_timeout = '5s'")); + + let admin = PgPool::connect(&admin_url().await).await.unwrap(); + let (pool, name) = create_scratch_db_through(&admin, "pr_fresh", Some(56)).await; + assert!(regclass_oid(&pool, "idx_thread_metadata_root") + .await + .is_some()); + migration::run_migrations(&pool).await.unwrap(); + assert!(regclass_oid(&pool, "idx_event_mentions_scope_received") + .await + .is_some()); + assert_eq!(regclass_oid(&pool, "idx_thread_metadata_root").await, None); + drop_scratch_db(&admin, pool, &name).await; +} diff --git a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs index 6e4248a740..ecdac3f3cf 100644 --- a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs @@ -270,9 +270,14 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer() .await .unwrap(); - // Also bind the public migrator: the advisory schema/destruction lock, - // ledger write and post-migration catalog checks must work with ingestion. - let production_result = migration::run_migrations(&pool).await; + // Also bind the migrator: the advisory schema/destruction lock and ledger + // write must work with ingestion through 0056. 0057 alters thread_metadata + // and so cannot finish while this writer is held; its bounded lock budget + // must fail it fast instead of queueing ingestion behind it. + let production_result = migration::run_migrations_through(&pool, 56).await; + let started = std::time::Instant::now(); + let blocked_0057 = migration::run_migrations(&pool).await; + let blocked_for = started.elapsed(); sqlx::query(insert) .bind(community) .bind(channel) @@ -311,6 +316,15 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer() production_result.is_ok(), "production migrator must preserve ingestion progress: {production_result:?}" ); + let blocked_0057 = blocked_0057.expect_err("0057 cannot take thread_metadata under a writer"); + assert!( + blocked_0057.to_string().contains("lock timeout"), + "0057 must fail on its lock budget: {blocked_0057}" + ); + assert!( + blocked_for < std::time::Duration::from_secs(4), + "{blocked_for:?}" + ); assert_eq!(version, 56); assert_eq!(final_oid, oid, "prebuild must not be replaced"); assert_eq!(count, 4, "all writer witnesses must persist"); diff --git a/docs/buzz-v1-read-state.md b/docs/buzz-v1-read-state.md index 58a5965cfa..0913448115 100644 --- a/docs/buzz-v1-read-state.md +++ b/docs/buzz-v1-read-state.md @@ -230,12 +230,56 @@ device-local. shared eight-second intent-processing deadline after admission. Limits are containment, not a production capacity claim. -Apply migration 0056 (or the equivalent desired schema). It creates two -private tables, adds `thread_metadata.last_reply_received_at`, +Apply migrations 0056 and 0057 (or the equivalent desired schema). 0056 +creates two private tables. 0057 adds `personal_read_accounts.started_at` +(NULL for accounts from 0056, which start at their next read intent), +`personal_read_frontiers.following`, `thread_metadata.last_reply_received_at`, `channels.last_timeline_received_at` and `event_mentions.received_at`/`root_id` (NULL for rows from before it, which no -position can reach), and one index on `event_mentions`. Ingest writes the -author's own position and follow rows in the same transaction as the message. +position can reach), and one index on `event_mentions`. It also drops +`idx_thread_metadata_root`, whose columns lead `idx_thread_metadata_window` +(0049). Ingest writes the author's own position and follow rows in the same +transaction as the message. + +Like 0049, 0057 bounds lock waits and statement time, so it fails +instead of blocking `event_mentions` writes while building its index on a +populated table. Brownfield deployments prebuild the index first. Adding +nullable columns is a catalog-only change; `CREATE INDEX CONCURRENTLY` cannot +run inside a transaction block: + +```sql +ALTER TABLE public.event_mentions + ADD COLUMN IF NOT EXISTS received_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS root_id BYTEA; +CREATE INDEX CONCURRENTLY idx_event_mentions_scope_received + ON public.event_mentions (community_id, pubkey_hex, channel_id, root_id, received_at); +``` + +Verify it with the `pg_index` query in +[thread-window-deployment.md](thread-window-deployment.md), using +`idx_event_mentions_scope_received`. All flags must be `true`, and the +definition must be exactly: + +```text +CREATE INDEX idx_event_mentions_scope_received ON public.event_mentions USING btree (community_id, pubkey_hex, channel_id, root_id, received_at) +``` + +Migration 0057 then skips the build and validates the catalog shape again. It +rejects an invalid or differently defined index. Recover the same way as for +0049: `DROP INDEX CONCURRENTLY`, then prebuild again. + +Its transactional `DROP INDEX IF EXISTS idx_thread_metadata_root` needs a brief +`ACCESS EXCLUSIVE` lock on `thread_metadata`, bounded by the same lock budget. +On a busy database, drop it beforehand, also outside a transaction: + +```sql +DROP INDEX CONCURRENTLY IF EXISTS public.idx_thread_metadata_root; +``` + +0057 still takes brief exclusive locks to add columns to the two read-state +tables, `thread_metadata`, `channels` and `event_mentions`. If a long reader holds one of them past the +lock budget, startup fails and can be retried; ingestion does not queue behind +it. During a rolling deploy, relay processes still on the old code write `event_mentions` rows without `received_at`/`root_id` and do not mark posting @@ -280,8 +324,8 @@ is no new public account export/reset endpoint. Operator-assisted erasure/export must use the established authenticated operational process and explicitly scope both community and actor; never equate the read-time horizon with data erasure. -Migration 0056 must be applied before this relay serves, enabled or not: started -without it and with auto-migration off, the relay stops before readiness. There +Migrations 0056 and 0057 must be applied before this relay serves, enabled or +not: started without them and with auto-migration off, the relay stops before readiness. There is no down migration, and disabling the API is not a rollback. A relay built before 0056 that restarts with `BUZZ_AUTO_MIGRATE=true` (the Helm default) refuses to start on the migrated schema. Whole-community deletion run from a diff --git a/migrations/0056_personal_read_state.sql b/migrations/0056_personal_read_state.sql index 2d51ae8cf0..41889e8890 100644 --- a/migrations/0056_personal_read_state.sql +++ b/migrations/0056_personal_read_state.sql @@ -2,18 +2,12 @@ -- A frontier is the relay arrival time (events.received_at) of the message a -- context was read through, never signed event time, which the sender chooses. -- An empty root_id covers only the channel timeline; a root-specific frontier --- covers that thread, without inheritance. A thread row exists for each --- thread the actor follows or has unfollowed. Ingest moves the author's own --- frontiers through each message they post. +-- covers that thread, without inheritance. -- threads_through_timestamp is the only cross-context cut: an explicit -- whole-channel read that also covers every thread in that channel. --- started_at is the actor's first read intent and the floor of every --- position: until then nothing counts, and from then the actor starts caught --- up. Ingest creates accounts for thread membership without starting them. CREATE TABLE personal_read_accounts ( community_id UUID NOT NULL REFERENCES communities(id), actor BYTEA NOT NULL CHECK (octet_length(actor) = 32), - started_at TIMESTAMPTZ, PRIMARY KEY (community_id, actor) ); @@ -26,9 +20,6 @@ CREATE TABLE personal_read_frontiers ( -- Whole-channel cut covering every thread; channel rows only. threads_through_timestamp TIMESTAMPTZ CHECK (threads_through_timestamp IS NULL OR root_id = ''::bytea), - -- Thread rows only: false after unfollow, kept so later replies don't - -- re-follow. Replying or a follow intent sets it again. - following BOOLEAN NOT NULL DEFAULT true CHECK (following OR root_id <> ''::bytea), PRIMARY KEY (community_id, actor, channel_id, root_id), FOREIGN KEY (community_id, actor) REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE, @@ -37,33 +28,6 @@ CREATE TABLE personal_read_frontiers ( ); --- Arrival time of a thread root's latest reply: the sidebar skips threads --- with nothing new since the reader's position without reading replies. -ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ; - --- Arrival time of a channel's latest eligible timeline message (top-level or --- broadcast reply): the sidebar answers "anything new?" for a caught-up --- reader with one comparison instead of walking past replies. Unindexed so --- updates stay HOT. NULL for channels with no timeline message since this --- migration, which no position can be behind: every position is floored at --- a started_at set after it. -ALTER TABLE channels ADD COLUMN last_timeline_received_at TIMESTAMPTZ; - --- Arrival time and read scope of each mention, so the sidebar counts a --- reader's unread mentions in one channel timeline or thread from its --- position. root_id is the thread a reply belongs to, NULL when the mention --- shows on the channel timeline (a top-level message or broadcast reply). --- Existing rows stay NULL: every position is floored at its account's --- started_at, set by a read intent after this migration, so no earlier --- mention can be unread. -ALTER TABLE event_mentions ADD COLUMN received_at TIMESTAMPTZ, ADD COLUMN root_id BYTEA; -CREATE INDEX idx_event_mentions_scope_received - ON event_mentions (community_id, pubkey_hex, channel_id, root_id, received_at); - --- idx_thread_metadata_window (0049) leads with the same columns. With both, --- the planner can take this narrower index for a thread's latest reply and --- then sort every reply in the thread instead of reading the first row. -DROP INDEX IF EXISTS idx_thread_metadata_root; SELECT attach_community_write_fence('personal_read_accounts'); SELECT attach_community_write_fence('personal_read_frontiers'); diff --git a/migrations/0057_personal_read_scopes.sql b/migrations/0057_personal_read_scopes.sql new file mode 100644 index 0000000000..37be6bfe3d --- /dev/null +++ b/migrations/0057_personal_read_scopes.sql @@ -0,0 +1,85 @@ +-- Channel and thread read scopes on top of 0056 (checksum-frozen). +-- Startup is bounded like 0049: this migration takes exclusive locks on +-- thread_metadata, channels and event_mentions, and must fail rather than +-- queue ingestion behind a long reader or an index build on a populated +-- table. Brownfield operators prebuild idx_event_mentions_scope_received and +-- pre-drop idx_thread_metadata_root as documented in +-- docs/buzz-v1-read-state.md. +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; + +-- started_at is the actor's first read intent and the floor of every +-- position: until then nothing counts, and from then the actor starts caught +-- up. Ingest creates accounts for thread membership without starting them. +-- Accounts from 0056 stay NULL and start at their next read intent. +ALTER TABLE personal_read_accounts ADD COLUMN started_at TIMESTAMPTZ; + +-- A thread row exists for each thread the actor follows or has unfollowed. +-- Ingest moves the author's own frontiers through each message they post. +-- following: thread rows only, false after unfollow, kept so later replies +-- don't re-follow. Replying or a follow intent sets it again. +ALTER TABLE personal_read_frontiers + ADD COLUMN following BOOLEAN NOT NULL DEFAULT true + CHECK (following OR root_id <> ''::bytea); + +-- Arrival time of a thread root's latest reply: the sidebar skips threads +-- with nothing new since the reader's position without reading replies. +ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ; + +-- Arrival time of a channel's latest eligible timeline message (top-level or +-- broadcast reply): the sidebar answers "anything new?" for a caught-up +-- reader with one comparison instead of walking past replies. Unindexed so +-- updates stay HOT. NULL for channels with no timeline message since this +-- migration, which no position can be behind: every position is floored at +-- a started_at set after it. +ALTER TABLE channels ADD COLUMN last_timeline_received_at TIMESTAMPTZ; + +-- Arrival time and read scope of each mention, so the sidebar counts a +-- reader's unread mentions in one channel timeline or thread from its +-- position. root_id is the thread a reply belongs to, NULL when the mention +-- shows on the channel timeline (a top-level message or broadcast reply). +-- Existing rows stay NULL: every position is floored at its account's +-- started_at, set by a read intent after this migration, so no earlier +-- mention can be unread. +-- IF NOT EXISTS admits the brownfield prebuild, which adds these columns +-- before building the index concurrently. As in 0049, a prebuilt index skips +-- CREATE INDEX and must pass the catalog check below. +ALTER TABLE event_mentions + ADD COLUMN IF NOT EXISTS received_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS root_id BYTEA; +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'event_mentions' + AND ((column_name = 'received_at' AND data_type = 'timestamp with time zone') + OR (column_name = 'root_id' AND data_type = 'bytea')) + HAVING count(*) = 2 + ) THEN + RAISE EXCEPTION 'event_mentions.received_at/root_id have the wrong type; see docs/buzz-v1-read-state.md'; + END IF; + + IF to_regclass('public.idx_event_mentions_scope_received') IS NULL THEN + CREATE INDEX idx_event_mentions_scope_received + ON public.event_mentions (community_id, pubkey_hex, channel_id, root_id, received_at); + END IF; + + IF NOT EXISTS ( + SELECT 1 FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relname = 'idx_event_mentions_scope_received' + AND i.indisvalid AND i.indisready AND i.indislive + AND pg_get_indexdef(i.indexrelid) = + 'CREATE INDEX idx_event_mentions_scope_received ON public.event_mentions USING btree (community_id, pubkey_hex, channel_id, root_id, received_at)' + ) THEN + RAISE EXCEPTION 'idx_event_mentions_scope_received invalid or wrong definition; see docs/buzz-v1-read-state.md'; + END IF; +END $$; + +-- idx_thread_metadata_window (0049) leads with the same columns. With both, +-- the planner can take this narrower index for a thread's latest reply and +-- then sort every reply in the thread instead of reading the first row. +-- This supersedes 0049's "legacy root/parent indexes remain" for the root +-- index (0049 is checksum-frozen); the parent index remains. +DROP INDEX IF EXISTS idx_thread_metadata_root; diff --git a/schema/schema.sql b/schema/schema.sql index aaff2018df..bc0780df60 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -101,7 +101,7 @@ CREATE TABLE channels ( participant_hash BYTEA, ttl_seconds INT, ttl_deadline TIMESTAMPTZ, - -- Arrival of the latest eligible timeline message; see 0056. + -- Arrival of the latest eligible timeline message; see 0057. last_timeline_received_at TIMESTAMPTZ, PRIMARY KEY (community_id, id), CONSTRAINT chk_channels_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid)