diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f3cd034..7ecb367f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,9 @@ return `ItemRemovedError`; use `isItemRemovedError()` to handle this case. Also included: +- `useSyncStatus()` for one canonical application-facing view of connectivity, active reads, + queued and retryable failed writes, event/reconnect reconciliation, and the last fully synced + time. - Import-safe schema bindings through `createHooks(schema)`. The generated hooks resolve their runtime from `FigbirdProvider`, and `useMutations()` returns that instance's typed write proxy. Imperative code uses `figbird.m`, `figbird.prepare`, and other instance methods directly, so diff --git a/README.md b/README.md index f87dcc5f..94fbc974 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ const figbird = new Figbird({ schema, }) -export const { useQuery, useMutations, useAction, q } = createHooks(schema) +export const { useQuery, useMutations, useAction, useSyncStatus, q } = createHooks(schema) function Notes() { const { data: notes } = useQuery(q.notes.where({ read: false }).related('author')) diff --git a/demo/README.md b/demo/README.md index a1d651a5..dbe3c88f 100644 --- a/demo/README.md +++ b/demo/README.md @@ -56,6 +56,8 @@ If you're here to learn figbird, read in this order — each file teaches one id 8. **`src/components/DemoControls.tsx`** — demo-server switches for latency, background traffic, forced failures, socket reconnects, and resets. The Figbird browser extension provides the query, timeline, event, and write views. +9. **`src/components/SyncStatusIndicator.tsx`** — one `useSyncStatus()` snapshot rendered + as the nav's offline/saving/restoring/saved indicator. Structure rule: `src/` root is wiring (`main`, `figbird`, `demoControl`, `App`); `src/components/` is the shell UI the workspace composes; `src/pages/` is routed screens. @@ -76,6 +78,7 @@ The bottom-right **Demo controls** menu changes server behavior: queue pauses instead, keeping its projection visible until you choose Retry or Discard. - **Drop socket** — kills the transport; socket.io auto-reconnects and figbird refetches every active query (and the materialized reference sets) to reconcile anything missed. + The nav indicator moves through offline → restoring → saved from `useSyncStatus()`. Install the extension from `extensions/build/chrome` or `extensions/build/firefox`, open the browser's developer tools, and select the **Figbird** panel. It exposes live queries, diff --git a/demo/src/App.tsx b/demo/src/App.tsx index c253683f..3a92e4e8 100644 --- a/demo/src/App.tsx +++ b/demo/src/App.tsx @@ -14,6 +14,7 @@ import { figbird } from './figbird' import { issueDetailRouteQueries } from './pages/IssueDetail/queries' import { TeamsPage } from './pages/Teams/screen' import { DetailSkeleton, SkeletonRows } from './components/ui' +import { SyncStatusIndicator } from './components/SyncStatusIndicator' function EmptyDetail() { return ( @@ -106,6 +107,7 @@ function Workspace({ children }: { children?: ReactNode }) { + New issue + tip: open two windows side by side {isFull ? ( diff --git a/demo/src/components/SyncStatusIndicator.tsx b/demo/src/components/SyncStatusIndicator.tsx new file mode 100644 index 00000000..ddef74ab --- /dev/null +++ b/demo/src/components/SyncStatusIndicator.tsx @@ -0,0 +1,40 @@ +import { useSyncStatus } from '../figbird' + +function plural(count: number, word: string): string { + return `${count} ${word}${count === 1 ? '' : 's'}` +} + +export function SyncStatusIndicator() { + const sync = useSyncStatus() + const label = + sync.phase === 'offline' + ? 'Working offline' + : sync.phase === 'error' + ? sync.failedWrites > 0 + ? `Couldn’t sync ${plural(sync.failedWrites, 'change')}` + : 'Couldn’t refresh data' + : sync.phase === 'restoring' + ? 'Refreshing stale data…' + : sync.phase === 'syncing' + ? `Saving ${plural(sync.pendingWrites, 'change')}…` + : 'Everything saved' + + const detail = [ + `${plural(sync.pendingWrites, 'pending write')}`, + `${plural(sync.failedWrites, 'failed write')}`, + `${plural(sync.fetchingQueries, 'fetching query')}`, + `${plural(sync.pendingReconciliations, 'pending reconciliation')}`, + sync.lastSyncedAt === null + ? 'Not synced yet' + : `Last synced ${new Date(sync.lastSyncedAt).toLocaleTimeString()}`, + ].join(' · ') + + return ( +
+
+ ) +} diff --git a/demo/src/figbird.ts b/demo/src/figbird.ts index 0ebb2952..8467f51b 100644 --- a/demo/src/figbird.ts +++ b/demo/src/figbird.ts @@ -170,6 +170,7 @@ export const { useAction, useMutating, useMutationQueue, + useSyncStatus, } = createHooks(schema) // Reference data: preload the complete sets once — realtime maintains them, and every diff --git a/demo/src/styles.css b/demo/src/styles.css index d837480c..013bf2b5 100644 --- a/demo/src/styles.css +++ b/demo/src/styles.css @@ -73,6 +73,83 @@ body { font-size: 12px; } +.sync-status { + --sync-color: var(--text-dim); + display: inline-flex; + align-items: center; + gap: 7px; + min-width: 118px; + padding: 3px 9px 3px 6px; + border: 1px solid var(--border); + border-radius: 999px; + color: var(--text-muted); + background: var(--bg); + font-size: 11.5px; + font-variant-numeric: tabular-nums; + transition: + color 140ms ease, + border-color 140ms ease, + background 140ms ease; +} + +.sync-status.synced { + --sync-color: var(--open); +} + +.sync-status.offline { + --sync-color: #d97706; + color: #92400e; + border-color: #fde2b7; + background: #fffaf0; +} + +.sync-status.error { + --sync-color: var(--danger); + color: #991b1b; + border-color: #fecaca; + background: #fff5f5; +} + +.sync-status-orbit { + position: relative; + width: 14px; + height: 14px; + flex: 0 0 auto; + border: 1px solid color-mix(in srgb, var(--sync-color) 42%, transparent); + border-radius: 50%; +} + +.sync-status-core { + position: absolute; + inset: 4px; + border-radius: 50%; + background: var(--sync-color); +} + +.sync-status.syncing .sync-status-orbit, +.sync-status.restoring .sync-status-orbit { + border-top-color: var(--sync-color); + animation: sync-orbit 800ms linear infinite; +} + +.sync-status.error .sync-status-core { + border-radius: 1px; + transform: rotate(45deg); +} + +@keyframes sync-orbit { + to { + transform: rotate(1turn); + } +} + +@media (prefers-reduced-motion: reduce) { + .sync-status.syncing .sync-status-orbit, + .sync-status.restoring .sync-status-orbit { + animation: none; + } +} + /* ----- Grid ----- */ .grid { diff --git a/docs/content/_index.md b/docs/content/_index.md index 50ac70fb..00a9685c 100644 --- a/docs/content/_index.md +++ b/docs/content/_index.md @@ -1511,6 +1511,35 @@ Returns a boolean, live via `useSyncExternalStore` over `figbird.mutating`, the synchronous tracker. It's correct for components that mount mid-mutation and it sees writes from any surface. See [Entity-level activity](#entity-level-activity-usemutating). +## useSyncStatus + +```ts +const sync = useSyncStatus() + +sync.phase // 'restoring' | 'offline' | 'syncing' | 'synced' | 'error' +sync.pendingWrites +sync.failedWrites +sync.fetchingQueries +sync.pendingReconciliations +sync.lastSyncedAt // epoch milliseconds, or null before the first successful settle +``` + +Returns Figbird's canonical, instance-wide sync snapshot. Unlike the observability event +stream, this state replays: a component mounting halfway through a fetch, scheduled mutation, +paused queue failure, disconnect, or hidden-tab reconciliation sees the correct answer +immediately. `offline` takes priority while the adapter transport is disconnected; `error` +covers failed writes or reconciliation refreshes; `restoring` covers connection setup and +event/reconnect reconciliation; pending writes are `syncing`; otherwise the instance is `synced`. +Ordinary query fetches update `fetchingQueries` without changing `phase`, so a global saved/saving +indicator does not flash during normal screen-level data loading. + +`pendingWrites` includes scheduled queue work and a failed queue item that still needs retry or +discard. `failedWrites` counts that retryable queue work; an ordinary rejected action settles out +of the global snapshot because its caller owns the error and Figbird no longer has work to retry. +`lastSyncedAt` advances only when successful work leaves the whole instance clean after a write or +reconciliation. The hook is backed by `figbird.sync` and +`useSyncExternalStore`, so it is also available from a schema-bound `createHooks` kit. + ## defineQuery ```ts @@ -1665,6 +1694,7 @@ const figbird = new Figbird({ | `m` | The instance’s write proxy: `figbird.m.issues.patch(...)`, or `figbird.m(service)` for dynamic names. In React, access the provider instance through `useMutations()`. See [m](#m). | | `createMutationQueue(config?)` | Explicitly owned serial writes across records or services. See [figbird.createMutationQueue](#figbirdcreatemutationqueue). | | `mutating` | Synchronous active-mutation tracker (`subscribe`/`getSnapshot`) — `useMutating` is its React binding. | +| `sync` | Canonical aggregate sync tracker (`subscribe`/`getSnapshot`) — `useSyncStatus` is its React binding. | | `explain(...)` | Static classification report — see [figbird.explain](#figbirdexplain). | | `inspect()` | Live-query snapshot — see [figbird.inspect](#figbirdinspect). | | `events` | Observability channel — see [figbird.events](#figbirdevents). | @@ -1699,13 +1729,22 @@ Meta behavior: `find` returns `{ data, meta }` (`FindMeta`: `{ total, limit, ski Binds a schema to import-safe, typed React hooks: ```ts -export const { useQuery, useFigbird, useMutations, q, defineQuery, useAction, useMutating } = - createHooks(schema) +export const { + useQuery, + useFigbird, + useMutations, + q, + defineQuery, + useAction, + useMutating, + useSyncStatus, +} = createHooks(schema) ``` Returns the daily-use kit: `useQuery`, `q` (the read proxy), schema-typed `defineQuery`, and the write side — `useMutations` (the provider instance's write proxy), -`useAction` (per-action state), and `useMutating` (in-flight activity). It also includes +`useAction` (per-action state), `useMutating` (in-flight activity), and `useSyncStatus` +(instance-wide connectivity and sync state). It also includes typed `useFigbird`, `useFeathers` (the raw-client escape hatch), and the deprecated legacy hooks (`useMutation`, `useFind`, `useGet`) for older codebases. diff --git a/lib/adapters/adapter.ts b/lib/adapters/adapter.ts index b84e3548..43caae82 100644 --- a/lib/adapters/adapter.ts +++ b/lib/adapters/adapter.ts @@ -61,6 +61,9 @@ export interface EventHandlers { removed: (item: unknown) => void } +/** Adapter transport state consumed by Figbird's canonical sync snapshot. */ +export type AdapterConnectionState = 'connecting' | 'connected' | 'disconnected' + /** Service context supplied when the adapter evaluates a query locally. */ export interface MatcherContext { serviceName: string @@ -102,6 +105,12 @@ export interface Adapter< // reconnects after a period where realtime events may have been missed. subscribeToReconnect?(handler: () => void): () => void + /** Current transport state. Omit when the adapter has no meaningful connection lifecycle. */ + getConnectionState?(): AdapterConnectionState + + /** Notify whenever `getConnectionState()` may have changed. */ + subscribeToConnectionState?(handler: () => void): () => void + /** * Read an item's id, or `undefined` when absent. Pure extraction — whether a * missing id is noteworthy is the store's call (it warns on event/fetch paths diff --git a/lib/adapters/feathers.ts b/lib/adapters/feathers.ts index bc7b6045..f8d45399 100644 --- a/lib/adapters/feathers.ts +++ b/lib/adapters/feathers.ts @@ -1,5 +1,6 @@ import type { Adapter, + AdapterConnectionState, EventHandlers, MatcherContext, PageCursor, @@ -228,6 +229,11 @@ interface ReconnectEventSource { removeListener?: (event: string, listener: () => void) => void } +interface ConnectionEventSource extends ReconnectEventSource { + connected?: boolean + active?: boolean +} + /** * Typed Feathers service for a specific service in the schema. * Provides full type safety for CRUD methods and custom methods. @@ -353,6 +359,7 @@ export class FeathersAdapter> implements Adapte #operators: Record #defaultPagination: FeathersPagination | undefined #pagination: Record + #observedConnectionState: AdapterConnectionState | undefined /** Names of custom operators registered for every service. */ get customOperators(): readonly string[] { @@ -633,6 +640,39 @@ export class FeathersAdapter> implements Adapte } } + getConnectionState(): AdapterConnectionState { + if (this.#observedConnectionState) return this.#observedConnectionState + const socket = this.#getConnectionEventSource() + if (!socket) return 'connected' + if (socket.connected === true) return 'connected' + if (socket.active === true) return 'connecting' + return socket.connected === false ? 'disconnected' : 'connected' + } + + subscribeToConnectionState(handler: () => void): () => void { + const socket = this.#getConnectionEventSource() + if (!socket) return () => {} + const connected = () => { + this.#observedConnectionState = 'connected' + handler() + } + const disconnected = () => { + this.#observedConnectionState = 'disconnected' + handler() + } + socket.on('connect', connected) + socket.on('disconnect', disconnected) + return () => { + if (socket.off) { + socket.off('connect', connected) + socket.off('disconnect', disconnected) + } else { + socket.removeListener?.('connect', connected) + socket.removeListener?.('disconnect', disconnected) + } + } + } + #getReconnectEventSource(): ReconnectEventSource | null { const io = (this.feathers as { io?: { io?: unknown } }).io const candidates = [ @@ -659,6 +699,25 @@ export class FeathersAdapter> implements Adapte return null } + #getConnectionEventSource(): ConnectionEventSource | null { + const candidates = [ + (this.feathers as { io?: unknown }).io, + (this.feathers as { socket?: unknown }).socket, + (this.feathers as { primus?: unknown }).primus, + ] + for (const candidate of candidates) { + if ( + candidate && + typeof candidate === 'object' && + 'on' in candidate && + typeof candidate.on === 'function' + ) { + return candidate as ConnectionEventSource + } + } + return null + } + getId(item: unknown): string | number | undefined { return typeof this.#idField === 'string' ? ((item as Record)[this.#idField] as string | number | undefined) diff --git a/lib/core/figbird.ts b/lib/core/figbird.ts index 834b1e36..5e9de72f 100644 --- a/lib/core/figbird.ts +++ b/lib/core/figbird.ts @@ -18,6 +18,7 @@ import { type MutationQueueHost, } from './mutationQueue.js' import type { MutationActivity } from './mutationTracker.js' +import type { SyncActivity } from './syncTracker.js' import { createQueryBuilderProxy, queryBuilderUsesSchema, @@ -108,6 +109,7 @@ export type { MutationSchedule, } from './mutationQueue.js' export type { InFlightMutation, MutationActivity } from './mutationTracker.js' +export type { SyncActivity, SyncPhase, SyncStatus } from './syncTracker.js' export { defineQuery, isQueryDefinition, @@ -792,6 +794,11 @@ export class Figbird< return this.queryStore.mutations } + /** Canonical aggregate state behind `useSyncStatus()`. */ + get sync(): SyncActivity { + return this.queryStore.sync + } + /** * Manually refetch cached queries — the escape hatch for changes figbird cannot * observe (custom methods on services without realtime events, out-of-band diff --git a/lib/core/mutationTracker.ts b/lib/core/mutationTracker.ts index e52509a4..501df13c 100644 --- a/lib/core/mutationTracker.ts +++ b/lib/core/mutationTracker.ts @@ -25,8 +25,8 @@ export interface InFlightMutation { } /** - * Read-only view of the tracker — what `figbird.mutating` exposes. `start`/`end` - * are internal to the store. + * Read-only view of the tracker — what `figbird.mutating` exposes. Mutation + * lifecycle methods are internal to the store. */ export interface MutationActivity { /** Notifies synchronously whenever the active set changes. */ @@ -35,22 +35,55 @@ export interface MutationActivity { getSnapshot(): readonly InFlightMutation[] } +type MutationSyncState = 'pending' | 'retry-paused' + +interface TrackedMutationEntry extends InFlightMutation { + syncState: MutationSyncState +} + +export interface MutationSyncSnapshot { + readonly pendingWrites: number + readonly failedWrites: number +} + +type MutationSettlement = 'success' | 'failure' | 'discarded' + +interface MutationSyncChange { + successful: boolean +} + export class MutationTracker implements MutationActivity { - #inFlight: Map = new Map() + #inFlight: Map = new Map() #listeners: Set<() => void> = new Set() + #syncListeners: Set<(change: MutationSyncChange) => void> = new Set() #nextId = 1 #snapshot: readonly InFlightMutation[] = [] + #syncSnapshot: MutationSyncSnapshot = { pendingWrites: 0, failedWrites: 0 } start(entry: { serviceName: string; method: string; id?: string | number }): number { const mutationId = this.#nextId++ - this.#inFlight.set(mutationId, { mutationId, ...entry }) + this.#inFlight.set(mutationId, { mutationId, ...entry, syncState: 'pending' }) this.#changed() return mutationId } - end(mutationId: number): void { + attemptFailed(mutationId: number): void { + const mutation = this.#inFlight.get(mutationId) + if (!mutation || mutation.syncState === 'retry-paused') return + mutation.syncState = 'retry-paused' + this.#changed({ activityChanged: false }) + } + + attemptRetrying(mutationId: number): void { + const mutation = this.#inFlight.get(mutationId) + if (!mutation || mutation.syncState === 'pending') return + mutation.syncState = 'pending' + this.#changed({ activityChanged: false }) + } + + settle(mutationId: number, outcome: MutationSettlement): void { if (this.#inFlight.delete(mutationId)) { - this.#changed() + this.#changed({ successful: outcome === 'success' }) } } @@ -65,8 +98,31 @@ export class MutationTracker implements MutationActivity { } } - #changed(): void { - this.#snapshot = Array.from(this.#inFlight.values()) + getSyncSnapshot(): MutationSyncSnapshot { + return this.#syncSnapshot + } + + subscribeToSync(listener: (change: MutationSyncChange) => void): () => void { + this.#syncListeners.add(listener) + return () => { + this.#syncListeners.delete(listener) + } + } + + #changed({ + successful = false, + activityChanged = true, + }: { successful?: boolean; activityChanged?: boolean } = {}): void { + const mutations = Array.from(this.#inFlight.values()) + if (activityChanged) { + this.#snapshot = mutations.map(({ syncState: _, ...mutation }) => mutation) + } + this.#syncSnapshot = { + pendingWrites: mutations.length, + failedWrites: mutations.filter(mutation => mutation.syncState === 'retry-paused').length, + } + for (const listener of this.#syncListeners) listener({ successful }) + if (!activityChanged) return for (const fn of this.#listeners) { try { fn() diff --git a/lib/core/queryStore.ts b/lib/core/queryStore.ts index 724b7e9e..3972e341 100644 --- a/lib/core/queryStore.ts +++ b/lib/core/queryStore.ts @@ -71,6 +71,7 @@ import { type ServiceState, } from './queryTypes.js' import { defaultRetryDelay, resolveRetryDelay } from './retryDelay.js' +import { SyncTracker } from './syncTracker.js' /** * Where the store learns whether the tab is visible. Injectable for tests and @@ -119,11 +120,18 @@ interface TrackedMutation { promise: Promise } +interface StartedMutation { + mutationId: number + startedAt: number + tracking: MutationTrackingEntry +} + interface QueuedMutation { desc: MutationDescriptor args: unknown[] optimistic: boolean attempt: GatedMutationAttempt + mutationId: number } interface AppliedEventEffect { @@ -169,6 +177,7 @@ export class QueryStore< #adapter: Adapter #events: FigbirdEventEmitter #mutations: MutationTracker + #sync: SyncTracker #realtime: Set = new Set() #listeners: Map) => void>> = new Map() @@ -200,6 +209,7 @@ export class QueryStore< #retryDelay: RetryDelay #reconnectJitter: readonly [number, number] #reconnectSweepTimer: ReturnType | null = null + #scheduledReconnectQueryIds: Set = new Set() #reconnectQueryIds: Set = new Set() #warnedMissingIdServices: Set = new Set() @@ -253,12 +263,19 @@ export class QueryStore< this.#eventBatchInterval = eventBatchInterval this.#events = new FigbirdEventEmitter() this.#mutations = new MutationTracker() + this.#sync = new SyncTracker( + this.#mutations, + this.#adapter.getConnectionState?.() ?? 'connected', + ) this.#reconcileCooldown = reconcileCooldown this.#retry = this.#normalizeRetry(retry) this.#retryDelay = retryDelay this.#reconnectJitter = this.#normalizeReconnectJitter(reconnectJitter) this.#visibility = visibility ?? documentVisibility() this.#visibility.onChange(() => this.#drainDeferredReconciles()) + this.#adapter.subscribeToConnectionState?.(() => { + this.#sync.connectionChanged(this.#adapter.getConnectionState?.() ?? 'connected') + }) this.#adapter.subscribeToReconnect?.(() => this.#scheduleReconnectSweep()) } @@ -273,6 +290,11 @@ export class QueryStore< return this.#mutations } + /** The instance's canonical aggregate sync state. */ + get sync(): SyncTracker { + return this.#sync + } + /** Returns the entire store state map keyed by service name. */ getState(): Map> { return this.#state @@ -661,39 +683,37 @@ export class QueryStore< this.#getEntity(desc.serviceName, id), ) + const tracking = { + serviceName: desc.serviceName, + method: desc.method, + id, + optimistic, + args, + } + const started = this.#beginMutation(tracking) const entry: QueuedMutation = { desc, args, optimistic, attempt: new GatedMutationAttempt(control), + mutationId: started.mutationId, } - const tracked = this.#trackMutation( - { - serviceName: desc.serviceName, - method: desc.method, - id, - optimistic, - args, - }, - () => entry.attempt.promise, - { - onSuccess: item => this.#settleQueuedMutation(lane, entry, { ok: true, item }), - onError: (error, mutationId) => { - this.#settleQueuedMutation(lane, entry, { ok: false, error }) - if (optimistic) { - this.#events.emit({ - kind: 'mutate:rollback', - mutationId, - serviceName: desc.serviceName, - method: desc.method, - id, - }) - } - }, + const tracked = this.#observeMutation(started, () => entry.attempt.promise, { + onSuccess: item => this.#settleQueuedMutation(lane, entry, { ok: true, item }), + onError: (error, mutationId) => { + this.#settleQueuedMutation(lane, entry, { ok: false, error }) + if (optimistic) { + this.#events.emit({ + kind: 'mutate:rollback', + mutationId, + serviceName: desc.serviceName, + method: desc.method, + id, + }) + } }, - ) - + }) this.#applyProjection(this.#mutationLanes.enqueue(lane, entry), true) entry.attempt.whenReady(() => { this.#expediteMutationPredecessors(lane, entry) @@ -733,8 +753,10 @@ export class QueryStore< return } entry.attempt.start(() => - this.#runControlledAttempt(entry.attempt.control, () => - this.#adapter.mutate(lane.serviceName, entry.desc.method, [...entry.args]), + this.#runControlledAttempt( + entry.attempt.control, + () => this.#adapter.mutate(lane.serviceName, entry.desc.method, [...entry.args]), + entry.mutationId, ), ) } @@ -749,6 +771,7 @@ export class QueryStore< async #runControlledAttempt( control: ScheduledMutationControl | undefined, run: () => Promise, + mutationId: number, ): Promise { let attempt = 0 while (true) { @@ -758,9 +781,12 @@ export class QueryStore< return await run() } catch (error) { const normalized = error instanceof Error ? error : new Error(String(error)) - if (!control || (await control.onAttemptFailure(normalized, attempt)) === 'discard') { + if (!control) throw normalized + this.#mutations.attemptFailed(mutationId) + if ((await control.onAttemptFailure(normalized, attempt)) === 'discard') { throw normalized } + this.#mutations.attemptRetrying(mutationId) } } } @@ -891,7 +917,7 @@ export class QueryStore< ) const start = () => { - attempt.start(() => this.#runControlledAttempt(control, run)) + attempt.start(() => this.#runControlledAttempt(control, run, tracked.mutationId)) } attempt.whenReady(start) @@ -918,7 +944,11 @@ export class QueryStore< run: () => Promise, hooks?: MutationTrackingHooks, ): TrackedMutation { - const { serviceName, method, id, optimistic, args } = entry + return this.#observeMutation(this.#beginMutation(entry), run, hooks) + } + + #beginMutation(tracking: MutationTrackingEntry): StartedMutation { + const { serviceName, method, id, optimistic, args } = tracking const idField = id !== undefined ? { id } : {} const startedAt = Date.now() const mutationId = this.#mutations.start({ serviceName, method, ...idField }) @@ -931,10 +961,20 @@ export class QueryStore< optimistic, args, }) + return { mutationId, startedAt, tracking } + } + + #observeMutation( + { mutationId, startedAt, tracking }: StartedMutation, + run: () => Promise, + hooks?: MutationTrackingHooks, + ): TrackedMutation { + const { serviceName, method, id, optimistic } = tracking + const idField = id !== undefined ? { id } : {} const promise = run().then( result => { hooks?.onSuccess?.(result) - this.#mutations.end(mutationId) + this.#mutations.settle(mutationId, 'success') this.#events.emit({ kind: 'mutate:end', mutationId, @@ -949,7 +989,10 @@ export class QueryStore< (err: unknown) => { const error = err instanceof Error ? err : new Error(String(err)) hooks?.onError?.(error, mutationId) - this.#mutations.end(mutationId) + this.#mutations.settle( + mutationId, + error instanceof MutationSupersededError ? 'discarded' : 'failure', + ) this.#events.emit({ kind: 'mutate:error', mutationId, @@ -968,38 +1011,51 @@ export class QueryStore< // Query lifecycle async #queue(queryId: string): Promise { - this.#fetching({ queryId }) - const generation = this.#queryGenerations.get(queryId) - if (generation === undefined) return + this.#sync.queryStarted(queryId) + let syncOutcome: 'success' | 'error' | 'cancelled' = 'cancelled' + try { + this.#fetching({ queryId }) + const generation = this.#queryGenerations.get(queryId) + if (generation === undefined) return + + let retryAttempt = 0 + while (true) { + const outcome = await this.#runFetchAttempt(queryId, generation) + if (outcome.kind === 'completed') { + syncOutcome = 'success' + return + } + if (outcome.kind === 'stale') return - let retryAttempt = 0 - while (true) { - const outcome = await this.#runFetchAttempt(queryId, generation) - if (outcome.kind !== 'failed') return - - const query = this.#getQuery(queryId) - if ( - !query || - this.#queryGenerations.get(queryId) !== generation || - !this.#hasRetryOwner(queryId) || - !this.#shouldRetry(query, retryAttempt, outcome.error) - ) { - if (query && this.#queryGenerations.get(queryId) === generation) { - this.#fetchFailed({ queryId, error: outcome.error }) + const query = this.#getQuery(queryId) + if ( + !query || + this.#queryGenerations.get(queryId) !== generation || + !this.#hasRetryOwner(queryId) || + !this.#shouldRetry(query, retryAttempt, outcome.error) + ) { + if (query && this.#queryGenerations.get(queryId) === generation) { + this.#fetchFailed({ queryId, error: outcome.error }) + syncOutcome = 'error' + } + return } - return - } - retryAttempt++ - const configuredDelay = query.config.retryDelay ?? this.#retryDelay - const delay = this.#resolveRetryDelay(configuredDelay, retryAttempt, outcome.error) - await new Promise(resolve => setTimeout(resolve, delay)) + retryAttempt++ + const configuredDelay = query.config.retryDelay ?? this.#retryDelay + const delay = this.#resolveRetryDelay(configuredDelay, retryAttempt, outcome.error) + await new Promise(resolve => setTimeout(resolve, delay)) - if (this.#queryGenerations.get(queryId) !== generation) return - if (!this.#hasRetryOwner(queryId)) { - this.#fetchFailed({ queryId, error: outcome.error }) - return + if (this.#queryGenerations.get(queryId) !== generation) return + if (!this.#hasRetryOwner(queryId)) { + this.#fetchFailed({ queryId, error: outcome.error }) + syncOutcome = 'error' + return + } } + } finally { + this.#sync.queryFinished(queryId, syncOutcome) + this.#maybeFinishReconciliation(queryId) } } @@ -1915,11 +1971,18 @@ export class QueryStore< * `reconcileCooldown` coalesce into one guaranteed trailing refetch. */ #requestReconcile(queryId: string, { force = false }: { force?: boolean } = {}): void { + if (!this.#getQuery(queryId)) { + this.#sync.reconciliationFinished(queryId) + return + } if (!force && this.#listenerCount(queryId) === 0) { this.#markQueryPending(queryId) + this.#sync.reconciliationFinished(queryId) return } + this.#sync.reconciliationStarted(queryId) + if (this.#visibility.isHidden()) { this.#deferredWhileHidden.add(queryId) this.#markQueryPending(queryId) @@ -1986,6 +2049,22 @@ export class QueryStore< if (window?.trailing) clearTimeout(window.trailing) this.#reconcileWindows.delete(queryId) this.#deferredWhileHidden.delete(queryId) + this.#scheduledReconnectQueryIds.delete(queryId) + this.#sync.forgetQuery(queryId) + } + + #maybeFinishReconciliation(queryId: string): void { + const query = this.#getQuery(queryId) + const window = this.#reconcileWindows.get(queryId) + if ( + this.#deferredWhileHidden.has(queryId) || + window?.trailing || + query?.state.isFetching || + query?.dirty + ) { + return + } + this.#sync.reconciliationFinished(queryId) } /** @@ -2017,12 +2096,13 @@ export class QueryStore< } } - #refetchActiveQueries(): void { + #activeReconnectQueries(): Map { + const targets = new Map() for (const service of this.getState().values()) { // Materialization roots reconcile even with no subscribers — every local read // depends on their completeness, and events may have been missed while offline. if (service.materialized) { - this.#requestReconcile(service.materialized.queryId, { force: true }) + targets.set(service.materialized.queryId, true) } for (const query of service.queries.values()) { if (query.queryId === service.materialized?.queryId) continue @@ -2031,13 +2111,29 @@ export class QueryStore< (query.config.realtime !== 'disabled' || this.#reconnectQueryIds.has(query.queryId)) && this.#listenerCount(query.queryId) > 0 ) { - this.#requestReconcile(query.queryId) + targets.set(query.queryId, targets.get(query.queryId) ?? false) } } } + return targets + } + + #refetchActiveQueries(): void { + const targets = this.#activeReconnectQueries() + for (const queryId of this.#scheduledReconnectQueryIds) { + if (!targets.has(queryId)) this.#sync.reconciliationFinished(queryId) + } + this.#scheduledReconnectQueryIds.clear() + for (const [queryId, force] of targets) { + this.#requestReconcile(queryId, { force }) + } } #scheduleReconnectSweep(): void { + for (const queryId of this.#activeReconnectQueries().keys()) { + this.#scheduledReconnectQueryIds.add(queryId) + this.#sync.reconciliationStarted(queryId) + } if (this.#reconnectSweepTimer) return const [min, max] = this.#reconnectJitter const delay = min === max ? min : min + Math.floor(Math.random() * (max - min + 1)) diff --git a/lib/core/syncTracker.ts b/lib/core/syncTracker.ts new file mode 100644 index 00000000..6d9d1380 --- /dev/null +++ b/lib/core/syncTracker.ts @@ -0,0 +1,157 @@ +import type { AdapterConnectionState } from '../adapters/adapter.js' +import type { MutationTracker } from './mutationTracker.js' + +export type SyncPhase = 'restoring' | 'offline' | 'syncing' | 'synced' | 'error' + +/** Canonical, instance-wide view of Figbird's progress toward server truth. */ +export interface SyncStatus { + readonly phase: SyncPhase + /** Writes that have not settled, including scheduled and paused queue work. */ + readonly pendingWrites: number + /** Pending queue writes whose latest attempt failed and can be retried or discarded. */ + readonly failedWrites: number + /** Distinct queries currently fetching or waiting to retry. */ + readonly fetchingQueries: number + /** Event/reconnect reconciliations that are fetching, gated, or deferred. */ + readonly pendingReconciliations: number + /** Epoch milliseconds of the last fully successful settle, or null before the first one. */ + readonly lastSyncedAt: number | null +} + +/** Read-only external-store surface exposed as `figbird.sync`. */ +export interface SyncActivity { + subscribe(listener: () => void): () => void + getSnapshot(): SyncStatus +} + +/** Synchronously maintained by QueryStore; events are deliberately not involved. */ +export class SyncTracker implements SyncActivity { + #mutations: MutationTracker + #connection: AdapterConnectionState + #fetchCounts = new Map() + #failedReconciliations = new Set() + #reconciliations = new Set() + #listeners = new Set<() => void>() + #snapshot: SyncStatus + + constructor(mutations: MutationTracker, connection: AdapterConnectionState) { + this.#mutations = mutations + this.#connection = connection + this.#snapshot = this.#createSnapshot(null) + this.#mutations.subscribeToSync(change => this.#changed(change)) + } + + getSnapshot = (): SyncStatus => this.#snapshot + + subscribe = (listener: () => void): (() => void) => { + this.#listeners.add(listener) + return () => this.#listeners.delete(listener) + } + + connectionChanged(connection: AdapterConnectionState): void { + if (connection === this.#connection) return + this.#connection = connection + this.#changed() + } + + queryStarted(queryId: string): void { + this.#fetchCounts.set(queryId, (this.#fetchCounts.get(queryId) ?? 0) + 1) + if (this.#failedReconciliations.delete(queryId)) { + this.#reconciliations.add(queryId) + } + this.#changed() + } + + queryFinished(queryId: string, outcome: 'success' | 'error' | 'cancelled'): void { + const count = this.#fetchCounts.get(queryId) ?? 0 + if (count <= 1) this.#fetchCounts.delete(queryId) + else this.#fetchCounts.set(queryId, count - 1) + if (outcome === 'error' && count <= 1 && this.#reconciliations.has(queryId)) { + this.#failedReconciliations.add(queryId) + } + // Ordinary query fetches remain observable through fetchingQueries, but do + // not change the interpreted, write-focused phase or lastSyncedAt. + this.#changed() + } + + reconciliationStarted(queryId: string): void { + const wasFailed = this.#failedReconciliations.delete(queryId) + if (this.#reconciliations.has(queryId)) { + if (wasFailed) this.#changed() + return + } + this.#reconciliations.add(queryId) + this.#changed() + } + + reconciliationFinished(queryId: string): void { + if (!this.#reconciliations.delete(queryId)) return + this.#changed({ successful: !this.#failedReconciliations.has(queryId) }) + } + + /** Remove every trace of a query that QueryStore is deleting. */ + forgetQuery(queryId: string): void { + const fetched = this.#fetchCounts.delete(queryId) + const failed = this.#failedReconciliations.delete(queryId) + const reconciling = this.#reconciliations.delete(queryId) + if (fetched || failed || reconciling) this.#changed() + } + + #changed({ successful = false }: { successful?: boolean } = {}): void { + const previous = this.#snapshot + const canStamp = successful && this.#isFullySynced() + const lastSyncedAt = canStamp ? Date.now() : previous.lastSyncedAt + const next = this.#createSnapshot(lastSyncedAt) + if ( + next.phase === previous.phase && + next.pendingWrites === previous.pendingWrites && + next.failedWrites === previous.failedWrites && + next.fetchingQueries === previous.fetchingQueries && + next.pendingReconciliations === previous.pendingReconciliations && + next.lastSyncedAt === previous.lastSyncedAt + ) { + return + } + this.#snapshot = next + for (const listener of this.#listeners) { + try { + listener() + } catch { + // Subscriber failures must never interrupt the store's lifecycle. + } + } + } + + #isFullySynced(): boolean { + const writes = this.#mutations.getSyncSnapshot() + return ( + this.#connection === 'connected' && + writes.pendingWrites === 0 && + writes.failedWrites === 0 && + this.#failedReconciliations.size === 0 && + this.#reconciliations.size === 0 + ) + } + + #createSnapshot(lastSyncedAt: number | null): SyncStatus { + const writes = this.#mutations.getSyncSnapshot() + const phase: SyncPhase = + this.#connection === 'disconnected' + ? 'offline' + : writes.failedWrites > 0 || this.#failedReconciliations.size > 0 + ? 'error' + : this.#connection === 'connecting' || this.#reconciliations.size > 0 + ? 'restoring' + : writes.pendingWrites > 0 + ? 'syncing' + : 'synced' + return Object.freeze({ + phase, + pendingWrites: writes.pendingWrites, + failedWrites: writes.failedWrites, + fetchingQueries: this.#fetchCounts.size, + pendingReconciliations: this.#reconciliations.size, + lastSyncedAt, + }) + } +} diff --git a/lib/index.ts b/lib/index.ts index 4e4df370..42fa2cd4 100644 --- a/lib/index.ts +++ b/lib/index.ts @@ -46,6 +46,9 @@ export type { StandardSchemaV1, VisibilitySource, WriteMutationOptions, + SyncActivity, + SyncPhase, + SyncStatus, } from './core/figbird.js' export { MutationQueueDiscardedError, @@ -106,6 +109,7 @@ export { matcher } from './adapters/matcher.js' // Adapter interface and types export type { Adapter, + AdapterConnectionState, AdapterFindMeta, AdapterParams, AdapterQuery, @@ -128,6 +132,7 @@ export { FigbirdProvider, useFigbird, useFigbirdMaybe } from './react/context.js export { useAction } from './react/useAction.js' export { useMutating } from './react/useMutating.js' export { useMutationQueue } from './react/useMutationQueue.js' +export { useSyncStatus } from './react/useSyncStatus.js' // Deprecated: superseded by useMutations + useAction + useMutating. export { useMutation } from './react/useMutation.js' // useFeathers is the raw-client escape hatch, typed via createHooks. diff --git a/lib/react/createHooks.ts b/lib/react/createHooks.ts index cbee7bc9..4ae1f999 100644 --- a/lib/react/createHooks.ts +++ b/lib/react/createHooks.ts @@ -37,6 +37,8 @@ import type { MutationQueueDefinition } from '../core/mutationQueue.js' import { useFind, useGet, type QueryResult } from './useQueryByDesc.js' import { useQueries, type UseQueriesHook } from './useQueries.js' import { useQuery, type UseQueryHook } from './useQuery.js' +import { useSyncStatusImpl } from './useSyncStatus.js' +import type { SyncStatus } from '../core/syncTracker.js' /** * Strongly-typed call signatures per service name. @@ -113,6 +115,8 @@ export interface FigbirdHooks { useAction: UseActionHook useMutating: UseMutatingForSchema useMutationQueue: UseMutationQueueHook + /** Unified reads, writes, connectivity, and reconciliation status. */ + useSyncStatus: () => SyncStatus } /** @@ -169,6 +173,10 @@ export function createHooks( function useTypedMutationQueue(definition?: MutationQueueDefinition, key?: string) { return useMutationQueueImpl(useBoundFigbird(), definition, key) } + + function useTypedSyncStatus(): SyncStatus { + return useSyncStatusImpl(useBoundFigbird()) + } function useTypedFeathers() { const adapter = useBoundFigbird().adapter as { feathers?: FeathersClient } if (!adapter.feathers) { @@ -209,5 +217,6 @@ export function createHooks( useAction, useMutating: useTypedMutating as UseMutatingForSchema, useMutationQueue: useTypedMutationQueue as UseMutationQueueHook, + useSyncStatus: useTypedSyncStatus, } } diff --git a/lib/react/useSyncStatus.ts b/lib/react/useSyncStatus.ts new file mode 100644 index 00000000..67b03c33 --- /dev/null +++ b/lib/react/useSyncStatus.ts @@ -0,0 +1,22 @@ +import { useSyncExternalStore } from 'react' +import type { SyncActivity, SyncStatus } from '../core/syncTracker.js' +import { useFigbird } from './context.js' + +/** The slice of a Figbird instance required by `useSyncStatus`. @internal */ +export interface SyncStatusHost { + sync: SyncActivity +} + +/** + * Unified application-facing view of reads, writes, connectivity, and + * reconciliation across the nearest Figbird instance. + */ +export function useSyncStatus(): SyncStatus { + return useSyncStatusImpl(useFigbird()) +} + +/** Instance-taking implementation used by schema-bound hook kits. @internal */ +export function useSyncStatusImpl(figbird: SyncStatusHost): SyncStatus { + const { sync } = figbird + return useSyncExternalStore(sync.subscribe, sync.getSnapshot, sync.getSnapshot) +} diff --git a/test/mutation-hooks.test.tsx b/test/mutation-hooks.test.tsx index a5cfb0c6..61c20604 100644 --- a/test/mutation-hooks.test.tsx +++ b/test/mutation-hooks.test.tsx @@ -489,3 +489,88 @@ test('useMutating: service filter resolves schema aliases to transport paths', a }) t.is(probe.read(), 'false') }) + +test('useSyncStatus: replays pending writes and settles one-shot failures', async t => { + const { App, figbird, feathers } = createTestApp(schema, services()) + const { useSyncStatus } = createHooks(schema) + const d = dom() + + function Probe() { + const sync = useSyncStatus() + return {JSON.stringify(sync)} + } + + const read = () => JSON.parse(d.$('output')!.textContent!) as ReturnType + const observedWriteCounts: Array = [] + const unsubscribeMutating = figbird.mutating.subscribe(() => { + observedWriteCounts.push([ + figbird.mutating.getSnapshot().length, + figbird.sync.getSnapshot().pendingWrites, + ]) + }) + + t.is(figbird.sync.getSnapshot().phase, 'synced') + t.is(figbird.sync.getSnapshot().lastSyncedAt, null) + + const queryGate = deferred<{ total: number; limit: number; skip: number; data: MockItem[] }>() + feathers.service('notes').find = () => queryGate.promise + const query = figbird.queryDesc({ serviceName: 'notes', method: 'find' }) + const unsubscribeQuery = query.subscribe(() => {}) + t.is(figbird.sync.getSnapshot().fetchingQueries, 1) + t.is(figbird.sync.getSnapshot().phase, 'synced', 'ordinary reads do not flash global sync UI') + queryGate.resolve({ + total: 1, + limit: 10, + skip: 0, + data: [{ id: 1, content: 'hello' }], + }) + await new Promise(resolve => setTimeout(resolve, 0)) + t.is(figbird.sync.getSnapshot().fetchingQueries, 0) + t.is(figbird.sync.getSnapshot().lastSyncedAt, null) + unsubscribeQuery() + + const first = deferred() + feathers.service('notes').patch = () => first.promise + const failedWrite = figbird.m.notes.patch(1, { content: 'offline' }) + + // The hook subscribes after the call and still receives the canonical active snapshot. + d.render( + + + , + ) + t.is(read().phase, 'syncing') + t.is(read().pendingWrites, 1) + + await d.flush(async () => { + first.reject(new Error('offline')) + await t.throwsAsync(failedWrite, { message: 'offline' }) + }) + t.is(read().phase, 'synced') + t.is(read().pendingWrites, 0) + t.is(read().failedWrites, 0, 'settled one-shot errors are owned by the calling action') + + const retry = deferred() + feathers.service('notes').patch = () => retry.promise + let retriedWrite!: Promise + await d.flush(() => { + retriedWrite = figbird.m.notes.patch(1, { content: 'online' }) + }) + t.is(read().phase, 'syncing') + t.is(read().pendingWrites, 1) + t.is(read().failedWrites, 0) + + await d.flush(async () => { + retry.resolve({ id: 1, content: 'online' }) + await retriedWrite + }) + t.is(read().phase, 'synced') + t.is(read().pendingWrites, 0) + t.is(read().failedWrites, 0) + t.is(typeof read().lastSyncedAt, 'number') + t.true( + observedWriteCounts.every(([mutating, pendingWrites]) => mutating === pendingWrites), + 'mutating and sync snapshots share one authoritative registry', + ) + unsubscribeMutating() +}) diff --git a/test/mutation-queue.test.ts b/test/mutation-queue.test.ts index 7858f8d7..07539b77 100644 --- a/test/mutation-queue.test.ts +++ b/test/mutation-queue.test.ts @@ -409,9 +409,10 @@ test('mutation queue: a terminal failure pauses with optimism intact until retry await new Promise(resolve => setTimeout(resolve, 10)) let calls = 0 - feathers.service('notes').patch = ((_id: number, data: Partial) => { + const retryGate = deferred() + feathers.service('notes').patch = ((_id: number, _data: Partial) => { calls += 1 - return calls === 1 ? Promise.reject(new Error('offline')) : Promise.resolve({ id: 1, ...data }) + return calls === 1 ? Promise.reject(new Error('offline')) : retryGate.promise }) as never const queue = figbird.createMutationQueue() @@ -421,11 +422,24 @@ test('mutation queue: a terminal failure pauses with optimism intact until retry t.is(queue.getSnapshot().status, 'failed') t.is(latest?.data?.find(note => note.id === 1)?.content, 'survives retry') t.is(calls, 1) + t.deepEqual( + { + phase: figbird.sync.getSnapshot().phase, + pendingWrites: figbird.sync.getSnapshot().pendingWrites, + failedWrites: figbird.sync.getSnapshot().failedWrites, + }, + { phase: 'error', pendingWrites: 1, failedWrites: 1 }, + ) queue.retry() + await new Promise(resolve => setTimeout(resolve, 0)) + t.is(figbird.sync.getSnapshot().phase, 'syncing') + t.is(figbird.sync.getSnapshot().failedWrites, 0) + retryGate.resolve({ id: 1, content: 'survives retry' }) await pending t.is(calls, 2) t.is(queue.getSnapshot().status, 'idle') + t.is(figbird.sync.getSnapshot().phase, 'synced') }) test('mutation queue: related creates are transported serially in call order', async t => { @@ -472,6 +486,8 @@ test('mutation queue: discard rolls back the failed and pending optimistic work' await new Promise(resolve => setTimeout(resolve, 0)) t.is(queue.getSnapshot().status, 'failed') + t.is(figbird.sync.getSnapshot().pendingWrites, 2) + t.is(figbird.sync.getSnapshot().failedWrites, 1) t.deepEqual( latest?.data?.map(note => note.content), ['pending one', 'pending two'], @@ -485,4 +501,7 @@ test('mutation queue: discard rolls back the failed and pending optimistic work' ['hello', 'world'], ) t.is(queue.getSnapshot().status, 'idle') + t.is(figbird.sync.getSnapshot().phase, 'synced') + t.is(figbird.sync.getSnapshot().pendingWrites, 0) + t.is(figbird.sync.getSnapshot().failedWrites, 0) }) diff --git a/test/reconcile.test.tsx b/test/reconcile.test.tsx index 5ade733e..d75a06fb 100644 --- a/test/reconcile.test.tsx +++ b/test/reconcile.test.tsx @@ -180,6 +180,23 @@ test('cooldown: a trailing refetch is skipped when the last subscriber left', as const state = figbird.getState().get('notes') const pending = Array.from(state!.queries.values()).some(q => q.pending) t.true(pending) + + // A failed ephemeral reconciliation must be forgotten with the query. Otherwise + // a query that no longer exists leaves the instance permanently in `error`. + const ephemeral = figbird.queryDesc( + { serviceName: 'notes', method: 'find' }, + { realtime: 'refetch', fetchPolicy: 'network-only' }, + ) + const unsubscribeEphemeral = ephemeral.subscribe(() => {}) + await sleep(20) + const failure = Object.assign(new Error('bad request'), { code: 400 }) + notes.find = () => Promise.reject(failure) + notes.emit('created', { id: 12, content: 'fails to reconcile' }) + await sleep(20) + t.is(figbird.sync.getSnapshot().phase, 'error') + + unsubscribeEphemeral() + t.is(figbird.sync.getSnapshot().phase, 'synced') }) test('hidden tabs: event-driven reconciliation defers; local-exact merges keep flowing', async t => { @@ -241,21 +258,33 @@ test('hidden tabs: a reconnect while hidden defers the refetch-all until visible await sleep(20) const baseline = notes.counts.find + io.emit('disconnect') + t.is(figbird.sync.getSnapshot().phase, 'offline') + io.emit('connect') + t.is(figbird.sync.getSnapshot().phase, 'synced') + // Visible reconnect: refetches immediately (existing behavior). io.emit('reconnect') + t.is(figbird.sync.getSnapshot().phase, 'restoring') + t.is(figbird.sync.getSnapshot().pendingReconciliations, 1) await sleep(20) t.is(notes.counts.find, baseline + 1) + t.is(figbird.sync.getSnapshot().phase, 'synced') // Hidden reconnect: deferred... visibility.set(true) io.emit('reconnect') await sleep(20) t.is(notes.counts.find, baseline + 1, 'hidden tabs do not replay the reconnect storm') + t.is(figbird.sync.getSnapshot().phase, 'restoring') + t.is(figbird.sync.getSnapshot().pendingReconciliations, 1) // ...and reconciled once on return. visibility.set(false) await sleep(20) t.is(notes.counts.find, baseline + 2) + t.is(figbird.sync.getSnapshot().phase, 'synced') + t.is(figbird.sync.getSnapshot().pendingReconciliations, 0) unsub() }) @@ -281,11 +310,14 @@ test('reconnect jitter delays and coalesces a visible-tab sweep', async t => { io.emit('reconnect') io.emit('reconnect') + t.is(figbird.sync.getSnapshot().phase, 'restoring') + t.is(figbird.sync.getSnapshot().pendingReconciliations, 1) await sleep(25) t.is(notes.counts.find, baseline, 'the sweep stays inside the configured delay') await sleep(35) t.is(notes.counts.find, baseline + 1, 'two reconnects coalesce into one sweep') + t.is(figbird.sync.getSnapshot().phase, 'synced') unsub() })