diff --git a/src/lib/offline/queue.test.ts b/src/lib/offline/queue.test.ts index f5e6d0f..550a0c3 100644 --- a/src/lib/offline/queue.test.ts +++ b/src/lib/offline/queue.test.ts @@ -2,7 +2,15 @@ // src/lib/offline/queue.test.ts import { describe, it, expect, beforeEach } from "vitest"; import "fake-indexeddb/auto"; -import { enqueueAction, dequeueAction, getQueuedActions, markSynced, markConflict } from "./queue"; +import { + enqueueAction, + dequeueAction, + getQueuedActions, + markSynced, + markConflict, + markInFlight, + markFailed, +} from "./queue"; beforeEach(async () => { // Clear the queue between tests @@ -61,4 +69,45 @@ describe("offline queue", () => { expect(all[0].status).toBe("conflict"); expect(all[0].error).toBe("stale updated_at"); }); + + it("markInFlight transitions the action to in-flight and records an attempt", async () => { + const a = await enqueueAction("markOrderReady", { orderId: "abc" }); + const before = await getQueuedActions(); + expect(before[0].attempts).toBe(0); + expect(before[0].status).toBe("pending"); + expect(before[0].lastAttemptAt).toBeNull(); + + const beforeMs = Date.now(); + await markInFlight(a.id); + const afterMs = Date.now(); + + const all = await getQueuedActions(); + expect(all).toHaveLength(1); + expect(all[0].status).toBe("in-flight"); + expect(all[0].attempts).toBe(1); + expect(typeof all[0].lastAttemptAt).toBe("number"); + expect(all[0].lastAttemptAt).not.toBeNull(); + expect(all[0].lastAttemptAt!).toBeGreaterThanOrEqual(beforeMs); + expect(all[0].lastAttemptAt!).toBeLessThanOrEqual(afterMs); + }); + + it("markFailed transitions the action to failed and increments attempts", async () => { + const a = await enqueueAction("markOrderReady", { orderId: "abc" }); + // Pretend it has already been tried once + await markInFlight(a.id); + + const beforeMs = Date.now(); + await markFailed(a.id, "server returned 500"); + const afterMs = Date.now(); + + const all = await getQueuedActions(); + expect(all).toHaveLength(1); + expect(all[0].status).toBe("failed"); + expect(all[0].attempts).toBe(2); // markInFlight set it to 1, markFailed adds 1 + expect(all[0].error).toBe("server returned 500"); + expect(typeof all[0].lastAttemptAt).toBe("number"); + expect(all[0].lastAttemptAt).not.toBeNull(); + expect(all[0].lastAttemptAt!).toBeGreaterThanOrEqual(beforeMs); + expect(all[0].lastAttemptAt!).toBeLessThanOrEqual(afterMs); + }); }); diff --git a/src/lib/offline/queue.ts b/src/lib/offline/queue.ts index f4d3b6d..1886754 100644 --- a/src/lib/offline/queue.ts +++ b/src/lib/offline/queue.ts @@ -13,8 +13,15 @@ export async function enqueueAction( ): Promise { const db = await getOfflineDB(); const id = options.id ?? uuidv4(); - const existing = await db.get("queue", id); - if (existing) return existing; + // Atomic read-then-write inside a single readwrite transaction so two + // concurrent calls with the same explicit `id` can't both pass the + // existence check and write twice. + const tx = db.transaction("queue", "readwrite"); + const existing = await tx.store.get(id); + if (existing) { + await tx.done; + return existing; + } const action: QueuedAction = { id, actionName, @@ -24,7 +31,8 @@ export async function enqueueAction( lastAttemptAt: null, status: "pending", }; - await db.put("queue", action); + await tx.store.put(action); + await tx.done; return action; } @@ -49,12 +57,17 @@ export async function dequeueAction(id: string): Promise { export async function markInFlight(id: string): Promise { const db = await getOfflineDB(); - const action = await db.get("queue", id); - if (!action) return; + const tx = db.transaction("queue", "readwrite"); + const action = await tx.store.get(id); + if (!action) { + await tx.done; + return; + } action.status = "in-flight"; action.lastAttemptAt = Date.now(); action.attempts += 1; - await db.put("queue", action); + await tx.store.put(action); + await tx.done; } export async function markSynced(id: string): Promise { @@ -64,20 +77,30 @@ export async function markSynced(id: string): Promise { export async function markConflict(id: string, error: string): Promise { const db = await getOfflineDB(); - const action = await db.get("queue", id); - if (!action) return; + const tx = db.transaction("queue", "readwrite"); + const action = await tx.store.get(id); + if (!action) { + await tx.done; + return; + } action.status = "conflict"; action.error = error; - await db.put("queue", action); + await tx.store.put(action); + await tx.done; } export async function markFailed(id: string, error: string): Promise { const db = await getOfflineDB(); - const action = await db.get("queue", id); - if (!action) return; + const tx = db.transaction("queue", "readwrite"); + const action = await tx.store.get(id); + if (!action) { + await tx.done; + return; + } action.status = "failed"; action.error = error; action.attempts += 1; action.lastAttemptAt = Date.now(); - await db.put("queue", action); + await tx.store.put(action); + await tx.done; } diff --git a/src/lib/offline/sync.test.ts b/src/lib/offline/sync.test.ts index 7a4489b..ce233dc 100644 --- a/src/lib/offline/sync.test.ts +++ b/src/lib/offline/sync.test.ts @@ -61,4 +61,100 @@ describe("offline sync — replay", () => { expect(dispatcher).not.toHaveBeenCalled(); expect(result.synced).toBe(0); }); + + it("syncPending is a no-op for an empty queue and does not call the dispatcher", async () => { + const dispatcher = vi.fn(); + const result = await syncPending(dispatcher, { online: true, sleep: () => Promise.resolve() }); + expect(dispatcher).not.toHaveBeenCalled(); + expect(result).toEqual({ synced: 0, conflicts: 0, failed: 0 }); + }); + + it("syncPending moves an action to failed once max attempts is reached", async () => { + // Seed an action that's already at the max-attempt threshold + await enqueueAction("markOrderReady", { orderId: "abc" }, { id: "max-attempts-id" }); + const { getOfflineDB } = await import("./db"); + const db = await getOfflineDB(); + const tx = db.transaction("queue", "readwrite"); + const a = await tx.store.get("max-attempts-id"); + if (!a) throw new Error("seed not found"); + a.attempts = MAX_ATTEMPTS; // 5 + await tx.store.put(a); + await tx.done; + + const dispatcher = vi.fn(); + const result = await syncPending(dispatcher, { online: true, sleep: () => Promise.resolve() }); + expect(dispatcher).not.toHaveBeenCalled(); + expect(result.failed).toBe(1); + expect(result.synced).toBe(0); + expect(result.conflicts).toBe(0); + + const { getQueuedActions } = await import("./queue"); + const all = await getQueuedActions(); + expect(all).toHaveLength(1); + expect(all[0].status).toBe("failed"); + expect(all[0].error).toBe("max attempts reached"); + }); + + it("syncPending skips actions whose backoff window has not yet elapsed", async () => { + await enqueueAction("markOrderReady", { orderId: "abc" }, { id: "backoff-id" }); + // Pretend a previous attempt happened 100ms ago. With attempts=0, the + // backoff schedule says we must wait 1000ms — so we should skip. + const fixedNow = 1_700_000_000_000; + const { getOfflineDB } = await import("./db"); + const db = await getOfflineDB(); + const tx = db.transaction("queue", "readwrite"); + const a = await tx.store.get("backoff-id"); + if (!a) throw new Error("seed not found"); + a.attempts = 0; + a.lastAttemptAt = fixedNow - 100; + await tx.store.put(a); + await tx.done; + + const dispatcher = vi.fn(); + const result = await syncPending(dispatcher, { + online: true, + sleep: () => Promise.resolve(), + now: () => fixedNow, + }); + expect(dispatcher).not.toHaveBeenCalled(); + expect(result.synced).toBe(0); + expect(result.conflicts).toBe(0); + expect(result.failed).toBe(0); + + // Action is still pending (not advanced to in-flight or failed) + const { getQueuedActions } = await import("./queue"); + const all = await getQueuedActions(); + expect(all).toHaveLength(1); + expect(all[0].status).toBe("pending"); + }); + + it("syncPending marks the action as failed when the dispatcher throws", async () => { + const a = await enqueueAction("markOrderReady", { orderId: "abc" }); + const dispatcher = vi.fn().mockRejectedValue(new Error("network down")); + const result = await syncPending(dispatcher, { online: true, sleep: () => Promise.resolve() }); + expect(dispatcher).toHaveBeenCalledWith("markOrderReady", { orderId: "abc" }, a.id); + expect(result.failed).toBe(1); + expect(result.synced).toBe(0); + expect(result.conflicts).toBe(0); + + const { getQueuedActions } = await import("./queue"); + const all = await getQueuedActions(); + expect(all).toHaveLength(1); + expect(all[0].status).toBe("failed"); + expect(all[0].error).toBe("network down"); + }); + + it("syncPending leaves queued actions untouched when offline even if the queue is non-empty", async () => { + await enqueueAction("markOrderReady", { orderId: "abc" }); + await enqueueAction("markOrderReady", { orderId: "def" }); + const dispatcher = vi.fn(); + const result = await syncPending(dispatcher, { online: false, sleep: () => Promise.resolve() }); + expect(dispatcher).not.toHaveBeenCalled(); + expect(result).toEqual({ synced: 0, conflicts: 0, failed: 0 }); + + const { getQueuedActions } = await import("./queue"); + const all = await getQueuedActions(); + expect(all).toHaveLength(2); + expect(all.every((x) => x.status === "pending")).toBe(true); + }); }); diff --git a/src/lib/offline/sync.ts b/src/lib/offline/sync.ts index 387e422..65c65ed 100644 --- a/src/lib/offline/sync.ts +++ b/src/lib/offline/sync.ts @@ -27,47 +27,67 @@ export interface SyncResult { failed: number; } +// Re-entrancy guard: concurrent `online` events shouldn't kick off parallel +// sync passes that race on the same `markInFlight`/`markSynced` records. +let inFlightSync: Promise | null = null; + export async function syncPending(dispatcher: Dispatcher, options: SyncOptions): Promise { - const result: SyncResult = { synced: 0, conflicts: 0, failed: 0 }; - if (!options.online) return result; + if (inFlightSync) return inFlightSync; + inFlightSync = (async () => { + const result: SyncResult = { synced: 0, conflicts: 0, failed: 0 }; + if (!options.online) return result; - const actions = await getQueuedActions(); - for (const action of actions) { - if (action.status === "conflict" || action.status === "failed") { - // Don't auto-retry conflicts/failures; require manual user intervention - continue; - } - if (action.attempts >= MAX_ATTEMPTS) { - await markFailed(action.id, "max attempts reached"); - result.failed += 1; - continue; - } + const actions = await getQueuedActions(); + for (const action of actions) { + if (action.status === "conflict" || action.status === "failed") { + // Don't auto-retry conflicts/failures; require manual user intervention + continue; + } + if (action.attempts >= MAX_ATTEMPTS) { + await markFailed(action.id, "max attempts reached"); + result.failed += 1; + continue; + } - const backoff = calculateBackoff(action.attempts); - if (action.lastAttemptAt) { - const elapsed = (options.now?.() ?? Date.now()) - action.lastAttemptAt; - if (elapsed < backoff) continue; // not time yet - } + const backoff = calculateBackoff(action.attempts); + if (action.lastAttemptAt) { + const elapsed = (options.now?.() ?? Date.now()) - action.lastAttemptAt; + if (elapsed < backoff) continue; // not time yet + } - await markInFlight(action.id); - try { - const dispatchResult = await dispatcher(action.actionName, action.payload, action.id); - if (dispatchResult.ok) { - await markSynced(action.id); - result.synced += 1; - } else if ("conflict" in dispatchResult) { - await markConflict(action.id, dispatchResult.conflict); - result.conflicts += 1; - } else { - await markFailed(action.id, dispatchResult.error); + await markInFlight(action.id); + try { + const dispatchResult = await dispatcher(action.actionName, action.payload, action.id); + if (dispatchResult.ok) { + await markSynced(action.id); + result.synced += 1; + } else if ("conflict" in dispatchResult) { + await markConflict(action.id, dispatchResult.conflict); + result.conflicts += 1; + } else if ("error" in dispatchResult) { + await markFailed(action.id, dispatchResult.error); + result.failed += 1; + } + } catch (err) { + // A single poisoned record (IDB closed, quota, etc.) must not abort + // the whole sync pass. Swallow the secondary failure with a warn so + // it stays observable in dev. + const message = err instanceof Error ? err.message : String(err); + try { + await markFailed(action.id, message); + } catch (innerErr) { + console.warn("[offline-sync] markFailed threw for", action.id, innerErr); + } result.failed += 1; } - } catch (err) { - await markFailed(action.id, (err as Error).message); - result.failed += 1; } + return result; + })(); + try { + return await inFlightSync; + } finally { + inFlightSync = null; } - return result; } /**