Recover a lost queue notification with a SQLite outbox

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 

#!/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 Source：https://dev.to/torxx666/recover-a-lost-queue-notification-with-a-sqlite-outbox-53a0
My favorites