Merge pull request #199 from theta42/feat-mesh-routing

feat(multi-site): route replication over the mesh when available

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

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

nodejs/utils/site_replicate.js

@@ -37,21 +37,48 @@ function replicateToSpokes(reason) {
 	})();
 }
 
-async function pingOne(spoke, reason) {
+// Cross-component routing (MULTI_SITE_SPEC.md): if this spoke reported a WG
-	const url = String(spoke.endpoint).replace(/\/+$/, '') + '/api/site/resync';
+// mesh IP when it registered (utils/proxy_client.js's no-inbound relay path
-	const controller = new AbortController();
+// populates the same field), prefer sending the resync push over the mesh
-	const timer = setTimeout(() => controller.abort(), RESYNC_TIMEOUT_MS);
+// tunnel instead of the open internet -- plain HTTP is fine here since the
-	try {
+// WG tunnel itself is already encrypted, same reasoning as the no-inbound
-		const resp = await fetch(url, {
+// relay terminating at the master. Falls back to the spoke's public endpoint
-			method: 'POST',
+// if the mesh attempt fails (mesh IP set but that particular tunnel isn't
-			headers: { Authorization: 'Bearer ' + spoke.pushToken, 'Content-Type': 'application/json' },
+// actually up yet, or unreachable for any other reason) -- never let a
-			body: JSON.stringify({ reason: reason || 'catalog-changed' }),
+// mesh-routing preference turn into "spoke never gets updates."
-			signal: controller.signal
+function resyncUrls(spoke) {
-		});
+	const urls = [];
-		if (!resp.ok) throw new Error('status ' + resp.status);
+	if (spoke.meshIp) {
-	} finally {
+		let port = '3001';
-		clearTimeout(timer);
+		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 };