diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md index 1eefffdd..633d9b52 100644 --- a/crates/taskito-core/BINDING_CONTRACT.md +++ b/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:`, 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:` (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` (opaque), `status`, `priority`, `retry_count`, `max_retries`, `timeout_ms`, `unique_key`, diff --git a/crates/taskito-core/src/lib.rs b/crates/taskito-core/src/lib.rs index c91785e9..beae8c0f 100644 --- a/crates/taskito-core/src/lib.rs +++ b/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, }; diff --git a/crates/taskito-core/src/scheduler/maintenance.rs b/crates/taskito-core/src/scheduler/maintenance.rs index 08f07147..8d59ade1 100644 --- a/crates/taskito-core/src/scheduler/maintenance.rs +++ b/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| ttl.map(|t| now.saturating_sub(t)); diff --git a/crates/taskito-core/src/scheduler/mod.rs b/crates/taskito-core/src/scheduler/mod.rs index 69fe9b6f..77f327be 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/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 diff --git a/crates/taskito-core/src/scheduler/retention.rs b/crates/taskito-core/src/scheduler/retention.rs index 76924c76..1f3143ad 100644 --- a/crates/taskito-core/src/scheduler/retention.rs +++ b/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( + 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( + storage: &S, + namespace: Option<&str>, +) -> Result> { + 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( + storage: &S, + namespace: Option<&str>, +) -> Result> { + 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::(&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")); + } } diff --git a/crates/taskito-java/src/queue/admin.rs b/crates/taskito-java/src/queue/admin.rs index ad7d6ffa..f4f8120b 100644 --- a/crates/taskito-java/src/queue/admin.rs +++ b/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] diff --git a/crates/taskito-node/src/queue/admin.rs b/crates/taskito-node/src/queue/admin.rs index a402d083..13c14709 100644 --- a/crates/taskito-node/src/queue/admin.rs +++ b/crates/taskito-node/src/queue/admin.rs @@ -227,4 +227,16 @@ impl JsQueue { pub fn list_settings(&self) -> Result> { 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> { + taskito_core::scheduler::retention::read_effective_retention_json( + &self.storage, + self.namespace.as_deref(), + ) + .map_err(to_napi_err) + } } diff --git a/crates/taskito-python/src/py_queue/mod.rs b/crates/taskito-python/src/py_queue/mod.rs index ba88af64..95a5691b 100644 --- a/crates/taskito-python/src/py_queue/mod.rs +++ b/crates/taskito-python/src/py_queue/mod.rs @@ -669,6 +669,17 @@ impl PyQueue { .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) } + /// The retention windows the elected cleaner last published for this + /// queue's namespace, as a JSON document, or ``None`` if no worker has + /// swept yet. See ``BINDING_CONTRACT.md``. + pub fn effective_retention(&self) -> PyResult> { + taskito_core::scheduler::retention::read_effective_retention_json( + &self.storage, + self.namespace.as_deref(), + ) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } + /// Return all dashboard settings as a ``{key: value}`` dict. pub fn list_settings(&self) -> PyResult> { self.storage diff --git a/dashboard/src/features/settings/api.ts b/dashboard/src/features/settings/api.ts index 65b84bdc..2d854927 100644 --- a/dashboard/src/features/settings/api.ts +++ b/dashboard/src/features/settings/api.ts @@ -1,10 +1,14 @@ import { api } from "@/lib/api-client"; -import type { SettingsSnapshot } from "./types"; +import type { RetentionSnapshot, SettingsSnapshot } from "./types"; export function fetchSettings(signal?: AbortSignal): Promise { return api.get("/api/settings", { signal }); } +export function fetchRetention(signal?: AbortSignal): Promise { + return api.get("/api/retention", { signal }); +} + export interface SettingResponse { key: string; value: string; diff --git a/dashboard/src/features/settings/components/retention-section.tsx b/dashboard/src/features/settings/components/retention-section.tsx new file mode 100644 index 00000000..6c5a265d --- /dev/null +++ b/dashboard/src/features/settings/components/retention-section.tsx @@ -0,0 +1,130 @@ +import { Trash2 } from "lucide-react"; +import { + Badge, + Callout, + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, + EmptyState, + ErrorState, + Skeleton, + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui"; +import { formatRelative } from "@/lib/time"; +import { formatRetentionWindow, RETENTION_TABLES } from "../derived"; +import { useRetention } from "../hooks"; +import type { RetentionSnapshot } from "../types"; + +/** On / off / unreported — the three states an operator has to tell apart. */ +function RetentionStatus({ snapshot }: { snapshot: RetentionSnapshot }) { + if (!snapshot.reported) return Not reported; + return snapshot.enabled ? ( + + Deleting history + + ) : ( + + Disabled + + ); +} + +function WindowTable({ snapshot }: { snapshot: RetentionSnapshot }) { + return ( + + + + Table + Deleted after + + + + {RETENTION_TABLES.map(({ key, label, description }) => ( + + +
{label}
+
{description}
+ + + {formatRetentionWindow(snapshot.windows[key])} + + + ))} + +
+ ); +} + +/** + * Why rows vanish from the job, log, and dead-letter listings. Retention runs + * in the worker process, so these are the windows the elected cleaner reported + * — not this dashboard's configuration. Read-only: the windows are set where + * the worker is started. + */ +export function RetentionSection() { + const { data, isLoading, error, refetch } = useRetention(); + + return ( + + +
+
+ Retention + + How long each history table keeps a row before auto-cleanup deletes it. Set where the + worker is started; reported here by the worker that enforces it. + +
+ {data ? : null} +
+ + + {isLoading ? ( + + ) : error || !data ? ( + // Never a bare skeleton: a failed read must say so and offer a retry, + // not spin forever looking like a policy that is still loading. + refetch()} + /> + ) : !data.reported ? ( + + ) : ( + <> + {data.defaulted ? ( + + Running on the recommended defaults — no windows were configured. History is being + deleted on the schedule below. + + ) : null} + {!data.enabled ? ( + + Retention is disabled: no table has a window, so history is kept until you delete + it. Jobs and dead letters carry their own per-entry TTLs, which still apply. + + ) : ( + + )} +
+ Namespace {data.namespace} · + reported {formatRelative(data.reported_at)} +
+ + )} + + + ); +} diff --git a/dashboard/src/features/settings/derived.test.ts b/dashboard/src/features/settings/derived.test.ts index e18189c1..d3575b40 100644 --- a/dashboard/src/features/settings/derived.test.ts +++ b/dashboard/src/features/settings/derived.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from "vitest"; -import { applyJobContext, isSafeLinkUrl, parseExternalLinks } from "./derived"; +import { + applyJobContext, + formatRetentionWindow, + isSafeLinkUrl, + parseExternalLinks, +} from "./derived"; describe("isSafeLinkUrl", () => { it("allows http and https URLs", () => { @@ -119,3 +124,30 @@ describe("applyJobContext", () => { expect(applyJobContext("/jobs/{job_id}", "")).toBe("/jobs/"); }); }); + +describe("formatRetentionWindow", () => { + it("renders whole days, hours, and minutes in the largest exact unit", () => { + expect(formatRetentionWindow(7 * 86_400_000)).toBe("7 days"); + expect(formatRetentionWindow(86_400_000)).toBe("1 day"); + expect(formatRetentionWindow(12 * 3_600_000)).toBe("12 hours"); + expect(formatRetentionWindow(90 * 60_000)).toBe("90 minutes"); + }); + + it("falls back to seconds for windows that divide no larger unit", () => { + expect(formatRetentionWindow(90_000)).toBe("90 seconds"); + expect(formatRetentionWindow(1_500)).toBe("2 seconds"); + }); + + it("floors a sub-second window at one second", () => { + // Rounding these to "0 seconds" would read as the immediate purge below, + // but the rows do survive for a moment. + expect(formatRetentionWindow(1)).toBe("1 second"); + expect(formatRetentionWindow(499)).toBe("1 second"); + }); + + it("distinguishes no window from a zero window", () => { + // null = the table is never swept; 0 = purged as soon as the cleaner sees it. + expect(formatRetentionWindow(null)).toBe("Kept forever"); + expect(formatRetentionWindow(0)).toBe("Purged immediately"); + }); +}); diff --git a/dashboard/src/features/settings/derived.ts b/dashboard/src/features/settings/derived.ts index 38acaea6..23bf4b2c 100644 --- a/dashboard/src/features/settings/derived.ts +++ b/dashboard/src/features/settings/derived.ts @@ -1,7 +1,12 @@ import { useEffect } from "react"; import { site } from "@/lib/site"; import { useSettings } from "./hooks"; -import { type ExternalLink, type IntegrationUrls, SETTING_KEYS } from "./types"; +import { + type ExternalLink, + type IntegrationUrls, + type RetentionWindows, + SETTING_KEYS, +} from "./types"; /** * Whether a configured URL may safely become an ``href``. Allows http(s) @@ -127,3 +132,61 @@ export function useApplyAccent(): void { return clear; }, [value]); } + +/** + * The history tables auto-cleanup purges, ordered the way an operator reads + * them: shortest-lived first, the dead-letter queue last. Each row explains + * what disappears from which page when its window elapses. + */ +export const RETENTION_TABLES: ReadonlyArray<{ + key: keyof RetentionWindows; + label: string; + description: string; +}> = [ + { + key: "task_logs_ttl_ms", + label: "Task logs", + description: "Per-job log lines shown on the Logs page and job detail.", + }, + { + key: "archived_jobs_ttl_ms", + label: "Archived jobs", + description: "Finished jobs of every terminal status — what job detail reads after completion.", + }, + { + key: "job_errors_ttl_ms", + label: "Job errors", + description: "Per-attempt failure records behind a job's error history.", + }, + { + key: "task_metrics_ttl_ms", + label: "Task metrics", + description: "The per-run timings the Metrics charts aggregate.", + }, + { + key: "dead_letter_ttl_ms", + label: "Dead letters", + description: "The only copy of a failed job's payload — deliberately kept longest.", + }, +]; + +const MINUTE_MS = 60_000; +const HOUR_MS = 60 * MINUTE_MS; +const DAY_MS = 24 * HOUR_MS; + +/** + * A retention window as an operator-readable age. ``null`` is "kept forever" + * (no window), which is not the same as a zero window — that purges a row as + * soon as the cleaner sees it. + */ +export function formatRetentionWindow(ms: number | null): string { + if (ms === null) return "Kept forever"; + if (ms <= 0) return "Purged immediately"; + const plural = (value: number, unit: string) => `${value} ${unit}${value === 1 ? "" : "s"}`; + if (ms % DAY_MS === 0) return plural(ms / DAY_MS, "day"); + if (ms % HOUR_MS === 0) return plural(ms / HOUR_MS, "hour"); + if (ms % MINUTE_MS === 0) return plural(ms / MINUTE_MS, "minute"); + // Floor at one second: a sub-second window still keeps rows for a moment, and + // rounding it to "0 seconds" would read as the immediate purge above. + return plural(Math.max(1, Math.round(ms / 1000)), "second"); +} diff --git a/dashboard/src/features/settings/hooks.ts b/dashboard/src/features/settings/hooks.ts index 6206f085..bde889d3 100644 --- a/dashboard/src/features/settings/hooks.ts +++ b/dashboard/src/features/settings/hooks.ts @@ -1,10 +1,11 @@ import { queryOptions, useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { toast } from "sonner"; import { ApiError } from "@/lib/api-client"; -import { deleteSetting, fetchSettings, setSetting } from "./api"; +import { deleteSetting, fetchRetention, fetchSettings, setSetting } from "./api"; import type { SettingsSnapshot } from "./types"; const KEY = ["settings"] as const; +const RETENTION_KEY = ["retention"] as const; /** * Return a user-friendly error description. @@ -31,6 +32,17 @@ export function useSettings() { return useQuery(settingsQuery()); } +export function retentionQuery() { + return queryOptions({ + queryKey: RETENTION_KEY, + queryFn: ({ signal }) => fetchRetention(signal), + }); +} + +export function useRetention() { + return useQuery(retentionQuery()); +} + interface UpdateInput { key: string; value: unknown; diff --git a/dashboard/src/features/settings/index.ts b/dashboard/src/features/settings/index.ts index 2725c79b..0e787b4e 100644 --- a/dashboard/src/features/settings/index.ts +++ b/dashboard/src/features/settings/index.ts @@ -2,14 +2,30 @@ export { BrandingSection } from "./components/branding-section"; export { ExternalLinksSection } from "./components/external-links-section"; export { IntegrationsSection } from "./components/integrations-section"; export { RefreshIntervalSection } from "./components/refresh-interval-section"; +export { RetentionSection } from "./components/retention-section"; export { applyJobContext, + formatRetentionWindow, parseExternalLinks, + RETENTION_TABLES, useApplyAccent, useBranding, useExternalLinks, useIntegrations, } from "./derived"; -export { settingsQuery, useDeleteSetting, useSettings, useUpdateSetting } from "./hooks"; -export type { ExternalLink, IntegrationUrls, SettingsSnapshot } from "./types"; +export { + retentionQuery, + settingsQuery, + useDeleteSetting, + useRetention, + useSettings, + useUpdateSetting, +} from "./hooks"; +export type { + ExternalLink, + IntegrationUrls, + RetentionSnapshot, + RetentionWindows, + SettingsSnapshot, +} from "./types"; export { SETTING_KEYS } from "./types"; diff --git a/dashboard/src/features/settings/types.ts b/dashboard/src/features/settings/types.ts index 429fc037..6a55b9ef 100644 --- a/dashboard/src/features/settings/types.ts +++ b/dashboard/src/features/settings/types.ts @@ -42,3 +42,26 @@ export interface BrandOverrides { /** Raw key→value snapshot returned by ``GET /api/settings``. */ export type SettingsSnapshot = Record; + +/** Per-table retention windows in milliseconds; `null` keeps a table forever. */ +export interface RetentionWindows { + task_logs_ttl_ms: number | null; + archived_jobs_ttl_ms: number | null; + job_errors_ttl_ms: number | null; + task_metrics_ttl_ms: number | null; + dead_letter_ttl_ms: number | null; +} + +/** + * What ``GET /api/retention`` reports: the windows the elected cleaner + * published, not this process's config. ``reported: false`` means no worker + * has swept yet — distinct from retention being off (``enabled: false``). + */ +export interface RetentionSnapshot { + reported: boolean; + enabled: boolean; + defaulted: boolean; + namespace: string | null; + reported_at: number | null; + windows: RetentionWindows; +} diff --git a/dashboard/src/routes/settings.tsx b/dashboard/src/routes/settings.tsx index 62998bda..6bcf6f20 100644 --- a/dashboard/src/routes/settings.tsx +++ b/dashboard/src/routes/settings.tsx @@ -6,12 +6,21 @@ import { ExternalLinksSection, IntegrationsSection, RefreshIntervalSection, + RetentionSection, + retentionQuery, settingsQuery, useSettings, } from "@/features/settings"; export const Route = createFileRoute("/settings")({ - loader: ({ context: { queryClient } }) => queryClient.ensureQueryData(settingsQuery()), + loader: ({ context: { queryClient } }) => { + // Retention is one read-only panel: prefetch it best-effort so a backend + // that can't report its windows never costs the operator the whole page + // (branding, integrations, links). `prefetchQuery` resolves on failure; + // the section renders its own error state from the cached rejection. + void queryClient.prefetchQuery(retentionQuery()); + return queryClient.ensureQueryData(settingsQuery()); + }, component: SettingsPage, }); @@ -23,7 +32,7 @@ function SettingsPage() { {isLoading || !data ? ( @@ -42,6 +51,7 @@ function SettingsPage() {
+
diff --git a/docs/content/docs/java/guides/operations/dashboard.mdx b/docs/content/docs/java/guides/operations/dashboard.mdx index c10f5a6b..1fd5ee6f 100644 --- a/docs/content/docs/java/guides/operations/dashboard.mdx +++ b/docs/content/docs/java/guides/operations/dashboard.mdx @@ -216,6 +216,7 @@ the same contract the bundled SPA consumes. All paths below are relative to | Webhooks | `webhooks` (+ `/{id}`, `/{id}/test`, `/{id}/rotate-secret`, `/{id}/deliveries`, `/{id}/deliveries/{deliveryId}`, `/{id}/deliveries/{deliveryId}/replay`) — see [Webhooks: Dashboard management](/java/guides/extensibility/webhooks#dashboard-management) | | Workflows | `workflows/runs` (+ `/{id}`, `/{id}/dag`, `/{id}/children`) | | Settings | `settings`, `settings/{key}` | +| Retention | `retention` — the windows the worker running cleanup reported, read-only | Gating depends on the [auth mode](#auth): open mode (the default) serves diff --git a/docs/content/docs/node/guides/operations/dashboard-api.mdx b/docs/content/docs/node/guides/operations/dashboard-api.mdx index 14bd60c1..48afd2c8 100644 --- a/docs/content/docs/node/guides/operations/dashboard-api.mdx +++ b/docs/content/docs/node/guides/operations/dashboard-api.mdx @@ -23,6 +23,7 @@ directly too. All timestamps are Unix milliseconds. | `GET /api/event-types` | Known job event names. | | `GET /api/workflows/runs` · `/:id` | Workflow runs, and one run. | | `GET /api/workflows/runs/:id/dag` · `/children` | A run's DAG graph and child runs. | +| `GET /api/retention` | Retention windows the worker running cleanup reported. | | `GET /api/webhooks` · `/:id` | Webhook config. | | `GET /api/webhooks/:id/deliveries` | Paged delivery log — see below. | diff --git a/docs/content/docs/python/guides/dashboard/rest-api.mdx b/docs/content/docs/python/guides/dashboard/rest-api.mdx index 043807c4..f24787f9 100644 --- a/docs/content/docs/python/guides/dashboard/rest-api.mdx +++ b/docs/content/docs/python/guides/dashboard/rest-api.mdx @@ -584,7 +584,42 @@ itself for branding, external links, and integration URLs — but you can write your own keys here too. Note that this is the same store where authentication, webhook subscriptions, delivery logs, and runtime overrides live, all under namespaced prefixes (`auth:*`, -`webhooks:*`, `overrides:*`, etc.). +`webhooks:*`, `overrides:*`, etc.). The reserved prefixes (`auth:`, +`webhooks:`, `retention:`) read as absent here and cannot be written. + +## Retention + +### `GET /api/retention` + +The retention windows a worker is applying to this queue's namespace — what +the Settings page echoes to explain why rows leave the listings. + +```json +{ + "reported": true, + "enabled": true, + "defaulted": true, + "namespace": "default", + "reported_at": 1753200000000, + "windows": { + "task_logs_ttl_ms": 259200000, + "archived_jobs_ttl_ms": 604800000, + "job_errors_ttl_ms": 604800000, + "task_metrics_ttl_ms": 604800000, + "dead_letter_ttl_ms": 2592000000 + } +} +``` + +Retention runs in the worker process, so this is not computed from the +dashboard's own config: the worker elected to run cleanup publishes its +windows on each sweep. `reported: false` means no worker has swept yet — +the policy is unknown, which is not the same as retention being off +(`enabled: false`). A `null` window keeps that table forever. + +Read-only. The published document lives under the reserved `retention:` +settings prefix, so it is neither listed nor writable through +`/api/settings`. ## Using the API programmatically diff --git a/docs/content/docs/python/guides/operations/job-management.mdx b/docs/content/docs/python/guides/operations/job-management.mdx index 119a2b39..a6b37e31 100644 --- a/docs/content/docs/python/guides/operations/job-management.mdx +++ b/docs/content/docs/python/guides/operations/job-management.mdx @@ -109,6 +109,26 @@ queue = Queue( control; when both are set, `retention` wins. A negative `result_ttl` is rejected at construction. +### Seeing the active windows + +Retention runs in the worker process, so the resolved windows are reported by +the worker elected to run cleanup rather than read from local config. The +dashboard's **Settings** page echoes them, and `effective_retention()` returns +the same report: + +```python +policy = queue.effective_retention() +if policy is None: + print("no worker has swept yet — the active policy is unknown") +else: + print(policy.enabled, policy.defaulted, policy.windows["task_logs"]) +``` + +Windows are in **milliseconds** here (the reporting unit), and `None` keeps a +table forever. `None` for the whole report means unreported — the first sweep +publishes it — which is deliberately distinct from retention being disabled +(`enabled=False`). + ### Cascade cleanup Manual `purge_completed()` / `purge_dead()` is a blunt delete — it removes a diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index 0da8fd7c..a0da48a5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -35,6 +35,7 @@ import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.model.CircuitBreakerState; import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.EffectiveRetention; import org.byteveda.taskito.model.Job; import org.byteveda.taskito.model.JobDag; import org.byteveda.taskito.model.JobError; @@ -677,6 +678,11 @@ public Optional getSetting(String key) { return backend.getSetting(key); } + @Override + public Optional effectiveRetention() { + return backend.effectiveRetentionJson().map(json -> decode(json, EffectiveRetention.class)); + } + @Override public void setSetting(String key, String value) { backend.setSetting(key, value); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index 61113af9..4ac40d68 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -24,6 +24,7 @@ import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.model.CircuitBreakerState; import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.EffectiveRetention; import org.byteveda.taskito.model.Job; import org.byteveda.taskito.model.JobDag; import org.byteveda.taskito.model.JobError; @@ -290,6 +291,15 @@ static Builder builder() { Map listSettings(); + /** + * The retention windows a worker is applying to this queue's namespace, or + * empty when no worker has swept yet — distinct from retention being + * disabled, which reports with {@code enabled = false}. + * + * @return the published policy, or empty if unreported + */ + Optional effectiveRetention(); + // ── Middleware toggles ────────────────────────────────────────── /** diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java index d73270cb..e17f766e 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java @@ -19,6 +19,7 @@ import org.byteveda.taskito.dashboard.api.MetricsHandlers; import org.byteveda.taskito.dashboard.api.OpsHandlers; import org.byteveda.taskito.dashboard.api.OverridesHandlers; +import org.byteveda.taskito.dashboard.api.RetentionHandlers; import org.byteveda.taskito.dashboard.api.SettingsHandlers; import org.byteveda.taskito.dashboard.api.WebhooksHandlers; import org.byteveda.taskito.dashboard.api.WorkflowsHandlers; @@ -335,6 +336,7 @@ private Router buildRouter() { CoreHandlers core = new CoreHandlers(queue); SettingsAccess settings = SettingsAccess.of(queue); SettingsHandlers settingsApi = new SettingsHandlers(settings); + RetentionHandlers retention = new RetentionHandlers(queue); OverridesHandlers overrides = new OverridesHandlers(queue, new OverridesStore(settings)); MetricsHandlers metrics = new MetricsHandlers(queue); WorkflowsHandlers workflows = new WorkflowsHandlers(queue); @@ -382,6 +384,7 @@ private Router buildRouter() { r.get("/api/workflows/runs/([^/]+)", req -> workflows.run(req.param(0))); // Settings KV + r.get("/api/retention", req -> retention.retention()); r.get("/api/settings", req -> settingsApi.list()); r.get("/api/settings/(.+)", req -> settingsApi.get(req.param(0))); r.put("/api/settings/(.+)", req -> settingsApi.put(req.param(0), req.jsonBody())); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java index 6f1908ae..10e87344 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java @@ -6,6 +6,7 @@ import org.byteveda.taskito.model.CircuitBreakerState; import org.byteveda.taskito.model.DagEdge; import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.EffectiveRetention; import org.byteveda.taskito.model.Job; import org.byteveda.taskito.model.JobDag; import org.byteveda.taskito.model.QueueStats; @@ -215,6 +216,29 @@ private static Map maskHeaderValues(Map headers) return masked; } + /** + * The published retention policy, or the unreported state when {@code r} is + * null. {@code reported} is distinct from {@code enabled}: no worker has + * swept yet, which is not the same as retention being switched off. + */ + static Map retention(EffectiveRetention r) { + Map windows = new LinkedHashMap<>(); + windows.put("task_logs_ttl_ms", r == null ? null : r.windows.taskLogsMs); + windows.put("archived_jobs_ttl_ms", r == null ? null : r.windows.archivedJobsMs); + windows.put("job_errors_ttl_ms", r == null ? null : r.windows.jobErrorsMs); + windows.put("task_metrics_ttl_ms", r == null ? null : r.windows.taskMetricsMs); + windows.put("dead_letter_ttl_ms", r == null ? null : r.windows.deadLetterMs); + + Map m = new LinkedHashMap<>(); + m.put("reported", r != null); + m.put("enabled", r != null && r.enabled); + m.put("defaulted", r != null && r.defaulted); + m.put("namespace", r == null ? null : r.namespace); + m.put("reported_at", r == null ? null : r.reportedAt); + m.put("windows", windows); + return m; + } + static Map workflowNode(NodeSnapshot n) { Map m = new LinkedHashMap<>(); m.put("node_name", n.nodeName); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/RetentionHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/RetentionHandlers.java new file mode 100644 index 00000000..77f6f0a4 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/RetentionHandlers.java @@ -0,0 +1,23 @@ +package org.byteveda.taskito.dashboard.api; + +import org.byteveda.taskito.Taskito; + +/** + * Echoes the retention windows the elected cleaner published for this + * namespace, so the dashboard can explain why rows disappear from its + * listings. Retention runs in the worker process, so this is never computed + * from local config — an unreported policy is reported as such. + */ +public final class RetentionHandlers { + + private final Taskito queue; + + public RetentionHandlers(Taskito queue) { + this.queue = queue; + } + + /** The published retention policy for this queue's namespace. */ + public Object retention() { + return Contract.retention(queue.effectiveRetention().orElse(null)); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java index 244c63a4..ec09c801 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java @@ -9,15 +9,18 @@ /** * Generic settings KV API. Keys under the reserved prefixes ({@code auth:}, - * {@code webhooks:}) are treated as absent everywhere — never listed, read, + * {@code webhooks:}, {@code retention:}) are treated as absent everywhere — never listed, read, * written, or deleted through this surface — so auth and webhook state cannot be * exposed or clobbered. Keys are capped at 256 chars, values at 64 KiB. */ public final class SettingsHandlers { static final int MAX_KEY_LENGTH = 256; static final int MAX_VALUE_LENGTH = 64 * 1024; - // Hide auth state and the webhook store (persisted under the "taskito.webhooks" key). - private static final List PROTECTED_PREFIXES = List.of("auth:", "webhooks:", "taskito.webhooks"); + // Hide auth state, the webhook store (persisted under the "taskito.webhooks" + // key), and the retention windows the cleaner publishes — a report of what + // the worker does, not a knob. + private static final List PROTECTED_PREFIXES = + List.of("auth:", "webhooks:", "taskito.webhooks", "retention:"); private final SettingsAccess settings; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index 49ee1f10..213df5d6 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -231,6 +231,11 @@ public boolean deleteSetting(String key) { return withOpenHandle(() -> NativeQueue.deleteSetting(handle, key)); } + @Override + public Optional effectiveRetentionJson() { + return withOpenHandle(() -> Optional.ofNullable(NativeQueue.effectiveRetention(handle))); + } + @Override public String listSettingsJson() { return withOpenHandle(() -> NativeQueue.listSettings(handle)); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java index 80e3a49e..7344cfd7 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java @@ -99,6 +99,9 @@ private NativeQueue() {} public static native String listSettings(long handle); + /** The published retention windows as JSON, or {@code null} if unreported. */ + public static native String effectiveRetention(long handle); + // ── Logs ──────────────────────────────────────────────────────── public static native void writeTaskLog( long handle, String jobId, String taskName, String level, String message, String extraOrNull); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/EffectiveRetention.java b/sdks/java/src/main/java/org/byteveda/taskito/model/EffectiveRetention.java new file mode 100644 index 00000000..54bd8a1e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/EffectiveRetention.java @@ -0,0 +1,80 @@ +package org.byteveda.taskito.model; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * The retention windows a worker is actually applying, as reported by the + * elected cleaner on each sweep. Retention runs in the worker process, so this + * is the policy that governs the deletes — not this process's configuration. + * Windows are milliseconds; {@code null} keeps a table forever. + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class EffectiveRetention { + + /** False when no table has a window — only per-entry TTLs are swept. */ + public final boolean enabled; + + /** True when the windows are the recommended defaults, set by no one. */ + public final boolean defaulted; + + /** Namespace the windows cover. The purges are not queue-scoped. */ + public final String namespace; + + /** When the cleaner last published this, in Unix milliseconds. */ + public final long reportedAt; + + /** Per-table windows in milliseconds. */ + public final Windows windows; + + @JsonCreator + public EffectiveRetention( + @JsonProperty("enabled") boolean enabled, + @JsonProperty("defaulted") boolean defaulted, + @JsonProperty("namespace") String namespace, + @JsonProperty("reported_at") long reportedAt, + @JsonProperty("windows") Windows windows) { + this.enabled = enabled; + this.defaulted = defaulted; + this.namespace = namespace; + this.reportedAt = reportedAt; + this.windows = windows == null ? Windows.EMPTY : windows; + } + + /** How long each history table keeps a row, in milliseconds. */ + @JsonIgnoreProperties(ignoreUnknown = true) + public static final class Windows { + + static final Windows EMPTY = new Windows(null, null, null, null, null); + + /** Terminal jobs, every terminal status. */ + public final Long archivedJobsMs; + + /** Dead-letter entries — the only copy of a payload a human must act on. */ + public final Long deadLetterMs; + + /** Task logs — highest write volume, lowest per-row value. */ + public final Long taskLogsMs; + + /** Task metrics — feeds the dashboard charts. */ + public final Long taskMetricsMs; + + /** Per-attempt job errors. */ + public final Long jobErrorsMs; + + @JsonCreator + public Windows( + @JsonProperty("archived_jobs_ttl_ms") Long archivedJobsMs, + @JsonProperty("dead_letter_ttl_ms") Long deadLetterMs, + @JsonProperty("task_logs_ttl_ms") Long taskLogsMs, + @JsonProperty("task_metrics_ttl_ms") Long taskMetricsMs, + @JsonProperty("job_errors_ttl_ms") Long jobErrorsMs) { + this.archivedJobsMs = archivedJobsMs; + this.deadLetterMs = deadLetterMs; + this.taskLogsMs = taskLogsMs; + this.taskMetricsMs = taskMetricsMs; + this.jobErrorsMs = jobErrorsMs; + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index 233fcaf5..b12f1447 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -131,6 +131,15 @@ default long purgeDeadByTask(String taskName) { String listSettingsJson(); + /** + * The retention windows the elected cleaner last published for this queue's + * namespace, as JSON, or empty when no worker has swept yet. Defaults to + * empty for backends that do not report one. + */ + default Optional effectiveRetentionJson() { + return Optional.empty(); + } + // ── Logs ──────────────────────────────────────────────────────── void writeTaskLog(String jobId, String taskName, String level, String message, String extraOrNull); diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardRetentionTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardRetentionTest.java new file mode 100644 index 00000000..7e706043 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardRetentionTest.java @@ -0,0 +1,104 @@ +package org.byteveda.taskito.dashboard; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.model.EffectiveRetention; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** The retention echo: the windows the elected cleaner publishes for this namespace. */ +@Timeout(30) +class DashboardRetentionTest { + + private static final long DAY_MS = 86_400_000L; + + /** The document the elected cleaner publishes — see {@code BINDING_CONTRACT.md}. */ + private static final String PUBLISHED_KEY = "retention:effective:default"; + + private static final String PUBLISHED = "{\"enabled\":true,\"defaulted\":true,\"namespace\":\"default\"," + + "\"reported_at\":1753200000000,\"windows\":{\"archived_jobs_ttl_ms\":604800000," + + "\"dead_letter_ttl_ms\":2592000000,\"task_logs_ttl_ms\":259200000," + + "\"task_metrics_ttl_ms\":604800000,\"job_errors_ttl_ms\":null}}"; + + private static Taskito queue(Path dir) { + return Taskito.builder().sqlite(dir.resolve("t.db").toString()).open(); + } + + @Test + void effectiveRetentionIsEmptyUntilACleanerPublishes(@TempDir Path dir) { + try (Taskito queue = queue(dir)) { + // Unreported is not "off": no worker has swept, so nothing is known yet. + assertTrue(queue.effectiveRetention().isEmpty()); + } + } + + @Test + void effectiveRetentionParsesThePublishedDocument(@TempDir Path dir) { + try (Taskito queue = queue(dir)) { + queue.setSetting(PUBLISHED_KEY, PUBLISHED); + + EffectiveRetention snapshot = queue.effectiveRetention().orElseThrow(); + assertTrue(snapshot.enabled); + assertTrue(snapshot.defaulted); + assertEquals("default", snapshot.namespace); + assertEquals(1753200000000L, snapshot.reportedAt); + assertEquals(3 * DAY_MS, snapshot.windows.taskLogsMs); + assertEquals(30 * DAY_MS, snapshot.windows.deadLetterMs); + // A table with no window is kept forever, not purged. + assertEquals(null, snapshot.windows.jobErrorsMs); + } + } + + @Test + void retentionEndpointReportsNothingBeforeASweep(@TempDir Path dir) throws Exception { + try (Taskito queue = queue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + + String body = client.get("/api/retention").body(); + assertTrue(body.contains("\"reported\":false"), body); + assertTrue(body.contains("\"enabled\":false"), body); + assertTrue(body.contains("\"namespace\":null"), body); + assertTrue(body.contains("\"task_logs_ttl_ms\":null"), body); + } + } + + @Test + void retentionEndpointEchoesThePublishedWindows(@TempDir Path dir) throws Exception { + try (Taskito queue = queue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + queue.setSetting(PUBLISHED_KEY, PUBLISHED); + + String body = client.get("/api/retention").body(); + assertTrue(body.contains("\"reported\":true"), body); + assertTrue(body.contains("\"defaulted\":true"), body); + assertTrue(body.contains("\"reported_at\":1753200000000"), body); + assertTrue(body.contains("\"task_logs_ttl_ms\":259200000"), body); + assertTrue(body.contains("\"dead_letter_ttl_ms\":2592000000"), body); + } + } + + @Test + void publishedWindowsAreNotAnEditableSetting(@TempDir Path dir) throws Exception { + try (Taskito queue = queue(dir); + DashboardServer server = DashboardServer.start(queue, 0)) { + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + queue.setSetting(PUBLISHED_KEY, PUBLISHED); + + // A report of what the worker does, not a knob: never listed as an + // editable row, never spoofable through the generic KV endpoints. + assertFalse(client.get("/api/settings").body().contains(PUBLISHED_KEY)); + assertEquals(404, client.get("/api/settings/" + PUBLISHED_KEY).statusCode()); + assertEquals( + 400, + client.put("/api/settings/" + PUBLISHED_KEY, "{\"value\":\"{}\"}") + .statusCode()); + } + } +} diff --git a/sdks/node/src/dashboard/handlers/index.ts b/sdks/node/src/dashboard/handlers/index.ts index 34f27269..3bd9e3d7 100644 --- a/sdks/node/src/dashboard/handlers/index.ts +++ b/sdks/node/src/dashboard/handlers/index.ts @@ -4,4 +4,5 @@ export * from "./metrics"; export * from "./middleware"; export * from "./ops"; export * from "./overrides"; +export * from "./retention"; export * from "./settings"; diff --git a/sdks/node/src/dashboard/handlers/retention.ts b/sdks/node/src/dashboard/handlers/retention.ts new file mode 100644 index 00000000..979a6db7 --- /dev/null +++ b/sdks/node/src/dashboard/handlers/retention.ts @@ -0,0 +1,34 @@ +// Retention route handler. Echoes the windows the elected cleaner published for +// this namespace, so the dashboard can explain why rows disappear from its +// listings. Retention runs in the worker process, so this is never computed +// from local config — an unreported policy is reported as such. + +import type { Queue } from "../../index"; +import type { EffectiveRetention } from "../../types"; + +/** Every table's window in ms, `null` for keep-forever and unreported. */ +function windowsToContract(snapshot: EffectiveRetention | null) { + const windows = snapshot?.windows; + return { + task_logs_ttl_ms: windows?.taskLogs ?? null, + archived_jobs_ttl_ms: windows?.archivedJobs ?? null, + job_errors_ttl_ms: windows?.jobErrors ?? null, + task_metrics_ttl_ms: windows?.taskMetrics ?? null, + dead_letter_ttl_ms: windows?.deadLetter ?? null, + }; +} + +/** The published retention policy for this queue's namespace. */ +export function retention(queue: Queue) { + const snapshot = queue.effectiveRetention(); + return { + // Distinct from `enabled`: no worker has swept yet, so nothing is known + // about the policy — not the same as retention being switched off. + reported: snapshot !== null, + enabled: snapshot?.enabled ?? false, + defaulted: snapshot?.defaulted ?? false, + namespace: snapshot?.namespace ?? null, + reported_at: snapshot?.reportedAt ?? null, + windows: windowsToContract(snapshot), + }; +} diff --git a/sdks/node/src/dashboard/handlers/settings.ts b/sdks/node/src/dashboard/handlers/settings.ts index 78869902..f7673a6f 100644 --- a/sdks/node/src/dashboard/handlers/settings.ts +++ b/sdks/node/src/dashboard/handlers/settings.ts @@ -12,8 +12,9 @@ const MAX_VALUE_LENGTH = 64 * 1024; // 64 KiB — enough for any dashboard confi // public settings API. `auth:` holds password hashes and live sessions // (reading or overwriting these is a full auth bypass); `webhook`-prefixed // keys hold subscription rows with plaintext HMAC secrets plus delivery -// logs. Protected keys are treated as absent. -const PROTECTED_PREFIXES = ["auth:", "webhooks:", "webhook:"]; +// logs. `retention:` holds the windows the cleaner publishes — a report of +// what the worker does, not a knob. Protected keys are treated as absent. +const PROTECTED_PREFIXES = ["auth:", "webhooks:", "webhook:", "retention:"]; const isProtected = (key: string): boolean => PROTECTED_PREFIXES.some((p) => key.startsWith(p)); diff --git a/sdks/node/src/dashboard/routes.ts b/sdks/node/src/dashboard/routes.ts index 5a668f07..f1c48ff5 100644 --- a/sdks/node/src/dashboard/routes.ts +++ b/sdks/node/src/dashboard/routes.ts @@ -89,6 +89,7 @@ export const routes: Route[] = [ handle: (q) => h.purgeDeadLetters(q), }, { method: "GET", pattern: /^\/api\/settings$/, handle: (q) => h.listSettings(q) }, + { method: "GET", pattern: /^\/api\/retention$/, handle: (q) => h.retention(q) }, { method: "GET", pattern: /^\/api\/settings\/(.+)$/, diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index 0979d5ab..2ec7526a 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -84,6 +84,7 @@ export type { CursorJobFilter, DeadJob, DetailedJobFilter, + EffectiveRetention, EnqueueOptions, Job, JobError, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 1f2041bd..fba2e610 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -42,6 +42,7 @@ import { ResourceRuntime, type ResourceScope, } from "./resources"; +import { parseEffectiveRetention } from "./retention"; import { CodecSerializer, deserializeCall, @@ -58,6 +59,7 @@ import type { DeadJob, DeclaredTopic, DetailedJobFilter, + EffectiveRetention, EnqueueOptions, Job, JobDag, @@ -1217,6 +1219,16 @@ export class Queue { return this.native.listSettings(); } + /** + * The retention windows a worker is applying to this namespace, or `null` + * when no worker has swept yet — distinct from retention being disabled, + * which reports with `enabled: false`. + */ + effectiveRetention(): EffectiveRetention | null { + const raw = this.native.effectiveRetention(); + return raw === null ? null : parseEffectiveRetention(raw); + } + // ── Task & queue overrides (dashboard-tunable runtime config) ───── /** Every persisted task override keyed by task name. */ diff --git a/sdks/node/src/retention.ts b/sdks/node/src/retention.ts new file mode 100644 index 00000000..9eb024c0 --- /dev/null +++ b/sdks/node/src/retention.ts @@ -0,0 +1,37 @@ +// Parses the retention document the elected cleaner publishes. Retention runs +// in the worker process, so this is how any other process — a dashboard, an +// admin script — learns the policy that governs the deletes. +// See `crates/taskito-core/BINDING_CONTRACT.md`. + +import type { EffectiveRetention } from "./types"; + +/** Core's window field for a table, in milliseconds. `null` = keep forever. */ +type PublishedWindows = Record; + +const window = (windows: PublishedWindows, table: string): number | null => + windows[`${table}_ttl_ms`] ?? null; + +/** Parse the published JSON document into the SDK's camelCase shape. */ +export function parseEffectiveRetention(raw: string): EffectiveRetention { + const doc = JSON.parse(raw) as { + enabled?: boolean; + defaulted?: boolean; + namespace?: string; + reported_at?: number; + windows?: PublishedWindows; + }; + const windows = doc.windows ?? {}; + return { + enabled: Boolean(doc.enabled), + defaulted: Boolean(doc.defaulted), + namespace: doc.namespace ?? "default", + reportedAt: doc.reported_at ?? 0, + windows: { + archivedJobs: window(windows, "archived_jobs"), + deadLetter: window(windows, "dead_letter"), + taskLogs: window(windows, "task_logs"), + taskMetrics: window(windows, "task_metrics"), + jobErrors: window(windows, "job_errors"), + }, + }; +} diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index 8ffee168..9f70c0d6 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -315,3 +315,28 @@ export interface RetentionOptions { /** Per-attempt job errors. */ jobErrors?: number; } + +/** + * The windows a worker is actually applying, as reported by the elected + * cleaner on each sweep. Retention runs in the worker process, so this is the + * policy that governs the deletes — not this process's configuration. + * Windows are **milliseconds**; `null` keeps a table forever. + */ +export interface EffectiveRetention { + /** False when no table has a window — only per-entry TTLs are swept. */ + enabled: boolean; + /** True when the windows are the recommended defaults, set by no one. */ + defaulted: boolean; + /** Namespace the windows cover. The purges are not queue-scoped. */ + namespace: string; + /** When the cleaner last published this, in Unix milliseconds. */ + reportedAt: number; + /** Per-table windows in milliseconds. */ + windows: { + archivedJobs: number | null; + deadLetter: number | null; + taskLogs: number | null; + taskMetrics: number | null; + jobErrors: number | null; + }; +} diff --git a/sdks/node/test/dashboard/retention.test.ts b/sdks/node/test/dashboard/retention.test.ts new file mode 100644 index 00000000..cc1eaafd --- /dev/null +++ b/sdks/node/test/dashboard/retention.test.ts @@ -0,0 +1,138 @@ +import { execSync } from "node:child_process"; +import { once } from "node:events"; +import { existsSync, mkdtempSync } from "node:fs"; +import type { Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { seedAdminAndSession } from "../../src/dashboard/testing"; +import { Queue, serveDashboard } from "../../src/index"; + +const pkgRoot = fileURLToPath(new URL("../..", import.meta.url)); +const staticDir = join(pkgRoot, "static", "dashboard"); +const DAY_MS = 86_400_000; + +// The document the elected cleaner publishes — see `BINDING_CONTRACT.md`. +const PUBLISHED_KEY = "retention:effective:default"; +const published = JSON.stringify({ + enabled: true, + defaulted: true, + namespace: "default", + reported_at: 1_753_200_000_000, + windows: { + archived_jobs_ttl_ms: 7 * DAY_MS, + dead_letter_ttl_ms: 30 * DAY_MS, + task_logs_ttl_ms: 3 * DAY_MS, + task_metrics_ttl_ms: 7 * DAY_MS, + job_errors_ttl_ms: null, + }, +}); + +beforeAll(() => { + if (!existsSync(join(staticDir, "index.html"))) { + execSync("pnpm run build:dashboard", { cwd: pkgRoot, stdio: "ignore" }); + } +}, 120_000); + +let server: Server | undefined; +let queue: Queue; +let base = ""; +let headers: Record = {}; + +beforeEach(async () => { + const db = join(mkdtempSync(join(tmpdir(), "taskito-retention-")), "q.db"); + queue = new Queue({ dbPath: db }); + ({ headers } = await seedAdminAndSession(queue)); + server = serveDashboard(queue, { port: 0, staticDir, secureCookies: false }); + await once(server, "listening"); + base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; +}); + +afterEach(() => { + server?.close(); + server = undefined; +}); + +const get = async (path: string) => + (await (await fetch(`${base}${path}`, { headers })).json()) as Record; + +describe("effectiveRetention", () => { + it("is null until a cleaner publishes", () => { + // Unreported is not "off": no worker has swept, so nothing is known yet. + expect(queue.effectiveRetention()).toBeNull(); + }); + + it("parses the published document", () => { + queue.setSetting(PUBLISHED_KEY, published); + + expect(queue.effectiveRetention()).toEqual({ + enabled: true, + defaulted: true, + namespace: "default", + reportedAt: 1_753_200_000_000, + windows: { + archivedJobs: 7 * DAY_MS, + deadLetter: 30 * DAY_MS, + taskLogs: 3 * DAY_MS, + taskMetrics: 7 * DAY_MS, + // A table with no window is kept forever, not purged. + jobErrors: null, + }, + }); + }); +}); + +describe("retention api", () => { + it("reports nothing before a sweep", async () => { + const body = await get("/api/retention"); + + expect(body.reported).toBe(false); + expect(body.enabled).toBe(false); + expect(body.namespace).toBeNull(); + expect(body.reported_at).toBeNull(); + expect(body.windows).toEqual({ + task_logs_ttl_ms: null, + archived_jobs_ttl_ms: null, + job_errors_ttl_ms: null, + task_metrics_ttl_ms: null, + dead_letter_ttl_ms: null, + }); + }); + + it("echoes the published windows", async () => { + queue.setSetting(PUBLISHED_KEY, published); + + const body = await get("/api/retention"); + + expect(body.reported).toBe(true); + expect(body.enabled).toBe(true); + expect(body.defaulted).toBe(true); + expect(body.namespace).toBe("default"); + expect(body.reported_at).toBe(1_753_200_000_000); + expect(body.windows).toEqual({ + task_logs_ttl_ms: 3 * DAY_MS, + archived_jobs_ttl_ms: 7 * DAY_MS, + job_errors_ttl_ms: null, + task_metrics_ttl_ms: 7 * DAY_MS, + dead_letter_ttl_ms: 30 * DAY_MS, + }); + }); + + it("keeps the published policy out of the settings api", async () => { + queue.setSetting(PUBLISHED_KEY, published); + + // A report of what the worker does, not a knob: never listed as an + // editable row, never spoofable through the generic KV endpoints. + expect(Object.keys(await get("/api/settings"))).not.toContain(PUBLISHED_KEY); + expect((await fetch(`${base}/api/settings/${PUBLISHED_KEY}`, { headers })).status).toBe(404); + + const spoof = await fetch(`${base}/api/settings/${PUBLISHED_KEY}`, { + method: "PUT", + headers: { ...headers, "content-type": "application/json" }, + body: JSON.stringify({ value: "{}" }), + }); + expect(spoof.status).toBe(400); + }); +}); diff --git a/sdks/python/taskito/__init__.py b/sdks/python/taskito/__init__.py index 8167cf77..a6ce34a2 100644 --- a/sdks/python/taskito/__init__.py +++ b/sdks/python/taskito/__init__.py @@ -49,7 +49,7 @@ from taskito.notes import MAX_NOTE_FIELDS from taskito.proxies.no_proxy import NoProxy from taskito.result import JobResult -from taskito.retention import Retention +from taskito.retention import EffectiveRetention, Retention from taskito.serializers import ( CborSerializer, CloudpickleSerializer, @@ -75,6 +75,7 @@ "CloudpickleSerializer", "CodecSerializer", "CryptoError", + "EffectiveRetention", "EncryptedSerializer", "EventType", "GzipCodec", diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi index b9f7c7be..39b6e6b4 100644 --- a/sdks/python/taskito/_taskito.pyi +++ b/sdks/python/taskito/_taskito.pyi @@ -300,6 +300,7 @@ class PyQueue: def set_setting(self, key: str, value: str) -> None: ... def delete_setting(self, key: str) -> bool: ... def list_settings(self) -> dict[str, str]: ... + def effective_retention(self) -> str | None: ... def archive_old_jobs(self, older_than_seconds: int) -> int: ... def list_archived(self, limit: int = 50, offset: int = 0) -> list[PyJob]: ... def list_archived_after( diff --git a/sdks/python/taskito/async_support/mixins.py b/sdks/python/taskito/async_support/mixins.py index e29f28a8..3279bee3 100644 --- a/sdks/python/taskito/async_support/mixins.py +++ b/sdks/python/taskito/async_support/mixins.py @@ -18,6 +18,7 @@ from taskito.context import LogLevel from taskito.pagination import Page from taskito.result import JobResult + from taskito.retention import EffectiveRetention logger = logging.getLogger("taskito") @@ -128,6 +129,7 @@ def list_archived(self, limit: int = ..., offset: int = ...) -> list[JobResult]: def pause(self, queue_name: str = ...) -> None: ... def resume(self, queue_name: str = ...) -> None: ... def paused_queues(self) -> list[str]: ... + def effective_retention(self) -> EffectiveRetention | None: ... def resource_status(self) -> list[dict[str, Any]]: ... def reload_resources(self, names: list[str] | None = ...) -> dict[str, bool]: ... @@ -291,6 +293,10 @@ async def apaused_queues(self) -> list[str]: """Async version of :meth:`paused_queues`.""" return await self._run_sync(self.paused_queues) + async def aeffective_retention(self) -> EffectiveRetention | None: + """Async version of :meth:`effective_retention`.""" + return await self._run_sync(self.effective_retention) + # -- Locks -- def alock( diff --git a/sdks/python/taskito/dashboard/handlers/retention.py b/sdks/python/taskito/dashboard/handlers/retention.py new file mode 100644 index 00000000..b68b169c --- /dev/null +++ b/sdks/python/taskito/dashboard/handlers/retention.py @@ -0,0 +1,37 @@ +"""Retention route handler. + +Echoes the windows the elected cleaner published for this namespace, so the +dashboard can explain why rows disappear from its listings. Retention runs in +the worker process, so this is never computed from local config — an +unreported policy is reported as such rather than guessed at. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from taskito.retention import RETENTION_TABLES, EffectiveRetention + +if TYPE_CHECKING: + from taskito.app import Queue + + +def _windows(snapshot: EffectiveRetention | None) -> dict[str, int | None]: + """Every table's window in ms, ``None`` for keep-forever and unreported.""" + windows = snapshot.windows if snapshot else {} + return {f"{table}_ttl_ms": windows.get(table) for table in RETENTION_TABLES} + + +def _handle_retention(queue: Queue, _qs: dict) -> dict[str, Any]: + """Return the published retention policy for this queue's namespace.""" + snapshot = queue.effective_retention() + return { + # Distinct from `enabled`: no worker has swept yet, so nothing is known + # about the policy — not the same as retention being switched off. + "reported": snapshot is not None, + "enabled": snapshot.enabled if snapshot else False, + "defaulted": snapshot.defaulted if snapshot else False, + "namespace": snapshot.namespace if snapshot else None, + "reported_at": snapshot.reported_at if snapshot else None, + "windows": _windows(snapshot), + } diff --git a/sdks/python/taskito/dashboard/handlers/settings.py b/sdks/python/taskito/dashboard/handlers/settings.py index 7ec3590a..5c91d411 100644 --- a/sdks/python/taskito/dashboard/handlers/settings.py +++ b/sdks/python/taskito/dashboard/handlers/settings.py @@ -27,9 +27,11 @@ # is a full auth bypass. ``webhooks:`` holds subscription rows with plaintext # HMAC signing secrets plus delivery logs (full request payloads / response # bodies); the dedicated webhooks API redacts the secret to ``has_secret``, so -# these must stay out of the generic settings endpoints too. The settings -# handlers treat every protected key as if absent. -_PROTECTED_PREFIXES = ("auth:", "webhooks:") +# these must stay out of the generic settings endpoints too. ``retention:`` +# holds the windows the cleaner publishes — a report of what the worker does, +# not a knob, and writable only by the cleaner itself. The settings handlers +# treat every protected key as if absent. +_PROTECTED_PREFIXES = ("auth:", "webhooks:", "retention:") def _is_protected(key: str) -> bool: diff --git a/sdks/python/taskito/dashboard/routes.py b/sdks/python/taskito/dashboard/routes.py index 3a66459d..34044b66 100644 --- a/sdks/python/taskito/dashboard/routes.py +++ b/sdks/python/taskito/dashboard/routes.py @@ -55,6 +55,7 @@ handle_put_task_override, ) from taskito.dashboard.handlers.queues import _handle_stats_queues +from taskito.dashboard.handlers.retention import _handle_retention from taskito.dashboard.handlers.scaler import build_scaler_response from taskito.dashboard.handlers.settings import ( _handle_delete_setting, @@ -145,6 +146,7 @@ def is_public_path(path: str) -> bool: "/api/stats/queues": _handle_stats_queues, "/api/scaler": lambda q, qs: build_scaler_response(q, queue_name=qs.get("queue", [None])[0]), "/api/settings": _handle_list_settings, + "/api/retention": _handle_retention, "/api/webhooks": handle_list_webhooks, "/api/event-types": handle_list_event_types, "/api/tasks": handle_list_tasks, diff --git a/sdks/python/taskito/mixins/inspection.py b/sdks/python/taskito/mixins/inspection.py index e6d75e04..b85a0a28 100644 --- a/sdks/python/taskito/mixins/inspection.py +++ b/sdks/python/taskito/mixins/inspection.py @@ -9,6 +9,7 @@ from taskito.mixins._shared import _UNSET from taskito.pagination import Page from taskito.result import JobResult +from taskito.retention import EffectiveRetention class QueueInspectionMixin: @@ -262,6 +263,19 @@ def purge_completed(self, older_than: int = 86400) -> int: """Delete completed jobs older than a given age.""" return self._inner.purge_completed(older_than) # type: ignore[no-any-return] + def effective_retention(self) -> EffectiveRetention | None: + """The retention windows a worker is applying to this namespace. + + Reported by the elected cleaner on each sweep, so it reflects the + worker's configuration rather than this process's. ``None`` means no + worker has swept yet — distinct from retention being disabled, which + reports with ``enabled=False``. + """ + raw = self._inner.effective_retention() + if raw is None: + return None + return EffectiveRetention._from_json(raw) + def _aggregate_metrics(raw: list[dict]) -> dict[str, Any]: """Aggregate raw metric rows into per-task statistics.""" diff --git a/sdks/python/taskito/retention.py b/sdks/python/taskito/retention.py index ccfe8e7f..a4370d4f 100644 --- a/sdks/python/taskito/retention.py +++ b/sdks/python/taskito/retention.py @@ -2,7 +2,9 @@ from __future__ import annotations +import json from dataclasses import dataclass +from typing import Any @dataclass(frozen=True) @@ -51,3 +53,49 @@ def _as_map(self) -> dict[str, int]: "job_errors": self.job_errors, } return {table: secs for table, secs in fields.items() if secs is not None} + + +#: History tables auto-cleanup purges, in the order operators read them — +#: shortest-lived first, the dead-letter queue last. +RETENTION_TABLES: tuple[str, ...] = ( + "task_logs", + "archived_jobs", + "job_errors", + "task_metrics", + "dead_letter", +) + + +@dataclass(frozen=True) +class EffectiveRetention: + """The windows a worker is actually applying, as reported by the cleaner. + + Retention runs in the worker process, so this is published by the elected + cleaner on each sweep rather than read from local config — a dashboard or + admin script in another process sees the policy that governs the deletes. + Windows are **milliseconds** (the wire unit); ``None`` keeps a table forever. + """ + + enabled: bool + """False when no table has a window — only per-entry TTLs are swept.""" + defaulted: bool + """True when the windows are the recommended defaults, set by no one.""" + namespace: str + """Namespace the windows cover. The purges are not queue-scoped.""" + reported_at: int + """When the cleaner last published this, in Unix milliseconds.""" + windows: dict[str, int | None] + """``{table: window_ms}`` for every table in :data:`RETENTION_TABLES`.""" + + @classmethod + def _from_json(cls, raw: str) -> EffectiveRetention: + """Parse the document the core publishes. See ``BINDING_CONTRACT.md``.""" + doc: dict[str, Any] = json.loads(raw) + published: dict[str, Any] = doc.get("windows") or {} + return cls( + enabled=bool(doc.get("enabled", False)), + defaulted=bool(doc.get("defaulted", False)), + namespace=str(doc.get("namespace", "default")), + reported_at=int(doc.get("reported_at", 0)), + windows={table: published.get(f"{table}_ttl_ms") for table in RETENTION_TABLES}, + ) diff --git a/sdks/python/tests/dashboard/test_dashboard_retention.py b/sdks/python/tests/dashboard/test_dashboard_retention.py new file mode 100644 index 00000000..59c1fcb8 --- /dev/null +++ b/sdks/python/tests/dashboard/test_dashboard_retention.py @@ -0,0 +1,199 @@ +"""Tests for the retention echo. + +Covers: +- ``Queue.effective_retention`` before and after a worker sweep +- ``GET /api/retention`` +- the reserved ``retention:`` settings namespace +""" + +from __future__ import annotations + +import threading +import urllib.error +from collections.abc import Callable, Generator +from contextlib import contextmanager +from pathlib import Path + +import pytest + +from taskito import Queue, Retention +from taskito.dashboard._testing import AuthedClient, seed_admin_and_session + +DAY_MS = 86_400_000 + +PollUntil = Callable[..., None] + + +@pytest.fixture +def queue(tmp_path: Path) -> Queue: + # cleanup_interval=1 sweeps on the first maintenance tick, so the cleaner + # publishes without the test waiting out the production cadence. + return Queue(db_path=str(tmp_path / "retention.db"), scheduler_cleanup_interval=1) + + +@pytest.fixture +def dashboard_server(queue: Queue) -> Generator[tuple[AuthedClient, Queue]]: + from http.server import ThreadingHTTPServer + + from taskito.dashboard import _make_handler + + handler = _make_handler(queue) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + session = seed_admin_and_session(queue) + client = AuthedClient(base=f"http://127.0.0.1:{port}", session=session) + try: + yield client, queue + finally: + server.shutdown() + + +@contextmanager +def _worker_that_published(queue: Queue, poll_until: PollUntil) -> Generator[None]: + """Run a worker until it has published its retention windows.""" + thread = threading.Thread(target=queue.run_worker, daemon=True) + thread.start() + try: + poll_until( + lambda: queue.effective_retention() is not None, + timeout=10, + message="the cleaner never published its retention windows", + ) + yield + finally: + queue.shutdown() + thread.join(timeout=5) + + +# ── Python API ────────────────────────────────────────── + + +def test_effective_retention_is_none_before_any_sweep(queue: Queue) -> None: + # Unreported is not "off": no worker has run, so nothing is known yet. + assert queue.effective_retention() is None + + +def test_worker_publishes_the_recommended_defaults(queue: Queue, poll_until: PollUntil) -> None: + with _worker_that_published(queue, poll_until): + snapshot = queue.effective_retention() + assert snapshot is not None + assert snapshot.enabled + assert snapshot.defaulted, "nothing was configured, so these are the defaults" + assert snapshot.namespace == "default" + assert snapshot.reported_at > 0 + assert snapshot.windows == { + "task_logs": 3 * DAY_MS, + "archived_jobs": 7 * DAY_MS, + "job_errors": 7 * DAY_MS, + "task_metrics": 7 * DAY_MS, + "dead_letter": 30 * DAY_MS, + } + + +def test_worker_publishes_explicit_windows(tmp_path: Path, poll_until: PollUntil) -> None: + queue = Queue( + db_path=str(tmp_path / "explicit.db"), + scheduler_cleanup_interval=1, + retention=Retention(task_logs=3600, dead_letter=86400), + ) + with _worker_that_published(queue, poll_until): + snapshot = queue.effective_retention() + assert snapshot is not None + assert snapshot.enabled + assert not snapshot.defaulted + assert snapshot.windows["task_logs"] == 3_600_000 + assert snapshot.windows["dead_letter"] == 86_400_000 + # Tables the operator left out keep forever. + assert snapshot.windows["archived_jobs"] is None + + +def test_worker_publishes_a_disabled_policy(tmp_path: Path, poll_until: PollUntil) -> None: + # An empty Retention() opts out — that is a policy worth showing, so it + # publishes as disabled rather than staying unreported. + queue = Queue( + db_path=str(tmp_path / "disabled.db"), + scheduler_cleanup_interval=1, + retention=Retention(), + ) + with _worker_that_published(queue, poll_until): + snapshot = queue.effective_retention() + assert snapshot is not None + assert not snapshot.enabled + assert not snapshot.defaulted + assert all(window is None for window in snapshot.windows.values()) + + +def test_effective_retention_is_namespace_scoped(tmp_path: Path, poll_until: PollUntil) -> None: + db = str(tmp_path / "shared.db") + worker_queue = Queue(db_path=db, scheduler_cleanup_interval=1, namespace="tenant-a") + with _worker_that_published(worker_queue, poll_until): + # A second tenant on the same storage has its own policy — and none yet. + other = Queue(db_path=db, namespace="tenant-b") + assert other.effective_retention() is None + assert worker_queue.effective_retention() is not None + + +# ── HTTP endpoint ─────────────────────────────────────── + + +def test_retention_endpoint_reports_nothing_before_a_sweep( + dashboard_server: tuple[AuthedClient, Queue], +) -> None: + client, _ = dashboard_server + body = client.get("/api/retention") + assert body["reported"] is False + assert body["enabled"] is False + assert body["namespace"] is None + assert body["reported_at"] is None + assert body["windows"] == { + "task_logs_ttl_ms": None, + "archived_jobs_ttl_ms": None, + "job_errors_ttl_ms": None, + "task_metrics_ttl_ms": None, + "dead_letter_ttl_ms": None, + } + + +def test_retention_endpoint_echoes_the_published_windows( + dashboard_server: tuple[AuthedClient, Queue], poll_until: PollUntil +) -> None: + client, queue = dashboard_server + with _worker_that_published(queue, poll_until): + body = client.get("/api/retention") + assert body["reported"] is True + assert body["enabled"] is True + assert body["defaulted"] is True + assert body["namespace"] == "default" + assert body["reported_at"] > 0 + assert body["windows"]["task_logs_ttl_ms"] == 3 * DAY_MS + assert body["windows"]["dead_letter_ttl_ms"] == 30 * DAY_MS + + +# ── Reserved settings namespace ───────────────────────── + + +def test_published_windows_are_not_an_editable_setting( + dashboard_server: tuple[AuthedClient, Queue], poll_until: PollUntil +) -> None: + client, queue = dashboard_server + with _worker_that_published(queue, poll_until): + # The published policy is a report, not a knob: it must not show up as + # an editable settings row, and must not be spoofable through the KV API. + assert not any(key.startswith("retention:") for key in client.get("/api/settings")) + + with pytest.raises(urllib.error.HTTPError) as read: + client.get("/api/settings/retention:effective:default") + assert read.value.code == 404 + + with pytest.raises(urllib.error.HTTPError) as write: + client.put("/api/settings/retention:effective:default", {"value": "{}"}) + assert write.value.code == 400 + + +async def test_async_effective_retention(queue: Queue, poll_until: PollUntil) -> None: + with _worker_that_published(queue, poll_until): + snapshot = await queue.aeffective_retention() + assert snapshot is not None + assert snapshot.enabled