Changeset 7ccdced in Klonkt


Ignore:
Timestamp:
10/26/2026 01:46:00 PM (24 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@…>

Files:
4 edited

Legend:

Unmodified
Added
Removed

  • src/config/database.js

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

  • src/services/ActivityPubService.js

    rbfa3ba3 r7ccdced  
    3434import {
    3535  safeFetch, getOrCreateKeys, deliver, fetchActor,
    36   enqueueDelivery, deliverWithRetry, deliverNowOrQueue, processDeliveryQueue, startDeliveryWorker,
     36  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, cancelUndelivered, processDeliveryQueue, startDeliveryWorker,
    3737  anySigningSlug, verifyRequest, signedGetHeaders, signedGetJson, apGetJson,
    3838} from './ap-transport.js';
    3939export {
    4040  safeFetch, getOrCreateKeys, deliver, fetchActor,
    41   enqueueDelivery, deliverWithRetry, deliverNowOrQueue, processDeliveryQueue, startDeliveryWorker,
     41  enqueueDelivery, deliverWithRetry, deliverNowOrQueue, cancelUndelivered, processDeliveryQueue, startDeliveryWorker,
    4242  verifyRequest, signedGetHeaders, signedGetJson,
    4343};
    … …  
    28502850  inboxes.delete(`${me}/inbox`); inboxes.delete(`${base}/ap/inbox`);
    28512851  let delivered = 0;
     2852  // Langs de volgorde per inbox (shaer-xzab.3): een bewerking die vóór de nog
     2853  // wachtende Create aankomt, wordt genegeerd, en daarna brengt de Create de
     2854  // oude tekst.
    28522855  for (const inbox of [...inboxes].filter(Boolean)) {
    2853     let ok = false;
    2854     try { const st = await deliver(inbox, update, `${me}#main-key`, keys.private_pem); ok = st >= 200 && st < 300; } catch { ok = false; }
    2855     if (ok) delivered++;
    2856     else enqueueDelivery(site.slug, inbox, update); // durable: retry the edit later (was silently dropped)
     2856    if ((await deliverNowOrQueue(site.slug, inbox, update, `${me}#main-key`, keys.private_pem)) === 'sent') delivered++;
    28572857  }
    28582858  console.log('[AP] outreply edit', site.slug, 'delivered', delivered);
    … …  
    41524152    return { delivered: false, inbox: null };
    41534153  }
    4154   try {
    4155     const st = await deliver(inbox, payload, `${me}#main-key`, keys.private_pem);
    4156     if (st >= 200 && st < 300) { console.log('[AP] guardianship', activity.type, 'delivered →', inbox, st); return { delivered: true, inbox }; }
    4157     console.warn('[AP] guardianship', activity.type, 'got', st, 'from', inbox, '→ queued for retry');
    4158   } catch (e) { console.warn('[AP] guardianship', activity.type, 'to', inbox, 'failed:', e.message, '→ queued for retry'); }
    4159   enqueueDelivery(site.slug, inbox, payload);
     4154  // Langs de volgorde per inbox (shaer-xzab.3), zoals elke bezorging.
     4155  const hoe = await deliverNowOrQueue(site.slug, inbox, payload, `${me}#main-key`, keys.private_pem);
     4156  if (hoe === 'sent') console.log('[AP] guardianship', activity.type, 'delivered →', inbox);
     4157  else console.warn('[AP] guardianship', activity.type, 'to', inbox, '→ queued for retry');
    41604158  return { delivered: true, inbox };   // queued: the retry worker gets it there
    41614159}

  • src/services/ap-transport.js

    rbfa3ba3 r7ccdced  
    241241function deliveryStmts() {
    242242  if (!_insDeliv) {
    243     _insDeliv = db.prepare('INSERT INTO ap_delivery (slug, inbox, host, body, attempts, next_at) VALUES (?,?,?,?,0,CURRENT_TIMESTAMP)');
     243    _insDeliv = db.prepare('INSERT INTO ap_delivery (slug, inbox, host, body, attempts, next_at, tried) VALUES (?,?,?,?,0,CURRENT_TIMESTAMP,?)');
    244244    // Alleen de OUDSTE openstaande rij per inbox: een nieuwere wacht op hem, ook
    245245    // als hij zelf al aan de beurt zou zijn. Op id, want dat is de volgorde
    … …  
    260260  return { ins: _insDeliv, due: _dueDeliv, del: _delDeliv, bump: _bumpDeliv, host: _hostDeliv, off: _offDeliv, pendInbox: _pendInbox, hostWaits: _hostWaits };
    261261}
    262 export function enqueueDelivery(slug, inbox, activity) {
     262// `tried`: ging er al een poging aan vooraf? Standaard ja, want zo riep iedereen
     263// dit aan: na een mislukte deliver(). Alleen wie in de rij zet zonder te
     264// proberen (deliverNowOrQueue bij een wachtende inbox of server) zegt nee.
     265export function enqueueDelivery(slug, inbox, activity, { tried = true } = {}) {
    263266  if (!slug || !inbox || !activity) return;
    264   try { deliveryStmts().ins.run(slug, inbox, inboxHost(inbox), JSON.stringify(activity)); } catch { /* ignore */ }
     267  try { deliveryStmts().ins.run(slug, inbox, inboxHost(inbox), JSON.stringify(activity), tried ? 1 : 0); } catch { /* ignore */ }
     268}
     269
     270function objectIdOf(activity) {
     271  const o = activity && activity.object;
     272  return typeof o === 'string' ? o : (o && typeof o.id === 'string' ? o.id : null);
     273}
     274
     275/**
     276 * Een Delete van iets dat voor deze inbox nog in de rij staat (Robins vraag,
     277 * 26-10: "shouldn't a Delete on an undelivered Create resolve internally?").
     278 *
     279 * Ja: haal de wachtende Create en Update van dat object uit de rij. Dan komt
     280 * wat gewist is nooit meer aan bij een server die het nog niet had -- de
     281 * volgorde per inbox alleen zorgde er wel voor dat de Delete NA de Create
     282 * kwam, maar de Create kwam dan nog steeds.
     283 *
     284 * Of de Delete daarna nog hoeft, hangt af van of er iets aangekomen KAN zijn:
     285 * een rij die al eens geprobeerd is, kan dat (een time-out nadat de ander hem
     286 * verwerkte). Dan gaat de Delete toch; die is onschuldig als ze hem niet
     287 * kenden, en zonder hem blijft de post daar misschien staan.
     288 */
     289export function cancelUndelivered(inbox, objectId) {
     290  if (!inbox || !objectId) return { cancelled: 0, mayHaveArrived: false };
     291  let cancelled = 0, mayHaveArrived = false;
     292  try {
     293    for (const r of db.prepare('SELECT id, body, attempts, tried FROM ap_delivery WHERE inbox = ?').all(inbox)) {
     294      let a; try { a = JSON.parse(r.body); } catch { continue; }
     295      if (!a || (a.type !== 'Create' && a.type !== 'Update') || objectIdOf(a) !== objectId) continue;
     296      if (r.tried || r.attempts > 0) mayHaveArrived = true;
     297      deliveryStmts().del.run(r.id);
     298      cancelled++;
     299    }
     300  } catch { /* geen wachtrij te lezen: dan gewoon de Delete */ }
     301  return { cancelled, mayHaveArrived };
    265302}
    266303// Record delivery health per follower so the followers list can flag dead accounts.
    … …  
    289326// hoort (shaer-xzab.3). Drie plekken deden eerst zelf deliver() en pas bij een
    290327// mislukking enqueueDelivery(), en gingen zo langs de volgorde per inbox.
     328//
     329// 'cancelled': een Delete van iets dat deze inbox nooit kreeg; er gaat niets
     330// heen (zie cancelUndelivered).
    291331export async function deliverNowOrQueue(slug, inbox, activity, keyId, privPem) {
    292332  if (!inbox) return 'queued';
     333  if (activity && activity.type === 'Delete') {
     334    const r = cancelUndelivered(inbox, objectIdOf(activity));
     335    if (r.cancelled && !r.mayHaveArrived) return 'cancelled';
     336  }
    293337  try {
    294338    const st = deliveryStmts();
    295     if (st.pendInbox.get(inbox) || st.hostWaits.get(inboxHost(inbox))) { enqueueDelivery(slug, inbox, activity); return 'queued'; }
     339    if (st.pendInbox.get(inbox) || st.hostWaits.get(inboxHost(inbox))) { enqueueDelivery(slug, inbox, activity, { tried: false }); return 'queued'; }
    296340  } catch { /* geen wachtrij te lezen: gewoon proberen */ }
    297341  try { const st = await deliver(inbox, activity, keyId, privPem); if (st >= 200 && st < 300) { markFollowerDelivery(slug, inbox, true); return 'sent'; } } catch { /* queue below */ }

  • test/direct-sends-queue.test.js

    rbfa3ba3 r7ccdced  
    5656const stil = (f) => async () => { const l = console.log, w = console.warn; console.log = () => {}; console.warn = () => {}; try { await f(); } finally { console.log = l; console.warn = w; } };
    5757
    58 test('een gewist antwoord komt niet aan vóór het antwoord zelf', stil(async () => {
     58test('een gewist antwoord dat nog wachtte, komt nooit meer aan', stil(async () => {
     59  // Robins vraag (26-10): een Delete van een Create die nog in de rij staat,
     60  // hoort intern opgelost te worden. Eerst zorgde de volgorde per inbox er
     61  // alleen voor dat de Delete NA de Create kwam -- maar de Create kwam dan nog.
    5962  leeg();
    6063  status = 503;                                            // de server van de volger hapert
    … …  
    6669  ontvangen.length = 0;
    6770  await AP.deliverOutboxDelete(site, r.id);
    68   assert.deepEqual(ontvangen, [], 'de Delete gaat niet voor');
    69   assert.deepEqual(typen(), ['Create', 'Delete'], 'hij wacht achter de Create');
     71  await AP.processDeliveryQueue();
     72  assert.ok(!ontvangen.some((o) => o.type === 'Create'), 'het gewiste antwoord gaat niet meer de deur uit');
     73  // Hij is wel één keer geprobeerd (de 503), en dan KAN hij aangekomen zijn:
     74  // dan gaat de Delete toch, voor de zekerheid.
     75  assert.deepEqual(ontvangen.map((o) => o.type), ['Delete']);
     76  assert.deepEqual(rij(), []);
     77}));
    7078
    71   await AP.processDeliveryQueue();
    72   await AP.processDeliveryQueue();
    73   assert.deepEqual(ontvangen.map((o) => o.type), ['Create', 'Delete'], 'en zo komen ze aan');
     79test('nooit geprobeerd en gewist: er gaat helemaal niets heen', stil(async () => {
     80  leeg();
     81  // De server zit in zijn backoff, dus het antwoord gaat meteen de rij in,
     82  // zonder poging; daarna wordt het gewist.
     83  AP.enqueueDelivery('me', `http://${host}/ander-pad`, { type: 'Create', id: 'https://klonkt.test/iets' });
     84  db.prepare("UPDATE ap_delivery SET next_at = datetime('now', '+30 minutes')").run();
     85  const r = await AP.deliverReply(site, { postId: '', postSlug: null, parent, text: 'nooit verstuurd' });
     86  assert.equal(rij().filter((x) => x.inbox === inbox).length, 1, 'het antwoord wacht, ongeprobeerd');
     87  await AP.deliverOutboxDelete(site, r.id);
     88  assert.deepEqual(rij().filter((x) => x.inbox === inbox), [], 'Create weg, en geen Delete erachteraan');
     89  assert.deepEqual(ontvangen, []);
     90  assert.equal(rij().length, 1, 'wat er voor een andere inbox wachtte, blijft');
     91}));
     92
     93test('alleen de Create en Update van DAT object, en alleen voor die inbox', () => {
     94  leeg();
     95  const obj = 'https://klonkt.test/ap/notes/x';
     96  AP.enqueueDelivery('me', inbox, { type: 'Create', id: `${obj}#create`, object: { id: obj } }, { tried: false });
     97  AP.enqueueDelivery('me', inbox, { type: 'Update', id: `${obj}#update`, object: { id: obj } }, { tried: false });
     98  AP.enqueueDelivery('me', inbox, { type: 'Create', id: 'y#create', object: { id: 'https://klonkt.test/ap/notes/y' } }, { tried: false });
     99  AP.enqueueDelivery('me', `http://${host}/elders`, { type: 'Create', id: `${obj}#create`, object: { id: obj } }, { tried: false });
     100  // Bewust ALLEEN wat de inhoud draagt. Een Announce of Add van een gewist
     101  // object is onschuldig: de ontvanger haalt het op en krijgt een 404.
     102  AP.enqueueDelivery('me', inbox, { type: 'Announce', id: `${obj}#boost`, object: obj }, { tried: false });
     103  const r = AP.cancelUndelivered(inbox, obj);
     104  assert.deepEqual(r, { cancelled: 2, mayHaveArrived: false });
     105  assert.deepEqual(rij().map((x) => [x.inbox.endsWith('/inbox') ? 'deze' : 'andere', JSON.parse(x.body).type, String(JSON.parse(x.body).object.id || JSON.parse(x.body).object).slice(-1)]),
     106    [['deze', 'Create', 'y'], ['andere', 'Create', 'x'], ['deze', 'Announce', 'x']]);
     107});
     108
     109test('een bewerking gaat niet voorbij het antwoord dat nog wacht', stil(async () => {
     110  leeg();
     111  status = 503;
     112  const r = await AP.deliverReply(site, { postId: '', postSlug: null, parent, text: 'origineel' });
     113  status = 202; ontvangen.length = 0;
     114  await AP.deliverOutboxUpdate(site, r.id, 'bewerkt');
     115  assert.deepEqual(ontvangen, [], 'de Update gaat niet voor: dan negeert de ander hem, en brengt de Create daarna de oude tekst');
     116  assert.deepEqual(typen(), ['Create', 'Update']);
     117}));
     118
     119test('een guardian-activiteit gaat niet voorbij wat er voor die inbox wacht', stil(async () => {
     120  leeg();
     121  const omaInbox = `http://${host}/users/oma/inbox`;
     122  AP.enqueueDelivery('me', omaInbox, { type: 'Create', id: 'https://klonkt.test/eerder' });
     123  await AP.deliverToActor(site, `http://${host}/users/oma`, { type: 'Offer', id: 'https://klonkt.test/offer/1' });
     124  assert.deepEqual(ontvangen, []);
     125  assert.equal(rij().filter((x) => x.inbox === omaInbox).length, 2);
    74126}));
    75127
Note: See TracChangeset for help on using the changeset viewer.