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-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..5af3f48 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| { 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. ///