mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-14 13:16:16 -08:00
fix(server-storage): replay edits behind travelling writes in order, hold writes after an offline start
- An edit queued behind a travelling write keeps its own before → after and carries the travelling write's edit as `prior`, also in the saved writes. A conflict replays both in order, so adding and then removing a favorite removes it whether or not the first write landed, and a lost write is not dropped. - Page-hide writes now report their new tag or a 412 like normal writes: a confirmed one rebases the next edit, a conflict is replayed. - After an offline start from a remembered instance, writes are held and the retry only checks /instance before reloading, so saved writes never reach another database behind the same address. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
a8b6b86e43
commit
487c98ab96
11 files changed
+383
-82
No files matched your search
@@ -182,7 +182,9 @@ instance under `iptvnator-server:instance`. When the probe gets no answer
|
||||
(network error, timeout, or a 5xx such as nginx's 502 while the backend
|
||||
restarts), a browser that remembers this backend still starts in server
|
||||
mode, offline, and reloads once the server answers; it never falls back to
|
||||
an empty browser library. An answer without server storage, or a 404 from
|
||||
an empty browser library. Until that reload it sends no writes
|
||||
(`holdWrites`): the address may lead to another database by then, which
|
||||
the reload's probe finds out. An answer without server storage, or a 404 from
|
||||
an older backend, forgets it (`server-storage-start.ts`). Everything else of the
|
||||
client lives behind the `@iptvnator/services/server-storage` sub-entry, so
|
||||
other builds never load it; `ServerStorageStatusService` and the token
|
||||
@@ -215,11 +217,16 @@ settings and guards read them in every build.
|
||||
as offline. Each write is sent on the tag of the entry it was made
|
||||
against (`UserDataWriteQueue`); after a `412` the edit (before → after) is
|
||||
replayed on the server's current value, so another device's change to the
|
||||
same playlist's entry is kept. An entry a re-read keeps from the cache
|
||||
gets no tag, so its next write conflicts and is replayed too. A page-hide
|
||||
write behind a travelling one is taken from the queue only when that one
|
||||
is answered. While a playlist is being deleted its edits wait; the
|
||||
delete drops them only when it succeeds. Edits made before the first successful read
|
||||
same playlist's entry is kept. An edit queued behind a travelling write
|
||||
keeps its own before → after and carries the travelling one as `prior`
|
||||
(also in the saved writes): if that write was lost, a conflict replays
|
||||
both in order, and if it landed, adding and then removing an item still
|
||||
removes it. An entry a re-read keeps from the cache gets no tag, so its
|
||||
next write conflicts and is replayed too. Page-hide writes report their
|
||||
new tag or conflict like any other, and one behind a travelling write is
|
||||
taken from the queue only when that one is answered. While a playlist is
|
||||
being deleted its edits wait; the delete drops them only when it
|
||||
succeeds. Edits made before the first successful read
|
||||
are held as before → after and applied to the server's value once it
|
||||
arrives (`applyEdit`): a new device or a stale cache adds its favorite
|
||||
instead of replacing the server's list. They are saved in the tab too
|
||||
|
||||
@@ -84,19 +84,56 @@ describe('ServerApiClient', () => {
|
||||
expect(status.reportOffline).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('reports whether a keepalive write was taken', async () => {
|
||||
const ok = jest.fn().mockResolvedValue(response(200, {}));
|
||||
it('reports how a keepalive write ended', async () => {
|
||||
const ok = jest.fn().mockResolvedValue(response(200, { tag: 't2' }));
|
||||
await expect(
|
||||
client(ok).sendKeepalive('PUT', '/user-state/a/b', { value: 1 })
|
||||
).resolves.toBe(true);
|
||||
).resolves.toEqual({ status: 'saved', tag: 't2' });
|
||||
expect(ok.mock.calls[0][1]).toMatchObject({ keepalive: true });
|
||||
|
||||
const conflict = jest.fn().mockResolvedValue(
|
||||
response(412, {
|
||||
code: 'conflict',
|
||||
message: 'changed',
|
||||
current: { value: [2], tag: 't3' },
|
||||
})
|
||||
);
|
||||
await expect(
|
||||
client(conflict).sendKeepalive('PUT', '/user-state/a/b', {
|
||||
value: 1,
|
||||
})
|
||||
).resolves.toEqual({
|
||||
status: 'conflict',
|
||||
current: { value: [2], tag: 't3' },
|
||||
});
|
||||
|
||||
const refused = jest
|
||||
.fn()
|
||||
.mockRejectedValue(new TypeError('body too large'));
|
||||
await expect(
|
||||
client(refused).sendKeepalive('DELETE', '/user-state/a/b')
|
||||
).resolves.toBe(false);
|
||||
).resolves.toEqual({ status: 'failed' });
|
||||
});
|
||||
|
||||
it('holds writes, not reads, until the instance is checked', async () => {
|
||||
const fetchFn = jest.fn().mockResolvedValue(response(200, {}));
|
||||
const held = new ServerApiClient({
|
||||
baseUrl: '/api',
|
||||
status,
|
||||
fetchFn: fetchFn as unknown as typeof fetch,
|
||||
holdWrites: true,
|
||||
});
|
||||
|
||||
await expect(
|
||||
held.sendJson('PUT', '/user-state/a/b', { value: 1 })
|
||||
).rejects.toMatchObject({ status: 0, code: 'network' });
|
||||
await expect(
|
||||
held.sendKeepalive('DELETE', '/user-state/a/b')
|
||||
).resolves.toEqual({ status: 'failed' });
|
||||
await held.getJson('/instance');
|
||||
|
||||
expect(fetchFn).toHaveBeenCalledTimes(1);
|
||||
expect(fetchFn.mock.calls[0][0]).toBe('/api/v1/instance');
|
||||
});
|
||||
|
||||
it('reports offline when a body is cut off after the headers', async () => {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import type {
|
||||
ServerStorageErrorBody,
|
||||
ServerUserStateCurrent,
|
||||
ServerUserStateWriteResponse,
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import {
|
||||
clearServerAccessToken,
|
||||
@@ -26,6 +27,12 @@ export class ServerApiError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/** How a page-hide write ended, as far as the page could see. */
|
||||
export type KeepaliveOutcome =
|
||||
| { readonly status: 'saved'; readonly tag?: string | null }
|
||||
| { readonly status: 'conflict'; readonly current: ServerUserStateCurrent }
|
||||
| { readonly status: 'failed' };
|
||||
|
||||
export interface ServerApiClientOptions {
|
||||
readonly baseUrl: string;
|
||||
readonly status: Pick<
|
||||
@@ -36,6 +43,12 @@ export interface ServerApiClientOptions {
|
||||
readonly onWriteFailure?: (error: ServerApiError) => void;
|
||||
readonly fetchFn?: typeof fetch;
|
||||
readonly timeoutMs?: number;
|
||||
/**
|
||||
* Refuse writes as if offline: a start from a remembered instance that
|
||||
* has not answered yet, whose address may lead to another database by
|
||||
* the time it does.
|
||||
*/
|
||||
readonly holdWrites?: boolean;
|
||||
}
|
||||
|
||||
/** `fetch` for `/v1`: token, timeout, error mapping, connection state. */
|
||||
@@ -54,6 +67,9 @@ export class ServerApiClient {
|
||||
const isWrite = (init.method ?? 'GET') !== 'GET';
|
||||
let response: Response;
|
||||
try {
|
||||
if (isWrite && this.options.holdWrites) {
|
||||
throw new Error('Writes wait for the instance to answer.');
|
||||
}
|
||||
response = await this.fetchFn()(this.url(path), {
|
||||
...init,
|
||||
cache: 'no-store',
|
||||
@@ -150,15 +166,18 @@ export class ServerApiClient {
|
||||
}
|
||||
|
||||
/**
|
||||
* A write that survives the page. Resolves whether the server took it;
|
||||
* browsers reject keepalive bodies above 64 KiB, which resolves false.
|
||||
* A write that survives the page. Browsers reject keepalive bodies
|
||||
* above 64 KiB, which ends as `failed`.
|
||||
*/
|
||||
async sendKeepalive(
|
||||
method: 'PUT' | 'DELETE',
|
||||
path: string,
|
||||
body?: unknown,
|
||||
headers: Record<string, string> = {}
|
||||
): Promise<boolean> {
|
||||
): Promise<KeepaliveOutcome> {
|
||||
if (this.options.holdWrites) {
|
||||
return { status: 'failed' };
|
||||
}
|
||||
try {
|
||||
const response = await this.fetchFn()(this.url(path), {
|
||||
method,
|
||||
@@ -172,9 +191,21 @@ export class ServerApiClient {
|
||||
},
|
||||
...(body === undefined ? {} : { body: JSON.stringify(body) }),
|
||||
});
|
||||
return response.ok;
|
||||
if (response.ok) {
|
||||
const body = (await response
|
||||
.json()
|
||||
.catch(
|
||||
() => ({})
|
||||
)) as Partial<ServerUserStateWriteResponse>;
|
||||
return { status: 'saved', tag: body?.tag };
|
||||
}
|
||||
const { current } =
|
||||
response.status === 412 ? await readErrorBody(response) : {};
|
||||
return current
|
||||
? { status: 'conflict', current }
|
||||
: { status: 'failed' };
|
||||
} catch {
|
||||
return false;
|
||||
return { status: 'failed' };
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ describe('ServerStorageSession', () => {
|
||||
jest.useRealTimers();
|
||||
fetchMock.mockReset();
|
||||
localStorage.clear();
|
||||
sessionStorage.clear();
|
||||
TestBed.resetTestingModule();
|
||||
});
|
||||
|
||||
@@ -126,6 +127,37 @@ describe('ServerStorageSession', () => {
|
||||
expect(reload).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('sends no write before the reload checks the instance', async () => {
|
||||
// A write the last visit could not send, for instance i-1.
|
||||
sessionStorage.setItem(
|
||||
'iptvnator-server:unsent-writes:i-1',
|
||||
JSON.stringify([
|
||||
{
|
||||
namespace: 'xtream-favorites',
|
||||
playlistId: 'p1',
|
||||
value: ['x'],
|
||||
},
|
||||
])
|
||||
);
|
||||
const { session } = setup('none', true);
|
||||
jest.spyOn(
|
||||
session as unknown as { reloadPage(): void },
|
||||
'reloadPage'
|
||||
).mockImplementation(() => undefined);
|
||||
|
||||
await session.start();
|
||||
session.userData.flushWithKeepalive();
|
||||
await session.userData.flush();
|
||||
|
||||
expect(
|
||||
fetchMock.mock.calls.map(([url, init]) => [
|
||||
url,
|
||||
init?.method ?? 'GET',
|
||||
])
|
||||
).toEqual([['/api/v1/instance', 'GET']]);
|
||||
expect(session.userData.hasUnsentWrites()).toBe(true);
|
||||
});
|
||||
|
||||
it('never reloads after a normal start', async () => {
|
||||
const { session } = setup('none');
|
||||
const reload = jest
|
||||
|
||||
@@ -2,7 +2,10 @@ import { effect, inject, Injectable } from '@angular/core';
|
||||
import { MatSnackBar } from '@angular/material/snack-bar';
|
||||
import { Router } from '@angular/router';
|
||||
import { TranslateService } from '@ngx-translate/core';
|
||||
import type { ServerStorageStatusResponse } from '@iptvnator/shared/interfaces';
|
||||
import type {
|
||||
ServerInstanceInfo,
|
||||
ServerStorageStatusResponse,
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import { PlaylistStorageBackend } from '../playlist-storage-backend';
|
||||
import {
|
||||
clearServerAccessToken,
|
||||
@@ -39,6 +42,9 @@ export class ServerStorageSession {
|
||||
baseUrl: this.config.baseUrl,
|
||||
status: this.status,
|
||||
onWriteFailure: (error) => this.announceWriteFailure(error),
|
||||
// A remembered instance may answer as another database: writes
|
||||
// wait for the reload that checks which one it is.
|
||||
holdWrites: this.config.startedOffline,
|
||||
});
|
||||
readonly documentStore = new ServerPlaylistDocumentStore(
|
||||
this.api,
|
||||
@@ -83,14 +89,28 @@ export class ServerStorageSession {
|
||||
|
||||
/** Re-reads synced user data; also the offline retry. */
|
||||
async refresh(): Promise<void> {
|
||||
if (this.config.startedOffline) {
|
||||
await this.reloadOnceReachable();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await this.userData.hydrate();
|
||||
} catch {
|
||||
// The API client already reported offline or unauthorized.
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Nothing was read at start: once the server answers, boot again, so
|
||||
* its instance is checked before anything is sent.
|
||||
*/
|
||||
private async reloadOnceReachable(): Promise<void> {
|
||||
try {
|
||||
await this.api.getJson<ServerInstanceInfo>('/instance');
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (this.config.startedOffline && !this.reloading) {
|
||||
// Nothing was read at start: boot again against the server.
|
||||
if (!this.reloading) {
|
||||
this.reloading = true;
|
||||
this.reloadPage();
|
||||
}
|
||||
|
||||
@@ -65,6 +65,8 @@ export interface UnsentWrite {
|
||||
readonly before?: unknown;
|
||||
/** The server version it is based on (see `PendingEntry.tag`). */
|
||||
readonly tag?: string | null;
|
||||
/** A write sent before it, maybe saved (see `PendingEntry.prior`). */
|
||||
readonly prior?: { readonly before: unknown; readonly after: unknown };
|
||||
}
|
||||
|
||||
interface StoredWrite {
|
||||
@@ -74,6 +76,8 @@ interface StoredWrite {
|
||||
readonly remove?: true;
|
||||
readonly before?: unknown;
|
||||
readonly tag?: string | null;
|
||||
/** JSON drops an undefined side: absent reads back as undefined. */
|
||||
readonly prior?: { readonly before?: unknown; readonly after?: unknown };
|
||||
}
|
||||
|
||||
export function readUnsentWrites(
|
||||
@@ -101,6 +105,13 @@ export function readUnsentWrites(
|
||||
value: write.remove ? undefined : write.value,
|
||||
before: write.before,
|
||||
tag: write.tag,
|
||||
prior:
|
||||
write.prior && typeof write.prior === 'object'
|
||||
? {
|
||||
before: write.prior.before,
|
||||
after: write.prior.after,
|
||||
}
|
||||
: undefined,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -118,12 +129,15 @@ export function saveUnsentWrites(
|
||||
return;
|
||||
}
|
||||
const stored: StoredWrite[] = writes.map(
|
||||
({ namespace, playlistId, value, before, tag }) => ({
|
||||
({ namespace, playlistId, value, before, tag, prior }) => ({
|
||||
namespace,
|
||||
playlistId,
|
||||
...(value === undefined ? { remove: true } : { value }),
|
||||
...(before === undefined ? {} : { before }),
|
||||
...(tag === undefined ? {} : { tag }),
|
||||
...(prior === undefined
|
||||
? {}
|
||||
: { prior: { before: prior.before, after: prior.after } }),
|
||||
})
|
||||
);
|
||||
local.setItem(writesKey(instanceId), JSON.stringify(stored));
|
||||
|
||||
@@ -15,6 +15,19 @@ export interface PendingEntry {
|
||||
* seen absent, `undefined` when unknown (an unconditional write).
|
||||
*/
|
||||
readonly tag?: string | null;
|
||||
/**
|
||||
* The edit of the write this one was queued behind, which may or may
|
||||
* not have reached the server: a conflict replays it first. Kept apart
|
||||
* from this edit, so adding and then removing an item is not one no-op.
|
||||
*/
|
||||
readonly prior?: PriorEdit;
|
||||
}
|
||||
|
||||
export interface PriorEdit {
|
||||
readonly before: unknown;
|
||||
readonly after: unknown;
|
||||
/** The write it was sent with; `0` after a reload. */
|
||||
readonly edit: number;
|
||||
}
|
||||
|
||||
/** `If-Match` / `If-None-Match` for a write based on `tag`. */
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
entryPath,
|
||||
type PendingEntry,
|
||||
preconditionHeaders,
|
||||
type PriorEdit,
|
||||
} from './server-user-data-entries';
|
||||
import { applyEdit } from './server-user-data-edit';
|
||||
import { newestWrites } from './server-unsent-writes';
|
||||
@@ -115,28 +116,49 @@ export class UserDataWriteQueue {
|
||||
}
|
||||
|
||||
/**
|
||||
* A newer edit of an entry that is still queued or travelling keeps
|
||||
* that write's baseline and tag, so a conflict replays both edits.
|
||||
* Queues an edit (`before` → `value`). A newer edit of an entry still
|
||||
* queued joins it; one behind a travelling write keeps its own edit,
|
||||
* carries the write's as `prior` and is based on the write's tag.
|
||||
*/
|
||||
queue(
|
||||
namespace: ServerUserStateNamespace,
|
||||
playlistId: string,
|
||||
value: unknown,
|
||||
before: unknown,
|
||||
tag = this.tagFor(namespace, playlistId)
|
||||
tag = this.tagFor(namespace, playlistId),
|
||||
prior?: Omit<PriorEdit, 'edit'>
|
||||
): void {
|
||||
const key = `${namespace}|${playlistId}`;
|
||||
const earlier = this.pending.get(key) ?? this.inFlight.get(key);
|
||||
const entry: PendingEntry = {
|
||||
namespace,
|
||||
playlistId,
|
||||
value,
|
||||
edit: ++this.editCount,
|
||||
before: earlier ? earlier.before : before,
|
||||
tag: earlier ? earlier.tag : tag,
|
||||
};
|
||||
const queued = this.pending.get(key);
|
||||
const travelling = this.inFlight.get(key);
|
||||
const edit = ++this.editCount;
|
||||
let entry: PendingEntry;
|
||||
if (queued) {
|
||||
entry = { ...queued, value, edit };
|
||||
} else if (travelling) {
|
||||
entry = {
|
||||
namespace,
|
||||
playlistId,
|
||||
value,
|
||||
edit,
|
||||
before,
|
||||
tag: travelling.tag,
|
||||
prior: {
|
||||
before: travelling.prior
|
||||
? travelling.prior.before
|
||||
: travelling.before,
|
||||
after: travelling.value,
|
||||
edit: travelling.edit,
|
||||
},
|
||||
};
|
||||
} else {
|
||||
entry = { namespace, playlistId, value, edit, before, tag };
|
||||
if (prior) {
|
||||
entry = { ...entry, prior: { ...prior, edit: 0 } };
|
||||
}
|
||||
}
|
||||
this.pending.set(key, entry);
|
||||
this.edits.set(key, entry.edit);
|
||||
this.edits.set(key, edit);
|
||||
}
|
||||
|
||||
scheduleFlush(): void {
|
||||
@@ -322,7 +344,7 @@ export class UserDataWriteQueue {
|
||||
private async sendKeepalive(entry: PendingEntry): Promise<void> {
|
||||
const path = entryPath(entry);
|
||||
const headers = preconditionHeaders(entry.tag);
|
||||
const ok = await (entry.value === undefined
|
||||
const outcome = await (entry.value === undefined
|
||||
? this.api.sendKeepalive('DELETE', path, undefined, headers)
|
||||
: this.api.sendKeepalive(
|
||||
'PUT',
|
||||
@@ -331,7 +353,14 @@ export class UserDataWriteQueue {
|
||||
headers
|
||||
));
|
||||
this.retireInFlight(entry);
|
||||
if (!ok) {
|
||||
if (outcome.status === 'saved') {
|
||||
// Without a tag a newer edit keeps this one as `prior`.
|
||||
if (outcome.tag !== undefined) {
|
||||
this.confirmed(entry, outcome.tag);
|
||||
}
|
||||
} else if (outcome.status === 'conflict') {
|
||||
this.replayOnConflict(entry, outcome.current);
|
||||
} else {
|
||||
this.requeue(entry);
|
||||
}
|
||||
this.hooks.changed();
|
||||
@@ -375,20 +404,21 @@ export class UserDataWriteQueue {
|
||||
this.hooks.changed();
|
||||
}
|
||||
|
||||
/** The server took `entry`: a newer queued edit now builds on it. */
|
||||
/** The server took `entry`: the edit queued behind it builds on it. */
|
||||
private confirmed(entry: PendingEntry, tag: string | null): void {
|
||||
this.setTag(entry.namespace, entry.playlistId, tag);
|
||||
const key = entryKey(entry);
|
||||
const next = this.pending.get(key);
|
||||
if (next && next.before === entry.before && next.tag === entry.tag) {
|
||||
this.pending.set(key, { ...next, before: entry.value, tag });
|
||||
if (next?.prior?.edit === entry.edit) {
|
||||
this.pending.set(key, { ...next, prior: undefined, tag });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Another device changed the entry: apply this edit (before → after) to
|
||||
* the server's current value, show the result and send it on that
|
||||
* version, unless a newer edit of the entry will.
|
||||
* Another device changed the entry and `entry` did not land: replay its
|
||||
* edits on the server's current value (with the newer queued edit that
|
||||
* carries it as `prior`, if any), show the result and send it on that
|
||||
* version.
|
||||
*/
|
||||
private replayOnConflict(
|
||||
entry: PendingEntry,
|
||||
@@ -396,36 +426,35 @@ export class UserDataWriteQueue {
|
||||
): void {
|
||||
this.setTag(entry.namespace, entry.playlistId, current.tag);
|
||||
const key = entryKey(entry);
|
||||
const next = this.pending.get(key);
|
||||
if (next) {
|
||||
// The newer edit kept this one's baseline: replay both at once.
|
||||
this.pending.set(key, {
|
||||
...next,
|
||||
value: applyEdit(current.value, next.before, next.value),
|
||||
before: current.value,
|
||||
tag: current.tag,
|
||||
});
|
||||
} else if (this.edits.get(key) === entry.edit) {
|
||||
const merged = applyEdit(current.value, entry.before, entry.value);
|
||||
if (JSON.stringify(merged) !== JSON.stringify(current.value)) {
|
||||
this.pending.set(key, {
|
||||
...entry,
|
||||
value: merged,
|
||||
before: current.value,
|
||||
tag: current.tag,
|
||||
});
|
||||
} else {
|
||||
this.hooks.show(entry.namespace, entry.playlistId, merged);
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
const latest =
|
||||
this.pending.get(key) ??
|
||||
(this.edits.get(key) === entry.edit ? entry : undefined);
|
||||
if (!latest) {
|
||||
return;
|
||||
}
|
||||
this.hooks.show(
|
||||
entry.namespace,
|
||||
entry.playlistId,
|
||||
this.pending.get(key)?.value
|
||||
);
|
||||
const merged = replay(current.value, latest);
|
||||
this.hooks.show(entry.namespace, entry.playlistId, merged);
|
||||
if (
|
||||
latest === entry &&
|
||||
JSON.stringify(merged) === JSON.stringify(current.value)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
this.pending.set(key, {
|
||||
...latest,
|
||||
value: merged,
|
||||
before: current.value,
|
||||
tag: current.tag,
|
||||
prior: undefined,
|
||||
});
|
||||
this.scheduleFlush();
|
||||
}
|
||||
}
|
||||
|
||||
/** An entry's edits, its `prior` first, applied to the server's value. */
|
||||
function replay(server: unknown, entry: PendingEntry): unknown {
|
||||
const base = entry.prior
|
||||
? applyEdit(server, entry.prior.before, entry.prior.after)
|
||||
: server;
|
||||
return applyEdit(base, entry.before, entry.value);
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import type { KeepaliveOutcome } from './server-api-client';
|
||||
import { ServerUserDataStorage } from './server-user-data-storage';
|
||||
|
||||
function memoryStorage() {
|
||||
@@ -34,7 +35,7 @@ describe('ServerUserDataStorage', () => {
|
||||
}),
|
||||
sendJson: jest.fn().mockResolvedValue(undefined),
|
||||
request: jest.fn().mockResolvedValue(undefined),
|
||||
sendKeepalive: jest.fn().mockResolvedValue(true),
|
||||
sendKeepalive: jest.fn().mockResolvedValue({ status: 'saved' }),
|
||||
readJson: jest.fn().mockResolvedValue({}),
|
||||
};
|
||||
const onWriteFailure = jest.fn();
|
||||
@@ -269,7 +270,7 @@ describe('ServerUserDataStorage', () => {
|
||||
it('queues a page-hide write again when the server did not take it', async () => {
|
||||
const { api, storage } = setup();
|
||||
await storage.hydrate();
|
||||
api.sendKeepalive.mockResolvedValueOnce(false);
|
||||
api.sendKeepalive.mockResolvedValueOnce({ status: 'failed' });
|
||||
|
||||
storage.setItem('xtream-favorites', JSON.stringify({ p1: ['big'] }));
|
||||
storage.flushWithKeepalive();
|
||||
@@ -310,9 +311,9 @@ describe('ServerUserDataStorage', () => {
|
||||
it('lets a newer write wait for a page-hide write still travelling', async () => {
|
||||
const { api, storage } = setup();
|
||||
await storage.hydrate();
|
||||
let acknowledge: (ok: boolean) => void = () => undefined;
|
||||
let acknowledge: (outcome: KeepaliveOutcome) => void = () => undefined;
|
||||
api.sendKeepalive.mockReturnValueOnce(
|
||||
new Promise<boolean>((resolve) => (acknowledge = resolve))
|
||||
new Promise<KeepaliveOutcome>((resolve) => (acknowledge = resolve))
|
||||
);
|
||||
|
||||
storage.setItem('xtream-favorites', JSON.stringify({ p1: ['old'] }));
|
||||
@@ -323,7 +324,7 @@ describe('ServerUserDataStorage', () => {
|
||||
await settle();
|
||||
expect(api.sendJson).not.toHaveBeenCalled();
|
||||
|
||||
acknowledge(true);
|
||||
acknowledge({ status: 'saved' });
|
||||
await newer;
|
||||
expect(api.sendJson).toHaveBeenCalledWith(
|
||||
'PUT',
|
||||
@@ -397,9 +398,9 @@ describe('ServerUserDataStorage', () => {
|
||||
it('never retries an older page-hide write over a newer one', async () => {
|
||||
const { api, storage } = setup();
|
||||
await storage.hydrate();
|
||||
const acks: ((ok: boolean) => void)[] = [];
|
||||
const acks: ((outcome: KeepaliveOutcome) => void)[] = [];
|
||||
api.sendKeepalive.mockImplementation(
|
||||
() => new Promise<boolean>((resolve) => acks.push(resolve))
|
||||
() => new Promise<KeepaliveOutcome>((resolve) => acks.push(resolve))
|
||||
);
|
||||
|
||||
for (const value of ['A', 'B', 'C']) {
|
||||
@@ -410,7 +411,9 @@ describe('ServerUserDataStorage', () => {
|
||||
storage.flushWithKeepalive();
|
||||
await settle();
|
||||
// Each page-hide write waits for the previous one.
|
||||
acks[acks.length - 1](value !== 'B');
|
||||
acks[acks.length - 1]({
|
||||
status: value === 'B' ? 'failed' : 'saved',
|
||||
});
|
||||
await settle();
|
||||
}
|
||||
|
||||
@@ -651,7 +654,7 @@ describe('ServerUserDataStorage', () => {
|
||||
}),
|
||||
sendJson: jest.fn().mockResolvedValue(undefined),
|
||||
request: jest.fn().mockResolvedValue(undefined),
|
||||
sendKeepalive: jest.fn().mockResolvedValue(true),
|
||||
sendKeepalive: jest.fn().mockResolvedValue({ status: 'saved' }),
|
||||
readJson: jest.fn().mockResolvedValue({}),
|
||||
};
|
||||
const tab = () =>
|
||||
|
||||
@@ -80,7 +80,8 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
write.playlistId,
|
||||
write.value,
|
||||
write.before,
|
||||
write.tag
|
||||
write.tag,
|
||||
write.prior
|
||||
);
|
||||
}
|
||||
this.unsentSaved = this.writes.hasUnsent();
|
||||
|
||||
@@ -84,10 +84,16 @@ function fakeServer(initial: Record<string, Record<string, unknown>> = {}) {
|
||||
headers: Record<string, string> = {}
|
||||
) => {
|
||||
try {
|
||||
write(path, headers, body?.value);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
const response = write(path, headers, body?.value);
|
||||
const { tag } = await response.json();
|
||||
return { status: 'saved' as const, tag };
|
||||
} catch (error) {
|
||||
return error instanceof ServerApiError && error.current
|
||||
? {
|
||||
status: 'conflict' as const,
|
||||
current: error.current,
|
||||
}
|
||||
: { status: 'failed' as const };
|
||||
}
|
||||
}
|
||||
),
|
||||
@@ -338,4 +344,112 @@ describe('ServerUserDataStorage with other devices', () => {
|
||||
expect(server.state['xtream-favorites']['p1']).toEqual(['S']);
|
||||
expect(phone.hasUnsentWrites()).toBe(false);
|
||||
});
|
||||
|
||||
it('removes an item again after a page-hide write added it', async () => {
|
||||
const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
|
||||
const phone = new ServerUserDataStorage(
|
||||
server.api as never,
|
||||
memoryStorage(),
|
||||
jest.fn()
|
||||
);
|
||||
await phone.hydrate();
|
||||
|
||||
let answer: () => void = () => undefined;
|
||||
const keepalive = server.api.sendKeepalive.getMockImplementation();
|
||||
if (!keepalive) {
|
||||
throw new Error('The fake server answers sendKeepalive');
|
||||
}
|
||||
server.api.sendKeepalive.mockImplementationOnce(
|
||||
(...args: Parameters<typeof keepalive>) =>
|
||||
new Promise(
|
||||
(resolve) => (answer = () => resolve(keepalive(...args)))
|
||||
)
|
||||
);
|
||||
phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
|
||||
phone.flushWithKeepalive();
|
||||
// Back on the page, X is removed while the page-hide write travels.
|
||||
phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] }));
|
||||
answer();
|
||||
await phone.flush();
|
||||
await phone.flush();
|
||||
|
||||
expect(server.state['xtream-favorites']['p1']).toEqual(['S']);
|
||||
expect(phone.hasUnsentWrites()).toBe(false);
|
||||
});
|
||||
|
||||
it('recovers a removal queued behind a write whose answer was lost', async () => {
|
||||
const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
|
||||
const local = memoryStorage();
|
||||
const tabStore = memoryStorage();
|
||||
const open = () =>
|
||||
new ServerUserDataStorage(
|
||||
server.api as never,
|
||||
local,
|
||||
jest.fn(),
|
||||
'default',
|
||||
tabStore
|
||||
);
|
||||
const before = open();
|
||||
await before.hydrate();
|
||||
|
||||
const send = server.api.sendJson.getMockImplementation();
|
||||
if (!send) {
|
||||
throw new Error('The fake server answers sendJson');
|
||||
}
|
||||
// The server saves X, but the page reloads before the answer.
|
||||
server.api.sendJson.mockImplementationOnce(
|
||||
(...args: Parameters<typeof send>) => {
|
||||
void send(...args);
|
||||
return new Promise(() => undefined);
|
||||
}
|
||||
);
|
||||
before.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
|
||||
void before.flush();
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
before.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] }));
|
||||
expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']);
|
||||
|
||||
const reloaded = open();
|
||||
await reloaded.hydrate();
|
||||
await reloaded.flush();
|
||||
|
||||
expect(server.state['xtream-favorites']['p1']).toEqual(['S']);
|
||||
expect(reloaded.hasUnsentWrites()).toBe(false);
|
||||
});
|
||||
|
||||
it('replays a failed write it was queued behind on a conflict', async () => {
|
||||
const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
|
||||
const phone = new ServerUserDataStorage(
|
||||
server.api as never,
|
||||
memoryStorage(),
|
||||
jest.fn()
|
||||
);
|
||||
await phone.hydrate();
|
||||
|
||||
let fail: () => void = () => undefined;
|
||||
server.api.sendJson.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((_, reject) => {
|
||||
fail = () =>
|
||||
reject(new ServerApiError(0, 'network', 'offline'));
|
||||
})
|
||||
);
|
||||
phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
|
||||
const first = phone.flush();
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
phone.setItem(
|
||||
'xtream-favorites',
|
||||
JSON.stringify({ p1: ['S', 'X', 'Y'] })
|
||||
);
|
||||
// X never reached the server; another device saves Z meanwhile.
|
||||
server.state['xtream-favorites']['p1'] = ['S', 'Z'];
|
||||
fail();
|
||||
await first;
|
||||
await phone.flush();
|
||||
await phone.flush();
|
||||
|
||||
expect(
|
||||
[...(server.state['xtream-favorites']['p1'] as string[])].sort()
|
||||
).toEqual(['S', 'X', 'Y', 'Z']);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user