From 68d64f5018a0daded9f2f1f6ee4ed799e2730e87 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sat, 31 Oct 2026 23:55:17 +0100 Subject: [PATCH] fix: pair the policy with its own version and keep config reloads and saves from undoing each other #1066 Fixes from the review of the admin part: - GET /api/admin/policy reads the policy and its version together; an identity check waiting on the store let another replica's change pair the old policy with the new ETag, and a save with it then won. - A policy save answers with the stored policy and its ETag, and the policy, organisation, plugins and themes pages continue from that instead of reading it back from an instance that may not have it yet. - A stored policy that doesn't parse keeps its version, so a save can replace it instead of answering 409 forever. - A reload of another replica's change throws when this replica's writes kept landing while it read, so the change poller tries again; a write whose reload raced it reads the document again instead of applying its older result. - Dropping a deleted plugin from the policy rereads policy.json first in local mode, as every other write does. - New updateAdminConfig(key, change) builds a value on the stored one. The policy route's test mocks gain getPolicyWithVersion. --- app/(main)/admin/_pages/plugins.tsx | 9 +- app/(main)/admin/_pages/policy.tsx | 9 +- app/(main)/admin/_pages/themes.tsx | 9 +- .../__tests__/policy-bearer-route.test.ts | 1 + .../admin/__tests__/policy-etag-race.test.ts | 84 +++++++++ .../admin/__tests__/policy-if-match.test.ts | 16 ++ .../__tests__/policy-scopes-route.test.ts | 1 + app/api/admin/policy/route.ts | 13 +- components/admin/org/use-org-policy-draft.ts | 23 ++- lib/__tests__/instance-details-gates.test.ts | 1 + .../__tests__/config-manager-shared.test.ts | 97 ++++++++++ lib/admin/config-manager.ts | 172 ++++++++++++++---- 12 files changed, 384 insertions(+), 51 deletions(-) create mode 100644 app/api/admin/__tests__/policy-etag-race.test.ts 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'); }