wmantly
d27763e556
The shipped join flow (v2.2.0-v2.3.0) was a one-time snapshot: a spoke's catalog never updated after joining. This adds the two pieces that were explicitly designed but missing: - Live replication: a spoke registers its own endpoint with the master right after joining (POST /api/site/spokes, Bearer join-key), receiving a pushToken. Every successful catalog write on the master now fires a fire-and-forget resync ping (utils/site_replicate.js) at every known spoke, concurrently -- one unreachable spoke never blocks or delays another (wired into the existing write-gate middleware in api_directory_admin.js). The spoke's POST /api/site/resync handler reuses the already-tested export+import path rather than applying a partial diff. - Identical directories: POST /api/site/export now best-effort includes the master's agent-signing key; a spoke adopts it via agent_keys.adopt() on both join and every resync, so every site's sso-manager can validly sign a command for any agent enrolled anywhere -- the accepted tradeoff discussed for this deployment's scale (blast radius for simplicity). New SiteSpoke model tracks registered spokes (endpoint + pushToken); registered it in models/index.js (a real bug the e2e test below caught -- SiteSpoke.list() 500'd with "Cannot read properties of null (reading 'adapter')" until the model was added to initORM's model list). Verified end-to-end against docker-compose.multisite-e2e.yml: mint join key -> join with selfUrl -> write a NEW resource on master post-join -> poll the spoke -> it shows up within a few seconds via the resync push, no manual re-join needed. MULTISITE E2E PASS. Unit tests: nodejs/tests/site_replicate.test.js (concurrent fan-out, one failing spoke doesn't block another, empty-registry and list()-throws edge cases).
91 lines
3.3 KiB
JavaScript
91 lines
3.3 KiB
JavaScript
'use strict';
|
|
|
|
// In-memory stand-in for the SiteSpoke ORM model.
|
|
let spokeStore;
|
|
function makeSpokeMock() {
|
|
spokeStore = [];
|
|
return {
|
|
list: jest.fn(async () => [...spokeStore]),
|
|
_seed(rows) { spokeStore.push(...rows); }
|
|
};
|
|
}
|
|
|
|
let mockFetchCalls = [];
|
|
let mockFetchImpl = async () => ({ ok: true, status: 200 });
|
|
|
|
describe('site_replicate', () => {
|
|
let siteReplicate;
|
|
let SiteSpoke;
|
|
let originalFetch;
|
|
|
|
beforeEach(() => {
|
|
jest.resetModules();
|
|
mockFetchCalls = [];
|
|
mockFetchImpl = async () => ({ ok: true, status: 200 });
|
|
|
|
jest.doMock('../models/site_spoke', () => ({ SiteSpoke: makeSpokeMock() }));
|
|
siteReplicate = require('../utils/site_replicate');
|
|
SiteSpoke = require('../models/site_spoke').SiteSpoke;
|
|
|
|
// site_replicate.js uses the global fetch (Node 18+ built-in), not
|
|
// node-fetch -- stub that directly.
|
|
originalFetch = global.fetch;
|
|
global.fetch = (...args) => { mockFetchCalls.push(args); return mockFetchImpl(...args); };
|
|
});
|
|
|
|
afterEach(() => {
|
|
global.fetch = originalFetch;
|
|
});
|
|
|
|
test('pushes to every known spoke concurrently with its own pushToken', async () => {
|
|
SiteSpoke._seed([
|
|
{ endpoint: 'https://spoke-a.example.com', pushToken: 'token-a' },
|
|
{ endpoint: 'https://spoke-b.example.com', pushToken: 'token-b' }
|
|
]);
|
|
|
|
await siteReplicate.replicateToSpokes('catalog-changed');
|
|
await new Promise((r) => setImmediate(r));
|
|
|
|
expect(mockFetchCalls.length).toBe(2);
|
|
const urls = mockFetchCalls.map((c) => c[0]).sort();
|
|
expect(urls).toEqual(['https://spoke-a.example.com/api/site/resync', 'https://spoke-b.example.com/api/site/resync']);
|
|
|
|
const [, optsA] = mockFetchCalls.find((c) => c[0].includes('spoke-a'));
|
|
expect(optsA.headers.Authorization).toBe('Bearer token-a');
|
|
expect(JSON.parse(optsA.body).reason).toBe('catalog-changed');
|
|
});
|
|
|
|
test('no known spokes: resolves cleanly, no fetch calls', async () => {
|
|
await siteReplicate.replicateToSpokes('catalog-changed');
|
|
await new Promise((r) => setImmediate(r));
|
|
expect(mockFetchCalls.length).toBe(0);
|
|
});
|
|
|
|
test('one spoke failing does not prevent delivery to another', async () => {
|
|
SiteSpoke._seed([
|
|
{ endpoint: 'https://dead-spoke.example.com', pushToken: 'token-dead' },
|
|
{ endpoint: 'https://live-spoke.example.com', pushToken: 'token-live' }
|
|
]);
|
|
mockFetchImpl = async (url) => {
|
|
if (url.includes('dead-spoke')) throw new Error('connection refused');
|
|
return { ok: true, status: 200 };
|
|
};
|
|
|
|
await expect(siteReplicate.replicateToSpokes('event')).resolves.toBeUndefined();
|
|
await new Promise((r) => setImmediate(r));
|
|
expect(mockFetchCalls.length).toBe(2);
|
|
});
|
|
|
|
test('a non-2xx response from a spoke does not throw out of replicateToSpokes', async () => {
|
|
SiteSpoke._seed([{ endpoint: 'https://spoke-a.example.com', pushToken: 'token-a' }]);
|
|
mockFetchImpl = async () => ({ ok: false, status: 500 });
|
|
await expect(siteReplicate.replicateToSpokes('event')).resolves.toBeUndefined();
|
|
});
|
|
|
|
test('SiteSpoke.list() throwing does not propagate to the caller', async () => {
|
|
SiteSpoke.list = jest.fn(async () => { throw new Error('db unavailable'); });
|
|
await expect(siteReplicate.replicateToSpokes('event')).resolves.toBeUndefined();
|
|
expect(mockFetchCalls.length).toBe(0);
|
|
});
|
|
});
|