History

Use Option and Duration in Effect server paths

Cursor Agent committed 4 months ago

Commit 66d9fcf1 parent 82a9bcc5 files · +337 −280Browse files

Unified · Split

5 files
  1. apps/server/integration/OrchestrationEngineHarness.integration.ts +52 −64
  2. apps/server/src/cloud/ManagedEndpointRuntime.test.ts +146 −93
  3. apps/server/src/cloud/ManagedEndpointRuntime.ts +132 −117
  4. apps/server/src/cloud/http.ts +2 −2
  5. apps/server/src/server.test.ts +5 −4
apps/server/integration/OrchestrationEngineHarness.integration.ts+52−64
@@ -9,6 +9,7 @@ import { · 8 unmodified lines
9 type OrchestrationEvent,9 type OrchestrationEvent,
10 type OrchestrationThread,10 type OrchestrationThread,
11} from "@t3tools/contracts";11} from "@t3tools/contracts";
12import * as Duration from "effect/Duration";
12import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
13import * as Exit from "effect/Exit";14import * as Exit from "effect/Exit";
14import * as FileSystem from "effect/FileSystem";15import * as FileSystem from "effect/FileSystem";
@@ -120,39 +121,45 @@ class WaitForTimeoutError extends Schema.TaggedErrorClass<WaitForTimeoutError>() · 105 unmodified lines
120 },121 },
121) {}122) {}
122 123
124class WaitForRetrySignal extends Schema.TaggedErrorClass<WaitForRetrySignal>()(
125 "WaitForRetrySignal",
126 {
127 description: Schema.String,
128 },
129) {}
130
131const isWaitForRetrySignal = Schema.is(WaitForRetrySignal);
132const WAIT_FOR_RETRY_INTERVAL = Duration.millis(10);
133const WAIT_FOR_DEFAULT_TIMEOUT = Duration.seconds(40);
134
123function waitFor<A, E>(135function waitFor<A, E>(
124 read: Effect.Effect<A, E>,136 read: Effect.Effect<A, E>,
125 predicate: (value: A) => boolean,137 predicate: (value: A) => boolean,
126 description: string,138 description: string,
127 timeoutMs?: number,139 timeout?: Duration.Input,
128): Effect.Effect<A, never>;140): Effect.Effect<A, never>;
129function waitFor<A, B extends A, E>(141function waitFor<A, B extends A, E>(
130 read: Effect.Effect<A, E>,142 read: Effect.Effect<A, E>,
131 predicate: (value: A) => value is B,143 predicate: (value: A) => value is B,
132 description: string,144 description: string,
133 timeoutMs?: number,145 timeout?: Duration.Input,
134): Effect.Effect<B, never>;146): Effect.Effect<B, never>;
135function waitFor<A, E>(147function waitFor<A, E>(
136 read: Effect.Effect<A, E>,148 read: Effect.Effect<A, E>,
137 predicate: (value: A) => boolean,149 predicate: (value: A) => boolean,
138 description: string,150 description: string,
139 timeoutMs = 40_000,151 timeout: Duration.Input = WAIT_FOR_DEFAULT_TIMEOUT,
140): Effect.Effect<A, never> {152): Effect.Effect<A, never> {
141 const RETRY_SIGNAL = "wait_for_retry";
142 const retryIntervalMs = 10;
143 const maxRetries = Math.max(0, Math.floor(timeoutMs / retryIntervalMs));
144 const retrySchedule = Schedule.spaced(`${retryIntervalMs} millis`);
145
146 return read.pipe(153 return read.pipe(
147 Effect.filterOrFail(predicate, () => RETRY_SIGNAL),154 Effect.filterOrFail(predicate, () => new WaitForRetrySignal({ description })),
148 Effect.retry({155 Effect.retry({
149 schedule: retrySchedule,156 schedule: Schedule.spaced(WAIT_FOR_RETRY_INTERVAL),
150 times: maxRetries,157 while: isWaitForRetrySignal,
151 while: (error) => error === RETRY_SIGNAL,158 }),
159 Effect.timeoutOrElse({
160 duration: timeout,
161 orElse: () => Effect.fail(new WaitForTimeoutError({ description })),
152 }),162 }),
153 Effect.mapError((error) =>
154 error === RETRY_SIGNAL ? new WaitForTimeoutError({ description }) : error,
155 ),
156 Effect.orDie,163 Effect.orDie,
157 );164 );
158}165}
@@ -185,11 +192,11 @@ export interface OrchestrationIntegrationHarness { · 26 unmodified lines
185 readonly waitForThread: (192 readonly waitForThread: (
186 threadId: string,193 threadId: string,
187 predicate: (thread: OrchestrationThread) => boolean,194 predicate: (thread: OrchestrationThread) => boolean,
188 timeoutMs?: number,195 timeout?: Duration.Input,
189 ) => Effect.Effect<OrchestrationThread, never>;196 ) => Effect.Effect<OrchestrationThread, never>;
190 readonly waitForDomainEvent: (197 readonly waitForDomainEvent: (
191 predicate: (event: OrchestrationEvent) => boolean,198 predicate: (event: OrchestrationEvent) => boolean,
192 timeoutMs?: number,199 timeout?: Duration.Input,
193 ) => Effect.Effect<ReadonlyArray<OrchestrationEvent>, never>;200 ) => Effect.Effect<ReadonlyArray<OrchestrationEvent>, never>;
194 readonly waitForPendingApproval: (201 readonly waitForPendingApproval: (
195 requestId: string,202 requestId: string,
@@ -198,7 +205,7 @@ export interface OrchestrationIntegrationHarness { · 2 unmodified lines
198 readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;205 readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;
199 readonly resolvedAt: string | null;206 readonly resolvedAt: string | null;
200 }) => boolean,207 }) => boolean,
201 timeoutMs?: number,208 timeout?: Duration.Input,
202 ) => Effect.Effect<209 ) => Effect.Effect<
203 {210 {
204 readonly status: "pending" | "resolved";211 readonly status: "pending" | "resolved";
@@ -210,11 +217,11 @@ export interface OrchestrationIntegrationHarness { · 5 unmodified lines
210 readonly waitForReceipt: {217 readonly waitForReceipt: {
211 (218 (
212 predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,219 predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,
213 timeoutMs?: number,220 timeout?: Duration.Input,
214 ): Effect.Effect<OrchestrationRuntimeReceipt, never>;221 ): Effect.Effect<OrchestrationRuntimeReceipt, never>;
215 <Receipt extends OrchestrationRuntimeReceipt>(222 <Receipt extends OrchestrationRuntimeReceipt>(
216 predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt,223 predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt,
217 timeoutMs?: number,224 timeout?: Duration.Input,
218 ): Effect.Effect<Receipt, never>;225 ): Effect.Effect<Receipt, never>;
219 };226 };
220 readonly dispose: Effect.Effect<void, never>;227 readonly dispose: Effect.Effect<void, never>;
@@ -420,29 +427,29 @@ export const makeOrchestrationIntegrationHarness = ( · 199 unmodified lines
420 yield* Stream.runForEach(runtimeReceiptBus.streamEventsForTest, (receipt) =>427 yield* Stream.runForEach(runtimeReceiptBus.streamEventsForTest, (receipt) =>
421 Ref.update(receiptHistory, (history) => [...history, receipt]).pipe(Effect.asVoid),428 Ref.update(receiptHistory, (history) => [...history, receipt]).pipe(Effect.asVoid),
422 ).pipe(Effect.forkIn(scope));429 ).pipe(Effect.forkIn(scope));
423 yield* Effect.sleep(10);430 yield* Effect.sleep(Duration.millis(10));
424 431
425 const waitForThread: OrchestrationIntegrationHarness["waitForThread"] = (432 const waitForThread: OrchestrationIntegrationHarness["waitForThread"] = (
426 threadId,433 threadId,
427 predicate,434 predicate,
428 timeoutMs,435 timeout,
429 ) =>436 ) =>
430 waitFor(437 waitFor(
431 snapshotQuery438 snapshotQuery
432 .getSnapshot()439 .getSnapshot()
433 .pipe(440 .pipe(
434 Effect.map(441 Effect.map((snapshot) =>
435 (snapshot) => snapshot.threads.find((thread) => thread.id === threadId) ?? null,442 Option.fromUndefinedOr(snapshot.threads.find((thread) => thread.id === threadId)),
436 ),443 ),
437 ),444 ),
438 (thread): thread is OrchestrationThread => thread !== null && predicate(thread),445 (thread) => Option.isSome(thread) && predicate(thread.value),
439 `projected thread '${threadId}'`,446 `projected thread '${threadId}'`,
440 timeoutMs,447 timeout,
441 ) as Effect.Effect<OrchestrationThread, never>;448 ).pipe(Effect.map(Option.getOrThrow));
442 449
443 const waitForDomainEvent: OrchestrationIntegrationHarness["waitForDomainEvent"] = (450 const waitForDomainEvent: OrchestrationIntegrationHarness["waitForDomainEvent"] = (
444 predicate,451 predicate,
445 timeoutMs,452 timeout,
446 ) =>453 ) =>
447 waitFor(454 waitFor(
448 Stream.runCollect(engine.readEvents(0)).pipe(455 Stream.runCollect(engine.readEvents(0)).pipe(
@@ -450,68 +457,49 @@ export const makeOrchestrationIntegrationHarness = ( · 1 unmodified line
450 ),457 ),
451 (events) => events.some(predicate),458 (events) => events.some(predicate),
452 "domain event",459 "domain event",
453 timeoutMs,460 timeout,
454 );461 );
455 462
456 const waitForPendingApproval: OrchestrationIntegrationHarness["waitForPendingApproval"] = (463 const waitForPendingApproval: OrchestrationIntegrationHarness["waitForPendingApproval"] = (
457 requestId,464 requestId,
458 predicate,465 predicate,
459 timeoutMs,466 timeout,
460 ) =>467 ) =>
461 waitFor(468 waitFor(
462 pendingApprovalRepository469 pendingApprovalRepository
463 .getByRequestId({ requestId: ApprovalRequestId.make(requestId) })470 .getByRequestId({ requestId: ApprovalRequestId.make(requestId) })
464 .pipe(471 .pipe(
465 Effect.map((row) =>472 Effect.map(
466 Option.match(row, {473 Option.map((value) => ({
467 onNone: () => null,474 status: value.status,
468 onSome: (value) => ({475 decision: value.decision,
469 status: value.status,476 resolvedAt: value.resolvedAt,
470 decision: value.decision,477 })),
471 resolvedAt: value.resolvedAt,
472 }),
473 }),
474 ),478 ),
475 ),479 ),
476 (480 (row) => Option.isSome(row) && predicate(row.value),
477 row,
478 ): row is {
479 readonly status: "pending" | "resolved";
480 readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;
481 readonly resolvedAt: string | null;
482 } => row !== null && predicate(row),
483 `pending approval '${requestId}'`,481 `pending approval '${requestId}'`,
484 timeoutMs,482 timeout,
485 ) as Effect.Effect<483 ).pipe(Effect.map(Option.getOrThrow));
486 {
487 readonly status: "pending" | "resolved";
488 readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;
489 readonly resolvedAt: string | null;
490 },
491 never
492 >;
493 484
494 function waitForReceipt(485 function waitForReceipt(
495 predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,486 predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,
496 timeoutMs?: number,487 timeout?: Duration.Input,
497 ): Effect.Effect<OrchestrationRuntimeReceipt, never>;488 ): Effect.Effect<OrchestrationRuntimeReceipt, never>;
498 function waitForReceipt<Receipt extends OrchestrationRuntimeReceipt>(489 function waitForReceipt<Receipt extends OrchestrationRuntimeReceipt>(
499 predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt,490 predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt,
500 timeoutMs?: number,491 timeout?: Duration.Input,
501 ): Effect.Effect<Receipt, never>;492 ): Effect.Effect<Receipt, never>;
502 function waitForReceipt(493 function waitForReceipt(
503 predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,494 predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,
504 timeoutMs?: number,495 timeout?: Duration.Input,
505 ) {496 ) {
506 const readMatchingReceipt = Ref.get(receiptHistory).pipe(497 const readMatchingReceipt = Ref.get(receiptHistory).pipe(
507 Effect.map((history) => history.find(predicate)),498 Effect.map((history) => Option.fromUndefinedOr(history.find(predicate))),
508 );499 );
509 500
510 return waitFor(501 return waitFor(readMatchingReceipt, Option.isSome, "runtime receipt", timeout).pipe(
511 readMatchingReceipt,502 Effect.map(Option.getOrThrow),
512 (receipt): receipt is OrchestrationRuntimeReceipt => receipt !== undefined,
513 "runtime receipt",
514 timeoutMs,
515 );503 );
516 }504 }
517 505
apps/server/src/cloud/ManagedEndpointRuntime.test.ts+146−93
@@ -1,4 +1,4 @@
1import { describe, expect, it } from "@effect/vitest";1import { assert, describe, it } from "@effect/vitest";
2import { vi } from "vite-plus/test";2import { vi } from "vite-plus/test";
3import * as Deferred from "effect/Deferred";3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";4import * as Effect from "effect/Effect";
@@ -81,21 +81,24 @@ function makeHandle(input: { · 76 unmodified lines
81 81
82describe("CloudManagedEndpointRuntime", () => {82describe("CloudManagedEndpointRuntime", () => {
83 it("classifies Cloudflare connection and warning output", () => {83 it("classifies Cloudflare connection and warning output", () => {
84 expect(84 assert.equal(
85 ManagedEndpointRuntime.classifyRelayClientOutput(85 ManagedEndpointRuntime.classifyRelayClientOutput(
86 "2026-06-17T02:00:00Z INF Registered tunnel connection connIndex=0",86 "2026-06-17T02:00:00Z INF Registered tunnel connection connIndex=0",
87 ),87 ),
88 ).toBe("connected");88 "connected",
89 expect(89 );
90 assert.equal(
90 ManagedEndpointRuntime.classifyRelayClientOutput(91 ManagedEndpointRuntime.classifyRelayClientOutput(
91 "2026-06-17T02:00:00Z ERR Failed to serve tunnel connection",92 "2026-06-17T02:00:00Z ERR Failed to serve tunnel connection",
92 ),93 ),
93 ).toBe("warning");94 "warning",
94 expect(95 );
96 assert.equal(
95 ManagedEndpointRuntime.classifyRelayClientOutput(97 ManagedEndpointRuntime.classifyRelayClientOutput(
96 "2026-06-17T02:00:00Z INF Starting metrics server",98 "2026-06-17T02:00:00Z INF Starting metrics server",
97 ),99 ),
98 ).toBe("debug");100 "debug",
101 );
99 });102 });
100 103
101 it.effect("starts, deduplicates, rotates, and stops the Cloudflare connector", () =>104 it.effect("starts, deduplicates, rotates, and stops the Cloudflare connector", () =>
@@ -123,41 +126,65 @@ describe("CloudManagedEndpointRuntime", () => { · 21 unmodified lines
123 );126 );
124 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);127 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
125 128
126 yield* runtime.applyConfig({129 yield* runtime.applyConfig(
127 providerKind: "cloudflare_tunnel",130 Option.some({
128 connectorToken: "token-1",131 providerKind: "cloudflare_tunnel",
129 tunnelId: "tunnel-1",132 connectorToken: "token-1",
130 tunnelName: "t3-code-env-1",133 tunnelId: "tunnel-1",
131 });134 tunnelName: "t3-code-env-1",
132 yield* runtime.applyConfig({135 }),
133 providerKind: "cloudflare_tunnel",136 );
134 connectorToken: "token-1",137 yield* runtime.applyConfig(
135 tunnelId: "tunnel-1",138 Option.some({
136 tunnelName: "t3-code-env-1",139 providerKind: "cloudflare_tunnel",
137 });140 connectorToken: "token-1",
138 yield* runtime.applyConfig({141 tunnelId: "tunnel-1",
139 providerKind: "cloudflare_tunnel",142 tunnelName: "t3-code-env-1",
140 connectorToken: "token-2",143 }),
141 tunnelId: "tunnel-1",144 );
142 tunnelName: "t3-code-env-1",145 yield* runtime.applyConfig(
143 });146 Option.some({
144 const stopped = yield* runtime.applyConfig(null);147 providerKind: "cloudflare_tunnel",
148 connectorToken: "token-2",
149 tunnelId: "tunnel-1",
150 tunnelName: "t3-code-env-1",
151 }),
152 );
153 const stopped = yield* runtime.applyConfig(Option.none());
145 154
146 expect(spawned.map((command) => command.command)).toEqual(["cloudflared", "cloudflared"]);155 assert.deepEqual(
147 expect(spawned.map((command) => command.args)).toEqual([156 spawned.map((command) => command.command),
148 ["tunnel", "run"],157 ["cloudflared", "cloudflared"],
149 ["tunnel", "run"],158 );
150 ]);159 assert.deepEqual(
151 expect(spawned.map((command) => command.options.env?.TUNNEL_TOKEN)).toEqual([160 spawned.map((command) => command.args),
152 "token-1",161 [
153 "token-2",162 ["tunnel", "run"],
154 ]);163 ["tunnel", "run"],
155 expect(spawned.map((command) => command.options.stdout)).toEqual(["pipe", "pipe"]);164 ],
156 expect(spawned.map((command) => command.options.stderr)).toEqual(["pipe", "pipe"]);165 );
157 expect(spawned.map((command) => command.options.detached)).toEqual([false, false]);166 assert.deepEqual(
158 expect(spawned.map((command) => command.options.shell)).toEqual([false, false]);167 spawned.map((command) => command.options.env?.TUNNEL_TOKEN),
159 expect(killed).toEqual([100, 101]);168 ["token-1", "token-2"],
160 expect(stopped).toEqual({ status: "disabled" });169 );
170 assert.deepEqual(
171 spawned.map((command) => command.options.stdout),
172 ["pipe", "pipe"],
173 );
174 assert.deepEqual(
175 spawned.map((command) => command.options.stderr),
176 ["pipe", "pipe"],
177 );
178 assert.deepEqual(
179 spawned.map((command) => command.options.detached),
180 [false, false],
181 );
182 assert.deepEqual(
183 spawned.map((command) => command.options.shell),
184 [false, false],
185 );
186 assert.deepEqual(killed, [100, 101]);
187 assert.deepEqual(stopped, { status: "disabled" });
161 }),188 }),
162 );189 );
163 190
@@ -178,18 +205,22 @@ describe("CloudManagedEndpointRuntime", () => { · 14 unmodified lines
178 );205 );
179 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);206 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
180 207
181 const started = yield* runtime.applyConfig({208 const started = yield* runtime.applyConfig(
182 providerKind: "cloudflare_tunnel",209 Option.some({
183 connectorToken: "token",210 providerKind: "cloudflare_tunnel",
184 });211 connectorToken: "token",
185 const unsupported = yield* runtime.applyConfig({212 }),
186 providerKind: "manual",213 );
187 connectorToken: "manual-token",214 const unsupported = yield* runtime.applyConfig(
188 });215 Option.some({
216 providerKind: "manual",
217 connectorToken: "manual-token",
218 }),
219 );
189 220
190 expect(started.status).toBe("running");221 assert.equal(started.status, "running");
191 expect(unsupported).toEqual({ status: "unsupported", providerKind: "manual" });222 assert.deepEqual(unsupported, { status: "unsupported", providerKind: "manual" });
192 expect(killed).toEqual([200]);223 assert.deepEqual(killed, [200]);
193 }),224 }),
194 );225 );
195 226
@@ -220,14 +251,20 @@ describe("CloudManagedEndpointRuntime", () => { · 24 unmodified lines
220 tunnelId: "tunnel-1",251 tunnelId: "tunnel-1",
221 };252 };
222 253
223 const first = yield* runtime.applyConfig(config);254 const first = yield* runtime.applyConfig(Option.some(config));
224 firstRunning = false;255 firstRunning = false;
225 const second = yield* runtime.applyConfig(config);256 const second = yield* runtime.applyConfig(Option.some(config));
226 257
227 expect(first).toMatchObject({ status: "running", pid: 300 });258 if (first.status !== "running") {
228 expect(second).toMatchObject({ status: "running", pid: 301 });259 assert.fail(`Expected first connector to be running, got ${first.status}`);
229 expect(spawned).toEqual([300, 301]);260 }
230 expect(killed).toEqual([300]);261 assert.equal(first.pid, 300);
262 if (second.status !== "running") {
263 assert.fail(`Expected second connector to be running, got ${second.status}`);
264 }
265 assert.equal(second.pid, 301);
266 assert.deepEqual(spawned, [300, 301]);
267 assert.deepEqual(killed, [300]);
231 }),268 }),
232 );269 );
233 270
@@ -260,17 +297,22 @@ describe("CloudManagedEndpointRuntime", () => { · 26 unmodified lines
260 );297 );
261 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);298 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
262 299
263 const started = yield* runtime.applyConfig({300 const started = yield* runtime.applyConfig(
264 providerKind: "cloudflare_tunnel",301 Option.some({
265 connectorToken: "token",302 providerKind: "cloudflare_tunnel",
266 tunnelId: "tunnel-1",303 connectorToken: "token",
267 });304 tunnelId: "tunnel-1",
305 }),
306 );
268 yield* Deferred.succeed(firstExit, ChildProcessSpawner.ExitCode(1));307 yield* Deferred.succeed(firstExit, ChildProcessSpawner.ExitCode(1));
269 yield* Deferred.await(secondSpawned);308 yield* Deferred.await(secondSpawned);
270 309
271 expect(started).toMatchObject({ status: "running", pid: 400 });310 if (started.status !== "running") {
272 expect(spawned).toEqual([400, 401]);311 assert.fail(`Expected connector to be running, got ${started.status}`);
273 expect(killed).toEqual([400]);312 }
313 assert.equal(started.pid, 400);
314 assert.deepEqual(spawned, [400, 401]);
315 assert.deepEqual(killed, [400]);
274 }),316 }),
275 );317 );
276 318
@@ -301,26 +343,33 @@ describe("CloudManagedEndpointRuntime", () => { · 24 unmodified lines
301 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);343 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
302 344
303 const first = yield* runtime345 const first = yield* runtime
304 .applyConfig({346 .applyConfig(
305 providerKind: "cloudflare_tunnel",347 Option.some({
306 connectorToken: "token-1",348 providerKind: "cloudflare_tunnel",
307 })349 connectorToken: "token-1",
350 }),
351 )
308 .pipe(Effect.forkChild);352 .pipe(Effect.forkChild);
309 yield* Deferred.await(firstSpawnEntered);353 yield* Deferred.await(firstSpawnEntered);
310 const second = yield* runtime354 const second = yield* runtime
311 .applyConfig({355 .applyConfig(
312 providerKind: "cloudflare_tunnel",356 Option.some({
313 connectorToken: "token-2",357 providerKind: "cloudflare_tunnel",
314 })358 connectorToken: "token-2",
359 }),
360 )
315 .pipe(Effect.forkChild);361 .pipe(Effect.forkChild);
316 yield* Deferred.succeed(releaseFirstSpawn, undefined);362 yield* Deferred.succeed(releaseFirstSpawn, undefined);
317 363
318 yield* Fiber.join(first);364 yield* Fiber.join(first);
319 const status = yield* Fiber.join(second);365 const status = yield* Fiber.join(second);
320 366
321 expect(status).toMatchObject({ status: "running", pid: 501 });367 if (status.status !== "running") {
322 expect(spawned).toEqual([500, 501]);368 assert.fail(`Expected connector to be running, got ${status.status}`);
323 expect(killed).toEqual([500]);369 }
370 assert.equal(status.pid, 501);
371 assert.deepEqual(spawned, [500, 501]);
372 assert.deepEqual(killed, [500]);
324 }),373 }),
325 );374 );
326 375
@@ -338,17 +387,19 @@ describe("CloudManagedEndpointRuntime", () => { · 11 unmodified lines
338 );387 );
339 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);388 const runtime = yield* buildCloudManagedEndpointRuntime(spawner);
340 389
341 const status = yield* runtime.applyConfig({390 const status = yield* runtime.applyConfig(
342 providerKind: "cloudflare_tunnel",391 Option.some({
343 connectorToken: "token",392 providerKind: "cloudflare_tunnel",
344 tunnelId: "tunnel-1",393 connectorToken: "token",
345 });394 tunnelId: "tunnel-1",
395 }),
396 );
346 397
347 expect(status).toMatchObject({398 if (status.status !== "failed") {
348 status: "failed",399 assert.fail(`Expected connector spawn to fail, got ${status.status}`);
349 providerKind: "cloudflare_tunnel",400 }
350 tunnelId: "tunnel-1",401 assert.equal(status.providerKind, "cloudflare_tunnel");
351 });402 assert.equal(status.tunnelId, "tunnel-1");
352 }),403 }),
353 );404 );
354 405
@@ -371,17 +422,19 @@ describe("CloudManagedEndpointRuntime", () => { · 16 unmodified lines
371 ),422 ),
372 );423 );
373 424
374 const status = yield* runtime.applyConfig({425 const status = yield* runtime.applyConfig(
375 providerKind: "cloudflare_tunnel",426 Option.some({
376 connectorToken: "token",427 providerKind: "cloudflare_tunnel",
377 });428 connectorToken: "token",
429 }),
430 );
378 431
379 expect(status).toEqual({432 assert.deepEqual(status, {
380 status: "failed",433 status: "failed",
381 providerKind: "cloudflare_tunnel",434 providerKind: "cloudflare_tunnel",
382 reason: "The relay client is not installed.",435 reason: "The relay client is not installed.",
383 });436 });
384 expect(spawn).not.toHaveBeenCalled();437 assert.equal(spawn.mock.calls.length, 0);
385 }),438 }),
386 );439 );
387});440});
apps/server/src/cloud/ManagedEndpointRuntime.ts+132−117
@@ -24,9 +24,9 @@ const readRuntimeConfig = Effect.gen(function* () { · 23 unmodified lines
24 const secrets = yield* ServerSecretStore.ServerSecretStore;24 const secrets = yield* ServerSecretStore.ServerSecretStore;
25 const bytes = yield* secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG);25 const bytes = yield* secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG);
26 if (Option.isNone(bytes)) {26 if (Option.isNone(bytes)) {
27 return null;27 return Option.none<RelayManagedEndpointRuntimeConfig>();
28 }28 }
29 return Option.getOrNull(decodeRuntimeConfig(bytesToString(bytes.value)));29 return decodeRuntimeConfig(bytesToString(bytes.value));
30});30});
31 31
32export type CloudManagedEndpointRuntimeStatus =32export type CloudManagedEndpointRuntimeStatus =
@@ -56,7 +56,7 @@ export class CloudManagedEndpointRuntime extends Context.Service< · 23 unmodified lines
56 CloudManagedEndpointRuntime,56 CloudManagedEndpointRuntime,
57 {57 {
58 readonly applyConfig: (58 readonly applyConfig: (
59 config: RelayManagedEndpointRuntimeConfig | null,59 config: Option.Option<RelayManagedEndpointRuntimeConfig>,
60 ) => Effect.Effect<CloudManagedEndpointRuntimeStatus>;60 ) => Effect.Effect<CloudManagedEndpointRuntimeStatus>;
61 }61 }
62>()("t3/cloud/ManagedEndpointRuntime/CloudManagedEndpointRuntime") {}62>()("t3/cloud/ManagedEndpointRuntime/CloudManagedEndpointRuntime") {}
@@ -84,28 +84,32 @@ function runtimeConfigKey(config: RelayManagedEndpointRuntimeConfig): string { · 21 unmodified lines
84 });84 });
85}85}
86 86
87const stopConnector = (connector: ActiveConnector | null) =>87const stopConnector = (connector: Option.Option<ActiveConnector>) =>
88 connector88 Option.match(connector, {
89 ? Scope.close(connector.scope, Exit.void).pipe(89 onNone: () => Effect.void,
90 onSome: (active) =>
91 Scope.close(active.scope, Exit.void).pipe(
90 Effect.tap(() =>92 Effect.tap(() =>
91 Effect.logInfo("Relay client stopped", {93 Effect.logInfo("Relay client stopped", {
92 pid: Number(connector.child.pid),94 pid: Number(active.child.pid),
93 }),95 }),
94 ),96 ),
95 Effect.ignore,97 Effect.ignore,
96 )98 ),
97 : Effect.void;99 });
98 100
99export const make = Effect.gen(function* () {101export const make = Effect.gen(function* () {
100 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;102 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
101 const relayClient = yield* RelayClient.RelayClient;103 const relayClient = yield* RelayClient.RelayClient;
102 const activeRef = yield* Ref.make<ActiveConnector | null>(null);104 const activeRef = yield* Ref.make<Option.Option<ActiveConnector>>(Option.none());
103 const desiredConfigRef = yield* Ref.make<RelayManagedEndpointRuntimeConfig | null>(null);105 const desiredConfigRef = yield* Ref.make<Option.Option<RelayManagedEndpointRuntimeConfig>>(
106 Option.none(),
107 );
104 const reconcileSemaphore = yield* Semaphore.make(1);108 const reconcileSemaphore = yield* Semaphore.make(1);
105 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];109 let reconcileConfig: CloudManagedEndpointRuntime["Service"]["applyConfig"];
106 110
107 const stopActive = Effect.gen(function* () {111 const stopActive = Effect.gen(function* () {
108 const active = yield* Ref.getAndSet(activeRef, null);112 const active = yield* Ref.getAndSet(activeRef, Option.none());
109 yield* stopConnector(active);113 yield* stopConnector(active);
110 });114 });
111 115
@@ -116,19 +120,20 @@ export const make = Effect.gen(function* () { · 4 unmodified lines
116 Effect.gen(function* () {120 Effect.gen(function* () {
117 const active = yield* Ref.get(activeRef);121 const active = yield* Ref.get(activeRef);
118 if (122 if (
119 active?.child.pid !== connector.child.pid ||123 Option.isNone(active) ||
120 active.configKey !== connector.configKey124 active.value.child.pid !== connector.child.pid ||
125 active.value.configKey !== connector.configKey
121 ) {126 ) {
122 return;127 return;
123 }128 }
124 yield* Ref.set(activeRef, null);129 yield* Ref.set(activeRef, Option.none());
125 yield* stopConnector(connector);130 yield* stopConnector(Option.some(connector));
126 131
127 const desiredConfig = yield* Ref.get(desiredConfigRef);132 const desiredConfig = yield* Ref.get(desiredConfigRef);
128 if (133 if (
129 !desiredConfig ||134 Option.isNone(desiredConfig) ||
130 desiredConfig.providerKind !== "cloudflare_tunnel" ||135 desiredConfig.value.providerKind !== "cloudflare_tunnel" ||
131 runtimeConfigKey(desiredConfig) !== connector.configKey136 runtimeConfigKey(desiredConfig.value) !== connector.configKey
132 ) {137 ) {
133 return;138 return;
134 }139 }
@@ -181,120 +186,130 @@ export const make = Effect.gen(function* () { · 46 unmodified lines
181 ),186 ),
182 );187 );
183 188
184 reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(function* (config) {189 reconcileConfig = Effect.fn("CloudManagedEndpointRuntime.reconcileConfig")(
185 if (!config || config.providerKind !== "cloudflare_tunnel") {190 function* (configOption) {
191 if (Option.isNone(configOption)) {
192 yield* stopActive;
193 return { status: "disabled" };
194 }
195
196 const config = configOption.value;
197 if (config.providerKind !== "cloudflare_tunnel") {
198 yield* stopActive;
199 return { status: "unsupported", providerKind: config.providerKind };
200 }
201
202 const nextConfigKey = runtimeConfigKey(config);
203 const active = yield* Ref.get(activeRef);
204 if (Option.isSome(active) && active.value.configKey === nextConfigKey) {
205 const isRunning = yield* active.value.child.isRunning.pipe(
206 Effect.orElseSucceed(() => false),
207 );
208 if (isRunning) {
209 return {
210 status: "running",
211 providerKind: "cloudflare_tunnel",
212 pid: Number(active.value.child.pid),
213 ...(active.value.config.tunnelId ? { tunnelId: active.value.config.tunnelId } : {}),
214 ...(active.value.config.tunnelName
215 ? { tunnelName: active.value.config.tunnelName }
216 : {}),
217 } satisfies CloudManagedEndpointRuntimeStatus;
218 }
219 }
220
186 yield* stopActive;221 yield* stopActive;
187 return config
188 ? { status: "unsupported", providerKind: config.providerKind }
189 : { status: "disabled" };
190 }
191 222
192 const nextConfigKey = runtimeConfigKey(config);223 const executable = yield* relayClient.resolve;
193 const active = yield* Ref.get(activeRef);224 if (executable.status !== "available") {
194 if (active?.configKey === nextConfigKey) {
195 const isRunning = yield* active.child.isRunning.pipe(Effect.orElseSucceed(() => false));
196 if (isRunning) {
197 return {225 return {
198 status: "running",226 status: "failed",
199 providerKind: "cloudflare_tunnel",227 providerKind: "cloudflare_tunnel",
200 pid: Number(active.child.pid),228 reason:
201 ...(active.config.tunnelId ? { tunnelId: active.config.tunnelId } : {}),229 executable.status === "unsupported"
202 ...(active.config.tunnelName ? { tunnelName: active.config.tunnelName } : {}),230 ? `Relay client is unsupported on ${executable.platform}-${executable.arch}.`
231 : "The relay client is not installed.",
232 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
233 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
203 } satisfies CloudManagedEndpointRuntimeStatus;234 } satisfies CloudManagedEndpointRuntimeStatus;
204 }235 }
205 }
206 236
207 yield* stopActive;237 const connectorScope = yield* Scope.make("sequential");
208 238 const child = yield* spawner
209 const executable = yield* relayClient.resolve;239 .spawn(
210 if (executable.status !== "available") {240 ChildProcess.make(executable.executablePath, ["tunnel", "run"], {
211 return {241 detached: false,
212 status: "failed",242 env: {
213 providerKind: "cloudflare_tunnel",243 ...process.env,
214 reason:244 TUNNEL_TOKEN: config.connectorToken,
215 executable.status === "unsupported"245 },
216 ? `Relay client is unsupported on ${executable.platform}-${executable.arch}.`246 shell: false,
217 : "The relay client is not installed.",247 stderr: "pipe",
218 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),248 stdout: "pipe",
219 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
220 } satisfies CloudManagedEndpointRuntimeStatus;
221 }
222
223 const connectorScope = yield* Scope.make("sequential");
224 const child = yield* spawner
225 .spawn(
226 ChildProcess.make(executable.executablePath, ["tunnel", "run"], {
227 detached: false,
228 env: {
229 ...process.env,
230 TUNNEL_TOKEN: config.connectorToken,
231 },
232 shell: false,
233 stderr: "pipe",
234 stdout: "pipe",
235 }),
236 )
237 .pipe(
238 Effect.provideService(Scope.Scope, connectorScope),
239 Effect.tap((child) =>
240 Effect.logInfo("Relay client process started; waiting for tunnel connection", {
241 pid: Number(child.pid),
242 tunnelId: config.tunnelId,
243 tunnelName: config.tunnelName,
244 }),249 }),
245 ),250 )
246 Effect.catch((cause) =>251 .pipe(
247 Effect.logWarning("Failed to start relay client", {252 Effect.provideService(Scope.Scope, connectorScope),
248 cause,253 Effect.tap((child) =>
249 tunnelId: config.tunnelId,254 Effect.logInfo("Relay client process started; waiting for tunnel connection", {
250 tunnelName: config.tunnelName,255 pid: Number(child.pid),
251 }).pipe(256 tunnelId: config.tunnelId,
252 Effect.andThen(Scope.close(connectorScope, Exit.void).pipe(Effect.ignore)),257 tunnelName: config.tunnelName,
253 Effect.as({258 }),
254 status: "failed",
255 providerKind: "cloudflare_tunnel",
256 reason: String(cause),
257 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
258 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
259 } satisfies CloudManagedEndpointRuntimeStatus),
260 ),259 ),
261 ),260 Effect.catch((cause) =>
262 );261 Effect.logWarning("Failed to start relay client", {
262 cause,
263 tunnelId: config.tunnelId,
264 tunnelName: config.tunnelName,
265 }).pipe(
266 Effect.andThen(Scope.close(connectorScope, Exit.void).pipe(Effect.ignore)),
267 Effect.as({
268 status: "failed",
269 providerKind: "cloudflare_tunnel",
270 reason: String(cause),
271 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
272 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
273 } satisfies CloudManagedEndpointRuntimeStatus),
274 ),
275 ),
276 );
263 277
264 if ("status" in child && child.status === "failed") {278 if ("status" in child && child.status === "failed") {
265 return child;279 return child;
266 }280 }
281
282 if (!("status" in child)) {
283 const connector = {
284 child,
285 scope: connectorScope,
286 configKey: nextConfigKey,
287 config,
288 } satisfies ActiveConnector;
289 yield* Ref.set(activeRef, Option.some(connector));
290 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
291 yield* Effect.forkIn(superviseConnector(connector), connectorScope);
292 return {
293 status: "running",
294 providerKind: "cloudflare_tunnel",
295 pid: Number(child.pid),
296 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
297 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
298 } satisfies CloudManagedEndpointRuntimeStatus;
299 }
267 300
268 if (!("status" in child)) {
269 const connector = {
270 child,
271 scope: connectorScope,
272 configKey: nextConfigKey,
273 config,
274 } satisfies ActiveConnector;
275 yield* Ref.set(activeRef, connector);
276 yield* Effect.forkIn(observeConnectorOutput(connector), connectorScope);
277 yield* Effect.forkIn(superviseConnector(connector), connectorScope);
278 return {301 return {
279 status: "running",302 status: "failed",
280 providerKind: "cloudflare_tunnel",303 providerKind: "cloudflare_tunnel",
281 pid: Number(child.pid),304 reason: "Relay client did not start.",
282 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),305 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
283 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),306 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
284 } satisfies CloudManagedEndpointRuntimeStatus;307 } satisfies CloudManagedEndpointRuntimeStatus;
285 }308 },
286 309 );
287 return {
288 status: "failed",
289 providerKind: "cloudflare_tunnel",
290 reason: "Relay client did not start.",
291 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),
292 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),
293 } satisfies CloudManagedEndpointRuntimeStatus;
294 });
295 310
296 const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")(311 const applyConfig = Effect.fn("CloudManagedEndpointRuntime.applyConfig")(
297 (config: RelayManagedEndpointRuntimeConfig | null) =>312 (config: Option.Option<RelayManagedEndpointRuntimeConfig>) =>
298 reconcileSemaphore.withPermits(1)(313 reconcileSemaphore.withPermits(1)(
299 Ref.set(desiredConfigRef, config).pipe(Effect.andThen(reconcileConfig(config))),314 Ref.set(desiredConfigRef, config).pipe(Effect.andThen(reconcileConfig(config))),
300 ),315 ),
@@ -307,12 +322,12 @@ export const make = Effect.gen(function* () { · 6 unmodified lines
307 const initialConfig = yield* readRuntimeConfig.pipe(322 const initialConfig = yield* readRuntimeConfig.pipe(
308 Effect.catch((cause) =>323 Effect.catch((cause) =>
309 Effect.logWarning("Failed to read managed endpoint runtime config", { cause }).pipe(324 Effect.logWarning("Failed to read managed endpoint runtime config", { cause }).pipe(
310 Effect.as(null),325 Effect.as(Option.none<RelayManagedEndpointRuntimeConfig>()),
311 ),326 ),
312 ),327 ),
313 );328 );
314 yield* runtime.applyConfig(initialConfig);329 yield* runtime.applyConfig(initialConfig);
315 yield* Effect.addFinalizer(() => runtime.applyConfig(null));330 yield* Effect.addFinalizer(() => runtime.applyConfig(Option.none()));
316 return runtime;331 return runtime;
317});332});
318 333
apps/server/src/cloud/http.ts+2−2
@@ -433,7 +433,7 @@ const applyCloudRelayConfig = Effect.fn("environment.cloud.applyRelayConfig")(fu · 432 unmodified lines
433 });433 });
434 yield* validateCloudMintPublicKey(payload.cloudMintPublicKey);434 yield* validateCloudMintPublicKey(payload.cloudMintPublicKey);
435 const endpointRuntimeStatus = yield* dependencies.endpointRuntime.applyConfig(435 const endpointRuntimeStatus = yield* dependencies.endpointRuntime.applyConfig(
436 payload.endpointRuntime,436 Option.fromNullOr(payload.endpointRuntime),
437 );437 );
438 const ok =438 const ok =
439 endpointRuntimeStatus.status === "disabled" || endpointRuntimeStatus.status === "running";439 endpointRuntimeStatus.status === "disabled" || endpointRuntimeStatus.status === "running";
@@ -642,7 +642,7 @@ const cloudLinkStateHandler = Effect.fn("environment.cloud.linkState")( · 202 unmodified lines
642const cloudUnlinkHandler = Effect.fn("environment.cloud.unlink")(642const cloudUnlinkHandler = Effect.fn("environment.cloud.unlink")(
643 function* (dependencies: CloudHttpDependencies) {643 function* (dependencies: CloudHttpDependencies) {
644 yield* requireEnvironmentScope(AuthRelayWriteScope);644 yield* requireEnvironmentScope(AuthRelayWriteScope);
645 const endpointRuntimeStatus = yield* dependencies.endpointRuntime.applyConfig(null);645 const endpointRuntimeStatus = yield* dependencies.endpointRuntime.applyConfig(Option.none());
646 yield* Effect.all(646 yield* Effect.all(
647 [647 [
648 dependencies.secrets.remove(CLOUD_LINKED_USER_ID),648 dependencies.secrets.remove(CLOUD_LINKED_USER_ID),
apps/server/src/server.test.ts+5−4
@@ -2253,16 +2253,17 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 2252 unmodified lines
2253 layers: {2253 layers: {
2254 cloudManagedEndpointRuntime: {2254 cloudManagedEndpointRuntime: {
2255 applyConfig: (config) => {2255 applyConfig: (config) => {
2256 appliedRuntimeConfigs.push(config);2256 appliedRuntimeConfigs.push(Option.getOrNull(config));
2257 if (!config) {2257 if (Option.isNone(config)) {
2258 return Effect.succeed({ status: "disabled" });2258 return Effect.succeed({ status: "disabled" });
2259 }2259 }
2260 const runtimeConfig = config.value;
2260 return Effect.succeed({2261 return Effect.succeed({
2261 status: "running",2262 status: "running",
2262 providerKind: "cloudflare_tunnel",2263 providerKind: "cloudflare_tunnel",
2263 pid: 123,2264 pid: 123,
2264 ...(config.tunnelId ? { tunnelId: config.tunnelId } : {}),2265 ...(runtimeConfig.tunnelId ? { tunnelId: runtimeConfig.tunnelId } : {}),
2265 ...(config.tunnelName ? { tunnelName: config.tunnelName } : {}),2266 ...(runtimeConfig.tunnelName ? { tunnelName: runtimeConfig.tunnelName } : {}),
2266 });2267 });
2267 },2268 },
2268 },2269 },