fix(mobile): decouple notification host ownership

This commit is contained in:
Jinwoo-H
2026-08-07 01:52:00 -04:00
parent 595097b5fc
commit b4aea2a4e9
27 changed files with 1543 additions and 75 deletions
+2
View File
@@ -13,6 +13,7 @@ import { useOpenNotificationRoute } from '../src/notifications/use-open-notifica
import { loadHostCatalog } from '../src/transport/host-store'
import { extractPairingCodeFromUrl } from '../src/transport/pairing'
import { recoverMobileRelayPairing } from '../src/transport/mobile-relay-pairing-recovery'
import { NotificationHostConnectionOwner } from '../src/notifications/notification-host-connection-owner'
// Why: keeps the native splash screen visible until the React tree is mounted
// and ready to render. Without this the user sees a blank white/black frame
@@ -159,6 +160,7 @@ export default function RootLayout() {
return (
<RpcClientProvider>
<NotificationHostConnectionOwner />
<View style={styles.root} onLayout={onNavigatorLayout}>
<StatusBar style="light" />
<Stack
+2 -12
View File
@@ -32,7 +32,6 @@ import {
selectHomeAutoConnectHostIds
} from '../src/transport/home-host-auto-connect'
import { classifyConnection } from '../src/transport/connection-health'
import { subscribeToDesktopNotifications } from '../src/notifications/mobile-notifications'
import {
loadMobileOnboardingSteps,
mobileOnboardingDestination
@@ -246,7 +245,7 @@ export default function HomeScreen() {
// Why: scoped to the paired hosts so an unpaired desktop's cached reply leaves the header total.
const stats = useMemo(() => totalHomeStats(statsByHost, hostIds), [statsByHost, hostIds])
const autoConnectHostIds = useMemo(() => selectHomeAutoConnectHostIds(hosts), [hosts])
const allClients = useAllHostClients(hostIds, {
const allClients = useAllHostClients(autoConnectHostIds, {
autoConnectHostIds,
closeUnusedOnRelease: true
})
@@ -432,7 +431,7 @@ export default function HomeScreen() {
})
}, [allClients, hostCatalog])
// Notif/accounts subs + a snapshot read per connect for one host. Lives outside the effect body
// Accounts sub + a snapshot read per connect for one host. Lives outside the effect body
// because react-doctor's effect-needs-cleanup false-positives on `subscribe` inside one; the
// returned disposer owns every handle allocated here.
const wireHostSubscriptions = (entry: {
@@ -440,15 +439,11 @@ export default function HomeScreen() {
client: RpcClient
state: ConnectionState
}) => {
let unsubNotif: (() => void) | null = null
let unsubAccounts: (() => void) | null = null
const refetchGate = createHostConnectRefetchGate()
const wireUp = (state: ConnectionState) => {
const reconnected = refetchGate.observe(state)
if (state === 'connected') {
if (!unsubNotif) {
unsubNotif = subscribeToDesktopNotifications(entry.client, entry.hostId)
}
if (!unsubAccounts) {
unsubAccounts = entry.client.subscribe('accounts.subscribe', null, (payload) => {
if (!payload || typeof payload !== 'object') {
@@ -474,10 +469,6 @@ export default function HomeScreen() {
fetchTaskProviders(entry.client, entry.hostId, setTaskProvidersByHost, () => false)
}
} else {
if (unsubNotif) {
unsubNotif()
unsubNotif = null
}
if (unsubAccounts) {
unsubAccounts()
unsubAccounts = null
@@ -488,7 +479,6 @@ export default function HomeScreen() {
const unsubState = entry.client.onStateChange(wireUp)
return () => {
unsubState()
unsubNotif?.()
unsubAccounts?.()
}
}
+5 -3
View File
@@ -9,7 +9,7 @@ import {
type PreProfilePairingAttempt
} from '../src/transport/pre-profile-pairing-coordinator'
import type { ConnectionLogEntry } from '../src/transport/types'
import { useCloseHost } from '../src/transport/client-context'
import { useCloseHost, useForceReconnect } from '../src/transport/client-context'
import { colors, spacing, radii, typography } from '../src/theme/mobile-theme'
import { ConnectionLog } from '../src/components/ConnectionLog'
import {
@@ -29,6 +29,7 @@ const PAIRING_OVERALL_TIMEOUT_MS = 25_000
export default function PairConfirmScreen() {
const router = useRouter()
const closeHost = useCloseHost()
const forceReconnect = useForceReconnect()
const insets = useSafeAreaInsets()
const params = useLocalSearchParams<{ code?: string }>()
const [status, setStatus] = useState<Status>('awaiting-confirm')
@@ -113,9 +114,10 @@ export default function PairConfirmScreen() {
// (STA-1840 dedup), so a client cached under that id from an earlier
// pairing would keep the stale endpoint/relay. Close it so the
// destination screen opens a fresh client with the newly-paired
// profile — the removeHost() path already refreshes on re-pair, and a
// brand-new host has no cached entry so this is a no-op.
// profile. Reopen immediately so root notification ownership does not
// stay clientless while onboarding is visible.
closeHost(hostId)
await forceReconnect(hostId)
const onboardingSteps = await loadMobileOnboardingSteps()
if (!mountedRef.current) {
return
+5 -3
View File
@@ -18,7 +18,7 @@ import {
type PreProfilePairingAttempt
} from '../src/transport/pre-profile-pairing-coordinator'
import type { ConnectionLogEntry, PairingOffer } from '../src/transport/types'
import { useCloseHost } from '../src/transport/client-context'
import { useCloseHost, useForceReconnect } from '../src/transport/client-context'
import { colors, spacing, radii, typography } from '../src/theme/mobile-theme'
import { TextInputModal } from '../src/components/TextInputModal'
import { ConnectionLog } from '../src/components/ConnectionLog'
@@ -48,6 +48,7 @@ function Step({ number, text }: { number: number; text: string }) {
export default function PairScanScreen() {
const router = useRouter()
const closeHost = useCloseHost()
const forceReconnect = useForceReconnect()
const insets = useSafeAreaInsets()
const [permission, requestPermission] = useCameraPermissions()
const [status, setStatus] = useState<'scanning' | 'connecting' | 'error'>('scanning')
@@ -157,9 +158,10 @@ export default function PairScanScreen() {
// (STA-1840 dedup), so a client cached under that id from an earlier
// pairing would keep the stale endpoint/relay. Close it so the
// destination screen opens a fresh client with the newly-paired
// profile — the removeHost() path already refreshes on re-pair, and a
// brand-new host has no cached entry so this is a no-op.
// profile. Reopen immediately so root notification ownership does not
// stay clientless while onboarding is visible.
closeHost(hostId)
await forceReconnect(hostId)
const onboardingSteps = await loadMobileOnboardingSteps()
if (!mountedRef.current) {
return
@@ -35,6 +35,8 @@ const scheduledNotificationsByHostAndNotificationId = new Map<string, ScheduledN
// Why: keys never repeat and are only freed on desktop dismiss (which remote users often miss), so bound the map to stop unbounded growth.
const MAX_SCHEDULED_NOTIFICATIONS = 256
let maxScheduledNotifications = MAX_SCHEDULED_NOTIFICATIONS
let notificationChannelConfigured = false
let notificationChannelConfiguring = false
function getStoredNotificationKey(hostId: string, notificationId: string): string {
return `${encodeURIComponent(hostId)}:${encodeURIComponent(notificationId)}`
@@ -62,15 +64,35 @@ export function setScheduledNotificationsMaxForTests(max?: number): void {
maxScheduledNotifications = max ?? MAX_SCHEDULED_NOTIFICATIONS
}
export function resetNotificationChannelConfigurationForTests(): void {
notificationChannelConfigured = false
notificationChannelConfiguring = false
}
export function configureNotificationChannel(): void {
if (Platform.OS === 'android') {
void Notifications.setNotificationChannelAsync('orca-desktop', {
if (
Platform.OS !== 'android' ||
notificationChannelConfigured ||
notificationChannelConfiguring
) {
return
}
notificationChannelConfiguring = true
void Promise.resolve(
Notifications.setNotificationChannelAsync('orca-desktop', {
name: 'Desktop Notifications',
importance: Notifications.AndroidImportance.HIGH,
vibrationPattern: [0, 250],
lightColor: '#6366f1'
})
}
)
.then(() => {
notificationChannelConfigured = true
})
.catch(() => {})
.finally(() => {
notificationChannelConfiguring = false
})
}
export async function showLocalNotification(
@@ -5,7 +5,10 @@ export {
getNotificationPermissionState,
type NotificationPermissionState
} from './notification-permissions'
export { setScheduledNotificationsMaxForTests } from './local-notification-scheduling'
export {
resetNotificationChannelConfigurationForTests,
setScheduledNotificationsMaxForTests
} from './local-notification-scheduling'
import {
configureNotificationChannel,
dismissLocalNotification,
@@ -65,6 +68,9 @@ export function subscribeToDesktopNotifications(client: RpcClient, hostId: strin
}
return enqueueHostDelivery(session, async () => {
try {
if (disposed) {
return
}
await deliverLive(type, event)
} finally {
if (type === 'notification') {
@@ -88,6 +94,9 @@ export function subscribeToDesktopNotifications(client: RpcClient, hostId: strin
} else {
await dismissLocalNotification(event as DismissNotificationEvent, hostId)
}
if (session.retired) {
return
}
// Why after the await, exactly like the watermark below: `seen` asserts this event
// reached the user (#8129). Marked before, a rejected show leaves the key behind and
// every later replay is dropped as a duplicate — loss the quarantine cannot recover,
@@ -141,13 +150,13 @@ export function subscribeToDesktopNotifications(client: RpcClient, hostId: strin
// Why: desktop cuts by seq > lastSeenSeq, so re-fetching from the watermark is idempotent (session.seen guards residual overlap).
async function fetchMissed(): Promise<void> {
if (disposed) {
if (session.retired) {
return
}
// Captured before the request: everything at or below it is known delivered, so
// it is the floor the watermark falls back to if this catch-up never completes.
const askFrom = catchUpWatermarkSeq(session)
const missed = await client
const catchUp = await client
.sendRequest('notifications.getMissedSince', {
lastSeenSeq: askFrom,
// Why: sending the epoch lets the desktop reject a watermark from a counter
@@ -159,17 +168,24 @@ export function subscribeToDesktopNotifications(client: RpcClient, hostId: strin
return null
}
const result = response.result as { notifications?: unknown[]; epoch?: string } | undefined
adoptNotificationEpoch(session, hostId, result?.epoch)
return Array.isArray(result?.notifications) ? result.notifications : []
return {
epoch: result?.epoch,
notifications: Array.isArray(result?.notifications) ? result.notifications : []
}
})
.catch(() => null)
if (missed == null) {
if (session.retired) {
return
}
if (catchUp == null) {
// Why quarantine rather than retry: the range this catch-up abandoned stays
// unrecovered until SOME later one succeeds, and a live seq persisting past it
// meanwhile would make the desktop cut it forever.
quarantineCatchUpWatermark(session, hostId, askFrom)
return
}
adoptNotificationEpoch(session, hostId, catchUp.epoch)
const missed = catchUp.notifications
// Why the whole batch is ONE queue entry (#8591): awaiting per event returns to
// the event loop between replays, so a live seq 11 slots into the chain between
// seq 6 and 7 and persists a watermark past a notification still unshown. Why the
@@ -181,6 +197,9 @@ export function subscribeToDesktopNotifications(client: RpcClient, hostId: strin
let contiguousSeq = askFrom
let drained = false
try {
if (disposed) {
return
}
for (const raw of missed) {
// Re-checked per event: the batch can start before a teardown and still be
// draining after it, and a torn-down host must stop pushing.
@@ -195,7 +214,7 @@ export function subscribeToDesktopNotifications(client: RpcClient, hostId: strin
} finally {
if (drained) {
resolveCatchUpQuarantine(session, hostId)
} else {
} else if (!session.retired) {
quarantineCatchUpWatermark(session, hostId, contiguousSeq)
}
}
@@ -0,0 +1,28 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import * as Notifications from 'expo-notifications'
import {
configureNotificationChannel,
resetNotificationChannelConfigurationForTests
} from './local-notification-scheduling'
vi.mock('expo-notifications', () => ({
AndroidImportance: { HIGH: 'high' },
setNotificationChannelAsync: vi.fn()
}))
vi.mock('react-native', () => ({ Platform: { OS: 'android', Version: 35 } }))
describe('Android notification channel configuration', () => {
beforeEach(() => {
vi.clearAllMocks()
resetNotificationChannelConfigurationForTests()
})
it('configures the process-wide channel once across host subscriptions', async () => {
configureNotificationChannel()
configureNotificationChannel()
await Promise.resolve()
expect(Notifications.setNotificationChannelAsync).toHaveBeenCalledOnce()
})
})
@@ -0,0 +1,198 @@
import { useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react'
import { AppState } from 'react-native'
import type { RpcClient } from '../transport/rpc-client'
import { subscribeHostCollectionChanges } from '../transport/host-collection-changes'
import { selectConnectableHostProfiles } from '../transport/host-catalog-selection'
import { loadHostCatalog } from '../transport/host-store'
import { usePrimeHosts } from '../transport/client-context'
import { connectionLogStore } from '../transport/connection-log-buffer'
import type { ConnectionState, HostProfile } from '../transport/types'
import { useAllHostClients } from '../transport/use-all-host-clients'
import { subscribeToDesktopNotifications } from './mobile-notifications'
import { retireHostNotificationState } from './notification-reconnect-catchup'
const HOST_REFRESH_RETRY_DELAYS_MS = [250, 1_000, 5_000, 15_000] as const
const clientIdentities = new WeakMap<RpcClient, number>()
let nextClientIdentity = 1
function clientIdentity(client: RpcClient): number {
let identity = clientIdentities.get(client)
if (identity == null) {
identity = nextClientIdentity++
clientIdentities.set(client, identity)
}
return identity
}
function subscribeToHostNotificationLifecycle(client: RpcClient, hostId: string): () => void {
let unsubscribeNotifications: (() => void) | null = null
const wire = (state: ConnectionState) => {
if (state === 'connected' && !unsubscribeNotifications) {
unsubscribeNotifications = subscribeToDesktopNotifications(client, hostId)
} else if (state !== 'connected') {
unsubscribeNotifications?.()
unsubscribeNotifications = null
}
}
const unsubscribeState = client.onStateChange(wire)
wire(client.getState())
return () => {
unsubscribeState()
unsubscribeNotifications?.()
}
}
function NotificationHostClientOwner({ client, hostId }: { client: RpcClient; hostId: string }) {
useEffect(() => subscribeToHostNotificationLifecycle(client, hostId), [client, hostId])
return null
}
function subscribeNotificationHostCatalog(
updateHosts: (hosts: HostProfile[]) => void,
retireHosts: (hostIds: readonly string[]) => void
): () => void {
let disposed = false
let loadRevision = 0
let retryIndex = 0
let retryTimer: ReturnType<typeof setTimeout> | null = null
let profilesByHostId = new Map<string, HostProfile>()
const scheduleRetry = () => {
if (retryTimer || disposed) {
return
}
const delay = HOST_REFRESH_RETRY_DELAYS_MS[Math.min(retryIndex, 3)]
retryIndex += 1
retryTimer = setTimeout(() => {
retryTimer = null
refreshHosts()
}, delay)
}
const refreshHosts = (resetRetry = false) => {
if (resetRetry) {
retryIndex = 0
if (retryTimer) {
clearTimeout(retryTimer)
retryTimer = null
}
}
const revision = ++loadRevision
void loadHostCatalog()
.then((catalog) => {
if (!disposed && revision === loadRevision) {
const nextHosts = selectConnectableHostProfiles(catalog)
const readyHostIds = new Set(nextHosts.map(({ id }) => id))
for (const entry of catalog) {
if (entry.credentialStatus === 'temporarily-unavailable') {
const retained = profilesByHostId.get(entry.id)
if (retained && !readyHostIds.has(entry.id)) {
nextHosts.push(retained)
}
}
}
profilesByHostId = new Map(nextHosts.map((host) => [host.id, host]))
updateHosts(nextHosts)
if (
catalog.some(({ credentialStatus }) => credentialStatus === 'temporarily-unavailable')
) {
scheduleRetry()
} else {
retryIndex = 0
}
}
})
.catch(scheduleRetry)
}
const unsubscribeHosts = subscribeHostCollectionChanges(({ retiredHostIds }) => {
if (retiredHostIds.length > 0) {
for (const hostId of retiredHostIds) {
profilesByHostId.delete(hostId)
}
retireHosts(retiredHostIds)
}
refreshHosts(true)
})
const appStateSubscription = AppState.addEventListener('change', (state) => {
if (state === 'active') {
refreshHosts(true)
}
})
refreshHosts()
return () => {
disposed = true
if (retryTimer) {
clearTimeout(retryTimer)
}
unsubscribeHosts()
appStateSubscription.remove()
}
}
export function NotificationHostConnectionOwner() {
const [hosts, setHosts] = useState<HostProfile[]>([])
const [retiringHostIds, setRetiringHostIds] = useState<string[]>([])
const retirementStartedRef = useRef(new Set<string>())
const mountedRef = useRef(true)
const primeHosts = usePrimeHosts()
useEffect(
() => () => {
mountedRef.current = false
},
[]
)
useEffect(
() =>
subscribeNotificationHostCatalog(setHosts, (retiredHostIds) => {
const retired = new Set(retiredHostIds)
setHosts((current) => current.filter(({ id }) => !retired.has(id)))
setRetiringHostIds((current) => [...new Set([...current, ...retiredHostIds])])
}),
[]
)
const retiringHostIdSet = useMemo(() => new Set(retiringHostIds), [retiringHostIds])
const activeHosts = useMemo(
() => hosts.filter(({ id }) => !retiringHostIdSet.has(id)),
[hosts, retiringHostIdSet]
)
useLayoutEffect(() => {
primeHosts(activeHosts)
}, [activeHosts, primeHosts])
const hostIds = useMemo(() => activeHosts.map(({ id }) => id), [activeHosts])
const clients = useAllHostClients(hostIds, {
closeUnusedOnRelease: true,
observeConnectionState: false
})
useEffect(() => {
for (const hostId of retiringHostIds) {
if (retirementStartedRef.current.has(hostId)) {
continue
}
retirementStartedRef.current.add(hostId)
connectionLogStore.clear(hostId)
void retireHostNotificationState(hostId).finally(() => {
retirementStartedRef.current.delete(hostId)
if (mountedRef.current) {
setRetiringHostIds((current) => current.filter((id) => id !== hostId))
}
})
}
}, [retiringHostIds])
return (
<>
{clients.map(({ client, hostId }) => (
<NotificationHostClientOwner
key={`${hostId}:${clientIdentity(client)}`}
client={client}
hostId={hostId}
/>
))}
</>
)
}
@@ -0,0 +1,669 @@
import { createElement, useEffect } from 'react'
import * as Notifications from 'expo-notifications'
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { RpcClient } from '../transport/rpc-client'
import type { ConnectionState, HostProfile } from '../transport/types'
const connectMock = vi.fn()
const loadHostsMock = vi.fn()
const loadHostCatalogMock = vi.fn()
const hostCollectionChange = vi.hoisted(() => ({
listener: null as null | ((change: { retiredHostIds: readonly string[] }) => void)
}))
vi.mock('expo-notifications', () => ({
AndroidImportance: { HIGH: 'high' },
setNotificationChannelAsync: vi.fn(),
getPermissionsAsync: vi.fn(),
requestPermissionsAsync: vi.fn(),
scheduleNotificationAsync: vi.fn(),
dismissNotificationAsync: vi.fn()
}))
vi.mock('react-native', () => ({
AppState: {
addEventListener: vi.fn(() => ({ remove: vi.fn() }))
},
Platform: { OS: 'ios', Version: 18 }
}))
vi.mock('@react-native-async-storage/async-storage', () => ({
default: {
getItem: vi.fn(async () => null),
setItem: vi.fn(async () => undefined),
removeItem: vi.fn(async () => undefined)
}
}))
vi.mock('../storage/preferences', () => ({
loadPushNotificationsEnabled: vi.fn(async () => true)
}))
vi.mock('../transport/host-logical-client', () => ({
openHostLogicalClient: (...args: unknown[]) => connectMock(...args)
}))
vi.mock('../transport/host-store', () => ({
loadHostCatalog: () => loadHostCatalogMock(),
loadHosts: () => loadHostsMock()
}))
vi.mock('../transport/host-collection-changes', () => ({
subscribeHostCollectionChanges: vi.fn(
(listener: (change: { retiredHostIds: readonly string[] }) => void) => {
hostCollectionChange.listener = listener
return () => {
hostCollectionChange.listener = null
}
}
)
}))
vi.mock('../transport/connection-revival-triggers', () => ({
subscribeConnectionRevivalTriggers: () => () => {}
}))
import { subscribeToDesktopNotifications } from './mobile-notifications'
import { NotificationHostConnectionOwner } from './notification-host-connection-owner'
import { resetHostNotificationSessionsForTests } from './notification-reconnect-catchup'
import { RpcClientProvider, useCloseHost, useForceReconnect } from '../transport/client-context'
import { useAllHostClients } from '../transport/use-all-host-clients'
type FakeClient = RpcClient & {
emitState: (state: ConnectionState) => void
emitNotification: (event: unknown) => void
emitNotificationReady: () => void
getMissedCalls: () => number
notificationSubscribeCalls: () => number
closeCalls: () => number
}
function makeHost(
id: string,
lastConnected: number,
extra: Partial<HostProfile> = {}
): HostProfile {
return {
id,
name: id,
endpoint: `ws://${id}.internal:8787`,
deviceToken: `token-${id}`,
publicKeyB64: `key-${id}`,
lastConnected,
...extra
}
}
const HOSTS = [
makeHost('local-recent', 40),
makeHost('relay-recent', 30, { relayHostId: 'AbCdEf0123_-xyZ9' }),
makeHost('ssh-recent', 20),
makeHost('older-fourth', 10)
]
function makeFakeClient(
initialState: ConnectionState = 'connected',
connectDuringListenerRegistration = false
): FakeClient {
let state = initialState
let notificationReady: ((data: unknown) => void) | null = null
let notificationSubscribes = 0
let missedCalls = 0
const stateListeners = new Set<(next: ConnectionState) => void>()
const close = vi.fn()
const client = {
sendRequest: vi.fn(async (method: string) => {
if (method === 'notifications.getMissedSince') {
missedCalls += 1
return { ok: true, result: { notifications: [], epoch: 'epoch-1' } }
}
return { ok: true, result: undefined }
}),
subscribe: vi.fn((method: string, _params: unknown, listener: (data: unknown) => void) => {
if (method === 'notifications.subscribe') {
notificationSubscribes += 1
notificationReady = listener
}
return () => {
if (notificationReady === listener) {
notificationReady = null
}
}
}),
updateTerminalSubscriptionViewport: vi.fn(),
getState: () => state,
getReconnectAttempt: () => 0,
getLastConnectedAt: () => null,
onStateChange: (listener: (next: ConnectionState) => void) => {
stateListeners.add(listener)
if (connectDuringListenerRegistration) {
state = 'connected'
}
return () => stateListeners.delete(listener)
},
notifyForeground: vi.fn(),
close,
emitState: (next: ConnectionState) => {
state = next
for (const listener of stateListeners) {
listener(next)
}
},
emitNotification: (event: unknown) => {
notificationReady?.(event)
},
emitNotificationReady: () => {
notificationReady?.({ type: 'ready', subscriptionId: 'subscription-1', epoch: 'epoch-1' })
},
getMissedCalls: () => missedCalls,
notificationSubscribeCalls: () => notificationSubscribes,
closeCalls: () => close.mock.calls.length
}
return client as unknown as FakeClient
}
function HomeNotificationProbe(): null {
const hostIds = HOSTS.map(({ id }) => id)
const clients = useAllHostClients(hostIds, {
autoConnectHostIds: hostIds.slice(0, 3),
closeUnusedOnRelease: true
})
const clientKey = clients
.map(({ hostId }) => hostId)
.sort()
.join(',')
useEffect(() => {
const cleanups = clients.map((entry) => {
let unsubscribeNotifications: (() => void) | null = null
const wire = (state: ConnectionState) => {
if (state === 'connected' && !unsubscribeNotifications) {
unsubscribeNotifications = subscribeToDesktopNotifications(entry.client, entry.hostId)
} else if (state !== 'connected') {
unsubscribeNotifications?.()
unsubscribeNotifications = null
}
}
wire(entry.state)
const unsubscribeState = entry.client.onStateChange(wire)
return () => {
unsubscribeState()
unsubscribeNotifications?.()
}
})
return () => {
for (const cleanup of cleanups) {
cleanup()
}
}
}, [clientKey])
return null
}
function HomeDataProbe({ onRender }: { onRender: () => void }): null {
onRender()
useAllHostClients(
HOSTS.slice(0, 3).map(({ id }) => id),
{ closeUnusedOnRelease: true }
)
return null
}
let closeOwnedHost: ((hostId: string) => void) | null = null
let reconnectOwnedHost: ((hostId: string) => Promise<void>) | null = null
function RePairProbe(): null {
closeOwnedHost = useCloseHost()
reconnectOwnedHost = useForceReconnect()
return null
}
function suppressRendererWarning(): () => void {
const original = console.error
const spy = vi.spyOn(console, 'error').mockImplementation((...args) => {
if (typeof args[0] === 'string' && args[0].includes('react-test-renderer is deprecated')) {
return
}
original(...args)
})
return () => spy.mockRestore()
}
function flushAsync(): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, 0))
}
describe('paired-host notification ownership', () => {
beforeEach(() => {
globalThis.IS_REACT_ACT_ENVIRONMENT = true
vi.clearAllMocks()
resetHostNotificationSessionsForTests()
hostCollectionChange.listener = null
closeOwnedHost = null
reconnectOwnedHost = null
loadHostCatalogMock.mockImplementation(async () =>
(await loadHostsMock()).map((profile: HostProfile) => ({
...profile,
credentialStatus: 'ready',
profile
}))
)
})
it('keeps live and reconnect catch-up delivery for the fourth paired host', async () => {
const clients = new Map<string, FakeClient>()
connectMock.mockImplementation((profile: HostProfile) => {
const client = makeFakeClient()
clients.set(profile.id, client)
return client
})
loadHostsMock.mockResolvedValue(HOSTS)
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
const notificationOwner =
process.env.ORCA_TEST_DISABLE_NOTIFICATION_OWNER === '1'
? null
: createElement(NotificationHostConnectionOwner)
renderer = create(
createElement(
RpcClientProvider,
null,
notificationOwner,
createElement(HomeNotificationProbe)
)
)
await flushAsync()
})
await act(async () => {
for (const client of clients.values()) {
client.emitNotificationReady()
client.emitState('reconnecting')
client.emitState('connected')
client.emitNotificationReady()
}
await flushAsync()
})
expect(
HOSTS.map(({ id }) => ({
hostId: id,
liveSubscribed: (clients.get(id)?.notificationSubscribeCalls() ?? 0) > 0,
reconnectCatchUp: (clients.get(id)?.getMissedCalls() ?? 0) > 0
}))
).toEqual(
HOSTS.map(({ id }) => ({ hostId: id, liveSubscribed: true, reconnectCatchUp: true }))
)
} finally {
restore()
act(() => renderer?.unmount())
}
})
it('subscribes every paired host without adding Home data streams at scale', async () => {
vi.useFakeTimers()
const hosts = Array.from({ length: 1_000 }, (_, index) =>
makeHost(`host-${index}`, 1_000 - index)
)
const clients: FakeClient[] = []
connectMock.mockImplementation(() => {
const client = makeFakeClient()
clients.push(client)
return client
})
loadHostsMock.mockResolvedValue(hosts)
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(RpcClientProvider, null, createElement(NotificationHostConnectionOwner))
)
await Promise.resolve()
})
await act(async () => {
await vi.runAllTimersAsync()
})
expect(clients).toHaveLength(1_000)
expect(loadHostCatalogMock).toHaveBeenCalledOnce()
expect(loadHostsMock).toHaveBeenCalledOnce()
expect(clients.every((client) => client.notificationSubscribeCalls() === 1)).toBe(true)
expect(
clients.some((client) =>
vi.mocked(client.subscribe).mock.calls.some(([method]) => method === 'accounts.subscribe')
)
).toBe(false)
} finally {
restore()
act(() => renderer?.unmount())
vi.useRealTimers()
}
})
it('acquires newly paired hosts and closes removed notification clients', async () => {
const clients = new Map<string, FakeClient>()
connectMock.mockImplementation((profile: HostProfile) => {
const client = makeFakeClient()
clients.set(profile.id, client)
return client
})
loadHostsMock.mockResolvedValue(HOSTS)
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(RpcClientProvider, null, createElement(NotificationHostConnectionOwner))
)
await flushAsync()
})
const newlyPaired = makeHost('newly-paired', 50)
loadHostsMock.mockResolvedValue([...HOSTS, newlyPaired])
await act(async () => {
hostCollectionChange.listener?.({ retiredHostIds: [] })
await flushAsync()
})
expect(clients.get(newlyPaired.id)?.notificationSubscribeCalls()).toBe(1)
expect(clients.get(HOSTS[0]!.id)?.notificationSubscribeCalls()).toBe(1)
loadHostsMock.mockResolvedValue(HOSTS.slice(0, 3))
await act(async () => {
hostCollectionChange.listener?.({
retiredHostIds: ['older-fourth', newlyPaired.id]
})
await flushAsync()
})
expect(clients.get('older-fourth')?.closeCalls()).toBe(1)
expect(clients.get(newlyPaired.id)?.closeCalls()).toBe(1)
expect(clients.get(HOSTS[0]!.id)?.notificationSubscribeCalls()).toBe(1)
} finally {
restore()
act(() => renderer?.unmount())
}
})
it('waits for same-id retirement before binding the replacement subscription', async () => {
let finishSchedule: (identifier: string) => void = () => {}
vi.mocked(Notifications.getPermissionsAsync).mockResolvedValue({
status: 'granted',
canAskAgain: true
} as never)
vi.mocked(Notifications.scheduleNotificationAsync).mockReturnValueOnce(
new Promise<string>((resolve) => {
finishSchedule = resolve
})
)
const clients: FakeClient[] = []
connectMock.mockImplementation(() => {
const client = makeFakeClient()
clients.push(client)
return client
})
loadHostsMock.mockResolvedValue([HOSTS[0]])
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(RpcClientProvider, null, createElement(NotificationHostConnectionOwner))
)
await flushAsync()
})
clients[0]?.emitNotification({
type: 'notification',
source: 'agent-task-complete',
title: 'Held delivery',
body: 'Keep retirement open',
notificationId: 'agent:held'
})
await flushAsync()
loadHostsMock.mockResolvedValue([])
await act(async () => {
hostCollectionChange.listener?.({ retiredHostIds: [HOSTS[0]!.id] })
await flushAsync()
})
expect(clients).toHaveLength(1)
expect(clients[0]?.closeCalls()).toBe(1)
loadHostsMock.mockResolvedValue([HOSTS[0]])
await act(async () => {
hostCollectionChange.listener?.({ retiredHostIds: [] })
await flushAsync()
})
expect(clients).toHaveLength(1)
await act(async () => {
finishSchedule('scheduled-held')
await flushAsync()
})
await vi.waitFor(() => expect(clients).toHaveLength(2))
await act(async () => {
clients[1]?.emitNotificationReady()
clients[1]?.emitState('reconnecting')
clients[1]?.emitState('connected')
clients[1]?.emitNotificationReady()
await flushAsync()
})
expect(clients[1]?.notificationSubscribeCalls()).toBe(2)
expect(clients[1]?.getMissedCalls()).toBe(1)
} finally {
finishSchedule('scheduled-held')
restore()
act(() => renderer?.unmount())
}
})
it('binds after a client connects during listener registration', async () => {
const client = makeFakeClient('connecting', true)
connectMock.mockReturnValue(client)
loadHostsMock.mockResolvedValue([HOSTS[0]])
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(RpcClientProvider, null, createElement(NotificationHostConnectionOwner))
)
await flushAsync()
})
expect(client.notificationSubscribeCalls()).toBe(1)
} finally {
restore()
act(() => renderer?.unmount())
}
})
it('reopens root notification ownership immediately after re-pair closes the old client', async () => {
const clients: FakeClient[] = []
connectMock.mockImplementation(() => {
const client = makeFakeClient()
clients.push(client)
return client
})
loadHostsMock.mockResolvedValue([HOSTS[0]])
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(
RpcClientProvider,
null,
createElement(NotificationHostConnectionOwner),
createElement(RePairProbe)
)
)
await flushAsync()
})
expect(clients[0]?.notificationSubscribeCalls()).toBe(1)
await act(async () => {
closeOwnedHost?.(HOSTS[0]!.id)
await reconnectOwnedHost?.(HOSTS[0]!.id)
await flushAsync()
})
expect(clients).toHaveLength(2)
expect(clients[0]?.closeCalls()).toBe(1)
expect(clients[1]?.notificationSubscribeCalls()).toBe(1)
} finally {
restore()
act(() => renderer?.unmount())
}
})
it('retains a live notification client during a temporary credential read', async () => {
const host = HOSTS[0]!
const client = makeFakeClient()
connectMock.mockReturnValue(client)
loadHostsMock.mockResolvedValue([host])
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(RpcClientProvider, null, createElement(NotificationHostConnectionOwner))
)
await flushAsync()
})
loadHostCatalogMock.mockResolvedValueOnce([
{ ...host, credentialStatus: 'temporarily-unavailable', profile: null }
])
await act(async () => {
hostCollectionChange.listener?.({ retiredHostIds: [] })
await flushAsync()
})
expect(client.closeCalls()).toBe(0)
expect(client.notificationSubscribeCalls()).toBe(1)
} finally {
restore()
act(() => renderer?.unmount())
}
})
it.each(['temporary credential read', 'catalog rejection'])(
'retries a %s while the app stays foregrounded',
async (failure) => {
vi.useFakeTimers()
const host = HOSTS[0]!
if (failure === 'temporary credential read') {
loadHostCatalogMock.mockResolvedValueOnce([
{ ...host, credentialStatus: 'temporarily-unavailable', profile: null }
])
} else {
loadHostCatalogMock.mockRejectedValueOnce(new Error('storage unavailable'))
}
loadHostCatalogMock.mockResolvedValueOnce([
{ ...host, credentialStatus: 'ready', profile: host }
])
const client = makeFakeClient()
connectMock.mockReturnValue(client)
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(RpcClientProvider, null, createElement(NotificationHostConnectionOwner))
)
await Promise.resolve()
})
expect(client.notificationSubscribeCalls()).toBe(0)
await act(async () => {
await vi.advanceTimersByTimeAsync(250)
})
expect(client.notificationSubscribeCalls()).toBe(1)
} finally {
restore()
act(() => renderer?.unmount())
vi.useRealTimers()
}
}
)
it('does not rerender Home for notification-only host state changes', async () => {
const clients = new Map<string, FakeClient>()
connectMock.mockImplementation((profile: HostProfile) => {
const client = makeFakeClient()
clients.set(profile.id, client)
return client
})
loadHostsMock.mockResolvedValue(HOSTS)
let homeRenders = 0
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(
RpcClientProvider,
null,
createElement(NotificationHostConnectionOwner),
createElement(HomeDataProbe, { onRender: () => (homeRenders += 1) })
)
)
await flushAsync()
})
const settledRenders = homeRenders
await act(async () => {
clients.get('older-fourth')?.emitState('reconnecting')
await Promise.resolve()
})
expect(homeRenders).toBe(settledRenders)
} finally {
restore()
act(() => renderer?.unmount())
}
})
it('does not rerender Home when notification-only clients are added at scale', async () => {
const initialHosts = HOSTS.slice(0, 3)
const expandedHosts = [
...initialHosts,
...Array.from({ length: 997 }, (_, index) => makeHost(`notification-only-${index}`, 0))
]
connectMock.mockImplementation(() => makeFakeClient())
loadHostsMock.mockResolvedValue(initialHosts)
let homeRenders = 0
let renderer: ReactTestRenderer | null = null
const restore = suppressRendererWarning()
try {
await act(async () => {
renderer = create(
createElement(
RpcClientProvider,
null,
createElement(NotificationHostConnectionOwner),
createElement(HomeDataProbe, { onRender: () => (homeRenders += 1) })
)
)
await flushAsync()
})
const settledRenders = homeRenders
loadHostsMock.mockResolvedValue(expandedHosts)
await act(async () => {
hostCollectionChange.listener?.({ retiredHostIds: [] })
await flushAsync()
})
expect(connectMock).toHaveBeenCalledTimes(1_000)
expect(homeRenders).toBe(settledRenders)
} finally {
restore()
act(() => renderer?.unmount())
}
})
})
@@ -129,6 +129,7 @@ export function createSeenNotificationGuard(): {
// at module scope is what makes the catch-up recognise a reconnect (instead of
// mistaking it for a cold open) and keeps dedup effective across the teardown.
export type HostNotificationSession = {
retired: boolean
// Highest desktop seq delivered for this host in this app process. Outranks
// the persisted value, which lags because saveLastSeenSeq is fire-and-forget.
lastDeliveredSeq: number
@@ -157,11 +158,13 @@ export type HostNotificationSession = {
}
const sessionsByHost = new Map<string, HostNotificationSession>()
const retirementByHost = new Map<string, Promise<void>>()
export function getHostNotificationSession(hostId: string): HostNotificationSession {
let session = sessionsByHost.get(hostId)
if (!session) {
session = {
retired: false,
lastDeliveredSeq: 0,
lastDeliveredEpoch: null,
catchUpQuarantineSeq: null,
@@ -241,6 +244,7 @@ export function releaseQueuedShowNotificationId(
/** Test-only: drop per-host session state so each test starts from a cold open. */
export function resetHostNotificationSessionsForTests(): void {
sessionsByHost.clear()
retirementByHost.clear()
}
/**
@@ -257,6 +261,9 @@ export function quarantineCatchUpWatermark(
hostId: string,
contiguousSeq: number
): void {
if (session.retired) {
return
}
session.catchUpQuarantineSeq =
session.catchUpQuarantineSeq == null
? contiguousSeq
@@ -272,7 +279,7 @@ export function quarantineCatchUpWatermark(
/** Lift the quarantine once a catch-up completes, persisting what it held back. */
export function resolveCatchUpQuarantine(session: HostNotificationSession, hostId: string): void {
if (session.catchUpQuarantineSeq == null) {
if (session.retired || session.catchUpQuarantineSeq == null) {
return
}
session.catchUpQuarantineSeq = null
@@ -302,7 +309,7 @@ export function adoptNotificationEpoch(
hostId: string,
epoch: string | undefined
): void {
if (!epoch || epoch === session.lastDeliveredEpoch) {
if (session.retired || !epoch || epoch === session.lastDeliveredEpoch) {
return
}
// Why reset on a FIRST observation too (lastDeliveredEpoch === null): a seq seeded
@@ -357,6 +364,9 @@ export function seedWatermarkFromStorage(session: HostNotificationSession, hostI
return
}
const seeded = loadWatermark(hostId).then(({ seq, epoch, stored }) => {
if (session.retired) {
return
}
// Why the record's existence and not `seq > 0`: adoptNotificationEpoch persists
// `{seq: 0, epoch}` when it voids a watermark, so a device that HAS delivered for
// this host reloads as seq 0. Keying on the seq would read that as a first pairing
@@ -388,6 +398,31 @@ export function forgetHostNotificationSession(hostId: string): void {
sessionsByHost.delete(hostId)
}
export function retireHostNotificationState(hostId: string): Promise<void> {
const existing = retirementByHost.get(hostId)
if (existing) {
return existing
}
const session = sessionsByHost.get(hostId)
if (session) {
session.retired = true
}
const retirement = (async () => {
await session?.deliveryTail
if (!session || sessionsByHost.get(hostId) === session) {
sessionsByHost.delete(hostId)
}
await clearWatermark(hostId)
})()
retirementByHost.set(hostId, retirement)
void retirement.finally(() => {
if (retirementByHost.get(hostId) === retirement) {
retirementByHost.delete(hostId)
}
})
return retirement
}
// Why: key for the replay dedup guard. Uses notificationId when present, but
// disambiguates by seq so a legitimate live re-delivery of the same id at a
// NEW seq (content refresh, allowed by the existing behaviour) is NOT treated
@@ -0,0 +1,133 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import * as Notifications from 'expo-notifications'
import type { RpcClient } from '../transport/rpc-client'
import { subscribeToDesktopNotifications } from './mobile-notifications'
import {
resetHostNotificationSessionsForTests,
retireHostNotificationState
} from './notification-reconnect-catchup'
import { loadPushNotificationsEnabled } from '../storage/preferences'
const asyncStorage = vi.hoisted(() => ({
getItem: vi.fn(async () => null),
setItem: vi.fn(async () => undefined),
removeItem: vi.fn(async () => undefined)
}))
vi.mock('expo-notifications', () => ({
AndroidImportance: { HIGH: 'high' },
setNotificationChannelAsync: vi.fn(),
getPermissionsAsync: vi.fn(),
requestPermissionsAsync: vi.fn(),
scheduleNotificationAsync: vi.fn(),
dismissNotificationAsync: vi.fn()
}))
vi.mock('react-native', () => ({ Platform: { OS: 'ios', Version: 18 } }))
vi.mock('@react-native-async-storage/async-storage', () => ({
default: asyncStorage
}))
vi.mock('../storage/preferences', () => ({ loadPushNotificationsEnabled: vi.fn() }))
function flushAsync(): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, 0))
}
describe('retired host live delivery', () => {
beforeEach(() => {
vi.clearAllMocks()
resetHostNotificationSessionsForTests()
})
it('drops queued delivery after the host subscription is retired', async () => {
vi.mocked(loadPushNotificationsEnabled).mockResolvedValue(true)
vi.mocked(Notifications.getPermissionsAsync).mockResolvedValue({
status: 'granted',
canAskAgain: true
} as never)
let finishFirstSchedule: (identifier: string) => void = () => {}
vi.mocked(Notifications.scheduleNotificationAsync).mockReturnValueOnce(
new Promise<string>((resolve) => {
finishFirstSchedule = resolve
})
)
let onEvent: ((data: unknown) => void) | null = null
const client = {
subscribe: vi.fn((_method, _params, callback: (data: unknown) => void) => {
onEvent = callback
return vi.fn()
}),
getState: vi.fn(() => 'connected'),
sendRequest: vi.fn()
} as unknown as RpcClient
const unsubscribe = subscribeToDesktopNotifications(client, 'host-retired')
onEvent?.({
type: 'notification',
source: 'agent-task-complete',
title: 'First',
body: 'First body',
notificationId: 'agent:first'
})
onEvent?.({
type: 'notification',
source: 'agent-task-complete',
title: 'Second',
body: 'Second body',
notificationId: 'agent:second'
})
await flushAsync()
unsubscribe()
finishFirstSchedule('scheduled-first')
await flushAsync()
expect(Notifications.scheduleNotificationAsync).toHaveBeenCalledOnce()
})
it('does not recreate a watermark when catch-up resolves after retirement', async () => {
let onEvent: ((data: unknown) => void) | null = null
let resolveCatchUp: (response: { ok: true; result: unknown }) => void = () => {}
const catchUp = new Promise<{ ok: true; result: unknown }>((resolve) => {
resolveCatchUp = resolve
})
const client = {
subscribe: vi.fn((_method, _params, callback: (data: unknown) => void) => {
onEvent = callback
return () => {
if (onEvent === callback) {
onEvent = null
}
}
}),
getState: vi.fn(() => 'connected'),
sendRequest: vi.fn((method: string) =>
method === 'notifications.getMissedSince'
? catchUp
: Promise.resolve({ ok: true, result: undefined })
)
} as unknown as RpcClient
const firstUnsubscribe = subscribeToDesktopNotifications(client, 'host-retired-catch-up')
onEvent?.({ type: 'ready', subscriptionId: 'first' })
await flushAsync()
firstUnsubscribe()
const secondUnsubscribe = subscribeToDesktopNotifications(client, 'host-retired-catch-up')
onEvent?.({ type: 'ready', subscriptionId: 'second' })
await vi.waitFor(() =>
expect(client.sendRequest).toHaveBeenCalledWith('notifications.getMissedSince', {
lastSeenSeq: 0
})
)
asyncStorage.setItem.mockClear()
secondUnsubscribe()
await retireHostNotificationState('host-retired-catch-up')
resolveCatchUp({ ok: true, result: { notifications: [], epoch: 'retired-epoch' } })
await flushAsync()
expect(asyncStorage.setItem).not.toHaveBeenCalled()
})
})
@@ -390,6 +390,36 @@ describe('useHostClient', () => {
})
describe('useAllHostClients', () => {
it('observes a client opened by an earlier sibling effect in the same commit', async () => {
connectMock.mockReturnValue(makeFakeClient('connected'))
loadHostsMock.mockResolvedValue([HOST])
let observerHasClient = false
function Acquirer(): null {
useAllHostClients([HOST.id])
return null
}
function Observer(): null {
observerHasClient = useAllHostClients([HOST.id]).length === 1
return null
}
let renderer: ReactTestRenderer | null = null
const restore = suppressReactTestRendererDeprecationWarning()
try {
await act(async () => {
renderer = create(
createElement(RpcClientProvider, null, createElement(Acquirer), createElement(Observer))
)
await Promise.resolve()
})
expect(observerHasClient).toBe(true)
} finally {
restore()
act(() => renderer?.unmount())
}
})
it('only opens the requested startup subset', async () => {
const host2 = { ...HOST, id: 'host-2', name: 'Host 2' }
connectMock.mockReturnValue(makeFakeClient('connected'))
+15 -15
View File
@@ -12,12 +12,11 @@ import {
} from 'react'
import type { RpcClient } from './rpc-client'
import { connectionLogStore } from './connection-log-buffer'
import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers'
import { HostClientOpenRegistry } from './host-client-open-registry'
import { decrementPendingAcquisition } from './host-client-acquisition-count'
import { subscribeHostClientRevival } from './host-client-revival-scheduler'
import {
clientActivePath,
listHostClients,
notifyAllHostListeners,
notifyHostStateListeners,
primeHostProfiles,
@@ -51,8 +50,8 @@ export type RpcClientContextValue = {
// Why: ms-epoch of the last 'connected' (null if never this session); UI escalates "Reconnecting…" into a re-pair prompt.
getLastConnectedAt: (hostId: string) => number | null
getActivePath: (hostId: string) => MobileConnectionPath
getClient: (hostId: string) => RpcClient | null
subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void
getAllClients: () => Array<{ hostId: string; client: RpcClient }>
subscribeAllHosts: (listener: () => void) => () => void
// Why: lets the home screen feed already-loaded HostProfiles so we don't pay loadHosts() latency twice.
primeHosts: (hosts: HostProfile[]) => void
@@ -302,14 +301,17 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
return clientActivePath(storeRef.current.get(hostId)?.client)
}, [])
const getClient = useCallback(
(hostId: string) => storeRef.current.get(hostId)?.client ?? null,
[]
)
const subscribeHostState = useCallback(
(hostId: string, listener: (state: ConnectionState) => void) =>
subscribeHostStateListener(stateListenersRef.current, hostId, listener),
[]
)
const getAllClients = useCallback(() => listHostClients(storeRef.current), [])
const subscribeAllHosts = useCallback(
(listener: () => void) => subscribeAllHostListener(allHostsListenersRef.current, listener),
[]
@@ -330,11 +332,9 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
// Why: nudge live clients when the OS signals the link may be back so sessions recover without a restart (issue #5049).
useEffect(() => {
return subscribeConnectionRevivalTriggers((reason) => {
for (const entry of storeRef.current.values()) {
entry.client.notifyForeground(reason)
}
})
return subscribeHostClientRevival(() =>
[...storeRef.current.values()].map(({ client }) => client)
)
}, [])
const value = useMemo<RpcClientContextValue>(
@@ -350,8 +350,8 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
getReconnectAttempt,
getLastConnectedAt,
getActivePath,
getClient,
subscribeHostState,
getAllClients,
subscribeAllHosts,
primeHosts
}),
@@ -367,8 +367,8 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
getReconnectAttempt,
getLastConnectedAt,
getActivePath,
getClient,
subscribeHostState,
getAllClients,
subscribeAllHosts,
primeHosts
]
@@ -416,9 +416,9 @@ export function useHostClient(hostId: string | undefined): {
}
setState(next)
// Why: async open and forceReconnect swap the client object; re-read each state change so screens never drive a stale one.
const found = ctx.getAllClients().find((entry) => entry.hostId === hostId)
if (found && found.client !== clientRef.current) {
clientRef.current = found.client
const found = ctx.getClient(hostId)
if (found && found !== clientRef.current) {
clientRef.current = found
force((n) => n + 1)
} else if (!found && clientRef.current) {
// Why: closeHost deletes the entry with no replacement; null it so screens don't drive a dead client (STA-1511).
@@ -53,4 +53,16 @@ describe('connection log buffer', () => {
store.append('host-a', entry(2))
expect(onA).toHaveBeenCalledTimes(1)
})
it('evicts a retired host and notifies its active observer', () => {
const store = createConnectionLogStore()
const listener = vi.fn()
store.append('host-a', entry(1))
store.subscribe('host-a', listener)
store.clear('host-a')
expect(store.get('host-a')).toEqual([])
expect(listener).toHaveBeenCalledOnce()
})
})
@@ -11,6 +11,7 @@ const MAX_ENTRIES_PER_HOST = 200
export type ConnectionLogStore = {
append: (hostId: string, entry: ConnectionLogEntry) => void
clear: (hostId: string) => void
get: (hostId: string) => readonly ConnectionLogEntry[]
subscribe: (hostId: string, listener: () => void) => () => void
}
@@ -46,6 +47,17 @@ export function createConnectionLogStore(
}
},
clear(hostId) {
entriesByHost.delete(hostId)
snapshotByHost.delete(hostId)
const listeners = listenersByHost.get(hostId)
if (listeners) {
for (const listener of listeners) {
listener()
}
}
},
get(hostId) {
const cached = snapshotByHost.get(hostId)
if (cached) {
@@ -0,0 +1,43 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { RpcClient } from './rpc-client'
import { HostClientRevivalScheduler } from './host-client-revival-scheduler'
vi.mock('./connection-revival-triggers', () => ({
subscribeConnectionRevivalTriggers: vi.fn(() => () => {})
}))
describe('host client revival scheduler', () => {
afterEach(() => vi.useRealTimers())
it('paces a large foreground recovery wave without dropping clients', async () => {
vi.useFakeTimers()
const notifyForeground = Array.from({ length: 1_000 }, () => vi.fn())
const clients = notifyForeground.map(
(notify) => ({ notifyForeground: notify }) as unknown as RpcClient
)
const scheduler = new HostClientRevivalScheduler()
scheduler.schedule(clients, 'app-resume')
expect(notifyForeground.filter((notify) => notify.mock.calls.length > 0)).toHaveLength(32)
await vi.runAllTimersAsync()
expect(notifyForeground.every((notify) => notify.mock.calls[0]?.[0] === 'app-resume')).toBe(
true
)
})
it('cancels batches that have not started', async () => {
vi.useFakeTimers()
const notifyForeground = Array.from({ length: 64 }, () => vi.fn())
const clients = notifyForeground.map(
(notify) => ({ notifyForeground: notify }) as unknown as RpcClient
)
const scheduler = new HostClientRevivalScheduler()
scheduler.schedule(clients, 'network-change')
scheduler.stop()
await vi.runAllTimersAsync()
expect(notifyForeground.filter((notify) => notify.mock.calls.length > 0)).toHaveLength(32)
})
})
@@ -0,0 +1,60 @@
import type { RpcClient } from './rpc-client'
import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers'
const REVIVAL_BATCH_SIZE = 32
const REVIVAL_BATCH_DELAY_MS = 25
type RevivalReason = 'app-resume' | 'network-change'
export class HostClientRevivalScheduler {
private readonly pending = new Map<RpcClient, RevivalReason>()
private timer: ReturnType<typeof setTimeout> | null = null
private stopped = false
schedule(clients: readonly RpcClient[], reason: RevivalReason): void {
if (this.stopped) {
return
}
for (const client of clients) {
this.pending.set(client, reason)
}
if (!this.timer) {
this.notifyNextBatch()
}
}
stop(): void {
this.stopped = true
this.pending.clear()
if (this.timer) {
clearTimeout(this.timer)
this.timer = null
}
}
private notifyNextBatch(): void {
this.timer = null
if (this.stopped) {
return
}
const batch = [...this.pending].slice(0, REVIVAL_BATCH_SIZE)
for (const [client, reason] of batch) {
this.pending.delete(client)
client.notifyForeground(reason)
}
if (this.pending.size > 0) {
this.timer = setTimeout(() => this.notifyNextBatch(), REVIVAL_BATCH_DELAY_MS)
}
}
}
export function subscribeHostClientRevival(listClients: () => readonly RpcClient[]): () => void {
const scheduler = new HostClientRevivalScheduler()
const unsubscribe = subscribeConnectionRevivalTriggers((reason) => {
scheduler.schedule(listClients(), reason)
})
return () => {
scheduler.stop()
unsubscribe()
}
}
@@ -0,0 +1,22 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import {
notifyHostCollectionChanged,
resetHostCollectionChangeListenersForTests,
subscribeHostCollectionChanges
} from './host-collection-changes'
describe('host collection changes', () => {
beforeEach(() => resetHostCollectionChangeListenersForTests())
it('notifies active listeners only', () => {
const listener = vi.fn()
const unsubscribe = subscribeHostCollectionChanges(listener)
notifyHostCollectionChanged({ retiredHostIds: ['retired-host'] })
unsubscribe()
notifyHostCollectionChanged({ retiredHostIds: [] })
expect(listener).toHaveBeenCalledOnce()
expect(listener).toHaveBeenCalledWith({ retiredHostIds: ['retired-host'] })
})
})
@@ -0,0 +1,20 @@
export type HostCollectionChange = { retiredHostIds: readonly string[] }
const listeners = new Set<(change: HostCollectionChange) => void>()
export function notifyHostCollectionChanged(change: HostCollectionChange): void {
for (const listener of listeners) {
listener(change)
}
}
export function subscribeHostCollectionChanges(
listener: (change: HostCollectionChange) => void
): () => void {
listeners.add(listener)
return () => listeners.delete(listener)
}
export function resetHostCollectionChangeListenersForTests(): void {
listeners.clear()
}
@@ -0,0 +1,29 @@
import { describe, expect, it } from 'vitest'
import type { StoredHostProfile } from './types'
import { recordHostLastConnected } from './host-last-connected'
const HOST: StoredHostProfile = {
id: 'host-1',
name: 'Host 1',
endpoint: 'ws://host-1',
publicKeyB64: 'key-1',
lastConnected: 1
}
describe('recordHostLastConnected', () => {
it('updates only the matching host without mutating the input', () => {
const other = { ...HOST, id: 'host-2' }
const hosts = [HOST, other]
expect(recordHostLastConnected(hosts, HOST.id, 42)).toEqual([
{ ...HOST, lastConnected: 42 },
other
])
expect(hosts).toEqual([HOST, other])
})
it('returns the same list when the host is absent', () => {
const hosts = [HOST]
expect(recordHostLastConnected(hosts, 'missing', 42)).toBe(hosts)
})
})
@@ -0,0 +1,15 @@
import type { StoredHostProfile } from './types'
export function recordHostLastConnected(
hosts: StoredHostProfile[],
hostId: string,
connectedAt: number
): StoredHostProfile[] {
const index = hosts.findIndex((host) => host.id === hostId)
if (index < 0) {
return hosts
}
const next = hosts.slice()
next[index] = { ...next[index]!, lastConnected: connectedAt }
return next
}
@@ -83,9 +83,26 @@ describe('host removal lifecycle', () => {
removeHostMock.mockResolvedValue(undefined)
await removeHostAndCloseClient('host-1', vi.fn())
// clearWatermark is fire-and-forget; let its microtask land.
await Promise.resolve()
expect(asyncStorage.removeItem).toHaveBeenCalledWith('orca:mobileNotificationsWatermark:host-1')
})
it('drains in-flight notification delivery before retiring a host for re-pair', async () => {
removeHostMock.mockResolvedValue(undefined)
const session = getHostNotificationSession('host-1')
let finishDelivery: () => void = () => {}
session.deliveryTail = new Promise<void>((resolve) => {
finishDelivery = resolve
})
const removal = removeHostAndCloseClient('host-1', vi.fn())
await Promise.resolve()
expect(getHostNotificationSession('host-1')).toBe(session)
finishDelivery()
await removal
expect(getHostNotificationSession('host-1')).not.toBe(session)
expect(asyncStorage.removeItem).toHaveBeenCalledWith('orca:mobileNotificationsWatermark:host-1')
})
})
@@ -1,7 +1,5 @@
import {
clearWatermark,
forgetHostNotificationSession
} from '../notifications/notification-reconnect-catchup'
import { retireHostNotificationState } from '../notifications/notification-reconnect-catchup'
import { connectionLogStore } from './connection-log-buffer'
import { removeHost } from './host-store'
export async function removeHostAndCloseClient(
@@ -12,9 +10,9 @@ export async function removeHostAndCloseClient(
// storage failure; closing immediately after success prevents socket leaks.
await removeHost(hostId)
closeHostClient(hostId)
connectionLogStore.clear(hostId)
// Why: the notification session outlives the socket by design (it must survive
// reconnects), so removal is the only thing that can retire it. Left behind, a
// re-pair of the same host would inherit a watermark for a counter it never saw.
forgetHostNotificationSession(hostId)
void clearWatermark(hostId)
await retireHostNotificationState(hostId)
}
@@ -0,0 +1,97 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
const asyncStorage = vi.hoisted(() => ({
getItem: vi.fn(),
setItem: vi.fn(),
removeItem: vi.fn()
}))
const secureStore = vi.hoisted(() => ({
deleteItemAsync: vi.fn(),
getItemAsync: vi.fn(),
setItemAsync: vi.fn()
}))
vi.mock('@react-native-async-storage/async-storage', () => ({ default: asyncStorage }))
vi.mock('expo-secure-store', () => ({
WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'WHEN_UNLOCKED_THIS_DEVICE_ONLY',
...secureStore
}))
vi.mock('react-native', () => ({ Platform: { OS: 'ios' } }))
vi.mock('./host-credential-cleanup', () => ({
cancelPendingHostCredentialCleanup: vi.fn(async () => undefined),
recordHostCredentialCleanupIntent: vi.fn(async () => undefined),
scheduleHostCredentialCleanup: vi.fn(async () => undefined),
retryPendingHostCredentialCleanups: vi.fn()
}))
import { loadHostCatalog, resetHostStoreForTests, saveHost } from './host-store'
import { subscribeHostCollectionChanges } from './host-collection-changes'
import { resetMobileRelayHostOverlayStoreForTests } from './mobile-relay-host-overlay-store'
const HOSTS_STORAGE_KEY = 'orca:hosts'
const HOST = {
id: 'host-1',
name: 'Host 1',
endpoint: 'ws://127.0.0.1:1',
publicKeyB64: 'key-1',
lastConnected: 0
}
describe('host store collection notifications', () => {
let storedHosts: Array<typeof HOST>
let credentialByHostId: Map<string, string>
beforeEach(() => {
vi.clearAllMocks()
resetHostStoreForTests()
resetMobileRelayHostOverlayStoreForTests()
storedHosts = [HOST]
credentialByHostId = new Map()
asyncStorage.getItem.mockImplementation(async (key: string) =>
key === HOSTS_STORAGE_KEY ? JSON.stringify(storedHosts) : null
)
asyncStorage.setItem.mockImplementation(async (key: string, raw: string) => {
if (key === HOSTS_STORAGE_KEY) {
storedHosts = JSON.parse(raw)
}
})
secureStore.getItemAsync.mockImplementation(async (key: string) => {
const hostId = [...credentialByHostId.keys()].find((id) => key.endsWith(id))
return hostId ? (credentialByHostId.get(hostId) ?? null) : null
})
secureStore.setItemAsync.mockImplementation(async (key: string, token: string) => {
const hostId = key.split('.').at(-1)
if (hostId) {
credentialByHostId.set(hostId, token)
}
})
})
it('announces new membership and restored credentials', async () => {
await loadHostCatalog()
const listener = vi.fn()
subscribeHostCollectionChanges(listener)
await saveHost({ ...HOST, deviceToken: 'restored-token' })
await saveHost({
...HOST,
id: 'host-2',
name: 'Host 2',
publicKeyB64: 'key-2',
deviceToken: 'new-token'
})
expect(listener).toHaveBeenNthCalledWith(1, { retiredHostIds: [] })
expect(listener).toHaveBeenNthCalledWith(2, { retiredHostIds: [] })
})
it('announces duplicate identities retired by an authoritative save', async () => {
storedHosts = [HOST, { ...HOST, id: 'host-duplicate', name: 'Duplicate' }]
const listener = vi.fn()
subscribeHostCollectionChanges(listener)
await saveHost({ ...HOST, deviceToken: 'replacement-token' })
expect(listener).toHaveBeenCalledWith({ retiredHostIds: ['host-duplicate'] })
})
})
+9 -9
View File
@@ -30,6 +30,8 @@ import {
toStoredHostProfile,
writeStoredHostProfiles
} from './host-metadata-store'
import * as hostCollectionChanges from './host-collection-changes'
import { recordHostLastConnected } from './host-last-connected'
async function commitDeviceToken(hostId: string, token: string): Promise<void> {
markHostCredentialWrite(hostId)
@@ -169,6 +171,7 @@ export const saveExistingHostRelayUpgrade = (host: HostProfile): Promise<void> =
async function persistHost(host: HostProfile, requireExisting: boolean): Promise<void> {
const validated = HostProfileSchema.parse(host)
const stored = toStoredHostProfile(validated)
const credentialWasCached = tokenCache.has(stored.id)
const duplicateHostIds = new Set<string>()
let updatedExistingHost = false
let cleanupIntentRecordedBeforeMetadata = false
@@ -252,6 +255,9 @@ async function persistHost(host: HostProfile, requireExisting: boolean): Promise
// Metadata is already deduplicated; orphan-token recovery is best-effort.
}
}
if (!updatedExistingHost || duplicateHostIds.size > 0 || !credentialWasCached) {
hostCollectionChanges.notifyHostCollectionChanged({ retiredHostIds: [...duplicateHostIds] })
}
}
export async function removeHost(hostId: string): Promise<void> {
@@ -285,6 +291,7 @@ export async function removeHost(hostId: string): Promise<void> {
} catch {
// Metadata is already committed; orphan-token recovery is best-effort.
}
hostCollectionChanges.notifyHostCollectionChanged({ retiredHostIds: [hostId] })
}
export async function retryPendingHostCredentialCleanup(): Promise<{
@@ -319,15 +326,7 @@ export async function updateHostNameAndEndpoint(
export async function updateLastConnected(hostId: string): Promise<void> {
try {
await mutateStoredHosts((hosts) => {
const index = hosts.findIndex((h) => h.id === hostId)
if (index < 0) {
return hosts
}
const next = hosts.slice()
next[index] = { ...next[index]!, lastConnected: Date.now() }
return next
})
await mutateStoredHosts((hosts) => recordHostLastConnected(hosts, hostId, Date.now()))
} catch {
// Why: best-effort timestamp fired with void; swallow so unreadable storage doesn't reject.
}
@@ -340,4 +339,5 @@ export function resetHostStoreForTests(): void {
resetHostCredentialWriteRevisionsForTests()
hostListLoads.dropSharedHostListLoad()
resetPairingKeychainForTests()
hostCollectionChanges.resetHostCollectionChangeListenersForTests()
}
@@ -196,10 +196,8 @@ function activeHostIds(): string[] {
if (!context) {
throw new Error('client context was not captured')
}
return context
.getAllClients()
.map(({ hostId }) => hostId)
.sort()
const openedHostIds = connectMock.mock.calls.map(([profile]) => profile.id as string)
return [...new Set(openedHostIds)].filter((hostId) => context?.getClient(hostId)).sort()
}
beforeEach(() => {
+24 -9
View File
@@ -7,24 +7,28 @@ import { useRpcClientContext } from './client-context'
type UseAllHostClientsOptions = {
autoConnectHostIds?: readonly string[]
closeUnusedOnRelease?: boolean
observeConnectionState?: boolean
}
export function useAllHostClients(hostIds: string[], options?: UseAllHostClientsOptions) {
const ctx = useRpcClientContext()
const autoConnectHostIds = options?.autoConnectHostIds ?? hostIds
const closeUnusedOnRelease = options?.closeUnusedOnRelease ?? false
const observeConnectionState = options?.observeConnectionState ?? true
const key = useMemo(
() =>
[
[...hostIds].sort().join(','),
[...autoConnectHostIds].sort().join(','),
closeUnusedOnRelease ? 'close' : 'keep'
closeUnusedOnRelease ? 'close' : 'keep',
observeConnectionState ? 'state' : 'client'
].join('|'),
[autoConnectHostIds, closeUnusedOnRelease, hostIds]
[autoConnectHostIds, closeUnusedOnRelease, hostIds, observeConnectionState]
)
const [tick, setTick] = useState(0)
const acquiredHostIdsRef = useRef<Set<string>>(new Set())
const hostUnsubscribesRef = useRef<Map<string, () => void>>(new Map())
const observedClientsRef = useRef<Map<string, RpcClient | null>>(new Map())
const closeUnusedRef = useRef(closeUnusedOnRelease)
useEffect(() => {
@@ -32,9 +36,7 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients
}, [closeUnusedOnRelease])
useEffect(() => {
const unsubscribeAllHosts = ctx.subscribeAllHosts(() => setTick((value) => value + 1))
return () => {
unsubscribeAllHosts()
const trackedHostIds = [...hostUnsubscribesRef.current.keys()]
const acquiredHostIds = new Set(acquiredHostIdsRef.current)
for (const unsubscribe of hostUnsubscribesRef.current.values()) {
@@ -56,6 +58,7 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients
}
}
acquiredHostIdsRef.current.clear()
observedClientsRef.current.clear()
}
}, [ctx])
@@ -63,23 +66,38 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients
const trackedHostIds = new Set(hostIds)
const nextAcquiredHostIds = new Set(autoConnectHostIds.filter((id) => trackedHostIds.has(id)))
const removedTrackedHostIds: string[] = []
let foundClientOpenedBeforeSubscription = false
for (const [id, unsubscribe] of hostUnsubscribesRef.current) {
if (!trackedHostIds.has(id)) {
unsubscribe()
hostUnsubscribesRef.current.delete(id)
observedClientsRef.current.delete(id)
removedTrackedHostIds.push(id)
}
}
for (const id of trackedHostIds) {
if (!hostUnsubscribesRef.current.has(id)) {
const existingClient = ctx.getClient(id)
observedClientsRef.current.set(id, existingClient)
foundClientOpenedBeforeSubscription ||= existingClient !== null
hostUnsubscribesRef.current.set(
id,
ctx.subscribeHostState(id, () => setTick((value) => value + 1))
ctx.subscribeHostState(id, () => {
const nextClient = ctx.getClient(id)
if (observeConnectionState || observedClientsRef.current.get(id) !== nextClient) {
observedClientsRef.current.set(id, nextClient)
setTick((value) => value + 1)
}
})
)
}
}
if (foundClientOpenedBeforeSubscription) {
setTick((value) => value + 1)
}
for (const id of acquiredHostIdsRef.current) {
if (!nextAcquiredHostIds.has(id)) {
if (closeUnusedOnRelease) {
@@ -108,16 +126,13 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients
}, [ctx, key])
return useMemo(() => {
const clientsByHostId = new Map(
ctx.getAllClients().map((entry) => [entry.hostId, entry.client])
)
return hostIds.flatMap<{
hostId: string
client: RpcClient
state: ConnectionState
path: MobileConnectionPath
}>((hostId) => {
const client = clientsByHostId.get(hostId)
const client = ctx.getClient(hostId)
return client
? [{ hostId, client, state: ctx.getState(hostId), path: ctx.getActivePath(hostId) }]
: []