feat: echo the active retention windows in the dashboard

crates/taskito-core/BINDING_CONTRACT.md

@@ -229,6 +229,28 @@ layout or each shell sees only its own hooks.
 - **Delivery log** is separate: key `webhooks:deliveries:<subscription_id>`,
   one JSON array per subscription, newest last.
 
+## Effective retention (cross-SDK)
+Retention windows live in `SchedulerConfig`, inside the worker process — a
+dashboard elsewhere cannot see them. So the elected retention leader publishes
+what it applies to the settings KV on every cleanup sweep, and every shell's
+dashboard echoes that document instead of guessing at the defaults.
+
+- **Key** `retention:effective:<namespace>` (unnamespaced queues use `default`).
+  The `retention:` prefix is **reserved**: a shell's settings API must treat it
+  the way it treats `auth:` — never listed, written, or deleted through the
+  generic KV endpoints, so the published policy cannot be spoofed.
+- **Document** (snake_case; windows in **milliseconds**, `null` = keep forever):
+  `enabled`, `defaulted`, `namespace`, `reported_at` (Unix ms), and `windows` with
+  `archived_jobs_ttl_ms`, `dead_letter_ttl_ms`, `task_logs_ttl_ms`,
+  `task_metrics_ttl_ms`, `job_errors_ttl_ms`.
+- **Only the leader writes it** — a peer's config does not govern the deletes.
+  `reported_at` is rewritten every sweep, so it doubles as proof a leader is
+  still enforcing the policy.
+- **Absent = unreported**, not "retention off": no leader has swept yet. A shell
+  surfaces that state distinctly; `enabled: false` is what "off" looks like.
+- Shells read it through `scheduler::retention::read_effective_retention_json`
+  rather than parsing the key themselves.
+
 ## Types the shell produces / consumes
 - **`Job`** — `job.rs`. Fields incl. `id`, `queue`, `task_name`, `payload: Vec<u8>` (opaque),
   `status`, `priority`, `retry_count`, `max_retries`, `timeout_ms`, `unique_key`,

crates/taskito-core/src/lib.rs

@@ -25,7 +25,7 @@ pub use job::{now_millis, Job, JobCompletion, JobStatus, NewJob};
 pub use resilience::circuit_breaker::{CircuitBreakerConfig, CircuitState};
 pub use resilience::rate_limiter::RateLimitConfig;
 pub use resilience::retry::RetryPolicy;
-pub use scheduler::retention::RetentionConfig;
+pub use scheduler::retention::{EffectiveRetention, RetentionConfig};
 pub use scheduler::{
     JobResult, QueueConfig, ResultOutcome, Scheduler, SchedulerConfig, TaskConfig,
 };

crates/taskito-core/src/scheduler/maintenance.rs

@@ -3,6 +3,9 @@ use log::{error, info, warn};
 use crate::error::Result;
 use crate::job::{now_millis, NewJob};
 use crate::periodic::{next_cron_time, next_cron_time_tz};
+use crate::scheduler::retention::{
+    publish_effective_retention, EffectiveRetention, RetentionConfig, DEFAULT_NAMESPACE,
+};
 use crate::storage::{
     dead_worker_cutoff, try_lead, Storage, RETENTION_LOCK, RETENTION_LOCK_TTL_MS,
 };
@@ -43,11 +46,7 @@ fn iso_cutoff(ms: i64) -> String {
 /// Warn, once, that retention is deleting history on the recommended defaults —
 /// a policy the operator did not set. Names the queue and every active window
 /// (duration + resolved cutoff) and how to opt out.
-fn announce_default_retention(
-    namespace: &str,
-    windows: &crate::scheduler::retention::RetentionConfig,
-    now: i64,
-) {
+fn announce_default_retention(namespace: &str, windows: &RetentionConfig, now: i64) {
     // `namespace`, not a queue name: the purge predicates are not queue-scoped,
     // so the windows below apply cluster-wide within this namespace.
     let mut msg = format!(
@@ -172,6 +171,22 @@ impl Scheduler {
         Ok(())
     }
 
+    /// Record the windows this leader applies under its namespace, so a
+    /// dashboard in any process can echo the live policy rather than guess at
+    /// the defaults. Rewritten every sweep: `reported_at` then doubles as proof
+    /// a leader is still enforcing it.
+    fn publish_retention(&self, windows: &RetentionConfig, now: i64) -> Result<()> {
+        let namespace = self.namespace.as_deref().unwrap_or(DEFAULT_NAMESPACE);
+        let snapshot = EffectiveRetention {
+            enabled: !windows.is_empty(),
+            defaulted: self.config.retention_is_defaulted(),
+            namespace: namespace.to_string(),
+            reported_at: now,
+            windows: windows.clone(),
+        };
+        publish_effective_retention(&self.storage, self.namespace.as_deref(), &snapshot)
+    }
+
     /// Purge expired rows in each history table per its retention window. The
     /// per-entry TTL sweeps inside the archived/dead purges run every tick — a
     /// job or DLQ entry can carry its own `result_ttl` even when its table has no
@@ -207,11 +222,18 @@ impl Scheduler {
         // process (the election means that's the leader). Skipped for an explicit
         // or legacy config — that is the operator's own choice.
         if self.config.retention_is_defaulted() {
-            let namespace = self.namespace.as_deref().unwrap_or("default");
+            let namespace = self.namespace.as_deref().unwrap_or(DEFAULT_NAMESPACE);
             self.retention_announced
                 .call_once(|| announce_default_retention(namespace, &windows, now));
         }
 
+        // Publish before deleting: a dashboard that shows the windows only after
+        // the rows are gone explains nothing. Log-and-continue — an unreported
+        // policy must never stop the sweeps.
+        if let Err(e) = self.publish_retention(&windows, now) {
+            warn!("publishing the retention windows failed: {e}");
+        }
+
         // A window `ttl` becomes the cutoff `now - ttl`; an absent window is a
         // `None` cutoff, which the fused purges read as "per-entry only".
         let cutoff = |ttl: Option<i64>| ttl.map(|t| now.saturating_sub(t));

crates/taskito-core/src/scheduler/mod.rs

@@ -2224,6 +2224,92 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_auto_cleanup_publishes_the_effective_windows() {
+        // The elected cleaner records what it applies so a dashboard in another
+        // process can echo the live policy.
+        let storage =
+            StorageBackend::Sqlite(crate::storage::sqlite::SqliteStorage::in_memory().unwrap());
+        let scheduler = Scheduler::new(
+            storage,
+            vec!["default".to_string()],
+            SchedulerConfig::default(),
+            Some("tenant-a".to_string()),
+        );
+
+        scheduler.auto_cleanup().unwrap();
+
+        let published =
+            retention::read_effective_retention(&scheduler.storage, Some("tenant-a")).unwrap();
+        let published = published.expect("the leader must publish its windows");
+        assert!(published.enabled);
+        assert!(published.defaulted, "nothing was configured");
+        assert_eq!(published.namespace, "tenant-a");
+        assert_eq!(published.windows, retention::RetentionConfig::recommended());
+        assert!(published.reported_at > 0);
+
+        // Namespace-scoped: another tenant's dashboard sees nothing from this one.
+        assert!(
+            retention::read_effective_retention(&scheduler.storage, None)
+                .unwrap()
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn test_auto_cleanup_publishes_a_disabled_policy() {
+        // "Retention is off" is a policy worth showing — an empty config must
+        // publish as disabled, not as unreported.
+        let storage =
+            StorageBackend::Sqlite(crate::storage::sqlite::SqliteStorage::in_memory().unwrap());
+        let config = SchedulerConfig {
+            retention: Some(retention::RetentionConfig::default()),
+            ..SchedulerConfig::default()
+        };
+        let scheduler = Scheduler::new(storage, vec!["default".to_string()], config, None);
+
+        scheduler.auto_cleanup().unwrap();
+
+        let published = retention::read_effective_retention(&scheduler.storage, None)
+            .unwrap()
+            .expect("a disabled policy is still published");
+        assert!(!published.enabled);
+        assert!(!published.defaulted);
+        assert_eq!(published.namespace, retention::DEFAULT_NAMESPACE);
+    }
+
+    #[test]
+    fn test_non_leader_does_not_publish_the_windows() {
+        // Only the elected cleaner's config governs the deletes, so only it may
+        // report — otherwise a peer with a different config overwrites the truth.
+        let storage =
+            StorageBackend::Sqlite(crate::storage::sqlite::SqliteStorage::in_memory().unwrap());
+        let scheduler = Scheduler::new(
+            storage,
+            vec!["default".to_string()],
+            SchedulerConfig::default(),
+            None,
+        );
+
+        assert!(scheduler
+            .storage
+            .acquire_lock(
+                crate::storage::RETENTION_LOCK,
+                "another-worker",
+                crate::storage::RETENTION_LOCK_TTL_MS
+            )
+            .unwrap());
+
+        scheduler.auto_cleanup().unwrap();
+
+        assert!(
+            retention::read_effective_retention(&scheduler.storage, None)
+                .unwrap()
+                .is_none(),
+            "a non-leader must not publish"
+        );
+    }
+
     #[test]
     fn test_archive_retention_keeps_side_tables_with_no_window() {
         // Per-table independence: archived_jobs has a window but task_logs does

crates/taskito-core/src/scheduler/retention.rs

@@ -5,8 +5,22 @@
 //! per-entry TTL a single job or DLQ entry can carry, which is honored
 //! independently of any window here.
 
+use serde::{Deserialize, Serialize};
+
+use crate::error::Result;
+use crate::storage::Storage;
+
 const DAY_MS: i64 = 86_400_000;
 
+/// Namespace label used when a queue runs unnamespaced.
+pub const DEFAULT_NAMESPACE: &str = "default";
+
+/// Settings-key prefix under which the elected retention leader publishes the
+/// windows it applies, one key per namespace. Dashboards read it to echo the
+/// live policy; the shells treat the prefix as reserved so it can neither be
+/// listed as an editable setting nor spoofed through the settings API.
+pub const RETENTION_SETTING_PREFIX: &str = "retention:effective:";
+
 /// Recommended default windows, applied when a queue configures no retention at
 /// all. Ordered by the invariant
 /// `task_logs ≤ archived_jobs = job_errors = task_metrics ≤ dead_letter`:
@@ -24,7 +38,8 @@ pub const DEFAULT_JOB_ERRORS_TTL_MS: i64 = 7 * DAY_MS;
 
 /// Retention window per history table, in milliseconds. `None` keeps a table
 /// forever.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(default)]
 pub struct RetentionConfig {
     /// Terminal jobs (`archived_jobs`) — the artifact `get_job` reads after
     /// completion. Covers every terminal status.
@@ -86,6 +101,68 @@ impl RetentionConfig {
     }
 }
 
+/// What the elected retention leader is actually applying, as published for
+/// dashboards to echo. Windows are milliseconds; a `null` window keeps that
+/// table forever.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct EffectiveRetention {
+    /// False when no table has a window — only per-entry TTLs are swept.
+    pub enabled: bool,
+    /// True when the windows are the recommended defaults because the operator
+    /// configured nothing.
+    pub defaulted: bool,
+    /// Namespace the windows apply to. The purges are not queue-scoped, so they
+    /// cover every queue in this namespace.
+    pub namespace: String,
+    /// When the leader last published this document, in Unix milliseconds.
+    pub reported_at: i64,
+    /// The windows themselves.
+    pub windows: RetentionConfig,
+}
+
+/// Settings key holding the published windows for `namespace`.
+pub fn retention_setting_key(namespace: Option<&str>) -> String {
+    format!(
+        "{RETENTION_SETTING_PREFIX}{}",
+        namespace.unwrap_or(DEFAULT_NAMESPACE)
+    )
+}
+
+/// Publish the windows this leader applies so any dashboard can echo the live
+/// policy instead of guessing at the defaults.
+pub fn publish_effective_retention<S: Storage>(
+    storage: &S,
+    namespace: Option<&str>,
+    snapshot: &EffectiveRetention,
+) -> Result<()> {
+    let json = serde_json::to_string(snapshot)?;
+    storage.set_setting(&retention_setting_key(namespace), &json)
+}
+
+/// The windows last published for `namespace`, or `None` when no leader has
+/// reported yet — the state a dashboard sees before the first cleanup sweep.
+/// An unparseable document reads as unreported rather than failing the caller.
+pub fn read_effective_retention<S: Storage>(
+    storage: &S,
+    namespace: Option<&str>,
+) -> Result<Option<EffectiveRetention>> {
+    let Some(raw) = storage.get_setting(&retention_setting_key(namespace))? else {
+        return Ok(None);
+    };
+    Ok(serde_json::from_str(&raw).ok())
+}
+
+/// The published document for `namespace` as JSON, re-encoded from the parsed
+/// form so a shell never forwards a malformed body. `None` when unreported.
+pub fn read_effective_retention_json<S: Storage>(
+    storage: &S,
+    namespace: Option<&str>,
+) -> Result<Option<String>> {
+    read_effective_retention(storage, namespace)?
+        .map(|snapshot| serde_json::to_string(&snapshot).map_err(Into::into))
+        .transpose()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -109,4 +186,41 @@ mod tests {
         );
         assert!(!r.is_empty(), "the recommended windows are not empty");
     }
+
+    #[test]
+    fn setting_key_is_namespaced() {
+        assert_eq!(
+            retention_setting_key(Some("tenant-a")),
+            "retention:effective:tenant-a"
+        );
+        // An unnamespaced queue publishes under the same label the cleanup
+        // announcement uses, so the two always name the same thing.
+        assert_eq!(
+            retention_setting_key(None),
+            "retention:effective:default".to_string()
+        );
+    }
+
+    #[test]
+    fn published_document_round_trips() {
+        let snapshot = EffectiveRetention {
+            enabled: true,
+            defaulted: false,
+            namespace: DEFAULT_NAMESPACE.to_string(),
+            reported_at: 1_753_200_000_000,
+            windows: RetentionConfig {
+                archived_jobs_ttl_ms: Some(DAY_MS),
+                ..RetentionConfig::default()
+            },
+        };
+
+        let json = serde_json::to_string(&snapshot).unwrap();
+        assert_eq!(
+            serde_json::from_str::<EffectiveRetention>(&json).unwrap(),
+            snapshot
+        );
+        // A window that keeps its table forever must survive as an explicit
+        // null — a shell reads absent and null identically.
+        assert!(json.contains("\"dead_letter_ttl_ms\":null"));
+    }
 }

crates/taskito-java/src/queue/admin.rs

@@ -232,6 +232,27 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listSettin
     })
 }
 
+/// `String effectiveRetention(long handle)` — the retention windows the elected
+/// cleaner last published for this queue's namespace as a JSON document, or
+/// `null` if no worker has swept yet. See `BINDING_CONTRACT.md`.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_effectiveRetention<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        match taskito_core::scheduler::retention::read_effective_retention_json(
+            &queue.storage,
+            queue.namespace.as_deref(),
+        )? {
+            Some(json) => new_string(env, json),
+            None => Ok(std::ptr::null_mut()),
+        }
+    })
+}
+
 /// `String replayJob(long handle, String jobId)` — re-enqueue a copy of an
 /// existing job and record it in the replay history. Returns the new job id.
 #[no_mangle]

crates/taskito-node/src/queue/admin.rs

@@ -227,4 +227,16 @@ impl JsQueue {
     pub fn list_settings(&self) -> Result<HashMap<String, String>> {
         self.storage.list_settings().map_err(to_napi_err)
     }
+
+    /// The retention windows the elected cleaner last published for this
+    /// queue's namespace, as a JSON document, or `null` if no worker has swept
+    /// yet. See `BINDING_CONTRACT.md`.
+    #[napi]
+    pub fn effective_retention(&self) -> Result<Option<String>> {
+        taskito_core::scheduler::retention::read_effective_retention_json(
+            &self.storage,
+            self.namespace.as_deref(),
+        )
+        .map_err(to_napi_err)
+    }
 }