From 77ad3439c5cf20993b67368677df8dd0dbd758b1 Mon Sep 17 00:00:00 2001 From: Shanu Date: Sun, 23 Aug 2026 21:59:13 +0530 Subject: [PATCH 1/2] Give the queue a retry door, and say when a backfill is still running MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two things OpenHuman still reaches into `tinymemory_core::queue` for, and both are the last blocker on the files that hold them. `retry_failed` requeues parked work and wakes the pool, as one operation. Both host call sites already did exactly that pair — the "retry failed" control calls `store::requeue_failed` then `wake_workers`, and the post-provider-change path calls the engine helper that does the same — so splitting them across two contract methods would offer a caller a way to requeue and forget the wake. Rows moved back to `ready` and left there until the next scheduled window are indistinguishable, from the user's side, from a retry that did not run. It answers a `MaintenanceReport` like its four siblings rather than a bare count, because the caller's question is the one that shape already answers: how many, out of how many looked at, and a line to show. The wake is conditional on something having moved; waking a pool to process nothing is a spurious wake-up on an idle laptop. `QueueStats` gains `backfill_in_progress`, which is not derivable from the counts it sits beside. A backfill is a chain that re-enqueues itself, so between one link settling and the next being written there is an instant with nothing ready, nothing running, and the backfill not finished. The consumer is absence-reasoning — deciding whether an empty semantic-recall result means "no memory" or "not embedded yet" — and it gets that wrong at exactly that instant without this. Registered in all four lists this time: the interface block, the manifest, `tinymemory_bus::METHODS` with its length, and the loader's `EXPECTED_METHODS`. The last change taught that the compiler checks only the array length, so the loader E2E is the thing that proves it: `every_declared_method_is_actually_routed` passes against the built cdylib. --- crates/tinymemory-api/src/provider/records.rs | 23 ++++++++ crates/tinymemory-bus/src/lib.rs | 2 +- crates/tinymemory-bus/src/names.rs | 5 +- crates/tinymemory-bus/src/provider/types.rs | 10 ++++ crates/tinymemory-module/src/lib.rs | 1 + crates/tinymemory-module/src/service/mod.rs | 7 +++ crates/tinymemory-module/src/service/test.rs | 2 +- crates/tinymemory-module/tests/module_e2e.rs | 1 + .../tinymemory-tinycortex/src/engine/mod.rs | 30 ++++++++++ .../tests/full_provider_conformance.rs | 58 +++++++++++++++++++ 10 files changed, 136 insertions(+), 3 deletions(-) diff --git a/crates/tinymemory-api/src/provider/records.rs b/crates/tinymemory-api/src/provider/records.rs index 467f182..80dc064 100644 --- a/crates/tinymemory-api/src/provider/records.rs +++ b/crates/tinymemory-api/src/provider/records.rs @@ -181,6 +181,29 @@ pub trait MemoryMaintenance: Send + Sync { Ok(StoreStats::default()) } + /// Give terminally-failed queue work another attempt, and nudge whatever + /// drains the queue. + /// + /// The nudge is part of the operation, not a separate call. A driver that + /// requeues without waking has moved rows from `failed` back to `ready` + /// and left them there until the next scheduled window, which looks + /// identical to a retry that did not work — and the caller has no way to + /// ask for the wake on its own. + /// + /// What counts as retryable is the driver's judgement. A failure it will + /// never recover from is one it should leave parked; the caller is asking + /// for another attempt, not asserting that one can succeed. + /// + /// Defaulted to an empty report — a driver with no queue has nothing to + /// retry, which is true of it rather than a refusal. + /// + /// # Errors + /// + /// Backend failures only. + async fn retry_failed(&self) -> Result { + Ok(MaintenanceReport::default()) + } + /// The ingest and re-embed queue's state. /// /// `kind` narrows to one job kind (the driver's own identifier); `None` diff --git a/crates/tinymemory-bus/src/lib.rs b/crates/tinymemory-bus/src/lib.rs index f74bf11..8c1ec93 100644 --- a/crates/tinymemory-bus/src/lib.rs +++ b/crates/tinymemory-bus/src/lib.rs @@ -2,7 +2,7 @@ //! the members that carry them. //! //! TinyMemory ships as a loadable `TinyBus` module: `crates/tinymemory-module` -//! exports one object with 92 members on it, built as a `cdylib`. A host that +//! exports one object with 93 members on it, built as a `cdylib`. A host that //! loads it — OpenHuman — can call into it but cannot `use` anything out of it, //! so the payload vocabulary has to be published as an ordinary library. This //! is that library. diff --git a/crates/tinymemory-bus/src/names.rs b/crates/tinymemory-bus/src/names.rs index a3f5f36..ea13c0a 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -150,6 +150,8 @@ pub mod methods { pub const CONSOLIDATE: &str = "Consolidate"; /// `Doctor` — doctor. pub const DOCTOR: &str = "Doctor"; + /// `RetryFailed` — give terminally-failed queue work another attempt. + pub const RETRY_FAILED: &str = "RetryFailed"; /// `StoreStats` — aggregate counts over what the driver has stored. pub const STORE_STATS: &str = "StoreStats"; /// `QueueStats` — the ingest and re-embed queue's state. @@ -245,7 +247,7 @@ pub mod methods { /// The order matters: `tinybus`'s `Interface::members()` returns declaration /// order, and the module compares the two sequences directly rather than as /// sets, so a reordering is caught alongside an addition or a removal. -pub const METHODS: [&str; 92] = [ +pub const METHODS: [&str; 93] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -297,6 +299,7 @@ pub const METHODS: [&str; 92] = [ methods::COMPACT, methods::CONSOLIDATE, methods::DOCTOR, + methods::RETRY_FAILED, methods::STORE_STATS, methods::QUEUE_STATS, methods::LATEST_QUEUE_FAILURE, diff --git a/crates/tinymemory-bus/src/provider/types.rs b/crates/tinymemory-bus/src/provider/types.rs index ab8a6b4..d0553bf 100644 --- a/crates/tinymemory-bus/src/provider/types.rs +++ b/crates/tinymemory-bus/src/provider/types.rs @@ -449,6 +449,16 @@ pub struct QueueStats { /// When the queue last settled a job. #[serde(default, skip_serializing_if = "Option::is_none")] pub last_completed_ms: Option, + /// Whether a re-embedding backfill chain is still working through its + /// rows. + /// + /// Not derivable from the counts. A backfill runs as a chain that + /// re-enqueues itself, so between one link finishing and the next being + /// written there is an instant where nothing is ready, nothing is running, + /// and the backfill is nevertheless not done. A caller deciding whether to + /// warn that recall is degraded has to know the difference. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub backfill_in_progress: bool, /// The scheduled time of the oldest job eligible to run now. /// /// With [`Self::last_completed_ms`] this is what an idle-time calculation diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index d7a8722..9e5647e 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -307,6 +307,7 @@ mod exports { "Compact", "Consolidate", "Doctor", + "RetryFailed", "StoreStats", "QueueStats", "LatestQueueFailure", diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 71370bd..420bc2d 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -887,6 +887,13 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + async fn retry_failed(&self) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .retry_failed() + .await + .map_err(|error| into_bus_error(&error)) + } + async fn store_stats(&self) -> BusResult { require_family!(self, as_maintenance, Capability::Maintenance) .store_stats() diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index 9d918bd..0fc1818 100644 --- a/crates/tinymemory-module/src/service/test.rs +++ b/crates/tinymemory-module/src/service/test.rs @@ -544,7 +544,7 @@ fn the_served_members_are_exactly_the_published_contract() { .map(|member| (*member).to_string()) .collect(); - // Reported as differences rather than as a 92-element inequality, so the + // Reported as differences rather than as a 93-element inequality, so the // failure names the method that moved instead of printing both lists. let missing: Vec<&String> = served.iter().filter(|m| !published.contains(m)).collect(); assert!( diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index cdfd4ed..fc32dba 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -671,6 +671,7 @@ const EXPECTED_METHODS: &[&str] = &[ "Compact", "Consolidate", "Doctor", + "RetryFailed", "StoreStats", "QueueStats", "LatestQueueFailure", diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 47196e4..c9fdfb2 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -1158,6 +1158,31 @@ impl MemoryMaintenance for TinycortexProvider { }) } + async fn retry_failed(&self) -> Result { + let (examined, changed) = blocking( + self.config.clone(), + "retry failed queue work", + move |config| { + let examined = tinymemory_core::queue::count_total(config).unwrap_or(0); + let changed = tinymemory_core::queue::store::requeue_failed(config)?; + // Inside the same call, and only when something moved: rows + // put back on `ready` sit until the next scheduled window + // otherwise, which reads as a retry that did nothing. + if changed > 0 { + tinymemory_core::queue::wake_workers(); + } + Ok((examined, changed)) + }, + ) + .await?; + Ok(MaintenanceReport { + operation: "retry_failed".to_string(), + examined, + changed, + findings: vec![format!("requeued {changed} failed job(s)")], + }) + } + async fn store_stats(&self) -> Result { blocking(self.config.clone(), "read store stats", move |config| { tinymemory_core::store::chunks::store::with_connection(config, |conn| { @@ -1250,7 +1275,12 @@ impl MemoryMaintenance for TinycortexProvider { }, )?; let count = |n: i64| u64::try_from(n).unwrap_or(0); + // A process-global the backfill chain owns, not a column — + // read here so it arrives with the counts it has to be + // interpreted next to. + let backfill_in_progress = tinymemory_core::queue::backfill_in_progress(); Ok(QueueStats { + backfill_in_progress, ready: count(ready), running: count(running), done: count(done), diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index 773b82e..d3cf0f9 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -277,6 +277,64 @@ async fn deferred_work_stays_ready_without_becoming_eligible() { ); } +/// Retrying moves parked work back to ready, and says how much it moved. +/// +/// The count is the point. A caller offering the user a "retry failed" control +/// has to tell them whether anything happened, and `0` from a queue with +/// nothing parked has to be distinguishable from a retry that silently did +/// not run — which is what the direct call this replaces gave them, since it +/// returned a count the host then had to pair with a separate wake. +#[tokio::test] +async fn retrying_moves_parked_work_back_to_ready_and_counts_it() { + use tinymemory_api::provider::MemoryMaintenance; + use tinymemory_core::queue::store as queue_store; + use tinymemory_core::queue::types::{FlushStalePayload, NewJob}; + use tinymemory_core::tree::health::{FailureCode, PipelineFailure}; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + let config = provider_config(workspace.path(), serde_json::Value::Null); + + // Nothing parked: a retry is honest about having moved nothing rather than + // refusing or inventing a number. + let report = provider.retry_failed().await.expect("retry"); + assert_eq!(report.operation, "retry_failed"); + assert_eq!(report.changed, 0, "an empty queue has nothing to requeue"); + + let new_job = + NewJob::flush_stale(&FlushStalePayload::default(), "2026-08-23", 1).expect("build a job"); + let id = queue_store::enqueue(&config, &new_job) + .expect("enqueue") + .expect("a fresh job is not a duplicate"); + let job = queue_store::get_job(&config, &id) + .expect("read the job") + .expect("the job exists"); + queue_store::mark_failed_typed( + &config, + &job, + "parked for the retry test", + Some(&PipelineFailure::new(FailureCode::BudgetExhausted)), + ) + .expect("park the job"); + + let before = provider.queue_stats(None).await.expect("queue stats"); + assert_eq!(before.failed, 1, "precondition: one job is parked"); + assert_eq!(before.ready, 0); + + let report = provider.retry_failed().await.expect("retry"); + assert_eq!( + report.changed, 1, + "the parked job was given another attempt, and the caller is told so" + ); + + let after = provider.queue_stats(None).await.expect("queue stats"); + assert_eq!(after.failed, 0, "nothing is parked any more"); + assert_eq!( + after.ready, 1, + "and the job is queued again rather than lost" + ); +} + /// A reported failure carries the success watermark that decides whether it is /// still worth showing. /// From f0de879513573de1af453957f69518b34f7936c1 Mon Sep 17 00:00:00 2001 From: Shanu Date: Sun, 23 Aug 2026 22:55:30 +0530 Subject: [PATCH 2/2] Drop backfill_in_progress rather than promise a scope it does not have MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review caught that `backfill_in_progress()` is a process-global with no workspace argument, while `open_store` serves one store per memory subtree in a single process. A profile with dedicated memory running a backfill would make every other store answer `true`. The bleed is not new — the host already called that same global directly — but putting it on `QueueStats` made it worse in a way worth naming. `queue_stats` is asked of one bound provider for one store, so a field on it reads as per-store. A global behind a per-store API is harder to notice than a global that looks like one, and the caller has no way to find out. Scoping it properly means tracking the state per workspace where it is set, which is in tinycortex, behind a vendored submodule. That is its own change in its own repo, and doing it badly here would be worse than not doing it. So the field goes and the host keeps calling the global, visibly, until the engine can answer per store. `retry_failed` is unaffected — it takes no such shortcut, and it is what the two OpenHuman files were actually waiting on. --- crates/tinymemory-bus/src/provider/types.rs | 10 ---------- crates/tinymemory-tinycortex/src/engine/mod.rs | 5 ----- 2 files changed, 15 deletions(-) diff --git a/crates/tinymemory-bus/src/provider/types.rs b/crates/tinymemory-bus/src/provider/types.rs index d0553bf..ab8a6b4 100644 --- a/crates/tinymemory-bus/src/provider/types.rs +++ b/crates/tinymemory-bus/src/provider/types.rs @@ -449,16 +449,6 @@ pub struct QueueStats { /// When the queue last settled a job. #[serde(default, skip_serializing_if = "Option::is_none")] pub last_completed_ms: Option, - /// Whether a re-embedding backfill chain is still working through its - /// rows. - /// - /// Not derivable from the counts. A backfill runs as a chain that - /// re-enqueues itself, so between one link finishing and the next being - /// written there is an instant where nothing is ready, nothing is running, - /// and the backfill is nevertheless not done. A caller deciding whether to - /// warn that recall is degraded has to know the difference. - #[serde(default, skip_serializing_if = "std::ops::Not::not")] - pub backfill_in_progress: bool, /// The scheduled time of the oldest job eligible to run now. /// /// With [`Self::last_completed_ms`] this is what an idle-time calculation diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index c9fdfb2..5af3f48 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -1275,12 +1275,7 @@ impl MemoryMaintenance for TinycortexProvider { }, )?; let count = |n: i64| u64::try_from(n).unwrap_or(0); - // A process-global the backfill chain owns, not a column — - // read here so it arrives with the counts it has to be - // interpreted next to. - let backfill_in_progress = tinymemory_core::queue::backfill_in_progress(); Ok(QueueStats { - backfill_in_progress, ready: count(ready), running: count(running), done: count(done),