Give the queue a retry door, and say when a backfill is still running

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<MaintenanceReport, MemoryError> {
+ Ok(MaintenanceReport::default())
+ }
+
  /// The ingest and re-embed queue's state.
  ///
  /// `kind` narrows to one job kind (the driver's own identifier); `None`

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.

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,

crates/tinymemory-module/src/lib.rs

@@ -307,6 +307,7 @@ mod exports {
  "Compact",
  "Consolidate",
  "Doctor",
+ "RetryFailed",
  "StoreStats",
  "QueueStats",
  "LatestQueueFailure",

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<MaintenanceReport> {
+ require_family!(self, as_maintenance, Capability::Maintenance)
+ .retry_failed()
+ .await
+ .map_err(|error| into_bus_error(&error))
+ }
+
  async fn store_stats(&self) -> BusResult<StoreStats> {
  require_family!(self, as_maintenance, Capability::Maintenance)
  .store_stats()

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!(

crates/tinymemory-module/tests/module_e2e.rs

@@ -671,6 +671,7 @@ const EXPECTED_METHODS: &[&str] = &[
  "Compact",
  "Consolidate",
  "Doctor",
+ "RetryFailed",
  "StoreStats",
  "QueueStats",
  "LatestQueueFailure",

crates/tinymemory-tinycortex/src/engine/mod.rs

@@ -1158,6 +1158,31 @@ impl MemoryMaintenance for TinycortexProvider {
  })
  }
 
+ async fn retry_failed(&self) -> Result<MaintenanceReport, MemoryError> {
+ 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<StoreStats, MemoryError> {
  blocking(self.config.clone(), "read store stats", move |config| {
  tinymemory_core::store::chunks::store::with_connection(config, |conn| {

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.
 ///