From c87a01cc14b1696746bb4ee0627c81d327f5534d Mon Sep 17 00:00:00 2001 From: Shanu Date: Wed, 9 Sep 2026 13:28:40 +0530 Subject: [PATCH 1/2] Update a stale-key row through its requested document id `memory_docs` upserts on `(namespace, key)` but its primary key is `document_id`, so a write whose key was new while a row of the same namespace already held its requested id failed with `UNIQUE constraint failed: memory_docs.document_id`. Sync providers keyed documents by title before openhuman#4953 while already passing their stable `{toolkit}:{id}`; re-syncing such an item under the id key hit its own old row, and the GitHub pipeline, which does not tolerate scope errors, aborted every run that reached it (openhuman#6147). Resolve the row before writing, in both the full and the metadata-only path: a row found by key keeps its id whatever was requested (the DO UPDATE never rewrote the id, so chunks and the graph job went under an id no row had); a requested id that names a same-namespace row under another key re-keys that row inside the write transaction and updates it; an id owned by another namespace falls back to the derived id instead of failing the write; a blank id is no request. --- .../tinymemory-core/src/store/client_tests.rs | 55 ++++ .../src/store/namespace_store/README.md | 23 ++ .../src/store/namespace_store/documents.rs | 218 ++++++++++--- .../documents_identity_tests.rs | 296 ++++++++++++++++++ 4 files changed, 541 insertions(+), 51 deletions(-) create mode 100644 crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs diff --git a/crates/tinymemory-core/src/store/client_tests.rs b/crates/tinymemory-core/src/store/client_tests.rs index e41a994..81daf32 100644 --- a/crates/tinymemory-core/src/store/client_tests.rs +++ b/crates/tinymemory-core/src/store/client_tests.rs @@ -181,6 +181,61 @@ async fn store_skill_sync_with_secret_like_title_uses_stable_document_id_as_key( ); } +#[tokio::test] +async fn store_skill_sync_updates_a_row_written_before_the_stable_key_rule() { + // Regression for openhuman#6147. Before openhuman#4953 a Composio + // provider's document was keyed by its TITLE while already carrying the + // stable `{toolkit}:{id}` as its document id; since then the key is that + // id. A pre-#4953 row whose item is updated later is re-fetched and + // written under the new key: a new `(namespace, key)` whose requested id + // the old row still holds. The insert used to fail with + // `upsert memory_docs: UNIQUE constraint failed: memory_docs.document_id`; + // the GitHub pipeline does not tolerate scope errors, so the whole sync + // run aborted and, its cursor never advancing, aborted again every tick. + let (_tmp, client) = make_client(); + let stable_id = "github:4892120323"; + let title = "feat(rewards): surface the Rewards page"; + + // The pre-#4953 write: title as key, stable id as document id. + let mut legacy = doc("skill-github", title, "issue body v1"); + legacy.document_id = Some(stable_id.to_string()); + legacy.taint = crate::MemoryTaint::ExternalSync; + assert_eq!(client.put_doc(legacy).await.unwrap(), stable_id); + + client + .store_skill_sync( + "github", + "conn-1", + title, + "issue body v2", + Some("tinycortex-sync".into()), + None, + Some("medium".into()), + None, + None, + Some(stable_id.into()), + ) + .await + .expect("a re-sync of an item stored under its title must update it in place"); + + let docs = client.list_documents(Some("skill-github")).await.unwrap(); + let arr = docs + .get("documents") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(); + assert_eq!( + arr.len(), + 1, + "the re-sync must update the legacy row, not duplicate it" + ); + assert_eq!(arr[0]["documentId"], stable_id); + assert_eq!( + arr[0]["key"], stable_id, + "the row is re-keyed to the stable id" + ); +} + #[tokio::test] async fn clear_skill_memory_targets_prefixed_namespace() { let (_tmp, client) = make_client(); diff --git a/crates/tinymemory-core/src/store/namespace_store/README.md b/crates/tinymemory-core/src/store/namespace_store/README.md index d64f572..037f5e2 100644 --- a/crates/tinymemory-core/src/store/namespace_store/README.md +++ b/crates/tinymemory-core/src/store/namespace_store/README.md @@ -40,6 +40,29 @@ caused #5164. `safety::canonical_identifier` (namespace, KV key) and which is what flooded Sentry (3,055 events / 1 user / 1 day). The rejections that remain deliberate (secret-shaped identifiers, empty keys) are demoted out of the error stream by `ExpectedErrorKind::MemoryIdentifierRejected`. +### Document identity (`document_id` vs `(namespace, key)`) + +`memory_docs` is keyed twice: `document_id` is the primary key, `(namespace, +key)` is the upsert's conflict target. A writer that supplies its own +`document_id` (sync providers pass `{toolkit}:{id}`) can therefore address a +row two ways, and `documents.rs` resolves both before writing +(`resolve_document_identity`, used by the full and the metadata-only path): + +- a row with the `(namespace, key)` exists → its id is used, whatever was + requested. `DO UPDATE` never rewrites `document_id`, so chunks and the graph + job must follow the row's id; +- no row has the key, but the requested id names a row of the **same** + namespace → the same document under a stale key (providers keyed by title + before openhuman#4953 while already passing their stable id). The row is + re-keyed inside the write transaction and updated. Before this, the insert + failed with `UNIQUE constraint failed: memory_docs.document_id` and a + provider that does not tolerate scope errors aborted every sync run that + reached the item (openhuman#6147); +- the requested id belongs to a row in **another** namespace → a derived id + is used and a warning logged. Ids are addressed per namespace everywhere + else (`delete_document`, chunk and graph lookups), so a foreign row is not + this document and never blocks the write. + - **`graph.rs`** — `graph_namespace` / `graph_global` upserts with attribute merging and evidence accumulation, plus namespace / global / cross-namespace queries and document-scoped relation removal. - **`query.rs`** — hybrid retrieval. Combines graph relevance, vector similarity, keyword overlap, episodic signal and freshness; exposes `query_namespace_*` (with query) and `recall_namespace_*` (query-less) entry points used by `MemoryClient`. - **`helpers.rs`** — shared utilities: f32-vector byte codecs, cosine similarity, markdown chunking, text/graph normalisation, JSON attribute merging, recency scoring. diff --git a/crates/tinymemory-core/src/store/namespace_store/documents.rs b/crates/tinymemory-core/src/store/namespace_store/documents.rs index 80edaed..e9dc7ff 100644 --- a/crates/tinymemory-core/src/store/namespace_store/documents.rs +++ b/crates/tinymemory-core/src/store/namespace_store/documents.rs @@ -30,6 +30,40 @@ const DOCUMENT_CHUNK_MAX_TOKENS: usize = 225; /// fewer round-trips than the one-per-document write path it replaces. pub(crate) const EMBED_REQUEST_MAX_TEXTS: usize = 64; +/// The row a document write addresses, resolved before anything is written. +/// +/// Two identities can name a row: the `(namespace, key)` dedup key every +/// write carries, and the document id — the table's primary key — when the +/// caller supplies one. They can disagree, and `memory_docs` only upserts on +/// the first: a write whose key is new but whose requested id an existing row +/// holds used to fail on the primary key (openhuman#6147). Both write paths +/// resolve the row up front so the row they update, the chunks they replace +/// and the id they hand back all agree. +enum DocumentIdentity { + /// A row with this `(namespace, key)` exists. Its id wins over any + /// requested one: the upsert's `DO UPDATE` never rewrites `document_id`, + /// so chunks written under a different id would be orphaned and the graph + /// job queued under an id no row has. + Existing { + document_id: String, + created_at: f64, + }, + /// No row has this key, but the requested id names a row of the same + /// namespace filed under another key: the same document, written when a + /// different key rule applied (sync providers keyed by title before + /// openhuman#4953 while already passing their stable id). The write + /// re-keys that row and updates it. + StaleKey { + document_id: String, + created_at: f64, + }, + /// A new row. `document_id` is the requested id when it is free; `None` + /// when the write must mint one — nothing usable was requested, or the + /// requested id belongs to a row in another namespace, which is not this + /// document and must not block it. + New { document_id: Option }, +} + impl UnifiedMemory { /// Insert or update a document by `(namespace, key)`. Writes the markdown /// sidecar, replaces vector chunks, and embeds them with the configured @@ -185,19 +219,20 @@ impl UnifiedMemory { let _write_guard = Self::document_write_lock(&self.db_path, &namespace, &key) .lock_owned() .await; - let existing_document_id = { + let now = Self::now_ts(); + let identity = { let conn = self.conn.lock(); - conn.query_row( - "SELECT document_id FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![namespace, key], - |row| row.get::<_, String>(0), - ) - .optional() - .map_err(|e| format!("lookup existing document_id: {e}"))? + Self::resolve_document_identity(&conn, &namespace, &key, input.document_id.as_deref())? }; - let document_id = input - .document_id - .or(existing_document_id) + let (document_id, created_at, rekey) = match identity { + DocumentIdentity::Existing { + document_id, + created_at, + } => (document_id, created_at, false), + DocumentIdentity::StaleKey { + document_id, + created_at, + } => (document_id, created_at, true), // Derived from (namespace, key), NOT random. The lookup above and // the write below are separated by `.await`s, so two concurrent // stores of a not-yet-existing key both miss and both mint an id. @@ -210,18 +245,11 @@ impl UnifiedMemory { // deleted. A deterministic id makes both writers choose the same // one, so the second write updates the first's chunks instead of // orphaning them. - .unwrap_or_else(|| Self::derive_document_id(&namespace, &key)); - let now = Self::now_ts(); - let created_at = { - let conn = self.conn.lock(); - conn.query_row( - "SELECT created_at FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![namespace, key], - |row| row.get::<_, f64>(0), - ) - .optional() - .map_err(|e| format!("lookup existing created_at: {e}"))? - .unwrap_or(now) + DocumentIdentity::New { document_id } => ( + document_id.unwrap_or_else(|| Self::derive_document_id(&namespace, &key)), + now, + false, + ), }; let updated_at = now; let markdown_rel = self @@ -249,6 +277,9 @@ impl UnifiedMemory { let tx = conn .unchecked_transaction() .map_err(|e| format!("begin tx: {e}"))?; + if rekey { + Self::rekey_document(&tx, &namespace, &document_id, &key)?; + } tx.execute( "INSERT INTO memory_docs (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path, taint, logical_namespace) @@ -361,35 +392,29 @@ impl UnifiedMemory { let _write_guard = Self::document_write_lock(&self.db_path, &namespace, &key) .lock_owned() .await; - let existing_document_id = { - let conn = self.conn.lock(); - conn.query_row( - "SELECT document_id FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![namespace, key], - |row| row.get::<_, String>(0), - ) - .optional() - .map_err(|e| format!("lookup existing document_id: {e}"))? - }; - let document_id = input - .document_id - .or(existing_document_id) - .unwrap_or_else(|| { - let ts = Self::now_ts() as u64; - let short = &Uuid::new_v4().to_string()[..8]; - format!("{ts}_{short}") - }); let now = Self::now_ts(); - let created_at = { + let identity = { let conn = self.conn.lock(); - conn.query_row( - "SELECT created_at FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![namespace, key], - |row| row.get::<_, f64>(0), - ) - .optional() - .map_err(|e| format!("lookup existing created_at: {e}"))? - .unwrap_or(now) + Self::resolve_document_identity(&conn, &namespace, &key, input.document_id.as_deref())? + }; + let (document_id, created_at, rekey) = match identity { + DocumentIdentity::Existing { + document_id, + created_at, + } => (document_id, created_at, false), + DocumentIdentity::StaleKey { + document_id, + created_at, + } => (document_id, created_at, true), + DocumentIdentity::New { document_id } => ( + document_id.unwrap_or_else(|| { + let ts = Self::now_ts() as u64; + let short = &Uuid::new_v4().to_string()[..8]; + format!("{ts}_{short}") + }), + now, + false, + ), }; let updated_at = now; let markdown_rel = self @@ -412,7 +437,13 @@ impl UnifiedMemory { { let conn = self.conn.lock(); - conn.execute( + let tx = conn + .unchecked_transaction() + .map_err(|e| format!("begin tx: {e}"))?; + if rekey { + Self::rekey_document(&tx, &namespace, &document_id, &key)?; + } + tx.execute( "INSERT INTO memory_docs (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path, taint, logical_namespace) VALUES @@ -450,6 +481,7 @@ impl UnifiedMemory { ], ) .map_err(|e| format!("upsert memory_docs: {e}"))?; + tx.commit().map_err(|e| format!("commit tx: {e}"))?; } Ok(document_id) @@ -853,6 +885,86 @@ impl UnifiedMemory { Arc::clone(table.entry(id).or_default()) } + /// Resolve the row a write of `(namespace, key)` addresses; see + /// [`DocumentIdentity`] for the cases. `requested_id` is the caller's + /// `document_id`; it is trimmed, and a blank one is no request. + fn resolve_document_identity( + conn: &rusqlite::Connection, + namespace: &str, + key: &str, + requested_id: Option<&str>, + ) -> Result { + let requested_id = requested_id.map(str::trim).filter(|id| !id.is_empty()); + let by_key = conn + .query_row( + "SELECT document_id, created_at FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", + params![namespace, key], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, f64>(1)?)), + ) + .optional() + .map_err(|e| format!("lookup existing document: {e}"))?; + if let Some((document_id, created_at)) = by_key { + if requested_id.is_some_and(|requested| requested != document_id) { + log::debug!( + "[memory] document write keeps the row's id over the requested one namespace={namespace}" + ); + } + return Ok(DocumentIdentity::Existing { + document_id, + created_at, + }); + } + let Some(requested_id) = requested_id else { + return Ok(DocumentIdentity::New { document_id: None }); + }; + let owner = conn + .query_row( + "SELECT namespace, created_at FROM memory_docs WHERE document_id = ?1 LIMIT 1", + params![requested_id], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, f64>(1)?)), + ) + .optional() + .map_err(|e| format!("lookup document id owner: {e}"))?; + Ok(match owner { + None => DocumentIdentity::New { + document_id: Some(requested_id.to_owned()), + }, + Some((owner, created_at)) if owner == namespace => DocumentIdentity::StaleKey { + document_id: requested_id.to_owned(), + created_at, + }, + Some((owner, _)) => { + log::warn!( + "[memory] requested document id belongs to another namespace; storing under a derived id namespace={namespace} owner_namespace={owner}" + ); + DocumentIdentity::New { document_id: None } + } + }) + } + + /// Move the row `document_id` of `namespace` to `key`, so the upsert that + /// follows updates it through `ON CONFLICT(namespace, key)` instead of + /// inserting a second row the primary key then rejects. Runs inside the + /// caller's transaction. Chunks, the markdown sidecar and graph relations + /// are keyed by the id and need no change. + fn rekey_document( + conn: &rusqlite::Connection, + namespace: &str, + document_id: &str, + key: &str, + ) -> Result<(), String> { + let rekeyed = conn + .execute( + "UPDATE memory_docs SET key = ?1 WHERE namespace = ?2 AND document_id = ?3", + params![key, namespace, document_id], + ) + .map_err(|e| format!("re-key memory_docs: {e}"))?; + log::info!( + "[memory] re-keyed a document to the key its write addressed it by namespace={namespace} rows={rekeyed}" + ); + Ok(()) + } + /// A document id derived from `(namespace, key)`. /// /// Deterministic so two concurrent first-writes of one key agree, which is @@ -889,3 +1001,7 @@ mod tests; #[cfg(test)] #[path = "documents_document_id_tests.rs"] mod document_id_tests; + +#[cfg(test)] +#[path = "documents_identity_tests.rs"] +mod identity_tests; diff --git a/crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs b/crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs new file mode 100644 index 0000000..bae6c02 --- /dev/null +++ b/crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs @@ -0,0 +1,296 @@ +//! Tests for how a document write resolves the row it addresses when the +//! caller supplies a `document_id` (the table's primary key) alongside the +//! `(namespace, key)` dedup key. The two can name different rows, and the +//! write must land on one row instead of failing on the primary key +//! (openhuman#6147). + +use std::sync::Arc; + +use serde_json::json; +use tempfile::TempDir; + +use crate::store::{NamespaceDocumentInput, UnifiedMemory}; +use tinymemory_api::host::NoopEmbedding; + +fn open() -> (TempDir, UnifiedMemory) { + let tmp = TempDir::new().unwrap(); + let memory = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); + (tmp, memory) +} + +fn input( + namespace: &str, + key: &str, + document_id: Option<&str>, + content: &str, +) -> NamespaceDocumentInput { + NamespaceDocumentInput { + namespace: namespace.to_string(), + key: key.to_string(), + title: key.to_string(), + content: content.to_string(), + source_type: "doc".to_string(), + priority: "medium".to_string(), + tags: vec![], + metadata: json!({}), + category: "core".to_string(), + session_id: None, + document_id: document_id.map(str::to_owned), + taint: crate::MemoryTaint::ExternalSync, + } +} + +/// `(document_id, key, content, created_at)` of every row in `namespace`, +/// ordered by key. +fn stored_rows(memory: &UnifiedMemory, namespace: &str) -> Vec<(String, String, String, f64)> { + let conn = memory.conn.lock(); + let mut statement = conn + .prepare( + "SELECT document_id, key, content, created_at FROM memory_docs + WHERE namespace = ?1 ORDER BY key", + ) + .unwrap(); + let rows = statement + .query_map( + rusqlite::params![UnifiedMemory::sanitize_namespace(namespace)], + |row| { + Ok::<_, rusqlite::Error>(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, f64>(3)?, + )) + }, + ) + .unwrap(); + rows.map(Result::unwrap).collect() +} + +/// The distinct document ids `vector_chunks` holds for `namespace`. +fn chunk_document_ids(memory: &UnifiedMemory, namespace: &str) -> Vec { + let conn = memory.conn.lock(); + let mut statement = conn + .prepare( + "SELECT DISTINCT document_id FROM vector_chunks + WHERE namespace = ?1 ORDER BY document_id", + ) + .unwrap(); + let ids = statement + .query_map( + rusqlite::params![UnifiedMemory::sanitize_namespace(namespace)], + |row| row.get::<_, String>(0), + ) + .unwrap(); + ids.map(Result::unwrap).collect() +} + +/// The reporter's shape (openhuman#6147). Sync providers used to key their +/// documents by TITLE while already passing the stable `{toolkit}:{id}` as +/// the document id; since openhuman#4953 the key is that id. Re-syncing such +/// a document writes a NEW `(namespace, key)` whose requested id the old row +/// still holds. Before this fix the insert failed with +/// `UNIQUE constraint failed: memory_docs.document_id`, and a provider that +/// does not tolerate scope errors aborted its whole run on every tick that +/// reached the item. +#[tokio::test] +async fn a_requested_id_that_names_a_row_under_a_stale_key_updates_that_row() { + let (_tmp, memory) = open(); + let namespace = "skill-github"; + let stable_id = "github:4892120323"; + + let legacy_id = memory + .upsert_document(input( + namespace, + "Fix the login page", + Some(stable_id), + "issue body v1", + )) + .await + .unwrap(); + assert_eq!(legacy_id, stable_id); + let legacy_created_at = stored_rows(&memory, namespace)[0].3; + + let resynced_id = memory + .upsert_document(input( + namespace, + stable_id, + Some(stable_id), + "issue body v2", + )) + .await + .expect("a re-sync keyed by the stable id must update the title-keyed row"); + assert_eq!(resynced_id, legacy_id); + + let rows = stored_rows(&memory, namespace); + assert_eq!( + rows.len(), + 1, + "the re-sync must update the row in place, not add a second one" + ); + let (document_id, key, content, created_at) = &rows[0]; + assert_eq!(document_id, stable_id); + assert_eq!( + key, stable_id, + "the row now carries the key the write addressed it by" + ); + assert_eq!(content, "issue body v2"); + assert_eq!( + *created_at, legacy_created_at, + "re-keying is an update: created_at survives" + ); + assert_eq!( + chunk_document_ids(&memory, namespace), + vec![stable_id.to_string()], + "the chunks stay addressable from the row's id" + ); + assert!( + memory + .get_document_by_key(namespace, stable_id) + .await + .unwrap() + .is_some(), + "the row resolves by its new key" + ); + assert!( + memory + .get_document_by_key(namespace, "Fix the login page") + .await + .unwrap() + .is_none(), + "the stale key no longer resolves" + ); +} + +/// A requested id that another namespace's row already holds must not block +/// this namespace's write: the store mints its usual derived id instead and +/// leaves the other row alone. Document ids are addressed per namespace +/// everywhere else (`delete_document`, chunk and graph lookups), so a foreign +/// row is not "the same document". +#[tokio::test] +async fn a_requested_id_owned_by_another_namespace_stores_under_a_derived_id() { + let (_tmp, memory) = open(); + memory + .upsert_document(input( + "skill-github", + "github:1", + Some("github:1"), + "issue in github", + )) + .await + .unwrap(); + + let stored = memory + .upsert_document(input( + "notes", + "github:1", + Some("github:1"), + "a note about it", + )) + .await + .expect("a foreign row must not block the write"); + + assert_eq!( + stored, + UnifiedMemory::derive_document_id("notes", "github:1") + ); + let notes = stored_rows(&memory, "notes"); + assert_eq!(notes.len(), 1); + assert_eq!(notes[0].0, stored); + assert_eq!(notes[0].2, "a note about it"); + assert_eq!(chunk_document_ids(&memory, "notes"), vec![stored]); + let github = stored_rows(&memory, "skill-github"); + assert_eq!(github.len(), 1); + assert_eq!( + ( + github[0].0.as_str(), + github[0].1.as_str(), + github[0].2.as_str() + ), + ("github:1", "github:1", "issue in github"), + "the other namespace's row is untouched" + ); +} + +/// When a row already exists for `(namespace, key)`, its id wins over a +/// different requested one. The upsert's `DO UPDATE` never rewrites +/// `document_id`, so honouring the request would hand back — and write the +/// chunks and queue the graph job under — an id no row has. +#[tokio::test] +async fn the_row_id_wins_over_a_conflicting_requested_id() { + let (_tmp, memory) = open(); + let derived = memory + .upsert_document(input("notes", "plan", None, "draft one")) + .await + .unwrap(); + assert_eq!(derived, UnifiedMemory::derive_document_id("notes", "plan")); + + let stored = memory + .upsert_document(input("notes", "plan", Some("plan-v2"), "draft two")) + .await + .unwrap(); + + assert_eq!(stored, derived, "the existing row's id is the write's id"); + let notes = stored_rows(&memory, "notes"); + assert_eq!(notes.len(), 1); + assert_eq!(notes[0].0, derived); + assert_eq!(notes[0].2, "draft two"); + assert_eq!( + chunk_document_ids(&memory, "notes"), + vec![derived], + "no chunks are written under an id the row does not have" + ); +} + +/// The metadata-only path writes the same row, so it resolves the row the +/// same way: through a stale key when the requested id names one. +#[tokio::test] +async fn a_metadata_only_write_reaches_a_row_under_a_stale_key_too() { + let (_tmp, memory) = open(); + let namespace = "skill-github"; + memory + .upsert_document(input( + namespace, + "Fix the login page", + Some("github:7"), + "issue body v1", + )) + .await + .unwrap(); + let legacy_created_at = stored_rows(&memory, namespace)[0].3; + + let stored = memory + .upsert_document_metadata_only(input( + namespace, + "github:7", + Some("github:7"), + "issue body v2", + )) + .await + .expect("the light write must update the title-keyed row"); + + assert_eq!(stored, "github:7"); + let rows = stored_rows(&memory, namespace); + assert_eq!(rows.len(), 1); + assert_eq!( + (rows[0].1.as_str(), rows[0].2.as_str()), + ("github:7", "issue body v2") + ); + assert_eq!(rows[0].3, legacy_created_at); + assert_eq!( + chunk_document_ids(&memory, namespace), + vec!["github:7".to_string()], + "a metadata-only write leaves the row's chunks in place" + ); +} + +/// A blank requested id is no id: the write mints its own rather than making +/// the empty string a primary key. +#[tokio::test] +async fn a_blank_requested_id_is_ignored() { + let (_tmp, memory) = open(); + let stored = memory + .upsert_document(input("notes", "plan", Some(" "), "draft")) + .await + .unwrap(); + assert_eq!(stored, UnifiedMemory::derive_document_id("notes", "plan")); +} From 794237f8daa87a52a5a2c89ff688cbc7128096a2 Mon Sep 17 00:00:00 2001 From: Shanu Date: Wed, 9 Sep 2026 13:41:24 +0530 Subject: [PATCH 2/2] Re-check the document's key inside the write transaction The identity resolution runs before the markdown sidecar is written and outside the connection lock, and the per-key write lock only serialises writers of one key. A writer reaching the same document through another key could re-key the row in between, and the upsert's ON CONFLICT(namespace, key) would then miss the row and trip the primary key. Look the row up by id inside the transaction and move it under the key being written there, on both write paths; the early resolution now only settles the id and created_at. --- .../src/store/namespace_store/README.md | 15 ++- .../src/store/namespace_store/documents.rs | 100 +++++++++--------- .../documents_identity_tests.rs | 72 +++++++++++++ 3 files changed, 135 insertions(+), 52 deletions(-) diff --git a/crates/tinymemory-core/src/store/namespace_store/README.md b/crates/tinymemory-core/src/store/namespace_store/README.md index 037f5e2..e191722 100644 --- a/crates/tinymemory-core/src/store/namespace_store/README.md +++ b/crates/tinymemory-core/src/store/namespace_store/README.md @@ -54,15 +54,22 @@ row two ways, and `documents.rs` resolves both before writing - no row has the key, but the requested id names a row of the **same** namespace → the same document under a stale key (providers keyed by title before openhuman#4953 while already passing their stable id). The row is - re-keyed inside the write transaction and updated. Before this, the insert - failed with `UNIQUE constraint failed: memory_docs.document_id` and a - provider that does not tolerate scope errors aborted every sync run that - reached the item (openhuman#6147); + re-keyed and updated. Before this, the insert failed with + `UNIQUE constraint failed: memory_docs.document_id` and a provider that does + not tolerate scope errors aborted every sync run that reached the item + (openhuman#6147); - the requested id belongs to a row in **another** namespace → a derived id is used and a warning logged. Ids are addressed per namespace everywhere else (`delete_document`, chunk and graph lookups), so a foreign row is not this document and never blocks the write. +The resolution runs before the markdown sidecar is written and outside the +connection lock, and the per-key write lock only serialises writers of *one* +key — so which key the row carries is checked again inside the write +transaction (`rekey_document_in_namespace`, one primary-key lookup per write), +where a row the same document reached through another key is moved under the +key being written before the upsert runs. + - **`graph.rs`** — `graph_namespace` / `graph_global` upserts with attribute merging and evidence accumulation, plus namespace / global / cross-namespace queries and document-scoped relation removal. - **`query.rs`** — hybrid retrieval. Combines graph relevance, vector similarity, keyword overlap, episodic signal and freshness; exposes `query_namespace_*` (with query) and `recall_namespace_*` (query-less) entry points used by `MemoryClient`. - **`helpers.rs`** — shared utilities: f32-vector byte codecs, cosine similarity, markdown chunking, text/graph normalisation, JSON attribute merging, recency scoring. diff --git a/crates/tinymemory-core/src/store/namespace_store/documents.rs b/crates/tinymemory-core/src/store/namespace_store/documents.rs index e9dc7ff..1d98ed8 100644 --- a/crates/tinymemory-core/src/store/namespace_store/documents.rs +++ b/crates/tinymemory-core/src/store/namespace_store/documents.rs @@ -40,23 +40,18 @@ pub(crate) const EMBED_REQUEST_MAX_TEXTS: usize = 64; /// resolve the row up front so the row they update, the chunks they replace /// and the id they hand back all agree. enum DocumentIdentity { - /// A row with this `(namespace, key)` exists. Its id wins over any - /// requested one: the upsert's `DO UPDATE` never rewrites `document_id`, - /// so chunks written under a different id would be orphaned and the graph - /// job queued under an id no row has. + /// A row of this namespace is this document: either it carries the + /// `(namespace, key)` — then its id wins over any requested one, because + /// the upsert's `DO UPDATE` never rewrites `document_id` and chunks + /// written under another id would be orphaned — or it carries the + /// requested id under another key, written when a different key rule + /// applied (sync providers keyed by title before openhuman#4953 while + /// already passing their stable id). The write transaction moves such a + /// row under the key being written ([`UnifiedMemory::rekey_document_in_namespace`]). Existing { document_id: String, created_at: f64, }, - /// No row has this key, but the requested id names a row of the same - /// namespace filed under another key: the same document, written when a - /// different key rule applied (sync providers keyed by title before - /// openhuman#4953 while already passing their stable id). The write - /// re-keys that row and updates it. - StaleKey { - document_id: String, - created_at: f64, - }, /// A new row. `document_id` is the requested id when it is free; `None` /// when the write must mint one — nothing usable was requested, or the /// requested id belongs to a row in another namespace, which is not this @@ -224,15 +219,11 @@ impl UnifiedMemory { let conn = self.conn.lock(); Self::resolve_document_identity(&conn, &namespace, &key, input.document_id.as_deref())? }; - let (document_id, created_at, rekey) = match identity { + let (document_id, created_at) = match identity { DocumentIdentity::Existing { document_id, created_at, - } => (document_id, created_at, false), - DocumentIdentity::StaleKey { - document_id, - created_at, - } => (document_id, created_at, true), + } => (document_id, created_at), // Derived from (namespace, key), NOT random. The lookup above and // the write below are separated by `.await`s, so two concurrent // stores of a not-yet-existing key both miss and both mint an id. @@ -248,7 +239,6 @@ impl UnifiedMemory { DocumentIdentity::New { document_id } => ( document_id.unwrap_or_else(|| Self::derive_document_id(&namespace, &key)), now, - false, ), }; let updated_at = now; @@ -277,9 +267,7 @@ impl UnifiedMemory { let tx = conn .unchecked_transaction() .map_err(|e| format!("begin tx: {e}"))?; - if rekey { - Self::rekey_document(&tx, &namespace, &document_id, &key)?; - } + Self::rekey_document_in_namespace(&tx, &namespace, &document_id, &key)?; tx.execute( "INSERT INTO memory_docs (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path, taint, logical_namespace) @@ -397,15 +385,11 @@ impl UnifiedMemory { let conn = self.conn.lock(); Self::resolve_document_identity(&conn, &namespace, &key, input.document_id.as_deref())? }; - let (document_id, created_at, rekey) = match identity { + let (document_id, created_at) = match identity { DocumentIdentity::Existing { document_id, created_at, - } => (document_id, created_at, false), - DocumentIdentity::StaleKey { - document_id, - created_at, - } => (document_id, created_at, true), + } => (document_id, created_at), DocumentIdentity::New { document_id } => ( document_id.unwrap_or_else(|| { let ts = Self::now_ts() as u64; @@ -413,7 +397,6 @@ impl UnifiedMemory { format!("{ts}_{short}") }), now, - false, ), }; let updated_at = now; @@ -440,9 +423,7 @@ impl UnifiedMemory { let tx = conn .unchecked_transaction() .map_err(|e| format!("begin tx: {e}"))?; - if rekey { - Self::rekey_document(&tx, &namespace, &document_id, &key)?; - } + Self::rekey_document_in_namespace(&tx, &namespace, &document_id, &key)?; tx.execute( "INSERT INTO memory_docs (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path, taint, logical_namespace) @@ -888,6 +869,11 @@ impl UnifiedMemory { /// Resolve the row a write of `(namespace, key)` addresses; see /// [`DocumentIdentity`] for the cases. `requested_id` is the caller's /// `document_id`; it is trimmed, and a blank one is no request. + /// + /// Runs before the sidecar write, outside the connection lock, so it only + /// settles the id and `created_at`. Which key the row carries is checked + /// again inside the write transaction by + /// [`Self::rekey_document_in_namespace`]. fn resolve_document_identity( conn: &rusqlite::Connection, namespace: &str, @@ -929,7 +915,7 @@ impl UnifiedMemory { None => DocumentIdentity::New { document_id: Some(requested_id.to_owned()), }, - Some((owner, created_at)) if owner == namespace => DocumentIdentity::StaleKey { + Some((owner, created_at)) if owner == namespace => DocumentIdentity::Existing { document_id: requested_id.to_owned(), created_at, }, @@ -942,27 +928,45 @@ impl UnifiedMemory { }) } - /// Move the row `document_id` of `namespace` to `key`, so the upsert that - /// follows updates it through `ON CONFLICT(namespace, key)` instead of - /// inserting a second row the primary key then rejects. Runs inside the - /// caller's transaction. Chunks, the markdown sidecar and graph relations - /// are keyed by the id and need no change. - fn rekey_document( + /// Inside the write transaction: if `namespace` holds the row + /// `document_id` under a key other than `key`, move it under `key`, so the + /// upsert that follows updates it through `ON CONFLICT(namespace, key)` + /// instead of inserting a second row the primary key then rejects. + /// + /// Checked here, at the moment of writing, rather than trusted from + /// [`Self::resolve_document_identity`]: that ran before the sidecar write + /// and outside the connection lock, and the per-key write lock only + /// serialises writers of *this* key. A writer addressing the same + /// document through another key can re-key the row in between, and the + /// upsert would then miss it. Every write pays one primary-key lookup for + /// that. Chunks, the markdown sidecar and graph relations are keyed by + /// the id and need no change. Returns whether the row was re-keyed. + fn rekey_document_in_namespace( conn: &rusqlite::Connection, namespace: &str, document_id: &str, key: &str, - ) -> Result<(), String> { - let rekeyed = conn - .execute( - "UPDATE memory_docs SET key = ?1 WHERE namespace = ?2 AND document_id = ?3", - params![key, namespace, document_id], + ) -> Result { + let current_key = conn + .query_row( + "SELECT key FROM memory_docs WHERE namespace = ?1 AND document_id = ?2 LIMIT 1", + params![namespace, document_id], + |row| row.get::<_, String>(0), ) - .map_err(|e| format!("re-key memory_docs: {e}"))?; + .optional() + .map_err(|e| format!("lookup document key: {e}"))?; + if current_key.as_deref().is_none_or(|current| current == key) { + return Ok(false); + } + conn.execute( + "UPDATE memory_docs SET key = ?1 WHERE namespace = ?2 AND document_id = ?3", + params![key, namespace, document_id], + ) + .map_err(|e| format!("re-key memory_docs: {e}"))?; log::info!( - "[memory] re-keyed a document to the key its write addressed it by namespace={namespace} rows={rekeyed}" + "[memory] re-keyed a document to the key its write addressed it by namespace={namespace}" ); - Ok(()) + Ok(true) } /// A document id derived from `(namespace, key)`. diff --git a/crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs b/crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs index bae6c02..76d5f4f 100644 --- a/crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs +++ b/crates/tinymemory-core/src/store/namespace_store/documents_identity_tests.rs @@ -283,6 +283,78 @@ async fn a_metadata_only_write_reaches_a_row_under_a_stale_key_too() { ); } +/// The per-key write lock serialises writers of one key only. A writer that +/// resolved its row before the transaction can find, inside it, that another +/// writer reached the same document through a different key and moved it +/// meanwhile; the transaction re-checks the key itself so the upsert lands on +/// the row instead of tripping the primary key. +#[tokio::test] +async fn the_write_transaction_rekeys_a_row_another_writer_moved_meanwhile() { + let (_tmp, memory) = open(); + let namespace = "skill-github"; + let stored_namespace = UnifiedMemory::sanitize_namespace(namespace); + memory + .upsert_document(input( + namespace, + "moved-key", + Some("github:9"), + "issue body v1", + )) + .await + .unwrap(); + + { + let conn = memory.conn.lock(); + assert!( + UnifiedMemory::rekey_document_in_namespace( + &conn, + &stored_namespace, + "github:9", + "github:9" + ) + .unwrap(), + "a row held under another key is moved under the key being written" + ); + assert!( + !UnifiedMemory::rekey_document_in_namespace( + &conn, + &stored_namespace, + "github:9", + "github:9" + ) + .unwrap(), + "a row already under the key is left alone" + ); + assert!( + !UnifiedMemory::rekey_document_in_namespace( + &conn, + &stored_namespace, + "github:404", + "x" + ) + .unwrap(), + "a document the namespace does not hold is nothing to re-key" + ); + } + + let stored = memory + .upsert_document(input( + namespace, + "github:9", + Some("github:9"), + "issue body v2", + )) + .await + .expect("the upsert lands on the moved row"); + assert_eq!(stored, "github:9"); + let rows = stored_rows(&memory, namespace); + assert_eq!(rows.len(), 1); + assert_eq!( + (rows[0].1.as_str(), rows[0].2.as_str()), + ("github:9", "issue body v2") + ); +} + /// A blank requested id is no id: the write mints its own rather than making /// the empty string a primary key. #[tokio::test]