From 0bc7457534193bc29e190bc883eb5c2269a651d5 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Thu, 7 May 2026 19:48:01 -0700 Subject: [PATCH] fix(memory_tree): gate ingest on source_id so summariser tree never sees a source twice (#1353) --- src/openhuman/memory/tree/ingest.rs | 135 +++++++++++++++++++++++++++- src/openhuman/memory/tree/store.rs | 56 ++++++++++++ 2 files changed, 189 insertions(+), 2 deletions(-) diff --git a/src/openhuman/memory/tree/ingest.rs b/src/openhuman/memory/tree/ingest.rs index 0312223a5..3475526a8 100644 --- a/src/openhuman/memory/tree/ingest.rs +++ b/src/openhuman/memory/tree/ingest.rs @@ -21,6 +21,8 @@ use crate::openhuman::memory::tree::content_store; use crate::openhuman::memory::tree::jobs::{self, ExtractChunkPayload, NewJob}; use crate::openhuman::memory::tree::score::{self, ScoreResult, ScoringConfig}; use crate::openhuman::memory::tree::store; +use crate::openhuman::memory::tree::types::SourceKind; +use crate::openhuman::memory::tree::util::redact::redact; /// Outcome of one ingest call. #[derive(Clone, Debug, Serialize, Deserialize)] @@ -33,6 +35,11 @@ pub struct IngestResult { pub chunks_dropped: usize, /// IDs of all chunks written and queued. pub chunk_ids: Vec, + /// True when this ingest was a no-op because `(source_kind, source_id)` + /// had already been ingested. Memory items are append-only — the + /// summariser tree must not see the same source twice. + #[serde(default)] + pub already_ingested: bool, } impl IngestResult { @@ -42,6 +49,17 @@ impl IngestResult { chunks_written: 0, chunks_dropped: 0, chunk_ids: Vec::new(), + already_ingested: false, + } + } + + fn already_ingested(source_id: &str) -> Self { + Self { + source_id: source_id.to_string(), + chunks_written: 0, + chunks_dropped: 0, + chunk_ids: Vec::new(), + already_ingested: true, } } } @@ -55,6 +73,12 @@ pub async fn ingest_chat( tags: Vec, batch: ChatBatch, ) -> Result { + // No source-level gate for chat: a chat `source_id` (e.g. + // `slack:{connection_id}`) is a stream identifier — many batches / + // buckets accumulate into the same source tree over time. The gate + // would make every bucket after the first a no-op. Chunk-level + // idempotency (`chunk_id` includes content) still prevents true + // replay duplicates from reaching the summariser. let canonical = match chat::canonicalise(source_id, owner, &tags, batch).map_err(anyhow::Error::msg)? { Some(c) => c, @@ -72,6 +96,11 @@ pub async fn ingest_email( tags: Vec, thread: EmailThread, ) -> Result { + // No source-level gate for email: gmail per-participant ingest + // groups many threads under one `source_id` (e.g. + // `gmail:{participants_hash}`) and appends as new threads arrive. + // The gate would block all but the first thread. Chunk-level + // idempotency still protects against true replays. let canonical = match email::canonicalise(source_id, owner, &tags, thread).map_err(anyhow::Error::msg)? { Some(c) => c, @@ -89,6 +118,13 @@ pub async fn ingest_document( tags: Vec, doc: DocumentInput, ) -> Result { + if already_ingested(config, SourceKind::Document, source_id).await? { + log::debug!( + "[memory_tree::ingest] skip ingest_document — source_id_hash={} already ingested", + redact(source_id) + ); + return Ok(IngestResult::already_ingested(source_id)); + } let canonical = match document::canonicalise(source_id, owner, &tags, doc).map_err(anyhow::Error::msg)? { Some(c) => c, @@ -97,11 +133,27 @@ pub async fn ingest_document( persist(config, source_id, canonical).await } +/// Best-effort pre-canonicalisation check. The transactional claim inside +/// [`persist`] is what actually serialises concurrent ingests; this lookup +/// just spares the canonicaliser when we already know the source is a dup. +async fn already_ingested( + config: &Config, + source_kind: SourceKind, + source_id: &str, +) -> Result { + let cfg = config.clone(); + let sid = source_id.to_string(); + tokio::task::spawn_blocking(move || store::is_source_ingested(&cfg, source_kind, &sid)) + .await + .map_err(|e| anyhow::anyhow!("already_ingested join error: {e}"))? +} + async fn persist( config: &Config, source_id: &str, canonical: CanonicalisedSource, ) -> Result { + let source_kind_for_store = canonical.metadata.source_kind; let input = ChunkerInput { source_kind: canonical.metadata.source_kind, source_id: source_id.to_string(), @@ -140,11 +192,45 @@ async fn persist( let config_owned = config.clone(); let staged_for_store = staged.clone(); let results_for_store = all_results.clone(); - let written = tokio::task::spawn_blocking(move || -> Result { + let source_id_for_store = source_id.to_string(); + let written = tokio::task::spawn_blocking(move || -> Result> { use std::collections::{HashMap, HashSet}; store::with_connection(&config_owned, |conn| { let tx = conn.unchecked_transaction()?; + // Authoritative source-level gate (documents only). + // + // For documents, `source_id` identifies a single immutable + // file (one notion page, one drive doc). `is_source_ingested` + // above is a best-effort fast-path; this row insert is what + // actually serialises concurrent ingests of the same + // document and prevents the same content from flowing + // through extract → admit → buffer → seal twice. + // + // Chat and email don't get this gate: their `source_id` + // is a *stream* identifier (e.g. slack workspace, gmail + // participant group) under which many batches / threads + // accumulate over time. The chunk-level idempotency in + // the rest of this transaction is enough to swallow + // genuine replays without blocking legitimate appends. + if source_kind_for_store == SourceKind::Document { + let now_ms = chrono::Utc::now().timestamp_millis(); + let claimed = store::claim_source_ingest_tx( + &tx, + source_kind_for_store, + &source_id_for_store, + now_ms, + )?; + if !claimed { + log::debug!( + "[memory_tree::ingest] persist gate: document already ingested source_id_hash={}", + redact(&source_id_for_store) + ); + // Drop the (empty) transaction implicitly; nothing to commit. + return Ok(None); + } + } + // Read each chunk's CURRENT lifecycle BEFORE the upsert. This // is the "did this chunk exist before this batch" snapshot, // because `upsert_staged_chunks_tx` will either preserve the @@ -205,12 +291,21 @@ async fn persist( let _ = jobs::enqueue_tx(&tx, &extract)?; } tx.commit()?; - Ok(n) + Ok(Some(n)) }) }) .await .map_err(|e| anyhow::anyhow!("persist join error: {e}"))??; + let written = match written { + Some(n) => n, + None => { + // Lost the race against a concurrent ingest of the same source — + // the other writer claimed the row first. No work was committed. + return Ok(IngestResult::already_ingested(source_id)); + } + }; + jobs::wake_workers(); Ok(IngestResult { @@ -218,6 +313,7 @@ async fn persist( chunks_written: written, chunks_dropped: dropped, chunk_ids: staged.iter().map(|s| s.chunk.id.clone()).collect(), + already_ingested: false, }) } @@ -345,6 +441,41 @@ mod tests { assert_eq!(count_scores(&cfg).unwrap(), 0); } + #[tokio::test] + async fn second_ingest_document_with_same_source_id_is_short_circuited() { + let (_tmp, cfg) = test_config(); + let doc = DocumentInput { + provider: "notion".into(), + title: "Launch plan".into(), + body: "Phoenix ships Friday after staging review. alice@example.com owns this.".into(), + modified_at: Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(), + source_ref: Some("notion://page/abc".into()), + }; + let first = ingest_document(&cfg, "notion:abc", "alice", vec![], doc.clone()) + .await + .unwrap(); + assert!(!first.already_ingested); + assert!(first.chunks_written >= 1); + + // Even with completely different content under the same source_id, + // the second ingest must not write anything: documents are + // append-only and the source_id is the dedup key. + let mutated = DocumentInput { + body: "totally different content that should NOT make it into the tree".into(), + ..doc + }; + let second = ingest_document(&cfg, "notion:abc", "alice", vec![], mutated) + .await + .unwrap(); + assert!(second.already_ingested); + assert_eq!(second.chunks_written, 0); + assert!(second.chunk_ids.is_empty()); + + drain_until_idle(&cfg).await.unwrap(); + // Only the first ingest's chunks made it into the store. + assert_eq!(count_chunks(&cfg).unwrap(), first.chunks_written as u64); + } + #[tokio::test] async fn re_ingest_is_idempotent_on_chunks_and_scores() { let (_tmp, cfg) = test_config(); diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs index 9a167b202..94c9ff34d 100644 --- a/src/openhuman/memory/tree/store.rs +++ b/src/openhuman/memory/tree/store.rs @@ -215,6 +215,19 @@ CREATE INDEX IF NOT EXISTS idx_mem_tree_jobs_kind CREATE UNIQUE INDEX IF NOT EXISTS idx_mem_tree_jobs_dedupe_active ON mem_tree_jobs(dedupe_key) WHERE dedupe_key IS NOT NULL AND status IN ('ready', 'running'); + +-- Source-level ingest gate. Memory items (documents, chat batches, email +-- threads) are append-only — once a `(source_kind, source_id)` is ingested +-- it must not be re-ingested, otherwise its chunks flow back through +-- extract → admit → buffer → seal and end up duplicated in the summariser +-- tree. The first ingest claims the row; subsequent ingest_* calls for the +-- same key short-circuit before canonicalisation. +CREATE TABLE IF NOT EXISTS mem_tree_ingested_sources ( + source_kind TEXT NOT NULL, + source_id TEXT NOT NULL, + ingested_at_ms INTEGER NOT NULL, + PRIMARY KEY (source_kind, source_id) +); "; /// Upsert a batch of chunks atomically. @@ -532,6 +545,49 @@ fn set_chunk_lifecycle_status_conn(conn: &Connection, chunk_id: &str, status: &s Ok(()) } +/// Best-effort, non-transactional check used by `ingest_*` to skip +/// canonicalisation when a source has already been ingested. The +/// authoritative gate is [`claim_source_ingest_tx`] inside the persist +/// transaction — this lookup just avoids burning canonicaliser work on +/// the obvious dup case. +pub fn is_source_ingested( + config: &Config, + source_kind: SourceKind, + source_id: &str, +) -> Result { + with_connection(config, |conn| { + let n: i64 = conn.query_row( + "SELECT COUNT(*) FROM mem_tree_ingested_sources \ + WHERE source_kind = ?1 AND source_id = ?2", + params![source_kind.as_str(), source_id], + |r| r.get(0), + )?; + Ok(n > 0) + }) +} + +/// Atomically claim `(source_kind, source_id)` for ingestion. Returns +/// `true` if the row was newly inserted (caller should proceed with the +/// rest of the persist transaction); `false` if a previous ingest already +/// claimed this source (caller must roll back / skip). +/// +/// Lives inside the same transaction as the chunk + job writes so two +/// concurrent ingests of the same source can't both pass the gate. +pub(crate) fn claim_source_ingest_tx( + tx: &Transaction<'_>, + source_kind: SourceKind, + source_id: &str, + now_ms: i64, +) -> Result { + let inserted = tx.execute( + "INSERT OR IGNORE INTO mem_tree_ingested_sources \ + (source_kind, source_id, ingested_at_ms) \ + VALUES (?1, ?2, ?3)", + params![source_kind.as_str(), source_id, now_ms], + )?; + Ok(inserted > 0) +} + fn row_to_chunk(row: &rusqlite::Row<'_>) -> rusqlite::Result { let id: String = row.get(0)?; let source_kind_s: String = row.get(1)?;