From 6e44aef3c8766617c9ba817ea6482f70bc7ce4cf Mon Sep 17 00:00:00 2001 From: 4gray Date: Wed, 14 Oct 2026 21:28:13 +0200 Subject: [PATCH] fix(services): writes wait for Delete All before they queue; restores replace only settled lists Delete All, third variant: a write that arrived after Delete All waited for it inside its playlist's queue. A Stalker repair that reserved that playlist meanwhile skipped the barrier, but its write still queued behind that waiting one, while the reset waited for the repair's reservation. While a Delete All runs, writes now wait for it before they join their queue, so no queue holds work blocked on the reset; a write under a reservation held here joins at once. A restore replaced lists while this page's own write for the playlist was still unsent (refused, or lost): that write's edits joined the replacement and a conflict replay kept items the backup did not have. Both restore paths now require the playlist's writes to be on the server first (flushed), and otherwise stay parked. Co-Authored-By: Claude Opus 5.5 --- docs/architecture/pwa-self-hosted.md | 6 ++- docs/architecture/stalker-portal.md | 6 ++- .../pwa-xtream-data-source.spec.ts | 46 +++++++++++++++- .../data-sources/pwa-xtream-data-source.ts | 7 +++ ...playlist-backup.service.pwa-xtream.spec.ts | 36 ++++++++++++- .../playlists.service.inventory-read.spec.ts | 2 + .../src/lib/playlists.service.spec.ts | 53 ++++++++++++++++--- libs/services/src/lib/playlists.service.ts | 37 ++++++++----- .../lib/pwa-xtream-backup-state.service.ts | 13 +++++ 9 files changed, 180 insertions(+), 26 deletions(-) diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 69b3ee581..b60169aa1 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -268,8 +268,10 @@ settings and guards read them in every build. first (`ready(true)`: another device may have changed the lists since) and fails when that read does not succeed: unread maps would export as empty lists, and a restore's lists would merge into the server's instead of replacing them (the restore - stays parked). A restore also counts only once the server took that - playlist's lists (`UserDataStorage.flushed`); a refused one stays parked. Replayed edits keep recently viewed lists newest first, + stays parked). A restore also replaces nothing while this page's own + writes for the playlist are unsent (they would join the replacement) and + counts only once the server took its lists (`UserDataStorage.flushed`); + either way it stays parked. Replayed edits keep recently viewed lists newest first, as the app cuts them at their limit from the end. Also, re-reads run one at a time and keep entries edited while they were in flight or not yet saved, and entries another tab saved meanwhile or had diff --git a/docs/architecture/stalker-portal.md b/docs/architecture/stalker-portal.md index 8ffedbf90..99d8a96d4 100644 --- a/docs/architecture/stalker-portal.md +++ b/docs/architecture/stalker-portal.md @@ -244,8 +244,10 @@ reservation stays held through its conditional row transform. The reservation blocks every new authentication (including a URL edit with the same normalized fingerprint) and repair, and drains any work already in flight. Delete All drains the playlist writes queued before it, then takes the exclusive reset; -later writes wait for the delete, except a write under a reservation this tab -holds (`holdsPlaylistAuthority`), which the reset waits for in turn. +later writes wait for the delete before they join their playlist's queue, so +no queue holds work blocked on the reset, except a write under a reservation +this tab holds (`holdsPlaylistAuthority`), which joins at once since the +reset waits for that reservation. When Save follows a lazy repair in the same tab, Edit publishes its local authentication owner first, drains that repair through the actual Web Lock request completion, and only then requests the row reservation itself. Repair diff --git a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts index 8bf976ea8..761316310 100644 --- a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts +++ b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts @@ -1052,7 +1052,11 @@ describe('PwaXtreamDataSource', () => { it('keeps a restore parked while the server refuses its lists', async () => { const values = new Map(); - const flushed = jest.fn(async () => false); + // Nothing unsent before; the restored lists are refused. + const flushed = jest + .fn() + .mockResolvedValueOnce(true) + .mockResolvedValue(false); const storage = { getItem: (key: string) => values.get(key) ?? null, setItem: (key: string, value: string) => { @@ -1086,5 +1090,45 @@ describe('PwaXtreamDataSource', () => { ).rejects.toThrow(/did not take the restored lists/); expect(flushed).toHaveBeenCalledWith('p1'); }); + + it('replaces nothing while earlier changes to the playlist are unsent', async () => { + const values = new Map([ + ['xtream-favorites', JSON.stringify({ p1: ['live:1'] })], + ]); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => { + values.set(key, value); + }, + removeItem: (key: string) => { + values.delete(key); + }, + ready: async () => true, + flushed: jest.fn().mockResolvedValueOnce(false), + }; + TestBed.resetTestingModule(); + TestBed.configureTestingModule({ + providers: [ + PwaXtreamDataSource, + { provide: USER_DATA_STORAGE, useValue: storage }, + { provide: ParentalLockService, useValue: parentalLock }, + { provide: XtreamApiService, useValue: apiService }, + { provide: PlaylistsService, useValue: playlistsService }, + ], + }); + const source = TestBed.inject(PwaXtreamDataSource); + + await expect( + source.restoreUserData('p1', { + hiddenCategories: [], + favorites: [], + recentlyViewed: [], + playbackPositions: [], + } as never) + ).rejects.toThrow(/not on the server yet/); + expect(values.get('xtream-favorites')).toBe( + JSON.stringify({ p1: ['live:1'] }) + ); + }); }); }); diff --git a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts index 6f9ddad36..484f28bac 100644 --- a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts +++ b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts @@ -1413,6 +1413,13 @@ export class PwaXtreamDataSource implements IXtreamDataSource { "The server's user data could not be read yet; the restore waits." ); } + // This page's unsent writes for the playlist would join the + // replacement: they go first, or the restore waits. + if ((await this.userDataStorage.flushed?.(playlistId)) === false) { + throw new Error( + 'Earlier changes to this playlist are not on the server yet; the restore waits.' + ); + } // Restore favorites const favorites = this.collectionStorage.getFavoritesFromStorage(); favorites[playlistId] = restoreState.favorites diff --git a/libs/services/src/lib/playlist-backup.service.pwa-xtream.spec.ts b/libs/services/src/lib/playlist-backup.service.pwa-xtream.spec.ts index d9c54858b..30e928a54 100644 --- a/libs/services/src/lib/playlist-backup.service.pwa-xtream.spec.ts +++ b/libs/services/src/lib/playlist-backup.service.pwa-xtream.spec.ts @@ -539,10 +539,44 @@ describe('PlaylistBackupService Xtream user state in the PWA', () => { ).not.toBeNull(); }); + it('replaces nothing while earlier changes to the playlist are unsent', async () => { + const manifest = await exportSeeded(); + localStorage.clear(); + localStorage.setItem( + KEYS.favorites, + JSON.stringify({ 'xtream-1': ['live:1'] }) + ); + // A write of this page for the playlist that the server refused. + const flushed = jest.fn().mockResolvedValueOnce(false); + const service = createPlaylistBackupService({ + playlistsService: playlistsService([existingXtreamPlaylist]), + pwaXtream: createPwaXtreamBackupState({ + getItem: (key) => localStorage.getItem(key), + setItem: (key, value) => localStorage.setItem(key, value), + removeItem: (key) => localStorage.removeItem(key), + ready: async () => true, + flushed, + }), + }); + + const summary = await service.importBackup(xtreamOnly(manifest)); + + expect(summary.failed).toBe(1); + // Its edits would have joined the replacement lists. + expect(read(KEYS.favorites)).toEqual({ 'xtream-1': ['live:1'] }); + expect( + pendingRestore(service).getOrThrow('xtream-1') + ).not.toBeNull(); + }); + it('keeps the restore parked until the server took the lists', async () => { const manifest = await exportSeeded(); localStorage.clear(); - const flushed = jest.fn(async () => false); + // Nothing unsent before; the restored lists are refused. + const flushed = jest + .fn() + .mockResolvedValueOnce(true) + .mockResolvedValue(false); const service = createPlaylistBackupService({ playlistsService: playlistsService([existingXtreamPlaylist]), pwaXtream: createPwaXtreamBackupState({ diff --git a/libs/services/src/lib/playlists.service.inventory-read.spec.ts b/libs/services/src/lib/playlists.service.inventory-read.spec.ts index 492b6e000..bc75bcd08 100644 --- a/libs/services/src/lib/playlists.service.inventory-read.spec.ts +++ b/libs/services/src/lib/playlists.service.inventory-read.spec.ts @@ -54,6 +54,8 @@ describe('PlaylistsService inventory reads', () => { }, electronMigrationPromise: null, playlistWriteQueues: new Map(), + deletingAll: Promise.resolve(), + resets: 0, }); const settle = async (index: number, playlists: Playlist[]) => { // The read starts after the memoized migration's awaits. diff --git a/libs/services/src/lib/playlists.service.spec.ts b/libs/services/src/lib/playlists.service.spec.ts index 103fdeb2e..3570f8182 100644 --- a/libs/services/src/lib/playlists.service.spec.ts +++ b/libs/services/src/lib/playlists.service.spec.ts @@ -68,6 +68,8 @@ describe('PlaylistsService', () => { electronMigrationPromise: null, indexedDbMigrationPromise: null, playlistWriteQueues: new Map(), + deletingAll: Promise.resolve(), + resets: 0, playlistDeleteCleanups: [], }); @@ -1873,13 +1875,16 @@ describe('PlaylistsService', () => { key === SQLITE_PLAYLIST_MIGRATION_FLAG ? '1' : null ), dbSetAppState: jest.fn(), - dbUpsertAppPlaylist: jest.fn(() => - slowWrites - ? new Promise( - (resolve) => (finishWrite = resolve) - ) - : Promise.resolve() - ), + // Only the first write is slow. + dbUpsertAppPlaylist: jest.fn(() => { + if (!slowWrites) { + return Promise.resolve(); + } + slowWrites = false; + return new Promise( + (resolve) => (finishWrite = resolve) + ); + }), dbUpsertAppPlaylists: jest.fn(), dbDeletePlaylist: jest.fn(async () => undefined), dbDeleteAllPlaylists: jest.fn(async () => undefined), @@ -1919,6 +1924,40 @@ describe('PlaylistsService', () => { expect(electron.dbDeleteAllPlaylists).toHaveBeenCalled(); }); + it('lets a repair write past later writes waiting for the reset', async () => { + fakeLocks(); + const electron = store(true); + const service = createService(); + // An earlier slow write that Delete All waits for. + const slow = firstValueFrom( + service.transformPlaylistFavorites('q1', (f) => [...f, 'a']) + ); + await settle(); + const removed = firstValueFrom(service.removeAll()); + // A later write for P, which waits for the reset. + const later = firstValueFrom( + service.transformPlaylistFavorites('p9', (f) => [...f, 'b']) + ); + // A repair reserves P before the reset takes its lock. + const reservation = await reservePlaylistAuthority('p9'); + if (reservation.status !== 'acquired') { + throw new Error('expected a reservation'); + } + let repaired = false; + const repair = firstValueFrom( + service.transformPlaylistFavorites('p9', (f) => [...f, 'r']) + ).then(() => { + repaired = true; + reservation.release(); + }); + await settle(); + + expect(repaired).toBe(true); + finishWrite(); + await Promise.all([slow, repair, removed, later]); + expect(electron.dbDeleteAllPlaylists).toHaveBeenCalled(); + }); + it('lets a deletion queued behind a slow write finish first', async () => { fakeLocks(); const electron = store(true); diff --git a/libs/services/src/lib/playlists.service.ts b/libs/services/src/lib/playlists.service.ts index 3376078d1..1de4240d4 100644 --- a/libs/services/src/lib/playlists.service.ts +++ b/libs/services/src/lib/playlists.service.ts @@ -106,6 +106,8 @@ export class PlaylistsService { private readonly playlistWriteQueues = new Map>(); /** "Remove all" in progress: every playlist write waits for it. */ private deletingAll: Promise = Promise.resolve(); + /** Delete All runs in progress (see serializePlaylistWrite). */ + private resets = 0; /** Electron bridge or server store; `null` means IndexedDB. */ private get documentStore(): PlaylistDocumentStore | null { @@ -430,14 +432,20 @@ export class PlaylistsService { operation: () => Promise ): Observable { return defer(() => { - const previous = Promise.all([ - this.playlistWriteQueues.get(playlistId), - // A write under a reservation held here (a Stalker repair or - // Edit) cannot wait for Delete All: its reset waits for it. - holdsPlaylistAuthority(playlistId) - ? undefined - : this.deletingAll, - ]); + // While Delete All runs, a write waits for it before it joins + // its playlist's queue, so no queue holds work blocked on the + // reset; a write under a reservation held here (a Stalker repair + // or Edit) joins at once, as the reset waits for that reservation. + if (this.resets > 0 && !holdsPlaylistAuthority(playlistId)) { + return this.deletingAll.then(() => + firstValueFrom( + this.serializePlaylistWrite(playlistId, operation) + ) + ); + } + const previous = Promise.resolve( + this.playlistWriteQueues.get(playlistId) + ); const next = previous.then(() => retryOnConflict(operation)); const tail = next.then( () => undefined, @@ -1348,13 +1356,16 @@ export class PlaylistsService { removeAll(): Observable { // Writes queued so far land first, outside the reset (some take the - // shared side of its lock); new ones wait for the delete, except - // those under a reservation held here (serializePlaylistWrite). + // shared side of its lock); new ones wait for the delete before they + // queue, except those under a reservation held here. return defer(() => { const pending = [...this.playlistWriteQueues.values()]; - const run = Promise.all(pending).then(() => - runWithPlaylistAuthorityReset(() => this.deleteAllRows()) - ); + this.resets += 1; + const run = Promise.all(pending) + .then(() => + runWithPlaylistAuthorityReset(() => this.deleteAllRows()) + ) + .finally(() => (this.resets -= 1)); this.deletingAll = run.then( () => undefined, () => undefined diff --git a/libs/services/src/lib/pwa-xtream-backup-state.service.ts b/libs/services/src/lib/pwa-xtream-backup-state.service.ts index c72b7810a..7bd228bf7 100644 --- a/libs/services/src/lib/pwa-xtream-backup-state.service.ts +++ b/libs/services/src/lib/pwa-xtream-backup-state.service.ts @@ -125,6 +125,19 @@ export class PwaXtreamBackupStateService { const snapshots = snapshotsByKey( Array.isArray(archived) ? archived : [] ); + // The lists replace what the server has now: this page's own unsent + // writes for the playlist would join them, so they go first or the + // restore waits (titles merged regardless, as below). + if ((await this.storage.flushed?.(playlistId)) === false) { + try { + this.mergeSnapshots(playlistId, snapshots); + } catch { + // The unsent writes are the failure to report. + } + throw new Error( + 'Earlier changes to this playlist are not on the server yet; the restore waits.' + ); + } try { this.replaceLists(playlistId, state); } catch (error) {