diff --git a/apps/server/src/peer/PeerLinks.test.ts b/apps/server/src/peer/PeerLinks.test.ts index f35e760bea4d7..319a6fb4a701c 100644 --- a/apps/server/src/peer/PeerLinks.test.ts +++ b/apps/server/src/peer/PeerLinks.test.ts @@ -330,6 +330,150 @@ const serveRaw = ( ), ); +it.effect("starts the fallback preference window after route verification", () => + Effect.scoped( + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const probes = yield* Ref.make(0); + const slow = yield* serveRaw(() => + Ref.update(probes, (count) => count + 1).pipe( + Effect.andThen(Deferred.succeed(entered, undefined)), + Effect.andThen(Deferred.await(release)), + Effect.as(HttpServerResponse.jsonUnsafe(descriptorOf("environment-other", "Other"))), + ), + ); + const b = yield* serveB(box); + const a = yield* makeA(laptop); + yield* a.links.link({ + url: b.url, + pairingCode: (yield* b.auth.issuePairingCredential()).credential, + access: "auto", + }); + const urls = JSON.stringify([slow, b.url]); + yield* a.sql`UPDATE peer_environment_links SET urls_json = ${urls}`; + const resolving = yield* a.links.resolve(box.environmentId).pipe(Effect.forkScoped); + yield* Deferred.await(entered); + yield* TestClock.adjust(Duration.seconds(2)); + yield* Deferred.succeed(release, undefined); + expect((yield* Fiber.join(resolving)).url).toBe(b.url); + + yield* TestClock.adjust(Duration.seconds(59)); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(b.url); + expect(yield* Ref.get(probes)).toBe(1); + yield* TestClock.adjust(Duration.seconds(2)); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(b.url); + expect(yield* Ref.get(probes)).toBe(2); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect( + "prefers a verified fallback briefly, rechecks identity, and retries the configured route", + () => + Effect.scoped( + Effect.gen(function* () { + const first = yield* serveApart(descriptorOf("environment-other", "Other")); + const b = yield* serveB(box); + const a = yield* makeA(laptop); + yield* a.links.link({ + url: b.url, + pairingCode: (yield* b.auth.issuePairingCredential()).credential, + access: "auto", + }); + const urls = JSON.stringify([first.url, b.url]); + yield* a.sql`UPDATE peer_environment_links SET urls_json = ${urls}`; + expect((yield* a.links.resolve(box.environmentId)).url).toBe(b.url); + yield* Ref.set(first.probes, 0); + yield* Ref.set(b.probes, 0); + yield* TestClock.adjust(Duration.seconds(30)); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(b.url); + expect(yield* Ref.get(first.probes)).toBe(0); + expect(yield* Ref.get(b.probes)).toBe(1); + + // A recovered preferred address gets a turn, even under repeated demand. + yield* Ref.set(first.descriptor, box); + yield* TestClock.adjust(Duration.seconds(31)); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(first.url); + expect(yield* Ref.get(first.probes)).toBe(1); + + // A remembered address that changes identity is skipped immediately. + yield* Ref.set(first.descriptor, descriptorOf("environment-other", "Other")); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(b.url); + expect(yield* Ref.get(first.bearers)).toEqual([]); + + // When the remembered fallback stops answering as this peer, primary recovery + // clears that preference so later calls do not retry the failed fallback. + yield* Ref.set(first.descriptor, box); + yield* Ref.set(b.descriptor, descriptorOf("environment-other", "Other")); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(first.url); + yield* Ref.set(b.probes, 0); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(first.url); + expect(yield* Ref.get(b.probes)).toBe(0); + + // Relinking restores the new configured order rather than the old preference. + yield* Ref.set(first.descriptor, box); + yield* a.links.link({ + url: first.url, + alternateUrls: [b.url], + pairingCode: (yield* first.auth.issuePairingCredential()).credential, + access: "auto", + }); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(first.url); + yield* a.links.unlink(box.environmentId); + expect((yield* a.links.resolve(box.environmentId).pipe(Effect.flip)).reason).toBe( + "unknown_link", + ); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("resolution ignores an old resolve's preference after relinking", () => + Effect.scoped( + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const probes = yield* Ref.make(0); + const fallback = yield* serveRaw(() => + Ref.update(probes, (count) => count + 1).pipe( + Effect.andThen(Deferred.succeed(entered, undefined)), + Effect.andThen(Deferred.await(release)), + Effect.as(HttpServerResponse.jsonUnsafe(box)), + ), + ); + const first = yield* serveApart(descriptorOf("environment-other", "Other")); + const b = yield* serveB(box); + const a = yield* makeA(laptop); + yield* a.links.link({ + url: b.url, + pairingCode: (yield* b.auth.issuePairingCredential()).credential, + access: "auto", + }); + const urls = JSON.stringify([first.url, fallback]); + yield* a.sql`UPDATE peer_environment_links SET urls_json = ${urls}`; + const resolving = yield* a.links.resolve(box.environmentId).pipe(Effect.forkScoped); + yield* Effect.gen(function* () { + yield* Deferred.await(entered); + yield* Ref.set(first.descriptor, box); + yield* a.links.link({ + url: first.url, + alternateUrls: [fallback], + pairingCode: (yield* first.auth.issuePairingCredential()).credential, + access: "auto", + }); + // The old resolve finishes after relinking cleared its preference. + yield* Deferred.succeed(release, undefined); + expect((yield* Fiber.join(resolving)).url).toBe(fallback); + yield* Ref.set(first.probes, 0); + yield* Ref.set(probes, 0); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(first.url); + expect(yield* Ref.get(first.probes)).toBe(1); + expect(yield* Ref.get(probes)).toBe(0); + }).pipe(Effect.ensuring(Deferred.succeed(release, undefined))); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + it.effect("never follows a peer's redirect with the pairing code", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts index 48f61e6554a17..5be84f857766d 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -195,6 +195,14 @@ const make = Effect.gen(function* () { const environment = yield* ServerEnvironment.ServerEnvironment; const serverSettings = yield* ServerSettings.ServerSettingsService; const isOwnLoopback = yield* LoopbackHost; + // Prefer a recently working fallback, without caching its identity or credentials. + // Retry the configured order after a minute so a recovered direct route wins again. + let preferredRoutes: + | Map< + EnvironmentId, + { readonly url: string; readonly secretName: string; readonly until: number } + > + | undefined; const enabled = serverSettings.getSettings.pipe( Effect.map((settings) => settings.linkedEnvironmentsEnabled), Effect.orElseSucceed(() => false), @@ -559,6 +567,7 @@ const make = Effect.gen(function* () { if (Option.isSome(previous) && previous.value.secret_name !== secretName) { yield* secrets.remove(previous.value.secret_name).pipe(Effect.ignore); } + preferredRoutes?.delete(peer.environmentId); return linked; }); @@ -614,6 +623,7 @@ const make = Effect.gen(function* () { for (const row of deleted) { yield* secrets.remove(row.secret_name).pipe(Effect.catch(failStorage)); } + preferredRoutes?.delete(environmentId); return deleted.length > 0; }, ); @@ -628,7 +638,8 @@ const make = Effect.gen(function* () { }); } const link = yield* toLink(row.value); - if (DateTime.isLessThanOrEqualTo(link.expiresAt, yield* DateTime.now)) { + const now = yield* DateTime.now; + if (DateTime.isLessThanOrEqualTo(link.expiresAt, now)) { return yield* new PeerLinkError({ reason: "expired", message: `The link to ${link.label} expired. Link it again with a new pairing code from ${link.label}.`, @@ -643,9 +654,18 @@ const make = Effect.gen(function* () { } // The token goes only to an address that still answers as this peer, // so one that now answers as another machine never receives it. + const preferred = link.urls.length > 1 ? preferredRoutes?.get(environmentId) : undefined; + const reuse = + preferred !== undefined && + preferred.secretName === row.value.secret_name && + preferred.until > DateTime.toEpochMillis(now) && + link.urls.includes(preferred.url); + const urls = reuse + ? [preferred.url, ...link.urls.filter((candidate) => candidate !== preferred.url)] + : link.urls; let fenced = true; let label = link.label; - const url = yield* Effect.findFirst(link.urls, (candidate) => + const url = yield* Effect.findFirst(urls, (candidate) => answersAs(candidate, environmentId).pipe( Effect.map((found) => { if (Option.isNone(found)) return false; @@ -669,6 +689,24 @@ const make = Effect.gen(function* () { message: `${link.label} runs a T3 Code version that cannot keep a linked agent's limits. Update it, then link again.`, }); } + if (url.value === link.urls[0]) { + preferredRoutes?.delete(environmentId); + } else if (!reuse || preferred.url !== url.value) { + const verifiedAt = yield* DateTime.now; + // Healthy primary routes need no preference state. + const routes = (preferredRoutes ??= new Map()); + routes.delete(environmentId); + // Bound retained state even if links are changed outside this service. + if (routes.size >= 256) { + const oldest = routes.keys().next(); + if (!oldest.done) routes.delete(oldest.value); + } + routes.set(environmentId, { + url: url.value, + secretName: row.value.secret_name, + until: DateTime.toEpochMillis(verifiedAt) + 60_000, + }); + } return { link: yield* withLabel(link, label), url: url.value,