Use managed runtimes for base processor facades

ts/packages/base/src/processor/async-processor.ts

@@ -8,7 +8,7 @@
 
 import type { PubSubBackend } from "../backend/types.js";
 import { makeNatsBackend } from "../backend/nats.js";
-import { Context, Effect } from "effect";
+import { Context, Effect, Layer, ManagedRuntime } from "effect";
 import { processorLifecycleError, type ProcessorLifecycleError } from "../errors.js";
 import { loadProcessorRuntimeConfig } from "../runtime/config.js";
 
@@ -74,6 +74,8 @@ interface RegisteredSignalHandler {
   readonly handler: () => void;
 }
 
+const asyncProcessorRuntime = ManagedRuntime.make(Layer.empty);
+
 export function makeAsyncProcessor<
   RunError = ProcessorLifecycleError,
   RunRequirements = never,
@@ -94,14 +96,10 @@ export function makeAsyncProcessor<
     }
 
     const shutdown = () => {
-      void Effect.runPromise(
+      void asyncProcessorRuntime.runPromise(
         Effect.log(`[${config.id}] Shutting down...`).pipe(
-          Effect.flatMap(() =>
-            Effect.tryPromise({
-              try: () => processor.stop(),
-              catch: (error) => processorLifecycleError(config.id, "signal-shutdown", error),
-            }),
-          ),
+          Effect.flatMap(() => processor.stopEffect),
+          Effect.mapError((error) => processorLifecycleError(config.id, "signal-shutdown", error)),
         ),
       ).then(() => process.exit(0), () => process.exit(1));
     };
@@ -133,8 +131,8 @@ export function makeAsyncProcessor<
     registerConfigHandler: (handler) => {
       configHandlers.push(handler);
     },
-    start: (context) => Effect.runPromiseWith(context)(processor.startEffect),
-    stop: () => Effect.runPromise(processor.stopEffect),
+    start: (context) => asyncProcessorRuntime.runPromise(Effect.provide(processor.startEffect, context)),
+    stop: () => asyncProcessorRuntime.runPromise(processor.stopEffect),
     onShutdown: (callback) => {
       shutdownCallbacks.push(callback);
     },
@@ -178,7 +176,7 @@ export function makeAsyncProcessor<
       });
       return stopProcessor();
     },
-    run: (context) => Effect.runPromiseWith(context)(processor.runEffect),
+    run: (context) => asyncProcessorRuntime.runPromise(Effect.provide(processor.runEffect, context)),
     get runEffect() {
       if (options.runEffect !== undefined) {
         return options.runEffect(processor);
@@ -208,7 +206,7 @@ export const AsyncProcessor = Object.assign(
       id: string,
     ): Promise<void> {
       const ProcessorCtor = this;
-      return Effect.runPromise(
+      return asyncProcessorRuntime.runPromise(
         Effect.gen(function* () {
           const config = yield* loadProcessorRuntimeConfig(id);
           const processor = new ProcessorCtor(config);

ts/packages/base/src/processor/flow-processor.ts

@@ -37,7 +37,7 @@ import {
 } from "../messaging/runtime.js";
 import { makePubSubService, PubSub } from "../backend/pubsub.js";
 import { loadMessagingRuntimeConfig } from "../runtime/messaging-config.js";
-import { Duration, Effect, Exit, Scope } from "effect";
+import { Duration, Effect, Exit, Layer, ManagedRuntime, Scope } from "effect";
 import * as Predicate from "effect/Predicate";
 import * as S from "effect/Schema";
 
@@ -346,6 +346,7 @@ export function makeFlowProcessor<FlowRequirements = never>(
   const specifications: Array<Spec<FlowRequirements>> = [
     ...(options.specifications ?? []),
   ];
+  const compatibilityRuntime = ManagedRuntime.make(Layer.empty);
   let processor: FlowProcessorRuntime<FlowRequirements>;
   const base: AsyncProcessorRuntime<
     PubSubError | FlowRuntimeError | ProcessorLifecycleError,
@@ -385,7 +386,7 @@ export function makeFlowProcessor<FlowRequirements = never>(
       return makeStartEffect();
     },
     start: (context) =>
-      Effect.runPromiseWith(context)(
+      compatibilityRuntime.runPromise(Effect.provide(
         Effect.gen(function* () {
           const pubsub = makePubSubService(base.pubsub);
           const messagingConfig = yield* loadMessagingRuntimeConfig();
@@ -401,7 +402,8 @@ export function makeFlowProcessor<FlowRequirements = never>(
           );
           yield* Effect.scoped(start);
         }),
-      ),
+        context,
+      )),
   };
 
   return processor;

ts/packages/base/src/processor/flow.ts

@@ -4,7 +4,7 @@
  * Python reference: trustgraph-base/trustgraph/base/flow.py
  */
 
-import { Context, Effect, Exit, Scope } from "effect";
+import { Context, Effect, Exit, Layer, ManagedRuntime, Scope } from "effect";
 import type { PubSubBackend } from "../backend/types.js";
 import { makePubSubService } from "../backend/pubsub.js";
 import {
@@ -69,6 +69,7 @@ export function makeFlow<Requirements = never>(
   const requestors = new Map<string, EffectRequestResponse<never, unknown>>();
   const parameters = new Map<string, unknown>();
   let compatibilityScope: Scope.Closeable | null = null;
+  const compatibilityRuntime = ManagedRuntime.make(Layer.empty);
 
   const ensureCompatibilityScopeEffect = Effect.fn("Flow.ensureCompatibilityScope")(function* () {
     if (compatibilityScope !== null) {
@@ -107,7 +108,7 @@ export function makeFlow<Requirements = never>(
       });
     },
     start(context: Context.Context<Requirements>): Promise<void> {
-      return Effect.runPromise(
+      return compatibilityRuntime.runPromise(
         Effect.gen(function* () {
           if (compatibilityScope !== null) {
             yield* flow.stopEffect();
@@ -117,7 +118,7 @@ export function makeFlow<Requirements = never>(
       );
     },
     stop(): Promise<void> {
-      return Effect.runPromise(flow.stopEffect());
+      return compatibilityRuntime.runPromise(flow.stopEffect());
     },
     stopEffect(): Effect.Effect<void> {
       return Effect.gen(function* () {
@@ -157,7 +158,7 @@ export function makeFlow<Requirements = never>(
       runtimePubsub: PubSubBackend,
       context: Context.Context<Requirements>,
     ): Promise<A> {
-      return Effect.runPromise(flow.runInCompatibilityScopeEffect(effect, runtimePubsub, context));
+      return compatibilityRuntime.runPromise(flow.runInCompatibilityScopeEffect(effect, runtimePubsub, context));
     },
     clearResources(): void {
       producers.clear();
@@ -207,16 +208,16 @@ export function makeFlow<Requirements = never>(
       const p = producers.get(producerName);
       if (p === undefined) throw flowResourceNotFoundError(name, "producer", producerName);
       return {
-        send: (id, message) => Effect.runPromise((p as EffectProducer<T>).send(id, message)),
-        flush: () => Effect.runPromise(p.flush),
-        stop: () => Effect.runPromise(p.flush.pipe(Effect.flatMap(() => p.close))),
+        send: (id, message) => compatibilityRuntime.runPromise((p as EffectProducer<T>).send(id, message)),
+        flush: () => compatibilityRuntime.runPromise(p.flush),
+        stop: () => compatibilityRuntime.runPromise(p.flush.pipe(Effect.flatMap(() => p.close))),
       };
     },
     consumer(consumerName: string): FlowConsumer {
       const c = consumers.get(consumerName);
       if (c === undefined) throw flowResourceNotFoundError(name, "consumer", consumerName);
       return {
-        stop: () => Effect.runPromise(c.stop),
+        stop: () => compatibilityRuntime.runPromise(c.stop),
       };
     },
     requestor<TReq, TRes>(requestorName: string): FlowRequestor<TReq, TRes> {
@@ -224,13 +225,13 @@ export function makeFlow<Requirements = never>(
       if (rr === undefined) throw flowResourceNotFoundError(name, "requestor", requestorName);
       return {
         request: (request, options) =>
-          Effect.runPromise(
+          compatibilityRuntime.runPromise(
             (rr as EffectRequestResponse<TReq, TRes>).request(
               request,
               toEffectRequestOptions(options),
             ),
           ),
-        stop: () => Effect.runPromise(rr.stop),
+        stop: () => compatibilityRuntime.runPromise(rr.stop),
       };
     },
     parameter<T>(parameterName: string): T {

ts/packages/base/src/spec/parameter-spec.ts

@@ -20,6 +20,7 @@ export function makeParameterSpec(name: string): ParameterSpec {
   return {
     name,
     addEffect,
-    add: (flow, _pubsub, definition) => Effect.runPromise(addEffect(flow, definition)),
+    add: (flow, pubsub, definition, context) =>
+      flow.runInCompatibilityScope(addEffect(flow, definition), pubsub, context),
   };
 }