Unified · Split
| @@ -2489,8 +2489,13 @@ describe("agent read revalidation (coreTools connections.list)", () => { · 2488 unmodified lines | |||
| 2489 | it.effect("concurrent lists past the freshness gate collapse to one probe", () => | 2489 | it.effect("concurrent lists past the freshness gate collapse to one probe", () => |
| 2490 | Effect.gen(function* () { | 2490 | Effect.gen(function* () { |
| 2491 | const gate = yield* Deferred.make<void>(); | 2491 | const gate = yield* Deferred.make<void>(); |
| 2492 | // Completed on probe entry, AFTER `counters.probes` has been bumped: | ||
| 2493 | // awaiting it is the explicit "a probe has started" ordering point (a | ||
| 2494 | // wall-clock sleep is a timing assumption that loses under suite load). | ||
| 2495 | const entered = yield* Deferred.make<void>(); | ||
| 2492 | const { executor, counters, stamp, persisted } = yield* makeHealthHarness({ | 2496 | const { executor, counters, stamp, persisted } = yield* makeHealthHarness({ |
| 2493 | probe: Deferred.await(gate).pipe( | 2497 | probe: Deferred.succeed(entered, void 0).pipe( |
| 2498 | Effect.andThen(Deferred.await(gate)), | ||
| 2494 | Effect.map(() => ({ | 2499 | Effect.map(() => ({ |
| 2495 | status: "healthy" as const, | 2500 | status: "healthy" as const, |
| 2496 | checkedAt: Date.now(), | 2501 | checkedAt: Date.now(), |
| @@ -2508,11 +2513,12 @@ describe("agent read revalidation (coreTools connections.list)", () => { · 11 unmodified lines | |||
| 2508 | { concurrency: "unbounded" }, | 2513 | { concurrency: "unbounded" }, |
| 2509 | ), | 2514 | ), |
| 2510 | ); | 2515 | ); |
| 2511 | // The probe is held open by the gate, so NOTHING has been persisted yet: | 2516 | // Wait until the first probe has provably started. The gate still holds |
| 2512 | // every one of the five lists must pass the freshness check and reach | 2517 | // it open, so NOTHING has been persisted: every list that runs must pass |
| 2513 | // the probe path. Give them real time to get there — the counter then | 2518 | // the freshness check into the probe path, and the in-flight gate must |
| 2514 | // says how many probes actually started. | 2519 | // keep the count at one — any duplicate a later list starts is caught by |
| 2515 | yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 25))); | 2520 | // the final count below. |
| 2521 | yield* Deferred.await(entered); | ||
| 2516 | expect(counters.probes).toBe(1); | 2522 | expect(counters.probes).toBe(1); |
| 2517 | 2523 | ||
| 2518 | yield* Deferred.succeed(gate, void 0); | 2524 | yield* Deferred.succeed(gate, void 0); |
| @@ -2977,6 +2983,12 @@ describe("credential-only health path", () => { · 458 unmodified lines | |||
| 2977 | it.effect("concurrent checks collapse to one credential resolution", () => | 2983 | it.effect("concurrent checks collapse to one credential resolution", () => |
| 2978 | Effect.gen(function* () { | 2984 | Effect.gen(function* () { |
| 2979 | const gate = yield* Deferred.make<void>(); | 2985 | const gate = yield* Deferred.make<void>(); |
| 2986 | // Completed by the provider on entry, AFTER the resolve counter has | ||
| 2987 | // been bumped: awaiting it is the explicit "a resolution has started" | ||
| 2988 | // ordering point. A wall-clock sleep here is a timing assumption — under | ||
| 2989 | // parallel suite load the forked checks may not have finished their row | ||
| 2990 | // loads yet, and the counter reads 0. | ||
| 2991 | const entered = yield* Deferred.make<void>(); | ||
| 2980 | const { executor, counters, stamp, persisted, hooks } = yield* makeHealthHarness(); | 2992 | const { executor, counters, stamp, persisted, hooks } = yield* makeHealthHarness(); |
| 2981 | // No declared probe spec + an OAuth client on the row routes checkHealth | 2993 | // No declared probe spec + an OAuth client on the row routes checkHealth |
| 2982 | // down the credential-only path: the verdict is "the credential | 2994 | // down the credential-only path: the verdict is "the credential |
| @@ -2984,7 +2996,9 @@ describe("credential-only health path", () => { · 1 unmodified line | |||
| 2984 | // behind the same in-flight gate as probing, so concurrent checks must | 2996 | // behind the same in-flight gate as probing, so concurrent checks must |
| 2985 | // collapse to ONE resolution. | 2997 | // collapse to ONE resolution. |
| 2986 | yield* stamp({ oauth_client: "acme", expires_at: null }); | 2998 | yield* stamp({ oauth_client: "acme", expires_at: null }); |
| 2987 | hooks.onResolve = Deferred.await(gate); | 2999 | hooks.onResolve = Deferred.succeed(entered, void 0).pipe( |
| 3000 | Effect.andThen(Deferred.await(gate)), | ||
| 3001 | ); | ||
| 2988 | counters.resolves = 0; | 3002 | counters.resolves = 0; |
| 2989 | const ref = { owner: "org", integration: INTEG, name: ConnectionName.make("main") } as const; | 3003 | const ref = { owner: "org", integration: INTEG, name: ConnectionName.make("main") } as const; |
| 2990 | 3004 | ||
| @@ -2993,10 +3007,12 @@ describe("credential-only health path", () => { · 2 unmodified lines | |||
| 2993 | concurrency: "unbounded", | 3007 | concurrency: "unbounded", |
| 2994 | }), | 3008 | }), |
| 2995 | ); | 3009 | ); |
| 2996 | // The resolution is held open by the gate, so nothing has been | 3010 | // Wait until the first resolution has provably entered the provider. |
| 2997 | // persisted: both checks must reach the credential path. Give them real | 3011 | // The gate still holds it open, so nothing has been persisted: while it |
| 2998 | // time to get there — the counter then says how many resolutions started. | 3012 | // holds, the in-flight gate must keep the count at exactly one — and any |
| 2999 | yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 25))); | 3013 | // duplicate the second check ever starts is caught by the final count |
| 3014 | // below. | ||
| 3015 | yield* Deferred.await(entered); | ||
| 3000 | expect(counters.resolves).toBe(1); | 3016 | expect(counters.resolves).toBe(1); |
| 3001 | 3017 | ||
| 3002 | yield* Deferred.succeed(gate, void 0); | 3018 | yield* Deferred.succeed(gate, void 0); |
| @@ -3056,6 +3072,11 @@ describe("health probe gate key integrity", () => { · 53 unmodified lines | |||
| 3056 | Effect.gen(function* () { | 3072 | Effect.gen(function* () { |
| 3057 | const counters = { probes: 0 }; | 3073 | const counters = { probes: 0 }; |
| 3058 | const gate = yield* Deferred.make<void>(); | 3074 | const gate = yield* Deferred.make<void>(); |
| 3075 | // Completed by the probe that brings the count to two: awaiting it is | ||
| 3076 | // the explicit "both probes started" ordering point. A collided gate | ||
| 3077 | // never completes it (the second check joins the first probe's Deferred | ||
| 3078 | // instead of starting its own), which fails the test by timeout. | ||
| 3079 | const bothEntered = yield* Deferred.make<void>(); | ||
| 3059 | // Every probe increments the shared counter and then parks on the gate, | 3080 | // Every probe increments the shared counter and then parks on the gate, |
| 3060 | // so both checks are provably in flight at once: nothing is persisted, | 3081 | // so both checks are provably in flight at once: nothing is persisted, |
| 3061 | // and a collided gate would let the second check join the first probe's | 3082 | // and a collided gate would let the second check join the first probe's |
| @@ -3071,7 +3092,10 @@ describe("health probe gate key integrity", () => { · 9 unmodified lines | |||
| 3071 | checkHealth: () => | 3092 | checkHealth: () => |
| 3072 | Effect.suspend(() => { | 3093 | Effect.suspend(() => { |
| 3073 | counters.probes += 1; | 3094 | counters.probes += 1; |
| 3074 | return Deferred.await(gate).pipe( | 3095 | const arrived = |
| 3096 | counters.probes >= 2 ? Deferred.succeed(bothEntered, void 0) : Effect.void; | ||
| 3097 | return arrived.pipe( | ||
| 3098 | Effect.andThen(Deferred.await(gate)), | ||
| 3075 | Effect.map(() => ({ | 3099 | Effect.map(() => ({ |
| 3076 | status: "healthy" as const, | 3100 | status: "healthy" as const, |
| 3077 | checkedAt: Date.now(), | 3101 | checkedAt: Date.now(), |
| @@ -3111,9 +3135,10 @@ describe("health probe gate key integrity", () => { · 33 unmodified lines | |||
| 3111 | 3135 | ||
| 3112 | const checkA = yield* Effect.forkChild(executorA.connections.checkHealth(ref)); | 3136 | const checkA = yield* Effect.forkChild(executorA.connections.checkHealth(ref)); |
| 3113 | const checkB = yield* Effect.forkChild(executorB.connections.checkHealth(ref)); | 3137 | const checkB = yield* Effect.forkChild(executorB.connections.checkHealth(ref)); |
| 3114 | // Give both fibers real time to reach the probe path while the gate | 3138 | // Wait until both probes have provably started; the gate holds every |
| 3115 | // holds every probe open; the counter then says how many started. | 3139 | // probe open, so both checks are in flight at once and nothing has been |
| 3116 | yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 25))); | 3140 | // persisted. |
| 3141 | yield* Deferred.await(bothEntered); | ||
| 3117 | expect(counters.probes).toBe(2); | 3142 | expect(counters.probes).toBe(2); |
| 3118 | 3143 | ||
| 3119 | yield* Deferred.succeed(gate, void 0); | 3144 | yield* Deferred.succeed(gate, void 0); |