diff --git a/App.tsx b/App.tsx index 24c78bf4..402eb3d4 100644 --- a/App.tsx +++ b/App.tsx @@ -17,6 +17,7 @@ import { startCalendarNotificationSync, startCalendarReminderTapHandling, } from './src/lib/calendar-notifications'; +import { startLivenessMonitor } from './src/lib/connection-liveness'; import { setPendingCalendarOpen, type CalendarReminderTarget, @@ -627,6 +628,23 @@ export default function App() { void useOutboxStore.getState().flush(); }; + // Foreground liveness for the per-account connection dot: the open + // stream's pings, or a `Core/echo` every 30 s while there is none (see + // startLivenessMonitor). NetInfo cannot tell an unreachable server. + const liveness = startLivenessMonitor({ + ping: () => jmapClient.ping(), + streamHealthy: () => handle?.healthy ?? false, + isActive: () => appActive, + isOnline: () => useNetworkStore.getState().online, + onConnected: (connected) => { + if (!mounted || !activeAccountId) return; + const account = useAccountStore.getState().getAccountById(activeAccountId); + if (account && account.isConnected !== connected) { + useAccountStore.getState().updateAccount(activeAccountId, { isConnected: connected }); + } + }, + }); + void start(); const subscription = AppState.addEventListener('change', (state) => { @@ -638,19 +656,12 @@ export default function App() { handle = null; return; } - // Coming back: verify the session is still alive before trusting the - // stream, then reconnect and catch up on what was missed. - void (async () => { - const alive = await jmapClient.ping().catch(() => false); - if (!mounted) return; - if (!alive && useNetworkStore.getState().online) { - const ok = await useAuthStore.getState().retrySession().catch(() => false); - if (!ok || !mounted) return; - } - if (!handle) await start(); - else handle.reconnect(); - refreshAfterResume(); - })(); + // Coming back: reconnect and catch up on what was missed at once; + // one echo alongside checks the server for the connection dot. + if (!handle) void start(); + else handle.reconnect(); + refreshAfterResume(); + void liveness.check(); }); // Reconnect when the network comes back while foregrounded. @@ -665,39 +676,12 @@ export default function App() { mounted = false; subscription.remove(); unsubscribeNetwork(); + liveness.stop(); handle?.close(); handle = null; }; }, [client, isAuthenticated, haveLiveSession, activeAccountId]); - // Foreground keep-alive: a `Core/echo` every 30 s tells us when the server - // is unreachable even though the device is online (NetInfo cannot), and - // drives the per-account connection dot. - React.useEffect(() => { - if (!isAuthenticated || !haveLiveSession || !activeAccountId) return; - let cancelled = false; - let failures = 0; - const timer = setInterval(() => { - if (AppState.currentState !== 'active') return; - if (!useNetworkStore.getState().online) return; - void jmapClient.ping().then((ok) => { - if (cancelled) return; - failures = ok ? 0 : failures + 1; - const account = useAccountStore.getState().getAccountById(activeAccountId); - if (!account) return; - // One missed echo can be a blip; two in a row is a lost connection. - const connected = failures < 2; - if (account.isConnected !== connected) { - useAccountStore.getState().updateAccount(activeAccountId, { isConnected: connected }); - } - }).catch(() => undefined); - }, 30_000); - return () => { - cancelled = true; - clearInterval(timer); - }; - }, [isAuthenticated, haveLiveSession, activeAccountId]); - // Skip the "Restoring session" flash for returning users: if we already // have a persisted active account, render the main UI immediately with // whatever the email-store hydrated from cache. restoreSession still runs diff --git a/src/api/__tests__/push-stream-health.test.ts b/src/api/__tests__/push-stream-health.test.ts new file mode 100644 index 00000000..bfb970d5 --- /dev/null +++ b/src/api/__tests__/push-stream-health.test.ts @@ -0,0 +1,82 @@ +import { createRequire } from 'node:module'; +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; + +// The keep-alive echo is only sent while the event stream is not healthy +// (PF6), so the stream has to say when it is. + +type Handler = (event: { xhrStatus?: number; data?: string }) => void; +const sources: FakeEventSource[] = []; +class FakeEventSource { + handlers = new Map(); + closed = false; + constructor() { sources.push(this); } + addEventListener(type: string, handler: Handler) { + this.handlers.set(type, [...(this.handlers.get(type) ?? []), handler]); + } + removeEventListener(type: string, handler: Handler) { + this.handlers.set(type, (this.handlers.get(type) ?? []).filter((h) => h !== handler)); + } + close() { this.closed = true; } + emit(type: string, event: { xhrStatus?: number; data?: string } = {}) { + for (const h of this.handlers.get(type) ?? []) h(event); + } +} + +// push-stream loads the library with require(), which vi.mock does not +// reach: seed Node's module cache instead. +const nodeRequire = createRequire(import.meta.url); +const ssePath = nodeRequire.resolve('react-native-sse'); +nodeRequire.cache[ssePath] = { id: ssePath, filename: ssePath, loaded: true, exports: { default: FakeEventSource } } as NodeJS.Module; +vi.mock('../../lib/client-cert', () => ({ getClientCertAlias: vi.fn(async () => null) })); +vi.mock('../jmap-client', () => ({ + jmapClient: { + accountId: 'acc-1', + currentSession: { eventSourceUrl: 'https://mail.example.com/eventsource/?types={types}&closeafter={closeafter}&ping={ping}' }, + authHeader: 'Basic dXNlcjpwYXNz', + ensureFreshToken: vi.fn(async () => undefined), + onTokenRefresh: vi.fn(() => () => undefined), + request: vi.fn(), + }, + AuthenticationError: class AuthenticationError extends Error {}, +})); + +import { startLiveUpdates } from '../push-stream'; + +beforeEach(() => { + sources.length = 0; + vi.useFakeTimers(); +}); +afterEach(() => { vi.useRealTimers(); }); + +describe('live updates health', () => { + it('is healthy only while the stream is open and pinging', async () => { + const handle = await startLiveUpdates({ onStateChange: () => undefined }); + expect(handle.mode).toBe('sse'); + expect(handle.healthy).toBe(false); + + sources[0].emit('open'); + expect(handle.healthy).toBe(true); + sources[0].emit('ping'); + await vi.advanceTimersByTimeAsync(60_000); + expect(handle.healthy).toBe(true); + + // Three missed pings: the watchdog drops the stream. + await vi.advanceTimersByTimeAsync(90_000); + expect(handle.healthy).toBe(false); + handle.close(); + }); + + it('is not healthy after a network error or once closed', async () => { + const handle = await startLiveUpdates({ onStateChange: () => undefined }); + sources[0].emit('open'); + sources[0].emit('error', { xhrStatus: 0 }); + expect(handle.healthy).toBe(false); + + // First retry after a dropped socket: 5 s. + await vi.advanceTimersByTimeAsync(5_000); + sources[1].emit('open'); + expect(handle.healthy).toBe(true); + handle.close(); + expect(handle.healthy).toBe(false); + }); +}); diff --git a/src/api/push-stream.ts b/src/api/push-stream.ts index 131eee5b..3b584946 100644 --- a/src/api/push-stream.ts +++ b/src/api/push-stream.ts @@ -73,6 +73,11 @@ export interface LiveUpdatesHandle { reconnect: () => void; /** 'sse' | 'polling' | 'closed' */ readonly mode: 'sse' | 'polling' | 'closed'; + /** + * True while the event stream is open and its pings keep arriving (the + * watchdog drops it after three missed ones): the server is reachable. + */ + readonly healthy: boolean; } function buildEventSourceUrl(template: string): string { @@ -168,6 +173,7 @@ export async function startLiveUpdates(opts: LiveUpdatesOptions): Promise | null = null; let consecutiveFailures = 0; let closed = false; + let streamOpen = false; let unsubscribeTokenRefresh: (() => void) | null = null; const session = jmapClient.currentSession; @@ -181,6 +187,7 @@ export async function startLiveUpdates(opts: LiveUpdatesOptions): Promise { + streamOpen = false; if (es) { for (const [type, handler] of esHandlers) { try { es.removeEventListener(type, handler); } catch { /* ignore */ } @@ -269,6 +276,7 @@ export async function startLiveUpdates(opts: LiveUpdatesOptions): Promise { consecutiveFailures = 0; + streamOpen = es === source; armWatchdog(); if (recycleTimer) clearTimeout(recycleTimer); recycleTimer = setTimeout(() => { @@ -352,6 +360,9 @@ export async function startLiveUpdates(opts: LiveUpdatesOptions): Promise { closed = true; mode = 'closed'; diff --git a/src/lib/__tests__/connection-liveness.test.ts b/src/lib/__tests__/connection-liveness.test.ts new file mode 100644 index 00000000..21739e10 --- /dev/null +++ b/src/lib/__tests__/connection-liveness.test.ts @@ -0,0 +1,84 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { startLivenessMonitor, LIVENESS_INTERVAL_MS } from '../connection-liveness'; + +// PF6: a Core/echo every 30 s on top of the event stream's own pings, and two +// echoes in front of the catch-up sync on every resume. + +function setup(overrides: { healthy?: boolean; active?: boolean; online?: boolean; ping?: () => Promise } = {}) { + const state = { healthy: false, active: true, online: true, ...overrides }; + const ping = vi.fn(overrides.ping ?? (async () => true)); + const onConnected = vi.fn(); + const monitor = startLivenessMonitor({ + ping, + streamHealthy: () => state.healthy, + isActive: () => state.active, + isOnline: () => state.online, + onConnected, + }); + return { state, ping, onConnected, monitor }; +} + +beforeEach(() => { vi.useFakeTimers(); }); +afterEach(() => { vi.useRealTimers(); }); + +describe('connection liveness', () => { + it('sends no echo while the event stream is open', async () => { + const { ping, onConnected, monitor } = setup({ healthy: true }); + await vi.advanceTimersByTimeAsync(LIVENESS_INTERVAL_MS * 4); + expect(ping).not.toHaveBeenCalled(); + expect(onConnected).toHaveBeenLastCalledWith(true); + monitor.stop(); + }); + + it('falls back to an echo every 30 s without a stream (polling, reconnecting)', async () => { + const { ping, monitor } = setup(); + await vi.advanceTimersByTimeAsync(LIVENESS_INTERVAL_MS * 3); + expect(ping).toHaveBeenCalledTimes(3); + monitor.stop(); + }); + + it('reports a lost connection after two missed echoes', async () => { + const { onConnected, monitor } = setup({ ping: async () => false }); + await vi.advanceTimersByTimeAsync(LIVENESS_INTERVAL_MS); + expect(onConnected).toHaveBeenLastCalledWith(true); + await vi.advanceTimersByTimeAsync(LIVENESS_INTERVAL_MS); + expect(onConnected).toHaveBeenLastCalledWith(false); + monitor.stop(); + }); + + it('pauses in the background and offline', async () => { + const { state, ping, monitor } = setup({ active: false }); + await vi.advanceTimersByTimeAsync(LIVENESS_INTERVAL_MS * 2); + state.active = true; + state.online = false; + await vi.advanceTimersByTimeAsync(LIVENESS_INTERVAL_MS * 2); + expect(ping).not.toHaveBeenCalled(); + monitor.stop(); + }); + + it('sends one echo on resume, even when the periodic one fires with it', async () => { + let answer!: (ok: boolean) => void; + const { ping, monitor } = setup({ ping: () => new Promise((r) => { answer = r; }) }); + await vi.advanceTimersByTimeAsync(LIVENESS_INTERVAL_MS - 10); + // Resume: an on-demand check, then the timer that was due meanwhile. + const resumed = monitor.check(); + void monitor.check(); + await vi.advanceTimersByTimeAsync(10); + expect(ping).toHaveBeenCalledTimes(1); + answer(true); + await resumed; + await vi.advanceTimersByTimeAsync(1); + expect(ping).toHaveBeenCalledTimes(1); + monitor.stop(); + }); + + it('ignores an answer that lands after it was stopped', async () => { + let answer!: (ok: boolean) => void; + const { onConnected, monitor } = setup({ ping: () => new Promise((r) => { answer = r; }) }); + const pending = monitor.check(); + monitor.stop(); + answer(false); + await pending; + expect(onConnected).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/connection-liveness.ts b/src/lib/connection-liveness.ts new file mode 100644 index 00000000..baf8480d --- /dev/null +++ b/src/lib/connection-liveness.ts @@ -0,0 +1,79 @@ +// Whether the JMAP server is reachable while the app is in the foreground, +// for the per-account connection dot. NetInfo only knows the device is +// online, not that the server answers. +// +// An open event stream already proves it: the server pings it every 30 s and +// the stream is dropped after three missed pings. A `Core/echo` every 30 s is +// only sent while the stream is not open (polling fallback, reconnecting, +// client certificate), and on demand (`check`, e.g. on resume, where it runs +// alongside the catch-up sync instead of in front of it). Checks never +// overlap, and a periodic one is skipped right after an on-demand one. + +export const LIVENESS_INTERVAL_MS = 30_000; + +export interface LivenessOptions { + /** One `Core/echo`: true when the server answered. */ + ping: () => Promise; + /** Whether the event stream is open and its pings arrive. */ + streamHealthy: () => boolean; + /** Whether the app is in the foreground. */ + isActive: () => boolean; + /** Whether the device has a network. */ + isOnline: () => boolean; + /** Called with the connection state after each verdict. */ + onConnected: (connected: boolean) => void; + intervalMs?: number; + now?: () => number; +} + +export interface LivenessMonitor { + /** Check now (one `Core/echo`), unless a check is already running. */ + check: () => Promise; + stop: () => void; +} + +export function startLivenessMonitor(opts: LivenessOptions): LivenessMonitor { + const interval = opts.intervalMs ?? LIVENESS_INTERVAL_MS; + const now = opts.now ?? Date.now; + let failures = 0; + let inFlight: Promise | null = null; + let lastCheckAt = -Infinity; + let stopped = false; + + const check = (): Promise => { + if (inFlight) return inFlight; + lastCheckAt = now(); + inFlight = opts.ping() + .catch(() => false) + .then((ok) => { + if (stopped) return; + failures = ok ? 0 : failures + 1; + // One missed echo can be a blip; two in a row is a lost connection. + opts.onConnected(failures < 2); + }) + .finally(() => { + inFlight = null; + }); + return inFlight; + }; + + const timer = setInterval(() => { + if (!opts.isActive() || !opts.isOnline()) return; + if (opts.streamHealthy()) { + failures = 0; + opts.onConnected(true); + return; + } + // A tick that fired late (on resume) right after the resume check. + if (now() - lastCheckAt < interval / 2) return; + void check(); + }, interval); + + return { + check, + stop: () => { + stopped = true; + clearInterval(timer); + }, + }; +}