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); + }); });