Software/platform
1
1
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix product next sync error bubbling bug

Browse Source
This commit is contained in:
Dominic Ferrando
2026-07-03 19:47:55 -04:00
parent 69cf4900dd
commit 97881d357e
2 changed files with 15 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/server.ts
+2 -2
View File
@@ -250,9 +250,9 @@ export const app = new Elysia()
}); });
app.listen(3000, async () => { app.listen(3000, async () => {
log.info({ port: 3000 }, "server started")
if (cluster.worker?.id === 1) { if (cluster.worker?.id === 1) {
await syncService.nextSync().catch((err) => log.error({ err }, "failed to resume product sync queue")); log.info({ port: 3000 }, "server started")
await syncService.syncNext().catch((err) => log.error({ err }, "failed to resume product sync queue"));
} }
}); });
apps/api/src/services/sync.ts
+13 -10
View File
@@ -21,6 +21,7 @@ export class SyncService {
async sync(queuedSync: QueuedSync) { async sync(queuedSync: QueuedSync) {
let id = queuedSync.id ?? null; let id = queuedSync.id ?? null;
let freedSlot = false;
try { try {
log.info("starting sync run"); log.info("starting sync run");
await this.productSyncer.sync({ await this.productSyncer.sync({
@@ -31,13 +32,14 @@ export class SyncService {
printful_product_id_filter: queuedSync.filter?.printfulProductIds ?? null, printful_product_id_filter: queuedSync.filter?.printfulProductIds ?? null,
}; };
if (id) { if (id) {
// Guard against another worker/machine having already claimed
// this queued row between `queue.front()` and now.
const result = await db.updateTable("product_syncs") const result = await db.updateTable("product_syncs")
.set(values) .set(values)
.where("id", "=", id) .where("id", "=", id)
.where("started_at", "is", null) .where("started_at", "is", null)
.executeTakeFirst(); .executeTakeFirst();
// Guard against another worker/machine having already claimed
// this queued row between `queue.front()` and now.
if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError(); if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError();
} }
else { else {
@@ -61,9 +63,7 @@ export class SyncService {
.set({ syncing_printful_product_ids: [], ended_at: completionDate }) .set({ syncing_printful_product_ids: [], ended_at: completionDate })
.where("id", "=", id) .where("id", "=", id)
.execute(); .execute();
freedSlot = true;
// Sync next item in queue
await this.nextSync();
} }
}); });
} }
@@ -77,9 +77,8 @@ export class SyncService {
const activeSyncConflict = activeSync && err instanceof DatabaseError && err.code === "23505"; const activeSyncConflict = activeSync && err instanceof DatabaseError && err.code === "23505";
if (activeSyncConflict) { if (activeSyncConflict) {
// If active sync has been syncing for too long, clear it and retry this sync // If active sync has been syncing for 10 min (600000ms), clear it and retry this sync
const TEN_MIN = 600000; if (Date.now() - (activeSync.started_at?.getTime() ?? 0) > 600000) {
if (Date.now() - (activeSync.started_at?.getTime() ?? 0) > TEN_MIN) {
log.info("already active sync exceeded timeout"); log.info("already active sync exceeded timeout");
await db.updateTable("product_syncs") await db.updateTable("product_syncs")
.set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true }) .set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true })
@@ -98,14 +97,18 @@ export class SyncService {
.set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true }) .set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true })
.where("id", "=", id) .where("id", "=", id)
.execute(); .execute();
freedSlot = true;
} }
throw err; throw err;
} }
}
finally {
if (freedSlot)
this.syncNext().catch((err) => log.error({ err }, "failed to process next queued sync"));
} }
} }
async nextSync() { async syncNext() {
const nextSync = await this.queue.front(); const nextSync = await this.queue.front();
if (nextSync) { if (nextSync) {
await this.sync({ await this.sync({