QueueForge Ops

A full-stack asynchronous queue observability and dead-letter queue (DLQ) triage platform that intercepts, visualizes, and safely replays failed background tasks.

QueueForge Ops eliminates dark-data dead-letter queues by combining automated failure classification, safe payload mutation, differential inspection, and rate-limited async execution into a single, cohesive operator interface.

---

📋 Problem Statement & Target Audience

The Problem

When distributed background task processing (Celery, RQ, RabbitMQ, SQS) encounters unhandled exceptions, poison pills, or transient network failures, non-retryable tasks drop into Dead-Letter Queues (DLQs). Traditional handling presents critical operational challenges:

• Opacity: Stack traces and task contexts remain locked in logs or raw JSON dumps.
• Risky Manual Patches: Debugging usually requires direct database edits or fragile ad-hoc Python scripts executed against production datastores.
• Amplification Risk: Naive bulk replaying of failed jobs risks triggering cascading failures or overwhelming downstream services without backoff protections or rate limits.
• Lack of Auditability: Mutation of task arguments and re-execution outcomes are rarely tracked, obscuring accountability.

Target Audience

• Site Reliability Engineers (SREs) requiring queue health telemetry, failure classification, and safe re-injection mechanisms.
• Backend Platform Engineers seeking a standardized, auditable framework for DLQ triage.
• Operations & Support Teams needing an interactive workbench to patch schema mismatches and resolve malformed task payloads without write access to production databases.

---

✨ Key Features

• Dead-Letter Ingestion & Root-Cause Classification: Ingests failed job payloads and stack traces via REST/Webhook endpoints. Automatically classifies errors (e.g., Schema Mismatch, Transient Timeout, Auth Error, Database Lock) using zero-dependency regex pattern parsing.
• Interactive Payload Workbench & Diff Editor: Integrated web interface built with HTML5, Jinja2, and JS. Offers real-time payload editing, validation against custom JSON schemas, and side-by-side visual diffing prior to replay.
• Deterministic Safe Replay Engine: Asynchronous HTTP re-injection powered by httpx. Supports configurable concurrency controls, exponential backoff with jitter, and dynamic circuit-breaker thresholds to block cascading destination outages.
• Audit Logging & Telemetry Dashboard: Complete visual audit history recording every operator edit, payload patch, and replay execution result. Real-time metrics track queue depth, recovery success rates, and failure distributions.

---

🏗 Architecture & System Design

                     +-------------------------------------------------+
                     |               DLQ / Producer App                |
                     +-------------------------------------------------+
                                              |
                                              | POST /api/v1/tasks/ingest
                                              v
+-----------------------------------------------------------------------------------------------+
| QueueForge Ops                                                                                |
|                                                                                               |
|   +-----------------------+     +------------------------+     +--------------------------+   |
|   |   FastAPI REST API    |     |  Automated Classifier  |     |  SQLAlchemy Data Store   |   |
|   |     & Jinja2 UI       | --> |   (Stack-Trace Regex)  | --> |     (Tasks, Audits,      |   |
|   +-----------------------+     +------------------------+     |    Circuit Breakers)     |   |
|               ^                                                +--------------------------+   |
|               |                                                             |                 |
|               v                                                             v                 |
|   +-----------------------+                                    +--------------------------+   |
|   |  Interactive Payload  |                                    |   Async Replay Engine    |   |
|   |       Workbench       |                                    | (HTTPX / Circuit Breaker)|   |
|   +-----------------------+                                    +--------------------------+   |
+-----------------------------------------------------------------------------|-----------------+
                                                                              |
                                                                              | Rate-limited HTTP
                                                                              v
                                                         +------------------------------------------+
                                                         |        Target Consumer / Worker          |
                                                         +------------------------------------------+

Module Responsibilities

Module	Location	Purpose
Data Models	src/models.py	SQLAlchemy ORM entities (DLQTask, AuditLog, CircuitBreakerState) and Pydantic validation schemas for incoming ingestion requests and API responses.
Core Utilities	src/core.py	Engine suite containing stack-trace regex patterns, deep JSON payload diff generator, circuit breaker state controller, and httpx-based async replay runner with exponential backoff.
API & Workbench Router	src/api.py	FastAPI endpoint handlers for REST API management, payload patch validation, telemetry analytics, and Jinja2 UI view rendering.
Application Lifecycle	src/main.py	Application entry point, FastAPI initialization, database table migration/bootstrap, static asset mounting, and Uvicorn server launcher.

---

🛠 Technology Stack

• Language: Python 3.10+
• Framework: FastAPI (REST endpoints & Jinja2 template rendering)
• ORM & Database: SQLAlchemy 2.0 / SQLite (configurable to PostgreSQL via SQLAlchemy connection strings)
• Validation Engine: Pydantic v2
• Client & HTTP Engine: HTTPX (Async HTTP requests for task replaying)
• Frontend UI: HTML5, Modern Vanilla JavaScript, CSS3, Jinja2 Templates
• Testing Suite: pytest, pytest-asyncio, HTTPX AsyncClient

---

⚙️ Installation & Virtual Environment Setup

Prerequisites

• Python 3.10 or higher
• git version control tool

Step-by-Step Setup

1. Clone the Repository:

   git clone https://github.com/your-org/queueforge-ops.git
   cd queueforge-ops

2. Create and Activate a Virtual Environment:

   # Linux / macOS
   python3 -m venv venv
   source venv/bin/activate
   
   # Windows
   python -m venv venv
   .\venv\Scripts\activate

3. Install Dependencies:

   pip install --upgrade pip
   pip install -r requirements.txt

---

🔐 Environment Variables Configuration

Copy .env.example to create your local .env file:

cp .env.example .env

Supported Configuration Keys

Variable	Description	Default Value
DATABASE_URL	SQLAlchemy connection string	sqlite:///./queueforge.db
APP_HOST	Host IP for Uvicorn server	127.0.0.1
APP_PORT	Listening port for server	8000
LOG_LEVEL	Application logging verbosity (DEBUG, INFO, WARNING, ERROR)	INFO
REPLAY_MAX_RETRIES	Max replay retry attempts before tripping breaker	3
REPLAY_BACKOFF_FACTOR	Multiplier for exponential backoff delay (seconds)	1.5
CIRCUIT_BREAKER_THRESHOLD	Consecutive replay failure count to trip breaker	5

---

🚀 Usage

1. Launching the Server

Run the application entry point:

python -m src.main

Or using uvicorn directly:

uvicorn src.main:app --host 127.0.0.1 --port 8000 --reload

Navigate to http://127.0.0.1:8000 in your browser to access the Interactive Payload Workbench.

---

2. API Endpoints Reference

Ingest a Failed DLQ Task

curl -X POST "http://127.0.0.1:8000/api/v1/tasks/ingest" \
     -H "Content-Type: application/json" \
     -d '{
           "queue_name": "payments_processing",
           "target_url": "http://127.0.0.1:9000/webhook/payment",
           "payload": {"order_id": "ORD-9912", "amount": "INVALID_FLOAT"},
           "error_message": "ValueError: Could not convert string to float: '\''INVALID_FLOAT'\''",
           "stack_trace": "Traceback (most recent call last):\n  File \"workers.py\", line 42, in process_payment\n    amt = float(payload[\"amount\"])\nValueError: Could not convert string to float: '\''INVALID_FLOAT'\''"
         }'

List Pending Dead-Letter Tasks

curl -X GET "http://127.0.0.1:8000/api/v1/tasks?status=PENDING&classification=Schema%20Mismatch"

Mutate a Malformed Task Payload

curl -X PATCH "http://127.0.0.1:8000/api/v1/tasks/1" \
     -H "Content-Type: application/json" \
     -d '{
           "updated_payload": {"order_id": "ORD-9912", "amount": 99.50},
           "operator_notes": "Fixed string to float representation error."
         }'

Trigger Deterministic Task Replay

curl -X POST "http://127.0.0.1:8000/api/v1/tasks/1/replay"

Retrieve System Telemetry Metrics

curl -X GET "http://127.0.0.1:8000/api/v1/telemetry"

---

🧪 Automated Testing Instructions

QueueForge Ops uses pytest and pytest-asyncio for unit, integration, and endpoint testing.

Execute All Tests

pytest

Execute Tests with Verbose Output & Code Coverage

pytest -v --cov=src --cov-report=term-missing

Run Core vs API Test Suites Separately

# Run core logic tests (Regex classifier, diffing, circuit breaker, replay backoff)
pytest tests/test_core.py

# Run API & endpoint integration tests
pytest tests/test_api.py

---

📁 Project Structure

queueforge-ops/
│
├── .env.example              # Environment configuration template
├── .gitignore                # Git exclusion rules
├── requirements.txt          # Python dependencies
│
├── src/                      # Application Source Code
│   ├── __init__.py           # Package marker
│   ├── main.py               # Application bootstrap, FastAPI setup & server start
│   ├── models.py             # SQLAlchemy ORM models & Pydantic request/response schemas
│   ├── core.py               # Regex classifier, diff engine, circuit breaker, replay client
│   └── api.py                # REST API endpoints & Jinja2 UI rendering routes
│
└── tests/                    # Automated Test Suite
    ├── __init__.py           # Test package marker
    ├── test_core.py          # Unit tests for core engine components
    └── test_api.py           # Integration tests for FastAPI endpoints & flows

---

🛤 Roadmap & Future Enhancements

• Multi-Broker Native Connectors: Direct ingestion integrations for RabbitMQ Dead Letter Exchanges (DLX), AWS SQS, and Redis Streams.
• Role-Based Access Control (RBAC): Fine-grained permissions distinguishing Viewer, Editor, and Replay Executer roles with OAuth2/OIDC integration.
• Automated Replay Rules Engine: Define customizable trigger conditions (e.g., auto-replay transient database timeout errors after 5 minutes).
• Prometheus Metrics Exporter: Expose /metrics path for open-telemetry queue tracking, alertmanager integration, and Grafana visualization.

---

📄 License

Distributed under the MIT License. See LICENSE for more information.