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', () => {