From fd5a644338a62b0c15071d77d7e9e72a9b30c24d Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Fri, 8 May 2026 07:08:25 +0530 Subject: [PATCH] feat(memory_tree/jobs): JobOutcome::Defer + mark_deferred (#1256) (#1345) --- .../memory/tree/jobs/handlers/mod.rs | 61 +++-- src/openhuman/memory/tree/jobs/mod.rs | 6 +- src/openhuman/memory/tree/jobs/store.rs | 252 ++++++++++++++++++ src/openhuman/memory/tree/jobs/types.rs | 22 ++ src/openhuman/memory/tree/jobs/worker.rs | 26 +- 5 files changed, 336 insertions(+), 31 deletions(-) diff --git a/src/openhuman/memory/tree/jobs/handlers/mod.rs b/src/openhuman/memory/tree/jobs/handlers/mod.rs index a0e594bca..ed0edfa93 100644 --- a/src/openhuman/memory/tree/jobs/handlers/mod.rs +++ b/src/openhuman/memory/tree/jobs/handlers/mod.rs @@ -2,8 +2,12 @@ //! //! Each handler parses its payload from `Job::payload_json`, performs its //! side effects (DB writes, LLM calls, follow-up enqueues), and returns -//! `Ok(())` on success or an `anyhow::Error` on retryable failure. -//! [`handle_job`] fans out to the handler matching the row's `kind`. +//! `Ok(JobOutcome::Done)` on success or an `anyhow::Error` on retryable +//! failure. A handler may also return `Ok(JobOutcome::Defer { … })` to +//! re-queue the job with a wake-up time without burning the failure +//! budget — useful for transient blockers like cloud rate limits or a +//! warming-up model. [`handle_job`] fans out to the handler matching the +//! row's `kind`. use anyhow::{Context, Result}; @@ -14,7 +18,7 @@ use crate::openhuman::memory::tree::content_store::{ use crate::openhuman::memory::tree::jobs::store; use crate::openhuman::memory::tree::jobs::types::{ AppendBufferPayload, AppendTarget, DigestDailyPayload, ExtractChunkPayload, FlushStalePayload, - Job, JobKind, NewJob, NodeRef, SealPayload, TopicRoutePayload, + Job, JobKind, JobOutcome, NewJob, NodeRef, SealPayload, TopicRoutePayload, }; use crate::openhuman::memory::tree::score; use crate::openhuman::memory::tree::score::embed::{build_embedder_from_config, pack_checked}; @@ -28,7 +32,12 @@ use crate::openhuman::memory::tree::tree_source::{ use crate::openhuman::memory::tree::tree_topic::curator; /// Dispatch a claimed job to the matching per-kind handler. -pub async fn handle_job(config: &Config, job: &Job) -> Result<()> { +/// +/// Existing handlers all return `Ok(JobOutcome::Done)` on success. The +/// `Defer` outcome is wired through the worker but not yet emitted by any +/// in-tree handler — consumers (cloud rate limiter, triage tiered +/// fallback, embed warmup) land in follow-up issues. +pub async fn handle_job(config: &Config, job: &Job) -> Result { match job.kind { JobKind::ExtractChunk => handle_extract(config, job).await, JobKind::AppendBuffer => handle_append_buffer(config, job).await, @@ -39,7 +48,7 @@ pub async fn handle_job(config: &Config, job: &Job) -> Result<()> { } } -async fn handle_extract(config: &Config, job: &Job) -> Result<()> { +async fn handle_extract(config: &Config, job: &Job) -> Result { let payload: ExtractChunkPayload = serde_json::from_str(&job.payload_json).context("parse ExtractChunk payload")?; let Some(chunk) = chunk_store::get_chunk(config, &payload.chunk_id)? else { @@ -47,7 +56,7 @@ async fn handle_extract(config: &Config, job: &Job) -> Result<()> { "[memory_tree::jobs] extract chunk missing chunk_id={}", payload.chunk_id ); - return Ok(()); + return Ok(JobOutcome::Done); }; // Read the full body from disk (the `content` column in SQLite holds a @@ -204,10 +213,10 @@ async fn handle_extract(config: &Config, job: &Job) -> Result<()> { super::worker::wake_workers(); } - Ok(()) + Ok(JobOutcome::Done) } -async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> { +async fn handle_append_buffer(config: &Config, job: &Job) -> Result { use crate::openhuman::memory::tree::tree_source::bucket_seal::should_seal; use crate::openhuman::memory::tree::tree_source::store as src_store; @@ -221,7 +230,7 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> { NodeRef::Leaf { chunk_id } => { let Some(chunk) = chunk_store::get_chunk(config, chunk_id)? else { log::warn!("[memory_tree::jobs] append_buffer chunk missing chunk_id={chunk_id}"); - return Ok(()); + return Ok(JobOutcome::Done); }; let score_row = score_store::get_score(config, &chunk.id)? .ok_or_else(|| anyhow::anyhow!("missing score row for chunk {}", chunk.id))?; @@ -247,7 +256,7 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> { log::warn!( "[memory_tree::jobs] append_buffer summary missing summary_id={summary_id}" ); - return Ok(()); + return Ok(JobOutcome::Done); }; // Read the full body from disk — `summary.content` is a ≤500-char // preview after the MD-on-disk migration. The summariser receives @@ -282,7 +291,7 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> { // topic_route and this append). Drop on the floor — the // topic_route was advisory and the source-tree path already // ran for this leaf. - return Ok(()); + return Ok(JobOutcome::Done); }; let is_source_target = matches!(payload.target, AppendTarget::Source { .. }); @@ -342,10 +351,10 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> { if did_enqueue_seal { super::worker::wake_workers(); } - Ok(()) + Ok(JobOutcome::Done) } -async fn handle_seal(config: &Config, job: &Job) -> Result<()> { +async fn handle_seal(config: &Config, job: &Job) -> Result { use crate::openhuman::memory::tree::tree_source::bucket_seal::{seal_one_level, should_seal}; use crate::openhuman::memory::tree::tree_source::store as src_store; use crate::openhuman::memory::tree::tree_source::types::TreeKind; @@ -357,7 +366,7 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> { "[memory_tree::jobs] seal tree missing tree_id={}", payload.tree_id ); - return Ok(()); + return Ok(JobOutcome::Done); }; // Seal exactly one level. Parents only get sealed via a follow-up job @@ -371,7 +380,7 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> { tree.id, payload.level ); - return Ok(()); + return Ok(JobOutcome::Done); } if !forced && !should_seal(&buf) { // Another job sealed this level out from under us (or the buffer @@ -382,7 +391,7 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> { payload.level, buf.token_sum ); - return Ok(()); + return Ok(JobOutcome::Done); } // Pick the labeling strategy for this tree kind. Source trees mint @@ -417,10 +426,10 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> { } super::worker::wake_workers(); - Ok(()) + Ok(JobOutcome::Done) } -async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> { +async fn handle_topic_route(config: &Config, job: &Job) -> Result { let payload: TopicRoutePayload = serde_json::from_str(&job.payload_json).context("parse TopicRoute payload")?; @@ -431,7 +440,7 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> { NodeRef::Leaf { chunk_id } => { if chunk_store::get_chunk(config, chunk_id)?.is_none() { log::warn!("[memory_tree::jobs] topic_route chunk missing chunk_id={chunk_id}"); - return Ok(()); + return Ok(JobOutcome::Done); } chunk_id.clone() } @@ -442,7 +451,7 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> { log::warn!( "[memory_tree::jobs] topic_route summary missing summary_id={summary_id}" ); - return Ok(()); + return Ok(JobOutcome::Done); } summary_id.clone() } @@ -451,7 +460,7 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> { let entity_ids = score_store::list_entity_ids_for_node(config, &node_id)?; if entity_ids.is_empty() { log::debug!("[memory_tree::jobs] topic_route no entities for node_id={node_id} — skipping"); - return Ok(()); + return Ok(JobOutcome::Done); } let summariser = build_summariser(config); @@ -473,10 +482,10 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> { } } } - Ok(()) + Ok(JobOutcome::Done) } -async fn handle_digest_daily(config: &Config, job: &Job) -> Result<()> { +async fn handle_digest_daily(config: &Config, job: &Job) -> Result { let payload: DigestDailyPayload = serde_json::from_str(&job.payload_json).context("parse DigestDaily payload")?; let day = chrono::NaiveDate::parse_from_str(&payload.date_iso, "%Y-%m-%d") @@ -491,10 +500,10 @@ async fn handle_digest_daily(config: &Config, job: &Job) -> Result<()> { log::debug!("[memory_tree::jobs] digest skipped existing_id={existing_id}"); } } - Ok(()) + Ok(JobOutcome::Done) } -async fn handle_flush_stale(config: &Config, job: &Job) -> Result<()> { +async fn handle_flush_stale(config: &Config, job: &Job) -> Result { let payload: FlushStalePayload = serde_json::from_str(&job.payload_json).context("parse FlushStale payload")?; let age_secs = payload @@ -513,7 +522,7 @@ async fn handle_flush_stale(config: &Config, job: &Job) -> Result<()> { super::worker::wake_workers(); } } - Ok(()) + Ok(JobOutcome::Done) } #[cfg(test)] diff --git a/src/openhuman/memory/tree/jobs/mod.rs b/src/openhuman/memory/tree/jobs/mod.rs index 8ce89ab2a..cdb71ccb0 100644 --- a/src/openhuman/memory/tree/jobs/mod.rs +++ b/src/openhuman/memory/tree/jobs/mod.rs @@ -34,12 +34,12 @@ mod worker; pub use scheduler::{backfill_missing_digests, trigger_digest}; pub use store::{ - claim_next, count_by_status, count_total, enqueue, enqueue_tx, get_job, mark_done, mark_failed, - recover_stale_locks, DEFAULT_LOCK_DURATION_MS, + claim_next, count_by_status, count_total, enqueue, enqueue_tx, get_job, mark_deferred, + mark_done, mark_failed, recover_stale_locks, DEFAULT_LOCK_DURATION_MS, }; pub use testing::drain_until_idle; pub use types::{ AppendBufferPayload, AppendTarget, DigestDailyPayload, ExtractChunkPayload, FlushStalePayload, - Job, JobKind, JobStatus, NewJob, NodeRef, SealPayload, TopicRoutePayload, + Job, JobKind, JobOutcome, JobStatus, NewJob, NodeRef, SealPayload, TopicRoutePayload, }; pub use worker::{start, wake_workers}; diff --git a/src/openhuman/memory/tree/jobs/store.rs b/src/openhuman/memory/tree/jobs/store.rs index 32e44d3a5..d733dea59 100644 --- a/src/openhuman/memory/tree/jobs/store.rs +++ b/src/openhuman/memory/tree/jobs/store.rs @@ -253,6 +253,57 @@ pub fn mark_failed(config: &Config, job: &Job, error: &str) -> Result<()> { }) } +/// Mark a claimed job as deferred: put it back to `ready` with +/// `available_at_ms = until_ms` so [`claim_next`] will re-pick it once the +/// wake-up time has passed. The handler ran successfully but chose not to +/// make progress (cloud rate-limited, dependency unavailable, model +/// warming up), so this path **does not** burn the failure budget — the +/// `attempts` bump that [`claim_next`] applied at claim time is reverted. +/// +/// `reason` is recorded in `last_error` for visibility and `started_at_ms` +/// is cleared (mirroring the retry branch of [`mark_failed`]) so the next +/// claim stamps a fresh start time. +/// +/// Like [`mark_done`] / [`mark_failed`], the UPDATE is gated on the +/// claim-token (`attempts` + `started_at_ms`) so a stale lessee's +/// settlement is a silent no-op rather than clobbering an active lessee's +/// row. +pub fn mark_deferred(config: &Config, job: &Job, until_ms: i64, reason: &str) -> Result<()> { + let job_id = &job.id; + let claim_attempts = job.attempts as i64; + let pre_claim_attempts = claim_attempts.saturating_sub(1); + let claim_started_at = job.started_at_ms; + with_connection(config, |conn| { + let n = conn.execute( + "UPDATE mem_tree_jobs + SET status = 'ready', + attempts = ?1, + available_at_ms = ?2, + locked_until_ms = NULL, + started_at_ms = NULL, + last_error = ?3 + WHERE id = ?4 + AND attempts = ?5 + AND started_at_ms IS ?6", + params![ + pre_claim_attempts, + until_ms, + reason, + job_id, + claim_attempts, + claim_started_at, + ], + )?; + if n == 0 { + log::warn!( + "[memory_tree::jobs] mark_deferred id={job_id} was a no-op \ + (stale lease: attempts={claim_attempts} started_at_ms={claim_started_at:?})" + ); + } + Ok(()) + }) +} + /// Flip any `running` row whose `locked_until_ms` has expired back to /// `ready`. Called once at worker startup so a process crash mid-job /// doesn't leave work stranded. Returns the number of rows recovered. @@ -627,4 +678,205 @@ mod tests { assert_eq!(count_by_status(&cfg, JobStatus::Done).unwrap(), 1); assert_eq!(count_by_status(&cfg, JobStatus::Ready).unwrap(), 2); } + + /// Defer must NOT advance the failure-attempt counter. After a claim + /// (which bumps attempts to 1) and a deferral, the next claim should + /// see attempts==2 (just the second claim's bump) — proving the + /// transient deferral did not burn a slot from the row's failure + /// budget. + #[test] + fn mark_deferred_does_not_increment_attempts() { + let (_tmp, cfg) = test_config(); + let payload = ExtractChunkPayload { + chunk_id: "c-defer-1".into(), + }; + let nj = NewJob::extract_chunk(&payload).unwrap(); + let id = enqueue(&cfg, &nj).unwrap().expect("inserted"); + + let claim1 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + assert_eq!(claim1.id, id); + assert_eq!(claim1.attempts, 1, "first claim bumps attempts to 1"); + + // Defer with a wake-up time already in the past so the next + // claim_next is immediately eligible without sleeping. + mark_deferred(&cfg, &claim1, 0, "rate_limited").unwrap(); + + let row = get_job(&cfg, &id).unwrap().unwrap(); + assert_eq!(row.status, JobStatus::Ready); + assert_eq!( + row.attempts, 0, + "Defer must revert the claim's attempts bump (back to pre-claim 0)" + ); + assert_eq!(row.last_error.as_deref(), Some("rate_limited")); + assert_eq!(row.available_at_ms, 0); + assert!(row.locked_until_ms.is_none()); + assert!(row.started_at_ms.is_none()); + + let claim2 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + assert_eq!(claim2.id, id); + assert_eq!( + claim2.attempts, 1, + "second claim bumps attempts to 1 again (Defer didn't count)" + ); + } + + /// A row deferred to a future `until_ms` must not be claimable until + /// the system clock crosses that threshold. + #[test] + fn deferred_row_not_claimable_until_until_ms() { + let (_tmp, cfg) = test_config(); + let payload = ExtractChunkPayload { + chunk_id: "c-defer-2".into(), + }; + let nj = NewJob::extract_chunk(&payload).unwrap(); + let id = enqueue(&cfg, &nj).unwrap().expect("inserted"); + + let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + let future_ms = Utc::now().timestamp_millis() + 60_000; + mark_deferred(&cfg, &claimed, future_ms, "warming_up").unwrap(); + + // Right now: not yet eligible. + let none = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap(); + assert!( + none.is_none(), + "deferred row must not be claimable before until_ms" + ); + + // Force the row available again (proxy for "wall clock advanced + // past until_ms") and confirm claim_next picks it up. + with_connection(&cfg, |c| { + c.execute( + "UPDATE mem_tree_jobs SET available_at_ms = 0 WHERE id = ?1", + params![id], + )?; + Ok(()) + }) + .unwrap(); + let claim2 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + assert_eq!(claim2.id, id); + assert_eq!(claim2.attempts, 1, "Defer left attempts at pre-claim 0"); + } + + /// Three rows with three different terminal verbs: Done, Failed (with + /// retry left, so it bounces back to ready with bumped attempts), and + /// Defer. After processing, each row's terminal state must reflect its + /// settlement verb. Critically, the Defer row keeps its pre-claim + /// `attempts` value while the failed row bumps. + #[test] + fn mixed_outcomes_stress() { + let (_tmp, cfg) = test_config(); + let p_a = ExtractChunkPayload { + chunk_id: "c-mix-a".into(), + }; + let p_b = ExtractChunkPayload { + chunk_id: "c-mix-b".into(), + }; + let p_c = ExtractChunkPayload { + chunk_id: "c-mix-c".into(), + }; + let id_a = enqueue(&cfg, &NewJob::extract_chunk(&p_a).unwrap()) + .unwrap() + .unwrap(); + let id_b = enqueue(&cfg, &NewJob::extract_chunk(&p_b).unwrap()) + .unwrap() + .unwrap(); + let id_c = enqueue(&cfg, &NewJob::extract_chunk(&p_c).unwrap()) + .unwrap() + .unwrap(); + + // Claim the three rows in turn (FIFO within same kind/priority). + let claim_a = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + let claim_b = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + let claim_c = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + // Sanity: three distinct ids covering A/B/C. + let mut got: Vec<&str> = vec![&claim_a.id, &claim_b.id, &claim_c.id]; + got.sort(); + let mut want = vec![id_a.as_str(), id_b.as_str(), id_c.as_str()]; + want.sort(); + assert_eq!(got, want); + + let until_ms = Utc::now().timestamp_millis() + 30_000; + + // Settle: A=done, B=err (retry path), C=defer. + mark_done(&cfg, &claim_a).unwrap(); + mark_failed(&cfg, &claim_b, "transient_error").unwrap(); + mark_deferred(&cfg, &claim_c, until_ms, "rate_limited").unwrap(); + + // A: terminal done. + let row_a = get_job(&cfg, &id_a).unwrap().unwrap(); + assert_eq!(row_a.status, JobStatus::Done); + assert!(row_a.completed_at_ms.is_some()); + + // B: retry path — back to ready with bumped attempts (1) and a + // future available_at_ms from exponential backoff. + let row_b = get_job(&cfg, &id_b).unwrap().unwrap(); + assert_eq!(row_b.status, JobStatus::Ready); + assert_eq!( + row_b.attempts, 1, + "Err settlement keeps the claim's attempts bump" + ); + assert!(row_b.available_at_ms > Utc::now().timestamp_millis()); + assert_eq!(row_b.last_error.as_deref(), Some("transient_error")); + + // C: deferred — back to ready with attempts reverted and + // available_at_ms == until_ms. + let row_c = get_job(&cfg, &id_c).unwrap().unwrap(); + assert_eq!(row_c.status, JobStatus::Ready); + assert_eq!( + row_c.attempts, 0, + "Defer settlement reverts the claim's attempts bump" + ); + assert_eq!(row_c.available_at_ms, until_ms); + assert_eq!(row_c.last_error.as_deref(), Some("rate_limited")); + assert!(row_c.locked_until_ms.is_none()); + assert!(row_c.started_at_ms.is_none()); + } + + /// Stale-worker `mark_deferred` must be a no-op — same lease-token + /// gating as `mark_done` / `mark_failed`. After Worker A's claim + /// expires and Worker B re-claims, Worker A's stale Defer must not + /// clobber B's running row. + #[test] + fn mark_deferred_stale_lease_is_noop() { + let (_tmp, cfg) = test_config(); + let payload = ExtractChunkPayload { + chunk_id: "c-defer-stale".into(), + }; + let nj = NewJob::extract_chunk(&payload).unwrap(); + let id = enqueue(&cfg, &nj).unwrap().expect("inserted"); + + // Worker A claims with already-expired lock. + let worker_a_job = claim_next(&cfg, -1).unwrap().unwrap(); + assert_eq!(worker_a_job.attempts, 1); + + // Lease expires; Worker B re-claims. + let recovered = recover_stale_locks(&cfg).unwrap(); + assert_eq!(recovered, 1); + let worker_b_job = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + assert_eq!(worker_b_job.attempts, 2); + + // Worker A (stale) tries to defer using its old snapshot — must + // be a silent no-op. + let stale_until_ms = Utc::now().timestamp_millis() + 999_000; + mark_deferred(&cfg, &worker_a_job, stale_until_ms, "stale_defer").unwrap(); + + // Worker B's row must be untouched: still 'running' with + // attempts=2 and no stale_defer side effect. + let row = get_job(&cfg, &id).unwrap().unwrap(); + assert_eq!( + row.status, + JobStatus::Running, + "stale mark_deferred must not clobber Worker B's running row" + ); + assert_eq!(row.attempts, 2); + assert_ne!( + row.last_error.as_deref(), + Some("stale_defer"), + "stale defer reason must not be persisted" + ); + assert_ne!( + row.available_at_ms, stale_until_ms, + "stale until_ms must not be persisted" + ); + } } diff --git a/src/openhuman/memory/tree/jobs/types.rs b/src/openhuman/memory/tree/jobs/types.rs index 676ea4d95..6aaab3582 100644 --- a/src/openhuman/memory/tree/jobs/types.rs +++ b/src/openhuman/memory/tree/jobs/types.rs @@ -65,6 +65,28 @@ impl JobKind { } } +/// Outcome of a successful handler run. Workers translate this into a +/// queue settlement: `Done` finalises the row, while `Defer` puts it back +/// to `ready` with `available_at_ms = until_ms` and **does not** count +/// toward the failure-attempt budget. +/// +/// `Defer` exists so a handler that is transiently unable to make +/// progress (cloud rate-limited, dependency unavailable, model warming +/// up) can re-queue its job with a wake-up time without marking it +/// failed. Handlers should still surface real errors via `Err(_)` — that +/// path runs the existing exponential-backoff retry logic which **does** +/// burn the failure budget. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum JobOutcome { + /// Handler ran to completion. Row is settled as `done`. + Done, + /// Handler chose not to make progress yet. Row is rescheduled to + /// `available_at_ms = until_ms` (UTC milliseconds) with `attempts` + /// reverted to its pre-claim value so the failure budget is not + /// touched. `reason` is recorded in `last_error` for visibility. + Defer { until_ms: i64, reason: String }, +} + /// Lifecycle states persisted on `mem_tree_jobs.status`. Workers transition /// `ready → running → done|failed`. `Cancelled` is reserved for explicit /// admin actions (none surfaced yet). diff --git a/src/openhuman/memory/tree/jobs/worker.rs b/src/openhuman/memory/tree/jobs/worker.rs index da4ffc877..79464f93b 100644 --- a/src/openhuman/memory/tree/jobs/worker.rs +++ b/src/openhuman/memory/tree/jobs/worker.rs @@ -17,8 +17,10 @@ use tokio::sync::Notify; use crate::openhuman::config::Config; use crate::openhuman::memory::tree::jobs::handlers; use crate::openhuman::memory::tree::jobs::store::{ - claim_next, mark_done, mark_failed, recover_stale_locks, DEFAULT_LOCK_DURATION_MS, + claim_next, mark_deferred, mark_done, mark_failed, recover_stale_locks, + DEFAULT_LOCK_DURATION_MS, }; +use crate::openhuman::memory::tree::jobs::types::JobOutcome; /// Number of concurrent job-worker tasks. Each worker claims one job /// at a time via `claim_next` (atomic UPDATE under SQLite WAL with @@ -148,9 +150,29 @@ pub async fn run_once(config: &Config) -> Result { drop(llm_permit); match result { - Ok(()) => { + Ok(JobOutcome::Done) => { + log::debug!( + "[memory_tree::jobs] done id={} kind={}", + job.id, + job.kind.as_str() + ); mark_done(config, &job)?; } + Ok(JobOutcome::Defer { until_ms, reason }) => { + // Defer is normal operation (transient blocker, e.g. rate + // limit) — log at info, not warn — and do NOT count this + // claim toward the failure-attempt budget. `mark_deferred` + // reverts the bump applied by `claim_next` so the row's + // attempts counter stays where it was before this claim. + log::info!( + "[memory_tree::jobs] deferred id={} kind={} until_ms={} reason={}", + job.id, + job.kind.as_str(), + until_ms, + reason + ); + mark_deferred(config, &job, until_ms, &reason)?; + } Err(err) => { // Preserve the full anyhow cause chain in the persisted // last_error so a reader of mem_tree_jobs can see the root