diff --git a/packages/provider-core/src/server/managedProvider.test.ts b/packages/provider-core/src/server/managedProvider.test.ts index 7794107824c9b..9df85130d6017 100644 --- a/packages/provider-core/src/server/managedProvider.test.ts +++ b/packages/provider-core/src/server/managedProvider.test.ts @@ -3,6 +3,7 @@ import { DEFAULT_SERVER_SETTINGS, ProviderDriverKind, ProviderInstanceId, + ServerSettingsError, type ServerProvider, } from "@t3tools/contracts"; import { createModelCapabilities } from "@t3tools/shared/model"; @@ -410,6 +411,67 @@ describe("makeManagedServerProvider", () => { ).pipe(Effect.provide(layerAlwaysRunTest)), ); + it.effect("keeps consuming settings changes after a provider check fails", () => + Effect.scoped( + Effect.gen(function* () { + const settingsChanges = yield* PubSub.unbounded(); + const checkCalls = yield* Ref.make(0); + const releaseInitialCheck = yield* Deferred.make(); + const thirdCheckStarted = yield* Deferred.make(); + const releaseRecoveredCheck = yield* Deferred.make(); + const provider = yield* makeManagedServerProvider({ + resolveMaintenance: () => Effect.succeed(maintenanceCapabilities), + getSettings: Effect.succeed({ enabled: true }), + streamSettings: Stream.fromPubSub(settingsChanges), + haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled, + initialSnapshot: () => Effect.succeed(initialSnapshot), + checkProvider: Ref.updateAndGet(checkCalls, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 1 + ? Deferred.await(releaseInitialCheck).pipe(Effect.as(refreshedSnapshot)) + : count === 2 + ? Effect.fail( + new ServerSettingsError({ + settingsPath: "/t3/userdata/settings.json", + operation: "read-file", + }), + ) + : Deferred.succeed(thirdCheckStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseRecoveredCheck)), + Effect.as(refreshedSnapshotSecond), + ), + ), + ), + refreshInterval: "1 hour", + }); + + const updatesFiber = yield* Stream.take(provider.streamChanges, 2).pipe( + Stream.runCollect, + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* Deferred.succeed(releaseInitialCheck, undefined); + + // The first change fails its provider check; the failed application never + // advances the stored settings, so the same change still differs next time. + yield* PubSub.publish(settingsChanges, { enabled: false }); + yield* Effect.yieldNow; + assert.strictEqual(yield* Ref.get(checkCalls), 2); + + // The subscription must survive that failure and retry on the next change; + // on main the dead consumer never starts the third check and this times out. + yield* PubSub.publish(settingsChanges, { enabled: false }); + yield* Deferred.await(thirdCheckStarted); + assert.strictEqual(yield* Ref.get(checkCalls), 3); + yield* Deferred.succeed(releaseRecoveredCheck, undefined); + + const updates = Array.from(yield* Fiber.join(updatesFiber)); + assert.deepStrictEqual(updates, [refreshedSnapshot, refreshedSnapshotSecond]); + assert.deepStrictEqual(yield* provider.getSnapshot, refreshedSnapshotSecond); + }), + ).pipe(Effect.provide(layerAlwaysRunTest)), + ); + it.effect("streams supplemental snapshot updates after the base provider check completes", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/provider-core/src/server/managedProvider.ts b/packages/provider-core/src/server/managedProvider.ts index 41ff65b0f294b..f83ba2408e6f9 100644 --- a/packages/provider-core/src/server/managedProvider.ts +++ b/packages/provider-core/src/server/managedProvider.ts @@ -295,8 +295,10 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( ); } + // A failed check must not end this subscription: the next settings change + // retries, like the periodic and initial checks below. yield* Stream.runForEach(input.streamSettings, (nextSettings) => - Effect.asVoid(applySnapshot(nextSettings)), + Effect.asVoid(applySnapshot(nextSettings)).pipe(Effect.ignoreCause({ log: true })), ).pipe(Effect.forkScoped); yield* Effect.forever(