QueueCTL: Python-based Job Queue Management System

QueueCTL is a lightweight, reliable CLI-driven distributed job queue system built in Python.
It supports asynchronous job execution, persistent storage, multiple worker management, retry & backoff, dead letter queues (DLQ), configurable settings, and an optional web dashboard for live monitoring.

Backend Record Link: https://1drv.ms/v/c/bc204b0860a4efb4/ERntO2GiDYxLkDa-5pjRKw0BbHFxEEaI5y18dVSv_YxyuA?e=ceUf0u

---

Features Overview

Category	Description
Core Functionality	Job enqueueing, execution, state tracking
Persistence	SQLite-backed durable storage (survives restarts)
Workers	Multiple worker processes with job locking
Retry & Backoff	Automatic retry with exponential delay
Dead Letter Queue	Failed jobs moved to DLQ after max_retries
Configuration Management	CLI-based config editing (max_retries, backoff)
Metrics	Average duration, job counts per state
Web Dashboard	Minimal monitoring UI (Flask-based)
Output Logging	Captures job stdout/stderr for review
Timeout Handling	Jobs automatically killed after timeout_sec
Priority Queues	Jobs processed based on priority value
Scheduled Jobs	Jobs delayed until specific run_at timestamps

---

Architecture Overview

System Components

QUEUECTL/
│
├── .venv/                        # Python virtual environment (dependencies isolated)
├── logs/                         # Application and worker log files
│
├── queuectl/
│   ├── __pycache__/              # Compiled Python bytecode
│   ├── templates/                # HTML templates for Flask dashboard
│   ├── __init__.py               # Package initializer
│   ├── cli.py                    # Command-line interface for job control
│   ├── config.py                 # Configuration and environment management
│   ├── dashboard.py              # Flask-based web dashboard for job monitoring
│   ├── db.py                     # SQLite persistence layer for job storage
│   ├── models.py                 # ORM-like model definitions for queue entities
│   ├── utils.py                  # Utility helpers, logging, and time functions
│   ├── worker.py                 # Core worker logic for job execution
│   └── queue.db                  # Auto-generated SQLite database file
│
├── queuectl.egg-info/            # Package metadata (auto-generated by setuptools)
│
├── tests/                        # Unit and smoke tests
│   └── smoke.sh                  # Basic smoke test script
│
├── venv/                         # Optional additional virtual environment
│
├── LICENSE                       # License information
├── pyproject.toml                # Project configuration (build system, dependencies)
├── README.md                     # Project documentation (this file)
├── smoke.db                      # Database created during smoke testing
└── queue.db                      # Root-level database instance (if used globally)

---

Job Lifecycle

State	Description
pending	Waiting to be picked up by a worker
processing	Currently being executed
completed	Successfully executed
failed	Failed, but retryable
dead	Permanently failed (moved to DLQ)

---

Setup Instructions

Clone Repository

git clone https://github.com/chandanboyina/FLam-Backend.git

cd queuectl

Create Virtual Environment

python -m venv venv
venv\Scripts\activate   # (Windows)
# or
source venv/bin/activate  # (Linux/Mac)

Install Dependencies

pip install -e .

Initialize Database

python -m queuectl.db

---

Usage Examples — CLI Commands with Output

This section demonstrates how to use QueueCTL from the command line for everyday operations.
Each command example is followed by an example output (as seen in the console).

---

1. Enqueue a New Job

Adds a new job to the queue.

queuectl enqueue "{\"id\":\"job1\",\"command\":\"cmd /c echo Hello QueueCTL!\"}"

queuectl enqueue "{\"id\":\"job_fail\",\"command\":\"cmd /c exit 1\"}"

Output

Enqueued job: job1
Enqueued job: job_fail

2. Stsrt Worker

Start one or more worker processes to execute jobs.

queuectl worker start --count 2

Output

Launching 2 worker(s)...
Started 2 worker(s). PIDs: 8544, 9752
[Worker w-1a2b3c] Picked job job1
[Worker w-1a2b3c] Completed job job1 in 0.02s
[16:03:01] [Worker w-e20109c9] Started.
[16:03:01] [Worker w-8b63227e] Started.
[16:03:01] [Worker w-8b63227e] Picked job job_fail
(Priority=100, Attempt=1/3)
[16:03:01] [Worker w-8b63227e] Job job_fail failed
(exit_code=1), scheduling retry or DLQ
[DEBUG] fail_or_retry for job=job_fail (attempt 1/3) reason=exit_code=1
[Retry DEBUG] Retrying job job_fail in 2s (attempt 1)
[16:03:03] [Worker w-e20109c9] Picked job job_fail
(Priority=100, Attempt=2/3)
[16:03:03] [Worker w-e20109c9] Job job_fail failed
(exit_code=1), scheduling retry or DLQ
[DEBUG] fail_or_retry for job=job_fail (attempt 2/3) reason=exit_code=1
[Retry DEBUG] Retrying job job_fail in 4s (attempt 2)
[16:03:07] [Worker w-8b63227e] Picked job job_fail
(Priority=100, Attempt=3/3)
[16:03:07] [Worker w-8b63227e] Job job_fail failed
(exit_code=1), scheduling retry or DLQ
[DEBUG] fail_or_retry for job=job_fail (attempt 3/3) reason=exit_code=1
============================================================
[DLQ DEBUG] Moving job job_fail to DLQ now!
============================================================
[DLQ DEBUG] Job job_fail inserted into DLQ

3. Check Queue Status

Displays the current state of all jobs

queuectl status

Output

   QueueCTL System Status
┏━━━━━━━━━━━━━━━━━━┳━━━━━━━┓
┃ State            ┃ Count ┃
┡━━━━━━━━━━━━━━━━━━╇━━━━━━━┩
│ pending          │     0 │
│ processing       │     0 │
│ completed        │     1 │
│ failed           │     0 │
│ dead             │     1 │
│ dlq              │     1 │
│ avg_duration_sec │  0.07 │
└──────────────────┴───────┘

4. List Jobs by State

To view all jobs currently in the queue by their lifecycle state

queuectl list --state completed

Output

┏━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━┓
┃ ID      ┃ Command                   ┃ State      ┃ Duration │
┡━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━┩
│ job1    │ cmd /c echo Hello QueueCTL│ completed  │ 0.02s    │
└─────────┴───────────────────────────┴────────────┴──────────┘

5. Retry Job from DLQ

queuectl dlq retry job_fail

Output

Retrying job job_fail from DLQ
Job job_fail re-enqueued successfully

6. View and Edit Configuration

Update the configuration if needed:

queuectl config set max-retries 5
queuectl config set base_backoff 3

Output

Configuration updated: max_retries = 5
Configuration updated: base_backoff = 3

Show current configurations such as max retries or backoff base.

queuectl config show

Output

Current Configuration
---------------------
max_retries: 5
base_backoff: 5
timeout_sec: 60

7. Enqueue a Scheduled (Delayed) Job

Schedule a job to run at a specific time in the future using the run_at parameter.

queuectl enqueue "{\"id\":\"job_future\",\"command\":\"cmd /c echo Future Job\",\"run_at\":\"2025-11-08T12:00:00Z\"}"

Output

Enqueued job: job_future (Scheduled for 2025-11-08T12:00:00Z)

The job remains in pending until its scheduled time.

8. View Dead Letter Queue (DLQ)

List all jobs that have permanently failed after maximum retry attempts.

queuectl dlq list

Output

┏━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━┓
┃ ID     ┃ Reason               ┃ Created              ┃
┡━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━┩
│ job22  │ exit_code=1          │ 2025-11-07 19:49:46  │
└────────┴───────────────────────┴──────────────────────┘

9. Launch Web Dashboard

To view real-time statistics in your browser:

python -m queuectl.dashboard

Open: http://localhost:8080 to monitor:

• Job counts per state
• Success/failure metrics
• Average job duration
• DLQ table with retry options

Architecture Explanation

QueueCTL is designed with a modular and extensible architecture following the Single Responsibility Principle (SRP) — each module handles one clear purpose.

Core Components

Module	Role	Detailed Description
cli.py	Command Line Interface	Handles all CLI commands (enqueue, list, status, worker, dlq, config). It uses Typer/Click to parse commands and send actions to the core modules.
db.py	Database Persistence Layer	Manages the SQLite database. Creates tables for jobs, dlq, and config. Ensures data durability, so jobs persist after restarts. Uses SQL transactions for safe updates.
worker.py	Worker Logic	Core engine of the system. Each worker picks jobs in pending state, locks them, executes their command, tracks exit codes, updates job states (processing → completed/failed), applies timeouts and retries, and pushes to DLQ when necessary.
scheduler.py	Job Scheduling	Manages delayed jobs using run_at timestamps. Ensures jobs are not executed before their scheduled time.
config.py	Configuration Manager	Stores and retrieves global configurations like max_retries and backoff_base. Config values are stored persistently in the database and can be changed dynamically from the CLI.
dashboard.py	Monitoring Dashboard	A Flask web server providing a live visualization of queue metrics, job counts, DLQ contents, and success/failure stats. It serves dashboard.html with Chart.js for live charts.
utils.py	Helper Functions	Contains reusable utilities for logging, time formatting, exponential backoff calculation, and process management. Keeps code DRY and clean.
queue.db	SQLite Database	Central persistent storage of all jobs, DLQ entries, configuration, and logs.

---

Data Flow Overview

1. CLI Command → (e.g., enqueue, worker start)
2. Job Stored → db.py inserts into jobs table.
3. Worker Fetches → worker.py locks and picks one pending job.
4. Job Execution → Worker executes command using subprocess.
5. State Transition → Updates job’s state in database.
6. Retry/Backoff → If failure, retries with exponential delay.
7. DLQ Transfer → After max retries, moves to DLQ.
8. Dashboard Visualization → Flask dashboard reads DB and shows live metrics.

---

Design Principles Used

• SRP (Single Responsibility Principle) → Each module has one clear function.
• Persistence First → Job data stored before execution ensures fault tolerance.
• Locking Mechanism → Prevents multiple workers from processing the same job.
• Graceful Shutdown → Workers finish running jobs before termination.
• Scalability → Multiple workers can run in parallel.
• Observability → Real-time dashboard shows internal metrics.

Result:
This architecture makes QueueCTL robust, maintainable, and easy to extend with new features like priority queues or remote execution.

Key Algorithms

QueueCTL implements several key control algorithms that ensure job execution is fault-tolerant, efficient, and predictable.

---

Exponential Backoff (Retry Delay)

Purpose:
Prevents retry storms when a job fails repeatedly by adding increasing delays.

Formula:

delay = base_backoff ^ attempts

Example: If base_backoff = 2 and max_retries = 3:

Attempt	Delay (seconds)
1	2
2	4
3	8

Implementation:

• Stored in worker.py
• When a job fails, it calculates next run_at using this delay before retrying.
• On final failure, moves job to DLQ.

---

⏱️ Job Timeout Handling

Purpose:
Ensures no job runs indefinitely (e.g., a stuck command).

Mechanism:

• Each job has a timeout_sec field (default = 60s).
• worker.py uses subprocess.Popen with a timer.
• If process runs longer than timeout_sec, it is force-terminated and marked failed.

Result:
Your queue remains responsive even if one job misbehaves.

---

Retry Logic

Purpose:
Ensures transient failures are automatically retried.

Mechanism:

• On failure, increment attempts count.
• If attempts < max_retries → retry job after exponential delay.
• If attempts == max_retries → move job to DLQ.

DLQ entries can be retried manually via:

queuectl dlq retry job2

Result: Jobs recover automatically from network issues or temporary failures.

Worker Locking

Purpose: Prevents two workers from picking the same job concurrently.

Mechanism:

• SQLite BEGIN IMMEDIATE TRANSACTION ensures row-level lock.
• Worker atomically updates job state to processing.
• Other workers see the job as unavailable.

Result: No duplicate job executions even when multiple workers run in parallel.

---

Metrics Aggregation

Purpose:
Continuously update and display real-time queue statistics for monitoring and analysis.

Metrics Tracked

• Job counts per state — Pending, Running, Completed, Failed
• DLQ (Dead Letter Queue) count — Tracks permanently failed jobs
• Average duration: AVG(duration_sec) for completed jobs

Used In

• CLI:
  Displayed via the status command for quick summary.
• Dashboard:
  Exposed through the /api/stats endpoint for live charts and analytics.

---

Delayed Job Scheduling

Purpose:
Enable scheduling of jobs to execute at a future timestamp using run_at.

Mechanism

• The schedular.py filters jobs where:

  run_at <= NOW()

• Jobs scheduled for future remain pending.

Result These algorithms together make QueueCTL a resilient, fault-tolerant job system capable of recovering from errors, scaling workers, and executing jobs predictably.

---

Assumptions & Trade-offs (Detailed Explanation)

This section demonstrates that you’ve made intentional engineering decisions.
It proves you understand why you designed it the way you did.

Assumptions & Trade-offs

Decision / Assumption	Why it was made (Trade-off)
SQLite for persistence	Lightweight, file-based, cross-platform. No server dependency. Ideal for single-node local queue. (Trade-off: not distributed, slower under extreme parallelism.)
Python multiprocessing for workers	Enables real parallel execution. Simpler than threads due to GIL. (Trade-off: slightly more memory use.)
CLI-based design	Easier to test, script, and automate. Works on any OS. (Trade-off: not a GUI-heavy application.)
Flask dashboard	Lightweight framework, good for visualization. (Trade-off: limited scalability for large real-time systems.)
JSON job definition	Simple to enqueue via CLI. Human-readable and flexible. (Trade-off: basic schema validation.)
Exponential backoff retry	Prevents overloading the system. (Trade-off: longer total processing time for persistent failures.)
Single DB file (queue.db)	Simplicity & portability. (Trade-off: not ideal for distributed worker nodes.)
Timeout handling via subprocess	Compatible with any OS command. (Trade-off: subprocess overhead vs direct function call.)

Summary:
Every design choice balances simplicity, reliability, and maintainability, keeping QueueCTL portable and easy to understand for interview evaluations.

Testing Instructions - How to Verify Functionality

These tests ensure that every part of the QueueCTL system works correctly, including all job lifecycles, retry logic, DLQ handling, and monitoring.

---

1. Successful Job Test

queuectl enqueue "{\"id\":\"job_success\",\"command\":\"cmd /c echo Hello Success!\"}"
queuectl worker start --count 1

Expected: Job runs successfully → processing → completed.

2. Failing Job with Retries and DLQ

queuectl enqueue "{\"id\":\"job_fail\",\"command\":\"cmd /c exit 1\"}"

Expected:

• Job fails initially
• Retries 3 times with exponential backoff (2s, 4s, 8s)
• Moves to DLQ after exceeding retries

Check DLQ:

queuectl dlq list

3. Scheduled (Delayed) Job

queuectl enqueue "{\"id\":\"job_future\",\"command\":\"cmd /c echo Scheduled Run\",\"run_at\":\"2025-11-08T12:00:00Z\"}"

Expected: Job stays in pending until the timestamp is reached.

4. Timeout Job

queuectl enqueue "{\"id\":\"timeout_test\",\"command\":\"cmd /c timeout 70\",\"timeout_sec\":5}"

Expected: Worker kills process after 5 seconds → job marked failed → moves to DLQ.

5. Multiple Worker Test

queuectl enqueue "{\"id\":\"jobA\",\"command\":\"cmd /c echo JobA\"}"
queuectl enqueue "{\"id\":\"jobB\",\"command\":\"cmd /c echo JobB\"}"
queuectl worker start --count 3

Expected: Multiple jobs processed in parallel — no duplication due to locking.

6. List Jobs by State

queuectl list --state pending

DLQ Management

queuectl dlq list
queuectl dlq retry job2

6. Persistence Test

• Stop all workers
• Restart app
• Run queuectl status

Example Output:

┏━━━━━━━━━━━━━━━━━━┳━━━━━━━┓
┃ State            ┃ Count ┃
┡━━━━━━━━━━━━━━━━━━╇━━━━━━━┩
│ pending          │     1 │
│ processing       │     0 │
│ completed        │     3 │
│ failed           │     0 │
│ dead             │     1 │
│ dlq              │     1 │
│ avg_duration_sec │  0.04 │
└──────────────────┴───────┘

Expected: Job data and DLQ entries still visible. Confirms persistence through restarts.

7. Config Change Verification

queuectl config set max-retries 5
queuectl config set base_backoff 3
queuectl config show

Expected: Updated configuration values reflected immediately in future job retries.

8. Dashboard Verification

python -m queuectl.dashboard

Open http://localhost:8080

Expected:

• KPI counters update live
• Bar charts show current job distribution
• DLQ table lists failed jobs

---

---

References & Citations

This project was built with reference to publicly available and officially licensed resources for educational and implementation purposes.
All referenced materials are open-source or freely available for learning and development.

Documentation & Framework References

• Python Official Documentation
• Flask Web Framework Documentation
• Typer CLI Framework Documentation
• SQLite3 Official Docs
• Chart.js Documentation
• OpenAI ChatGPT — used for project guidance, learning explanations, and documentation structure.

All external references were used solely for educational and implementation understanding purposes.
The complete source code, logic, and integration were written, tested, and verified by Chandan Boyina.

---

Compliance & Integrity Declaration

• I hereby declare that this project is my original work, developed independently as part of a company technical evaluation.
• I have not received any unauthorized assistance or copied content from other individuals or projects.
• All referenced materials (Python documentation, Flask guides, SQLite tutorials) were used solely for conceptual understanding.
• I fully understand every component of this project and can explain its logic and architecture in detail.

---

Developer’s Note

• This project was developed independently for a placement technical assignment.
• Throughout the development process, I referred to official documentation for Python, Flask, SQLite, and other standard libraries.
• Automated tools were used only for research and learning reference, and all final implementation logic, architecture, and debugging were done independently by me.
• I take complete responsibility for the design decisions and correctness of the submitted code.

---