diff --git a/app/(main)/admin/_pages/plugins.tsx b/app/(main)/admin/_pages/plugins.tsx
index eb09a808e..4dda13db7 100644
--- a/app/(main)/admin/_pages/plugins.tsx
+++ b/app/(main)/admin/_pages/plugins.tsx
@@ -85,7 +85,14 @@ export function PluginsPage() {
});
if (res.ok) {
policyETag.current = res.headers.get('ETag');
- setSavedPolicy(policy);
+ // The server sanitises; continue from what it stored.
+ const stored = ((await res.json().catch(() => null)) as { policy?: SettingsPolicy } | null)?.policy;
+ if (stored) {
+ setPolicy(stored);
+ setSavedPolicy(stored);
+ } else {
+ setSavedPolicy(policy);
+ }
return null;
}
const data = await res.json().catch(() => ({}));
diff --git a/app/(main)/admin/_pages/policy.tsx b/app/(main)/admin/_pages/policy.tsx
index 3922976c4..4e765d468 100644
--- a/app/(main)/admin/_pages/policy.tsx
+++ b/app/(main)/admin/_pages/policy.tsx
@@ -229,7 +229,14 @@ export function PolicyPage() {
if (res.ok) {
policyETag.current = res.headers.get('ETag');
- setSavedPolicy(policy);
+ // The server sanitises; continue from what it stored.
+ const stored = ((await res.json().catch(() => null)) as { policy?: SettingsPolicy } | null)?.policy;
+ if (stored) {
+ setPolicy(stored);
+ setSavedPolicy(stored);
+ } else {
+ setSavedPolicy(policy);
+ }
setMessage({ type: 'success', text: 'Policy saved. Users see the change on their next login.' });
setDirty(false);
} else {
diff --git a/app/(main)/admin/_pages/themes.tsx b/app/(main)/admin/_pages/themes.tsx
index 8d472a43b..6bffbb308 100644
--- a/app/(main)/admin/_pages/themes.tsx
+++ b/app/(main)/admin/_pages/themes.tsx
@@ -160,7 +160,14 @@ export function ThemesPage() {
});
if (res.ok) {
policyETag.current = res.headers.get('ETag');
- setSavedPolicy(policy);
+ // The server sanitises; continue from what it stored.
+ const stored = ((await res.json().catch(() => null)) as { policy?: SettingsPolicy } | null)?.policy;
+ if (stored) {
+ setPolicy(withThemeDefaults(stored));
+ setSavedPolicy(withThemeDefaults(stored));
+ } else {
+ setSavedPolicy(policy);
+ }
return null;
}
const data = await res.json().catch(() => ({}));
diff --git a/app/api/admin/__tests__/policy-bearer-route.test.ts b/app/api/admin/__tests__/policy-bearer-route.test.ts
index 4bf99a360..fafe3ec9d 100644
--- a/app/api/admin/__tests__/policy-bearer-route.test.ts
+++ b/app/api/admin/__tests__/policy-bearer-route.test.ts
@@ -28,6 +28,7 @@ vi.mock('@/lib/admin/config-manager', () => ({
organisation: { disclaimer: { html: '
Acme GmbH
', text: 'Acme GmbH' } },
scopes: SCOPES,
}),
+ getPolicyWithVersion() { return { policy: this.getPolicy(), version: 'v1' }; },
},
}));
const logs = vi.hoisted(() => ({ calls: [] as unknown[][] }));
diff --git a/app/api/admin/__tests__/policy-etag-race.test.ts b/app/api/admin/__tests__/policy-etag-race.test.ts
new file mode 100644
index 000000000..23da59c54
--- /dev/null
+++ b/app/api/admin/__tests__/policy-etag-race.test.ts
@@ -0,0 +1,84 @@
+// @vitest-environment node
+// GET /api/admin/policy waits on the store for the identity checks. A
+// change from another replica applied meanwhile must not pair the policy
+// read before with the version read after: a page saving that body with
+// that ETag would put the old policy back over the other replica's save.
+import { afterEach, beforeEach, expect, it, vi } from 'vitest';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import path from 'node:path';
+import { NextRequest } from 'next/server';
+
+const hooks: { duringIdentity?: () => Promise } = {};
+vi.mock('@/lib/auth/instance-details', () => ({
+ canSeeInstanceDetails: async () => true,
+ signedInIdentity: async () => {
+ // Stands in for the store round trip verifyCurrentAdminSession makes.
+ await hooks.duringIdentity?.();
+ return { admin: true };
+ },
+}));
+vi.mock('@/lib/admin/session', () => ({
+ requireAdminAuth: async () => ({ payload: { role: 'admin' } }),
+ getClientIP: () => '127.0.0.1',
+}));
+vi.mock('@/lib/admin/audit', () => ({ auditLog: async () => {} }));
+
+import { configManager } from '@/lib/admin/config-manager';
+import { adminConfigDocs } from '@/lib/admin/admin-store';
+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 { separateFakeStore } from '@/lib/admin/__tests__/shared-store-fake';
+
+let dir: string;
+let store: ReturnType;
+
+beforeEach(async () => {
+ dir = await mkdtemp(path.join(tmpdir(), 'bw-review-etag-'));
+ 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('policy.json', JSON.stringify({ features: { calendarEnabled: true } }));
+ await markEpochsSeen();
+ await configManager.load();
+ setBootState('ready');
+});
+
+afterEach(async () => {
+ hooks.duringIdentity = undefined;
+ setStoreForTests(null);
+ resetChangesForTests();
+ resetLocalStoreForTests();
+ vi.unstubAllEnvs();
+ await rm(dir, { recursive: true, force: true });
+});
+
+it('sends the version of the policy it sends, when a change lands during the request', async () => {
+ hooks.duringIdentity = async () => {
+ // Another replica saves calendarEnabled:false; the poll lands mid-GET.
+ await adminConfigDocs().put('policy.json', JSON.stringify({ features: { calendarEnabled: false, filesEnabled: false } }));
+ await store.epochs.bump('policy');
+ await pollChanges();
+ };
+ const res = await GET(new NextRequest('https://mail.example/api/admin/policy'));
+ hooks.duringIdentity = undefined;
+ const body = await res.json();
+ const etag = res.headers.get('ETag')!;
+ // Old body...
+ expect(body.features.calendarEnabled).toBe(true);
+ // ...saved with the ETag it came with: refused, the other save stays.
+ const put = await PUT(new NextRequest('https://mail.example/api/admin/policy', {
+ method: 'PUT',
+ headers: { 'content-type': 'application/json', 'if-match': etag },
+ body: JSON.stringify(body),
+ }));
+ expect(put.status).toBe(409);
+ const stored = JSON.parse((await adminConfigDocs().get('policy.json'))!.data.toString('utf-8'));
+ expect(stored.features).toEqual({ calendarEnabled: false, filesEnabled: false });
+});
diff --git a/app/api/admin/__tests__/policy-if-match.test.ts b/app/api/admin/__tests__/policy-if-match.test.ts
index f98ce7b4f..2671b86cb 100644
--- a/app/api/admin/__tests__/policy-if-match.test.ts
+++ b/app/api/admin/__tests__/policy-if-match.test.ts
@@ -94,6 +94,22 @@ describe('policy save with If-Match', () => {
expect((await stored()).forceEnabledPlugins).toEqual([]);
});
+ it('drops a deleted plugin without undoing a save another replica made on the same volume', async () => {
+ await load();
+ await writeFile(path.join(dir, 'policy.json'), JSON.stringify({ features: { calendarEnabled: false }, forceEnabledPlugins: ['demo', 'other'] }));
+ await forgetExtensionInPolicy('plugin', 'demo');
+ expect((await stored()).forceEnabledPlugins).toEqual(['other']);
+ expect((await stored()).features.calendarEnabled).toBe(false);
+ });
+
+ it('answers a save with what it stored', async () => {
+ const page = await load();
+ const res = await save({ ...page.policy, features: { ...(page.policy.features as object), calendarEnabled: false } }, page.etag);
+ const body = await res.json();
+ expect(body.policy.features.calendarEnabled).toBe(false);
+ expect(res.headers.get('ETag')).toBe((await load()).etag);
+ });
+
it('replaces the policy without If-Match, as before', async () => {
await load();
await configManager.setPolicy({ ...configManager.getPolicy(), forceEnabledPlugins: [] });
diff --git a/app/api/admin/__tests__/policy-scopes-route.test.ts b/app/api/admin/__tests__/policy-scopes-route.test.ts
index 70c6f3b70..40ed2582d 100644
--- a/app/api/admin/__tests__/policy-scopes-route.test.ts
+++ b/app/api/admin/__tests__/policy-scopes-route.test.ts
@@ -29,6 +29,7 @@ vi.mock('@/lib/admin/config-manager', () => ({
scopes: SCOPES,
}),
getPolicyVersion: () => 'v1',
+ getPolicyWithVersion() { return { policy: this.getPolicy(), version: 'v1' }; },
},
}));
vi.mock('@/lib/logger', () => ({ logger: { error: vi.fn(), warn: vi.fn(), info: vi.fn(), debug: vi.fn() } }));
diff --git a/app/api/admin/policy/route.ts b/app/api/admin/policy/route.ts
index a799231df..06f160cb2 100644
--- a/app/api/admin/policy/route.ts
+++ b/app/api/admin/policy/route.ts
@@ -1,5 +1,5 @@
import { NextRequest, NextResponse } from 'next/server';
-import { configManager, isPolicyConflict } from '@/lib/admin/config-manager';
+import { completePolicy, configManager, isPolicyConflict } from '@/lib/admin/config-manager';
import { requireAdminAuth, getClientIP } from '@/lib/admin/session';
import { auditLog } from '@/lib/admin/audit';
import { logger } from '@/lib/logger';
@@ -37,7 +37,9 @@ function ifMatchVersion(request: NextRequest): string | undefined {
export async function GET(request: NextRequest) {
try {
await configManager.ensureLoaded();
- const policy = configManager.getPolicy();
+ // Together: the identity checks below wait on the store, and a change
+ // applied meanwhile must not pair this policy with a newer version.
+ const { policy, version } = configManager.getPolicyWithVersion();
if (!(await canSeeInstanceDetails(request))) {
const tokenIdentity = await bearerIdentity(request);
if (tokenIdentity) return userPolicyResponse(policy, tokenIdentity.addresses);
@@ -70,7 +72,7 @@ export async function GET(request: NextRequest) {
// The whole policy, as the admin pages edit it: with its version, which
// they send back as If-Match when they save.
return NextResponse.json({ ...policy, scopes: policy.scopes ?? [] }, {
- headers: { 'Cache-Control': 'no-store', ETag: policyETag(configManager.getPolicyVersion()) },
+ headers: { 'Cache-Control': 'no-store', ETag: policyETag(version) },
});
} catch (error) {
logger.error('Policy read error', { error: error instanceof Error ? error.message : 'Unknown error' });
@@ -146,7 +148,10 @@ export async function PUT(request: NextRequest) {
defaultSidebarAppCount: configManager.getPolicy().defaultSidebarApps?.length ?? 0,
}, ip);
- return NextResponse.json({ ok: true }, { headers: { ETag: policyETag(version) } });
+ // What was stored, with its version: the page continues from this
+ // instead of reading it back from a replica that may not have it yet.
+ const saved = completePolicy(policy);
+ return NextResponse.json({ ok: true, policy: { ...saved, scopes: saved.scopes ?? [] } }, { headers: { ETag: policyETag(version) } });
} catch (error) {
const readOnly = configReadOnlyResponse(error);
if (readOnly) return readOnly;
diff --git a/components/admin/org/use-org-policy-draft.ts b/components/admin/org/use-org-policy-draft.ts
index 283f0d831..ac2470427 100644
--- a/components/admin/org/use-org-policy-draft.ts
+++ b/components/admin/org/use-org-policy-draft.ts
@@ -58,20 +58,24 @@ export function useOrgPolicyDraft(dirtyId: string) {
const scopeId = useOrgScopeStore((s) => s.scopeId);
const setScopeId = useOrgScopeStore((s) => s.setScopeId);
+ /** Start editing from `data`, the policy stored at version `etag`. */
+ const adopt = useCallback((data: SettingsPolicy, etag: string | null) => {
+ policyETag.current = etag;
+ const normalised = { ...DEFAULT_POLICY, ...data, organisation: sanitizeOrganisationPolicy(data.organisation), scopes: data.scopes ?? [] };
+ setPolicy(normalised);
+ setSaved(clone(normalised));
+ }, []);
+
const load = useCallback(async () => {
try {
const res = await apiFetch('/api/admin/policy');
if (!res.ok) throw new Error();
- const data = (await res.json()) as SettingsPolicy;
- policyETag.current = res.headers.get('ETag');
- const normalised = { ...DEFAULT_POLICY, ...data, organisation: sanitizeOrganisationPolicy(data.organisation), scopes: data.scopes ?? [] };
- setPolicy(normalised);
- setSaved(clone(normalised));
+ adopt((await res.json()) as SettingsPolicy, res.headers.get('ETag'));
} catch {
setMessage({ type: 'error', text: 'Could not load the policy' });
setPolicy((p) => p ?? { ...DEFAULT_POLICY, organisation: clone(DEFAULT_ORGANISATION_POLICY), scopes: [] });
}
- }, [setMessage]);
+ }, [adopt, setMessage]);
useEffect(() => { void load(); }, [load]);
@@ -194,8 +198,11 @@ export function useOrgPolicyDraft(dirtyId: string) {
body: JSON.stringify(policy),
});
if (res.ok) {
- // The server sanitises; show what it kept.
- await load();
+ // The server sanitises; show what it kept. From the answer, not a new
+ // read, which another instance might serve before it has the change.
+ const stored = ((await res.json().catch(() => null)) as { policy?: SettingsPolicy } | null)?.policy;
+ if (stored) adopt(stored, res.headers.get('ETag'));
+ else await load();
setMessage({ type: 'success', text: 'Saved. Users get the change the next time Bulwark loads.' });
} else {
const data = await res.json().catch(() => ({}));
diff --git a/lib/__tests__/instance-details-gates.test.ts b/lib/__tests__/instance-details-gates.test.ts
index 2da5a6969..8be1eb7da 100644
--- a/lib/__tests__/instance-details-gates.test.ts
+++ b/lib/__tests__/instance-details-gates.test.ts
@@ -33,6 +33,7 @@ vi.mock('@/lib/admin/config-manager', () => ({
get: (key: string, fallback: unknown) => (key in config.values ? config.values[key] : fallback),
getPolicy: () => POLICY,
getPolicyVersion: () => 'v1',
+ getPolicyWithVersion() { return { policy: this.getPolicy(), version: 'v1' }; },
},
}));
vi.mock('@/lib/logger', () => ({ logger: { error: vi.fn(), warn: vi.fn(), info: vi.fn(), debug: vi.fn() } }));
diff --git a/lib/admin/__tests__/config-manager-shared.test.ts b/lib/admin/__tests__/config-manager-shared.test.ts
index 4d66303b9..3021b93b1 100644
--- a/lib/admin/__tests__/config-manager-shared.test.ts
+++ b/lib/admin/__tests__/config-manager-shared.test.ts
@@ -39,6 +39,8 @@ import { separateFakeStore } from './shared-store-fake';
let dir: string;
let store: ReturnType;
+/** Runs inside the store's document calls, to land another replica's change mid-request. */
+const hooks: { onGet?: (id: string) => Promise; afterUpdate?: (id: string) => Promise } = {};
const stored = async (id: string) => {
const doc = await adminConfigDocs().get(id);
@@ -71,6 +73,24 @@ beforeEach(async () => {
resetLocalStoreForTests();
resetChangesForTests();
store = separateFakeStore(path.join(dir, 'store'));
+ const collection = store.collection.bind(store);
+ store.collection = (name, options) => {
+ const docs = collection(name, options);
+ return {
+ ...docs,
+ get: async (id) => {
+ await hooks.onGet?.(id);
+ return docs.get(id);
+ },
+ update: async (id, fn, updateOptions) => {
+ const result = await docs.update(id, fn, updateOptions);
+ const hook = hooks.afterUpdate;
+ hooks.afterUpdate = undefined;
+ await hook?.(id);
+ return result;
+ },
+ };
+ };
setStoreForTests(store);
await store.epochs.bump(INIT_EPOCH);
await adminConfigDocs().put('config.json', JSON.stringify({ appName: 'First', setupComplete: true }));
@@ -81,6 +101,8 @@ beforeEach(async () => {
});
afterEach(async () => {
+ hooks.onGet = undefined;
+ hooks.afterUpdate = undefined;
setStoreForTests(null);
resetChangesForTests();
resetLocalStoreForTests();
@@ -160,7 +182,82 @@ describe('config manager with a shared store', () => {
});
});
+describe('change notices racing writes', () => {
+ it('fails a reload when writes of this replica kept landing while it read, so the poller tries again', async () => {
+ let reads = 0;
+ let n = 0;
+ hooks.onGet = async (id) => {
+ if (id !== 'config.json') return;
+ reads++;
+ await configManager.setAdminConfig({ loginShowVersion: ++n % 2 === 0 });
+ };
+ // The listener the change poller runs for a remote change.
+ const reload = (configManager as unknown as { reloadShared(area: string): Promise }).reloadShared('config');
+ await expect(reload).rejects.toThrow(/kept changing/);
+ expect(reads).toBe(3);
+ });
+
+ it('keeps another replica\'s change that was reloaded while a save of this one was in flight', async () => {
+ hooks.afterUpdate = async () => {
+ const current = await stored('config.json');
+ await otherReplicaSaves('config.json', { ...current, appName: 'Saved elsewhere' });
+ await pollChanges();
+ };
+ await configManager.setAdminConfig({ loginShowVersion: false });
+ expect(configManager.get('appName')).toBe('Saved elsewhere');
+ expect(configManager.get('loginShowVersion')).toBe(false);
+ });
+});
+
+describe('updateAdminConfig with a shared store', () => {
+ it('builds each of two saves on two replicas on the other\'s result', async () => {
+ const other = await otherReplicaManager();
+ const add = (host: string) => (list: Array<{ host: string }>) => [...list, { host }];
+ await Promise.all([
+ configManager.updateAdminConfig('domainBranding', add('a.example'), []),
+ other.updateAdminConfig('domainBranding', add('b.example'), []),
+ ]);
+ const hosts = (await stored('config.json')).domainBranding.map((e: { host: string }) => e.host).sort();
+ expect(hosts).toEqual(['a.example', 'b.example']);
+ });
+
+ it('starts from the value get() would give and leaves the config alone on undefined', async () => {
+ vi.stubEnv('APP_NAME', 'From env');
+ expect(await configManager.updateAdminConfig('appName', (current) => `${current}!`)).toBe('First!');
+ await configManager.removeAdminOverride('appName');
+ expect(await configManager.updateAdminConfig('appName', (current) => `${current}!`)).toBe('From env!');
+ const before = await adminConfigDocs().get('config.json');
+ expect(await configManager.updateAdminConfig('appName', () => undefined)).toBeUndefined();
+ expect((await adminConfigDocs().get('config.json'))!.version).toBe(before!.version);
+ });
+});
+
describe('policy saves with a shared store', () => {
+ it('lets a save replace a stored policy that doesn\'t parse', async () => {
+ await adminConfigDocs().put('policy.json', '{ not json');
+ await store.epochs.bump('policy');
+ await pollChanges();
+ const page = await load();
+ expect(page.etag).toBe(`"${(await adminConfigDocs().get('policy.json'))!.version}"`);
+ expect((await save(page.policy, page.etag)).status).toBe(200);
+ expect(await stored('policy.json')).toMatchObject({ features: page.policy.features });
+
+ // The same after a restart.
+ await adminConfigDocs().put('policy.json', '{ not json');
+ await configManager.reload();
+ const again = await load();
+ expect((await save(again.policy, again.etag)).status).toBe(200);
+ });
+
+ it('answers a save with the stored policy and its version', async () => {
+ const page = await load();
+ const res = await save({ ...page.policy, features: { ...page.policy.features, calendarEnabled: false } }, page.etag);
+ const body = await res.json();
+ expect(body.policy.features.calendarEnabled).toBe(false);
+ expect(body.policy.scopes).toEqual([]);
+ expect(res.headers.get('ETag')).toBe(`"${(await adminConfigDocs().get('policy.json'))!.version}"`);
+ });
+
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')! };
diff --git a/lib/admin/config-manager.ts b/lib/admin/config-manager.ts
index 5ffe686eb..1bb5d3f35 100644
--- a/lib/admin/config-manager.ts
+++ b/lib/admin/config-manager.ts
@@ -70,9 +70,12 @@ export function isPolicyConflict(error: unknown): error is PolicyConflictError {
/** Policy writes in this process, one after another, so a version check and its write can't interleave. */
const POLICY_WRITE_QUEUE = 'admin-config\0policy.json';
+/** Config read-modify-writes (updateAdminConfig) in this process, one after another. */
+const CONFIG_WRITE_QUEUE = 'admin-config\0config.json';
interface ParsedDoc {
- data: Record;
+ /** null when the stored document doesn't parse (shared mode). */
+ data: Record | null;
/** As the shared store computes it: changes whenever the bytes do. */
version: string;
}
@@ -81,9 +84,19 @@ 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;
+/**
+ * A document from the shared store. Data that doesn't parse comes back as
+ * null with the document's version, so an If-Match save can still replace
+ * it.
+ */
+function parseStored(doc: StoredDoc | null, onError?: (error: unknown) => void): ParsedDoc | null {
+ if (!doc) return null;
+ try {
+ return { data: JSON.parse(doc.data.toString('utf-8')), version: doc.version };
+ } catch (error) {
+ onError?.(error);
+ return { data: null, version: doc.version };
+ }
}
/** Which stored document a change area reloads. */
@@ -113,6 +126,8 @@ class ConfigManager {
private sharedLoad: Promise | null = null;
/** Shared mode: counts writes applied from their result, so a reload that raced one reads again. */
private writes = 0;
+ /** Shared mode: counts reloads applied, so a write that raced one reads its result again. */
+ private reloads = 0;
private unlisten: (() => void) | null = null;
/**
@@ -159,12 +174,9 @@ class ConfigManager {
/** 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) {
+ return parseStored(doc, (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. */
@@ -178,24 +190,53 @@ class ConfigManager {
}
/**
- * 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.
+ * Read one document again. Throws when the store can't be read, or when
+ * writes of this process kept landing while reading; the change poller
+ * then tries again. A document that doesn't parse keeps the copy in
+ * memory (the policy takes its version, so a save can replace it).
*/
private async reloadShared(area: ConfigArea): Promise {
for (let attempt = 0; attempt < 3; attempt++) {
const writes = this.writes;
- const doc = parseStored(await adminConfigDocs().get(AREA_FILES[area]));
+ const doc = this.parseOrWarn(AREA_FILES[area], 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);
- }
+ this.applyReloaded(area, doc);
return;
}
+ throw new Error(`${AREA_FILES[area]} kept changing while it was reloaded`);
+ }
+
+ private applyReloaded(area: ConfigArea, doc: ParsedDoc | null): void {
+ this.reloads++;
+ if (doc && doc.data === null) {
+ if (area === 'policy') this.policyVersion = doc.version;
+ return;
+ }
+ if (area === 'config') {
+ this.adminConfig = doc?.data || {};
+ this.applyLoggerSettings();
+ } else {
+ this.applyPolicyDoc(doc);
+ }
+ }
+
+ /**
+ * After a write of this process: apply its result, unless a reload of
+ * another replica's change was applied while it was in flight. That one
+ * may be newer than the write, so the document is read again instead.
+ */
+ private async applyWritten(area: ConfigArea, reloadsBefore: number, apply: () => void): Promise {
+ this.writes++;
+ if (this.reloads === reloadsBefore) {
+ apply();
+ return;
+ }
+ try {
+ this.applyReloaded(area, this.parseOrWarn(AREA_FILES[area], await adminConfigDocs().get(AREA_FILES[area])));
+ } catch {
+ apply();
+ }
}
/**
@@ -203,8 +244,9 @@ class ConfigManager {
* 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) => void): Promise {
+ private async updateSharedConfig(change: (config: Record) => void | false): Promise {
const inMemory = this.adminConfig;
+ const reloads = this.reloads;
const result = await adminConfigDocs().update('config.json', (current) => {
let base: Record = {};
if (current) {
@@ -215,12 +257,14 @@ class ConfigManager {
base = { ...inMemory };
}
}
- change(base);
+ if (change(base) === false) return undefined;
return JSON.stringify(base, null, 2);
});
- this.writes++;
- if (result.data) this.adminConfig = JSON.parse(result.data.toString('utf-8'));
- this.applyLoggerSettings();
+ if (result.data === undefined) return;
+ await this.applyWritten('config', reloads, () => {
+ if (result.data) this.adminConfig = JSON.parse(result.data.toString('utf-8'));
+ this.applyLoggerSettings();
+ });
await announceChange('config');
}
@@ -298,12 +342,17 @@ class ConfigManager {
// Synchronous callers can't wait: a due check runs in the background and
// the next read sees its result.
void this.checkForChanges();
+ return ConfigManager.resolve(this.adminConfig, key, defaultValue);
+ }
+
+ /** A value as get() reads it, from the admin overrides in `adminConfig`. */
+ private static resolve(adminConfig: Record, key: string, defaultValue?: T): T {
// Admin override (highest priority)
const mapping = CONFIG_ENV_MAP[key];
- if (key in this.adminConfig) {
- return this.adminConfig[key] as T;
- } else if (mapping && mapping.fileKey && mapping.fileKey in this.adminConfig) {
- const fileVal = readFileEnv(this.adminConfig[mapping.fileKey]);
+ if (key in adminConfig) {
+ return adminConfig[key] as T;
+ } else if (mapping && mapping.fileKey && mapping.fileKey in adminConfig) {
+ const fileVal = readFileEnv(adminConfig[mapping.fileKey]);
if (fileVal !== null) {
return parseEnvValue(fileVal, mapping.type) as T;
}
@@ -380,7 +429,11 @@ class ConfigManager {
*/
async setAdminConfig(updates: Record): Promise {
assertWritable('update admin config');
- if (this.usesStore()) return this.updateSharedConfig((config) => Object.assign(config, updates));
+ 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();
@@ -389,6 +442,36 @@ class ConfigManager {
this.applyLoggerSettings();
}
+ /**
+ * Set one admin override to `change(current)`, where `current` is the value
+ * as get() would return it from the stored config (override, else env,
+ * else `defaultValue`). For values built on the current one, such as a
+ * list with one entry changed: two saves at once, here or on another
+ * replica, each build on the other's result. Return undefined to leave
+ * it. Returns the value stored.
+ */
+ async updateAdminConfig(key: string, change: (current: T) => T | undefined, defaultValue?: T): Promise {
+ assertWritable('update admin config');
+ let stored: T | undefined;
+ if (this.usesStore()) {
+ await this.updateSharedConfig((config) => {
+ stored = change(ConfigManager.resolve(config, key, defaultValue));
+ if (stored === undefined) return false;
+ config[key] = stored;
+ });
+ return stored;
+ }
+ return serializeLocal(CONFIG_WRITE_QUEUE, async () => {
+ await this.reloadIfChanged();
+ stored = change(ConfigManager.resolve(this.adminConfig, key, defaultValue));
+ if (stored === undefined) return undefined;
+ this.adminConfig[key] = stored;
+ await this.writeJsonFile('config.json', this.adminConfig);
+ this.applyLoggerSettings();
+ return stored;
+ });
+ }
+
/**
* Remove an admin override, reverting to env/default.
*/
@@ -446,6 +529,15 @@ class ConfigManager {
return this.policyVersion ?? DEFAULT_POLICY_VERSION;
}
+ /**
+ * The policy and its version, read together. Use this whenever both go
+ * out (a body with its ETag): an await between two separate reads lets a
+ * change from another replica pair an old policy with the new version.
+ */
+ getPolicyWithVersion(): { policy: SettingsPolicy; version: string } {
+ return { policy: this.getPolicy(), version: this.getPolicyVersion() };
+ }
+
/**
* Update the settings policy. Writes to disk. With `ifVersion` (from
* getPolicyVersion when the caller read the policy) it refuses with a
@@ -477,6 +569,8 @@ class ConfigManager {
assertWritable('update settings policy');
if (this.usesStore()) return this.updateSharedPolicy(change);
await serializeLocal(POLICY_WRITE_QUEUE, async () => {
+ // Change what is on disk now, not what this process saw last.
+ await this.reloadIfChanged();
const next = change(this.policyCache);
if (next === undefined) return;
this.policyCache = ConfigManager.completePolicy(next);
@@ -487,24 +581,29 @@ class ConfigManager {
private async setSharedPolicy(policy: SettingsPolicy, ifVersion: string | undefined): Promise {
const completed = ConfigManager.completePolicy(policy);
const expected = ifVersion === DEFAULT_POLICY_VERSION ? null : ifVersion;
+ const reloads = this.reloads;
const result = await adminConfigDocs().put(
'policy.json',
JSON.stringify(completed, null, 2),
expected === undefined ? {} : { ifVersion: expected },
);
- this.writes++;
if (!result.ok) {
+ this.writes++;
// Show what is stored now, so a retry or a reload compares with it.
- this.applyPolicyDoc(this.parseOrWarn('policy.json', result.current));
+ this.applyReloaded('policy', this.parseOrWarn('policy.json', result.current));
throw new PolicyConflictError();
}
- this.policyCache = completed;
- this.policyVersion = result.version;
+ const version = result.version;
+ await this.applyWritten('policy', reloads, () => {
+ this.policyCache = completed;
+ this.policyVersion = version;
+ });
await announceChange('policy');
- return result.version;
+ return version;
}
private async updateSharedPolicy(change: (current: SettingsPolicy) => SettingsPolicy | undefined): Promise {
+ const reloads = this.reloads;
const result = await adminConfigDocs().update('policy.json', (current) => {
const stored = current
? ConfigManager.completePolicy(JSON.parse(current.toString('utf-8')) as Partial)
@@ -513,8 +612,9 @@ class ConfigManager {
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 this.applyWritten('policy', reloads, () => {
+ this.applyPolicyDoc(result.data && result.version ? parseStored({ data: result.data, version: result.version }) : null);
+ });
await announceChange('policy');
}