fix(memory): gmail canonicalisation, signature unification, adaptive page retry

src/memory/chunks/connection.rs

@@ -23,7 +23,8 @@ pub(crate) use super::connection_breaker::CB_THRESHOLD;
 
 use super::migrations::{migrate_legacy_embeddings_to_sidecar, purge_global_topic_trees};
 use super::recovery::{
-    is_corrupt_error, is_io_open_error, quarantine_corrupt_files, try_cleanup_stale_files,
+    is_corrupt_error, is_io_open_error, is_transient_cold_start, quarantine_corrupt_files,
+    try_cleanup_stale_files,
 };
 use super::schema::SCHEMA;
 use super::{db_path_for, SQLITE_BUSY_TIMEOUT};
@@ -366,6 +367,40 @@ pub(crate) fn get_or_init_connection(config: &MemoryConfig) -> Result<Arc<PMutex
 /// `db_path` is always produced by [`super::db_path_for`], which joins onto
 /// `config.workspace`.
 fn open_and_init(db_path: &Path, config: &MemoryConfig) -> Result<Connection> {
+    // Bootstrap I/O errors (`CANTOPEN`, the `-shm`/`-wal` cold-start races,
+    // `IOERR_FSTAT`) come from a sibling connection creating or truncating the
+    // same file at the same instant — a fresh open a moment later succeeds. Give
+    // a transient failure a couple of short, backed-off retries before
+    // surfacing it, which is what removes the flaky parallel-open failure.
+    const MAX_ATTEMPTS: u32 = 3;
+    let mut attempt = 0;
+    loop {
+        match try_open_and_init(db_path, config) {
+            Ok(conn) => return Ok(conn),
+            Err(error) if attempt + 1 < MAX_ATTEMPTS && is_transient_cold_start(&error) => {
+                let backoff = std::time::Duration::from_millis(20 * (1 << attempt));
+                // `tracing` is optional and only arrives with `sync`; the chunk
+                // store compiles in the dependency-light default build too, so
+                // the call has to be gated or that build cannot link.
+                #[cfg(feature = "sync")]
+                tracing::debug!(
+                    db_path = %db_path.display(),
+                    attempt = attempt + 1,
+                    backoff_ms = backoff.as_millis() as u64,
+                    "[chunks] transient cold-start opening chunk DB; retrying"
+                );
+                std::thread::sleep(backoff);
+                attempt += 1;
+            }
+            Err(error) => return Err(error),
+        }
+    }
+}
+
+/// One open+init attempt. Retried by [`open_and_init`] on a transient
+/// cold-start failure. Does not clean up a partially created directory or file
+/// on failure — a subsequent attempt simply retries against the same path.
+fn try_open_and_init(db_path: &Path, config: &MemoryConfig) -> Result<Connection> {
     let dir = db_path.parent().expect("db_path always has a parent");
     std::fs::create_dir_all(dir)
         .with_context(|| format!("Failed to create chunk DB dir: {}", dir.display()))?;

src/memory/chunks/embeddings.rs

@@ -6,6 +6,7 @@
 //! ported here) — callers pass vectors in.
 
 use super::connection::with_connection;
+use super::signature::{format_signature, signature_in_clause, signature_variants};
 use anyhow::{Context, Result};
 use chrono::Utc;
 use rusqlite::{Connection, OptionalExtension};
@@ -20,11 +21,20 @@ pub(crate) fn active_embedding_dims(config: &MemoryConfig) -> usize {
 }
 
 /// Resolve the active embedding signature — the canonical key every per-model
-/// sidecar read/write is scoped by. Derived from the configured model + dim so
-/// a provider/model/dimension switch becomes a query-time filter rather than a
-/// destructive rewrite.
+/// sidecar read/write is scoped by. Derived from the configured provider, model
+/// and dim so a provider/model/dimension switch becomes a query-time filter
+/// rather than a destructive rewrite.
+///
+/// This is the canonical `provider=…;model=…;dims=…` spelling, shared with the
+/// namespace store. Rows written under the tree's older `{model}@{dims}`
+/// spelling are still found, because per-signature reads match every variant
+/// (see [`signature_variants`]) rather than one exact string.
 pub fn tree_active_signature(config: &MemoryConfig) -> String {
-    format!("{}@{}", config.embedding.model, config.embedding.dim)
+    format_signature(
+        &config.embedding.provider,
+        &config.embedding.model,
+        config.embedding.dim,
+    )
 }
 
 /// Store a chunk's embedding under the active model signature (see
@@ -292,7 +302,8 @@ pub(crate) fn validate_reembed_skip_key<'a>(label: &str, value: &'a str) -> Resu
     Ok(trimmed)
 }
 
-/// Fetch a chunk embedding for exactly one provider/model/dimension signature.
+/// Fetch a chunk embedding for one provider/model/dimension signature — under
+/// any of its spellings (see [`signature_variants`]).
 ///
 /// Returns `Ok(None)` when no row exists for `(chunk_id, model_signature)` —
 /// absence is not an error.
@@ -307,13 +318,22 @@ pub fn get_chunk_embedding_for_signature(
     chunk_id: &str,
     model_signature: &str,
 ) -> Result<Option<Vec<f32>>> {
+    let variants = signature_variants(model_signature);
     with_connection(config, |conn| {
+        let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(variants.len() + 1);
+        params.push(&chunk_id as &dyn rusqlite::ToSql);
+        for variant in &variants {
+            params.push(variant as &dyn rusqlite::ToSql);
+        }
         let row: Option<(Vec<u8>, i64)> = conn
             .query_row(
-                "SELECT vector, dim
-                   FROM mem_tree_chunk_embeddings
-                  WHERE chunk_id = ?1 AND model_signature = ?2",
-                rusqlite::params![chunk_id, model_signature],
+                &format!(
+                    "SELECT vector, dim
+                       FROM mem_tree_chunk_embeddings
+                      WHERE chunk_id = ?1 AND model_signature {}",
+                    signature_in_clause(variants.len(), 2)
+                ),
+                params.as_slice(),
                 |r| Ok((r.get(0)?, r.get(1)?)),
             )
             .optional()?;
@@ -374,26 +394,40 @@ fn embedding_from_blob(bytes: &[u8], dim: i64, label: &str) -> Result<Option<Vec
     Ok(Some(floats))
 }
 
-/// Whether any live chunk or summary lacks both an embedding and terminal tombstone.
+/// Whether any live chunk or summary lacks both an embedding and terminal
+/// tombstone.
+///
+/// Coverage counts a vector written under *any* spelling of the signature: a
+/// row that only looks uncovered because the convention changed is not work,
+/// and re-embedding it would spend provider quota to reproduce a vector that
+/// is already on disk.
 pub fn has_uncovered_reembed_work(
     conn: &Connection,
     model_signature: &str,
 ) -> rusqlite::Result<bool> {
+    let variants = signature_variants(model_signature);
+    let sig_clause = signature_in_clause(variants.len(), 1);
+    let params: Vec<&dyn rusqlite::ToSql> = variants
+        .iter()
+        .map(|variant| variant as &dyn rusqlite::ToSql)
+        .collect();
     conn.query_row(
-        "SELECT EXISTS(
-             SELECT 1 FROM mem_tree_chunks c
-              WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e
-                                 WHERE e.chunk_id = c.id AND e.model_signature = ?1)
-                AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk
-                                 WHERE sk.chunk_id = c.id AND sk.model_signature = ?1))
-           OR EXISTS(
-             SELECT 1 FROM mem_tree_summaries s
-              WHERE s.deleted = 0
-                AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e
-                                 WHERE e.summary_id = s.id AND e.model_signature = ?1)
-                AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk
-                                 WHERE sk.summary_id = s.id AND sk.model_signature = ?1))",
-        rusqlite::params![model_signature],
+        &format!(
+            "SELECT EXISTS(
+                 SELECT 1 FROM mem_tree_chunks c
+                  WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e
+                                     WHERE e.chunk_id = c.id AND e.model_signature {sig_clause})
+                    AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk
+                                     WHERE sk.chunk_id = c.id AND sk.model_signature {sig_clause}))
+               OR EXISTS(
+                 SELECT 1 FROM mem_tree_summaries s
+                  WHERE s.deleted = 0
+                    AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e
+                                     WHERE e.summary_id = s.id AND e.model_signature {sig_clause})
+                    AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk
+                                     WHERE sk.summary_id = s.id AND sk.model_signature {sig_clause}))"
+        ),
+        params.as_slice(),
         |row| row.get(0),
     )
 }
@@ -427,6 +461,7 @@ pub fn get_chunk_embeddings_for_signature_batch(
     if chunk_ids.is_empty() {
         return Ok(HashMap::new());
     }
+    let variants = signature_variants(model_signature);
     with_connection(config, |conn| {
         let mut out: HashMap<String, Vec<f32>> = HashMap::with_capacity(chunk_ids.len());
         for window in chunk_ids.chunks(MAX_EMBEDDING_BATCH) {
@@ -437,17 +472,20 @@ pub fn get_chunk_embeddings_for_signature_batch(
                 "SELECT chunk_id, vector, dim
                    FROM mem_tree_chunk_embeddings
                   WHERE chunk_id IN ({placeholders})
-                    AND model_signature = ?{sig_idx}",
-                sig_idx = window.len() + 1,
+                    AND model_signature {sig_clause}",
+                sig_clause = signature_in_clause(variants.len(), window.len() + 1),
             );
             let mut stmt = conn
                 .prepare(&sql)
                 .context("prepare get_chunk_embeddings_for_signature_batch")?;
-            let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(window.len() + 1);
+            let mut params: Vec<&dyn rusqlite::ToSql> =
+                Vec::with_capacity(window.len() + variants.len());
             for id in window {
                 params.push(id as &dyn rusqlite::ToSql);
             }
-            params.push(&model_signature as &dyn rusqlite::ToSql);
+            for variant in &variants {
+                params.push(variant as &dyn rusqlite::ToSql);
+            }
             let rows = stmt
                 .query_map(params.as_slice(), |row| {
                     Ok((

src/memory/chunks/mod.rs

@@ -51,6 +51,8 @@ mod produce_split;
 mod raw_refs;
 #[path = "semantic.rs"]
 mod semantic;
+#[path = "signature.rs"]
+mod signature;
 #[path = "store.rs"]
 mod store;
 #[path = "store_delete.rs"]
@@ -98,6 +100,10 @@ pub use raw_refs::{
 pub use recovery::{
     is_io_open_error, is_transient_cold_start, recover_corrupt_db, try_cleanup_stale_files,
 };
+pub(crate) use signature::signature_in_clause;
+pub use signature::{
+    format_signature, parse_signature, signature_variants, signatures_equivalent, SignatureParts,
+};
 pub use store::{
     claim_source_ingest_tx, count_chunks, count_chunks_by_lifecycle_status,
     count_raw_paths_ingested_with_prefix, delete_source_ingest, extraction_coverage,

src/memory/chunks/recovery.rs

@@ -43,6 +43,10 @@ const SQLITE_IOERR_SHMSIZE: i32 = 4874;
 const SQLITE_IOERR_SHMMAP: i32 = 5386;
 /// `IOERR_IN_PAGE` — an mmap-page I/O fault, also seen under WAL cold-start.
 const SQLITE_IOERR_IN_PAGE: i32 = 8714;
+/// `IOERR_FSTAT` — an `fstat()` on the db/side-file failed, observed when a
+/// sibling connection is creating or truncating the file at the same instant
+/// (seen as flaky `Error code 1802: disk I/O error` under parallel cold opens).
+const SQLITE_IOERR_FSTAT: i32 = 1802;
 
 /// True if `err` (or anything in its cause chain) is one of the SQLite codes
 /// that fire during cold-start WAL/SHM bootstrap races.
@@ -67,6 +71,7 @@ pub fn is_transient_cold_start(err: &anyhow::Error) -> bool {
                     | SQLITE_IOERR_SHMSIZE
                     | SQLITE_IOERR_SHMMAP
                     | SQLITE_IOERR_IN_PAGE
+                    | SQLITE_IOERR_FSTAT
             );
         }
         false

src/memory/chunks/recovery_tests.rs

@@ -147,3 +147,22 @@ fn recover_corrupt_db_is_noop_on_healthy_db() {
         .any(|e| e.file_name().to_string_lossy().contains(".corrupt-"));
     assert!(!quarantined, "no quarantine file should be created");
 }
+
+#[test]
+fn is_transient_cold_start_classifies_ioerr_fstat() {
+    // `IOERR_FSTAT` (1802) is the flaky `disk I/O error` a parallel cold open
+    // hits; it must be treated as transient so `open_and_init` retries instead
+    // of surfacing it. A non-transient sqlite error must stay non-transient.
+    use super::is_transient_cold_start;
+    let fstat = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error::new(1802),
+        Some("disk I/O error".to_string()),
+    );
+    assert!(is_transient_cold_start(&anyhow::Error::from(fstat)));
+
+    let constraint = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error::new(19),
+        Some("constraint".to_string()),
+    );
+    assert!(!is_transient_cold_start(&anyhow::Error::from(constraint)));
+}