mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 00:03:20 +00:00
Merge remote-tracking branch 'origin/main' into brennanb2025/bug-macos-app-freezes-on-the-phone-session-ended
This commit is contained in:
+11
-12
@@ -28,6 +28,7 @@ import { loadHosts } from '../src/transport/host-store'
|
||||
import { removeHostAndCloseClient } from '../src/transport/host-removal-lifecycle'
|
||||
import { pickResumeWorktree } from '../src/worktree/resume-worktree'
|
||||
import type { RpcClient } from '../src/transport/rpc-client'
|
||||
import { sendSingleFlightRequest } from '../src/transport/request-single-flight'
|
||||
import {
|
||||
useAllHostClients,
|
||||
useCloseHost,
|
||||
@@ -140,11 +141,11 @@ function clientKey(client: RpcClient): number {
|
||||
|
||||
function fetchStats(
|
||||
client: RpcClient,
|
||||
hostId: string,
|
||||
setStats: (s: StatsSummary) => void,
|
||||
disposed: () => boolean
|
||||
) {
|
||||
client
|
||||
.sendRequest('stats.summary')
|
||||
sendSingleFlightRequest(client, hostId, 'stats.summary')
|
||||
.then((response) => {
|
||||
if (disposed()) {
|
||||
return
|
||||
@@ -182,9 +183,8 @@ function fetchWorktreeInfo(
|
||||
})
|
||||
}
|
||||
|
||||
client
|
||||
// Why: worktree.ps defaults to 200 and silently truncates; request all so counts are accurate.
|
||||
.sendRequest('worktree.ps', { limit: 10000 })
|
||||
// Why: worktree.ps defaults to 200 and silently truncates; request all so counts are accurate.
|
||||
sendSingleFlightRequest(client, hostId, 'worktree.ps', { limit: 10000 })
|
||||
.then((response) => {
|
||||
if (disposed()) {
|
||||
return
|
||||
@@ -225,8 +225,7 @@ function fetchAccountsSnapshot(
|
||||
) => void,
|
||||
disposed: () => boolean
|
||||
) {
|
||||
client
|
||||
.sendRequest('accounts.list')
|
||||
sendSingleFlightRequest(client, hostId, 'accounts.list')
|
||||
.then((response) => {
|
||||
if (disposed()) {
|
||||
return
|
||||
@@ -248,9 +247,9 @@ function fetchTaskProviders(
|
||||
disposed: () => boolean
|
||||
) {
|
||||
Promise.all([
|
||||
client.sendRequest('settings.get'),
|
||||
client.sendRequest('preflight.check'),
|
||||
client.sendRequest('linear.status')
|
||||
sendSingleFlightRequest(client, hostId, 'settings.get'),
|
||||
sendSingleFlightRequest(client, hostId, 'preflight.check'),
|
||||
sendSingleFlightRequest(client, hostId, 'linear.status')
|
||||
])
|
||||
.then(([settingsResponse, preflightResponse, linearResponse]) => {
|
||||
if (disposed()) {
|
||||
@@ -405,7 +404,7 @@ export default function HomeScreen() {
|
||||
})
|
||||
for (const entry of allClientsRef.current) {
|
||||
if (entry.client.getState() === 'connected') {
|
||||
fetchStats(entry.client, setStats, () => stale)
|
||||
fetchStats(entry.client, entry.hostId, setStats, () => stale)
|
||||
fetchWorktreeInfo(entry.client, entry.hostId, setWorktreeInfo, () => stale)
|
||||
fetchAccountsSnapshot(entry.client, entry.hostId, setAccountsByHost, () => stale)
|
||||
fetchTaskProviders(entry.client, entry.hostId, setTaskProvidersByHost, () => stale)
|
||||
@@ -512,7 +511,7 @@ export default function HomeScreen() {
|
||||
}
|
||||
if (!statsFetched) {
|
||||
statsFetched = true
|
||||
fetchStats(entry.client, setStats, () => false)
|
||||
fetchStats(entry.client, entry.hostId, setStats, () => false)
|
||||
fetchWorktreeInfo(entry.client, entry.hostId, setWorktreeInfo, () => false)
|
||||
fetchTaskProviders(entry.client, entry.hostId, setTaskProvidersByHost, () => false)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
|
||||
import { useCallback, useEffect, useMemo, useState } from 'react'
|
||||
import {
|
||||
ActivityIndicator,
|
||||
Pressable,
|
||||
@@ -9,7 +9,7 @@ import {
|
||||
View
|
||||
} from 'react-native'
|
||||
import { useSafeAreaInsets } from 'react-native-safe-area-context'
|
||||
import { useRouter } from 'expo-router'
|
||||
import { useFocusEffect, useRouter } from 'expo-router'
|
||||
import { ChevronLeft, ChevronRight } from 'lucide-react-native'
|
||||
import { colors, radii, spacing, typography } from '../src/theme/mobile-theme'
|
||||
import { loadHosts } from '../src/transport/host-store'
|
||||
@@ -18,6 +18,7 @@ import { useAllHostClients } from '../src/transport/client-context'
|
||||
import type { RpcClient } from '../src/transport/rpc-client'
|
||||
import { BottomDrawer } from '../src/components/BottomDrawer'
|
||||
import { VoiceModelList } from '../src/components/VoiceModelList'
|
||||
import { useDictationSetupPoller } from '../src/dictation/use-dictation-setup-poller'
|
||||
import {
|
||||
deleteDictationModel,
|
||||
downloadDictationModel,
|
||||
@@ -58,44 +59,45 @@ export default function VoiceSettingsScreen(): React.JSX.Element {
|
||||
const [error, setError] = useState<string | null>(null)
|
||||
const [busyAction, setBusyAction] = useState<ModelBusyAction | null>(null)
|
||||
const [modelDrawerOpen, setModelDrawerOpen] = useState(false)
|
||||
const pollRef = useRef<ReturnType<typeof setInterval> | null>(null)
|
||||
const [routeFocused, setRouteFocused] = useState(false)
|
||||
|
||||
const refresh = useCallback(async () => {
|
||||
useFocusEffect(
|
||||
useCallback(() => {
|
||||
setRouteFocused(true)
|
||||
return () => setRouteFocused(false)
|
||||
}, [])
|
||||
)
|
||||
|
||||
const refresh = useCallback(async (): Promise<boolean | undefined> => {
|
||||
if (!client) {
|
||||
return
|
||||
return false
|
||||
}
|
||||
try {
|
||||
setSetup(await fetchDictationSetup(client))
|
||||
const next = await fetchDictationSetup(client)
|
||||
setSetup(next)
|
||||
setError(null)
|
||||
return next.models.some(isModelInFlight)
|
||||
} catch (err) {
|
||||
setError(err instanceof Error ? err.message : 'Failed to load voice settings')
|
||||
return undefined
|
||||
} finally {
|
||||
setLoading(false)
|
||||
}
|
||||
}, [client])
|
||||
|
||||
// Initial load once a connected client is available.
|
||||
useEffect(() => {
|
||||
if (!client) {
|
||||
return
|
||||
}
|
||||
setLoading(true)
|
||||
setError(null)
|
||||
void refresh().finally(() => setLoading(false))
|
||||
}, [client, refresh])
|
||||
const polling = setup?.models.some(isModelInFlight) ?? false
|
||||
const refreshSetup = useDictationSetupPoller({
|
||||
visible: routeFocused && client !== null,
|
||||
polling,
|
||||
refresh,
|
||||
intervalMs: POLL_INTERVAL_MS
|
||||
})
|
||||
|
||||
// Poll only while a model is downloading/extracting; stop otherwise.
|
||||
useEffect(() => {
|
||||
const inFlight = setup?.models.some(isModelInFlight) ?? false
|
||||
if (inFlight && client) {
|
||||
pollRef.current = setInterval(() => void refresh(), POLL_INTERVAL_MS)
|
||||
return () => {
|
||||
if (pollRef.current) {
|
||||
clearInterval(pollRef.current)
|
||||
pollRef.current = null
|
||||
}
|
||||
}
|
||||
if (routeFocused && client && setup === null) {
|
||||
setLoading(true)
|
||||
}
|
||||
return undefined
|
||||
}, [setup, client, refresh])
|
||||
}, [routeFocused, client, setup])
|
||||
|
||||
const handleToggleEnabled = useCallback(
|
||||
async (enabled: boolean) => {
|
||||
@@ -109,10 +111,10 @@ export default function VoiceSettingsScreen(): React.JSX.Element {
|
||||
setSetup(await setDictationConfig(client, { enabled }))
|
||||
} catch (err) {
|
||||
setError(err instanceof Error ? err.message : 'Could not update')
|
||||
void refresh()
|
||||
void refreshSetup()
|
||||
}
|
||||
},
|
||||
[client, refresh]
|
||||
[client, refreshSetup]
|
||||
)
|
||||
|
||||
const handleSelectMode = useCallback(
|
||||
@@ -126,10 +128,10 @@ export default function VoiceSettingsScreen(): React.JSX.Element {
|
||||
setSetup(await setDictationConfig(client, { dictationMode }))
|
||||
} catch (err) {
|
||||
setError(err instanceof Error ? err.message : 'Could not update')
|
||||
void refresh()
|
||||
void refreshSetup()
|
||||
}
|
||||
},
|
||||
[client, refresh]
|
||||
[client, refreshSetup]
|
||||
)
|
||||
|
||||
const handleUseModel = useCallback(
|
||||
@@ -160,14 +162,14 @@ export default function VoiceSettingsScreen(): React.JSX.Element {
|
||||
setError(null)
|
||||
try {
|
||||
await downloadDictationModel(client, model.id)
|
||||
await refresh()
|
||||
await refreshSetup()
|
||||
} catch (err) {
|
||||
setError(err instanceof Error ? err.message : 'Download failed')
|
||||
} finally {
|
||||
setBusyAction(null)
|
||||
}
|
||||
},
|
||||
[client, refresh]
|
||||
[client, refreshSetup]
|
||||
)
|
||||
|
||||
const handleDelete = useCallback(
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
import { useCallback, useEffect, useRef, useState } from 'react'
|
||||
import { useCallback, useEffect, useState } from 'react'
|
||||
import { ActivityIndicator, Pressable, StyleSheet, Switch, Text, View } from 'react-native'
|
||||
import { Check, Download } from 'lucide-react-native'
|
||||
import { BottomDrawer } from './BottomDrawer'
|
||||
import { colors, radii, spacing, typography } from '../theme/mobile-theme'
|
||||
import type { RpcClient } from '../transport/rpc-client'
|
||||
import { triggerError, triggerSuccess } from '../platform/haptics'
|
||||
import { useDictationSetupPoller } from '../dictation/use-dictation-setup-poller'
|
||||
import {
|
||||
downloadDictationModel,
|
||||
fetchDictationSetup,
|
||||
@@ -37,40 +38,34 @@ export function MobileDictationSetupSheet({ visible, client, onClose, onReady }:
|
||||
const [setup, setSetup] = useState<MobileSpeechSetup | null>(null)
|
||||
const [error, setError] = useState<string | null>(null)
|
||||
const [busy, setBusy] = useState<string | null>(null)
|
||||
const pollRef = useRef<ReturnType<typeof setInterval> | null>(null)
|
||||
|
||||
const refresh = useCallback(async () => {
|
||||
const refresh = useCallback(async (): Promise<boolean | undefined> => {
|
||||
if (!client) {
|
||||
return
|
||||
return false
|
||||
}
|
||||
try {
|
||||
setSetup(await fetchDictationSetup(client))
|
||||
const next = await fetchDictationSetup(client)
|
||||
setSetup(next)
|
||||
setError(null)
|
||||
return next.models.some(isModelInFlight)
|
||||
} catch (err) {
|
||||
setError(err instanceof Error ? err.message : 'Failed to load')
|
||||
return undefined
|
||||
}
|
||||
}, [client])
|
||||
|
||||
const polling = setup?.models.some(isModelInFlight) ?? false
|
||||
const refreshSetup = useDictationSetupPoller({
|
||||
visible: visible && client !== null,
|
||||
polling,
|
||||
refresh,
|
||||
intervalMs: POLL_INTERVAL_MS
|
||||
})
|
||||
|
||||
useEffect(() => {
|
||||
if (visible) {
|
||||
setError(null)
|
||||
void refresh()
|
||||
}
|
||||
}, [visible, refresh])
|
||||
|
||||
// Poll only while something is downloading/extracting; stop otherwise.
|
||||
useEffect(() => {
|
||||
const inFlight = setup?.models.some(isModelInFlight) ?? false
|
||||
if (visible && inFlight && client) {
|
||||
pollRef.current = setInterval(() => void refresh(), POLL_INTERVAL_MS)
|
||||
return () => {
|
||||
if (pollRef.current) {
|
||||
clearInterval(pollRef.current)
|
||||
pollRef.current = null
|
||||
}
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
}, [visible, setup, client, refresh])
|
||||
}, [visible])
|
||||
|
||||
const handleDownload = useCallback(
|
||||
async (model: MobileSpeechModel) => {
|
||||
@@ -81,7 +76,7 @@ export function MobileDictationSetupSheet({ visible, client, onClose, onReady }:
|
||||
setError(null)
|
||||
try {
|
||||
await downloadDictationModel(client, model.id)
|
||||
await refresh()
|
||||
await refreshSetup()
|
||||
} catch (err) {
|
||||
triggerError()
|
||||
setError(err instanceof Error ? err.message : 'Download failed')
|
||||
@@ -89,7 +84,7 @@ export function MobileDictationSetupSheet({ visible, client, onClose, onReady }:
|
||||
setBusy(null)
|
||||
}
|
||||
},
|
||||
[client, refresh]
|
||||
[client, refreshSetup]
|
||||
)
|
||||
|
||||
const handleUseModel = useCallback(
|
||||
|
||||
@@ -0,0 +1,197 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { DictationSetupPollController } from './dictation-setup-poll-controller'
|
||||
|
||||
const POLL_INTERVAL_MS = 1500
|
||||
|
||||
async function flushPromises(): Promise<void> {
|
||||
// Why: the refresh mock wraps its result in `.finally()` and the resume path chains
|
||||
// runRefresh → requestRefresh → runRefresh, so the follow-up refresh is several microtask
|
||||
// hops deep — drain generously rather than a fixed two ticks.
|
||||
for (let i = 0; i < 8; i += 1) {
|
||||
await Promise.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
function deferred<T>(): {
|
||||
promise: Promise<T>
|
||||
resolve: (value: T) => void
|
||||
} {
|
||||
let resolve!: (value: T) => void
|
||||
return {
|
||||
promise: new Promise<T>((next) => {
|
||||
resolve = next
|
||||
}),
|
||||
resolve
|
||||
}
|
||||
}
|
||||
|
||||
describe('DictationSetupPollController', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('does not refresh while hidden, unfocused, or backgrounded', async () => {
|
||||
const refresh = vi.fn().mockResolvedValue(true)
|
||||
const poller = new DictationSetupPollController(refresh, POLL_INTERVAL_MS)
|
||||
poller.setPolling(true)
|
||||
|
||||
poller.setForeground(true)
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 2)
|
||||
expect(refresh).not.toHaveBeenCalled()
|
||||
|
||||
poller.setVisible(true)
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
await flushPromises()
|
||||
poller.setVisible(false)
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 2)
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
|
||||
poller.setVisible(true)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
await flushPromises()
|
||||
poller.setForeground(false)
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 2)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
|
||||
poller.dispose()
|
||||
})
|
||||
|
||||
it('keeps slow refreshes single-flight and waits a full delay after each response', async () => {
|
||||
const requests = [deferred<boolean>(), deferred<boolean>(), deferred<boolean>()]
|
||||
let active = 0
|
||||
let maxActive = 0
|
||||
const refresh = vi.fn(() => {
|
||||
const request = requests[refresh.mock.calls.length - 1]
|
||||
active += 1
|
||||
maxActive = Math.max(maxActive, active)
|
||||
return request.promise.finally(() => {
|
||||
active -= 1
|
||||
})
|
||||
})
|
||||
const poller = new DictationSetupPollController(refresh, POLL_INTERVAL_MS)
|
||||
poller.setPolling(true)
|
||||
poller.setVisible(true)
|
||||
poller.setForeground(true)
|
||||
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 4)
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
|
||||
requests[0].resolve(true)
|
||||
await flushPromises()
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 4)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
|
||||
requests[1].resolve(true)
|
||||
await flushPromises()
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS - 1)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
expect(refresh).toHaveBeenCalledTimes(3)
|
||||
expect(maxActive).toBe(1)
|
||||
|
||||
requests[2].resolve(false)
|
||||
await flushPromises()
|
||||
poller.dispose()
|
||||
})
|
||||
|
||||
it('coalesces an immediate resume refresh behind a slow request', async () => {
|
||||
const requests = [deferred<boolean>(), deferred<boolean>()]
|
||||
let active = 0
|
||||
let maxActive = 0
|
||||
const refresh = vi.fn(() => {
|
||||
const request = requests[refresh.mock.calls.length - 1]
|
||||
active += 1
|
||||
maxActive = Math.max(maxActive, active)
|
||||
return request.promise.finally(() => {
|
||||
active -= 1
|
||||
})
|
||||
})
|
||||
const poller = new DictationSetupPollController(refresh, POLL_INTERVAL_MS)
|
||||
poller.setPolling(true)
|
||||
poller.setVisible(true)
|
||||
poller.setForeground(true)
|
||||
|
||||
poller.setForeground(false)
|
||||
poller.setForeground(true)
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
|
||||
requests[0].resolve(true)
|
||||
await flushPromises()
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
expect(maxActive).toBe(1)
|
||||
|
||||
requests[1].resolve(false)
|
||||
await flushPromises()
|
||||
poller.dispose()
|
||||
})
|
||||
|
||||
it('refreshes immediately when visibility or foreground eligibility resumes', async () => {
|
||||
const refresh = vi.fn().mockResolvedValue(true)
|
||||
const poller = new DictationSetupPollController(refresh, POLL_INTERVAL_MS)
|
||||
poller.setPolling(true)
|
||||
poller.setVisible(true)
|
||||
poller.setForeground(true)
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
await flushPromises()
|
||||
|
||||
poller.setForeground(false)
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 2)
|
||||
poller.setForeground(true)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
await flushPromises()
|
||||
|
||||
poller.setVisible(false)
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 2)
|
||||
poller.setVisible(true)
|
||||
expect(refresh).toHaveBeenCalledTimes(3)
|
||||
|
||||
poller.dispose()
|
||||
})
|
||||
|
||||
it('stops after setup leaves the download or extraction lifecycle', async () => {
|
||||
const refresh = vi.fn().mockResolvedValueOnce(true).mockResolvedValueOnce(false)
|
||||
const poller = new DictationSetupPollController(refresh, POLL_INTERVAL_MS)
|
||||
poller.setPolling(true)
|
||||
poller.setVisible(true)
|
||||
poller.setForeground(true)
|
||||
await flushPromises()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
await flushPromises()
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 4)
|
||||
expect(refresh).toHaveBeenCalledTimes(2)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
poller.dispose()
|
||||
})
|
||||
|
||||
it('does not resurrect polling when an in-flight refresh resolves true after setPolling(false)', async () => {
|
||||
const request = deferred<boolean>()
|
||||
const refresh = vi.fn(() => request.promise)
|
||||
const poller = new DictationSetupPollController(refresh, POLL_INTERVAL_MS)
|
||||
poller.setPolling(true)
|
||||
poller.setVisible(true)
|
||||
poller.setForeground(true)
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
|
||||
// Explicit stop lands while the read is still on the wire.
|
||||
poller.setPolling(false)
|
||||
// The stale read then resolves "keep polling" — the fence must drop it, not restart the poll.
|
||||
request.resolve(true)
|
||||
await flushPromises()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 4)
|
||||
expect(refresh).toHaveBeenCalledOnce()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
poller.dispose()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,153 @@
|
||||
type PollState = {
|
||||
visible: boolean
|
||||
foreground: boolean
|
||||
polling: boolean
|
||||
}
|
||||
|
||||
type RefreshResult = boolean | undefined
|
||||
|
||||
export class DictationSetupPollController {
|
||||
private state: PollState = { visible: false, foreground: false, polling: false }
|
||||
private timer: ReturnType<typeof setTimeout> | null = null
|
||||
private inFlight = false
|
||||
private immediateRefreshPending = false
|
||||
private refreshWaiters: Array<() => void> = []
|
||||
private disposed = false
|
||||
// Why: an explicit setPolling is a newer lifecycle intent than a read that was already on the wire.
|
||||
// Bumped on every setPolling so an in-flight refresh resolving after an explicit stop/start can be
|
||||
// fenced out instead of clobbering that intent (e.g. a late `true` resurrecting a just-stopped poll).
|
||||
private pollingRevision = 0
|
||||
|
||||
constructor(
|
||||
private readonly refresh: () => Promise<RefreshResult>,
|
||||
private readonly intervalMs: number
|
||||
) {}
|
||||
|
||||
setVisible(visible: boolean): void {
|
||||
this.update({ visible })
|
||||
}
|
||||
|
||||
setForeground(foreground: boolean): void {
|
||||
this.update({ foreground })
|
||||
}
|
||||
|
||||
setPolling(polling: boolean): void {
|
||||
this.pollingRevision += 1
|
||||
this.update({ polling })
|
||||
}
|
||||
|
||||
refreshNow(): Promise<void> {
|
||||
if (this.disposed || !this.isEligible()) {
|
||||
return Promise.resolve()
|
||||
}
|
||||
return new Promise((resolve) => {
|
||||
this.refreshWaiters.push(resolve)
|
||||
this.requestRefresh(true)
|
||||
})
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.disposed = true
|
||||
this.immediateRefreshPending = false
|
||||
this.clearTimer()
|
||||
this.resolveRefreshWaiters()
|
||||
}
|
||||
|
||||
private update(next: Partial<PollState>): void {
|
||||
if (this.disposed) {
|
||||
return
|
||||
}
|
||||
const wasEligible = this.isEligible()
|
||||
const wasPolling = this.state.polling
|
||||
this.state = { ...this.state, ...next }
|
||||
|
||||
if (!this.isEligible()) {
|
||||
this.immediateRefreshPending = false
|
||||
this.clearTimer()
|
||||
return
|
||||
}
|
||||
if (!wasEligible) {
|
||||
this.requestRefresh(true)
|
||||
return
|
||||
}
|
||||
if (!this.state.polling) {
|
||||
this.clearTimer()
|
||||
return
|
||||
}
|
||||
if (!wasPolling) {
|
||||
this.scheduleRefresh()
|
||||
}
|
||||
}
|
||||
|
||||
private isEligible(): boolean {
|
||||
return this.state.visible && this.state.foreground
|
||||
}
|
||||
|
||||
private requestRefresh(immediate: boolean): void {
|
||||
if (this.inFlight) {
|
||||
this.immediateRefreshPending ||= immediate
|
||||
return
|
||||
}
|
||||
this.clearTimer()
|
||||
this.inFlight = true
|
||||
void this.runRefresh()
|
||||
}
|
||||
|
||||
private async runRefresh(): Promise<void> {
|
||||
// Snapshot the lifecycle intent this read is answering; an explicit setPolling during the read makes
|
||||
// its result stale.
|
||||
const revisionAtStart = this.pollingRevision
|
||||
let shouldContinue: RefreshResult
|
||||
try {
|
||||
shouldContinue = await this.refresh()
|
||||
} catch {
|
||||
// A transient read failure preserves the current lifecycle for a later retry.
|
||||
shouldContinue = undefined
|
||||
} finally {
|
||||
this.inFlight = false
|
||||
}
|
||||
|
||||
// Fence: only let the read drive polling if no explicit setPolling superseded it mid-flight, so a
|
||||
// late `true` can't resurrect a poll the caller just stopped (nor a late `false` cancel a restart).
|
||||
if (shouldContinue !== undefined && this.pollingRevision === revisionAtStart) {
|
||||
this.state.polling = shouldContinue
|
||||
}
|
||||
if (this.disposed || !this.isEligible()) {
|
||||
this.resolveRefreshWaiters()
|
||||
return
|
||||
}
|
||||
if (this.immediateRefreshPending) {
|
||||
this.immediateRefreshPending = false
|
||||
this.requestRefresh(true)
|
||||
return
|
||||
}
|
||||
this.resolveRefreshWaiters()
|
||||
if (this.state.polling) {
|
||||
this.scheduleRefresh()
|
||||
}
|
||||
}
|
||||
|
||||
private scheduleRefresh(): void {
|
||||
if (this.timer !== null || this.inFlight || !this.isEligible() || !this.state.polling) {
|
||||
return
|
||||
}
|
||||
this.timer = setTimeout(() => {
|
||||
this.timer = null
|
||||
this.requestRefresh(false)
|
||||
}, this.intervalMs)
|
||||
}
|
||||
|
||||
private clearTimer(): void {
|
||||
if (this.timer !== null) {
|
||||
clearTimeout(this.timer)
|
||||
this.timer = null
|
||||
}
|
||||
}
|
||||
|
||||
private resolveRefreshWaiters(): void {
|
||||
const waiters = this.refreshWaiters.splice(0)
|
||||
for (const resolve of waiters) {
|
||||
resolve()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
import { useCallback, useEffect, useMemo, useRef } from 'react'
|
||||
import { AppState } from 'react-native'
|
||||
import { DictationSetupPollController } from './dictation-setup-poll-controller'
|
||||
|
||||
type PollerOptions = {
|
||||
visible: boolean
|
||||
polling: boolean
|
||||
refresh: () => Promise<boolean | undefined>
|
||||
intervalMs: number
|
||||
}
|
||||
|
||||
export function useDictationSetupPoller({
|
||||
visible,
|
||||
polling,
|
||||
refresh,
|
||||
intervalMs
|
||||
}: PollerOptions): () => Promise<void> {
|
||||
const refreshRef = useRef(refresh)
|
||||
refreshRef.current = refresh
|
||||
const poller = useMemo(
|
||||
() => new DictationSetupPollController(() => refreshRef.current(), intervalMs),
|
||||
[intervalMs]
|
||||
)
|
||||
|
||||
useEffect(() => () => poller.dispose(), [poller])
|
||||
|
||||
useEffect(() => {
|
||||
void poller.refreshNow()
|
||||
}, [poller, refresh])
|
||||
|
||||
useEffect(() => {
|
||||
poller.setPolling(polling)
|
||||
}, [poller, polling])
|
||||
|
||||
useEffect(() => {
|
||||
poller.setVisible(visible)
|
||||
if (!visible) {
|
||||
poller.setForeground(false)
|
||||
return undefined
|
||||
}
|
||||
|
||||
poller.setForeground(AppState.currentState === 'active')
|
||||
const subscription = AppState.addEventListener('change', (state) => {
|
||||
poller.setForeground(state === 'active')
|
||||
})
|
||||
return () => {
|
||||
subscription.remove()
|
||||
poller.setVisible(false)
|
||||
poller.setForeground(false)
|
||||
}
|
||||
}, [poller, visible])
|
||||
|
||||
return useCallback(() => poller.refreshNow(), [poller])
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { RpcClient } from './rpc-client'
|
||||
import type { RpcResponse } from './types'
|
||||
import { sendSingleFlightRequest } from './request-single-flight'
|
||||
|
||||
function makeResponse(id: string): RpcResponse {
|
||||
return { id, ok: true, result: {}, _meta: { runtimeId: 'runtime-1' } }
|
||||
}
|
||||
|
||||
const response = makeResponse('request-1')
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void
|
||||
let reject!: (reason?: unknown) => void
|
||||
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
|
||||
resolve = resolvePromise
|
||||
reject = rejectPromise
|
||||
})
|
||||
return { promise, resolve, reject }
|
||||
}
|
||||
|
||||
function rpcClient(sendRequest: RpcClient['sendRequest']): RpcClient {
|
||||
return { sendRequest } as RpcClient
|
||||
}
|
||||
|
||||
describe('sendSingleFlightRequest', () => {
|
||||
it('coalesces triggers that arrive during an in-flight read into one trailing follow-up', async () => {
|
||||
const leading = deferred<RpcResponse>()
|
||||
const trailing = deferred<RpcResponse>()
|
||||
const leadingResponse = makeResponse('leading')
|
||||
const trailingResponse = makeResponse('trailing')
|
||||
const sendRequest = vi
|
||||
.fn<() => Promise<RpcResponse>>()
|
||||
.mockReturnValueOnce(leading.promise)
|
||||
.mockReturnValueOnce(trailing.promise)
|
||||
const client = rpcClient(sendRequest)
|
||||
|
||||
const first = sendSingleFlightRequest(client, 'host-1', 'worktree.ps', { limit: 10000 })
|
||||
// Two more triggers arrive while the read is on the wire: no duplicate now, and they share ONE
|
||||
// trailing follow-up (not the older in-flight response).
|
||||
const second = sendSingleFlightRequest(client, 'host-1', 'worktree.ps', { limit: 10000 })
|
||||
const third = sendSingleFlightRequest(client, 'host-1', 'worktree.ps', { limit: 10000 })
|
||||
|
||||
expect(second).toBe(third)
|
||||
expect(second).not.toBe(first)
|
||||
expect(sendRequest).toHaveBeenCalledTimes(1)
|
||||
|
||||
leading.resolve(leadingResponse)
|
||||
expect(await first).toBe(leadingResponse)
|
||||
|
||||
// The leading read settled → the coalesced follow-up fires exactly once.
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
trailing.resolve(trailingResponse)
|
||||
expect(await second).toBe(trailingResponse)
|
||||
expect(await third).toBe(trailingResponse)
|
||||
})
|
||||
|
||||
it('starts a fresh request once nothing is in flight', async () => {
|
||||
const leading = deferred<RpcResponse>()
|
||||
const sendRequest = vi
|
||||
.fn<() => Promise<RpcResponse>>()
|
||||
.mockReturnValueOnce(leading.promise)
|
||||
.mockResolvedValueOnce(response)
|
||||
const client = rpcClient(sendRequest)
|
||||
|
||||
const first = sendSingleFlightRequest(client, 'host-1', 'worktree.ps', { limit: 10000 })
|
||||
leading.resolve(response)
|
||||
await first
|
||||
|
||||
const next = sendSingleFlightRequest(client, 'host-1', 'worktree.ps', { limit: 10000 })
|
||||
expect(next).not.toBe(first)
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
await next
|
||||
})
|
||||
|
||||
it('rejects the leading caller on failure but still runs a queued follow-up', async () => {
|
||||
const leading = deferred<RpcResponse>()
|
||||
const failure = new Error('request failed')
|
||||
const sendRequest = vi
|
||||
.fn<() => Promise<RpcResponse>>()
|
||||
.mockReturnValueOnce(leading.promise)
|
||||
.mockResolvedValueOnce(response)
|
||||
const client = rpcClient(sendRequest)
|
||||
|
||||
const first = sendSingleFlightRequest(client, 'host-1', 'accounts.list')
|
||||
const second = sendSingleFlightRequest(client, 'host-1', 'accounts.list')
|
||||
|
||||
leading.reject(failure)
|
||||
await expect(first).rejects.toBe(failure)
|
||||
// A trigger that arrived mid-flight is not poisoned by the leading failure: its follow-up runs.
|
||||
await expect(second).resolves.toBe(response)
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('clears a failed leading request so the next call retries', async () => {
|
||||
const failure = new Error('request failed')
|
||||
const sendRequest = vi
|
||||
.fn<() => Promise<RpcResponse>>()
|
||||
.mockRejectedValueOnce(failure)
|
||||
.mockResolvedValueOnce(response)
|
||||
const client = rpcClient(sendRequest)
|
||||
|
||||
await expect(sendSingleFlightRequest(client, 'host-1', 'accounts.list')).rejects.toBe(failure)
|
||||
await expect(sendSingleFlightRequest(client, 'host-1', 'accounts.list')).resolves.toBe(response)
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('does not share requests across clients, hosts, or request kinds', async () => {
|
||||
const pending = deferred<RpcResponse>()
|
||||
const firstSend = vi.fn(() => pending.promise)
|
||||
const secondSend = vi.fn(() => pending.promise)
|
||||
const firstClient = rpcClient(firstSend)
|
||||
const secondClient = rpcClient(secondSend)
|
||||
|
||||
const requests = [
|
||||
sendSingleFlightRequest(firstClient, 'host-1', 'settings.get'),
|
||||
sendSingleFlightRequest(firstClient, 'host-2', 'settings.get'),
|
||||
sendSingleFlightRequest(firstClient, 'host-1', 'preflight.check'),
|
||||
sendSingleFlightRequest(secondClient, 'host-1', 'settings.get')
|
||||
]
|
||||
|
||||
expect(firstSend).toHaveBeenCalledTimes(3)
|
||||
expect(secondSend).toHaveBeenCalledTimes(1)
|
||||
pending.resolve(response)
|
||||
await Promise.all(requests)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,110 @@
|
||||
import type { RpcClient } from './rpc-client'
|
||||
import type { RpcResponse } from './types'
|
||||
|
||||
type Deferred = {
|
||||
promise: Promise<RpcResponse>
|
||||
resolve: (value: RpcResponse) => void
|
||||
reject: (reason?: unknown) => void
|
||||
}
|
||||
|
||||
type SingleFlightEntry = {
|
||||
// The request currently on the wire for this (client, host, kind).
|
||||
current: Promise<RpcResponse>
|
||||
// At most one trailing follow-up: every trigger that arrives while `current` is in flight coalesces
|
||||
// here so a refresh requested mid-read still re-reads the latest state (latest params win) instead of
|
||||
// being silently answered by the older in-flight response.
|
||||
followUp: { deferred: Deferred; params: unknown } | null
|
||||
}
|
||||
|
||||
const inFlightRequests = new WeakMap<RpcClient, Map<string, Map<string, SingleFlightEntry>>>()
|
||||
|
||||
function makeDeferred(): Deferred {
|
||||
let resolve!: (value: RpcResponse) => void
|
||||
let reject!: (reason?: unknown) => void
|
||||
const promise = new Promise<RpcResponse>((resolvePromise, rejectPromise) => {
|
||||
resolve = resolvePromise
|
||||
reject = rejectPromise
|
||||
})
|
||||
return { promise, resolve, reject }
|
||||
}
|
||||
|
||||
export function sendSingleFlightRequest(
|
||||
client: RpcClient,
|
||||
hostId: string,
|
||||
requestKind: string,
|
||||
params?: unknown
|
||||
): Promise<RpcResponse> {
|
||||
let requestsByHost = inFlightRequests.get(client)
|
||||
if (!requestsByHost) {
|
||||
requestsByHost = new Map()
|
||||
inFlightRequests.set(client, requestsByHost)
|
||||
}
|
||||
let requestsByKind = requestsByHost.get(hostId)
|
||||
if (!requestsByKind) {
|
||||
requestsByKind = new Map()
|
||||
requestsByHost.set(hostId, requestsByKind)
|
||||
}
|
||||
|
||||
const send = (): Promise<RpcResponse> => {
|
||||
try {
|
||||
return client.sendRequest(requestKind, params)
|
||||
} catch (error) {
|
||||
return Promise.reject(error)
|
||||
}
|
||||
}
|
||||
|
||||
const existing = requestsByKind.get(requestKind)
|
||||
if (existing) {
|
||||
// A read is already on the wire: don't fire a duplicate now, but don't drop this trigger either.
|
||||
// Record (or refresh) a single trailing follow-up that runs once the current read settles.
|
||||
if (existing.followUp) {
|
||||
existing.followUp.params = params
|
||||
} else {
|
||||
existing.followUp = { deferred: makeDeferred(), params }
|
||||
}
|
||||
return existing.followUp.deferred.promise
|
||||
}
|
||||
|
||||
const entry: SingleFlightEntry = { current: send(), followUp: null }
|
||||
requestsByKind.set(requestKind, entry)
|
||||
|
||||
const cleanup = (): void => {
|
||||
if (requestsByKind.get(requestKind) !== entry) {
|
||||
return
|
||||
}
|
||||
requestsByKind.delete(requestKind)
|
||||
if (requestsByKind.size === 0) {
|
||||
requestsByHost.delete(hostId)
|
||||
}
|
||||
if (requestsByHost.size === 0) {
|
||||
inFlightRequests.delete(client)
|
||||
}
|
||||
}
|
||||
|
||||
// Chain each settled request into either its queued follow-up (delivering the fresh result to every
|
||||
// caller that awaited it) or entry teardown. Recurses so triggers arriving during a follow-up queue
|
||||
// the next one.
|
||||
const onSettled = (): void => {
|
||||
const followUp = entry.followUp
|
||||
if (!followUp) {
|
||||
cleanup()
|
||||
return
|
||||
}
|
||||
entry.followUp = null
|
||||
let next: Promise<RpcResponse>
|
||||
try {
|
||||
next = client.sendRequest(requestKind, followUp.params)
|
||||
} catch (error) {
|
||||
next = Promise.reject(error)
|
||||
}
|
||||
entry.current = next
|
||||
next.then(
|
||||
(response) => followUp.deferred.resolve(response),
|
||||
(error) => followUp.deferred.reject(error)
|
||||
)
|
||||
void next.then(onSettled, onSettled)
|
||||
}
|
||||
|
||||
void entry.current.then(onSettled, onSettled)
|
||||
return entry.current
|
||||
}
|
||||
@@ -7,20 +7,46 @@ afterEach(() => {
|
||||
})
|
||||
|
||||
describe('RemoteRuntimeServerHeartbeat', () => {
|
||||
it('still reaps a client that misses a probe while another remains alive', async () => {
|
||||
vi.useFakeTimers()
|
||||
let now = 1_000
|
||||
const responsiveSocket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket
|
||||
const deadSocket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket
|
||||
const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now)
|
||||
heartbeat.noteAlive(responsiveSocket)
|
||||
heartbeat.noteAlive(deadSocket)
|
||||
// start() probes immediately: both are pinged now (probe #1) and cleared to await a pong.
|
||||
heartbeat.start(() => [responsiveSocket, deadSocket])
|
||||
// Only the responsive socket pongs the immediate probe.
|
||||
heartbeat.noteAlive(responsiveSocket)
|
||||
|
||||
now += 100
|
||||
await vi.advanceTimersByTimeAsync(100)
|
||||
|
||||
expect(responsiveSocket.ping).toHaveBeenCalledTimes(2)
|
||||
expect(responsiveSocket.terminate).not.toHaveBeenCalled()
|
||||
expect(deadSocket.ping).toHaveBeenCalledTimes(1)
|
||||
expect(deadSocket.terminate).toHaveBeenCalledTimes(1)
|
||||
heartbeat.stop()
|
||||
})
|
||||
|
||||
it('grants clients a fresh probe after the server event loop resumes', async () => {
|
||||
vi.useFakeTimers()
|
||||
let now = 1_000
|
||||
const socket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket
|
||||
const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now)
|
||||
heartbeat.noteAlive(socket)
|
||||
// start() probes immediately (ping #1); the socket pongs it.
|
||||
heartbeat.start(() => [socket])
|
||||
heartbeat.noteAlive(socket)
|
||||
|
||||
now += 100
|
||||
await vi.advanceTimersByTimeAsync(100)
|
||||
await vi.advanceTimersByTimeAsync(100) // ping #2, socket pongs
|
||||
heartbeat.noteAlive(socket)
|
||||
now += 3_600_000
|
||||
await vi.advanceTimersByTimeAsync(100)
|
||||
await vi.advanceTimersByTimeAsync(100) // resumed-from-pause: re-grants a probe (ping #3), no reap
|
||||
|
||||
expect(socket.ping).toHaveBeenCalledTimes(2)
|
||||
expect(socket.ping).toHaveBeenCalledTimes(3)
|
||||
expect(socket.terminate).not.toHaveBeenCalled()
|
||||
|
||||
now += 100
|
||||
|
||||
@@ -22,6 +22,11 @@ export class RemoteRuntimeServerHeartbeat {
|
||||
this.lastTickAt = this.now()
|
||||
this.timer = setInterval(() => this.sweep(getClients()), this.intervalMs)
|
||||
this.timer.unref?.()
|
||||
// Why: the interval's first tick is a full intervalMs (~15s) out, so arming on the first accepted
|
||||
// connection would leave that socket unprobed for the whole window. Sweep once now so the first
|
||||
// liveness ping goes out immediately; seeded-alive sockets are pinged (not reaped) and have until
|
||||
// the next tick to pong. WS pong is answered at the protocol level, so a live socket always survives.
|
||||
this.sweep(getClients())
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { mkdtempSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
@@ -14,6 +15,30 @@ function makeTls() {
|
||||
return loadOrCreateTlsCertificate(userDataPath)
|
||||
}
|
||||
|
||||
function heartbeatLifecycle(transport: WebSocketTransport) {
|
||||
return transport as unknown as {
|
||||
heartbeat: {
|
||||
timer: ReturnType<typeof setInterval> | null
|
||||
alive: WeakSet<WebSocket>
|
||||
}
|
||||
heartbeatConnections: Set<WebSocket>
|
||||
wss: { clients: Set<WebSocket> }
|
||||
handleConnection(ws: WebSocket): void
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForHeartbeatLifecycle(
|
||||
transport: WebSocketTransport,
|
||||
connectionCount: number,
|
||||
armed: boolean
|
||||
): Promise<void> {
|
||||
await vi.waitFor(() => {
|
||||
const lifecycle = heartbeatLifecycle(transport)
|
||||
expect(lifecycle.heartbeatConnections.size).toBe(connectionCount)
|
||||
expect(lifecycle.heartbeat.timer !== null).toBe(armed)
|
||||
})
|
||||
}
|
||||
|
||||
describe('WebSocketTransport', () => {
|
||||
const transports: WebSocketTransport[] = []
|
||||
|
||||
@@ -70,6 +95,68 @@ describe('WebSocketTransport', () => {
|
||||
await transport.stop()
|
||||
})
|
||||
|
||||
it('arms heartbeat only while accepted connections exist', async () => {
|
||||
const { transport } = await createTransport()
|
||||
await transport.start()
|
||||
|
||||
const lifecycle = heartbeatLifecycle(transport)
|
||||
expect(lifecycle.heartbeat.timer).toBeNull()
|
||||
expect(lifecycle.heartbeatConnections.size).toBe(0)
|
||||
|
||||
const firstClient = await connectWs(transport)
|
||||
await waitForHeartbeatLifecycle(transport, 1, true)
|
||||
const firstServerSocket = Array.from(lifecycle.wss.clients)[0]
|
||||
expect(firstServerSocket).toBeDefined()
|
||||
// Note: arming probes immediately, so `alive` membership is racy here (the client's protocol-level
|
||||
// pong re-adds the socket right after the arm sweep clears it). Assert the arm/disarm lifecycle only.
|
||||
const firstTimer = lifecycle.heartbeat.timer
|
||||
|
||||
const secondClient = await connectWs(transport)
|
||||
await waitForHeartbeatLifecycle(transport, 2, true)
|
||||
expect(lifecycle.heartbeat.timer).toBe(firstTimer)
|
||||
|
||||
firstClient.close()
|
||||
await waitForHeartbeatLifecycle(transport, 1, true)
|
||||
expect(lifecycle.heartbeat.timer).toBe(firstTimer)
|
||||
|
||||
secondClient.close()
|
||||
await waitForHeartbeatLifecycle(transport, 0, false)
|
||||
|
||||
const thirdClient = await connectWs(transport)
|
||||
await waitForHeartbeatLifecycle(transport, 1, true)
|
||||
expect(lifecycle.heartbeat.timer).not.toBe(firstTimer)
|
||||
|
||||
thirdClient.close()
|
||||
await waitForHeartbeatLifecycle(transport, 0, false)
|
||||
})
|
||||
|
||||
it('finalizes heartbeat membership once when error and close race', () => {
|
||||
const transport = new WebSocketTransport({ host: '127.0.0.1', port: 0 })
|
||||
transports.push(transport)
|
||||
const lifecycle = heartbeatLifecycle(transport)
|
||||
const socket = Object.assign(new EventEmitter(), {
|
||||
OPEN: WebSocket.OPEN,
|
||||
readyState: WebSocket.OPEN,
|
||||
close: vi.fn(),
|
||||
ping: vi.fn(),
|
||||
terminate: vi.fn()
|
||||
}) as unknown as WebSocket
|
||||
const closeHandler = vi.fn()
|
||||
transport.onConnectionClose(closeHandler)
|
||||
|
||||
lifecycle.handleConnection(socket)
|
||||
expect(lifecycle.heartbeatConnections.size).toBe(1)
|
||||
expect(lifecycle.heartbeat.timer).not.toBeNull()
|
||||
|
||||
socket.emit('error', new Error('connection reset'))
|
||||
socket.emit('close')
|
||||
|
||||
expect(closeHandler).toHaveBeenCalledTimes(1)
|
||||
expect(socket.close).toHaveBeenCalledTimes(1)
|
||||
expect(lifecycle.heartbeatConnections.size).toBe(0)
|
||||
expect(lifecycle.heartbeat.timer).toBeNull()
|
||||
})
|
||||
|
||||
it('handles request/response round-trip', async () => {
|
||||
const { transport } = await createTransport((msg, reply) => {
|
||||
const request = JSON.parse(msg)
|
||||
|
||||
@@ -61,6 +61,7 @@ export class WebSocketTransport implements RpcTransport {
|
||||
| null = null
|
||||
// Why: maps each socket to its authenticated clientId so close can report which device disconnected.
|
||||
private wsClientIds = new Map<WebSocket, string>()
|
||||
private heartbeatConnections = new Set<WebSocket>()
|
||||
private preAuthTimers = new WeakMap<WebSocket, ReturnType<typeof setTimeout>>()
|
||||
|
||||
constructor({
|
||||
@@ -199,7 +200,6 @@ export class WebSocketTransport implements RpcTransport {
|
||||
|
||||
this.httpServer = httpServer
|
||||
this.wss = wss
|
||||
this.heartbeat.start(() => this.wss?.clients ?? [])
|
||||
}
|
||||
|
||||
// Why: force-terminate soon after the 1013 close since a half-open phone may never ack and would hold the descriptor past the WS cap; the 'error' listener absorbs a reset while closing.
|
||||
@@ -217,6 +217,7 @@ export class WebSocketTransport implements RpcTransport {
|
||||
this.wss = null
|
||||
this.httpServer = null
|
||||
this.heartbeat.stop()
|
||||
this.heartbeatConnections.clear()
|
||||
|
||||
if (wss) {
|
||||
for (const client of wss.clients) {
|
||||
@@ -280,6 +281,10 @@ export class WebSocketTransport implements RpcTransport {
|
||||
ws.off('close', finalizeConnection)
|
||||
ws.off('error', onError)
|
||||
this.clearPreAuthTimer(ws)
|
||||
this.heartbeatConnections.delete(ws)
|
||||
if (this.heartbeatConnections.size === 0) {
|
||||
this.heartbeat.stop()
|
||||
}
|
||||
const clientId = this.wsClientIds.get(ws) ?? null
|
||||
this.wsClientIds.delete(ws)
|
||||
const hasOtherConnections =
|
||||
@@ -287,6 +292,13 @@ export class WebSocketTransport implements RpcTransport {
|
||||
this.connectionCloseHandler?.(clientId, ws, hasOtherConnections)
|
||||
}
|
||||
|
||||
// Why: seed before arming so a fresh first socket survives its initial sweep.
|
||||
this.heartbeatConnections.add(ws)
|
||||
this.heartbeat.noteAlive(ws)
|
||||
if (this.heartbeatConnections.size === 1) {
|
||||
this.heartbeat.start(() => this.wss?.clients ?? [])
|
||||
}
|
||||
|
||||
const preAuthTimer = setTimeout(() => {
|
||||
if (!this.wsClientIds.has(ws)) {
|
||||
// Why: a silent auto-ponging client would otherwise hold a finite mobile slot forever without starting the E2EE handshake.
|
||||
@@ -298,9 +310,6 @@ export class WebSocketTransport implements RpcTransport {
|
||||
}
|
||||
this.preAuthTimers.set(ws, preAuthTimer)
|
||||
|
||||
// Why: seed alive so the first heartbeat tick doesn't reap a fresh socket before its first pong.
|
||||
this.heartbeat.noteAlive(ws)
|
||||
|
||||
ws.on('pong', onPong)
|
||||
ws.on('message', onMessage)
|
||||
|
||||
|
||||
@@ -2,11 +2,30 @@ import { describe, expect, it, vi, afterEach, beforeEach } from 'vitest'
|
||||
import { WebRuntimeClient } from './web-runtime-client'
|
||||
|
||||
// Why: a half-open browser WebSocket stays readyState===OPEN with no
|
||||
// onclose/onerror, so the client must actively detect server silence and force
|
||||
// a reconnect. These tests drive the private heartbeat with controllable time +
|
||||
// visibility (the real timers/visibility are faked away).
|
||||
// onclose/onerror, so the client must actively detect server silence without
|
||||
// keeping its timer armed while the window is hidden.
|
||||
|
||||
const fakeSockets: FakeWebSocket[] = []
|
||||
let visibilityState: DocumentVisibilityState = 'visible'
|
||||
let nextIntervalId = 1
|
||||
const documentListeners = new Map<string, () => void>()
|
||||
const intervalCallbacks = new Map<number, () => void>()
|
||||
const setIntervalMock = vi.fn((callback: () => void, _intervalMs: number): number => {
|
||||
const intervalId = nextIntervalId++
|
||||
intervalCallbacks.set(intervalId, callback)
|
||||
return intervalId
|
||||
})
|
||||
const clearIntervalMock = vi.fn((intervalId: number): void => {
|
||||
intervalCallbacks.delete(intervalId)
|
||||
})
|
||||
const addDocumentEventListenerMock = vi.fn((event: string, listener: () => void): void => {
|
||||
documentListeners.set(event, listener)
|
||||
})
|
||||
const removeDocumentEventListenerMock = vi.fn((event: string, listener: () => void): void => {
|
||||
if (documentListeners.get(event) === listener) {
|
||||
documentListeners.delete(event)
|
||||
}
|
||||
})
|
||||
|
||||
class FakeWebSocket {
|
||||
static readonly CONNECTING = 0
|
||||
@@ -34,6 +53,7 @@ type HeartbeatInternals = {
|
||||
lastInboundFrameAt: number
|
||||
lastHeartbeatTickAt: number
|
||||
heartbeatProbeSentAt: number | null
|
||||
startHeartbeat: () => void
|
||||
runHeartbeatTick: () => void
|
||||
now: () => number
|
||||
isDocumentVisible: () => boolean
|
||||
@@ -47,7 +67,6 @@ function makeConnectedClient(): {
|
||||
setVisible: (visible: boolean) => void
|
||||
} {
|
||||
let nowMs = 1_000
|
||||
let visible = true
|
||||
const client = new WebRuntimeClient({
|
||||
v: 2,
|
||||
endpoint: 'ws://127.0.0.1:6768',
|
||||
@@ -57,7 +76,7 @@ function makeConnectedClient(): {
|
||||
const internals = client as unknown as HeartbeatInternals
|
||||
// Override the protected time/visibility seams deterministically.
|
||||
internals.now = () => nowMs
|
||||
internals.isDocumentVisible = () => visible
|
||||
internals.isDocumentVisible = () => visibilityState === 'visible'
|
||||
const socket = fakeSockets[0]!
|
||||
socket.readyState = FakeWebSocket.OPEN
|
||||
internals.ws = socket
|
||||
@@ -73,8 +92,8 @@ function makeConnectedClient(): {
|
||||
setNow: (ms) => {
|
||||
nowMs = ms
|
||||
},
|
||||
setVisible: (next) => {
|
||||
visible = next
|
||||
setVisible: (visible) => {
|
||||
visibilityState = visible ? 'visible' : 'hidden'
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -82,14 +101,31 @@ function makeConnectedClient(): {
|
||||
describe('WebRuntimeClient liveness heartbeat', () => {
|
||||
beforeEach(() => {
|
||||
fakeSockets.length = 0
|
||||
visibilityState = 'visible'
|
||||
nextIntervalId = 1
|
||||
documentListeners.clear()
|
||||
intervalCallbacks.clear()
|
||||
setIntervalMock.mockClear()
|
||||
clearIntervalMock.mockClear()
|
||||
addDocumentEventListenerMock.mockClear()
|
||||
removeDocumentEventListenerMock.mockClear()
|
||||
vi.stubGlobal('window', {
|
||||
setTimeout,
|
||||
clearTimeout,
|
||||
setInterval,
|
||||
clearInterval,
|
||||
setInterval: setIntervalMock,
|
||||
clearInterval: clearIntervalMock,
|
||||
atob: (value: string) => Buffer.from(value, 'base64').toString('binary'),
|
||||
btoa: (value: string) => Buffer.from(value, 'binary').toString('base64')
|
||||
})
|
||||
vi.stubGlobal('document', {
|
||||
get visibilityState() {
|
||||
return visibilityState
|
||||
},
|
||||
addEventListener: addDocumentEventListenerMock,
|
||||
removeEventListener: removeDocumentEventListenerMock
|
||||
})
|
||||
vi.stubGlobal('setInterval', setIntervalMock)
|
||||
vi.stubGlobal('clearInterval', clearIntervalMock)
|
||||
vi.stubGlobal('WebSocket', FakeWebSocket)
|
||||
})
|
||||
|
||||
@@ -97,6 +133,15 @@ describe('WebRuntimeClient liveness heartbeat', () => {
|
||||
vi.unstubAllGlobals()
|
||||
})
|
||||
|
||||
function dispatchVisibilityChange(): void {
|
||||
documentListeners.get('visibilitychange')?.()
|
||||
}
|
||||
|
||||
function runActiveHeartbeatTick(): void {
|
||||
expect(intervalCallbacks.size).toBe(1)
|
||||
intervalCallbacks.values().next().value?.()
|
||||
}
|
||||
|
||||
// Advance time AND record a tick boundary so the suspended-loop detector
|
||||
// (sinceLastTick) sees a normal cadence, mirroring back-to-back real ticks.
|
||||
function advanceOneTick(internals: HeartbeatInternals, setNow: (ms: number) => void): void {
|
||||
@@ -104,6 +149,107 @@ describe('WebRuntimeClient liveness heartbeat', () => {
|
||||
setNow(next)
|
||||
}
|
||||
|
||||
it('disarms the heartbeat interval while hidden', () => {
|
||||
const { client, internals, setVisible } = makeConnectedClient()
|
||||
|
||||
internals.startHeartbeat()
|
||||
expect(setIntervalMock).toHaveBeenCalledWith(expect.any(Function), 10_000)
|
||||
expect(intervalCallbacks.size).toBe(1)
|
||||
|
||||
setVisible(false)
|
||||
dispatchVisibilityChange()
|
||||
|
||||
expect(clearIntervalMock).toHaveBeenCalledTimes(1)
|
||||
expect(intervalCallbacks.size).toBe(0)
|
||||
client.close()
|
||||
})
|
||||
|
||||
it('re-arms once and resets the tick clock but preserves the liveness baseline when visible again', () => {
|
||||
const { client, internals, setNow, setVisible } = makeConnectedClient()
|
||||
internals.startHeartbeat()
|
||||
internals.heartbeatProbeSentAt = 1_000
|
||||
|
||||
setVisible(false)
|
||||
dispatchVisibilityChange()
|
||||
setNow(601_000)
|
||||
setVisible(true)
|
||||
dispatchVisibilityChange()
|
||||
dispatchVisibilityChange()
|
||||
|
||||
expect(setIntervalMock).toHaveBeenCalledTimes(2)
|
||||
expect(intervalCallbacks.size).toBe(1)
|
||||
// lastInboundFrameAt is NOT rebaselined on a visible re-arm: it stays at the fresh-connect baseline
|
||||
// (1_000) so a socket that went silent while hidden is still detectable on the next tick.
|
||||
expect(internals.lastInboundFrameAt).toBe(1_000)
|
||||
// The tick clock resets so the parked hidden gap isn't misread as a suspended loop; the probe clears.
|
||||
expect(internals.lastHeartbeatTickAt).toBe(601_000)
|
||||
expect(internals.heartbeatProbeSentAt).toBeNull()
|
||||
client.close()
|
||||
})
|
||||
|
||||
it('probes and then closes a connection that went silent while hidden, once visible again', () => {
|
||||
const { client, internals, socket, setNow, setVisible } = makeConnectedClient()
|
||||
internals.startHeartbeat()
|
||||
|
||||
// Hide, let a long silent gap elapse, then become visible again. The connection produced no inbound
|
||||
// frames the whole time, so the preserved baseline (1_000) must drive prompt liveness detection.
|
||||
setVisible(false)
|
||||
dispatchVisibilityChange()
|
||||
setNow(601_000)
|
||||
setVisible(true)
|
||||
dispatchVisibilityChange()
|
||||
|
||||
// First visible tick: idle far exceeds the threshold, so send a liveness probe (not an immediate close).
|
||||
setNow(611_000)
|
||||
runActiveHeartbeatTick()
|
||||
expect(socket.close).not.toHaveBeenCalled()
|
||||
expect(socket.send).toHaveBeenCalledTimes(1)
|
||||
expect(internals.heartbeatProbeSentAt).toBe(611_000)
|
||||
|
||||
// Normal-cadence ticks: still within grace → no close yet.
|
||||
setNow(621_000)
|
||||
runActiveHeartbeatTick()
|
||||
expect(socket.close).not.toHaveBeenCalled()
|
||||
|
||||
// Probe now unanswered past grace (20s) → force the reconnect.
|
||||
setNow(631_000)
|
||||
runActiveHeartbeatTick()
|
||||
expect(socket.close).toHaveBeenCalledTimes(1)
|
||||
client.close()
|
||||
})
|
||||
|
||||
it('keeps the visible heartbeat cadence unchanged', () => {
|
||||
const { client, internals, socket, setNow } = makeConnectedClient()
|
||||
internals.startHeartbeat()
|
||||
|
||||
setNow(11_000)
|
||||
runActiveHeartbeatTick()
|
||||
setNow(21_000)
|
||||
runActiveHeartbeatTick()
|
||||
expect(socket.send).not.toHaveBeenCalled()
|
||||
|
||||
setNow(31_000)
|
||||
runActiveHeartbeatTick()
|
||||
expect(socket.send).toHaveBeenCalledTimes(1)
|
||||
expect(internals.heartbeatProbeSentAt).toBe(31_000)
|
||||
client.close()
|
||||
})
|
||||
|
||||
it('cleans up the heartbeat interval and visibility listener on close', () => {
|
||||
const { client, internals } = makeConnectedClient()
|
||||
internals.startHeartbeat()
|
||||
const visibilityListener = documentListeners.get('visibilitychange')
|
||||
|
||||
client.close()
|
||||
|
||||
expect(intervalCallbacks.size).toBe(0)
|
||||
expect(removeDocumentEventListenerMock).toHaveBeenCalledWith(
|
||||
'visibilitychange',
|
||||
visibilityListener
|
||||
)
|
||||
expect(documentListeners.has('visibilitychange')).toBe(false)
|
||||
})
|
||||
|
||||
it('does nothing while the socket keeps receiving frames', () => {
|
||||
const { internals, socket } = makeConnectedClient()
|
||||
// Just under the idle threshold → no probe, no close.
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
import type { RuntimeRpcResponse, RuntimeRpcSuccess } from '../../../shared/runtime-rpc-envelope'
|
||||
import { isKeepaliveFrame } from '../../../shared/runtime-rpc-envelope'
|
||||
import type { WebPairingOffer } from './web-pairing'
|
||||
import { installWindowVisibilityInterval } from '../lib/window-visibility-interval'
|
||||
import { withRemoteRuntimeTailscaleHint } from '../../../shared/remote-runtime-tailscale-hint'
|
||||
import {
|
||||
decrypt,
|
||||
@@ -76,7 +77,7 @@ export class WebRuntimeClient {
|
||||
private connectTimer: number | null = null
|
||||
private handshakeTimer: number | null = null
|
||||
private reconnectTimer: number | null = null
|
||||
private heartbeatTimer: number | null = null
|
||||
private heartbeatCleanup: (() => void) | null = null
|
||||
private lastInboundFrameAt = 0
|
||||
// Timestamp of an outstanding liveness probe (null = none); dead-close fires only on an unanswered sent probe.
|
||||
private heartbeatProbeSentAt: number | null = null
|
||||
@@ -765,18 +766,32 @@ export class WebRuntimeClient {
|
||||
|
||||
private startHeartbeat(): void {
|
||||
this.clearHeartbeatTimer()
|
||||
// Why: this runs at 'connected', right after the handshake's inbound frames — a genuine liveness
|
||||
// baseline. Only the fresh-connect moment resets lastInboundFrameAt; the visible re-arm below must not.
|
||||
const now = this.now()
|
||||
this.lastInboundFrameAt = now
|
||||
this.lastHeartbeatTickAt = now
|
||||
this.heartbeatProbeSentAt = null
|
||||
this.heartbeatTimer = window.setInterval(() => this.runHeartbeatTick(), HEARTBEAT_INTERVAL_MS)
|
||||
this.heartbeatCleanup = installWindowVisibilityInterval({
|
||||
run: () => this.runHeartbeatTick(),
|
||||
runOnVisible: () => this.rebaselineHeartbeat(),
|
||||
intervalMs: HEARTBEAT_INTERVAL_MS
|
||||
})
|
||||
}
|
||||
|
||||
private rebaselineHeartbeat(): void {
|
||||
// Why: the interval is merely parked while hidden, so on becoming visible reset the tick clock (don't
|
||||
// let the parked gap trip the suspended-loop rebaseline) and drop a probe that was in flight when we
|
||||
// hid. But PRESERVE lastInboundFrameAt: if the socket went silent while hidden, keeping the real
|
||||
// last-heard time lets the next tick detect the staleness and probe promptly, instead of masking a
|
||||
// dead connection for another full idle window (#9883 review).
|
||||
this.lastHeartbeatTickAt = this.now()
|
||||
this.heartbeatProbeSentAt = null
|
||||
}
|
||||
|
||||
private clearHeartbeatTimer(): void {
|
||||
if (this.heartbeatTimer) {
|
||||
window.clearInterval(this.heartbeatTimer)
|
||||
this.heartbeatTimer = null
|
||||
}
|
||||
this.heartbeatCleanup?.()
|
||||
this.heartbeatCleanup = null
|
||||
this.heartbeatProbeSentAt = null
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user