From 360751f04623d2c6a782342974b7de5a79771f77 Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sat, 17 Oct 2026 13:59:41 +0000 Subject: [PATCH 1/2] feat(TC-1274): maintain epoch_head so writes stop scanning epoch history transact found a space's head epochs (a new epoch's parents) with an epoch/epoch_order anti-join over the space's whole history on every write, under the space's sequence lock. It now reads them from a new epoch_head table (PK (space, id)) and records the new epoch as the space's only head in the same transaction: one UPDATE in the usual one-row case. The cache is trusted only when its newest seq is the space's newest (MAX(event_order.seq) + 1, which transact already computes). Every epoch, including one appended by a release that does not write epoch_head, outranks all existing ones under the same lock, so a stale cache is always detected; that space falls back to the anti-join once and the write repairs it. Heads are taken in id byte order; nothing re-derives an epoch hash from its parents. m20261018_000000_epoch_head creates only the empty table, with no foreign keys, so it takes no lock on existing tables. A gentle online backfill ([epoch_head_backfill], on by default at 2 spaces/s, pausing at least as long as each space took, stopping at its first error) fills spaces written before the table existed; writes are correct without it. The TC-951 round-trip budgets rise by the measured cache write: sign-in 75 -> 77, kv put 37 -> 38. Also folds in two #341 review nits: the in-progress check's pg_locks is limited to the current database, and a failed index plus a non-building SHARE UPDATE EXCLUSIVE holder still gets the DROP advice. --- CHANGELOG.md | 1 + tinycloud-core/src/db.rs | 80 +- tinycloud-core/src/epoch_head.rs | 692 ++++++++++++++++++ tinycloud-core/src/kv_sync.rs | 16 +- tinycloud-core/src/lib.rs | 1 + ...20261017_000000_epoch_order_space_index.rs | 42 ++ .../migrations/m20261018_000000_epoch_head.rs | 60 ++ tinycloud-core/src/migrations/mod.rs | 2 + tinycloud-core/src/models/epoch_head.rs | 23 + tinycloud-core/src/models/mod.rs | 1 + tinycloud-node-server/src/config.rs | 58 ++ .../src/epoch_head_backfill.rs | 293 ++++++++ tinycloud-node-server/src/lib.rs | 24 + .../tests/postgres_round_trip_budget.rs | 15 +- 14 files changed, 1281 insertions(+), 27 deletions(-) create mode 100644 tinycloud-core/src/epoch_head.rs create mode 100644 tinycloud-core/src/migrations/m20261018_000000_epoch_head.rs create mode 100644 tinycloud-core/src/models/epoch_head.rs create mode 100644 tinycloud-node-server/src/epoch_head_backfill.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 3941a362..b36f0b62 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## [Unreleased] - New migration `m20261017_000000_epoch_order_space_index` adds `idx_epoch_order_space_parent_child ON epoch_order (space, parent, child)` for the head-epoch anti-join that every write runs under its space lock (TC-1274). On a fresh database the build is instant. On PostgreSQL, if a valid index of that name and shape already exists, the migration issues no DDL and takes no lock on `epoch_order`. If the index of that name is INVALID or has a different shape, the boot fails, and nothing is applied, with an error naming the fix. On an existing database **without** the index, the migration builds it inside the boot's migration transaction. That holds a SHARE lock that blocks writes for the whole build, and the lock wait is bounded by the 30 s `lock_timeout` (TC-1176). **Runbook (PostgreSQL, before upgrading a node with history):** in autocommit, run `CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_epoch_order_space_parent_child ON epoch_order (space, parent, child);`. Then confirm `SELECT indisvalid FROM pg_index WHERE indexrelid = 'idx_epoch_order_space_parent_child'::regclass;` returns `t`. If it returns `f`, run `DROP INDEX CONCURRENTLY idx_epoch_order_space_parent_child;` and build it again. `down` keeps the index. MySQL is covered only by the generated statement, because the full MySQL migration chain does not run (TC-1314). +- Writes no longer scan a space's whole epoch history to find its head epochs (TC-1274). A new `epoch_head` table (migration `m20261018_000000_epoch_head`) caches each space's heads. Every write maintains it in the same transaction, under the space's existing sequence lock. A write now reads O(heads) rows, normally one, instead of running the `epoch`/`epoch_order` anti-join, and usually costs one more statement, an `UPDATE` of the cached head. The cache is used only when its newest `seq` is the space's newest. Otherwise that space falls back to the old anti-join once and the write repairs the cache, so epochs appended by older releases (a rolling deploy, or a rollback and upgrade) never produce wrong parents. Heads are taken in `id` byte order. Previously the anti-join returned them in no defined order. This affects only a merge after a pre-TC-732 fork, and nothing re-derives an epoch hash from its parents. The migration only creates the empty table, without foreign keys, so it takes no locks on existing tables. After the node starts serving, a gentle online backfill checks each space once. It runs in one short transaction per space, under that space's lock, at no more than `epoch_head_backfill.spaces_per_second` (default 2) and pauses after each space for at least as long as the space took. It stops at its first error with one log line, and logs progress every `progress_every` spaces (default 500). Turn it off with `[epoch_head_backfill] enabled = false` or `TINYCLOUD_EPOCH_HEAD_BACKFILL__ENABLED=false`. Writes stay correct without it. Keep the `idx_epoch_order_space_parent_child` index: the fallback and the backfill use it. - Request events include validated/echoed UUID correlation, typed replacement flags, node-keyed target account/space pseudonyms and per-request database acquisition/execution timings and attempt counts. Driver/actor instrumentation includes failures and request-awaited audit batches; forwarding requires explicit first-party origin classification. A shared core `Pseudonymizer` preserves existing account/space hashes, adds the service/space-scoped database hash primitive for F6, and is derived once at startup for injection into consumers. See `docs/request-telemetry.md` for the wire contract, key custody, restart joinability and coverage/exclusions (TC-959 N2). - CACAO/SIWE delegation ingestion normalizes ECDSA recovery IDs 0/1 to 27/28 before hashing and persisting the canonical DAG-CBOR bytes (TC-1071). Equivalent low-s signatures now share one delegation CID, including the CID used for revocation lookup. High-s signatures are rejected during decoding, matching the existing EIP-191 verifier and EIP-2 low-s convention; UCAN/JWS bytes remain unchanged. Existing stored delegation IDs are not rewritten. - Synchronize the standalone M1, W5 and load-signer Cargo locks with CACAO's `k256` dependency so `--locked` dependency resolution succeeds (TC-1071). Keep the load signer compatible with the SDK's default invocation lifetime. diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index c2d97193..91c32043 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -541,6 +541,44 @@ where pub async fn begin_graph_transaction(&self) -> Result { begin_write_transaction(&self.conn, chain_isolation_level(&self.conn)).await } + + /// TC-1274 backfill: up to `limit` space IDs after `after`, in ID order. + pub async fn epoch_head_backfill_page( + &self, + after: Option<&SpaceId>, + limit: u64, + ) -> Result, DbErr> { + let mut query = space::Entity::find().order_by_asc(space::Column::Id); + if let Some(after) = after { + query = query.filter(space::Column::Id.gt(SpaceIdWrap(after.clone()))); + } + Ok(query + .limit(limit) + .all(&self.conn) + .await? + .into_iter() + .map(|space| space.id.0) + .collect()) + } + + /// TC-1274 backfill of one space: rewrite its cached head epochs from + /// history unless they are provably current. Takes the space's sequence + /// locks as a write does, so it serializes with that space's writers and + /// holds no other space. Returns whether it rewrote the cache. + pub async fn repair_epoch_heads(&self, space: &SpaceId) -> Result { + let _sequence = self + .acquire_space_sequence_guards(std::slice::from_ref(space)) + .await; + let _writer = match &self.writer_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; + let tx = self.begin_graph_transaction().await?; + crate::kv_sync::lock_space_sequences(&tx, [space]).await?; + let repaired = crate::epoch_head::repair(&tx, &SpaceIdWrap(space.clone())).await?; + tx.commit().await?; + Ok(repaired) + } } impl SpaceDatabase @@ -3344,34 +3382,18 @@ pub(crate) async fn transact( } } - // get 'most recent' epochs for each of the spaces - let mut most_recent = epoch::Entity::find() - .select_only() - .left_join(epoch_order::Entity) - .filter( - Condition::all() - .add(epoch::Column::Space.is_in(event_spaces.keys().cloned().map(SpaceIdWrap))) - .add(epoch_order::Column::Child.is_null()), - ) - .column(epoch::Column::Space) - .column(epoch::Column::Id) - .into_tuple::<(SpaceIdWrap, Hash)>() - .all(db) - .await? - .into_iter() - .fold( - HashMap::new(), - |mut m: HashMap>, (space, epoch)| { - m.entry(space).or_default().push(epoch); - m - }, - ); + // The current heads of each space (TC-1274): O(#heads) from the + // `epoch_head` cache, falling back to history only for a space whose + // cache is not provably current. The sequence locks taken above are + // held to commit. + let head_spaces: Vec = event_spaces.keys().cloned().map(SpaceIdWrap).collect(); + let mut heads = crate::epoch_head::current(db, &head_spaces, &max_seqs).await?; // get all the orderings and associated data let (epoch_order, space_order, event_order, epochs) = event_spaces .into_iter() .map(|(space, events)| { - let parents = most_recent.remove(&space).unwrap_or_default(); + let parents = heads.parents.remove(&space).unwrap_or_default(); let epoch = epoch_hash(&space, &events, &parents)?; let seq = max_seqs.remove(&space).unwrap_or(0); Ok((space, (epoch, events, seq, parents))) @@ -3450,6 +3472,18 @@ pub(crate) async fn transact( TxError::EpochInsert(e) })?; + // Each space's only head is now the epoch just appended: its parents + // were all of the previous heads. + crate::epoch_head::advance( + db, + &heads, + space_order + .iter() + .map(|(space, (seq, epoch, _, _))| (SpaceIdWrap(space.clone()), *epoch, *seq)) + .collect(), + ) + .await?; + // save epoch orderings if !epoch_order.is_empty() { epoch_order::Entity::insert_many(epoch_order) diff --git a/tinycloud-core/src/epoch_head.rs b/tinycloud-core/src/epoch_head.rs new file mode 100644 index 00000000..1a7a09e7 --- /dev/null +++ b/tinycloud-core/src/epoch_head.rs @@ -0,0 +1,692 @@ +//! TC-1274: per-space head epochs without scanning history. +//! +//! A new epoch's parents are all of its space's current heads: the epochs +//! with no child in `epoch_order`. Finding them with the `epoch` / +//! `epoch_order` anti-join costs O(history) on every write. `epoch_head` +//! caches them instead, and `transact` rewrites a space's rows, under the +//! space's sequence lock, to the one epoch it just appended. +//! +//! The cache is trusted only when it is provably current. Every epoch is +//! appended under the space's sequence lock with `seq = MAX(event_order.seq) +//! + 1` (TC-732), so it outranks every existing epoch, including those +//! appended by releases that never write `epoch_head`. Rows written under the +//! lock are exact at that moment, and the newest epoch is always a head (its +//! child would have a larger `seq`). So the rows are still exact if and only +//! if their largest `seq` is the space's largest. Otherwise the space falls +//! back to the anti-join once and the caller rewrites its rows. Correctness +//! never depends on a backfill having run, or on every writer maintaining the +//! table. +//! +//! Heads are returned in `id` (byte) order. The anti-join had no order, and +//! the parent order only feeds the hash of a new merge epoch; nothing +//! re-derives an epoch hash from its parents. + +use std::collections::HashMap; + +use sea_orm::{ + sea_query::Expr, ColumnTrait, ConnectionTrait, DbErr, EntityTrait, QueryFilter, QueryOrder, + QuerySelect, +}; + +use crate::hash::Hash; +use crate::models::{epoch, epoch_head}; +use crate::relationships::{epoch_order, event_order}; +use crate::types::SpaceIdWrap; + +/// The next `seq` of `space`: one past its largest `event_order.seq`, or +/// `None` for a space with no events yet. `transact` computes the same value. +async fn next_seq(db: &C, space: &SpaceIdWrap) -> Result, DbErr> { + let max = event_order::Entity::find() + .filter(event_order::Column::Space.eq(space.clone())) + .select_only() + .column_as(event_order::Column::Seq.max(), "max_seq") + .into_tuple::>() + .one(db) + .await? + .flatten(); + max.map(|seq| { + seq.checked_add(1) + .ok_or_else(|| DbErr::Custom(format!("event_order seq overflow in space {}", space.0))) + }) + .transpose() +} + +/// Whether cached heads with these seqs are exact for a space whose next seq +/// is `next` (see the module docs). +fn is_current(cached: &[(Hash, i64)], next: Option) -> bool { + match (cached.iter().map(|(_, seq)| *seq).max(), next) { + (None, None) => true, + (Some(newest), Some(next)) => newest.checked_add(1) == Some(next), + _ => false, + } +} + +/// The heads of `spaces` from history: the anti-join `transact` used to run +/// on every write. O(history); with `idx_epoch_order_space_parent_child` it +/// is two index-only range scans per space. +async fn heads_from_history( + db: &C, + spaces: &[SpaceIdWrap], +) -> Result, DbErr> { + epoch::Entity::find() + .select_only() + .left_join(epoch_order::Entity) + .filter(epoch::Column::Space.is_in(spaces.iter().cloned())) + .filter(epoch_order::Column::Child.is_null()) + .column(epoch::Column::Space) + .column(epoch::Column::Id) + .column(epoch::Column::Seq) + .order_by_asc(epoch::Column::Space) + .order_by_asc(epoch::Column::Id) + .into_tuple::<(SpaceIdWrap, Hash, i64)>() + .all(db) + .await +} + +async fn cached_heads( + db: &C, + spaces: &[SpaceIdWrap], +) -> Result>, DbErr> { + let mut cached: HashMap> = HashMap::new(); + for row in epoch_head::Entity::find() + .filter(epoch_head::Column::Space.is_in(spaces.iter().cloned())) + .order_by_asc(epoch_head::Column::Space) + .order_by_asc(epoch_head::Column::Id) + .all(db) + .await? + { + cached.entry(row.space).or_default().push((row.id, row.seq)); + } + Ok(cached) +} + +/// The heads `transact` builds on, read under the sequence locks of `spaces`. +pub(crate) struct Heads { + /// The current heads of each space, in `id` order; a space without + /// epochs is absent. + pub(crate) parents: HashMap>, + /// How many cache rows each space has now (absent: none). + cached_rows: HashMap, +} + +/// The current heads of each of `spaces`. `next_seqs` holds each space's next +/// seq as `transact` computed it (absent for a space with no events). The +/// caller holds the sequence lock of every space and, after appending one +/// epoch per space, records them with [`advance`]. +pub(crate) async fn current( + db: &C, + spaces: &[SpaceIdWrap], + next_seqs: &HashMap, +) -> Result { + let mut cached = cached_heads(db, spaces).await?; + let mut parents = HashMap::new(); + let mut cached_rows = HashMap::new(); + let mut stale = Vec::new(); + for space in spaces { + let rows = cached.remove(space).unwrap_or_default(); + if !rows.is_empty() { + cached_rows.insert(space.clone(), rows.len()); + } + if is_current(&rows, next_seqs.get(space).copied()) { + if !rows.is_empty() { + parents.insert(space.clone(), rows.into_iter().map(|(id, _)| id).collect()); + } + } else { + stale.push(space.clone()); + } + } + if !stale.is_empty() { + for (space, id, _) in heads_from_history(db, &stale).await? { + parents.entry(space).or_insert_with(Vec::new).push(id); + } + } + Ok(Heads { + parents, + cached_rows, + }) +} + +/// Record each `(space, id, seq)` as its space's only head: the epoch just +/// appended, whose parents were all of the space's heads. One statement per +/// space in the usual case of one cached row. +pub(crate) async fn advance( + db: &C, + heads: &Heads, + appended: Vec<(SpaceIdWrap, Hash, i64)>, +) -> Result<(), DbErr> { + for (space, id, seq) in appended { + match heads.cached_rows.get(&space).copied().unwrap_or(0) { + 0 => insert(db, [(space, id, seq)]).await?, + 1 => { + epoch_head::Entity::update_many() + .col_expr(epoch_head::Column::Id, Expr::value(id)) + .col_expr(epoch_head::Column::Seq, Expr::value(seq)) + .filter(epoch_head::Column::Space.eq(space)) + .exec(db) + .await?; + } + _ => replace(db, std::slice::from_ref(&space), [(space.clone(), id, seq)]).await?, + } + } + Ok(()) +} + +async fn insert( + db: &C, + heads: impl IntoIterator, +) -> Result<(), DbErr> { + let rows = heads + .into_iter() + .map(|(space, id, seq)| epoch_head::ActiveModel { + space: sea_orm::Set(space), + id: sea_orm::Set(id), + seq: sea_orm::Set(seq), + }) + .collect::>(); + if !rows.is_empty() { + epoch_head::Entity::insert_many(rows).exec(db).await?; + } + Ok(()) +} + +/// Make `heads` (`(space, id, seq)`) the only cached heads of `spaces`. +async fn replace( + db: &C, + spaces: &[SpaceIdWrap], + heads: impl IntoIterator, +) -> Result<(), DbErr> { + epoch_head::Entity::delete_many() + .filter(epoch_head::Column::Space.is_in(spaces.iter().cloned())) + .exec(db) + .await?; + insert(db, heads).await +} + +/// Backfill one space: if its cached heads are not provably current, rewrite +/// them from history. Returns whether it rewrote them. The caller holds the +/// space's sequence lock. +pub(crate) async fn repair(db: &C, space: &SpaceIdWrap) -> Result { + let spaces = std::slice::from_ref(space); + let next = next_seq(db, space).await?; + let cached = cached_heads(db, spaces) + .await? + .remove(space) + .unwrap_or_default(); + if is_current(&cached, next) { + return Ok(false); + } + let heads = heads_from_history(db, spaces).await?; + replace(db, spaces, heads).await?; + Ok(true) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::migrate; + use crate::hash::hash; + use crate::keys::StaticSecret; + use crate::kv_sync::tests::{host, run, sqlite_db, Op, Owner, TestDb}; + use crate::migrations::Migrator; + use crate::models::space; + use crate::storage::memory::MemoryStore; + use crate::test_support::PostgresTestSchema; + use crate::SpaceDatabase; + use sea_orm::{DatabaseConnection, TransactionTrait}; + use sea_orm_migration::MigratorTrait; + use tinycloud_auth::resource::SpaceId; + + async fn postgres_db(fixture: &PostgresTestSchema) -> TestDb { + SpaceDatabase::new( + fixture.connect(15).await, + MemoryStore::default(), + StaticSecret::new([0u8; 32].to_vec()).unwrap(), + ) + .await + .unwrap() + } + + fn wrap(space: &SpaceId) -> SpaceIdWrap { + SpaceIdWrap(space.clone()) + } + + /// The cached heads of `space`, in id order. + async fn cached(db: &DatabaseConnection, space: &SpaceId) -> Vec { + cached_heads(db, &[wrap(space)]) + .await + .unwrap() + .remove(&wrap(space)) + .unwrap_or_default() + .into_iter() + .map(|(id, _)| id) + .collect() + } + + /// The true heads of `space`, from history, in id order. + async fn truth(db: &DatabaseConnection, space: &SpaceId) -> Vec { + heads_from_history(db, &[wrap(space)]) + .await + .unwrap() + .into_iter() + .map(|(_, id, _)| id) + .collect() + } + + /// The epoch with the largest seq in `space`. + async fn newest(db: &DatabaseConnection, space: &SpaceId) -> epoch::Model { + epoch::Entity::find() + .filter(epoch::Column::Space.eq(wrap(space))) + .order_by_desc(epoch::Column::Seq) + .one(db) + .await + .unwrap() + .unwrap() + } + + /// The parents recorded for `child`, in id order. + async fn parents_of(db: &DatabaseConnection, space: &SpaceId, child: Hash) -> Vec { + let mut parents: Vec = epoch_order::Entity::find() + .filter(epoch_order::Column::Space.eq(wrap(space))) + .filter(epoch_order::Column::Child.eq(child)) + .all(db) + .await + .unwrap() + .into_iter() + .map(|edge| edge.parent) + .collect(); + parents.sort(); + parents + } + + /// Append an epoch the way a release without `epoch_head` (v1.20.x) + /// does: `epoch`, `epoch_order` and `event_order` rows only. + async fn old_writer_append( + db: &DatabaseConnection, + space: &SpaceId, + parents: &[Hash], + seq: i64, + tag: &str, + ) -> Hash { + let id = hash(format!("tc1274-old-writer-epoch-{tag}").as_bytes()); + epoch::Entity::insert(epoch::ActiveModel { + seq: sea_orm::Set(seq), + id: sea_orm::Set(id), + space: sea_orm::Set(wrap(space)), + }) + .exec(db) + .await + .unwrap(); + for parent in parents { + epoch_order::Entity::insert(epoch_order::ActiveModel { + parent: sea_orm::Set(*parent), + child: sea_orm::Set(id), + space: sea_orm::Set(wrap(space)), + }) + .exec(db) + .await + .unwrap(); + } + event_order::Entity::insert(event_order::ActiveModel { + seq: sea_orm::Set(seq), + epoch: sea_orm::Set(id), + epoch_seq: sea_orm::Set(0), + event: sea_orm::Set(hash(format!("tc1274-old-writer-event-{tag}").as_bytes())), + space: sea_orm::Set(wrap(space)), + }) + .exec(db) + .await + .unwrap(); + id + } + + /// After every write the cache holds exactly the epoch just appended, + /// which is also the only head in history, and that epoch's parent is the + /// previous head. + async fn heads_follow_every_write(db: &TestDb) { + let conn = db.connection(); + let owner = Owner::generate(); + host(db, &owner).await; + let mut previous = Vec::new(); + for (round, key) in ["a", "b", "a", "c"].into_iter().enumerate() { + run(db, &owner, &[Op::Put(key, b"value")]).await; + let head = newest(conn, &owner.space).await; + assert_eq!(head.seq, round as i64); + assert_eq!( + cached(conn, &owner.space).await, + vec![head.id], + "round {round}" + ); + assert_eq!( + truth(conn, &owner.space).await, + vec![head.id], + "round {round}" + ); + assert_eq!(parents_of(conn, &owner.space, head.id).await, previous); + previous = vec![head.id]; + } + } + + #[tokio::test] + async fn sqlite_heads_follow_every_write() { + heads_follow_every_write(&sqlite_db().await).await; + } + + #[tokio::test] + async fn postgres_heads_follow_every_write() { + let Some(fixture) = PostgresTestSchema::new("postgres_heads_follow_every_write").await + else { + return; + }; + heads_follow_every_write(&postgres_db(&fixture).await).await; + } + + /// A release that does not maintain `epoch_head` appends several epochs, + /// ending in a fork (two epochs with the same parent and seq, as + /// concurrent pre-TC-732 writers left). The next new-release write must + /// take exactly the history's heads as parents, in id order, not the + /// stale cached head, and leave only itself cached. + async fn old_writer_history_with_a_fork_is_detected(db: &TestDb) { + let conn = db.connection(); + let owner = Owner::generate(); + host(db, &owner).await; + run(db, &owner, &[Op::Put("a", b"new release")]).await; + let first = newest(conn, &owner.space).await; + assert_eq!(cached(conn, &owner.space).await, vec![first.id]); + + let second = old_writer_append(conn, &owner.space, &[first.id], 1, "second").await; + let third = old_writer_append(conn, &owner.space, &[second], 2, "third").await; + let fork_a = old_writer_append(conn, &owner.space, &[third], 3, "fork-a").await; + let fork_b = old_writer_append(conn, &owner.space, &[third], 3, "fork-b").await; + let mut fork = vec![fork_a, fork_b]; + fork.sort(); + assert_eq!(truth(conn, &owner.space).await, fork); + assert_eq!( + cached(conn, &owner.space).await, + vec![first.id], + "cache is stale" + ); + + let tx = conn.begin().await.unwrap(); + let next = HashMap::from([(wrap(&owner.space), 4)]); + let heads = current(&tx, &[wrap(&owner.space)], &next).await.unwrap(); + tx.rollback().await.unwrap(); + assert_eq!( + heads.parents.get(&wrap(&owner.space)), + Some(&fork), + "id order" + ); + + run(db, &owner, &[Op::Put("b", b"new release again")]).await; + let merge = newest(conn, &owner.space).await; + assert_eq!(merge.seq, 4); + assert_eq!(parents_of(conn, &owner.space, merge.id).await, fork); + assert_eq!(cached(conn, &owner.space).await, vec![merge.id]); + assert_eq!(truth(conn, &owner.space).await, vec![merge.id]); + } + + #[tokio::test] + async fn sqlite_old_writer_history_with_a_fork_is_detected() { + old_writer_history_with_a_fork_is_detected(&sqlite_db().await).await; + } + + #[tokio::test] + async fn postgres_old_writer_history_with_a_fork_is_detected() { + let Some(fixture) = + PostgresTestSchema::new("postgres_old_writer_history_with_a_fork_is_detected").await + else { + return; + }; + old_writer_history_with_a_fork_is_detected(&postgres_db(&fixture).await).await; + } + + /// Writers in two node processes (separate pools and in-process locks) + /// race on one space. The history stays one chain and the cache ends + /// holding exactly its single head. + #[tokio::test] + async fn postgres_concurrent_writers_keep_exactly_one_head() { + let Some(fixture) = + PostgresTestSchema::new("postgres_concurrent_writers_keep_exactly_one_head").await + else { + return; + }; + let nodes = [postgres_db(&fixture).await, postgres_db(&fixture).await]; + let owner = std::sync::Arc::new(Owner::generate()); + host(&nodes[0], &owner).await; + const WRITES: usize = 24; + let writes = (0..WRITES).map(|i| { + let db = nodes[i % 2].clone(); + let owner = owner.clone(); + async move { + let key = format!("k{i}"); + run(&db, &owner, &[Op::Put(&key, b"v")]).await; + } + }); + futures::future::join_all(writes).await; + + let conn = nodes[0].connection(); + let head = newest(conn, &owner.space).await; + assert_eq!(head.seq, WRITES as i64 - 1); + assert_eq!(truth(conn, &owner.space).await, vec![head.id]); + assert_eq!(cached(conn, &owner.space).await, vec![head.id]); + let edges = epoch_order::Entity::find() + .filter(epoch_order::Column::Space.eq(wrap(&owner.space))) + .all(conn) + .await + .unwrap(); + assert_eq!(edges.len(), WRITES - 1, "one chain, no fork"); + } + + /// A database upgraded from v1.20.6 has history but an empty cache. + /// Repair makes every space's cache exact, including a forked space, and + /// a second pass changes nothing. + async fn repair_fills_upgraded_history_once(db: &TestDb) { + let conn = db.connection(); + let (linear, forked) = (Owner::generate(), Owner::generate()); + for owner in [&linear, &forked] { + host(db, owner).await; + run(db, owner, &[Op::Put("a", b"1")]).await; + run(db, owner, &[Op::Put("b", b"2")]).await; + } + let base = newest(conn, &forked.space).await; + old_writer_append(conn, &forked.space, &[base.id], 2, "upgrade-fork-a").await; + old_writer_append(conn, &forked.space, &[base.id], 2, "upgrade-fork-b").await; + // As left by v1.20.6, which has no epoch_head. + epoch_head::Entity::delete_many().exec(conn).await.unwrap(); + let empty = HostedNoHistory::new(db).await; + + for (owner, expected) in [(&linear.space, true), (&forked.space, true)] { + assert!(cached(conn, owner).await.is_empty()); + assert_eq!(db.repair_epoch_heads(owner).await.unwrap(), expected); + assert_eq!(cached(conn, owner).await, truth(conn, owner).await); + } + assert_eq!(cached(conn, &forked.space).await.len(), 2); + assert!( + !db.repair_epoch_heads(&empty.0).await.unwrap(), + "nothing to do" + ); + for owner in [&linear.space, &forked.space] { + let before = cached(conn, owner).await; + assert!(!db.repair_epoch_heads(owner).await.unwrap(), "idempotent"); + assert_eq!(cached(conn, owner).await, before); + } + // A write on the backfilled fork merges both heads and leaves one. + let fork = cached(conn, &forked.space).await; + run(db, &forked, &[Op::Put("merge", b"3")]).await; + let merge = newest(conn, &forked.space).await; + assert_eq!(parents_of(conn, &forked.space, merge.id).await, fork); + assert_eq!(cached(conn, &forked.space).await, vec![merge.id]); + } + + /// A hosted space with no writes yet. + struct HostedNoHistory(SpaceId); + + impl HostedNoHistory { + async fn new(db: &TestDb) -> Self { + let owner = Owner::generate(); + host(db, &owner).await; + Self(owner.space) + } + } + + #[tokio::test] + async fn sqlite_repair_fills_upgraded_history_once() { + repair_fills_upgraded_history_once(&sqlite_db().await).await; + } + + #[tokio::test] + async fn postgres_repair_fills_upgraded_history_once() { + let Some(fixture) = + PostgresTestSchema::new("postgres_repair_fills_upgraded_history_once").await + else { + return; + }; + repair_fills_upgraded_history_once(&postgres_db(&fixture).await).await; + } + + /// The backfill and writers on another node interleave on one space, + /// starting from a stale cache. Every repair runs under the space's + /// lock, so the cache ends exact. + #[tokio::test] + async fn postgres_repair_concurrent_with_writers_ends_exact() { + let Some(fixture) = + PostgresTestSchema::new("postgres_repair_concurrent_with_writers_ends_exact").await + else { + return; + }; + let (writer, backfill) = (postgres_db(&fixture).await, postgres_db(&fixture).await); + let owner = std::sync::Arc::new(Owner::generate()); + host(&writer, &owner).await; + run(&writer, &owner, &[Op::Put("seed", b"v")]).await; + epoch_head::Entity::delete_many() + .exec(writer.connection()) + .await + .unwrap(); + + let writes = { + let owner = owner.clone(); + async move { + for i in 0..12 { + let key = format!("k{i}"); + run(&writer, &owner, &[Op::Put(&key, b"v")]).await; + } + writer + } + }; + let repairs = async { + let mut repaired = 0; + for _ in 0..12 { + repaired += backfill.repair_epoch_heads(&owner.space).await.unwrap() as usize; + tokio::task::yield_now().await; + } + repaired + }; + let (writer, _) = tokio::join!(writes, repairs); + + let conn = writer.connection(); + let head = newest(conn, &owner.space).await; + assert_eq!(head.seq, 12); + assert_eq!(truth(conn, &owner.space).await, vec![head.id]); + assert_eq!(cached(conn, &owner.space).await, vec![head.id]); + } + + /// Upgrading a v1.20.6 database, with history and with the operator's + /// online `epoch_order (space, parent, child)` index (TC-1274 PR A) + /// already built, adds the empty table at boot; the backfill and the next + /// write then pick the history's exact heads. + #[tokio::test] + async fn postgres_v1_20_6_upgrade_with_operator_index() { + let test = "postgres_v1_20_6_upgrade_with_operator_index"; + let v1_20_6 = Migrator::migrations() + .iter() + .position(|migration| migration.name() == "m20261007_010000_database_identity_fence") + .unwrap() as u32 + + 1; + let Some(fixture) = PostgresTestSchema::migrated_through(test, Some(v1_20_6)).await else { + return; + }; + let conn = &fixture.conn; + let (linear, forked) = (Owner::generate(), Owner::generate()); + for owner in [&linear, &forked] { + space::Entity::insert(space::ActiveModel { + id: sea_orm::Set(wrap(&owner.space)), + }) + .exec(conn) + .await + .unwrap(); + let root = + old_writer_append(conn, &owner.space, &[], 0, &format!("{}-0", owner.space)).await; + old_writer_append( + conn, + &owner.space, + &[root], + 1, + &format!("{}-1", owner.space), + ) + .await; + } + let base = newest(conn, &forked.space).await; + old_writer_append(conn, &forked.space, &[base.id], 2, "v1206-fork-a").await; + old_writer_append(conn, &forked.space, &[base.id], 2, "v1206-fork-b").await; + conn.execute_unprepared( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_epoch_order_space_parent_child \ + ON epoch_order (space, parent, child)", + ) + .await + .unwrap(); + + migrate(&conn.clone().into()).await.unwrap(); + + let db = postgres_db(&fixture).await; + assert!(db.repair_epoch_heads(&linear.space).await.unwrap()); + assert_eq!( + cached(conn, &linear.space).await, + truth(conn, &linear.space).await + ); + let fork = truth(conn, &forked.space).await; + assert_eq!(fork.len(), 2); + // The forked space is left to its next write (no backfill yet). + run(&db, &forked, &[Op::Put("after", b"upgrade")]).await; + let merge = newest(conn, &forked.space).await; + assert_eq!(parents_of(conn, &forked.space, merge.id).await, fork); + assert_eq!(cached(conn, &forked.space).await, vec![merge.id]); + } + + #[tokio::test] + async fn backfill_pages_every_space_in_id_order() { + let db = sqlite_db().await; + let mut spaces = Vec::new(); + for _ in 0..5 { + let owner = Owner::generate(); + host(&db, &owner).await; + spaces.push(owner.space); + } + spaces.sort_by_key(|space| space.to_string()); + let mut seen = Vec::new(); + let mut after = None; + loop { + let page = db + .epoch_head_backfill_page(after.as_ref(), 2) + .await + .unwrap(); + if page.is_empty() { + break; + } + assert!(page.len() <= 2); + after = page.last().cloned(); + seen.extend(page); + } + assert_eq!(seen, spaces); + } + + #[test] + fn cache_is_current_only_when_it_holds_the_newest_seq() { + let id = hash(b"head"); + assert!(is_current(&[], None), "no history, nothing cached"); + assert!(!is_current(&[], Some(1)), "history, nothing cached"); + assert!(!is_current(&[(id, 0)], None), "cached heads, no history"); + assert!(is_current(&[(id, 4)], Some(5))); + assert!(!is_current(&[(id, 3)], Some(5)), "another writer appended"); + assert!(is_current(&[(id, 2), (hash(b"fork"), 4)], Some(5))); + assert!(!is_current(&[(id, i64::MAX)], Some(i64::MIN))); + } +} diff --git a/tinycloud-core/src/kv_sync.rs b/tinycloud-core/src/kv_sync.rs index 891d308d..76da742f 100644 --- a/tinycloud-core/src/kv_sync.rs +++ b/tinycloud-core/src/kv_sync.rs @@ -1812,7 +1812,7 @@ pub(crate) mod tests { .map(|epoch| (epoch.id, epoch.seq)) .collect::>(); for event in event_order::Entity::find() - .filter(event_order::Column::Space.eq(space)) + .filter(event_order::Column::Space.eq(space.clone())) .all(conn) .await .unwrap() @@ -1823,6 +1823,20 @@ pub(crate) mod tests { "{label}: an event must carry its epoch's seq" ); } + // TC-1274: the cached heads are exactly the chain's last epoch. + let cached = crate::models::epoch_head::Entity::find() + .filter(crate::models::epoch_head::Column::Space.eq(space)) + .all(conn) + .await + .unwrap() + .into_iter() + .map(|head| (head.id, head.seq)) + .collect::>(); + assert_eq!( + cached, + Vec::from_iter(epochs.last().map(|epoch| (epoch.id, epoch.seq))), + "{label}: the cached head must be the newest epoch" + ); epochs.len() } diff --git a/tinycloud-core/src/lib.rs b/tinycloud-core/src/lib.rs index 8bd82574..db327f91 100644 --- a/tinycloud-core/src/lib.rs +++ b/tinycloud-core/src/lib.rs @@ -11,6 +11,7 @@ pub mod did_resolution; pub mod duckdb; pub mod encryption; pub mod encryption_network; +pub(crate) mod epoch_head; pub mod events; pub mod hash; pub mod keys; diff --git a/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs b/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs index 8c81b45f..443b4feb 100644 --- a/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs +++ b/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs @@ -69,6 +69,8 @@ async fn postgres_index_present(manager: &SchemaManager<'_>) -> Result Result<(), DbErr> { + let mut id = ColumnDef::new(epoch_head::Column::Id); + if manager.get_database_backend() == sea_orm::DbBackend::MySql { + // As in m20261014: MySQL's bare BINARY is one byte and fixed-width + // BINARY would pad the multihash. + id.var_binary(255); + } else { + id.binary(); + } + manager + .create_table( + Table::create() + .table(epoch_head::Entity) + .if_not_exists() + .col( + ColumnDef::new(epoch_head::Column::Space) + .string() + .not_null(), + ) + .col(id.not_null()) + .col( + ColumnDef::new(epoch_head::Column::Seq) + .big_integer() + .not_null(), + ) + .primary_key( + Index::create() + .col(epoch_head::Column::Space) + .col(epoch_head::Column::Id), + ) + .to_owned(), + ) + .await + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(epoch_head::Entity).to_owned()) + .await + } +} diff --git a/tinycloud-core/src/migrations/mod.rs b/tinycloud-core/src/migrations/mod.rs index 0bb5bbdc..3d0618ba 100644 --- a/tinycloud-core/src/migrations/mod.rs +++ b/tinycloud-core/src/migrations/mod.rs @@ -27,6 +27,7 @@ pub mod m20261007_010000_database_identity_fence; pub mod m20261014_000000_delegation_cid_alias; pub mod m20261015_000000_delegation_cursor_index; pub mod m20261017_000000_epoch_order_space_index; +pub mod m20261018_000000_epoch_head; pub struct Migrator; @@ -62,6 +63,7 @@ impl MigratorTrait for Migrator { Box::new(m20261014_000000_delegation_cid_alias::Migration), Box::new(m20261015_000000_delegation_cursor_index::Migration), Box::new(m20261017_000000_epoch_order_space_index::Migration), + Box::new(m20261018_000000_epoch_head::Migration), ] } } diff --git a/tinycloud-core/src/models/epoch_head.rs b/tinycloud-core/src/models/epoch_head.rs new file mode 100644 index 00000000..82b9a81a --- /dev/null +++ b/tinycloud-core/src/models/epoch_head.rs @@ -0,0 +1,23 @@ +//! TC-1274: the current head epochs of each space, maintained by `transact` +//! under the space's sequence lock. A derived cache of the `epoch` / +//! `epoch_order` anti-join; see `crate::epoch_head` for when it is trusted. + +use crate::hash::Hash; +use crate::types::SpaceIdWrap; +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "epoch_head")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub space: SpaceIdWrap, + #[sea_orm(primary_key, auto_increment = false)] + pub id: Hash, + /// The head epoch's `seq`. + pub seq: i64, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/tinycloud-core/src/models/mod.rs b/tinycloud-core/src/models/mod.rs index afefa687..8e7f85a0 100644 --- a/tinycloud-core/src/models/mod.rs +++ b/tinycloud-core/src/models/mod.rs @@ -14,6 +14,7 @@ pub mod encryption_network; pub mod encryption_network_member; pub mod encryption_nonce; pub mod epoch; +pub mod epoch_head; pub mod hook_delivery; pub mod hook_subscription; pub mod invocation; diff --git a/tinycloud-node-server/src/config.rs b/tinycloud-node-server/src/config.rs index 9c2746c9..07b88785 100644 --- a/tinycloud-node-server/src/config.rs +++ b/tinycloud-node-server/src/config.rs @@ -50,6 +50,8 @@ pub struct Config { pub share_email: ShareEmailConfig, #[serde(default)] pub retention: RetentionConfig, + #[serde(default)] + pub epoch_head_backfill: EpochHeadBackfillConfig, } /// Production exact-email composition. The capability remains unavailable @@ -922,6 +924,62 @@ impl Default for RetentionConfig { } } +/// The online `epoch_head` backfill (TC-1274): one pass over every space +/// after the node starts serving, rewriting a space's cached head epochs +/// only when they are not provably current. Writes stay correct without it; +/// it moves each space's one-time history scan off its first write. +/// +/// On by default and deliberately slow. Section `[epoch_head_backfill]`; +/// env `TINYCLOUD_EPOCH_HEAD_BACKFILL__ENABLED`, +/// `TINYCLOUD_EPOCH_HEAD_BACKFILL__SPACES_PER_SECOND`, +/// `TINYCLOUD_EPOCH_HEAD_BACKFILL__PROGRESS_EVERY`. +#[derive(Serialize, Deserialize, Debug, Clone, Hash, PartialEq, Eq)] +pub struct EpochHeadBackfillConfig { + #[serde(default = "default_epoch_head_backfill_enabled")] + pub enabled: bool, + /// Upper bound on spaces checked per second. After each space the task + /// also pauses at least as long as that space took. + #[serde(default = "default_epoch_head_backfill_spaces_per_second")] + pub spaces_per_second: u32, + /// Log a progress line every this many spaces. + #[serde(default = "default_epoch_head_backfill_progress_every")] + pub progress_every: u32, +} + +fn default_epoch_head_backfill_enabled() -> bool { + true +} + +fn default_epoch_head_backfill_spaces_per_second() -> u32 { + 2 +} + +fn default_epoch_head_backfill_progress_every() -> u32 { + 500 +} + +impl Default for EpochHeadBackfillConfig { + fn default() -> Self { + Self { + enabled: default_epoch_head_backfill_enabled(), + spaces_per_second: default_epoch_head_backfill_spaces_per_second(), + progress_every: default_epoch_head_backfill_progress_every(), + } + } +} + +impl EpochHeadBackfillConfig { + pub fn validate(&self) -> Result<(), &'static str> { + if self.spaces_per_second == 0 { + return Err("epoch_head_backfill.spaces_per_second must be at least 1; set enabled = false to turn the backfill off"); + } + if self.progress_every == 0 { + return Err("epoch_head_backfill.progress_every must be at least 1"); + } + Ok(()) + } +} + #[derive(Serialize, Deserialize, Debug, Clone, Hash, PartialEq, Eq, Default)] #[serde(tag = "type")] pub enum Keys { diff --git a/tinycloud-node-server/src/epoch_head_backfill.rs b/tinycloud-node-server/src/epoch_head_backfill.rs new file mode 100644 index 00000000..44e372be --- /dev/null +++ b/tinycloud-node-server/src/epoch_head_backfill.rs @@ -0,0 +1,293 @@ +//! TC-1274: the gentle online `epoch_head` backfill. +//! +//! One pass over every space, in ID order, after the node starts serving. +//! Each space is checked in its own short transaction under its sequence lock +//! (`SpaceDatabase::repair_epoch_heads`), and rewritten only when its cached +//! heads are not provably current, so the pass is idempotent and safe to run +//! on several nodes at once. Writes are correct without it; it only moves +//! each space's one-time history scan off that space's first write. +//! +//! Gentle on a loaded database: at most `spaces_per_second` spaces, and after +//! each space a pause at least as long as that space took. The first error +//! stops the pass with one log line; there is no retry. Logs carry counts +//! only, never space IDs or database error text. + +use std::{future::Future, time::Duration}; + +use tinycloud_auth::resource::SpaceId; +use tinycloud_core::sea_orm::error::{DbErr, RuntimeErr, SqlxError}; +use tokio::time::Instant; + +use crate::{config::EpochHeadBackfillConfig, TinyCloud}; + +/// Spaces fetched per page. +const PAGE: u64 = 100; + +/// What the backfill needs from the node's database. +#[async_trait::async_trait] +pub trait EpochHeadBackfillTarget { + async fn page(&self, after: Option<&SpaceId>, limit: u64) -> Result, DbErr>; + async fn repair(&self, space: &SpaceId) -> Result; +} + +#[async_trait::async_trait] +impl EpochHeadBackfillTarget for TinyCloud { + async fn page(&self, after: Option<&SpaceId>, limit: u64) -> Result, DbErr> { + self.epoch_head_backfill_page(after, limit).await + } + + async fn repair(&self, space: &SpaceId) -> Result { + self.repair_epoch_heads(space).await + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BackfillEnd { + /// Every space was checked. + Done, + /// The node is shutting down. + Shutdown, + /// The first database error; the pass stopped there. + Failed, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BackfillOutcome { + pub end: BackfillEnd, + pub checked: u64, + pub repaired: u64, +} + +/// SQLSTATE of a database error, if it has one: safe to log, unlike the +/// error text, which can quote row values. +fn sqlstate(error: &DbErr) -> Option { + match error { + DbErr::Conn(RuntimeErr::SqlxError(SqlxError::Database(database))) + | DbErr::Exec(RuntimeErr::SqlxError(SqlxError::Database(database))) + | DbErr::Query(RuntimeErr::SqlxError(SqlxError::Database(database))) => { + database.code().map(|code| code.into_owned()) + } + _ => None, + } +} + +fn failed(stage: &'static str, error: &DbErr, checked: u64, repaired: u64) -> BackfillOutcome { + ::tracing::error!( + event = "epoch_head_backfill", + phase = "stopped", + stage, + sqlstate = sqlstate(error).as_deref().unwrap_or("none"), + checked, + repaired, + "epoch head backfill stopped at its first error; writes stay correct and repair each space on its next write" + ); + BackfillOutcome { + end: BackfillEnd::Failed, + checked, + repaired, + } +} + +/// Run one pass. `config` is validated (`spaces_per_second` and +/// `progress_every` are at least 1). +pub async fn run( + target: &T, + config: &EpochHeadBackfillConfig, + shutdown: impl Future, +) -> BackfillOutcome { + tokio::pin!(shutdown); + let interval = Duration::from_secs(1) / config.spaces_per_second.max(1); + let progress_every = u64::from(config.progress_every.max(1)); + let started = Instant::now(); + let (mut checked, mut repaired) = (0u64, 0u64); + let mut after: Option = None; + ::tracing::info!( + event = "epoch_head_backfill", + phase = "start", + spaces_per_second = config.spaces_per_second, + "epoch head backfill started" + ); + loop { + let page = match target.page(after.as_ref(), PAGE).await { + Ok(page) => page, + Err(error) => return failed("page", &error, checked, repaired), + }; + if page.is_empty() { + ::tracing::info!( + event = "epoch_head_backfill", + phase = "done", + checked, + repaired, + elapsed_ms = started.elapsed().as_millis() as u64, + "epoch head backfill finished" + ); + return BackfillOutcome { + end: BackfillEnd::Done, + checked, + repaired, + }; + } + for space in page { + let space_started = Instant::now(); + match target.repair(&space).await { + Ok(rewrote) => repaired += u64::from(rewrote), + Err(error) => return failed("repair", &error, checked, repaired), + } + checked += 1; + if checked % progress_every == 0 { + ::tracing::info!( + event = "epoch_head_backfill", + phase = "progress", + checked, + repaired, + elapsed_ms = started.elapsed().as_millis() as u64, + "epoch head backfill progress" + ); + } + after = Some(space); + let pause = interval.max(space_started.elapsed()); + tokio::select! { + _ = tokio::time::sleep(pause) => {} + _ = &mut shutdown => { + ::tracing::info!( + event = "epoch_head_backfill", + phase = "shutdown", + checked, + repaired, + "epoch head backfill stopped for shutdown" + ); + return BackfillOutcome { end: BackfillEnd::Shutdown, checked, repaired }; + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use parking_lot::Mutex; + + /// `spaces` in pages; `repair` takes `repair_cost` of (paused) time, + /// rewrites every other space, and fails on space index `fail_at`. + struct Fake { + spaces: Vec, + repair_cost: Duration, + fail_at: Option, + repaired: Mutex>, + } + + impl Fake { + fn new(count: usize) -> Self { + let mut spaces = (0..count) + .map(|_| { + let jwk = tinycloud_auth::ssi::jwk::JWK::generate_ed25519().unwrap(); + let did = tinycloud_auth::resolver::DID_METHODS + .generate(&jwk, "key") + .unwrap(); + SpaceId::new(did, "default".parse().unwrap()) + }) + .collect::>(); + spaces.sort_by_key(|space| space.to_string()); + Self { + spaces, + repair_cost: Duration::ZERO, + fail_at: None, + repaired: Mutex::new(Vec::new()), + } + } + } + + #[async_trait::async_trait] + impl EpochHeadBackfillTarget for Fake { + async fn page(&self, after: Option<&SpaceId>, limit: u64) -> Result, DbErr> { + let start = match after { + None => 0, + Some(after) => self.spaces.iter().position(|s| s == after).unwrap() + 1, + }; + Ok(self.spaces[start..] + .iter() + .take(limit as usize) + .cloned() + .collect()) + } + + async fn repair(&self, space: &SpaceId) -> Result { + tokio::time::sleep(self.repair_cost).await; + let mut repaired = self.repaired.lock(); + if Some(repaired.len()) == self.fail_at { + return Err(DbErr::Custom("boom".into())); + } + repaired.push(space.clone()); + Ok(repaired.len() % 2 == 1) + } + } + + fn config(spaces_per_second: u32) -> EpochHeadBackfillConfig { + EpochHeadBackfillConfig { + enabled: true, + spaces_per_second, + progress_every: 50, + } + } + + #[tokio::test(start_paused = true)] + async fn checks_every_space_once_across_pages_at_the_configured_rate() { + let fake = Fake::new(250); + let started = Instant::now(); + let outcome = run(&fake, &config(50), std::future::pending()).await; + assert_eq!( + outcome, + BackfillOutcome { + end: BackfillEnd::Done, + checked: 250, + repaired: 125 + } + ); + assert_eq!(*fake.repaired.lock(), fake.spaces); + assert!(started.elapsed() >= Duration::from_secs(5), "250 at 50/s"); + } + + #[tokio::test(start_paused = true)] + async fn pauses_at_least_as_long_as_each_space_took() { + let mut fake = Fake::new(3); + fake.repair_cost = Duration::from_secs(2); + let started = Instant::now(); + run(&fake, &config(100), std::future::pending()).await; + // 3 × (2 s of work + at least 2 s of pause). + assert!( + started.elapsed() >= Duration::from_secs(12), + "{:?}", + started.elapsed() + ); + } + + #[tokio::test(start_paused = true)] + async fn stops_at_the_first_error_without_retrying() { + let mut fake = Fake::new(10); + fake.fail_at = Some(4); + let outcome = run(&fake, &config(100), std::future::pending()).await; + assert_eq!( + outcome, + BackfillOutcome { + end: BackfillEnd::Failed, + checked: 4, + repaired: 2 + } + ); + assert_eq!(fake.repaired.lock().len(), 4); + } + + #[tokio::test(start_paused = true)] + async fn stops_on_shutdown() { + let fake = Fake::new(10); + let outcome = run( + &fake, + &config(1), + tokio::time::sleep(Duration::from_millis(2_500)), + ) + .await; + assert_eq!(outcome.end, BackfillEnd::Shutdown); + assert_eq!(outcome.checked, 3); + } +} diff --git a/tinycloud-node-server/src/lib.rs b/tinycloud-node-server/src/lib.rs index 8a4c40a1..d9f12c4e 100644 --- a/tinycloud-node-server/src/lib.rs +++ b/tinycloud-node-server/src/lib.rs @@ -30,6 +30,7 @@ pub mod authorization; pub mod config; pub mod dstack; pub mod egress; +pub mod epoch_head_backfill; pub mod hooks; pub mod invocation_replay; pub mod link; @@ -424,6 +425,10 @@ pub async fn app(config: &Figment) -> Result> { pub fn resolve_runtime_config(tinycloud_config: &Config) -> Result { let mut resolved = tinycloud_config.clone(); resolved.hooks.validate().map_err(anyhow::Error::msg)?; + resolved + .epoch_head_backfill + .validate() + .map_err(anyhow::Error::msg)?; tracing::RequestIdOrigins::new(&resolved.log.request_id_first_party_origins) .map_err(anyhow::Error::msg)?; resolved.storage.resolve(); @@ -735,6 +740,7 @@ pub async fn app_with_control( // is moved into Rocket-managed state below. let telemetry_enabled = tinycloud_config.telemetry.enabled; let telemetry_tinycloud = tinycloud.clone(); + let backfill_tinycloud = tinycloud.clone(); let rocket = rocket::custom(config) .attach(AdHoc::on_liftoff("webhook-egress-refresh", { @@ -840,6 +846,24 @@ pub async fn app_with_control( rocket }; + // TC-1274: one gentle pass that fills the `epoch_head` cache of spaces + // written before it existed. Writes stay correct without it. + let rocket = if tinycloud_config.epoch_head_backfill.enabled { + let backfill_config = tinycloud_config.epoch_head_backfill.clone(); + rocket.attach(AdHoc::on_liftoff("epoch-head-backfill", move |rocket| { + let tinycloud = backfill_tinycloud.clone(); + let backfill_config = backfill_config.clone(); + let shutdown = rocket.shutdown(); + Box::pin(async move { + tokio::spawn(async move { + epoch_head_backfill::run(&tinycloud, &backfill_config, shutdown).await; + }); + }) + })) + } else { + rocket + }; + // TC-326: pool + read-audit telemetry sampler. Gated on telemetry.enabled; // an on-liftoff task that samples the DB pool gauges, probes pool-acquire // latency, and advances the read-audit commit counters every 5 seconds, diff --git a/tinycloud-node-server/tests/postgres_round_trip_budget.rs b/tinycloud-node-server/tests/postgres_round_trip_budget.rs index 461a0a5d..96b8ea14 100644 --- a/tinycloud-node-server/tests/postgres_round_trip_budget.rs +++ b/tinycloud-node-server/tests/postgres_round_trip_budget.rs @@ -125,11 +125,16 @@ use tokio::{ /// `test_before_acquire(false)`, database identity resolve as one statement, /// the read audit as one statement and no second sequence lock (TC-951): /// -/// - sign-in 75 = peer key 0 + host `/delegate` 17 + session `/delegate` 18 +/// - sign-in 77 = peer key 0 + host `/delegate` 18 + session `/delegate` 19 /// + account-space SQL write 40; -/// - kv put 37; +/// - kv put 38; /// - kv get 22, including the read-audit commit. -const BUDGETS: [(&str, u64); 3] = [("sign-in", 75), ("kv put", 37), ("kv get", 22)]; +/// +/// TC-1274: each epoch-writing commit (both `/delegate`s and the kv put) +/// gained one round trip, the `epoch_head` cache write, which replaces the +/// head anti-join over the space's whole history (measured on main: 75, 37, +/// 22; with the cache: 77, 38, 22). +const BUDGETS: [(&str, u64); 3] = [("sign-in", 77), ("kv put", 38), ("kv get", 22)]; /// Full sign-in plus put plus get rounds run before the measured one. const WARM_UP_ROUNDS: u8 = 3; @@ -960,6 +965,10 @@ database = "{}" [keys] type = "Static" secret = "{}" +# Background traffic that is not a request's (TC-1274): off, so it cannot +# land inside a measured window. +[epoch_head_backfill] +enabled = false "#, datadir.path().join("data").display(), proxied_url(url, proxy, schema)?, From 447ad70584b3e32a3458ea8af10b7b1eec714329 Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sat, 17 Oct 2026 14:30:29 +0000 Subject: [PATCH 2/2] fix(TC-1274): epoch_head review delta: doc reflow, rollback note, disabled pacing - Reflow the epoch_head module doc so no line starts with '+ 1' (clippy doc_lazy_continuation). - CHANGELOG: rolling back to v1.20.x first needs the seaql_migrations rows of m20261017 and m20261018 deleted (sea-orm-migration refuses unknown applied versions); keep the index and the table. - A disabled backfill boots whatever its pacing fields say; an enabled one still needs spaces_per_second and progress_every of at least 1. --- CHANGELOG.md | 4 ++-- tinycloud-core/src/epoch_head.rs | 18 ++++++++--------- tinycloud-node-server/src/config.rs | 31 +++++++++++++++++++++++++++++ 3 files changed, 42 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b36f0b62..aa994077 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,8 +2,8 @@ ## [Unreleased] -- New migration `m20261017_000000_epoch_order_space_index` adds `idx_epoch_order_space_parent_child ON epoch_order (space, parent, child)` for the head-epoch anti-join that every write runs under its space lock (TC-1274). On a fresh database the build is instant. On PostgreSQL, if a valid index of that name and shape already exists, the migration issues no DDL and takes no lock on `epoch_order`. If the index of that name is INVALID or has a different shape, the boot fails, and nothing is applied, with an error naming the fix. On an existing database **without** the index, the migration builds it inside the boot's migration transaction. That holds a SHARE lock that blocks writes for the whole build, and the lock wait is bounded by the 30 s `lock_timeout` (TC-1176). **Runbook (PostgreSQL, before upgrading a node with history):** in autocommit, run `CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_epoch_order_space_parent_child ON epoch_order (space, parent, child);`. Then confirm `SELECT indisvalid FROM pg_index WHERE indexrelid = 'idx_epoch_order_space_parent_child'::regclass;` returns `t`. If it returns `f`, run `DROP INDEX CONCURRENTLY idx_epoch_order_space_parent_child;` and build it again. `down` keeps the index. MySQL is covered only by the generated statement, because the full MySQL migration chain does not run (TC-1314). -- Writes no longer scan a space's whole epoch history to find its head epochs (TC-1274). A new `epoch_head` table (migration `m20261018_000000_epoch_head`) caches each space's heads. Every write maintains it in the same transaction, under the space's existing sequence lock. A write now reads O(heads) rows, normally one, instead of running the `epoch`/`epoch_order` anti-join, and usually costs one more statement, an `UPDATE` of the cached head. The cache is used only when its newest `seq` is the space's newest. Otherwise that space falls back to the old anti-join once and the write repairs the cache, so epochs appended by older releases (a rolling deploy, or a rollback and upgrade) never produce wrong parents. Heads are taken in `id` byte order. Previously the anti-join returned them in no defined order. This affects only a merge after a pre-TC-732 fork, and nothing re-derives an epoch hash from its parents. The migration only creates the empty table, without foreign keys, so it takes no locks on existing tables. After the node starts serving, a gentle online backfill checks each space once. It runs in one short transaction per space, under that space's lock, at no more than `epoch_head_backfill.spaces_per_second` (default 2) and pauses after each space for at least as long as the space took. It stops at its first error with one log line, and logs progress every `progress_every` spaces (default 500). Turn it off with `[epoch_head_backfill] enabled = false` or `TINYCLOUD_EPOCH_HEAD_BACKFILL__ENABLED=false`. Writes stay correct without it. Keep the `idx_epoch_order_space_parent_child` index: the fallback and the backfill use it. +- New migration `m20261017_000000_epoch_order_space_index` adds `idx_epoch_order_space_parent_child ON epoch_order (space, parent, child)` for the head-epoch anti-join that every write runs under its space lock (TC-1274). On a fresh database the build is instant. On PostgreSQL, if a valid index of that name and shape already exists, the migration issues no DDL and takes no lock on `epoch_order`. If the index of that name is INVALID or has a different shape, the boot fails, and nothing is applied, with an error naming the fix. On an existing database **without** the index, the migration builds it inside the boot's migration transaction. That holds a SHARE lock that blocks writes for the whole build, and the lock wait is bounded by the 30 s `lock_timeout` (TC-1176). **Runbook (PostgreSQL, before upgrading a node with history):** in autocommit, run `CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_epoch_order_space_parent_child ON epoch_order (space, parent, child);`. Then confirm `SELECT indisvalid FROM pg_index WHERE indexrelid = 'idx_epoch_order_space_parent_child'::regclass;` returns `t`. If it returns `f`, run `DROP INDEX CONCURRENTLY idx_epoch_order_space_parent_child;` and build it again. `down` keeps the index. MySQL is covered only by the generated statement, because the full MySQL migration chain does not run (TC-1314). Rolling back to v1.20.x: the older node refuses to start while `seaql_migrations` lists this migration (`Migration file of version 'm20261017_000000_epoch_order_space_index' is missing`). Delete that row first, and keep the index: v1.20.x runs the same anti-join, and a later upgrade finds the valid index and does nothing. +- Writes no longer scan a space's whole epoch history to find its head epochs (TC-1274). A new `epoch_head` table (migration `m20261018_000000_epoch_head`) caches each space's heads. Every write maintains it in the same transaction, under the space's existing sequence lock. A write now reads O(heads) rows, normally one, instead of running the `epoch`/`epoch_order` anti-join, and usually costs one more statement, an `UPDATE` of the cached head. The cache is used only when its newest `seq` is the space's newest. Otherwise that space falls back to the old anti-join once and the write repairs the cache, so epochs appended by older releases (a rolling deploy, or a rollback and upgrade) never produce wrong parents. Heads are taken in `id` byte order. Previously the anti-join returned them in no defined order. This affects only a merge after a pre-TC-732 fork, and nothing re-derives an epoch hash from its parents. The migration only creates the empty table, without foreign keys, so it takes no locks on existing tables. After the node starts serving, a gentle online backfill checks each space once. It runs in one short transaction per space, under that space's lock, at no more than `epoch_head_backfill.spaces_per_second` (default 2) and pauses after each space for at least as long as the space took. It stops at its first error with one log line, and logs progress every `progress_every` spaces (default 500). Turn it off with `[epoch_head_backfill] enabled = false` or `TINYCLOUD_EPOCH_HEAD_BACKFILL__ENABLED=false`. Writes stay correct without it. Keep the `idx_epoch_order_space_parent_child` index: the fallback and the backfill use it. Rolling back to v1.20.x: as for `m20261017_000000_epoch_order_space_index`, the older node refuses to start until its rows are gone. Run `DELETE FROM seaql_migrations WHERE version IN ('m20261017_000000_epoch_order_space_index', 'm20261018_000000_epoch_head');` first. Keep the index and the `epoch_head` table: v1.20.x ignores the table. A later upgrade re-applies both migrations as no-ops (the index check and `CREATE TABLE IF NOT EXISTS`), and the `seq` check repairs every space the older node wrote to. - Request events include validated/echoed UUID correlation, typed replacement flags, node-keyed target account/space pseudonyms and per-request database acquisition/execution timings and attempt counts. Driver/actor instrumentation includes failures and request-awaited audit batches; forwarding requires explicit first-party origin classification. A shared core `Pseudonymizer` preserves existing account/space hashes, adds the service/space-scoped database hash primitive for F6, and is derived once at startup for injection into consumers. See `docs/request-telemetry.md` for the wire contract, key custody, restart joinability and coverage/exclusions (TC-959 N2). - CACAO/SIWE delegation ingestion normalizes ECDSA recovery IDs 0/1 to 27/28 before hashing and persisting the canonical DAG-CBOR bytes (TC-1071). Equivalent low-s signatures now share one delegation CID, including the CID used for revocation lookup. High-s signatures are rejected during decoding, matching the existing EIP-191 verifier and EIP-2 low-s convention; UCAN/JWS bytes remain unchanged. Existing stored delegation IDs are not rewritten. - Synchronize the standalone M1, W5 and load-signer Cargo locks with CACAO's `k256` dependency so `--locked` dependency resolution succeeds (TC-1071). Keep the load signer compatible with the SDK's default invocation lifetime. diff --git a/tinycloud-core/src/epoch_head.rs b/tinycloud-core/src/epoch_head.rs index 1a7a09e7..43ccb566 100644 --- a/tinycloud-core/src/epoch_head.rs +++ b/tinycloud-core/src/epoch_head.rs @@ -7,15 +7,15 @@ //! space's sequence lock, to the one epoch it just appended. //! //! The cache is trusted only when it is provably current. Every epoch is -//! appended under the space's sequence lock with `seq = MAX(event_order.seq) -//! + 1` (TC-732), so it outranks every existing epoch, including those -//! appended by releases that never write `epoch_head`. Rows written under the -//! lock are exact at that moment, and the newest epoch is always a head (its -//! child would have a larger `seq`). So the rows are still exact if and only -//! if their largest `seq` is the space's largest. Otherwise the space falls -//! back to the anti-join once and the caller rewrites its rows. Correctness -//! never depends on a backfill having run, or on every writer maintaining the -//! table. +//! appended under the space's sequence lock with +//! `seq = MAX(event_order.seq) + 1` (TC-732), so it outranks every existing +//! epoch, including those appended by releases that never write +//! `epoch_head`. Rows written under the lock are exact at that moment, and +//! the newest epoch is always a head (its child would have a larger `seq`). +//! So the rows are still exact if and only if their largest `seq` is the +//! space's largest. Otherwise the space falls back to the anti-join once and +//! the caller rewrites its rows. Correctness never depends on a backfill +//! having run, or on every writer maintaining the table. //! //! Heads are returned in `id` (byte) order. The anti-join had no order, and //! the parent order only feeds the hash of a new merge epoch; nothing diff --git a/tinycloud-node-server/src/config.rs b/tinycloud-node-server/src/config.rs index 07b88785..464c951b 100644 --- a/tinycloud-node-server/src/config.rs +++ b/tinycloud-node-server/src/config.rs @@ -969,7 +969,12 @@ impl Default for EpochHeadBackfillConfig { } impl EpochHeadBackfillConfig { + /// The pacing fields matter only when the backfill runs, so a disabled + /// backfill accepts any values. pub fn validate(&self) -> Result<(), &'static str> { + if !self.enabled { + return Ok(()); + } if self.spaces_per_second == 0 { return Err("epoch_head_backfill.spaces_per_second must be at least 1; set enabled = false to turn the backfill off"); } @@ -1511,6 +1516,32 @@ mod tests { } } + /// TC-1274: a running backfill needs a positive rate and progress + /// interval; a disabled one boots whatever its pacing says. + #[test] + fn epoch_head_backfill_pacing_is_checked_only_when_enabled() { + for (enabled, spaces_per_second, progress_every, ok) in [ + (true, 1, 1, true), + (true, 0, 500, false), + (true, 2, 0, false), + (false, 0, 0, true), + ] { + let config = Config { + epoch_head_backfill: EpochHeadBackfillConfig { + enabled, + spaces_per_second, + progress_every, + }, + ..Config::default() + }; + assert_eq!( + crate::resolve_runtime_config(&config).is_ok(), + ok, + "enabled {enabled}, {spaces_per_second}/s, every {progress_every}" + ); + } + } + fn enabled_config() -> ShareEmailConfig { ShareEmailConfig { enabled: true,