diff --git a/apps/server/src/peer/PeerLinks.test.ts b/apps/server/src/peer/PeerLinks.test.ts index 2ff34859c8a97..7a653365ea211 100644 --- a/apps/server/src/peer/PeerLinks.test.ts +++ b/apps/server/src/peer/PeerLinks.test.ts @@ -12,6 +12,7 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as HttpRouter from "effect/http/HttpRouter"; @@ -389,6 +390,62 @@ it.effect("never sends a link's token to an address that stops answering as that ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); +it.effect("lists a peer asleep at every address as unreachable within one probe budget", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + // Another address of the same machine, asleep too. + const wake = yield* Deferred.make(); + const elsewhere = yield* serveRaw(() => + Deferred.await(wake).pipe(Effect.as(HttpServerResponse.empty({ status: 503 }))), + ); + yield* Effect.addFinalizer(() => Deferred.succeed(wake, undefined)); + const a = yield* makeA(laptop); + const pairing = yield* b.auth.issuePairingCredential(); + yield* a.links.link({ + url: b.url, + alternateUrls: [elsewhere], + pairingCode: pairing.credential, + access: "auto", + }); + + yield* Ref.set(b.asleep, true); + const listing = yield* a.links.list.pipe(Effect.forkScoped); + yield* Queue.take(b.unanswered); + yield* TestClock.adjust(Duration.seconds(3)); + const listed = yield* Fiber.join(listing); + expect(listed.map((link) => link.status)).toEqual(["unreachable"]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("lists a peer as reachable at another address while its first one sleeps", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const described = yield* HttpClient.get(`${b.url}/.well-known/t3/environment`).pipe( + Effect.flatMap((response) => response.text), + Effect.provide(FetchHttpClient.layer), + ); + // Another address of the same machine that answers. + const elsewhere = yield* serveRaw(() => + Effect.succeed(HttpServerResponse.text(described, { contentType: "application/json" })), + ); + const a = yield* makeA(laptop); + const pairing = yield* b.auth.issuePairingCredential(); + yield* a.links.link({ + url: b.url, + alternateUrls: [elsewhere], + pairingCode: pairing.credential, + access: "auto", + }); + + yield* Ref.set(b.asleep, true); + expect((yield* a.links.list).map((link) => link.status)).toEqual(["reachable"]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + it.effect("cancelling a descriptor probe aborts its request and leaves later resolves fresh", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/peer/PeerLinks.testkit.ts b/apps/server/src/peer/PeerLinks.testkit.ts index 7ba3d144ef104..a9e15472eec93 100644 --- a/apps/server/src/peer/PeerLinks.testkit.ts +++ b/apps/server/src/peer/PeerLinks.testkit.ts @@ -7,8 +7,10 @@ import { type ExecutionEnvironmentDescriptor, } from "@t3tools/contracts"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as FetchHttpClient from "effect/http/FetchHttpClient"; import * as HttpRouter from "effect/http/HttpRouter"; @@ -70,6 +72,9 @@ export const servePeer = ( const descriptor = yield* Ref.make(initial); const bearers = yield* Ref.make>([]); const probes = yield* Ref.make(0); + const asleep = yield* Ref.make(false); + const unanswered = yield* Queue.unbounded(); + const wake = yield* Deferred.make(); const layerEnvironment = Layer.succeed(ServerEnvironment.ServerEnvironment, { getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)), getDescriptor: Ref.get(descriptor), @@ -91,6 +96,11 @@ export const servePeer = ( const request = yield* HttpServerRequest.HttpServerRequest; if (request.url.startsWith("/.well-known/t3/environment")) { yield* Ref.update(probes, (count) => count + 1); + if (yield* Ref.get(asleep)) { + yield* Queue.offer(unanswered, undefined); + yield* Deferred.await(wake); + return yield* Effect.die("the peer was closed while asleep"); + } } const authorization = request.headers.authorization; if (authorization !== undefined) { @@ -141,6 +151,8 @@ export const servePeer = ( Layer.fresh, Layer.build, ); + // An unanswered request would hold the server open past the test. + yield* Effect.addFinalizer(() => Deferred.succeed(wake, undefined)); const address = (yield* HttpServer.HttpServer).address; if (address._tag === "UnixPathAddress") return yield* Effect.die("expected a TCP address"); const url = `http://127.0.0.1:${address.port}`; @@ -154,6 +166,10 @@ export const servePeer = ( bearers, /** How often its descriptor was asked for, as a probe of whether it answers. */ probes, + /** While set, its descriptor never answers, as a machine that is asleep. */ + asleep, + /** One entry per descriptor request it left unanswered while asleep. */ + unanswered, /** The sessions linking created there, which its Connections lists. */ linkedSessions: auth .listSessions() diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts index f5796d7a6994d..82eaf945af710 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -51,6 +51,8 @@ const LINK_REDIRECT_URI = "http://127.0.0.1/t3code-peer-link"; */ const probeTimeout = (origin: string) => isLocalLoopbackHost(new URL(origin).hostname) ? Duration.seconds(3) : Duration.seconds(10); +/** How long a listing waits on a link's addresses; a real call still gives a waking peer `probeTimeout`. */ +const LIST_PROBE_TIMEOUT = Duration.seconds(3); /** Node's TLS verification code in a failed request's cause chain, if any. */ export const certificateErrorCode = (error: unknown) => { @@ -567,18 +569,15 @@ const make = Effect.gen(function* () { now: DateTime.Utc, ): Effect.fn.Return { if (DateTime.isLessThanOrEqualTo(link.expiresAt, now)) return { ...link, status: "expired" }; - let label = link.label; - const answered = yield* Effect.findFirst(link.urls, (url) => - answersAs(url, link.environmentId).pipe( - Effect.map((found) => { - if (Option.isNone(found)) return false; - label = found.value.label; - return true; - }), + // A race over no addresses never settles. + if (link.urls.length === 0) return { ...link, status: "unreachable" }; + const answered = yield* Effect.raceAll( + link.urls.map((url) => + answersAs(url, link.environmentId).pipe(Effect.flatMap(Effect.fromOption)), ), - ); + ).pipe(Effect.timeoutOption(LIST_PROBE_TIMEOUT), Effect.orElseSucceed(Option.none)); if (Option.isNone(answered)) return { ...link, status: "unreachable" }; - return { ...(yield* withLabel(link, label)), status: "reachable" }; + return { ...(yield* withLabel(link, answered.value.label)), status: "reachable" }; }); const list: PeerLinks["Service"]["list"] = Effect.gen(function* () {