From a8b6b86e43d93c856f7ea0fc0077030160130378 Mon Sep 17 00:00:00 2001 From: 4gray Date: Mon, 12 Oct 2026 09:10:48 +0200 Subject: [PATCH] fix(server-storage): keep tags with their values, rebase page-hide writes, hold deletes - A re-read that keeps an entry from the cache (another tab's or this tab's unconfirmed value) no longer takes the server's tag for it, so the next edit conflicts and is replayed instead of overwriting another device's change. - Page-hide writes behind a travelling one are taken from the queue when it is answered, so its new version becomes their baseline. - A playlist delete holds that playlist's user-data edits instead of discarding them up front; only a successful delete drops them, a failed one sends them as before. Co-Authored-By: Claude Opus 5.5 --- docs/architecture/pwa-self-hosted.md | 7 +- .../server-playlist-document-store.spec.ts | 14 ++- .../server-playlist-document-store.ts | 25 ++-- .../server-storage/server-storage-session.ts | 2 +- .../server-storage/server-user-data-queue.ts | 87 ++++++++----- .../server-user-data-storage.ts | 37 +++++- .../server-user-data-sync.spec.ts | 116 +++++++++++++++++- 7 files changed, 231 insertions(+), 57 deletions(-) diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 4ba2fddd4..b69eeb347 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -215,8 +215,11 @@ settings and guards read them in every build. as offline. Each write is sent on the tag of the entry it was made against (`UserDataWriteQueue`); after a `412` the edit (before → after) is replayed on the server's current value, so another device's change to the - same playlist's entry is kept. Deleting a playlist first drops its queued - edits and waits for writes still travelling. Edits made before the first successful read + same playlist's entry is kept. An entry a re-read keeps from the cache + gets no tag, so its next write conflicts and is replayed too. A page-hide + write behind a travelling one is taken from the queue only when that one + is answered. While a playlist is being deleted its edits wait; the + delete drops them only when it succeeds. Edits made before the first successful read are held as before → after and applied to the server's value once it arrives (`applyEdit`): a new device or a stale cache adds its favorite instead of replacing the server's list. They are saved in the tab too diff --git a/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts b/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts index 8a3c034b2..307ce05bb 100644 --- a/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts +++ b/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts @@ -142,7 +142,7 @@ describe('ServerPlaylistDocumentStore', () => { }); }); - it('clears user-data writes before deleting on the server', async () => { + it('runs deletes through the user-data hold', async () => { const order: string[] = []; const api = { request: jest.fn(async (path: string) => { @@ -155,17 +155,23 @@ describe('ServerPlaylistDocumentStore', () => { }; const store = new ServerPlaylistDocumentStore( api as never, - async (playlistId) => void order.push(`discard ${playlistId}`) + async (playlistId, remove) => { + order.push(`hold ${playlistId}`); + await remove(); + order.push(`release ${playlistId}`); + } ); await store.dbDeletePlaylist('p1'); await store.dbDeleteAllPlaylists(); expect(order).toEqual([ - 'discard p1', + 'hold p1', 'DELETE /playlists/p1', - 'discard undefined', + 'release p1', + 'hold undefined', 'DELETE /playlists', + 'release undefined', ]); }); }); diff --git a/libs/services/src/lib/server-storage/server-playlist-document-store.ts b/libs/services/src/lib/server-storage/server-playlist-document-store.ts index 076260ec1..e0bd5975d 100644 --- a/libs/services/src/lib/server-storage/server-playlist-document-store.ts +++ b/libs/services/src/lib/server-storage/server-playlist-document-store.ts @@ -33,12 +33,13 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { constructor( private readonly api: ServerApi, /** - * Runs before a playlist (or, without an id, every playlist) is - * deleted on the server, so no user-data write lands afterwards. + * Runs the server delete of a playlist (without an id, of every + * playlist), so user-data writes cannot land after it. */ - private readonly beforeDelete: ( - playlistId?: string - ) => Promise = async () => undefined + private readonly deleting: ( + playlistId: string | undefined, + remove: () => Promise + ) => Promise = async (_, remove) => void (await remove()) ) {} dbGetAppPlaylists(): Promise { @@ -120,17 +121,19 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { } async dbDeletePlaylist(playlistId: string): Promise { - await this.beforeDelete(playlistId); - await this.api.request(`/playlists/${encodeURIComponent(playlistId)}`, { - method: 'DELETE', - }); + await this.deleting(playlistId, () => + this.api.request(`/playlists/${encodeURIComponent(playlistId)}`, { + method: 'DELETE', + }) + ); this.cache.delete(playlistId); return { success: true }; } async dbDeleteAllPlaylists(): Promise { - await this.beforeDelete(); - await this.api.request('/playlists', { method: 'DELETE' }); + await this.deleting(undefined, () => + this.api.request('/playlists', { method: 'DELETE' }) + ); this.cache.clear(); return { success: true }; } diff --git a/libs/services/src/lib/server-storage/server-storage-session.ts b/libs/services/src/lib/server-storage/server-storage-session.ts index b741bfb29..222a67485 100644 --- a/libs/services/src/lib/server-storage/server-storage-session.ts +++ b/libs/services/src/lib/server-storage/server-storage-session.ts @@ -42,7 +42,7 @@ export class ServerStorageSession { }); readonly documentStore = new ServerPlaylistDocumentStore( this.api, - (playlistId) => this.userData.discard(playlistId) + (playlistId, remove) => this.userData.deleting(playlistId, remove) ); readonly userData = new ServerUserDataStorage( this.api, diff --git a/libs/services/src/lib/server-storage/server-user-data-queue.ts b/libs/services/src/lib/server-storage/server-user-data-queue.ts index c346dfff1..6a889a5d1 100644 --- a/libs/services/src/lib/server-storage/server-user-data-queue.ts +++ b/libs/services/src/lib/server-storage/server-user-data-queue.ts @@ -55,6 +55,8 @@ export class UserDataWriteQueue { /** Edit counter per `namespace|playlistId`, for re-reads to keep them. */ private readonly edits = new Map(); private editCount = 0; + /** Entries a delete is waiting on: they stay queued and are not sent. */ + private readonly held: ((key: string) => boolean)[] = []; constructor( private readonly api: ServerApi, @@ -182,29 +184,18 @@ export class UserDataWriteQueue { * later writes wait for them in turn. */ flushWithKeepalive(): void { - const entries = this.takePending(); - if (entries.length === 0) { - // Writes still travelling may be cut off with the page. - this.hooks.changed(); + // Saved first: the page may be gone before anything is answered. + this.hooks.changed(); + if (this.pending.size === 0) { return; } - const sendAll = async (): Promise => { - this.writesInFlight += 1; - try { - await Promise.all( - entries.map((entry) => this.sendKeepalive(entry)) - ); - } finally { - this.writesInFlight -= 1; - } - }; - for (const entry of entries) { - this.inFlight.set(entryKey(entry), entry); - } - this.hooks.changed(); - // Idle: send synchronously, while the hiding page still runs. + // Idle: send synchronously, while the hiding page still runs. Behind + // a travelling write the entries stay queued until it is answered, + // so the version it created can still become their baseline. const run = - this.writesInFlight === 0 ? sendAll() : this.sending.then(sendAll); + this.writesInFlight === 0 + ? this.sendKeepaliveBatch() + : this.sending.then(() => this.sendKeepaliveBatch()); this.sending = run.then( () => undefined, () => undefined @@ -212,14 +203,27 @@ export class UserDataWriteQueue { } /** - * Drops queued edits whose key `matches` and forgets their edit numbers, - * so a travelling write that fails is not retried; resolves once the - * travelling writes have landed. Without a playlist id, tags go too. + * Keeps the entries whose key `matches` queued but unsent until the + * returned release is called (a delete is running). */ - async discard( - matches: (key: string) => boolean, - playlistId?: string - ): Promise { + hold(matches: (key: string) => boolean): () => void { + this.held.push(matches); + return () => { + this.held.splice(this.held.indexOf(matches), 1); + this.scheduleFlush(); + }; + } + + /** Resolves once the writes travelling now have been answered. */ + whenSent(): Promise { + return this.sending; + } + + /** + * After a delete: forgets the queued edits whose key `matches`, their + * edit numbers and their tags (all tags without a playlist id). + */ + drop(matches: (key: string) => boolean, playlistId?: string): void { for (const store of [this.pending, this.edits]) { for (const key of [...store.keys()]) { if (matches(key)) { @@ -227,7 +231,6 @@ export class UserDataWriteQueue { } } } - await this.sending; for (const [namespace, tags] of this.tags) { if (playlistId === undefined) { this.tags.set(namespace, {}); @@ -268,8 +271,13 @@ export class UserDataWriteQueue { clearTimeout(this.flushTimer); this.flushTimer = null; } - const batch = [...this.pending.values()]; - this.pending.clear(); + const batch: PendingEntry[] = []; + for (const [key, entry] of this.pending) { + if (!this.held.some((matches) => matches(key))) { + batch.push(entry); + this.pending.delete(key); + } + } return batch; } @@ -292,6 +300,25 @@ export class UserDataWriteQueue { } } + private async sendKeepaliveBatch(): Promise { + const entries = this.takePending(); + if (entries.length === 0) { + return; + } + for (const entry of entries) { + this.inFlight.set(entryKey(entry), entry); + } + this.hooks.changed(); + this.writesInFlight += 1; + try { + await Promise.all( + entries.map((entry) => this.sendKeepalive(entry)) + ); + } finally { + this.writesInFlight -= 1; + } + } + private async sendKeepalive(entry: PendingEntry): Promise { const path = entryPath(entry); const headers = preconditionHeaders(entry.tag); diff --git a/libs/services/src/lib/server-storage/server-user-data-storage.ts b/libs/services/src/lib/server-storage/server-user-data-storage.ts index 2973f94e7..9d2a1fc5f 100644 --- a/libs/services/src/lib/server-storage/server-user-data-storage.ts +++ b/libs/services/src/lib/server-storage/server-user-data-storage.ts @@ -145,13 +145,32 @@ export class ServerUserDataStorage implements UserDataStorage { } /** - * Before a playlist (or every playlist) is deleted on the server: drop - * its queued and early edits and its cache, and let writes already - * travelling land first, so none recreates deleted data afterwards. + * Runs the server delete of a playlist (without an id, of every + * playlist). Its user-data edits wait meanwhile and writes already + * travelling land first, so none recreates deleted data; a delete that + * succeeded drops its queued and early edits and its cache, a failed + * one leaves them to be sent as before. */ - async discard(playlistId?: string): Promise { + async deleting( + playlistId: string | undefined, + remove: () => Promise + ): Promise { const matches = (key: string) => playlistId === undefined || splitEditKey(key)[1] === playlistId; + const release = this.writes.hold(matches); + try { + await this.writes.whenSent(); + await remove(); + this.forget(matches, playlistId); + } finally { + release(); + } + } + + private forget( + matches: (key: string) => boolean, + playlistId: string | undefined + ): void { for (const key of [...this.deferred.keys()]) { if (matches(key)) { this.deferred.delete(key); @@ -160,7 +179,7 @@ export class ServerUserDataStorage implements UserDataStorage { saveDeferredEdits(this.tabStore, this.instanceId, [ ...this.deferred.values(), ]); - await this.writes.discard(matches, playlistId); + this.writes.drop(matches, playlistId); for (const namespace of SERVER_USER_STATE_NAMESPACES) { const map = this.baseline(namespace); if (playlistId === undefined) { @@ -208,7 +227,13 @@ export class ServerUserDataStorage implements UserDataStorage { // so callers and the diff baseline see the same map. this.writeCache(namespace, JSON.stringify(value)); if (tags) { - this.writes.setTags(namespace, tags[namespace] ?? {}); + // A kept value is not what the server's tag describes: its + // next write must conflict and be replayed on the server's. + const fresh = { ...tags[namespace] }; + for (const playlistId of keepCached) { + delete fresh[playlistId]; + } + this.writes.setTags(namespace, fresh); } } this.writes.settledUpTo(editsBefore); diff --git a/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts b/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts index c1c4cbc22..514498173 100644 --- a/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts +++ b/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts @@ -76,7 +76,21 @@ function fakeServer(initial: Record> = {}) { write(path, init.headers ?? {}) ), readJson: jest.fn((response: Response) => response.json()), - sendKeepalive: jest.fn().mockResolvedValue(true), + sendKeepalive: jest.fn( + async ( + _method: string, + path: string, + body?: { value: unknown }, + headers: Record = {} + ) => { + try { + write(path, headers, body?.value); + return true; + } catch { + return false; + } + } + ), }; return { state, api }; } @@ -211,7 +225,7 @@ describe('ServerUserDataStorage with other devices', () => { ]); }); - it('drops queued edits and cache before a playlist is deleted', async () => { + it('drops queued edits and cache once a playlist is deleted', async () => { const server = fakeServer({ 'xtream-favorites': { p1: ['a'] } }); const phone = new ServerUserDataStorage( server.api as never, @@ -221,11 +235,107 @@ describe('ServerUserDataStorage with other devices', () => { await phone.hydrate(); phone.setItem('xtream-favorites', JSON.stringify({ p1: ['a', 'b'] })); - await phone.discard('p1'); + await phone.deleting('p1', async () => { + await phone.flush(); + delete server.state['xtream-favorites']['p1']; + }); await phone.flush(); expect(server.api.sendJson).not.toHaveBeenCalled(); expect(favorites(phone)).toEqual({}); expect(phone.hasUnsentWrites()).toBe(false); }); + + it('keeps the edits of a playlist whose delete failed', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['a'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['a', 'b'] })); + + await expect( + phone.deleting('p1', async () => { + throw new Error('offline'); + }) + ).rejects.toThrow('offline'); + await phone.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual(['a', 'b']); + expect(favorites(phone)).toEqual({ p1: ['a', 'b'] }); + }); + + it("re-reads another tab's unsent value without the server's tag", async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const shared = memoryStorage(); + const tab = () => + new ServerUserDataStorage( + server.api as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + const tabA = tab(); + const tabB = tab(); + await tabA.hydrate(); + await tabB.hydrate(); + // Tab A's favorite is not sent yet when another device saves one. + tabA.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + server.state['xtream-favorites']['p1'] = ['S', 'Z']; + + await tabB.hydrate(); + expect(favorites(tabB)).toEqual({ p1: ['S', 'X'] }); + tabB.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['S', 'X', 'Y'] }) + ); + // Each conflict replays the edit on the newest value: the two tabs + // settle after a few rounds. + for (let round = 0; round < 4; round += 1) { + await tabB.flush(); + await tabA.flush(); + } + + expect(tabA.hasUnsentWrites() || tabB.hasUnsentWrites()).toBe(false); + expect( + [...(server.state['xtream-favorites']['p1'] as string[])].sort() + ).toEqual(['S', 'X', 'Y', 'Z']); + }); + + it('sends a page-hide edit on the version a travelling write created', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + + let answer: () => void = () => undefined; + const send = server.api.sendJson.getMockImplementation(); + if (!send) { + throw new Error('The fake server answers sendJson'); + } + server.api.sendJson.mockImplementationOnce( + (...args: Parameters) => + new Promise( + (resolve) => (answer = () => resolve(send(...args))) + ) + ); + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + const first = phone.flush(); + await new Promise((resolve) => setTimeout(resolve, 0)); + // X is removed again while its addition is travelling; the page hides. + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] })); + phone.flushWithKeepalive(); + answer(); + await first; + await phone.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S']); + expect(phone.hasUnsentWrites()).toBe(false); + }); });