From 7508aa4c4bbf650ba8180fa1a55dc92ba5352ba8 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 3 Nov 2026 00:02:22 -0800 Subject: [PATCH 1/3] Reduce tool-history duplication and repair small retention payloads --- crates/tracedecay-lcm/src/compression.rs | 29 +- crates/tracedecay-lcm/src/retention.rs | 251 ++++++++++++- crates/tracedecay-lcm/src/retention/tests.rs | 344 +++++++++++++++++- crates/tracedecay/src/daemon/maintenance.rs | 3 +- .../session_suite/lcm_compression/replay.rs | 69 ++++ 5 files changed, 668 insertions(+), 28 deletions(-) diff --git a/crates/tracedecay-lcm/src/compression.rs b/crates/tracedecay-lcm/src/compression.rs index 41d77e8fda..8debaa39f7 100644 --- a/crates/tracedecay-lcm/src/compression.rs +++ b/crates/tracedecay-lcm/src/compression.rs @@ -2098,7 +2098,8 @@ async fn ingest_active_messages( let mut replay = message.clone(); replay["role"] = Value::String(role.clone()); replay["content"] = original_content.clone(); - let initial_metadata_json = active_message_metadata(message, &replay); + let initial_metadata_json = + active_message_metadata(message, &replay, &storage_text); let expected_content_hash = projected_content_hash(&storage_text); if let Some(existing) = existing_state { let matches_stored_row = existing.ordinal == ordinal @@ -2167,6 +2168,7 @@ async fn ingest_active_messages( let metadata_json = active_replay_metadata_json( upsert.projection_metadata_json.as_deref(), &replay, + &raw, ); if metadata_json != initial_metadata_json { update_active_replay_metadata(conn, provider, &message_id, &metadata_json) @@ -2354,7 +2356,7 @@ fn default_message_kind(role: &str) -> String { } } -fn active_message_metadata(message: &Value, replay: &Value) -> String { +fn active_message_metadata(message: &Value, replay: &Value, storage_text: &str) -> String { let mut metadata = Map::new(); metadata.insert( replay_transactions::ACTIVE_REPLAY_METADATA_KEY.to_string(), @@ -2362,7 +2364,7 @@ fn active_message_metadata(message: &Value, replay: &Value) -> String { ); metadata.insert( replay_transactions::ACTIVE_REPLAY_MESSAGE_KEY.to_string(), - active_replay_for_metadata(replay), + active_replay_for_metadata(replay, Some(storage_text)), ); if let Some(lcm_ingest) = message.get("lcm_ingest") { metadata.insert("lcm_ingest".to_string(), lcm_ingest.clone()); @@ -2384,7 +2386,11 @@ fn replay_content_value( serde_json::from_str(&raw.content).unwrap_or_else(|_| Value::String(raw.content.clone())) } -fn active_replay_metadata_json(existing_metadata_json: Option<&str>, replay: &Value) -> String { +fn active_replay_metadata_json( + existing_metadata_json: Option<&str>, + replay: &Value, + raw: &LcmRawMessage, +) -> String { let mut metadata = existing_metadata_json .and_then(|text| serde_json::from_str::(text).ok()) .and_then(|value| value.as_object().cloned()) @@ -2395,18 +2401,29 @@ fn active_replay_metadata_json(existing_metadata_json: Option<&str>, replay: &Va ); metadata.insert( replay_transactions::ACTIVE_REPLAY_MESSAGE_KEY.to_string(), - active_replay_for_metadata(replay), + active_replay_for_metadata( + replay, + (raw.storage_kind == LcmStorageKind::Inline).then_some(raw.content.as_str()), + ), ); Value::Object(metadata).to_string() } -fn active_replay_for_metadata(replay: &Value) -> Value { +fn active_replay_for_metadata(replay: &Value, canonical_content: Option<&str>) -> Value { let mut replay = replay.clone(); if let Some(object) = replay.as_object_mut() { // This locator is regenerated from the canonical raw row at replay. // Persisting it makes our own replay look like a source metadata edit // and invalidates otherwise unchanged summary lineage on re-ingest. object.remove("store_id"); + // The canonical raw body already owns this exact string. The replay + // reader restores omitted content from it; structured content and + // external payload placeholders retain their separate representation. + if canonical_content + .is_some_and(|content| object.get("content").and_then(Value::as_str) == Some(content)) + { + object.remove("content"); + } replay_transactions::strip_disposable_assistant_replay_sidecars(object, ""); } replay diff --git a/crates/tracedecay-lcm/src/retention.rs b/crates/tracedecay-lcm/src/retention.rs index 3bf7b09734..05a6af3ee9 100644 --- a/crates/tracedecay-lcm/src/retention.rs +++ b/crates/tracedecay-lcm/src/retention.rs @@ -20,18 +20,21 @@ //! //! # Passes //! -//! Both passes reuse the store's content-addressed external payload lifecycle -//! (`lcm_external_payloads` keyed by a content-hash-derived `payload_ref`, -//! with a full reaping GC in [`super::gc`]). In reclaim order: +//! External payloads use the store's owner-scoped lifecycle +//! (`lcm_external_payloads`, with a reaping GC in [`super::gc`]). In order: //! //! 1. **Drop** (terminal, longest window): projection-durable raw rows past //! `drop_after_days` are deleted. Any now unreferenced external payload is //! reaped by the existing payload GC. -//! 2. **Offload** (recoverable, shorter window): projection-durable *inline* -//! raw rows past `offload_after_days` have their bulky `content` -//! externalized to the content-addressed store (deduplicated by hash) and -//! replaced with a recoverable placeholder, reclaiming the inline column and -//! its FTS shadow. +//! 2. **Inline** (lossless): small bodies previously externalized by retention +//! return to their raw rows. The released files follow the existing GC grace +//! period; no disk reclamation is claimed before GC runs. +//! 3. **Offload** (recoverable, shorter window): projection-durable *inline* +//! raw rows past `offload_after_days` have large `content` externalized and +//! replaced with a recoverable placeholder. Bodies below the normal inline +//! threshold stay inline, avoiding a separate uncompressed file and manifest +//! for each small message. Generated search text and its index follow the +//! retained body or placeholder; inlining restores the original search text. //! //! Every pass is bounded (`max_batch_size`) and incremental so the daemon can //! schedule it off the hot path without competing with foreground writes, and a @@ -52,9 +55,10 @@ use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, Row, params}; use tracedecay_runtime_core::db::{ Database, DatabaseEngineReadConnection, DatabaseMemoryTransaction, }; +use tracedecay_store::INLINE_BODY_BYTES; use super::payload::{ExternalPayloadWrite, PayloadFileRollback}; -use super::{LcmError, payload, schema, util}; +use super::{LcmError, payload, raw, schema, util}; const SECONDS_PER_DAY: i64 = 24 * 60 * 60; @@ -91,9 +95,9 @@ pub struct LcmRetentionConfig { /// no-op even in [`RetentionMode::Apply`]. #[serde(default = "default_retention_enabled")] pub enabled: bool, - /// Window after which a projection-durable, still-inline raw row has its - /// content offloaded to the content-addressed store. `None` disables the - /// offload pass. + /// Window after which a projection-durable raw row with a large inline body + /// is offloaded. Small retention-offloaded bodies past this window return + /// inline instead. `None` disables both lossless storage passes. #[serde(default = "default_offload_after_days")] pub offload_after_days: Option, /// Window after which a projection-durable raw row is dropped. `None` @@ -171,7 +175,9 @@ pub struct LcmRetentionPhaseReport { pub eligible: u64, /// Rows actually acted on (`0` in a dry run). pub acted: u64, - /// Bytes of message content reclaimed from the database by this pass. + /// Logical inline-content bytes removed from the database by this pass. + /// This excludes metadata, indexes, canonical bodies and payload files, and + /// does not measure physical disk reclamation. Inlining reports zero. pub bytes_reclaimed: u64, /// Oldest timestamp among the bounded eligible rows, when any. #[serde(default)] @@ -195,6 +201,10 @@ pub struct LcmRetentionReport { pub ended_at: i64, pub dropped: LcmRetentionPhaseReport, pub offloaded: LcmRetentionPhaseReport, + /// Small retention-offloaded bodies restored inline. Released files remain + /// available until the existing payload GC verifies them after its grace. + #[serde(default)] + pub inlined: LcmRetentionPhaseReport, /// Database `PRAGMA freelist_count` before/after (freed pages are the /// measurable, VACUUM-free signal that space was reclaimed). pub freelist_before: u64, @@ -245,7 +255,7 @@ pub async fn read_session_retention_backlog( .unwrap_or(now); let sql = format!( "SELECT MIN(r.timestamp), - COALESCE(SUM(LENGTH(COALESCE(r.content, ''))), 0) + COALESCE(SUM(LENGTH(CAST(COALESCE(r.content, '') AS BLOB))), 0) FROM lcm_raw_messages r WHERE {DURABLE_ROW_IDS} AND r.timestamp IS NOT NULL @@ -254,7 +264,7 @@ pub async fn read_session_retention_backlog( OR (?3 = 1 AND +r.timestamp < ?4 AND r.storage_kind = 'inline' AND r.content IS NOT NULL - AND LENGTH(r.content) > 0) + AND LENGTH(CAST(r.content AS BLOB)) >= {INLINE_BODY_BYTES}) )" ); let mut rows = conn @@ -383,6 +393,7 @@ async fn run_session_retention_inner( ended_at: now, dropped: LcmRetentionPhaseReport::disabled(), offloaded: LcmRetentionPhaseReport::disabled(), + inlined: LcmRetentionPhaseReport::disabled(), freelist_before, freelist_after: freelist_before, page_count_before, @@ -393,6 +404,7 @@ async fn run_session_retention_inner( if !config.enabled || !config.any_window() { report.dropped.window_days = config.drop_after_days; report.offloaded.window_days = config.offload_after_days; + report.inlined.window_days = config.offload_after_days; return Ok(report); } @@ -413,6 +425,18 @@ async fn run_session_retention_inner( ) .await?; report.dropped = dropped; + let (inlined, inline_cursor) = run_inline_offloaded_pass( + store, + storage_root, + scope, + config, + mode, + now, + &mut report.errors, + authorize, + ) + .await?; + report.inlined = inlined; let (offloaded, offload_cursor) = run_offload_pass( store, storage_root, @@ -428,7 +452,11 @@ async fn run_session_retention_inner( if mode.is_apply() { // Consume the staged GC/reporting meta cards: record the last run so a // scheduler and Doctor can report retention backlog without a rescan. - let acted = report.dropped.acted.saturating_add(report.offloaded.acted); + let acted = report + .dropped + .acted + .saturating_add(report.offloaded.acted) + .saturating_add(report.inlined.acted); let transaction = store .begin_memory_write_transaction("begin session retention metadata", authorize) .await?; @@ -436,7 +464,11 @@ async fn run_session_retention_inner( // A scoped run examined only its own rows, so only an unscoped run may // move the cursors past what it examined. if scope.is_unscoped() { - for (pass, cursor) in [("drop", drop_cursor), ("offload", offload_cursor)] { + for (pass, cursor) in [ + ("drop", drop_cursor), + ("offload", offload_cursor), + ("inline_offloaded", inline_cursor), + ] { if let Some(cursor) = cursor { cursor.persist(&transaction, pass).await?; } @@ -863,7 +895,7 @@ async fn run_drop_pass( cursor, cutoff, config.batch_limit(), - "r.store_id, r.timestamp, LENGTH(COALESCE(r.content, ''))", + "r.store_id, r.timestamp, LENGTH(CAST(COALESCE(r.content, '') AS BLOB))", "1 = 1", |row| { Ok(DropRow { @@ -922,6 +954,184 @@ async fn run_drop_pass( Ok((report, (errors.len() == errors_before).then_some(next))) } +struct InlineOffloadedRow { + store_id: i64, + timestamp: i64, + provider: String, + session_id: String, + message_id: String, + payload_ref: String, + content_hash: String, + placeholder_text: Option, + metadata_json: Option, +} + +fn small_offload_predicate() -> String { + format!( + "r.storage_kind = 'external' AND r.content IS NULL + AND EXISTS ( + SELECT 1 FROM lcm_external_payloads e + WHERE e.payload_ref = r.payload_ref + AND e.provider = r.provider AND e.session_id = r.session_id + AND e.message_id = r.message_id AND e.content_hash = r.content_hash + AND e.kind = '{OFFLOAD_KIND}' + AND e.byte_count > 0 AND e.byte_count < {INLINE_BODY_BYTES} + )" + ) +} + +#[allow(clippy::too_many_arguments)] +async fn run_inline_offloaded_pass( + store: RetentionStore<'_>, + storage_root: &Path, + scope: RetentionScope<'_>, + config: &LcmRetentionConfig, + mode: RetentionMode, + now: i64, + errors: &mut Vec, + authorize: Option<&RetentionAuthorization<'_>>, +) -> Result<(LcmRetentionPhaseReport, Option), LcmError> { + let mut report = LcmRetentionPhaseReport { + window_days: config.offload_after_days, + ..LcmRetentionPhaseReport::default() + }; + let Some(window) = config.offload_after_days else { + return Ok((report, None)); + }; + let read = store.read_connection(); + // A distinct cursor reaches small files created before the offload cursor + // advanced. It uses the same bounded age/summary-lineage traversal. + let cursor = RetentionCursor::read(&read, "inline_offloaded").await?; + let (targets, next) = select_retention_candidates( + &read, + scope, + cursor, + cutoff_secs(window, now), + config.batch_limit(), + "r.store_id, r.timestamp, r.provider, r.session_id, r.message_id, + r.payload_ref, r.content_hash, r.placeholder_text, r.metadata_json", + &small_offload_predicate(), + |row| { + Ok(InlineOffloadedRow { + store_id: row.get(0)?, + timestamp: row.get(1)?, + provider: row.get(2)?, + session_id: row.get(3)?, + message_id: row.get(4)?, + payload_ref: row.get(5)?, + content_hash: row.get(6)?, + placeholder_text: row.get(7)?, + metadata_json: row.get(8)?, + }) + }, + ) + .await?; + report.eligible = targets.len() as u64; + report.oldest_eligible_at = targets.iter().map(|target| target.timestamp).min(); + if !mode.is_apply() { + return Ok((report, None)); + } + + let errors_before = errors.len(); + for target in targets { + match inline_offloaded_one(store, storage_root, &target, authorize).await { + Ok(()) => report.acted += 1, + Err(error) => errors.push(format!("inline raw row {}: {error}", target.store_id)), + } + } + // Inlining releases a file reference, not physical bytes. Existing owner + // GC proves it is unreferenced after grace and reports its own reclamation. + Ok((report, (errors.len() == errors_before).then_some(next))) +} + +async fn inline_offloaded_one( + store: RetentionStore<'_>, + storage_root: &Path, + target: &InlineOffloadedRow, + authorize: Option<&RetentionAuthorization<'_>>, +) -> Result<(), LcmError> { + let read = store.read_connection(); + let metadata = payload::load_payload_metadata(&read, &target.payload_ref).await?; + if metadata.provider != target.provider + || metadata.session_id != target.session_id + || metadata.message_id != target.message_id + { + return Err(LcmError::PayloadNotOwnedBySession); + } + let bytes = + usize::try_from(metadata.byte_count).map_err(|_| LcmError::PayloadIntegrityMismatch)?; + let chars = + usize::try_from(metadata.char_count).map_err(|_| LcmError::PayloadIntegrityMismatch)?; + if metadata.kind != OFFLOAD_KIND + || metadata.content_hash != target.content_hash + || bytes == 0 + || bytes >= INLINE_BODY_BYTES + || chars > bytes + { + return Err(LcmError::PayloadIntegrityMismatch); + } + let content = payload::read_verified_payload_content_with_checkpoint( + storage_root, + &target.payload_ref, + &target.content_hash, + bytes, + chars, + &mut || { + if let Some(authorize) = authorize { + authorize("read session retention inline payload")?; + } + Ok(()) + }, + )?; + let txn = store + .begin_memory_write_transaction("begin session retention inline pass", authorize) + .await?; + if let Some(authorize) = authorize { + authorize("compare and swap session retention inline row")?; + } + let update_sql = format!( + "UPDATE lcm_raw_messages AS r + SET content = ?2, storage_kind = 'inline', payload_ref = NULL, placeholder_text = NULL + WHERE r.store_id = ?1 AND r.content_hash = ?3 AND r.payload_ref = ?4 + AND r.provider = ?5 AND r.session_id = ?6 AND r.message_id = ?7 + AND r.timestamp = ?8 AND r.placeholder_text IS ?9 AND r.metadata_json IS ?10 + AND {PROJECTION_DURABLE} AND {}", + small_offload_predicate() + ); + let changed = txn + .execute( + &update_sql, + params![ + target.store_id, + content, + target.content_hash.as_str(), + target.payload_ref.as_str(), + target.provider.as_str(), + target.session_id.as_str(), + target.message_id.as_str(), + target.timestamp, + util::opt_text(target.placeholder_text.as_deref()), + util::opt_text(target.metadata_json.as_deref()) + ], + ) + .await?; + if changed != 1 { + txn.rollback().await?; + return Err(LcmError::Db(format!( + "inline compare-and-swap rejected changed row {}", + target.store_id + ))); + } + // The normal raw read authority verifies the original sanitizer receipt + // against the restored full text. A corrupt receipt never becomes visible + // as inline text, and its external file remains intact on rollback. + if let Err(error) = raw::load_raw_message_by_store_id(&txn, target.store_id).await { + txn.rollback().await?; + return Err(error); + } + commit_authorized(txn, authorize, "commit session retention inline pass").await +} + struct OffloadRow { store_id: i64, provider: String, @@ -959,7 +1169,10 @@ async fn run_offload_pass( cutoff, config.batch_limit(), "r.store_id, r.timestamp, r.provider, r.session_id, r.message_id, r.content", - "r.storage_kind = 'inline' AND r.content IS NOT NULL AND LENGTH(r.content) > 0", + &format!( + "r.storage_kind = 'inline' AND r.content IS NOT NULL + AND LENGTH(CAST(r.content AS BLOB)) >= {INLINE_BODY_BYTES}" + ), |row| { Ok(OffloadRow { store_id: row.get(0)?, diff --git a/crates/tracedecay-lcm/src/retention/tests.rs b/crates/tracedecay-lcm/src/retention/tests.rs index 6978cf853b..27831c3087 100644 --- a/crates/tracedecay-lcm/src/retention/tests.rs +++ b/crates/tracedecay-lcm/src/retention/tests.rs @@ -402,7 +402,8 @@ async fn dry_run_counts_without_mutating() -> Result<(), String> { #[asupersync::test] async fn backlog_read_reports_real_eligible_bytes_and_watermark() -> Result<(), String> { let store = test_store().await?; - let durable = insert_message(&store.conn, 1, 90, "retention backlog bytes").await?; + let content = "retention backlog café ☕"; + let durable = insert_message(&store.conn, 1, 90, content).await?; make_projection_durable(&store.conn, durable).await?; let records = read_session_retention_backlog( @@ -417,7 +418,7 @@ async fn backlog_read_reports_real_eligible_bytes_and_watermark() -> Result<(), assert_eq!(records.len(), 1); assert_eq!(records[0].table.as_str(), "lcm_raw_messages"); - assert!(records[0].past_window_bytes.get() > 0); + assert_eq!(records[0].past_window_bytes.get(), content.len() as u64); assert_eq!( records[0].oldest_past_window_at, tracedecay_domain::UtcMicros((NOW - 90 * DAY) * 1_000_000) @@ -533,6 +534,295 @@ async fn offload_externalizes_durable_content_after_durability() -> Result<(), S Ok(()) } +#[asupersync::test] +async fn offload_keeps_small_bodies_inline_and_uses_byte_threshold() -> Result<(), String> { + let store = test_store().await?; + let small = "é".repeat(INLINE_BODY_BYTES / 2 - 1); + let small_id = insert_message(&store.conn, 1, 90, &small).await?; + make_projection_durable(&store.conn, small_id).await?; + let config = LcmRetentionConfig { + drop_after_days: None, + ..LcmRetentionConfig::default() + }; + let report = run_apply(&store.conn, &store.storage_root, &config).await?; + assert!(report.errors.is_empty(), "{:?}", report.errors); + assert_eq!(report.offloaded.eligible, 0); + assert_eq!(count(&store.conn, "lcm_external_payloads").await?, 0); + assert!(!payload::payload_dir(&store.storage_root).exists()); + let backlog = read_session_retention_backlog( + &store.conn, + tracedecay_contracts::storage::StoreKeyV1::new("sessions.db") + .map_err(|error| error.to_string())?, + &config, + NOW, + ) + .await + .map_err(|error| error.to_string())?; + assert_eq!(backlog[0].past_window_bytes.get(), 0); + + let boundary = "é".repeat(INLINE_BODY_BYTES / 2); + let boundary_id = insert_message(&store.conn, 2, 90, &boundary).await?; + make_projection_durable(&store.conn, boundary_id).await?; + let report = run_apply(&store.conn, &store.storage_root, &config).await?; + assert!(report.errors.is_empty(), "{:?}", report.errors); + assert_eq!(report.offloaded.acted, 1); + assert_eq!(report.offloaded.bytes_reclaimed, boundary.len() as u64); + let unchanged = raw::load_raw_message_by_store_id(&store.conn, small_id) + .await + .map_err(|error| error.to_string())?; + assert_eq!(unchanged.content, small); + Ok(()) +} + +#[asupersync::test] +async fn small_offload_restores_verified_text_and_releases_file_only_after_gc_grace() +-> Result<(), String> { + let store = test_store().await?; + let content = "small tool output: café completed"; + let target = seed_small_offload(&store, 1, content).await?; + let path = payload::payload_dir(&store.storage_root).join(&target.payload_ref); + let config = LcmRetentionConfig { + drop_after_days: None, + ..LcmRetentionConfig::default() + }; + let preview = run_session_retention_authorized( + &store.conn, + &store.storage_root, + "all", + None, + &config, + RetentionMode::DryRun, + NOW, + &|_| Ok(()), + ) + .await + .map_err(|error| error.to_string())?; + assert_eq!(preview.inlined.eligible, 1); + assert_eq!(preview.inlined.acted, 0); + assert_eq!(preview.bytes_reclaimed(), 0); + assert_eq!(count(&store.conn, "lcm_external_payloads").await?, 1); + assert!(path.exists()); + + let report = run_apply(&store.conn, &store.storage_root, &config).await?; + assert!(report.errors.is_empty(), "{:?}", report.errors); + assert_eq!(report.inlined.acted, 1); + assert_eq!(report.offloaded.acted, 0); + assert_eq!(report.bytes_reclaimed(), 0); + assert_eq!(count(&store.conn, "session_canonical_bodies").await?, 0); + let restored = raw::load_raw_message_by_store_id(&store.conn, target.store_id) + .await + .map_err(|error| error.to_string())?; + assert_eq!(restored.content, content); + assert_eq!(restored.content_hash, target.content_hash); + assert_eq!(restored.metadata_json, target.metadata_json); + assert!(restored.payload_ref.is_none()); + assert!(path.exists(), "inlining never deletes the payload file"); + assert_eq!( + fetch_i64( + &store.conn, + "SELECT COUNT(*) FROM lcm_raw_messages_fts + WHERE lcm_raw_messages_fts MATCH 'completed'", + (), + ) + .await?, + 1, + "restoration also makes the original tool text searchable again" + ); + + let marked_at = fetch_i64( + &store.conn, + "SELECT first_seen_at FROM lcm_gc_marks WHERE payload_ref = ?1", + params![target.payload_ref.as_str()], + ) + .await?; + let gc_config = crate::LcmGcConfig::default(); + for (now, retained) in [ + (marked_at + gc_config.grace_seconds as i64 - 1, true), + (marked_at + gc_config.grace_seconds as i64 + 1, false), + ] { + let gc = crate::gc::run_payload_gc_with_apply( + &store.conn, + &store.storage_root, + PROVIDER, + Some(SESSION), + &gc_config, + true, + now, + ) + .await + .map_err(|error| error.to_string())?; + assert!(gc.errors.is_empty(), "{:?}", gc.errors); + assert_eq!(path.exists(), retained); + assert_eq!( + count(&store.conn, "lcm_external_payloads").await?, + i64::from(retained) + ); + } + let restored = raw::load_raw_message_by_store_id(&store.conn, target.store_id) + .await + .map_err(|error| error.to_string())?; + assert_eq!(restored.content, content, "full text survives payload GC"); + Ok(()) +} + +#[asupersync::test] +async fn small_offload_restoration_rejects_corruption_staleness_and_lost_authority() +-> Result<(), String> { + for failure in ["payload", "receipt", "stale", "authority"] { + let store = test_store().await?; + let mut target = seed_small_offload(&store, 1, "original complete text").await?; + let path = payload::payload_dir(&store.storage_root).join(&target.payload_ref); + match failure { + "payload" => { + std::fs::write(&path, "tampered bytes").map_err(|error| error.to_string())? + } + "receipt" => { + store + .conn + .execute( + "UPDATE lcm_raw_messages SET metadata_json = '{}' WHERE store_id = ?1", + params![target.store_id], + ) + .await + .map_err(|error| error.to_string())?; + target.metadata_json = Some("{}".to_owned()); + } + "stale" => { + store + .conn + .execute( + "UPDATE lcm_raw_messages SET timestamp = ?2 WHERE store_id = ?1", + params![target.store_id, NOW], + ) + .await + .map_err(|error| error.to_string())?; + } + _ => {} + } + let error = inline_offloaded_one( + RetentionStore::Connection(&store.conn), + &store.storage_root, + &target, + Some(&|intent| { + if failure == "authority" && intent == "commit session retention inline pass" { + Err(LcmError::Cancelled) + } else { + Ok(()) + } + }), + ) + .await + .expect_err("unproven or changed payload must remain external"); + match failure { + "payload" | "receipt" => assert_eq!(error, LcmError::PayloadIntegrityMismatch), + "stale" => assert!(error.to_string().contains("compare-and-swap rejected")), + "authority" => assert_eq!(error, LcmError::Cancelled), + _ => unreachable!(), + } + assert_eq!( + fetch_i64( + &store.conn, + "SELECT COUNT(*) FROM lcm_raw_messages + WHERE store_id = ?1 AND storage_kind = 'external' + AND content IS NULL AND payload_ref = ?2", + params![target.store_id, target.payload_ref.as_str()], + ) + .await?, + 1, + "{failure} must preserve the original row" + ); + assert_eq!(count(&store.conn, "lcm_external_payloads").await?, 1); + assert!(path.exists(), "{failure} must preserve the payload file"); + } + Ok(()) +} + +#[asupersync::test] +async fn small_offload_restoration_is_bounded_scoped_and_leaves_protected_payloads() +-> Result<(), String> { + let store = test_store().await?; + let first = seed_small_offload(&store, 1, "first old body").await?; + let second = seed_small_offload(&store, 2, "second old body").await?; + let protected = seed_small_offload(&store, 3, "protected body").await?; + store + .conn + .execute( + "UPDATE lcm_external_payloads SET kind = 'quarantined_assistant_output' + WHERE payload_ref = ?1", + params![protected.payload_ref.as_str()], + ) + .await + .map_err(|error| error.to_string())?; + let config = LcmRetentionConfig { + drop_after_days: None, + max_batch_size: 1, + ..LcmRetentionConfig::default() + }; + for disabled in [ + LcmRetentionConfig { + enabled: false, + ..config + }, + LcmRetentionConfig { + offload_after_days: None, + ..config + }, + ] { + let report = run_apply(&store.conn, &store.storage_root, &disabled).await?; + assert!(report.errors.is_empty(), "{:?}", report.errors); + assert_eq!(report.inlined.eligible, 0); + assert_eq!(report.inlined.acted, 0); + } + for (provider, session_id, expected) in [ + ("other-provider", None, 0), + (PROVIDER, Some("other-session"), 0), + (PROVIDER, Some(SESSION), 1), + ("all", None, 1), + ("all", None, 0), + ] { + let report = run_session_retention_authorized( + &store.conn, + &store.storage_root, + provider, + session_id, + &config, + RetentionMode::Apply, + NOW, + &|_| Ok(()), + ) + .await + .map_err(|error| error.to_string())?; + assert!(report.errors.is_empty(), "{:?}", report.errors); + assert_eq!(report.inlined.acted, expected); + assert!(report.inlined.eligible <= 1); + assert_eq!(report.bytes_reclaimed(), 0); + } + for target in [&first, &second] { + assert!( + raw::load_raw_message_by_store_id(&store.conn, target.store_id) + .await + .map_err(|error| error.to_string())? + .payload_ref + .is_none() + ); + } + assert_eq!( + fetch_str( + &store.conn, + "SELECT payload_ref FROM lcm_raw_messages WHERE store_id = ?1", + params![protected.store_id], + ) + .await?, + protected.payload_ref + ); + assert!( + payload::payload_dir(&store.storage_root) + .join(&protected.payload_ref) + .exists() + ); + Ok(()) +} + #[asupersync::test] async fn offload_cas_preserves_revived_row_and_rolls_back_payload() -> Result<(), String> { let store = test_store().await?; @@ -709,3 +999,53 @@ async fn fetch_str( .ok_or_else(|| "no row".to_string())?; row.get::(0).map_err(|e| e.to_string()) } + +/// Seeds the prior retention representation through its actual payload writer. +async fn seed_small_offload( + store: &TestStore, + ordinal: i64, + content: &str, +) -> Result { + let store_id = insert_message(&store.conn, ordinal, 90, content).await?; + make_projection_durable(&store.conn, store_id).await?; + offload_one( + RetentionStore::Connection(&store.conn), + &store.storage_root, + &OffloadRow { + store_id, + provider: PROVIDER.to_owned(), + session_id: SESSION.to_owned(), + message_id: format!("msg-{ordinal}"), + timestamp: NOW - 90 * DAY, + content: content.to_owned(), + }, + None, + ) + .await + .map_err(|error| error.to_string())?; + let mut rows = store + .conn + .query( + "SELECT payload_ref, content_hash, placeholder_text, metadata_json + FROM lcm_raw_messages WHERE store_id = ?1", + params![store_id], + ) + .await + .map_err(|error| error.to_string())?; + let row = rows + .next() + .await + .map_err(|error| error.to_string())? + .ok_or_else(|| "missing offloaded row".to_string())?; + Ok(InlineOffloadedRow { + store_id, + timestamp: NOW - 90 * DAY, + provider: PROVIDER.to_owned(), + session_id: SESSION.to_owned(), + message_id: format!("msg-{ordinal}"), + payload_ref: row.get(0).map_err(|error| error.to_string())?, + content_hash: row.get(1).map_err(|error| error.to_string())?, + placeholder_text: row.get(2).map_err(|error| error.to_string())?, + metadata_json: row.get(3).map_err(|error| error.to_string())?, + }) +} diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 35fdc012d2..a557ec0b24 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -232,12 +232,13 @@ async fn run_registered_store_retention( match &report.session_lcm { Some(Ok(session_lcm)) => { let reclaimed = session_lcm.bytes_reclaimed(); - if reclaimed > 0 || !session_lcm.errors.is_empty() { + if reclaimed > 0 || session_lcm.inlined.acted > 0 || !session_lcm.errors.is_empty() { log_daemon_event( "retention_session_lcm", &[ ("store", "mounted_sessions".to_owned()), ("bytes_reclaimed", reclaimed.to_string()), + ("inlined_rows", session_lcm.inlined.acted.to_string()), ("errors", session_lcm.errors.len().to_string()), ], ); diff --git a/crates/tracedecay/tests/session_suite/lcm_compression/replay.rs b/crates/tracedecay/tests/session_suite/lcm_compression/replay.rs index 72d17d0eff..429fb6c79b 100644 --- a/crates/tracedecay/tests/session_suite/lcm_compression/replay.rs +++ b/crates/tracedecay/tests/session_suite/lcm_compression/replay.rs @@ -214,6 +214,75 @@ async fn threshold_pressure_summarizes_short_huge_active_context() { ); } +#[tokio::test] +async fn active_tool_result_body_is_stored_once_with_lossless_replay() { + let tmp = TempDir::new().unwrap(); + let db = open_lcm_db(&tmp).await; + insert_session(&db, "cursor", "session-tool-body").await; + + let body = format!( + "tool-output-start\n{}tool-output-tail", + "test result: module completed successfully\n".repeat(2_048) + ); + assert!(!tracedecay_lcm::security::should_externalize( + "tool", + Some("tool_result"), + &body, + )); + let mut messages = active_multi_tool_transaction(); + messages[1]["content"] = json!(body); + let initial = + ingest_active_messages(&db, "cursor", "session-tool-body", messages.clone()).await; + let raw = db + .lcm_load_raw_message("cursor", "tool-a") + .await + .expect("tool result should be stored"); + assert_eq!(raw.storage_kind, LcmStorageKind::Inline); + assert_eq!(raw.content, body); + let stored_metadata = raw + .metadata_json + .as_deref() + .expect("active replay metadata"); + let metadata: Value = serde_json::from_str(stored_metadata).unwrap(); + assert!( + metadata["active_replay"].get("content").is_none(), + "the canonical body must not also be stored in SQL metadata" + ); + assert!(stored_metadata.len() < tracedecay_store::INLINE_BODY_BYTES); + assert_eq!(metadata["active_replay"]["tool_call_id"], "call-a"); + assert_eq!(initial.replay_messages[1]["content"], body); + assert_eq!(initial.replay_messages[1]["tool_call_id"], "call-a"); + + let preflight = db + .lcm_preflight(preflight_request( + "cursor", + "session-tool-body", + Vec::new(), + Some(100), + )) + .await + .unwrap(); + assert_eq!(preflight.replay_messages, initial.replay_messages); + + for input in [messages, initial.replay_messages.clone()] { + let replay = ingest_active_messages(&db, "cursor", "session-tool-body", input).await; + assert_eq!(replay.replay_messages, initial.replay_messages); + assert_eq!(replay.replay_token_estimate, initial.replay_token_estimate); + assert_eq!( + db.lcm_load_raw_message("cursor", "tool-a").await.unwrap(), + raw, + "original input and replay must preserve body, identity, and metadata" + ); + assert_eq!( + db.lcm_status("cursor", Some("session-tool-body")) + .await + .unwrap() + .raw_message_count, + 3 + ); + } +} + #[tokio::test] async fn active_structured_content_survives_compress_ingest_and_preflight_replay() { let tmp = TempDir::new().unwrap(); From 233d8087550dfe794ff4d58ada2e26a96f652011 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 3 Nov 2026 00:49:28 -0800 Subject: [PATCH 2/3] build: consume Grafeo fork main without a downstream patch --- Cargo.lock | 10 +- Cargo.toml | 18 +- MODULE.bazel | 7 - .../patches/grafeo-core-read-allocation.patch | 473 ------------------ 4 files changed, 15 insertions(+), 493 deletions(-) delete mode 100644 scripts/bazel/patches/grafeo-core-read-allocation.patch diff --git a/Cargo.lock b/Cargo.lock index 12682e8a5f..f4394e5755 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2885,7 +2885,7 @@ dependencies = [ [[package]] name = "grafeo-adapters" version = "0.5.43" -source = "git+https://github.com/ScriptedAlchemy/grafeo.git?rev=be4713a5c2ee9df7e4639671c1e1244b7680f103#be4713a5c2ee9df7e4639671c1e1244b7680f103" +source = "git+https://github.com/ScriptedAlchemy/grafeo.git?branch=main#5116be3c7b9392006ef4d229634963e1306a7b6b" dependencies = [ "bincode", "grafeo-common", @@ -2900,7 +2900,7 @@ dependencies = [ [[package]] name = "grafeo-common" version = "0.5.43" -source = "git+https://github.com/ScriptedAlchemy/grafeo.git?rev=be4713a5c2ee9df7e4639671c1e1244b7680f103#be4713a5c2ee9df7e4639671c1e1244b7680f103" +source = "git+https://github.com/ScriptedAlchemy/grafeo.git?branch=main#5116be3c7b9392006ef4d229634963e1306a7b6b" dependencies = [ "arcstr", "bincode", @@ -2920,7 +2920,7 @@ dependencies = [ [[package]] name = "grafeo-core" version = "0.5.43" -source = "git+https://github.com/ScriptedAlchemy/grafeo.git?rev=be4713a5c2ee9df7e4639671c1e1244b7680f103#be4713a5c2ee9df7e4639671c1e1244b7680f103" +source = "git+https://github.com/ScriptedAlchemy/grafeo.git?branch=main#5116be3c7b9392006ef4d229634963e1306a7b6b" dependencies = [ "arc-swap", "arcstr", @@ -2943,7 +2943,7 @@ dependencies = [ [[package]] name = "grafeo-engine" version = "0.5.43" -source = "git+https://github.com/ScriptedAlchemy/grafeo.git?rev=be4713a5c2ee9df7e4639671c1e1244b7680f103#be4713a5c2ee9df7e4639671c1e1244b7680f103" +source = "git+https://github.com/ScriptedAlchemy/grafeo.git?branch=main#5116be3c7b9392006ef4d229634963e1306a7b6b" dependencies = [ "arcstr", "bincode", @@ -2965,7 +2965,7 @@ dependencies = [ [[package]] name = "grafeo-storage" version = "0.5.43" -source = "git+https://github.com/ScriptedAlchemy/grafeo.git?rev=be4713a5c2ee9df7e4639671c1e1244b7680f103#be4713a5c2ee9df7e4639671c1e1244b7680f103" +source = "git+https://github.com/ScriptedAlchemy/grafeo.git?branch=main#5116be3c7b9392006ef4d229634963e1306a7b6b" dependencies = [ "bincode", "byteorder", diff --git a/Cargo.toml b/Cargo.toml index f181a8429d..8d448c3b59 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -81,14 +81,16 @@ asupersync = { git = "https://github.com/Dicklesworthstone/asupersync", rev = "e # and in-place, page-verified sealed sections that sealed generation stores # are built and reopened with. Its `memory_usage()` also counts the catalog, # empty property-index shards and plan-cache tables, which the serving graph -# heap charge reads. The fork's workspace resolves sibling crates -# by path, so all five crates must name the same `rev`; mixed revisions build -# type-incompatible instances. -grafeo-adapters = { git = "https://github.com/ScriptedAlchemy/grafeo.git", rev = "be4713a5c2ee9df7e4639671c1e1244b7680f103", default-features = false, features = ["algos"] } -grafeo-common = { git = "https://github.com/ScriptedAlchemy/grafeo.git", rev = "be4713a5c2ee9df7e4639671c1e1244b7680f103", default-features = false } -grafeo-core = { git = "https://github.com/ScriptedAlchemy/grafeo.git", rev = "be4713a5c2ee9df7e4639671c1e1244b7680f103", default-features = false, features = ["lpg"] } -grafeo-engine = { git = "https://github.com/ScriptedAlchemy/grafeo.git", rev = "be4713a5c2ee9df7e4639671c1e1244b7680f103", default-features = false, features = ["lpg", "gql", "algos", "wal", "grafeo-file"] } -grafeo-storage = { git = "https://github.com/ScriptedAlchemy/grafeo.git", rev = "be4713a5c2ee9df7e4639671c1e1244b7680f103", default-features = false, features = ["wal"] } +# heap charge reads, and reports native read allocation demand before +# materializing rows. These APIs live in the fork so every build uses the +# same implementation. The fork resolves sibling crates by path: keep all +# five dependencies on main together and lock their exact commit in Cargo.lock +# to avoid type-incompatible instances from mixed sources. +grafeo-adapters = { git = "https://github.com/ScriptedAlchemy/grafeo.git", branch = "main", default-features = false, features = ["algos"] } +grafeo-common = { git = "https://github.com/ScriptedAlchemy/grafeo.git", branch = "main", default-features = false } +grafeo-core = { git = "https://github.com/ScriptedAlchemy/grafeo.git", branch = "main", default-features = false, features = ["lpg"] } +grafeo-engine = { git = "https://github.com/ScriptedAlchemy/grafeo.git", branch = "main", default-features = false, features = ["lpg", "gql", "algos", "wal", "grafeo-file"] } +grafeo-storage = { git = "https://github.com/ScriptedAlchemy/grafeo.git", branch = "main", default-features = false, features = ["wal"] } # Pinned once: the MCP git handlers and the root's edit/lifecycle readers must # resolve the same object database and status implementation, so a feature # added for one is never missing for the other. diff --git a/MODULE.bazel b/MODULE.bazel index c4f38a74ae..4cb5ca79a8 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -103,13 +103,6 @@ crate.annotation( version = "0.46.0", ) -crate.annotation( - crate = "grafeo-core", - patch_args = ["-p1"], - patches = ["//scripts:bazel/patches/grafeo-core-read-allocation.patch"], - version = "0.5.43", -) - # arborium grammar build scripts read `env!("CARGO_MANIFEST_DIR")`, which bakes # the compiling output base's absolute path into the build-script binary. The # action key does not include that path, so the disk cache served one output diff --git a/scripts/bazel/patches/grafeo-core-read-allocation.patch b/scripts/bazel/patches/grafeo-core-read-allocation.patch deleted file mode 100644 index 1effd1e0ff..0000000000 --- a/scripts/bazel/patches/grafeo-core-read-allocation.patch +++ /dev/null @@ -1,473 +0,0 @@ -diff --git a/src/graph/compact/column.rs b/src/graph/compact/column.rs -index 861a9595..098e900f 100644 ---- a/src/graph/compact/column.rs -+++ b/src/graph/compact/column.rs -@@ -291,6 +291,28 @@ pub enum ColumnCodec { - } - - impl ColumnCodec { -+ /// Reads borrowed codec storage. It never decodes a Value. -+ pub fn allocation_bytes(&self, index: usize) -> usize { -+ use std::mem::size_of; -+ use grafeo_common::memory::heap::arc_slice_bytes; -+ match self { -+ Self::Dict(dict) => dict.get(index).map_or(0, |entry| { -+ // A malformed marker falls back to a literal string. Include -+ // the partial hex vector as well as that fallback allocation. -+ entry.len().saturating_mul(2).saturating_add(2 * size_of::()) -+ }), -+ Self::Int8Vector { dimensions, .. } => { -+ usize::from(*dimensions).saturating_mul(size_of::()) -+ .saturating_add(arc_slice_bytes::(usize::from(*dimensions))) -+ } -+ Self::Float32Vector { dimensions, .. } => { -+ usize::from(*dimensions).saturating_mul(size_of::()) -+ .saturating_add(arc_slice_bytes::(usize::from(*dimensions))) -+ } -+ _ => 0, -+ } -+ } -+ - // ── Phase 3a: Bytes-backed constructors ───────────────────────── - - /// Constructs a [`RawI64`](Self::RawI64) column from a `Vec` -diff --git a/src/graph/compact/graph_store_impl.rs b/src/graph/compact/graph_store_impl.rs -index 341460db..75dfac1b 100644 ---- a/src/graph/compact/graph_store_impl.rs -+++ b/src/graph/compact/graph_store_impl.rs -@@ -19,6 +19,40 @@ use crate::graph::traits::{GraphStore, GraphStoreSearch}; - use crate::statistics::Statistics; - - impl GraphStore for CompactStore { -+ fn node_allocation_bytes(&self, id: NodeId) -> Option { -+ let Some((table, offset)) = self.resolve_node(id) else { return Some(0); }; -+ let table = self.resolve_node_table(table)?; -+ Some(table.allocation_bytes(usize::try_from(offset).ok()?).saturating_add(table.label().len())) -+ } -+ -+ fn edge_allocation_bytes(&self, id: EdgeId) -> Option { -+ let Some((table, offset)) = self.resolve_edge(id) else { return Some(0); }; -+ Some(self.resolve_rel_table(table)?.allocation_bytes(usize::try_from(offset).ok()?)) -+ } -+ -+ fn node_property_allocation_bytes(&self, id: NodeId, key: &PropertyKey) -> Option { -+ let Some((table, offset)) = self.resolve_node(id) else { return Some(0); }; -+ Some(self.resolve_node_table(table)?.property_allocation_bytes(usize::try_from(offset).ok()?, key)) -+ } -+ -+ fn label_allocation_bytes(&self) -> Option { -+ Some(self.table_id_to_label.iter().fold( -+ self.table_id_to_label.len().saturating_mul(std::mem::size_of::()), -+ |bytes, label| bytes.saturating_add(label.len()), -+ )) -+ } -+ -+ fn property_lookup_allocation_bytes(&self, key: &PropertyKey, value: &Value) -> Option { -+ let index = self.property_value_index(key)?; -+ let mut count = 0_usize; -+ for (table, order) in index.iter() { -+ let col = self.resolve_node_table(*table)?.column(key)?; -+ count = count.saturating_add(super::value_order::equality_count(col, order, value)?); -+ } -+ let payload = match value { Value::String(value) => value.len(), Value::Bytes(value) => value.len() * 2, _ => 0 }; -+ Some(count.saturating_mul(8 * std::mem::size_of::()).saturating_add(payload.saturating_mul(3)).saturating_add(128)) -+ } -+ - fn get_node(&self, id: NodeId) -> Option { - let (table_id, offset) = self.resolve_node(id)?; - let nt = self.resolve_node_table(table_id)?; -@@ -29,10 +63,7 @@ impl GraphStore for CompactStore { - - let mut node = Node::new(id); - node.add_label(nt.label()); -- let props = nt.get_all_properties(row); -- for (k, v) in props { -- node.set_property(k, v); -- } -+ nt.for_each_property(row, |key, value| node.set_property(key.clone(), value)); - Some(node) - } - -@@ -48,10 +79,7 @@ impl GraphStore for CompactStore { - let edge_type = rt.edge_type().clone(); - - let mut edge = Edge::new(id, src, dst, edge_type); -- let props = rt.get_all_edge_properties(pos as usize); -- for (k, v) in props { -- edge.set_property(k, v); -- } -+ rt.for_each_property(pos as usize, |key, value| edge.set_property(key.clone(), value)); - Some(edge) - } - -diff --git a/src/graph/compact/layered.rs b/src/graph/compact/layered.rs -index 6eb25cb7..ed189b50 100644 ---- a/src/graph/compact/layered.rs -+++ b/src/graph/compact/layered.rs -@@ -437,6 +437,31 @@ impl LayeredStore { - // ── GraphStore implementation ────────────────────────────────────── - - impl GraphStore for LayeredStore { -+ fn node_allocation_bytes(&self, id: NodeId) -> Option { -+ Some(self.base.load().node_allocation_bytes(id)? -+ .saturating_add(self.overlay.load().node_allocation_bytes(id)?)) -+ } -+ -+ fn edge_allocation_bytes(&self, id: EdgeId) -> Option { -+ Some(self.base.load().edge_allocation_bytes(id)? -+ .saturating_add(self.overlay.load().edge_allocation_bytes(id)?)) -+ } -+ -+ fn node_property_allocation_bytes(&self, id: NodeId, key: &PropertyKey) -> Option { -+ Some(self.base.load().node_property_allocation_bytes(id, key)? -+ .saturating_add(self.overlay.load().node_property_allocation_bytes(id, key)?)) -+ } -+ -+ fn label_allocation_bytes(&self) -> Option { -+ Some(self.base.load().label_allocation_bytes()? -+ .saturating_add(self.overlay.load().label_allocation_bytes()?).saturating_mul(4)) -+ } -+ -+ fn property_lookup_allocation_bytes(&self, key: &PropertyKey, value: &Value) -> Option { -+ Some(self.base.load().property_lookup_allocation_bytes(key, value)? -+ .saturating_add(self.overlay.load().property_lookup_allocation_bytes(key, value)?).saturating_mul(2)) -+ } -+ - fn get_node(&self, id: NodeId) -> Option { - if self.is_node_deleted_from_base(id) { - return None; -diff --git a/src/graph/compact/node_table.rs b/src/graph/compact/node_table.rs -index 95ee80e5..103db9d0 100644 ---- a/src/graph/compact/node_table.rs -+++ b/src/graph/compact/node_table.rs -@@ -135,6 +135,34 @@ impl NodeTable { - /// A `true` row is one the builder padded because the node never had the - /// property; its column body still decodes to a value there, which no - /// read may surface. -+ pub fn allocation_bytes(&self, offset: usize) -> usize { -+ use std::mem::size_of; -+ if offset >= self.len { return 0; } -+ self.columns.iter().fold( -+ self.columns.len().saturating_mul(4 * size_of::<(PropertyKey, Value)>()), -+ |bytes, (key, col)| { -+ if self.is_null(offset, key) { bytes } else { -+ bytes.saturating_add(col.allocation_bytes(offset)).saturating_add(key.as_str().len()) -+ } -+ }, -+ ) -+ } -+ -+ pub fn for_each_property(&self, offset: usize, mut visit: impl FnMut(&PropertyKey, Value)) { -+ for (key, col) in &self.columns { -+ if !self.is_null(offset, key) { -+ if let Some(value) = col.get(offset) { -+ visit(key, value); -+ } -+ } -+ } -+ } -+ -+ pub fn property_allocation_bytes(&self, offset: usize, key: &PropertyKey) -> usize { -+ if self.is_null(offset, key) { return 0; } -+ self.columns.get(key).map_or(0, |col| col.allocation_bytes(offset)) -+ } -+ - #[must_use] - pub fn is_null(&self, offset: usize, key: &PropertyKey) -> bool { - self.null_masks -diff --git a/src/graph/compact/rel_table.rs b/src/graph/compact/rel_table.rs -index f5e5df76..55eb73e3 100644 ---- a/src/graph/compact/rel_table.rs -+++ b/src/graph/compact/rel_table.rs -@@ -182,6 +182,28 @@ impl RelTable { - - /// Returns the property value for a specific edge (by CSR position) and - /// key; `None` when the edge holds no value for it. -+ pub fn allocation_bytes(&self, offset: usize) -> usize { -+ use std::mem::size_of; -+ self.properties.iter().fold( -+ self.properties.len().saturating_mul(4 * size_of::<(PropertyKey, Value)>()), -+ |bytes, (key, col)| { -+ if self.is_null(offset, key) { bytes } else { -+ bytes.saturating_add(col.allocation_bytes(offset)).saturating_add(key.as_str().len()) -+ } -+ }, -+ ) -+ } -+ -+ pub fn for_each_property(&self, offset: usize, mut visit: impl FnMut(&PropertyKey, Value)) { -+ for (key, col) in &self.properties { -+ if !self.is_null(offset, key) { -+ if let Some(value) = col.get(offset) { -+ visit(key, value); -+ } -+ } -+ } -+ } -+ - #[must_use] - pub fn get_edge_property(&self, csr_position: usize, key: &PropertyKey) -> Option { - if self.is_null(csr_position, key) { -diff --git a/src/graph/compact/value_order.rs b/src/graph/compact/value_order.rs -index 230b3a4d..83a3c523 100644 ---- a/src/graph/compact/value_order.rs -+++ b/src/graph/compact/value_order.rs -@@ -137,6 +137,46 @@ pub(crate) fn build_row_order( - Some(RowOrder::Inline(rows)) - } - -+/// The number of rows whose value equals `value`, found by binary search -+/// over `order`, mirroring [`find_eq`] without collecting the rows. -+/// -+/// `None` when the codec and value are not an orderable pair this index -+/// serves, so the caller can decline rather than assume zero. -+pub(crate) fn equality_count(codec: &ColumnCodec, order: &RowOrder, value: &Value) -> Option { -+ let entry; -+ let compare: Box Option + '_> = match (codec, value) { -+ (ColumnCodec::Dict(dict), Value::String(_) | Value::Bytes(_)) => { -+ entry = encode_dict_entry(value); -+ Box::new(|row| { -+ let code = dict.code_at(row)?; -+ Some(dict.entry(code as usize)?.cmp(entry.as_str())) -+ }) -+ } -+ (ColumnCodec::BitPacked(packed), &Value::Int64(target)) => { -+ let target = u64::try_from(target).ok()?; -+ Box::new(move |row| Some(packed.get(row)?.cmp(&target))) -+ } -+ (ColumnCodec::RawI64(store), &Value::Int64(target)) => { -+ Box::new(move |row| Some(store.get(row)?.cmp(&target))) -+ } -+ _ => return None, -+ }; -+ let bound = |inclusive: bool| -> Option { -+ let (mut low, mut high) = (0, order.len()); -+ while low < high { -+ let mid = low + (high - low) / 2; -+ let ordering = order.row(mid).and_then(&compare)?; -+ if ordering == Ordering::Less || (inclusive && ordering == Ordering::Equal) { -+ low = mid + 1; -+ } else { -+ high = mid; -+ } -+ } -+ Some(low) -+ }; -+ Some(bound(true)?.saturating_sub(bound(false)?)) -+} -+ - /// The rows of `codec` whose value equals `value`, in row order, found by - /// binary search over `order`. - pub(crate) fn find_eq(codec: &ColumnCodec, order: &RowOrder, value: &Value) -> Vec { -diff --git a/src/graph/lpg/property.rs b/src/graph/lpg/property.rs -index 415d1498..e22be7fd 100644 ---- a/src/graph/lpg/property.rs -+++ b/src/graph/lpg/property.rs -@@ -302,6 +302,27 @@ impl PropertyStorage { - } - - /// Gets all properties for an entity. -+ #[must_use] -+ /// Borrowed current values, without a map, Value clone, or payload decode. -+ pub fn allocation_bytes(&self, id: Id, key: Option<&PropertyKey>) -> usize { -+ use std::mem::size_of; -+ let columns = self.columns.read(); -+ let mut bytes = 0_usize; -+ for (name, col) in columns.iter() { -+ if key.is_some_and(|key| key != name) { continue; } -+ if let Some(value) = col.borrowed(id) { -+ bytes = bytes.saturating_add(4 * size_of::<(PropertyKey, Value)>()) -+ .saturating_add(name.as_str().len()) -+ .saturating_add(match value { -+ Value::String(value) => value.len(), -+ Value::Bytes(value) => value.len(), -+ _ => 0, -+ }); -+ } -+ } -+ bytes -+ } -+ - #[must_use] - pub fn get_all(&self, id: Id) -> FxHashMap { - let columns = self.columns.read(); -@@ -705,6 +726,16 @@ impl PropertyStorage { - } - } - -+ /// Visits the same visible epoch values as get_all_at. -+ pub fn for_each_at(&self, id: Id, epoch: EpochId, mut visit: impl FnMut(&PropertyKey, Value)) { -+ let columns = self.columns.read(); -+ for (key, col) in columns.iter() { -+ if let Some(value) = col.get_at(id, epoch) { -+ visit(key, value); -+ } -+ } -+ } -+ - /// Removes all PENDING entries from all columns (transaction rollback). - pub fn remove_pending(&self) { - let mut columns = self.columns.write(); -@@ -1020,6 +1051,12 @@ impl PropertyColumn { - /// - /// First checks the hot buffer (uncompressed values), then falls back - /// to the compressed data if present. -+ #[must_use] -+ #[cfg(not(feature = "temporal"))] -+ fn borrowed(&self, id: Id) -> Option<&Value> { -+ self.values.get(&id) -+ } -+ - #[must_use] - pub fn get(&self, id: Id) -> Option { - // First check hot buffer -@@ -1685,6 +1722,12 @@ impl PropertyColumn { - } - - /// Gets the latest value for an entity, filtering out tombstones (Null). -+ #[must_use] -+ #[cfg(feature = "temporal")] -+ fn borrowed(&self, id: Id) -> Option<&Value> { -+ self.values.get(&id).and_then(|log| log.latest()).filter(|value| !value.is_null()) -+ } -+ - #[must_use] - pub fn get(&self, id: Id) -> Option { - self.values -diff --git a/src/graph/lpg/store/edge_ops.rs b/src/graph/lpg/store/edge_ops.rs -index 2aa96b6c..e9a49569 100644 ---- a/src/graph/lpg/store/edge_ops.rs -+++ b/src/graph/lpg/store/edge_ops.rs -@@ -20,7 +20,9 @@ impl LpgStore { - id_to_type.get(record.type_id as usize)?.clone() - }; - let mut edge = Edge::new(id, record.src, record.dst, edge_type); -- edge.properties = self.edge_properties.get_all(id).into_iter().collect(); -+ self.edge_properties.for_each(id, |key, value| { -+ edge.set_property(key.clone(), value); -+ }); - Some(edge) - } - -@@ -76,11 +78,7 @@ impl LpgStore { - id_to_type.get(record.type_id as usize)?.clone() - }; - let mut edge = Edge::new(id, record.src, record.dst, edge_type); -- edge.properties = self -- .edge_properties -- .get_all_at(id, epoch) -- .into_iter() -- .collect(); -+ self.edge_properties.for_each_at(id, epoch, |key, value| edge.set_property(key.clone(), value)); - Some(edge) - } - -diff --git a/src/graph/lpg/store/graph_store_impl.rs b/src/graph/lpg/store/graph_store_impl.rs -index f19a3e9e..8fb6e2e8 100644 ---- a/src/graph/lpg/store/graph_store_impl.rs -+++ b/src/graph/lpg/store/graph_store_impl.rs -@@ -20,6 +20,41 @@ use grafeo_common::utils::hash::FxHashMap; - use std::sync::Arc; - - impl GraphStore for LpgStore { -+ fn node_allocation_bytes(&self, id: NodeId) -> Option { -+ if !self.node_visible(id) { return Some(0); } -+ let labels = self.node_labels.read(); -+ #[cfg(not(feature = "temporal"))] -+ let count = labels.get(&id).map_or(0, |labels| labels.len()); -+ #[cfg(feature = "temporal")] -+ let count = labels.get(&id).and_then(|log| log.latest()).map_or(0, |labels| labels.len()); -+ Some(self.node_properties.allocation_bytes(id, None) -+ .saturating_add(count.saturating_mul(4 * std::mem::size_of::()))) -+ } -+ -+ fn property_lookup_allocation_bytes(&self, key: &PropertyKey, value: &Value) -> Option { -+ let indexes = self.property_indexes.read(); -+ let index = indexes.get(key)?; -+ let value = grafeo_common::types::HashableValue::new(value.clone()); -+ Some(index.get(&value).map_or(0, |nodes| nodes.len()).saturating_mul(std::mem::size_of::())) -+ } -+ -+ fn edge_allocation_bytes(&self, id: EdgeId) -> Option { -+ if self.edge_header(id).is_none() { return Some(0); } -+ Some(self.edge_properties.allocation_bytes(id, None)) -+ } -+ -+ fn node_property_allocation_bytes(&self, id: NodeId, key: &PropertyKey) -> Option { -+ Some(self.node_properties.allocation_bytes(id, Some(key))) -+ } -+ -+ fn label_allocation_bytes(&self) -> Option { -+ let registry = self.label_registry.read(); -+ Some(registry.names().iter().fold( -+ registry.names().len().saturating_mul(std::mem::size_of::()), -+ |bytes, label| bytes.saturating_add(label.len()), -+ )) -+ } -+ - fn get_node(&self, id: NodeId) -> Option { - LpgStore::get_node(self, id) - } -diff --git a/src/graph/lpg/store/node_ops.rs b/src/graph/lpg/store/node_ops.rs -index 3cd1b525..bd3a02d9 100644 ---- a/src/graph/lpg/store/node_ops.rs -+++ b/src/graph/lpg/store/node_ops.rs -@@ -102,7 +102,9 @@ impl LpgStore { - } - } - -- node.properties = self.node_properties.get_all(id).into_iter().collect(); -+ self.node_properties.for_each(id, |key, value| { -+ node.set_property(key.clone(), value); -+ }); - node - } - -@@ -181,11 +183,7 @@ impl LpgStore { - } - } - -- node.properties = self -- .node_properties -- .get_all_at(id, epoch) -- .into_iter() -- .collect(); -+ self.node_properties.for_each_at(id, epoch, |key, value| node.set_property(key.clone(), value)); - node - } - -diff --git a/src/graph/traits.rs b/src/graph/traits.rs -index 4b226ec2..cb2154e6 100644 ---- a/src/graph/traits.rs -+++ b/src/graph/traits.rs -@@ -44,6 +44,30 @@ use std::sync::Arc; - pub trait GraphStore: Send + Sync { - // --- Point lookups --- - -+ /// Allocation demand of materializing a native row, without materializing it. -+ /// None means this store cannot supply allocation metadata. -+ fn node_allocation_bytes(&self, _id: NodeId) -> Option { -+ None -+ } -+ -+ fn edge_allocation_bytes(&self, _id: EdgeId) -> Option { -+ None -+ } -+ -+ /// Allocation demand of one property lookup, including dictionary decoding. -+ fn node_property_allocation_bytes(&self, _id: NodeId, _key: &PropertyKey) -> Option { -+ None -+ } -+ -+ /// Allocation demand of enumerating the label universe. -+ fn label_allocation_bytes(&self) -> Option { -+ None -+ } -+ -+ fn property_lookup_allocation_bytes(&self, _key: &PropertyKey, _value: &Value) -> Option { -+ None -+ } -+ - /// Returns a node by ID (latest visible version at current epoch). - fn get_node(&self, id: NodeId) -> Option; - From 4aebbd32fbfc33f5f092b7aa569025e396cf06a2 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 3 Nov 2026 01:56:25 -0800 Subject: [PATCH 3/3] test(storage): use SQLite integer types in vacuum regressions --- crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index cff2f896f4..59a98dd175 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -2055,7 +2055,7 @@ mod auxiliary_scheduling_tests { connection } - fn vacuum_page_counts(connection: &rusqlite::Connection) -> (u64, u64) { + fn vacuum_page_counts(connection: &rusqlite::Connection) -> (i64, i64) { connection .query_row( "SELECT page_count, freelist_count FROM pragma_page_count, pragma_freelist_count", @@ -2083,7 +2083,7 @@ mod auxiliary_scheduling_tests { assert!(response.try_recv().unwrap().is_ok()); let after = vacuum_page_counts(&connection); - assert_eq!(before.1 - after.1, u64::from(max_pages.max(1))); + assert_eq!(before.1 - after.1, i64::from(max_pages.max(1))); assert!(after.0 < before.0, "reclaimed pages leave the database"); assert!(after.1 > 0, "one call must not drain the entire freelist"); assert_eq!(