fix(server-storage): keep duplicated tabs off live journals, drain writes around remove-all, normalize legacy Stalker rows on the move

- A page marks its unconfirmed-writes record closed on pagehide (not into
  the back-forward cache); a new page takes over the previous page load's
  saved writes only when that record is closed, stale or missing, so a
  duplicated tab, which copies sessionStorage, leaves the running tab its
  writes.
- PlaylistsService.removeAll lets queued playlist writes land first and
  makes writes issued meanwhile wait until it finished; the server store
  refuses a document without an id.
- Moving browser data sets the explicit Stalker portal-mode flag on rows
  from before it (withExplicitStalkerPortalFlag, now shared with
  PlaylistsService).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
4grayandClaude Opus 5.5 committed 2026-10-12 23:59:18 +02:00
1 parent f956758cce
commit 37c14d7d89
13 files changed
+278 -50

No files matched your search

+10 -3
View File
@@ -206,7 +206,10 @@ settings and guards read them in every build.
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). Imports and start-up
migrations (`dbUpsertAppPlaylists`) write unconditionally. An M3U favorite
migrations (`dbUpsertAppPlaylists`) write unconditionally, and the store
refuses a document without an id. "Remove all" first lets queued
playlist writes land, and writes issued meanwhile wait until it is done.
An M3U favorite
toggle goes through `transformPlaylistFavorites`, adding or removing that
one channel on the favorites read at write time. After a write the store
keeps its copy only when `X-Replaced-ETag` names the version its payload
@@ -224,7 +227,10 @@ settings and guards read them in every build.
per page load (`iptvnator-server:unsent-writes:<instanceId>:<page>`). The
tab remembers its page load in `sessionStorage`, so a reload takes its
writes over and sends them first (the old copy goes only once the new one
is saved, at the latest with this page's next saved copy); the writes of a tab closed before it
is saved, at the latest with this page's next saved copy). A page marks
its record closed on `pagehide` (not into the back-forward cache), and
only a closed, stale or missing record is taken over: a duplicated tab,
which copies `sessionStorage`, leaves the running tab its writes; the writes of a tab closed before it
could send them are queued, ahead of its own, by the next re-read in
another tab once that tab's record has gone stale; the result replays the
closed tab's edits first and this tab's newer ones last. Tabs never clear each other's.
@@ -289,7 +295,8 @@ settings and guards read them in every build.
late 401 for a token replaced since (a new sign-in) changes nothing.
- **Moving browser data.** `ServerMigrationService` offers, once per server
instance (marker `iptvnator-server:migration:<instanceId>`), to move the
playlists this browser kept in IndexedDB (`planServerMigration`). A
playlists this browser kept in IndexedDB (`planServerMigration`), Stalker
rows from before the portal-mode flag normalized first. A
playlist the server already has, by id or by source (Xtream server and
user, Stalker portal and MAC, M3U URL), merges into the server's copy; the
others upload with their ids. A merged playlist gains the browser
@@ -1636,6 +1636,61 @@ describe('PlaylistsService', () => {
);
});
it('deletes everything only after queued writes, before later ones', async () => {
const order: string[] = [];
let finishWrite: () => void = () => undefined;
const electron = {
dbGetAppPlaylist: jest.fn(async () => {
order.push('read');
return { _id: 'p9', title: 'P', favorites: [] };
}),
dbGetAppPlaylists: jest.fn(async () => []),
dbGetAppState: jest.fn(async (key: string) =>
key === SQLITE_PLAYLIST_MIGRATION_FLAG ? '1' : null
),
dbSetAppState: jest.fn(),
dbUpsertAppPlaylist: jest.fn(
() =>
new Promise<void>((resolve) => {
order.push('write');
finishWrite = () => resolve();
})
),
dbUpsertAppPlaylists: jest.fn(),
dbDeleteAllPlaylists: jest.fn(async () => {
order.push('delete all');
}),
};
testWindow.electron = electron;
const service = createService();
// Microtasks only: a real timer would let other tests' late errors in.
const settle = async () => {
for (let i = 0; i < 50; i += 1) {
await Promise.resolve();
}
};
const first = firstValueFrom(
service.transformPlaylistFavorites('p9', (f) => [...f, 'a'])
);
await settle();
const removed = firstValueFrom(service.removeAll());
const later = firstValueFrom(
service.transformPlaylistFavorites('p9', (f) => [...f, 'b'])
);
await settle();
expect(order).toEqual(['read', 'write']);
finishWrite();
await first;
await removed;
await settle();
finishWrite();
await later;
expect(order).toEqual(['read', 'write', 'delete all', 'read', 'write']);
});
it('runs a mutation again on a fresh read when the server refused it', async () => {
const versions = [
{ _id: 'p9', title: 'P', favorites: ['a'] },
+34 -31
View File
@@ -28,7 +28,6 @@ import {
Channel,
DbStores,
extractStalkerItemId,
isFullStalkerPortalUrl,
isM3uRecentlyViewedItem,
M3uFavoriteChannel,
M3uRecentlyViewedItem,
@@ -38,6 +37,7 @@ import {
PlaylistUpdateState,
StalkerPortalItem,
normalizeStalkerDate,
withExplicitStalkerPortalFlag,
} from '@iptvnator/shared/interfaces';
import { PLAYLIST_DELETE_CLEANUP } from './playlist-delete-cleanup.token';
import {
@@ -101,6 +101,8 @@ export class PlaylistsService {
private startupMetas?: Promise<Playlist[]> | null;
private indexedDbMigrationPromise: Promise<void> | null = null;
private readonly playlistWriteQueues = new Map<string, Promise<unknown>>();
/** "Remove all" in progress: every playlist write waits for it. */
private deletingAll: Promise<void> = Promise.resolve();
/** Electron bridge or server store; `null` means IndexedDB. */
private get documentStore(): PlaylistDocumentStore | null {
@@ -242,19 +244,7 @@ export class PlaylistsService {
}
private withExplicitLegacyStalkerPortalFlag(playlist: Playlist): Playlist {
if (
!playlist?.macAddress ||
playlist.isFullStalkerPortal !== undefined
) {
return playlist;
}
const portalUrl = playlist.portalUrl ?? playlist.url ?? '';
return {
...playlist,
isFullStalkerPortal: isFullStalkerPortalUrl(portalUrl),
};
return withExplicitStalkerPortalFlag(playlist);
}
private collectStalkerMetadataMigrationUpdates(
@@ -412,8 +402,10 @@ export class PlaylistsService {
operation: () => Promise<T>
): Observable<T> {
return defer(() => {
const previous =
this.playlistWriteQueues.get(playlistId) ?? Promise.resolve();
const previous = Promise.all([
this.playlistWriteQueues.get(playlistId),
this.deletingAll,
]);
const next = previous.then(() => retryOnConflict(operation));
const tail = next.then(
() => undefined,
@@ -1319,23 +1311,34 @@ export class PlaylistsService {
}
removeAll(): Observable<void> {
return defer(() =>
runWithPlaylistAuthorityReset(async () => {
if (this.usesDocumentStore) {
await this.ensureElectronPlaylistMigrations();
const store = this.documentStore;
if (store) {
this.startupMetas = null;
await store.dbDeleteAllPlaylists();
this.healthEvidence?.connections.next({});
}
return undefined;
}
// Writes already queued land first; new ones wait for the delete.
return defer(() => {
const pending = [...this.playlistWriteQueues.values()];
const run = Promise.all(pending).then(() => this.removeAllNow());
this.deletingAll = run.then(
() => undefined,
() => undefined
);
return run;
});
}
await firstValueFrom(this.dbService.clear(DbStores.Playlists));
private removeAllNow(): Promise<void> {
return runWithPlaylistAuthorityReset(async () => {
if (this.usesDocumentStore) {
await this.ensureElectronPlaylistMigrations();
const store = this.documentStore;
if (store) {
this.startupMetas = null;
await store.dbDeleteAllPlaylists();
this.healthEvidence?.connections.next({});
}
return undefined;
})
);
}
await firstValueFrom(this.dbService.clear(DbStores.Playlists));
return undefined;
});
}
private normalizePortalRecentIdentity(value: unknown): string {
@@ -396,4 +396,26 @@ describe('ServerMigrationService', () => {
expect(result.failed).toBeGreaterThan(0);
expect(migration.browserDataMoved()).toBe(false);
});
it('uploads a legacy Stalker row with its portal flag set', async () => {
TestBed.overrideProvider(NgxIndexedDBService, {
useValue: {
getAll: () =>
of([
{
_id: 's1',
title: 'Portal',
macAddress: '00:1A:79:00:00:01',
portalUrl: 'http://portal.test/stalker_portal/c',
},
]),
},
});
await TestBed.inject(ServerMigrationService).moveBrowserData();
expect(session.documentStore.dbUpsertAppPlaylist).toHaveBeenCalledWith(
expect.objectContaining({ _id: 's1', isFullStalkerPortal: true })
);
});
});
@@ -7,6 +7,7 @@ import {
type Playlist,
type ServerUserStateNamespace,
type ServerUserStateResponse,
withExplicitStalkerPortalFlag,
} from '@iptvnator/shared/interfaces';
import {
combineUserData,
@@ -322,10 +323,13 @@ export class ServerMigrationService {
return failed;
}
private readBrowserPlaylists(): Promise<Playlist[]> {
return firstValueFrom(
private async readBrowserPlaylists(): Promise<Playlist[]> {
const playlists = await firstValueFrom(
this.dbService.getAll<Playlist>(DbStores.Playlists)
);
// A browser that skipped the Stalker metadata migration still has
// rows without the explicit portal flag.
return playlists.map(withExplicitStalkerPortalFlag);
}
private readMarker(): string | null {
@@ -233,6 +233,15 @@ describe('ServerPlaylistDocumentStore', () => {
});
});
it('refuses a document without an id', async () => {
const { api, store } = setup();
await expect(
store.dbUpsertAppPlaylist({ title: 'Orphan' } as Playlist)
).rejects.toThrow('without an id');
expect(api.sendJson).not.toHaveBeenCalled();
});
it('writes imports as they are', async () => {
const { api, store } = setup();
api.request.mockResolvedValueOnce(fakeResponse(200, doc, '"b.0.1"'));
@@ -147,7 +147,11 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
playlist: Playlist,
conditional: boolean
): Promise<ElectronBridgeResult> {
const id = playlist._id;
const id = playlist?._id;
if (!id) {
// A mutation that ran after its playlist was deleted.
throw new Error('A playlist without an id cannot be stored.');
}
const path = `/playlists/${encodeURIComponent(id)}`;
const payloadEtag = this.servedPayloadEtag(playlist);
// The version it was read at: of its payload, else of the copy it
@@ -137,9 +137,13 @@ export class ServerStorageSession {
void this.refresh();
}
});
window.addEventListener('pagehide', () =>
this.userData.flushWithKeepalive()
);
window.addEventListener('pagehide', (event) => {
this.userData.flushWithKeepalive();
// Not into the back-forward cache: the page is gone for good.
if (!event.persisted) {
this.userData.close();
}
});
setInterval(() => {
this.userData.touch();
// Also online: a write the server refused (a 5xx) or a first
@@ -292,7 +292,48 @@ export function saveDeferredEdits(
}
}
type UnconfirmedRegistry = Record<string, { at: number; keys: string[] }>;
/** `closed`: the page unloaded (not into the back-forward cache). */
type UnconfirmedRegistry = Record<
string,
{ at: number; keys: string[]; closed?: boolean }
>;
/** Marks a page load's record as unloaded: a reload may take it over. */
export function markClosed(
local: UserDataStorage,
instanceId: string,
page: string
): void {
const registry = readRegistry(local, instanceId);
if (!registry[page]) {
return;
}
registry[page] = { ...registry[page], closed: true };
try {
local.setItem(unconfirmedKey(instanceId), JSON.stringify(registry));
} catch {
// Storage full: the record goes stale instead.
}
}
/**
* Whether a page load still runs: it has a fresh record and did not
* unload. A duplicated tab copies the tab's page load but must not take
* over its saved writes.
*/
export function pageIsLive(
local: UserDataStorage,
instanceId: string,
page: string,
now = Date.now()
): boolean {
const record = readRegistry(local, instanceId)[page];
return (
!!record &&
!record.closed &&
now - Number(record.at) < UNCONFIRMED_TTL_MS
);
}
function readRegistry(
local: UserDataStorage,
@@ -490,7 +490,8 @@ describe('ServerUserDataStorage', () => {
},
]);
// Reload: a new instance over the same browser storage.
// Reload: the page unloads, a new instance over the same storage.
first.storage.close();
const api = {
...first.api,
sendJson: jest.fn().mockResolvedValue(undefined),
@@ -621,6 +622,7 @@ describe('ServerUserDataStorage', () => {
'xtream-favorites',
JSON.stringify({ p1: ['mine'] })
);
first.storage.close();
const api = {
...first.api,
@@ -22,8 +22,10 @@ import {
import { UserDataWriteQueue } from './server-user-data-queue';
import {
claimCache,
markClosed,
markUnconfirmed,
orphanedPages,
pageIsLive,
othersUnconfirmed,
readDeferredEdits,
readUnsentWrites,
@@ -98,11 +100,17 @@ export class ServerUserDataStorage implements UserDataStorage {
});
claimCache(local, instanceId);
const previous = swapPage(tabStore, instanceId, this.tabId);
if (previous) {
// A duplicated tab copies the page load of a tab that still runs.
if (previous && !pageIsLive(local, instanceId, previous)) {
this.takeOver(previous);
}
}
/** The page unloads: a reload of this tab may take its writes over. */
close(): void {
markClosed(this.local, this.instanceId, this.tabId);
}
/**
* A reload of this tab: the writes its last page load could not confirm
* go out with the first flush, and the first read keeps them.
@@ -191,7 +199,7 @@ export class ServerUserDataStorage implements UserDataStorage {
}
getItem(key: string): string | null {
if (!this.isSynced(key)) {
if (!isServerUserStateNamespace(key)) {
return this.local.getItem(key);
}
const value = this.cache.read(key);
@@ -200,7 +208,7 @@ export class ServerUserDataStorage implements UserDataStorage {
}
setItem(key: string, value: string): void {
if (this.isSynced(key)) {
if (isServerUserStateNamespace(key)) {
this.write(key, value);
} else {
this.local.setItem(key, value);
@@ -208,7 +216,7 @@ export class ServerUserDataStorage implements UserDataStorage {
}
removeItem(key: string): void {
if (this.isSynced(key)) {
if (isServerUserStateNamespace(key)) {
this.write(key, null);
} else {
this.local.removeItem(key);
@@ -395,10 +403,6 @@ export class ServerUserDataStorage implements UserDataStorage {
this.adoptClosedTabs();
}
private isSynced(key: string): key is ServerUserStateNamespace {
return isServerUserStateNamespace(key);
}
private queueChanges(
namespace: ServerUserStateNamespace,
next: Record<string, unknown>,
@@ -421,6 +421,7 @@ describe('ServerUserDataStorage with other devices', () => {
before.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] }));
expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']);
before.close();
const reloaded = open();
await reloaded.hydrate();
await reloaded.flush();
@@ -823,6 +824,7 @@ describe('ServerUserDataStorage with other devices', () => {
setItem(key, value);
};
const kept = saved()[0];
before.close();
const reloaded = new ServerUserDataStorage(
offline as never,
local,
@@ -869,6 +871,7 @@ describe('ServerUserDataStorage with other devices', () => {
await before.flush();
// Reloaded: the recovered write fails once, the read works.
before.close();
const send = server.api.sendJson.getMockImplementation();
const api = {
...server.api,
@@ -920,6 +923,7 @@ describe('ServerUserDataStorage with other devices', () => {
await before.hydrate();
before.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
await before.flush();
before.close();
const send = server.api.sendJson.getMockImplementation();
const reloaded = new ServerUserDataStorage(
@@ -967,4 +971,50 @@ describe('ServerUserDataStorage with other devices', () => {
expect(phone.hasUnsentWrites()).toBe(false);
expect(favorites(phone)).toEqual({});
});
it('leaves a still running tab its saved writes when the tab is duplicated', async () => {
const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
const local = memoryStorage();
const tabStore = memoryStorage();
const offline = {
...server.api,
sendJson: jest
.fn()
.mockRejectedValue(new ServerApiError(0, 'network', 'offline')),
};
const original = new ServerUserDataStorage(
offline as never,
local,
jest.fn(),
'default',
tabStore
);
await original.hydrate();
original.setItem(
'xtream-favorites',
JSON.stringify({ p1: ['S', 'X'] })
);
await original.flush();
// A duplicated tab starts with a copy of the tab's own storage.
const copy = memoryStorage();
for (const [key, value] of tabStore.values) {
copy.setItem(key, value);
}
const duplicate = new ServerUserDataStorage(
offline as never,
local,
jest.fn(),
'default',
copy
);
expect(duplicate.hasUnsentWrites()).toBe(false);
expect(original.hasUnsentWrites()).toBe(true);
expect(
[...local.values.keys()].filter((key) =>
key.startsWith('iptvnator-server:unsent-writes:')
)
).toHaveLength(1);
});
});
@@ -26,6 +26,29 @@ export function isFullStalkerPortalUrl(url: string): boolean {
return url.includes('/stalker_portal') || url.includes('/server/load.php');
}
/**
* A Stalker playlist saved before the explicit flag existed, with the flag
* set from its URL; any other playlist comes back unchanged.
*/
export function withExplicitStalkerPortalFlag<
T extends {
macAddress?: string;
isFullStalkerPortal?: boolean;
portalUrl?: string;
url?: string;
},
>(playlist: T): T {
if (!playlist?.macAddress || playlist.isFullStalkerPortal !== undefined) {
return playlist;
}
return {
...playlist,
isFullStalkerPortal: isFullStalkerPortalUrl(
playlist.portalUrl ?? playlist.url ?? ''
),
};
}
/**
* Whether a playlist should use the full-portal auth lifecycle
* (handshake, Bearer token, watchdog).