feat: share installed plugins and themes between instances #1066

app/api/admin/extensions/[kind]/[id]/rollback/route.ts

@@ -6,6 +6,7 @@ import { HistoryError, readHistoryEntry } from '@/lib/admin/extension-history';
 import { getPlugin, getTheme, savePlugin, saveTheme } from '@/lib/admin/plugin-registry';
 import { invalidateFrameOriginsCache } from '@/lib/admin/csp-frame-origins';
 import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response';
+import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response';
 import { EXTENSION_ID_RE } from '@/lib/extension-package';
 
 /**
@@ -87,6 +88,8 @@ export async function POST(
     }
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Extension rollback error', { error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Rollback failed' }, { status: 500 });
   }

app/api/admin/marketplace/route.ts

@@ -23,6 +23,7 @@ import {
 import { configManager } from '@/lib/admin/config-manager';
 import { checkBundleSignature } from '@/lib/admin/directory-keys';
 import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response';
+import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response';
 import { compareVersions } from '@/lib/version-compare';
 import { RELEASE_VERSION } from '@/lib/release-version';
 import {
@@ -386,6 +387,8 @@ export async function POST(request: NextRequest) {
   } catch (error) {
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     if (error instanceof DirectoryError) {
       return NextResponse.json({ error: error.message }, { status: 502 });
     }

app/api/admin/plugins/[id]/bundle/route.ts

@@ -3,6 +3,7 @@ import { getPluginBundle, getPlugin, getPluginSignature } from '@/lib/admin/plug
 import { getDevPlugin, readDevBundle } from '@/lib/admin/plugin-dev';
 import { signBytes } from '@/lib/admin/plugin-signing';
 import { canSeeInstanceDetails } from '@/lib/auth/instance-details';
+import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response';
 
 async function safeSign(code: string): Promise<string | null> {
   try { return await signBytes(code); } catch { return null; }
@@ -56,7 +57,7 @@ export async function GET(
       return NextResponse.json({ error: 'Plugin is disabled' }, { status: 403 });
     }
 
-    const code = await getPluginBundle(id);
+    const code = await getPluginBundle(id, plugin.bundleHash);
     if (!code) {
       return NextResponse.json({ error: 'Bundle not found' }, { status: 404 });
     }
@@ -71,7 +72,7 @@ export async function GET(
     };
     if (etag) headers['ETag'] = etag;
 
-    const signature = await getPluginSignature(id, code);
+    const signature = await getPluginSignature(id, code, plugin.bundleHash);
     if (signature) headers['X-Bundle-Signature'] = signature;
 
     if (etag && request.headers.get('if-none-match') === etag) {
@@ -80,7 +81,9 @@ export async function GET(
 
     headers['Content-Length'] = String(Buffer.byteLength(code, 'utf-8'));
     return new NextResponse(code, { headers });
-  } catch {
+  } catch (error) {
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     return NextResponse.json({ error: 'Not found' }, { status: 404 });
   }
 }

app/api/admin/plugins/route.ts

@@ -16,6 +16,7 @@ import { keptVersions, listHistory } from '@/lib/admin/extension-history';
 import { ExtensionPackageError, readPluginPackage, suspiciousCode } from '@/lib/extension-package';
 import { MAX_PLUGIN_SIZE } from '@/lib/plugin-types';
 import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response';
+import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response';
 
 /**
  * GET /api/admin/plugins - List all admin-managed plugins
@@ -133,6 +134,8 @@ export async function POST(request: NextRequest) {
   } catch (error) {
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Plugin install error', { error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
   }
@@ -186,6 +189,8 @@ export async function PATCH(request: NextRequest) {
   } catch (error) {
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Plugin update error', { error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
   }
@@ -219,6 +224,8 @@ export async function DELETE(request: NextRequest) {
   } catch (error) {
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Plugin delete error', { error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
   }

app/api/admin/themes/route.ts

@@ -15,6 +15,7 @@ import { keptVersions, listHistory } from '@/lib/admin/extension-history';
 
 import { readThemePackage } from '@/lib/extension-package';
 import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response';
+import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response';
 
 /**
  * GET /api/admin/themes - List all admin-managed themes
@@ -85,6 +86,8 @@ export async function POST(request: NextRequest) {
   } catch (error) {
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Theme install error', { error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
   }
@@ -124,6 +127,8 @@ export async function PATCH(request: NextRequest) {
   } catch (error) {
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Theme update error', { error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
   }
@@ -156,6 +161,8 @@ export async function DELETE(request: NextRequest) {
   } catch (error) {
     const readOnly = configReadOnlyResponse(error);
     if (readOnly) return readOnly;
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Theme delete error', { error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
   }

lib/admin/__tests__/bundle-cache.test.ts

@@ -0,0 +1,33 @@
+// @vitest-environment node
+import { beforeEach, describe, expect, it } from 'vitest';
+import { bundleCacheSize, cacheBundle, clearBundleCache, getCachedBundle, MAX_BYTES } from '../bundle-cache';
+
+// The per-replica bundle cache (#1066) must stay under its 64 MB cap and
+// drop the bundle used longest ago first.
+
+beforeEach(() => clearBundleCache());
+
+const big = (c: string) => c.repeat(20 * 1024 * 1024);
+
+describe('bundle cache', () => {
+  it('drops the least recently used bundle once it holds more than the cap', () => {
+    cacheBundle('a', big('a'));
+    cacheBundle('b', big('b'));
+    cacheBundle('c', big('c'));
+    expect(getCachedBundle('a')).toBeTruthy(); // a is now the most recently used
+    cacheBundle('d', big('d'));
+
+    expect(getCachedBundle('b')).toBeUndefined();
+    expect(getCachedBundle('a')?.code[0]).toBe('a');
+    expect(getCachedBundle('c')).toBeTruthy();
+    expect(getCachedBundle('d')).toBeTruthy();
+    expect(bundleCacheSize().bytes).toBeLessThanOrEqual(MAX_BYTES);
+  });
+
+  it('keeps a signature learned later with the bundle', () => {
+    cacheBundle('h', 'code()');
+    getCachedBundle('h')!.signature = 'sig';
+    cacheBundle('h', 'code()');
+    expect(getCachedBundle('h')).toMatchObject({ code: 'code()', signature: 'sig' });
+  });
+});

lib/admin/__tests__/plugin-registry-shared.test.ts

@@ -0,0 +1,250 @@
+// @vitest-environment node
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { createHash } from 'node:crypto';
+import { existsSync } from 'node:fs';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import path from 'node:path';
+import { NextRequest } from 'next/server';
+import JSZip from 'jszip';
+
+// Several replicas share one store (#1066): a plugin or theme installed
+// through one replica must be listed and served by every other one, two
+// installs at once on different replicas must both stay in the registry,
+// and an install while the store is down must answer 503 and leave nothing
+// half-written behind.
+
+vi.mock('@/lib/admin/session', () => ({
+  requireAdminAuth: vi.fn(async () => ({ session: { sub: 'admin' } })),
+  getClientIP: () => '127.0.0.1',
+}));
+vi.mock('@/lib/admin/audit', () => ({ auditLog: vi.fn(async () => {}) }));
+vi.mock('@/lib/admin/config-manager', () => ({
+  configManager: {
+    ensureLoaded: vi.fn(async () => {}),
+    get: vi.fn(() => undefined),
+    getPolicy: vi.fn(() => ({ forceEnabledPlugins: [] })),
+    updatePolicy: vi.fn(async () => {}),
+  },
+}));
+vi.mock('@/lib/admin/plugin-dev', () => ({
+  listDevPlugins: vi.fn(async () => []),
+  getDevPlugin: vi.fn(async () => null),
+  readDevBundle: vi.fn(),
+}));
+vi.mock('@/lib/auth/instance-details', () => ({ canSeeInstanceDetails: vi.fn(async () => true) }));
+
+import { setStoreForTests } from '@/lib/shared-store';
+import { resetLocalStoreForTests } from '@/lib/shared-store/local';
+import { INIT_EPOCH, resetChangesForTests } from '@/lib/shared-store/changes';
+import { separateFakeStore } from './shared-store-fake';
+
+const MANIFEST = {
+  id: 'demo-plugin', name: 'Demo', version: '1.0.0', author: 'me', description: 'd',
+  type: 'hook', permissions: ['email:read'], entrypoint: 'index.js',
+};
+const PLUGIN = {
+  id: 'pp', name: 'P', version: '1.0.0', author: 'a', description: '', type: 'hook',
+  permissions: [], entrypoint: 'index.js', enabled: true, installedAt: 'x', updatedAt: 'x',
+};
+
+let dir: string;
+let store: ReturnType<typeof separateFakeStore>;
+
+/**
+ * Another replica: fresh module copies and nothing of this process's
+ * memory (registry copy, bundle cache, write queue, signing key). Only the
+ * shared store is common.
+ */
+async function otherReplica() {
+  const g = globalThis as Record<symbol, unknown>;
+  for (const name of ['bulwark.admin.pluginRegistry', 'bulwark.admin.bundleCache', 'bulwark.admin.pluginSigning']) {
+    delete g[Symbol.for(name)];
+  }
+  vi.resetModules();
+  return {
+    registry: await import('../plugin-registry'),
+    plugins: await import('@/app/api/admin/plugins/route'),
+    themes: await import('@/app/api/admin/themes/route'),
+    bundle: await import('@/app/api/admin/plugins/[id]/bundle/route'),
+    themeCss: await import('@/app/api/admin/themes/[id]/css/route'),
+  };
+}
+
+async function zipOf(files: Record<string, string>) {
+  const zip = new JSZip();
+  for (const [n, d] of Object.entries(files)) zip.file(n, d);
+  return zip.generateAsync({ type: 'uint8array', compression: 'DEFLATE' });
+}
+
+async function install(replica: Awaited<ReturnType<typeof otherReplica>>, code: string) {
+  const form = new FormData();
+  const bytes = await zipOf({ 'manifest.json': JSON.stringify(MANIFEST), 'index.js': code });
+  form.append('file', new File([bytes as BlobPart], 'p.zip', { type: 'application/zip' }));
+  return replica.plugins.POST(new NextRequest('http://localhost/api/admin/plugins', { method: 'POST', body: form }));
+}
+
+function getBundle(replica: Awaited<ReturnType<typeof otherReplica>>, id = MANIFEST.id) {
+  return replica.bundle.GET(new NextRequest(`http://localhost/api/admin/plugins/${id}/bundle`), { params: Promise.resolve({ id }) });
+}
+
+const localPluginsDir = () => path.join(dir, 'config', 'plugins');
+
+beforeEach(async () => {
+  dir = await mkdtemp(path.join(tmpdir(), 'bw-registry-shared-'));
+  vi.stubEnv('ADMIN_CONFIG_DIR', path.join(dir, 'config'));
+  vi.stubEnv('ADMIN_STATE_DIR', path.join(dir, 'state'));
+  vi.stubEnv('PLUGIN_SIGNING_KEY_FILE', '');
+  resetLocalStoreForTests();
+  resetChangesForTests();
+  store = separateFakeStore(path.join(dir, 'store'));
+  setStoreForTests(store);
+  await store.epochs.bump(INIT_EPOCH);
+});
+
+afterEach(async () => {
+  setStoreForTests(null);
+  resetLocalStoreForTests();
+  resetChangesForTests();
+  vi.unstubAllEnvs();
+  await rm(dir, { recursive: true, force: true });
+});
+
+describe('plugins in the shared store', () => {
+  it('lists and serves on another replica a plugin installed through one', async () => {
+    const a = await otherReplica();
+    const code = 'module.exports = { hello: 1 }';
+    expect((await install(a, code)).status).toBe(200);
+    expect((await store.epochs.all()).plugins).toBeGreaterThan(0);
+
+    const b = await otherReplica();
+    const list = await (await b.plugins.GET(new NextRequest('http://localhost/api/admin/plugins'))).json();
+    expect(list.map((p: { id: string }) => p.id)).toEqual([MANIFEST.id]);
+
+    const res = await getBundle(b);
+    expect(res.status).toBe(200);
+    expect(await res.text()).toBe(code);
+    expect(res.headers.get('etag')).toBe(`"${createHash('sha256').update(code).digest('hex')}"`);
+    expect(res.headers.get('x-bundle-signature')).toBeTruthy();
+
+    // Nothing went to this replica's own config dir.
+    expect(existsSync(localPluginsDir())).toBe(false);
+  });
+
+  it('keeps serving a bundle it already served while the store is down', async () => {
+    const a = await otherReplica();
+    await install(a, 'module.exports = { cached: true }');
+    const b = await otherReplica();
+    expect((await getBundle(b)).status).toBe(200);
+
+    store.down = true;
+    const res = await getBundle(b);
+    expect(res.status).toBe(200);
+    expect(await res.text()).toBe('module.exports = { cached: true }');
+
+    // A replica that never read it can't: 503, not "not found".
+    const c = await otherReplica();
+    expect((await getBundle(c)).status).toBe(503);
+  });
+
+  it('serves the new bundle on another replica after an update', async () => {
+    const a = await otherReplica();
+    const b = await otherReplica();
+    await install(a, 'v1()');
+    expect(await (await getBundle(b)).text()).toBe('v1()');
+    await install(a, 'v2()');
+    expect(await (await getBundle(b)).text()).toBe('v2()');
+    expect(existsSync(localPluginsDir())).toBe(false);
+  });
+
+  it('keeps both plugins when two replicas install different ones at once', async () => {
+    const replicas = await Promise.all([otherReplica(), otherReplica()]);
+    // Each starts its install with nothing shared in memory.
+    const runs = replicas.map((r, i) => {
+      delete (globalThis as Record<symbol, unknown>)[Symbol.for('bulwark.admin.pluginRegistry')];
+      return r.registry.savePlugin({ ...PLUGIN, id: `plugin-${i}a` }, `code${i}()`);
+    });
+    await Promise.all(runs);
+
+    const c = await otherReplica();
+    expect((await c.registry.getPluginRegistry()).plugins.map((p) => p.id).sort()).toEqual(['plugin-0a', 'plugin-1a']);
+    const docs = (await import('../extension-store')).pluginDocs();
+    expect(JSON.parse((await docs.get('registry.json'))!.data.toString()).plugins).toHaveLength(2);
+  });
+
+  it('answers 503 for an install while the store is down and writes nothing', async () => {
+    const a = await otherReplica();
+    store.down = true;
+    const res = await install(a, 'module.exports = {}');
+    expect(res.status).toBe(503);
+
+    store.down = false;
+    const docs = (await import('../extension-store')).pluginDocs();
+    expect(await docs.list('')).toEqual([]);
+    expect(existsSync(localPluginsDir())).toBe(false);
+  });
+
+  it('answers 503 and writes nothing when the store goes down after the registry was read', async () => {
+    const a = await otherReplica();
+    await a.registry.getPluginRegistry();
+    store.down = true;
+    expect((await install(a, 'module.exports = {}')).status).toBe(503);
+
+    store.down = false;
+    const docs = (await import('../extension-store')).pluginDocs();
+    expect(await docs.list('')).toEqual([]);
+  });
+
+  it('fixes an old short bundle hash without losing an install that lands at the same time', async () => {
+    const docs = (await import('../extension-store')).pluginDocs();
+    await docs.put('old-one.js', 'old()');
+    await docs.put('registry.json', JSON.stringify({ plugins: [{ ...PLUGIN, id: 'old-one', bundleHash: 'abcdef0123456789' }] }));
+
+    const [a, b] = await Promise.all([otherReplica(), otherReplica()]);
+    await Promise.all([a.registry.getPluginRegistry(), b.registry.savePlugin({ ...PLUGIN, id: 'new-one' }, 'new()')]);
+
+    const stored = JSON.parse((await docs.get('registry.json'))!.data.toString());
+    expect(stored.plugins.map((p: { id: string }) => p.id).sort()).toEqual(['new-one', 'old-one']);
+    expect(stored.plugins.find((p: { id: string }) => p.id === 'old-one').bundleHash)
+      .toBe(createHash('sha256').update('old()').digest('hex'));
+  });
+
+  it('keeps the history in the store and rolls back on another replica', async () => {
+    const a = await otherReplica();
+    await a.registry.savePlugin(PLUGIN, 'v1()');
+    await a.registry.savePlugin({ ...PLUGIN, version: '1.1.0' }, 'v2()');
+
+    const b = await otherReplica();
+    const history = await import('../extension-history');
+    expect((await history.listHistory('plugin', 'pp')).map((h) => h.version)).toEqual(['1.0.0']);
+    const kept = await history.readHistoryEntry('plugin', 'pp', '1.0.0');
+    expect(kept.code).toBe('v1()');
+    await b.registry.deletePlugin('pp');
+    expect(await history.listHistory('plugin', 'pp')).toEqual([]);
+    expect(existsSync(localPluginsDir())).toBe(false);
+  });
+});
+
+describe('themes in the shared store', () => {
+  it('serves on another replica a theme installed through one', async () => {
+    const a = await otherReplica();
+    await a.registry.saveTheme(
+      { id: 'dark-blue', name: 'Dark blue', version: '1.0.0', author: 'a', description: '', variants: ['dark'], enabled: true, installedAt: 'x', updatedAt: 'x' },
+      ':root { --x: 1 }',
+      '.skin {}',
+    );
+    expect((await store.epochs.all()).themes).toBeGreaterThan(0);
+
+    const b = await otherReplica();
+    const res = await b.themeCss.GET(new NextRequest('http://localhost/api/admin/themes/dark-blue/css'), { params: Promise.resolve({ id: 'dark-blue' }) });
+    expect(res.status).toBe(200);
+    expect(await res.text()).toBe(':root { --x: 1 }');
+    expect(await b.registry.getThemeSkin('dark-blue')).toBe('.skin {}');
+    expect(existsSync(path.join(dir, 'config', 'themes'))).toBe(false);
+
+    store.down = true;
+    const c = await otherReplica();
+    const down = await c.themeCss.GET(new NextRequest('http://localhost/api/admin/themes/dark-blue/css'), { params: Promise.resolve({ id: 'dark-blue' }) });
+    expect(down.status).toBe(503);
+  });
+});

lib/admin/bundle-cache.ts

@@ -0,0 +1,70 @@
+/**
+ * Plugin bundles this replica served lately, keyed by their SHA-256
+ * (`bundleHash`), so serving a bundle in shared mode (#1066) doesn't fetch
+ * up to 5 MB from the shared store on every request. Content-addressed: a
+ * new install has a new hash, so nothing here can go stale; entries are only
+ * added after the bytes were checked against the hash. Least recently used
+ * entries go first once the cache holds more than MAX_BYTES.
+ *
+ * Per process, on globalThis (one cache for every module copy).
+ */
+
+export const MAX_BYTES = 64 * 1024 * 1024;
+
+export interface CachedBundle {
+  code: string;
+  /** The stored signature; undefined until it was read once. */
+  signature?: string | null;
+}
+
+interface BundleCacheState {
+  entries: Map<string, CachedBundle & { bytes: number }>;
+  bytes: number;
+}
+
+const KEY = Symbol.for('bulwark.admin.bundleCache');
+
+function state(): BundleCacheState {
+  const g = globalThis as typeof globalThis & { [KEY]?: BundleCacheState };
+  return (g[KEY] ??= { entries: new Map(), bytes: 0 });
+}
+
+export function getCachedBundle(hash: string): CachedBundle | undefined {
+  const s = state();
+  const entry = s.entries.get(hash);
+  if (!entry) return undefined;
+  // Most recently used last.
+  s.entries.delete(hash);
+  s.entries.set(hash, entry);
+  return entry;
+}
+
+export function cacheBundle(hash: string, code: string, signature?: string | null): void {
+  const s = state();
+  const previous = s.entries.get(hash);
+  if (previous) {
+    s.entries.delete(hash);
+    s.bytes -= previous.bytes;
+  }
+  const bytes = Buffer.byteLength(code, 'utf-8');
+  if (bytes > MAX_BYTES) return;
+  s.entries.set(hash, { code, signature: signature ?? previous?.signature, bytes });
+  s.bytes += bytes;
+  for (const [oldest, entry] of s.entries) {
+    if (s.bytes <= MAX_BYTES) break;
+    s.entries.delete(oldest);
+    s.bytes -= entry.bytes;
+  }
+}
+
+/** How much the cache holds (tests). */
+export function bundleCacheSize(): { entries: number; bytes: number } {
+  const s = state();
+  return { entries: s.entries.size, bytes: s.bytes };
+}
+
+export function clearBundleCache(): void {
+  const s = state();
+  s.entries.clear();
+  s.bytes = 0;
+}

lib/admin/extension-history.ts

@@ -22,13 +22,19 @@
  * current enabled/forced flags are kept. The snapshot's hash is checked
  * before it is restored: a history file edited on disk is refused rather
  * than installed for every user.
+ *
+ * The files are documents of the `plugins` and `themes` collections
+ * (extension-store.ts) under `history/<id>/<version>/`: the paths above in
+ * local mode, the shared store in shared mode (#1066).
  */
 
-import { mkdir, readFile, readdir, rm, writeFile } from 'node:fs/promises';
+import { readdir, rm } from 'node:fs/promises';
 import { createHash } from 'node:crypto';
 import path from 'node:path';
 import { logger } from '@/lib/logger';
-import { getConfigDir } from './paths';
+import { isSharedMode, type DocCollection } from '@/lib/shared-store';
+import { isStoreUnavailable } from './admin-store';
+import { pluginDocs, pluginsOptions, themeDocs, themesOptions } from './extension-store';
 import type { ServerPlugin, ServerTheme } from './plugin-registry';
 
 export type ExtensionKind = 'plugin' | 'theme';
@@ -49,9 +55,39 @@ export interface HistoryEntry {
 
 const SAFE_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/;
 
-function historyRoot(kind: ExtensionKind, id: string): string {
+/** The extension's history folder: a document id prefix, without the trailing slash. */
+function historyRoot(id: string): string {
   if (!SAFE_ID_RE.test(id)) throw new Error('Invalid extension id');
-  return path.join(getConfigDir(), kind === 'plugin' ? 'plugins' : 'themes', 'history', id);
+  return `history/${id}`;
+}
+
+const docsOf = (kind: ExtensionKind): DocCollection => (kind === 'plugin' ? pluginDocs() : themeDocs());
+const localDir = (kind: ExtensionKind): string => (kind === 'plugin' ? pluginsOptions : themesOptions).local.dir();
+
+async function readText(kind: ExtensionKind, id: string): Promise<string | null> {
+  try {
+    const doc = await docsOf(kind).get(id);
+    return doc ? doc.data.toString('utf-8') : null;
+  } catch (error) {
+    if (isStoreUnavailable(error)) throw error;
+    return null;
+  }
+}
+
+/**
+ * Remove a history folder. Local mode: `rm -r` of the folder, as before.
+ * Shared mode: entry.json first, so no reader sees a kept version without
+ * its files, then the rest.
+ */
+async function removeFolder(kind: ExtensionKind, folder: string): Promise<void> {
+  if (!isSharedMode()) {
+    await rm(path.join(localDir(kind), ...folder.split('/')), { recursive: true, force: true });
+    return;
+  }
+  const docs = docsOf(kind);
+  const ids = await docs.list(`${folder}/`);
+  const entry = `${folder}/entry.json`;
+  for (const id of [...ids.filter((i) => i === entry), ...ids.filter((i) => i !== entry)]) await docs.delete(id);
 }
 
 /**
@@ -78,9 +114,11 @@ interface StoredEntry extends HistoryEntry {
   entry: ServerPlugin | ServerTheme;
 }
 
-async function readEntry(dir: string): Promise<StoredEntry | null> {
+async function readEntry(kind: ExtensionKind, dir: string): Promise<StoredEntry | null> {
+  const text = await readText(kind, `${dir}/entry.json`);
+  if (text === null) return null;
   try {
-    const raw = JSON.parse(await readFile(path.join(dir, 'entry.json'), 'utf-8')) as StoredEntry;
+    const raw = JSON.parse(text) as StoredEntry;
     if (!raw || typeof raw.version !== 'string' || typeof raw.savedAt !== 'string' || !raw.entry) return null;
     return raw;
   } catch {
@@ -88,18 +126,36 @@ async function readEntry(dir: string): Promise<StoredEntry | null> {
   }
 }
 
+/**
+ * The version folders under an extension's history. Local mode reads the
+ * one directory, as before (listing the collection would walk every file
+ * under C/plugins on each install).
+ */
+async function versionFolders(kind: ExtensionKind, root: string): Promise<string[]> {
+  if (!isSharedMode()) {
+    return readdir(path.join(localDir(kind), ...root.split('/')));
+  }
+  const names = new Set<string>();
+  for (const docId of await docsOf(kind).list(`${root}/`)) {
+    const name = docId.slice(root.length + 1).split('/')[0];
+    if (name) names.add(name);
+  }
+  return [...names];
+}
+
 async function readAll(kind: ExtensionKind, id: string): Promise<Array<StoredEntry & { dir: string }>> {
-  const root = historyRoot(kind, id);
+  const root = historyRoot(id);
   let names: string[];
   try {
-    names = await readdir(root);
-  } catch {
+    names = await versionFolders(kind, root);
+  } catch (error) {
+    if (isStoreUnavailable(error)) throw error;
     return [];
   }
   const out: Array<StoredEntry & { dir: string }> = [];
   for (const name of names) {
-    const dir = path.join(root, name);
-    const entry = await readEntry(dir);
+    const dir = `${root}/${name}`;
+    const entry = await readEntry(kind, dir);
     if (entry) out.push({ ...entry, dir });
   }
   // Newest replacement first.
@@ -110,7 +166,7 @@ async function readAll(kind: ExtensionKind, id: string): Promise<Array<StoredEnt
 async function prune(kind: ExtensionKind, id: string): Promise<void> {
   const all = await readAll(kind, id);
   for (const stale of all.slice(KEEP)) {
-    await rm(stale.dir, { recursive: true, force: true });
+    await removeFolder(kind, stale.dir);
   }
 }
 
@@ -142,8 +198,7 @@ export async function listHistory(kind: ExtensionKind, id: string): Promise<Hist
  * inside its write lock, with the entry and code currently installed.
  */
 export async function snapshotPlugin(current: ServerPlugin, code: string): Promise<void> {
-  const dir = path.join(historyRoot('plugin', current.id), dirNameFor(current.version));
-  await mkdir(dir, { recursive: true });
+  const dir = `${historyRoot(current.id)}/${dirNameFor(current.version)}`;
   const stored: StoredEntry = {
     kind: 'plugin',
     version: current.version,
@@ -152,15 +207,15 @@ export async function snapshotPlugin(current: ServerPlugin, code: string): Promi
     ...(current.source ? { source: current.source } : {}),
     entry: current,
   };
-  await writeFile(path.join(dir, 'bundle.js'), code, 'utf-8');
-  await writeFile(path.join(dir, 'entry.json'), JSON.stringify(stored, null, 2), 'utf-8');
+  const docs = docsOf('plugin');
+  await docs.put(`${dir}/bundle.js`, code);
+  await docs.put(`${dir}/entry.json`, JSON.stringify(stored, null, 2));
   await prune('plugin', current.id);
 }
 
 /** Same for a theme: its registry entry, CSS and skin. */
 export async function snapshotTheme(current: ServerTheme, css: string, skin: string | null): Promise<void> {
-  const dir = path.join(historyRoot('theme', current.id), dirNameFor(current.version));
-  await mkdir(dir, { recursive: true });
+  const dir = `${historyRoot(current.id)}/${dirNameFor(current.version)}`;
   const stored: StoredEntry = {
     kind: 'theme',
     version: current.version,
@@ -169,10 +224,11 @@ export async function snapshotTheme(current: ServerTheme, css: string, skin: str
     ...(current.source ? { source: current.source } : {}),
     entry: current,
   };
-  await writeFile(path.join(dir, 'theme.css'), css, 'utf-8');
-  if (skin !== null) await writeFile(path.join(dir, 'skin.css'), skin, 'utf-8');
-  else await rm(path.join(dir, 'skin.css'), { force: true });
-  await writeFile(path.join(dir, 'entry.json'), JSON.stringify(stored, null, 2), 'utf-8');
+  const docs = docsOf('theme');
+  await docs.put(`${dir}/theme.css`, css);
+  if (skin !== null) await docs.put(`${dir}/skin.css`, skin);
+  else await docs.delete(`${dir}/skin.css`);
+  await docs.put(`${dir}/entry.json`, JSON.stringify(stored, null, 2));
   await prune('theme', current.id);
 }
 
@@ -186,7 +242,7 @@ export async function snapshotTheme(current: ServerTheme, css: string, skin: str
 export async function forgetVersion(kind: ExtensionKind, id: string, version: string): Promise<void> {
   if (!SAFE_ID_RE.test(id)) return;
   for (const kept of await readAll(kind, id)) {
-    if (kept.version === version) await rm(kept.dir, { recursive: true, force: true });
+    if (kept.version === version) await removeFolder(kind, kept.dir);
   }
 }
 
@@ -194,7 +250,7 @@ export async function forgetVersion(kind: ExtensionKind, id: string, version: st
 export async function purgeHistory(kind: ExtensionKind, id: string): Promise<void> {
   if (!SAFE_ID_RE.test(id)) return;
   try {
-    await rm(historyRoot(kind, id), { recursive: true, force: true });
+    await removeFolder(kind, historyRoot(id));
   } catch (err) {
     logger.warn('[extension-history] could not remove history', { kind, id, error: err instanceof Error ? err.message : String(err) });
   }
@@ -228,13 +284,13 @@ export async function readHistoryEntry<K extends ExtensionKind>(kind: K, id: str
   const stored = await find(kind, id, version);
   const damaged = () => new HistoryError(`The kept copy of ${id} ${version} does not match what was saved; it was not restored`, 409);
   if (kind === 'plugin') {
-    const code = await readFile(path.join(stored.dir, 'bundle.js'), 'utf-8').catch(() => null);
+    const code = await readText(kind, `${stored.dir}/bundle.js`);
     if (code === null || hashOf([code]) !== stored.contentHash) throw damaged();
     const restored: RestoredPlugin = { kind: 'plugin', entry: stored.entry as ServerPlugin, code };
     return restored as Restored<K>;
   }
-  const css = await readFile(path.join(stored.dir, 'theme.css'), 'utf-8').catch(() => null);
-  const skin = await readFile(path.join(stored.dir, 'skin.css'), 'utf-8').catch(() => null);
+  const css = await readText(kind, `${stored.dir}/theme.css`);
+  const skin = await readText(kind, `${stored.dir}/skin.css`);
   if (css === null || hashOf([css, skin]) !== stored.contentHash) throw damaged();
   const restored: RestoredTheme = { kind: 'theme', entry: stored.entry as ServerTheme, css, skin };
   return restored as Restored<K>;

lib/admin/plugin-registry.ts

@@ -1,20 +1,14 @@
 import type { CredentialSpec } from '@/lib/plugin-secrets/credentials';
-import { readFile, writeFile, mkdir, rename, unlink } from 'node:fs/promises';
-import { existsSync } from 'node:fs';
-import { createHash } from 'node:crypto';
-import path from 'node:path';
+import { createHash, randomBytes } from 'node:crypto';
 import { logger } from '@/lib/logger';
-import { getConfigDir, assertWritable } from './paths';
+import { getStore, isSharedMode, replicaId } from '@/lib/shared-store';
+import { announceChange } from '@/lib/shared-store/changes';
+import { assertWritable, isConfigReadOnly } from './paths';
+import { isStoreUnavailable } from './admin-store';
+import { cacheBundle, getCachedBundle } from './bundle-cache';
+import { pluginDocs, themeDocs } from './extension-store';
 import { forgetVersion, purgeHistory, snapshotPlugin, snapshotTheme } from './extension-history';
 
-function getPluginsDir(): string {
-  return path.join(getConfigDir(), 'plugins');
-}
-
-function getThemesDir(): string {
-  return path.join(getConfigDir(), 'themes');
-}
-
 // ─── Types ───────────────────────────────────────────────────
 
 export interface PluginConfigField {
@@ -145,48 +139,142 @@ interface ThemeRegistry {
   themes: ServerTheme[];
 }
 
-// ─── Plugin Registry ─────────────────────────────────────────
+// ─── Storage ─────────────────────────────────────────────────
+//
+// Both registries, their bundles, signatures, CSS and history are
+// documents of the `plugins` and `themes` collections (extension-store.ts):
+// the files under C/plugins and C/themes in local mode, the shared store in
+// shared mode (#1066). Every change writes the bundle (and its signature)
+// before the registry entry that points at it, changes the registry with
+// update() so a concurrent change of another entry stays, and then
+// announces area `plugins` or `themes` so every replica drops what it
+// derived from the old registry (the CSP frame origins, for one).
 
-async function ensureDir(dir: string): Promise<void> {
-  if (!existsSync(dir)) {
-    await mkdir(dir, { recursive: true });
-  }
+type RegistryKind = 'plugins' | 'themes';
+
+interface RegistryGlobals {
+  /** One change at a time per registry in this process. */
+  queues: Map<string, Promise<unknown>>;
+  /** Shared mode: the last registry read, served while the store is down. */
+  lastGood: Map<RegistryKind, string>;
 }
 
-async function readJsonFile<T>(filePath: string, fallback: T): Promise<T> {
+const GLOBAL_KEY = Symbol.for('bulwark.admin.pluginRegistry');
+
+function registryGlobals(): RegistryGlobals {
+  const g = globalThis as typeof globalThis & { [GLOBAL_KEY]?: RegistryGlobals };
+  return (g[GLOBAL_KEY] ??= { queues: new Map(), lastGood: new Map() });
+}
+
+const docsFor = (kind: RegistryKind) => (kind === 'plugins' ? pluginDocs() : themeDocs());
+
+const REGISTRY_ID = 'registry.json';
+
+function serialize(registry: unknown): string {
+  return JSON.stringify(registry, null, 2);
+}
+
+/** A registry document, or `fallback` when it is missing or doesn't parse (as a broken file always did). */
+function parseRegistry<T>(kind: RegistryKind, data: Buffer | null, fallback: () => T): T {
+  if (!data) return fallback();
   try {
-    const raw = await readFile(filePath, 'utf-8');
-    return JSON.parse(raw);
+    return JSON.parse(data.toString('utf-8')) as T;
   } catch (error) {
-    if ((error as NodeJS.ErrnoException).code === 'ENOENT') return fallback;
-    logger.warn(`Failed to read ${filePath}`, { error: error instanceof Error ? error.message : 'Unknown error' });
-    return fallback;
+    logger.warn(`Failed to read the ${kind} registry`, { error: error instanceof Error ? error.message : 'Unknown error' });
+    return fallback();
   }
 }
 
-async function writeJsonFile(filePath: string, data: unknown): Promise<void> {
-  const dir = path.dirname(filePath);
-  await ensureDir(dir);
-  // A temp name per write: two writers sharing one ".tmp" overwrote each
-  // other's file and the second rename failed (a 500 on concurrent installs).
-  const tmpPath = `${filePath}.${process.pid}.${Date.now().toString(36)}${Math.random().toString(36).slice(2, 8)}.tmp`;
-  await writeFile(tmpPath, JSON.stringify(data, null, 2), 'utf-8');
-  await rename(tmpPath, filePath);
+/**
+ * Read a registry. A store outage serves the last copy this replica read
+ * (config reads keep the last good copy, plan §6); with none it throws.
+ */
+async function readRegistry<T>(kind: RegistryKind, fallback: () => T): Promise<T> {
+  const g = registryGlobals();
+  let data: Buffer | null;
+  try {
+    data = (await docsFor(kind).get(REGISTRY_ID))?.data ?? null;
+  } catch (error) {
+    if (isStoreUnavailable(error)) {
+      const last = g.lastGood.get(kind);
+      if (last !== undefined) return JSON.parse(last) as T;
+      throw error;
+    }
+    logger.warn(`Failed to read the ${kind} registry`, { error: error instanceof Error ? error.message : 'Unknown error' });
+    return fallback();
+  }
+  const registry = parseRegistry(kind, data, fallback);
+  if (isSharedMode()) g.lastGood.set(kind, serialize(registry));
+  return registry;
 }
 
-// Changes to one registry file run one after another, so two admin actions
-// at once can't read the same old list and drop each other's change.
-const registryQueues = new Map<string, Promise<unknown>>();
-function withRegistryLock<T>(name: 'plugins' | 'themes', task: () => Promise<T>): Promise<T> {
-  const run = (registryQueues.get(name) ?? Promise.resolve()).then(task, task);
-  const tail = run.catch(() => {});
-  registryQueues.set(name, tail);
+/** Change a registry with update(); `change` may run more than once. */
+async function updateRegistry<T>(kind: RegistryKind, fallback: () => T, change: (registry: T) => boolean): Promise<void> {
+  const result = await docsFor(kind).update(REGISTRY_ID, (current) => {
+    const registry = parseRegistry(kind, current, fallback);
+    return change(registry) ? serialize(registry) : undefined;
+  });
+  if (result.data && isSharedMode()) registryGlobals().lastGood.set(kind, result.data.toString('utf-8'));
+}
+
+async function readText(kind: RegistryKind, id: string): Promise<string | null> {
+  const doc = await docsFor(kind).get(id);
+  return doc ? doc.data.toString('utf-8') : null;
+}
+
+/** Delete a document; only a store outage counts as a failure (a file that is gone already is fine). */
+async function removeDoc(kind: RegistryKind, id: string): Promise<void> {
+  try {
+    await docsFor(kind).delete(id);
+  } catch (error) {
+    if (isStoreUnavailable(error)) throw error;
+  }
+}
+
+const LEASE_TTL_MS = 30_000;
+const LEASE_WAIT_MS = 15_000;
+
+/**
+ * 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.
+ */
+function withRegistryLock<T>(kind: RegistryKind, task: () => Promise<T>): Promise<T> {
+  const queues = registryGlobals().queues;
+  const locked = () => withLease(kind, task);
+  const run = (queues.get(kind) ?? Promise.resolve()).then(locked, locked);
+  queues.set(kind, run.catch(() => {}));
   return run;
 }
 
+async function withLease<T>(kind: RegistryKind, task: () => Promise<T>): Promise<T> {
+  if (!isSharedMode() || isConfigReadOnly()) return task();
+  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);
+  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);
+  }
+  if (!held) logger.warn(`Another instance is still changing the ${kind}; going ahead without waiting longer`);
+  try {
+    return await task();
+  } finally {
+    if (held) await store.leases.release(name, holder).catch(() => {});
+  }
+}
+
+const sha256Hex = (data: string) => createHash('sha256').update(data).digest('hex');
+
 // ─── Plugin Operations ───────────────────────────────────────
 
-const pluginRegistryPath = () => path.join(getPluginsDir(), 'registry.json');
+const emptyPluginRegistry = (): PluginRegistry => ({ plugins: [] });
 
 const FULL_HASH_RE = /^[0-9a-f]{64}$/;
 
@@ -194,30 +282,48 @@ const FULL_HASH_RE = /^[0-9a-f]{64}$/;
  * Older builds wrote a 16-char SHA-256 prefix into `bundleHash`. The current
  * client-side verifyBundle requires equal-length hex (and the full digest for
  * real integrity), so any registry entry with a truncated or otherwise
- * malformed hash needs to be re-hashed from the on-disk bundle. If the bundle
- * file is missing the hash is cleared so verifyBundle skips the check rather
+ * malformed hash needs to be re-hashed from the stored bundle. If the bundle
+ * is missing the hash is cleared so verifyBundle skips the check rather
  * than refusing to load.
+ *
+ * Returns the fixed hash per plugin id and the hash it replaces (null =
+ * clear it), or null when nothing needs fixing.
  */
-async function migrateBundleHashes(registry: PluginRegistry): Promise<boolean> {
-  let changed = false;
+async function bundleHashFixes(registry: PluginRegistry): Promise<Map<string, { from: string; to: string | null }> | null> {
+  const fixes = new Map<string, { from: string; to: string | null }>();
   for (const plugin of registry.plugins) {
     if (!plugin.bundleHash || FULL_HASH_RE.test(plugin.bundleHash)) continue;
-    const bundlePath = path.join(getPluginsDir(), `${plugin.id}.js`);
+    let code: string | null = null;
     try {
-      const code = await readFile(bundlePath);
-      plugin.bundleHash = createHash('sha256').update(code).digest('hex');
+      code = await readText('plugins', `${plugin.id}.js`);
     } catch {
-      delete plugin.bundleHash;
+      /* unreadable: cleared, as a missing bundle */
     }
+    fixes.set(plugin.id, { from: plugin.bundleHash, to: code === null ? null : sha256Hex(code) });
+  }
+  return fixes.size > 0 ? fixes : null;
+}
+
+function applyHashFixes(registry: PluginRegistry, fixes: Map<string, { from: string; to: string | null }>): boolean {
+  let changed = false;
+  for (const plugin of registry.plugins) {
+    const fix = fixes.get(plugin.id);
+    // Only an entry that still has the old hash: a save in between wins.
+    if (!fix || plugin.bundleHash !== fix.from) continue;
+    if (fix.to) plugin.bundleHash = fix.to;
+    else delete plugin.bundleHash;
     changed = true;
   }
   return changed;
 }
 
 export async function getPluginRegistry(): Promise<PluginRegistry> {
-  const registry = await readJsonFile<PluginRegistry>(pluginRegistryPath(), { plugins: [] });
-  if (await migrateBundleHashes(registry)) {
-    try { await writeJsonFile(pluginRegistryPath(), registry); } catch { /* read-only fs ok */ }
+  const registry = await readRegistry('plugins', emptyPluginRegistry);
+  const fixes = await bundleHashFixes(registry);
+  if (fixes) {
+    applyHashFixes(registry, fixes);
+    // Stored with update(), so an install that lands meanwhile isn't undone.
+    try { await updateRegistry('plugins', emptyPluginRegistry, (stored) => applyHashFixes(stored, fixes)); } catch { /* read-only fs ok */ }
   }
   return registry;
 }
@@ -229,8 +335,9 @@ export async function getPlugin(id: string): Promise<ServerPlugin | null> {
 
 const SAFE_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/;
 
-export function savePlugin(plugin: ServerPlugin, code: string): Promise<void> {
-  return withRegistryLock('plugins', () => savePluginNow(plugin, code));
+export async function savePlugin(plugin: ServerPlugin, code: string): Promise<void> {
+  await withRegistryLock('plugins', () => savePluginNow(plugin, code));
+  await announceChange('plugins');
 }
 
 async function savePluginNow(
@@ -241,134 +348,179 @@ async function savePluginNow(
   if (!SAFE_ID_RE.test(plugin.id)) {
     throw new Error('Invalid plugin id');
   }
-  const dir = getPluginsDir();
-  await ensureDir(dir);
 
-  const bundleHash = createHash('sha256').update(code).digest('hex');
-  const registry = await getPluginRegistry();
-  const idx = registry.plugins.findIndex(p => p.id === plugin.id);
+  const bundleHash = sha256Hex(code);
+  const current = (await getPluginRegistry()).plugins.find(p => p.id === plugin.id);
+  // Sign now and keep the signature with the bundle. Signing at serve time
+  // would sign whatever is stored, so a bundle swapped after install would
+  // still verify. Signed before anything is written: a key that can't be
+  // loaded from the shared store fails the install, not half of it.
+  const signature = await signBundle(plugin.id, code);
 
   // Keep what is about to be replaced, so the update can be undone
   // (extension-history.ts). The same bytes under the same version are not
   // a replacement worth keeping.
-  const bundlePath = path.join(dir, `${plugin.id}.js`);
-  if (idx >= 0) {
-    const current = registry.plugins[idx];
+  if (current) {
     await forgetVersion('plugin', plugin.id, plugin.version);
     if (current.version !== plugin.version || current.bundleHash !== bundleHash) {
-      const currentCode = await readFile(bundlePath, 'utf-8').catch(() => null);
+      const currentCode = await readText('plugins', `${plugin.id}.js`);
       if (currentCode !== null) await snapshotPlugin(current, currentCode);
     }
   }
 
-  // Save code bundle
-  await writeFile(bundlePath, code, 'utf-8');
-  // Sign now and keep the signature with the bundle. Signing at serve time
-  // would sign whatever is on disk, so a bundle file swapped after install
-  // would still verify.
-  await storePluginSignature(plugin.id, code);
+  // The bundle and its signature first, then the registry entry pointing at them.
+  const docs = pluginDocs();
+  await docs.put(`${plugin.id}.js`, code);
+  await storeSignature(plugin.id, signature);
 
   // Stamp content hash + updatedAt so clients can detect re-uploads even
   // when the manifest version hasn't changed. Preserve the original
   // 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();
-
-  const next: ServerPlugin = {
-    ...plugin,
-    bundleHash,
-    updatedAt: now,
-    installedAt: idx >= 0 ? registry.plugins[idx].installedAt : plugin.installedAt,
-  };
-  if (idx >= 0) {
-    registry.plugins[idx] = next;
-  } else {
-    registry.plugins.push(next);
-  }
-  await writeJsonFile(pluginRegistryPath(), registry);
+  await updateRegistry('plugins', emptyPluginRegistry, (registry) => {
+    const idx = registry.plugins.findIndex(p => p.id === plugin.id);
+    const next: ServerPlugin = {
+      ...plugin,
+      bundleHash,
+      updatedAt: now,
+      installedAt: idx >= 0 ? registry.plugins[idx].installedAt : plugin.installedAt,
+    };
+    if (idx >= 0) {
+      registry.plugins[idx] = next;
+    } else {
+      registry.plugins.push(next);
+    }
+    return true;
+  });
+  if (isSharedMode()) cacheBundle(bundleHash, code, signature);
 }
 
-export function updatePluginMeta(id: string, updates: Partial<Pick<ServerPlugin, 'enabled' | 'forceEnabled'>>): Promise<ServerPlugin | null> {
-  return withRegistryLock('plugins', () => updatePluginMetaNow(id, updates));
+export async function updatePluginMeta(id: string, updates: Partial<Pick<ServerPlugin, 'enabled' | 'forceEnabled'>>): Promise<ServerPlugin | null> {
+  const updated = await withRegistryLock('plugins', () => updatePluginMetaNow(id, updates));
+  if (updated) await announceChange('plugins');
+  return updated;
 }
 
 async function updatePluginMetaNow(id: string, updates: Partial<Pick<ServerPlugin, 'enabled' | 'forceEnabled'>>): Promise<ServerPlugin | null> {
   assertWritable('update plugin metadata');
-  const registry = await getPluginRegistry();
-  const idx = registry.plugins.findIndex(p => p.id === id);
-  if (idx < 0) return null;
-
-  registry.plugins[idx] = { ...registry.plugins[idx], ...updates, updatedAt: new Date().toISOString() };
-  await writeJsonFile(pluginRegistryPath(), registry);
-  return registry.plugins[idx];
+  const now = new Date().toISOString();
+  let updated: ServerPlugin | null = null;
+  await updateRegistry('plugins', emptyPluginRegistry, (registry) => {
+    updated = null;
+    const idx = registry.plugins.findIndex(p => p.id === id);
+    if (idx < 0) return false;
+    registry.plugins[idx] = { ...registry.plugins[idx], ...updates, updatedAt: now };
+    updated = registry.plugins[idx];
+    return true;
+  });
+  return updated;
 }
 
-export function deletePlugin(id: string): Promise<boolean> {
-  return withRegistryLock('plugins', () => deletePluginNow(id));
+export async function deletePlugin(id: string): Promise<boolean> {
+  const deleted = await withRegistryLock('plugins', () => deletePluginNow(id));
+  if (deleted) await announceChange('plugins');
+  return deleted;
 }
 
 async function deletePluginNow(id: string): Promise<boolean> {
   assertWritable('delete plugin');
-  const registry = await getPluginRegistry();
-  const idx = registry.plugins.findIndex(p => p.id === id);
-  if (idx < 0) return false;
+  let found = false;
+  await updateRegistry('plugins', emptyPluginRegistry, (registry) => {
+    const idx = registry.plugins.findIndex(p => p.id === id);
+    found = idx >= 0;
+    if (!found) return false;
+    registry.plugins.splice(idx, 1);
+    return true;
+  });
+  if (!found) return false;
 
-  registry.plugins.splice(idx, 1);
-  await writeJsonFile(pluginRegistryPath(), registry);
-
-  // Remove bundle and signature files
-  for (const file of [`${id}.js`, `${id}.sig`]) {
-    try { await unlink(path.join(getPluginsDir(), file)); } catch { /* ok if missing */ }
+  // Remove bundle and signature (the registry no longer points at them)
+  if (SAFE_ID_RE.test(id)) {
+    for (const file of [`${id}.js`, `${id}.sig`]) await removeDoc('plugins', file);
   }
   await purgeHistory('plugin', id);
 
   return true;
 }
 
-async function storePluginSignature(id: string, code: string): Promise<string | null> {
-  const sigPath = path.join(getPluginsDir(), `${id}.sig`);
+/** Sign a bundle; null when no key can be loaded (the plugin then can't run privileged). */
+async function signBundle(id: string, code: string): Promise<string | null> {
   try {
     const { signBytes } = await import('./plugin-signing');
-    const signature = await signBytes(code);
-    await writeFile(sigPath, signature, 'utf-8');
-    return signature;
+    return await signBytes(code);
   } catch (err) {
+    if (isStoreUnavailable(err)) throw err;
     // Without a key the bundle still installs; it just can't run privileged.
-    try { await unlink(sigPath); } catch { /* none */ }
     logger.warn('[plugin-registry] could not sign plugin bundle', { id, error: err instanceof Error ? err.message : String(err) });
     return null;
   }
 }
 
+async function storeSignature(id: string, signature: string | null): Promise<void> {
+  if (signature) await pluginDocs().put(`${id}.sig`, signature);
+  else await removeDoc('plugins', `${id}.sig`);
+}
+
 /**
  * The signature made when the bundle was installed. Bundles installed before
  * signatures were stored get one on first request (trust on first use), so
- * a later change to the file on disk stops verifying from then on.
+ * a later change to the stored bundle stops verifying from then on.
+ *
+ * `bundleHash` (the registry entry's) lets shared mode answer from this
+ * replica's bundle cache.
  */
-export async function getPluginSignature(id: string, code: string): Promise<string | null> {
+export async function getPluginSignature(id: string, code: string, bundleHash?: string): Promise<string | null> {
   if (!SAFE_ID_RE.test(id)) return null;
+  const cached = bundleHash && isSharedMode() ? getCachedBundle(bundleHash) : undefined;
+  if (cached && cached.code === code && cached.signature !== undefined) return cached.signature;
+  let stored: string | null;
   try {
-    return (await readFile(path.join(getPluginsDir(), `${id}.sig`), 'utf-8')).trim() || null;
-  } catch {
-    return storePluginSignature(id, code);
+    stored = await readText('plugins', `${id}.sig`);
+  } catch (error) {
+    if (isStoreUnavailable(error)) throw error;
+    stored = null;
   }
+  let signature: string | null;
+  if (stored !== null) {
+    signature = stored.trim() || null;
+  } else {
+    signature = await signBundle(id, code);
+    await storeSignature(id, signature);
+  }
+  if (cached && cached.code === code) cached.signature = signature;
+  return signature;
 }
 
-export async function getPluginBundle(id: string): Promise<string | null> {
-  const bundlePath = path.join(getPluginsDir(), `${id}.js`);
+/**
+ * The stored bundle of a plugin. With `bundleHash` (the registry entry's),
+ * shared mode serves it from this replica's cache once it was fetched and
+ * found to match.
+ */
+export async function getPluginBundle(id: string, bundleHash?: string): Promise<string | null> {
+  if (!SAFE_ID_RE.test(id)) return null;
+  const useCache = !!bundleHash && isSharedMode();
+  if (useCache) {
+    const cached = getCachedBundle(bundleHash!);
+    if (cached) return cached.code;
+  }
+  let code: string | null;
   try {
-    return await readFile(bundlePath, 'utf-8');
-  } catch {
+    code = await readText('plugins', `${id}.js`);
+  } catch (error) {
+    if (isStoreUnavailable(error)) throw error;
     return null;
   }
+  if (code !== null && useCache && sha256Hex(code) === bundleHash) cacheBundle(bundleHash!, code);
+  return code;
 }
 
 // ─── Theme Operations ────────────────────────────────────────
 
-const themeRegistryPath = () => path.join(getThemesDir(), 'registry.json');
+const emptyThemeRegistry = (): ThemeRegistry => ({ themes: [] });
 
 export async function getThemeRegistry(): Promise<ThemeRegistry> {
-  return readJsonFile<ThemeRegistry>(themeRegistryPath(), { themes: [] });
+  return readRegistry('themes', emptyThemeRegistry);
 }
 
 export async function getTheme(id: string): Promise<ServerTheme | null> {
@@ -376,8 +528,9 @@ export async function getTheme(id: string): Promise<ServerTheme | null> {
   return registry.themes.find(t => t.id === id) || null;
 }
 
-export function saveTheme(theme: ServerTheme, css: string, skin: string | null = null): Promise<void> {
-  return withRegistryLock('themes', () => saveThemeNow(theme, css, skin));
+export async function saveTheme(theme: ServerTheme, css: string, skin: string | null = null): Promise<void> {
+  await withRegistryLock('themes', () => saveThemeNow(theme, css, skin));
+  await announceChange('themes');
 }
 
 async function saveThemeNow(
@@ -389,75 +542,85 @@ async function saveThemeNow(
   if (!SAFE_ID_RE.test(theme.id)) {
     throw new Error('Invalid theme id');
   }
-  const dir = getThemesDir();
-  await ensureDir(dir);
 
   const contentHash = createHash('sha256').update(css).update('\0').update(skin ?? '').digest('hex');
-  const registry = await getThemeRegistry();
-  const idx = registry.themes.findIndex(t => t.id === theme.id);
-  const cssPath = path.join(dir, `${theme.id}.css`);
-  const skinPath = path.join(dir, `${theme.id}.skin.css`);
+  const current = (await getThemeRegistry()).themes.find(t => t.id === theme.id);
+  const cssId = `${theme.id}.css`;
+  const skinId = `${theme.id}.skin.css`;
 
   // Keep the theme being replaced for a rollback (extension-history.ts).
-  if (idx >= 0) {
-    const current = registry.themes[idx];
+  if (current) {
     await forgetVersion('theme', theme.id, theme.version);
     if (current.version !== theme.version || current.contentHash !== contentHash) {
-      const currentCss = await readFile(cssPath, 'utf-8').catch(() => null);
-      const currentSkin = current.hasSkin ? await readFile(skinPath, 'utf-8').catch(() => null) : null;
+      const currentCss = await readText('themes', cssId);
+      const currentSkin = current.hasSkin ? await readText('themes', skinId) : null;
       if (currentCss !== null) await snapshotTheme(current, currentCss, currentSkin);
     }
   }
 
-  // Save CSS file
-  await writeFile(cssPath, css, 'utf-8');
+  // The CSS and skin first, then the registry entry.
+  const docs = themeDocs();
+  await docs.put(cssId, css);
   if (skin) {
-    await writeFile(skinPath, skin, 'utf-8');
+    await docs.put(skinId, skin);
   } else {
-    try { await unlink(skinPath); } catch { /* none stored */ }
+    await removeDoc('themes', skinId);
   }
-  theme = { ...theme, hasSkin: !!skin, contentHash };
+  const next: ServerTheme = { ...theme, hasSkin: !!skin, contentHash };
 
-  // Update registry
-  if (idx >= 0) {
-    registry.themes[idx] = theme;
-  } else {
-    registry.themes.push(theme);
-  }
-  await writeJsonFile(themeRegistryPath(), registry);
+  await updateRegistry('themes', emptyThemeRegistry, (registry) => {
+    const idx = registry.themes.findIndex(t => t.id === theme.id);
+    if (idx >= 0) {
+      registry.themes[idx] = next;
+    } else {
+      registry.themes.push(next);
+    }
+    return true;
+  });
 }
 
-export function updateThemeMeta(id: string, updates: Partial<Pick<ServerTheme, 'enabled' | 'forceEnabled'>>): Promise<ServerTheme | null> {
-  return withRegistryLock('themes', () => updateThemeMetaNow(id, updates));
+export async function updateThemeMeta(id: string, updates: Partial<Pick<ServerTheme, 'enabled' | 'forceEnabled'>>): Promise<ServerTheme | null> {
+  const updated = await withRegistryLock('themes', () => updateThemeMetaNow(id, updates));
+  if (updated) await announceChange('themes');
+  return updated;
 }
 
 async function updateThemeMetaNow(id: string, updates: Partial<Pick<ServerTheme, 'enabled' | 'forceEnabled'>>): Promise<ServerTheme | null> {
   assertWritable('update theme metadata');
-  const registry = await getThemeRegistry();
-  const idx = registry.themes.findIndex(t => t.id === id);
-  if (idx < 0) return null;
-
-  registry.themes[idx] = { ...registry.themes[idx], ...updates, updatedAt: new Date().toISOString() };
-  await writeJsonFile(themeRegistryPath(), registry);
-  return registry.themes[idx];
+  const now = new Date().toISOString();
+  let updated: ServerTheme | null = null;
+  await updateRegistry('themes', emptyThemeRegistry, (registry) => {
+    updated = null;
+    const idx = registry.themes.findIndex(t => t.id === id);
+    if (idx < 0) return false;
+    registry.themes[idx] = { ...registry.themes[idx], ...updates, updatedAt: now };
+    updated = registry.themes[idx];
+    return true;
+  });
+  return updated;
 }
 
-export function deleteTheme(id: string): Promise<boolean> {
-  return withRegistryLock('themes', () => deleteThemeNow(id));
+export async function deleteTheme(id: string): Promise<boolean> {
+  const deleted = await withRegistryLock('themes', () => deleteThemeNow(id));
+  if (deleted) await announceChange('themes');
+  return deleted;
 }
 
 async function deleteThemeNow(id: string): Promise<boolean> {
   assertWritable('delete theme');
-  const registry = await getThemeRegistry();
-  const idx = registry.themes.findIndex(t => t.id === id);
-  if (idx < 0) return false;
+  let found = false;
+  await updateRegistry('themes', emptyThemeRegistry, (registry) => {
+    const idx = registry.themes.findIndex(t => t.id === id);
+    found = idx >= 0;
+    if (!found) return false;
+    registry.themes.splice(idx, 1);
+    return true;
+  });
+  if (!found) return false;
 
-  registry.themes.splice(idx, 1);
-  await writeJsonFile(themeRegistryPath(), registry);
-
-  // Remove CSS and skin files
-  for (const file of [`${id}.css`, `${id}.skin.css`]) {
-    try { await unlink(path.join(getThemesDir(), file)); } catch { /* ok if missing */ }
+  // Remove CSS and skin
+  if (SAFE_ID_RE.test(id)) {
+    for (const file of [`${id}.css`, `${id}.skin.css`]) await removeDoc('themes', file);
   }
   await purgeHistory('theme', id);
 
@@ -466,9 +629,14 @@ async function deleteThemeNow(id: string): Promise<boolean> {
 
 export async function getThemeSkin(id: string): Promise<string | null> {
   if (!SAFE_ID_RE.test(id)) return null;
+  return readThemeAsset(`${id}.skin.css`);
+}
+
+async function readThemeAsset(file: string): Promise<string | null> {
   try {
-    return await readFile(path.join(getThemesDir(), `${id}.skin.css`), 'utf-8');
-  } catch {
+    return await readText('themes', file);
+  } catch (error) {
+    if (isStoreUnavailable(error)) throw error;
     return null;
   }
 }
@@ -489,10 +657,13 @@ export async function themeCSSResolver(): Promise<(id: string) => string | null>
 }
 
 export async function getThemeCSS(id: string): Promise<string | null> {
-  const cssPath = path.join(getThemesDir(), `${id}.css`);
-  try {
-    return await readFile(cssPath, 'utf-8');
-  } catch {
-    return null;
-  }
+  if (!SAFE_ID_RE.test(id)) return null;
+  return readThemeAsset(`${id}.css`);
+}
+
+/** Forget the registries this replica keeps for an outage (tests). */
+export function resetRegistryCacheForTests(): void {
+  const g = registryGlobals();
+  g.lastGood.clear();
+  g.queues.clear();
 }

lib/admin/store-unavailable-response.ts

@@ -0,0 +1,15 @@
+import { NextResponse } from 'next/server';
+import { isStoreUnavailable } from './admin-store';
+
+/**
+ * The 503 a route answers when the shared store (#1066) couldn't be
+ * reached: nothing was written, and the client may try again shortly.
+ * Returns null for any other error.
+ */
+export function storeUnavailableResponse(error: unknown): NextResponse | null {
+  if (!isStoreUnavailable(error)) return null;
+  return NextResponse.json(
+    { error: 'Temporarily unavailable, try again in a moment' },
+    { status: 503, headers: { 'Retry-After': '10' } },
+  );
+}

lib/admin/theme-asset.ts

@@ -1,6 +1,7 @@
 import { NextResponse } from 'next/server';
 import { getThemeCSS, getThemeRegistry, getThemeSkin } from './plugin-registry';
 import { logger } from '@/lib/logger';
+import { storeUnavailableResponse } from './store-unavailable-response';
 
 /**
  * Serve an enabled theme's CSS or skin. Clients put the theme's content hash
@@ -33,6 +34,8 @@ export async function serveThemeAsset(id: string, kind: 'css' | 'skin'): Promise
       },
     });
   } catch (error) {
+    const unavailable = storeUnavailableResponse(error);
+    if (unavailable) return unavailable;
     logger.error('Theme asset serve error', { kind, error: error instanceof Error ? error.message : 'Unknown error' });
     return NextResponse.json({ error: 'Internal server error' }, { status: 500 });
   }