# Qrusty 🦀

> The trusty priority queue server that never forgets

Qrusty is a high-performance priority queue server written in Rust, designed for
reliable message processing with filesystem persistence and acknowledgment support.

## Features

- 🚀 **Fast** - 50,000+ messages/second
- 💾 **Persistent** - Survives restarts with RocksDB storage (or run in-memory)
- 🎯 **Priority-based** - Numeric (u64) or text (lexicographic string) priorities
- ⚡ **Configurable Ordering** - Min-first or max-first priority queues
- ✅ **Acknowledgments** - Messages requeue on timeout
- 🔄 **Multiple queues** - Run dozens of isolated queues
- 🌐 **Simple HTTP API** - Easy integration with any language
- 🐳 **Docker-ready** - Single container deployment

## Priority Ordering

Qrusty supports two priority ordering modes that can be configured per queue:

### Max-First (Default)

- Higher priority values are processed first
- Priority 100 comes before priority 50
- Traditional "high priority = urgent" model
- Perfect for urgent/critical task processing

### Min-First

- Lower priority values are processed first
- Priority 10 comes before priority 50
- Unix process priority model (lower = higher priority)
- Ideal for batch processing and background jobs

### Priority Kinds

Each queue is configured with a `priority_kind` that determines which type of priority values it accepts:

- **Numeric** (default) — unsigned 64-bit integers (0 to 2^64-1)
- **Text** — arbitrary UTF-8 strings for lexicographic ordering

```bash
# Create a queue with text priorities
curl -X POST http://localhost:6784/create-queue \
  -H "Content-Type: application/json" \
  -d '{"name":"categories","config":{"ordering":"MinFirst","priority_kind":"Text"}}'

# Publish with a string priority
curl -X POST http://localhost:6784/publish \
  -H "Content-Type: application/json" \
  -d '{"queue":"categories","priority":"alpha","payload":"First category"}'
```

## Quick Start

### Create Queues with Different Orderings

```bash
# Create a max-first queue for urgent tasks
curl -X POST http://localhost:6784/create-queue \
  -H "Content-Type: application/json" \
  -d '{"name":"urgent","config":{"ordering":"MaxFirst"}}'

# Create a min-first queue for batch processing
curl -X POST http://localhost:6784/create-queue \
  -H "Content-Type: application/json" \
  -d '{"name":"batch","config":{"ordering":"MinFirst"}}'
```

### Publish Messages

```bash
# Publish to urgent queue (high priority = 100)
curl -X POST http://localhost:6784/publish \
  -H "Content-Type: application/json" \
  -d '{"queue":"urgent","priority":100,"payload":"Critical task"}'

# Publish to batch queue (low priority = 10)
curl -X POST http://localhost:6784/publish \
  -H "Content-Type: application/json" \
  -d '{"queue":"batch","priority":10,"payload":"Background job"}'
```

### Consume Messages

```bash
# Consume from urgent queue (gets highest priority first)
curl -X POST http://localhost:6784/consume/urgent \
  -H "Content-Type: application/json" \
  -d '{"consumer_id":"worker-1","timeout_seconds":30}'

# Consume from batch queue (gets lowest priority first)
curl -X POST http://localhost:6784/consume/batch \
  -H "Content-Type: application/json" \
  -d '{"consumer_id":"worker-2","timeout_seconds":30}'
```

### Lock Timeout

By default, consumed messages are locked for 30 seconds. For long-running
processing, you can specify a longer lock timeout at consume time (HTTP) or
subscribe time (WebSocket):

```bash
# HTTP: lock for 5 minutes
curl -X POST http://localhost:6784/consume/slow_queue \
  -H "Content-Type: application/json" \
  -d '{"consumer_id":"worker-1","timeout_seconds":300}'
```

```json
// WebSocket: subscribe with 5-minute lock
{"type": "subscribe", "queue": "slow_queue", "lock_timeout_secs": 300}

// WebSocket: renew with custom timeout
{"type": "renew", "queue": "slow_queue", "id": "msg-uuid", "lock_timeout_secs": 300}
```

### Queue Management

```bash
# Get queue statistics
curl http://localhost:6784/stats

# Update queue configuration (change allow_duplicates)
curl -X POST http://localhost:6784/update-queue \
  -H "Content-Type: application/json" \
  -d '{"name":"urgent","config":{"allow_duplicates":false}}'

# Rename a queue and update allow_duplicates
curl -X POST http://localhost:6784/update-queue \
  -H "Content-Type: application/json" \
  -d '{"name":"urgent","config":{"name":"high_priority","allow_duplicates":false}}'

# Purge all messages from a queue (preserves configuration)
curl -X POST http://localhost:6784/purge-queue/urgent

# Delete a queue and all its messages
curl -X DELETE http://localhost:6784/delete-queue/urgent
```

## API Endpoints

- `POST /create-queue` - Create a new queue with specific configuration
- `POST /update-queue` - Update queue name and/or allow_duplicates setting (queue type is immutable)
- `POST /publish` - Publish a message to a queue
- `POST /consume/{queue}` - Consume a message from a queue
- `POST /ack/{queue}/{id}` - Acknowledge a message
- `POST /nack/{queue}/{id}` - Negative acknowledge (requeue)
- `GET /stats` - Get queue statistics
- `POST /purge-queue/{queue}` - Remove all messages (keep configuration)
- `DELETE /delete-queue/{queue}` - Delete queue and all messages
- `GET /health` - Health / readiness check — always `200 OK`; body is `{"status":"ok"}` when fully ready, or `{"status":"starting","message":"..."}` while the persistent storage cache is being loaded in the background on startup

### Startup memory safety

Storage initialisation is designed to avoid OOM during startup on large
databases. Hot tiers are populated lazily on first pop (not pre-loaded),
and the integrity scan does not re-inline externalized payloads (which
would inflate the RocksDB write buffer). RocksDB memtables are flushed
between init phases to cap write-buffer memory.

### Startup mutation gate

While the server is initialising its persistent storage cache, **all mutation
endpoints** (create/update/delete queue, publish, consume, ack, nack, purge,
batch operations) return `503 Service Unavailable` with a `Retry-After: 2`
header and a JSON body:

```json
{
  "status": "starting",
  "message": "Qrusty is initialising. Retry shortly.",
  "retry_after_secs": 2
}
```

Read-only endpoints (`/health`, `/stats`, `/queues`, `/queue-stats/{queue}`,
`/operation-timings`, `/ws`) and static files remain accessible during
initialisation. This prevents bogus empty payloads caused by clients publishing
before the storage cache is fully reconstructed.

### Broken payload recovery

When the payload store is enabled and segment files are missing or corrupted
(e.g. after toggling `QRUSTY_PAYLOAD_STORE=disabled` and back), messages whose
payloads cannot be recovered are automatically deleted:

- **At startup**: the integrity scan detects and removes broken messages
- **At consume time**: if a broken message is encountered, it is deleted and
  the next valid message is returned instead

Consumers will never receive a message with an empty payload caused by a
missing segment file.

### Payload externalization threshold

Payloads smaller than `QRUSTY_EXTERNALIZE_MIN_BYTES` (default: 4096) stay
inline in RocksDB rather than being written to the payload store. This avoids
segment file overhead for small payloads where the `PayloadRef` metadata is
comparable in size to the payload itself.

### Memory pressure reclamation

When memory usage exceeds configurable thresholds, Qrusty progressively
reclaims memory:

- **Warning** (80%): flushes RocksDB memtables, shrinks block cache to 25%,
  reclaims per-queue structures for empty queues (dedup sets, hot tier entries,
  locked index entries), calls `shrink_to_fit()` on all dedup HashSets and
  the locked index to release over-allocated capacity, and runs a full-range
  RocksDB compaction to reclaim tombstone space from acked/deleted messages.
- **Critical** (90%): all Warning-level actions, then shrinks block cache
  further to 1 MB, shrinks hot tiers to 10%, compacts the payload store
  (including cleanup of all segments when no live payloads remain), and
  rejects new publishes until pressure subsides.

## Examples and Integration

For comprehensive examples and integration guides, see:

- [Queue Management Guide](integrations/examples/queue_management_guide.md) - Complete cURL, Python, and JavaScript examples
- [Python Integration Examples](integrations/python/) - Production-ready Python integration
- [Node-RED Integration](integrations/node-red/) - For IoT and workflow automation
- [Priority Usage Examples](integrations/examples/priority_examples.md) - Understanding priority ordering
- [API Usage Examples](integrations/examples/api_usage_examples.md) - Comprehensive API documentation

Example consume with a longer timeout:

```bash
curl -X POST http://localhost:6784/consume/batch \
  -H "Content-Type: application/json" \
  -d '{"consumer_id":"worker-2","timeout_seconds":60}'
```

## Make Targets

Run these from the dev container, or an equivalent environment:

- `make all` - Build, generate docs, run machete, and test
- `make build` - Build the Rust project with cargo
- `make doc` - Generate documentation with cargo doc
- `make machete` - Run cargo machete (dependency analysis)
- `make test` - Run tests with cargo test
- `make smoke-test` - Run the live end-to-end smoke test (see below)
- `make smoke-test-memory` - Same smoke test using in-memory storage
- `make traceability` - Print traceability report to stdout
- `make traceability-report` - Save traceability report to `docs/traceability_report.md`

### Live Smoke Test

`make smoke-test` builds the binary, starts a real qrusty server on an
isolated port, runs concurrent publishers and consumers against three queues
for 30 seconds, polls `/stats` every 3 seconds to display a live throughput
table, and asserts that `publish_rate_per_sec` and `ack_rate_per_sec` are
non-zero — proving the server actually processes messages and reports accurate
statistics.

```bash
# Default run (30 s workload, port 17784)
make smoke-test

# Longer run on a custom port
make smoke-test SMOKE_ARGS="--duration 120 --port 19000"

# Point at a release build
python scripts/smoke_test.py --binary target/release/qrusty --duration 60
```

Requirements: SYS-0011, SYS-0012

Coverage targets (requires `cargo llvm-cov` in your environment):

- `make coverage` - Run the full Rust test suite with coverage and print a summary
- `make coverage-missing` - Generate `target/llvm-cov/missing.txt` with uncovered line numbers
- `make coverage-summary-json` - Generate `target/llvm-cov/summary.json` (machine-readable totals)
- `make coverage-html` - Generate an HTML report at `target/llvm-cov/html/index.html`
- `make coverage-clean` - Remove coverage artifacts generated by `cargo llvm-cov`

Run this outside the dev container, unless it's been altered to allow docker-in-docker:

- `make show-docs` - Build and open the docs in the default browser
- `make build-qrusty` - Build and push Docker image for ARM64 platform

## Requirements (Doorstop)

This repo uses [Doorstop](https://doorstop.info/) for Requirements-Based Verification (RBV).

### Document hierarchy

| Document                    | Prefix | Parent   | Purpose                                                |
| --------------------------- | ------ | -------- | ------------------------------------------------------ |
| `requirements/system/`      | SYS    | _(root)_ | System-level requirements (deployment, health, web UI) |
| `requirements/persistence/` | PER    | SYS      | Persistence and recovery                               |
| `requirements/scheduling/`  | SCH    | SYS      | Priority and scheduling                                |
| `requirements/delivery/`    | DLV    | SYS      | Delivery semantics (lock, ack, nack, retry)            |
| `requirements/api/`         | API    | SYS      | HTTP API endpoints                                     |
| `requirements/websocket/`   | WS     | SYS      | WebSocket API                                          |

Each requirement YAML item includes:

- **text** – "The system shall …" requirement statement.
- **verification** – IADT method: `inspection`, `analysis`, `demonstration`, or `test`.
- **links** – parent requirement UIDs (for traceability).

### Run Doorstop in the dev container

The devcontainer includes a helper script that bootstraps a local venv at `.venv/` (if missing), installs Doorstop, and starts `doorstop-server`:

```bash
bash .devcontainer/start-doorstop.sh
```

Then open:

- <http://localhost:17867/>
- <http://localhost:17867/documents/SYS.html>

To stop the server:

```bash
pkill -f 'doorstop-server --host 0.0.0.0 --port 17867' || true
```

### Traceability

Requirements are linked to implementation code and tests via special comment tags:

**Implementation tagging** (Rust doc comments):

```rust
/// Requirements: PER-0001, PER-0002, SCH-0001
pub async fn push(&self, message: Message) -> Result<String, StorageError> {
```

**Test tagging** (Rust comments):

```rust
// Verifies: DLV-0001, DLV-0002, API-0003
#[tokio::test]
async fn test_push_and_pop_message() {
```

**Generate traceability report:**

```bash
# Print to stdout
make traceability

# Save to docs/traceability_report.md
make traceability-report
```

The report shows:

- Which requirements have implementation and test coverage
- Untraced requirements (gaps in coverage)
- Orphan tags (tags referencing non-existent requirements)
- Detailed links to source locations

See [docs/traceability_report.md](docs/traceability_report.md) for the current report.

## In-Memory Storage Mode

Qrusty supports an optional in-memory storage backend (`STORAGE_MODE=memory`)
that eliminates all disk I/O. This is useful when persistent storage is
unavailable or unreliable (e.g. a failing drive).

All functional behaviour (priority ordering, locking, ACK/NACK, dead-letter
queues, duplicate detection, timeout monitoring) is identical to RocksDB mode.

**All data is lost on restart.** A warning is logged at startup.

```bash
# Docker
docker run -p 6784:6784 -e STORAGE_MODE=memory greeng340or/qrusty

# Cargo
STORAGE_MODE=memory cargo run
```

To switch back, remove `STORAGE_MODE=memory` (or set it to `rocksdb`) and
ensure a volume is mounted at `/data`.

Requirements: SYS-0013, PER-0011

## Using the Docker Image

```bash
# Pull the image
docker pull greeng340or/qrusty

# Run with default settings (RocksDB persistence)
docker run -p 6784:6784 greeng340or/qrusty

# Run with custom data path and bind address
docker run -p 6784:6784 \
  -e DATA_PATH=/data \
  -e BIND_ADDR=0.0.0.0:6784 \
  -v qrusty-data:/data \
  greeng340or/qrusty

# Run with custom RocksDB block cache (default 64 MB)
docker run -p 6784:6784 -e ROCKSDB_CACHE_MB=128 greeng340or/qrusty

# Run with custom hot tier size (default 1000 messages per queue)
docker run -p 6784:6784 -e QRUSTY_HOT_TIER_SIZE=2000 -e QRUSTY_REFILL_THRESHOLD=500 greeng340or/qrusty

# Run with in-memory storage (no disk I/O)
docker run -p 6784:6784 -e STORAGE_MODE=memory greeng340or/qrusty

# Build locally (from a prompt that can run docker)
make build-qrusty
```

### Environment Variables

All settings have sensible defaults. See [`example.env`](example.env) for a template.

| Variable                                | Default             | Description                                               |
| --------------------------------------- | ------------------- | --------------------------------------------------------- |
| `DATA_PATH`                             | `/data`             | Directory for RocksDB and payload store data              |
| `BIND_ADDR`                             | `0.0.0.0:6784`      | Server bind address                                       |
| `STORAGE_MODE`                          | `rocksdb`           | Storage backend: `rocksdb` or `memory`                    |
| `MAX_CONNECTIONS`                       | `10000`             | Max concurrent HTTP/WS connections                        |
| `WEBUI_DIR`                             | `/opt/qrusty/webui` | Web UI static files directory                             |
| `ROCKSDB_CACHE_MB`                      | `128`               | Block cache size in MB (includes index/filter)            |
| `ROCKSDB_WRITE_BUFFER_MB`               | `16`                | RocksDB write buffer per memtable in MB                   |
| `ROCKSDB_MAX_OPEN_FILES`                | `128`               | Max open SST file handles                                 |
| `QRUSTY_HOT_TIER_SIZE`                  | `1000`              | Max available messages per queue in hot tier              |
| `QRUSTY_REFILL_THRESHOLD`               | `250`               | Refill hot tier when it drops to this count               |
| `QRUSTY_SEGMENT_MAX_MB`                 | `256`               | Payload store segment rotation size in MB                 |
| `QRUSTY_COMPACT_INTERVAL_SECS`          | `300`               | Payload compaction interval (seconds)                     |
| `QRUSTY_MEMORY_LIMIT_MB`                | auto                | Memory limit override (auto-detected from cgroup)         |
| `QRUSTY_MEMORY_PRESSURE_THRESHOLD`      | `0.80`              | Usage ratio that enters pressure (Warning level)          |
| `QRUSTY_MEMORY_PRESSURE_EXIT_THRESHOLD` | `0.70`              | Usage ratio that exits pressure (hysteresis)              |
| `QRUSTY_MEMORY_CRITICAL_THRESHOLD`      | `0.90`              | Usage ratio that triggers Critical level (shed publishes) |
| `QRUSTY_MAX_LOCKED_INDEX`               | `500000`            | Max locked message index entries                          |
| `WS_PING_INTERVAL_SECS`                 | `30`                | WebSocket ping interval                                   |
| `WS_PING_TIMEOUT_SECS`                  | `10`                | WebSocket ping timeout                                    |
