diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 35a4d8d2b..756b272ff 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -197,11 +197,13 @@ settings and guards read them in every build. `electron`, `server`, `indexeddb`). The store caches full documents by ETag and uploads a document whose payload it served unchanged with `keepPayload=1`, so a favorite or a rename does not resend a large M3U. - Such a write carries `If-Match` with the version it was read at: when + A mutation carries `If-Match` with the version it was read at (of its + payload, else of the playlist's last read, as for portal documents): when another device wrote in between, the server answers `412`, the store throws `PlaylistWriteConflict`, and `PlaylistsService`'s per-playlist write queue runs the mutation again on a fresh read (a retry on the same, - unread copy writes unconditionally, as before). After a write the store + unread copy writes unconditionally, as before). Imports and start-up + migrations (`dbUpsertAppPlaylists`) write unconditionally. After a write the store keeps its copy only when `X-Replaced-ETag` names the version its payload came from. - **Xtream user data.** `USER_DATA_STORAGE` becomes `ServerUserDataStorage`: @@ -216,9 +218,11 @@ settings and guards read them in every build. every unconfirmed write (queued or travelling) is saved in `localStorage` per page load (`iptvnator-server:unsent-writes::`). The tab remembers its page load in `sessionStorage`, so a reload takes its - writes over and sends them first; the writes of a tab closed before it + writes over and sends them first (the old copy goes only once the new one + is saved); the writes of a tab closed before it could send them are queued, ahead of its own, by the next re-read in - another tab once that tab's record has gone stale. Tabs never clear each other's. + another tab once that tab's record has gone stale; the result replays the + closed tab's edits first and this tab's newer ones last. Tabs never clear each other's. Cache and saved writes belong to one server database (`iptvnator-server:cache-owner`); another one behind the same address starts clean. Also, diff --git a/docs/superpowers/specs/2026-10-08-server-storage-design.md b/docs/superpowers/specs/2026-10-08-server-storage-design.md index b02b5bbb2..7ab5df6e1 100644 --- a/docs/superpowers/specs/2026-10-08-server-storage-design.md +++ b/docs/superpowers/specs/2026-10-08-server-storage-design.md @@ -209,11 +209,10 @@ https://claude.ai/artifact/3WheJAgeNL5ag6kmtPrJcD ## 11. Risks -- **Concurrent edits on two devices**: a metadata edit (`keepPayload`) is - sent with `If-Match` on the version it was read at; on a `412` the - per-playlist write queue runs the mutation again on a fresh read. Other - document writes (refresh, import) stay last write wins; every playlist - mutation re-reads first, so the window is one round trip. +- **Concurrent edits on two devices**: a playlist mutation is sent with + `If-Match` on the version it was read at; on a `412` the per-playlist + write queue runs the mutation again on a fresh read. Imports and start-up + migrations stay last write wins. Xtream map entries are written on the tag they were read at and replayed on the server's value after a `412`, so both devices' edits are kept. - **Large payloads**: M3U global favorites and backup export read all full 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 ebe7523ea..9a56c4959 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 @@ -165,6 +165,59 @@ describe('ServerPlaylistDocumentStore', () => { expect(api.sendJson.mock.calls[1][3]).toEqual({}); }); + it('writes a portal document on the version it read', async () => { + const { api, store } = setup(); + const portal = { _id: 'p2', title: 'Portal', macAddress: 'm' }; + api.request.mockResolvedValueOnce(fakeResponse(200, portal, '"b.0.1"')); + const served = await store.dbGetAppPlaylist('p2'); + if (!served) throw new Error('expected a document'); + api.sendJson.mockResolvedValueOnce(fakeResponse(200, {}, '"b.0.2"')); + + await store.dbUpsertAppPlaylist({ ...served, title: 'Renamed' }); + + expect(api.sendJson.mock.calls[0][1]).toBe('/playlists/p2'); + expect(api.sendJson.mock.calls[0][3]).toEqual({ + 'If-Match': '"b.0.1"', + }); + }); + + it("keeps checking a playlist after another one's refusal", async () => { + const { api, store } = setup(); + // Both playlists are at revision 1 of the same boot: one ETag. + for (const id of ['p1', 'p2']) { + api.request.mockResolvedValueOnce( + fakeResponse(200, { ...doc, _id: id }, '"b.0.1"') + ); + } + const first = await store.dbGetAppPlaylist('p1'); + const second = await store.dbGetAppPlaylist('p2'); + if (!first || !second) throw new Error('expected documents'); + api.sendJson.mockRejectedValueOnce( + new ServerApiError(412, 'conflict', 'changed') + ); + await expect( + store.dbUpsertAppPlaylist({ ...first, title: 'A' }) + ).rejects.toBeInstanceOf(PlaylistWriteConflict); + + api.sendJson.mockResolvedValueOnce(putResponse('"b.0.2"', '"b.0.1"')); + await store.dbUpsertAppPlaylist({ ...second, title: 'B' }); + + expect(api.sendJson.mock.calls[1][3]).toEqual({ + 'If-Match': '"b.0.1"', + }); + }); + + it('writes imports as they are', async () => { + const { api, store } = setup(); + api.request.mockResolvedValueOnce(fakeResponse(200, doc, '"b.0.1"')); + await store.dbGetAppPlaylist('p1'); + api.sendJson.mockResolvedValue(fakeResponse(200, {}, '"b.0.2"')); + + await store.dbUpsertAppPlaylists([{ ...doc, title: 'Imported' }]); + + expect(api.sendJson.mock.calls[0][3]).toEqual({}); + }); + it('runs deletes through the user-data hold', async () => { const order: string[] = []; const api = { 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 2b9cc64c6..aff132337 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 @@ -30,7 +30,9 @@ type ServerApi = Pick< */ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { private readonly cache = new Map(); - /** Versions a write was refused on (another device wrote since). */ + /** The version each playlist was last read or written at. */ + private readonly basis = new Map(); + /** `id|version` a write was refused on (another device wrote since). */ private readonly refused = new Set(); /** Served payload object → the ETag of the document it came from. */ private readonly servedPayloads = new WeakMap(); @@ -66,6 +68,7 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { } catch (error) { if (error instanceof ServerApiError && error.status === 404) { this.cache.delete(playlistId); + this.basis.delete(playlistId); return null; } throw error; @@ -79,58 +82,17 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { return this.serve(document, etag); } - async dbUpsertAppPlaylist( - playlist: Playlist - ): Promise { - const path = `/playlists/${encodeURIComponent(playlist._id)}`; - const payloadEtag = this.servedPayloadEtag(playlist); - if (payloadEtag !== null) { - try { - const response = await this.api.sendJson( - 'PUT', - `${path}?keepPayload=1`, - withoutPayload(playlist), - // Built on the version it was read at; a retry on the - // same copy, not read again, writes as before. - this.refused.has(payloadEtag) - ? {} - : { 'If-Match': payloadEtag } - ); - // The server kept its stored payload; it is ours only if - // nothing replaced it since we read it. - const replaced = response.headers.get( - SERVER_REPLACED_ETAG_HEADER - ); - if (replaced && sameEtag(replaced, payloadEtag)) { - this.remember(playlist._id, response, playlist); - } else { - this.cache.delete(playlist._id); - } - return { success: true }; - } catch (error) { - if (error instanceof ServerApiError && error.status === 412) { - this.refused.add(payloadEtag); - this.cache.delete(playlist._id); - throw new PlaylistWriteConflict(playlist._id); - } - if (!( - error instanceof ServerApiError && - error.code === 'payload_missing' - )) { - throw error; - } - } - } - const response = await this.api.sendJson('PUT', path, playlist); - this.remember(playlist._id, response, playlist); - return { success: true }; + /** A mutation: built on the version it read, refused if it changed. */ + dbUpsertAppPlaylist(playlist: Playlist): Promise { + return this.upsert(playlist, true); } + /** Imports and migrations: written as they are, outside any retry. */ async dbUpsertAppPlaylists( playlists: Playlist[] ): Promise { for (const playlist of playlists) { - await this.dbUpsertAppPlaylist(playlist); + await this.upsert(playlist, false); } return { success: true, count: playlists.length }; } @@ -142,6 +104,7 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { }) ); this.cache.delete(playlistId); + this.basis.delete(playlistId); return { success: true }; } @@ -150,6 +113,7 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { this.api.request('/playlists', { method: 'DELETE' }) ); this.cache.clear(); + this.basis.clear(); return { success: true }; } @@ -174,6 +138,86 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { return { success: true }; } + private async upsert( + playlist: Playlist, + conditional: boolean + ): Promise { + const id = playlist._id; + const path = `/playlists/${encodeURIComponent(id)}`; + const payloadEtag = this.servedPayloadEtag(playlist); + // The version it was read at: of its payload, else of the last read + // (portal documents have no payload). A retry on the same copy, not + // read again after a refusal, writes as before. + const read = payloadEtag ?? this.basis.get(id) ?? null; + const headers: Record = + conditional && read && !this.refused.has(`${id}|${read}`) + ? { 'If-Match': read } + : {}; + try { + if (payloadEtag !== null) { + const response = await this.keepPayload( + playlist, + payloadEtag, + headers + ); + if (response) { + return { success: true }; + } + } + const response = await this.api.sendJson( + 'PUT', + path, + playlist, + payloadEtag === null ? headers : {} + ); + this.remember(id, response, playlist); + return { success: true }; + } catch (error) { + if (error instanceof ServerApiError && error.status === 412) { + this.refused.add(`${id}|${read}`); + this.cache.delete(id); + this.basis.delete(id); + throw new PlaylistWriteConflict(id); + } + throw error; + } + } + + /** `null` when the server has no payload to keep (write it in full). */ + private async keepPayload( + playlist: Playlist, + payloadEtag: string, + headers: Record + ): Promise { + let response: Response; + try { + response = await this.api.sendJson( + 'PUT', + `/playlists/${encodeURIComponent(playlist._id)}?keepPayload=1`, + withoutPayload(playlist), + headers + ); + } catch (error) { + if ( + error instanceof ServerApiError && + error.code === 'payload_missing' + ) { + return null; + } + throw error; + } + // The server kept its stored payload; it is ours only if nothing + // replaced it since we read it. + const replaced = response.headers.get(SERVER_REPLACED_ETAG_HEADER); + if (replaced && sameEtag(replaced, payloadEtag)) { + this.remember(playlist._id, response, playlist); + } else { + this.cache.delete(playlist._id); + this.noteWrite(playlist._id, response); + } + return response; + } + private remember( id: string, response: Response, @@ -185,14 +229,26 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { } else { this.cache.delete(id); } + this.noteWrite(id, response); return etag; } + /** The version a read or write left the playlist at. */ + private noteWrite(id: string, response: Response): void { + const etag = response.headers.get('etag'); + if (etag) { + this.basis.set(id, etag); + } else { + this.basis.delete(id); + } + } + private serve(document: Playlist, etag: string | null): Playlist { const copy = structuredClone(document); if (!etag) { return copy; } + this.basis.set(document._id, etag); const record = copy as unknown as Record; for (const field of SERVER_PAYLOAD_FIELDS) { const value = record[field]; diff --git a/libs/services/src/lib/server-storage/server-unsent-writes.ts b/libs/services/src/lib/server-storage/server-unsent-writes.ts index 7506b09e6..42a9a9a0d 100644 --- a/libs/services/src/lib/server-storage/server-unsent-writes.ts +++ b/libs/services/src/lib/server-storage/server-unsent-writes.ts @@ -186,11 +186,11 @@ export function saveUnsentWrites( instanceId: string, page: string, writes: readonly UnsentWrite[] -): void { +): boolean { try { if (writes.length === 0) { local.removeItem(writesKey(instanceId, page)); - return; + return true; } const stored: StoredWrite[] = writes.map( ({ namespace, playlistId, value, tag, steps }) => ({ @@ -202,8 +202,10 @@ export function saveUnsentWrites( }) ); local.setItem(writesKey(instanceId, page), JSON.stringify(stored)); + return true; } catch { // Storage unavailable or full: the queue stays in memory only. + return false; } } @@ -267,11 +269,11 @@ export function saveDeferredEdits( instanceId: string, page: string, edits: readonly SavedDeferredEdit[] -): void { +): boolean { try { if (edits.length === 0) { local.removeItem(deferredKey(instanceId, page)); - return; + return true; } local.setItem( deferredKey(instanceId, page), @@ -283,8 +285,10 @@ export function saveDeferredEdits( })) ) ); + return true; } catch { // Storage unavailable or full: the edits stay in memory only. + return false; } } 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 80aa4b961..5e2aba41e 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 @@ -187,19 +187,38 @@ export class UserDataWriteQueue { adopt( namespace: ServerUserStateNamespace, playlistId: string, - value: unknown, + current: unknown, steps: readonly EditStep[], foreign: boolean - ): void { - this.queue(namespace, playlistId, value, value, { foreign }); + ): unknown { const key = `${namespace}|${playlistId}`; + const adopted = uncertain(steps); + const queued = this.pending.get(key); + if (queued) { + // This tab's queued edits are newer: they replay last. + const all = [...adopted, ...queued.steps]; + const value = replaySteps(queued.steps[0]?.before, all); + const edit = ++this.editCount; + this.pending.set(key, { + ...queued, + value, + edit, + steps: all, + tag: foreign ? UNKNOWN_TAG : queued.tag, + }); + this.edits.set(key, edit); + return value; + } + const value = replaySteps(current, steps); + this.queue(namespace, playlistId, value, value, { foreign }); const entry = this.pending.get(key); if (entry) { this.pending.set(key, { ...entry, - steps: [...uncertain(steps), ...entry.steps], + steps: [...adopted, ...entry.steps], }); } + return value; } scheduleFlush(): void { 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 ba9a4289a..5a29d8adb 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 @@ -17,7 +17,6 @@ import { applyEdit, type DeferredEdit, refreshedNamespace, - replaySteps, } from './server-user-data-edit'; import { UserDataWriteQueue } from './server-user-data-queue'; import { @@ -58,6 +57,8 @@ export class ServerUserDataStorage implements UserDataStorage { */ private readonly deferred = new Map(); private hydrated = false; + /** Closed tabs' page loads this one took over (once each). */ + private readonly adopted = new Set(); /** This page load: names its saved writes and its registry record. */ private readonly tabId = Math.random().toString(36).slice(2); private readonly cache: UserDataCache; @@ -124,11 +125,13 @@ export class ServerUserDataStorage implements UserDataStorage { )) { this.deferred.set(`${edit.namespace}|${edit.playlistId}`, edit); } - // Saved under this page load before the old one is forgotten. - this.saveDeferred(); - this.saveUnsent(); + // Saved under this page load before the old one is forgotten; if + // storage refuses the copy, the old one stays. + const saved = [this.saveDeferred(), this.saveUnsent()].every(Boolean); this.publishUnconfirmed(); - removePage(this.local, this.instanceId, page); + if (saved) { + removePage(this.local, this.instanceId, page); + } } /** @@ -140,7 +143,7 @@ export class ServerUserDataStorage implements UserDataStorage { this.local, this.instanceId, this.tabId - )) { + ).filter((page) => !this.adopted.has(page))) { const edits = [ ...readDeferredEdits(this.local, this.instanceId, page), ...readUnsentWrites(this.local, this.instanceId, page), @@ -151,23 +154,22 @@ export class ServerUserDataStorage implements UserDataStorage { this.tabId ); for (const { namespace, playlistId, steps } of edits) { - const value = replaySteps( - this.cache.map(namespace)[playlistId], - steps - ); - this.cache.setEntry(namespace, playlistId, value); // Queued even when the cache shows it already. - this.writes.adopt( + const value = this.writes.adopt( namespace, playlistId, - value, + this.cache.map(namespace)[playlistId], steps, foreign.has(`${namespace}|${playlistId}`) ); + this.cache.setEntry(namespace, playlistId, value); + } + this.adopted.add(page); + // The closed tab's copy goes only once this tab saved its own. + if (this.saveUnsent()) { + removePage(this.local, this.instanceId, page); } - // Saved here before the closed tab's copy is forgotten. this.writes.scheduleFlush(); - removePage(this.local, this.instanceId, page); } } @@ -423,19 +425,26 @@ export class ServerUserDataStorage implements UserDataStorage { } /** Persists what the server has not confirmed; skips idle rewrites. */ - private saveUnsent(): void { + /** False when browser storage refused the copy. */ + private saveUnsent(): boolean { // A newer edit can be queued while an older one is still travelling. const writes = this.writes.unsentWrites(); if (writes.length === 0 && !this.unsentSaved) { - return; + return true; } - saveUnsentWrites(this.local, this.instanceId, this.tabId, writes); - this.unsentSaved = writes.length > 0; + const saved = saveUnsentWrites( + this.local, + this.instanceId, + this.tabId, + writes + ); + this.unsentSaved = writes.length > 0 || !saved; this.publishUnconfirmed(); + return saved; } - private saveDeferred(): void { - saveDeferredEdits(this.local, this.instanceId, this.tabId, [ + private saveDeferred(): boolean { + return saveDeferredEdits(this.local, this.instanceId, this.tabId, [ ...this.deferred.values(), ]); } 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 c54e62d65..ca92b6542 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 @@ -717,4 +717,109 @@ describe('ServerUserDataStorage with other devices', () => { }); expect(favorites(tabA)).toEqual({ p1: ['a', 'A1'], p2: ['b', 'B2'] }); }); + + it("keeps this tab's newer progress over a closed tab's older one", async () => { + const at = (seconds: number) => [ + { + contentType: 'vod', + contentXtreamId: 1, + positionSeconds: seconds, + }, + ]; + const server = fakeServer({ + 'xtream-playback-positions': { p1: at(10) }, + }); + const shared = memoryStorage(); + const offline = () => + Promise.reject(new ServerApiError(0, 'network', 'offline')); + const clock = jest.spyOn(Date, 'now').mockReturnValue(1_000); + const closed = new ServerUserDataStorage( + { ...server.api, sendJson: jest.fn(offline) } as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + await closed.hydrate(); + closed.setItem( + 'xtream-playback-positions', + JSON.stringify({ p1: at(20) }) + ); + await closed.flush(); + + // This tab starts while the server cannot be read: its edit waits. + let readable = false; + const tab = new ServerUserDataStorage( + { + ...server.api, + getJson: jest.fn(() => + readable ? server.api.getJson() : offline() + ), + } as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + await expect(tab.hydrate()).rejects.toBeInstanceOf(ServerApiError); + tab.setItem( + 'xtream-playback-positions', + JSON.stringify({ p1: at(30) }) + ); + // Later the closed tab's record is stale and the server answers. + clock.mockReturnValue(1_000 + UNCONFIRMED_TTL_MS + 1); + readable = true; + await tab.hydrate(); + await tab.flush(); + await tab.flush(); + clock.mockRestore(); + + expect(server.state['xtream-playback-positions']['p1']).toEqual(at(30)); + }); + + it("keeps the last page load's saved writes when storage refuses the copy", async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const local = memoryStorage(); + const tabStore = memoryStorage(); + const offline = { + ...server.api, + sendJson: jest + .fn() + .mockRejectedValue(new ServerApiError(0, 'network', 'offline')), + }; + const before = new ServerUserDataStorage( + offline as never, + local, + jest.fn(), + 'default', + tabStore + ); + await before.hydrate(); + before.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + await before.flush(); + const saved = () => + [...local.values.keys()].filter((key) => + key.startsWith('iptvnator-server:unsent-writes:') + ); + expect(saved()).toHaveLength(1); + + // Storage is full now: the reloaded page cannot save its copy. + const setItem = local.setItem; + local.setItem = (key: string, value: string) => { + if (key.startsWith('iptvnator-server:unsent-writes:')) { + throw new DOMException('full', 'QuotaExceededError'); + } + setItem(key, value); + }; + const reloaded = new ServerUserDataStorage( + offline as never, + local, + jest.fn(), + 'default', + tabStore + ); + + expect(reloaded.hasUnsentWrites()).toBe(true); + expect(saved()).toHaveLength(1); + }); });