From d2be56fae4104e3c085fa250bc95b564c0944c05 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:21:33 +0530 Subject: [PATCH 1/7] fix(node): treat an empty retention config as disabled Presence, not emptiness, decides: an empty retention now reaches the core as Some(empty) to disable auto-cleanup, distinct from omitting it (recommended defaults). Prereq for retention-on-by-default. --- crates/taskito-node/src/config.rs | 19 +++++++++++-------- crates/taskito-node/src/worker.rs | 4 +++- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs index 7c28463c..95d43f62 100644 --- a/crates/taskito-node/src/config.rs +++ b/crates/taskito-node/src/config.rs @@ -196,17 +196,18 @@ pub struct RetentionInput { } impl RetentionInput { - /// Build a core [`RetentionConfig`], or `None` when no window is set. - pub fn to_config(&self) -> Option { + /// Build a core [`RetentionConfig`]. An all-unset input yields an empty + /// config, which disables retention — deliberately distinct from *omitting* + /// the `retention` option, which falls back to the recommended defaults. + pub fn to_config(&self) -> taskito_core::scheduler::retention::RetentionConfig { let to_ms = |secs: Option| secs.map(|s| s as i64 * 1000); - let config = taskito_core::scheduler::retention::RetentionConfig { + taskito_core::scheduler::retention::RetentionConfig { archived_jobs_ttl_ms: to_ms(self.archived_jobs), dead_letter_ttl_ms: to_ms(self.dead_letter), task_logs_ttl_ms: to_ms(self.task_logs), task_metrics_ttl_ms: to_ms(self.task_metrics), job_errors_ttl_ms: to_ms(self.job_errors), - }; - (!config.is_empty()).then_some(config) + } } } @@ -243,14 +244,16 @@ mod tests { task_logs: Some(3), ..RetentionInput::default() }; - let config = input.to_config().expect("some windows set"); + let config = input.to_config(); assert_eq!(config.archived_jobs_ttl_ms, Some(7_000)); assert_eq!(config.task_logs_ttl_ms, Some(3_000)); assert_eq!(config.dead_letter_ttl_ms, None); } #[test] - fn retention_to_config_is_none_when_empty() { - assert!(RetentionInput::default().to_config().is_none()); + fn retention_to_config_is_empty_when_no_windows() { + // A present-but-empty input disables retention (an empty config), rather + // than collapsing to the omitted case that would use the defaults. + assert!(RetentionInput::default().to_config().is_empty()); } } diff --git a/crates/taskito-node/src/worker.rs b/crates/taskito-node/src/worker.rs index ee80e6c5..31685c91 100644 --- a/crates/taskito-node/src/worker.rs +++ b/crates/taskito-node/src/worker.rs @@ -95,8 +95,10 @@ pub fn start_worker( if let Some(batch) = options.batch_size { config.batch_size = batch.max(1) as usize; } + // Present (even empty) → an explicit config: an empty one disables retention. + // Absent → leave `None`, so the core applies the recommended defaults. if let Some(retention) = &options.retention { - config.retention = retention.to_config(); + config.retention = Some(retention.to_config()); } // Bound in-flight work to what this worker will actually run, so it never // claims more and strands the surplus Running while peers sharing the From aaa33eeb3f84b7f39f8fdfb789abe318cc467b8d Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:22:17 +0530 Subject: [PATCH 2/7] fix(java): treat an empty retention config as disabled Presence, not emptiness, decides: an empty retention now reaches the core as Some(empty) to disable auto-cleanup, distinct from omitting it (recommended defaults). Prereq for retention-on-by-default. --- crates/taskito-java/src/convert.rs | 19 +++++++++++-------- crates/taskito-java/src/worker.rs | 4 +++- .../org/byteveda/taskito/worker/Worker.java | 3 +++ 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index 79a6669b..d166f150 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -315,17 +315,18 @@ pub struct RetentionSpec { } impl RetentionSpec { - /// Build a core [`RetentionConfig`], or `None` when no window is set. - pub fn to_config(&self) -> Option { + /// Build a core [`RetentionConfig`]. An all-unset spec yields an empty + /// config, which disables retention — deliberately distinct from *omitting* + /// the `retention` option, which falls back to the recommended defaults. + pub fn to_config(&self) -> taskito_core::scheduler::retention::RetentionConfig { let to_ms = |secs: Option| secs.map(|s| s as i64 * 1000); - let config = taskito_core::scheduler::retention::RetentionConfig { + taskito_core::scheduler::retention::RetentionConfig { archived_jobs_ttl_ms: to_ms(self.archived_jobs), dead_letter_ttl_ms: to_ms(self.dead_letter), task_logs_ttl_ms: to_ms(self.task_logs), task_metrics_ttl_ms: to_ms(self.task_metrics), job_errors_ttl_ms: to_ms(self.job_errors), - }; - (!config.is_empty()).then_some(config) + } } } @@ -643,14 +644,16 @@ mod tests { fn retention_spec_converts_and_deserializes() { let spec: RetentionSpec = serde_json::from_str(r#"{"archivedJobs":7,"taskLogs":3}"#).unwrap(); - let config = spec.to_config().expect("some windows set"); + let config = spec.to_config(); assert_eq!(config.archived_jobs_ttl_ms, Some(7_000)); assert_eq!(config.task_logs_ttl_ms, Some(3_000)); assert_eq!(config.dead_letter_ttl_ms, None); } #[test] - fn retention_spec_is_none_when_empty() { - assert!(RetentionSpec::default().to_config().is_none()); + fn retention_spec_is_empty_when_no_windows() { + // A present-but-empty spec disables retention (an empty config), rather + // than collapsing to the omitted case that would use the defaults. + assert!(RetentionSpec::default().to_config().is_empty()); } } diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index e3530c47..7f0f5f69 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -76,8 +76,10 @@ fn start_worker( if let Some(batch) = options.batch_size { config.batch_size = batch.max(1) as usize; } + // Present (even empty) → an explicit config: an empty one disables retention. + // Absent → leave `None`, so the core applies the recommended defaults. if let Some(retention) = &options.retention { - config.retention = retention.to_config(); + config.retention = Some(retention.to_config()); } // Bound in-flight work to the worker's execution parallelism so a // drain-until-empty poll can't claim more than the pool runs and starve diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index bec32fc1..d5731116 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -457,6 +457,9 @@ private String encodeOptions() { if (!subscriptions.isEmpty()) { options.put("subscriptions", encodeSubscriptions()); } + // Presence, not emptiness: an empty Retention encodes as `{}` and + // disables retention, which the core distinguishes from an omitted + // option (recommended defaults). Do NOT change to a non-empty check. if (retention != null) { options.put("retention", retention.toMap()); } From 72bcb8a453ff64755a9c2bb8adf38e1f47132f2e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:22:32 +0530 Subject: [PATCH 3/7] feat(retention): enable retention by default With no retention configured, effective_retention now returns recommended per-table windows (archived/metrics/errors 7d, logs 3d, DLQ 30d) instead of nothing. A once-per-process warn announces the defaulted policy before the first delete; explicit or legacy config stays silent. --- .../taskito-core/src/scheduler/maintenance.rs | 64 +++++++++++++++++ crates/taskito-core/src/scheduler/mod.rs | 71 +++++++++++++++++-- .../taskito-core/src/scheduler/retention.rs | 50 +++++++++++++ 3 files changed, 179 insertions(+), 6 deletions(-) diff --git a/crates/taskito-core/src/scheduler/maintenance.rs b/crates/taskito-core/src/scheduler/maintenance.rs index 5b9a6ad9..ec2d86e5 100644 --- a/crates/taskito-core/src/scheduler/maintenance.rs +++ b/crates/taskito-core/src/scheduler/maintenance.rs @@ -15,6 +15,59 @@ const PERIODIC_DEFAULT_MAX_RETRIES: i32 = 3; /// Default timeout for periodic tasks (ms). const PERIODIC_DEFAULT_TIMEOUT_MS: i64 = 300_000; +/// A retention window in ms as a compact human duration (`7d`, `12h`, `30m`). +fn humanize_ms(ms: i64) -> String { + const S: i64 = 1000; + const M: i64 = 60 * S; + const H: i64 = 60 * M; + const D: i64 = 24 * H; + match ms { + _ if ms % D == 0 => format!("{}d", ms / D), + _ if ms % H == 0 => format!("{}h", ms / H), + _ if ms % M == 0 => format!("{}m", ms / M), + _ if ms % S == 0 => format!("{}s", ms / S), + _ => format!("{ms}ms"), + } +} + +/// An epoch-ms cutoff as RFC 3339, falling back to raw ms if out of range. +fn iso_cutoff(ms: i64) -> String { + chrono::DateTime::from_timestamp_millis(ms) + .map(|dt| dt.to_rfc3339()) + .unwrap_or_else(|| format!("{ms}ms")) +} + +/// 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( + queue: &str, + windows: &crate::scheduler::retention::RetentionConfig, + now: i64, +) { + let mut msg = format!( + "retention is ON by default for queue '{queue}': history is auto-deleted on the \ + recommended windows below. Set explicit windows, or pass an empty retention config to \ + disable." + ); + for (table, ttl) in [ + ("archived_jobs", windows.archived_jobs_ttl_ms), + ("dead_letter", windows.dead_letter_ttl_ms), + ("task_metrics", windows.task_metrics_ttl_ms), + ("job_errors", windows.job_errors_ttl_ms), + ("task_logs", windows.task_logs_ttl_ms), + ] { + if let Some(ttl) = ttl { + msg.push_str(&format!( + "\n {table}: delete after {} (cutoff {})", + humanize_ms(ttl), + iso_cutoff(now.saturating_sub(ttl)), + )); + } + } + warn!("{msg}"); +} + /// Run one retention sweep, logging and continuing on failure. /// /// The five purges are independent, so one table's error must never abort the @@ -143,6 +196,17 @@ impl Scheduler { let now = now_millis(); let windows = self.config.effective_retention(); + + // The one moment we tell an operator their history will be auto-deleted + // on a policy they did not choose. Before the first delete, once per + // 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 queue = self.namespace.as_deref().unwrap_or("default"); + self.retention_announced + .call_once(|| announce_default_retention(queue, &windows, now)); + } + // 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 7c5f579d..19241957 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/crates/taskito-core/src/scheduler/mod.rs @@ -81,20 +81,34 @@ impl Default for SchedulerConfig { impl SchedulerConfig { /// The retention windows this config actually applies. An explicit - /// `retention` wins; otherwise the deprecated queue-wide `result_ttl_ms` - /// maps onto every table, which is exactly what it used to mean. Negative - /// windows are dropped on both paths — one would invert the cutoff into the - /// future and match every row (the bindings reject them; this is defense in - /// depth for a directly-constructed config). + /// `retention` wins (an empty one disables retention); otherwise the + /// deprecated queue-wide `result_ttl_ms` maps onto every table, which is + /// exactly what it used to mean. With neither set, retention falls back to + /// the recommended defaults — history is bounded out of the box. Negative + /// windows are dropped on the explicit and legacy paths (a negative would + /// invert the cutoff into the future and match every row; the bindings + /// reject them, this is defense in depth for a directly-constructed config). pub fn effective_retention(&self) -> retention::RetentionConfig { if let Some(retention) = &self.retention { return retention.sanitized(); } match self.result_ttl_ms { Some(ttl) if ttl >= 0 => retention::RetentionConfig::uniform(ttl), - _ => retention::RetentionConfig::default(), + // A negative legacy TTL is a set-but-invalid choice: refuse to be + // destructive on it (disable), rather than inverting the cutoff or + // silently upgrading it to the recommended windows. + Some(_) => retention::RetentionConfig::default(), + None => retention::RetentionConfig::recommended(), } } + + /// True when retention runs on the recommended defaults because nothing was + /// configured — no per-table map and no legacy `result_ttl_ms` at all. The + /// one-time cleanup announcement fires only in this case: an explicit, legacy, + /// or set-but-invalid config is the operator's own doing and needs no warning. + pub fn retention_is_defaulted(&self) -> bool { + self.retention.is_none() && self.result_ttl_ms.is_none() + } } /// Result of executing a job (sent back from worker threads). @@ -260,6 +274,11 @@ pub struct Scheduler { /// can refill immediately instead of waiting out its backoff — keeping /// throughput up despite the in-flight cap. dispatch_wake: Arc, + /// Fires the default-retention announcement exactly once per process, before + /// the first delete. Only the elected cleaner reaches it, so in practice it + /// is one line per cluster; a leadership change re-announces, which is + /// informative. Per-instance (`Scheduler` is not `Clone`). + retention_announced: std::sync::Once, /// Wake source for push-dispatch, installed before `run()`. Taken (moved /// out) once when the loop starts. `None` means the loop polls as today. #[cfg(feature = "push-dispatch")] @@ -310,6 +329,7 @@ impl Scheduler { claim_owner: format!("{}-{}", poller::SCHEDULER_CLAIM_OWNER, uuid::Uuid::now_v7()), in_flight: Mutex::new(InFlight::default()), dispatch_wake: Arc::new(Notify::new()), + retention_announced: std::sync::Once::new(), #[cfg(feature = "push-dispatch")] wake_source: Mutex::new(None), #[cfg(feature = "push-dispatch")] @@ -2080,6 +2100,45 @@ mod tests { ); } + #[test] + fn test_default_retention_is_enabled() { + // The default flip: a config that sets neither the per-table map nor a + // legacy result_ttl now bounds history on the recommended windows. This + // is the single behavioral change of the default-on release. + let config = SchedulerConfig::default(); + assert!(config.retention_is_defaulted()); + assert_eq!( + config.effective_retention(), + retention::RetentionConfig::recommended() + ); + assert!(!config.effective_retention().is_empty()); + } + + #[test] + fn test_retention_is_defaulted_only_without_config() { + // Defaulted (announced + recommended) only when nothing is chosen. + assert!(SchedulerConfig::default().retention_is_defaulted()); + assert!(!SchedulerConfig { + result_ttl_ms: Some(3_600_000), + ..SchedulerConfig::default() + } + .retention_is_defaulted()); + assert!(!SchedulerConfig { + retention: Some(retention::RetentionConfig::default()), + ..SchedulerConfig::default() + } + .retention_is_defaulted()); + // A negative legacy TTL is a set-but-invalid choice, not the absence of + // one: not defaulted (no announcement), and it disables retention rather + // than falling through to the recommended windows. + let negative = SchedulerConfig { + result_ttl_ms: Some(-1), + ..SchedulerConfig::default() + }; + assert!(!negative.retention_is_defaulted()); + assert!(negative.effective_retention().is_empty()); + } + #[test] fn test_effective_retention_maps_legacy_result_ttl() { // The legacy knob means "every table, same window" — the mapping must be diff --git a/crates/taskito-core/src/scheduler/retention.rs b/crates/taskito-core/src/scheduler/retention.rs index 149fafb7..9a9985c7 100644 --- a/crates/taskito-core/src/scheduler/retention.rs +++ b/crates/taskito-core/src/scheduler/retention.rs @@ -5,6 +5,19 @@ //! per-entry TTL a single job or DLQ entry can carry, which is honored //! independently of any window here. +const DAY_MS: i64 = 86_400_000; + +/// 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`: +/// logs churn hardest and are worth least (shortest); the DLQ is the only copy +/// of a payload a human must act on, so it is deleted last (longest). +pub const DEFAULT_ARCHIVED_JOBS_TTL_MS: i64 = 7 * DAY_MS; +pub const DEFAULT_DEAD_LETTER_TTL_MS: i64 = 30 * DAY_MS; +pub const DEFAULT_TASK_LOGS_TTL_MS: i64 = 3 * DAY_MS; +pub const DEFAULT_TASK_METRICS_TTL_MS: i64 = 7 * DAY_MS; +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)] @@ -36,6 +49,18 @@ impl RetentionConfig { } } + /// The recommended default windows, applied when a queue sets no retention. + /// See the module-level `DEFAULT_*_TTL_MS` constants and their invariant. + pub fn recommended() -> Self { + Self { + archived_jobs_ttl_ms: Some(DEFAULT_ARCHIVED_JOBS_TTL_MS), + dead_letter_ttl_ms: Some(DEFAULT_DEAD_LETTER_TTL_MS), + task_logs_ttl_ms: Some(DEFAULT_TASK_LOGS_TTL_MS), + task_metrics_ttl_ms: Some(DEFAULT_TASK_METRICS_TTL_MS), + job_errors_ttl_ms: Some(DEFAULT_JOB_ERRORS_TTL_MS), + } + } + /// True when no table has a window — auto-cleanup only runs the per-entry /// TTL sweeps. pub fn is_empty(&self) -> bool { @@ -56,3 +81,28 @@ impl RetentionConfig { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn recommended_respects_the_ordering_invariant() { + let r = RetentionConfig::recommended(); + let logs = r.task_logs_ttl_ms.unwrap(); + let archived = r.archived_jobs_ttl_ms.unwrap(); + let errors = r.job_errors_ttl_ms.unwrap(); + let metrics = r.task_metrics_ttl_ms.unwrap(); + let dlq = r.dead_letter_ttl_ms.unwrap(); + + // task_logs ≤ archived_jobs = job_errors = task_metrics ≤ dead_letter. + assert!(logs <= archived, "logs must not outlive their archived job"); + assert_eq!(archived, errors, "job_errors backs its archived job"); + assert_eq!(archived, metrics, "metrics must outlive their charted job"); + assert!( + archived <= dlq, + "the DLQ is the longest-lived, deliberately" + ); + assert!(!r.is_empty(), "the recommended windows are not empty"); + } +} From 5fc332298bcd21d942f99387a6db32329b21aca4 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:22:48 +0530 Subject: [PATCH 4/7] docs(python): document retention default and opt-out Retention is now on by default; note the recommended windows and that an empty Retention() disables auto-cleanup. --- sdks/python/taskito/retention.py | 6 ++++++ sdks/python/tests/core/test_retention.py | 7 +++++++ 2 files changed, 13 insertions(+) diff --git a/sdks/python/taskito/retention.py b/sdks/python/taskito/retention.py index e7322b4f..404e78b9 100644 --- a/sdks/python/taskito/retention.py +++ b/sdks/python/taskito/retention.py @@ -12,6 +12,12 @@ class Retention: Values are in seconds; ``None`` keeps a table forever. A job or DLQ entry can still carry its own per-entry ``result_ttl``, which is honored independently of these windows. + + Retention is **on by default**: a worker started with no ``retention`` + applies the recommended windows (archived jobs and metrics 7d, job errors + 7d, task logs 3d, dead-letter 30d). Pass a fully empty ``Retention()`` to + disable auto-cleanup and keep every table forever; pass specific windows to + override only those tables (the rest then keep forever). """ archived_jobs: int | None = None diff --git a/sdks/python/tests/core/test_retention.py b/sdks/python/tests/core/test_retention.py index fd8067ee..0fe2ef34 100644 --- a/sdks/python/tests/core/test_retention.py +++ b/sdks/python/tests/core/test_retention.py @@ -58,3 +58,10 @@ def test_retention_map_omits_none_fields() -> None: "archived_jobs": 7, "task_logs": 3, } + + +def test_empty_retention_is_the_opt_out_signal() -> None: + # Retention is on by default, so opting out is passing an empty Retention(): + # it maps to an empty window set, which the native layer reads as "disable" + # rather than the omitted case that applies the recommended defaults. + assert Retention()._as_map() == {} From 5a8cdcbc7aacc4990f8cb87f4de00457bb9cf55e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:23:08 +0530 Subject: [PATCH 5/7] docs: update retention guide for on-by-default Document the recommended default windows, per-table Retention overrides, the empty-Retention opt-out, and the one-time announcement. Clarify that automatic retention does not cascade-delete side tables (each has its own window). --- .../docs/python/api-reference/queue/index.mdx | 4 +- .../guides/operations/job-management.mdx | 88 +++++++++++++++---- .../python/guides/operations/postgres.mdx | 15 +++- 3 files changed, 88 insertions(+), 19 deletions(-) diff --git a/docs/content/docs/python/api-reference/queue/index.mdx b/docs/content/docs/python/api-reference/queue/index.mdx index c16e5d0c..028f402d 100644 --- a/docs/content/docs/python/api-reference/queue/index.mdx +++ b/docs/content/docs/python/api-reference/queue/index.mdx @@ -26,6 +26,7 @@ Queue( default_timeout: int = 300, default_priority: int = 0, result_ttl: int | None = None, + retention: Retention | None = None, middleware: list[TaskMiddleware] | None = None, drain_timeout: int = 30, interception: str = "off", @@ -50,7 +51,8 @@ Queue( | `default_retry` | `int` | `3` | Default max retry attempts for tasks | | `default_timeout` | `int` | `300` | Default task timeout in seconds | | `default_priority` | `int` | `0` | Default task priority (higher = more urgent) | -| `result_ttl` | `int \| None` | `None` | Auto-cleanup completed/dead jobs older than this many seconds. `None` disables. | +| `result_ttl` | `int \| None` | `None` | Legacy uniform auto-cleanup window in seconds, applied to every retention table. Superseded by `retention`. | +| `retention` | `Retention \| None` | `None` | Per-table retention windows. Retention is **on by default** — leaving both `result_ttl` and `retention` unset applies the recommended windows (`archived_jobs`/`task_metrics`/`job_errors` 7d, `task_logs` 3d, `dead_letter` 30d). Pass an empty `Retention()` to disable auto-cleanup and keep every table forever. See [Retention & auto-cleanup](/python/guides/operations/job-management#retention--auto-cleanup). | | `middleware` | `list[TaskMiddleware] \| None` | `None` | Queue-level middleware applied to all tasks. | | `drain_timeout` | `int` | `30` | Seconds to wait for in-flight tasks during graceful shutdown. | | `interception` | `str` | `"off"` | Argument interception mode: `"strict"`, `"lenient"`, or `"off"`. See [Resource System](/python/guides/resources). | diff --git a/docs/content/docs/python/guides/operations/job-management.mdx b/docs/content/docs/python/guides/operations/job-management.mdx index d27ee6e2..f121f78f 100644 --- a/docs/content/docs/python/guides/operations/job-management.mdx +++ b/docs/content/docs/python/guides/operations/job-management.mdx @@ -20,7 +20,7 @@ cancelled = queue.cancel_job(job.id) # True if was pending - Returns `False` if the job was already running, completed, or in another non-pending state - Cancelled jobs cannot be un-cancelled -## Result TTL & auto-cleanup +## Retention & auto-cleanup ### Manual cleanup @@ -32,30 +32,83 @@ deleted = queue.purge_completed(older_than=3600) deleted = queue.purge_dead(older_than=86400) ``` -### Automatic cleanup +### Automatic cleanup (on by default) + +A queue that configures no retention at all automatically applies +recommended per-table windows — history is bounded out of the box, with no +configuration required: + +| Table | Default window | +|---|---| +| `archived_jobs` (completed/failed/cancelled jobs) | 7 days | +| `task_metrics` | 7 days | +| `job_errors` | 7 days | +| `task_logs` | 3 days | +| `dead_letter` (DLQ) | 30 days | + +The DLQ is deliberately the longest-lived — it's the only copy of a payload +a human must act on. The scheduler checks every ~60 seconds and purges rows +older than their table's window. + + + The first time a worker running on this defaulted policy runs a cleanup + sweep, it logs a one-time `warn`-level message naming the queue and every + window it resolved, plus how to opt out. A queue with an explicit + `retention` or legacy `result_ttl` is silent — the announcement exists only + because nothing was configured. + + +### Per-table overrides with `Retention` -Set `result_ttl` on the Queue to automatically purge old jobs while the -worker runs: +Pass a `Retention` to override only the tables you care about — any field +you leave unset (`None`, the default) keeps that table forever: ```python +from taskito import Queue, Retention + queue = Queue( db_path="myapp.db", - result_ttl=3600, # Auto-purge completed/dead jobs older than 1 hour + retention=Retention( + archived_jobs=604_800, # 7 days + dead_letter=2_592_000, # 30 days + task_logs=259_200, # 3 days + ), ) ``` -The scheduler checks every ~60 seconds and purges: +All windows are in seconds. + +### Opting out + +Pass a fully empty `Retention()` to disable auto-cleanup entirely and keep +every table forever. This is deliberately different from *omitting* +`retention`, which applies the recommended defaults above: + +```python +# Never auto-delete anything +queue = Queue(db_path="myapp.db", retention=Retention()) +``` -- Completed jobs older than `result_ttl` -- Dead letter (DLQ — a holding area for jobs that exhausted their retries) entries older than `result_ttl` -- Error history records older than `result_ttl` +### Legacy: `result_ttl` + +`result_ttl` (seconds) is the older, coarser knob — it applies the same +window to every table: + +```python +queue = Queue( + db_path="myapp.db", + result_ttl=3600, # every table purges rows older than 1 hour +) +``` -Set to `None` (default) to disable auto-cleanup. +`result_ttl` still works but is superseded by `Retention` for per-table +control; when both are set, `retention` wins. A negative `result_ttl` is +rejected at construction. ### Cascade cleanup -When jobs are purged — either manually via `purge_completed()` or -automatically via `result_ttl` — their related child records are also deleted: +Manual `purge_completed()` / `purge_dead()` is a blunt delete — it removes a +job together with all of its child records: - Error history (`job_errors`) - Task logs (`task_logs`) @@ -63,8 +116,11 @@ automatically via `result_ttl` — their related child records are also deleted: - Job dependencies (`job_dependencies`) - Replay history (`replay_history`) -This prevents orphaned records from accumulating when parent jobs are -removed. +Automatic retention is per-table by design: purging an archived job does +**not** cascade-delete its logs, metrics, or errors — each of those tables +ages out on its own window instead, so `archived_jobs=7d` with `task_logs` +left unset keeps the logs forever. Only the structural relations +(`job_dependencies`, `replay_history`) always follow the job. ```python # Manual purge — child records are cleaned up automatically @@ -82,8 +138,8 @@ job = resize_image.apply_async( Dead letter entries are **not** cascade-deleted — they have their own - lifecycle managed by `purge_dead()`. Timestamp-based cleanup (`result_ttl`) - of error history, logs, and metrics also continues to run independently, + lifecycle managed by `purge_dead()`. Retention's timestamp-based cleanup of + error history, logs, and metrics also continues to run independently, catching old records regardless of whether the parent job still exists. diff --git a/docs/content/docs/python/guides/operations/postgres.mdx b/docs/content/docs/python/guides/operations/postgres.mdx index ba1099ff..3c8e062e 100644 --- a/docs/content/docs/python/guides/operations/postgres.mdx +++ b/docs/content/docs/python/guides/operations/postgres.mdx @@ -3,6 +3,8 @@ title: Postgres Backend description: "Multi-machine workers and concurrent writes with the PostgreSQL backend." --- +import { Callout } from "fumadocs-ui/components/callout"; + taskito supports PostgreSQL as an alternative storage backend for production deployments that need multi-machine workers or higher write throughput. @@ -43,7 +45,8 @@ queue = Queue( | `workers` | `int` | `0` (auto) | Number of worker threads | All other `Queue` parameters (`default_retry`, `default_timeout`, -`default_priority`, `result_ttl`) work identically to the SQLite backend. +`default_priority`, `result_ttl`, `retention`) work identically to the +SQLite backend. ## Django integration @@ -76,7 +79,15 @@ All Django settings: | `TASKITO_DEFAULT_RETRY` | `3` | Default max retries | | `TASKITO_DEFAULT_TIMEOUT` | `300` | Default task timeout in seconds | | `TASKITO_DEFAULT_PRIORITY` | `0` | Default task priority | -| `TASKITO_RESULT_TTL` | `None` | Result TTL in seconds | +| `TASKITO_RESULT_TTL` | `None` | Legacy queue-wide result TTL in seconds | + + + Retention is on by default: leaving `TASKITO_RESULT_TTL` unset does **not** + disable auto-cleanup — the queue applies the recommended per-table + retention windows automatically. See [Retention & + auto-cleanup](/python/guides/operations/job-management#retention--auto-cleanup) + for the defaults and how to opt out. + ## Schema isolation From 7f311384b5461e8abd1cf8a117076215f8c9d900 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:33:33 +0530 Subject: [PATCH 6/7] fix(retention): label the announcement scope as namespace The purge predicates are not queue-scoped; the value logged is the scheduler's namespace, so name it as such instead of 'queue'. --- crates/taskito-core/src/scheduler/maintenance.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/crates/taskito-core/src/scheduler/maintenance.rs b/crates/taskito-core/src/scheduler/maintenance.rs index ec2d86e5..e0b54bb1 100644 --- a/crates/taskito-core/src/scheduler/maintenance.rs +++ b/crates/taskito-core/src/scheduler/maintenance.rs @@ -41,12 +41,14 @@ fn iso_cutoff(ms: i64) -> String { /// 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( - queue: &str, + namespace: &str, windows: &crate::scheduler::retention::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!( - "retention is ON by default for queue '{queue}': history is auto-deleted on the \ + "retention is ON by default for namespace '{namespace}': history is auto-deleted on the \ recommended windows below. Set explicit windows, or pass an empty retention config to \ disable." ); @@ -202,9 +204,9 @@ 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 queue = self.namespace.as_deref().unwrap_or("default"); + let namespace = self.namespace.as_deref().unwrap_or("default"); self.retention_announced - .call_once(|| announce_default_retention(queue, &windows, now)); + .call_once(|| announce_default_retention(namespace, &windows, now)); } // A window `ttl` becomes the cutoff `now - ttl`; an absent window is a From 5e4b5ee5978f94b96acce1b47590755bcc61ac0f Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:33:43 +0530 Subject: [PATCH 7/7] docs: correct retention cadence and opt-out wording Cleanup runs on the cleanup tick, not a fixed ~60s. An empty Retention() disables the table-wide windows but a per-entry result_ttl is still honored. --- .../docs/python/api-reference/queue/index.mdx | 2 +- .../python/guides/operations/job-management.mdx | 14 +++++++++----- sdks/python/taskito/retention.py | 5 +++-- 3 files changed, 13 insertions(+), 8 deletions(-) diff --git a/docs/content/docs/python/api-reference/queue/index.mdx b/docs/content/docs/python/api-reference/queue/index.mdx index 028f402d..205bd029 100644 --- a/docs/content/docs/python/api-reference/queue/index.mdx +++ b/docs/content/docs/python/api-reference/queue/index.mdx @@ -52,7 +52,7 @@ Queue( | `default_timeout` | `int` | `300` | Default task timeout in seconds | | `default_priority` | `int` | `0` | Default task priority (higher = more urgent) | | `result_ttl` | `int \| None` | `None` | Legacy uniform auto-cleanup window in seconds, applied to every retention table. Superseded by `retention`. | -| `retention` | `Retention \| None` | `None` | Per-table retention windows. Retention is **on by default** — leaving both `result_ttl` and `retention` unset applies the recommended windows (`archived_jobs`/`task_metrics`/`job_errors` 7d, `task_logs` 3d, `dead_letter` 30d). Pass an empty `Retention()` to disable auto-cleanup and keep every table forever. See [Retention & auto-cleanup](/python/guides/operations/job-management#retention--auto-cleanup). | +| `retention` | `Retention \| None` | `None` | Per-table retention windows. Retention is **on by default** — leaving both `result_ttl` and `retention` unset applies the recommended windows (`archived_jobs`/`task_metrics`/`job_errors` 7d, `task_logs` 3d, `dead_letter` 30d). Pass an empty `Retention()` to disable the table-wide windows (per-entry `result_ttl` is still honored). See [Retention & auto-cleanup](/python/guides/operations/job-management#retention--auto-cleanup). | | `middleware` | `list[TaskMiddleware] \| None` | `None` | Queue-level middleware applied to all tasks. | | `drain_timeout` | `int` | `30` | Seconds to wait for in-flight tasks during graceful shutdown. | | `interception` | `str` | `"off"` | Argument interception mode: `"strict"`, `"lenient"`, or `"off"`. See [Resource System](/python/guides/resources). | diff --git a/docs/content/docs/python/guides/operations/job-management.mdx b/docs/content/docs/python/guides/operations/job-management.mdx index f121f78f..119a2b39 100644 --- a/docs/content/docs/python/guides/operations/job-management.mdx +++ b/docs/content/docs/python/guides/operations/job-management.mdx @@ -47,8 +47,9 @@ configuration required: | `dead_letter` (DLQ) | 30 days | The DLQ is deliberately the longest-lived — it's the only copy of a payload -a human must act on. The scheduler checks every ~60 seconds and purges rows -older than their table's window. +a human must act on. The scheduler runs cleanup periodically (on its cleanup +tick, not a fixed wall-clock interval) and purges rows older than their +table's window. The first time a worker running on this defaulted policy runs a cleanup @@ -80,15 +81,18 @@ All windows are in seconds. ### Opting out -Pass a fully empty `Retention()` to disable auto-cleanup entirely and keep -every table forever. This is deliberately different from *omitting* +Pass a fully empty `Retention()` to disable the table-wide windows, so no table +is auto-cleaned on a schedule. This is deliberately different from *omitting* `retention`, which applies the recommended defaults above: ```python -# Never auto-delete anything +# No table-wide retention windows queue = Queue(db_path="myapp.db", retention=Retention()) ``` +A per-entry `result_ttl` on an individual job or DLQ entry is still honored even +with the windows disabled — the per-entry sweep runs every cleanup tick. + ### Legacy: `result_ttl` `result_ttl` (seconds) is the older, coarser knob — it applies the same diff --git a/sdks/python/taskito/retention.py b/sdks/python/taskito/retention.py index 404e78b9..ccfe8e7f 100644 --- a/sdks/python/taskito/retention.py +++ b/sdks/python/taskito/retention.py @@ -16,8 +16,9 @@ class Retention: Retention is **on by default**: a worker started with no ``retention`` applies the recommended windows (archived jobs and metrics 7d, job errors 7d, task logs 3d, dead-letter 30d). Pass a fully empty ``Retention()`` to - disable auto-cleanup and keep every table forever; pass specific windows to - override only those tables (the rest then keep forever). + disable the table-wide windows (a per-entry ``result_ttl`` is still honored); + pass specific windows to override only those tables (the rest then keep + forever). """ archived_jobs: int | None = None