diff --git a/apps/server/src/peer/PeerLinks.test.ts b/apps/server/src/peer/PeerLinks.test.ts index 3c510c35e6904..04eaa4d2d1e23 100644 --- a/apps/server/src/peer/PeerLinks.test.ts +++ b/apps/server/src/peer/PeerLinks.test.ts @@ -5,9 +5,11 @@ import * as Context from "effect/Context"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; 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"; @@ -342,6 +344,40 @@ 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)); + yield* Effect.yieldNow; + const listed = listing.pollUnsafe(); + expect( + listed === undefined + ? "still waiting on the asleep peer" + : Exit.map(listed, (links) => links.map((link) => link.status)), + ).toEqual(Exit.succeed(["unreachable"])); + }), + ).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 c72c5ba8c81bc..11f0e36a4bf7c 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"; @@ -68,6 +70,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), @@ -89,6 +94,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) { @@ -124,6 +134,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}`; @@ -137,6 +149,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 adbc8cbd1edea..10cf3c09cfb9d 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -49,6 +49,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); /** A usable link: where to reach the peer and the token to send there. */ export interface ResolvedPeerLink { @@ -497,16 +499,18 @@ const make = Effect.gen(function* () { const statusOf = (link: PeerLink, now: DateTime.Utc) => DateTime.isLessThanOrEqualTo(link.expiresAt, now) ? Effect.succeed("expired") - : Effect.findFirst(link.urls, (url) => - describe(url).pipe( - Effect.map( - (found) => Option.isSome(found) && found.value.environmentId === link.environmentId, + : Effect.raceAll( + link.urls.map((url) => + describe(url).pipe( + Effect.filterOrFail( + (found) => Option.isSome(found) && found.value.environmentId === link.environmentId, + ), ), ), ).pipe( - Effect.map((answered): PeerLinkStatus => - Option.isSome(answered) ? "reachable" : "unreachable", - ), + Effect.timeout(LIST_PROBE_TIMEOUT), + Effect.as("reachable"), + Effect.orElseSucceed((): PeerLinkStatus => "unreachable"), ); const list: PeerLinks["Service"]["list"] = Effect.gen(function* () {