diff --git a/Cargo.lock b/Cargo.lock index 9565dc4a..0fa6782a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2067,6 +2067,7 @@ dependencies = [ "async-trait", "chrono", "log", + "rusqlite", "serde", "serde_json", "tempfile", diff --git a/crates/tinymemory-api/src/provider/records.rs b/crates/tinymemory-api/src/provider/records.rs index 064697fc..467f1825 100644 --- a/crates/tinymemory-api/src/provider/records.rs +++ b/crates/tinymemory-api/src/provider/records.rs @@ -18,7 +18,9 @@ use async_trait::async_trait; use crate::error::MemoryError; use crate::goals::GoalsDoc; -use crate::provider::types::{IngestOutcome, MaintenanceReport, SourceItem}; +use crate::provider::types::{ + IngestOutcome, MaintenanceReport, QueueFailure, QueueStats, SourceItem, StoreStats, +}; use crate::tool_memory::ToolMemoryRule; use crate::types::MemoryTaint; @@ -164,4 +166,48 @@ pub trait MemoryMaintenance: Send + Sync { /// Backend failures only. A *finding* is not an error: a store with /// problems still returns `Ok` with the problems listed. async fn doctor(&self) -> Result; + + /// Aggregate counts over what this driver has stored. + /// + /// Defaulted to an empty [`StoreStats`] rather than `Unsupported`, and the + /// difference matters: this is a diagnostic, and a caller asking "how much + /// is stored" can do something sensible with "nothing reported" while + /// having nothing to do with an error. A driver that can answer should. + /// + /// # Errors + /// + /// Backend failures only. + async fn store_stats(&self) -> Result { + Ok(StoreStats::default()) + } + + /// The ingest and re-embed queue's state. + /// + /// `kind` narrows to one job kind (the driver's own identifier); `None` + /// counts every kind. A driver with no queue answers all-zero, which is + /// true of it rather than a refusal — and so does a kind this driver does + /// not have, since "no jobs of a kind I never enqueue" is the honest + /// count. A caller that does not know the driver's vocabulary passes + /// `None`; that is what the `Option` is for. + /// + /// # Errors + /// + /// Backend failures only. + async fn queue_stats(&self, kind: Option<&str>) -> Result { + let _ = kind; + Ok(QueueStats::default()) + } + + /// The most recent terminal queue failure, if the driver records one. + /// + /// `Ok(None)` means "nothing has failed", which is why this is not an + /// error: a healthy queue and a driver that keeps no failure history give + /// the same answer, and neither is a fault the caller can act on. + /// + /// # Errors + /// + /// Backend failures only. + async fn latest_queue_failure(&self) -> Result, MemoryError> { + Ok(None) + } } diff --git a/crates/tinymemory-bus/src/lib.rs b/crates/tinymemory-bus/src/lib.rs index 6e852977..f74bf116 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 89 members on it, built as a `cdylib`. A host that +//! exports one object with 92 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 87a78daf..a3f5f363 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -150,6 +150,12 @@ pub mod methods { pub const CONSOLIDATE: &str = "Consolidate"; /// `Doctor` — doctor. pub const DOCTOR: &str = "Doctor"; + /// `StoreStats` — aggregate counts over what the driver has stored. + pub const STORE_STATS: &str = "StoreStats"; + /// `QueueStats` — the ingest and re-embed queue's state. + pub const QUEUE_STATS: &str = "QueueStats"; + /// `LatestQueueFailure` — the most recent terminal queue failure. + pub const LATEST_QUEUE_FAILURE: &str = "LatestQueueFailure"; // The people store: ranking, handles, scores and interactions. /// `ListPeople` — list people. @@ -239,7 +245,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; 89] = [ +pub const METHODS: [&str; 92] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -291,6 +297,9 @@ pub const METHODS: [&str; 89] = [ methods::COMPACT, methods::CONSOLIDATE, methods::DOCTOR, + methods::STORE_STATS, + methods::QUEUE_STATS, + methods::LATEST_QUEUE_FAILURE, methods::LIST_PEOPLE, methods::GET_PERSON, methods::RESOLVE_HANDLE, diff --git a/crates/tinymemory-bus/src/provider/types.rs b/crates/tinymemory-bus/src/provider/types.rs index 29ed4deb..ab8a6b47 100644 --- a/crates/tinymemory-bus/src/provider/types.rs +++ b/crates/tinymemory-bus/src/provider/types.rs @@ -388,6 +388,107 @@ pub struct MaintenanceReport { pub findings: Vec, } +/// Aggregate counts over what the driver has stored. +/// +/// Separate from [`MaintenanceReport`] because the caller does something +/// different with it: a report is read by an operator, these are read by code. +/// `findings: Vec` cannot answer "how far behind is the pipeline" +/// without parsing prose back into numbers. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct StoreStats { + /// Chunks the driver holds. + pub chunks: u64, + /// Of those chunks, how many the driver has extracted structure from. + /// + /// A count rather than the ratio a caller displays, because the ratio is + /// only meaningful against the denominator it was measured with. Read + /// separately, the two can be sampled either side of a write and produce a + /// coverage above 1.0; read together they cannot. + /// + /// A driver that does not extract structure leaves this at zero, which + /// reads as "nothing extracted" — correct for it, and the reason a caller + /// should show the pair rather than the ratio alone. + pub chunks_with_structure: u64, + /// Timestamp of the most recently stored chunk, if any. + /// + /// `None` for an empty store — distinct from `Some(0)`, which would be a + /// chunk stamped at the epoch. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub most_recent_chunk_ms: Option, +} + +/// The ingest and re-embed queue's state, as counts rather than rows. +/// +/// Every field answers a question an operator or a health probe asks about +/// throughput. A driver with no queue answers all-zero rather than refusing: +/// "nothing is backed up" is true of a driver that cannot back up. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct QueueStats { + /// Jobs waiting, whatever their scheduled time. + pub ready: u64, + /// Jobs a worker currently holds. + pub running: u64, + /// Jobs that finished successfully. + pub done: u64, + /// Jobs that ended in a terminal failure. + pub failed: u64, + /// Of those failures, how many the driver will not retry on its own. + /// + /// The distinction is what separates an alert from a shrug: transient + /// failures self-heal on the next attempt, and a caller that escalates on + /// [`Self::failed`] alone pages someone for a queue that is already + /// recovering. Counted with `failed` rather than beside it, because two + /// reads can land either side of a retry and report more unrecoverable + /// failures than there are failures. + pub failed_unrecoverable: u64, + /// Ready jobs whose scheduled time has already passed. + /// + /// The difference between this and [`Self::ready`] is deferred work, and + /// conflating them reads a healthy backlog of future jobs as a stall. + pub eligible_now: u64, + /// When the queue last settled a job. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_completed_ms: Option, + /// The scheduled time of the oldest job eligible to run now. + /// + /// With [`Self::last_completed_ms`] this is what an idle-time calculation + /// needs: how long something runnable has been waiting. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub oldest_eligible_ms: Option, +} + +/// The most recent terminal queue failure. +/// +/// Carries the driver's own words rather than a class this contract invents: +/// the caller shows it to an operator, and a re-classification here would lose +/// what the engine actually said. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct QueueFailure { + /// The failure's own message. Must carry no memory content. + pub reason: String, + /// The driver's classification, when it has one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub class: Option, + /// When the failing job settled. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub completed_at_ms: Option, + /// When the queue last completed a job *successfully*, read together with + /// the failure above rather than in a second call. + /// + /// A caller deciding whether to show this failure asks whether anything + /// has succeeded since it — a success after the failure means the queue + /// recovered and the failure is stale. Answering that from two separate + /// calls lets a job settle in between and flip the decision, so the two + /// values are read as one observation. + /// + /// This is not [`QueueStats::last_completed_ms`]: that one counts a + /// failure as progress, because a fast-failing queue is not a stalled + /// one. Supersession needs the opposite reading — only a success clears a + /// failure — so it takes the newest *successful* completion. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_success_ms: Option, +} + #[cfg(test)] #[path = "types_tests.rs"] mod tests; diff --git a/crates/tinymemory-module/Cargo.lock b/crates/tinymemory-module/Cargo.lock index 6f852cf5..e64052e4 100644 --- a/crates/tinymemory-module/Cargo.lock +++ b/crates/tinymemory-module/Cargo.lock @@ -1902,6 +1902,7 @@ dependencies = [ "async-trait", "chrono", "log", + "rusqlite", "serde", "serde_json", "tinycortex", diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 30e705fb..d7a87223 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -307,6 +307,9 @@ mod exports { "Compact", "Consolidate", "Doctor", + "StoreStats", + "QueueStats", + "LatestQueueFailure", ], signals = [], // The host's embedder is deliberately NOT declared as `requires`. That diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 009125f5..71370bd6 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -121,7 +121,7 @@ use tinymemory_api::goals::GoalsDoc; use tinymemory_api::health::MemoryHealth; use tinymemory_api::provider::types::{ DiffReport, EntityHit, ExportPage, ExportRecord, ImportOutcome, IngestItem, IngestOutcome, - MaintenanceReport, SnapshotRef, SourceItem, SourceScope, + MaintenanceReport, QueueFailure, QueueStats, SnapshotRef, SourceItem, SourceScope, StoreStats, }; // `MemoryCore`, `MemoryRecall` and `MemoryPortability` are deliberately not // imported: they are supertraits of `MemoryProvider`, so their methods are @@ -887,6 +887,27 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + async fn store_stats(&self) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .store_stats() + .await + .map_err(|error| into_bus_error(&error)) + } + + async fn queue_stats(&self, kind: Option) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .queue_stats(kind.as_deref()) + .await + .map_err(|error| into_bus_error(&error)) + } + + async fn latest_queue_failure(&self) -> BusResult> { + require_family!(self, as_maintenance, Capability::Maintenance) + .latest_queue_failure() + .await + .map_err(|error| into_bus_error(&error)) + } + // ── People ────────────────────────────────────────────────────────────── /// Known people, ranked by closeness. diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index ced3e49b..9d918bdd 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 89-element inequality, so the + // Reported as differences rather than as a 92-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 db3a34b9..cdfd4ede 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -671,6 +671,9 @@ const EXPECTED_METHODS: &[&str] = &[ "Compact", "Consolidate", "Doctor", + "StoreStats", + "QueueStats", + "LatestQueueFailure", ]; #[tokio::test] diff --git a/crates/tinymemory-tinycortex/Cargo.toml b/crates/tinymemory-tinycortex/Cargo.toml index 22ae6eff..b4a70da8 100644 --- a/crates/tinymemory-tinycortex/Cargo.toml +++ b/crates/tinymemory-tinycortex/Cargo.toml @@ -14,6 +14,15 @@ repository = "https://github.com/tinyhumansai/tinymemory" [dependencies] # The contract this adapter targets. tinymemory-api = { path = "../tinymemory-api" } +# The diagnostic reads in `MemoryMaintenance` query TinyCortex's own tables +# directly, which is this crate's job — it exists to adapt that schema to the +# contract, and the schema is not something the contract can describe. +# +# Pinned to the exact version `tinymemory-core` uses, deliberately. `rusqlite` +# carries a `links` key through `libsqlite3-sys`, so two different versions in +# one graph is a hard cargo error rather than a silent duplicate; matching the +# pin makes cargo unify them onto the one bundled copy already present. +rusqlite = { version = "=0.40.2", features = ["bundled"] } # The engine being adapted. A version requirement rather than a path, so a host # that already pins its own TinyCortex checkout unifies both onto one copy # through its `[patch.crates-io]`; the workspace root patches it to the nested diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 7772f554..47196e4e 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -33,7 +33,7 @@ use tinymemory_api::host::{ use tinymemory_api::mandatory::MemoryTraitProvider; use tinymemory_api::provider::types::{ EntityHit, EntityRef, ExportPage, ExportRecord, ImportOutcome, IngestItem, IngestOutcome, - MaintenanceReport, SourceItem, SourceScope, + MaintenanceReport, QueueFailure, QueueStats, SourceItem, SourceScope, StoreStats, }; // Diff-family value types, used only by the `MemoryDiff` impl below — which is // compiled out without the git-backed snapshot store. @@ -1158,6 +1158,171 @@ impl MemoryMaintenance for TinycortexProvider { }) } + async fn store_stats(&self) -> Result { + blocking(self.config.clone(), "read store stats", move |config| { + tinymemory_core::store::chunks::store::with_connection(config, |conn| { + // One statement for all three. The count and the extracted + // count are a ratio the caller displays, and sampling them + // either side of a write can put the numerator above the + // denominator — a coverage over 100%. + // + // `MAX` over an empty table is SQL NULL, which is the same + // answer as "no chunks" and must stay distinguishable from a + // chunk stamped at the epoch — hence `Option`, not `0`. + let (chunks, chunks_with_structure, most_recent_chunk_ms): (i64, i64, Option) = + conn.query_row( + "SELECT + COUNT(*), + COALESCE(SUM(EXISTS ( + SELECT 1 FROM mem_tree_entity_index e + WHERE e.node_id = c.id + )), 0), + MAX(timestamp_ms) + FROM mem_tree_chunks c", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + )?; + let count = |n: i64| u64::try_from(n).unwrap_or(0); + Ok(StoreStats { + chunks: count(chunks), + chunks_with_structure: count(chunks_with_structure), + most_recent_chunk_ms, + }) + }) + .map_err(|error| anyhow::anyhow!("store stats: {error}")) + }) + .await + } + + async fn queue_stats(&self, kind: Option<&str>) -> Result { + let kind = kind.map(str::to_string); + blocking(self.config.clone(), "read queue stats", move |config| { + let now_ms = chrono::Utc::now().timestamp_millis(); + tinymemory_core::store::chunks::store::with_connection(config, |conn| { + // One statement rather than six round trips, and one `now` + // rather than one per sub-query: counts taken at different + // instants can disagree with each other, and an idle-time + // calculation built on that reads as a stall that never + // happened. + let ( + ready, + running, + done, + failed, + failed_unrecoverable, + eligible_now, + last_completed_ms, + oldest_eligible_ms, + ): (i64, i64, i64, i64, i64, i64, Option, Option) = conn.query_row( + "SELECT + COALESCE(SUM(status = 'ready'), 0), + COALESCE(SUM(status = 'running'), 0), + COALESCE(SUM(status = 'done'), 0), + COALESCE(SUM(status = 'failed'), 0), + COALESCE(SUM(status = 'failed' + AND failure_class = 'unrecoverable'), 0), + COALESCE(SUM(status = 'ready' AND available_at_ms <= ?1), 0), + -- Every status, not just 'done'. A job that fails + -- stamps `completed_at_ms` too, and a queue failing + -- fast is making progress in the only sense this + -- field measures: it is not stuck. Filtering to + -- 'done' would report a fast-failing pipeline as + -- idle, which is the misdiagnosis this exists to + -- avoid. Supersession wants the opposite reading and + -- gets its own field on `QueueFailure`. + MAX(completed_at_ms), + MIN(CASE WHEN status = 'ready' AND available_at_ms <= ?1 + THEN available_at_ms END) + FROM mem_tree_jobs + WHERE (?2 IS NULL OR kind = ?2)", + rusqlite::params![now_ms, kind], + |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + row.get(5)?, + row.get(6)?, + row.get(7)?, + )) + }, + )?; + let count = |n: i64| u64::try_from(n).unwrap_or(0); + Ok(QueueStats { + ready: count(ready), + running: count(running), + done: count(done), + failed: count(failed), + failed_unrecoverable: count(failed_unrecoverable), + eligible_now: count(eligible_now), + last_completed_ms, + oldest_eligible_ms, + }) + }) + .map_err(|error| anyhow::anyhow!("queue stats: {error}")) + }) + .await + } + + async fn latest_queue_failure(&self) -> Result, MemoryError> { + blocking( + self.config.clone(), + "read latest queue failure", + move |config| { + tinymemory_core::store::chunks::store::with_connection(config, |conn| { + use rusqlite::OptionalExtension; + let Some(mut failure) = conn + .query_row( + "SELECT failure_reason, failure_class, completed_at_ms + FROM mem_tree_jobs + WHERE status = 'failed' AND failure_reason IS NOT NULL + ORDER BY completed_at_ms DESC + LIMIT 1", + [], + |row| { + Ok(QueueFailure { + reason: row.get(0)?, + class: row.get(1)?, + completed_at_ms: row.get(2)?, + last_success_ms: None, + }) + }, + ) + .optional()? + else { + return Ok(None); + }; + + // Still inside the same `with_connection`, which holds the + // connection for the whole closure: no job can settle + // between the two reads and flip a supersession decision + // made from them. + // + // Only worth asking when the failure is timestamped — + // without one there is nothing to compare a success + // against, and the caller has to surface it either way. + if failure.completed_at_ms.is_some() { + failure.last_success_ms = conn + .query_row( + "SELECT MAX(completed_at_ms) FROM mem_tree_jobs + WHERE status = 'done'", + [], + |row| row.get(0), + ) + .optional() + .map(Option::flatten)?; + } + + Ok(Some(failure)) + }) + .map_err(|error| anyhow::anyhow!("latest queue failure: {error}")) + }, + ) + .await + } + async fn compact(&self) -> Result { let (examined, changed) = blocking(self.config.clone(), "compact memory queue", move |config| { diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index 67717ae5..773b82e8 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -133,6 +133,248 @@ async fn the_full_provider_actually_retains() { ); } +/// The maintenance diagnostics answer from the store, not from their defaults. +/// +/// `store_stats`, `queue_stats` and `latest_queue_failure` are defaulted on +/// the trait so a driver without a queue compiles and answers "nothing" rather +/// than refusing. That default is also the failure this test exists to catch: +/// an engine that inherits it reports an empty store and an idle queue forever, +/// which is indistinguishable from a healthy quiet one and is exactly the shape +/// a caller cannot detect. Asserting a zero would pass against the default, so +/// this stores something first and requires the numbers to move. +#[tokio::test(flavor = "multi_thread")] +async fn maintenance_diagnostics_read_the_store_rather_than_their_defaults() { + use tinymemory_api::chunks::DataSource; + use tinymemory_api::provider::{MemoryMaintenance, MemoryProvider}; + use tinymemory_api::types::MemoryTaint; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + + let before = provider.store_stats().await.expect("store stats"); + assert_eq!(before.chunks, 0, "a fresh workspace holds nothing"); + assert_eq!( + before.most_recent_chunk_ms, None, + "an empty store has no newest chunk — `None`, never `Some(0)`, which \ + would be a chunk stamped at the epoch" + ); + + // Ingested, not `store`d: `MemoryCore::store` writes a memory entry, and + // `store_stats.chunks` counts the CHUNK tier, which only ingest fills. + // Asserting against `store` here passed a zero against a zero and proved + // nothing — the first version of this test did exactly that. + let outcome = provider + .as_ingest() + .expect("Ingest") + .ingest_document(tinymemory_api::provider::types::IngestItem { + namespace: None, + source: DataSource::Upload, + source_id: "diagnostics-upload".into(), + owner: "owner".into(), + source_ref: None, + content: "A deterministic sentence for the diagnostics test.".into(), + mime: Some("text/plain".into()), + timestamp: Some( + chrono::DateTime::from_timestamp(1_700_000_000, 0).expect("fixed timestamp"), + ), + tags: Vec::new(), + taint: MemoryTaint::Internal, + path_scope: None, + }) + .await + .expect("ingest a document"); + assert!(outcome.written > 0, "the ingest must persist a chunk"); + + let after = provider.store_stats().await.expect("store stats"); + assert!( + after.chunks > before.chunks, + "the count must follow the store; got {} after writing to {}", + after.chunks, + before.chunks + ); + assert!( + after.chunks_with_structure <= after.chunks, + "extracted chunks are a subset of stored ones; a numerator above its \ + denominator is a coverage over 100%, which is what reading the two \ + separately can produce" + ); + + // A queue this engine has not been asked to fill is legitimately empty, so + // the assertion is that the call answers from the queue at all rather than + // erroring — the numbers themselves are only meaningful once work exists. + let queue = provider.queue_stats(None).await.expect("queue stats"); + assert_eq!( + queue.eligible_now.min(queue.ready), + queue.eligible_now, + "jobs eligible now are a subset of jobs ready; conflating the two \ + reads a backlog of deferred work as a stall" + ); + + // Nothing has failed, and that must read as `None` rather than an error: + // a healthy queue and a driver keeping no failure history give the same + // answer, and neither is something a caller can act on. + assert!( + provider + .latest_queue_failure() + .await + .expect("latest queue failure") + .is_none(), + "a store that has run no failing job reports no failure" + ); +} + +/// Work the queue has deliberately parked is ready, but not eligible now. +/// +/// This is the difference between a backlog and a stall. A job backing off +/// after a transient failure stays `ready` with its next attempt scheduled +/// forward; counting it as runnable means a queue that is behaving correctly +/// reports as one that has stopped, and the caller escalates on it. The +/// caller cannot make the distinction itself — it sees counts, not schedules +/// — so the split has to be made here. +#[tokio::test] +async fn deferred_work_stays_ready_without_becoming_eligible() { + use tinymemory_api::provider::MemoryMaintenance; + use tinymemory_core::queue::store as queue_store; + use tinymemory_core::queue::types::{FlushStalePayload, NewJob}; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + let config = provider_config(workspace.path(), serde_json::Value::Null); + + let new_job = + NewJob::flush_stale(&FlushStalePayload::default(), "2026-08-21", 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"); + + let queue = provider.queue_stats(None).await.expect("queue stats"); + assert_eq!(queue.ready, 1); + assert_eq!( + queue.eligible_now, 1, + "precondition: a freshly enqueued job is runnable now" + ); + + // Park it the way a backing-off retry does: still `ready`, scheduled + // forward. Far enough forward that no plausible clock lands after it. + let until_ms = chrono::Utc::now().timestamp_millis() + 60 * 60 * 1000; + queue_store::mark_deferred(&config, &job, until_ms, "backing off").expect("defer the job"); + + let queue = provider.queue_stats(None).await.expect("queue stats"); + assert_eq!( + queue.ready, 1, + "deferred work is still queued — it has not been abandoned" + ); + assert_eq!( + queue.eligible_now, 0, + "but nothing is runnable, so nothing is being held up" + ); + assert_eq!( + queue.oldest_eligible_ms, None, + "and there is no waiting job to measure an idle window from" + ); +} + +/// A reported failure carries the success watermark that decides whether it is +/// still worth showing. +/// +/// A caller asking "has anything succeeded since this failed?" out of two +/// separate calls lets a job settle in between and flip the answer. So the +/// watermark rides along with the failure, and what this test pins is that it +/// is populated from the store rather than left at `None` — the shape that +/// would push the caller back into asking twice. +#[tokio::test] +async fn a_reported_failure_carries_the_success_watermark_that_supersedes_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); + + // Fail one job for real rather than writing the row by hand: the query + // filters on `failure_reason IS NOT NULL`, which only the typed failure + // path fills, and a hand-written row would not prove that filter matches + // what the engine actually persists. + let failing = + NewJob::flush_stale(&FlushStalePayload::default(), "2026-08-21", 1).expect("build a job"); + let failing_id = queue_store::enqueue(&config, &failing) + .expect("enqueue") + .expect("a fresh job is not a duplicate"); + let failing = queue_store::get_job(&config, &failing_id) + .expect("read the job") + .expect("the job exists"); + queue_store::mark_failed_typed( + &config, + &failing, + "the failure the operator would see", + Some(&PipelineFailure::new(FailureCode::BudgetExhausted)), + ) + .expect("fail the job"); + + let failure = provider + .latest_queue_failure() + .await + .expect("latest queue failure") + .expect("the failed job is reported"); + assert_eq!(failure.reason, "budget_exhausted"); + assert_eq!( + failure.last_success_ms, None, + "nothing has succeeded yet, so there is no watermark to supersede it" + ); + + let queue = provider.queue_stats(None).await.expect("queue stats"); + assert_eq!(queue.failed, 1, "the job is parked as failed"); + assert_eq!( + queue.failed_unrecoverable, 1, + "an exhausted budget is not retried on its own, and a caller that \ + escalates on `failed` alone cannot tell that apart from a failure \ + about to self-heal" + ); + assert!( + queue.last_completed_ms.is_some(), + "a failure settles a job too. Counting only successes here reports a \ + queue that is failing fast — as fast as it can run — as one that has \ + gone idle, which is the opposite diagnosis" + ); + + // Now settle one successfully. Same queue, same connection the failure is + // read on. + let done = + NewJob::flush_stale(&FlushStalePayload::default(), "2026-08-22", 1).expect("build a job"); + let done_id = queue_store::enqueue(&config, &done) + .expect("enqueue") + .expect("a fresh job is not a duplicate"); + let done = queue_store::get_job(&config, &done_id) + .expect("read the job") + .expect("the job exists"); + queue_store::mark_done(&config, &done).expect("settle the job"); + + let failure = provider + .latest_queue_failure() + .await + .expect("latest queue failure") + .expect("the failed job is still the newest failure"); + let watermark = failure + .last_success_ms + .expect("a completed job must show up as the success watermark"); + let failed_at = failure + .completed_at_ms + .expect("the typed failure path stamps a completion time"); + // `>=`, not `>`: both settle from the wall clock and can land on the same + // millisecond. What is under test is that the two values arrive together + // and are comparable at all, not the resolution of the clock. + assert!( + watermark >= failed_at, + "the success settled after the failure; got watermark {watermark} against failure \ + {failed_at}" + ); +} + /// The KV write path canonicalizes identifiers (the shim in `tinymemory-core` /// routes every `set_*`/`delete_*` through `canonical_identifier`), so a read /// path that compares the raw caller key misses every rewritten key: put→get