Changeset bfa3ba3 in Klonkt for src/services

Timestamp:

10/26/2026 01:35:31 PM (27 hours ago)

Author:

Robin <roboburr@…>

Branches:

main

Children:

7ccdced

Parents:

31288dc

Message:

Ook de directe verzenders langs de volgorde per inbox

Rest van shaer-xzab.3. Drie plekken verstuurden zelf met deliver() en zetten
pas bij een mislukking in de rij: een antwoord (deliverReply), het wissen
daarvan (deliverOutboxDelete) en een direct bericht (deliverDirectNote, ook de
hulpvraag en de zwaai). Zo kon de Delete van een antwoord aankomen vóór het
antwoord zelf als dat nog in de rij wachtte, en stond het antwoord daarna weer
op die server.

Nu één weg: deliverNowOrQueue in ap-transport, met dezelfde regel als
deliverWithRetry (wacht er voor die inbox al iets, of zit de server in zijn
backoff, dan achteraan), en hij zegt of het 'sent' of 'queued' werd, want
deliverReply en deliverDirectNote tellen wat er meteen aankwam.
deliverWithRetry is er nu een dunne laag over. guardianship/delivery.js krijgt
deliverNowOrQueue via wireDelivery in plaats van deliver en enqueueDelivery.

test/direct-sends-queue.test.js: vijf toetsen tegen een echte HTTP-server,
waaronder het hele scenario: antwoord lukt niet en wacht, de server is terug,
het antwoord wordt gewist -- de Delete wacht achter de Create en ze komen in
die volgorde aan. Vier mutaties apart gedraaid (elke plek terug naar zelf
versturen, en 'sent' dat niet telt), elk gevangen. Volledige suite 1376 groen.

Co-Authored-By: Claude Opus 5.5 <noreply@…>

Location:

src/services

Files:

• ActivityPubService.js (modified) (4 diffs)
• ap-transport.js (modified) (2 diffs)
• guardianship/delivery.js (modified) (2 diffs)

src/services/ActivityPubService.js

@@ -34 +34 @@
 import {
   safeFetch, getOrCreateKeys, deliver, fetchActor,
-  enqueueDelivery, deliverWithRetry, processDeliveryQueue, startDeliveryWorker,
+  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, processDeliveryQueue, startDeliveryWorker,
   anySigningSlug, verifyRequest, signedGetHeaders, signedGetJson, apGetJson,
 } from './ap-transport.js';
 export {
   safeFetch, getOrCreateKeys, deliver, fetchActor,
-  enqueueDelivery, deliverWithRetry, processDeliveryQueue, startDeliveryWorker,
+  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, processDeliveryQueue, startDeliveryWorker,
   verifyRequest, signedGetHeaders, signedGetJson,
 };
@@ -2416 +2416 @@
   let delivered = 0;
   for (const inbox of [...inboxes].filter(Boolean)) {
-    let ok = false;
-    try { const st = await deliver(inbox, create, keyId, keys.private_pem); ok = st >= 200 && st < 300; } catch { ok = false; }
-    if (ok) delivered++;
-    else enqueueDelivery(site.slug, inbox, create); // durable: retry a briefly-offline recipient (was silently dropped)
+    // Langs de volgorde per inbox (shaer-xzab.3); wat niet meteen kan, gaat
+    // in de rij en wordt later opnieuw geprobeerd.
+    if ((await deliverNowOrQueue(site.slug, inbox, create, keyId, keys.private_pem)) === 'sent') delivered++;
   }
   console.log('[AP] outreply', site.slug, '→', parent.actor_uri, 'delivered', delivered);
@@ -2789 +2788 @@
     if (row.to_actor) { const a = await fetchActor(row.to_actor).catch(() => null); if (a) inboxes.add((a.endpoints && a.endpoints.sharedInbox) || a.inbox); }
     for (const f of fStmts().list.all(site.slug)) inboxes.add(f.shared_inbox || f.inbox);
-    for (const inbox of [...inboxes].filter(Boolean)) {
-      try { const st = await deliver(inbox, del, `${me}#main-key`, keys.private_pem); if (st >= 200 && st < 300) continue; } catch { /* queue below */ }
-      enqueueDelivery(site.slug, inbox, del); // durable: a failed comment-delete now retries (was silently dropped)
-    }
+    // Langs de volgorde per inbox (shaer-xzab.3): deze Delete mag niet vóór
+    // de Create van dezelfde reactie aankomen als die nog in de rij wacht --
+    // dan stond de reactie daarna weer op die server.
+    for (const inbox of [...inboxes].filter(Boolean)) await deliverNowOrQueue(site.slug, inbox, del, `${me}#main-key`, keys.private_pem);
   }
   db.prepare('DELETE FROM ap_outbox WHERE id = ?').run(outboxId);
@@ -4164 +4163 @@
   actorId, fetchActor, localActor, deliverTo: deliverToActor, deriveHandle, escHtml, linkUrls, linkHashtags,
   getOutboxRow: (id) => iStmts().getO.get(id),
-  buildReplyNote, AP_CONTEXT, getOrCreateKeys, deliver, enqueueDelivery,
+  buildReplyNote, AP_CONTEXT, getOrCreateKeys, deliverNowOrQueue,
   // Rijke directe berichten: dezelfde sanitizer als deliverReply gebruikt, zodat
   // een antwoord uit Berichten door precies één poort gaat.

src/services/ap-transport.js

@@ -278 +278 @@
   } catch { /* health tracking is non-fatal */ }
 }
-// Deliver now; queue for retry if it fails.
+// Deliver now, or queue: zegt welke van de twee het werd ('sent' | 'queued'),
+// voor de aanroepers die tellen wat er meteen aankwam.
 //
 // Niet meteen als er voor deze inbox al iets in de wachtrij staat (dan zou dit
@@ -284 +285 @@
 // server net faalde en in zijn backoff zit (dan kost het alleen acht seconden
 // time-out om hetzelfde te horen). Dan achteraan in de rij.
+//
+// DE ENIGE WEG om iets direct te versturen dat bij een mislukking in de rij
+// hoort (shaer-xzab.3). Drie plekken deden eerst zelf deliver() en pas bij een
+// mislukking enqueueDelivery(), en gingen zo langs de volgorde per inbox.
+export async function deliverNowOrQueue(slug, inbox, activity, keyId, privPem) {
+  if (!inbox) return 'queued';
+  try {
+    const st = deliveryStmts();
+    if (st.pendInbox.get(inbox) || st.hostWaits.get(inboxHost(inbox))) { enqueueDelivery(slug, inbox, activity); return 'queued'; }
+  } catch { /* geen wachtrij te lezen: gewoon proberen */ }
+  try { const st = await deliver(inbox, activity, keyId, privPem); if (st >= 200 && st < 300) { markFollowerDelivery(slug, inbox, true); return 'sent'; } } catch { /* queue below */ }
+  enqueueDelivery(slug, inbox, activity);
+  return 'queued';
+}
+// Deliver now; queue for retry if it fails.
 export async function deliverWithRetry(slug, inbox, activity, keyId, privPem) {
-  if (!inbox) return;
-  try {
-    const st = deliveryStmts();
-    if (st.pendInbox.get(inbox) || st.hostWaits.get(inboxHost(inbox))) { enqueueDelivery(slug, inbox, activity); return; }
-  } catch { /* geen wachtrij te lezen: gewoon proberen */ }
-  try { const st = await deliver(inbox, activity, keyId, privPem); if (st >= 200 && st < 300) { markFollowerDelivery(slug, inbox, true); return; } } catch { /* queue below */ }
-  enqueueDelivery(slug, inbox, activity);
+  await deliverNowOrQueue(slug, inbox, activity, keyId, privPem);
 }
 let _processingDeliv = false;

src/services/guardianship/delivery.js

@@ -93 +93 @@
   if (!inReplyTo && helpMark && helpMark.noteUri) inReplyTo = helpMark.noteUri;
   const { actorId, fetchActor, localActor, deliverTo, deriveHandle, escHtml, linkUrls, linkHashtags,
-          getOutboxRow, buildReplyNote, AP_CONTEXT, getOrCreateKeys, deliver, enqueueDelivery } = deps;
+          getOutboxRow, buildReplyNote, AP_CONTEXT, getOrCreateKeys, deliverNowOrQueue } = deps;
   const base = (process.env.PUBLIC_BASE_URL || '').replace(/\/+$/, '');
   const list = [...new Set((recipients || []).filter((u) => /^https?:\/\//i.test(String(u || ''))))].slice(0, 8);
@@ -207 +207 @@
   }
   // Remote: one POST per inbox, so two guardians on the same server share it.
+  // Langs de volgorde per inbox (shaer-xzab.3), zoals elke bezorging.
   for (const inbox of [...new Set(resolved.filter((x) => !x.local).map((r) => r.inbox))]) {
-    let ok = false;
-    try { const st = await deliver(inbox, create, keyId, keys.private_pem); ok = st >= 200 && st < 300; } catch { ok = false; }
-    if (ok) delivered++;
-    else enqueueDelivery(site.slug, inbox, create);
+    if ((await deliverNowOrQueue(site.slug, inbox, create, keyId, keys.private_pem)) === 'sent') delivered++;
   }
   console.log('[AP] direct note', site.slug, '→', resolved.length, 'recipient(s), delivered', delivered);