Changeset 7ccdced in Klonkt for src

Timestamp:

10/26/2026 01:46:00 PM (26 hours ago)

Author:

Robin <roboburr@…>

Branches:

main

Parents:

bfa3ba3

Message:

Een Delete van wat nog in de rij wacht: intern oplossen

Robins vraag (26-10): "Shouldn't a Delete on an undelivered Create resolve
internally for all unfederated ones?" Ja. bfa3ba3 zorgde er alleen voor dat de
Delete NA de wachtende Create aankwam -- maar de Create kwam dan nog steeds,
en wat gewist was bereikte zo alsnog een server die het nooit had gehad.

Nu haalt een Delete per inbox de wachtende Create en Update van datzelfde
object uit de rij (cancelUndelivered). Bewust alleen wat de inhoud draagt: een
Announce of Add van een gewist object is onschuldig, de ontvanger krijgt een
404 als hij het ophaalt.

Of de Delete zelf dan nog moet, hangt af van of er iets aangekomen KAN zijn.
Daarvoor de nieuwe kolom ap_delivery.tried: een rij die in de rij ging zonder
poging (de inbox wachtte al, of de server zat in zijn backoff) is nooit de deur
uit geweest; dan gaat er voor die inbox helemaal niets ('cancelled'). Is hij
wel geprobeerd, dan kan een time-out na verwerking hem hebben laten aankomen,
en gaat de Delete toch -- onschuldig als ze hem niet kenden. Rijen van voor de
kolom tellen als geprobeerd: dat is de veilige kant.

Daarbij twee directe verzenders die ik in bfa3ba3 had gemist, nu ook langs
deliverNowOrQueue: het bewerken van een antwoord (deliverOutboxUpdate -- een
Update vóór de wachtende Create wordt genegeerd, waarna de Create de oude tekst
brengt) en guardian-activiteiten (deliverToActor).

test/direct-sends-queue.test.js, nu negen toetsen: het gewiste antwoord dat
wachtte komt nooit aan (en de Delete gaat wel, want het was geprobeerd), een
nooit geprobeerd en gewist antwoord laat niets vertrekken, alleen Create en
Update van dat object en die inbox, en de bewerking en de guardian-activiteit
wachten. Acht mutaties apart gedraaid, elk gevangen; één (ook andere soorten
weghalen) eerst niet, waarop de bedoelde grens in de toets is vastgelegd.
Volledige suite 1380 groen.

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

Location:

src

Files:

• config/database.js (modified) (1 diff)
• services/ActivityPubService.js (modified) (3 diffs)
• services/ap-transport.js (modified) (3 diffs)

src/config/database.js

@@ -895 +895 @@
   // Rijen van voor deze kolom krijgen hun host hier, eenmalig.
   ensureColumn('ap_delivery', 'host', 'TEXT');
+  // Is deze rij al eens geprobeerd? Dan KAN hij aangekomen zijn, ook al gaf het
+  // een fout (een time-out nadat de ander hem verwerkte). Een Delete voor iets
+  // dat nooit de deur uitging, hoeft niet; voor iets dat misschien aankwam wel.
+  // Standaard 1: van rijen van voor deze kolom weten we het niet, en dan is
+  // "misschien aangekomen" de veilige kant. Nieuwe rijen zeggen het zelf.
+  ensureColumn('ap_delivery', 'tried', 'INTEGER DEFAULT 1');
   try {
     const zonder = db.prepare('SELECT id, inbox FROM ap_delivery WHERE host IS NULL').all();

src/services/ActivityPubService.js

@@ -34 +34 @@
 import {
   safeFetch, getOrCreateKeys, deliver, fetchActor,
-  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, processDeliveryQueue, startDeliveryWorker,
+  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, cancelUndelivered, processDeliveryQueue, startDeliveryWorker,
   anySigningSlug, verifyRequest, signedGetHeaders, signedGetJson, apGetJson,
 } from './ap-transport.js';
 export {
   safeFetch, getOrCreateKeys, deliver, fetchActor,
-  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, processDeliveryQueue, startDeliveryWorker,
+  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, cancelUndelivered, processDeliveryQueue, startDeliveryWorker,
   verifyRequest, signedGetHeaders, signedGetJson,
 };
@@ -2850 +2850 @@
   inboxes.delete(`${me}/inbox`); inboxes.delete(`${base}/ap/inbox`);
   let delivered = 0;
+  // Langs de volgorde per inbox (shaer-xzab.3): een bewerking die vóór de nog
+  // wachtende Create aankomt, wordt genegeerd, en daarna brengt de Create de
+  // oude tekst.
   for (const inbox of [...inboxes].filter(Boolean)) {
-    let ok = false;
-    try { const st = await deliver(inbox, update, `${me}#main-key`, keys.private_pem); ok = st >= 200 && st < 300; } catch { ok = false; }
-    if (ok) delivered++;
-    else enqueueDelivery(site.slug, inbox, update); // durable: retry the edit later (was silently dropped)
+    if ((await deliverNowOrQueue(site.slug, inbox, update, `${me}#main-key`, keys.private_pem)) === 'sent') delivered++;
   }
   console.log('[AP] outreply edit', site.slug, 'delivered', delivered);
@@ -4152 +4152 @@
     return { delivered: false, inbox: null };
   }
-  try {
-    const st = await deliver(inbox, payload, `${me}#main-key`, keys.private_pem);
-    if (st >= 200 && st < 300) { console.log('[AP] guardianship', activity.type, 'delivered →', inbox, st); return { delivered: true, inbox }; }
-    console.warn('[AP] guardianship', activity.type, 'got', st, 'from', inbox, '→ queued for retry');
-  } catch (e) { console.warn('[AP] guardianship', activity.type, 'to', inbox, 'failed:', e.message, '→ queued for retry'); }
-  enqueueDelivery(site.slug, inbox, payload);
+  // Langs de volgorde per inbox (shaer-xzab.3), zoals elke bezorging.
+  const hoe = await deliverNowOrQueue(site.slug, inbox, payload, `${me}#main-key`, keys.private_pem);
+  if (hoe === 'sent') console.log('[AP] guardianship', activity.type, 'delivered →', inbox);
+  else console.warn('[AP] guardianship', activity.type, 'to', inbox, '→ queued for retry');
   return { delivered: true, inbox };   // queued: the retry worker gets it there
 }

src/services/ap-transport.js

@@ -241 +241 @@
 function deliveryStmts() {
   if (!_insDeliv) {
-    _insDeliv = db.prepare('INSERT INTO ap_delivery (slug, inbox, host, body, attempts, next_at) VALUES (?,?,?,?,0,CURRENT_TIMESTAMP)');
+    _insDeliv = db.prepare('INSERT INTO ap_delivery (slug, inbox, host, body, attempts, next_at, tried) VALUES (?,?,?,?,0,CURRENT_TIMESTAMP,?)');
     // Alleen de OUDSTE openstaande rij per inbox: een nieuwere wacht op hem, ook
     // als hij zelf al aan de beurt zou zijn. Op id, want dat is de volgorde
@@ -260 +260 @@
   return { ins: _insDeliv, due: _dueDeliv, del: _delDeliv, bump: _bumpDeliv, host: _hostDeliv, off: _offDeliv, pendInbox: _pendInbox, hostWaits: _hostWaits };
 }
-export function enqueueDelivery(slug, inbox, activity) {
+// `tried`: ging er al een poging aan vooraf? Standaard ja, want zo riep iedereen
+// dit aan: na een mislukte deliver(). Alleen wie in de rij zet zonder te
+// proberen (deliverNowOrQueue bij een wachtende inbox of server) zegt nee.
+export function enqueueDelivery(slug, inbox, activity, { tried = true } = {}) {
   if (!slug || !inbox || !activity) return;
-  try { deliveryStmts().ins.run(slug, inbox, inboxHost(inbox), JSON.stringify(activity)); } catch { /* ignore */ }
+  try { deliveryStmts().ins.run(slug, inbox, inboxHost(inbox), JSON.stringify(activity), tried ? 1 : 0); } catch { /* ignore */ }
+}
+
+function objectIdOf(activity) {
+  const o = activity && activity.object;
+  return typeof o === 'string' ? o : (o && typeof o.id === 'string' ? o.id : null);
+}
+
+/**
+ * Een Delete van iets dat voor deze inbox nog in de rij staat (Robins vraag,
+ * 26-10: "shouldn't a Delete on an undelivered Create resolve internally?").
+ *
+ * Ja: haal de wachtende Create en Update van dat object uit de rij. Dan komt
+ * wat gewist is nooit meer aan bij een server die het nog niet had -- de
+ * volgorde per inbox alleen zorgde er wel voor dat de Delete NA de Create
+ * kwam, maar de Create kwam dan nog steeds.
+ *
+ * Of de Delete daarna nog hoeft, hangt af van of er iets aangekomen KAN zijn:
+ * een rij die al eens geprobeerd is, kan dat (een time-out nadat de ander hem
+ * verwerkte). Dan gaat de Delete toch; die is onschuldig als ze hem niet
+ * kenden, en zonder hem blijft de post daar misschien staan.
+ */
+export function cancelUndelivered(inbox, objectId) {
+  if (!inbox || !objectId) return { cancelled: 0, mayHaveArrived: false };
+  let cancelled = 0, mayHaveArrived = false;
+  try {
+    for (const r of db.prepare('SELECT id, body, attempts, tried FROM ap_delivery WHERE inbox = ?').all(inbox)) {
+      let a; try { a = JSON.parse(r.body); } catch { continue; }
+      if (!a || (a.type !== 'Create' && a.type !== 'Update') || objectIdOf(a) !== objectId) continue;
+      if (r.tried || r.attempts > 0) mayHaveArrived = true;
+      deliveryStmts().del.run(r.id);
+      cancelled++;
+    }
+  } catch { /* geen wachtrij te lezen: dan gewoon de Delete */ }
+  return { cancelled, mayHaveArrived };
 }
 // Record delivery health per follower so the followers list can flag dead accounts.
@@ -289 +326 @@
 // hoort (shaer-xzab.3). Drie plekken deden eerst zelf deliver() en pas bij een
 // mislukking enqueueDelivery(), en gingen zo langs de volgorde per inbox.
+//
+// 'cancelled': een Delete van iets dat deze inbox nooit kreeg; er gaat niets
+// heen (zie cancelUndelivered).
 export async function deliverNowOrQueue(slug, inbox, activity, keyId, privPem) {
   if (!inbox) return 'queued';
+  if (activity && activity.type === 'Delete') {
+    const r = cancelUndelivered(inbox, objectIdOf(activity));
+    if (r.cancelled && !r.mayHaveArrived) return 'cancelled';
+  }
   try {
     const st = deliveryStmts();
-    if (st.pendInbox.get(inbox) || st.hostWaits.get(inboxHost(inbox))) { enqueueDelivery(slug, inbox, activity); return 'queued'; }
+    if (st.pendInbox.get(inbox) || st.hostWaits.get(inboxHost(inbox))) { enqueueDelivery(slug, inbox, activity, { tried: false }); 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 */ }