From da9f99f3f247a7de0ae0675db9e5bc5ffe4f57d8 Mon Sep 17 00:00:00 2001 From: 4gray Date: Wed, 14 Oct 2026 22:42:26 +0200 Subject: [PATCH] fix(server-storage): a restore merged by a conflict stays parked; first-read tests wait for the read A restore whose write conflicted with an edit another device made during it had the conflict replay merge that edit in, and flushed() still said the lists were taken, so the restore was consumed with lists that were not the backup's. Both restore paths now compare the stored lists with the ones they wrote once the server took them, and a difference keeps the restore parked to run again on a fresh read. The two first-read tests released the mocked read after a fixed number of microtask turns, which could come before the read started; they now wait until it has. Co-Authored-By: Claude Opus 5.5 --- docs/architecture/pwa-self-hosted.md | 5 +- .../pwa-xtream-data-source.spec.ts | 54 +++++++++++++++++++ .../data-sources/pwa-xtream-data-source.ts | 17 ++++++ ...playlist-backup.service.pwa-xtream.spec.ts | 35 ++++++++++++ .../lib/pwa-xtream-backup-state.service.ts | 34 +++++++++--- .../server-user-data-adopt.spec.ts | 24 ++++++--- 6 files changed, 155 insertions(+), 14 deletions(-) diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 393cb6079..41c6ee002 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -273,8 +273,9 @@ settings and guards read them in every build. would merge into the server's instead of replacing them (the restore 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, + counts only once the server took its lists (`UserDataStorage.flushed`) + as they are: a conflict that merged another device's edit in leaves it + parked to run again on a fresh read. 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/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 761316310..28fd338a7 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 @@ -1091,6 +1091,60 @@ describe('PwaXtreamDataSource', () => { expect(flushed).toHaveBeenCalledWith('p1'); }); + it('replaces the lists, and runs again when a conflict merged them', async () => { + const values = new Map([ + ['xtream-favorites', JSON.stringify({ p1: ['live:1'] })], + ]); + let merge = false; + 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(async () => { + if (merge) { + // Another device added one meanwhile: replayed in. + values.set( + 'xtream-favorites', + JSON.stringify({ p1: ['live:9', 'movie:5'] }) + ); + } + return true; + }), + }; + 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); + const state = { + hiddenCategories: [], + favorites: [{ xtreamId: 5, contentType: 'movie' }], + recentlyViewed: [], + playbackPositions: [], + } as never; + + await source.restoreUserData('p1', state); + expect(JSON.parse(values.get('xtream-favorites') ?? '{}')).toEqual({ + p1: ['movie:5'], + }); + + merge = true; + await expect(source.restoreUserData('p1', state)).rejects.toThrow( + /changed elsewhere/ + ); + }); + it('replaces nothing while earlier changes to the playlist are unsent', async () => { const values = new Map([ ['xtream-favorites', 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 484f28bac..9c55b7972 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 @@ -1465,6 +1465,23 @@ export class PwaXtreamDataSource implements IXtreamDataSource { 'The server did not take the restored lists; the restore waits.' ); } + // A conflict merges an edit another device made meanwhile into the + // lists: the restore replaces them, so it runs again on a fresh read. + const stored = [ + this.collectionStorage.getFavoritesFromStorage()[playlistId], + this.collectionStorage.getRecentItemsFromStorage()[playlistId], + this.getPlaybackPositionsFromStorage()[playlistId], + ]; + const restored = [ + favorites[playlistId], + recentItems[playlistId], + playbackPositions[playlistId], + ]; + if (JSON.stringify(stored) !== JSON.stringify(restored)) { + throw new Error( + 'The lists changed elsewhere during the restore; it runs again.' + ); + } } // ========================================================================= 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 30e928a54..1ba082396 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,6 +539,41 @@ describe('PlaylistBackupService Xtream user state in the PWA', () => { ).not.toBeNull(); }); + it('keeps the restore parked when a conflict merged other lists in', async () => { + const manifest = await exportSeeded(); + localStorage.clear(); + // Taken, but merged with an edit another device made meanwhile. + const flushed = jest + .fn() + .mockResolvedValueOnce(true) + .mockImplementation(async () => { + const map = JSON.parse( + localStorage.getItem(KEYS.favorites) ?? '{}' + ); + map['xtream-1'] = [...map['xtream-1'], 'live:9']; + localStorage.setItem(KEYS.favorites, JSON.stringify(map)); + return true; + }); + 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); + expect(summary.errors[0]).toContain('changed elsewhere'); + expect( + pendingRestore(service).getOrThrow('xtream-1') + ).not.toBeNull(); + }); + it('replaces nothing while earlier changes to the playlist are unsent', async () => { const manifest = await exportSeeded(); localStorage.clear(); 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 7bd228bf7..5946bda36 100644 --- a/libs/services/src/lib/pwa-xtream-backup-state.service.ts +++ b/libs/services/src/lib/pwa-xtream-backup-state.service.ts @@ -138,8 +138,9 @@ export class PwaXtreamBackupStateService { 'Earlier changes to this playlist are not on the server yet; the restore waits.' ); } + let written: Map; try { - this.replaceLists(playlistId, state); + written = this.replaceLists(playlistId, state); } catch (error) { if (!isQuotaExceeded(error)) { try { @@ -153,22 +154,43 @@ export class PwaXtreamBackupStateService { // Server mode: the lists count once the server took them; a refused // one keeps the restore parked (titles merged regardless, as above). const saved = (await this.storage.flushed?.(playlistId)) !== false; + // A conflict merges an edit another device made meanwhile into the + // lists: the backup replaces them, so it waits and runs again on a + // fresh read. + const replaced = + saved && + [...written].every( + ([name, list]) => + JSON.stringify(this.listFor(name, playlistId)) === + JSON.stringify(list) + ); this.mergeSnapshots(playlistId, snapshots); if (!saved) { throw new Error( 'The server did not take the restored lists; the restore waits.' ); } + if (!replaced) { + throw new Error( + 'The lists changed elsewhere during the restore; it runs again.' + ); + } } + /** The lists it wrote, by map. */ private replaceLists( playlistId: string, state: XtreamPendingRestoreState - ): void { + ): Map { + const written = new Map(); + const write = (name: UserDataMap, list: unknown[]) => { + this.writeFor(name, playlistId, list); + written.set(name, list); + }; const favorites = state.favorites .map((item) => typedKey(item.contentType, item.xtreamId)) .filter((key): key is string => key !== null); - this.writeFor('favorites', playlistId, [...new Set(favorites)]); + write('favorites', [...new Set(favorites)]); const recent: StoredMap[] = []; for (const item of state.recentlyViewed) { @@ -181,18 +203,18 @@ export class PwaXtreamBackupStateService { }); } } - this.writeFor('recentItems', playlistId, recent); + write('recentItems', recent); const now = new Date().toISOString(); - this.writeFor( + write( 'playbackPositions', - playlistId, state.playbackPositions.filter(isPlaybackPosition).map((item) => ({ ...item, playlistId, updatedAt: item.updatedAt ?? now, })) ); + return written; } /** diff --git a/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts b/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts index 5b61d9461..249e5f206 100644 --- a/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts +++ b/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts @@ -3,6 +3,16 @@ import { UNCONFIRMED_TTL_MS } from './server-unsent-writes'; import { ServerUserDataStorage } from './server-user-data-storage'; import { fakeServer, memoryStorage } from './server-user-data.test-helpers'; +/** Lets microtasks run until `condition` holds (a mocked call started). */ +async function until(condition: () => boolean): Promise { + for (let turn = 0; turn < 1000 && !condition(); turn += 1) { + await Promise.resolve(); + } + if (!condition()) { + throw new Error('The awaited call never started'); + } +} + describe('ServerUserDataStorage taking over other page loads', () => { it("keeps progress this tab saved over a closed tab's older edit", async () => { const at = (seconds: number) => [ @@ -222,11 +232,13 @@ describe('ServerUserDataStorage taking over other page loads', () => { ); await a.hydrate(); let answer = (): void => undefined; + let asked = false; const b = new ServerUserDataStorage( { ...server.api, // Answered with the state of when it was asked. getJson: jest.fn(() => { + asked = true; const old = server.api.getJson(); return new Promise((resolve) => { answer = () => resolve(old); @@ -239,9 +251,8 @@ describe('ServerUserDataStorage taking over other page loads', () => { memoryStorage() ); const reading = b.hydrate(); - for (let turn = 0; turn < 10; turn += 1) { - await Promise.resolve(); - } + // Its read is out before the other tab's write lands. + await until(() => asked); a.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); await a.flush(); answer(); @@ -426,10 +437,12 @@ describe('ServerUserDataStorage taking over other page loads', () => { a.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); const sending = a.flush(); let answer = (): void => undefined; + let asked = false; const b = new ServerUserDataStorage( { ...server.api, getJson: jest.fn(() => { + asked = true; const old = server.api.getJson(); return new Promise((resolve) => { answer = () => resolve(old); @@ -442,9 +455,8 @@ describe('ServerUserDataStorage taking over other page loads', () => { memoryStorage() ); const reading = b.hydrate(); - for (let turn = 0; turn < 10; turn += 1) { - await Promise.resolve(); - } + // Its read is out before the other tab's write lands. + await until(() => asked); // A's write lands (its record is cleared), then B's old answer. land(); await sending;