feat: send telemetry error counts within a minute when an admin turns on instant mode

app/(main)/admin/_pages/telemetry.tsx

@@ -16,7 +16,8 @@ import { cn } from '@/lib/utils';
 interface TelemetryStatus {
   consent: 'pending' | 'on' | 'off';
   consentSource: 'no-third-party' | 'env' | 'file';
-  errorsConsent?: 'on' | 'off';
+  /** 'on' sends them daily; 'instant' also sends today's as they change. */
+  errorsConsent?: ErrorsMode;
   errorsConsentSource?: 'telemetry' | 'no-third-party' | 'env' | 'file';
   endpoint: string;
   defaultEndpoint: string;
@@ -25,9 +26,12 @@ interface TelemetryStatus {
   nextScheduledAt: string | null;
   payloadPreview: Record<string, unknown>;
   errorsPreview?: unknown[];
+  instantErrorsPreview?: unknown[];
   accountCounts: { total: number; active7d: number };
 }
 
+type ErrorsMode = 'on' | 'off' | 'instant';
+
 const PAGE = adminPageDef('telemetry');
 
 function timeAgo(iso: string | null): string {
@@ -59,7 +63,7 @@ export function TelemetryPage() {
   // The heartbeat switch waits for Save like the endpoint (one page, one
   // save bar).
   const [consentDraft, setConsentDraft] = useState<'on' | 'off' | null>(null);
-  const [errorsDraft, setErrorsDraft] = useState<'on' | 'off' | null>(null);
+  const [errorsDraft, setErrorsDraft] = useState<ErrorsMode | null>(null);
 
   const endpointDirty = status !== null && endpointDraft !== status.endpoint;
   const consentDirty = status !== null && consentDraft !== null && consentDraft !== status.consent;
@@ -159,12 +163,17 @@ export function TelemetryPage() {
 
   // Error counts follow the heartbeat switch as it is about to be saved.
   const errorsLocked = noThirdParty || status.errorsConsentSource === 'env' || !shownOn;
-  const errorsShownOn = shownOn && (errorsDraft ?? status.errorsConsent ?? 'off') === 'on';
+  const errorsMode: ErrorsMode = shownOn ? (errorsDraft ?? status.errorsConsent ?? 'off') : 'off';
+  const errorsShownOn = errorsMode !== 'off';
   const errorsDescription = [
     'Adds how often each error happened yesterday, the error name and where in the code it happened. No message text, no URLs, no instance id goes to the error tracker.',
     !shownOn && 'Needs the heartbeat on.',
     status.errorsConsentSource === 'env' && 'Locked by the BULWARK_TELEMETRY_ERRORS env var.',
   ].filter(Boolean).join(' ');
+  const instantDescription = [
+    'Also sends today\'s counts within about a minute when an error is new or happens noticeably more often, so a broken release shows up the same day. The same fields as the daily counts; the collector learns roughly when errors happen.',
+    !errorsShownOn && 'Needs error counts on.',
+  ].filter(Boolean).join(' ');
 
   // The payload as it will be once the switches are saved.
   const previewPayload = errorsShownOn
@@ -220,6 +229,13 @@ export function TelemetryPage() {
               onChange={(checked) => { setErrorsDraft(checked ? 'on' : 'off'); setMessage(null); }}
             />
           </SettingItem>
+          <SettingItem label="Send error counts right away" description={instantDescription}>
+            <ToggleSwitch
+              checked={errorsMode === 'instant'}
+              disabled={busy === 'save' || errorsLocked || !errorsShownOn}
+              onChange={(checked) => { setErrorsDraft(checked ? 'instant' : 'on'); setMessage(null); }}
+            />
+          </SettingItem>
           <SettingItem label="Last sent">
             <span className="text-sm text-foreground">{timeAgo(status.lastSentAt)}</span>
           </SettingItem>
@@ -290,6 +306,22 @@ export function TelemetryPage() {
           <pre className="text-xs bg-muted/50 rounded-md p-3 overflow-x-auto max-h-96">
             {JSON.stringify(previewPayload, null, 2)}
           </pre>
+          {errorsMode === 'instant' && (
+            <>
+              <p className="text-sm text-muted-foreground">
+                Sent right away: today&apos;s error counts as they would go out now. Each minute only
+                the errors that are new or reached a higher count range are sent.
+              </p>
+              <pre className="text-xs bg-muted/50 rounded-md p-3 overflow-x-auto max-h-96">
+                {JSON.stringify({
+                  schema: '1',
+                  instance_id: status.payloadPreview.instance_id,
+                  day: new Date().toISOString().slice(0, 10),
+                  errors: status.instantErrorsPreview ?? [],
+                }, null, 2)}
+              </pre>
+            </>
+          )}
         </SettingsSection>
       </div>
 

app/api/admin/telemetry/route.ts

@@ -14,8 +14,8 @@ import {
   resolveEndpointAllowed,
 } from '@/lib/telemetry';
 import { storedDataUnavailableResponse } from '@/lib/stored-data-errors';
-import { errorsForPayload } from '@/lib/telemetry/payload';
-import { readErrors } from '@/lib/error-reporting/aggregate';
+import { errorsForDay, errorsForPayload } from '@/lib/telemetry/payload';
+import { readErrors, utcDay } from '@/lib/error-reporting/aggregate';
 
 /**
  * GET /api/admin/telemetry
@@ -51,6 +51,8 @@ export async function GET(request: NextRequest) {
         // What `errors` would hold, so the page can show the v2 payload as soon
         // as the switch is flipped, before anything is saved or sent.
         errorsPreview: errorsForPayload(counted),
+        // What instant mode would send now: today's counts so far.
+        instantErrorsPreview: errorsForDay(counted, utcDay()),
         accountCounts,
       },
       { headers: { 'Cache-Control': 'no-store' } },
@@ -69,7 +71,7 @@ export async function GET(request: NextRequest) {
  * POST /api/admin/telemetry
  * Body: { action: 'set-consent' | 'set-endpoint' | 'send-now', ... }
  *   set-consent  : { action, consent: 'on' | 'off' } (off also switches error counts off)
- *   set-errors-consent : { action, consent: 'on' | 'off' }
+ *   set-errors-consent : { action, consent: 'on' | 'instant' | 'off' } ('on' is daily)
  *   set-endpoint : { action, endpoint: string }
  *   send-now     : { action }
  */
@@ -129,10 +131,10 @@ export async function POST(request: NextRequest) {
       if (errors.source === 'no-third-party') {
         return NextResponse.json({ error: 'telemetry is off while NO_THIRD_PARTY_TRAFFIC is on' }, { status: 409 });
       }
-      if (body.consent !== 'on' && body.consent !== 'off') {
-        return NextResponse.json({ error: 'consent must be "on" or "off"' }, { status: 400 });
+      if (body.consent !== 'on' && body.consent !== 'off' && body.consent !== 'instant') {
+        return NextResponse.json({ error: 'consent must be "on", "instant" or "off"' }, { status: 400 });
       }
-      if (body.consent === 'on' && errors.source === 'telemetry') {
+      if (body.consent !== 'off' && errors.source === 'telemetry') {
         return NextResponse.json({ error: 'turn on the heartbeat first' }, { status: 409 });
       }
       const consent = body.consent;

instrumentation.node.ts

@@ -119,15 +119,19 @@ migrateLegacyAdminLayout()
     // See https://bulwarkmail.org/docs/legal/privacy/telemetry
     const { startScheduler, stopScheduler, markProcessStart } = await import("./lib/telemetry");
     markProcessStart();
+    // Instant error counts: a one-minute tick that sends only while the
+    // error-count consent is "instant".
+    const instantErrors = await import("./lib/telemetry/instant-errors");
     // Hourly check against version.telemetry.bulwarkmail.org. Disable with
     // BULWARK_UPDATE_CHECK=off or override the endpoint with
     // BULWARK_UPDATE_CHECK_URL.
     const versionCheck = await import("./lib/version-check");
-    // Both run on one replica of the deployment, the scheduler leader
+    // All run on one replica of the deployment, the scheduler leader
     // (#1066); a single instance always leads.
     const { startSchedulerLeader } = await import("./lib/scheduler-leader");
     await startSchedulerLeader([
       { name: "telemetry", start: startScheduler, stop: stopScheduler },
+      { name: "telemetry-errors", start: instantErrors.startInstantErrors, stop: instantErrors.stopInstantErrors },
       { name: "version-check", start: versionCheck.startScheduler, stop: versionCheck.stopScheduler },
     ]);
   })

lib/__tests__/telemetry-instant-errors.test.ts

@@ -0,0 +1,189 @@
+// @vitest-environment node
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import path from 'node:path';
+import { NextRequest } from 'next/server';
+
+// Instant error counts: with the error-count consent set to "instant", the
+// scheduler leader sends the current day's counts to the collector's
+// /v1/errors once a minute, but only errors that are new or whose count
+// bucket changed since the last send. Daily mode sends nothing extra.
+
+const thirdParty = vi.hoisted(() => ({ on: false }));
+vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => thirdParty.on }));
+vi.mock('@/lib/admin/session', () => ({
+  requireAdminAuth: vi.fn(async () => ({ session: { sub: 'admin' } })),
+  getClientIP: () => '127.0.0.1',
+}));
+vi.mock('@/lib/admin/audit', () => ({ auditLog: vi.fn(async () => {}) }));
+vi.mock('@/lib/admin/plugin-registry', () => ({
+  getPluginRegistry: async () => ({ plugins: [] }),
+  getThemeRegistry: async () => ({ themes: [] }),
+}));
+
+const sent = vi.fn(async (_url: string, _init?: { body?: string }) => new Response('{"ok":true}'));
+vi.mock('@/lib/telemetry/endpoint-guard', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@/lib/telemetry/endpoint-guard')>();
+  return {
+    ...actual,
+    resolveEndpointAllowed: vi.fn(async () => ({ ok: true })),
+    fetchTelemetryTarget: (url: string, init?: { body?: string }) => sent(url, init),
+  };
+});
+
+import { POST } from '@/app/api/admin/telemetry/route';
+import { DEFAULT_ENDPOINT, buildPayload, effectiveErrorsConsent, loadState, saveState } from '@/lib/telemetry';
+import { resetInstantErrorsForTests, sendInstantErrors } from '@/lib/telemetry/instant-errors';
+import { resetAggregateForTests, utcDay } from '@/lib/error-reporting/aggregate';
+
+let dir: string;
+const debugId = 'bc67f7db-7f3b-7be2-6f94-966cd7825cc2';
+
+function entry(count: number, name = 'TypeError') {
+  return {
+    name,
+    frames: [
+      { path: '/_next/static/chunks/1.js', line: 2, column: 30, debugId },
+      { path: 'node_modules/next/dist/x.js', line: 1, column: 1 },
+    ],
+    version: '1.20.0',
+    runtime: 'browser' as const,
+    count,
+  };
+}
+
+function storeCounts(day: Record<string, ReturnType<typeof entry>>): void {
+  writeFileSync(path.join(dir, 'telemetry', 'errors.json'), JSON.stringify({ [utcDay()]: day }));
+}
+
+function bodies(): Array<Record<string, unknown> & { errors: Array<Record<string, unknown>> }> {
+  return sent.mock.calls.map(([, init]) => JSON.parse(String(init?.body)));
+}
+
+function post(body: Record<string, unknown>) {
+  return POST(new NextRequest('http://localhost/api/admin/telemetry', {
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    body: JSON.stringify(body),
+  }));
+}
+
+beforeAll(() => {
+  dir = mkdtempSync(path.join(tmpdir(), 'bulwark-telemetry-instant-'));
+  vi.stubEnv('ADMIN_CONFIG_DIR', dir);
+  vi.stubEnv('TELEMETRY_DATA_DIR', path.join(dir, 'telemetry'));
+  mkdirSync(path.join(dir, 'telemetry'), { recursive: true });
+});
+afterAll(() => {
+  vi.unstubAllEnvs();
+  rmSync(dir, { recursive: true, force: true });
+});
+beforeEach(async () => {
+  thirdParty.on = false;
+  vi.stubEnv('BULWARK_TELEMETRY', '');
+  vi.stubEnv('BULWARK_TELEMETRY_ERRORS', '');
+  sent.mockClear();
+  sent.mockImplementation(async () => new Response('{"ok":true}'));
+  resetAggregateForTests();
+  resetInstantErrorsForTests();
+  await saveState({ ...(await loadState()), consent: 'on', endpoint: DEFAULT_ENDPOINT, errorsConsent: 'instant' });
+  storeCounts({ aaaa000011112222: entry(1) });
+});
+afterEach(() => {
+  resetAggregateForTests();
+  resetInstantErrorsForTests();
+});
+
+describe('instant error counts', () => {
+  it('sends today\'s counts to the collector\'s /v1/errors, nothing but the allowed fields', async () => {
+    await sendInstantErrors();
+    expect(sent).toHaveBeenCalledTimes(1);
+    expect(sent.mock.calls[0][0]).toBe('https://telemetry.bulwarkmail.org/v1/errors');
+    const [body] = bodies();
+    expect(Object.keys(body).sort()).toEqual(['day', 'errors', 'instance_id', 'schema']);
+    expect(body.schema).toBe('1');
+    expect(body.day).toBe(utcDay());
+    expect(body.instance_id).toMatch(/^[0-9a-f-]{36}$/);
+    expect(body.errors).toEqual([{
+      name: 'TypeError',
+      fingerprint: 'aaaa000011112222',
+      frames: [{ path: '/_next/static/chunks/1.js', line: 2, col: 30, debug_id: debugId }],
+      version: '1.20.0',
+      count: '1',
+      days: 1,
+    }]);
+  });
+
+  it('sends an error again only when its count bucket changes', async () => {
+    storeCounts({ aaaa000011112222: entry(1), bbbb000011112222: entry(1, 'RangeError') });
+    await sendInstantErrors();
+    expect(bodies()[0].errors).toHaveLength(2);
+
+    await sendInstantErrors();
+    expect(sent).toHaveBeenCalledTimes(1);
+
+    // 1 -> 3 moves from '1' to '2-5'.
+    storeCounts({ aaaa000011112222: entry(3), bbbb000011112222: entry(1, 'RangeError') });
+    await sendInstantErrors();
+    expect(sent).toHaveBeenCalledTimes(2);
+    expect(bodies()[1].errors.map((e) => [e.fingerprint, e.count])).toEqual([['aaaa000011112222', '2-5']]);
+
+    // 3 -> 4 is still '2-5': nothing new to say.
+    storeCounts({ aaaa000011112222: entry(4), bbbb000011112222: entry(1, 'RangeError') });
+    await sendInstantErrors();
+    expect(sent).toHaveBeenCalledTimes(2);
+  });
+
+  it('tries a failed send again on the next tick', async () => {
+    sent.mockImplementationOnce(async () => new Response('{}', { status: 503 }));
+    await sendInstantErrors();
+    await sendInstantErrors();
+    expect(sent).toHaveBeenCalledTimes(2);
+    expect(bodies()[1].errors).toHaveLength(1);
+  });
+
+  it('sends nothing in daily mode, with error counts off, with the heartbeat off or in no-third-party mode', async () => {
+    await saveState({ ...(await loadState()), errorsConsent: 'on' });
+    await sendInstantErrors();
+    await saveState({ ...(await loadState()), errorsConsent: 'off' });
+    await sendInstantErrors();
+    await saveState({ ...(await loadState()), consent: 'off', errorsConsent: 'instant' });
+    await sendInstantErrors();
+    await saveState({ ...(await loadState()), consent: 'on' });
+    thirdParty.on = true;
+    await sendInstantErrors();
+    expect(sent).not.toHaveBeenCalled();
+  });
+
+  it('sends nothing when the collector URL is empty, and follows a custom one', async () => {
+    await saveState({ ...(await loadState()), endpoint: '' });
+    await sendInstantErrors();
+    expect(sent).not.toHaveBeenCalled();
+    await saveState({ ...(await loadState()), endpoint: 'https://collector.example.org/v1/heartbeat' });
+    await sendInstantErrors();
+    expect(sent.mock.calls[0][0]).toBe('https://collector.example.org/v1/errors');
+  });
+
+  it('also sends the daily heartbeat with error counts (schema 2)', async () => {
+    expect((await buildPayload()).schema).toBe('2');
+  });
+});
+
+describe('instant consent', () => {
+  it('can be chosen in the admin panel and needs the heartbeat', async () => {
+    await saveState({ ...(await loadState()), errorsConsent: 'off' });
+    expect((await post({ action: 'set-errors-consent', consent: 'instant' })).status).toBe(200);
+    expect((await loadState()).errorsConsent).toBe('instant');
+    expect((await effectiveErrorsConsent()).consent).toBe('instant');
+
+    await saveState({ ...(await loadState()), consent: 'off', errorsConsent: 'off' });
+    expect((await post({ action: 'set-errors-consent', consent: 'instant' })).status).toBe(409);
+  });
+
+  it('can be set with BULWARK_TELEMETRY_ERRORS=instant', async () => {
+    await saveState({ ...(await loadState()), errorsConsent: 'off' });
+    vi.stubEnv('BULWARK_TELEMETRY_ERRORS', 'instant');
+    expect(await effectiveErrorsConsent()).toEqual({ consent: 'instant', source: 'env' });
+  });
+});

lib/admin/outgoing-hosts-registry.ts

@@ -39,7 +39,7 @@ export const OUTGOING_HOSTS: readonly OutgoingHost[] = [
     host: 'telemetry.bulwarkmail.org',
     files: ['lib/telemetry/types.ts'],
     side: 'server',
-    usedBy: 'Anonymous daily heartbeat, sent only after an admin turned telemetry on.',
+    usedBy: 'Anonymous daily heartbeat, sent only after an admin turned telemetry on; with instant error counts on, also the current day\'s error counts as they change.',
     off: 'Off by default. BULWARK_TELEMETRY=off, or System > Usage stats (switch it off or clear the collector URL).',
     inMode: 'Never sent, whatever the telemetry setting says.',
     lite: false,

lib/error-reporting/__tests__/collect-consent.test.ts

@@ -1,7 +1,7 @@
 // @vitest-environment node
 import { describe, expect, it, vi } from 'vitest';
 
-const consent = vi.hoisted(() => ({ value: 'off' as 'on' | 'off' }));
+const consent = vi.hoisted(() => ({ value: 'off' as 'on' | 'off' | 'instant' }));
 vi.mock('@/lib/admin/config-manager', () => ({
   configManager: { ensureLoaded: async () => {}, get: (_key: string, fallback: unknown) => fallback },
 }));
@@ -23,4 +23,9 @@ describe('error collection', () => {
     consent.value = 'on';
     expect(await shouldCollectErrors()).toBe(true);
   });
+
+  it('is on in instant mode too', async () => {
+    consent.value = 'instant';
+    expect(await shouldCollectErrors()).toBe(true);
+  });
 });

lib/error-reporting/settings.ts

@@ -46,7 +46,7 @@ export function errorReportingSettings(): ErrorReportingSettings {
 export async function shouldCollectErrors(): Promise<boolean> {
   await configManager.ensureLoaded();
   if (errorReportingSettings().dsn || metricsEnabled()) return true;
-  return (await effectiveErrorsConsent()).consent === 'on';
+  return (await effectiveErrorsConsent()).consent !== 'off';
 }
 
 /** What /api/config hands the browser. */

lib/telemetry/index.ts

@@ -1,5 +1,5 @@
 export { startScheduler, stopScheduler, reschedule, sendOnce } from './sender';
-export { buildPayload, errorsForPayload, markProcessStart } from './payload';
+export { buildPayload, errorsForDay, errorsForPayload, markProcessStart } from './payload';
 export {
   loadState, saveState, updateState, getInstanceId, effectiveConsent, effectiveErrorsConsent,
 } from './state';
@@ -10,6 +10,7 @@ export {
 export type { EndpointCheck } from './endpoint-guard';
 export type {
   TelemetryPayload, TelemetryPayloadV2, AnyTelemetryPayload, TelemetryError, TelemetryStateFile, ConsentState,
+  ErrorsConsent, TelemetryErrorBatch,
   Platform, OsFamily, CountBucket, TelemetryFeatures,
 } from './types';
 export { DEFAULT_ENDPOINT } from './types';

lib/telemetry/instant-errors.ts

@@ -0,0 +1,123 @@
+import { logger } from '@/lib/logger';
+import { readErrors, utcDay } from '@/lib/error-reporting/aggregate';
+import { effectiveConsent, effectiveErrorsConsent, endpointEnabled, getInstanceId } from './state';
+import { errorsForDay } from './payload';
+import { fetchTelemetryTarget, resolveEndpointAllowed } from './endpoint-guard';
+import { DEFAULT_ENDPOINT, errorsEndpointFor, type TelemetryErrorBatch } from './types';
+
+/**
+ * Instant error counts. With the error-count consent on "instant", the
+ * scheduler leader posts the current UTC day's counts to the collector's
+ * /v1/errors every TICK_MS. Only errors that are new or whose count bucket
+ * changed since the last send go out, so a known error repeating doesn't
+ * cause traffic until it reaches the next bucket. The daily heartbeat still
+ * carries the complete day as before.
+ *
+ * Each tick reads the consent, so switching it in the admin takes effect on
+ * the next tick without restarting anything. Other replicas' counts reach
+ * the store with their own flush (every minute), so they go out a minute later.
+ */
+
+const TICK_MS = 60_000;
+
+interface InstantState {
+  timer: NodeJS.Timeout | null;
+  /** The day `sent` is for. */
+  day: string;
+  /** `fingerprint|version` -> the count bucket last sent. */
+  sent: Map<string, string>;
+  running: Promise<void> | null;
+}
+
+// On globalThis, like the heartbeat timer: one per process, whichever
+// module copy starts or stops it.
+const KEY = Symbol.for('bulwark.telemetry.instant-errors');
+function state(): InstantState {
+  const g = globalThis as typeof globalThis & { [KEY]?: InstantState };
+  return (g[KEY] ??= { timer: null, day: '', sent: new Map(), running: null });
+}
+
+async function tick(now: number): Promise<void> {
+  const s = state();
+  const { consent } = await effectiveErrorsConsent();
+  if (consent !== 'instant') {
+    s.sent.clear();
+    return;
+  }
+  const { state: stored } = await effectiveConsent();
+  const endpoint = stored.endpoint ?? DEFAULT_ENDPOINT;
+  if (!endpointEnabled(endpoint)) return;
+  let target: string;
+  try {
+    target = errorsEndpointFor(endpoint);
+  } catch {
+    return;
+  }
+
+  const day = utcDay(now);
+  if (s.day !== day) {
+    s.day = day;
+    s.sent.clear();
+  }
+  const changed = errorsForDay(await readErrors(), day)
+    .filter((error) => s.sent.get(`${error.fingerprint}|${error.version}`) !== error.count);
+  if (changed.length === 0) return;
+
+  // Checked at send time, like the heartbeat: defeats DNS rebinding and a
+  // state file edited past the admin API.
+  const guard = await resolveEndpointAllowed(target);
+  if (!guard.ok) {
+    logger.warn('telemetry: errors endpoint blocked', { reason: guard.reason });
+    return;
+  }
+
+  const batch: TelemetryErrorBatch = { schema: '1', instance_id: await getInstanceId(), day, errors: changed };
+  const res = await fetchTelemetryTarget(target, {
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    body: JSON.stringify(batch),
+    signal: AbortSignal.timeout(5000),
+  });
+  if (!res.ok) {
+    // Not marked as sent: the next tick tries them again.
+    logger.warn('telemetry: instant error counts rejected', { status: res.status });
+    return;
+  }
+  for (const error of changed) s.sent.set(`${error.fingerprint}|${error.version}`, error.count);
+}
+
+/** One tick: send what changed, if instant mode is on. Never throws. */
+export function sendInstantErrors(now = Date.now()): Promise<void> {
+  const s = state();
+  if (s.running) return s.running;
+  s.running = tick(now)
+    .catch((error) => {
+      logger.warn('telemetry: instant error counts failed', { error: error instanceof Error ? error.message : String(error) });
+    })
+    .finally(() => { s.running = null; });
+  return s.running;
+}
+
+// Run by the scheduler leader (lib/scheduler-leader.ts). Idempotent.
+export async function startInstantErrors(): Promise<void> {
+  const s = state();
+  if (s.timer) return;
+  s.timer = setInterval(() => { void sendInstantErrors(); }, TICK_MS);
+  s.timer.unref?.();
+}
+
+export async function stopInstantErrors(): Promise<void> {
+  const s = state();
+  if (s.timer) clearInterval(s.timer);
+  s.timer = null;
+  // A new leader starts from nothing and sends the day's counts once.
+  s.sent.clear();
+}
+
+/** For tests. */
+export function resetInstantErrorsForTests(): void {
+  const s = state();
+  if (s.timer) clearInterval(s.timer);
+  const g = globalThis as typeof globalThis & { [KEY]?: InstantState };
+  delete g[KEY];
+}

lib/telemetry/payload.ts

@@ -115,8 +115,17 @@ const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
  * and a message only when it is one of Bulwark's fixed error codes.
  */
 export function errorsForPayload(agg: Aggregate, now = Date.now()): TelemetryError[] {
-  const day = utcDay(now - 86_400_000);
-  const recentDays = Array.from({ length: 7 }, (_, i) => utcDay(now - (i + 1) * 86_400_000));
+  return errorsForDay(agg, utcDay(now - 86_400_000));
+}
+
+/**
+ * The error counts of one UTC day, shaped as in errorsForPayload. `days`
+ * counts the days of the week ending on `day` the error occurred on.
+ * Instant mode sends the current day's this way.
+ */
+export function errorsForDay(agg: Aggregate, day: string): TelemetryError[] {
+  const end = Date.parse(`${day}T00:00:00Z`);
+  const recentDays = Array.from({ length: 7 }, (_, i) => utcDay(end - i * 86_400_000));
   return Object.entries(agg[day] ?? {})
     .sort(([, a], [, b]) => b.count - a.count)
     .slice(0, MAX_TELEMETRY_ERRORS)
@@ -145,7 +154,8 @@ export function errorsForPayload(agg: Aggregate, now = Date.now()): TelemetryErr
  */
 export async function buildPayload(): Promise<AnyTelemetryPayload> {
   const base = await buildPayloadV1();
-  if ((await effectiveErrorsConsent()).consent !== 'on') return base;
+  // Daily and instant mode both send the last complete day with the heartbeat.
+  if ((await effectiveErrorsConsent()).consent === 'off') return base;
   return { ...base, schema: '2', errors: errorsForPayload(await readErrors()) };
 }
 

lib/telemetry/state.ts

@@ -3,7 +3,7 @@ import { randomUUID } from 'node:crypto';
 import { logger } from '@/lib/logger';
 import { getStore, type CollectionOptions } from '@/lib/shared-store';
 import { isStoreUnavailable, StoredDataUnreadableError } from '@/lib/stored-data-errors';
-import type { TelemetryStateFile, ConsentState } from './types';
+import type { TelemetryStateFile, ConsentState, ErrorsConsent } from './types';
 import { DEFAULT_ENDPOINT } from './types';
 import { isNoThirdPartyTrafficOn } from '@/lib/admin/third-party-mode';
 
@@ -174,10 +174,11 @@ export async function saveStateIfUnchanged(state: TelemetryStateFile, version: s
   return (await telemetryDocs().put(STATE_DOC, serialize(state), { ifVersion: version })).ok;
 }
 
-function errorsEnvOverride(): 'on' | 'off' | null {
+function errorsEnvOverride(): ErrorsConsent | null {
   const v = (process.env.BULWARK_TELEMETRY_ERRORS ?? '').toLowerCase();
   if (v === 'off' || v === 'false' || v === '0' || v === 'no') return 'off';
-  if (v === 'on' || v === 'true' || v === '1' || v === 'yes') return 'on';
+  if (v === 'on' || v === 'true' || v === '1' || v === 'yes' || v === 'daily') return 'on';
+  if (v === 'instant') return 'instant';
   return null;
 }
 
@@ -185,9 +186,11 @@ function errorsEnvOverride(): 'on' | 'off' | null {
  * Effective consent to send error counts (schema v2): off whenever
  * telemetry itself is off (which covers the no-third-party mode), then
  * BULWARK_TELEMETRY_ERRORS, then the stored choice, off by default.
+ * 'on' sends them daily with the heartbeat; 'instant' does that too and
+ * also sends the current day's counts as they change.
  */
 export async function effectiveErrorsConsent(): Promise<{
-  consent: 'on' | 'off';
+  consent: ErrorsConsent;
   source: 'telemetry' | 'no-third-party' | 'env' | 'file';
 }> {
   const telemetry = await effectiveConsent();
@@ -195,7 +198,8 @@ export async function effectiveErrorsConsent(): Promise<{
   if (telemetry.consent !== 'on') return { consent: 'off', source: 'telemetry' };
   const env = errorsEnvOverride();
   if (env) return { consent: env, source: 'env' };
-  return { consent: telemetry.state.errorsConsent === 'on' ? 'on' : 'off', source: 'file' };
+  const stored = telemetry.state.errorsConsent;
+  return { consent: stored === 'on' || stored === 'instant' ? stored : 'off', source: 'file' };
 }
 
 // Effective consent: the no-third-party mode wins over everything (#903),

lib/telemetry/types.ts

@@ -73,6 +73,30 @@ export interface TelemetryPayloadV2 extends Omit<TelemetryPayload, 'schema'> {
 
 export type AnyTelemetryPayload = TelemetryPayload | TelemetryPayloadV2;
 
+/**
+ * Error counts: off, daily (with the heartbeat, for the day before) or
+ * instant (daily, plus the current day's counts within about a minute).
+ * 'on' is the daily mode; the stored value predates the instant one.
+ */
+export type ErrorsConsent = 'on' | 'off' | 'instant';
+
+/**
+ * What instant mode posts to the collector's /v1/errors: the current UTC
+ * day's errors that are new or whose count bucket changed since the last send.
+ */
+export interface TelemetryErrorBatch {
+  schema: '1';
+  instance_id: string;
+  /** The UTC day the counts are for (YYYY-MM-DD). */
+  day: string;
+  errors: TelemetryError[];
+}
+
+/** The collector's errors URL next to its heartbeat URL: `.../v1/heartbeat` -> `.../v1/errors`. */
+export function errorsEndpointFor(heartbeatEndpoint: string): string {
+  return new URL('errors', heartbeatEndpoint).toString();
+}
+
 export interface TelemetryStateFile {
   consent: ConsentState;
   endpoint: string;
@@ -85,10 +109,11 @@ export interface TelemetryStateFile {
   lastScheduledSentAt?: string | null;
   nextScheduledAt: string | null;
   /**
-   * The separate consent to add error counts (schema v2). Off unless an
-   * admin turns it on; switching telemetry off switches it off too.
+   * The separate consent to add error counts (schema v2), and whether they
+   * also go out instantly. Off unless an admin turns it on; switching
+   * telemetry off switches it off too.
    */
-  errorsConsent?: 'on' | 'off';
+  errorsConsent?: ErrorsConsent;
 }
 
 export const DEFAULT_ENDPOINT = 'https://telemetry.bulwarkmail.org/v1/heartbeat';