A worker pops a notification, then stops before committing its work. If the queue is only a dispatch accelerator, recovery must start from a durable record.
This runnable example stores three jobs in SQLite, removes one Spedo notification to simulate a crash, redispatches pending rows, and deliberately adds a duplicate. Four deliveries produce three local effects. The effect and processed-state transition share a single SQLite transaction.
Run the failure case
Requirements: Docker, Python 3, a free local port 6380. Save the full example below as transient_idempotent_job_accelerator.py.
docker run --rm -d --name spedo-demo \
-p 127.0.0.1:6380:6380 \
-e SPEDO_ADDR=0.0.0.0:6380 \
-e SPEDO_REQUIREPASS=demo-local-only \
torxx666/spedo:v0.74.0
SPEDO_HOST=127.0.0.1 SPEDO_PORT=6380 SPEDO_PASSWORD=demo-local-only \
python3 transient_idempotent_job_accelerator.py
docker stop spedo-demo
The example was tested against this public image on 9 November 2026. It checks the server before creating data, uses a unique queue and does not call FLUSHDB. The website advertised v0.75.0, which was unavailable during that check; this reproduction deliberately pins the tested v0.74.0 tag.
What the result means
The script validates three durable outbox rows and exactly three local effects despite a missing notification and a duplicate. Those effects are rows in the same SQLite database. Email, webhooks and external databases need their own delivery and idempotency contract. This is not a durable-message-broker or an exactly-once guarantee for external effects, and it is not a performance benchmark.
Complete example
#!/usr/bin/env python3
"""Spedo as a transient, idempotent job accelerator.
Run from the repository root:
docker compose up -d --build spedo
SPEDO_HOST=127.0.0.1 SPEDO_PORT=6380 \
python3 examples/transient_idempotent_job_accelerator.py
Optional authentication is read from ``SPEDO_USERNAME`` and ``SPEDO_PASSWORD``.
The demo deliberately uses only the Python standard library.
What this demonstrates
----------------------
* A SQLite outbox is the durable source of truth in this example.
* Spedo's ``QPUSH``/``QPOP`` path is used only to accelerate dispatch.
* A task popped just before a simulated worker crash disappears from Spedo, then
is recovered by redispatching the still-pending SQLite outbox record.
* Duplicate delivery is harmless because the simulated business effect and the
``pending -> processed`` transition happen in one SQLite transaction.
What this does *not* demonstrate
--------------------------------
It is not a reliable or durable queue implementation, and it does not provide
an exactly-once guarantee for external side effects. A production integration
must keep its own durable source of truth and use application-level idempotency
keys for effects outside that transaction.
"""
from __future__ import annotations
import argparse
import datetime as dt
import json
import os
import socket
import sqlite3
import sys
import tempfile
import uuid
from pathlib import Path
from typing import Any, Iterable
class RespError(RuntimeError):
"""A Redis/RESP error returned by the server."""
class RespClient:
"""Small RESP2 client kept here so this example has no package dependency."""
def __init__(self, host: str, port: int, timeout_seconds: float) -> None:
self._socket = socket.create_connection((host, port), timeout=timeout_seconds)
self._socket.settimeout(timeout_seconds)
self._reader = self._socket.makefile("rb")
def close(self) -> None:
try:
self._reader.close()
finally:
self._socket.close()
def __enter__(self) -> "RespClient":
return self
def __exit__(self, *_: object) -> None:
self.close()
def command(self, *parts: object) -> Any:
encoded_parts = [self._as_bytes(part) for part in parts]
payload = [f"*{len(encoded_parts)}\r\n".encode("ascii")]
for part in encoded_parts:
payload.extend((f"${len(part)}\r\n".encode("ascii"), part, b"\r\n"))
self._socket.sendall(b"".join(payload))
return self._read_response()
@staticmethod
def _as_bytes(value: object) -> bytes:
if isinstance(value, bytes):
return value
return str(value).encode("utf-8")
def _readline(self) -> bytes:
line = self._reader.readline()
if not line:
raise ConnectionError("Spedo closed the connection while reading a RESP response")
if not line.endswith(b"\r\n"):
raise RespError(f"malformed RESP line: {line!r}")
return line[:-2]
def _read_response(self) -> Any:
prefix = self._reader.read(1)
if not prefix:
raise ConnectionError("Spedo closed the connection while reading a RESP response")
line = self._readline()
if prefix == b"+":
return line.decode("utf-8", errors="replace")
if prefix == b"-":
raise RespError(line.decode("utf-8", errors="replace"))
if prefix == b":":
return int(line)
if prefix == b"$":
length = int(line)
if length == -1:
return None
if length < -1:
raise RespError(f"invalid RESP bulk length: {length}")
value = self._reader.read(length)
terminator = self._reader.read(2)
if len(value) != length or terminator != b"\r\n":
raise RespError("truncated RESP bulk response")
return value
if prefix == b"*":
count = int(line)
if count == -1:
return None
if count < -1:
raise RespError(f"invalid RESP array length: {count}")
return [self._read_response() for _ in range(count)]
raise RespError(f"unsupported RESP response prefix: {prefix!r}")
def utc_now() -> str:
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
def create_outbox(connection: sqlite3.Connection) -> None:
"""Create a deliberately small, local stand-in for a durable application DB."""
connection.execute("PRAGMA journal_mode=WAL")
connection.execute("PRAGMA synchronous=FULL")
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS outbox (
id TEXT PRIMARY KEY,
event_type TEXT NOT NULL,
payload_json TEXT NOT NULL,
state TEXT NOT NULL CHECK (state IN ('pending', 'processed')),
dispatch_count INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
processed_at TEXT
);
-- This represents the business side effect. The primary key is the
-- idempotency key: one effect may exist for each source outbox record.
CREATE TABLE IF NOT EXISTS delivered_notifications (
outbox_id TEXT PRIMARY KEY REFERENCES outbox(id),
body_json TEXT NOT NULL,
delivered_at TEXT NOT NULL
);
"""
)
def seed_outbox(connection: sqlite3.Connection, run_id: str) -> list[dict[str, str]]:
records = [
{
"id": f"invoice-email:{run_id}:001",
"event_type": "invoice.ready",
"payload_json": json.dumps(
{"invoice_id": "INV-1001", "recipient": "alice@example.test"},
separators=(",", ":"),
),
},
{
"id": f"invoice-email:{run_id}:002",
"event_type": "invoice.ready",
"payload_json": json.dumps(
{"invoice_id": "INV-1002", "recipient": "bob@example.test"},
separators=(",", ":"),
),
},
{
"id": f"invoice-email:{run_id}:003",
"event_type": "invoice.ready",
"payload_json": json.dumps(
{"invoice_id": "INV-1003", "recipient": "carol@example.test"},
separators=(",", ":"),
),
},
]
connection.executemany(
"""
INSERT INTO outbox (id, event_type, payload_json, state, created_at)
VALUES (:id, :event_type, :payload_json, 'pending', :created_at)
""",
[{**record, "created_at": utc_now()} for record in records],
)
return records
def pending_outbox_records(
connection: sqlite3.Connection, run_id: str, limit: int | None = None
) -> list[dict[str, str]]:
statement = """
SELECT id, event_type, payload_json
FROM outbox
WHERE state = 'pending' AND id LIKE ?
ORDER BY id
"""
params: list[object] = [f"%:{run_id}:%"]
if limit is not None:
statement += " LIMIT ?"
params.append(limit)
rows = connection.execute(statement, params).fetchall()
return [
{"id": row[0], "event_type": row[1], "payload_json": row[2]}
for row in rows
]
def to_queue_payload(record: dict[str, str]) -> str:
return json.dumps(record, separators=(",", ":"), sort_keys=True)
def enqueue_records(
client: RespClient,
connection: sqlite3.Connection,
queue_key: str,
records: Iterable[dict[str, str]],
) -> int:
"""Publish transient copies; the SQLite source remains pending until work commits."""
count = 0
for record in records:
queue_length = client.command(
"QPUSH",
queue_key,
to_queue_payload(record),
"PRIORITY",
0,
"FAMILY",
"billing-demo",
)
if not isinstance(queue_length, int):
raise RespError(f"QPUSH returned an unexpected response: {queue_length!r}")
connection.execute(
"UPDATE outbox SET dispatch_count = dispatch_count + 1 WHERE id = ?",
(record["id"],),
)
count += 1
return count
def pop_many(client: RespClient, queue_key: str, count: int) -> list[dict[str, str]]:
"""QPOP with a count always returns a RESP array in Spedo."""
response = client.command("QPOP", queue_key, count)
if response is None:
return []
if not isinstance(response, list):
raise RespError(f"QPOP returned an unexpected response: {response!r}")
records = []
for value in response:
if not isinstance(value, bytes):
raise RespError(f"QPOP item is not a bulk string: {value!r}")
record = json.loads(value.decode("utf-8"))
if not isinstance(record, dict) or not {"id", "event_type", "payload_json"} <= record.keys():
raise RespError(f"QPOP payload has the wrong shape: {record!r}")
records.append(record)
return records
def apply_idempotently(connection: sqlite3.Connection, record: dict[str, str]) -> bool:
"""Return True only when this delivery creates the business side effect.
The durable state transition and the sample effect share one local SQLite
transaction. In a real system, use the same pattern (or an idempotency key
accepted by the external provider) for side effects outside the database.
"""
connection.execute("BEGIN IMMEDIATE")
try:
changed = connection.execute(
"""
UPDATE outbox
SET state = 'processed', processed_at = ?
WHERE id = ? AND state = 'pending'
""",
(utc_now(), record["id"]),
).rowcount
if changed != 1:
connection.rollback()
return False
connection.execute(
"""
INSERT INTO delivered_notifications (outbox_id, body_json, delivered_at)
VALUES (?, ?, ?)
""",
(record["id"], record["payload_json"], utc_now()),
)
connection.commit()
return True
except BaseException:
connection.rollback()
raise
def validate_result(connection: sqlite3.Connection, run_id: str, expected_ids: set[str]) -> None:
states = dict(
connection.execute(
"SELECT id, state FROM outbox WHERE id LIKE ?", (f"%:{run_id}:%",)
).fetchall()
)
effect_ids = {
row[0]
for row in connection.execute(
"""
SELECT outbox_id
FROM delivered_notifications
WHERE outbox_id LIKE ?
""",
(f"%:{run_id}:%",),
)
}
assert set(states) == expected_ids, f"unexpected outbox rows: {states!r}"
assert set(states.values()) == {"processed"}, f"not all records were processed: {states!r}"
assert effect_ids == expected_ids, f"effects were not exactly one per source record: {effect_ids!r}"
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Demonstrate Spedo as a transient job accelerator over a SQLite outbox."
)
parser.add_argument(
"--outbox-path",
type=Path,
help="SQLite file used as the simulated durable source (default: a unique file in the OS temp directory).",
)
parser.add_argument(
"--timeout-seconds",
type=float,
default=float(os.environ.get("SPEDO_TIMEOUT_SECONDS", "3")),
help="TCP connect/read timeout; default: SPEDO_TIMEOUT_SECONDS or 3.",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
host = os.environ.get("SPEDO_HOST", "127.0.0.1")
try:
port = int(os.environ.get("SPEDO_PORT", "6380"))
except ValueError:
print("SPEDO_PORT must be an integer.", file=sys.stderr)
return 2
if args.timeout_seconds <= 0:
print("--timeout-seconds must be greater than zero.", file=sys.stderr)
return 2
run_id = uuid.uuid4().hex[:12]
queue_key = f"demo:transient-outbox:{run_id}:queue"
outbox_path = args.outbox_path or Path(tempfile.gettempdir()) / f"spedo-outbox-demo-{run_id}.sqlite3"
password = os.environ.get("SPEDO_PASSWORD")
username = os.environ.get("SPEDO_USERNAME")
print("Spedo transient job accelerator demo", flush=True)
print("=" * 42, flush=True)
print(f"Checking Spedo at {host}:{port} before creating demo data …", flush=True)
print("The SQLite outbox is the durable source in this demo; Spedo is transient.", flush=True)
try:
with RespClient(host, port, args.timeout_seconds) as client:
if password:
auth_result = (
client.command("AUTH", username, password)
if username
else client.command("AUTH", password)
)
if auth_result != "OK":
raise RespError(f"AUTH returned {auth_result!r}")
pong = client.command("PING")
if pong != "PONG":
raise RespError(f"PING returned {pong!r}, expected 'PONG'")
print("Startup check passed. No FLUSHDB is used; the queue key is unique for this run.")
outbox_path.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(outbox_path, isolation_level=None) as connection:
create_outbox(connection)
records = seed_outbox(connection, run_id)
expected_ids = {record["id"] for record in records}
print(f"SQLite durable outbox: {outbox_path}")
print(f"Transient Spedo queue: {queue_key}")
# Step 1: Spedo is fast transport only. Once QPOP removes this
# item, a crash before the transaction loses this transient copy.
first_dispatch = enqueue_records(client, connection, queue_key, records[:1])
assert first_dispatch == 1
lost_delivery = pop_many(client, queue_key, 1)
assert len(lost_delivery) == 1 and lost_delivery[0]["id"] == records[0]["id"]
print("1. Simulated worker crash after QPOP: transient copy is gone; SQLite row is still pending.")
still_pending = connection.execute(
"SELECT state FROM outbox WHERE id = ?", (records[0]["id"],)
).fetchone()
assert still_pending == ("pending",)
# Step 2: recovery reads the durable source, so it can restore
# the lost delivery. It is safe to enqueue more than once.
recovered = pending_outbox_records(connection, run_id)
assert {record["id"] for record in recovered} == expected_ids
assert enqueue_records(client, connection, queue_key, recovered) == len(records)
print("2. Recovery redispatches all pending SQLite rows, including the lost job.")
# Deliberately model a retry/duplicate. The worker must make it
# harmless rather than assuming QPOP grants exactly-once delivery.
assert enqueue_records(client, connection, queue_key, [records[0]]) == 1
print("3. A duplicate delivery is added deliberately to exercise idempotency.")
deliveries = pop_many(client, queue_key, 16)
assert len(deliveries) == len(records) + 1
created_effects = 0
ignored_duplicates = 0
for delivery in deliveries:
if apply_idempotently(connection, delivery):
created_effects += 1
else:
ignored_duplicates += 1
print(
f"4. Worker handled {len(deliveries)} deliveries: "
f"{created_effects} effects created, {ignored_duplicates} duplicate ignored."
)
assert created_effects == len(records)
assert ignored_duplicates == 1
validate_result(connection, run_id, expected_ids)
except (OSError, ConnectionError, RespError) as exc:
print(f"\nCould not run the demo against Spedo at {host}:{port}: {exc}", file=sys.stderr)
print(
"Start it with `docker compose up -d --build spedo`, set SPEDO_HOST/SPEDO_PORT "
"(and optional SPEDO_USERNAME/SPEDO_PASSWORD), then retry.",
file=sys.stderr,
)
return 1
except (AssertionError, json.JSONDecodeError, sqlite3.Error) as exc:
print(f"\nDemo validation failed: {exc}", file=sys.stderr)
return 1
print("\nVALIDATED: 3 durable outbox records produced exactly 3 local effects despite a lost QPOP and a duplicate.")
print("Spedo accelerated delivery; SQLite remained the recoverable source of truth.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
The records are synthetic. This walkthrough was prepared with AI assistance; the execution output was checked against a fresh local run and a separate run of the public Docker image.
If you try it, check which transaction owns the durable job in your system and whether your side effect can share that transaction. Record that boundary before adapting the example to an external service.
Project: https://spedo.dev?utm_source=dev-to&utm_medium=social&utm_campaign=spedo-launch&utm_content=post-45
Top comments (0)