Skip to content

fix(global-db): defer LCM index convergence after daemon admission #769

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
BleedingDev wants to merge 1 commit into
base: codex/tracedecay-total-redesign-plan-reopened
Choose a base branch
from
Open

fix(global-db): defer LCM index convergence after daemon admission #769

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 43 additions & 26 deletions crates/tracedecay-global-db/src/schema_stages.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -558,6 +557,7 @@ pub async fn ensure_registered_schema_for_admission(
Ok(RegisteredSchemaConvergence {
force_exhaustive,
is_fresh,
lcm_status_performance_indexes: false,
})
}

Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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?;
Expand All @@ -808,6 +829,7 @@ pub async fn converge_attached_registered_schema(
RegisteredSchemaConvergence {
force_exhaustive,
is_fresh: false,
lcm_status_performance_indexes: false,
},
)
.await?;
Expand All @@ -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,
Expand Down Expand Up @@ -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,
})
}

Expand Down
89 changes: 79 additions & 10 deletions crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -106,31 +137,69 @@ 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;
assert!(
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);
}
9 changes: 5 additions & 4 deletions crates/tracedecay-lcm/src/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading