diff --git a/crates/tracedecay-global-db/src/schema_stages.rs b/crates/tracedecay-global-db/src/schema_stages.rs index 201fde53b7..9626fb9d3b 100644 --- a/crates/tracedecay-global-db/src/schema_stages.rs +++ b/crates/tracedecay-global-db/src/schema_stages.rs @@ -405,6 +405,7 @@ pub async fn ensure_registered_schema( pub struct RegisteredSchemaConvergence { force_exhaustive: bool, is_fresh: bool, + lcm_status_performance_indexes: bool, } /// Typed schema states an admissible store was classified into, carried from @@ -540,12 +541,10 @@ pub async fn ensure_registered_schema_for_admission( .map_err(|error| { global_db_operation_error("initialize observation projection indexes", error) })?; - // Stores installed before the LCM status indexes existed are already at - // the current LCM schema version, so the in-transaction LCM stage above - // returned without touching them. Each build is one idempotent - // authority-revalidated batch outside the shared schema transaction: a - // real-scale index build over the raw-message store gets the long lease - // instead of holding every other admission stage open. + // Fresh installations publish the final LCM index shape. Keep each build + // independently durable outside the shared schema transaction so a + // real-scale index build gets the long lease without holding every other + // installation stage open. for sql in tracedecay_lcm::schema::LCM_STATUS_PERFORMANCE_INDEX_SQL { installation .execute_authority_revalidated_batch(sql) @@ -558,6 +557,7 @@ pub async fn ensure_registered_schema_for_admission( Ok(RegisteredSchemaConvergence { force_exhaustive, is_fresh, + lcm_status_performance_indexes: false, }) } @@ -753,13 +753,16 @@ async fn install_registered_schema_stages( /// an older shape forward: the historical projection-anchor binding, retrieval /// anchor, repository provenance, projector version migration, and session /// project-path passes were all one-time legacy upgrades and have been removed. -/// Only the authority invariant audit remains, and it stays out of line because -/// it pages real authority rows on a large store. +/// Existing daemon stores also build the LCM status indexes here, after +/// admission, before the authority invariant audit pages historical rows. #[hotpath::measure(future = true, label = "global_db.schema.persist.converge")] pub async fn converge_registered_schema( database: &Database, convergence: RegisteredSchemaConvergence, ) -> tracedecay_domain::errors::Result<()> { + if convergence.lcm_status_performance_indexes { + converge_lcm_status_performance_indexes(database).await?; + } // The invariant pass pages historical authority rows and can legitimately // outlive an ordinary open on a large store. The admission phase has // already installed and validated its guard triggers, so daemon reads and @@ -773,6 +776,22 @@ pub async fn converge_registered_schema( transaction.commit().await } +async fn converge_lcm_status_performance_indexes( + database: &Database, +) -> tracedecay_domain::errors::Result<()> { + // One independently durable batch per index lets an interrupted daemon + // resume without rebuilding indexes that already completed. + for sql in tracedecay_lcm::schema::LCM_STATUS_PERFORMANCE_INDEX_SQL { + database + .execute_authority_revalidated_batch("install LCM status performance index", sql) + .await + .map_err(|error| { + global_db_operation_error("converge LCM status performance indexes", error) + })?; + } + Ok(()) +} + #[hotpath::measure(future = true, label = "global_db.schema.persist.converge")] async fn converge_registered_schema_on( connection: &impl Executor, @@ -786,19 +805,21 @@ async fn converge_registered_schema_on( .await } -/// Synchronously converges an attached existing store's authority invariants. +/// Synchronously converges an attached existing store's historical schema. /// -/// Short-lived attaches have no background maintenance task to run the audit -/// later, so tamper evidence must fail the attach itself: projection-output -/// tamper triggers delete the trusted audit checkpoint (arming an exhaustive -/// re-audit here), and altered guard triggers force the exhaustive pass with -/// its foreign-key sweep, mirroring +/// Short-lived attaches have no background maintenance task, so they build the +/// LCM status indexes and run the authority audit before returning. Tamper +/// evidence must fail the attach itself: projection-output tamper triggers +/// delete the trusted audit checkpoint (arming an exhaustive re-audit here), +/// and altered guard triggers force the exhaustive pass with its foreign-key +/// sweep, mirroring /// [`ensure_registered_schema_for_admission`]. An untampered store resumes /// from its plausible checkpoint and pays only the bounded suffix audit. #[hotpath::measure(future = true, label = "global_db.schema.persist.converge_attached")] pub async fn converge_attached_registered_schema( database: &Database, ) -> tracedecay_domain::errors::Result<()> { + converge_lcm_status_performance_indexes(database).await?; let transaction = database .begin_bulk_write_transaction("converge attached global database authority schema") .await?; @@ -808,6 +829,7 @@ pub async fn converge_attached_registered_schema( RegisteredSchemaConvergence { force_exhaustive, is_fresh: false, + lcm_status_performance_indexes: false, }, ) .await?; @@ -822,9 +844,12 @@ pub async fn converge_attached_registered_schema( /// authority's exact typed reset identity (LCM `ProfileResetRequired`, /// temporal / workflow / configuration / remote-deletion resets) while a /// refused store stays untouched for the operator's explicit reset decision. -/// An admissible store then re-ensures every idempotent schema stage, so an -/// admissibly-fresh store — an existing database file with no schema objects — -/// receives the full install exactly as initialization would have. +/// An admissible store then re-ensures every admission-critical idempotent +/// schema stage. An admissibly-fresh store — an existing database file with no +/// schema objects — receives the same admission-critical install as +/// initialization. The returned convergence plan carries the LCM status-index +/// work for lifecycle-owned daemon maintenance; short-lived callers run that +/// same work synchronously through [`converge_attached_registered_schema`]. #[hotpath::measure(future = true, label = "global_db.schema.persist.attach")] pub async fn ensure_attached_registered_schema( database: &Database, @@ -872,19 +897,11 @@ pub async fn ensure_attached_registered_schema( })?; transaction.commit().await?; } - for sql in tracedecay_lcm::schema::LCM_STATUS_PERFORMANCE_INDEX_SQL { - let transaction = database - .begin_bulk_write_transaction("install LCM status performance index") - .await?; - transaction.execute_batch(sql).await.map_err(|error| { - global_db_operation_error("initialize LCM status performance indexes", error) - })?; - transaction.commit().await?; - } validate_authority_schema_contract(&read_connection).await?; Ok(RegisteredSchemaConvergence { force_exhaustive, is_fresh: configuration_fresh.is_some(), + lcm_status_performance_indexes: true, }) } diff --git a/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs b/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs index 12ce316da2..267543634f 100644 --- a/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs +++ b/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs @@ -73,27 +73,58 @@ async fn current_lcm_schema_reopens_without_republishing_its_marker() { /// A store installed before the LCM status performance indexes existed is /// already at the current LCM schema version, so the in-transaction LCM -/// stage skips it. Admission must still build the indexes in place — and -/// retire the superseded plain payload owner index — without touching the -/// version marker. +/// stage skips it. A short-lived attach must still converge the indexes in +/// place — and retire the superseded plain payload owner index — without +/// touching the version marker or the rows those indexes cover. #[tokio::test] -async fn admission_builds_status_performance_indexes_on_current_stores() { +async fn short_lived_attach_convergence_rebuilds_queryable_lcm_status_indexes() { let tmp = TempDir::new().unwrap(); let db_path = tmp.path().join(".tracedecay").join("sessions.db"); let db = open_global_db(&db_path).await.expect("fresh global db"); drop(db); + set_migration_applied_at(&db_path, 123).await; { // Rewind to the pre-index store shape while keeping the current // version marker, exactly as a store installed by an older binary. let db = TestConnection::open(&db_path); let conn = (*db).clone(); conn.execute_batch( - "DROP INDEX idx_lcm_raw_legacy_truncated; + r#"INSERT INTO sessions(provider, session_id, project_key, project_path) + VALUES ('cursor', 'status-index-session', 'project.status-index', '/status-index'); + INSERT INTO lcm_raw_messages ( + provider, message_id, session_id, role, ordinal, content, + content_hash, storage_kind, snippet_text, index_text, + legacy_truncated, metadata_json + ) VALUES + ( + 'cursor', 'legacy-message', 'status-index-session', 'assistant', 1, + 'legacy body', 'legacy-hash', 'inline', 'legacy', 'legacy', 1, NULL + ), + ( + 'cursor', 'lossy-message', 'status-index-session', 'assistant', 2, + 'lossy body', 'lossy-hash', 'inline', 'lossy', 'lossy', 0, + '{"ingest_protection":{"lossy":true}}' + ); + INSERT INTO lcm_summary_nodes ( + node_id, provider, conversation_id, session_id, depth, + summary_text, summary_hash, summary_token_count, source_token_count + ) VALUES ( + 'summary-node', 'cursor', 'conversation', 'status-index-session', 2, + 'summary body', 'summary-hash', 3, 5 + ); + INSERT INTO lcm_external_payloads ( + payload_ref, provider, session_id, message_id, kind, + content_hash, byte_count, char_count + ) VALUES ( + 'payload-ref', 'cursor', 'status-index-session', 'payload-message', 'text', + 'payload-hash', 11, 7 + ); + DROP INDEX idx_lcm_raw_legacy_truncated; DROP INDEX idx_lcm_raw_lossy_ingest; DROP INDEX idx_lcm_summary_nodes_depth_tokens; DROP INDEX idx_lcm_external_payloads_owner_bytes; CREATE INDEX idx_lcm_external_payloads_owner - ON lcm_external_payloads(provider, session_id);", + ON lcm_external_payloads(provider, session_id);"#, ) .await .expect("rewind to the pre-index schema shape"); @@ -106,7 +137,7 @@ async fn admission_builds_status_performance_indexes_on_current_stores() { for index in ["idx_lcm_raw_legacy_truncated", "idx_lcm_raw_lossy_ingest"] { assert!( raw_indexes.iter().any(|name| name == index), - "admission did not build {index}; raw message indexes: {raw_indexes:?}" + "short-lived convergence did not build {index}; raw message indexes: {raw_indexes:?}" ); } let summary_indexes = table_index_names(&reopened, "lcm_summary_nodes").await; @@ -114,23 +145,61 @@ async fn admission_builds_status_performance_indexes_on_current_stores() { summary_indexes .iter() .any(|name| name == "idx_lcm_summary_nodes_depth_tokens"), - "admission did not build the summary depth/token index: {summary_indexes:?}" + "short-lived convergence did not build the summary depth/token index: {summary_indexes:?}" ); let payload_indexes = table_index_names(&reopened, "lcm_external_payloads").await; assert!( payload_indexes .iter() .any(|name| name == "idx_lcm_external_payloads_owner_bytes"), - "admission did not build the payload owner/bytes index: {payload_indexes:?}" + "short-lived convergence did not build the payload owner/bytes index: {payload_indexes:?}" ); assert!( !payload_indexes .iter() .any(|name| name == "idx_lcm_external_payloads_owner"), - "admission left the superseded payload owner index in place: {payload_indexes:?}" + "short-lived convergence left the superseded payload owner index in place: {payload_indexes:?}" ); + for (index, query) in [ + ( + "idx_lcm_raw_legacy_truncated", + "SELECT COUNT(*) + FROM lcm_raw_messages INDEXED BY idx_lcm_raw_legacy_truncated + WHERE provider = ?1 AND session_id = ?2 AND legacy_truncated != 0", + ), + ( + "idx_lcm_raw_lossy_ingest", + "SELECT COUNT(*) + FROM lcm_raw_messages INDEXED BY idx_lcm_raw_lossy_ingest + WHERE provider = ?1 AND session_id = ?2 + AND metadata_json IS NOT NULL + AND json_valid(metadata_json) + AND json_type(metadata_json, '$.ingest_protection.lossy') = 'true'", + ), + ( + "idx_lcm_summary_nodes_depth_tokens", + "SELECT COUNT(*) + FROM lcm_summary_nodes INDEXED BY idx_lcm_summary_nodes_depth_tokens + WHERE provider = ?1 AND session_id = ?2", + ), + ( + "idx_lcm_external_payloads_owner_bytes", + "SELECT COUNT(*) + FROM lcm_external_payloads INDEXED BY idx_lcm_external_payloads_owner_bytes + WHERE provider = ?1 AND session_id = ?2", + ), + ] { + let mut rows = reopened + .query(query, params!["cursor", "status-index-session"]) + .await + .unwrap_or_else(|error| panic!("rebuilt index {index} must remain queryable: {error}")); + let count: i64 = rows.next().await.unwrap().unwrap().get(0).unwrap(); + assert_eq!(count, 1, "rebuilt index {index} lost its covered row"); + } assert_eq!( schema_version_on(&reopened).await, tracedecay_lcm::LCM_SCHEMA_VERSION ); + drop(reopened); + assert_eq!(migration_applied_at(&db_path).await, 123); } diff --git a/crates/tracedecay-lcm/src/schema.rs b/crates/tracedecay-lcm/src/schema.rs index 8d8a04652f..e0c9005f5e 100644 --- a/crates/tracedecay-lcm/src/schema.rs +++ b/crates/tracedecay-lcm/src/schema.rs @@ -20,10 +20,11 @@ const MIGRATION_NAME: &str = "lcm"; /// `lcm_summary_nodes` / `lcm_external_payloads` records — multi-gigabyte /// body reads on a long-lived profile store for a one-row answer (issue #767 /// measured 10.65 s daemon-side). Each entry is one independently committed -/// idempotent batch: fresh stores install them with the schema, and the -/// registered-schema admission ensures them on already-current stores, where -/// the one-time build cost is a bounded scan per index instead of that same -/// scan on every status call. +/// idempotent batch. Fresh stores install the final index shape with the +/// schema. Already-current daemon stores build missing indexes through +/// lifecycle-owned post-admission convergence, while short-lived attaches +/// converge synchronously. The one-time work is one full-table build per +/// missing index instead of that same scan on every status call. /// /// The partial-index predicates must stay byte-identical to the WHERE terms /// in the status count queries ([`super::query`] status counts): SQLite only diff --git a/crates/tracedecay/src/daemon/store_runtime_tests.rs b/crates/tracedecay/src/daemon/store_runtime_tests.rs index 2f49e1ee9a..c7884b5f5b 100644 --- a/crates/tracedecay/src/daemon/store_runtime_tests.rs +++ b/crates/tracedecay/src/daemon/store_runtime_tests.rs @@ -13,22 +13,16 @@ use tracedecay_domain::{ BrainNodeId, Confidence, FactCategoryV1, FactCurationActionV1, FactLineageEventKindV1, FactOwnerV1, FactRelationKindV1, }; +use tracedecay_graph_db::{ + GraphDbError, GraphGenerationId, GraphGenerationManifest, GraphIdempotencyKey, GraphNamespace, + GraphProjectionId, GraphProjectionIdentity, GraphWatermark, SourceGeneration, +}; +use tracedecay_runtime_core::db::engine::{QueryExecutor, TestConnection}; use tracedecay_runtime_core::db::{DatabaseAccessMode, DatabaseAuthority}; use tracedecay_runtime_core::store_runtime::registry::StoreRuntimeRegistryFailure; use tracedecay_rusqlite_runtime::remote::{ RemoteSpoolKeyV1, RemoteSpoolKeyringV1, RemoteSqliteStorageErrorV1, }; -use tracedecay_code_index_runtime::CodeGraphSeatRuntimePortV1; -use tracedecay_store::{ProjectId, StoreShardIdV1}; -use tracedecay_store_runtime::{ - DaemonSessionRuntimeRegistryV1, RegisteredSchemaConvergenceStatus, process_runtime_generation, - register_registered_schema_installer, registry_open_error, -}; -use tracedecay_graph_db::{ - GraphDbError, GraphGenerationId, GraphGenerationManifest, GraphIdempotencyKey, GraphNamespace, - GraphProjectionId, GraphProjectionIdentity, GraphWatermark, SourceGeneration, -}; -use tracedecay_runtime_core::db::engine::TestConnection; use tracedecay_session_memory::memory::{ MemoryOperationContext, ProjectMemoryCurationMutationTarget, ProjectMemoryCurationOperation, ProjectMemoryFactAddRequest, ProjectMemoryFactAddRequestOutcome, memory_application_for_db, @@ -37,6 +31,11 @@ use tracedecay_store::{ FactReadControl, FactWriteControl, ProjectMemoryFactHistoryQueryV1, ProjectMemoryFactIdV1, ProjectMemoryFactProjectionV1, RetainedGraphStoreLeaseV1, }; +use tracedecay_store::{ProjectId, StoreShardIdV1}; +use tracedecay_store_runtime::{ + DaemonSessionRuntimeRegistryV1, RegisteredSchemaConvergenceStatus, process_runtime_generation, + register_registered_schema_installer, registry_open_error, +}; struct TestRemoteKeyring(Arc); @@ -178,6 +177,81 @@ async fn wait_for_schema_convergence( .expect("registered schema convergence must reach a terminal state") } +const LCM_STATUS_PERFORMANCE_INDEX_NAMES: [&str; 4] = [ + "idx_lcm_raw_legacy_truncated", + "idx_lcm_raw_lossy_ingest", + "idx_lcm_summary_nodes_depth_tokens", + "idx_lcm_external_payloads_owner_bytes", +]; +const SUPERSEDED_LCM_PAYLOAD_OWNER_INDEX: &str = "idx_lcm_external_payloads_owner"; + +async fn installed_lcm_status_index_names( + connection: &(impl QueryExecutor + ?Sized), +) -> Vec { + let mut rows = connection + .query( + "SELECT name + FROM sqlite_master + WHERE type = 'index' + AND name IN ( + 'idx_lcm_raw_legacy_truncated', + 'idx_lcm_raw_lossy_ingest', + 'idx_lcm_summary_nodes_depth_tokens', + 'idx_lcm_external_payloads_owner_bytes', + 'idx_lcm_external_payloads_owner' + ) + ORDER BY name", + (), + ) + .await + .expect("read LCM status index names"); + let mut names = Vec::new(); + while let Some(row) = rows.next().await.expect("read LCM status index name") { + names.push(row.get::(0).expect("decode LCM status index name")); + } + names +} + +async fn deferred_lcm_fixture_row_count(connection: &(impl QueryExecutor + ?Sized)) -> i64 { + let mut rows = connection + .query( + "SELECT COUNT(*) + FROM sessions AS session + JOIN lcm_raw_messages AS message + ON message.provider = session.provider + AND message.session_id = session.session_id + WHERE session.session_id = 'deferred-index-session' + AND message.message_id = 'deferred-index-message'", + (), + ) + .await + .expect("read seeded session and LCM row"); + rows.next() + .await + .expect("read seeded session and LCM count") + .expect("seeded session and LCM count row") + .get::(0) + .expect("decode seeded session and LCM count") +} + +async fn lcm_migration_applied_at(connection: &(impl QueryExecutor + ?Sized)) -> i64 { + let mut rows = connection + .query( + "SELECT applied_at + FROM session_schema_migrations + WHERE name = 'lcm'", + (), + ) + .await + .expect("read LCM migration applied_at"); + rows.next() + .await + .expect("read LCM migration row") + .expect("LCM migration row") + .get::(0) + .expect("decode LCM migration applied_at") +} + fn accepting_memory_write_control() -> FactWriteControl { FactWriteControl::new(Arc::new(|| false), Arc::new(|| true)) } @@ -676,9 +750,34 @@ async fn project_sessions_mount_uses_typed_enrollment_and_is_idempotent() { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn daemon_admission_returns_while_historical_convergence_is_blocked() { - let (_temporary, identity, project_id, project_root, _sessions_path, _database_scope) = +async fn daemon_admission_remains_ready_while_lcm_indexes_converge_in_background() { + let (_temporary, identity, project_id, project_root, sessions_path, _database_scope) = project_sessions_pending_convergence("project.schema-admission").await; + let seed = TestConnection::open(&sessions_path); + seed.execute_batch( + r#"INSERT INTO sessions(provider, session_id, project_key, project_path) + VALUES ('cursor', 'deferred-index-session', 'project.schema-admission', '/deferred'); + INSERT INTO lcm_raw_messages ( + provider, message_id, session_id, role, ordinal, content, + content_hash, storage_kind, snippet_text, index_text, + legacy_truncated, metadata_json + ) VALUES ( + 'cursor', 'deferred-index-message', 'deferred-index-session', 'assistant', 1, + 'deferred body', 'deferred-hash', 'inline', 'deferred', 'deferred', 1, NULL + ); + UPDATE session_schema_migrations + SET applied_at = 123 + WHERE name = 'lcm'; + DROP INDEX idx_lcm_raw_legacy_truncated; + DROP INDEX idx_lcm_raw_lossy_ingest; + DROP INDEX idx_lcm_summary_nodes_depth_tokens; + DROP INDEX idx_lcm_external_payloads_owner_bytes; + CREATE INDEX idx_lcm_external_payloads_owner + ON lcm_external_payloads(provider, session_id);"#, + ) + .await + .expect("seed an already-current store without the LCM status indexes"); + drop(seed); let shard_id = StoreShardIdV1::project_sessions( identity.brain_id().clone(), identity.profile_id().clone(), @@ -693,42 +792,81 @@ async fn daemon_admission_returns_while_historical_convergence_is_blocked() { tokio::pin!(admission); let convergence_blocked = convergence_gate.wait_until_blocked(); tokio::pin!(convergence_blocked); - let database = tokio::select! { - result = &mut admission => { - let database = result.expect("registered project sessions"); - convergence_blocked.await; - database - } - () = &mut convergence_blocked => { - tokio::time::timeout(std::time::Duration::from_secs(1), admission) - .await - .expect("daemon admission must not wait for blocked historical convergence") - .expect("registered project sessions") + let database = tokio::time::timeout(std::time::Duration::from_secs(10), async { + tokio::select! { + result = &mut admission => { + let database = result.expect("registered project sessions"); + convergence_blocked.await; + database + } + () = &mut convergence_blocked => { + admission.await.expect("registered project sessions") + } } - }; + }) + .await + .expect("daemon admission and convergence scheduling must not stall"); assert_eq!( registry.registered_schema_convergence_status(&shard_id), Some(RegisteredSchemaConvergenceStatus::Pending) ); + { + let snapshot = database + .read_snapshot() + .await + .expect("ordinary read snapshot while convergence is pending"); + let indexes = installed_lcm_status_index_names(&snapshot).await; + for index in LCM_STATUS_PERFORMANCE_INDEX_NAMES { + assert!( + !indexes.iter().any(|installed| installed == index), + "daemon admission synchronously built {index}: {indexes:?}" + ); + } + assert!( + indexes + .iter() + .any(|index| index == SUPERSEDED_LCM_PAYLOAD_OWNER_INDEX), + "daemon admission retired the old payload index before convergence: {indexes:?}" + ); + assert_eq!( + deferred_lcm_fixture_row_count(&snapshot).await, + 1, + "ordinary session and LCM reads must remain available while convergence is pending" + ); + } + convergence_gate.release(); + assert_eq!( + wait_for_schema_convergence(®istry, &shard_id).await, + RegisteredSchemaConvergenceStatus::Complete + ); let snapshot = database .read_snapshot() .await - .expect("ordinary read snapshot while convergence is pending"); - let mut rows = snapshot - .query("SELECT COUNT(*) FROM sessions", ()) - .await - .expect("ordinary read while convergence is pending"); + .expect("ordinary read snapshot after convergence"); + let indexes = installed_lcm_status_index_names(&snapshot).await; + for index in LCM_STATUS_PERFORMANCE_INDEX_NAMES { + assert!( + indexes.iter().any(|installed| installed == index), + "background convergence did not build {index}: {indexes:?}" + ); + } + assert!( + !indexes + .iter() + .any(|index| index == SUPERSEDED_LCM_PAYLOAD_OWNER_INDEX), + "background convergence left the superseded payload index in place: {indexes:?}" + ); assert_eq!( - rows.next() - .await - .expect("read session count") - .expect("session count row") - .get::(0) - .expect("decode session count"), - 0 + deferred_lcm_fixture_row_count(&snapshot).await, + 1, + "background convergence must preserve seeded session and LCM rows" + ); + assert_eq!( + lcm_migration_applied_at(&snapshot).await, + 123, + "background convergence must not rewrite the current LCM migration marker" ); - convergence_gate.release(); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)]