Honker

Honker: Postgres-Style Messaging for SQLite

Honker is a SQLite extension plus language bindings that brings Postgres NOTIFY/LISTEN semantics to SQLite. You get push-style event delivery with single-digit millisecond latency, durable work queues with retries and dead-letter handling, event streams with per-consumer offsets, and a leader-elected periodic task scheduler — all stored in your existing .db file with no external daemon or broker.

When to use this skill

• You want cross-process pub/sub without a message broker
• You need a background job queue with retries, priorities, and dead-letter
• You want event streams with per-consumer offset tracking (Kafka-like)
• You need transactional outbox — enqueue jobs/events atomically with business writes
• You want periodic/cron tasks with leader election
• Your application already uses SQLite and you don't want to add Postgres or Redis just for messaging

When NOT to use this skill

• You need multi-machine replication — SQLite is single-host; use Postgres
• You need workflow DAGs (task chaining, pipelines, chords) — use Temporal
• You're already on Postgres — use the pg-messaging skill instead
• You need sub-millisecond latency — honker's wake latency is 1-5ms typical

Architecture

WAL-based signaling

Honker requires PRAGMA journal_mode = WAL. It monitors the .db-wal sidecar file using stat(2) polling at ~1ms granularity. When any process commits a write, the WAL file's size/mtime changes, and honker fans out async notifications to all subscribers on that database.

stat(2) takes under 1 microsecond per call — 100 listeners cost the same as

1. The cost is one syscall per millisecond per database, regardless of subscriber count.

Single writer

SQLite supports one writer and concurrent readers. Honker's claim operations use indexed UPDATE ... RETURNING statements; acknowledgments use DELETE. This avoids the exclusive locks that would serialize work in DELETE/TRUNCATE journal modes.

Transactional outbox (built-in)

Jobs, events, and notifications are INSERTs within the caller's transaction. Business writes and side-effect enqueues commit or rollback atomically — no separate dispatch table or daemon process. This eliminates the classic "notification sent but transaction rolled back" problem.

Over-triggering

All subscribers wake on any commit. Filtering happens in the SELECT path, not trigger dispatch. Each wasted wake is one indexed SELECT (microseconds). This design prioritizes never missing a real event over eliminating false positives.

Three Primitives

1. Notify (Ephemeral Pub/Sub)

Fire-and-forget channel notifications with no replay. Listeners attach at MAX(id) of the notification table — history is not replayed.

Not durable: if a listener is offline when a notification fires, it misses the event. Use streams for durable replay.

Stored in _honker_notifications table.

2. Stream (Durable Pub/Sub)

Persistent event log with per-consumer offset tracking. Each named consumer tracks its offset in _honker_stream_consumers. On subscribe, the iterator replays events past the saved offset, then transitions to live delivery on WAL wakes.

At-least-once semantics: crash re-delivers events up to the last flushed offset. Auto-saves offsets every 1000 events or 1 second (configurable via save_every_n / save_every_s), or manual flush.

Stored in _honker_streams and _honker_stream_consumers tables.

3. Queue (At-Least-Once Work)

Durable task queue with retries, priorities, delayed jobs, and dead-letter handling. Claims lock jobs for a visibility window (default 300s); if a worker crashes, the job re-enters the queue after expiry. Exhausted retries move to _honker_dead.

A partial index on (queue, priority DESC, run_at, id) WHERE state IN ('pending','processing') keeps the claim hot path bounded by working-set size, not historical size.

Stored in _honker_live (active) and _honker_dead (failed) tables.

Installation

Python

pip install honker

import honker
db = honker.open("app.db")

Node.js

npm install @russellthehippo/honker-node

const { open } = require('@russellthehippo/honker-node');
const db = open('app.db');

Rust

[dependencies]
honker = "0.x"

Go, Ruby, Bun, Elixir

Bindings available — all wrap the same loadable extension binary. Install via the language's package manager.

Raw SQLite extension (any language)

.load ./libhonker_ext
SELECT honker_bootstrap();

Any SQLite 3.9+ client can load the extension. The honker_bootstrap() call creates the schema tables if they don't exist.

Python API

Queue: enqueue and claim

emails = db.queue("emails")
emails.enqueue({"to": "alice@example.com"})

async for job in emails.claim("worker-1"):
    try:
        send(job.payload)
        job.ack()
    except Exception as e:
        job.retry(delay_s=60, error=str(e))

Transactional enqueue

Enqueue atomically with a business write — both commit or both rollback:

with db.transaction() as tx:
    tx.execute("INSERT INTO orders (user_id) VALUES (?)", [42])
    emails.enqueue({"to": "alice@example.com"}, tx=tx)

Task decorator

@emails.task(retries=3, timeout_s=30)
def send_email(to: str, subject: str) -> dict:
    ...
    return {"sent_at": time.time()}

# Caller — blocks until result is available
r = send_email("alice@example.com", "Hi")
print(r.get(timeout=10))

Run workers via CLI:

python -m honker worker myapp.tasks:db --queue=emails --concurrency=4

Stream: publish and subscribe

stream = db.stream("user-events")

# Publish atomically with a business write
with db.transaction() as tx:
    tx.execute("UPDATE users SET name=? WHERE id=?", [name, uid])
    stream.publish({"user_id": uid, "change": "name"}, tx=tx)

# Subscribe with automatic offset tracking
async for event in stream.subscribe(consumer="dashboard"):
    await push_to_browser(event)

Notify: ephemeral pub/sub

# Listen
async for n in db.listen("orders"):
    print(n.channel, n.payload)

# Notify atomically with a business write
with db.transaction() as tx:
    tx.execute("INSERT INTO orders (id, total) VALUES (?, ?)", [42, 99.99])
    tx.notify("orders", {"id": 42})

Periodic tasks

@emails.periodic_task(crontab("0 3 * * *"))
def nightly_backup():
    ...

Priority and delayed jobs

queue = db.queue("tasks")

# Higher priority runs first (default 0)
queue.enqueue({"type": "urgent"}, priority=10)
queue.enqueue({"type": "background"}, priority=0)

# Delayed job — won't be claimable until run_at
from datetime import datetime, timedelta, timezone

run_at = datetime.now(timezone.utc) + timedelta(minutes=30)
queue.enqueue({"type": "scheduled_report"}, run_at=run_at)

Dead-letter inspection and replay

import sqlite3

conn = sqlite3.connect("app.db")

# Inspect failed jobs
dead = conn.execute("""
    SELECT id, queue, payload, error, attempts
    FROM _honker_dead
    WHERE queue = ?
    ORDER BY id DESC
    LIMIT 20
""", ["emails"]).fetchall()

for job in dead:
    print(f"Job {job[0]}: {job[3]} (attempts={job[4]})")

# Replay a dead job by moving it back to _honker_live
def replay_dead_job(conn, job_id):
    with conn:
        conn.execute("""
            INSERT INTO _honker_live (queue, payload, state, priority, run_at, attempts)
            SELECT queue, payload, 'pending', priority, unixepoch(), 0
            FROM _honker_dead WHERE id = ?
        """, [job_id])
        conn.execute("DELETE FROM _honker_dead WHERE id = ?", [job_id])

Named locks

# Acquire a lock for 60 seconds
acquired = db.lock_acquire("data-export", holder="worker-1", ttl_s=60)
if acquired:
    try:
        run_export()
    finally:
        db.lock_release("data-export", holder="worker-1")

Rate limiting

# Allow 10 requests per 60-second window
allowed = db.rate_limit_try("api:user:42", limit=10, window_s=60)
if not allowed:
    raise RateLimitExceeded()

# Periodic cleanup of expired windows
db.rate_limit_sweep(max_age_s=3600)

Result storage with TTL

@emails.task(retries=3, timeout_s=30)
def generate_report(user_id: int) -> dict:
    report = build_report(user_id)
    return {"url": report.url, "rows": report.row_count}

# Caller retrieves the result
result = generate_report(42)
data = result.get(timeout=60)  # Blocks until result or timeout
print(data["url"])

# Results auto-expire per the TTL set on the task/queue

Go API

Go bindings wrap the same loadable extension. The idiomatic approach is to load the extension via your SQLite driver and call the SQL functions directly. This works with any Go SQLite driver (mattn/go-sqlite3, zombiezen/go-sqlite, modernc.org/sqlite).

The examples below use mattn/go-sqlite3 (CGo) since it has the widest adoption. The same SQL calls work identically with other drivers.

Setup: load the extension

package main

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/mattn/go-sqlite3"
)

func init() {
	// Register a driver that auto-loads the honker extension
	sql.Register("sqlite3_honker", &sqlite3.SQLiteDriver{
		Extensions: []string{"libhonker_ext"},
	})
}

func openDB(path string) (*sql.DB, error) {
	db, err := sql.Open("sqlite3_honker", path+"?_journal_mode=WAL")
	if err != nil {
		return nil, fmt.Errorf("open db: %w", err)
	}
	// Bootstrap honker tables
	if _, err := db.Exec("SELECT honker_bootstrap()"); err != nil {
		return nil, fmt.Errorf("honker bootstrap: %w", err)
	}
	return db, nil
}

Queue: enqueue and claim

func enqueue(db *sql.DB, queue string, payload any) error {
	data, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal payload: %w", err)
	}
	_, err = db.Exec(`
		INSERT INTO _honker_live (queue, payload, state, priority, run_at, attempts)
		VALUES (?, ?, 'pending', 0, unixepoch(), 0)
	`, queue, string(data))
	if err != nil {
		return fmt.Errorf("enqueue: %w", err)
	}
	return nil
}

func claimAndProcess(db *sql.DB, queue, workerID string) error {
	for {
		var raw string
		row := db.QueryRow(
			"SELECT honker_claim_batch(?, ?, 1, 300)",
			queue, workerID,
		)
		if err := row.Scan(&raw); err != nil {
			return fmt.Errorf("claim: %w", err)
		}

		var jobs []struct {
			ID      int             `json:"id"`
			Payload json.RawMessage `json:"payload"`
		}
		if err := json.Unmarshal([]byte(raw), &jobs); err != nil {
			return fmt.Errorf("unmarshal claim: %w", err)
		}
		if len(jobs) == 0 {
			time.Sleep(50 * time.Millisecond)
			continue
		}

		for _, job := range jobs {
			if err := processJob(job.Payload); err != nil {
				log.Printf("job %d failed: %v", job.ID, err)
				// Job will be reclaimed after visibility timeout expires
				continue
			}
			ackIDs, _ := json.Marshal([]int{job.ID})
			db.Exec("SELECT honker_ack_batch(?, ?)", string(ackIDs), workerID)
		}
	}
}

Transactional enqueue

func createOrderWithEmail(db *sql.DB, userID int, email string) error {
	tx, err := db.Begin()
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	_, err = tx.Exec("INSERT INTO orders (user_id) VALUES (?)", userID)
	if err != nil {
		return fmt.Errorf("insert order: %w", err)
	}

	payload, _ := json.Marshal(map[string]string{
		"to":       email,
		"template": "order_confirmation",
	})
	_, err = tx.Exec(`
		INSERT INTO _honker_live (queue, payload, state, priority, run_at, attempts)
		VALUES ('emails', ?, 'pending', 0, unixepoch(), 0)
	`, string(payload))
	if err != nil {
		return fmt.Errorf("enqueue email: %w", err)
	}

	// Both the order and the email job commit or rollback together
	return tx.Commit()
}

Stream: publish and subscribe

func publishEvent(db *sql.DB, stream, key string, payload any) (int64, error) {
	data, err := json.Marshal(payload)
	if err != nil {
		return 0, fmt.Errorf("marshal: %w", err)
	}
	var offset int64
	err = db.QueryRow(
		"SELECT honker_stream_publish(?, ?, ?)",
		stream, key, string(data),
	).Scan(&offset)
	if err != nil {
		return 0, fmt.Errorf("publish: %w", err)
	}
	return offset, nil
}

func subscribe(ctx context.Context, db *sql.DB, stream, consumer string, handler func(json.RawMessage) error) error {
	// Get last saved offset
	var offset int64
	db.QueryRow(
		"SELECT honker_stream_get_offset(?, ?)", consumer, stream,
	).Scan(&offset)

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		var raw string
		err := db.QueryRow(
			"SELECT honker_stream_read_since(?, ?, 100)",
			stream, offset,
		).Scan(&raw)
		if err != nil {
			return fmt.Errorf("read stream: %w", err)
		}

		var events []struct {
			ID      int64           `json:"id"`
			Payload json.RawMessage `json:"payload"`
		}
		json.Unmarshal([]byte(raw), &events)

		for _, evt := range events {
			if err := handler(evt.Payload); err != nil {
				return fmt.Errorf("handle event %d: %w", evt.ID, err)
			}
			offset = evt.ID
		}

		// Periodically save offset
		if len(events) > 0 {
			db.Exec("SELECT honker_stream_save_offset(?, ?, ?)",
				consumer, stream, offset)
		}

		if len(events) == 0 {
			time.Sleep(5 * time.Millisecond) // Wait for WAL change
		}
	}
}

Notify: ephemeral pub/sub with WAL watcher

// Publish a notification inside a transaction
func notifyInTx(tx *sql.Tx, channel string, payload any) error {
	data, _ := json.Marshal(payload)
	_, err := tx.Exec("SELECT notify(?, ?)", channel, string(data))
	return err
}

// Poll for notifications (pair with stat(2) watcher for push-style)
func pollNotifications(db *sql.DB, channel string, sinceID int64) ([]json.RawMessage, int64, error) {
	rows, err := db.Query(`
		SELECT id, payload FROM _honker_notifications
		WHERE channel = ? AND id > ?
		ORDER BY id
	`, channel, sinceID)
	if err != nil {
		return nil, sinceID, fmt.Errorf("poll: %w", err)
	}
	defer rows.Close()

	var results []json.RawMessage
	var lastID int64 = sinceID
	for rows.Next() {
		var id int64
		var payload string
		rows.Scan(&id, &payload)
		results = append(results, json.RawMessage(payload))
		lastID = id
	}
	return results, lastID, rows.Err()
}

Named locks

func withLock(db *sql.DB, name, holder string, ttlS int, fn func() error) error {
	var acquired int
	db.QueryRow(
		"SELECT honker_lock_acquire(?, ?, ?)", name, holder, ttlS,
	).Scan(&acquired)
	if acquired == 0 {
		return fmt.Errorf("lock %q held by another process", name)
	}
	defer db.Exec("SELECT honker_lock_release(?, ?)", name, holder)
	return fn()
}

Rate limiting

func checkRateLimit(db *sql.DB, key string, limit, windowS int) (bool, error) {
	var allowed int
	err := db.QueryRow(
		"SELECT honker_rate_limit_try(?, ?, ?)", key, limit, windowS,
	).Scan(&allowed)
	if err != nil {
		return false, fmt.Errorf("rate limit check: %w", err)
	}
	return allowed == 1, nil
}

Scheduler

func registerPeriodicTask(db *sql.DB, name, queue, cron string, payload any) error {
	data, _ := json.Marshal(payload)
	_, err := db.Exec(
		"SELECT honker_scheduler_register(?, ?, ?, ?, 0, NULL)",
		name, queue, cron, string(data),
	)
	return err
}

// Run this in a goroutine — only one leader should tick
func runScheduler(ctx context.Context, db *sql.DB, leaderID string) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		var acquired int
		db.QueryRow(
			"SELECT honker_lock_acquire('_honker_scheduler', ?, 60)", leaderID,
		).Scan(&acquired)
		if acquired == 0 {
			time.Sleep(5 * time.Second)
			continue
		}

		var raw string
		db.QueryRow(
			"SELECT honker_scheduler_tick(?)", time.Now().Unix(),
		).Scan(&raw)
		// raw is a JSON array of due tasks that were enqueued

		time.Sleep(1 * time.Second)
	}
}

Full Go worker example

func main() {
	db, err := openDB("app.db")
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Trap SIGTERM for graceful shutdown
	go func() {
		sigCh := make(chan os.Signal, 1)
		signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
		<-sigCh
		cancel()
	}()

	workerID := fmt.Sprintf("worker-%d", os.Getpid())
	log.Printf("starting worker %s", workerID)

	// Sweep expired jobs periodically
	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case <-time.After(30 * time.Second):
				db.Exec("SELECT honker_sweep_expired('emails')")
			}
		}
	}()

	// Process jobs until shutdown
	for {
		select {
		case <-ctx.Done():
			log.Println("shutting down")
			return
		default:
		}

		var raw string
		db.QueryRow(
			"SELECT honker_claim_batch('emails', ?, 1, 300)", workerID,
		).Scan(&raw)

		var jobs []struct {
			ID      int             `json:"id"`
			Payload json.RawMessage `json:"payload"`
		}
		json.Unmarshal([]byte(raw), &jobs)

		if len(jobs) == 0 {
			time.Sleep(50 * time.Millisecond)
			continue
		}

		for _, job := range jobs {
			if err := processJob(job.Payload); err != nil {
				log.Printf("job %d failed: %v", job.ID, err)
				continue
			}
			ackIDs, _ := json.Marshal([]int{job.ID})
			db.Exec("SELECT honker_ack_batch(?, ?)", string(ackIDs), workerID)
		}
	}
}

Node.js API

const { open } = require('@russellthehippo/honker-node');
const db = open('app.db');

// Transactional enqueue + notify
const tx = db.transaction();
tx.execute('INSERT INTO orders (id) VALUES (?)', [42]);
tx.notify('orders', { id: 42 });
tx.commit();

// Listen for notifications
for await (const n of db.listen('orders')) {
    handle(n.payload);
}

Raw SQL API

All functions return JSON or integers for cross-language compatibility. Use these directly from any SQLite client after loading the extension.

Queue operations

-- Claim up to 32 jobs, 300s visibility timeout
SELECT honker_claim_batch('emails', 'worker-1', 32, 300);  -- JSON array

-- Acknowledge completed jobs
SELECT honker_ack_batch('[1,2,3]', 'worker-1');             -- Count

-- Move expired jobs back to pending or to dead-letter
SELECT honker_sweep_expired('emails');                      -- Dead count

Stream operations

-- Publish an event
SELECT honker_stream_publish('orders', 'key', '{"id":42}'); -- Offset

-- Read events since offset
SELECT honker_stream_read_since('orders', 0, 1000);         -- JSON array

-- Save/get consumer offset
SELECT honker_stream_save_offset('worker', 'orders', 42);
SELECT honker_stream_get_offset('worker', 'orders');        -- Offset or 0

Notification operations

SELECT notify('orders', '{"id":42}');
SELECT honker_prune_notifications(3600, 10000);  -- older_than_s, max_keep

Scheduler (leader-elected periodic tasks)

-- Register a periodic task
SELECT honker_scheduler_register(
    'nightly',      -- name
    'backups',      -- queue
    '0 3 * * *',   -- cron expression
    '"go"',        -- payload (JSON)
    0,             -- paused (0=active)
    NULL           -- description
);

-- Tick the scheduler (call once per second from leader)
SELECT honker_scheduler_tick(unixepoch());      -- JSON: due tasks

-- Query next fire time
SELECT honker_cron_next_after('0 3 * * *', unixepoch());

-- Get soonest next fire across all tasks
SELECT honker_scheduler_soonest();

-- Unregister
SELECT honker_scheduler_unregister('nightly');

Leader election uses honker_lock_acquire('_honker_scheduler', leader_id, 60) internally — only one active scheduler leader runs honker_scheduler_tick().

Named locks

SELECT honker_lock_acquire('backup', 'me', 60);  -- 1=acquired, 0=held
SELECT honker_lock_release('backup', 'me');       -- 1=released

Rate limiting

SELECT honker_rate_limit_try('api', 10, 60);   -- 1=under limit, 0=at limit
SELECT honker_rate_limit_sweep(3600);           -- Drop old windows

Result storage

SELECT honker_result_save(42, '{"ok":true}', 3600);  -- task_id, value, TTL
SELECT honker_result_get(42);                         -- Value or NULL
SELECT honker_result_sweep();                         -- Prune expired

Schema

Honker creates these tables via honker_bootstrap():

Configuration

WAL autocheckpoint

db = honker.open("app.db", wal_autocheckpoint=10000)

Default wal_autocheckpoint=10000 batches 10k page writes per fsync, dramatically improving throughput over per-commit fsync in DELETE mode.

Queue visibility timeout

async for job in queue.claim("worker", visibility_timeout_s=600):
    ...

Default 300s. If a worker crashes, the job re-enters the queue after expiry.

Stream auto-save interval

async for event in stream.subscribe(
    consumer="worker",
    save_every_n=1000,
    save_every_s=5,
):
    ...

Default flushes every 1000 events or 5 seconds.

Retries and backoff

@queue.task(retries=5, timeout_s=30)
def work():
    ...

# Or manually on retry
job.retry(delay_s=60, backoff=2.0)  # Exponential backoff

Default max_attempts is 3. After exhaustion, the job moves to _honker_dead.

Crash Recovery

Transaction safety: Rollback drops jobs/events/notifications with the business write (SQLite ACID). SIGKILL mid-transaction is safe — WAL rollback on next open leaves no stale state.

Claim expiry: If a worker crashes with a claimed job, it expires after visibility_timeout_s and another worker reclaims. attempts increments; after max_attempts, the row moves to _honker_dead.

Stream replay: A crashed subscriber replays events from the last flushed offset (at-least-once). Tune save_every_n / save_every_s to control the replay window.

Notify durability: Ephemeral — listeners offline during publish miss the event. Use db.stream() for durable replay.

Performance

Limitations

• Single-machine only: SQLite's locking is per-host; NFS multi-writer corrupts the database
• No workflow DAGs: No task chaining, pipelines, groups, or chords
• No multi-writer replication: Use Postgres for distributed setups
• WAL required: Cannot use DELETE or TRUNCATE journal modes
• WAL sidecars required: .db-wal and .db-shm files must remain alongside the database file
• Over-triggering: All subscribers wake on any commit to the database, not just their channel — filtering happens after wake

Patterns

Background email queue with transactional outbox

import honker

db = honker.open("app.db")
emails = db.queue("emails")

# In your request handler — atomic with the order insert
def create_order(user_id, items):
    with db.transaction() as tx:
        tx.execute(
            "INSERT INTO orders (user_id, items) VALUES (?, ?)",
            [user_id, json.dumps(items)],
        )
        emails.enqueue(
            {"to": get_email(user_id), "template": "order_confirmation"},
            tx=tx,
        )

# Worker process
async def run_worker():
    async for job in emails.claim("email-worker-1"):
        try:
            send_email(**job.payload)
            job.ack()
        except Exception as e:
            job.retry(delay_s=60, error=str(e))

Event sourcing with streams

stream = db.stream("account-events")

def transfer(from_id, to_id, amount):
    with db.transaction() as tx:
        tx.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?",
                   [amount, from_id])
        tx.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?",
                   [amount, to_id])
        stream.publish({
            "type": "transfer",
            "from": from_id,
            "to": to_id,
            "amount": amount,
        }, tx=tx)

# Multiple independent consumers, each with their own offset
async def audit_consumer():
    async for event in stream.subscribe(consumer="audit-log"):
        append_to_audit_log(event)

async def analytics_consumer():
    async for event in stream.subscribe(consumer="analytics"):
        update_dashboard(event)

Combining notify + queue for fast wake

Use ephemeral notify for instant wake, queue for durable processing:

def enqueue_with_notify(queue_name, payload):
    with db.transaction() as tx:
        db.queue(queue_name).enqueue(payload, tx=tx)
        tx.notify(queue_name, {"hint": "new_job"})

This avoids polling the queue table — workers listen on the channel and only check the queue when notified.