From e0a8738980965411f514f4a62c09f941efdea90c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 28 Jul 2026 11:43:55 +0300 Subject: [PATCH 1/4] fix(memory): guard chunk tag rewrites --- src/memory/store/content/tags.rs | 28 +++++++++++++++++++++++ src/memory/store/content/tags_tests.rs | 31 ++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/src/memory/store/content/tags.rs b/src/memory/store/content/tags.rs index 0a171dc..1f15ad9 100644 --- a/src/memory/store/content/tags.rs +++ b/src/memory/store/content/tags.rs @@ -35,6 +35,8 @@ pub fn update_chunk_tags(abs_path: &Path, tags: &[String]) -> anyhow::Result<()> let new_bytes = rewrite_tags(&old_bytes, &augmented) .map_err(|e| anyhow::anyhow!("rewrite_tags {:?}: {e}", abs_path))?; + ensure_tag_rewrite_preserves_body(&old_bytes, &new_bytes, abs_path)?; + let parent = abs_path.parent().unwrap_or_else(|| Path::new(".")); let tmp_name = format!(".tmp_tags_{}.md", crate_temp_id()); let tmp_path = parent.join(&tmp_name); @@ -57,6 +59,32 @@ pub fn update_chunk_tags(abs_path: &Path, tags: &[String]) -> anyhow::Result<()> Ok(()) } +/// Reject a front-matter rewrite unless both files parse and their bodies are +/// byte-identical. +/// +/// The body hash is persisted separately from the markdown file, so silently +/// accepting an invalid or changed body would corrupt later retrieval. +fn ensure_tag_rewrite_preserves_body( + old_bytes: &[u8], + new_bytes: &[u8], + abs_path: &Path, +) -> anyhow::Result<()> { + let body = |bytes: &[u8]| -> Option { + std::str::from_utf8(bytes) + .ok() + .and_then(split_front_matter) + .map(|(_, body)| body.to_owned()) + }; + + match (body(old_bytes), body(new_bytes)) { + (Some(old), Some(new)) if old == new => Ok(()), + _ => Err(anyhow::anyhow!( + "tag rewrite would mutate or invalidate the body for {:?}", + abs_path + )), + } +} + /// Slugify an entity kind string for an Obsidian hierarchical tag. /// /// Output: lowercase, non-alphanumeric → `-`, collapsed, trimmed. diff --git a/src/memory/store/content/tags_tests.rs b/src/memory/store/content/tags_tests.rs index cd8bec1..1cac1e5 100644 --- a/src/memory/store/content/tags_tests.rs +++ b/src/memory/store/content/tags_tests.rs @@ -88,6 +88,37 @@ fn update_chunk_tags_is_noop_for_missing_file() { assert!(update_chunk_tags(&path, &["p/X".into()]).is_ok()); } +#[test] +fn update_chunk_tags_rejects_unparseable_front_matter_without_overwriting() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("malformed.md"); + let original = b"not front matter\nBODY"; + std::fs::write(&path, original).unwrap(); + + let result = update_chunk_tags(&path, &["person/Alice".into()]); + + assert!(result.is_err()); + assert_eq!(std::fs::read(&path).unwrap(), original); +} + +#[test] +fn body_guard_accepts_front_matter_only_changes() { + let path = Path::new("chunk.md"); + let old = b"---\ntags: []\n---\nBODY"; + let new = b"---\ntags:\n - person/Alice\n---\nBODY"; + + assert!(ensure_tag_rewrite_preserves_body(old, new, path).is_ok()); +} + +#[test] +fn body_guard_rejects_body_drift() { + let path = Path::new("chunk.md"); + let old = b"---\ntags: []\n---\nBODY"; + let new = b"---\ntags:\n - person/Alice\n---\nDIFFERENT"; + + assert!(ensure_tag_rewrite_preserves_body(old, new, path).is_err()); +} + #[test] fn slugify_tag_kind_examples() { assert_eq!(slugify_tag_kind("Person"), "person"); From 5fabcf18d9e3907d6b26b59528ad49cebfc1c271 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 6 Aug 2026 17:44:00 +0530 Subject: [PATCH 2/4] fix(queue): requeue failed jobs per row so a dedupe collision cannot abort the batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `requeue_failed_where` flipped every matching `failed` row to `ready` in one UPDATE. `idx_mem_tree_jobs_dedupe_active` is a partial unique index over `dedupe_key` covering only `ready`/`running`, so failed rows sit outside it and the same key can legitimately accumulate several failed rows over time. The moment that UPDATE moved two siblings into the index at once they collided, SQLite aborted the whole statement, and nothing was requeued. Observed in production as `UNIQUE constraint failed: mem_tree_jobs.dedupe_key` from the periodic self-heal on every boot and every 3h tick, with the queue permanently parked. The manual retry path shares the same helper and aborted identically, so there was no way out from either side. Select the candidates first, then apply per row: - skip a key already held by a live `ready`/`running` row (that work is already in flight, and requeueing onto it is the collision); - among matched failed rows sharing a key, requeue only the newest and settle its older siblings as `cancelled` — they are duplicates of the same unit of work, and leaving them `failed` would keep them counted as failures the user must act on; - rows with a NULL `dedupe_key` are outside the index and always requeue. Selection and mutation share one transaction so a concurrent claim cannot insert an active row between the read and the write. The return value still counts only rows actually flipped to `ready`. --- src/memory/queue/store_settle.rs | 139 ++++++++++++++++++++++--- src/memory/queue/store_settle_tests.rs | 123 ++++++++++++++++++++++ 2 files changed, 248 insertions(+), 14 deletions(-) diff --git a/src/memory/queue/store_settle.rs b/src/memory/queue/store_settle.rs index 896d8fe..ca11f48 100644 --- a/src/memory/queue/store_settle.rs +++ b/src/memory/queue/store_settle.rs @@ -255,24 +255,135 @@ pub fn retry_all_failed(config: &MemoryConfig) -> Result { requeue_failed(config) } +/// Shared requeue worker for [`requeue_failed`] / [`requeue_transient_failed`]. +/// +/// # Why this is not one blanket `UPDATE` +/// +/// `idx_mem_tree_jobs_dedupe_active` is a **partial unique index** over +/// `dedupe_key` restricted to `status IN ('ready', 'running')`. Failed rows sit +/// outside it, so the same `dedupe_key` can legitimately accumulate several +/// `failed` rows over time (the same seal / re-embed unit of work failing on +/// successive attempts). The moment a single `UPDATE` flips two such siblings to +/// `ready` they both enter the index and collide, SQLite aborts the **whole** +/// statement, and **nothing** is requeued. +/// +/// That is not hypothetical: it is what the periodic self-heal hit in +/// production, logging `UNIQUE constraint failed: mem_tree_jobs.dedupe_key` +/// every three hours and on every boot while the queue stayed permanently +/// parked — and the manual "retry failed" path aborted the same way, so the +/// user had no way out either. +/// +/// So the requeue is filtered before it is applied: +/// +/// - a key that **already has an active row** (`ready`/`running`) is skipped +/// entirely — live work already covers it, and requeueing would collide; +/// - among matched failed rows **sharing a key**, only the newest is requeued; +/// its siblings are settled as `cancelled` (superseded duplicates of the same +/// unit of work) so they stop being counted as failures the user must act on. +/// - rows with `dedupe_key IS NULL` are outside the index and always requeue. +/// +/// Selection and mutation run in one transaction, so a concurrent claim can't +/// slip an active row in between the read and the write. +/// +/// Returns the number of jobs actually flipped back to `ready` (cancelled +/// duplicates are not counted — they were not requeued). fn requeue_failed_where(config: &MemoryConfig, predicate: &str) -> Result { with_connection(config, |conn| { let now_ms = Utc::now().timestamp_millis(); - let sql = format!( - "UPDATE mem_tree_jobs - SET status = 'ready', - attempts = 0, - available_at_ms = ?1, - locked_until_ms = NULL, - started_at_ms = NULL, - completed_at_ms = NULL, - last_error = NULL, - failure_reason = NULL, - failure_class = NULL - WHERE {predicate}" + let tx = conn.unchecked_transaction()?; + + // Newest-first so the first row seen for a key is the one to keep. + // `completed_at_ms` is stamped on failure; fall back to `created_at_ms` + // for legacy rows that predate the column. + let select = format!( + "SELECT id, dedupe_key FROM mem_tree_jobs + WHERE {predicate} + ORDER BY COALESCE(completed_at_ms, created_at_ms) DESC, id DESC" ); - let n = conn.execute(&sql, params![now_ms])?; - Ok(n as u64) + let candidates: Vec<(String, Option)> = { + let mut stmt = tx.prepare(&select)?; + let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?; + rows.collect::>>()? + }; + if candidates.is_empty() { + return Ok(0); + } + + // Keys already held by a live row: requeueing onto one of these is the + // collision case, and the live row is already doing the work. + let active_keys: std::collections::HashSet = { + let mut stmt = tx.prepare( + "SELECT DISTINCT dedupe_key FROM mem_tree_jobs + WHERE dedupe_key IS NOT NULL AND status IN ('ready', 'running')", + )?; + let rows = stmt.query_map([], |r| r.get::<_, String>(0))?; + rows.collect::>>()? + }; + + let mut claimed_keys: std::collections::HashSet = std::collections::HashSet::new(); + let mut to_requeue: Vec = Vec::new(); + let mut to_cancel: Vec = Vec::new(); + let mut skipped_active = 0_usize; + + for (id, dedupe_key) in candidates { + match dedupe_key { + // Outside the partial unique index — never collides. + None => to_requeue.push(id), + Some(key) if active_keys.contains(&key) => skipped_active += 1, + // First (newest) row for this key wins the requeue; every later + // sibling is the same unit of work, so settle it rather than + // leaving it counted as a failure the user must act on. + Some(key) => { + if claimed_keys.insert(key) { + to_requeue.push(id); + } else { + to_cancel.push(id); + } + } + } + } + + for id in &to_cancel { + tx.execute( + "UPDATE mem_tree_jobs + SET status = 'cancelled', + locked_until_ms = NULL, + completed_at_ms = ?2, + last_error = 'superseded by a newer job with the same dedupe_key' + WHERE id = ?1", + params![id, now_ms], + )?; + } + + let mut requeued = 0_u64; + for id in &to_requeue { + requeued += tx.execute( + "UPDATE mem_tree_jobs + SET status = 'ready', + attempts = 0, + available_at_ms = ?2, + locked_until_ms = NULL, + started_at_ms = NULL, + completed_at_ms = NULL, + last_error = NULL, + failure_reason = NULL, + failure_class = NULL + WHERE id = ?1", + params![id, now_ms], + )? as u64; + } + + tx.commit()?; + + if !to_cancel.is_empty() || skipped_active > 0 { + log::debug!( + "[queue::requeue] action=requeue_failed requeued={requeued} \ + cancelled_duplicates={} skipped_active_key={skipped_active}", + to_cancel.len() + ); + } + + Ok(requeued) }) } diff --git a/src/memory/queue/store_settle_tests.rs b/src/memory/queue/store_settle_tests.rs index a022923..dedbe8f 100644 --- a/src/memory/queue/store_settle_tests.rs +++ b/src/memory/queue/store_settle_tests.rs @@ -246,6 +246,129 @@ fn requeue_transient_failed_skips_unrecoverable_jobs() { assert_eq!(row_b.failure_class.as_deref(), Some("unrecoverable")); } +/// Fail the same unit of work twice so two `failed` rows end up sharing one +/// `dedupe_key`. Legal on disk: `idx_mem_tree_jobs_dedupe_active` only covers +/// `ready`/`running`, so a failed row leaves the key free for a re-enqueue. +/// Returns `(older_id, newer_id)`. +fn two_failed_rows_sharing_a_dedupe_key(cfg: &MemoryConfig, chunk: &str) -> (String, String) { + let mut ids = Vec::new(); + for _ in 0..2 { + let id = enqueue(cfg, &extract_job(chunk, 1)) + .unwrap() + .expect("key is free while the previous attempt sits in `failed`"); + let claimed = claim_next(cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + mark_failed_typed( + cfg, + &claimed, + "No backend session for cloud embeddings", + Some(&JobFailure::unrecoverable("auth_missing")), + ) + .unwrap(); + ids.push(id); + } + (ids[0].clone(), ids[1].clone()) +} + +/// The production defect: a single blanket `UPDATE … SET status='ready'` over +/// two failed rows sharing a `dedupe_key` puts both into the partial unique +/// index at once, SQLite aborts the whole statement, and NOTHING is requeued — +/// forever, on every boot and every 3h self-heal tick. +/// +/// The requeue must succeed and requeue the newest row for the key. +#[test] +fn requeue_failed_survives_duplicate_dedupe_keys() { + let (_tmp, cfg) = test_config(); + let (older, newer) = two_failed_rows_sharing_a_dedupe_key(&cfg, "c-dup"); + assert_eq!(count_by_status(&cfg, JobStatus::Failed).unwrap(), 2); + + let requeued = requeue_failed(&cfg).expect("requeue must not abort on a dedupe collision"); + + assert_eq!(requeued, 1, "exactly one row per dedupe_key may go ready"); + assert_eq!( + get_job(&cfg, &newer).unwrap().unwrap().status, + JobStatus::Ready, + "the newest attempt is the one that retries" + ); + assert_eq!( + get_job(&cfg, &older).unwrap().unwrap().status, + JobStatus::Cancelled, + "the superseded duplicate must settle, not stay counted as failed" + ); + assert_eq!( + count_by_status(&cfg, JobStatus::Failed).unwrap(), + 0, + "no failure may survive a manual retry — that is what pins the panel red" + ); +} + +/// Same collision on the automatic self-heal path (the one that logged +/// `UNIQUE constraint failed: mem_tree_jobs.dedupe_key` on every boot). +#[test] +fn requeue_transient_failed_survives_duplicate_dedupe_keys() { + let (_tmp, cfg) = test_config(); + let mut ids = Vec::new(); + for _ in 0..2 { + let id = enqueue(&cfg, &extract_job("c-dup-transient", 1)) + .unwrap() + .expect("inserted"); + let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + mark_failed(&cfg, &claimed, "connection reset by peer").unwrap(); + ids.push(id); + } + + let requeued = requeue_transient_failed(&cfg).expect("self-heal must not abort"); + + assert_eq!(requeued, 1); + assert_eq!( + get_job(&cfg, &ids[1]).unwrap().unwrap().status, + JobStatus::Ready + ); + assert_eq!( + get_job(&cfg, &ids[0]).unwrap().unwrap().status, + JobStatus::Cancelled + ); +} + +/// A failed row whose `dedupe_key` is already held by a live `ready` row must +/// be left alone: requeueing it would collide, and the live row is already +/// doing that work. It settles as a superseded duplicate rather than blocking +/// the rest of the batch. +#[test] +fn requeue_failed_skips_keys_already_held_by_an_active_row() { + let (_tmp, cfg) = test_config(); + + // An unrelated failed row, which must still requeue in the same batch. + let other_id = enqueue(&cfg, &extract_job("c-other", 1)).unwrap().unwrap(); + let claimed_other = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + mark_failed(&cfg, &claimed_other, "boom").unwrap(); + + // A failed attempt at `c-live`... + let failed_id = enqueue(&cfg, &extract_job("c-live", 1)).unwrap().unwrap(); + let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + mark_failed(&cfg, &claimed, "boom").unwrap(); + // ...and a fresh live row that re-took the same key. Enqueued last so no + // `claim_next` above can consume it. + let live_id = enqueue(&cfg, &extract_job("c-live", 5)).unwrap().unwrap(); + + let requeued = requeue_failed(&cfg).unwrap(); + + assert_eq!(requeued, 1, "only the unrelated row requeues"); + assert_eq!( + get_job(&cfg, &other_id).unwrap().unwrap().status, + JobStatus::Ready + ); + assert_eq!( + get_job(&cfg, &live_id).unwrap().unwrap().status, + JobStatus::Ready, + "the live row keeps its claim on the key, untouched" + ); + assert_ne!( + get_job(&cfg, &failed_id).unwrap().unwrap().status, + JobStatus::Ready, + "requeueing onto an occupied key is exactly the collision to avoid" + ); +} + #[test] fn recover_stale_locks_resets_running_rows() { let (_tmp, cfg) = test_config(); From 03e6de7f69861f67d5740d72382b1012ff79b467 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 6 Aug 2026 20:40:28 +0530 Subject: [PATCH 3/4] fix(queue): deterministic requeue tiebreaker + accurate skip semantics (PR #139 review) - ORDER BY breaks exact completed_at_ms ties by created_at_ms then id, so the "keep newest per dedupe_key" pick is reproducible instead of hinging on a random UUID. Rows sharing a dedupe_key are the same unit of work, so the pick is immaterial to correctness -- no monotonic-sequence migration needed. - Correct the skip-active-key test doc (the occupied-key row stays failed, skipped, not settled) and assert JobStatus::Failed exactly. - Add a test for equal-millisecond completions asserting the one-Ready / one-Cancelled invariant holds regardless of tiebreak. - Refresh the JobStatus::Cancelled doc: requeue supersession is its first real producer. Co-Authored-By: Claude Opus 4.8 --- src/memory/queue/store_settle.rs | 9 ++++- src/memory/queue/store_settle_tests.rs | 53 +++++++++++++++++++++++--- src/memory/queue/types.rs | 10 +++-- 3 files changed, 61 insertions(+), 11 deletions(-) diff --git a/src/memory/queue/store_settle.rs b/src/memory/queue/store_settle.rs index ca11f48..ea39c0b 100644 --- a/src/memory/queue/store_settle.rs +++ b/src/memory/queue/store_settle.rs @@ -294,11 +294,16 @@ fn requeue_failed_where(config: &MemoryConfig, predicate: &str) -> Result { // Newest-first so the first row seen for a key is the one to keep. // `completed_at_ms` is stamped on failure; fall back to `created_at_ms` - // for legacy rows that predate the column. + // for legacy rows that predate the column. `created_at_ms` then `id` + // break exact-timestamp ties by enqueue order — job ids are random + // UUIDs, so `id` alone carries no ordering signal. For rows sharing a + // `dedupe_key` the pick is immaterial to correctness anyway (they are + // the same unit of work), but a stable order keeps the choice + // reproducible. let select = format!( "SELECT id, dedupe_key FROM mem_tree_jobs WHERE {predicate} - ORDER BY COALESCE(completed_at_ms, created_at_ms) DESC, id DESC" + ORDER BY COALESCE(completed_at_ms, created_at_ms) DESC, created_at_ms DESC, id DESC" ); let candidates: Vec<(String, Option)> = { let mut stmt = tx.prepare(&select)?; diff --git a/src/memory/queue/store_settle_tests.rs b/src/memory/queue/store_settle_tests.rs index dedbe8f..9d2aa47 100644 --- a/src/memory/queue/store_settle_tests.rs +++ b/src/memory/queue/store_settle_tests.rs @@ -331,8 +331,8 @@ fn requeue_transient_failed_survives_duplicate_dedupe_keys() { /// A failed row whose `dedupe_key` is already held by a live `ready` row must /// be left alone: requeueing it would collide, and the live row is already -/// doing that work. It settles as a superseded duplicate rather than blocking -/// the rest of the batch. +/// doing that work. It stays in `failed` — skipped, not requeued and not +/// cancelled — so it neither collides nor blocks the rest of the batch. #[test] fn requeue_failed_skips_keys_already_held_by_an_active_row() { let (_tmp, cfg) = test_config(); @@ -362,10 +362,53 @@ fn requeue_failed_skips_keys_already_held_by_an_active_row() { JobStatus::Ready, "the live row keeps its claim on the key, untouched" ); - assert_ne!( + assert_eq!( get_job(&cfg, &failed_id).unwrap().unwrap().status, - JobStatus::Ready, - "requeueing onto an occupied key is exactly the collision to avoid" + JobStatus::Failed, + "the occupied-key row is skipped and left in `failed` (requeueing onto \ + an occupied key is exactly the collision to avoid), not cancelled" + ); +} + +/// Two failed rows sharing a `dedupe_key` AND the exact same `completed_at_ms`. +/// Job ids are random UUIDs, so the millisecond tie falls to the +/// `created_at_ms`/`id` tiebreaker. Whichever row wins, the invariant that +/// matters must hold: exactly one row is requeued and the other settled — never +/// both requeued (the UNIQUE-index collision) and never zero. +#[test] +fn requeue_failed_is_deterministic_on_equal_millisecond_completions() { + let (_tmp, cfg) = test_config(); + let (a, b) = two_failed_rows_sharing_a_dedupe_key(&cfg, "c-tie"); + + // Force identical completion timestamps so ordering falls to the tiebreaker. + let same_ms = 1_800_000_000_000_i64; + with_connection(&cfg, |conn| { + conn.execute( + "UPDATE mem_tree_jobs SET completed_at_ms = ?1 WHERE id IN (?2, ?3)", + params![same_ms, a, b], + )?; + Ok(()) + }) + .unwrap(); + + let requeued = requeue_failed(&cfg).expect("requeue must not abort on a millisecond tie"); + + assert_eq!( + requeued, 1, + "exactly one row per key requeues, even on a tie" + ); + assert_eq!( + count_by_status(&cfg, JobStatus::Failed).unwrap(), + 0, + "no failure may survive the retry" + ); + let statuses = [ + get_job(&cfg, &a).unwrap().unwrap().status, + get_job(&cfg, &b).unwrap().unwrap().status, + ]; + assert!( + statuses.contains(&JobStatus::Ready) && statuses.contains(&JobStatus::Cancelled), + "exactly one Ready + one Cancelled regardless of which won the tie, got {statuses:?}" ); } diff --git a/src/memory/queue/types.rs b/src/memory/queue/types.rs index ca07356..2c04053 100644 --- a/src/memory/queue/types.rs +++ b/src/memory/queue/types.rs @@ -119,8 +119,9 @@ pub enum JobOutcome { } /// Lifecycle states persisted on `mem_tree_jobs.status`. Workers transition -/// `ready → running → done|failed`. `Cancelled` is reserved for explicit admin -/// actions (none surfaced yet). +/// `ready → running → done|failed`. `Cancelled` settles a job outside the +/// worker path — currently a failed row superseded by a newer one with the same +/// `dedupe_key` during requeue (see `requeue_failed_where`). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum JobStatus { /// Claimable: waiting for a worker (wire string `ready`). @@ -132,8 +133,9 @@ pub enum JobStatus { /// Settled as failed after exhausting retries or on an unrecoverable /// classification (wire string `failed`). Failed, - /// Cancelled by explicit admin action (wire string `cancelled`); reserved, - /// no producer yet. + /// Settled without running to completion (wire string `cancelled`) — e.g. a + /// failed row superseded by a newer duplicate of the same `dedupe_key` when + /// the queue is requeued. Excluded from the `failed` counters. Cancelled, } From 4cad022df61b983bcddb45b3158042d4ff844108 Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 7 Aug 2026 11:59:05 +0530 Subject: [PATCH 4/4] ci: re-trigger CI after GitHub Actions outage (no code change) The 03e6de7 CI run failed only on GitHub's action-resolution outage ("Service Unavailable" resolving actions/checkout, rust-toolchain, rust-cache); no job compiled. GitHub Actions is Operational again but dropped events can't replay, so this empty commit re-triggers the pipeline. Co-Authored-By: Claude Opus 4.8