mirror of
https://github.com/bulwarkmail/native.git
synced 2026-11-03 08:06:59 +00:00
fix: sync at once on resume and stop pinging the server while live updates run
Coming back to the app first waited for a `Core/echo` (sometimes two: the keep-alive timer that was due fired alongside) and for the event stream to be set up before the catch-up sync started. A failed echo only led to `retrySession`, which does nothing while the session is live. On top of that, a `Core/echo` went out every 30 s although the event stream's own pings already show the server is reachable. - On resume the stream reconnects and the folders and open list are synced right away; one echo runs alongside for the connection dot. - The 30 s echo is only sent while the event stream is not open (polling fallback, reconnecting, client certificate). An open stream whose pings arrive (the live-updates handle's new `healthy`) marks the account connected instead. - Echoes never overlap, and a periodic one right after the resume one is skipped.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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<string, Handler[]>();
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<LiveUp
|
||||
let recycleTimer: ReturnType<typeof setTimeout> | 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<LiveUp
|
||||
};
|
||||
|
||||
const dropStream = () => {
|
||||
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<LiveUp
|
||||
|
||||
on('open', () => {
|
||||
consecutiveFailures = 0;
|
||||
streamOpen = es === source;
|
||||
armWatchdog();
|
||||
if (recycleTimer) clearTimeout(recycleTimer);
|
||||
recycleTimer = setTimeout(() => {
|
||||
@@ -352,6 +360,9 @@ export async function startLiveUpdates(opts: LiveUpdatesOptions): Promise<LiveUp
|
||||
get mode() {
|
||||
return mode;
|
||||
},
|
||||
get healthy() {
|
||||
return mode === 'sse' && streamOpen;
|
||||
},
|
||||
close: () => {
|
||||
closed = true;
|
||||
mode = 'closed';
|
||||
|
||||
@@ -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<boolean> } = {}) {
|
||||
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<boolean>((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<boolean>((r) => { answer = r; }) });
|
||||
const pending = monitor.check();
|
||||
monitor.stop();
|
||||
answer(false);
|
||||
await pending;
|
||||
expect(onConnected).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -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<boolean>;
|
||||
/** 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<void>;
|
||||
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<void> | null = null;
|
||||
let lastCheckAt = -Infinity;
|
||||
let stopped = false;
|
||||
|
||||
const check = (): Promise<void> => {
|
||||
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);
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user