// queue.js — the durable outbound delivery queue.
//
// ⛔ WHY THIS EXISTS. deliver.js's retries are in-process: three attempts over ~21 seconds, then the
// activity is gone. Its own header has said since M1 that "a delivery lost to a restart is lost
// silently — acceptable for one test actor, NOT acceptable once anything real depends on it."
// DESIGN.md §5 puts this state here: "Push state — queue, retries, follower list — lives on
// ap-bridge. The bridge holds data; the VPS does not."
//
// ⛔ WHAT IS STORED IS UNSIGNED. An HTTP Signature covers `date`, and receivers reject stale dates,
// so a signature made now is worthless to a retry an hour from now — each attempt must sign afresh.
// The queue therefore holds the ACTIVITY and the destination, nothing else. That is also what keeps
// it orthogonal to the key-custody question: the queue holds nothing secret and does not care
// whether the signer is local, on a device, or across a socket.
//
// ★ THE OFFLINE CASE IS LATENCY, NOT LOSS. Signing happens when you send, and you only send when
// you act — so being offline mostly means there is nothing to sign. What is genuinely owed while
// you sleep is an `Accept` for a follow that arrived, and retries of a delivery that began before
// you went away. Both are late rather than lost, and this queue is what makes that true.
//
// ⚠ SCOPE: a JSON file, matching followers.js and inbound-store.js. followers.js notes that a
// delivery queue "needs secondary indexes by host/status/next-retry, which a flat file cannot
// give" — true at scale, and not true at one actor with tens of pending entries, which is what a
// full scan handles fine. The INTERFACE is the part that matters: it is async throughout, so
// moving the backing store later changes nothing above it. Per-host rate limiting
// (`bucket-rate-limit`, DESIGN.md §5) is NOT here yet and is the next thing this needs.
import fs from 'bare-fs'
import { dirname } from 'bare-path'
// 1m, 5m, 15m, 1h, 3h, 6h, 12h, 24h — eight attempts spanning about two days, then bury.
// Fediverse servers go down for maintenance windows measured in hours; a schedule that gives up in
// minutes would drop deliveries that were always going to succeed.
export const BACKOFF = [60_000, 300_000, 900_000, 3_600_000, 10_800_000, 21_600_000, 43_200_000, 86_400_000]
// One activity to one inbox. The pair is the identity: the same Create to two inboxes is two
// deliveries, and the same Create to one inbox twice is a duplicate in that person's timeline.
const keyFor = (inbox, activity) => `${activity && activity.id ? activity.id : '?'}|${inbox}`
class BaseQueue {
async add ({ to, inbox, activity, signerId = null }) {
const state = await this._read()
const id = keyFor(inbox, activity)
if (state.pending[id] || state.dead[id]) return { id, added: false }
state.pending[id] = {
id,
to,
inbox,
activity,
// ⛔ WHICH IDENTITY SIGNS THIS, by its PUBLIC key id, resolved to a live signer at send time.
// A signer is a live object and cannot survive a restart in a JSON file, so carrying it by
// reference was never possible — but carrying NOTHING meant every queued activity fell back to
// the PROCESS signer. Every post and profile Update for a registered user went out signed with
// the primary's key, against the primary's key id, while the activity's `actor` was the
// registered user. Mastodon verifies that signature happily — it is a real key — resolves the
// signing account to the primary, sees an actor that does not match, and DROPS the activity
// after already answering 202. Seventeen accepted, none stored, no error anywhere. Measured on
// the wire: actor .../users/ashatron/actor signed with keyId https://host/ap/actor#main-key.
//
// A keyId is public — it is in every actor document — so this keeps the queue's "holds nothing
// secret" property while making the durable path know whose delivery it carries.
signerId,
attempts: 0,
nextAt: 0, // due immediately
createdAt: Date.now(),
lastError: null
}
await this._write(state)
return { id, added: true }
}
async list () { return Object.values((await this._read()).pending) }
async dead () { return Object.values((await this._read()).dead) }
async size () { return (await this.list()).length }
/** Entries whose backoff has elapsed, oldest first so a backlog drains in the order it accrued. */
async due (now) {
return (await this.list())
.filter((e) => e.nextAt <= now)
.sort((a, b) => a.createdAt - b.createdAt)
}
async succeed (id) {
const state = await this._read()
if (!state.pending[id]) return false
delete state.pending[id]
await this._write(state)
return true
}
/**
* A transient failure. Schedules the next attempt, or buries the entry once the schedule is
* exhausted. ⛔ Returns { dead } rather than throwing, because a caller that forgets to check
* would otherwise turn "we gave up" back into the silent drop this module exists to prevent.
*/
async retry (id, { error, now = Date.now() } = {}) {
const state = await this._read()
const e = state.pending[id]
if (!e) return { dead: false, missing: true }
e.attempts++
e.lastError = error == null ? null : String(error)
if (e.attempts > BACKOFF.length) {
delete state.pending[id]
state.dead[id] = { ...e, buriedAt: now, buriedBecause: 'gave up after ' + (e.attempts - 1) + ' attempts' }
await this._write(state)
return { dead: true }
}
e.nextAt = now + BACKOFF[e.attempts - 1]
await this._write(state)
return { dead: false }
}
/** A permanent rejection — a 4xx that is not 429. Retrying annoys the receiver and never works. */
async bury (id, reason, { now = Date.now() } = {}) {
const state = await this._read()
const e = state.pending[id]
if (!e) return false
delete state.pending[id]
state.dead[id] = { ...e, lastError: reason == null ? null : String(reason), buriedAt: now, buriedBecause: 'rejected permanently' }
await this._write(state)
return true
}
}
export class DeliveryQueue extends BaseQueue {
constructor (path) {
super()
this._path = path
this._state = null
}
async _read () {
if (this._state) return this._state
try {
const raw = JSON.parse(fs.readFileSync(this._path, 'utf8'))
this._state = { pending: raw.pending || {}, dead: raw.dead || {} }
} catch {
// Absent or corrupt: start empty rather than refusing to boot. Same call as followers.js —
// a daemon that will not start is a worse outage than a queue that lost its backlog, and the
// file is rewritten atomically so corruption should not happen in the first place.
this._state = { pending: {}, dead: {} }
}
return this._state
}
async _write (state) {
this._state = state
fs.mkdirSync(dirname(this._path), { recursive: true })
// Write-then-rename, like followers.js: a crash mid-write must not truncate the backlog to
// nothing, which would be the very failure this file exists to stop.
const tmp = this._path + '.tmp'
fs.writeFileSync(tmp, JSON.stringify(state, null, 2))
fs.renameSync(tmp, this._path)
}
}
/** Same interface, no disk. For tests. */
export class MemoryDeliveryQueue extends BaseQueue {
constructor () { super(); this._state = { pending: {}, dead: {} } }
async _read () { return this._state }
async _write (state) { this._state = state }
}
/**
* Wraps a deliverer so activities are made durable BEFORE anything is attempted.
*
* `deliver({ to, inbox, activity })` keeps createDeliverer's signature, so Publisher — which calls
* it fire-and-forget — needs no change at all.
*
* ⚠ Two retry layers on purpose. The inner deliverer retries fast (~21s) for a connection blip;
* this one retries slow (hours) for a server that is actually down. They answer different
* questions and collapsing them would lose one of the answers.
*/
export function createQueuedDeliverer ({ queue, deliver, resolveSigner = null, log = () => {}, now = Date.now }) {
if (!queue) throw new Error('createQueuedDeliverer: a queue is required')
if (typeof deliver !== 'function') throw new Error('createQueuedDeliverer: a deliver function is required')
let inflight = null
let again = false
async function attempt (e) {
try {
// ⛔ AN UNRESOLVABLE SIGNER IS NOT A REASON TO USE ANOTHER. Falling through to the process
// signer is precisely the defect above: it signs one person's activity with another's key.
// The entry stays queued — a released user, or a key not loaded yet, is a state to wait in,
// not one to send from somebody else's identity.
let signer = null
if (e.signerId) {
signer = resolveSigner ? resolveSigner(e.signerId) : null
if (!signer) {
log({ msg: 'queued delivery HELD — cannot resolve its signer', to: e.to, signerId: e.signerId })
return
}
}
const res = await deliver({ to: e.to, inbox: e.inbox, activity: e.activity, signer })
if (res && res.ok) {
await queue.succeed(e.id)
log({ msg: 'queued delivery sent', to: e.to, inbox: e.inbox, attempts: e.attempts })
return
}
// deliver() returns rather than throws for a permanent 4xx — that classification is its job.
if (res && res.permanent) {
await queue.bury(e.id, 'HTTP ' + res.status, { now: now() })
log({ msg: 'queued delivery REJECTED permanently', to: e.to, inbox: e.inbox, status: res.status })
return
}
const { dead } = await queue.retry(e.id, { error: 'HTTP ' + (res && res.status), now: now() })
if (dead) log({ msg: 'queued delivery GAVE UP', to: e.to, inbox: e.inbox, status: res && res.status })
} catch (err) {
// The inner deliverer throws once its own fast retries are exhausted — transient.
const { dead } = await queue.retry(e.id, { error: err.message, now: now() })
if (dead) log({ msg: 'queued delivery GAVE UP', to: e.to, inbox: e.inbox, error: err.message })
}
}
// ⛔ A KICK THAT ARRIVES MID-DRAIN IS NOT DISCARDED. `if (draining) return` dropped it, which is
// safe only when a timer is coming round again to pick up what was missed — i.e. it silently
// depended on the sweep this file no longer has. Without that, an activity enqueued while a drain
// was in flight would wait for the NEXT unrelated event. Re-running once at the end costs one
// `due()` on an empty queue and makes the event sufficient on its own.
//
// ⛔ AND IT RETURNS THE IN-FLIGHT PROMISE rather than undefined, so `await drain()` still means
// "a drain covering my call has finished". Returning early broke exactly that: a caller awaited,
// got a resolved promise while the real pass was still running, and asserted against state that
// had not been written yet.
function drain () {
if (inflight) { again = true; return inflight }
inflight = (async () => {
try {
do {
again = false
for (const e of await queue.due(now())) await attempt(e)
} while (again)
} finally {
inflight = null
}
})()
return inflight
}
return {
/** Enqueue. Durable before the network is touched, and does not block the caller on it. */
async deliver ({ to, inbox = null, activity, signer = null }) {
const { id, added } = await queue.add({ to, inbox, activity, signerId: (signer && signer.keyId) || null })
if (!added) {
log({ msg: 'delivery already queued, not duplicating', to, id })
return { queued: false, id }
}
log({ msg: 'delivery queued', to, inbox, type: activity && activity.type, id })
// ★ THE ENQUEUE IS THE EVENT. Delivery used to happen only on a 60s sweep, so everything we
// already held in hand waited an average of thirty seconds — measured live 2026-09-09:
// published 22:33:54, on the wire 22:34:25. The single-actor bridge called deliver() directly
// and was instant; the queue bought durability and quietly charged a minute of latency for it.
// Those are not a trade: the entry is durable BEFORE the network is touched either way.
//
// Fire-and-forget, so the caller still never blocks on the network — the property the comment
// above this method has always claimed.
drain().catch((err) => log({ msg: 'drain failed', error: err.message }))
return { queued: true, id }
},
drain,
/** How much is owed. Logged at boot, because a backlog nobody can see is the old bug wearing a hat. */
async pending () { return queue.size() },
/**
* Boot reconciliation. NOT a timer.
*
* ⛔ THE SWEEP CONCEALED A DEAD HOOK, which is the same defect the mirror trigger shipped three
* times (hyper-social #66 §3): an unconditional periodic pass makes delivery appear to work and
* hides the fact that nothing fires on the event. Delivery was ONLY ever the timer here, and it
* took a latency measurement rather than a failure to notice.
*
* So the shape is event-driven plus reconciliation: the enqueue kicks a drain, this reconciles a
* restart's backlog, and `drain()` is exposed for an operator verb — the same posture
* `reproject` already has for notes. Nothing polls.
*
* ⚠ What this gives up, stated plainly: an entry whose retry backoff elapses while the process
* is otherwise idle waits for the next enqueue or an explicit reconcile rather than for a tick.
* That is the honest cost of not polling, and it is visible in `pending()` rather than silent.
*/
start () {
drain().catch((err) => log({ msg: 'drain failed', error: err.message }))
},
/** Nothing to stop any more — kept so callers that shut down cleanly do not need to change. */
stop () {}
}
}