((listener) => {
+ listeners.add(listener)
+ return () => {
+ listeners.delete(listener)
+ unsubscribe()
+ }
+ })
+
+ return {
+ subscribe,
+ unsubscribe,
+ listenerCount: () => listeners.size,
+ restart() {
+ act(() => {
+ for (const listener of [...listeners]) {
+ listener()
+ }
+ })
+ },
+ }
+}
+
+/**
+ * A client API whose `getProject()` hands out generation-tagged project
+ * instances. Bumping the generation stands in for a backend restart, and is
+ * deliberately the harshest shape the reset has to cope with: every instance
+ * handed out before the bump starts rejecting, as a project wrapper does under
+ * `@comapeo/ipc` v9. A v10 reference stays usable instead, which only makes the
+ * same reset easier — so passing here covers both.
+ */
+function createGenerationalClientApi() {
+ let generation = 0
+
+ const getProject = vi.fn(async (projectId: string) => {
+ const instanceGeneration = generation
+ function assertLive() {
+ if (instanceGeneration !== generation) {
+ throw new Error(
+ `ProjectClosed: instance from generation ${instanceGeneration}`,
+ )
+ }
+ }
+ return {
+ generation: instanceGeneration,
+ projectId,
+ $getProjectSettings: async () => {
+ assertLive()
+ return { name: `settings-gen-${instanceGeneration}` }
+ },
+ $blobs: {
+ // The media server port is ephemeral and changes on every restart
+ getUrl: async () => {
+ assertLive()
+ return `http://127.0.0.1:${5000 + instanceGeneration}/blob`
+ },
+ },
+ }
+ })
+
+ const clientApi = Object.assign(createMockClientApi(), { getProject })
+
+ return {
+ clientApi: clientApi as unknown as ComapeoCoreClientApi,
+ getProject,
+ bumpGeneration() {
+ generation += 1
+ },
+ }
+}
+
+function ProjectScreen() {
+ const { data: projectApi } = useSingleProject({ projectId: PROJECT_ID })
+ const { data: settings } = useProjectSettings({ projectId: PROJECT_ID })
+ const { data: attachmentUrl } = useAttachmentUrl({
+ projectId: PROJECT_ID,
+ blobId: {
+ type: 'photo',
+ variant: 'thumbnail',
+ name: 'name',
+ driveId: 'drive-id',
+ },
+ })
+
+ return (
+
+
+ {String((projectApi as unknown as { generation: number }).generation)}
+
+ {settings.name}
+ {attachmentUrl}
+
+ )
+}
+
+function renderProvider({
+ queryClient,
+ clientApi = createMockClientApi() as unknown as ComapeoCoreClientApi,
+ subscribeToBackendRestart,
+ children,
+}: {
+ queryClient: QueryClient
+ clientApi?: ComapeoCoreClientApi
+ subscribeToBackendRestart?: SubscribeToBackendRestart
+ children?: ReactNode
+}) {
+ function Tree(props: {
+ subscribeToBackendRestart?: SubscribeToBackendRestart
+ }) {
+ return (
+
+ new URL('http://localhost:3000')}
+ subscribeToBackendRestart={props.subscribeToBackendRestart}
+ >
+ loading}>
+ {children}
+
+
+
+ )
+ }
+
+ const utils = render(
+ ,
+ )
+
+ return {
+ ...utils,
+ screen: within(utils.container),
+ rerenderTree() {
+ utils.rerender(
+ ,
+ )
+ },
+ setSubscribeToBackendRestart(next?: SubscribeToBackendRestart) {
+ utils.rerender()
+ },
+ }
+}
+
+describe('subscribeToBackendRestart', () => {
+ test('is optional', () => {
+ const queryClient = new QueryClient()
+
+ expect(() => {
+ renderProvider({ queryClient })
+ }).not.toThrow()
+ })
+
+ // Call counts are not asserted anywhere in this file: the test setup enables
+ // `reactStrictMode`, so effects are mounted, cleaned up and re-mounted. The
+ // number of live listeners is what actually matters.
+ test('subscribes on mount', () => {
+ const queryClient = new QueryClient()
+ const backend = createBackendRestartSource()
+
+ renderProvider({
+ queryClient,
+ subscribeToBackendRestart: backend.subscribe,
+ })
+
+ expect(backend.subscribe).toHaveBeenCalled()
+ expect(backend.listenerCount()).toBe(1)
+ })
+
+ test('unsubscribes on unmount', () => {
+ const queryClient = new QueryClient()
+ const backend = createBackendRestartSource()
+
+ const { unmount } = renderProvider({
+ queryClient,
+ subscribeToBackendRestart: backend.subscribe,
+ })
+
+ expect(backend.listenerCount()).toBe(1)
+
+ unmount()
+
+ expect(backend.unsubscribe).toHaveBeenCalled()
+ expect(backend.listenerCount()).toBe(0)
+ })
+
+ test('unsubscribes from the previous function when the prop changes', () => {
+ const queryClient = new QueryClient()
+ const first = createBackendRestartSource()
+ const second = createBackendRestartSource()
+
+ const { setSubscribeToBackendRestart } = renderProvider({
+ queryClient,
+ subscribeToBackendRestart: first.subscribe,
+ })
+
+ setSubscribeToBackendRestart(second.subscribe)
+
+ expect(first.unsubscribe).toHaveBeenCalled()
+ expect(first.listenerCount()).toBe(0)
+ expect(second.listenerCount()).toBe(1)
+ })
+
+ test('does not resubscribe when the tree re-renders with the same function', () => {
+ const queryClient = new QueryClient()
+ const backend = createBackendRestartSource()
+
+ const { rerenderTree } = renderProvider({
+ queryClient,
+ subscribeToBackendRestart: backend.subscribe,
+ })
+
+ const subscribeCalls = backend.subscribe.mock.calls.length
+
+ rerenderTree()
+ rerenderTree()
+
+ expect(backend.subscribe.mock.calls.length).toBe(subscribeCalls)
+ expect(backend.listenerCount()).toBe(1)
+ })
+})
+
+describe('recovery after a restart', () => {
+ test('a mounted project-derived query ends up with data from the new backend', async () => {
+ const queryClient = new QueryClient()
+ const backend = createGenerationalClientApi()
+ const restarts = createBackendRestartSource()
+
+ const { screen } = renderProvider({
+ queryClient,
+ clientApi: backend.clientApi,
+ subscribeToBackendRestart: restarts.subscribe,
+ children: ,
+ })
+
+ await waitFor(() => {
+ expect(screen.getByTestId('settings-name').textContent).toBe(
+ 'settings-gen-0',
+ )
+ })
+
+ backend.bumpGeneration()
+ restarts.restart()
+
+ await waitFor(() => {
+ expect(screen.queryByTestId('loading')).toBeNull()
+ expect(screen.getByTestId('generation').textContent).toBe('1')
+ expect(screen.getByTestId('settings-name').textContent).toBe(
+ 'settings-gen-1',
+ )
+ })
+
+ // A `useSuspenseQuery` with `retry: false` that fails once never refetches
+ // again, so the derived query must never have run against the closed
+ // instance in the first place
+ expect(
+ queryClient.getQueryState([
+ ROOT_QUERY_KEY,
+ 'projects',
+ PROJECT_ID,
+ 'project_settings',
+ ])?.status,
+ ).toBe('success')
+ })
+
+ test('a static-staleTime query re-runs after a restart', async () => {
+ const queryClient = new QueryClient()
+ const backend = createGenerationalClientApi()
+ const restarts = createBackendRestartSource()
+
+ const { screen } = renderProvider({
+ queryClient,
+ clientApi: backend.clientApi,
+ subscribeToBackendRestart: restarts.subscribe,
+ children: ,
+ })
+
+ await waitFor(() => {
+ expect(screen.getByTestId('attachment-url').textContent).toContain(
+ '127.0.0.1:5000',
+ )
+ })
+
+ backend.bumpGeneration()
+ restarts.restart()
+
+ // The media server origin is cached with `staleTime: 'static'`, so
+ // invalidation cannot reach it: without being removed, every image URL
+ // would point at the dead port for the life of the app
+ await waitFor(() => {
+ expect(screen.getByTestId('attachment-url').textContent).toContain(
+ '127.0.0.1:5001',
+ )
+ })
+ })
+
+ test('manager-level queries are refetched without dropping their data', async () => {
+ const queryClient = new QueryClient()
+ const restarts = createBackendRestartSource()
+ const queryFn = vi.fn(async () => 'device-info')
+
+ function ManagerLevelProbe() {
+ const { data } = useQuery({
+ queryKey: [ROOT_QUERY_KEY, 'client', 'device_info'],
+ queryFn,
+ networkMode: 'always',
+ retry: false,
+ })
+ return {data ?? 'none'}
+ }
+
+ const { screen } = renderProvider({
+ queryClient,
+ subscribeToBackendRestart: restarts.subscribe,
+ children: ,
+ })
+
+ await waitFor(() => {
+ expect(queryFn).toHaveBeenCalledTimes(1)
+ })
+
+ restarts.restart()
+
+ await waitFor(() => {
+ expect(queryFn).toHaveBeenCalledTimes(2)
+ })
+ // Invalidated, not removed: the data stays put while it refetches
+ expect(screen.getByTestId('device-info').textContent).toBe('device-info')
+ })
+
+ // Invites are in-memory actors on the backend: a restart drops every one of
+ // them, and `invite.getMany()` is the only way to find out what the new
+ // backend has. Nothing invalidates the invites query on its own - the
+ // `invite-received` / `invite-updated` events that normally do were raised
+ // (if at all) while the app was disconnected - so the root invalidation is
+ // what has to reach it.
+ test('the invites query is refetched', async () => {
+ const queryClient = new QueryClient()
+ const restarts = createBackendRestartSource()
+ const clientApi = createMockClientApi()
+ const getMany = vi.fn(async () => [])
+ Object.assign(clientApi.invite, { getMany })
+
+ function InvitesProbe() {
+ const { data } = useManyInvites()
+ return {data.length}
+ }
+
+ const { screen } = renderProvider({
+ queryClient,
+ clientApi: clientApi as unknown as ComapeoCoreClientApi,
+ subscribeToBackendRestart: restarts.subscribe,
+ children: ,
+ })
+
+ await waitFor(() => {
+ expect(screen.getByTestId('invite-count').textContent).toBe('0')
+ })
+ const callsBeforeRestart = getMany.mock.calls.length
+
+ restarts.restart()
+
+ await waitFor(() => {
+ expect(getMany.mock.calls.length).toBeGreaterThan(callsBeforeRestart)
+ })
+ })
+
+ test('queries owned by the consuming app are left alone', () => {
+ const queryClient = new QueryClient()
+ const appQueryKey = ['app-owned-query']
+ queryClient.setQueryData(appQueryKey, 'not ours')
+
+ const restarts = createBackendRestartSource()
+
+ renderProvider({
+ queryClient,
+ subscribeToBackendRestart: restarts.subscribe,
+ })
+
+ restarts.restart()
+
+ expect(queryClient.getQueryData(appQueryKey)).toBe('not ours')
+ expect(queryClient.getQueryState(appQueryKey)?.isInvalidated).toBe(false)
+ })
+
+ test('a restart no longer resets the cache after unmount', () => {
+ const queryClient = new QueryClient()
+ const projectQueryKey = [ROOT_QUERY_KEY, 'projects', PROJECT_ID]
+ queryClient.setQueryData(projectQueryKey, 'from the previous backend')
+
+ const restarts = createBackendRestartSource()
+
+ const { unmount } = renderProvider({
+ queryClient,
+ subscribeToBackendRestart: restarts.subscribe,
+ })
+
+ unmount()
+ restarts.restart()
+
+ expect(queryClient.getQueryData(projectQueryKey)).toBe(
+ 'from the previous backend',
+ )
+ })
+})
+
+class TestErrorBoundary extends Component<
+ { children: ReactNode },
+ { error: Error | null }
+> {
+ override state: { error: Error | null } = { error: null }
+ static getDerivedStateFromError(error: Error) {
+ return { error }
+ }
+ override render() {
+ if (this.state.error) {
+ return (
+
+ {this.state.error.message}
+
+ )
+ }
+ return this.props.children
+ }
+}
+
+function DeviceInfoScreen() {
+ const { data } = useOwnDeviceInfo()
+ return (
+
+ {(data as unknown as { name: string }).name}
+
+ )
+}
+
+describe('channel-closed retry', () => {
+ // A query in flight when the backend's RPC transport drops rejects with
+ // code RPC_CHANNEL_CLOSED (a read whose response will never arrive).
+ // `baseQueryOptions` retries only that code, so the query keeps loading
+ // through the restart instead of latching into an error state.
+ test('a channel-closed rejection is retried and resolves', async () => {
+ const queryClient = new QueryClient()
+ const clientApi =
+ createMockClientApi() as unknown as ComapeoCoreClientApi & {
+ getDeviceInfo: ReturnType
+ }
+ let failuresLeft = 1
+ clientApi.getDeviceInfo = vi.fn(async () => {
+ if (failuresLeft > 0) {
+ failuresLeft -= 1
+ throw Object.assign(new Error('Channel closed'), {
+ code: 'RPC_CHANNEL_CLOSED',
+ })
+ }
+ return {
+ deviceId: 'device-id',
+ name: 'gecko',
+ deviceType: 'mobile' as const,
+ }
+ })
+
+ const { screen } = renderProvider({
+ queryClient,
+ clientApi,
+ children: (
+
+
+
+ ),
+ })
+
+ await waitFor(
+ () => {
+ expect(screen.getByTestId('device-name').textContent).toBe('gecko')
+ },
+ { timeout: 5_000 },
+ )
+ expect(screen.queryByTestId('boundary-error')).toBeNull()
+ expect(clientApi.getDeviceInfo.mock.calls.length).toBeGreaterThanOrEqual(2)
+ }, 10_000)
+
+ test('other errors are not retried', async () => {
+ const queryClient = new QueryClient()
+ const clientApi =
+ createMockClientApi() as unknown as ComapeoCoreClientApi & {
+ getDeviceInfo: ReturnType
+ }
+ clientApi.getDeviceInfo = vi.fn(async () => {
+ throw new Error('genuinely broken')
+ })
+
+ const { screen } = renderProvider({
+ queryClient,
+ clientApi,
+ children: (
+
+
+
+ ),
+ })
+
+ await waitFor(() => {
+ expect(screen.getByTestId('boundary-error').textContent).toBe(
+ 'genuinely broken',
+ )
+ })
+ })
+})
+
+// @comapeo/ipc v10 hands out permanent project references. Leaving a project
+// does not invalidate the reference the app is holding: every call on it -
+// and `getProject()` itself, for a project left before it was ever acquired -
+// rejects with `ProjectLeftError` until the project is re-joined via an
+// invite. There is nothing to recover from, so the error has to reach the
+// consuming app's error boundary with its code intact and without being
+// retried first.
+describe('a project this device has left', () => {
+ function createLeftProjectError() {
+ return Object.assign(new Error('This device has left the project'), {
+ code: 'PROJECT_LEFT',
+ })
+ }
+
+ function createLeftProjectClientApi({ leftBefore }: { leftBefore: boolean }) {
+ const getProject = vi.fn(async () => {
+ if (leftBefore) throw createLeftProjectError()
+ return {
+ $getProjectSettings: async () => {
+ throw createLeftProjectError()
+ },
+ }
+ })
+ const clientApi = Object.assign(createMockClientApi(), { getProject })
+ return {
+ clientApi: clientApi as unknown as ComapeoCoreClientApi,
+ getProject,
+ }
+ }
+
+ function LeftProjectScreen() {
+ const { data } = useProjectSettings({ projectId: PROJECT_ID })
+ return {data.name}
+ }
+
+ test('a call on a held reference reaches the error boundary', async () => {
+ const { clientApi } = createLeftProjectClientApi({ leftBefore: false })
+
+ const { screen } = renderProvider({
+ queryClient: new QueryClient(),
+ clientApi,
+ children: (
+
+
+
+ ),
+ })
+
+ await waitFor(() => {
+ expect(screen.getByTestId('boundary-error').textContent).toBe(
+ 'This device has left the project',
+ )
+ })
+ expect(screen.getByTestId('boundary-error').dataset.code).toBe(
+ 'PROJECT_LEFT',
+ )
+ })
+
+ test('a first acquisition reaches the error boundary', async () => {
+ const { clientApi, getProject } = createLeftProjectClientApi({
+ leftBefore: true,
+ })
+
+ const { screen } = renderProvider({
+ queryClient: new QueryClient(),
+ clientApi,
+ children: (
+
+
+
+ ),
+ })
+
+ await waitFor(() => {
+ expect(screen.getByTestId('boundary-error').dataset.code).toBe(
+ 'PROJECT_LEFT',
+ )
+ })
+ // Not retried: a retry would push the failure past the 1s retry delay
+ const callsWhenSettled = getProject.mock.calls.length
+ await new Promise((resolve) => setTimeout(resolve, 1_200))
+ expect(getProject.mock.calls.length).toBe(callsWhenSettled)
+ }, 10_000)
+})
diff --git a/test/hooks/invite-rejoin.test.ts b/test/hooks/invite-rejoin.test.ts
index c3557c9..ac29cb0 100644
--- a/test/hooks/invite-rejoin.test.ts
+++ b/test/hooks/invite-rejoin.test.ts
@@ -62,11 +62,10 @@ async function waitForPeers(managers: Managers) {
// Regression test for digidem/comapeo-mobile#2042 and #2041: a member is
// removed from a project, leaves it, and is re-invited. Accepting the new
-// invite closes the old project instance on the manager
-// (`MapeoManager.addProject`) and opens a fresh one. The project client
-// wrapper is cached with `staleTime: Infinity`, so without invalidation the
-// hooks keep using the closed instance and every project call rejects with
-// ProjectClosed until app restart.
+// invite re-adds the project on the manager (`MapeoManager.addProject`),
+// which closes the instance left behind by leaving and opens a fresh one.
+// The project must be usable again from the hooks straight afterwards,
+// rather than every call rejecting until the app restarts.
test(
're-joining a project after leaving yields a working project instance',
{ timeout: 60_000 },
@@ -198,11 +197,18 @@ test(
)
assert.strictEqual(settingsHook.result.current.data.name, 'mapeo')
- // The re-joined project must be a fresh instance — calls on the wrapper
- // cached before the re-join reject because that instance is closed.
- assert.notStrictEqual(
- rejoinedProjectHook.result.current.data,
- originalWrapper,
- )
+ // The regression itself: a call made directly through the reference the
+ // hooks hand out has to reach a live instance, not the closed one that
+ // was cached before the leave. Asserted through the reference rather
+ // than only through the settings query, because it is the cached
+ // reference that #199 was about.
+ //
+ // What the reference *is* differs by major and is deliberately not
+ // asserted: @comapeo/ipc v9 hands out a fresh wrapper and leaves the
+ // pre-leave one permanently closed, while v10 hands back the same
+ // permanent reference, re-routed to a re-opened instance.
+ const rejoinedProjectApi = rejoinedProjectHook.result.current.data
+ const settingsFromReference = await rejoinedProjectApi.$getProjectSettings()
+ assert.strictEqual(settingsFromReference.name, 'mapeo')
},
)
diff --git a/test/lib/react-query.test.ts b/test/lib/react-query.test.ts
new file mode 100644
index 0000000..af96df4
--- /dev/null
+++ b/test/lib/react-query.test.ts
@@ -0,0 +1,47 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+
+import {
+ baseMutationOptions,
+ baseQueryOptions,
+} from '../../src/lib/react-query.js'
+
+function errorWithCode(code: string) {
+ return Object.assign(new Error(code), { code })
+}
+
+describe('baseQueryOptions() retry', () => {
+ const { retry } = baseQueryOptions()
+
+ // `RPC_CHANNEL_CLOSED` is what rpc-reflector rejects in-flight calls with
+ // when the transport to the backend drops. There is no other transport-level
+ // code - `RPC_TRANSPORT_CLOSED` has never existed.
+ it('retries a channel-closed rejection a bounded number of times', () => {
+ const error = errorWithCode('RPC_CHANNEL_CLOSED')
+
+ expect(retry(0, error)).toBe(true)
+ expect(retry(2, error)).toBe(true)
+ expect(retry(3, error)).toBe(false)
+ })
+
+ // A left project is only usable again after re-joining via an invite, which
+ // no amount of retrying brings about
+ it('does not retry a left project', () => {
+ expect(retry(0, errorWithCode('PROJECT_LEFT'))).toBe(false)
+ })
+
+ it('does not retry anything else', () => {
+ expect(retry(0, new Error('genuinely broken'))).toBe(false)
+ expect(retry(0, errorWithCode('CLIENT_CLOSED'))).toBe(false)
+ expect(retry(0, errorWithCode('RPC_TIMEOUT'))).toBe(false)
+ })
+})
+
+describe('baseMutationOptions()', () => {
+ // Re-issuing a write whose response was lost is never safe
+ it('never retries', () => {
+ expect(baseMutationOptions().retry).toBe(false)
+ })
+})
diff --git a/test/lib/received-map-shares-store.test.ts b/test/lib/received-map-shares-store.test.ts
index f4b3e38..0438fdd 100644
--- a/test/lib/received-map-shares-store.test.ts
+++ b/test/lib/received-map-shares-store.test.ts
@@ -1,6 +1,7 @@
/**
* @vitest-environment node
*/
+import { QueryClient } from '@tanstack/react-query'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { createMapServerApi } from '../../src/contexts/MapServer.js'
@@ -40,6 +41,88 @@ describe('ReceivedMapSharesStore', () => {
getBaseUrl: async () => new URL(receiver.localBaseUrl),
}),
})
+ t.onTestFinished(store.listen())
+ })
+
+ describe('client api listener', () => {
+ function createStore() {
+ return createReceivedMapSharesStore({
+ // @ts-expect-error - We're only mocking what we need
+ clientApi: mockClientApi,
+ mapServerApi: createMapServerApi({
+ getBaseUrl: async () => new URL(receiver.localBaseUrl),
+ }),
+ })
+ }
+
+ it('should not attach a listener until listen() is called', () => {
+ const before = mockClientApi.listeners.get('map-share')?.length ?? 0
+
+ const teardown = createStore().listen()
+
+ expect(mockClientApi.listeners.get('map-share')).toHaveLength(before + 1)
+
+ teardown()
+ })
+
+ it('should remove its listener on teardown', () => {
+ const before = mockClientApi.listeners.get('map-share')?.length ?? 0
+
+ const teardown = createStore().listen()
+ teardown()
+
+ expect(mockClientApi.listeners.get('map-share')).toHaveLength(before)
+ })
+
+ it('should not register a second listener when listen() is called twice', async () => {
+ const before = mockClientApi.listeners.get('map-share')?.length ?? 0
+
+ const doubleListeningStore = createStore()
+ const teardown = doubleListeningStore.listen()
+ const secondTeardown = doubleListeningStore.listen()
+
+ expect(mockClientApi.listeners.get('map-share')).toHaveLength(before + 1)
+ expect(secondTeardown).toBe(teardown)
+
+ const serverShare = await createShare(sender, receiver)
+ mockClientApi.emit(
+ 'map-share',
+ createMapShareFromServerShare(sender.deviceId, serverShare),
+ )
+ expect(doubleListeningStore.getSnapshot()).toHaveLength(1)
+
+ teardown()
+ expect(mockClientApi.listeners.get('map-share')).toHaveLength(before)
+ })
+
+ it('should re-attach when listen() is called again after teardown', () => {
+ const before = mockClientApi.listeners.get('map-share')?.length ?? 0
+
+ const relisteningStore = createStore()
+ relisteningStore.listen()()
+
+ const teardown = relisteningStore.listen()
+ expect(mockClientApi.listeners.get('map-share')).toHaveLength(before + 1)
+
+ teardown()
+ expect(mockClientApi.listeners.get('map-share')).toHaveLength(before)
+ })
+
+ it('should ignore map-share events after teardown', async () => {
+ const serverShare = await createShare(sender, receiver)
+ const mapShare = createMapShareFromServerShare(
+ sender.deviceId,
+ serverShare,
+ )
+
+ const torndownStore = createStore()
+ const teardown = torndownStore.listen()
+ teardown()
+
+ mockClientApi.emit('map-share', mapShare)
+
+ expect(torndownStore.getSnapshot()).toHaveLength(0)
+ })
})
describe('subscription', () => {
@@ -139,3 +222,156 @@ describe('ReceivedMapSharesStore', () => {
})
})
})
+
+// The event stream is the only thing that ever moves a share out of
+// `downloading`. If the map server goes away for good — it dies with the
+// backend on Android — the stream never re-establishes, and without an error
+// path the share would sit in `downloading` for the life of the app.
+describe('map share event stream failures', () => {
+ const MAP_SHARE = {
+ shareId: 'share-id',
+ senderDeviceId: 'sender-device-id',
+ senderDeviceName: 'Sender',
+ mapShareReceivedAt: Date.now(),
+ mapId: 'custom',
+ estimatedSizeBytes: 100,
+ mapShareUrls: ['http://127.0.0.1:1/'],
+ }
+
+ // eventsource-client's own default, and what the map server gets unless it
+ // sends a `retry:` field
+ const DEFAULT_RECONNECT_DELAY_MS = 2_000
+
+ function createStoreWithFakeEventSource() {
+ const close = vi.fn()
+ const createEventSource = vi.fn()
+ let onScheduleReconnect: ((info: { delay: number }) => void) | undefined
+ let onMessage: ((event: { data: string }) => void) | undefined
+
+ const mapServerApi = {
+ post: () => ({ json: async () => ({ downloadId: 'download-id' }) }),
+ createEventSource: (options: {
+ onScheduleReconnect?: (info: { delay: number }) => void
+ onMessage?: (event: { data: string }) => void
+ }) => {
+ createEventSource(options)
+ onScheduleReconnect = options.onScheduleReconnect
+ onMessage = options.onMessage
+ return { close }
+ },
+ }
+
+ const mockClientApi = createMockClientApi()
+ const store = createReceivedMapSharesStore({
+ // @ts-expect-error - We're only mocking what we need
+ clientApi: mockClientApi,
+ // @ts-expect-error - We're only mocking what we need
+ mapServerApi,
+ queryClient: new QueryClient(),
+ })
+
+ return {
+ store,
+ close,
+ createEventSource,
+ async startDownload() {
+ await store.actions.download({ shareId: MAP_SHARE.shareId })
+ },
+ receiveShare() {
+ mockClientApi.emit('map-share', MAP_SHARE)
+ },
+ dropStream: (delay = DEFAULT_RECONNECT_DELAY_MS) =>
+ onScheduleReconnect?.({ delay }),
+ emitProgress: (bytesDownloaded: number) =>
+ onMessage?.({
+ data: JSON.stringify({ status: 'downloading', bytesDownloaded }),
+ }),
+ }
+ }
+
+ // The budget is measured in seconds of outage rather than reconnect
+ // attempts: at the 2s default with no backoff, a handful of attempts is only
+ // a few seconds, which an Android backend restart routinely exceeds.
+ it('tolerates a stream outage far longer than a few reconnects', async (t) => {
+ const fake = createStoreWithFakeEventSource()
+ t.onTestFinished(fake.store.listen())
+ fake.receiveShare()
+
+ await fake.startDownload()
+
+ for (let i = 0; i < 15; i++) fake.dropStream()
+
+ expect(fake.store.getSnapshot()[0]).toHaveProperty('status', 'downloading')
+ })
+
+ it('moves the share to error once the outage budget is spent', async (t) => {
+ const fake = createStoreWithFakeEventSource()
+ t.onTestFinished(fake.store.listen())
+ fake.receiveShare()
+
+ await fake.startDownload()
+ expect(fake.store.getSnapshot()[0]).toHaveProperty('status', 'downloading')
+
+ // 30s and 60s of accumulated downtime are both still within budget
+ fake.dropStream(30_000)
+ fake.dropStream(30_000)
+ expect(fake.store.getSnapshot()[0]).toHaveProperty('status', 'downloading')
+
+ fake.dropStream(30_000)
+
+ expect(fake.store.getSnapshot()[0]).toMatchObject({
+ status: 'error',
+ error: { code: 'EVENT_STREAM_ERROR' },
+ })
+ expect(fake.close).toHaveBeenCalled()
+ })
+
+ it('keeps monitoring a stream that recovers between drops', async (t) => {
+ const fake = createStoreWithFakeEventSource()
+ t.onTestFinished(fake.store.listen())
+ fake.receiveShare()
+
+ await fake.startDownload()
+
+ // The map server sends the current state as the first message on a new
+ // stream, so a delivered event means the outage is over
+ for (let i = 1; i <= 10; i++) {
+ fake.dropStream(30_000)
+ fake.dropStream(30_000)
+ fake.emitProgress(i)
+ }
+
+ expect(fake.store.getSnapshot()[0]).toMatchObject({
+ status: 'downloading',
+ bytesDownloaded: 10,
+ })
+ })
+
+ // The stores are plain in-memory state and are never reset, so a share left
+ // in `error` would be a dead row for the life of the app
+ it('lets a share that lost its stream be downloaded again', async (t) => {
+ const fake = createStoreWithFakeEventSource()
+ t.onTestFinished(fake.store.listen())
+ fake.receiveShare()
+
+ await fake.startDownload()
+ fake.dropStream(90_000)
+ expect(fake.store.getSnapshot()[0]).toHaveProperty('status', 'error')
+
+ await fake.startDownload()
+
+ expect(fake.store.getSnapshot()[0]).toMatchObject({
+ status: 'downloading',
+ bytesDownloaded: 0,
+ })
+ // A second monitor, so the retry is watching a live stream rather than
+ // riding on the closed one
+ expect(fake.createEventSource).toHaveBeenCalledTimes(2)
+
+ fake.emitProgress(42)
+ expect(fake.store.getSnapshot()[0]).toMatchObject({
+ status: 'downloading',
+ bytesDownloaded: 42,
+ })
+ })
+})
diff --git a/test/lib/sent-map-shares-store.test.ts b/test/lib/sent-map-shares-store.test.ts
index 0a09171..e888e69 100644
--- a/test/lib/sent-map-shares-store.test.ts
+++ b/test/lib/sent-map-shares-store.test.ts
@@ -95,7 +95,7 @@ describe('SentMapSharesStore', () => {
let receiverMockClientApi: MockClientApi
let receivedStore: ReceivedMapSharesStore
- beforeEach(() => {
+ beforeEach((t) => {
receiverMockClientApi = createMockClientApi()
receivedStore = createReceivedMapSharesStore({
// @ts-expect-error - We're only mocking what we need
@@ -104,6 +104,7 @@ describe('SentMapSharesStore', () => {
getBaseUrl: async () => new URL(receiver.localBaseUrl),
}),
})
+ t.onTestFinished(receivedStore.listen())
})
it('should update status when receiver declines the share', async () => {
@@ -348,7 +349,7 @@ describe('SentMapSharesStore', () => {
let receiverMockClientApi: MockClientApi
let receivedStore: ReceivedMapSharesStore
- beforeEach(() => {
+ beforeEach((t) => {
receiverMockClientApi = createMockClientApi()
receivedStore = createReceivedMapSharesStore({
// @ts-expect-error - We're only mocking what we need
@@ -357,6 +358,7 @@ describe('SentMapSharesStore', () => {
getBaseUrl: async () => new URL(receiver.localBaseUrl),
}),
})
+ t.onTestFinished(receivedStore.listen())
})
it('should not allow cancel after decline', async () => {
diff --git a/test/lib/sync.test.ts b/test/lib/sync.test.ts
new file mode 100644
index 0000000..1bb5757
--- /dev/null
+++ b/test/lib/sync.test.ts
@@ -0,0 +1,286 @@
+/**
+ * @vitest-environment node
+ */
+import type { ComapeoProjectClientApi } from '@comapeo/ipc'
+import { describe, expect, it, vi } from 'vitest'
+
+import {
+ refreshActiveSyncStores,
+ SyncStore,
+ type SyncState,
+} from '../../src/lib/sync.js'
+
+function createSyncState({
+ isSyncEnabled = false,
+ devices = {},
+}: {
+ isSyncEnabled?: boolean
+ /** Sync-enabled remote devices, by outstanding `want` count */
+ devices?: Record
+} = {}): SyncState {
+ return {
+ data: { isSyncEnabled },
+ remoteDeviceSyncState: Object.fromEntries(
+ Object.entries(devices).map(([deviceId, want]) => [
+ deviceId,
+ { data: { isSyncEnabled: true, want, wanted: 0 } },
+ ]),
+ ),
+ } as unknown as SyncState
+}
+
+function createProject({
+ throwOnOff = false,
+ throwOnOn = false,
+}: { throwOnOff?: boolean; throwOnOn?: boolean } = {}) {
+ const on = vi.fn(() => {
+ if (throwOnOn) throw new Error('ProjectClosed')
+ })
+ const off = vi.fn(() => {
+ if (throwOnOff) throw new Error('ProjectClosed')
+ })
+ const getState = vi.fn(async () => {
+ throw new Error('ProjectClosed')
+ })
+
+ return {
+ project: {
+ $sync: { on, off, getState },
+ } as unknown as ComapeoProjectClientApi,
+ on,
+ off,
+ }
+}
+
+// `on`/`off` throw on a project wrapper closed by a backend restart in
+// @comapeo/ipc v9, and on any reference after the whole client is closed. The
+// unsubscribe runs in React effect cleanup, where a throw would take down the
+// tree, so both emitter calls have to tolerate it.
+describe('SyncStore with a closed project wrapper', () => {
+ it('does not throw from the unsubscribe returned by subscribe()', () => {
+ const { project, off } = createProject({ throwOnOff: true })
+ const store = new SyncStore(project)
+
+ const unsubscribe = store.subscribe(() => {})
+
+ expect(() => unsubscribe()).not.toThrow()
+ expect(off).toHaveBeenCalled()
+ })
+
+ it('does not throw from subscribe() when adding the listener fails', () => {
+ const { project, on } = createProject({ throwOnOn: true })
+ const store = new SyncStore(project)
+
+ expect(() => store.subscribe(() => {})).not.toThrow()
+ expect(on).toHaveBeenCalled()
+ // The failure is surfaced through the snapshot, the same way a rejected
+ // `getState()` is
+ expect(() => store.getStateSnapshot()).toThrow(/ProjectClosed/)
+ })
+})
+
+function createLiveProject() {
+ const listeners = new Set<(state: SyncState) => void>()
+ let nextState = createSyncState()
+
+ const on = vi.fn((_event: string, listener: (state: SyncState) => void) => {
+ listeners.add(listener)
+ })
+ const off = vi.fn((_event: string, listener: (state: SyncState) => void) => {
+ listeners.delete(listener)
+ })
+ const getState = vi.fn(async () => nextState)
+
+ return {
+ project: {
+ $sync: { on, off, getState },
+ } as unknown as ComapeoProjectClientApi,
+ on,
+ off,
+ getState,
+ listenerCount: () => listeners.size,
+ emit(state: SyncState) {
+ nextState = state
+ for (const listener of listeners) listener(state)
+ },
+ setNextState(state: SyncState) {
+ nextState = state
+ },
+ }
+}
+
+// With a permanent project client reference (@comapeo/ipc v10) the per-project
+// store cache hands back the same `SyncStore` after a backend restart, so
+// anything it latched from the previous backend has to be cleared explicitly.
+describe('recovering a SyncStore after a backend restart', () => {
+ it('clears a sticky error on a store that is still listened to', async (t) => {
+ const project = createLiveProject()
+ project.getState.mockRejectedValueOnce(new Error('ProjectClosed'))
+
+ const store = new SyncStore(project.project)
+ t.onTestFinished(store.subscribe(() => {}))
+
+ await vi.waitFor(() => {
+ expect(() => store.getStateSnapshot()).toThrow(/ProjectClosed/)
+ })
+
+ const fresh = createSyncState({ isSyncEnabled: true })
+ project.setNextState(fresh)
+ refreshActiveSyncStores()
+
+ expect(() => store.getStateSnapshot()).not.toThrow()
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).toBe(fresh)
+ })
+ })
+
+ // The real sequence when the error reaches a boundary: the snapshot throws
+ // during render, the boundary unmounts the subtree, and that removes the
+ // last listener. The remount is handed back this same store and reads the
+ // snapshot during render, before `subscribe` can run — so the error must
+ // not still be there.
+ it('recovers a store whose error boundary unmounted every listener', async () => {
+ const project = createLiveProject()
+ project.getState.mockRejectedValueOnce(new Error('ProjectClosed'))
+
+ const store = new SyncStore(project.project)
+ const unsubscribe = store.subscribe(() => {})
+
+ await vi.waitFor(() => {
+ expect(() => store.getStateSnapshot()).toThrow(/ProjectClosed/)
+ })
+
+ unsubscribe()
+ const fresh = createSyncState({ isSyncEnabled: true })
+ project.setNextState(fresh)
+
+ refreshActiveSyncStores()
+
+ expect(() => store.getStateSnapshot()).not.toThrow()
+
+ const unsubscribeAfterRemount = store.subscribe(() => {})
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).toBe(fresh)
+ })
+ unsubscribeAfterRemount()
+ })
+
+ // Same wedge, without a restart: a boundary that offers a retry button has
+ // to be able to get somewhere on its own.
+ it('recovers a remount with no restart notification at all', async () => {
+ const project = createLiveProject()
+ project.getState.mockRejectedValueOnce(new Error('ProjectClosed'))
+
+ const store = new SyncStore(project.project)
+ const unsubscribe = store.subscribe(() => {})
+
+ await vi.waitFor(() => {
+ expect(() => store.getStateSnapshot()).toThrow(/ProjectClosed/)
+ })
+
+ unsubscribe()
+
+ expect(() => store.getStateSnapshot()).not.toThrow()
+
+ const fresh = createSyncState({ isSyncEnabled: true })
+ project.setNextState(fresh)
+ const unsubscribeAfterRemount = store.subscribe(() => {})
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).toBe(fresh)
+ })
+ unsubscribeAfterRemount()
+ })
+
+ it('notifies listeners with the state from the new backend', async (t) => {
+ const project = createLiveProject()
+ const store = new SyncStore(project.project)
+ const listener = vi.fn()
+ t.onTestFinished(store.subscribe(listener))
+
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).not.toBeNull()
+ })
+ listener.mockClear()
+
+ const fresh = createSyncState({ isSyncEnabled: true })
+ project.setNextState(fresh)
+ refreshActiveSyncStores()
+
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).toBe(fresh)
+ })
+ expect(listener).toHaveBeenCalled()
+ // The transport owner re-sends its subscriptions before announcing the
+ // restart, so the store must not attach a second listener
+ expect(project.listenerCount()).toBe(1)
+ })
+
+ // Progress is a ratio against the largest sync count seen so far. Clearing
+ // those baselines while keeping the previous backend's state makes the
+ // ratio read 1, which the UI shows as "sync complete".
+ it('never reports sync as complete on the way through a restart', async (t) => {
+ const project = createLiveProject()
+ const store = new SyncStore(project.project)
+ const seen: Array = []
+ t.onTestFinished(
+ store.subscribe(() => {
+ seen.push(store.getDataProgressSnapshot())
+ }),
+ )
+
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).not.toBeNull()
+ })
+ project.emit(createSyncState({ devices: { 'device-a': 10 } }))
+ project.emit(createSyncState({ devices: { 'device-a': 5 } }))
+ expect(store.getDataProgressSnapshot()).toBe(0.5)
+
+ project.setNextState(createSyncState({ devices: { 'device-a': 8 } }))
+ refreshActiveSyncStores()
+
+ // Not known yet, rather than a number derived from the dead backend
+ expect(store.getDataProgressSnapshot()).toBeNull()
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).not.toBeNull()
+ })
+ expect(store.getDataProgressSnapshot()).toBe(0)
+ expect(seen).not.toContain(1)
+ })
+
+ it('does not touch a store with no listeners', async () => {
+ const project = createLiveProject()
+ const store = new SyncStore(project.project)
+ const unsubscribe = store.subscribe(() => {})
+
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).not.toBeNull()
+ })
+
+ unsubscribe()
+ project.getState.mockClear()
+
+ refreshActiveSyncStores()
+
+ expect(project.getState).not.toHaveBeenCalled()
+ })
+
+ it('re-attaches the listener when the first attach failed', async (t) => {
+ const project = createLiveProject()
+ project.on.mockImplementationOnce(() => {
+ throw new Error('ProjectClosed')
+ })
+
+ const store = new SyncStore(project.project)
+ t.onTestFinished(store.subscribe(() => {}))
+
+ expect(() => store.getStateSnapshot()).toThrow(/ProjectClosed/)
+ expect(project.listenerCount()).toBe(0)
+
+ refreshActiveSyncStores()
+
+ expect(project.listenerCount()).toBe(1)
+ await vi.waitFor(() => {
+ expect(store.getStateSnapshot()).not.toBeNull()
+ })
+ })
+})