From 7afb9758f4b3ac994d774ee86b99439dff74688d Mon Sep 17 00:00:00 2001 From: 4gray Date: Mon, 12 Oct 2026 21:17:18 +0200 Subject: [PATCH] fix(server-storage): conditional playlist metadata writes, retried on a fresh read A favorite or rename (keepPayload PUT) is sent with If-Match on the version it was read at. The server answers 412 when another device wrote since; the store throws PlaylistWriteConflict and PlaylistsService's per-playlist write queue runs the mutation again, which reads the playlist afresh, so two devices' metadata edits both survive. A retry on the same unread copy writes unconditionally, as before; refresh and import writes stay last write wins. Co-Authored-By: Claude Opus 5.5 --- .../src/app/storage/playlists-router.ts | 9 +++++ .../src/app/storage/storage-routes.spec.ts | 27 ++++++++++++++ docs/architecture/pwa-self-hosted.md | 19 ++++++---- .../specs/2026-10-08-server-storage-design.md | 7 +++- .../src/lib/playlist-storage-backend.ts | 11 ++++++ .../src/lib/playlists.service.spec.ts | 37 +++++++++++++++++++ libs/services/src/lib/playlists.service.ts | 19 +++++++++- .../server-playlist-document-store.spec.ts | 23 ++++++++++++ .../server-playlist-document-store.ts | 19 +++++++++- 9 files changed, 159 insertions(+), 12 deletions(-) diff --git a/apps/web-backend/src/app/storage/playlists-router.ts b/apps/web-backend/src/app/storage/playlists-router.ts index 97d39df99..d755fe4f2 100644 --- a/apps/web-backend/src/app/storage/playlists-router.ts +++ b/apps/web-backend/src/app/storage/playlists-router.ts @@ -55,6 +55,15 @@ export function createPlaylistsRouter( router.put('/playlists/:id', json, (req, res) => { const id = requirePlaylistId(req.params.id); const replaced = revisions.etag(id); + // An edit built on a version another device has replaced since. + const ifMatch = req.get('if-match'); + if (ifMatch && !matchesEtag(ifMatch, replaced)) { + throw new StorageRequestError( + 412, + 'conflict', + 'The playlist changed since it was read.' + ); + } playlists.upsert( { ...requireDocument(req.body), _id: id }, { keepPayload: req.query['keepPayload'] === '1' } diff --git a/apps/web-backend/src/app/storage/storage-routes.spec.ts b/apps/web-backend/src/app/storage/storage-routes.spec.ts index 83b354491..bbb32092b 100644 --- a/apps/web-backend/src/app/storage/storage-routes.spec.ts +++ b/apps/web-backend/src/app/storage/storage-routes.spec.ts @@ -205,6 +205,33 @@ describe('server storage routes', () => { }) )); + it('refuses a playlist write built on a replaced version', () => + withTempStorage((storage) => + withServer(app(storage), async (baseUrl) => { + const first = await fetch( + `${baseUrl}/v1/playlists/p1`, + json(m3uDocument('p1'), { method: 'PUT' }) + ); + const read = first.headers.get('etag') ?? ''; + const edit = (ifMatch: string) => + fetch( + `${baseUrl}/v1/playlists/p1?keepPayload=1`, + json( + { _id: 'p1', title: 'Renamed' }, + { method: 'PUT', headers: { 'if-match': ifMatch } } + ) + ); + + expect((await edit(read)).status).toBe(200); + // Another device wrote since `read`. + const stale = await edit(read); + expect(stale.status).toBe(412); + await expect(stale.json()).resolves.toMatchObject({ + code: 'conflict', + }); + }) + )); + it('writes user state only on the tag it was read at', () => withTempStorage((storage) => withServer(app(storage), async (baseUrl) => { diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 35c29e424..35a4d8d2b 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -85,7 +85,8 @@ Server storage adds versioned routes under `/v1`, which the PWA uses in `PUT ...?keepPayload=1` keeps the stored `playlist`, `items` and `header` fields (`409 payload_missing` without a stored row); every `PUT` answers with the new `ETag` and `X-Replaced-ETag`, the version it - replaced + replaced, and one with `If-Match` answers `412 conflict` when another + write replaced that version - `GET /v1/user-state`, `PUT|DELETE /v1/user-state/:namespace/:playlistId`: Xtream favorites, recently viewed, collection snapshots and playback positions, one JSON value per namespace and playlist; `GET` adds a @@ -193,12 +194,16 @@ settings and guards read them in every build. - **Playlists.** `ServerStorageSession.start()` (an app initializer) plugs `ServerPlaylistDocumentStore` into `PlaylistStorageBackend` (kinds - `electron`, `server`, `indexeddb`), so `PlaylistsService` is unchanged. 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. It keeps its copy after such a write only when - `X-Replaced-ETag` names the version its payload came from; otherwise - another device wrote in between and the next read fetches the document. + `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 + 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 + keeps its copy only when `X-Replaced-ETag` names the version its payload + came from. - **Xtream user data.** `USER_DATA_STORAGE` becomes `ServerUserDataStorage`: the four synced namespaces are cached under the `iptvnator-server:` key prefix (the legacy unprefixed keys are never written in server mode) and 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 6b500c3a6..b02b5bbb2 100644 --- a/docs/superpowers/specs/2026-10-08-server-storage-design.md +++ b/docs/superpowers/specs/2026-10-08-server-storage-design.md @@ -209,8 +209,11 @@ https://claude.ai/artifact/3WheJAgeNL5ag6kmtPrJcD ## 11. Risks -- **Concurrent edits on two devices**: last write wins per document. - Every playlist mutation re-reads first, so the window is one round trip. +- **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. 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/playlist-storage-backend.ts b/libs/services/src/lib/playlist-storage-backend.ts index b6ab6a478..5766176f6 100644 --- a/libs/services/src/lib/playlist-storage-backend.ts +++ b/libs/services/src/lib/playlist-storage-backend.ts @@ -28,6 +28,17 @@ export type PlaylistDocumentStore = Pick< > >; +/** + * The server refused a write built on a playlist version another device has + * replaced since; the write queue runs the mutation again on a fresh read. + */ +export class PlaylistWriteConflict extends Error { + constructor(readonly playlistId: string) { + super(`Playlist ${playlistId} changed since it was read.`); + this.name = 'PlaylistWriteConflict'; + } +} + export type PlaylistStorageKind = 'electron' | 'server' | 'indexeddb'; /** Single place that decides where `PlaylistsService` persists playlists. */ diff --git a/libs/services/src/lib/playlists.service.spec.ts b/libs/services/src/lib/playlists.service.spec.ts index 7c44c6acf..08f5abd8e 100644 --- a/libs/services/src/lib/playlists.service.spec.ts +++ b/libs/services/src/lib/playlists.service.spec.ts @@ -1,5 +1,6 @@ import { EMPTY, firstValueFrom, from, of } from 'rxjs'; import { DbStores, Playlist, PlaylistMeta } from '@iptvnator/shared/interfaces'; +import { PlaylistWriteConflict } from './playlist-storage-backend'; import { PlaylistsService, resolvePlaylistParser } from './playlists.service'; const SQLITE_PLAYLIST_MIGRATION_FLAG = 'm3u-playlists-indexeddb-to-sqlite-v1'; @@ -1635,6 +1636,42 @@ describe('PlaylistsService', () => { ); }); + it('runs a mutation again on a fresh read when the server refused it', async () => { + const versions = [ + { _id: 'p9', title: 'P', favorites: ['a'] }, + { _id: 'p9', title: 'P', favorites: ['a', 'theirs'] }, + ] as unknown as Playlist[]; + let reads = 0; + const electron = { + dbGetAppPlaylist: jest.fn( + async () => versions[Math.min(reads++, 1)] + ), + dbGetAppPlaylists: jest.fn(async () => []), + dbGetAppState: jest.fn(async (key: string) => + key === SQLITE_PLAYLIST_MIGRATION_FLAG ? '1' : null + ), + dbSetAppState: jest.fn(), + dbUpsertAppPlaylist: jest + .fn() + .mockRejectedValueOnce(new PlaylistWriteConflict('p9')) + .mockResolvedValue(undefined), + dbUpsertAppPlaylists: jest.fn(), + }; + testWindow.electron = electron; + + await firstValueFrom( + createService().transformPlaylistFavorites('p9', (favorites) => [ + ...favorites, + 'mine', + ]) + ); + + expect(electron.dbUpsertAppPlaylist).toHaveBeenCalledTimes(2); + expect(electron.dbUpsertAppPlaylist).toHaveBeenLastCalledWith( + expect.objectContaining({ favorites: ['a', 'theirs', 'mine'] }) + ); + }); + it('short-circuits removeFromPlaylistRecentlyViewedBatch when identities is empty', async () => { const existingPlaylist: Playlist = { _id: 'playlist-5', diff --git a/libs/services/src/lib/playlists.service.ts b/libs/services/src/lib/playlists.service.ts index 6a138d144..cbd1dd3f6 100644 --- a/libs/services/src/lib/playlists.service.ts +++ b/libs/services/src/lib/playlists.service.ts @@ -47,6 +47,7 @@ import { import { PlaylistDocumentStore, PlaylistStorageBackend, + PlaylistWriteConflict, } from './playlist-storage-backend'; const SQLITE_PLAYLIST_MIGRATION_FLAG = 'm3u-playlists-indexeddb-to-sqlite-v1'; @@ -413,7 +414,7 @@ export class PlaylistsService { return defer(() => { const previous = this.playlistWriteQueues.get(playlistId) ?? Promise.resolve(); - const next = previous.then(() => operation()); + const next = previous.then(() => retryOnConflict(operation)); const tail = next.then( () => undefined, () => undefined @@ -1553,3 +1554,19 @@ export class PlaylistsService { return this.clearPlaylistRecentlyViewed(playlistId); } } + +/** + * A mutation the server refused as built on an older version (another device + * wrote since) runs again: it reads the playlist afresh first. + */ +async function retryOnConflict(operation: () => Promise): Promise { + for (let attempt = 1; ; attempt += 1) { + try { + return await operation(); + } catch (error) { + if (!(error instanceof PlaylistWriteConflict) || attempt >= 3) { + throw error; + } + } + } +} 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 307ce05bb..ebe7523ea 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 @@ -1,4 +1,5 @@ import type { Playlist } from '@iptvnator/shared/interfaces'; +import { PlaylistWriteConflict } from '../playlist-storage-backend'; import { ServerApiError } from './server-api-client'; import { ServerPlaylistDocumentStore } from './server-playlist-document-store'; @@ -142,6 +143,28 @@ describe('ServerPlaylistDocumentStore', () => { }); }); + it('writes an edit on the version it read and reports a refusal', async () => { + const { api, store } = setup(); + api.request.mockResolvedValueOnce(fakeResponse(200, doc, '"b.0.1"')); + const served = await store.dbGetAppPlaylist('p1'); + if (!served) throw new Error('expected a document'); + api.sendJson.mockRejectedValueOnce( + new ServerApiError(412, 'conflict', 'changed') + ); + + await expect( + store.dbUpsertAppPlaylist({ ...served, favorites: ['a'] }) + ).rejects.toBeInstanceOf(PlaylistWriteConflict); + expect(api.sendJson.mock.calls[0][3]).toEqual({ + 'If-Match': '"b.0.1"', + }); + + // A retry on the same copy, not read again, writes as before. + api.sendJson.mockResolvedValueOnce(putResponse('"b.0.3"', '"b.0.2"')); + await store.dbUpsertAppPlaylist({ ...served, favorites: ['a'] }); + expect(api.sendJson.mock.calls[1][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 e0bd5975d..2b9cc64c6 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 @@ -5,7 +5,10 @@ import { type ElectronBridgeResult, type Playlist, } from '@iptvnator/shared/interfaces'; -import type { PlaylistDocumentStore } from '../playlist-storage-backend'; +import { + type PlaylistDocumentStore, + PlaylistWriteConflict, +} from '../playlist-storage-backend'; import { ServerApiError, type ServerApiClient } from './server-api-client'; interface CachedDocument { @@ -27,6 +30,8 @@ type ServerApi = Pick< */ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { private readonly cache = new Map(); + /** Versions 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(); @@ -84,7 +89,12 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { const response = await this.api.sendJson( 'PUT', `${path}?keepPayload=1`, - withoutPayload(playlist) + 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. @@ -98,6 +108,11 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { } 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'