.webp)
Replace Kafka with a production SQLite queue on a $5 VPS: batched CTE claims, WAL checkpoints, backoff, and dead-letter handling.
Kafka is a distributed log system. A $5 VPS background-job problem is usually not. For webhook retries, email dispatch, image processing, and other single-node workloads, a SQLite-backed queue can provide durable, at-least-once job delivery without brokers, partitions, or a second machine.
This is not a claim that SQLite outperforms Kafka at cluster scale. It is a claim that many teams install Kafka where a WAL-mode table and a disciplined worker would have been enough.
When the Replacement Is Honest
Good fit:
- one primary node
- tens to thousands of jobs per minute (and sometimes more with batching)
- at-least-once processing with idempotent handlers
- operational simplicity over multi-subscriber log fan-out
- many independent consumer groups across teams
- multi-region replay and long retained streams
- very high sustained multi-producer log traffic
- strict cross-service event-bus semantics
Failure Modes in Naive SQLite Queues
- Write contention (SQLITE_BUSY)
BEGIN IMMEDIATE takes a reserved/write lock. Multiple claim workers without backoff will collide. The failure mode is not theoretical; it shows up as database is locked under modest concurrency. - Unbatched claim IOPS waste
One transaction per job ties throughput to single-row commit cost. On small VPS disks, that ceiling is much lower than CPU would suggest. - Unbounded WAL growth
Long-lived readers or missing checkpoints let queue.db-wal grow until the disk fills. On a 20 GB VPS this becomes an outage. - Poison-pill retries
A stale-lock reaper that only resets processing → pending can create an infinite crash loop. Dead-letter thresholds must be enforced in the data model.
Schema and PRAGMAs
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA busy_timeout = 5000;
PRAGMA temp_store = MEMORY;
CREATE TABLE jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
queue TEXT NOT NULL DEFAULT 'default',
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending', -- pending|processing|done|failed
attempts INTEGER NOT NULL DEFAULT 0,
max_attempts INTEGER NOT NULL DEFAULT 5,
available_at INTEGER NOT NULL,
locked_at INTEGER,
locked_by TEXT,
created_at INTEGER NOT NULL,
error TEXT
);
CREATE INDEX idx_jobs_poll
ON jobs(queue, status, available_at, id)
WHERE status = 'pending';
CREATE INDEX idx_jobs_reaper
ON jobs(status, locked_at)
WHERE status = 'processing';Throughput Model
Single-job claim throughput is bounded by lock + write + flush cost:
Batching (B) claims into one transaction amortizes commit cost:
With B = 50 and modest contention efficiency
eta approx 0.95 , throughput can rise by an order of magnitude relative to single-row claims. Exact numbers depend on disk, payload size, and competing writers; the scaling shape is the important part.Production Python Engine
import json
import random
import sqlite3
import time
import uuid
from typing import Any, Dict, List
class ProductionSQLiteQueue:
def __init__(self, db_path: str = "queue.db", max_busy_timeout_ms: int = 5000):
self.db_path = db_path
self.worker_id = f"worker-{uuid.uuid4().hex[:8]}"
self.max_busy_timeout_ms = max_busy_timeout_ms
self._init_db()
def _get_connection(self) -> sqlite3.Connection:
conn = sqlite3.connect(
self.db_path,
isolation_level=None,
timeout=self.max_busy_timeout_ms / 1000.0,
check_same_thread=False,
)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode = WAL;")
conn.execute("PRAGMA synchronous = NORMAL;")
conn.execute(f"PRAGMA busy_timeout = {self.max_busy_timeout_ms};")
conn.execute("PRAGMA temp_store = MEMORY;")
return conn
def _init_db(self) -> None:
with self._get_connection() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
queue TEXT NOT NULL DEFAULT 'default',
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0,
max_attempts INTEGER NOT NULL DEFAULT 5,
available_at INTEGER NOT NULL,
locked_at INTEGER,
locked_by TEXT,
created_at INTEGER NOT NULL,
error TEXT
);
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_jobs_poll
ON jobs(queue, status, available_at, id)
WHERE status = 'pending';
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_jobs_reaper
ON jobs(status, locked_at)
WHERE status = 'processing';
""")
def enqueue_batch(
self,
payloads: List[Dict[str, Any]],
queue: str = "default",
delay_sec: int = 0,
) -> None:
now = int(time.time())
available_at = now + delay_sec
rows = [(queue, json.dumps(p), "pending", available_at, now) for p in payloads]
conn = self._get_connection()
try:
conn.execute("BEGIN IMMEDIATE;")
conn.executemany(
"""
INSERT INTO jobs (queue, payload, status, available_at, created_at)
VALUES (?, ?, ?, ?, ?)
""",
rows,
)
conn.execute("COMMIT;")
except Exception:
conn.execute("ROLLBACK;")
raise
finally:
conn.close()
def claim_batch(self, queue: str = "default", batch_size: int = 10) -> List[sqlite3.Row]:
conn = self._get_connection()
now = int(time.time())
for attempt in range(5):
try:
conn.execute("BEGIN IMMEDIATE;")
cur = conn.execute(
"""
WITH claimable AS (
SELECT id FROM jobs
WHERE queue = ?
AND status = 'pending'
AND available_at <= ?
ORDER BY id ASC
LIMIT ?
)
UPDATE jobs
SET status = 'processing',
locked_at = ?,
locked_by = ?,
attempts = attempts + 1
WHERE id IN (SELECT id FROM claimable)
RETURNING *;
""",
(queue, now, batch_size, now, self.worker_id),
)
rows = cur.fetchall()
conn.execute("COMMIT;")
return rows
except sqlite3.OperationalError as e:
conn.execute("ROLLBACK;")
msg = str(e).lower()
if "locked" in msg or "busy" in msg:
# exponential backoff + jitter
time.sleep((2 ** attempt) * 0.01 + random.uniform(0, 0.05))
else:
raise
finally:
if attempt == 4:
conn.close()
conn.close()
return []
def ack_batch(self, job_ids: List[int]) -> None:
if not job_ids:
return
placeholders = ",".join("?" * len(job_ids))
conn = self._get_connection()
try:
conn.execute("BEGIN IMMEDIATE;")
conn.execute(
f"UPDATE jobs SET status = 'done' WHERE id IN ({placeholders})",
job_ids,
)
conn.execute("COMMIT;")
except Exception:
conn.execute("ROLLBACK;")
raise
finally:
conn.close()
def nack_or_fail(self, job_id: int, error_msg: str, retry_delay: int = 30) -> None:
conn = self._get_connection()
now = int(time.time())
try:
conn.execute("BEGIN IMMEDIATE;")
row = conn.execute(
"SELECT attempts, max_attempts FROM jobs WHERE id = ?",
(job_id,),
).fetchone()
if row and row["attempts"] >= row["max_attempts"]:
# Dead-letter / terminal failure
conn.execute(
"""
UPDATE jobs
SET status = 'failed',
error = ?,
locked_at = NULL,
locked_by = NULL
WHERE id = ?
""",
(error_msg, job_id),
)
else:
conn.execute(
"""
UPDATE jobs
SET status = 'pending',
error = ?,
locked_at = NULL,
locked_by = NULL,
available_at = ?
WHERE id = ?
""",
(error_msg, now + retry_delay, job_id),
)
conn.execute("COMMIT;")
except Exception:
conn.execute("ROLLBACK;")
raise
finally:
conn.close()
def reap_stale_locks(self, visibility_timeout_sec: int = 300) -> int:
conn = self._get_connection()
cutoff = int(time.time()) - visibility_timeout_sec
try:
conn.execute("BEGIN IMMEDIATE;")
cur = conn.execute(
"""
UPDATE jobs
SET status = 'pending',
locked_at = NULL,
locked_by = NULL
WHERE status = 'processing'
AND locked_at < ?
RETURNING id;
""",
(cutoff,),
)
n = len(cur.fetchall())
conn.execute("COMMIT;")
return n
except Exception:
conn.execute("ROLLBACK;")
raise
finally:
conn.close()
def checkpoint_wal(self) -> None:
conn = self._get_connection()
try:
conn.execute("PRAGMA wal_checkpoint(PASSIVE);")
finally:
conn.close()
Worker Loop Recommendations
- Claim in batches, process, then ack_batch.
- Use jittered backoff on lock errors; do not spin.
- Run reap_stale_locks() on an interval.
- Run checkpoint_wal() periodically under write load.
- Keep payloads small; store large blobs elsewhere and queue references.
- Make handlers idempotent—at-least-once delivery implies duplicates after crashes.
Durability and Backup
synchronous = NORMAL with WAL is a common performance/durability balance for job queues. If every job is financial-critical, evaluate stronger durability settings and continuous backup. Tools such as Litestream can replicate SQLite WAL changes to object storage for low-cost DR on small VPS setups.
What You Give Up Relative to Kafka
- independent consumer groups at org scale
- long retained multi-subscriber logs
- broker-level partition elasticity
- mature multi-tenant event-bus tooling
Operational Metrics
Track at minimum:
- pending depth by queue
- age of oldest pending job
- claim wait / lock retries
- failed DLQ rate
- WAL size / checkpoint age
Trade-offs
- single-node durability boundary unless replicated
- write serialization under heavy multi-writer contention
- not a cross-language enterprise bus by default
- exactly-once requires idempotent consumers and careful side effects
Final Notes
Replacing Kafka with SQLite on a $5 VPS is sound when the problem is durable jobs, not distributed log transport. The production version is not “one table and a while loop.” It is batched claims, backoff under SQLITE_BUSY, WAL checkpoint discipline, visibility timeouts, and dead-letter thresholds.Get those mechanics right and SQLite becomes a boring, inexpensive queue that keeps the architecture proportional to the workload. Reach for Kafka when you need Kafka’s distributed properties—not as a default background-worker dependency.