mirror of
https://github.com/bulwarkmail/webmail.git
synced 2026-11-03 08:15:38 +00:00
feat: share admin config and policy between instances and spread changes within seconds #1066
This commit is contained in:
@@ -8,6 +8,8 @@ import { clearSetupToken } from '@/lib/setup/token';
|
||||
import { ensureConfigDir, getConfigPath } from '@/lib/admin/paths';
|
||||
import { auditLog } from '@/lib/admin/audit';
|
||||
import { logger } from '@/lib/logger';
|
||||
import { isSharedMode } from '@/lib/shared-store';
|
||||
import { adminConfigDocs } from '@/lib/admin/admin-store';
|
||||
|
||||
export const dynamic = 'force-dynamic';
|
||||
|
||||
@@ -77,12 +79,16 @@ export async function POST(request: NextRequest) {
|
||||
|
||||
// 3. Optional advisory lock marker.
|
||||
if (lockConfig) {
|
||||
await ensureConfigDir();
|
||||
await writeFile(
|
||||
getConfigPath('.config-locked'),
|
||||
new Date().toISOString(),
|
||||
'utf-8',
|
||||
);
|
||||
if (isSharedMode()) {
|
||||
await adminConfigDocs().put('.config-locked', new Date().toISOString());
|
||||
} else {
|
||||
await ensureConfigDir();
|
||||
await writeFile(
|
||||
getConfigPath('.config-locked'),
|
||||
new Date().toISOString(),
|
||||
'utf-8',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// 4. Destroy the setup token. Any other browser holding the cookie is
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
// @vitest-environment node
|
||||
import { afterAll, describe, expect, it, vi } from 'vitest';
|
||||
import { rename, rm, writeFile } from 'node:fs/promises';
|
||||
import path from 'node:path';
|
||||
|
||||
// With ADMIN_CONFIG_READONLY (a Kubernetes ConfigMap) every replica reads
|
||||
// config.json and policy.json from the mounted files, also with a shared
|
||||
// store, and a changed ConfigMap reaches them without a restart.
|
||||
|
||||
// Hoisted: isConfigReadOnly() reads the env once.
|
||||
const dir = await vi.hoisted(async () => {
|
||||
const { mkdtempSync, writeFileSync } = await import('node:fs');
|
||||
const { tmpdir } = await import('node:os');
|
||||
const { join } = await import('node:path');
|
||||
const config = mkdtempSync(join(tmpdir(), 'bw-config-shared-ro-'));
|
||||
writeFileSync(join(config, 'config.json'), JSON.stringify({ appName: 'Mounted' }));
|
||||
process.env.ADMIN_CONFIG_DIR = config;
|
||||
process.env.ADMIN_CONFIG_READONLY = 'true';
|
||||
return config;
|
||||
});
|
||||
|
||||
import { configManager } from '../config-manager';
|
||||
import { setStoreForTests } from '@/lib/shared-store';
|
||||
import { separateFakeStore } from './shared-store-fake';
|
||||
|
||||
afterAll(async () => {
|
||||
setStoreForTests(null);
|
||||
vi.restoreAllMocks();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe('read-only config with a shared store', () => {
|
||||
it('reads the mounted files and follows a change to them', async () => {
|
||||
setStoreForTests(separateFakeStore(path.join(dir, 'store')));
|
||||
let now = 1_000_000;
|
||||
vi.spyOn(Date, 'now').mockImplementation(() => now);
|
||||
await configManager.load();
|
||||
expect(configManager.get('appName')).toBe('Mounted');
|
||||
|
||||
await writeFile(path.join(dir, 'config.json.new'), JSON.stringify({ appName: 'Updated ConfigMap' }));
|
||||
await rename(path.join(dir, 'config.json.new'), path.join(dir, 'config.json'));
|
||||
now += 10_000;
|
||||
await configManager.ensureLoaded();
|
||||
expect(configManager.get('appName')).toBe('Updated ConfigMap');
|
||||
|
||||
await expect(configManager.setAdminConfig({ appName: 'x' })).rejects.toThrow(/read-only/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,214 @@
|
||||
// @vitest-environment node
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { mkdtemp, readFile, rm } from 'node:fs/promises';
|
||||
import { existsSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import path from 'node:path';
|
||||
import { NextRequest } from 'next/server';
|
||||
|
||||
// Several replicas share one store (#1066). An admin change saved on one
|
||||
// has to reach the others within seconds, two admins saving different keys
|
||||
// at once both have to win, and a policy page opened before someone else's
|
||||
// save must not put the old policy back.
|
||||
|
||||
vi.mock('@/lib/auth/instance-details', () => ({
|
||||
canSeeInstanceDetails: async () => true,
|
||||
signedInIdentity: async () => ({ admin: true }),
|
||||
}));
|
||||
vi.mock('@/lib/admin/session', () => ({
|
||||
requireAdminAuth: async () => ({ payload: { role: 'admin' } }),
|
||||
getClientIP: () => '127.0.0.1',
|
||||
}));
|
||||
vi.mock('@/lib/admin/audit', () => ({ auditLog: async () => {} }));
|
||||
vi.mock('@/lib/admin/csp-frame-origins', () => ({ getEnabledPluginFrameOrigins: vi.fn(async () => []) }));
|
||||
vi.mock('@/lib/wopi/discovery', () => ({ getWopiEditorOrigins: vi.fn(async () => []) }));
|
||||
vi.mock('next-intl/middleware', async () => {
|
||||
const { NextResponse } = await import('next/server');
|
||||
return { default: () => () => NextResponse.next() };
|
||||
});
|
||||
|
||||
import { configManager } from '../config-manager';
|
||||
import { adminConfigDocs } from '../admin-store';
|
||||
import { forgetExtensionInPolicy } from '../extension-cleanup';
|
||||
import { setBootState, setStoreForTests } from '@/lib/shared-store';
|
||||
import { resetLocalStoreForTests } from '@/lib/shared-store/local';
|
||||
import { INIT_EPOCH, markEpochsSeen, pollChanges, resetChangesForTests } from '@/lib/shared-store/changes';
|
||||
import { GET, PUT } from '@/app/api/admin/policy/route';
|
||||
import { proxy } from '@/proxy';
|
||||
import { separateFakeStore } from './shared-store-fake';
|
||||
|
||||
let dir: string;
|
||||
let store: ReturnType<typeof separateFakeStore>;
|
||||
|
||||
const stored = async (id: string) => {
|
||||
const doc = await adminConfigDocs().get(id);
|
||||
return doc ? JSON.parse(doc.data.toString('utf-8')) : null;
|
||||
};
|
||||
|
||||
/** What another replica's save does: write the stored document, then announce it. */
|
||||
async function otherReplicaSaves(id: 'config.json' | 'policy.json', data: Record<string, unknown>, announce = true) {
|
||||
await adminConfigDocs().put(id, JSON.stringify(data, null, 2));
|
||||
if (announce) await store.epochs.bump(id === 'config.json' ? 'config' : 'policy');
|
||||
}
|
||||
|
||||
/** A second replica's config manager: a fresh module copy with its own instance. */
|
||||
async function otherReplicaManager() {
|
||||
const KEY = Symbol.for('bulwark.admin.configManager');
|
||||
const g = globalThis as Record<symbol, unknown>;
|
||||
const mine = g[KEY];
|
||||
delete g[KEY];
|
||||
vi.resetModules();
|
||||
const { configManager: other } = await import('../config-manager');
|
||||
g[KEY] = mine;
|
||||
await other.load();
|
||||
return other;
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
dir = await mkdtemp(path.join(tmpdir(), 'bw-config-shared-'));
|
||||
vi.stubEnv('ADMIN_CONFIG_DIR', path.join(dir, 'config'));
|
||||
vi.stubEnv('ADMIN_STATE_DIR', path.join(dir, 'state'));
|
||||
resetLocalStoreForTests();
|
||||
resetChangesForTests();
|
||||
store = separateFakeStore(path.join(dir, 'store'));
|
||||
setStoreForTests(store);
|
||||
await store.epochs.bump(INIT_EPOCH);
|
||||
await adminConfigDocs().put('config.json', JSON.stringify({ appName: 'First', setupComplete: true }));
|
||||
await adminConfigDocs().put('policy.json', JSON.stringify({ features: { pluginsEnabled: false } }));
|
||||
await markEpochsSeen();
|
||||
await configManager.load();
|
||||
setBootState('ready');
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
setStoreForTests(null);
|
||||
resetChangesForTests();
|
||||
resetLocalStoreForTests();
|
||||
vi.unstubAllEnvs();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe('config manager with a shared store', () => {
|
||||
it('loads from the store and writes nothing to the local config dir', async () => {
|
||||
expect(configManager.get('appName')).toBe('First');
|
||||
await configManager.setAdminConfig({ loginShowVersion: false });
|
||||
expect(await stored('config.json')).toEqual({ appName: 'First', setupComplete: true, loginShowVersion: false });
|
||||
expect(existsSync(path.join(dir, 'config', 'config.json'))).toBe(false);
|
||||
});
|
||||
|
||||
it('picks up a config change from another replica on the next poll', async () => {
|
||||
await otherReplicaSaves('config.json', { appName: 'Second name', setupComplete: true });
|
||||
await pollChanges();
|
||||
expect(configManager.get('appName')).toBe('Second name');
|
||||
});
|
||||
|
||||
it('gives the proxy a policy change from another replica', async () => {
|
||||
const sandbox = () => proxy(new NextRequest('http://localhost:3000/plugin-sandbox/frame', { headers: { 'sec-fetch-dest': 'iframe' } }));
|
||||
expect((await sandbox()).status).toBe(403);
|
||||
|
||||
await otherReplicaSaves('policy.json', { features: { pluginsEnabled: true } });
|
||||
await pollChanges();
|
||||
expect(configManager.getPolicy().features.pluginsEnabled).toBe(true);
|
||||
expect((await sandbox()).status).not.toBe(403);
|
||||
});
|
||||
|
||||
it('keeps both of two admins\' saves of different keys on two replicas', async () => {
|
||||
const other = await otherReplicaManager();
|
||||
await Promise.all([
|
||||
configManager.setAdminConfig({ appName: 'From A' }),
|
||||
other.setAdminConfig({ loginShowVersion: false }),
|
||||
]);
|
||||
expect(await stored('config.json')).toEqual({ appName: 'From A', setupComplete: true, loginShowVersion: false });
|
||||
|
||||
// A replica that starts now reads both.
|
||||
const third = await otherReplicaManager();
|
||||
expect(third.get('appName')).toBe('From A');
|
||||
expect(third.get('loginShowVersion')).toBe(false);
|
||||
});
|
||||
|
||||
it('merges a save into the stored config, not into this replica\'s older copy', async () => {
|
||||
// Another replica saved, and its notice hasn't arrived yet.
|
||||
await otherReplicaSaves('config.json', { appName: 'Second name', setupComplete: true }, false);
|
||||
await configManager.removeAdminOverride('loginShowVersion');
|
||||
await configManager.setAdminConfig({ loginShowVersion: true });
|
||||
expect(await stored('config.json')).toEqual({ appName: 'Second name', setupComplete: true, loginShowVersion: true });
|
||||
expect(configManager.get('appName')).toBe('Second name');
|
||||
});
|
||||
|
||||
it('marks setup complete in the store, so every replica leaves the wizard', async () => {
|
||||
await otherReplicaSaves('config.json', { appName: 'First' });
|
||||
await pollChanges();
|
||||
expect(configManager.isSetupComplete()).toBe(false);
|
||||
await configManager.markSetupComplete();
|
||||
const other = await otherReplicaManager();
|
||||
expect(other.isSetupComplete()).toBe(true);
|
||||
});
|
||||
|
||||
it('keeps serving the last loaded config while the store is down', async () => {
|
||||
store.down = true;
|
||||
await otherReplicaSaves('config.json', { appName: 'Unreachable' }).catch(() => {});
|
||||
await store.epochs.bump('config');
|
||||
await pollChanges();
|
||||
await configManager.ensureLoaded();
|
||||
expect(configManager.get('appName')).toBe('First');
|
||||
await expect(configManager.setAdminConfig({ appName: 'Nope' })).rejects.toThrow(/down/);
|
||||
|
||||
store.down = false;
|
||||
await otherReplicaSaves('config.json', { appName: 'Back', setupComplete: true });
|
||||
await pollChanges();
|
||||
expect(configManager.get('appName')).toBe('Back');
|
||||
});
|
||||
});
|
||||
|
||||
describe('policy saves with a shared store', () => {
|
||||
async function load() {
|
||||
const res = await GET(new NextRequest('https://mail.example/api/admin/policy'));
|
||||
return { policy: await res.json(), etag: res.headers.get('ETag')! };
|
||||
}
|
||||
const save = (policy: Record<string, unknown>, etag?: string) => PUT(new NextRequest('https://mail.example/api/admin/policy', {
|
||||
method: 'PUT',
|
||||
headers: { 'content-type': 'application/json', ...(etag ? { 'if-match': etag } : {}) },
|
||||
body: JSON.stringify(policy),
|
||||
}));
|
||||
|
||||
it('refuses a save from a page opened before another replica saved, and keeps that save', async () => {
|
||||
const page = await load();
|
||||
// Saved on another replica; this one hasn't heard of it yet.
|
||||
await otherReplicaSaves('policy.json', { features: { pluginsEnabled: true, calendarEnabled: false } }, false);
|
||||
|
||||
const res = await save({ ...page.policy, features: { ...page.policy.features, filesEnabled: false } }, page.etag);
|
||||
expect(res.status).toBe(409);
|
||||
expect((await res.json()).error).toMatch(/^Someone else changed the policy after you opened this page/);
|
||||
expect(await stored('policy.json')).toEqual({ features: { pluginsEnabled: true, calendarEnabled: false } });
|
||||
// The refusal brought this replica up to date.
|
||||
expect(configManager.getPolicy().features.pluginsEnabled).toBe(true);
|
||||
|
||||
const fresh = await load();
|
||||
const ok = await save({ ...fresh.policy, features: { ...fresh.policy.features, filesEnabled: false } }, fresh.etag);
|
||||
expect(ok.status).toBe(200);
|
||||
expect((await stored('policy.json')).features).toMatchObject({ pluginsEnabled: true, calendarEnabled: false, filesEnabled: false });
|
||||
});
|
||||
|
||||
it('drops a deleted plugin from the stored policy without undoing another replica\'s save', async () => {
|
||||
await otherReplicaSaves('policy.json', { features: { calendarEnabled: false }, forceEnabledPlugins: ['demo', 'other'] }, false);
|
||||
await forgetExtensionInPolicy('plugin', 'demo');
|
||||
const policy = await stored('policy.json');
|
||||
expect(policy.forceEnabledPlugins).toEqual(['other']);
|
||||
expect(policy.features.calendarEnabled).toBe(false);
|
||||
});
|
||||
|
||||
it('answers a page that loaded before any policy was saved with 409 once one is', async () => {
|
||||
await adminConfigDocs().delete('policy.json');
|
||||
await configManager.reload();
|
||||
const page = await load();
|
||||
expect(page.etag).toBe('"default"');
|
||||
await otherReplicaSaves('policy.json', { features: { calendarEnabled: false } }, false);
|
||||
expect((await save(page.policy, page.etag)).status).toBe(409);
|
||||
});
|
||||
|
||||
it('leaves the local policy file alone', async () => {
|
||||
const page = await load();
|
||||
expect((await save(page.policy, page.etag)).status).toBe(200);
|
||||
await expect(readFile(path.join(dir, 'config', 'policy.json'))).rejects.toThrow();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,50 @@
|
||||
import path from 'node:path';
|
||||
import { isConfigReadOnly } from '../paths';
|
||||
import { localCollection, localLog } from '@/lib/shared-store/local';
|
||||
import { StoreUnavailableError, type CollectionOptions, type LogOptions, type SharedStore } from '@/lib/shared-store';
|
||||
import { fakeSharedStore } from '@/lib/shared-store/__tests__/fake-shared';
|
||||
|
||||
/**
|
||||
* The fake shared store (lib/shared-store/__tests__/fake-shared.ts) with its
|
||||
* documents and logs under `root` instead of the dirs local mode uses, as a
|
||||
* Redis would hold them apart from this replica's files. Without that, code
|
||||
* that still read and wrote its local files would pass for shared.
|
||||
*
|
||||
* `down` makes every document, log and counter call fail as an outage would.
|
||||
*/
|
||||
export function separateFakeStore(root: string): SharedStore & { down: boolean } {
|
||||
const fake = fakeSharedStore();
|
||||
const guard = <T>(fn: () => Promise<T>): Promise<T> =>
|
||||
fake.down ? Promise.reject(new StoreUnavailableError('down (test)')) : fn();
|
||||
const counters = fake.counters;
|
||||
return Object.assign(fake, {
|
||||
collection(name: string, options: CollectionOptions) {
|
||||
// Read-only config comes from the mounted files, as in the Redis store.
|
||||
if (options.configDir && isConfigReadOnly()) return localCollection(name, options);
|
||||
const docs = localCollection(name, { ...options, local: { ...options.local, dir: () => path.join(root, 'docs', name) } });
|
||||
return {
|
||||
name,
|
||||
get: (id: string) => guard(() => docs.get(id)),
|
||||
put: (...args: Parameters<typeof docs.put>) => guard(() => docs.put(...args)),
|
||||
update: (...args: Parameters<typeof docs.update>) => guard(() => docs.update(...args)),
|
||||
delete: (...args: Parameters<typeof docs.delete>) => guard(() => docs.delete(...args)),
|
||||
list: (prefix?: string) => guard(() => docs.list(prefix)),
|
||||
};
|
||||
},
|
||||
log(name: string, options: LogOptions) {
|
||||
const log = localLog(name, { ...options, local: { ...options.local, path: () => path.join(root, 'logs', `${name}.log`) } });
|
||||
return {
|
||||
append: (line: string) => guard(() => log.append(line)),
|
||||
readAll: () => guard(() => log.readAll()),
|
||||
length: () => guard(() => log.length()),
|
||||
};
|
||||
},
|
||||
counters: {
|
||||
reserve: (...args: Parameters<typeof counters.reserve>) => guard(() => counters.reserve(...args)),
|
||||
refund: (...args: Parameters<typeof counters.refund>) => guard(() => counters.refund(...args)),
|
||||
hit: (...args: Parameters<typeof counters.hit>) => guard(() => counters.hit(...args)),
|
||||
get: (...args: Parameters<typeof counters.get>) => guard(() => counters.get(...args)),
|
||||
reset: (key: string) => guard(() => counters.reset(key)),
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
import { getStore, type CollectionOptions, type DocCollection } from '@/lib/shared-store';
|
||||
import { getConfigDir, getStateDir } from './paths';
|
||||
|
||||
/**
|
||||
* Where the admin dashboard's own documents live (#1066). Local mode: the
|
||||
* admin config dir and state dir, with today's file names and temp names
|
||||
* (`<file>.tmp`). Shared mode: the shared store, except that read-only
|
||||
* config (ADMIN_CONFIG_READONLY) is read from the mounted files.
|
||||
*/
|
||||
|
||||
const fixedTmpName = (target: string) => `${target}.tmp`;
|
||||
|
||||
/** config.json, policy.json, admin.json, .config-locked. */
|
||||
export const ADMIN_CONFIG_COLLECTION = 'admin-config';
|
||||
export const adminConfigOptions: CollectionOptions = {
|
||||
// Owner-only: config.json carries the session secret, admin.json the password hash.
|
||||
local: { dir: () => getConfigDir(), mode: 0o600, tmpName: fixedTmpName },
|
||||
configDir: true,
|
||||
};
|
||||
|
||||
/** admin-state.json, and other runtime state of the admin dashboard. */
|
||||
export const ADMIN_STATE_COLLECTION = 'admin-state';
|
||||
export const adminStateOptions: CollectionOptions = {
|
||||
local: { dir: () => getStateDir(), tmpName: fixedTmpName },
|
||||
};
|
||||
|
||||
export function adminConfigDocs(): DocCollection {
|
||||
return getStore().collection(ADMIN_CONFIG_COLLECTION, adminConfigOptions);
|
||||
}
|
||||
|
||||
export function adminStateDocs(): DocCollection {
|
||||
return getStore().collection(ADMIN_STATE_COLLECTION, adminStateOptions);
|
||||
}
|
||||
+165
-5
@@ -2,7 +2,7 @@ import { readFile, writeFile, rename, stat } from 'node:fs/promises';
|
||||
import { configureLogger, logger } from '@/lib/logger';
|
||||
import { readFileEnv } from '@/lib/read-file-env';
|
||||
import { CONFIG_ENV_MAP, DEFAULT_FEATURE_GATES, DEFAULT_POLICY, DEFAULT_THEME_POLICY, type SettingsPolicy } from './types';
|
||||
import { ensureConfigDir, getConfigPath, assertWritable } from './paths';
|
||||
import { ensureConfigDir, getConfigPath, assertWritable, isConfigReadOnly } from './paths';
|
||||
import { isValidRelayUrl, normalizeRelayUrl } from '@/lib/push-relays';
|
||||
import { sanitizeDefaultSidebarApps } from '@/lib/sidebar-apps';
|
||||
import { sanitizePolicyDefaults } from './policy-defaults';
|
||||
@@ -10,6 +10,9 @@ import { sanitizeOrganisationPolicy } from './organisation-policy';
|
||||
import { sanitizeRestrictions, sanitizeScopes } from './policy-scopes';
|
||||
import { versionOf } from '@/lib/shared-store/keys';
|
||||
import { serializeLocal } from '@/lib/shared-store/local';
|
||||
import { isSharedMode, type StoredDoc } from '@/lib/shared-store';
|
||||
import { announceChange, onSharedChange } from '@/lib/shared-store/changes';
|
||||
import { adminConfigDocs } from './admin-store';
|
||||
|
||||
function parseEnvValue(value: string, type: string): unknown {
|
||||
switch (type) {
|
||||
@@ -69,6 +72,15 @@ function parseDoc(raw: string): ParsedDoc {
|
||||
return { data: JSON.parse(raw), version: versionOf(Buffer.from(raw, 'utf-8')) };
|
||||
}
|
||||
|
||||
/** A document from the shared store; throws when it doesn't parse. */
|
||||
function parseStored(doc: StoredDoc | null): ParsedDoc | null {
|
||||
return doc ? { data: JSON.parse(doc.data.toString('utf-8')), version: doc.version } : null;
|
||||
}
|
||||
|
||||
/** Which stored document a change area reloads. */
|
||||
const AREA_FILES = { config: 'config.json', policy: 'policy.json' } as const;
|
||||
type ConfigArea = keyof typeof AREA_FILES;
|
||||
|
||||
/** Changes whenever the file is replaced or rewritten; null when it is missing. */
|
||||
async function fileStamp(filename: string): Promise<string | null> {
|
||||
try {
|
||||
@@ -88,9 +100,24 @@ class ConfigManager {
|
||||
private fileStamps: Partial<Record<WatchedFile, string | null>> = {};
|
||||
private lastChangeCheck = 0;
|
||||
private changeCheck: Promise<void> | null = null;
|
||||
/** Shared mode: the first load, shared by every request that waits for it. */
|
||||
private sharedLoad: Promise<void> | null = null;
|
||||
/** Shared mode: counts writes applied from their result, so a reload that raced one reads again. */
|
||||
private writes = 0;
|
||||
private unlisten: (() => void) | null = null;
|
||||
|
||||
/**
|
||||
* Shared mode (#1066): config and policy live in the shared store, and
|
||||
* every replica reloads them when another one announces a change.
|
||||
* Read-only config keeps reading the mounted files, as in local mode.
|
||||
*/
|
||||
private usesStore(): boolean {
|
||||
return isSharedMode() && !isConfigReadOnly();
|
||||
}
|
||||
|
||||
/** Load admin config and policy from disk. Called once at startup and on reload. */
|
||||
async load(): Promise<void> {
|
||||
if (this.usesStore()) return this.loadShared();
|
||||
// Stamped before reading, so a write that lands mid-read is seen next time.
|
||||
const stamps = await this.readStamps();
|
||||
this.apply(await this.readJsonFile('config.json'), await this.readJsonFile('policy.json'));
|
||||
@@ -101,12 +128,91 @@ class ConfigManager {
|
||||
|
||||
private apply(config: ParsedDoc | null, policy: ParsedDoc | null): void {
|
||||
this.adminConfig = config?.data || {};
|
||||
this.applyPolicyDoc(policy);
|
||||
this.loaded = true;
|
||||
this.applyLoggerSettings();
|
||||
}
|
||||
|
||||
private applyPolicyDoc(policy: ParsedDoc | null): void {
|
||||
this.policyCache = policy?.data
|
||||
? ConfigManager.completePolicy(policy.data as Partial<SettingsPolicy>)
|
||||
: { ...DEFAULT_POLICY };
|
||||
this.policyVersion = policy?.version ?? null;
|
||||
this.loaded = true;
|
||||
}
|
||||
|
||||
private async loadShared(): Promise<void> {
|
||||
this.listenForChanges();
|
||||
const docs = adminConfigDocs();
|
||||
const [config, policy] = await Promise.all([docs.get('config.json'), docs.get('policy.json')]);
|
||||
this.apply(this.parseOrWarn('config.json', config), this.parseOrWarn('policy.json', policy));
|
||||
logger.debug('ConfigManager loaded from the shared store', { configKeys: Object.keys(this.adminConfig).length });
|
||||
}
|
||||
|
||||
/** As a broken file on disk: logged, and the defaults apply. */
|
||||
private parseOrWarn(filename: string, doc: StoredDoc | null): ParsedDoc | null {
|
||||
try {
|
||||
return parseStored(doc);
|
||||
} catch (error) {
|
||||
logger.warn(`Failed to read ${filename} from the shared store`, { error: error instanceof Error ? error.message : 'Unknown error' });
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Reload config or policy when another replica changed it. Registered again on every load. */
|
||||
private listenForChanges(): void {
|
||||
this.unlisten?.();
|
||||
const off = (Object.keys(AREA_FILES) as ConfigArea[]).map((area) =>
|
||||
// A change of this process already replaced the copy in memory.
|
||||
onSharedChange(area, (origin) => (origin === 'remote' ? this.reloadShared(area) : undefined)),
|
||||
);
|
||||
this.unlisten = () => off.forEach((fn) => fn());
|
||||
}
|
||||
|
||||
/**
|
||||
* Read one document again. Throws when the store can't be read or the
|
||||
* document doesn't parse, keeping the copy in memory; the change poller
|
||||
* then tries again.
|
||||
*/
|
||||
private async reloadShared(area: ConfigArea): Promise<void> {
|
||||
for (let attempt = 0; attempt < 3; attempt++) {
|
||||
const writes = this.writes;
|
||||
const doc = parseStored(await adminConfigDocs().get(AREA_FILES[area]));
|
||||
// A write of this process landed while reading: its result may be newer.
|
||||
if (writes !== this.writes) continue;
|
||||
if (area === 'config') {
|
||||
this.adminConfig = doc?.data || {};
|
||||
this.applyLoggerSettings();
|
||||
} else {
|
||||
this.applyPolicyDoc(doc);
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared mode: change the stored config.json with `change` (which edits
|
||||
* the object it gets), so only the keys a request names change and a
|
||||
* concurrent save of other keys on another replica stays.
|
||||
*/
|
||||
private async updateSharedConfig(change: (config: Record<string, unknown>) => void): Promise<void> {
|
||||
const inMemory = this.adminConfig;
|
||||
const result = await adminConfigDocs().update('config.json', (current) => {
|
||||
let base: Record<string, unknown> = {};
|
||||
if (current) {
|
||||
try {
|
||||
base = JSON.parse(current.toString('utf-8'));
|
||||
} catch {
|
||||
// Broken in the store: start from this replica's copy, as a broken file would.
|
||||
base = { ...inMemory };
|
||||
}
|
||||
}
|
||||
change(base);
|
||||
return JSON.stringify(base, null, 2);
|
||||
});
|
||||
this.writes++;
|
||||
if (result.data) this.adminConfig = JSON.parse(result.data.toString('utf-8'));
|
||||
this.applyLoggerSettings();
|
||||
await announceChange('config');
|
||||
}
|
||||
|
||||
private async readStamps(): Promise<Partial<Record<WatchedFile, string | null>>> {
|
||||
@@ -141,7 +247,8 @@ class ConfigManager {
|
||||
|
||||
/** At most every CHANGE_CHECK_INTERVAL_MS: one stat per watched file. */
|
||||
private checkForChanges(): Promise<void> {
|
||||
if (!this.loaded) return Promise.resolve();
|
||||
// Shared mode hears of changes from the store instead (listenForChanges).
|
||||
if (!this.loaded || this.usesStore()) return Promise.resolve();
|
||||
if (this.changeCheck) return this.changeCheck;
|
||||
const now = Date.now();
|
||||
if (now - this.lastChangeCheck < CHANGE_CHECK_INTERVAL_MS) return Promise.resolve();
|
||||
@@ -163,10 +270,16 @@ class ConfigManager {
|
||||
/**
|
||||
* Ensure config is loaded. Once loaded, waits for the periodic check for
|
||||
* changes on disk when one is due (a stat per file, every few seconds).
|
||||
* In shared mode a request never waits on the store once the config is
|
||||
* loaded: changes arrive through the change notices.
|
||||
*/
|
||||
async ensureLoaded(): Promise<void> {
|
||||
if (!this.loaded) await this.load();
|
||||
else await this.checkForChanges();
|
||||
if (this.loaded) return this.checkForChanges();
|
||||
if (!this.usesStore()) return this.load();
|
||||
this.sharedLoad ??= this.load().finally(() => {
|
||||
this.sharedLoad = null;
|
||||
});
|
||||
await this.sharedLoad;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -258,6 +371,7 @@ class ConfigManager {
|
||||
*/
|
||||
async setAdminConfig(updates: Record<string, unknown>): Promise<void> {
|
||||
assertWritable('update admin config');
|
||||
if (this.usesStore()) return this.updateSharedConfig((config) => Object.assign(config, updates));
|
||||
// Start from what is on disk, so a save here doesn't undo a change another
|
||||
// replica saved a moment ago. Same before every write below.
|
||||
await this.reloadIfChanged();
|
||||
@@ -271,6 +385,11 @@ class ConfigManager {
|
||||
*/
|
||||
async removeAdminOverride(key: string): Promise<void> {
|
||||
assertWritable('remove admin override');
|
||||
if (this.usesStore()) {
|
||||
return this.updateSharedConfig((config) => {
|
||||
delete config[key];
|
||||
});
|
||||
}
|
||||
await this.reloadIfChanged();
|
||||
delete this.adminConfig[key];
|
||||
await this.writeJsonFile('config.json', this.adminConfig);
|
||||
@@ -292,6 +411,11 @@ class ConfigManager {
|
||||
*/
|
||||
async markSetupComplete(): Promise<void> {
|
||||
assertWritable('mark setup complete');
|
||||
if (this.usesStore()) {
|
||||
return this.updateSharedConfig((config) => {
|
||||
config.setupComplete = true;
|
||||
});
|
||||
}
|
||||
await this.reloadIfChanged();
|
||||
this.adminConfig.setupComplete = true;
|
||||
await this.writeJsonFile('config.json', this.adminConfig);
|
||||
@@ -322,6 +446,7 @@ class ConfigManager {
|
||||
*/
|
||||
async setPolicy(policy: SettingsPolicy, options: { ifVersion?: string } = {}): Promise<string> {
|
||||
assertWritable('update settings policy');
|
||||
if (this.usesStore()) return this.setSharedPolicy(policy, options.ifVersion);
|
||||
return serializeLocal(POLICY_WRITE_QUEUE, async () => {
|
||||
if (options.ifVersion !== undefined) {
|
||||
// Compare with what is on disk now, not with what this process saw last.
|
||||
@@ -341,6 +466,7 @@ class ConfigManager {
|
||||
*/
|
||||
async updatePolicy(change: (current: SettingsPolicy) => SettingsPolicy | undefined): Promise<void> {
|
||||
assertWritable('update settings policy');
|
||||
if (this.usesStore()) return this.updateSharedPolicy(change);
|
||||
await serializeLocal(POLICY_WRITE_QUEUE, async () => {
|
||||
const next = change(this.policyCache);
|
||||
if (next === undefined) return;
|
||||
@@ -349,6 +475,40 @@ class ConfigManager {
|
||||
});
|
||||
}
|
||||
|
||||
private async setSharedPolicy(policy: SettingsPolicy, ifVersion: string | undefined): Promise<string> {
|
||||
const completed = ConfigManager.completePolicy(policy);
|
||||
const expected = ifVersion === DEFAULT_POLICY_VERSION ? null : ifVersion;
|
||||
const result = await adminConfigDocs().put(
|
||||
'policy.json',
|
||||
JSON.stringify(completed, null, 2),
|
||||
expected === undefined ? {} : { ifVersion: expected },
|
||||
);
|
||||
this.writes++;
|
||||
if (!result.ok) {
|
||||
// Show what is stored now, so a retry or a reload compares with it.
|
||||
this.applyPolicyDoc(this.parseOrWarn('policy.json', result.current));
|
||||
throw new PolicyConflictError();
|
||||
}
|
||||
this.policyCache = completed;
|
||||
this.policyVersion = result.version;
|
||||
await announceChange('policy');
|
||||
return result.version;
|
||||
}
|
||||
|
||||
private async updateSharedPolicy(change: (current: SettingsPolicy) => SettingsPolicy | undefined): Promise<void> {
|
||||
const result = await adminConfigDocs().update('policy.json', (current) => {
|
||||
const stored = current
|
||||
? ConfigManager.completePolicy(JSON.parse(current.toString('utf-8')) as Partial<SettingsPolicy>)
|
||||
: ConfigManager.completePolicy({});
|
||||
const next = change(stored);
|
||||
return next === undefined ? undefined : JSON.stringify(ConfigManager.completePolicy(next), null, 2);
|
||||
});
|
||||
if (result.data === undefined) return;
|
||||
this.writes++;
|
||||
this.applyPolicyDoc(result.data && result.version ? parseStored({ data: result.data, version: result.version }) : null);
|
||||
await announceChange('policy');
|
||||
}
|
||||
|
||||
/**
|
||||
* A policy as it would be stored: defaults filled in, migrated and
|
||||
* sanitised, exactly as policy.json is on load. For previewing an import.
|
||||
|
||||
Reference in New Issue
Block a user