From df2768d30e968f2c3ffd1b1d79689a2c2c1f7bf8 Mon Sep 17 00:00:00 2001 From: Maurice Date: Thu, 8 Oct 2026 13:47:54 +0200 Subject: [PATCH] fix(client): let a flush called mid-pass resolve with that pass flush() returned at once when this tab already had a flush going, and that includes one still waiting for another tab's Web Lock. The socket's reconnect hook chains a refetch onto flush(), so a reconnect landing in that window rehydrated the trip while a replay was still on its way, the race the lock wait was meant to close. sendSoon and the online, periodic and visibility triggers make such an overlap common. flush() now keeps the running flush as a promise. A second call still asks it for one more pass and then gets that same promise back, so it resolves only once every pass, including the one it asked for, is done. The pass itself moves to _flushPass, and the repeat runs as a loop instead of a nested flush() call. The client CLAUDE.md sync bullet now says so, and loses its dash. --- client/CLAUDE.md | 2 +- client/src/sync/mutationQueue.ts | 65 +++++++++++++------- client/tests/unit/sync/mutationQueue.test.ts | 51 +++++++++++++++ 3 files changed, 95 insertions(+), 23 deletions(-) diff --git a/client/CLAUDE.md b/client/CLAUDE.md index bef0198a8..a897f9478 100644 --- a/client/CLAUDE.md +++ b/client/CLAUDE.md @@ -41,7 +41,7 @@ The layering is **component → feature hook → store/slice → `repo/` → `ap - **`src/repo/`** — per-entity repositories mediating between the network and the offline cache: online, call REST and upsert into Dexie; offline, read Dexie. Always go through a repo for trip data. - **`src/api/client.ts`** — the single Axios instance, typed from `@trek/shared`. New API surface goes in a per-domain `api/.ts`, not the god module. - **`src/db/offlineDb.ts`** — the Dexie schema. Database and table names are the on-device contract — renaming one orphans every user's cache. -- **`src/sync/`** — `mutationQueue.ts`: offline writes do an optimistic Dexie write (temporary **negative id** for creates), then `flush()` replays REST on reconnect with the mutation's UUID as the **`X-Idempotency-Key`** header (matching the server interceptor) and reconciles the row. Writes to one entity keep their order: a write parked as `failed` or `conflict` holds back the later writes to the same `resource` + entity id until the user retries, resolves or discards it, and only that action deletes a parked row (trip eviction never does). Repos whose writes carry no version token (visits, day clears, tours, driving settings) ask `mutationQueue.mustQueue()` before an online write and queue it while an older write to the entity is still queued, so a retried write cannot land over a newer one made here online; places and packing items send `X-Base-Updated-At`, so the server refuses the stale replay with a 409 instead. `flush()` takes the Web Lock `trek-mutation-flush:` so only one tab replays at a time; a tab that finds it taken waits for it, so `flush()` never resolves while another tab is still replaying and a re-seed chained onto it cannot lay older server rows over those writes (the in-tab flag alone where the API is missing). The rest of the directory drives reconnection sync, the auth gate that stops background loops before logout swaps the DB, offline preferences, and map pre-download (raster and vector prefetchers). +- **`src/sync/`**: in `mutationQueue.ts`, offline writes do an optimistic Dexie write (temporary **negative id** for creates), then `flush()` replays REST on reconnect with the mutation's UUID as the **`X-Idempotency-Key`** header (matching the server interceptor) and reconciles the row. Writes to one entity keep their order: a write parked as `failed` or `conflict` holds back the later writes to the same `resource` + entity id until the user retries, resolves or discards it, and only that action deletes a parked row (trip eviction never does). Repos whose writes carry no version token (visits, day clears, tours, driving settings) ask `mutationQueue.mustQueue()` before an online write and queue it while an older write to the entity is still queued, so a retried write cannot land over a newer one made here online; places and packing items send `X-Base-Updated-At`, so the server refuses the stale replay with a 409 instead. `flush()` takes the Web Lock `trek-mutation-flush:` so only one tab replays at a time; a tab that finds it taken waits for it, and a `flush()` called while this tab's flush runs or waits for the lock gets that same flush back (with one more pass asked for), so `flush()` never resolves while a tab is still replaying and a re-seed chained onto it cannot lay older server rows over those writes (the in-tab flush alone where the API is missing). The rest of the directory drives reconnection sync, the auth gate that stops background loops before logout swaps the DB, offline preferences, and map pre-download (raster and vector prefetchers). ## Rules for new code diff --git a/client/src/sync/mutationQueue.ts b/client/src/sync/mutationQueue.ts index ad5f02ba0..e30d3e31b 100644 --- a/client/src/sync/mutationQueue.ts +++ b/client/src/sync/mutationQueue.ts @@ -40,16 +40,19 @@ export function generateUUID(): string { return randomId() } -let _flushing = false +// The flush in progress, from the moment flush() is called until its last pass is +// done, including the time it waits for another tab's lock. A second call gets +// this same promise, so no caller resolves while a pass is still to come. +let _flushRun: Promise | null = null // Set when flush() is called while a pass is already running. That pass read its // pending rows when it began, so a write queued since then waits for one more -// pass, which the running one starts once it is done. +// pass, which the running flush starts once that pass is done. let _flushAgain = false /** * Take the cross-tab flush lock, waiting while another tab holds it. Every open * tab of the account shares the queue and runs its own triggers, and the - * _flushing flag only covers its own tab: two tabs read the same pending rows + * in-progress flush only covers its own tab: two tabs read the same pending rows * and replay each of them twice, leaning on the server's replay cache to answer * the second. Waiting rather than skipping matters to the callers: the online * trigger re-seeds Dexie from the server once flush() resolves, and the socket @@ -276,18 +279,44 @@ export const mutationQueue = { * failed or as a conflict holds back the later writes to the same entity, so * they keep their order whatever the user decides about it. * - * A call that arrives while a pass is running returns at once, but asks that - * pass for one more round: the running pass read the pending rows when it - * began and would not see a write queued since. + * A call that arrives while a flush is running, or still waiting for another + * tab's lock, asks it for one more round (the running pass read the pending + * rows when it began and would not see a write queued since) and resolves with + * it. So whoever refetches once flush() resolves, like the socket's reconnect + * hook, never does so while this tab or another one is still replaying. */ - async flush(): Promise { - if (isEffectivelyOffline() || !isAuthed()) return - if (_flushing) { + flush(): Promise { + if (isEffectivelyOffline() || !isAuthed()) return Promise.resolve() + if (_flushRun) { _flushAgain = true - return + return _flushRun } - _flushing = true - _flushAgain = false + const run = (async () => { + try { + // A "mine wins" auto-resolution dropped its base token; one more pass now + // overwrites the server unconditionally. Bounded: the retried write carries + // no token, so it cannot 409 for the same reason. A flush asked for while a + // pass ran gets its pass here too. Bounded as well: the flag is only set by + // a call made during a pass, and every pass clears it when it starts. + let again = true + while (again) { + _flushAgain = false + const needsRetry = await this._flushPass() + again = (needsRetry || _flushAgain) && !isEffectivelyOffline() && isAuthed() + } + } finally { + _flushRun = null + } + })() + _flushRun = run + return run + }, + + /** + * One pass over the pending rows, under the cross-tab lock. Only flush() calls + * it. Resolves to true when a write was re-queued for another pass. + */ + async _flushPass(): Promise { const release = await acquireFlushLock() // tempId → realId learned during this flush, so a dependent edit/delete // queued against an offline-created entity (still holding the negative id) @@ -528,16 +557,8 @@ export const mutationQueue = { } } finally { release() - _flushing = false - } - // A "mine wins" auto-resolution dropped its base token; one more pass now - // overwrites the server unconditionally. Bounded: the retried write carries - // no token, so it cannot 409 for the same reason. A flush asked for while - // this pass ran gets its pass here too. Bounded as well: the flag is only set - // by a call made during a pass, and every pass clears it when it starts. - if ((needsRetry || _flushAgain) && !isEffectivelyOffline()) { - await this.flush() } + return needsRetry }, /** @@ -647,7 +668,7 @@ export const mutationQueue = { /** Reset internal flushing flag and timestamp counters — useful in tests. */ _resetFlushing(): void { - _flushing = false + _flushRun = null _flushAgain = false _lastTs = 0 _lastTempId = 0 diff --git a/client/tests/unit/sync/mutationQueue.test.ts b/client/tests/unit/sync/mutationQueue.test.ts index 5701587bc..dfd8bbc96 100644 --- a/client/tests/unit/sync/mutationQueue.test.ts +++ b/client/tests/unit/sync/mutationQueue.test.ts @@ -810,6 +810,57 @@ describe('mutationQueue.flush: one tab at a time', () => { expect(await offlineDb.mutationQueue.get(id)).toBeUndefined(); }); + it('a second flush while the first waits for the lock resolves only with it', async () => { + const locks = fakeLocks(); + const releaseOther = locks.holdElsewhere(); + const id = generateUUID(); + await mutationQueue.enqueue(makeMutation({ id })); + let sent = 0; + server.use(http.post('/api/trips/1/places', () => { sent++; return HttpResponse.json({ place: buildPlace({ trip_id: 1, id: 64 }) }); })); + + const first = mutationQueue.flush(); + await settle(); + // The socket's reconnect hook calls in here and refetches once this resolves. + let secondDone = false; + const second = mutationQueue.flush().then(() => { secondDone = true; }); + await settle(); + expect(secondDone).toBe(false); + + releaseOther(); + await Promise.all([first, second]); + + expect(secondDone).toBe(true); + expect(sent).toBe(1); + // One lock request for the waiting pass and one for the extra pass it was asked for. + expect(locks.request).toHaveBeenCalledTimes(3); + }); + + it('a write queued during a pass goes out before the second caller resolves', async () => { + const first = generateUUID(); + const later = generateUUID(); + await mutationQueue.enqueue(makeMutation({ id: first })); + const sent: string[] = []; + let second: Promise | undefined; + server.use(http.post('/api/trips/1/places', async ({ request }) => { + const { name } = await request.json() as { name: string }; + sent.push(name); + if (name === 'Eiffel Tower') { + await mutationQueue.enqueue(makeMutation({ id: later, body: { name: 'Louvre' } })); + // What had been sent at the moment this second call resolved. + second = mutationQueue.flush().then(() => [...sent]); + } + return HttpResponse.json({ place: buildPlace({ trip_id: 1, id: sent.length + 70 }) }); + })); + + const firstRun = mutationQueue.flush(); + await vi.waitFor(() => expect(second).toBeDefined()); + + // Resolved only after the extra pass sent the later write, not when it was asked for. + expect(await second).toEqual(['Eiffel Tower', 'Louvre']); + expect(await offlineDb.mutationQueue.count()).toBe(0); + await firstRun; + }); + it('flushes under the flag alone when the lock request is refused', async () => { Object.defineProperty(navigator, 'locks', { value: { request: vi.fn().mockRejectedValue(new DOMException('denied', 'SecurityError')) },