Skip to content

fix(projection): avoid TEMP output ownership preflight during rebuild #4802

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

Closed
ScriptedAlchemy wants to merge 7 commits into from
Closed

fix(projection): avoid TEMP output ownership preflight during rebuild #4802

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
0e979f9
fix(projection): validate rebuild outputs without TEMP cache
ScriptedAlchemy Nov 8, 2026
61eefab
test(projection): check durable SQLite integrity on preflight
ScriptedAlchemy Nov 8, 2026
82b126a
style(projection): format output preflight regression
ScriptedAlchemy Nov 8, 2026
c178ff5
test(projection): prove persisted output and TEMP pages survive rollback
ScriptedAlchemy Nov 8, 2026
b14ced8
test(projection): compare TEMP page allocation to preflight baseline
ScriptedAlchemy Nov 8, 2026
4025e77
test(projection): stage normalized stored messages in preflight regre…
ScriptedAlchemy Nov 8, 2026
976e858
fix(projection): import normalized message helper in regression
ScriptedAlchemy Nov 8, 2026
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
259 changes: 201 additions & 58 deletions crates/tracedecay-global-db/src/observation_projection/rebuild.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2328,57 +2328,6 @@ async fn prepare_rebuild_output_activation(
conn: &impl Executor,
generation: &str,
) -> ProjectionStoreResult<()> {
conn.execute_batch(
"DROP TABLE IF EXISTS temp.observation_projection_rebuild_preexisting_outputs;
CREATE TEMP TABLE observation_projection_rebuild_preexisting_outputs (
output_provider TEXT NOT NULL,
output_message_id TEXT NOT NULL,
active_exists INTEGER NOT NULL CHECK(active_exists IN (0, 1)),
current_owned INTEGER NOT NULL CHECK(current_owned IN (0, 1)),
cross_owned INTEGER NOT NULL CHECK(cross_owned IN (0, 1)),
staged_created INTEGER NOT NULL CHECK(staged_created IN (0, 1)),
PRIMARY KEY(output_provider, output_message_id)
) WITHOUT ROWID;",
)
.await
.map_err(|error| storage("prepare staged projection output activation", error))?;
conn.execute(
"INSERT INTO temp.observation_projection_rebuild_preexisting_outputs (
output_provider, output_message_id, active_exists,
current_owned, cross_owned, staged_created
)
SELECT staged.output_provider, staged.output_message_id,
EXISTS (
SELECT 1 FROM lcm_raw_messages AS active
WHERE active.provider = staged.output_provider
AND active.message_id = staged.output_message_id
),
EXISTS (
SELECT 1 FROM observation_projection_provenance AS owner
WHERE owner.projector_version = ?1 AND owner.message_created = 1
AND owner.output_provider = staged.output_provider
AND owner.output_message_id = staged.output_message_id
),
EXISTS (
SELECT 1 FROM observation_projection_provenance AS owner
WHERE owner.projector_version <> ?1
AND owner.output_provider = staged.output_provider
AND owner.output_message_id = staged.output_message_id
),
EXISTS (
SELECT 1 FROM observation_projection_rebuild_provenance AS owner
WHERE owner.projector_version = ?1 AND owner.generation = ?2
AND owner.message_created = 1
AND owner.output_provider = staged.output_provider
AND owner.output_message_id = staged.output_message_id
)
FROM observation_projection_rebuild_messages AS staged
WHERE staged.projector_version = ?1 AND staged.generation = ?2",
params![SESSION_MESSAGE_PROJECTOR_VERSION, generation],
)
.await
.map_err(|error| storage("materialize preexisting projection outputs", error))?;

let message_conflicts = format!(
"{}
OR (active.provider <> 'hermes' AND (
Expand All @@ -2393,26 +2342,46 @@ async fn prepare_rebuild_output_activation(
json_extract_expr(STAGED_MESSAGE_JSON_COLUMN, "text"),
json_extract_expr(STAGED_MESSAGE_JSON_COLUMN, "metadata_json"),
);
// Validate against live indexed ownership while the old projector still owns
// its outputs. A long-lived writer must not materialize connection-local
// TEMP tables: they can keep unlinked SQLite files allocated between rebuilds.
let mut conflicts = conn
.query(
&format!(
"SELECT staged.output_provider, staged.output_message_id
FROM observation_projection_rebuild_messages AS staged
JOIN temp.observation_projection_rebuild_preexisting_outputs AS ownership
ON ownership.output_provider = staged.output_provider
AND ownership.output_message_id = staged.output_message_id
LEFT JOIN lcm_raw_messages AS active
ON active.provider = staged.output_provider
AND active.message_id = staged.output_message_id
WHERE staged.projector_version = ?1 AND staged.generation = ?2
AND (
(ownership.active_exists = 0 AND ownership.staged_created = 0)
(active.message_id IS NULL
AND NOT EXISTS (
SELECT 1 FROM observation_projection_rebuild_provenance AS staged_owner
WHERE staged_owner.projector_version = ?1
AND staged_owner.generation = ?2
AND staged_owner.message_created = 1
AND staged_owner.output_provider = staged.output_provider
AND staged_owner.output_message_id = staged.output_message_id
))
OR (
ownership.active_exists = 1
AND NOT (ownership.current_owned = 1 AND ownership.cross_owned = 0)
active.message_id IS NOT NULL
AND (
{message_conflicts}
NOT EXISTS (
SELECT 1 FROM observation_projection_provenance AS current_owner
WHERE current_owner.projector_version = ?1
AND current_owner.message_created = 1
AND current_owner.output_provider = staged.output_provider
AND current_owner.output_message_id = staged.output_message_id
)
OR EXISTS (
SELECT 1 FROM observation_projection_provenance AS foreign_owner
WHERE foreign_owner.projector_version <> ?1
AND foreign_owner.output_provider = staged.output_provider
AND foreign_owner.output_message_id = staged.output_message_id
)
)
AND ({message_conflicts})
)
)
LIMIT 1"
Expand Down Expand Up @@ -2609,6 +2578,7 @@ mod activation_tests {
use super::{
REBUILD_PAGE_SIZE, SESSION_MESSAGE_PROJECTOR_VERSION,
activate_projection_rebuild_transaction, activate_rebuild_sessions,
prepare_rebuild_output_activation, projected_stored_message,
reconcile_overlapping_rebuild_sessions, write_staged_message,
};
use crate::tests::harness::RegisteredGlobalDbHarness;
Expand Down Expand Up @@ -2796,6 +2766,179 @@ mod activation_tests {
);
}

#[asupersync::test]
async fn output_preflight_rejects_unowned_changes_without_retaining_temp_storage() {
const GENERATION: &str = "generation.output-preflight";
let harness = RegisteredGlobalDbHarness::open("output-preflight-no-temp").await;
let active = session(None, None);
assert!(harness.registered.upsert_session(&active).await);

let foreign = SessionMessageRecord {
provider: active.provider.clone(),
session_id: active.session_id.clone(),
message_id: "message.unowned".to_owned(),
role: "assistant".to_owned(),
timestamp: Some(1),
ordinal: 1,
text: "foreign durable body".to_owned(),
kind: None,
model: None,
tool_names: None,
source_path: None,
source_offset: None,
metadata_json: None,
};
let transaction = harness.registered.begin_write_transaction().await.unwrap();
tracedecay_lcm::raw::upsert_projection_raw_message(
&transaction,
&foreign,
Some("/foreign"),
)
.await
.unwrap();
transaction.commit().await.unwrap();

// The foreign message must exist in the committed store before a
// separate rebuild transaction attempts to replace it.
let transaction = harness.registered.begin_write_transaction().await.unwrap();
stage(&transaction, GENERATION, &active).await;
let changed = SessionMessageRecord {
text: "overwritten body".to_owned(),
..foreign.clone()
};
let changed_stored = projected_stored_message(&changed).unwrap();
write_staged_message(&transaction, GENERATION, &changed_stored)
.await
.unwrap();
// Schema initialization and other writer operations can already
// allocate TEMP pages. The preflight must not grow that allocation,
// regardless of unrelated TEMP storage already held by this connection.
let mut baseline_pages = transaction
.query("PRAGMA temp.page_count", ())
.await
.unwrap();
let baseline_page_count = baseline_pages
.next()
.await
.unwrap()
.unwrap()
.get::<i64>(0)
.unwrap();
drop(baseline_pages);

let error = prepare_rebuild_output_activation(&transaction, GENERATION)
.await
.unwrap_err();
assert!(
matches!(error, ProjectionStoreError::OutputCollision { provider, message_id }
if provider == "cursor" && message_id == "message.unowned"),
"rebuild must refuse a foreign output with different content"
);

// Matching foreign bodies are compatible; rejecting every unowned row
// would also avoid spills, but would break legitimate rebuilds.
let compatible_stored = projected_stored_message(&foreign).unwrap();
write_staged_message(&transaction, GENERATION, &compatible_stored)
.await
.unwrap();
prepare_rebuild_output_activation(&transaction, GENERATION)
.await
.expect("unchanged foreign output must remain eligible");

let missing = SessionMessageRecord {
message_id: "message.not-created".to_owned(),
..foreign.clone()
};
let missing_stored = projected_stored_message(&missing).unwrap();
write_staged_message(&transaction, GENERATION, &missing_stored)
.await
.unwrap();
let error = prepare_rebuild_output_activation(&transaction, GENERATION)
.await
.unwrap_err();
assert!(
matches!(error, ProjectionStoreError::OutputCollision { provider, message_id }
if provider == "cursor" && message_id == "message.not-created"),
"an absent output without staged creation evidence must be rejected"
);

let mut rows = transaction
.query(
"SELECT name FROM sqlite_temp_master
WHERE name = 'observation_projection_rebuild_preexisting_outputs'",
(),
)
.await
.unwrap();
assert!(
rows.next().await.unwrap().is_none(),
"repeated preflight checks must not retain a connection-local TEMP table"
);
drop(rows);

// A dropped TEMP b-tree can disappear from sqlite_temp_master while
// its pages remain allocated on this writer. Compare page counts on
// this same long-lived connection before and after repeated preflight
// calls; requiring an absolute zero would conflate earlier storage.
let mut pages = transaction
.query("PRAGMA temp.page_count", ())
.await
.unwrap();
assert_eq!(
pages.next().await.unwrap().unwrap().get::<i64>(0).unwrap(),
baseline_page_count,
"repeated preflight checks must not grow SQLite TEMP database pages"
);
drop(pages);

transaction.rollback().await.unwrap();
let snapshot = harness.registered.read_snapshot().await.unwrap();
let mut rows = snapshot
.query(
"SELECT content, location_path FROM lcm_raw_messages
WHERE provider = ?1 AND message_id = ?2",
params![foreign.provider.as_str(), foreign.message_id.as_str()],
)
.await
.unwrap();
let row = rows.next().await.unwrap().unwrap();
assert_eq!(
row.get::<String>(0).unwrap(),
foreign.text,
"a denied rebuild must not replace a committed foreign body"
);
assert_eq!(row.get::<String>(1).unwrap(), "/foreign");
drop(rows);

let mut rows = snapshot
.query(
"SELECT COUNT(*) FROM observation_projection_rebuilds
WHERE generation = ?1",
params![GENERATION],
)
.await
.unwrap();
assert_eq!(
rows.next().await.unwrap().unwrap().get::<i64>(0).unwrap(),
0,
"a failed rebuild must leave no durable staging generation"
);
drop(rows);

let mut integrity = snapshot.query("PRAGMA integrity_check", ()).await.unwrap();
assert_eq!(
integrity
.next()
.await
.unwrap()
.unwrap()
.get::<String>(0)
.unwrap(),
"ok",
"the committed SQLite store must remain structurally sound after rollback"
);
}

#[asupersync::test]
async fn activation_names_the_session_field_instead_of_a_message_collision() {
let harness = RegisteredGlobalDbHarness::open("session-collision-field").await;
Expand Down
Loading