Job management

Manage running jobs — cancel, pause queues, archive, revoke, replay, and clean up.

Job cancellation

Cancel a pending job before it starts:

job = send_email.delay("user@example.com", "Hello", "World")
cancelled = queue.cancel_job(job.id)  # True if was pending

• Returns True if the job was pending and is now cancelled
• Returns False if the job was already running, completed, or in another non-pending state
• Cancelled jobs cannot be un-cancelled

Retention & auto-cleanup

Manual cleanup

# Purge completed jobs older than 1 hour
deleted = queue.purge_completed(older_than=3600)

# Purge dead letters older than 24 hours
deleted = queue.purge_dead(older_than=86400)

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:

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.

Per-table overrides with Retention

Pass a Retention to override only the tables you care about — any field you leave unset (None, the default) keeps that table forever:

from flexiq import Queue, Retention

queue = Queue(
    db_path="myapp.db",
    retention=Retention(
        archived_jobs=604_800,  # 7 days
        dead_letter=2_592_000,  # 30 days
        task_logs=259_200,      # 3 days
    ),
)

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:

# 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 window to every table:

queue = Queue(
    db_path="myapp.db",
    result_ttl=3600,  # every table purges rows older than 1 hour
)

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.

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:

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).

Previewing a purge

Tuning a window without knowing how much it deletes is guesswork. dry_run_retention() counts, per table, the rows a purge would remove right now — without deleting anything:

preview = queue.dry_run_retention()
print(f"{preview.total} rows would be deleted")
print(preview.counts["archived_jobs"], preview.counts["task_logs"])

With no argument it previews this queue's configured (or default-recommended) windows. Pass candidate windows to size one before committing to it — no worker restart needed:

from flexiq import Retention

preview = queue.dry_run_retention(Retention(archived_jobs=3 * 86_400))
print(f"a 3-day archive window would delete {preview.counts['archived_jobs']} jobs")

The counts are a point-in-time snapshot; nothing is deleted, so it is safe to run against production. Unlike effective_retention(), it is computed in-process against the live data and always returns a result. The async twin is adry_run_retention().

Cascade cleanup

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)
• Task metrics (task_metrics)
• Job dependencies (job_dependencies)
• Replay history (replay_history)

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.

# Manual purge — child records are cleaned up automatically
deleted = queue.purge_completed(older_than=3600)
print(f"Purged {deleted} jobs and their related records")

# With per-job TTL — cascade cleanup still applies
job = resize_image.apply_async(
    args=("photo.jpg",),
    result_ttl=600,  # This job's results expire after 10 minutes
)
# When this job is purged (after 10 min), its errors, logs,
# metrics, dependencies, and replay history are also removed.

Queue pause/resume

Temporarily pause job processing on a queue without stopping the worker:

# Pause the "emails" queue
queue.pause("emails")

# Check which queues are paused
print(queue.paused_queues())  # ["emails"]

# Resume processing
queue.resume("emails")

Paused queues still accept new jobs — they just won't be dequeued until resumed.

Maintenance window example

# Before maintenance: pause all queues
for q in ["default", "emails", "reports"]:
    queue.pause(q)
print(f"Paused: {queue.paused_queues()}")

# ... perform maintenance ...

# After maintenance: resume all queues
for q in ["default", "emails", "reports"]:
    queue.resume(q)

Job archival

Move old completed jobs to an archive table to keep the main jobs table lean:

# Archive completed jobs older than 24 hours
archived_count = queue.archive(older_than=86400)
print(f"Archived {archived_count} jobs")

# Browse archived jobs
archived = queue.list_archived(limit=50, offset=0)
for job in archived:
    print(f"{job.id}: {job.task_name} ({job.status})")

Archived jobs are no longer returned by queue.stats() or queue.list_jobs(), but remain queryable via queue.list_archived().

Scheduled archival

from flexiq import current_job

@queue.periodic(cron="0 0 2 * * *")  # Daily at 2 AM
def nightly_archival():
    archived = queue.archive(older_than=7 * 86400)  # Archive jobs older than 7 days
    current_job.log(f"Archived {archived} jobs")

Task revocation

Cancel all pending jobs for a specific task:

# Revoke all pending "send_newsletter" jobs
cancelled = queue.revoke_task("myapp.tasks.send_newsletter")
print(f"Revoked {cancelled} jobs")

Queue purge

Remove all pending jobs from a specific queue:

purged = queue.purge("emails")
print(f"Purged {purged} jobs from the emails queue")

Job replay

Replay a completed or dead job with the same arguments:

new_job = queue.replay(job_id)
print(f"Replayed as {new_job.id}")

# Check replay history
history = queue.replay_history(job_id)

Retry from dead letter with replay

# List dead letters and replay them
dead = queue.dead_letters()
for entry in dead:
    print(f"Replaying dead job {entry['original_job_id']}: {entry['task_name']}")
    new_id = queue.retry_dead(entry["id"])
    print(f"  -> New job: {new_id}")

SQLite configuration

Left at their defaults, SQLite's own pragmas favor safety over throughput and let concurrent readers block on writers — so flexiq sets these at connection time to give WAL-mode concurrency and predictable behavior under lock contention without any configuration on your part:

The connection pool uses up to 8 connections via r2d2.