From aa41c67a207e5a506fe896dd09010076eee2a595 Mon Sep 17 00:00:00 2001 From: Maurice Date: Fri, 30 Oct 2026 23:14:32 +0000 Subject: [PATCH] place-contrib: settle a sent value only while its queue row still holds it A value saved while the request for its predecessor was out rewrote the same outbox row with a new cid; the answer for the old value then deleted the row, or charged it an attempt, and the newer value was never sent. The settle writes now match the cid that was sent as well as the id, so such a row stays queued for the next run. A replaced value also counts its age from the replacement, so a long pause cannot expire a value queued days ago. --- .../PlaceContribOutbox.repository.ts | 44 ++++++++++++----- .../place-contrib-sender.service.ts | 14 ++---- .../PlaceContribOutbox.repository.test.ts | 47 +++++++++++++++---- .../place-contrib-sender.service.test.ts | 20 ++++++++ 4 files changed, 95 insertions(+), 30 deletions(-) diff --git a/server/src/db/repositories/PlaceContribOutbox.repository.ts b/server/src/db/repositories/PlaceContribOutbox.repository.ts index 2fbee9c0d..142692324 100644 --- a/server/src/db/repositories/PlaceContribOutbox.repository.ts +++ b/server/src/db/repositories/PlaceContribOutbox.repository.ts @@ -20,6 +20,12 @@ export interface OutboxRow { created_at: string; } +/** + * A value the sender read: its settle writes touch the row only while it still + * holds that value, since a newer one keeps the row's id but brings a new cid. + */ +export type OutboxSent = Pick; + /** A value to queue: the newest one for its key and field replaces what waits there. */ interface OutboxWrite { contrib_key: string; @@ -73,7 +79,7 @@ export class PlaceContribOutboxRepository extends TrekRepository(); } - /** Queues a value, replacing whatever waited for the same key and field, attempts included. */ + /** Queues a value, replacing whatever waited for the same key and field, attempts and age included. */ async upsertEntry(row: OutboxWrite): Promise { await this.db() .insertInto('place_contrib_outbox') @@ -88,6 +94,7 @@ export class PlaceContribOutboxRepository extends TrekRepository eb.ref('excluded.country'), cid: (eb) => eb.ref('excluded.cid'), attempts: 0, + created_at: (eb) => eb.ref('excluded.created_at'), eligible_at: (eb) => eb.ref('excluded.eligible_at'), next_attempt_at: (eb) => eb.ref('excluded.next_attempt_at'), }), @@ -124,9 +131,14 @@ export class PlaceContribOutboxRepository extends TrekRepository { - if (ids.length === 0) return 0; - const result = await this.db().deleteFrom('place_contrib_outbox').where('id', 'in', ids).executeTakeFirst(); + /** Drops the values the sender settled; a row that took a newer value since stays queued. */ + async deleteSent(sent: OutboxSent[]): Promise { + if (sent.length === 0) return 0; + const result = await this.db() + .deleteFrom('place_contrib_outbox') + .where('id', 'in', sent.map((row) => row.id)) + .where('cid', 'in', sent.map((row) => row.cid)) + .executeTakeFirst(); return Number(result.numDeletedRows); } @@ -182,23 +194,31 @@ export class PlaceContribOutboxRepository extends TrekRepository { - if (ids.length === 0) return; + /** + * Counts the attempt and moves the values to their next try; a row that took + * a newer value since is left as it is. + */ + async scheduleRetry(sent: OutboxSent[], nextAttemptAt: string): Promise { + if (sent.length === 0) return; await this.db() .updateTable('place_contrib_outbox') .set((eb) => ({ attempts: eb('attempts', '+', 1), next_attempt_at: nextAttemptAt })) - .where('id', 'in', ids) + .where('id', 'in', sent.map((row) => row.id)) + .where('cid', 'in', sent.map((row) => row.cid)) .execute(); } - /** Moves the values to a later try without counting one (the service is pausing, not failing). */ - async postpone(ids: number[], nextAttemptAt: string): Promise { - if (ids.length === 0) return; + /** + * Moves the values to a later try without counting one (the service is + * pausing, not failing); a row that took a newer value since is left as it is. + */ + async postpone(sent: OutboxSent[], nextAttemptAt: string): Promise { + if (sent.length === 0) return; await this.db() .updateTable('place_contrib_outbox') .set({ next_attempt_at: nextAttemptAt }) - .where('id', 'in', ids) + .where('id', 'in', sent.map((row) => row.id)) + .where('cid', 'in', sent.map((row) => row.cid)) .execute(); } diff --git a/server/src/nest/place-contrib/place-contrib-sender.service.ts b/server/src/nest/place-contrib/place-contrib-sender.service.ts index 1b1f06174..ba2719064 100644 --- a/server/src/nest/place-contrib/place-contrib-sender.service.ts +++ b/server/src/nest/place-contrib/place-contrib-sender.service.ts @@ -235,7 +235,7 @@ export class PlaceContribSender { if (refusal.status === 401 || refusal.status === 403) { // The key is not taken: wait for another one, never fall back to sending without. await this.store.pauseFor(refusal.status === 401 ? 'key_invalid' : 'key_refused', null); - await this.outbox.postpone(ids(entries), dbNow(new Date(now.getTime() + NOT_ACCEPTING_WAIT_MS))); + await this.outbox.postpone(entries, dbNow(new Date(now.getTime() + NOT_ACCEPTING_WAIT_MS))); return 'stop'; } if (refusal.status === 429) { @@ -245,7 +245,7 @@ export class PlaceContribSender { if (refusal.status === 503) { const until = new Date(now.getTime() + (refusal.retryAfterMs ?? NOT_ACCEPTING_WAIT_MS)); await this.store.pauseFor('not_accepting', until); - await this.outbox.postpone(ids(entries), dbNow(until)); + await this.outbox.postpone(entries, dbNow(until)); return 'stop'; } const refusedField = /^field_not_accepted:([a-z_]+)$/.exec(refusal.error)?.[1]; @@ -257,7 +257,7 @@ export class PlaceContribSender { taken.map((entry): Final => ({ entry, result: 'rejected', reason: refusal.error })), now, ); - await this.outbox.postpone(ids(others), dbNow(now)); + await this.outbox.postpone(others, dbNow(now)); return 'next'; } await this.finish( @@ -273,7 +273,7 @@ export class PlaceContribSender { const attempts = Math.max(1, ...again.map((entry) => entry.attempts + 1)); const next = dbNow(new Date(opts.now.getTime() + (opts.retryAfterMs ?? retryDelayMs(attempts)))); await this.uow.transactional(async () => { - await this.outbox.scheduleRetry(ids(again), next); + await this.outbox.scheduleRetry(again, next); await this.finishInTx( givenUp.map((entry): Final => ({ entry, result: 'withheld', reason: 'gave_up' })), opts.now, @@ -291,7 +291,7 @@ export class PlaceContribSender { private async finishInTx(finals: Final[], now: Date): Promise { if (finals.length === 0) return; await this.log.insertLines(finals.map((final) => logLine(final, now))); - await this.outbox.deleteIds(finals.map((final) => final.entry.id)); + await this.outbox.deleteSent(finals.map((final) => final.entry)); } /** The bytes of an uploaded place picture, null when it is gone or heavier than any upload could be. */ @@ -385,10 +385,6 @@ function logLine(final: Final, now: Date): PlaceContribLogWrite { }; } -function ids(entries: OutboxRow[]): number[] { - return entries.map((entry) => entry.id); -} - function partition(items: T[], test: (item: T) => boolean): [T[], T[]] { const yes: T[] = []; const no: T[] = []; diff --git a/server/tests/unit/db/repositories/PlaceContribOutbox.repository.test.ts b/server/tests/unit/db/repositories/PlaceContribOutbox.repository.test.ts index 756f2eb9e..3fe52059a 100644 --- a/server/tests/unit/db/repositories/PlaceContribOutbox.repository.test.ts +++ b/server/tests/unit/db/repositories/PlaceContribOutbox.repository.test.ts @@ -66,10 +66,7 @@ describe('PlaceContribOutboxRepository', () => { it('PCOUTREPO-001: a newer value for the same key and field replaces the queued one and starts its attempts over', async () => { const placeId = await aPlace(); await queue(placeId); - await outbox.scheduleRetry( - (await outbox.dueFor('osm:node/1', NOW)).map((row) => row.id), - PAST, - ); + await outbox.scheduleRetry(await outbox.dueFor('osm:node/1', NOW), PAST); await queue(placeId, { value: 'https://luna.example', cid: 'cid-website-0002' }); const rows = await outbox.dueFor('osm:node/1', NOW); expect(rows).toEqual([ @@ -94,9 +91,9 @@ describe('PlaceContribOutboxRepository', () => { await queue(placeId); const [row] = await outbox.dueFor('osm:node/1', NOW); if (!row) throw new Error('the queued value is missing'); - await outbox.scheduleRetry([row.id], LATER); + await outbox.scheduleRetry([row], LATER); expect(await outbox.dueFor('osm:node/1', NOW)).toEqual([]); - await outbox.postpone([row.id], PAST); + await outbox.postpone([row], PAST); const [again] = await outbox.dueFor('osm:node/1', NOW); expect(again).toEqual(expect.objectContaining({ attempts: 1 })); await outbox.scheduleRetry([], LATER); @@ -132,12 +129,11 @@ describe('PlaceContribOutboxRepository', () => { expect(await outbox.deleteForPlace(one, [])).toBe(0); expect(await outbox.deleteFields([])).toBe(0); - expect(await outbox.deleteIds([])).toBe(0); + expect(await outbox.deleteSent([])).toBe(0); expect(await outbox.deleteForPlace(one, ['phone'])).toBe(1); expect(await outbox.deleteFields(['email'])).toBe(1); - const left = (await outbox.dueFor('osm:node/2', NOW)).map((row) => row.id); - expect(await outbox.deleteIds(left)).toBe(1); + expect(await outbox.deleteSent(await outbox.dueFor('osm:node/2', NOW))).toBe(1); expect(await outbox.deleteAll()).toBe(1); expect(await countRows(t, PlaceContribOutbox)).toBe(0); }); @@ -162,4 +158,37 @@ describe('PlaceContribOutboxRepository', () => { ]), ); }); + + it('PCOUTREPO-007: a value replaced after the sender read it is neither deleted nor charged nor postponed', async () => { + const placeId = await aPlace(); + await queue(placeId); + const read = await outbox.dueFor('osm:node/1', NOW); + await queue(placeId, { value: 'https://luna.example', cid: 'cid-website-0002' }); + + await outbox.scheduleRetry(read, LATER); + await outbox.postpone(read, LATER); + expect(await outbox.deleteSent(read)).toBe(0); + + expect(await outbox.dueFor('osm:node/1', NOW)).toEqual([ + expect.objectContaining({ value: 'https://luna.example', cid: 'cid-website-0002', attempts: 0 }), + ]); + }); + + it('PCOUTREPO-008: a replaced value counts its age from the replacement, not from the value it replaced', async () => { + const placeId = await aPlace(); + await insertRow(t, PlaceContribOutbox, { + contrib_key: 'osm:node/1', + field: 'website', + value: 'https://cafe-luna.de', + place: placeId, + place_name: 'Café Luna', + cid: 'cid-website-0001', + eligible_at: PAST, + next_attempt_at: PAST, + created_at: '2026-01-01 00:00:00', + }); + expect(await outbox.queuedBefore('2026-06-01 00:00:00', 10)).toHaveLength(1); + await queue(placeId, { value: 'https://luna.example', cid: 'cid-website-0002' }); + expect(await outbox.queuedBefore('2026-06-01 00:00:00', 10)).toEqual([]); + }); }); diff --git a/server/tests/unit/nest/place-contrib/place-contrib-sender.service.test.ts b/server/tests/unit/nest/place-contrib/place-contrib-sender.service.test.ts index d5a87fe9b..fb0cf420c 100644 --- a/server/tests/unit/nest/place-contrib/place-contrib-sender.service.test.ts +++ b/server/tests/unit/nest/place-contrib/place-contrib-sender.service.test.ts @@ -292,6 +292,26 @@ describe('a place request', () => { expect(client.servedPlace).not.toHaveBeenCalled(); expect(await countRows(orm(), PlaceContribOutbox)).toBe(2); }); + + it('PCSD-015: a value the user replaces while its request is out stays queued for the next run', async () => { + await seed({ website: 'https://cafe-luna.de' }); + client.postPlace.mockImplementation(async (body) => { + // The user saves a newer website while the request for the old one is out. + await updateRows( + orm(), + PlaceContribOutbox, + { field: 'website' }, + { value: 'https://luna.example', cid: 'cid-website-0002' }, + ); + return accepted(resultsFor(body, 'published')); + }); + await sender().run(NOW); + // The old value went out and is logged; the newer one waits for its own request. + expect(await logLines()).toEqual([{ field: 'website', result: 'accepted', reason: 'published' }]); + expect(await findRows(orm(), PlaceContribOutbox, {})).toEqual([ + expect.objectContaining({ value: 'https://luna.example', cid: 'cid-website-0002', attempts: 0 }), + ]); + }); }); describe('refusals and failures', () => {