| … | … | @@ -173,12 +173,12 @@ export class MemoryDeliveryQueue extends BaseQueue { |
| 173 | 173 | * this one retries slow (hours) for a server that is actually down. They answer different |
| 174 | 174 | * questions and collapsing them would lose one of the answers. |
| 175 | 175 | */ |
| 176 | | export function createQueuedDeliverer ({ queue, deliver, resolveSigner = null, log = () => {}, now = Date.now, intervalMs = 60_000, setTimer = setInterval, clearTimer = clearInterval }) { |
| 176 | export function createQueuedDeliverer ({ queue, deliver, resolveSigner = null, log = () => {}, now = Date.now }) { |
| 177 | 177 | if (!queue) throw new Error('createQueuedDeliverer: a queue is required') |
| 178 | 178 | if (typeof deliver !== 'function') throw new Error('createQueuedDeliverer: a deliver function is required') |
| 179 | 179 | |
| 180 | | let timer = null |
| 181 | | let draining = false |
| 180 | let inflight = null |
| 181 | let again = false |
| 182 | 182 | |
| 183 | 183 | async function attempt (e) { |
| 184 | 184 | try { |
| … | … | @@ -215,14 +215,29 @@ export function createQueuedDeliverer ({ queue, deliver, resolveSigner = null, l |
| 215 | 215 | } |
| 216 | 216 | } |
| 217 | 217 | |
| 218 | | async function drain () { |
| 219 | | if (draining) return |
| 220 | | draining = true |
| 221 | | try { |
| 222 | | for (const e of await queue.due(now())) await attempt(e) |
| 223 | | } finally { |
| 224 | | draining = false |
| 225 | | } |
| 218 | // ⛔ A KICK THAT ARRIVES MID-DRAIN IS NOT DISCARDED. `if (draining) return` dropped it, which is |
| 219 | // safe only when a timer is coming round again to pick up what was missed — i.e. it silently |
| 220 | // depended on the sweep this file no longer has. Without that, an activity enqueued while a drain |
| 221 | // was in flight would wait for the NEXT unrelated event. Re-running once at the end costs one |
| 222 | // `due()` on an empty queue and makes the event sufficient on its own. |
| 223 | // |
| 224 | // ⛔ AND IT RETURNS THE IN-FLIGHT PROMISE rather than undefined, so `await drain()` still means |
| 225 | // "a drain covering my call has finished". Returning early broke exactly that: a caller awaited, |
| 226 | // got a resolved promise while the real pass was still running, and asserted against state that |
| 227 | // had not been written yet. |
| 228 | function drain () { |
| 229 | if (inflight) { again = true; return inflight } |
| 230 | inflight = (async () => { |
| 231 | try { |
| 232 | do { |
| 233 | again = false |
| 234 | for (const e of await queue.due(now())) await attempt(e) |
| 235 | } while (again) |
| 236 | } finally { |
| 237 | inflight = null |
| 238 | } |
| 239 | })() |
| 240 | return inflight |
| 226 | 241 | } |
| 227 | 242 | |
| 228 | 243 | return { |
| … | … | @@ -234,6 +249,15 @@ export function createQueuedDeliverer ({ queue, deliver, resolveSigner = null, l |
| 234 | 249 | return { queued: false, id } |
| 235 | 250 | } |
| 236 | 251 | log({ msg: 'delivery queued', to, inbox, type: activity && activity.type, id }) |
| 252 | // ★ THE ENQUEUE IS THE EVENT. Delivery used to happen only on a 60s sweep, so everything we |
| 253 | // already held in hand waited an average of thirty seconds — measured live 2026-09-09: |
| 254 | // published 22:33:54, on the wire 22:34:25. The single-actor bridge called deliver() directly |
| 255 | // and was instant; the queue bought durability and quietly charged a minute of latency for it. |
| 256 | // Those are not a trade: the entry is durable BEFORE the network is touched either way. |
| 257 | // |
| 258 | // Fire-and-forget, so the caller still never blocks on the network — the property the comment |
| 259 | // above this method has always claimed. |
| 260 | drain().catch((err) => log({ msg: 'drain failed', error: err.message })) |
| 237 | 261 | return { queued: true, id } |
| 238 | 262 | }, |
| 239 | 263 | |
| … | … | @@ -242,14 +266,27 @@ export function createQueuedDeliverer ({ queue, deliver, resolveSigner = null, l |
| 242 | 266 | /** How much is owed. Logged at boot, because a backlog nobody can see is the old bug wearing a hat. */ |
| 243 | 267 | async pending () { return queue.size() }, |
| 244 | 268 | |
| 245 | | /** Start draining on a timer. Call once at boot; the first pass picks up a restart's backlog. */ |
| 269 | /** |
| 270 | * Boot reconciliation. NOT a timer. |
| 271 | * |
| 272 | * ⛔ THE SWEEP CONCEALED A DEAD HOOK, which is the same defect the mirror trigger shipped three |
| 273 | * times (hyper-social #66 §3): an unconditional periodic pass makes delivery appear to work and |
| 274 | * hides the fact that nothing fires on the event. Delivery was ONLY ever the timer here, and it |
| 275 | * took a latency measurement rather than a failure to notice. |
| 276 | * |
| 277 | * So the shape is event-driven plus reconciliation: the enqueue kicks a drain, this reconciles a |
| 278 | * restart's backlog, and `drain()` is exposed for an operator verb — the same posture |
| 279 | * `reproject` already has for notes. Nothing polls. |
| 280 | * |
| 281 | * ⚠ What this gives up, stated plainly: an entry whose retry backoff elapses while the process |
| 282 | * is otherwise idle waits for the next enqueue or an explicit reconcile rather than for a tick. |
| 283 | * That is the honest cost of not polling, and it is visible in `pending()` rather than silent. |
| 284 | */ |
| 246 | 285 | start () { |
| 247 | | if (timer) return |
| 248 | 286 | drain().catch((err) => log({ msg: 'drain failed', error: err.message })) |
| 249 | | timer = setTimer(() => { drain().catch((err) => log({ msg: 'drain failed', error: err.message })) }, intervalMs) |
| 250 | | if (timer && timer.unref) timer.unref() |
| 251 | 287 | }, |
| 252 | 288 | |
| 253 | | stop () { if (timer) { clearTimer(timer); timer = null } } |
| 289 | /** Nothing to stop any more — kept so callers that shut down cleanly do not need to change. */ |
| 290 | stop () {} |
| 254 | 291 | } |
| 255 | 292 | } |