diff --git a/app/api/admin/plugins/[id]/bundle/route.ts b/app/api/admin/plugins/[id]/bundle/route.ts index 963d2eecf..7264a74e9 100644 --- a/app/api/admin/plugins/[id]/bundle/route.ts +++ b/app/api/admin/plugins/[id]/bundle/route.ts @@ -1,5 +1,5 @@ import { NextRequest, NextResponse } from 'next/server'; -import { getPluginBundle, getPlugin, getPluginSignature } from '@/lib/admin/plugin-registry'; +import { getPluginBundle, getPlugin, getPluginSignature, isBundleMismatch } from '@/lib/admin/plugin-registry'; import { getDevPlugin, readDevBundle } from '@/lib/admin/plugin-dev'; import { signBytes } from '@/lib/admin/plugin-signing'; import { canSeeInstanceDetails } from '@/lib/auth/instance-details'; @@ -82,6 +82,10 @@ export async function GET( headers['Content-Length'] = String(Buffer.byteLength(code, 'utf-8')); return new NextResponse(code, { headers }); } catch (error) { + // Logged where it was found. 409 on every replica alike. + if (isBundleMismatch(error)) { + return NextResponse.json({ error: (error as Error).message }, { status: 409, headers: { 'Cache-Control': 'no-store' } }); + } const unavailable = storeUnavailableResponse(error); if (unavailable) return unavailable; return NextResponse.json({ error: 'Not found' }, { status: 404 }); diff --git a/lib/admin/__tests__/plugin-registry-shared.test.ts b/lib/admin/__tests__/plugin-registry-shared.test.ts index e81f213ca..f2b0f4c45 100644 --- a/lib/admin/__tests__/plugin-registry-shared.test.ts +++ b/lib/admin/__tests__/plugin-registry-shared.test.ts @@ -225,6 +225,64 @@ describe('plugins in the shared store', () => { }); }); +describe('one change of the registry at a time across replicas', () => { + const LEASE = 'extension-registry-plugins'; + + it('refuses an install with 503 while another replica holds the registry, and writes nothing', async () => { + const a = await otherReplica(); + a.registry.setRegistryLeaseTimingForTests({ waitMs: 200 }); + expect(await store.leases.acquire(LEASE, 'replica-b-install', 60_000)).toBe(true); + + expect((await install(a, 'module.exports = {}')).status).toBe(503); + const docs = (await import('../extension-store')).pluginDocs(); + expect(await docs.list('')).toEqual([]); + + await store.leases.release(LEASE, 'replica-b-install'); + expect((await install(a, 'module.exports = {}')).status).toBe(200); + }); + + it('keeps the lease while a slow install runs', async () => { + const a = await otherReplica(); + a.registry.setRegistryLeaseTimingForTests({ ttlMs: 90, waitMs: 50 }); + // Writing the bundle takes longer than the lease lasts. + const collection = store.collection.bind(store); + store.collection = (name, options) => { + const docs = collection(name, options); + return { ...docs, put: async (...args: Parameters) => { + if (args[0] === `${PLUGIN.id}.js`) await new Promise((r) => setTimeout(r, 300)); + return docs.put(...args); + } }; + }; + const saving = a.registry.savePlugin(PLUGIN, 'slow()'); + await new Promise((r) => setTimeout(r, 200)); + // Another replica tries to take over the registry meanwhile. + expect(await store.leases.acquire(LEASE, 'replica-b', 1_000)).toBe(false); + await expect(saving).resolves.toBeUndefined(); + store.collection = collection; + expect((await a.registry.getPlugin(PLUGIN.id))?.bundleHash).toBe(createHash('sha256').update('slow()').digest('hex')); + }); + + it('leaves the registry alone when the lease was lost, and every replica refuses the torn bundle', async () => { + const a = await otherReplica(); + await a.registry.savePlugin(PLUGIN, 'v1()'); + // The lease runs out mid-install (a stall) and another replica takes it. + const holder = store.leases.holder.bind(store.leases); + store.leases.holder = async () => 'replica-b'; + await expect(a.registry.savePlugin({ ...PLUGIN, version: '2.0.0' }, 'v2()')).rejects.toMatchObject({ name: 'StoreUnavailableError' }); + store.leases.holder = holder; + + const registry = JSON.parse((await (await import('../extension-store')).pluginDocs().get('registry.json'))!.data.toString()); + expect(registry.plugins[0]).toMatchObject({ version: '1.0.0', bundleHash: createHash('sha256').update('v1()').digest('hex') }); + + for (let i = 0; i < 2; i++) { + const replica = await otherReplica(); + const res = await getBundle(replica, PLUGIN.id); + expect(res.status).toBe(409); + expect((await res.json()).error).toContain('Install it again'); + } + }); +}); + describe('themes in the shared store', () => { it('serves on another replica a theme installed through one', async () => { const a = await otherReplica(); diff --git a/lib/admin/plugin-registry.ts b/lib/admin/plugin-registry.ts index 05d0ee170..6cd241c00 100644 --- a/lib/admin/plugin-registry.ts +++ b/lib/admin/plugin-registry.ts @@ -1,7 +1,7 @@ import type { CredentialSpec } from '@/lib/plugin-secrets/credentials'; import { createHash, randomBytes } from 'node:crypto'; import { logger } from '@/lib/logger'; -import { getStore, isSharedMode, replicaId } from '@/lib/shared-store'; +import { getStore, isSharedMode, replicaId, StoreUnavailableError } from '@/lib/shared-store'; import { announceChange } from '@/lib/shared-store/changes'; import { assertWritable, isConfigReadOnly } from './paths'; import { isStoreUnavailable } from './admin-store'; @@ -231,19 +231,32 @@ async function removeDoc(kind: RegistryKind, id: string): Promise { } } -const LEASE_TTL_MS = 30_000; -const LEASE_WAIT_MS = 15_000; +/** How long a registry lease lasts (renewed every third of it) and how long a change waits for one. */ +const leaseTiming = { ttlMs: 30_000, waitMs: 15_000 }; + +/** Shorter lease times (tests). */ +export function setRegistryLeaseTimingForTests(timing: Partial | null): void { + Object.assign(leaseTiming, timing ?? { ttlMs: 30_000, waitMs: 15_000 }); +} + +interface RegistryLease { + /** Throws (StoreUnavailableError, so a 503) unless this change still holds the lease. */ + assertHeld(): Promise; +} + +const NO_LEASE: RegistryLease = { async assertHeld() {} }; /** * Changes to one registry run one after another: in this process through a * promise chain on globalThis (every module copy shares it), and across * replicas under a lease, so two installs of the same extension at once - * can't leave one's bundle under the other's registry entry. The registry - * itself is changed with update(), so even a change that couldn't get the - * lease in time keeps every other entry. A store outage fails here, before - * anything is written. + * can't leave one's bundle under the other's registry entry. A change that + * can't get the lease in time is refused (503) before anything is written; + * the lease is renewed while the change runs, and an install checks it + * still holds it right before it writes the registry entry. The registry + * itself is changed with update(), so other entries always stay. */ -function withRegistryLock(kind: RegistryKind, task: () => Promise): Promise { +function withRegistryLock(kind: RegistryKind, task: (lease: RegistryLease) => Promise): Promise { const queues = registryGlobals().queues; const locked = () => withLease(kind, task); const run = (queues.get(kind) ?? Promise.resolve()).then(locked, locked); @@ -251,25 +264,58 @@ function withRegistryLock(kind: RegistryKind, task: () => Promise): Promis return run; } -async function withLease(kind: RegistryKind, task: () => Promise): Promise { - if (!isSharedMode() || isConfigReadOnly()) return task(); +async function withLease(kind: RegistryKind, task: (lease: RegistryLease) => Promise): Promise { + if (!isSharedMode() || isConfigReadOnly()) return task(NO_LEASE); const store = getStore(); const name = `extension-registry-${kind}`; const holder = `${replicaId()}-${randomBytes(4).toString('hex')}`; - const deadline = Date.now() + LEASE_WAIT_MS; - let held = await store.leases.acquire(name, holder, LEASE_TTL_MS); + const { ttlMs, waitMs } = leaseTiming; + const deadline = Date.now() + waitMs; + let held = await store.leases.acquire(name, holder, ttlMs); while (!held && Date.now() < deadline) { - await new Promise((resolve) => setTimeout(resolve, 100 + Math.random() * 100)); - held = await store.leases.acquire(name, holder, LEASE_TTL_MS); + await new Promise((resolve) => setTimeout(resolve, Math.min(100, waitMs / 4) * (1 + Math.random()))); + held = await store.leases.acquire(name, holder, ttlMs); } - if (!held) logger.warn(`Another instance is still changing the ${kind}; going ahead without waiting longer`); + if (!held) { + throw new StoreUnavailableError(`Another instance is still changing the ${kind}; nothing was changed. Try again in a moment.`); + } + let lost = false; + const renew = setInterval(() => { + store.leases.acquire(name, holder, ttlMs).then((ok) => { if (!ok) lost = true; }, () => { /* checked before the write */ }); + }, Math.max(10, Math.floor(ttlMs / 3))); + renew.unref?.(); + const lease: RegistryLease = { + async assertHeld() { + if (lost || (await store.leases.holder(name)) !== holder) { + logger.error(`Lost the lock on the ${kind} in the middle of a change; the registry was left as it was`); + throw new StoreUnavailableError(`This instance lost the lock on the ${kind} in the middle of a change; the registry was left as it was. Try again.`); + } + }, + }; try { - return await task(); + return await task(lease); } finally { - if (held) await store.leases.release(name, holder).catch(() => {}); + clearInterval(renew); + await store.leases.release(name, holder).catch(() => {}); } } +/** + * The stored bundle isn't the one the registry entry names: an install that + * stopped between writing the bundle and the entry. Refused the same way on + * every replica, so it is visibly broken instead of working on some. + */ +export class BundleMismatchError extends Error { + constructor(id: string) { + super(`The stored code of plugin ${id} does not match its registry entry (an install was interrupted). Install it again.`); + this.name = 'BundleMismatchError'; + } +} + +export function isBundleMismatch(error: unknown): boolean { + return (error as Error | null)?.name === 'BundleMismatchError'; +} + const sha256Hex = (data: string) => createHash('sha256').update(data).digest('hex'); // ─── Plugin Operations ─────────────────────────────────────── @@ -338,13 +384,14 @@ export async function getPlugin(id: string): Promise { const SAFE_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/; export async function savePlugin(plugin: ServerPlugin, code: string): Promise { - await withRegistryLock('plugins', () => savePluginNow(plugin, code)); + await withRegistryLock('plugins', (lease) => savePluginNow(plugin, code, lease)); await announceChange('plugins'); } async function savePluginNow( plugin: ServerPlugin, code: string, + lease: RegistryLease, ): Promise { assertWritable('install plugin'); if (!SAFE_ID_RE.test(plugin.id)) { @@ -380,6 +427,7 @@ async function savePluginNow( // installedAt across re-uploads. The full SHA-256 is required because the // client-side verifyBundle compares the entire digest length-checked. const now = new Date().toISOString(); + await lease.assertHeld(); await updateRegistry('plugins', emptyPluginRegistry, (registry) => { const idx = registry.plugins.findIndex(p => p.id === plugin.id); const next: ServerPlugin = { @@ -524,7 +572,8 @@ async function signatureVerifies(code: string, signature: string): Promise { if (!SAFE_ID_RE.test(id)) return null; @@ -540,7 +589,13 @@ export async function getPluginBundle(id: string, bundleHash?: string): Promise< if (isStoreUnavailable(error)) throw error; return null; } - if (code !== null && useCache && sha256Hex(code) === bundleHash) cacheBundle(bundleHash!, code); + if (code !== null && useCache && FULL_HASH_RE.test(bundleHash!)) { + if (sha256Hex(code) !== bundleHash) { + logger.error('[plugin-registry] stored plugin bundle does not match its registry entry', { id }); + throw new BundleMismatchError(id); + } + cacheBundle(bundleHash!, code); + } return code; } @@ -558,14 +613,15 @@ export async function getTheme(id: string): Promise { } export async function saveTheme(theme: ServerTheme, css: string, skin: string | null = null): Promise { - await withRegistryLock('themes', () => saveThemeNow(theme, css, skin)); + await withRegistryLock('themes', (lease) => saveThemeNow(theme, css, skin, lease)); await announceChange('themes'); } async function saveThemeNow( theme: ServerTheme, css: string, - skin: string | null = null, + skin: string | null, + lease: RegistryLease, ): Promise { assertWritable('install theme'); if (!SAFE_ID_RE.test(theme.id)) { @@ -597,6 +653,7 @@ async function saveThemeNow( } const next: ServerTheme = { ...theme, hasSkin: !!skin, contentHash }; + await lease.assertHeld(); await updateRegistry('themes', emptyThemeRegistry, (registry) => { const idx = registry.themes.findIndex(t => t.id === theme.id); if (idx >= 0) {