diff --git a/crates/taskito-core/src/scheduler/maintenance.rs b/crates/taskito-core/src/scheduler/maintenance.rs index 5b9a6ad9..e0b54bb1 100644 --- a/crates/taskito-core/src/scheduler/maintenance.rs +++ b/crates/taskito-core/src/scheduler/maintenance.rs @@ -15,6 +15,61 @@ 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( + 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 namespace '{namespace}': 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 +198,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 namespace = self.namespace.as_deref().unwrap_or("default"); + self.retention_announced + .call_once(|| announce_default_retention(namespace, &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"); + } +} 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/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 diff --git a/docs/content/docs/python/api-reference/queue/index.mdx b/docs/content/docs/python/api-reference/queue/index.mdx index c16e5d0c..205bd029 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 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 d27ee6e2..119a2b39 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,87 @@ 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 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 + 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 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 +# 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` -- 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` +`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 +120,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 +142,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 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()); } diff --git a/sdks/python/taskito/retention.py b/sdks/python/taskito/retention.py index e7322b4f..ccfe8e7f 100644 --- a/sdks/python/taskito/retention.py +++ b/sdks/python/taskito/retention.py @@ -12,6 +12,13 @@ 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 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 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() == {}