From a0964ce350ad72f417ccea7f41fad2d17f3197f2 Mon Sep 17 00:00:00 2001 From: William Mantly Date: Mon, 10 Aug 2026 20:39:27 -0400 Subject: [PATCH 1/2] feat(multi-site): route replication traffic over the mesh when available Cross-component routing TODO item: a spoke's resync push now prefers its WG mesh IP (reported via the noInbound/meshIp fields added for the relay automation) over the public endpoint, falling back to the public endpoint if the mesh attempt fails for any reason (tunnel not actually up between these two particular gateways yet, transient failure, etc.) -- a mesh-routing preference must never turn into "spoke stops getting updates." Plain HTTP over the mesh IP, not HTTPS: the WG tunnel is already encrypted, same reasoning already applied to the no-inbound relay terminating at the master. A spoke with no meshIp on file behaves exactly as before (public endpoint only) -- this is additive, not a behavior change for spokes that haven't opted into mesh registration. --- nodejs/tests/site_replicate.test.js | 39 ++++++++++++++++++++ nodejs/utils/site_replicate.js | 57 +++++++++++++++++++++-------- 2 files changed, 81 insertions(+), 15 deletions(-) diff --git a/nodejs/tests/site_replicate.test.js b/nodejs/tests/site_replicate.test.js index d36d5fe..4126bcd 100644 --- a/nodejs/tests/site_replicate.test.js +++ b/nodejs/tests/site_replicate.test.js @@ -55,6 +55,45 @@ describe('site_replicate', () => { expect(JSON.parse(optsA.body).reason).toBe('catalog-changed'); }); + test('prefers the mesh IP over the public endpoint when the spoke reported one', async () => { + SiteSpoke._seed([ + { endpoint: 'https://spoke-a.example.com:8443', pushToken: 'token-a', meshIp: '172.24.5.1' } + ]); + + await siteReplicate.replicateToSpokes('catalog-changed'); + await new Promise((r) => setImmediate(r)); + + expect(mockFetchCalls.length).toBe(1); + expect(mockFetchCalls[0][0]).toBe('http://172.24.5.1:8443/api/site/resync'); + }); + + test('falls back to the public endpoint if the mesh attempt fails', async () => { + SiteSpoke._seed([ + { endpoint: 'https://spoke-a.example.com', pushToken: 'token-a', meshIp: '172.24.5.1' } + ]); + mockFetchImpl = async (url) => { + if (url.startsWith('http://172.24.5.1')) throw new Error('mesh unreachable'); + return { ok: true, status: 200 }; + }; + + await siteReplicate.replicateToSpokes('catalog-changed'); + await new Promise((r) => setImmediate(r)); + + expect(mockFetchCalls.length).toBe(2); + expect(mockFetchCalls[0][0]).toMatch(/^http:\/\/172\.24\.5\.1/); + expect(mockFetchCalls[1][0]).toBe('https://spoke-a.example.com/api/site/resync'); + }); + + test('a spoke with no meshIp only ever tries the public endpoint', async () => { + SiteSpoke._seed([{ endpoint: 'https://spoke-a.example.com', pushToken: 'token-a' }]); + + await siteReplicate.replicateToSpokes('catalog-changed'); + await new Promise((r) => setImmediate(r)); + + expect(mockFetchCalls.length).toBe(1); + expect(mockFetchCalls[0][0]).toBe('https://spoke-a.example.com/api/site/resync'); + }); + test('no known spokes: resolves cleanly, no fetch calls', async () => { await siteReplicate.replicateToSpokes('catalog-changed'); await new Promise((r) => setImmediate(r)); diff --git a/nodejs/utils/site_replicate.js b/nodejs/utils/site_replicate.js index 4421d65..40a507b 100644 --- a/nodejs/utils/site_replicate.js +++ b/nodejs/utils/site_replicate.js @@ -37,21 +37,48 @@ function replicateToSpokes(reason) { })(); } -async function pingOne(spoke, reason) { - const url = String(spoke.endpoint).replace(/\/+$/, '') + '/api/site/resync'; - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(), RESYNC_TIMEOUT_MS); - try { - const resp = await fetch(url, { - method: 'POST', - headers: { Authorization: 'Bearer ' + spoke.pushToken, 'Content-Type': 'application/json' }, - body: JSON.stringify({ reason: reason || 'catalog-changed' }), - signal: controller.signal - }); - if (!resp.ok) throw new Error('status ' + resp.status); - } finally { - clearTimeout(timer); +// Cross-component routing (MULTI_SITE_SPEC.md): if this spoke reported a WG +// mesh IP when it registered (utils/proxy_client.js's no-inbound relay path +// populates the same field), prefer sending the resync push over the mesh +// tunnel instead of the open internet -- plain HTTP is fine here since the +// WG tunnel itself is already encrypted, same reasoning as the no-inbound +// relay terminating at the master. Falls back to the spoke's public endpoint +// if the mesh attempt fails (mesh IP set but that particular tunnel isn't +// actually up yet, or unreachable for any other reason) -- never let a +// mesh-routing preference turn into "spoke never gets updates." +function resyncUrls(spoke) { + const urls = []; + if (spoke.meshIp) { + let port = '3001'; + try { port = new URL(spoke.endpoint).port || port; } catch (_) { /* keep default */ } + urls.push(`http://${spoke.meshIp}:${port}/api/site/resync`); } + urls.push(String(spoke.endpoint).replace(/\/+$/, '') + '/api/site/resync'); + return urls; } -module.exports = { replicateToSpokes }; +async function pingOne(spoke, reason) { + const urls = resyncUrls(spoke); + let lastErr; + for (const url of urls) { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), RESYNC_TIMEOUT_MS); + try { + const resp = await fetch(url, { + method: 'POST', + headers: { Authorization: 'Bearer ' + spoke.pushToken, 'Content-Type': 'application/json' }, + body: JSON.stringify({ reason: reason || 'catalog-changed' }), + signal: controller.signal + }); + if (!resp.ok) throw new Error('status ' + resp.status); + return; // success -- don't try the next (fallback) URL + } catch (err) { + lastErr = err; + } finally { + clearTimeout(timer); + } + } + throw lastErr; +} + +module.exports = { replicateToSpokes, resyncUrls }; From 7bc6f470703b90ea8e1c525faf5e434e1fc89e04 Mon Sep 17 00:00:00 2001 From: William Mantly Date: Mon, 10 Aug 2026 20:52:35 -0400 Subject: [PATCH 2/2] feat(multi-site): forward noInbound/meshIp/publicHost through /api/site/join /api/site/spokes already accepted these and drove proxy_client's relay automation, but nothing on the real operator join path ever passed them through -- a no-inbound spoke calling /join had no way to trigger its own relay route. Forward them into the internal /spokes call and surface the resulting relay note in /join's response. --- nodejs/routes/api_site.js | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/nodejs/routes/api_site.js b/nodejs/routes/api_site.js index 509abe8..63c66bb 100644 --- a/nodejs/routes/api_site.js +++ b/nodejs/routes/api_site.js @@ -370,7 +370,7 @@ async function adoptFromMaster({ masterUrl, joinKey }) { router.post('/join', async (req, res, next) => { try { - const { masterUrl, joinKey, selfUrl } = req.body || {}; + const { masterUrl, joinKey, selfUrl, noInbound, meshIp, publicHost } = req.body || {}; if (!masterUrl || !joinKey) { return res.status(400).json({ status: 'error', message: 'masterUrl and joinKey are required' }); } @@ -404,17 +404,29 @@ router.post('/join', async (req, res, next) => { // snapshot for that spoke, not a hard failure). let replicationPushToken = null; let replicationNote = 'not registered (no selfUrl given)'; + let relayNote = noInbound ? 'not attempted (registration did not run)' : 'not applicable (this spoke has inbound access)'; if (selfUrl) { try { const regResp = await fetch(base + '/api/site/spokes', { method: 'POST', headers: { Authorization: 'Bearer ' + joinKey, 'Content-Type': 'application/json' }, - body: JSON.stringify({ endpoint: selfUrl, siteSlug: exportData.siteSlug || cfg.siteSlug }) + // noInbound/meshIp/publicHost: this spoke has no public IP of its + // own; forwarded so the master can best-effort auto-create a relay + // route on its own theta-proxy (utils/proxy_client.js). Previously + // accepted by /spokes but never actually reachable from here -- + // nothing forwarded them, so the automation existed but no real + // join flow could ever trigger it. + body: JSON.stringify({ + endpoint: selfUrl, + siteSlug: exportData.siteSlug || cfg.siteSlug, + ...(noInbound ? { noInbound: true, meshIp, publicHost } : {}) + }) }); if (regResp.ok) { const regBody = await regResp.json(); replicationPushToken = regBody.pushToken; replicationNote = 'registered for live replication'; + if (regBody.relay) relayNote = regBody.relay.note; } else { replicationNote = 'registration failed: HTTP ' + regResp.status; } @@ -454,7 +466,8 @@ router.post('/join', async (req, res, next) => { resources: { created: imp.created, updated: imp.updated, edges: imp.edgeCount }, ldap: { note: ldapNote }, signingKey: { note: signingKeyNote }, - replication: { note: replicationNote, live: !!replicationPushToken } + replication: { note: replicationNote, live: !!replicationPushToken }, + relay: { note: relayNote } }); } catch (e) { next(e); } });