From bfaab5db3d4fd25ad3e3d0f744c5169e7b235307 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sun, 6 Sep 2026 15:07:26 +0530 Subject: [PATCH 1/2] fix(sdk): prevent canceled health checks from stranding probes --- packages/core/sdk/src/connections.test.ts | 67 +++++++++- packages/core/sdk/src/executor.ts | 152 +++++++++++----------- 2 files changed, 145 insertions(+), 74 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 2070befcb..0d03fb1dd 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "@effect/vitest"; +import { describe, expect, it, vi } from "@effect/vitest"; import { Cause, Deferred, @@ -11,6 +11,7 @@ import { Predicate, Result, Schema, + Scheduler, Tracer, } from "effect"; @@ -3524,6 +3525,70 @@ describe("credential-only health path", () => { }); describe("health probe gate lifecycle", () => { + it.effect("cancellation between gate registration and fork cannot strand later checks", () => + Effect.gen(function* () { + const { executor, counters, stamp } = yield* makeHealthHarness(); + yield* stamp({ oauth_client: "acme", expires_at: null }); + const ref = { owner: "org", integration: INTEG, name: ConnectionName.make("main") } as const; + const controller = new AbortController(); + let registered = false; + let interrupted = false; + let cancelOnResume = false; + const scheduler = new Scheduler.MixedScheduler("async", (resume) => { + const timer = setTimeout(() => { + if (cancelOnResume) { + cancelOnResume = false; + controller.abort(); + } + resume(); + }, 0); + return () => clearTimeout(timer); + }); + scheduler.shouldYield = () => { + if (!registered || interrupted) return false; + interrupted = true; + cancelOnResume = true; + return true; + }; + + // Observe registration only to place cancellation in the otherwise tiny + // register/fork window. All writes retain their normal behavior, and the + // assertions below use the public health API. + const originalSet = Map.prototype.set; + const registration = vi.spyOn(Map.prototype, "set").mockImplementation(function ( + this: Map, + key: unknown, + value: unknown, + ) { + const result = originalSet.call(this, key, value); + if ( + typeof key === "string" && + key.endsWith(',"vercel","main"]') && + Deferred.isDeferred(value) + ) { + registered = true; + } + return result; + }); + yield* Effect.promise(() => + Effect.runPromiseExit(executor.connections.checkHealth(ref), { + signal: controller.signal, + scheduler, + }), + ).pipe(Effect.ensuring(Effect.sync(() => registration.mockRestore()))); + + expect(interrupted).toBe(true); + const next = yield* Effect.promise(() => + Effect.runPromise( + executor.connections.checkHealth(ref).pipe(Effect.timeoutOption("1 second")), + ), + ); + expect(Option.isSome(next)).toBe(true); + expect(Option.getOrThrow(next).status).toBe("healthy"); + expect(counters.resolves).toBeGreaterThan(0); + }), + ); + it.effect("a failed probe clears its gate entry for the next check", () => Effect.gen(function* () { let firstProbe = true; diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index bdc755f31..f262f7ded 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -6053,79 +6053,85 @@ export const createExecutor = { - const key = healthProbeGateKey(tenant, connectionRow); - const existing = healthProbeInFlight.get(key); - if (existing) return Deferred.await(existing); - const deferred = Deferred.makeUnsafe(); - // Nothing suspends between the lookup above and this registration, - // so check-and-set is atomic against peer fibers. - healthProbeInFlight.set(key, deferred); - const freshVerdict: Effect.Effect = - spec === undefined && connectionRow.oauth_client != null - ? // No probe operation is declared, so "healthy" here means only - // "the credential resolved (refreshing if due)" — a refresh - // failure is the one real signal this path can produce, and it - // must not hide inside a green span. - oauthCredentialHealthWithoutProbe(connectionRow).pipe( - Effect.tap((result) => persistProbeHealthResult(ref, result)), - Effect.map((result) => ({ - source: "credential_only" as const, - result, - })), - ) - : foldCredentialResolutionIntoVerdict( - Effect.gen(function* () { - const values = yield* resolveConnectionValues(connectionRow); - const record = rowToIntegrationRecord( - integrationRow, - yield* describeAuthMethodsForRow(integrationRow), - ); - const grantedScopes = grantedScopesFromRow(connectionRow); - const credential: ToolInvocationCredential = { - owner: connectionRow.owner as Owner, - integration: ref.integration, - connection: ConnectionName.make(connectionRow.name), - template: AuthTemplateSlug.make(connectionRow.template), - value: values[PRIMARY_INPUT_VARIABLE] ?? null, - values, - config: record.config, - ...(grantedScopes ? { grantedScopes } : {}), - }; - // Core resolves the declared spec (its own column) and - // hands it to the plugin; plugins no longer read it out of - // their config. - return yield* foldPluginFailure( - check({ - ctx: runtime.ctx, - integration: record, - credential, - spec, - }), - `Health check for connection "${ref.name}" failed.`, - ); - }), - ).pipe( - // Persist the verdict on the connection row so the accounts - // list shows alive/expired at a glance, AND so the freshness - // gate above has something to serve. A probe that could not - // resolve its credential persists too: it is the connection - // most likely to be re-probed by every surface on every - // mount, so leaving it unwritten is what turns one broken - // connection into unbounded upstream and error traffic. - Effect.tap((result) => persistProbeHealthResult(ref, result)), - Effect.map((result) => ({ - source: "probe" as const, - result, - })), - ); - const run = freshVerdict.pipe( - Effect.exit, - Effect.flatMap((exit) => Deferred.done(deferred, exit)), - Effect.ensuring(Effect.sync(() => void healthProbeInFlight.delete(key))), - ); - return Effect.forkDetach(run).pipe(Effect.andThen(Deferred.await(deferred))); - }); + // Registration and startup must complete together. Cancellation between them + // would leave a Deferred that every later check waits on forever. + const outcome = yield* Effect.uninterruptibleMask((restore) => + Effect.suspend(() => { + const key = healthProbeGateKey(tenant, connectionRow); + const existing = healthProbeInFlight.get(key); + if (existing) return restore(Deferred.await(existing)); + const deferred = Deferred.makeUnsafe(); + // Nothing suspends between the lookup above and this registration, + // so check-and-set is atomic against peer fibers. + healthProbeInFlight.set(key, deferred); + const freshVerdict: Effect.Effect = + spec === undefined && connectionRow.oauth_client != null + ? // No probe operation is declared, so "healthy" here means only + // "the credential resolved (refreshing if due)" — a refresh + // failure is the one real signal this path can produce, and it + // must not hide inside a green span. + oauthCredentialHealthWithoutProbe(connectionRow).pipe( + Effect.tap((result) => persistProbeHealthResult(ref, result)), + Effect.map((result) => ({ + source: "credential_only" as const, + result, + })), + ) + : foldCredentialResolutionIntoVerdict( + Effect.gen(function* () { + const values = yield* resolveConnectionValues(connectionRow); + const record = rowToIntegrationRecord( + integrationRow, + yield* describeAuthMethodsForRow(integrationRow), + ); + const grantedScopes = grantedScopesFromRow(connectionRow); + const credential: ToolInvocationCredential = { + owner: connectionRow.owner as Owner, + integration: ref.integration, + connection: ConnectionName.make(connectionRow.name), + template: AuthTemplateSlug.make(connectionRow.template), + value: values[PRIMARY_INPUT_VARIABLE] ?? null, + values, + config: record.config, + ...(grantedScopes ? { grantedScopes } : {}), + }; + // Core resolves the declared spec (its own column) and + // hands it to the plugin; plugins no longer read it out of + // their config. + return yield* foldPluginFailure( + check({ + ctx: runtime.ctx, + integration: record, + credential, + spec, + }), + `Health check for connection "${ref.name}" failed.`, + ); + }), + ).pipe( + // Persist the verdict on the connection row so the accounts + // list shows alive/expired at a glance, AND so the freshness + // gate above has something to serve. A probe that could not + // resolve its credential persists too: it is the connection + // most likely to be re-probed by every surface on every + // mount, so leaving it unwritten is what turns one broken + // connection into unbounded upstream and error traffic. + Effect.tap((result) => persistProbeHealthResult(ref, result)), + Effect.map((result) => ({ + source: "probe" as const, + result, + })), + ); + const run = freshVerdict.pipe( + Effect.exit, + Effect.flatMap((exit) => Deferred.done(deferred, exit)), + Effect.ensuring(Effect.sync(() => void healthProbeInFlight.delete(key))), + ); + return Effect.forkDetach(run, { startImmediately: true }).pipe( + Effect.andThen(restore(Deferred.await(deferred))), + ); + }), + ); yield* annotateHealthVerdict(outcome.source, outcome.result, previous); return outcome.result; }).pipe( From 430c5139e8765047b4d200a03ff069dbb80af4aa Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sun, 6 Sep 2026 15:12:47 +0530 Subject: [PATCH 2/2] test(sdk): verify detached health probe timeouts (greptile) --- packages/core/sdk/src/connections.test.ts | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 0d03fb1dd..dbb897d00 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -3589,6 +3589,27 @@ describe("health probe gate lifecycle", () => { }), ); + it.live("a detached probe remains interruptible and releases its gate after timeout", () => + Effect.gen(function* () { + let firstProbe = true; + const { executor, counters } = yield* makeHealthHarness({ + probe: Effect.suspend(() => { + if (firstProbe) { + firstProbe = false; + return Effect.never.pipe(Effect.timeout("10 millis")); + } + return Effect.succeed({ status: "healthy" as const, checkedAt: Date.now() }); + }), + }); + const ref = { owner: "org", integration: INTEG, name: ConnectionName.make("main") } as const; + const first = yield* executor.connections.checkHealth(ref).pipe(Effect.exit); + expect(Exit.isFailure(first)).toBe(true); + const next = yield* executor.connections.checkHealth(ref); + expect(next.status).toBe("healthy"); + expect(counters.probes).toBe(2); + }), + ); + it.effect("a failed probe clears its gate entry for the next check", () => Effect.gen(function* () { let firstProbe = true;