mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-14 13:16:16 -08:00
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
dd143c8aca
commit
6e44aef3c8
9 files changed
+180
-26
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1052,7 +1052,11 @@ describe('PwaXtreamDataSource', () => {
|
||||
|
||||
it('keeps a restore parked while the server refuses its lists', async () => {
|
||||
const values = new Map<string, string>();
|
||||
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<string, string>([
|
||||
['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'] })
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<void>(
|
||||
(resolve) => (finishWrite = resolve)
|
||||
)
|
||||
: Promise.resolve()
|
||||
),
|
||||
// Only the first write is slow.
|
||||
dbUpsertAppPlaylist: jest.fn(() => {
|
||||
if (!slowWrites) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
slowWrites = false;
|
||||
return new Promise<void>(
|
||||
(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);
|
||||
|
||||
@@ -106,6 +106,8 @@ export class PlaylistsService {
|
||||
private readonly playlistWriteQueues = new Map<string, Promise<unknown>>();
|
||||
/** "Remove all" in progress: every playlist write waits for it. */
|
||||
private deletingAll: Promise<void> = 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<T>
|
||||
): Observable<T> {
|
||||
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<void> {
|
||||
// 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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in new issue
Block a user