k1nq/TREK
1
0
Fork 0
mirror of https://github.com/mauriceboe/TREK.git synced 2026-11-13 15:51:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

place-contrib: settle a sent value only while its queue row still holds it

Browse Source 
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.
This commit is contained in:
Maurice
2026-10-30 23:14:32 +00:00
parent c10a698b9f
commit aa41c67a20
4 changed files with 95 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/db/repositories/PlaceContribOutbox.repository.ts
+32 -12
View File
@@ -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<OutboxRow, 'id' | 'cid'>;
/** 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<PlaceContribOut
return this.kysely<OutboxKyselyDB>();
}
/** 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<void> {
await this.db()
.insertInto('place_contrib_outbox')
@@ -88,6 +94,7 @@ export class PlaceContribOutboxRepository extends TrekRepository<PlaceContribOut
country: (eb) => 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<PlaceContribOut
return Number(result.numDeletedRows);
}
async deleteIds(ids: number[]): Promise<number> {
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<number> {
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<PlaceContribOut
.execute();
}
/** Counts the attempt and moves the values to their next try. */
async scheduleRetry(ids: number[], nextAttemptAt: string): Promise<void> {
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<void> {
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<void> {
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<void> {
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();
}
server/src/nest/place-contrib/place-contrib-sender.service.ts
+5 -9
View File
@@ -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<void> {
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<T>(items: T[], test: (item: T) => boolean): [T[], T[]] {
const yes: T[] = [];
const no: T[] = [];