// queue.test.js — the durable delivery queue.
//
// ⛔ THE BUG THIS EXISTS TO CLOSE. deliver.js:8-11 has said since M1: "retries are in-process and
// bounded… a delivery lost to a restart is lost silently — acceptable for one test actor, NOT
// acceptable once anything real depends on it." Restart the bridge mid-retry and the activity is
// gone, with nothing anywhere recording that it was owed.
//
// ⛔ AND THE ENTRY IS STORED UNSIGNED. An HTTP Signature covers `date` and receivers reject stale
// ones, so a signature made now is worthless to a retry an hour from now. The queue holds the
// ACTIVITY; signing happens per attempt, at send time. That is also what keeps the queue orthogonal
// to the key-custody question — it holds nothing secret and does not care where the signer lives.
import test from 'brittle'
import fs from 'bare-fs'
import { join } from 'bare-path'
import { DeliveryQueue, MemoryDeliveryQueue, createQueuedDeliverer, BACKOFF } from '../lib/queue.js'
const INBOX = 'https://remote.example/users/bob/inbox'
const TO = 'https://remote.example/users/bob'
const activity = (id = 'https://ap.7rnx.net/ap/activities/1') => ({
'@context': 'https://www.w3.org/ns/activitystreams', type: 'Create', id
})
let n = 0
function tmp () {
const dir = join('/tmp/claude-1000/-storage-projects-code-hypercore-kb/c3f731e9-2bec-4052-9d87-28d0c39fbacf/scratchpad', 'q' + (n++) + '-' + Date.now())
return join(dir, 'queue.json')
}
// ── the property the whole module exists for ─────────────────────────────────────────────────
test('★★ an undelivered activity SURVIVES a restart', async (t) => {
const path = tmp()
const first = new DeliveryQueue(path)
await first.add({ to: TO, inbox: INBOX, activity: activity() })
// A whole new instance over the same file — this is what a daemon restart looks like.
const second = new DeliveryQueue(path)
const pending = await second.list()
t.is(pending.length, 1, 'the activity is still owed after a restart')
t.is(pending[0].inbox, INBOX)
t.alike(pending[0].activity, activity(), 'and the activity itself round-tripped intact')
})
test('★★ the queue stores the activity UNSIGNED — signatures expire, so they cannot be stored', async (t) => {
const path = tmp()
const q = new DeliveryQueue(path)
await q.add({ to: TO, inbox: INBOX, activity: activity() })
const raw = fs.readFileSync(path, 'utf8')
t.absent(/signature/i.test(raw), 'no signature was persisted')
t.absent(/keyId/i.test(raw), 'and no keyId — the queue holds nothing that identifies a key')
})
// ── retry policy ──────────────────────────────────────────────────────────────────────────────
test('★ a transient failure is retried later, with growing backoff', async (t) => {
const q = new MemoryDeliveryQueue()
const { id } = await q.add({ to: TO, inbox: INBOX, activity: activity() })
const now = 1_000_000
await q.retry(id, { error: 'connect ETIMEDOUT', now })
const [e1] = await q.list()
t.is(e1.attempts, 1)
t.is(e1.nextAt, now + BACKOFF[0], 'first retry uses the first backoff step')
t.is(e1.lastError, 'connect ETIMEDOUT', 'and the reason is recorded, not discarded')
await q.retry(id, { error: 'again', now })
const [e2] = await q.list()
t.is(e2.nextAt, now + BACKOFF[1], 'the second wait is longer')
t.ok(BACKOFF[1] > BACKOFF[0])
})
test('★ nothing is delivered before it is due', async (t) => {
const q = new MemoryDeliveryQueue()
const { id } = await q.add({ to: TO, inbox: INBOX, activity: activity() })
t.is((await q.due(0)).length, 1, 'a fresh entry is due immediately')
await q.retry(id, { error: 'x', now: 0 })
t.is((await q.due(0)).length, 0, 'after a failure it waits')
t.is((await q.due(BACKOFF[0])).length, 1, 'and becomes due again when the backoff elapses')
})
// ⛔ "Lost silently" is the exact failure being fixed. A give-up must leave EVIDENCE.
test('⛔ giving up buries the entry with its reason — it is never silently dropped', async (t) => {
const q = new MemoryDeliveryQueue()
const { id } = await q.add({ to: TO, inbox: INBOX, activity: activity() })
let dead = false
for (let i = 0; i <= BACKOFF.length; i++) dead = (await q.retry(id, { error: 'gone', now: i })).dead
t.ok(dead, 'the schedule was exhausted')
t.is((await q.list()).length, 0, 'no longer pending')
const buried = await q.dead()
t.is(buried.length, 1, 'but kept where a human can see it')
t.is(buried[0].lastError, 'gone')
})
test('★ a permanent rejection is buried immediately, never retried', async (t) => {
const q = new MemoryDeliveryQueue()
const { id } = await q.add({ to: TO, inbox: INBOX, activity: activity() })
await q.bury(id, 'HTTP 401')
t.is((await q.list()).length, 0)
t.is((await q.due(Date.now())).length, 0, 'a 4xx is not retried — it will never succeed')
t.is((await q.dead())[0].lastError, 'HTTP 401')
})
test('★ the same activity to the same inbox enqueues once', async (t) => {
const q = new MemoryDeliveryQueue()
const a = await q.add({ to: TO, inbox: INBOX, activity: activity() })
const b = await q.add({ to: TO, inbox: INBOX, activity: activity() })
t.ok(a.added)
t.absent(b.added, 'a duplicate is refused')
t.is(a.id, b.id)
t.is((await q.list()).length, 1)
})
test('a corrupt queue file does not stop the daemon booting', async (t) => {
const path = tmp()
fs.mkdirSync(join(path, '..'), { recursive: true })
fs.writeFileSync(path, '{ this is not json')
const q = new DeliveryQueue(path)
t.is((await q.list()).length, 0, 'starts empty rather than throwing')
await q.add({ to: TO, inbox: INBOX, activity: activity() })
t.is((await q.list()).length, 1, 'and is usable afterwards')
})
// ── the drain, which is what makes it a queue rather than a list ──────────────────────────────
test('★★ a queued delivery is retried across a RESTART, not just in-process', async (t) => {
const path = tmp()
const attempts = []
// First run: the remote is down. The entry survives the process ending.
{
const queue = new DeliveryQueue(path)
const qd = createQueuedDeliverer({
queue,
deliver: async () => { attempts.push('down'); throw new Error('ECONNREFUSED') },
log: () => {},
now: () => 0
})
await qd.deliver({ to: TO, inbox: INBOX, activity: activity() })
await qd.drain()
t.is((await queue.list()).length, 1, 'still owed when the process ends')
}
// Second run: same file, remote is back.
{
const queue = new DeliveryQueue(path)
const qd = createQueuedDeliverer({
queue,
deliver: async () => { attempts.push('up'); return { ok: true, status: 202 } },
log: () => {},
now: () => BACKOFF[0]
})
await qd.drain()
t.is((await queue.list()).length, 0, 'delivered after the restart')
}
t.alike(attempts, ['down', 'up'], 'one attempt per run')
})
test('★ drain buries a permanent rejection instead of retrying it forever', async (t) => {
const queue = new MemoryDeliveryQueue()
let calls = 0
const qd = createQueuedDeliverer({
queue,
deliver: async () => { calls++; return { ok: false, status: 401, permanent: true } },
log: () => {},
now: () => 0
})
await qd.deliver({ to: TO, inbox: INBOX, activity: activity() })
await qd.drain()
await qd.drain()
t.is(calls, 1, 'tried once and never again')
t.is((await queue.dead()).length, 1)
})
test('★ enqueue returns immediately — Publisher fires and forgets', async (t) => {
const queue = new MemoryDeliveryQueue()
let started = false
const qd = createQueuedDeliverer({
queue,
deliver: async () => { started = true; return { ok: true, status: 202 } },
log: () => {},
now: () => 0
})
await qd.deliver({ to: TO, inbox: INBOX, activity: activity() })
t.absent(started, 'accepting the activity does not block on the network')
t.is((await queue.list()).length, 1, 'it is durable before anything is attempted')
})
// ⛔⛔ THE QUEUE ERASED WHO WAS SIGNING, AND EVERY PER-USER DELIVERY WENT OUT AS THE PRIMARY.
//
// createDeliverer grew a per-delivery `signer` in 0.9.52 so one bridge could front many people.
// This wrapper kept "createDeliverer's signature" as it was written for a single actor and called
// `deliver({ to, inbox, activity })` — dropping it. A signer is a live object and cannot be written
// to a JSON file, so nothing about the durable path could have carried it as-is.
//
// The result was invisible from our side and fatal on theirs: every activity for a registered user
// was signed with the PRIMARY's key against the primary's key id, while the activity's `actor` was
// the registered user. Mastodon verifies the signature (it is a real key), resolves the signing
// account to the primary, sees an actor that does not match, and DROPS the activity — after having
// already returned 202. Measured on the wire: actor .../users/ashatron/actor signed with
// keyId https://ap.7rnx.net/ap/actor#main-key. 17 activities accepted, none stored, no refetch.
//
// ⛔ AND AN UNRESOLVABLE SIGNER MUST NOT FALL BACK. Falling back to the process signer is exactly
// the bug; signing as somebody else is worse than not sending.
test('⛔⛔ a queued delivery is signed by the identity that queued it', async (t) => {
const seen = []
const q = new MemoryDeliveryQueue()
const signers = { ash: { keyId: 'https://h/ap/users/ash/actor#main-key' } }
const qd = createQueuedDeliverer({
queue: q,
deliver: async (d) => { seen.push(d); return { ok: true } },
resolveSigner: (id) => Object.values(signers).find((s) => s.keyId === id) || null,
setTimer: () => null,
clearTimer: () => {},
log: () => {}
})
const deliver = qd.deliver
await deliver({
to: 'https://remote/users/bob',
inbox: 'https://remote/inbox',
activity: { id: 'a1', type: 'Create', actor: 'https://h/ap/users/ash/actor' },
signer: signers.ash
})
await qd.drain()
t.is(seen.length, 1)
t.ok(seen[0].signer, 'a signer reached the wire')
t.is(seen[0].signer.keyId, signers.ash.keyId, 'and it is THEIRS, not the process default')
})
test('⛔ a queued delivery whose signer cannot be resolved is NOT sent as somebody else', async (t) => {
const seen = []
const q = new MemoryDeliveryQueue()
const qd = createQueuedDeliverer({
queue: q,
deliver: async (d) => { seen.push(d); return { ok: true } },
resolveSigner: () => null, // the user was released, or their key is gone
setTimer: () => null,
clearTimer: () => {},
log: () => {}
})
const deliver = qd.deliver
await deliver({
to: 'https://remote/users/bob',
inbox: 'https://remote/inbox',
activity: { id: 'a2', type: 'Create', actor: 'https://h/ap/users/gone/actor' },
signer: { keyId: 'https://h/ap/users/gone/actor#main-key' }
})
await qd.drain()
t.is(seen.length, 0, 'nothing went out under the wrong identity')
t.is(await q.size(), 1, 'and it is still queued rather than silently dropped')
})
test('★ a delivery with no signer still uses the process default, exactly as before', async (t) => {
const seen = []
const qd = createQueuedDeliverer({
queue: new MemoryDeliveryQueue(),
deliver: async (d) => { seen.push(d); return { ok: true } },
setTimer: () => null,
clearTimer: () => {},
log: () => {}
})
await qd.deliver({ to: 'https://remote/users/bob', inbox: 'https://remote/inbox', activity: { id: 'a3', type: 'Create' } })
await qd.drain()
t.is(seen.length, 1)
t.absent(seen[0].signer, 'the primary keeps the deliverer it always had')
})
// ⛔ THE TIMER IS FOR RETRY, NOT FOR FIRST DELIVERY. `deliver()` enqueued and returned, and the only
// things that ever drained were boot and a 60s interval — so every activity waited up to a minute
// before its first attempt, averaging thirty seconds. Measured live on 2026-09-09: published at
// 22:33:54, on the wire at 22:34:25.
//
// The single-actor bridge called deliver() directly and was instant; the queue was added for
// durability and silently bought a minute of latency with it. Durability and latency are not a
// trade — the entry is written before the network is touched either way, and the drain is kicked
// straight after, fire-and-forget so the caller still never blocks on the network.
test('★ enqueueing kicks a drain rather than waiting for the timer', async (t) => {
const sent = []
const q = new MemoryDeliveryQueue()
const d = createQueuedDeliverer({
queue: q,
deliver: async (x) => { sent.push(x); return { ok: true } },
log: () => {}
})
await d.deliver({ to: 'https://prsm.space/users/fleeky', activity: { type: 'Create', id: 'a1' } })
// Let the fire-and-forget drain settle.
await new Promise((r) => setImmediate(r))
await new Promise((r) => setImmediate(r))
t.is(sent.length, 1, '⛔ delivered on the enqueue event — there is no timer to fall back on')
t.is(await d.pending(), 0, 'and the entry is no longer owed')
})