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); } }); 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 };