From a0964ce350ad72f417ccea7f41fad2d17f3197f2 Mon Sep 17 00:00:00 2001 From: William Mantly Date: Mon, 10 Aug 2026 20:39:27 -0400 Subject: [PATCH] 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 };