Merge branch 'feat/multi-instance-data' into feat/multi-instance

This commit is contained in:
Linus Rath
2026-11-01 00:05:03 +01:00
42 changed files with 2625 additions and 321 deletions
+9 -3
View File
@@ -11,6 +11,7 @@ import { rejectCrossOriginRequest } from '@/lib/security/same-origin';
import { MAX_ACCOUNT_SLOTS } from '@/lib/account-utils';
import { archiveIdentity, parseArchiveEnvelope, ARCHIVE_MAX_BYTES, type ArchiveOwner } from '@/lib/account-archive';
import { deleteArchive, loadArchive, saveArchive } from '@/lib/account-archive-storage';
import { isStoreUnavailable } from '@/lib/stored-data-errors';
export const runtime = 'nodejs';
@@ -18,6 +19,11 @@ const reply = (data: unknown, status = 200) => NextResponse.json(data, {
status, headers: { 'Cache-Control': 'no-store' },
});
/** A shared store that can't be reached is a 503 the client retries; nothing was written. */
const storageFailed = (error: unknown) => isStoreUnavailable(error)
? NextResponse.json({ error: 'storage_unavailable' }, { status: 503, headers: { 'Cache-Control': 'no-store', 'Retry-After': '10' } })
: reply({ error: 'storage_failed' }, 500);
/**
* The owner whose archive this request may touch, or the response refusing it.
*
@@ -67,7 +73,7 @@ export async function GET(request: NextRequest) {
const owner = await verifiedOwner(request);
if (owner instanceof NextResponse) return owner;
try { return reply({ envelope: await loadArchive(owner) }); }
catch { return reply({ error: 'storage_failed' }, 500); }
catch (error) { return storageFailed(error); }
}
export async function PUT(request: NextRequest) {
@@ -83,7 +89,7 @@ export async function PUT(request: NextRequest) {
envelope = parseArchiveEnvelope((JSON.parse(text) as { envelope?: unknown })?.envelope);
} catch { return reply({ error: 'invalid_archive' }, 400); }
try { await saveArchive(owner, envelope); return reply({ ok: true }); }
catch { return reply({ error: 'storage_failed' }, 500); }
catch (error) { return storageFailed(error); }
}
export async function DELETE(request: NextRequest) {
@@ -92,5 +98,5 @@ export async function DELETE(request: NextRequest) {
const owner = await verifiedOwner(request);
if (owner instanceof NextResponse) return owner;
try { await deleteArchive(owner); return reply({ ok: true }); }
catch { return reply({ error: 'storage_failed' }, 500); }
catch (error) { return storageFailed(error); }
}
+20 -13
View File
@@ -4,8 +4,7 @@ import { auditLog } from '@/lib/admin/audit';
import { logger } from '@/lib/logger';
import {
effectiveConsent,
loadState,
saveState,
updateState,
buildPayload,
sendOnce,
reschedule,
@@ -13,6 +12,7 @@ import {
getLoginCounts,
resolveEndpointAllowed,
} from '@/lib/telemetry';
import { storedDataUnavailableResponse } from '@/lib/stored-data-errors';
/**
* GET /api/admin/telemetry
@@ -48,6 +48,8 @@ export async function GET(request: NextRequest) {
logger.error('telemetry GET error', {
error: err instanceof Error ? err.message : 'unknown',
});
const unavailable = storedDataUnavailableResponse(err);
if (unavailable) return unavailable;
return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
}
}
@@ -90,13 +92,15 @@ export async function POST(request: NextRequest) {
if (body.consent !== 'on' && body.consent !== 'off') {
return NextResponse.json({ error: 'consent must be "on" or "off"' }, { status: 400 });
}
const state = await loadState();
const before = state.consent;
state.consent = body.consent;
if (body.consent === 'on' && !state.consentedAt) {
state.consentedAt = new Date().toISOString();
}
await saveState(state);
const consent = body.consent;
const now = new Date().toISOString();
let before: string | undefined;
// Against the stored state, so a schedule written meanwhile stays.
await updateState((state) => {
before = state.consent;
state.consent = consent;
if (consent === 'on' && !state.consentedAt) state.consentedAt = now;
});
await reschedule();
await auditLog('telemetry.set-consent', { from: before, to: body.consent }, ip);
return NextResponse.json({ ok: true });
@@ -113,11 +117,12 @@ export async function POST(request: NextRequest) {
return NextResponse.json({ error: check.reason }, { status: 400 });
}
}
const state = await loadState();
const before = state.endpoint;
let before: string | undefined;
// Empty means no collector: heartbeats stop until a URL is set again.
state.endpoint = trimmed;
await saveState(state);
const state = await updateState((current) => {
before = current.endpoint;
current.endpoint = trimmed;
});
await reschedule();
await auditLog('telemetry.set-endpoint', { from: before, to: state.endpoint }, ip);
return NextResponse.json({ ok: true, endpoint: state.endpoint });
@@ -138,6 +143,8 @@ export async function POST(request: NextRequest) {
logger.error('telemetry POST error', {
error: err instanceof Error ? err.message : 'unknown',
});
const unavailable = storedDataUnavailableResponse(err);
if (unavailable) return unavailable;
return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
}
}
+5
View File
@@ -10,6 +10,7 @@ import {
DEFAULT_VERSION_ENDPOINT,
} from '@/lib/version-check';
import { isNoThirdPartyTrafficOn } from '@/lib/admin/third-party-mode';
import { storedDataUnavailableResponse } from '@/lib/stored-data-errors';
/**
* GET /api/admin/version
@@ -40,6 +41,8 @@ export async function GET(request: NextRequest) {
logger.error('version admin GET error', {
error: err instanceof Error ? err.message : 'unknown',
});
const unavailable = storedDataUnavailableResponse(err, { 'Cache-Control': 'no-store' });
if (unavailable) return unavailable;
return NextResponse.json({ error: 'failed' }, { status: 500 });
}
}
@@ -64,6 +67,8 @@ export async function POST(req: NextRequest) {
logger.error('version admin POST error', {
error: err instanceof Error ? err.message : 'unknown',
});
const unavailable = storedDataUnavailableResponse(err, { 'Cache-Control': 'no-store' });
if (unavailable) return unavailable;
return NextResponse.json({ error: 'failed' }, { status: 500 });
}
}
+18 -2
View File
@@ -12,6 +12,7 @@ import {
putUserCredential,
type VaultUser,
} from '@/lib/plugin-secrets/vault';
import { storedDataUnavailableResponse } from '@/lib/stored-data-errors';
const PLUGIN_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/;
const MAX_SECRET_LENGTH = 4096;
@@ -35,7 +36,14 @@ export async function GET(request: NextRequest) {
if (pluginId !== undefined && !PLUGIN_ID_RE.test(pluginId)) {
return NextResponse.json({ error: 'Invalid plugin id' }, { status: 400 });
}
const credentials = await listUserCredentials(user, pluginId);
let credentials;
try {
credentials = await listUserCredentials(user, pluginId);
} catch (error) {
const unavailable = storedDataUnavailableResponse(error, NO_STORE);
if (unavailable) return unavailable;
throw error;
}
// For one plugin, also say which of its admin credentials are set up, and
// on which servers: the host page checks the links a plugin shows against
// them (the plugin itself only gets yes/no, see proxyAccounts).
@@ -93,6 +101,8 @@ export async function POST(request: NextRequest) {
});
} catch (error) {
if (error instanceof VaultFullError) return NextResponse.json({ error: error.message }, { status: 409 });
const unavailable = storedDataUnavailableResponse(error, NO_STORE);
if (unavailable) return unavailable;
throw error;
}
return NextResponse.json({ ok: true }, { headers: NO_STORE });
@@ -108,7 +118,13 @@ export async function DELETE(request: NextRequest) {
const credentialId = request.nextUrl.searchParams.get('credential') ?? undefined;
if (!PLUGIN_ID_RE.test(pluginId)) return bad('Invalid plugin id');
if (credentialId !== undefined && !CREDENTIAL_ID_RE.test(credentialId)) return bad('Invalid credential id');
await deleteUserCredentials(user, pluginId, credentialId);
try {
await deleteUserCredentials(user, pluginId, credentialId);
} catch (error) {
const unavailable = storedDataUnavailableResponse(error, NO_STORE);
if (unavailable) return unavailable;
throw error;
}
return NextResponse.json({ ok: true }, { headers: NO_STORE });
}
+14 -4
View File
@@ -5,13 +5,14 @@ import { logger } from '@/lib/logger';
import { decryptSession } from '@/lib/auth/crypto';
import { sessionCookieName } from '@/lib/auth/session-cookie';
import { readStalwartAuthContextFromStore } from '@/lib/stalwart/auth-context';
import { saveUserSettings, loadUserSettings, deleteUserSettings } from '@/lib/settings-sync';
import { updateUserSettings, loadUserSettings, deleteUserSettings } from '@/lib/settings-sync';
import { configManager } from '@/lib/admin/config-manager';
import { resolvePolicy } from '@/lib/admin/policy-scopes';
import { hasSessionSecret } from '@/lib/auth/session-secret';
import { MAX_ACCOUNT_SLOTS } from '@/lib/account-utils';
import { mergeStoredTemplates } from '@/lib/template-merge';
import { mergeStoredSubscriptions } from '@/lib/calendar-subscription-sync';
import { storedDataUnavailableResponse } from '@/lib/stored-data-errors';
/** Largest settings blob a client may store (templates and signatures included). */
const SETTINGS_MAX_BYTES = 2 * 1024 * 1024;
@@ -141,6 +142,8 @@ export async function GET(request: NextRequest) {
const message = error instanceof Error ? error.message : 'Unknown error';
const code = (error as NodeJS.ErrnoException).code;
logger.error('Settings load error', { error: message, code });
const unavailable = storedDataUnavailableResponse(error);
if (unavailable) return unavailable;
const classified = classifyError(error);
return NextResponse.json({ error: classified.message }, { status: classified.status });
}
@@ -210,17 +213,22 @@ export async function POST(request: NextRequest) {
// Templates and iCal subscriptions are merged into the stored blob rather
// than replaced, so a stale tab or device cannot drop ones another added.
const stored = await loadUserSettings(accountName, storageServerUrl(serverUrl));
await saveUserSettings(
// In one step against what is stored now: two saves at once (two tabs,
// two replicas) each keep the other's additions.
await updateUserSettings(
accountName,
storageServerUrl(serverUrl),
mergeStoredSubscriptions(stored, mergeStoredTemplates(stored, filteredSettings)),
(stored) => mergeStoredSubscriptions(stored, mergeStoredTemplates(stored, filteredSettings)),
);
return NextResponse.json({ ok: true });
} catch (error) {
const message = error instanceof Error ? error.message : 'Unknown error';
const code = (error as NodeJS.ErrnoException).code;
logger.error('Settings save error', { error: message, code });
// Stored settings that don't decrypt, or no store: write nothing, so
// the user's real settings survive and the client keeps its copy.
const unavailable = storedDataUnavailableResponse(error);
if (unavailable) return unavailable;
const classified = classifyError(error);
return NextResponse.json({ error: classified.message }, { status: classified.status });
}
@@ -252,6 +260,8 @@ export async function DELETE(request: NextRequest) {
const message = error instanceof Error ? error.message : 'Unknown error';
const code = (error as NodeJS.ErrnoException).code;
logger.error('Settings delete error', { error: message, code });
const unavailable = storedDataUnavailableResponse(error);
if (unavailable) return unavailable;
const classified = classifyError(error);
return NextResponse.json({ error: classified.message }, { status: classified.status });
}
@@ -19,6 +19,7 @@ vi.mock('@/lib/telemetry', () => ({
effectiveConsent: async () => ({ source: 'env' }),
loadState: async () => ({}),
saveState: async () => {},
updateState: async () => ({}),
reschedule: async () => {},
}));
@@ -0,0 +1,39 @@
/**
* @vitest-environment node
*/
import { describe, it, expect, vi } from 'vitest';
import { NextRequest } from 'next/server';
// The wizard's telemetry choice used to load the whole state, change it and
// save it back, putting its copy over anything written meanwhile (the
// scheduler leader storing the next heartbeat, on this or another replica).
// It changes only the consent fields of what is stored now.
vi.mock('@/lib/setup/state', () => ({ detectSetupState: () => 'bootstrap' }));
vi.mock('@/lib/setup/session', () => ({ authenticateWizardRequest: async () => true }));
vi.mock('@/lib/admin/config-manager', () => ({
configManager: { ensureLoaded: async () => {}, setAdminConfig: async () => {} },
}));
const stored = vi.hoisted(() => ({ consent: 'off', consentedAt: null as string | null, nextScheduledAt: 'kept' }));
vi.mock('@/lib/telemetry', () => ({
effectiveConsent: async () => ({ source: 'file' }),
loadState: async () => { throw new Error('read-modify-write of the whole state'); },
saveState: async () => { throw new Error('read-modify-write of the whole state'); },
updateState: async (change: (s: typeof stored) => void) => { change(stored); return stored; },
reschedule: async () => {},
}));
import { POST } from '../step/route';
describe('POST /api/setup/step (security, telemetry)', () => {
it('turns consent on against the stored state', async () => {
const res = await POST(new NextRequest('http://localhost/api/setup/step', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ step: 'security', values: { telemetryConsent: 'on' } }),
}));
expect(res.status).toBe(200);
expect(stored).toMatchObject({ consent: 'on', nextScheduledAt: 'kept' });
expect(stored.consentedAt).not.toBeNull();
});
});
+7 -7
View File
@@ -4,7 +4,7 @@ import { authenticateWizardRequest } from '@/lib/setup/session';
import { configManager } from '@/lib/admin/config-manager';
import { CONFIG_ENV_MAP } from '@/lib/admin/types';
import { parseJmapServers } from '@/lib/admin/jmap-servers';
import { effectiveConsent, loadState, saveState, reschedule } from '@/lib/telemetry';
import { effectiveConsent, updateState, reschedule } from '@/lib/telemetry';
import { logger } from '@/lib/logger';
export const dynamic = 'force-dynamic';
@@ -93,12 +93,12 @@ export async function POST(request: NextRequest) {
// A BULWARK_TELEMETRY env var hard-locks the choice; don't fight it.
const { source } = await effectiveConsent();
if (source !== 'env') {
const tstate = await loadState();
tstate.consent = consent;
if (consent === 'on' && !tstate.consentedAt) {
tstate.consentedAt = new Date().toISOString();
}
await saveState(tstate);
const now = new Date().toISOString();
// Against the stored state, so a schedule written meanwhile stays.
await updateState((tstate) => {
tstate.consent = consent;
if (consent === 'on' && !tstate.consentedAt) tstate.consentedAt = now;
});
await reschedule();
}
}
+12 -4
View File
@@ -2,6 +2,7 @@ import { NextRequest, NextResponse } from 'next/server';
import { checkOnce, freshStatus, loadState } from '@/lib/version-check';
import { configManager } from '@/lib/admin/config-manager';
import { canSeeInstanceDetails } from '@/lib/auth/instance-details';
import { storedDataUnavailableResponse } from '@/lib/stored-data-errors';
// Public endpoint that returns the latest cached update status. Fed by the
// background scheduler started in instrumentation.node.ts; in production we
@@ -13,11 +14,18 @@ import { canSeeInstanceDetails } from '@/lib/auth/instance-details';
// requiring a dev-server restart. The 5s upstream timeout in fetchStatus
// caps the worst-case latency added to a dev reload.
export async function GET(request: NextRequest) {
if (process.env.NODE_ENV === 'development') {
await checkOnce({ reason: 'dev-reload' });
let state;
try {
if (process.env.NODE_ENV === 'development') {
await checkOnce({ reason: 'dev-reload' });
}
state = await loadState();
} catch (err) {
// The shared store is down: the banner stays as the browser has it.
const unavailable = storedDataUnavailableResponse(err, { 'Cache-Control': 'no-store' });
if (unavailable) return unavailable;
throw err;
}
const state = await loadState();
// freshStatus drops a status persisted by an earlier build - after an
// upgrade it would still advertise the version we just upgraded from (#913).
let status = freshStatus(state);
+9 -6
View File
@@ -80,16 +80,19 @@ migrateLegacyAdminLayout()
// or with BULWARK_TELEMETRY=on; BULWARK_TELEMETRY=off or an empty
// collector URL keeps it from sending.
// See https://bulwarkmail.org/docs/legal/privacy/telemetry
const { startScheduler, markProcessStart } = await import("./lib/telemetry");
const { startScheduler, stopScheduler, markProcessStart } = await import("./lib/telemetry");
markProcessStart();
await startScheduler();
})
.then(async () => {
// Hourly check against version.telemetry.bulwarkmail.org. Disable with
// BULWARK_UPDATE_CHECK=off or override the endpoint with
// BULWARK_UPDATE_CHECK_URL.
const { startScheduler } = await import("./lib/version-check");
await startScheduler();
const versionCheck = await import("./lib/version-check");
// Both 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: "version-check", start: versionCheck.startScheduler, stop: versionCheck.stopScheduler },
]);
})
.catch((err) => {
console.warn("Admin dashboard init skipped:", err instanceof Error ? err.message : err);
@@ -0,0 +1,120 @@
// @vitest-environment node
import { mkdtempSync, readdirSync, readFileSync, rmSync, statSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { open } from 'node:fs/promises';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { NextRequest } from 'next/server';
import { setStoreForTests } from '@/lib/shared-store';
import { resetLocalStoreForTests } from '@/lib/shared-store/local';
import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store';
// Account archives with a shared store (#1066): one document per owner,
// replaced whole, the same on every replica.
const mocks = vi.hoisted(() => ({ contexts: new Map<number, unknown>(), verify: vi.fn() }));
vi.mock('@/lib/admin/config-manager', () => ({ configManager: {
ensureLoaded: async () => {},
get: (_key: string, fallback: unknown) => fallback,
getPolicy: () => ({ features: { accountArchiveEnabled: true } }),
} }));
vi.mock('@/lib/auth/session-secret', () => ({ hasSessionSecret: () => true }));
vi.mock('next/headers', () => ({ cookies: async () => ({}) }));
vi.mock('@/lib/stalwart/auth-context', () => ({
readStalwartAuthContextFromStore: (_: unknown, slot: number) => mocks.contexts.get(slot) ?? null,
}));
vi.mock('@/lib/auth/verify-jmap-auth', async importOriginal => ({
...await importOriginal<typeof import('@/lib/auth/verify-jmap-auth')>(), verifyJmapAuth: mocks.verify,
}));
import { deleteArchive, loadArchive, saveArchive } from '@/lib/account-archive-storage';
import { GET, PUT } from '@/app/api/account-archive/route';
import type { ArchiveEnvelope } from '@/lib/account-archive';
const owner = { username: 'owner@example.com', serverUrl: 'https://mail.example.com' };
const envelope: ArchiveEnvelope = { version: 1, iterations: 600000, salt: 'A'.repeat(22) + '==', iv: 'A'.repeat(16), ciphertext: 'A'.repeat(43) + '=' };
const other: ArchiveEnvelope = { ...envelope, ciphertext: 'B'.repeat(43) + '=' };
let disks: string;
let store: SeparateDiskStore;
beforeEach(() => {
disks = mkdtempSync(path.join(tmpdir(), 'bw-archive-replicas-'));
vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, 'a'));
store = separateDiskStore();
setStoreForTests(store);
mocks.contexts.clear();
mocks.contexts.set(0, { username: owner.username, serverUrl: owner.serverUrl, authHeader: 'Basic ' + Buffer.from(`${owner.username}:pw`).toString('base64') });
mocks.verify.mockReset().mockResolvedValue(owner.serverUrl);
});
afterEach(() => {
setStoreForTests(null);
resetLocalStoreForTests();
vi.unstubAllEnvs();
store.cleanup();
rmSync(disks, { recursive: true, force: true });
});
function request(method: string, body?: unknown) {
return new NextRequest('https://webmail.example.com/api/account-archive', {
method,
headers: { 'x-archive-username': owner.username, 'x-archive-server': owner.serverUrl, 'content-type': 'application/json', 'sec-fetch-site': 'same-origin' },
...(body === undefined ? {} : { body: JSON.stringify(body) }),
});
}
describe('account archives with a shared store', () => {
it('returns on replica B what replica A stored, and the latest save wins', async () => {
await saveArchive(owner, envelope);
vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, 'b'));
expect(await loadArchive(owner)).toEqual(envelope);
await saveArchive(owner, other);
vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, 'a'));
expect(await loadArchive(owner)).toEqual(other);
await deleteArchive(owner);
expect(await loadArchive(owner)).toBeNull();
});
it('answers 503 while the store is down and keeps what is stored', async () => {
await saveArchive(owner, envelope);
store.down = true;
expect((await PUT(request('PUT', { envelope: other }))).status).toBe(503);
expect((await GET(request('GET'))).status).toBe(503);
store.down = false;
expect(await loadArchive(owner)).toEqual(envelope);
});
});
describe('account archives on a single instance', () => {
it('flush the new file to disk before it replaces the old one', async () => {
setStoreForTests(null);
const handle = await open(path.join(disks, 'probe'), 'w');
const proto = Object.getPrototypeOf(handle) as { sync: () => Promise<void> };
await handle.close();
const sync = vi.spyOn(proto, 'sync');
try {
await saveArchive(owner, envelope);
expect(sync).toHaveBeenCalledTimes(1);
expect(await loadArchive(owner)).toEqual(envelope);
} finally {
sync.mockRestore();
}
});
it('keep today\'s file: <sha256 of the owner>.json under account-archives, mode 0600', async () => {
setStoreForTests(null);
await saveArchive(owner, envelope);
const dir = path.join(disks, 'a', 'account-archives');
const files = readdirSync(dir);
expect(files).toHaveLength(1);
expect(JSON.parse(readFileSync(path.join(dir, files[0]), 'utf8'))).toEqual(envelope);
if (process.platform !== 'win32') {
expect(statSync(path.join(dir, files[0])).mode & 0o777).toBe(0o600);
expect(statSync(dir).mode & 0o777).toBe(0o700);
}
expect(files[0]).toMatch(/^[0-9a-f]{64}\.json$/);
});
});
@@ -0,0 +1,69 @@
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { localCollection } from '@/lib/shared-store/local';
import { fakeSharedStore } from '@/lib/shared-store/__tests__/fake-shared';
import {
StoreUnavailableError,
type DocCollection,
type KvStore,
type LeaseStore,
type SharedStore,
} from '@/lib/shared-store/types';
export type SeparateDiskStore = SharedStore & { down: boolean; root: string; cleanup(): void };
/**
* A fake shared store (mode 'redis') whose documents live in a folder of
* their own, not in the dirs a store names for local mode. Every replica has
* its own disk: a store that still wrote its local files would not find on
* "replica B" (another SETTINGS_DATA_DIR, …) what it saved on "replica A".
* `down` makes every document, kv and lease call fail as an outage would.
*/
export function separateDiskStore(): SeparateDiskStore {
const root = mkdtempSync(path.join(tmpdir(), 'bw-shared-store-'));
const base = fakeSharedStore();
const guard = <T>(run: () => Promise<T>): Promise<T> =>
store.down ? Promise.reject(new StoreUnavailableError('down (test)')) : run();
const kv: KvStore = {
get: (key) => guard(() => base.kv.get(key)),
set: (key, value, options) => guard(() => base.kv.set(key, value, options)),
del: (key) => guard(() => base.kv.del(key)),
take: (key) => guard(() => base.kv.take(key)),
update: (key, fn, options) => guard(() => base.kv.update(key, fn, options)),
};
const leases: LeaseStore = {
acquire: (name, holder, ttlMs) => guard(() => base.leases.acquire(name, holder, ttlMs)),
release: (name, holder) => guard(() => base.leases.release(name, holder)),
holder: (name) => guard(() => base.leases.holder(name)),
};
const store: SeparateDiskStore = {
...base,
kv,
leases,
root,
get down() {
return base.down;
},
set down(value: boolean) {
base.down = value;
},
collection(name, options): DocCollection {
const inner = localCollection(name, { ...options, local: { ...options.local, dir: () => path.join(root, name) } });
return {
name,
get: (id) => guard(() => inner.get(id)),
put: (id, data, putOptions) => guard(() => inner.put(id, data, putOptions)),
update: (id, fn, updateOptions) => guard(() => inner.update(id, fn, updateOptions)),
delete: (id, deleteOptions) => guard(() => inner.delete(id, deleteOptions)),
list: (prefix) => guard(() => inner.list(prefix)),
};
},
cleanup() {
rmSync(root, { recursive: true, force: true });
},
};
return store;
}
@@ -0,0 +1,99 @@
// @vitest-environment node
import { mkdtempSync, readdirSync, rmSync, statSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { setStoreForTests } from '@/lib/shared-store';
import { resetLocalStoreForTests } from '@/lib/shared-store/local';
import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store';
// Users' plugin credentials with a shared store (#1066). The vault used to
// serialise writes with a queue in its own module, which no other process
// (and no other module copy) sees: two writes on two replicas lost one.
vi.mock('@/lib/auth/session-secret', () => ({
getSessionSecret: () => 's'.repeat(64),
hasSessionSecret: () => true,
}));
vi.mock('@/lib/logger', () => ({
logger: { warn: () => {}, error: () => {}, info: () => {}, debug: () => {} },
}));
const user = { username: 'alice@example.org', serverUrl: 'https://mail.example.org' };
const input = (secret: string) => ({ origin: 'https://cloud.example.org', inject: { type: 'basic' } as const, label: 'Cloud', secret });
type Vault = typeof import('@/lib/plugin-secrets/vault');
/** A fresh copy of the vault module, as another process (or bundle) has. */
async function replica(): Promise<Vault> {
vi.resetModules();
return import('@/lib/plugin-secrets/vault');
}
let disks: string;
let store: SeparateDiskStore;
beforeEach(() => {
disks = mkdtempSync(path.join(tmpdir(), 'bw-vault-replicas-'));
vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(disks, 'a'));
store = separateDiskStore();
setStoreForTests(store);
});
afterEach(() => {
setStoreForTests(null);
resetLocalStoreForTests();
vi.unstubAllEnvs();
store.cleanup();
rmSync(disks, { recursive: true, force: true });
});
describe('plugin credentials with a shared store', () => {
it('reads on replica B what replica A stored', async () => {
const a = await replica();
await a.putUserCredential(user, 'nextcloud', 'cloud', input('pw'));
vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(disks, 'b'));
const b = await replica();
expect(await b.getUserCredential(user, 'nextcloud', 'cloud')).toMatchObject({ secret: 'pw' });
});
it('keeps both of two credentials saved at once through two replicas', async () => {
const a = await replica();
const b = await replica();
await Promise.all([
a.putUserCredential(user, 'nextcloud', 'cloud', input('one')),
b.putUserCredential(user, 'paperless', 'docs', input('two')),
a.putUserCredential(user, 'radicale', 'dav', input('three')),
b.putUserCredential(user, 'immich', 'photos', input('four')),
]);
const ids = (await a.listUserCredentials(user)).map((c) => c.pluginId).sort();
expect(ids).toEqual(['immich', 'nextcloud', 'paperless', 'radicale']);
});
it('answers with StoreUnavailableError while the store is down and writes nothing', async () => {
const vault = await replica();
await vault.putUserCredential(user, 'nextcloud', 'cloud', input('pw'));
store.down = true;
await expect(vault.putUserCredential(user, 'paperless', 'docs', input('x'))).rejects.toThrow('down (test)');
await expect(vault.listUserCredentials(user)).rejects.toThrow('down (test)');
store.down = false;
expect((await vault.listUserCredentials(user)).map((c) => c.pluginId)).toEqual(['nextcloud']);
});
});
describe('plugin credentials on a single instance', () => {
it('keep today\'s file: <sha256>.enc in the vault dir, mode 0600, gone when the last one goes', async () => {
setStoreForTests(null);
const vault = await replica();
await vault.putUserCredential(user, 'nextcloud', 'cloud', input('pw'));
const dir = path.join(disks, 'a');
const files = readdirSync(dir);
expect(files).toEqual([`${vault.userKey(user)}.enc`]);
if (process.platform !== 'win32') expect(statSync(path.join(dir, files[0])).mode & 0o777).toBe(0o600);
await vault.deleteUserCredentials(user, 'nextcloud');
expect(readdirSync(dir)).toEqual([]);
});
});
+381
View File
@@ -0,0 +1,381 @@
// @vitest-environment node
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { setTimeout as realSleep } from 'node:timers/promises';
import { NextRequest } from 'next/server';
import { replicaId, setStoreForTests } from '@/lib/shared-store';
import { resetLocalStoreForTests } from '@/lib/shared-store/local';
import { INIT_EPOCH, announceChange, pollChanges, resetChangesForTests } from '@/lib/shared-store/changes';
import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store';
// The background jobs with several replicas (#1066): only the holder of the
// scheduler-leader lease runs them, another replica takes over when the
// holder's lease runs out and keeps the stored schedule, and a scheduled
// heartbeat is claimed with a compare-and-set so two leaders can't both send.
const log = vi.hoisted(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }));
vi.mock('@/lib/logger', () => ({ logger: log }));
vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false }));
const buildPayload = vi.hoisted(() => vi.fn(async () => ({ schema: '1' })));
vi.mock('@/lib/telemetry/payload', () => ({ buildPayload, markProcessStart: () => {} }));
const sent = vi.hoisted(() => vi.fn(async (_url: string, _init?: unknown) => new Response('ok')));
vi.mock('@/lib/telemetry/endpoint-guard', () => ({
resolveEndpointAllowed: async () => ({ ok: true }),
fetchTelemetryTarget: (url: string, init?: unknown) => sent(url, init),
}));
vi.mock('@/lib/admin/session', () => ({
requireAdminAuth: async () => ({ session: { sub: 'admin' } }),
getClientIP: () => '127.0.0.1',
}));
vi.mock('@/lib/admin/audit', () => ({ auditLog: async () => {} }));
import {
LEADER_LEASE, isSchedulerLeader, resetSchedulerLeaderForTests, schedulerLeaderSettled, startSchedulerLeader, stopSchedulerLeader,
} from '@/lib/scheduler-leader';
import { loadState, saveState, sendOnce, startScheduler, stopScheduler, updateState } from '@/lib/telemetry';
import { POST as telemetryRoute } from '@/app/api/admin/telemetry/route';
const MIN = 60_000;
const COLLECTOR = 'https://collector.example.org/v1/heartbeat';
const jobs = [{ name: 'telemetry', start: startScheduler, stop: stopScheduler }];
let dir: string;
let store: SeparateDiskStore;
beforeEach(() => {
dir = mkdtempSync(path.join(tmpdir(), 'bw-leader-'));
vi.stubEnv('TELEMETRY_DATA_DIR', path.join(dir, 'telemetry'));
vi.stubEnv('BULWARK_TELEMETRY', '');
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'setInterval', 'clearInterval', 'Date'] });
vi.setSystemTime(new Date('2026-10-31T12:00:00.000Z'));
store = separateDiskStore();
setStoreForTests(store);
sent.mockClear();
log.info.mockClear();
});
afterEach(async () => {
await stopSchedulerLeader();
resetSchedulerLeaderForTests();
await stopScheduler();
resetChangesForTests();
vi.useRealTimers();
setStoreForTests(null);
resetLocalStoreForTests();
vi.unstubAllEnvs();
store.cleanup();
rmSync(dir, { recursive: true, force: true });
});
async function consentOn(nextScheduledAt: string | null) {
await saveState({ ...(await loadState()), consent: 'on', endpoint: COLLECTOR, nextScheduledAt });
}
/** Let time pass; the leader's tries and job starts do real file I/O, so wait for them too. */
async function pass(ms: number) {
for (let passed = 0; passed < ms; passed += 10_000) {
await vi.advanceTimersByTimeAsync(Math.min(10_000, ms - passed));
await schedulerLeaderSettled();
}
}
/** Another replica holding (and renewing) the lease while `ms` pass here. */
async function otherReplicaLeads(ms: number) {
for (let passed = 0; passed < ms; passed += 10_000) {
expect(await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000)).toBe(true);
await pass(10_000);
}
}
/** Let a heartbeat a timer set off finish its real file I/O, so "not sent" means it. */
async function ioSettles() {
await realSleep(200);
}
/** The heartbeat the timer set off does real I/O before it sends. */
async function sends(times: number) {
await vi.waitFor(() => expect(sent).toHaveBeenCalledTimes(times));
}
describe('scheduler leader with a shared store', () => {
it('only the holder sends; when its lease runs out another replica takes over and keeps the schedule', async () => {
const slot = new Date(Date.now() + 20 * MIN).toISOString();
await consentOn(slot);
await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000);
await startSchedulerLeader(jobs);
expect(isSchedulerLeader()).toBe(false);
await otherReplicaLeads(5 * MIN);
await ioSettles();
expect(isSchedulerLeader()).toBe(false);
// replica-a is gone: its lease expires after 30 s, the next try takes it.
await pass(40_000);
expect(isSchedulerLeader()).toBe(true);
expect(log.info).toHaveBeenCalledWith('scheduler: leading', expect.objectContaining({ replica: expect.any(String) }));
expect((await loadState()).nextScheduledAt).toBe(slot);
expect(sent).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(new Date(slot).getTime() - Date.now() + 1000);
await sends(1);
});
it('stops its timers when it can no longer reach the store', async () => {
const slot = new Date(Date.now() + 5 * MIN).toISOString();
await consentOn(slot);
await startSchedulerLeader(jobs);
expect(isSchedulerLeader()).toBe(true);
store.down = true;
await pass(10_000);
expect(isSchedulerLeader()).toBe(false);
expect(log.info).toHaveBeenCalledWith('scheduler: stopped leading', expect.objectContaining({ reason: expect.stringContaining('shared store') }));
// Past the slot with the store still down: no heartbeat from here.
await pass(6 * MIN);
await ioSettles();
expect(sent).not.toHaveBeenCalled();
expect(isSchedulerLeader()).toBe(false);
});
it('stops leading when the store lost its data', async () => {
await store.epochs.bump(INIT_EPOCH);
await consentOn(new Date(Date.now() + 5 * MIN).toISOString());
await startSchedulerLeader(jobs);
expect(isSchedulerLeader()).toBe(true);
// Flushed: the init epoch is gone, and the next poll notices.
resetLocalStoreForTests();
await pollChanges();
await pass(10_000);
expect(isSchedulerLeader()).toBe(false);
expect(log.info).toHaveBeenCalledWith('scheduler: stopped leading', expect.objectContaining({ reason: 'the shared store lost its data' }));
});
it('stops its timers when another replica has taken the lease', async () => {
const slot = new Date(Date.now() + 5 * MIN).toISOString();
await consentOn(slot);
await startSchedulerLeader(jobs);
expect(isSchedulerLeader()).toBe(true);
// This replica's lease is gone (it stalled past the TTL) and replica-a took it.
await store.leases.release(LEADER_LEASE, replicaId());
await otherReplicaLeads(6 * MIN);
await ioSettles();
expect(isSchedulerLeader()).toBe(false);
expect(sent).not.toHaveBeenCalled();
});
it('does not arm a new timer after a send that was in flight when the lead moved', async () => {
const slot = new Date(Date.now() + MIN).toISOString();
await consentOn(slot);
let finish!: (res: Response) => void;
sent.mockImplementationOnce(() => new Promise<Response>((resolve) => { finish = resolve; }));
await startSchedulerLeader(jobs);
await vi.advanceTimersByTimeAsync(MIN + 1000);
await sends(1);
// The lead moves to replica-a while the heartbeat hangs.
await store.leases.release(LEADER_LEASE, replicaId());
expect(await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000)).toBe(true);
await pass(10_000);
expect(isSchedulerLeader()).toBe(false);
await stopSchedulerLeader(); // no more tries from this replica
finish(new Response('ok'));
await ioSettles();
await vi.advanceTimersByTimeAsync(27 * 60 * MIN);
await ioSettles();
expect(sent).toHaveBeenCalledTimes(1);
expect(vi.getTimerCount()).toBe(0);
});
it('stops a job whose restart finished after the replica stepped down', async () => {
let running = false;
let starts = 0;
let finishStart: (() => void) | null = null;
const job = {
name: 'telemetry',
start: async () => {
if (++starts === 2) await new Promise<void>((resolve) => { finishStart = resolve; });
running = true;
},
stop: async () => { running = false; },
};
await startSchedulerLeader([job]);
expect(running).toBe(true);
// Consent changed: the restart's start hangs while the lead is lost.
const restarted = announceChange('telemetry');
await vi.waitFor(() => expect(finishStart).not.toBeNull());
store.down = true;
await pass(10_000);
expect(isSchedulerLeader()).toBe(false);
finishStart!();
await restarted;
expect(running).toBe(false);
});
it('version check: does not arm a new timer after a check that was in flight when it stopped', async () => {
vi.stubEnv('VERSION_CHECK_DATA_DIR', path.join(dir, 'version-check'));
vi.stubEnv('BULWARK_UPDATE_CHECK', '');
vi.stubEnv('BULWARK_UPDATE_CHECK_URL', 'https://version.example/');
vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.20.0');
let finish!: (res: Response) => void;
const fetchMock = vi.fn(() => new Promise<Response>((resolve) => { finish = resolve; }));
vi.stubGlobal('fetch', fetchMock);
const versionCheck = await import('@/lib/version-check');
try {
await versionCheck.startScheduler();
await vi.advanceTimersByTimeAsync(31_000);
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1));
await versionCheck.stopScheduler();
finish(new Response('{}', { status: 500 }));
await ioSettles();
await vi.advanceTimersByTimeAsync(2 * 60 * MIN);
await ioSettles();
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(vi.getTimerCount()).toBe(0);
} finally {
await versionCheck.stopScheduler();
vi.unstubAllGlobals();
}
});
it('sends once when two leaders fire for the same slot, and not again after it', async () => {
const slot = new Date(Date.now() - 1000).toISOString();
await consentOn(slot);
const results = await Promise.all([
sendOnce({ reason: 'scheduled', slot }),
sendOnce({ reason: 'scheduled', slot }),
]);
expect(results.filter((r) => r.ok)).toHaveLength(1);
expect(sent).toHaveBeenCalledTimes(1);
// The paused leader wakes up an hour later, still holding that slot.
vi.advanceTimersByTime(60 * MIN);
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false);
expect(sent).toHaveBeenCalledTimes(1);
});
it('sends once when the old leader stalled past its slot and the new one picked a later slot', async () => {
const oldSlot = new Date(Date.now() - 1000).toISOString();
await consentOn(oldSlot);
// The new leader found the slot passed and scheduled an hour out.
const newSlot = new Date(Date.now() + 60 * MIN).toISOString();
// The stalled leader wakes up and sends for its slot first.
expect((await sendOnce({ reason: 'scheduled', slot: oldSlot })).ok).toBe(true);
vi.advanceTimersByTime(61 * MIN);
expect((await sendOnce({ reason: 'scheduled', slot: newSlot })).ok).toBe(false);
expect(sent).toHaveBeenCalledTimes(1);
});
it('sends once when a leader died after sending but before storing the next slot', async () => {
const slot = new Date(Date.now() - 1000).toISOString();
await consentOn(slot);
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
// nextScheduledAt still names the slot just sent; the next leader
// finds it passed and fires an hour later.
vi.advanceTimersByTime(60 * MIN);
expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(false);
expect(sent).toHaveBeenCalledTimes(1);
});
it('still sends the next scheduled heartbeat after a manual "send now"', async () => {
const slot = new Date(Date.now() - 1000).toISOString();
await consentOn(slot);
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
vi.advanceTimersByTime(20 * 60 * MIN);
expect((await sendOnce({ reason: 'admin-manual' })).ok).toBe(true);
vi.advanceTimersByTime(2 * 60 * MIN);
expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(true);
expect(sent).toHaveBeenCalledTimes(3);
});
it('gives the slot back when the payload cannot be built', async () => {
const slot = new Date(Date.now() - 1000).toISOString();
await consentOn(slot);
buildPayload.mockRejectedValueOnce(new Error('store down'));
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false);
expect((await loadState()).lastSentAt).toBeNull();
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
expect(sent).toHaveBeenCalledTimes(1);
});
it('gives the slot back when the send fails', async () => {
const slot = new Date(Date.now() - 1000).toISOString();
await consentOn(slot);
sent.mockResolvedValueOnce(new Response('no', { status: 500 }));
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false);
expect((await loadState()).lastSentAt).toBeNull();
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
});
it('sends on "send now" from a replica that does not lead, even right after a heartbeat', async () => {
await consentOn(null);
await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000);
await startSchedulerLeader(jobs);
expect(isSchedulerLeader()).toBe(false);
await updateState((state) => { state.lastSentAt = new Date().toISOString(); });
const res = await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ action: 'send-now' }),
}));
expect(res.status).toBe(200);
expect(sent).toHaveBeenCalledTimes(1);
});
it('starts the heartbeat on the leader when consent is turned on through another replica', async () => {
await store.epochs.bump(INIT_EPOCH);
await startSchedulerLeader(jobs);
expect(isSchedulerLeader()).toBe(true);
expect((await loadState()).nextScheduledAt).toBeNull();
// The other replica's admin route: write, then bump the epoch.
await updateState((state) => { state.consent = 'on'; state.endpoint = COLLECTOR; });
await store.epochs.bump('telemetry');
await pollChanges();
expect((await loadState()).nextScheduledAt).not.toBeNull();
await vi.advanceTimersByTimeAsync(61 * MIN);
await sends(1);
});
});
describe('scheduler leader on a single instance', () => {
it('sends a scheduled heartbeat an hour after the last one, as before', async () => {
setStoreForTests(null);
const slot = new Date(Date.now() - 1000).toISOString();
await consentOn(slot);
expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
vi.advanceTimersByTime(60 * MIN);
expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(true);
expect((await loadState()).lastScheduledSentAt).toBeUndefined();
});
it('leads at once and restarts the heartbeat when consent changes in this process', async () => {
setStoreForTests(null);
await startSchedulerLeader(jobs);
expect(isSchedulerLeader()).toBe(true);
expect(log.info).toHaveBeenCalledWith('scheduler: leading', expect.anything());
const res = await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ action: 'set-consent', consent: 'on' }),
}));
expect(res.status).toBe(200);
expect(vi.getTimerCount()).toBe(1);
await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ action: 'set-consent', consent: 'off' }),
}));
expect(vi.getTimerCount()).toBe(0);
});
});
@@ -0,0 +1,62 @@
// @vitest-environment node
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
// Next loads lib/telemetry and lib/version-check once per bundle. The timer
// sat in a module-level variable, so the admin route's reschedule() acted on
// its own copy: the timer instrumentation started kept running next to the
// new one (two heartbeats), and turning consent off didn't stop it.
vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false }));
let dir: string;
beforeEach(() => {
dir = mkdtempSync(path.join(tmpdir(), 'bw-scheduler-copies-'));
vi.stubEnv('TELEMETRY_DATA_DIR', path.join(dir, 'telemetry'));
vi.stubEnv('VERSION_CHECK_DATA_DIR', path.join(dir, 'version-check'));
vi.stubEnv('BULWARK_TELEMETRY', 'on');
vi.stubEnv('BULWARK_UPDATE_CHECK', '');
vi.stubEnv('BULWARK_UPDATE_CHECK_URL', 'https://version.example/');
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] });
});
afterEach(() => {
vi.useRealTimers();
vi.unstubAllEnvs();
vi.resetModules();
rmSync(dir, { recursive: true, force: true });
});
async function copy<T>(load: () => Promise<T>): Promise<T> {
vi.resetModules();
return load();
}
describe('scheduler timers across module copies', () => {
it('telemetry: a stop from another copy clears the running timer, a restart leaves one', async () => {
const instrumentation = await copy(() => import('@/lib/telemetry/sender'));
const route = await copy(() => import('@/lib/telemetry/sender'));
await instrumentation.startScheduler();
expect(vi.getTimerCount()).toBe(1);
await route.startScheduler();
expect(vi.getTimerCount()).toBe(1);
await route.stopScheduler();
expect(vi.getTimerCount()).toBe(0);
});
it('version check: the same', async () => {
const instrumentation = await copy(() => import('@/lib/version-check/sender'));
const route = await copy(() => import('@/lib/version-check/sender'));
await instrumentation.startScheduler();
expect(vi.getTimerCount()).toBe(1);
await route.startScheduler();
expect(vi.getTimerCount()).toBe(1);
await route.stopScheduler();
expect(vi.getTimerCount()).toBe(0);
});
});
@@ -39,6 +39,9 @@ vi.mock('@/lib/settings-sync', () => ({
saveUserSettings: async (username: string, serverUrl: string, settings: Record<string, unknown>) => {
store.set(`${username}|${serverUrl}`, settings);
},
updateUserSettings: async (username: string, serverUrl: string, change: (stored: Record<string, unknown> | null) => Record<string, unknown>) => {
store.set(`${username}|${serverUrl}`, change(store.get(`${username}|${serverUrl}`) ?? null));
},
deleteUserSettings: async (username: string, serverUrl: string) => { store.delete(`${username}|${serverUrl}`); },
}));
@@ -38,6 +38,9 @@ vi.mock('@/lib/settings-sync', () => ({
saveUserSettings: async (username: string, serverUrl: string, settings: Record<string, unknown>) => {
store.set(`${username}|${serverUrl}`, settings);
},
updateUserSettings: async (username: string, serverUrl: string, change: (stored: Record<string, unknown> | null) => Record<string, unknown>) => {
store.set(`${username}|${serverUrl}`, change(store.get(`${username}|${serverUrl}`) ?? null));
},
deleteUserSettings: async () => {},
}));
+176
View File
@@ -0,0 +1,176 @@
// @vitest-environment node
import { mkdtempSync, readdirSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { NextRequest } from 'next/server';
import { setStoreForTests } from '@/lib/shared-store';
import { resetLocalStoreForTests } from '@/lib/shared-store/local';
import { SETTINGS_COLLECTION, settingsId } from '@/lib/settings-sync';
import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store';
// Synced settings with a shared store (#1066): every replica reads and
// writes the one store, never its own disk.
vi.mock('@/lib/auth/session-secret', () => ({
getSessionSecret: () => 's'.repeat(64),
hasSessionSecret: () => true,
}));
vi.mock('@/lib/logger', () => ({
logger: { warn: () => {}, error: () => {}, info: () => {}, debug: () => {} },
}));
vi.mock('@/lib/admin/config-manager', () => ({
configManager: {
ensureLoaded: async () => {},
get: (key: string, fallback: unknown) => (key === 'settingsSyncEnabled' ? true : fallback),
getPolicy: () => ({ restrictions: {} }),
},
}));
const jar = new Map<string, string>();
vi.mock('next/headers', () => ({
cookies: async () => ({
get: (name: string) => (jar.has(name) ? { name, value: jar.get(name)! } : undefined),
set: (name: string, value: string) => { jar.set(name, value); },
delete: (name: string) => { jar.delete(name); },
getAll: () => [...jar].map(([name, value]) => ({ name, value })),
}),
}));
const SERVER = 'https://mail.example.org';
const USER = 'alice@example.org';
let disks: string;
let store: SeparateDiskStore;
/** Point this process at one replica's own data dir. */
function onReplica(name: 'a' | 'b') {
vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, name, 'settings'));
}
beforeEach(async () => {
disks = mkdtempSync(path.join(tmpdir(), 'bw-settings-replicas-'));
store = separateDiskStore();
setStoreForTests(store);
vi.stubEnv('SETTINGS_SYNC_ENABLED', 'true');
onReplica('a');
jar.clear();
const { setStalwartAuthContextInStore } = await import('@/lib/stalwart/auth-context');
setStalwartAuthContextInStore({
get: (n: string) => (jar.has(n) ? { name: n, value: jar.get(n)! } : undefined),
set: (n: string, v: string) => { jar.set(n, v); },
delete: (n: string) => { jar.delete(n); },
} as never, 0, { serverUrl: SERVER, authHeader: 'Bearer t', username: USER });
});
afterEach(() => {
setStoreForTests(null);
resetLocalStoreForTests();
vi.unstubAllEnvs();
store.cleanup();
rmSync(disks, { recursive: true, force: true });
});
function template(id: string) {
const at = '2026-10-01T00:00:00.000Z';
return { id, name: id, subject: '', body: '', isHTML: false, category: '', isFavorite: false, createdAt: at, updatedAt: at };
}
async function push(settings: Record<string, unknown>) {
const { POST } = await import('@/app/api/settings/route');
return POST(new NextRequest('https://webmail.example/api/settings', {
method: 'POST',
headers: { 'sec-fetch-site': 'same-origin', 'content-type': 'application/json' },
body: JSON.stringify({ username: USER, serverUrl: SERVER, settings }),
}));
}
async function fetchSettings() {
const { GET } = await import('@/app/api/settings/route');
return GET(new NextRequest('https://webmail.example/api/settings', {
headers: { 'x-settings-username': USER, 'x-settings-server': SERVER },
}));
}
function storedDoc() {
return store.collection('settings', SETTINGS_COLLECTION).get(settingsId(USER, SERVER));
}
describe('settings with a shared store', () => {
it('loads on replica B what was saved through replica A', async () => {
expect((await push({ theme: 'dark' })).status).toBe(200);
onReplica('b');
const res = await fetchSettings();
expect(res.status).toBe(200);
expect((await res.json()).settings).toEqual({ theme: 'dark' });
});
it('deletes them for every replica', async () => {
await push({ theme: 'dark' });
onReplica('b');
const { DELETE } = await import('@/app/api/settings/route');
const res = await DELETE(new NextRequest('https://webmail.example/api/settings', {
method: 'DELETE',
headers: { 'sec-fetch-site': 'same-origin', 'content-type': 'application/json' },
body: JSON.stringify({ username: USER, serverUrl: SERVER }),
}));
expect(res.status).toBe(200);
onReplica('a');
expect((await (await fetchSettings()).json()).settings).toBeNull();
});
it('moves a stored document that does not decrypt aside in the store before a save, keeping its bytes', async () => {
const garbage = Buffer.alloc(64, 7);
const id = settingsId(USER, SERVER);
await store.collection('settings', SETTINGS_COLLECTION).put(id, garbage);
expect((await (await fetchSettings()).json()).settings).toBeNull();
expect((await push({ theme: 'light' })).status).toBe(200);
expect((await (await fetchSettings()).json()).settings).toEqual({ theme: 'light' });
// In the store, not on this replica's disk.
const aside = readdirSync(path.join(store.root, 'settings')).filter((f) => f.startsWith(`${id}.unreadable-`));
expect(aside).toHaveLength(1);
const asideId = aside[0].slice(0, -'.enc'.length);
expect((await store.collection('settings', SETTINGS_COLLECTION).get(asideId))?.data.equals(garbage)).toBe(true);
});
it('answers 503 and writes nothing while the store is down', async () => {
await push({ theme: 'dark' });
const before = (await storedDoc())!.data;
store.down = true;
expect((await push({ theme: 'light' })).status).toBe(503);
expect((await fetchSettings()).status).toBe(503);
store.down = false;
expect((await storedDoc())!.data.equals(before)).toBe(true);
});
});
// Two saves at once, each from a tab that added a template the other
// doesn't know yet. The route used to load, merge and save without a lock,
// so the second save put its merge over the first: one template was lost,
// with two tabs on one instance as well as across replicas.
async function twoConcurrentSaves() {
await push({ templates: [], deletedTemplateIds: {} });
const [a, b] = await Promise.all([
push({ templates: [template('from-a')], deletedTemplateIds: {} }),
push({ templates: [template('from-b')], deletedTemplateIds: {} }),
]);
expect([a.status, b.status]).toEqual([200, 200]);
const saved = (await (await fetchSettings()).json()).settings as { templates: { id: string }[] };
return saved.templates.map((t) => t.id).sort();
}
describe('two saves at once', () => {
it('keep both templates with a shared store', async () => {
expect(await twoConcurrentSaves()).toEqual(['from-a', 'from-b']);
});
it('keep both templates on a single instance', async () => {
setStoreForTests(null);
expect(await twoConcurrentSaves()).toEqual(['from-a', 'from-b']);
});
});
@@ -0,0 +1,212 @@
// @vitest-environment node
import { mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { NextRequest } from 'next/server';
// A settings file or plugin vault that exists but doesn't decrypt (written
// with another session secret) used to read as "nothing stored", and the
// next save wrote over it: the user's real data was gone for good. Now the
// save first moves it aside (`<file>.unreadable-<time>`, bytes unchanged)
// and goes on as if nothing was stored, so the user can save again and the
// old data can still be restored with the old secret.
let secret = 'a'.repeat(64);
vi.mock('@/lib/auth/session-secret', () => ({
getSessionSecret: () => secret,
hasSessionSecret: () => true,
}));
vi.mock('@/lib/logger', () => ({
logger: { warn: () => {}, error: () => {}, info: () => {}, debug: () => {} },
}));
vi.mock('@/lib/admin/config-manager', () => ({
configManager: {
ensureLoaded: async () => {},
get: (key: string, fallback: unknown) => (key === 'settingsSyncEnabled' ? true : fallback),
getPolicy: () => ({ restrictions: {}, features: { pluginsEnabled: true } }),
},
}));
vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false, thirdPartyRefusal: () => null }));
// One installed plugin with a credential the admin set up.
vi.mock('@/lib/admin/plugin-registry', () => ({
getPlugin: async (id: string) => (id === 'paperless'
? { id, enabled: true, credentials: { shared: { scope: 'admin', label: 'Paperless', inject: { type: 'bearer' }, origin: 'https://paperless.example.com', secretKey: 'token' } } }
: null),
}));
vi.mock('@/lib/admin/plugin-config', () => ({ getPluginConfig: async () => ({ token: 'admin-token' }) }));
const jar = new Map<string, string>();
vi.mock('next/headers', () => ({
cookies: async () => ({
get: (name: string) => (jar.has(name) ? { name, value: jar.get(name)! } : undefined),
set: (name: string, value: string) => { jar.set(name, value); },
delete: (name: string) => { jar.delete(name); },
getAll: () => [...jar].map(([name, value]) => ({ name, value })),
}),
}));
const SERVER = 'https://mail.example.org';
const USER = 'alice@example.org';
vi.mock('@/lib/stalwart/credentials', () => ({
getStalwartCredentials: async () => ({ username: USER, accountName: USER, serverUrl: SERVER, authHeader: 'x' }),
}));
vi.mock('@/lib/plugin-secrets/known-plugins', () => ({
declaredUserCredential: async () => ({ scope: 'user', label: 'Nextcloud', inject: { type: 'basic' } }),
}));
let dataDir: string;
beforeEach(async () => {
dataDir = mkdtempSync(path.join(tmpdir(), 'bw-unreadable-'));
vi.stubEnv('SETTINGS_DATA_DIR', path.join(dataDir, 'settings'));
vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(dataDir, 'plugin-secrets'));
vi.stubEnv('SETTINGS_SYNC_ENABLED', 'true');
secret = 'a'.repeat(64);
jar.clear();
});
afterEach(() => {
vi.unstubAllEnvs();
rmSync(dataDir, { recursive: true, force: true });
});
/** Sign in after the secret change: the context cookie is sealed with the current secret. */
async function signIn() {
const { setStalwartAuthContextInStore } = await import('@/lib/stalwart/auth-context');
setStalwartAuthContextInStore({
get: (n: string) => (jar.has(n) ? { name: n, value: jar.get(n)! } : undefined),
set: (n: string, v: string) => { jar.set(n, v); },
delete: (n: string) => { jar.delete(n); },
} as never, 0, { serverUrl: SERVER, authHeader: 'Bearer t', username: USER });
}
function files(dir: string): string[] {
return readdirSync(dir).sort();
}
function onlyFile(dir: string): string {
const [file] = files(dir).filter((name) => name.endsWith('.enc'));
return path.join(dir, file);
}
function asideOf(dir: string, file: string): string | undefined {
const name = files(dir).find((n) => n.startsWith(`${path.basename(file)}.unreadable-`));
return name && path.join(dir, name);
}
describe('settings that do not decrypt', () => {
it('read as nothing stored, and a file too short to hold any as well', async () => {
const { loadUserSettings, saveUserSettings } = await import('@/lib/settings-sync');
expect(await loadUserSettings(USER, SERVER)).toBeNull();
await saveUserSettings(USER, SERVER, { theme: 'dark' });
secret = 'b'.repeat(64);
expect(await loadUserSettings(USER, SERVER)).toBeNull();
writeFileSync(onlyFile(path.join(dataDir, 'settings')), Buffer.alloc(10));
expect(await loadUserSettings(USER, SERVER)).toBeNull();
});
it('are moved aside unchanged by the next save, which then goes through', async () => {
const { saveUserSettings, loadUserSettings } = await import('@/lib/settings-sync');
await saveUserSettings(USER, SERVER, { theme: 'dark', templates: [{ id: 'a' }] });
const dir = path.join(dataDir, 'settings');
const file = onlyFile(dir);
const before = readFileSync(file);
secret = 'b'.repeat(64);
await signIn();
const { GET, POST } = await import('@/app/api/settings/route');
const got = await GET(new NextRequest('https://webmail.example/api/settings', {
headers: { 'x-settings-username': USER, 'x-settings-server': SERVER },
}));
expect(got.status).toBe(200);
expect((await got.json()).settings).toBeNull();
const saved = await POST(new NextRequest('https://webmail.example/api/settings', {
method: 'POST',
headers: { 'sec-fetch-site': 'same-origin', 'content-type': 'application/json' },
body: JSON.stringify({ username: USER, serverUrl: SERVER, settings: { theme: 'light' } }),
}));
expect(saved.status).toBe(200);
expect(await loadUserSettings(USER, SERVER)).toEqual({ theme: 'light' });
const aside = asideOf(dir, file);
expect(aside).toBeDefined();
expect(readFileSync(aside!).equals(before)).toBe(true);
// With the old secret back, the moved file still holds the old settings.
secret = 'a'.repeat(64);
writeFileSync(file, readFileSync(aside!));
expect(await loadUserSettings(USER, SERVER)).toEqual({ theme: 'dark', templates: [{ id: 'a' }] });
});
it('a file too short to hold any is written over, not moved', async () => {
const { updateUserSettings, loadUserSettings, settingsId } = await import('@/lib/settings-sync');
const dir = path.join(dataDir, 'settings');
mkdirSync(dir, { recursive: true });
writeFileSync(path.join(dir, `${settingsId(USER, SERVER)}.enc`), Buffer.alloc(0));
await updateUserSettings(USER, SERVER, (stored) => ({ ...(stored ?? {}), theme: 'dark' }));
expect(files(dir)).toEqual([`${settingsId(USER, SERVER)}.enc`]);
expect(await loadUserSettings(USER, SERVER)).toEqual({ theme: 'dark' });
});
});
describe('a plugin vault that does not decrypt', () => {
const user = { username: USER, serverUrl: SERVER };
const input = { origin: 'https://cloud.example.org', inject: { type: 'basic' } as const, label: 'Nextcloud', secret: 'pw' };
it('reads as empty, and the next write moves it aside unchanged and goes through', async () => {
const { listUserCredentials, putUserCredential } = await import('@/lib/plugin-secrets/vault');
expect(await listUserCredentials(user)).toEqual([]);
await putUserCredential(user, 'nextcloud', 'cloud', input);
const dir = path.join(dataDir, 'plugin-secrets');
const file = onlyFile(dir);
const before = readFileSync(file);
secret = 'b'.repeat(64);
expect(await listUserCredentials(user)).toEqual([]);
await putUserCredential(user, 'other', 'x', input);
expect((await listUserCredentials(user)).map((c) => c.pluginId)).toEqual(['other']);
const aside = asideOf(dir, file);
expect(aside).toBeDefined();
expect(readFileSync(aside!).equals(before)).toBe(true);
});
it('lets the routes work again: list, connect, disconnect, and the proxy finds no credential', async () => {
const { putUserCredential } = await import('@/lib/plugin-secrets/vault');
await putUserCredential(user, 'nextcloud', 'cloud', input);
secret = 'b'.repeat(64);
const { GET, POST, DELETE } = await import('@/app/api/plugin-secrets/route');
const { POST: proxy } = await import('@/app/api/plugin-proxy/route');
const same = { 'sec-fetch-site': 'same-origin' };
const listed = await GET(new NextRequest('http://localhost/api/plugin-secrets?plugin=paperless'));
expect(listed.status).toBe(200);
expect(await listed.json()).toMatchObject({ credentials: [], admin: [{ credentialId: 'shared', configured: true }] });
const proxied = await proxy(new NextRequest(
'http://localhost/api/plugin-proxy?plugin=nextcloud&credential=cloud&url=https%3A%2F%2Fcloud.example.org%2F',
{ method: 'POST', headers: same },
));
expect(proxied.status).toBe(404);
const post = await POST(new NextRequest('http://localhost/api/plugin-secrets', {
method: 'POST',
headers: same,
body: JSON.stringify({ pluginId: 'nextcloud', credentialId: 'cloud', label: 'x', origin: 'https://cloud.example.org', inject: { type: 'basic' }, secret: 's' }),
}));
expect(post.status).toBe(200);
const del = await DELETE(new NextRequest('http://localhost/api/plugin-secrets?plugin=nextcloud', { method: 'DELETE', headers: same }));
expect(del.status).toBe(200);
});
it('resolves an admin credential without reading the user vault', async () => {
const { resolveCredential } = await import('@/lib/plugin-secrets/proxy');
// A vault that can't even be read (a folder where the file should be).
const { userKey } = await import('@/lib/plugin-secrets/vault');
mkdirSync(path.join(dataDir, 'plugin-secrets', `${userKey(user)}.enc`), { recursive: true });
expect(await resolveCredential(user, 'paperless', 'shared')).toMatchObject({ scope: 'admin', secret: 'admin-token' });
await expect(resolveCredential(user, 'nextcloud', 'cloud')).rejects.toThrow();
});
});
@@ -0,0 +1,58 @@
// @vitest-environment node
import { mkdtempSync, readFileSync, rmSync, writeFileSync, mkdirSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
// A telemetry state file that couldn't be read (a torn write, a passing I/O
// error) used to be replaced with the defaults, which turned the admin's
// consent off for good. Now the read answers with the defaults and leaves
// the file alone.
vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false }));
let dir: string;
beforeEach(() => {
dir = mkdtempSync(path.join(tmpdir(), 'bw-telemetry-state-'));
vi.stubEnv('TELEMETRY_DATA_DIR', dir);
vi.stubEnv('BULWARK_TELEMETRY', '');
});
afterEach(() => {
vi.unstubAllEnvs();
rmSync(dir, { recursive: true, force: true });
});
describe('telemetry state that cannot be read', () => {
it('keeps the stored consent', async () => {
const { effectiveConsent, loadState } = await import('@/lib/telemetry/state');
const good = JSON.stringify({ consent: 'on', endpoint: 'https://collector.example.org/', consentedAt: '2026-10-01T00:00:00.000Z', lastSentAt: null, nextScheduledAt: null });
mkdirSync(dir, { recursive: true });
const file = path.join(dir, 'state.json');
writeFileSync(file, good.slice(0, 20));
expect((await loadState()).consent).toBe('off');
expect((await effectiveConsent()).consent).toBe('off');
expect(readFileSync(file, 'utf8')).toBe(good.slice(0, 20));
writeFileSync(file, good);
expect((await effectiveConsent()).consent).toBe('on');
});
it('refuses to write over it', async () => {
const { updateState } = await import('@/lib/telemetry/state');
mkdirSync(dir, { recursive: true });
const file = path.join(dir, 'state.json');
writeFileSync(file, '{"consent":"on","endp');
await expect(updateState((state) => { state.nextScheduledAt = 'x'; })).rejects.toThrow('could not be read');
expect(readFileSync(file, 'utf8')).toBe('{"consent":"on","endp');
});
it('still writes the default state on a fresh install', async () => {
const { loadState } = await import('@/lib/telemetry/state');
expect((await loadState()).consent).toBe('off');
expect(JSON.parse(readFileSync(path.join(dir, 'state.json'), 'utf8')).consent).toBe('off');
});
});
@@ -0,0 +1,128 @@
// @vitest-environment node
import { mkdtempSync, readdirSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { setStoreForTests } from '@/lib/shared-store';
import { resetLocalStoreForTests } from '@/lib/shared-store/local';
import type { UpdateStatus } from '@/lib/version-check/types';
import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store';
// Telemetry and the version check with a shared store (#1066): one instance
// id, one login list and one state for the deployment.
vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false }));
let disks: string;
let store: SeparateDiskStore;
function onReplica(name: 'a' | 'b') {
vi.stubEnv('TELEMETRY_DATA_DIR', path.join(disks, name, 'telemetry'));
vi.stubEnv('VERSION_CHECK_DATA_DIR', path.join(disks, name, 'version-check'));
}
beforeEach(() => {
disks = mkdtempSync(path.join(tmpdir(), 'bw-telemetry-replicas-'));
store = separateDiskStore();
setStoreForTests(store);
vi.stubEnv('BULWARK_TELEMETRY', '');
onReplica('a');
});
afterEach(() => {
setStoreForTests(null);
resetLocalStoreForTests();
vi.unstubAllEnvs();
vi.unstubAllGlobals();
store.cleanup();
rmSync(disks, { recursive: true, force: true });
});
describe('telemetry with a shared store', () => {
it('gives every replica one instance id, even when they ask at once', async () => {
const { getInstanceId } = await import('@/lib/telemetry/state');
const [first, second] = await Promise.all([getInstanceId(), getInstanceId()]);
expect(second).toBe(first);
onReplica('b');
expect(await getInstanceId()).toBe(first);
});
it('counts logins on every replica together and reports the same numbers everywhere', async () => {
// A fresh module copy per replica: nothing carried over in memory.
const tracker = async () => {
vi.resetModules();
return import('@/lib/telemetry/login-tracker');
};
const a = await tracker();
await Promise.all([
a.recordLogin('alice@example.org', 'https://mail.example.org'),
a.recordLogin('bob@example.org', 'https://mail.example.org'),
]);
onReplica('b');
const b = await tracker();
await b.recordLogin('carol@example.org', 'https://mail.example.org');
expect(await b.getLoginCounts()).toEqual({ total: 3, active7d: 3 });
expect(await a.getLoginCounts()).toEqual({ total: 3, active7d: 3 });
});
it('does not report zero accounts while the store is down', async () => {
const { recordLogin, getLoginCounts } = await import('@/lib/telemetry/login-tracker');
await recordLogin('alice@example.org', 'https://mail.example.org');
store.down = true;
await expect(getLoginCounts()).rejects.toThrow('down (test)');
store.down = false;
expect(await getLoginCounts()).toEqual({ total: 1, active7d: 1 });
});
it('shows on replica B the consent set on replica A', async () => {
const { updateState, effectiveConsent } = await import('@/lib/telemetry/state');
await updateState((state) => { state.consent = 'on'; });
onReplica('b');
expect((await effectiveConsent()).consent).toBe('on');
});
});
function status(current: string, latest: string): UpdateStatus {
return {
schema: 1, current, latest, updateAvailable: current !== latest, severity: current !== latest ? 'normal' : 'none',
url: null, advisory: null, checkedAt: new Date().toISOString(),
};
}
describe('version check with a shared store', () => {
it('keeps each running version\'s status during a rolling update', async () => {
vi.stubEnv('BULWARK_UPDATE_CHECK_URL', 'https://version.example/');
const answer = (current: string) => vi.stubGlobal('fetch', vi.fn(async () => new Response(JSON.stringify(status(current, '1.21.0')))));
const { checkOnce, loadState, freshStatus } = await import('@/lib/version-check');
vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.20.0');
answer('1.20.0');
expect((await checkOnce()).ok).toBe(true);
onReplica('b');
vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.21.0');
answer('1.21.0');
expect((await checkOnce()).ok).toBe(true);
expect(freshStatus(await loadState())).toMatchObject({ current: '1.21.0', updateAvailable: false });
// A replica still on the old version keeps showing its banner.
onReplica('a');
vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.20.0');
expect(freshStatus(await loadState())).toMatchObject({ current: '1.20.0', updateAvailable: true });
});
});
describe('telemetry and version check on a single instance', () => {
it('keep today\'s files', async () => {
setStoreForTests(null);
const { getInstanceId, loadState } = await import('@/lib/telemetry/state');
const { recordLogin } = await import('@/lib/telemetry/login-tracker');
const versionCheck = await import('@/lib/version-check');
await getInstanceId();
await loadState();
await recordLogin('alice@example.org', 'https://mail.example.org');
await versionCheck.saveState({ ...(await versionCheck.loadState()), lastCheckedAt: 'x' });
expect(readdirSync(path.join(disks, 'a', 'telemetry')).sort()).toEqual(['.telemetry-id', 'logins.json', 'state.json']);
expect(readdirSync(path.join(disks, 'a', 'version-check'))).toEqual(['state.json']);
});
});
@@ -0,0 +1,54 @@
// @vitest-environment node
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { NextRequest } from 'next/server';
import { setStoreForTests } from '@/lib/shared-store';
import { resetLocalStoreForTests } from '@/lib/shared-store/local';
import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store';
// With the shared store down, the update banner's routes answered 500. They
// answer 503 with Retry-After now, like the other routes that need the store.
vi.mock('@/lib/admin/session', () => ({ requireAdminAuth: async () => ({ session: { sub: 'admin' } }) }));
vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false }));
vi.mock('@/lib/auth/instance-details', () => ({ canSeeInstanceDetails: async () => true }));
vi.mock('@/lib/logger', () => ({ logger: { warn: () => {}, error: () => {}, info: () => {}, debug: () => {} } }));
import { GET as updateStatus } from '@/app/api/system/update-status/route';
import { GET as adminVersion, POST as adminCheck } from '@/app/api/admin/version/route';
let store: SeparateDiskStore;
beforeEach(() => {
store = separateDiskStore();
setStoreForTests(store);
vi.stubEnv('BULWARK_UPDATE_CHECK_URL', 'https://version.example/');
vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.20.0');
});
afterEach(() => {
setStoreForTests(null);
resetLocalStoreForTests();
vi.unstubAllEnvs();
store.cleanup();
});
const req = (url: string, init?: RequestInit) => new NextRequest(`http://localhost${url}`, init as never);
describe('update status routes while the shared store is down', () => {
it('answer 503 with Retry-After, and 200 again once it is back', async () => {
store.down = true;
const responses = [
await updateStatus(req('/api/system/update-status')),
await adminVersion(req('/api/admin/version')),
await adminCheck(req('/api/admin/version', { method: 'POST', body: JSON.stringify({ action: 'check-now' }) })),
];
for (const res of responses) {
expect(res.status).toBe(503);
expect(res.headers.get('retry-after')).toBe('10');
}
store.down = false;
expect((await updateStatus(req('/api/system/update-status'))).status).toBe(200);
expect((await adminVersion(req('/api/admin/version'))).status).toBe(200);
});
});
+36 -24
View File
@@ -1,41 +1,53 @@
import { createHash, randomBytes } from 'node:crypto';
import { mkdir, open, readFile, rename, unlink } from 'node:fs/promises';
import { createHash } from 'node:crypto';
import { mkdir } from 'node:fs/promises';
import path from 'node:path';
import { getStore, isSharedMode, type CollectionOptions } from '@/lib/shared-store';
import { archiveIdentity, parseArchiveEnvelope, type ArchiveEnvelope, type ArchiveOwner } from './account-archive';
/** One file per owner: `SETTINGS_DATA_DIR/account-archives/<sha256 of the owner>.json`. */
function archiveDir(): string {
return path.join(process.env.SETTINGS_DATA_DIR || path.join(process.cwd(), 'data', 'settings'), 'account-archives');
}
function archivePath(owner: ArchiveOwner): string {
return path.join(archiveDir(), createHash('sha256').update(archiveIdentity(owner)).digest('hex') + '.json');
const ARCHIVE_FILE = /^[0-9a-f]{64}\.json$/;
/**
* One document per owner (#1066), replaced whole on every save. Local mode:
* today's files, mode 0600. The archive is encrypted in the browser with
* the user's password; the server holds no key.
*/
export const ARCHIVE_COLLECTION: CollectionOptions = {
local: {
dir: archiveDir,
file: (id) => `${id}.json`,
idOf: (relative) => (ARCHIVE_FILE.test(relative) ? relative.slice(0, -'.json'.length) : null),
mode: 0o600,
// Written to disk before the rename, as it always was.
fsync: true,
},
};
function collection() {
return getStore().collection('account-archives', ARCHIVE_COLLECTION);
}
function archiveId(owner: ArchiveOwner): string {
return createHash('sha256').update(archiveIdentity(owner)).digest('hex');
}
export async function loadArchive(owner: ArchiveOwner): Promise<ArchiveEnvelope | null> {
try { return parseArchiveEnvelope(JSON.parse(await readFile(archivePath(owner), 'utf8'))); }
catch (err) {
if ((err as NodeJS.ErrnoException).code === 'ENOENT') return null;
throw err;
}
const stored = await collection().get(archiveId(owner));
return stored ? parseArchiveEnvelope(JSON.parse(stored.data.toString('utf8'))) : null;
}
/** Replaces the owner's archive atomically: a reader sees the old file or the new one, never half of it. */
/** Replaces the owner's archive atomically: a reader sees the old one or the new one, never half of it. */
export async function saveArchive(owner: ArchiveOwner, envelope: ArchiveEnvelope): Promise<void> {
const target = archivePath(owner);
const temporary = `${target}.${randomBytes(8).toString('hex')}.tmp`;
await mkdir(archiveDir(), { recursive: true, mode: 0o700 });
try {
const file = await open(temporary, 'wx', 0o600);
try { await file.writeFile(JSON.stringify(parseArchiveEnvelope(envelope))); await file.sync(); }
finally { await file.close(); }
await rename(temporary, target);
} finally {
await unlink(temporary).catch(() => {});
}
const data = JSON.stringify(parseArchiveEnvelope(envelope));
// The folder only its owner may list, as it has always been created.
if (!isSharedMode()) await mkdir(archiveDir(), { recursive: true, mode: 0o700 });
await collection().put(archiveId(owner), data);
}
export async function deleteArchive(owner: ArchiveOwner): Promise<void> {
await unlink(archivePath(owner)).catch(err => {
if ((err as NodeJS.ErrnoException).code !== 'ENOENT') throw err;
});
await collection().delete(archiveId(owner));
}
+22 -4
View File
@@ -12,6 +12,7 @@ import { DisallowedUrlError, fetchPublicUrl } from '@/lib/security/url-guard';
import { authHeaderFor, credentialOrigin, privateProxyOrigins, type CredentialInject, type CredentialSpec } from './credentials';
import { getUserCredential, setUserCredentialSecret, userKey, type VaultUser } from './vault';
import { declaredUserCredential } from './known-plugins';
import { isStoredDataUnavailable } from '@/lib/stored-data-errors';
/** Largest request body the proxy forwards (one upload chunk). */
export const MAX_PROXY_REQUEST_BYTES = 100 * 1024 * 1024;
@@ -68,7 +69,12 @@ export interface ResolvedCredential {
* admin configured for an installed plugin that declares it.
*/
export async function resolveCredential(user: VaultUser, pluginId: string, credentialId: string): Promise<ResolvedCredential | null> {
const stored = await getUserCredential(user, pluginId, credentialId);
const plugin = await getPlugin(pluginId);
const spec = (plugin as { credentials?: Record<string, CredentialSpec> } | null)?.credentials?.[credentialId];
// A credential the admin set up never depends on the user's own vault: a
// user can only store credentials a plugin declares for users, so there
// is nothing of theirs to find under this id.
const stored = spec?.scope === 'admin' ? null : await getUserCredential(user, pluginId, credentialId);
if (stored) {
// Only for a plugin this server knows declares it, on the server the
// manifest fixes (see declaredUserCredential).
@@ -80,8 +86,6 @@ export async function resolveCredential(user: VaultUser, pluginId: string, crede
trusted: privateProxyOrigins().includes(stored.origin), scope: 'user',
};
}
const plugin = await getPlugin(pluginId);
const spec = (plugin as { credentials?: Record<string, CredentialSpec> } | null)?.credentials?.[credentialId];
if (!plugin || !plugin.enabled || !spec || spec.scope !== 'admin') return null;
const config = await getPluginConfig(pluginId);
const secret = config[spec.secretKey];
@@ -269,7 +273,15 @@ export async function proxyRequest(input: ProxyInput): Promise<Response> {
const method = input.method.toUpperCase();
if (!PROXY_METHODS.has(method)) throw new ProxyError(`Method ${method} is not allowed`, 405, 'bad_method');
const credential = await resolveCredential(input.user, input.pluginId, input.credentialId);
let credential: ResolvedCredential | null;
try {
credential = await resolveCredential(input.user, input.pluginId, input.credentialId);
} catch (error) {
if (isStoredDataUnavailable(error)) {
throw Object.assign(new ProxyError('The connected accounts could not be read right now', 503, 'store_unavailable'), { retryAfter: 10 });
}
throw error;
}
if (!credential) throw new ProxyError('No such credential', 404, 'no_credential');
let target: URL;
@@ -381,6 +393,12 @@ export async function proxyRequest(input: ProxyInput): Promise<Response> {
audit(error.status, 0, error.code);
throw error;
}
// A captured secret that couldn't be stored: the vault doesn't read,
// or the shared store is down. Nothing was written.
if (isStoredDataUnavailable(error)) {
audit(503, 0, 'store unavailable');
throw Object.assign(new ProxyError('The connected accounts could not be read or saved right now', 503, 'store_unavailable'), { retryAfter: 10 });
}
if (error instanceof DisallowedUrlError) {
audit(403, 0, 'blocked address');
throw new ProxyError('This server address is not allowed', 403, 'blocked_address');
+85 -49
View File
@@ -1,12 +1,14 @@
// Per-user store of the credentials users connected for plugins. One
// encrypted file per mail account; the secrets never leave the server again
// except as an auth header on a request the proxy forwards to the
// encrypted document per mail account; the secrets never leave the server
// again except as an auth header on a request the proxy forwards to the
// credential's own origin.
import { createHash } from 'node:crypto';
import { readFile, writeFile, mkdir, rename, unlink } from 'node:fs/promises';
import path from 'node:path';
import { decryptPayload, encryptPayload } from '@/lib/auth/crypto';
import { getStore, type CollectionOptions } from '@/lib/shared-store';
import { isStoreUnavailable, isStoredDataUnreadable, StoredDataUnreadableError } from '@/lib/stored-data-errors';
import { moveUnreadableAside } from '@/lib/stored-data-quarantine';
import type { CredentialInject } from './credentials';
export interface VaultUser {
@@ -49,51 +51,89 @@ export function userKey(user: VaultUser): string {
return createHash('sha256').update(`${user.username}:${user.serverUrl}`).digest('hex');
}
function vaultPath(user: VaultUser): string {
const dir = path.resolve(vaultDir());
const file = path.resolve(dir, `${userKey(user)}.enc`);
if (!file.startsWith(dir + path.sep)) throw new Error('Invalid vault path');
return file;
const VAULT_FILE = /^[0-9a-f]{64}\.enc$/;
/**
* One document per account, keyed by userKey (#1066). Local mode: today's
* `<vault dir>/<key>.enc`, base64 `encryptPayload` text, mode 0600.
*/
export const VAULT_COLLECTION: CollectionOptions = {
local: {
dir: () => vaultDir(),
file: (id) => `${id}.enc`,
idOf: (relative) => (VAULT_FILE.test(relative) ? relative.slice(0, -'.enc'.length) : null),
mode: 0o600,
},
};
function collection() {
return getStore().collection('plugin-secrets', VAULT_COLLECTION);
}
async function load(user: VaultUser): Promise<VaultData> {
let raw: string;
try {
raw = await readFile(vaultPath(user), 'utf8');
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return {};
throw error;
}
const payload = decryptPayload(raw, 'plugin-secrets');
/**
* The user's credentials from a stored document; {} for none. One that
* doesn't decrypt (another session secret) throws StoredDataUnreadableError,
* so no write lands on top of it before it is moved aside (see modify).
*/
function decode(user: VaultUser, raw: Buffer | null): VaultData {
if (!raw) return {};
const payload = decryptPayload(raw.toString('utf8'), 'plugin-secrets');
if (!payload) throw new StoredDataUnreadableError('plugin credentials');
// Bound to this user: a file copied over from another account's name
// decrypts, but carries the other account's key and is ignored.
if (!payload || payload.u !== userKey(user)) return {};
if (payload.u !== userKey(user)) return {};
const data = payload.d;
return data && typeof data === 'object' ? (data as VaultData) : {};
}
async function save(user: VaultUser, data: VaultData): Promise<void> {
const file = vaultPath(user);
if (Object.keys(data).length === 0) {
try { await unlink(file); } catch { /* nothing stored */ }
return;
async function load(user: VaultUser): Promise<VaultData> {
let stored;
try {
stored = await collection().get(userKey(user));
} catch (error) {
if (isStoreUnavailable(error)) throw error;
throw new StoredDataUnreadableError('plugin credentials', { cause: error });
}
try {
return decode(user, stored?.data ?? null);
} catch (error) {
// Reads as no credentials; the next write moves it aside.
if (isStoredDataUnreadable(error)) return {};
throw error;
}
await mkdir(path.dirname(file), { recursive: true });
const tmp = `${file}.${process.pid}.tmp`;
await writeFile(tmp, encryptPayload({ u: userKey(user), d: data }, 'plugin-secrets'), { encoding: 'utf8', mode: 0o600 });
await rename(tmp, file);
}
// Writes for one user run one after another, so two saves can't drop each
// other's changes.
const queues = new Map<string, Promise<unknown>>();
function serialize<T>(user: VaultUser, task: () => Promise<T>): Promise<T> {
const key = userKey(user);
const run = (queues.get(key) ?? Promise.resolve()).then(task, task);
const tail = run.catch(() => {});
queues.set(key, tail);
void tail.then(() => { if (queues.get(key) === tail) queues.delete(key); });
return run;
/**
* Change the user's credentials in one step against what is stored now, so
* two writes at once (two tabs, two replicas) both land. `change` edits the
* data it gets and says whether it changed anything; it may run more than
* once. An empty vault is deleted.
*
* A vault that doesn't decrypt is moved aside first (kept next to the
* original as `.unreadable-<time>`, never overwritten), and the write goes
* on as if nothing was stored.
*/
async function modify<T>(user: VaultUser, change: (data: VaultData) => { result: T; changed: boolean }): Promise<T> {
const docs = collection();
const id = userKey(user);
for (let attempt = 0; ; attempt++) {
try {
let result!: T;
await docs.update(id, (current) => {
const data = decode(user, current);
const out = change(data);
result = out.result;
if (!out.changed) return undefined;
if (Object.keys(data).length === 0) return current ? null : undefined;
return encryptPayload({ u: userKey(user), d: data }, 'plugin-secrets');
});
return result;
} catch (error) {
if (!isStoredDataUnreadable(error) || attempt >= 2) throw error;
const current = await docs.get(id);
if (current) await moveUnreadableAside(docs, id, current, path.join(vaultDir(), `${id}.enc`), 'plugin credentials');
}
}
}
export async function getUserCredential(user: VaultUser, pluginId: string, credentialId: string): Promise<StoredCredential | null> {
@@ -134,8 +174,7 @@ export function putUserCredential(
credentialId: string,
input: { origin: string; inject: CredentialInject; label: string; username?: string; secret?: string },
): Promise<void> {
return serialize(user, async () => {
const data = await load(user);
return modify(user, (data) => {
const existing = data[pluginId]?.[credentialId];
const count = Object.values(data).reduce((n, c) => n + Object.keys(c).length, 0);
if (!existing && count >= MAX_USER_CREDENTIALS) throw new VaultFullError();
@@ -153,7 +192,7 @@ export function putUserCredential(
updatedAt: now,
},
};
await save(user, data);
return { result: undefined, changed: true };
});
}
@@ -165,32 +204,29 @@ export function setUserCredentialSecret(
secret: string,
username?: string,
): Promise<boolean> {
return serialize(user, async () => {
const data = await load(user);
return modify(user, (data) => {
const existing = data[pluginId]?.[credentialId];
if (!existing) return false;
if (!existing) return { result: false, changed: false };
data[pluginId][credentialId] = {
...existing,
secret,
...(username ? { username } : {}),
updatedAt: new Date().toISOString(),
};
await save(user, data);
return true;
return { result: true, changed: true };
});
}
/** Remove one credential, or all of a plugin's when credentialId is omitted. */
export function deleteUserCredentials(user: VaultUser, pluginId: string, credentialId?: string): Promise<void> {
return serialize(user, async () => {
const data = await load(user);
if (!data[pluginId]) return;
return modify(user, (data) => {
if (!data[pluginId]) return { result: undefined, changed: false };
if (credentialId) {
delete data[pluginId][credentialId];
if (Object.keys(data[pluginId]).length === 0) delete data[pluginId];
} else {
delete data[pluginId];
}
await save(user, data);
return { result: undefined, changed: true };
});
}
+186
View File
@@ -0,0 +1,186 @@
import { logger } from '@/lib/logger';
import { getStore, isSharedMode, replicaId } from '@/lib/shared-store';
import { onSharedChange, storeLostItsData } from '@/lib/shared-store/changes';
/**
* One replica runs the background jobs (telemetry heartbeat, update check)
* for the whole deployment (#1066). It holds the `scheduler-leader` lease,
* renewing it every 10 s; every other replica tries to take it every 10 s,
* so when the leader dies or loses the store another one takes over within
* the 30 s lease plus one try. A leader that can't renew stops its jobs at
* once. The jobs keep their schedule in the shared state, so a new leader
* continues it instead of starting over.
*
* Local mode: the one process is always the leader.
*
* Everything lives on globalThis: instrumentation starts it, and the
* `telemetry` change listener may be run from a route's module copy.
*/
export const LEADER_LEASE = 'scheduler-leader';
export const LEADER_TTL_MS = 30_000;
export const LEADER_TRY_MS = 10_000;
export interface ScheduledJob {
name: string;
/** Start the job's timer from its stored schedule. Must be idempotent. */
start(): Promise<void>;
stop(): Promise<void>;
}
interface LeaderGlobals {
jobs: ScheduledJob[];
/** Jobs whose start went through while leading; the others are retried on each renewal. */
started: Set<string>;
leading: boolean;
interval?: NodeJS.Timeout;
trying?: Promise<boolean>;
unlisten?: () => void;
}
const KEY = Symbol.for('bulwark.schedulerLeader');
function globals(): LeaderGlobals {
const g = globalThis as typeof globalThis & { [KEY]?: LeaderGlobals };
return (g[KEY] ??= { jobs: [], started: new Set(), leading: false });
}
function message(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
/** True while this process runs the background jobs. */
export function isSchedulerLeader(): boolean {
return globals().leading;
}
async function startJobs(g: LeaderGlobals): Promise<void> {
for (const job of g.jobs) {
if (g.started.has(job.name)) continue;
try {
await job.start();
// Stepped down while it started: don't leave its timer running.
if (g.leading) g.started.add(job.name);
else await job.stop();
} catch (error) {
logger.warn('scheduler: could not start a job; retrying', { job: job.name, error: message(error) });
}
}
}
async function stopJobs(g: LeaderGlobals): Promise<void> {
g.started.clear();
for (const job of g.jobs) {
try {
await job.stop();
} catch (error) {
logger.warn('scheduler: could not stop a job', { job: job.name, error: message(error) });
}
}
}
async function becomeLeader(g: LeaderGlobals): Promise<void> {
g.leading = true;
logger.info('scheduler: leading', { replica: replicaId() });
await startJobs(g);
}
async function stepDown(g: LeaderGlobals, reason: string): Promise<void> {
g.leading = false;
logger.info('scheduler: stopped leading', { replica: replicaId(), reason });
await stopJobs(g);
}
/**
* One try: take or renew the lease, start the jobs on taking it, stop them
* on losing it (or on not reaching the store). Returns whether this replica
* leads afterwards.
*/
export async function tryLead(): Promise<boolean> {
const g = globals();
if (g.trying) return g.trying;
g.trying = (async () => {
let holds: boolean;
let reason = 'another replica holds the lease';
if (storeLostItsData()) {
// The store was emptied under us: a lease taken there now proves
// nothing, and the jobs' stored schedule is gone. Wait for it back.
holds = false;
reason = 'the shared store lost its data';
} else {
try {
holds = await getStore().leases.acquire(LEADER_LEASE, replicaId(), LEADER_TTL_MS);
} catch (error) {
holds = false;
reason = `could not reach the shared store: ${message(error)}`;
}
}
if (holds && !g.leading) await becomeLeader(g);
else if (holds) await startJobs(g);
else if (g.leading) await stepDown(g, reason);
return g.leading;
})().finally(() => {
g.trying = undefined;
});
return g.trying;
}
/** Restart one job on the leader (its settings changed), keeping its stored schedule. */
async function restartJob(name: string): Promise<void> {
const g = globals();
if (!g.leading) return;
const job = g.jobs.find((j) => j.name === name);
if (!job) return;
g.started.delete(name);
await job.stop();
await job.start();
if (g.leading) g.started.add(name);
else await job.stop();
}
/**
* Start leading or trying to (instrumentation, once per process). In shared
* mode every replica calls it; only the lease holder runs `jobs`.
*/
export async function startSchedulerLeader(jobs: ScheduledJob[]): Promise<void> {
const g = globals();
g.jobs = jobs;
// Consent or the collector changed on some replica: the leader picks it up.
g.unlisten ??= onSharedChange('telemetry', () => restartJob('telemetry'));
if (!isSharedMode()) {
if (!g.leading) await becomeLeader(g);
return;
}
if (!g.interval) {
g.interval = setInterval(() => void tryLead(), LEADER_TRY_MS);
g.interval.unref();
}
await tryLead();
}
/** Stop the jobs and give the lease up (shutdown, tests). */
export async function stopSchedulerLeader(): Promise<void> {
const g = globals();
if (g.interval) clearInterval(g.interval);
g.interval = undefined;
g.unlisten?.();
g.unlisten = undefined;
if (g.leading) {
if (isSharedMode()) await getStore().leases.release(LEADER_LEASE, replicaId()).catch(() => {});
await stepDown(g, 'stopped');
}
}
/** Resolves when the current try (and the job starts it set off) is done (tests). */
export function schedulerLeaderSettled(): Promise<unknown> {
return globals().trying ?? Promise.resolve();
}
/** Forget everything (tests). */
export function resetSchedulerLeaderForTests(): void {
const g = globals();
if (g.interval) clearInterval(g.interval);
g.unlisten?.();
const holder = globalThis as typeof globalThis & { [KEY]?: LeaderGlobals };
delete holder[KEY];
}
+101 -43
View File
@@ -1,14 +1,17 @@
import { createHash, createCipheriv, createDecipheriv, randomBytes } from 'node:crypto';
import { readFile, writeFile, unlink, mkdir, rename } from 'node:fs/promises';
import { existsSync } from 'node:fs';
import path from 'node:path';
import { logger } from '@/lib/logger';
import { getSessionSecret } from '@/lib/auth/session-secret';
import { getStore, type CollectionOptions } from '@/lib/shared-store';
import { isStoreUnavailable, isStoredDataUnreadable, StoredDataUnreadableError } from '@/lib/stored-data-errors';
import { moveUnreadableAside } from '@/lib/stored-data-quarantine';
const ALGORITHM = 'aes-256-gcm';
const IV_LENGTH = 12;
const TAG_LENGTH = 16;
type Settings = Record<string, unknown>;
function getKey(): Buffer {
const secret = getSessionSecret();
if (!secret) throw new Error('SESSION_SECRET not configured');
@@ -19,47 +22,46 @@ export function getSettingsDir(): string {
return process.env.SETTINGS_DATA_DIR || path.join(process.cwd(), 'data', 'settings');
}
function getSettingsPath(username: string, serverUrl: string): string {
const hash = createHash('sha256').update(`${username}:${serverUrl}`).digest('hex');
const filePath = path.join(getSettingsDir(), `${hash}.enc`);
// Defense in depth: ensure path stays within the settings directory
const resolvedDir = path.resolve(getSettingsDir());
const resolvedPath = path.resolve(filePath);
if (!resolvedPath.startsWith(resolvedDir + path.sep) && resolvedPath !== resolvedDir) {
throw new Error('Invalid settings path');
}
return resolvedPath;
/** The document id of an account's settings: sha256 of `user:server`, hex. */
export function settingsId(username: string, serverUrl: string): string {
return createHash('sha256').update(`${username}:${serverUrl}`).digest('hex');
}
export async function saveUserSettings(username: string, serverUrl: string, settings: Record<string, unknown>): Promise<void> {
const dir = getSettingsDir();
if (!existsSync(dir)) {
await mkdir(dir, { recursive: true });
}
const SETTINGS_FILE = /^[0-9a-f]{64}\.enc$/;
const key = getKey();
/**
* One encrypted document per account (#1066). Local mode: today's
* `<settings dir>/<id>.enc`, `iv‖tag‖ciphertext`. The account archives live
* in a subfolder of the same dir and are not part of this collection.
*/
export const SETTINGS_COLLECTION: CollectionOptions = {
local: {
dir: () => getSettingsDir(),
file: (id) => `${id}.enc`,
idOf: (relative) => (SETTINGS_FILE.test(relative) ? relative.slice(0, -'.enc'.length) : null),
},
};
function collection() {
return getStore().collection('settings', SETTINGS_COLLECTION);
}
function encryptSettings(settings: Settings): Buffer {
const iv = randomBytes(IV_LENGTH);
const cipher = createCipheriv(ALGORITHM, key, iv);
const payload = JSON.stringify(settings);
const encrypted = Buffer.concat([cipher.update(payload, 'utf8'), cipher.final()]);
const tag = cipher.getAuthTag();
const data = Buffer.concat([iv, tag, encrypted]);
const targetPath = getSettingsPath(username, serverUrl);
const tmpPath = targetPath + '.tmp';
await writeFile(tmpPath, data);
await rename(tmpPath, targetPath);
const cipher = createCipheriv(ALGORITHM, getKey(), iv);
const encrypted = Buffer.concat([cipher.update(JSON.stringify(settings), 'utf8'), cipher.final()]);
return Buffer.concat([iv, cipher.getAuthTag(), encrypted]);
}
export async function loadUserSettings(username: string, serverUrl: string): Promise<Record<string, unknown> | null> {
const filePath = getSettingsPath(username, serverUrl);
/**
* Settings from stored bytes. Too short to hold anything reads as nothing
* stored, as it always has; bytes that don't decrypt throw
* StoredDataUnreadableError.
*/
function decryptSettings(data: Buffer): Settings | null {
if (data.length < IV_LENGTH + TAG_LENGTH) return null;
const key = getKey();
try {
const data = await readFile(filePath);
if (data.length < IV_LENGTH + TAG_LENGTH) return null;
const key = getKey();
const iv = data.subarray(0, IV_LENGTH);
const tag = data.subarray(IV_LENGTH, IV_LENGTH + TAG_LENGTH);
const encrypted = data.subarray(IV_LENGTH + TAG_LENGTH);
@@ -70,18 +72,74 @@ export async function loadUserSettings(username: string, serverUrl: string): Pro
const decrypted = Buffer.concat([decipher.update(encrypted), decipher.final()]);
return JSON.parse(decrypted.toString('utf8'));
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null;
logger.warn('Failed to load user settings', { error: error instanceof Error ? error.message : 'Unknown error' });
return null;
logger.warn('Failed to decrypt user settings', { error: error instanceof Error ? error.message : 'Unknown error' });
throw new StoredDataUnreadableError('settings', { cause: error });
}
}
export async function saveUserSettings(username: string, serverUrl: string, settings: Settings): Promise<void> {
await collection().put(settingsId(username, serverUrl), encryptSettings(settings));
}
/**
* The user's stored settings, or null when none are stored. Settings that
* don't decrypt (written with another session secret) read as none too; the
* next save moves them aside before it writes (see updateUserSettings). A
* file that can't be read throws StoredDataUnreadableError, a shared store
* that can't be reached StoreUnavailableError.
*/
export async function loadUserSettings(username: string, serverUrl: string): Promise<Settings | null> {
let stored;
try {
stored = await collection().get(settingsId(username, serverUrl));
} catch (error) {
if (isStoreUnavailable(error)) throw error;
logger.warn('Failed to read user settings', { error: error instanceof Error ? error.message : 'Unknown error' });
throw new StoredDataUnreadableError('settings', { cause: error });
}
if (!stored) return null;
try {
return decryptSettings(stored.data);
} catch (error) {
if (isStoredDataUnreadable(error)) return null;
throw error;
}
}
/**
* Read, change and write the user's settings in one step, so a save that
* lands at the same time (another tab, another replica) isn't lost: `change`
* gets what is stored now (null for nothing) and returns what to store. It
* may run more than once.
*
* Stored settings that don't decrypt are moved aside first (kept next to
* the original as `.unreadable-<time>`, never overwritten), and the save
* goes on as if nothing was stored.
*/
export async function updateUserSettings(
username: string,
serverUrl: string,
change: (stored: Settings | null) => Settings,
): Promise<void> {
const docs = collection();
const id = settingsId(username, serverUrl);
for (let attempt = 0; ; attempt++) {
try {
await docs.update(id, (current) => encryptSettings(change(current ? decryptSettings(current) : null)));
return;
} catch (error) {
if (!isStoredDataUnreadable(error) || attempt >= 2) throw error;
const current = await docs.get(id);
if (current) await moveUnreadableAside(docs, id, current, path.join(getSettingsDir(), `${id}.enc`), 'settings');
}
}
}
export async function deleteUserSettings(username: string, serverUrl: string): Promise<void> {
try {
await unlink(getSettingsPath(username, serverUrl));
await collection().delete(settingsId(username, serverUrl));
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
logger.warn('Failed to delete user settings', { error: error instanceof Error ? error.message : 'Unknown error' });
}
if (isStoreUnavailable(error)) throw error;
logger.warn('Failed to delete user settings', { error: error instanceof Error ? error.message : 'Unknown error' });
}
}
@@ -0,0 +1,103 @@
/**
* @vitest-environment node
*/
import { mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, statSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { setStoreForTests } from '../index';
import { resetLocalStoreForTests } from '../local';
import { separateDiskStore, type SeparateDiskStore } from '@/lib/__tests__/helpers/separate-disk-store';
import { steps } from '../import/data';
import { SETTINGS_COLLECTION } from '@/lib/settings-sync';
import { VAULT_COLLECTION } from '@/lib/plugin-secrets/vault';
import { ARCHIVE_COLLECTION } from '@/lib/account-archive-storage';
import { TELEMETRY_COLLECTION } from '@/lib/telemetry/state';
import { VERSION_CHECK_COLLECTION } from '@/lib/version-check/state';
// An existing single instance's settings, plugin credentials, archives,
// telemetry and update-check state go into an empty shared store once, as
// they are; the local files stay untouched.
let data: string;
let store: SeparateDiskStore;
const hash = (c: string) => c.repeat(64);
function snapshot(dir: string): Record<string, string> {
const out: Record<string, string> = {};
const walk = (d: string) => {
for (const name of readdirSync(d)) {
const full = path.join(d, name);
if (statSync(full).isDirectory()) walk(full);
else out[path.relative(dir, full)] = readFileSync(full).toString('base64');
}
};
walk(dir);
return out;
}
beforeEach(() => {
data = mkdtempSync(path.join(tmpdir(), 'bw-import-data-'));
vi.stubEnv('SETTINGS_DATA_DIR', path.join(data, 'settings'));
vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(data, 'plugin-secrets'));
vi.stubEnv('TELEMETRY_DATA_DIR', path.join(data, 'telemetry'));
vi.stubEnv('VERSION_CHECK_DATA_DIR', path.join(data, 'version-check'));
for (const dir of ['settings/account-archives', 'plugin-secrets', 'telemetry', 'version-check']) {
mkdirSync(path.join(data, dir), { recursive: true });
}
writeFileSync(path.join(data, 'settings', `${hash('a')}.enc`), Buffer.from([1, 2, 3, 4]));
writeFileSync(path.join(data, 'settings', `${hash('b')}.enc`), Buffer.from([5, 6]));
writeFileSync(path.join(data, 'settings', `${hash('c')}.enc.tmp`), 'half written');
writeFileSync(path.join(data, 'settings', 'account-archives', `${hash('d')}.json`), '{"version":1}');
writeFileSync(path.join(data, 'plugin-secrets', `${hash('e')}.enc`), 'c2VhbGVk');
writeFileSync(path.join(data, 'telemetry', 'state.json'), '{"consent":"on"}');
writeFileSync(path.join(data, 'telemetry', '.telemetry-id'), '0b9f1f3e-8b8e-4c4b-9a35-3c1b8a7a1e11');
writeFileSync(path.join(data, 'telemetry', 'logins.json'), '{"records":[]}');
writeFileSync(path.join(data, 'version-check', 'state.json'), '{"status":null}');
store = separateDiskStore();
setStoreForTests(store);
});
afterEach(() => {
setStoreForTests(null);
resetLocalStoreForTests();
vi.unstubAllEnvs();
store.cleanup();
rmSync(data, { recursive: true, force: true });
});
async function runAll(): Promise<Record<string, number>> {
const counts: Record<string, number> = {};
for (const step of steps) counts[step.name] = await step.run();
return counts;
}
describe('importing part data', () => {
it('copies every document once and leaves the local files alone', async () => {
const before = snapshot(data);
expect(await runAll()).toEqual({
settings: 2, 'plugin-secrets': 1, 'account-archives': 1, telemetry: 3, 'version-check': 1,
});
expect(await runAll()).toEqual({
settings: 0, 'plugin-secrets': 0, 'account-archives': 0, telemetry: 0, 'version-check': 0,
});
expect(snapshot(data)).toEqual(before);
const settings = store.collection('settings', SETTINGS_COLLECTION);
expect(await settings.list()).toEqual([hash('a'), hash('b')]);
expect((await settings.get(hash('a')))?.data).toEqual(Buffer.from([1, 2, 3, 4]));
expect((await store.collection('plugin-secrets', VAULT_COLLECTION).get(hash('e')))?.data.toString()).toBe('c2VhbGVk');
expect((await store.collection('account-archives', ARCHIVE_COLLECTION).get(hash('d')))?.data.toString()).toBe('{"version":1}');
expect((await store.collection('telemetry', TELEMETRY_COLLECTION).list()).sort()).toEqual(['.telemetry-id', 'logins.json', 'state.json']);
expect((await store.collection('version-check', VERSION_CHECK_COLLECTION).get('state.json'))?.data.toString()).toBe('{"status":null}');
});
it('keeps what the store already holds', async () => {
const settings = store.collection('settings', SETTINGS_COLLECTION);
await settings.put(hash('a'), Buffer.from([9]));
expect((await runAll()).settings).toBe(1);
expect((await settings.get(hash('a')))?.data).toEqual(Buffer.from([9]));
});
});
+42
View File
@@ -0,0 +1,42 @@
import { getStore } from '@/lib/shared-store';
import { localCollection } from '@/lib/shared-store/local';
import type { CollectionOptions } from '@/lib/shared-store/types';
import type { ImportStep } from '@/lib/shared-store/import';
import { SETTINGS_COLLECTION } from '@/lib/settings-sync';
import { VAULT_COLLECTION } from '@/lib/plugin-secrets/vault';
import { ARCHIVE_COLLECTION } from '@/lib/account-archive-storage';
import { TELEMETRY_COLLECTION } from '@/lib/telemetry/state';
import { VERSION_CHECK_COLLECTION } from '@/lib/version-check/state';
/**
* Part `data` of the shared-store import (#1066): every document of these
* collections, read from the local files exactly as local mode reads them,
* written to the shared store only where it holds nothing yet. The bytes go
* over as they are: settings and plugin credentials stay encrypted with the
* session secret, which the boot imports first.
*/
function copyCollection(name: string, options: CollectionOptions): ImportStep {
return {
name,
async run() {
const local = localCollection(name, options);
const shared = getStore().collection(name, options);
let written = 0;
for (const id of await local.list()) {
const doc = await local.get(id);
if (!doc) continue;
if ((await shared.put(id, doc.data, { ifVersion: null })).ok) written++;
}
return written;
},
};
}
export const steps: ImportStep[] = [
copyCollection('settings', SETTINGS_COLLECTION),
copyCollection('plugin-secrets', VAULT_COLLECTION),
copyCollection('account-archives', ARCHIVE_COLLECTION),
// state.json, logins.json and the instance id .telemetry-id
copyCollection('telemetry', TELEMETRY_COLLECTION),
copyCollection('version-check', VERSION_CHECK_COLLECTION),
];
+15 -2
View File
@@ -1,6 +1,6 @@
import { randomBytes } from 'node:crypto';
import { existsSync } from 'node:fs';
import { appendFile, mkdir, readFile, readdir, rename, stat, unlink, writeFile } from 'node:fs/promises';
import { appendFile, mkdir, open, readFile, readdir, rename, stat, unlink, writeFile } from 'node:fs/promises';
import path from 'node:path';
import { assertDocId, assertKey, assertName, toBuffer, versionOf } from './keys';
import type {
@@ -245,7 +245,20 @@ export function localCollection(name: string, options: CollectionOptions): DocCo
async function writeCurrent(file: string, data: Buffer): Promise<void> {
await mkdir(path.dirname(file), { recursive: true });
const tmp = layout.tmpName ? layout.tmpName(file) : `${file}.${process.pid}.${randomBytes(4).toString('hex')}.tmp`;
await writeFile(tmp, data, layout.mode !== undefined ? { mode: layout.mode } : undefined);
if (layout.fsync) {
const handle = await open(tmp, 'wx', layout.mode ?? 0o666);
try {
await handle.writeFile(data);
await handle.sync();
} catch (error) {
await handle.close().catch(() => {});
await unlink(tmp).catch(() => {});
throw error;
}
await handle.close();
} else {
await writeFile(tmp, data, layout.mode !== undefined ? { mode: layout.mode } : undefined);
}
try {
await rename(tmp, file);
} catch (error) {
+5
View File
@@ -130,6 +130,11 @@ export interface LocalLayout {
* depend on it. Default: `<target>.<pid>.<random>.tmp`.
*/
tmpName?: (target: string) => string;
/**
* Flush the temp file to disk before the rename (open with `wx`, write,
* fsync, close), for documents that must survive a power cut.
*/
fsync?: boolean;
}
export interface CollectionOptions {
+59
View File
@@ -0,0 +1,59 @@
import { NextResponse } from 'next/server';
import { isStoreUnavailableError } from '@/lib/shared-store/types';
/**
* A user's stored document exists but can't be read: it doesn't decrypt
* (another session secret), it is cut short, or the read failed. It is not
* the same as "nothing stored": a route that treated it as empty would merge
* into nothing and overwrite the user's real data on the next save. Routes
* answer 503 and write nothing.
*/
export class StoredDataUnreadableError extends Error {
constructor(what: string, options?: { cause?: unknown }) {
super(`Stored ${what} could not be read`, options);
this.name = 'StoredDataUnreadableError';
}
}
// By name, not instanceof: Next bundles each entry with its own copy of a
// module, so the store's error class may not be the one this file imports.
function named(error: unknown, name: string): boolean {
return error instanceof Error && error.name === name;
}
/** The shared store couldn't be reached (StoreUnavailableError from lib/shared-store). */
export function isStoreUnavailable(error: unknown): boolean {
return isStoreUnavailableError(error);
}
export function isStoredDataUnreadable(error: unknown): boolean {
return named(error, 'StoredDataUnreadableError');
}
/** True for the errors that mean "try again later, nothing was written". */
export function isStoredDataUnavailable(error: unknown): boolean {
return isStoredDataUnreadable(error) || isStoreUnavailable(error);
}
/**
* The 503 for an unreadable document or an unreachable shared store, or null
* for any other error. The client keeps its own copy and retries.
*/
export function storedDataUnavailableResponse(
error: unknown,
headers?: Record<string, string>,
): NextResponse | null {
if (isStoredDataUnreadable(error)) {
return NextResponse.json(
{ error: 'Your stored data could not be read on the server. Nothing was changed.', code: 'stored_data_unreadable' },
{ status: 503, headers: { 'Retry-After': '10', ...headers } },
);
}
if (isStoreUnavailable(error)) {
return NextResponse.json(
{ error: 'Storage is temporarily unavailable. Nothing was changed.', code: 'store_unavailable' },
{ status: 503, headers: { 'Retry-After': '10', ...headers } },
);
}
return null;
}
+50
View File
@@ -0,0 +1,50 @@
import { readFile, rename } from 'node:fs/promises';
import path from 'node:path';
import { logger } from '@/lib/logger';
import { isSharedMode } from '@/lib/shared-store';
import { versionOf } from '@/lib/shared-store/keys';
import { serializeLocal } from '@/lib/shared-store/local';
import type { DocCollection, StoredDoc } from '@/lib/shared-store/types';
/**
* Put a user's document that no longer decrypts out of the way, so the
* user can save again, without losing its bytes (#1066). It most likely was
* written with another session secret (rotated, or a replica that had its
* own); with that secret back it can be restored by hand.
*
* - Local mode: renamed to `<file>.unreadable-<time>` next to it.
* - Shared mode: copied to `<id>.unreadable-<time>`, then the original is
* deleted, only if it is still the same document.
*
* Either way a concurrent write that replaced it meanwhile is left alone.
*/
export async function moveUnreadableAside(
docs: DocCollection,
id: string,
current: StoredDoc,
localFile: string,
what: string,
): Promise<void> {
const stamp = new Date().toISOString().replace(/[:.]/g, '-');
if (!isSharedMode()) {
const file = path.resolve(localFile);
const target = `${file}.unreadable-${stamp}`;
// On the collection's own queue for this file, so no write lands between
// the check and the rename.
const moved = await serializeLocal(`doc\0${file}`, async () => {
const now = await readFile(file).catch(() => null);
if (!now || versionOf(now) !== current.version) return false;
await rename(file, target);
return true;
});
if (moved) logger.warn(`Moved aside stored ${what} that could not be decrypted`, { path: target });
return;
}
const aside = `${id}.unreadable-${stamp}`;
await docs.put(aside, current.data, { ifVersion: null });
if (await docs.delete(id, { ifVersion: current.version })) {
logger.warn(`Moved aside stored ${what} that could not be decrypted`, { collection: docs.name, id: aside });
} else {
await docs.delete(aside).catch(() => {});
}
}
+1 -1
View File
@@ -1,7 +1,7 @@
export { startScheduler, stopScheduler, reschedule, sendOnce } from './sender';
export { buildPayload, markProcessStart } from './payload';
export {
loadState, saveState, getInstanceId, effectiveConsent,
loadState, saveState, updateState, getInstanceId, effectiveConsent,
} from './state';
export { recordLogin, getLoginCounts } from './login-tracker';
export {
+33 -52
View File
@@ -1,9 +1,7 @@
import { readFile, writeFile, mkdir, rename } from 'node:fs/promises';
import { existsSync } from 'node:fs';
import path from 'node:path';
import { createHmac } from 'node:crypto';
import { logger } from '@/lib/logger';
import { getInstanceId } from './state';
import { isStoreUnavailableError } from '@/lib/shared-store/types';
import { getInstanceId, telemetryDocs } from './state';
// We never store usernames or server URLs in the clear. Each login is
// recorded as HMAC-SHA256(username + '@' + serverUrl, instance_id), so the
@@ -22,39 +20,17 @@ interface LoginsFile {
const SEVEN_DAYS_MS = 7 * 24 * 60 * 60 * 1000;
const RETENTION_MS = 90 * 24 * 60 * 60 * 1000;
let cache: LoginsFile | null = null;
/** `logins.json` in the telemetry collection (#1066): one list for every replica. */
const LOGINS_DOC = 'logins.json';
function getDir(): string {
return process.env.TELEMETRY_DATA_DIR || path.join(process.cwd(), 'data', 'telemetry');
}
function loginsPath(): string {
return path.join(getDir(), 'logins.json');
}
async function ensureDir(): Promise<void> {
const dir = getDir();
if (!existsSync(dir)) await mkdir(dir, { recursive: true });
}
async function loadFile(): Promise<LoginsFile> {
if (cache) return cache;
function parse(data: Buffer | null): LoginsFile {
if (!data) return { records: [] };
try {
const raw = await readFile(loginsPath(), 'utf8');
const parsed = JSON.parse(raw) as Partial<LoginsFile>;
cache = Array.isArray(parsed?.records) ? { records: parsed.records as LoginRecord[] } : { records: [] };
const parsed = JSON.parse(data.toString('utf8')) as Partial<LoginsFile>;
return Array.isArray(parsed?.records) ? { records: parsed.records as LoginRecord[] } : { records: [] };
} catch {
cache = { records: [] };
return { records: [] };
}
return cache;
}
async function saveFile(file: LoginsFile): Promise<void> {
await ensureDir();
cache = file;
const tmp = loginsPath() + '.tmp';
await writeFile(tmp, JSON.stringify(file), 'utf8');
await rename(tmp, loginsPath());
}
function normalizeServer(serverUrl: string): string {
@@ -71,29 +47,31 @@ async function hashIdentity(username: string, serverUrl: string): Promise<string
* Record a successful login. Best-effort; never throws. Updates the
* existing record's timestamp if the same identity has logged in before,
* otherwise appends a new record. Records older than the retention window
* are pruned on every write.
* are pruned on every write. Read and written in one step, so logins on
* several replicas at once all count.
*/
export async function recordLogin(username: string, serverUrl: string): Promise<void> {
if (!username || !serverUrl) return;
try {
const id = await hashIdentity(username, serverUrl);
const file = await loadFile();
const now = new Date().toISOString();
const cutoff = Date.now() - RETENTION_MS;
const next: LoginRecord[] = [];
let updated = false;
for (const rec of file.records) {
const ts = new Date(rec.lastLoginAt).getTime();
if (Number.isNaN(ts) || ts < cutoff) continue;
if (rec.id === id) {
next.push({ id, lastLoginAt: now });
updated = true;
} else {
next.push(rec);
await telemetryDocs().update(LOGINS_DOC, (current) => {
const cutoff = Date.now() - RETENTION_MS;
const next: LoginRecord[] = [];
let updated = false;
for (const rec of parse(current).records) {
const ts = new Date(rec.lastLoginAt).getTime();
if (Number.isNaN(ts) || ts < cutoff) continue;
if (rec.id === id) {
next.push({ id, lastLoginAt: now });
updated = true;
} else {
next.push(rec);
}
}
}
if (!updated) next.push({ id, lastLoginAt: now });
await saveFile({ records: next });
if (!updated) next.push({ id, lastLoginAt: now });
return JSON.stringify({ records: next });
});
} catch (err) {
logger.debug?.('telemetry: recordLogin failed', {
error: err instanceof Error ? err.message : String(err),
@@ -103,11 +81,13 @@ export async function recordLogin(username: string, serverUrl: string): Promise<
/**
* Total distinct accounts seen in the 90-day retention window, plus those
* with a login in the last 7 days.
* with a login in the last 7 days. Read from the store each time, so every
* replica reports the same numbers. Throws while the shared store is down
* rather than reporting zero accounts.
*/
export async function getLoginCounts(): Promise<{ total: number; active7d: number }> {
try {
const file = await loadFile();
const file = parse((await telemetryDocs().get(LOGINS_DOC))?.data ?? null);
const cutoff = Date.now() - RETENTION_MS;
const sevenAgo = Date.now() - SEVEN_DAYS_MS;
let total = 0;
@@ -119,7 +99,8 @@ export async function getLoginCounts(): Promise<{ total: number; active7d: numbe
if (ts >= sevenAgo) active7d++;
}
return { total, active7d };
} catch {
} catch (err) {
if (isStoreUnavailableError(err)) throw err;
return { total: 0, active7d: 0 };
}
}
+120 -24
View File
@@ -1,5 +1,9 @@
import { logger } from '@/lib/logger';
import { effectiveConsent, endpointEnabled, loadState, saveState } from './state';
import { isSharedMode } from '@/lib/shared-store';
import { announceChange } from '@/lib/shared-store/changes';
import {
effectiveConsent, endpointEnabled, loadState, loadStateVersioned, saveStateIfUnchanged, updateState,
} from './state';
import { buildPayload } from './payload';
import { fetchTelemetryTarget, resolveEndpointAllowed } from './endpoint-guard';
import { DEFAULT_ENDPOINT } from './types';
@@ -7,15 +11,81 @@ import { DEFAULT_ENDPOINT } from './types';
const DAY_MS = 24 * 60 * 60 * 1000;
const JITTER_MS = 2 * 60 * 60 * 1000; // ± 2 hours
const FIRST_DELAY_MS = 60 * 60 * 1000; // 1 hour after consent
/** A send this long before a scheduled slot already counts for that slot. */
const SLOT_SLACK_MS = 60 * 1000;
/**
* With a shared store, a scheduled heartbeat this recent means the day's is
* done. A new leader computes its own slot when the old one stalled past its
* slot or died before storing the next one, so the slot alone can't tell.
*/
const SCHEDULED_GAP_MS = 20 * 60 * 60 * 1000;
let currentTimer: NodeJS.Timeout | null = null;
// On globalThis: Next loads this module once per bundle (instrumentation,
// each route), and a reschedule from a route must clear the timer the
// instrumentation copy started, not a copy of its own.
const TIMER_KEY = Symbol.for('bulwark.telemetry.timer');
/**
* `generation` goes up on every stop. A start or a tick captures it and
* arms a timer only while it is unchanged, so a check or send still in
* flight when the scheduler stops (this replica lost the lead) can't set a
* new timer afterwards.
*/
interface TimerState { current: NodeJS.Timeout | null; generation: number }
function timer(): TimerState {
const g = globalThis as typeof globalThis & { [TIMER_KEY]?: TimerState };
return (g[TIMER_KEY] ??= { current: null, generation: 0 });
}
function jitteredDelay(base: number): number {
const j = (Math.random() * 2 - 1) * JITTER_MS;
return Math.max(60_000, base + j);
}
export async function sendOnce(opts?: { reason?: string }): Promise<{
type Claim =
| { claimed: true; previous: string | null; previousScheduled: string | null | undefined; at: string; shared: boolean }
| { claimed: false };
/**
* Take the scheduled send for `slot` (its nextScheduledAt): set lastSentAt
* with a compare-and-set on the state, unless a send at or after the slot is
* already recorded. Of two leaders firing for one slot (a handover after a
* long pause), only one gets it.
*/
async function claimSlot(slot: string): Promise<Claim> {
const slotMs = new Date(slot).getTime();
const shared = isSharedMode();
for (let attempt = 0; attempt < 5; attempt++) {
const { state, version } = await loadStateVersioned();
if (version === null) return { claimed: false };
const last = state.lastSentAt ? new Date(state.lastSentAt).getTime() : NaN;
if (!Number.isNaN(last) && last >= slotMs - SLOT_SLACK_MS) return { claimed: false };
if (shared && state.lastScheduledSentAt) {
const scheduled = new Date(state.lastScheduledSentAt).getTime();
if (!Number.isNaN(scheduled) && Date.now() - scheduled < SCHEDULED_GAP_MS) return { claimed: false };
}
const at = new Date().toISOString();
const next = shared ? { ...state, lastSentAt: at, lastScheduledSentAt: at } : { ...state, lastSentAt: at };
if (await saveStateIfUnchanged(next, version)) {
return { claimed: true, previous: state.lastSentAt, previousScheduled: state.lastScheduledSentAt, at, shared };
}
}
return { claimed: false };
}
/** Give a claimed slot back after a failed send, unless another send landed meanwhile. */
async function releaseSlot(claim: Extract<Claim, { claimed: true }>): Promise<void> {
await updateState((state) => {
if (state.lastSentAt === claim.at) state.lastSentAt = claim.previous;
if (claim.shared && state.lastScheduledSentAt === claim.at) state.lastScheduledSentAt = claim.previousScheduled ?? null;
}).catch(() => {});
}
/**
* Send one heartbeat. A scheduled send passes its `slot` and is skipped
* when another replica already sent for it; the admin's "send now" passes
* none and always sends.
*/
export async function sendOnce(opts?: { reason?: string; slot?: string }): Promise<{
ok: boolean;
status?: number;
error?: string;
@@ -37,8 +107,19 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{
return { ok: false, error: `endpoint blocked: ${guard.reason}` };
}
const payload = await buildPayload();
let claim: Claim | null = null;
if (opts?.slot) {
claim = await claimSlot(opts.slot);
if (!claim.claimed) {
logger.info('telemetry: heartbeat skipped, already sent for this slot', { slot: opts.slot });
return { ok: false, error: 'already sent' };
}
}
try {
// Inside the try: a payload that can't be built (store down) gives the
// claimed slot back instead of losing the day's heartbeat.
const payload = await buildPayload();
// Never follows redirects: the endpoint guard only vetted the URL we
// were given, so a 3xx to an internal host would bypass it.
const res = await fetchTelemetryTarget(endpoint, {
@@ -48,11 +129,11 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{
signal: AbortSignal.timeout(5000),
});
const ok = res.ok;
if (ok) {
const next = await loadState();
next.lastSentAt = new Date().toISOString();
await saveState(next);
if (ok && !claim) {
const at = new Date().toISOString();
await updateState((next) => { next.lastSentAt = at; });
}
if (!ok && claim?.claimed) await releaseSlot(claim);
logger.info('telemetry: heartbeat', {
ok, status: res.status, reason: opts?.reason ?? 'scheduled',
});
@@ -60,28 +141,40 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
logger.warn('telemetry: heartbeat failed', { error: msg });
if (claim?.claimed) await releaseSlot(claim);
return { ok: false, error: msg };
}
}
async function scheduleNext(delayMs: number): Promise<void> {
if (currentTimer) clearTimeout(currentTimer);
async function scheduleNext(delayMs: number, generation: number): Promise<void> {
const t = timer();
if (t.generation !== generation) return; // stopped meanwhile
if (t.current) clearTimeout(t.current);
const at = new Date(Date.now() + delayMs).toISOString();
const state = await loadState();
state.nextScheduledAt = at;
await saveState(state);
currentTimer = setTimeout(() => { void tick(); }, delayMs);
// The timer first: a state write that fails (store down) must not end
// the chain of heartbeats.
t.current = setTimeout(() => { void tick(at, generation); }, delayMs);
// Don't keep the process alive just for this.
currentTimer.unref?.();
t.current.unref?.();
try {
await updateState((state) => { state.nextScheduledAt = at; });
} catch (err) {
logger.warn('telemetry: could not store the schedule', { error: err instanceof Error ? err.message : String(err) });
}
}
async function tick(): Promise<void> {
await sendOnce({ reason: 'scheduled' });
await scheduleNext(jitteredDelay(DAY_MS));
async function tick(slot: string, generation: number): Promise<void> {
try {
await sendOnce({ reason: 'scheduled', slot });
} catch (err) {
logger.warn('telemetry: heartbeat failed', { error: err instanceof Error ? err.message : String(err) });
}
await scheduleNext(jitteredDelay(DAY_MS), generation);
}
// Called from instrumentation. Idempotent.
// Run by the scheduler leader (lib/scheduler-leader.ts). Idempotent.
export async function startScheduler(): Promise<void> {
const generation = timer().generation;
const { consent } = await effectiveConsent();
if (consent !== 'on') {
logger.info('telemetry: scheduler not started', { consent });
@@ -99,7 +192,7 @@ export async function startScheduler(): Promise<void> {
const remaining = new Date(state.nextScheduledAt).getTime() - Date.now();
if (remaining > 0) delay = Math.min(remaining, DAY_MS + JITTER_MS);
}
await scheduleNext(delay);
await scheduleNext(delay, generation);
logger.info('telemetry: scheduler started', {
nextInMs: delay,
endpoint: state.endpoint,
@@ -107,12 +200,15 @@ export async function startScheduler(): Promise<void> {
}
export async function stopScheduler(): Promise<void> {
if (currentTimer) clearTimeout(currentTimer);
currentTimer = null;
const t = timer();
t.generation++;
if (t.current) clearTimeout(t.current);
t.current = null;
}
// Called when consent flips on/off or the collector URL changes via the UI.
// The replica that runs the jobs (any replica's change reaches it) restarts
// the timer from the stored state; in local mode that is this process, at once.
export async function reschedule(): Promise<void> {
await stopScheduler();
await startScheduler();
await announceChange('telemetry');
}
+127 -38
View File
@@ -1,8 +1,8 @@
import { readFile, writeFile, mkdir, rename } from 'node:fs/promises';
import { existsSync } from 'node:fs';
import path from 'node:path';
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 { DEFAULT_ENDPOINT } from './types';
import { isNoThirdPartyTrafficOn } from '@/lib/admin/third-party-mode';
@@ -12,8 +12,26 @@ export function getDir(): string {
path.join(process.cwd(), 'data', 'telemetry');
}
function statePath(): string { return path.join(getDir(), 'state.json'); }
function idPath(): string { return path.join(getDir(), '.telemetry-id'); }
/**
* Telemetry's documents (#1066): `state.json`, `logins.json` and the
* instance id `.telemetry-id`. Local mode: today's files in the telemetry
* dir. Shared mode: one copy for the whole deployment, so every replica
* reports as the same instance with the same counts.
*/
export const TELEMETRY_COLLECTION: CollectionOptions = {
local: {
dir: () => getDir(),
idOf: (relative) => (['state.json', 'logins.json', '.telemetry-id'].includes(relative) ? relative : null),
},
};
export const STATE_DOC = 'state.json';
const ID_DOC = '.telemetry-id';
const UUID = /^[0-9a-f-]{36}$/i;
export function telemetryDocs() {
return getStore().collection('telemetry', TELEMETRY_COLLECTION);
}
function envOverride(): ConsentState | null {
const v = (process.env.BULWARK_TELEMETRY ?? '').toLowerCase();
@@ -26,21 +44,29 @@ function envOverride(): ConsentState | null {
return null;
}
export async function ensureDir(): Promise<void> {
if (!existsSync(getDir())) await mkdir(getDir(), { recursive: true });
}
/**
* The instance id, created once: of several replicas (or module copies)
* asking at once, every one gets the id the first one stored.
*/
export async function getInstanceId(): Promise<string> {
await ensureDir();
try {
const id = (await readFile(idPath(), 'utf8')).trim();
if (/^[0-9a-f-]{36}$/i.test(id)) return id;
} catch { /* generate fresh */ }
const fresh = randomUUID();
const tmp = idPath() + '.tmp';
await writeFile(tmp, fresh, 'utf8');
await rename(tmp, idPath());
return fresh;
const docs = telemetryDocs();
for (let attempt = 0; attempt < 3; attempt++) {
let current;
try {
current = await docs.get(ID_DOC);
} catch (error) {
if (isStoreUnavailable(error)) throw error;
current = null; // unreadable: make a fresh one, as before
}
const id = current?.data.toString('utf8').trim();
if (id && UUID.test(id)) return id;
const fresh = randomUUID();
const result = await docs.put(ID_DOC, fresh, { ifVersion: current ? current.version : null });
if (result.ok) return fresh;
const won = result.current?.data.toString('utf8').trim();
if (won && UUID.test(won)) return won;
}
throw new Error('telemetry: could not store an instance id');
}
// Default consent is 'off' - telemetry is opt-in. Admins can enable it during
@@ -56,32 +82,95 @@ const DEFAULTS: TelemetryStateFile = {
nextScheduledAt: null,
};
export async function loadState(): Promise<TelemetryStateFile> {
await ensureDir();
function serialize(state: TelemetryStateFile): string {
return JSON.stringify(state, null, 2);
}
function parse(data: Buffer): TelemetryStateFile {
return { ...DEFAULTS, ...(JSON.parse(data.toString('utf8')) as Partial<TelemetryStateFile>) };
}
function readFailed(err: unknown): TelemetryStateFile {
// Unreadable is not missing: answer with the defaults (so nothing is
// sent on a guess) but leave the file alone, or a passing read error
// would switch the admin's consent off for good.
logger.warn('telemetry: state read failed', {
error: err instanceof Error ? err.message : String(err),
});
return { ...DEFAULTS };
}
/**
* The state with its stored version (null when it couldn't be read), for a
* compare-and-set. Throws StoreUnavailableError when the shared store is down.
*/
export async function loadStateVersioned(): Promise<{ state: TelemetryStateFile; version: string | null }> {
const docs = telemetryDocs();
let current;
try {
const raw = await readFile(statePath(), 'utf8');
const parsed = JSON.parse(raw) as Partial<TelemetryStateFile>;
return { ...DEFAULTS, ...parsed };
current = await docs.get(STATE_DOC);
} catch (err) {
if ((err as NodeJS.ErrnoException).code !== 'ENOENT') {
logger.warn('telemetry: state read failed', {
error: err instanceof Error ? err.message : String(err),
});
}
// First-ever load on a fresh install: persist the default-off state so the
// instance id and (lack of) consent are stable across restarts. The admin
// can opt in later via the UI or the BULWARK_TELEMETRY env var.
const fresh: TelemetryStateFile = { ...DEFAULTS };
await saveState(fresh);
return fresh;
if (isStoreUnavailable(err)) throw err;
return { state: readFailed(err), version: null };
}
if (current) {
try {
return { state: parse(current.data), version: current.version };
} catch (err) {
return { state: readFailed(err), version: null };
}
}
// First-ever load on a fresh install: persist the default-off state so the
// instance id and (lack of) consent are stable across restarts. The admin
// can opt in later via the UI or the BULWARK_TELEMETRY env var.
const fresh: TelemetryStateFile = { ...DEFAULTS };
const written = await docs.put(STATE_DOC, serialize(fresh), { ifVersion: null });
if (written.ok) return { state: fresh, version: written.version };
// Another replica created it first.
return written.current
? { state: parse(written.current.data), version: written.current.version }
: { state: fresh, version: null };
}
export async function loadState(): Promise<TelemetryStateFile> {
return (await loadStateVersioned()).state;
}
export async function saveState(state: TelemetryStateFile): Promise<void> {
await ensureDir();
const tmp = statePath() + '.tmp';
await writeFile(tmp, JSON.stringify(state, null, 2), 'utf8');
await rename(tmp, statePath());
await telemetryDocs().put(STATE_DOC, serialize(state));
}
/**
* Change some fields of the stored state without undoing a write that
* landed meanwhile (another replica scheduling, the admin changing consent).
* `change` may run more than once. Returns the state written. A state file
* that doesn't parse throws StoredDataUnreadableError and is left as it is.
*/
export async function updateState(change: (state: TelemetryStateFile) => void): Promise<TelemetryStateFile> {
let next: TelemetryStateFile = { ...DEFAULTS };
await telemetryDocs().update(STATE_DOC, (current) => {
let state: TelemetryStateFile;
try {
state = current ? parse(current) : { ...DEFAULTS };
} catch (err) {
// Unreadable is not missing, here as on reads: writing the defaults
// with one change would throw away the admin's consent and endpoint.
throw new StoredDataUnreadableError('telemetry state', { cause: err });
}
change(state);
next = state;
return serialize(state);
});
return next;
}
/**
* Write `state` only if the stored state is still at `version` (from
* loadStateVersioned). False when something else wrote first.
*/
export async function saveStateIfUnchanged(state: TelemetryStateFile, version: string | null): Promise<boolean> {
if (version === null) return false;
return (await telemetryDocs().put(STATE_DOC, serialize(state), { ifVersion: version })).ok;
}
// Effective consent: the no-third-party mode wins over everything (#903),
+5
View File
@@ -41,6 +41,11 @@ export interface TelemetryStateFile {
endpoint: string;
consentedAt: string | null;
lastSentAt: string | null;
/**
* With a shared store only: when the last scheduled heartbeat went out.
* Unlike lastSentAt, the admin's "send now" doesn't set it.
*/
lastScheduledSentAt?: string | null;
nextScheduledAt: string | null;
}
+1 -1
View File
@@ -1,6 +1,6 @@
export { startScheduler, stopScheduler, checkOnce } from './sender';
export {
loadState, saveState, effectiveEndpoint, disabledByEnv,
loadState, saveState, updateState, effectiveEndpoint, disabledByEnv,
getCurrentVersion, isStatusStale, freshStatus,
} from './state';
export { fetchStatus, parseStatus } from './fetcher';
+59 -25
View File
@@ -1,7 +1,9 @@
import { logger } from '@/lib/logger';
import {
disabledByEnv, effectiveEndpoint, getCurrentVersion, isStatusStale, loadState, saveState,
disabledByEnv, effectiveEndpoint, freshStatus, getCurrentVersion, isStatusStale, loadState, saveState,
updateState, withVersionStatus,
} from './state';
import { isSharedMode } from '@/lib/shared-store';
import { fetchStatus } from './fetcher';
import type { UpdateStatus } from './types';
import { isNoThirdPartyTrafficOn } from '@/lib/admin/third-party-mode';
@@ -11,7 +13,21 @@ const JITTER_MS = 5 * 60 * 1000; // ± 5 min, keeps containers from syncin
const FIRST_DELAY_MS = 30 * 1000; // first check ~30s after boot
const FAILURE_BACKOFF_MS = 15 * 60 * 1000; // after a failed fetch, retry in 15 min
let currentTimer: NodeJS.Timeout | null = null;
// On globalThis: Next loads this module once per bundle (instrumentation,
// each route), and a reschedule from a route must clear the timer the
// instrumentation copy started, not a copy of its own.
const TIMER_KEY = Symbol.for('bulwark.versionCheck.timer');
/**
* `generation` goes up on every stop. A start or a tick captures it and
* arms a timer only while it is unchanged, so a check or send still in
* flight when the scheduler stops (this replica lost the lead) can't set a
* new timer afterwards.
*/
interface TimerState { current: NodeJS.Timeout | null; generation: number }
function timer(): TimerState {
const g = globalThis as typeof globalThis & { [TIMER_KEY]?: TimerState };
return (g[TIMER_KEY] ??= { current: null, generation: 0 });
}
/** What checkOnce reports while NO_THIRD_PARTY_TRAFFIC is on. */
export const NO_THIRD_PARTY_ERROR = 'off: no third-party traffic';
@@ -39,13 +55,15 @@ export async function checkOnce(opts?: { reason?: string }): Promise<{
const now = new Date().toISOString();
const result = await fetchStatus(endpoint, current);
const next = await loadState();
next.lastCheckedAt = now;
if (result.ok) {
next.lastSuccessAt = now;
next.status = result.status;
}
await saveState(next);
const shared = isSharedMode();
await updateState((next) => {
next.lastCheckedAt = now;
if (result.ok) {
next.lastSuccessAt = now;
next.status = result.status;
if (shared) next.statuses = withVersionStatus(next.statuses, result.status);
}
});
logger.info('version-check: ran', {
ok: result.ok,
@@ -57,28 +75,38 @@ export async function checkOnce(opts?: { reason?: string }): Promise<{
return { ok: false, error: result.error };
}
async function scheduleNext(delayMs: number): Promise<void> {
if (currentTimer) clearTimeout(currentTimer);
async function scheduleNext(delayMs: number, generation: number): Promise<void> {
const t = timer();
if (t.generation !== generation) return; // stopped meanwhile
if (t.current) clearTimeout(t.current);
const at = new Date(Date.now() + delayMs).toISOString();
const state = await loadState();
state.nextScheduledAt = at;
await saveState(state);
currentTimer = setTimeout(() => { void tick(); }, delayMs);
// The timer first: a state write that fails (store down) must not end
// the chain of checks.
t.current = setTimeout(() => { void tick(generation); }, delayMs);
// Don't keep the process alive just for this.
currentTimer.unref?.();
t.current.unref?.();
try {
await updateState((state) => { state.nextScheduledAt = at; });
} catch (err) {
logger.warn('version-check: could not store the schedule', { error: err instanceof Error ? err.message : String(err) });
}
}
async function tick(): Promise<void> {
const result = await checkOnce({ reason: 'scheduled' });
async function tick(generation: number): Promise<void> {
const result = await checkOnce({ reason: 'scheduled' }).catch((err: unknown) => ({
ok: false as const, error: err instanceof Error ? err.message : String(err),
}));
// The scheduler keeps running while third-party traffic is off, so turning
// the switch off at runtime resumes the checks; checkOnce skips meanwhile.
const skipped = result.error === NO_THIRD_PARTY_ERROR;
const delay = result.ok || skipped ? jitteredDelay(HOUR_MS) : FAILURE_BACKOFF_MS;
await scheduleNext(delay);
await scheduleNext(delay, generation);
}
// Idempotent - safe to call from instrumentation hot-reload in dev.
// Run by the scheduler leader (lib/scheduler-leader.ts). Idempotent - safe
// to call from instrumentation hot-reload in dev.
export async function startScheduler(): Promise<void> {
const generation = timer().generation;
if (disabledByEnv()) {
logger.info('version-check: scheduler not started (disabled by env)');
return;
@@ -92,8 +120,12 @@ export async function startScheduler(): Promise<void> {
// version we no longer run, and its schedule would keep that stale answer
// (e.g. "update available" for the version we just upgraded from) on screen
// for up to an hour. Drop it and check promptly instead (#913).
const stale = isStatusStale(state.status);
if (stale) {
// With a shared store, replicas on another version may still be serving
// that status: keep it, and only check promptly when there is none for
// this version yet.
const shared = isSharedMode();
const stale = shared ? !freshStatus(state) && state.status !== null : isStatusStale(state.status);
if (stale && !shared) {
logger.info('version-check: discarding status from a previous version', {
cached: state.status?.current ?? null,
running: getCurrentVersion(),
@@ -111,11 +143,13 @@ export async function startScheduler(): Promise<void> {
const remaining = new Date(state.nextScheduledAt).getTime() - Date.now();
if (remaining > 0) delay = Math.min(remaining, HOUR_MS + JITTER_MS);
}
await scheduleNext(delay);
await scheduleNext(delay, generation);
logger.info('version-check: scheduler started', { nextInMs: delay });
}
export async function stopScheduler(): Promise<void> {
if (currentTimer) clearTimeout(currentTimer);
currentTimer = null;
const t = timer();
t.generation++;
if (t.current) clearTimeout(t.current);
t.current = null;
}
+71 -19
View File
@@ -1,7 +1,7 @@
import { readFile, writeFile, mkdir, rename } from 'node:fs/promises';
import { existsSync } from 'node:fs';
import path from 'node:path';
import { logger } from '@/lib/logger';
import { getStore, type CollectionOptions } from '@/lib/shared-store';
import { isStoreUnavailable } from '@/lib/stored-data-errors';
import type { UpdateStatus, VersionCheckStateFile } from './types';
import { DEFAULT_VERSION_ENDPOINT } from './types';
@@ -10,7 +10,23 @@ function getDir(): string {
path.join(process.cwd(), 'data', 'version-check');
}
function statePath(): string { return path.join(getDir(), 'state.json'); }
/**
* `state.json` (#1066). Local mode: today's file in the version-check dir.
* Shared mode: one state for the deployment, written by the replica that
* runs the scheduled check and read by all.
*/
export const VERSION_CHECK_COLLECTION: CollectionOptions = {
local: {
dir: () => getDir(),
idOf: (relative) => (relative === 'state.json' ? relative : null),
},
};
export const STATE_DOC = 'state.json';
function docs() {
return getStore().collection('version-check', VERSION_CHECK_COLLECTION);
}
const DEFAULTS: VersionCheckStateFile = {
endpoint: DEFAULT_VERSION_ENDPOINT,
@@ -20,31 +36,44 @@ const DEFAULTS: VersionCheckStateFile = {
status: null,
};
export async function ensureDir(): Promise<void> {
if (!existsSync(getDir())) await mkdir(getDir(), { recursive: true });
function parse(data: Buffer): VersionCheckStateFile {
return { ...DEFAULTS, ...(JSON.parse(data.toString('utf8')) as Partial<VersionCheckStateFile>) };
}
function serialize(state: VersionCheckStateFile): string {
return JSON.stringify(state, null, 2);
}
/** The stored state, or the defaults. Throws StoreUnavailableError when the shared store is down. */
export async function loadState(): Promise<VersionCheckStateFile> {
await ensureDir();
try {
const raw = await readFile(statePath(), 'utf8');
const parsed = JSON.parse(raw) as Partial<VersionCheckStateFile>;
return { ...DEFAULTS, ...parsed };
const current = await docs().get(STATE_DOC);
return current ? parse(current.data) : { ...DEFAULTS };
} catch (err) {
if ((err as NodeJS.ErrnoException).code !== 'ENOENT') {
logger.warn('version-check: state read failed', {
error: err instanceof Error ? err.message : String(err),
});
}
if (isStoreUnavailable(err)) throw err;
logger.warn('version-check: state read failed', {
error: err instanceof Error ? err.message : String(err),
});
return { ...DEFAULTS };
}
}
export async function saveState(state: VersionCheckStateFile): Promise<void> {
await ensureDir();
const tmp = statePath() + '.tmp';
await writeFile(tmp, JSON.stringify(state, null, 2), 'utf8');
await rename(tmp, statePath());
await docs().put(STATE_DOC, serialize(state));
}
/** Change some fields against what is stored now. `change` may run more than once. */
export async function updateState(change: (state: VersionCheckStateFile) => void): Promise<void> {
await docs().update(STATE_DOC, (current) => {
let state: VersionCheckStateFile;
try {
state = current ? parse(current) : { ...DEFAULTS };
} catch {
state = { ...DEFAULTS };
}
change(state);
return serialize(state);
});
}
export function disabledByEnv(): boolean {
@@ -78,6 +107,29 @@ export function isStatusStale(
return status.current !== current;
}
/** How many running versions keep their own status (a rolling update runs two). */
const KEPT_VERSIONS = 4;
/**
* The status for the version this replica runs. With a shared store,
* replicas on two versions during a rolling update share one state; each
* reads the status checked for its own version from `statuses`, so the
* banner doesn't come and go as the other version's checks land.
*/
export function freshStatus(state: VersionCheckStateFile): UpdateStatus | null {
return isStatusStale(state.status) ? null : state.status;
if (!isStatusStale(state.status)) return state.status;
const own = state.statuses?.[getCurrentVersion()];
return own && !isStatusStale(own) ? own : null;
}
/** `statuses` with `status` added for its version, keeping the newest few. */
export function withVersionStatus(
statuses: Record<string, UpdateStatus> | undefined,
status: UpdateStatus,
): Record<string, UpdateStatus> {
const next = { ...(statuses ?? {}), [status.current]: status };
const newest = Object.entries(next)
.sort(([, a], [, b]) => b.checkedAt.localeCompare(a.checkedAt))
.slice(0, KEPT_VERSIONS);
return Object.fromEntries(newest);
}
+5
View File
@@ -20,6 +20,11 @@ export interface VersionCheckStateFile {
lastSuccessAt: string | null;
nextScheduledAt: string | null;
status: UpdateStatus | null;
/**
* With a shared store only: the last status per running version, so
* replicas on two versions during a rolling update each keep their own.
*/
statuses?: Record<string, UpdateStatus>;
}
export const DEFAULT_VERSION_ENDPOINT = 'https://version.telemetry.bulwarkmail.org/';