mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-14 13:16:16 -08:00
fix(server-storage): conditional playlist metadata writes, retried on a fresh read
A favorite or rename (keepPayload PUT) is sent with If-Match on the version it was read at. The server answers 412 when another device wrote since; the store throws PlaylistWriteConflict and PlaylistsService's per-playlist write queue runs the mutation again, which reads the playlist afresh, so two devices' metadata edits both survive. A retry on the same unread copy writes unconditionally, as before; refresh and import writes stay last write wins. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
2d6f28f8c8
commit
7afb9758f4
9 files changed
+159
-12
No files matched your search
@@ -55,6 +55,15 @@ export function createPlaylistsRouter(
|
||||
router.put('/playlists/:id', json, (req, res) => {
|
||||
const id = requirePlaylistId(req.params.id);
|
||||
const replaced = revisions.etag(id);
|
||||
// An edit built on a version another device has replaced since.
|
||||
const ifMatch = req.get('if-match');
|
||||
if (ifMatch && !matchesEtag(ifMatch, replaced)) {
|
||||
throw new StorageRequestError(
|
||||
412,
|
||||
'conflict',
|
||||
'The playlist changed since it was read.'
|
||||
);
|
||||
}
|
||||
playlists.upsert(
|
||||
{ ...requireDocument(req.body), _id: id },
|
||||
{ keepPayload: req.query['keepPayload'] === '1' }
|
||||
|
||||
@@ -205,6 +205,33 @@ describe('server storage routes', () => {
|
||||
})
|
||||
));
|
||||
|
||||
it('refuses a playlist write built on a replaced version', () =>
|
||||
withTempStorage((storage) =>
|
||||
withServer(app(storage), async (baseUrl) => {
|
||||
const first = await fetch(
|
||||
`${baseUrl}/v1/playlists/p1`,
|
||||
json(m3uDocument('p1'), { method: 'PUT' })
|
||||
);
|
||||
const read = first.headers.get('etag') ?? '';
|
||||
const edit = (ifMatch: string) =>
|
||||
fetch(
|
||||
`${baseUrl}/v1/playlists/p1?keepPayload=1`,
|
||||
json(
|
||||
{ _id: 'p1', title: 'Renamed' },
|
||||
{ method: 'PUT', headers: { 'if-match': ifMatch } }
|
||||
)
|
||||
);
|
||||
|
||||
expect((await edit(read)).status).toBe(200);
|
||||
// Another device wrote since `read`.
|
||||
const stale = await edit(read);
|
||||
expect(stale.status).toBe(412);
|
||||
await expect(stale.json()).resolves.toMatchObject({
|
||||
code: 'conflict',
|
||||
});
|
||||
})
|
||||
));
|
||||
|
||||
it('writes user state only on the tag it was read at', () =>
|
||||
withTempStorage((storage) =>
|
||||
withServer(app(storage), async (baseUrl) => {
|
||||
|
||||
@@ -85,7 +85,8 @@ Server storage adds versioned routes under `/v1`, which the PWA uses in
|
||||
`PUT ...?keepPayload=1` keeps the stored `playlist`, `items` and
|
||||
`header` fields (`409 payload_missing` without a stored row); every `PUT`
|
||||
answers with the new `ETag` and `X-Replaced-ETag`, the version it
|
||||
replaced
|
||||
replaced, and one with `If-Match` answers `412 conflict` when another
|
||||
write replaced that version
|
||||
- `GET /v1/user-state`, `PUT|DELETE /v1/user-state/:namespace/:playlistId`:
|
||||
Xtream favorites, recently viewed, collection snapshots and playback
|
||||
positions, one JSON value per namespace and playlist; `GET` adds a
|
||||
@@ -193,12 +194,16 @@ settings and guards read them in every build.
|
||||
|
||||
- **Playlists.** `ServerStorageSession.start()` (an app initializer) plugs
|
||||
`ServerPlaylistDocumentStore` into `PlaylistStorageBackend` (kinds
|
||||
`electron`, `server`, `indexeddb`), so `PlaylistsService` is unchanged. The
|
||||
store caches full documents by ETag and uploads a document whose payload it
|
||||
served unchanged with `keepPayload=1`, so a favorite or a rename does not
|
||||
resend a large M3U. It keeps its copy after such a write only when
|
||||
`X-Replaced-ETag` names the version its payload came from; otherwise
|
||||
another device wrote in between and the next read fetches the document.
|
||||
`electron`, `server`, `indexeddb`). The store caches full documents by
|
||||
ETag and uploads a document whose payload it served unchanged with
|
||||
`keepPayload=1`, so a favorite or a rename does not resend a large M3U.
|
||||
Such a write carries `If-Match` with the version it was read at: when
|
||||
another device wrote in between, the server answers `412`, the store
|
||||
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). After a write the store
|
||||
keeps its copy only when `X-Replaced-ETag` names the version its payload
|
||||
came from.
|
||||
- **Xtream user data.** `USER_DATA_STORAGE` becomes `ServerUserDataStorage`:
|
||||
the four synced namespaces are cached under the `iptvnator-server:` key
|
||||
prefix (the legacy unprefixed keys are never written in server mode) and
|
||||
|
||||
@@ -209,8 +209,11 @@ https://claude.ai/artifact/3WheJAgeNL5ag6kmtPrJcD
|
||||
|
||||
## 11. Risks
|
||||
|
||||
- **Concurrent edits on two devices**: last write wins per document.
|
||||
Every playlist mutation re-reads first, so the window is one round trip.
|
||||
- **Concurrent edits on two devices**: a metadata edit (`keepPayload`) is
|
||||
sent with `If-Match` on the version it was read at; on a `412` the
|
||||
per-playlist write queue runs the mutation again on a fresh read. Other
|
||||
document writes (refresh, import) stay last write wins; every playlist
|
||||
mutation re-reads first, so the window is one round trip.
|
||||
Xtream map entries are written on the tag they were read at and replayed
|
||||
on the server's value after a `412`, so both devices' edits are kept.
|
||||
- **Large payloads**: M3U global favorites and backup export read all full
|
||||
|
||||
@@ -28,6 +28,17 @@ export type PlaylistDocumentStore = Pick<
|
||||
>
|
||||
>;
|
||||
|
||||
/**
|
||||
* The server refused a write built on a playlist version another device has
|
||||
* replaced since; the write queue runs the mutation again on a fresh read.
|
||||
*/
|
||||
export class PlaylistWriteConflict extends Error {
|
||||
constructor(readonly playlistId: string) {
|
||||
super(`Playlist ${playlistId} changed since it was read.`);
|
||||
this.name = 'PlaylistWriteConflict';
|
||||
}
|
||||
}
|
||||
|
||||
export type PlaylistStorageKind = 'electron' | 'server' | 'indexeddb';
|
||||
|
||||
/** Single place that decides where `PlaylistsService` persists playlists. */
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { EMPTY, firstValueFrom, from, of } from 'rxjs';
|
||||
import { DbStores, Playlist, PlaylistMeta } from '@iptvnator/shared/interfaces';
|
||||
import { PlaylistWriteConflict } from './playlist-storage-backend';
|
||||
import { PlaylistsService, resolvePlaylistParser } from './playlists.service';
|
||||
|
||||
const SQLITE_PLAYLIST_MIGRATION_FLAG = 'm3u-playlists-indexeddb-to-sqlite-v1';
|
||||
@@ -1635,6 +1636,42 @@ describe('PlaylistsService', () => {
|
||||
);
|
||||
});
|
||||
|
||||
it('runs a mutation again on a fresh read when the server refused it', async () => {
|
||||
const versions = [
|
||||
{ _id: 'p9', title: 'P', favorites: ['a'] },
|
||||
{ _id: 'p9', title: 'P', favorites: ['a', 'theirs'] },
|
||||
] as unknown as Playlist[];
|
||||
let reads = 0;
|
||||
const electron = {
|
||||
dbGetAppPlaylist: jest.fn(
|
||||
async () => versions[Math.min(reads++, 1)]
|
||||
),
|
||||
dbGetAppPlaylists: jest.fn(async () => []),
|
||||
dbGetAppState: jest.fn(async (key: string) =>
|
||||
key === SQLITE_PLAYLIST_MIGRATION_FLAG ? '1' : null
|
||||
),
|
||||
dbSetAppState: jest.fn(),
|
||||
dbUpsertAppPlaylist: jest
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new PlaylistWriteConflict('p9'))
|
||||
.mockResolvedValue(undefined),
|
||||
dbUpsertAppPlaylists: jest.fn(),
|
||||
};
|
||||
testWindow.electron = electron;
|
||||
|
||||
await firstValueFrom(
|
||||
createService().transformPlaylistFavorites('p9', (favorites) => [
|
||||
...favorites,
|
||||
'mine',
|
||||
])
|
||||
);
|
||||
|
||||
expect(electron.dbUpsertAppPlaylist).toHaveBeenCalledTimes(2);
|
||||
expect(electron.dbUpsertAppPlaylist).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({ favorites: ['a', 'theirs', 'mine'] })
|
||||
);
|
||||
});
|
||||
|
||||
it('short-circuits removeFromPlaylistRecentlyViewedBatch when identities is empty', async () => {
|
||||
const existingPlaylist: Playlist = {
|
||||
_id: 'playlist-5',
|
||||
|
||||
@@ -47,6 +47,7 @@ import {
|
||||
import {
|
||||
PlaylistDocumentStore,
|
||||
PlaylistStorageBackend,
|
||||
PlaylistWriteConflict,
|
||||
} from './playlist-storage-backend';
|
||||
|
||||
const SQLITE_PLAYLIST_MIGRATION_FLAG = 'm3u-playlists-indexeddb-to-sqlite-v1';
|
||||
@@ -413,7 +414,7 @@ export class PlaylistsService {
|
||||
return defer(() => {
|
||||
const previous =
|
||||
this.playlistWriteQueues.get(playlistId) ?? Promise.resolve();
|
||||
const next = previous.then(() => operation());
|
||||
const next = previous.then(() => retryOnConflict(operation));
|
||||
const tail = next.then(
|
||||
() => undefined,
|
||||
() => undefined
|
||||
@@ -1553,3 +1554,19 @@ export class PlaylistsService {
|
||||
return this.clearPlaylistRecentlyViewed(playlistId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A mutation the server refused as built on an older version (another device
|
||||
* wrote since) runs again: it reads the playlist afresh first.
|
||||
*/
|
||||
async function retryOnConflict<T>(operation: () => Promise<T>): Promise<T> {
|
||||
for (let attempt = 1; ; attempt += 1) {
|
||||
try {
|
||||
return await operation();
|
||||
} catch (error) {
|
||||
if (!(error instanceof PlaylistWriteConflict) || attempt >= 3) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { Playlist } from '@iptvnator/shared/interfaces';
|
||||
import { PlaylistWriteConflict } from '../playlist-storage-backend';
|
||||
import { ServerApiError } from './server-api-client';
|
||||
import { ServerPlaylistDocumentStore } from './server-playlist-document-store';
|
||||
|
||||
@@ -142,6 +143,28 @@ describe('ServerPlaylistDocumentStore', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('writes an edit on the version it read and reports a refusal', async () => {
|
||||
const { api, store } = setup();
|
||||
api.request.mockResolvedValueOnce(fakeResponse(200, doc, '"b.0.1"'));
|
||||
const served = await store.dbGetAppPlaylist('p1');
|
||||
if (!served) throw new Error('expected a document');
|
||||
api.sendJson.mockRejectedValueOnce(
|
||||
new ServerApiError(412, 'conflict', 'changed')
|
||||
);
|
||||
|
||||
await expect(
|
||||
store.dbUpsertAppPlaylist({ ...served, favorites: ['a'] })
|
||||
).rejects.toBeInstanceOf(PlaylistWriteConflict);
|
||||
expect(api.sendJson.mock.calls[0][3]).toEqual({
|
||||
'If-Match': '"b.0.1"',
|
||||
});
|
||||
|
||||
// A retry on the same copy, not read again, writes as before.
|
||||
api.sendJson.mockResolvedValueOnce(putResponse('"b.0.3"', '"b.0.2"'));
|
||||
await store.dbUpsertAppPlaylist({ ...served, favorites: ['a'] });
|
||||
expect(api.sendJson.mock.calls[1][3]).toEqual({});
|
||||
});
|
||||
|
||||
it('runs deletes through the user-data hold', async () => {
|
||||
const order: string[] = [];
|
||||
const api = {
|
||||
|
||||
@@ -5,7 +5,10 @@ import {
|
||||
type ElectronBridgeResult,
|
||||
type Playlist,
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import type { PlaylistDocumentStore } from '../playlist-storage-backend';
|
||||
import {
|
||||
type PlaylistDocumentStore,
|
||||
PlaylistWriteConflict,
|
||||
} from '../playlist-storage-backend';
|
||||
import { ServerApiError, type ServerApiClient } from './server-api-client';
|
||||
|
||||
interface CachedDocument {
|
||||
@@ -27,6 +30,8 @@ type ServerApi = Pick<
|
||||
*/
|
||||
export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
|
||||
private readonly cache = new Map<string, CachedDocument>();
|
||||
/** Versions a write was refused on (another device wrote since). */
|
||||
private readonly refused = new Set<string>();
|
||||
/** Served payload object → the ETag of the document it came from. */
|
||||
private readonly servedPayloads = new WeakMap<object, string>();
|
||||
|
||||
@@ -84,7 +89,12 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
|
||||
const response = await this.api.sendJson(
|
||||
'PUT',
|
||||
`${path}?keepPayload=1`,
|
||||
withoutPayload(playlist)
|
||||
withoutPayload(playlist),
|
||||
// Built on the version it was read at; a retry on the
|
||||
// same copy, not read again, writes as before.
|
||||
this.refused.has(payloadEtag)
|
||||
? {}
|
||||
: { 'If-Match': payloadEtag }
|
||||
);
|
||||
// The server kept its stored payload; it is ours only if
|
||||
// nothing replaced it since we read it.
|
||||
@@ -98,6 +108,11 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
|
||||
}
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
if (error instanceof ServerApiError && error.status === 412) {
|
||||
this.refused.add(payloadEtag);
|
||||
this.cache.delete(playlist._id);
|
||||
throw new PlaylistWriteConflict(playlist._id);
|
||||
}
|
||||
if (!(
|
||||
error instanceof ServerApiError &&
|
||||
error.code === 'payload_missing'
|
||||
|
||||
Reference in new issue
Block a user