feat(sync): batch replica sync into one /api/sync/replicas request (#4109)

Auto-sync triggers (boot non-settings, focus, visibilitychange, online,
periodic) used to fan out N parallel `GET /api/sync/replicas?kind=…`
requests, one per replica kind. With 5 kinds today and the focus path
firing on every foreground transition, that's 5x the Cloudflare Worker
invocations of what the work actually requires.

Server: extend POST /api/sync/replicas to accept a batched-pull body
(`{ cursors: [{kind, since}, …] }`) alongside the existing push
(`{ rows: […] }`). Per-kind queries fan out via Promise.all — Supabase
calls inside the Worker aren't billed as Cloudflare requests, so DB
load is unchanged while Worker invocations collapse from N to 1.

Client/manager: add `client.pullBatch` and `manager.pullMany` that
share the existing cursor/HLC machinery. The boot path's `since=null`
override carries over via `pullMany(kinds, { since: null })`.

Orchestrator: `triggerIncrementalPullAll` now does ONE pullMany call
then fans out per-kind apply via Promise.allSettled. Boot does the
same for non-settings kinds (settings stays a single call to preserve
its apply-first ordering invariant).

Foreground triggers: listen to BOTH `focus` and `visibilitychange`,
sharing one throttle. focus is fastest on iOS Tauri WKWebView (~T=0,
~400ms ahead of visibilitychange). visibilitychange is the only
signal that fires on browser tab switching — focus does not. Drops
the Supabase user-ref-change listener (was the slowest of the three
foreground signals; redundant with the DOM events).

Bonus: `useBooksSync` now serializes `handleAutoSync` against
`pullLibrary` via the shared `isPullingRef` gate. The two paths used
to fire two concurrent `/api/sync?type=books` requests on the same
`since` value at startup; now whichever runs first claims the gate
and the other skips (throttle's `emitLast` retries afterwards).

Per session: boot 5→2 Worker calls. Per foreground trigger: 5→1.
This commit is contained in:
Huang Xin
2026-05-09 22:18:46 +08:00
committed by GitHub
parent 295a588988
commit 1eae2af23e
11 changed files with 869 additions and 163 deletions
@@ -23,6 +23,8 @@ let envValue: { envConfig: unknown; appService: unknown } = {
appService: null,
};
let authValue: { user: { id: string } | null } = { user: { id: 'test-user' } };
vi.mock('@/services/sync/replicaPullAndApply', () => ({
replicaPullAndApply: (...args: unknown[]) => pullSpy(...args),
}));
@@ -40,6 +42,10 @@ vi.mock('@/context/EnvContext', () => ({
useEnv: () => envValue,
}));
vi.mock('@/context/AuthContext', () => ({
useAuth: () => authValue,
}));
vi.mock('@/services/transferManager', () => ({
transferManager: { queueReplicaDownload: vi.fn() },
}));
@@ -103,6 +109,20 @@ import { useReplicaPull, __resetReplicaPullForTests } from '@/hooks/useReplicaPu
const fakeService = { createDir: vi.fn(), name: 'fake' };
// Mock manager exposing both per-kind `pull` (used by the boot path) and
// the batched `pullMany` (used by the incremental triggers). Each test
// recreates these so individual call counts don't bleed across cases.
const makeManagerMock = () => ({
pull: vi.fn<(...args: unknown[]) => Promise<unknown[]>>(async () => []),
pullMany: vi.fn<
(kinds: string[], opts?: { since?: string | null }) => Promise<Map<string, unknown[]>>
>(async (kinds) => {
const out = new Map<string, unknown[]>();
for (const k of kinds) out.set(k, []);
return out;
}),
});
/**
* Settings is implicitly pulled at boot regardless of which kinds the
* caller asked for (so other kinds' applyRemote auto-saves don't
@@ -124,6 +144,7 @@ beforeEach(() => {
readyListeners.clear();
__resetReplicaPullForTests();
envValue = { envConfig: { name: 'env' }, appService: fakeService };
authValue = { user: { id: 'test-user' } };
});
afterEach(() => {
@@ -133,7 +154,7 @@ afterEach(() => {
describe('useReplicaPull', () => {
test('does not pull before delayMs elapses', () => {
getReplicaSyncSpy.mockReturnValue({ manager: {} });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 5_000 }));
vi.advanceTimersByTime(4_999);
@@ -141,7 +162,7 @@ describe('useReplicaPull', () => {
});
test('fires pull after delayMs', async () => {
getReplicaSyncSpy.mockReturnValue({ manager: {} });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 1_000 }));
await act(async () => {
@@ -154,9 +175,40 @@ describe('useReplicaPull', () => {
expect(dictionaryPullCount()).toBe(1);
});
test('boot batches non-settings kinds into one pullMany with since=null', async () => {
// Boot today: 1 settings call (single-kind, sequential) + 1 batched
// pullMany call for the rest. Was previously 1 + N parallel
// calls. The batched call must use `{ since: null }` so each kind
// does a full refetch — same recovery semantics as the old
// per-kind boot.
const managerMock = makeManagerMock();
getReplicaSyncSpy.mockReturnValue({ manager: managerMock });
renderHook(() => useReplicaPull({ kinds: ['dictionary', 'font', 'texture'], delayMs: 100 }));
await act(async () => {
vi.advanceTimersByTime(200);
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
});
// Settings still gets its own `replicaPullAndApply` call (the
// `dictionaryPullCount`-style counter approach is what asserts the
// ordering elsewhere). The pullMany path collapses dictionary +
// font + texture into a single round-trip with the since=null
// override applied uniformly.
expect(managerMock.pullMany).toHaveBeenCalledTimes(1);
const [batchedKinds, batchedOpts] = managerMock.pullMany.mock.calls[0]!;
expect([...batchedKinds].sort()).toEqual(['dictionary', 'font', 'texture']);
expect(batchedOpts).toEqual({ since: null });
// pullSpy is `replicaPullAndApply`; expect 1 invocation per kind
// (settings + the three from the batch) all running through the
// apply path.
expect(pullSpy).toHaveBeenCalledTimes(4);
});
test('skips when appService is null', () => {
envValue = { envConfig: { name: 'env' }, appService: null };
getReplicaSyncSpy.mockReturnValue({ manager: {} });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
vi.advanceTimersByTime(500);
@@ -183,7 +235,7 @@ describe('useReplicaPull', () => {
// initReplicaSync now finishes; getReplicaSync starts returning the
// singleton, and the ready listener fires.
getReplicaSyncSpy.mockReturnValue({ manager: {} });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
fireReplicaSyncReady();
await act(async () => {
@@ -204,7 +256,7 @@ describe('useReplicaPull', () => {
});
test('only pulls once per kind across multiple mounts', async () => {
getReplicaSyncSpy.mockReturnValue({ manager: {} });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
const first = renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await act(async () => {
vi.advanceTimersByTime(200);
@@ -227,8 +279,8 @@ describe('useReplicaPull', () => {
});
test('failed pull releases the dedup slot so a later navigation can retry', async () => {
getReplicaSyncSpy.mockReturnValue({ manager: {} });
// Settings pull resolves; dictionary pull (the second call) rejects.
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
// Settings pull resolves; dictionary apply (the second call) rejects.
pullSpy.mockResolvedValueOnce(undefined);
pullSpy.mockRejectedValueOnce(new Error('flaky'));
vi.spyOn(console, 'warn').mockImplementation(() => {});
@@ -256,7 +308,7 @@ describe('useReplicaPull', () => {
});
test('cleanup cancels a pending pull when the component unmounts before delayMs', () => {
getReplicaSyncSpy.mockReturnValue({ manager: {} });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
const view = renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 5_000 }));
vi.advanceTimersByTime(2_000);
view.unmount();
@@ -274,63 +326,57 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
});
};
test('visibilitychange to visible fires an incremental pull (cursor-based, not since=null)', async () => {
test('window focus fires one batched incremental pull (pullMany, cursor-based)', async () => {
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
expect(dictionaryPullCount()).toBe(1);
// Simulate tab going hidden, then visible. The "visible" transition
// is the trigger.
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
// Simulate the window regaining focus (foreground transition).
await act(async () => {
document.dispatchEvent(new Event('visibilitychange'));
window.dispatchEvent(new Event('focus'));
await Promise.resolve();
await Promise.resolve();
});
expect(dictionaryPullCount()).toBe(2);
// The incremental call's pull deps should NOT pin since=null —
// that's the boot-only behavior. Inspect the LAST dict pull deps.
const dictCalls = pullSpy.mock.calls.filter((c) => {
const deps = c[0] as { adapter?: { kind?: string } } | undefined;
return deps?.adapter?.kind === 'dictionary';
});
const incrementalCall = dictCalls.at(-1)![0] as { pull: () => Promise<unknown[]> };
await incrementalCall.pull();
const managerPull = (
getReplicaSyncSpy.mock.results[0]!.value as { manager: { pull: ReturnType<typeof vi.fn> } }
).manager.pull;
// Boot call uses { since: null }; incremental call passes undefined
// (or no opts) so manager.pull falls back to the cursor.
expect(managerPull).toHaveBeenCalled();
const lastArgs = managerPull.mock.calls.at(-1);
expect(lastArgs?.[1]).toBeUndefined();
// pullMany fires twice: once at boot for the non-settings kinds
// (settings boot stays a single-kind `manager.pull`), and once for
// the focus-triggered incremental over every registered kind.
const managerMock = (
getReplicaSyncSpy.mock.results[0]!.value as {
manager: ReturnType<typeof makeManagerMock>;
}
).manager;
expect(managerMock.pullMany).toHaveBeenCalledTimes(2);
// The first call (boot) gets just the non-settings kinds with
// since=null override; the focus call gets all registered kinds
// with cursor-based defaults.
const bootArgs = managerMock.pullMany.mock.calls[0]!;
expect(bootArgs[0]).toEqual(['dictionary']);
expect(bootArgs[1]).toEqual({ since: null });
const focusArgs = managerMock.pullMany.mock.calls[1]!;
expect([...focusArgs[0]].sort()).toEqual(['dictionary', 'settings']);
expect(focusArgs[1]).toBeUndefined();
});
test('visibilitychange is throttled to at most one fire per 30 seconds', async () => {
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
test('focus is throttled to collapse iOS focus-fires-twice bursts', async () => {
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
expect(dictionaryPullCount()).toBe(1); // boot pull only
// Burst of visibility changes within the throttle window — only the
// first should fire an incremental pull. Rapid alt-tab cycling is
// the real-world trigger. Advance ~10s total (well under 30s).
// Burst of focus events within the throttle window — only the first
// fires an incremental pull. iOS Tauri's two-back-to-back focus
// events on a single foreground transition are the real-world
// trigger; rapid alt-tab cycling is the desktop equivalent.
for (let i = 0; i < 5; i++) {
await act(async () => {
document.dispatchEvent(new Event('visibilitychange'));
window.dispatchEvent(new Event('focus'));
vi.advanceTimersByTime(2_000);
await Promise.resolve();
await Promise.resolve();
@@ -338,40 +384,107 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
}
expect(dictionaryPullCount()).toBe(2); // boot + 1 throttled
// Cross the 30s boundary (we already advanced 10s above; advance
// the remaining time and a touch more).
// Cross the throttle boundary; the next focus is allowed through.
await act(async () => {
vi.advanceTimersByTime(20_500);
document.dispatchEvent(new Event('visibilitychange'));
window.dispatchEvent(new Event('focus'));
await Promise.resolve();
await Promise.resolve();
});
expect(dictionaryPullCount()).toBe(3);
});
test('online and periodic triggers are NOT subject to the visibility throttle', async () => {
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
test('visibilitychange to visible fires an incremental pull (browser tab-switch path)', async () => {
// Browser tab switching (cmd+1 / cmd+2) fires `visibilitychange`
// but NOT `window.focus`. Without this listener, replica sync
// wouldn't catch up on tab switch.
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
expect(dictionaryPullCount()).toBe(1);
// Visibility fires once, consuming the throttle slot.
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
await act(async () => {
document.dispatchEvent(new Event('visibilitychange'));
await Promise.resolve();
await Promise.resolve();
});
expect(dictionaryPullCount()).toBe(2);
});
// Online event within the visibility throttle window must STILL
// fire — it's a different signal (we may have just regained
// network) and shouldn't be silenced by recent focus activity.
test('visibilitychange to hidden does NOT fire a pull', async () => {
getReplicaSyncSpy.mockReturnValue({
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
expect(dictionaryPullCount()).toBe(1);
Object.defineProperty(document, 'visibilityState', {
value: 'hidden',
configurable: true,
});
await act(async () => {
document.dispatchEvent(new Event('visibilitychange'));
await Promise.resolve();
await Promise.resolve();
});
expect(dictionaryPullCount()).toBe(1); // unchanged
});
test('focus and visibilitychange share one throttle (no double-pump)', async () => {
// iOS Tauri WKWebView fires both events on the same foreground
// transition (focus first, visibilitychange ~400ms later). One
// throttle gate prevents a double-pull.
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
getReplicaSyncSpy.mockReturnValue({
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
expect(dictionaryPullCount()).toBe(1);
await act(async () => {
window.dispatchEvent(new Event('focus'));
document.dispatchEvent(new Event('visibilitychange'));
await Promise.resolve();
await Promise.resolve();
});
// Boot + ONE incremental, not two.
expect(dictionaryPullCount()).toBe(2);
});
test('online and periodic triggers are NOT subject to the focus throttle', async () => {
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
getReplicaSyncSpy.mockReturnValue({
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
expect(dictionaryPullCount()).toBe(1);
// Focus fires once, consuming the throttle slot.
await act(async () => {
window.dispatchEvent(new Event('focus'));
await Promise.resolve();
await Promise.resolve();
});
expect(dictionaryPullCount()).toBe(2);
// Online event within the focus throttle window must STILL fire —
// it's a different signal (we may have just regained network) and
// shouldn't be silenced by recent foreground activity.
await act(async () => {
window.dispatchEvent(new Event('online'));
await Promise.resolve();
@@ -382,7 +495,7 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
test('online event fires an incremental pull', async () => {
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
@@ -402,7 +515,7 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
configurable: true,
});
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
@@ -425,7 +538,7 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
configurable: true,
});
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
@@ -451,7 +564,7 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
}),
);
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
// Boot pull starts. Settings is dispatched first (in flight, pending);
@@ -462,17 +575,13 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
});
expect(pullSpy).toHaveBeenCalledTimes(1); // settings, awaiting
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
// Fire several triggers while settings boot pull is still in flight.
// pullInFlight has 'settings' → incremental settings is gated;
// dictionary boot hasn't started yet so its incremental is gated by
// pulledKinds (not yet added).
for (let i = 0; i < 5; i++) {
await act(async () => {
document.dispatchEvent(new Event('visibilitychange'));
window.dispatchEvent(new Event('focus'));
window.dispatchEvent(new Event('online'));
await Promise.resolve();
});
@@ -487,13 +596,52 @@ describe('useReplicaPull — incremental auto-pull (visibility / online / interv
});
});
test('focus / online / periodic triggers no-op when there is no user', async () => {
// Logged-out users shouldn't burn /api/sync round-trips. Gate at the
// dispatch layer so the module-level listeners (which run for the
// life of the tab) don't fire pulls after sign-out.
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
authValue = { user: { id: 'user-1' } };
getReplicaSyncSpy.mockReturnValue({
manager: makeManagerMock(),
});
const { rerender } = renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
expect(dictionaryPullCount()).toBe(1);
// Sign out.
authValue = { user: null };
rerender();
await act(async () => {
await Promise.resolve();
});
// None of the auto-sync triggers should fire pulls now. Advance past
// the focus throttle so it doesn't mask the real gate.
await act(async () => {
vi.advanceTimersByTime(15_000);
await Promise.resolve();
});
await act(async () => {
window.dispatchEvent(new Event('focus'));
window.dispatchEvent(new Event('online'));
vi.advanceTimersByTime(5 * 60 * 1000);
await Promise.resolve();
await Promise.resolve();
});
expect(dictionaryPullCount()).toBe(1); // still just the original boot pull
});
test('listeners stay installed across mounts so a long-lived tab keeps pulling', async () => {
Object.defineProperty(document, 'visibilityState', {
value: 'visible',
configurable: true,
});
getReplicaSyncSpy.mockReturnValue({
manager: { pull: vi.fn(async () => []) },
manager: makeManagerMock(),
});
const first = renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
await advancePastBootPull();
@@ -530,7 +678,7 @@ describe('useReplicaPull — settings boot pull sequencing', () => {
}
return Promise.resolve();
});
getReplicaSyncSpy.mockReturnValue({ manager: { pull: vi.fn(async () => []) } });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
// Boot delay elapses; settings pull starts and is pending.
@@ -554,7 +702,7 @@ describe('useReplicaPull — settings boot pull sequencing', () => {
});
test('settings pull is shared across mounts (single network round-trip)', async () => {
getReplicaSyncSpy.mockReturnValue({ manager: { pull: vi.fn(async () => []) } });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
const a = renderHook(() => useReplicaPull({ kinds: ['dictionary'], delayMs: 100 }));
const b = renderHook(() => useReplicaPull({ kinds: ['font'], delayMs: 100 }));
await act(async () => {
@@ -572,7 +720,7 @@ describe('useReplicaPull — settings boot pull sequencing', () => {
});
test('caller asking for settings explicitly does not double-pull settings', async () => {
getReplicaSyncSpy.mockReturnValue({ manager: { pull: vi.fn(async () => []) } });
getReplicaSyncSpy.mockReturnValue({ manager: makeManagerMock() });
renderHook(() => useReplicaPull({ kinds: ['settings', 'dictionary'], delayMs: 100 }));
await act(async () => {
vi.advanceTimersByTime(200);
@@ -264,3 +264,60 @@ describe('ReplicaSyncClient.listReplicaKeys (cache + dedupe)', () => {
expect(mockFetch).toHaveBeenCalledTimes(2);
});
});
describe('ReplicaSyncClient.pullBatch', () => {
test('POSTs cursors to /sync/replicas and returns the per-kind results', async () => {
const fontRow: ReplicaRow = { ...sampleRow, kind: 'font', replica_id: 'f1' };
mockFetch.mockResolvedValueOnce(
new Response(
JSON.stringify({
results: [
{ kind: 'dictionary', rows: [sampleRow] },
{ kind: 'font', rows: [fontRow] },
{ kind: 'texture', rows: [] },
],
}),
{ status: 200 },
),
);
const client = new ReplicaSyncClient();
const result = await client.pullBatch([
{ kind: 'dictionary', since: HLC },
{ kind: 'font', since: null },
{ kind: 'texture', since: null },
]);
expect(mockFetch).toHaveBeenCalledOnce();
const [url, init] = mockFetch.mock.calls[0]!;
// Reuses the existing /sync/replicas route — body shape
// (`{ cursors }` vs `{ rows }`) is the discriminator.
expect(url).toBe('https://example.test/sync/replicas');
expect(init.method).toBe('POST');
expect(JSON.parse(init.body)).toEqual({
cursors: [
{ kind: 'dictionary', since: HLC },
{ kind: 'font', since: null },
{ kind: 'texture', since: null },
],
});
expect(result).toEqual([
{ kind: 'dictionary', rows: [sampleRow] },
{ kind: 'font', rows: [fontRow] },
{ kind: 'texture', rows: [] },
]);
});
test('empty cursors is a no-op (no fetch call)', async () => {
const client = new ReplicaSyncClient();
const result = await client.pullBatch([]);
expect(result).toEqual([]);
expect(mockFetch).not.toHaveBeenCalled();
});
test('5xx → SyncError SERVER', async () => {
mockFetch.mockResolvedValueOnce(new Response('{}', { status: 500 }));
const client = new ReplicaSyncClient();
await expect(client.pullBatch([{ kind: 'dictionary', since: null }])).rejects.toBeInstanceOf(
SyncError,
);
});
});
@@ -1,8 +1,10 @@
import { describe, expect, test } from 'vitest';
import {
HLC_SKEW_TOLERANCE_MS,
MAX_PULL_BATCH,
MAX_PUSH_BATCH,
clampHlcSkew,
validatePullBatch,
validatePullParams,
validatePushBatch,
} from '@/libs/replicaSyncServer';
@@ -153,3 +155,82 @@ describe('validatePullParams', () => {
}
});
});
describe('validatePullBatch', () => {
test('accepts an empty cursors array', () => {
const result = validatePullBatch({ cursors: [] });
expect(result.ok).toBe(true);
if (result.ok) expect(result.params.cursors).toEqual([]);
});
test('accepts mixed since values (string + null) for allowed kinds', () => {
const result = validatePullBatch({
cursors: [
{ kind: 'dictionary', since: '0000000000064-00000000-dev-a' },
{ kind: 'font', since: null },
{ kind: 'settings', since: null },
],
});
expect(result.ok).toBe(true);
});
test('rejects body that is not an object', () => {
const result = validatePullBatch(null);
expect(result.ok).toBe(false);
if (!result.ok) expect(result.status).toBe(400);
});
test('rejects body where cursors is not an array', () => {
const result = validatePullBatch({ cursors: 'not-an-array' });
expect(result.ok).toBe(false);
if (!result.ok) expect(result.status).toBe(400);
});
test('rejects unknown kind in cursors', () => {
const result = validatePullBatch({ cursors: [{ kind: 'not_a_kind', since: null }] });
expect(result.ok).toBe(false);
if (!result.ok) {
expect(result.status).toBe(422);
expect(result.code).toBe('UNKNOWN_KIND');
expect(result.offendingIndex).toBe(0);
}
});
test('rejects since that is not a string or null', () => {
const result = validatePullBatch({
cursors: [{ kind: 'dictionary', since: 1234 as unknown as null }],
});
expect(result.ok).toBe(false);
if (!result.ok) {
expect(result.status).toBe(400);
expect(result.offendingIndex).toBe(0);
}
});
test('rejects duplicated kind in the same batch', () => {
// Avoids ambiguity in the response — a server merging two queries
// for the same kind would have to pick one, and the client has no
// way to express which `since` won.
const result = validatePullBatch({
cursors: [
{ kind: 'dictionary', since: null },
{ kind: 'dictionary', since: '0000000000064-00000000-dev-a' },
],
});
expect(result.ok).toBe(false);
if (!result.ok) {
expect(result.status).toBe(400);
expect(result.offendingIndex).toBe(1);
}
});
test('rejects oversized batches with status 413', () => {
const cursors = Array.from({ length: MAX_PULL_BATCH + 1 }, () => ({
kind: 'dictionary',
since: null,
}));
const result = validatePullBatch({ cursors });
expect(result.ok).toBe(false);
if (!result.ok) expect(result.status).toBe(413);
});
});
@@ -23,6 +23,10 @@ const makeRow = (id: string, hlcStr: Hlc = HLC_NOW): ReplicaRow => ({
const makeFakeClient = () => ({
push: vi.fn(async (rows: ReplicaRow[]) => rows),
pull: vi.fn(async (_kind: string, _since: Hlc | null) => [] as ReplicaRow[]),
pullBatch: vi.fn(
async (_cursors: { kind: string; since: Hlc | null }[]) =>
[] as { kind: string; rows: ReplicaRow[] }[],
),
});
const makeManager = (clientOverrides: Partial<ReturnType<typeof makeFakeClient>> = {}) => {
@@ -294,3 +298,76 @@ describe('ReplicaSyncManager.pull', () => {
expect(cursors.get('dictionary')).toBe(r2.updated_at_ts);
});
});
describe('ReplicaSyncManager.pullMany (batched incremental)', () => {
test("one pullBatch round-trip per call, with each kind's persisted cursor", async () => {
// Seed cursors so we can verify each kind\'s cursor reaches the
// batched request — this is the core saving over per-kind GETs.
const dictCursor = hlcPack(NOW + 50, 0, DEV) as Hlc;
const fontCursor = hlcPack(NOW + 60, 0, DEV) as Hlc;
const client = makeFakeClient();
const { manager, cursors } = makeManager(client);
cursors.set('dictionary', dictCursor);
cursors.set('font', fontCursor);
// texture has no cursor yet — should arrive as null (initial pull).
await manager.pullMany(['dictionary', 'font', 'texture']);
expect(client.pullBatch).toHaveBeenCalledTimes(1);
expect(client.pull).not.toHaveBeenCalled();
const cursorsArg = client.pullBatch.mock.calls[0]![0];
expect(cursorsArg).toEqual([
{ kind: 'dictionary', since: dictCursor },
{ kind: 'font', since: fontCursor },
{ kind: 'texture', since: null },
]);
});
test("advances each kind's cursor independently from its own rows", async () => {
const dictRow = makeRow('d1', hlcPack(NOW + 1000, 0, DEV) as Hlc);
const fontRow = makeRow('f1', hlcPack(NOW + 2000, 0, DEV) as Hlc);
const client = {
...makeFakeClient(),
pullBatch: vi.fn(async () => [
{ kind: 'dictionary', rows: [dictRow] as ReplicaRow[] },
{ kind: 'font', rows: [fontRow] as ReplicaRow[] },
{ kind: 'texture', rows: [] as ReplicaRow[] },
]),
};
const { manager, cursors } = makeManager(client);
const result = await manager.pullMany(['dictionary', 'font', 'texture']);
expect(cursors.get('dictionary')).toBe(dictRow.updated_at_ts);
expect(cursors.get('font')).toBe(fontRow.updated_at_ts);
// Empty result must NOT advance the cursor (would skip future rows).
expect(cursors.get('texture')).toBeUndefined();
// Returned map covers every requested kind, including empty ones.
expect(result.get('dictionary')).toEqual([dictRow]);
expect(result.get('font')).toEqual([fontRow]);
expect(result.get('texture')).toEqual([]);
});
test('observes remote HLCs into the local generator (cross-device clock)', async () => {
const remoteHlc = hlcPack(NOW + 90_000, 3, 'dev-other') as Hlc;
const client = {
...makeFakeClient(),
pullBatch: vi.fn(async () => [
{ kind: 'dictionary', rows: [makeRow('r1', remoteHlc)] as ReplicaRow[] },
]),
};
const { manager, hlc } = makeManager(client);
await manager.pullMany(['dictionary']);
const next = hlc.next();
// Local generator must be ahead of the observed remote stamp.
expect(next > remoteHlc).toBe(true);
});
test('empty kinds list short-circuits without hitting the wire', async () => {
const client = makeFakeClient();
const { manager } = makeManager(client);
const result = await manager.pullMany([]);
expect(result.size).toBe(0);
expect(client.pullBatch).not.toHaveBeenCalled();
});
});
@@ -37,18 +37,12 @@ export const useBooksSync = () => {
const pullLibrary = useCallback(
async (fullRefresh = false, verbose = false) => {
if (!user) return;
if (isPullingRef.current) {
console.log('Pull already in progress, skipping...');
return;
}
if (isPullingRef.current) return;
try {
isPullingRef.current = true;
const library = useLibraryStore.getState().library;
const syncedBooksCount = await syncBooks(
[],
'pull',
(libraryLoaded && library.length === 0) || fullRefresh ? 0 : undefined,
);
const since = (libraryLoaded && library.length === 0) || fullRefresh ? 0 : undefined;
const syncedBooksCount = await syncBooks([], 'pull', since);
if (verbose) {
eventDispatcher.dispatch('toast', {
type: 'info',
@@ -65,10 +59,15 @@ export const useBooksSync = () => {
// eslint-disable-next-line react-hooks/exhaustive-deps
const handleAutoSync = useCallback(
throttle(
() => {
async () => {
if (isPullingRef.current) return;
const newBooks = getNewBooks();
if (newBooks.lastSyncedAt) {
syncBooks(newBooks.books, 'both');
if (!newBooks.lastSyncedAt) return;
isPullingRef.current = true;
try {
await syncBooks(newBooks.books, 'both');
} finally {
isPullingRef.current = false;
}
},
SYNC_BOOKS_INTERVAL_SEC * 1000,
@@ -79,9 +78,7 @@ export const useBooksSync = () => {
useEffect(() => {
if (!user) return;
if (isPullingRef.current) {
return;
}
if (isPullingRef.current) return;
handleAutoSync();
}, [user, library, handleAutoSync]);
+213 -74
View File
@@ -1,4 +1,5 @@
import { useEffect } from 'react';
import { useAuth } from '@/context/AuthContext';
import { useEnv } from '@/context/EnvContext';
import { useCustomDictionaryStore, findDictionaryByContentId } from '@/store/customDictionaryStore';
import {
@@ -43,7 +44,7 @@ import type { ImportedDictionary } from '@/services/dictionaries/types';
import type { CustomFont } from '@/styles/fonts';
import type { CustomTexture } from '@/styles/textures';
import type { OPDSCatalog } from '@/types/opds';
import type { Hlc } from '@/types/replica';
import type { Hlc, ReplicaRow } from '@/types/replica';
import type { SystemSettings } from '@/types/settings';
export type ReplicaKind = 'dictionary' | 'font' | 'texture' | 'opds_catalog' | 'settings';
@@ -60,12 +61,14 @@ const REPLICA_PULL_DEFAULT_DELAY_MS = 5_000;
/** Periodic incremental-pull cadence for long-lived foreground tabs. */
const REPLICA_PULL_PERIODIC_INTERVAL_MS = 5 * 60 * 1000;
/**
* Minimum gap between visibility-triggered pulls. Rapid alt-tab cycles
* (or browsers that fire `visibilitychange` on every workspace switch)
* would otherwise pump a pull on every focus event. Online and periodic
* triggers bypass this — they signal genuinely new conditions.
* Minimum gap between focus-triggered pulls. iOS Tauri WKWebView fires
* `focus` twice during a single foreground transition (~8ms apart) and
* `visibilitychange` an additional ~400ms later — the throttle
* collapses that burst to one pull. Rapid alt-tab cycles on desktop
* are handled by the same gate. `online` and the periodic timer
* bypass this — they're independent signals.
*/
const REPLICA_PULL_VISIBILITY_THROTTLE_MS = 10_000;
const REPLICA_PULL_FOREGROUND_THROTTLE_MS = 10_000;
// Module-level dedup so navigating between pages (library → reader → …)
// doesn't fire a fresh boot pull every time. Once a kind has had its
@@ -89,7 +92,12 @@ const registeredKinds = new Set<ReplicaKind>();
let autoSyncContext: { service: AppService; envConfig: EnvConfigType } | null = null;
let autoSyncListenersInstalled = false;
let periodicTimer: ReturnType<typeof setInterval> | null = null;
let lastVisibilityPullAt = 0;
let lastForegroundPullAt = 0;
// Module-level mirror of `useAuth().user`. The auto-sync listeners run
// at module scope and have no React access; they consult this flag to
// short-circuit when there is no signed-in user (no point hitting
// /api/sync — it would 401 anyway). Kept in sync by the hook.
let hasCurrentUser = false;
// Shared promise for the boot-time settings pull. All other kinds'
// boot pulls await this so applyRemoteSettings has a chance to seed
// `lastPublishedFields` with server-authoritative values before
@@ -123,19 +131,32 @@ interface ReplicaPullConfig<T extends ReplicaLocalRecord> {
onSaltNotFound?: (paths: readonly string[]) => void;
}
/**
* Build the deps the orchestrator hands to `replicaPullAndApply`.
*
* - Boot path: omits `pullOverride`, so `pull` calls `manager.pull(kind)`
* per kind (network round-trip per kind). The boot path also keeps
* the `isAuthenticated` gate, which doubles as the per-kind sync-
* category gate.
* - Incremental path: passes a `pullOverride` returning rows the
* batched `manager.pullMany` already fetched. `isAuthenticated` is
* skipped because the orchestrator pre-filtered by
* `isSyncCategoryEnabled` before issuing the batched request.
*/
const buildReplicaPullDeps = <T extends ReplicaLocalRecord>(
manager: ReplicaSyncManager,
service: AppService,
envConfig: EnvConfigType,
config: ReplicaPullConfig<T>,
pullOpts?: { since?: Hlc | null },
pullOverride?: () => Promise<ReplicaRow[]>,
): PullAndApplyDeps<T> => ({
adapter: config.adapter,
// Boot path passes { since: null } so we always re-fetch and apply
// locally, ignoring any previously-advanced cursor. The visibility /
// online / periodic incremental triggers omit pullOpts so manager.pull
// falls back to the persisted cursor — cheap delta fetch.
pull: () => manager.pull(config.kind, pullOpts),
// locally, ignoring any previously-advanced cursor. The incremental
// path passes pullOverride so a single batched `manager.pullMany`
// result is fanned out to per-kind apply without re-hitting the wire.
pull: pullOverride ?? (() => manager.pull(config.kind, pullOpts)),
findByContentId: config.findByContentId,
hydrateLocalStore: config.hydrateLocalStore
? () => config.hydrateLocalStore!(envConfig)
@@ -171,10 +192,15 @@ const buildReplicaPullDeps = <T extends ReplicaLocalRecord>(
// user-facing category gate here so disabling a kind in
// `User → Manage Sync` no-ops the pull (no HTTP, no warnings)
// alongside the auth precheck — same effect, half the wiring.
isAuthenticated: async () => {
if (!isSyncCategoryEnabled(config.kind)) return false;
return !!(await getAccessToken());
},
// Skipped on the incremental path: the orchestrator pre-filters
// by category before dispatching the batched request, so this
// would just re-check what's already been gated.
isAuthenticated: pullOverride
? undefined
: async () => {
if (!isSyncCategoryEnabled(config.kind)) return false;
return !!(await getAccessToken());
},
});
const dictionaryPullConfig: ReplicaPullConfig<ImportedDictionary> = {
@@ -270,36 +296,69 @@ const settingsPullConfig = (envConfig: EnvConfigType): ReplicaPullConfig<Setting
},
});
/**
* Per-kind dispatch for both the boot pull (one HTTP per kind) and the
* incremental apply (rows already fetched in a batch). Keeping the
* switch keeps the generic record type sound — collapsing the configs
* into a Record<ReplicaKind, ReplicaPullConfig<...>> would force a
* contravariant cast.
*/
const runPullForKind = async (
kind: ReplicaKind,
service: AppService,
envConfig: EnvConfigType,
pullOpts?: { since?: Hlc | null },
pullOverride?: () => Promise<ReplicaRow[]>,
): Promise<void> => {
const ctx = getReplicaSync();
if (!ctx) return;
// Per-kind dispatch keeps the generic record type sound — collapsing
// the three configs into a Record<ReplicaKind, ReplicaPullConfig<...>>
// would force a contravariant cast that loses type safety.
switch (kind) {
case 'dictionary':
await replicaPullAndApply(
buildReplicaPullDeps(ctx.manager, service, envConfig, dictionaryPullConfig, pullOpts),
buildReplicaPullDeps(
ctx.manager,
service,
envConfig,
dictionaryPullConfig,
pullOpts,
pullOverride,
),
);
return;
case 'font':
await replicaPullAndApply(
buildReplicaPullDeps(ctx.manager, service, envConfig, fontPullConfig, pullOpts),
buildReplicaPullDeps(
ctx.manager,
service,
envConfig,
fontPullConfig,
pullOpts,
pullOverride,
),
);
return;
case 'texture':
await replicaPullAndApply(
buildReplicaPullDeps(ctx.manager, service, envConfig, texturePullConfig, pullOpts),
buildReplicaPullDeps(
ctx.manager,
service,
envConfig,
texturePullConfig,
pullOpts,
pullOverride,
),
);
return;
case 'opds_catalog':
await replicaPullAndApply(
buildReplicaPullDeps(ctx.manager, service, envConfig, opdsCatalogPullConfig, pullOpts),
buildReplicaPullDeps(
ctx.manager,
service,
envConfig,
opdsCatalogPullConfig,
pullOpts,
pullOverride,
),
);
return;
case 'settings':
@@ -310,6 +369,7 @@ const runPullForKind = async (
envConfig,
settingsPullConfig(envConfig),
pullOpts,
pullOverride,
),
);
return;
@@ -317,56 +377,87 @@ const runPullForKind = async (
};
/**
* Cursor-based incremental pull, gated by a per-kind in-flight set so
* concurrent triggers (visibility + online firing in the same tick,
* periodic timer racing with a focus event) collapse to one network
* round-trip.
* Cursor-based incremental pull. One batched HTTP round-trip for every
* registered kind that has had its boot pull initiated and whose sync
* category is enabled. Concurrent triggers (focus + online firing in
* the same tick, periodic timer racing with a focus event) collapse
* via the per-kind `pullInFlight` set: if any of the kinds we'd batch
* is already in flight, we skip the redundant call entirely.
*/
const runIncrementalPullForKind = async (
kind: ReplicaKind,
service: AppService,
envConfig: EnvConfigType,
): Promise<void> => {
if (pullInFlight.has(kind)) return;
// Skip until the kind's boot pull has at least been initiated. Boot
// does the full re-fetch (since=null); incremental builds on whatever
// cursor that pull advanced.
if (!pulledKinds.has(kind)) return;
pullInFlight.add(kind);
try {
await runPullForKind(kind, service, envConfig);
} catch (err) {
console.warn(`replica ${kind} incremental pull failed`, err);
} finally {
pullInFlight.delete(kind);
}
};
const triggerIncrementalPullAll = (): void => {
if (!hasCurrentUser) return;
if (!autoSyncContext) return;
const ctx = getReplicaSync();
if (!ctx) return;
const { service, envConfig } = autoSyncContext;
const kindsToPull: ReplicaKind[] = [];
for (const kind of registeredKinds) {
void runIncrementalPullForKind(kind, service, envConfig);
// Skip until the kind's boot pull has at least been initiated. Boot
// does the full re-fetch (since=null); incremental builds on whatever
// cursor that pull advanced.
if (!pulledKinds.has(kind)) continue;
if (pullInFlight.has(kind)) continue;
if (!isSyncCategoryEnabled(kind)) continue;
kindsToPull.push(kind);
}
if (kindsToPull.length === 0) return;
for (const kind of kindsToPull) pullInFlight.add(kind);
void (async () => {
try {
// ONE network round-trip for every eligible kind. Per-kind apply
// runs in parallel below — adapters are independent, and one
// kind's apply error doesn't poison the others.
let rowsByKind: Map<string, ReplicaRow[]>;
try {
rowsByKind = await ctx.manager.pullMany(kindsToPull);
} catch (err) {
console.warn('replica batch pull failed', err);
return;
}
await Promise.allSettled(
kindsToPull.map(async (kind) => {
const rows = rowsByKind.get(kind) ?? [];
try {
await runPullForKind(kind, service, envConfig, undefined, async () => rows);
} catch (err) {
console.warn(`replica ${kind} incremental apply failed`, err);
}
}),
);
} finally {
for (const kind of kindsToPull) pullInFlight.delete(kind);
}
})();
};
const onVisibilityChange = (): void => {
if (typeof document === 'undefined') return;
if (document.visibilityState !== 'visible') return;
// Two foreground signals, sharing one throttle:
// Listening to both with one throttle catches every transition without
// double-pulling. Some debug logs are kept on purpose: foreground-sync
// regressions have historically been hard to reproduce.
const onForegroundReturn = (source: 'focus' | 'visibilitychange'): void => {
// Visibility events fire both directions; only act on the visible side.
if (source === 'visibilitychange' && typeof document !== 'undefined') {
if (document.visibilityState !== 'visible') return;
}
const now = Date.now();
if (now - lastVisibilityPullAt < REPLICA_PULL_VISIBILITY_THROTTLE_MS) return;
lastVisibilityPullAt = now;
if (now - lastForegroundPullAt < REPLICA_PULL_FOREGROUND_THROTTLE_MS) return;
lastForegroundPullAt = now;
triggerIncrementalPullAll();
};
const onFocus = (): void => onForegroundReturn('focus');
const onVisibilityChange = (): void => onForegroundReturn('visibilitychange');
const onOnline = (): void => {
triggerIncrementalPullAll();
};
const onPeriodicTick = (): void => {
// Don't burn battery on a backgrounded tab — the visibilitychange
// listener will fire a catch-up pull when the tab returns to
// foreground.
// Don't burn battery on a backgrounded tab — the focus listener
// will fire a catch-up pull when the window returns to foreground.
if (typeof document !== 'undefined' && document.visibilityState === 'hidden') return;
triggerIncrementalPullAll();
};
@@ -399,7 +490,7 @@ const ensureSettingsBootPulled = (service: AppService, envConfig: EnvConfigType)
* pulls. Idempotent — first call wires everything, subsequent calls
* just refresh `autoSyncContext` (so listeners always see the latest
* appService / envConfig). Listeners stay attached for the lifetime of
* the page.
* the page; in production this runs exactly once.
*/
const installAutoSyncListeners = (service: AppService, envConfig: EnvConfigType): void => {
autoSyncContext = { service, envConfig };
@@ -409,6 +500,7 @@ const installAutoSyncListeners = (service: AppService, envConfig: EnvConfigType)
document.addEventListener('visibilitychange', onVisibilityChange);
}
if (typeof window !== 'undefined') {
window.addEventListener('focus', onFocus);
window.addEventListener('online', onOnline);
}
periodicTimer = setInterval(onPeriodicTick, REPLICA_PULL_PERIODIC_INTERVAL_MS);
@@ -441,12 +533,22 @@ export const useReplicaPull = ({
delayMs = REPLICA_PULL_DEFAULT_DELAY_MS,
}: UseReplicaPullOpts): void => {
const { envConfig, appService } = useEnv();
const { user } = useAuth();
// Stable cache key so the effect doesn't re-run when the caller
// passes a freshly-allocated array literal each render.
const kindsKey = kinds.join(',');
// Mirror the React-side auth state into the module-level flag the
// auto-sync listeners read. Kept in its own effect so a user ref
// change (Supabase TOKEN_REFRESHED reissues the user object on
// foreground / refresh) doesn't tear down the boot-pull effect.
useEffect(() => {
hasCurrentUser = !!user;
}, [user]);
useEffect(() => {
if (!appService) return;
if (!user) return;
for (const kind of kinds) registeredKinds.add(kind);
@@ -472,25 +574,60 @@ export const useReplicaPull = ({
// Subsequent mounts share `settingsBootPullPromise` so the
// pull only happens once per session.
await ensureSettingsBootPulled(appService, envConfig);
for (const kind of otherPending) {
if (pulledKinds.has(kind)) continue;
// Claim both slots up front so a concurrently-scheduled mount
// (e.g., library + reader mounting back-to-back) doesn't
// double-pull, and so a visibility/online trigger landing
// mid-boot doesn't fire a second pull alongside this one.
// On failure we release `pulledKinds` so a subsequent
// navigation can retry; `pullInFlight` is always cleared.
// Boot path skips disabled kinds: enabling a category later
// re-fires `triggerIncrementalPullAll` from a focus event,
// which will fetch the missed rows. This keeps boot bandwidth
// proportional to what the user actually sync's.
const eligible = otherPending.filter(
(k) => !pulledKinds.has(k) && isSyncCategoryEnabled(k),
);
if (eligible.length === 0) return;
// Claim both slots up front so a concurrently-scheduled mount
// (e.g., library + reader mounting back-to-back) doesn't
// double-pull, and so a focus / online trigger landing
// mid-boot doesn't fire a second pull alongside this one.
// On failure we release `pulledKinds` so a subsequent
// navigation can retry; `pullInFlight` is always cleared.
for (const kind of eligible) {
pulledKinds.add(kind);
pullInFlight.add(kind);
void runPullForKind(kind, appService, envConfig, { since: null })
.catch((err) => {
console.warn(`replica ${kind} pull failed`, err);
pulledKinds.delete(kind);
})
.finally(() => {
pullInFlight.delete(kind);
});
}
// ONE batched HTTP round-trip for all non-settings kinds at
// boot, with `since=null` so each kind does a full re-fetch
// (mirrors the old per-kind `runPullForKind(kind, …, {since: null})`
// semantics). Per-kind apply runs in parallel afterwards.
const ctx = getReplicaSync();
if (!ctx) {
for (const kind of eligible) {
pulledKinds.delete(kind);
pullInFlight.delete(kind);
}
return;
}
let rowsByKind: Map<string, ReplicaRow[]>;
try {
rowsByKind = await ctx.manager.pullMany(eligible, { since: null });
} catch (err) {
console.warn('replica boot batch pull failed', err);
for (const kind of eligible) {
pulledKinds.delete(kind);
pullInFlight.delete(kind);
}
return;
}
await Promise.allSettled(
eligible.map(async (kind) => {
try {
const rows = rowsByKind.get(kind) ?? [];
await runPullForKind(kind, appService, envConfig, undefined, async () => rows);
} catch (err) {
console.warn(`replica ${kind} boot apply failed`, err);
pulledKinds.delete(kind);
} finally {
pullInFlight.delete(kind);
}
}),
);
})();
}, delayMs);
};
@@ -510,7 +647,7 @@ export const useReplicaPull = ({
if (unsubscribe) unsubscribe();
};
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [kindsKey, appService, envConfig, delayMs]);
}, [kindsKey, appService, envConfig, delayMs, user]);
};
/** Test seam — clear all module-level state and tear down listeners. */
@@ -523,6 +660,7 @@ export const __resetReplicaPullForTests = (): void => {
document.removeEventListener('visibilitychange', onVisibilityChange);
}
if (typeof window !== 'undefined') {
window.removeEventListener('focus', onFocus);
window.removeEventListener('online', onOnline);
}
if (periodicTimer) {
@@ -530,6 +668,7 @@ export const __resetReplicaPullForTests = (): void => {
periodicTimer = null;
}
autoSyncListenersInstalled = false;
lastVisibilityPullAt = 0;
lastForegroundPullAt = 0;
settingsBootPullPromise = null;
hasCurrentUser = false;
};
@@ -98,6 +98,46 @@ export class ReplicaSyncClient {
return data.rows ?? [];
}
/**
* Batched pull for the incremental auto-sync path. Collapses what
* used to be N parallel `GET /api/sync/replicas?kind=…&since=…`
* requests (one per replica kind, fired on every focus / online /
* periodic trigger) into a single POST round-trip. The boot path
* still uses `pull()` per kind to preserve the settings-first
* ordering invariant.
*
* The endpoint is the same `/sync/replicas` route — `{ cursors: [...] }`
* in the body discriminates batched-pull from `{ rows: [...] }` push.
*/
async pullBatch(
cursors: { kind: string; since: Hlc | null }[],
): Promise<{ kind: string; rows: ReplicaRow[] }[]> {
if (cursors.length === 0) return [];
const token = await requireToken();
let response: Response;
try {
response = await fetch(ENDPOINT(), {
method: 'POST',
headers: {
Authorization: `Bearer ${token}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({ cursors }),
});
} catch (cause) {
throw new SyncError('SERVER', 'Network failure during batch pull', { cause });
}
if (!response.ok) {
const body = await parseErrorBody(response);
const code = body.code ?? statusToDefaultCode(response.status);
throw new SyncError(code, body.error ?? `Batch pull failed with status ${response.status}`, {
status: response.status,
});
}
const data = (await response.json()) as { results: { kind: string; rows: ReplicaRow[] }[] };
return data.results ?? [];
}
/**
* The replica_keys list rarely changes — only on `createReplicaKey`
* (passphrase setup / rotation) and `forgetReplicaKeys` (forgot-
@@ -5,6 +5,11 @@ import type { SyncErrorCode } from '@/libs/errors';
export const HLC_SKEW_TOLERANCE_MS = 60_000;
export const MAX_PUSH_BATCH = 100;
// Cap the batched-pull cursor list. Today there are 5 kinds; this leaves
// generous headroom for future replica kinds while keeping the request
// body bounded so a malicious caller can't burn server time on
// pathological queries.
export const MAX_PULL_BATCH = 50;
export interface PushReplicasBody {
rows: ReplicaRow[];
@@ -119,3 +124,83 @@ export const validatePullParams = (kind: string | null, since: string | null): P
},
};
};
export interface PullBatchEntry {
kind: string;
since: Hlc | null;
}
export interface PullBatchBody {
cursors: PullBatchEntry[];
}
export type PullBatchValidation =
| { ok: true; params: PullBatchBody }
| { ok: false; status: number; code: SyncErrorCode; message: string; offendingIndex?: number };
/**
* Validate the batched-pull body. Same per-row checks as the per-kind
* GET (`isAllowedKind`), but applied to every cursor entry.
*/
export const validatePullBatch = (body: unknown): PullBatchValidation => {
if (typeof body !== 'object' || body === null) {
return { ok: false, status: 400, code: 'VALIDATION', message: 'body must be an object' };
}
const cursors = (body as PullBatchBody).cursors;
if (!Array.isArray(cursors)) {
return { ok: false, status: 400, code: 'VALIDATION', message: 'body.cursors must be an array' };
}
if (cursors.length === 0) {
return { ok: true, params: { cursors: [] } };
}
if (cursors.length > MAX_PULL_BATCH) {
return {
ok: false,
status: 413,
code: 'VALIDATION',
message: `cursors.length ${cursors.length} exceeds MAX_PULL_BATCH=${MAX_PULL_BATCH}`,
};
}
const seen = new Set<string>();
for (let i = 0; i < cursors.length; i++) {
const c = cursors[i]!;
if (typeof c.kind !== 'string' || c.kind.length === 0) {
return {
ok: false,
status: 400,
code: 'VALIDATION',
message: `cursors[${i}].kind must be a non-empty string`,
offendingIndex: i,
};
}
if (!isAllowedKind(c.kind)) {
return {
ok: false,
status: 422,
code: 'UNKNOWN_KIND',
message: `cursors[${i}].kind=${c.kind} is not in the server allowlist`,
offendingIndex: i,
};
}
if (seen.has(c.kind)) {
return {
ok: false,
status: 400,
code: 'VALIDATION',
message: `cursors[${i}].kind=${c.kind} is duplicated in the batch`,
offendingIndex: i,
};
}
seen.add(c.kind);
if (c.since !== null && typeof c.since !== 'string') {
return {
ok: false,
status: 400,
code: 'VALIDATION',
message: `cursors[${i}].since must be a string or null`,
offendingIndex: i,
};
}
}
return { ok: true, params: { cursors } };
};
@@ -3,7 +3,7 @@ import { NextRequest, NextResponse } from 'next/server';
import { createSupabaseClient } from '@/utils/supabase';
import { validateUserAndToken } from '@/utils/access';
import { runMiddleware, corsAllMethods } from '@/utils/cors';
import { validatePullParams, validatePushBatch } from '@/libs/replicaSyncServer';
import { validatePullBatch, validatePullParams, validatePushBatch } from '@/libs/replicaSyncServer';
import type { ReplicaRow } from '@/types/replica';
const errorResponse = (status: number, code: string, message: string, offendingIndex?: number) =>
@@ -30,6 +30,50 @@ export async function POST(req: NextRequest) {
return errorResponse(400, 'VALIDATION', 'Invalid JSON body');
}
// Body discriminator: `{ cursors: [...] }` is a batched pull (replaces
// N parallel `GET ?kind=K&since=…` calls with a single Worker
// invocation); `{ rows: [...] }` is the existing push.
if (typeof body === 'object' && body !== null && 'cursors' in body) {
const validation = validatePullBatch(body);
if (!validation.ok) {
return errorResponse(
validation.status,
validation.code,
validation.message,
validation.offendingIndex,
);
}
const { cursors } = validation.params;
if (cursors.length === 0) {
return NextResponse.json({ results: [] }, { status: 200 });
}
// Per-kind queries run in parallel: each is the same SELECT the
// single-kind GET issues, just dispatched together. Supabase calls
// inside a Worker aren't billed as Cloudflare requests, so this
// collapses N Worker invocations to 1 without changing DB load.
try {
const tasks = cursors.map(async ({ kind, since }) => {
let query = supabase
.from('replicas')
.select('*')
.eq('user_id', user.id)
.eq('kind', kind)
.order('updated_at_ts', { ascending: true })
.limit(1000);
if (since) query = query.gt('updated_at_ts', since);
const { data, error } = await query;
if (error) throw new Error(`pull replicas (kind=${kind}) failed: ${error.message}`);
return { kind, rows: (data ?? []) as ReplicaRow[] };
});
const results = await Promise.all(tasks);
return NextResponse.json({ results }, { status: 200 });
} catch (error) {
console.error('batch pull replicas failed', { cursors, error });
const message = error instanceof Error ? error.message : 'unknown error';
return errorResponse(500, 'SERVER', message);
}
}
const validation = validatePushBatch(body, user.id, Date.now());
if (!validation.ok) {
return errorResponse(
@@ -10,7 +10,7 @@ export interface ReplicaSyncInitOpts {
deviceId: string;
cursorStore: CursorStore;
hlcStore?: HlcSnapshotStore;
client?: Pick<ReplicaSyncClient, 'push' | 'pull'>;
client?: Pick<ReplicaSyncClient, 'push' | 'pull' | 'pullBatch'>;
}
export interface ReplicaSyncContext {
@@ -9,7 +9,7 @@ export interface CursorStore {
export interface ReplicaSyncManagerOpts {
hlc: HlcGenerator;
client: Pick<ReplicaSyncClient, 'push' | 'pull'>;
client: Pick<ReplicaSyncClient, 'push' | 'pull' | 'pullBatch'>;
cursorStore: CursorStore;
debounceMs?: number;
}
@@ -131,14 +131,52 @@ export class ReplicaSyncManager {
// sync (visibility / online) keeps using the cursor.
const since = opts && 'since' in opts ? (opts.since ?? null) : this.opts.cursorStore.get(kind);
const rows = await this.opts.client.pull(kind, since);
if (rows.length === 0) return rows;
this.observeAndAdvanceCursor(kind, rows);
return rows;
}
/**
* Batched pull for the incremental auto-sync path AND the boot full
* re-fetch. Default behaviour (no opts): each kind uses its
* persisted cursor — the cheap delta path used by focus / online /
* periodic. Boot path passes `{ since: null }` so the same single
* round-trip refreshes every kind from scratch, mirroring the
* recovery semantics of the per-kind `pull(kind, { since: null })`
* boot call.
*
* Returns a `Map<kind, rows>`. Kinds present in the input but missing
* from the response (because they had no rows past the cursor) are
* still mapped to an empty array, so callers can iterate over the
* input kinds without checking for `undefined`.
*/
async pullMany(
kinds: string[],
opts?: { since?: Hlc | null },
): Promise<Map<string, ReplicaRow[]>> {
const out = new Map<string, ReplicaRow[]>();
if (kinds.length === 0) return out;
const overrideSince = opts && 'since' in opts;
const cursors = kinds.map((kind) => ({
kind,
since: overrideSince ? (opts.since ?? null) : this.opts.cursorStore.get(kind),
}));
const results = await this.opts.client.pullBatch(cursors);
for (const kind of kinds) out.set(kind, []);
for (const { kind, rows } of results) {
this.observeAndAdvanceCursor(kind, rows);
out.set(kind, rows);
}
return out;
}
private observeAndAdvanceCursor(kind: string, rows: ReplicaRow[]): void {
if (rows.length === 0) return;
let maxHlc: Hlc = rows[0]!.updated_at_ts;
for (const row of rows) {
if (hlcCompare(row.updated_at_ts, maxHlc) > 0) maxHlc = row.updated_at_ts;
this.opts.hlc.observe(row.updated_at_ts);
}
this.opts.cursorStore.set(kind, maxHlc);
return rows;
}
startAutoSync(): void {