From 5c2b3c235b31a3169fee352b742e7f2ec0ca405e Mon Sep 17 00:00:00 2001 From: 4gray Date: Wed, 14 Oct 2026 23:24:16 +0200 Subject: [PATCH] fix(server-storage): kept page loads replay oldest first; restore checks compare normalized reads A page load and an older one it kept could both hold steps for one entry. Adopting them, each adopted page went ahead of what was queued, so adopting the oldest first put the newer page's steps before the older's; restoring them after a reload collapsed the newer page's steps into one, so its removal of an item the older page had added was lost on a conflict replay. Adoption now goes newest first, and a reload appends a later page's steps (and early edits) after an older copy's. restoreUserData() compared the lists it wrote with reads that drop duplicate favorites, so a saved restore listing one twice never finished, also in browser-only mode. It now compares reads right after writing with reads after the save. Co-Authored-By: Claude Opus 5.5 --- .../pwa-xtream-data-source.spec.ts | 11 +++ .../data-sources/pwa-xtream-data-source.ts | 20 ++-- .../server-user-data-adopt.spec.ts | 91 ++++++++++++++++++- .../server-storage/server-user-data-adopt.ts | 18 +++- .../server-user-data-storage.ts | 3 +- 5 files changed, 129 insertions(+), 14 deletions(-) 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 28fd338a7..39ba106de 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 @@ -1138,6 +1138,17 @@ describe('PwaXtreamDataSource', () => { expect(JSON.parse(values.get('xtream-favorites') ?? '{}')).toEqual({ p1: ['movie:5'], }); + // A saved restore may list one favorite twice: reads drop the + // copy on both sides of the comparison. + await expect( + source.restoreUserData('p1', { + ...(state as object), + favorites: [ + { xtreamId: 5, contentType: 'movie' }, + { xtreamId: 5, contentType: 'movie' }, + ], + } as never) + ).resolves.toBeUndefined(); merge = true; await expect(source.restoreUserData('p1', state)).rejects.toThrow( 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 9c55b7972..23f77d2c2 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 @@ -1458,6 +1458,14 @@ export class PwaXtreamDataSource implements IXtreamDataSource { }) ); this.savePlaybackPositionsToStorage(playbackPositions); + // The lists as reads return them (deduplicated, normalized). + const restoredLists = () => + JSON.stringify([ + this.collectionStorage.getFavoritesFromStorage()[playlistId], + this.collectionStorage.getRecentItemsFromStorage()[playlistId], + this.getPlaybackPositionsFromStorage()[playlistId], + ]); + const restored = restoredLists(); // Server mode: done once the server took the lists, else it stays // parked for the next open. if ((await this.userDataStorage.flushed?.(playlistId)) === false) { @@ -1467,17 +1475,7 @@ export class PwaXtreamDataSource implements IXtreamDataSource { } // 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)) { + if (restoredLists() !== restored) { throw new Error( 'The lists changed elsewhere during the restore; it runs again.' ); 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 249e5f206..7e761988f 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 @@ -1,5 +1,11 @@ import { ServerApiError } from './server-api-client'; -import { UNCONFIRMED_TTL_MS } from './server-unsent-writes'; +import { keepPages } from './server-page-pointer'; +import { + markClosed, + markUnconfirmed, + saveUnsentWrites, + UNCONFIRMED_TTL_MS, +} from './server-unsent-writes'; import { ServerUserDataStorage } from './server-user-data-storage'; import { fakeServer, memoryStorage } from './server-user-data.test-helpers'; @@ -593,4 +599,87 @@ describe('ServerUserDataStorage taking over other page loads', () => { duplicate.touch(); expect(duplicate.hasUnsentWrites()).toBe(true); }); + + describe('a page load and the older one it kept', () => { + // The older page added X; the newer one took that over and then + // removed X again; neither copy could replace the other. + function journals() { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const local = memoryStorage(); + const tag = `t${JSON.stringify(['S']).replace(/"/g, "'")}`; + const entry = (steps: { before: unknown; after: unknown }[]) => [ + { + namespace: 'xtream-favorites' as const, + playlistId: 'p1', + value: steps[steps.length - 1].after, + tag, + steps, + }, + ]; + saveUnsentWrites( + local, + 'default', + 'older', + entry([{ before: ['S'], after: ['S', 'X'] }]) + ); + saveUnsentWrites( + local, + 'default', + 'newer', + entry([ + { before: ['S'], after: ['S', 'X'] }, + { before: ['S', 'X'], after: ['S'] }, + ]) + ); + // Another device adds Y meanwhile: the replay decides. + server.state['xtream-favorites']['p1'] = ['S', 'Y']; + return { server, local }; + } + + async function send(page: ServerUserDataStorage) { + await page.hydrate(); + await page.flush(); + await page.flush(); + } + + it('replays them oldest first on a reload', async () => { + const { server, local } = journals(); + const tabStore = memoryStorage(); + keepPages(tabStore, 'default', 'newer', ['older']); + + await send( + new ServerUserDataStorage( + server.api as never, + local, + jest.fn(), + 'default', + tabStore + ) + ); + + // X stays removed. + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'Y']); + }); + + it('replays them oldest first when a duplicate adopts them', async () => { + const { server, local } = journals(); + const tabStore = memoryStorage(); + keepPages(tabStore, 'default', 'newer', ['older']); + // The newer page load still runs when the duplicate opens. + markUnconfirmed(local, 'default', 'newer', ['xtream-favorites|p1']); + const duplicate = new ServerUserDataStorage( + server.api as never, + local, + jest.fn(), + 'default', + tabStore + ); + markClosed(local, 'default', 'newer'); + duplicate.touch(); + + await send(duplicate); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'Y']); + }); + }); }); diff --git a/libs/services/src/lib/server-storage/server-user-data-adopt.ts b/libs/services/src/lib/server-storage/server-user-data-adopt.ts index b596ca092..d8b047c7f 100644 --- a/libs/services/src/lib/server-storage/server-user-data-adopt.ts +++ b/libs/services/src/lib/server-storage/server-user-data-adopt.ts @@ -3,6 +3,7 @@ import { behind, type EditStep, splitEditKey, + uncertain, UNKNOWN_TAG, yielding, } from './server-user-data-entries'; @@ -95,6 +96,17 @@ export function restorePage( page: string ): void { for (const write of readUnsentWrites(local, instanceId, page)) { + const key = `${write.namespace}|${write.playlistId}`; + const queued = writes.isQueued(key) ? writes.unsent(key) : undefined; + if (queued) { + // An older page load's copy came first: these steps follow it. + writes.put({ + ...queued, + value: write.value, + steps: [...queued.steps, ...uncertain(write.steps)], + }); + continue; + } writes.queue( write.namespace, write.playlistId, @@ -107,7 +119,11 @@ export function restorePage( ); } for (const edit of readDeferredEdits(local, instanceId, page)) { - deferred.set(`${edit.namespace}|${edit.playlistId}`, edit); + const key = `${edit.namespace}|${edit.playlistId}`; + deferred.set(key, { + ...edit, + steps: [...(deferred.get(key)?.steps ?? []), ...edit.steps], + }); } } 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 99f50c635..4970b20e7 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 @@ -177,8 +177,9 @@ export class ServerUserDataStorage implements UserDataStorage { const owner = pages[pages.length - 1]; if (owner && !pageIsLive(this.local, this.instanceId, owner)) { // This page may have read and edited since: adopt, not restore. + // Each adopted page goes ahead of what is queued: newest first. this.awaited = []; - pages.forEach((page) => this.adoptPage(page)); + [...pages].reverse().forEach((page) => this.adoptPage(page)); } // A tab with nothing to sync never re-reads, which is where closed // tabs' saved writes are otherwise picked up.