.opencode/node_modules/effect/dist/unstable/cluster/Runners.js

/**
 * Handles communication between Effect Cluster runners.
 *
 * `Runners` sits between sharding decisions and runner execution. It can ping a
 * runner, send requests or control envelopes, notify a runner that persisted
 * work is available, and record that a runner address is unavailable. This
 * module defines the runner communication service, its RPC protocol, no-op and
 * RPC-backed implementations, local persistence support, reply recovery, and
 * the protocol service used by transport-specific runner layers.
 *
 * @since 4.0.0
 */
import * as Context from "../../Context.js";
import * as Effect from "../../Effect.js";
import * as Exit from "../../Exit.js";
import * as Latch from "../../Latch.js";
import * as Layer from "../../Layer.js";
import * as Option from "../../Option.js";
import * as Queue from "../../Queue.js";
import * as RcMap from "../../RcMap.js";
import * as Schema from "../../Schema.js";
import * as Rpc from "../rpc/Rpc.js";
import * as RpcClient_ from "../rpc/RpcClient.js";
import * as RpcGroup from "../rpc/RpcGroup.js";
import * as RpcSchema from "../rpc/RpcSchema.js";
import { AlreadyProcessingMessage, EntityNotAssignedToRunner, MailboxFull, RunnerUnavailable } from "./ClusterError.js";
import { Persisted } from "./ClusterSchema.js";
import * as Envelope from "./Envelope.js";
import * as Message from "./Message.js";
import * as MessageStorage from "./MessageStorage.js";
import * as Reply from "./Reply.js";
import { ShardingConfig } from "./ShardingConfig.js";
import * as Snowflake from "./Snowflake.js";
/**
 * Service for communicating with cluster runners, including pinging runners,
 * sending and notifying messages, coordinating persisted replies, and marking
 * runners unavailable.
 *
 * @category context
 * @since 4.0.0
 */
export class Runners extends /*#__PURE__*/Context.Service()("effect/cluster/Runners") {}
/**
 * Builds the `Runners` service from remote runner callbacks and adds local
 * message persistence, duplicate request handling, optional local serialization
 * simulation, and polling for persisted replies.
 *
 * **When to use**
 *
 * Use when you need a custom `Runners` service around remote `ping`, `send`,
 * `notify`, and `onRunnerUnavailable` callbacks, with standard local
 * persistence and reply recovery behavior.
 *
 * **Details**
 *
 * `make` uses the supplied remote callbacks for runner communication and
 * derives `sendLocal` and `notifyLocal`. Local sends can optionally simulate
 * remote serialization, persisted notifications are saved through
 * `MessageStorage`, duplicate requests are resumed from stored replies when
 * possible, and pending replies are polled according to
 * `ShardingConfig.entityReplyPollInterval`.
 *
 * **Gotchas**
 *
 * `notify` and `notifyLocal` only support RPCs annotated as persisted; calling
 * either path with a non-persisted message dies instead of returning a typed
 * error.
 *
 * @see {@link makeRpc} for the RPC-backed implementation built on top of this constructor
 * @see {@link makeNoop} for a no-op implementation when remote runner communication is not needed
 *
 * @category constructors
 * @since 4.0.0
 */
export const make = /*#__PURE__*/Effect.fnUntraced(function* (options) {
  const storage = yield* MessageStorage.MessageStorage;
  const runnersScope = yield* Effect.scope;
  const snowflakeGen = yield* Snowflake.Generator;
  const config = yield* ShardingConfig;
  const requestIdRewrites = new Map();
  function notifyWith(message, afterPersist) {
    const rpc = message.rpc;
    const persisted = Context.get(rpc.annotations, Persisted);
    if (!persisted) {
      return Effect.die("Runners.notify only supports persisted messages");
    }
    if (message._tag === "OutgoingEnvelope") {
      const rewriteId = requestIdRewrites.get(message.envelope.requestId);
      const requestId = rewriteId ?? message.envelope.requestId;
      const entry = storageRequests.get(requestId);
      if (rewriteId) {
        message = new Message.OutgoingEnvelope({
          ...message,
          envelope: message.envelope.withRequestId(rewriteId)
        });
      }
      return storage.saveEnvelope(message).pipe(Effect.catchTag("MalformedMessage", Effect.die), Effect.andThen(entry ? Effect.andThen(entry.latch.open, afterPersist(message, false)) : afterPersist(message, false)));
    }
    // For requests, after persisting the request, we need to check if the
    // request is a duplicate. If it is, we need to resume from the last
    // received reply.
    //
    // Otherwise, we notify the remote entity and then reply from storage.
    return Effect.flatMap(Effect.catchTag(storage.saveRequest(message), "MalformedMessage", Effect.die), MessageStorage.SaveResult.$match({
      Success: () => afterPersist(message, false),
      Duplicate: ({
        lastReceivedReply,
        originalId
      }) => {
        // If the last received reply is an exit, we can just return it
        // as the response.
        if (Option.isSome(lastReceivedReply) && lastReceivedReply.value._tag === "WithExit") {
          return message.respond(lastReceivedReply.value.withRequestId(message.envelope.requestId));
        }
        requestIdRewrites.set(message.envelope.requestId, originalId);
        return afterPersist(new Message.OutgoingRequest({
          ...message,
          lastReceivedReply,
          envelope: Envelope.makeRequest({
            ...message.envelope,
            requestId: originalId
          }),
          respond(reply) {
            if (reply._tag === "WithExit") {
              requestIdRewrites.delete(message.envelope.requestId);
            }
            return message.respond(reply.withRequestId(message.envelope.requestId));
          }
        }), true);
      }
    }));
  }
  const storageRequests = new Map();
  const waitingStorageRequests = new Map();
  const replyFromStorage = Effect.fnUntraced(function* (message) {
    let entry = storageRequests.get(message.envelope.requestId);
    if (entry) {
      entry.messages.add(message);
      entry.doneLatch ??= Latch.makeUnsafe(false);
      return yield* entry.doneLatch.await;
    } else {
      entry = {
        latch: Latch.makeUnsafe(false),
        doneLatch: undefined,
        replies: [],
        messages: new Set([message])
      };
      storageRequests.set(message.envelope.requestId, entry);
    }
    while (true) {
      // wait for the storage loop to notify us
      entry.latch.closeUnsafe();
      waitingStorageRequests.set(message.envelope.requestId, message);
      storageLatch.openUnsafe();
      yield* entry.latch.await;
      // send the replies back
      for (let i = 0; i < entry.replies.length; i++) {
        const reply = entry.replies[i];
        // we have reached the end
        if (reply._tag === "WithExit") {
          for (const message of entry.messages) {
            yield* message.respond(reply);
          }
          entry.doneLatch?.openUnsafe();
          return;
        }
        entry.latch.closeUnsafe();
        for (const message of entry.messages) {
          yield* message.respond(reply);
        }
        // wait for ack
        yield* entry.latch.await;
      }
      entry.replies = [];
    }
  }, (effect, message) => Effect.ensuring(effect, Effect.sync(() => {
    const entry = storageRequests.get(message.envelope.requestId);
    if (!entry || entry.messages.size > 1) {
      entry?.messages.delete(message);
      return;
    }
    storageRequests.delete(message.envelope.requestId);
    waitingStorageRequests.delete(message.envelope.requestId);
  })));
  const storageLatch = Latch.makeUnsafe(false);
  if (storage !== MessageStorage.noop) {
    yield* Effect.gen(function* () {
      const foundRequests = new Set();
      while (true) {
        yield* storageLatch.await;
        storageLatch.closeUnsafe();
        const replies = yield* storage.repliesFor(waitingStorageRequests.values()).pipe(Effect.catchCause(cause => Effect.as(Effect.annotateLogs(Effect.logDebug(cause), {
          package: "@effect/cluster",
          module: "Runners",
          fiber: "Read replies loop"
        }), [])));
        // put the replies into the storage requests and then open the latches
        for (let i = 0; i < replies.length; i++) {
          const reply = replies[i];
          const entry = storageRequests.get(reply.requestId);
          if (!entry) continue;
          entry.replies.push(reply);
          waitingStorageRequests.delete(reply.requestId);
          foundRequests.add(entry);
        }
        foundRequests.forEach(entry => entry.latch.openUnsafe());
        foundRequests.clear();
      }
    }).pipe(Effect.forkIn(runnersScope));
    yield* Effect.suspend(() => {
      if (waitingStorageRequests.size === 0) {
        return storageLatch.await;
      }
      return storageLatch.open;
    }).pipe(Effect.delay(config.entityReplyPollInterval), Effect.forever, Effect.forkIn(runnersScope));
  }
  return Runners.of({
    ...options,
    sendLocal(options) {
      const message = options.message;
      if (!options.simulateRemoteSerialization) {
        return options.send(Message.incomingLocalFromOutgoing(message));
      }
      return Message.serialize(message).pipe(Effect.flatMap(encoded => Message.deserializeLocal(message, encoded)), Effect.flatMap(options.send), Effect.catchTag("MalformedMessage", error => {
        if (message._tag === "OutgoingEnvelope") {
          return Effect.die(error);
        }
        return message.respond(new Reply.WithExit({
          id: snowflakeGen.nextUnsafe(),
          requestId: message.envelope.requestId,
          exit: Exit.die(error)
        }));
      }));
    },
    notify(options_) {
      const {
        discard,
        message
      } = options_;
      return notifyWith(message, (message, duplicate) => {
        if (discard || message._tag === "OutgoingEnvelope") {
          return options.notify(options_);
        } else if (!duplicate && Option.isSome(options_.address)) {
          return Effect.catch(options.send({
            address: options_.address.value,
            message
          }), _ => replyFromStorage(message));
        }
        return options.notify(options_).pipe(Effect.andThen(replyFromStorage(message)));
      });
    },
    notifyLocal(options) {
      return notifyWith(options.message, (message, duplicate) => {
        if (options.discard || message._tag === "OutgoingEnvelope") {
          return Effect.catchTag(options.notify(Message.incomingLocalFromOutgoing(message)), "EntityNotAssignedToRunner", () => Effect.void);
        } else if (!duplicate && options.storageOnly !== true) {
          return options.notify(Message.incomingLocalFromOutgoing(message)).pipe(Effect.andThen(storage.registerReplyHandler(message)), Effect.catchTag("EntityNotAssignedToRunner", () => replyFromStorage(message)));
        }
        return options.notify(Message.incomingLocalFromOutgoing(message)).pipe(Effect.catchTag("EntityNotAssignedToRunner", () => Effect.void), Effect.andThen(replyFromStorage(message)));
      });
    }
  });
});
/**
 * Creates a no-op `Runners` service that rejects sends with
 * `EntityNotAssignedToRunner` and ignores notifications, pings, and unavailable
 * runner reports.
 *
 * @category No-op
 * @since 4.0.0
 */
export const makeNoop = /*#__PURE__*/make({
  send: ({
    message
  }) => Effect.fail(new EntityNotAssignedToRunner({
    address: message.envelope.address
  })),
  notify: () => Effect.void,
  ping: () => Effect.void,
  onRunnerUnavailable: () => Effect.void
});
/**
 * Layer that provides the no-op `Runners` service, using the default snowflake
 * generator.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerNoop = /*#__PURE__*/Layer.effect(Runners, makeNoop).pipe(/*#__PURE__*/Layer.provide([Snowflake.layerGenerator]));
const rpcErrors = /*#__PURE__*/Schema.Union([EntityNotAssignedToRunner, MailboxFull, AlreadyProcessingMessage]);
/**
 * RPC group used for runner-to-runner communication, including ping, notify,
 * effect, stream, and envelope messages.
 *
 * @category Rpcs
 * @since 4.0.0
 */
export class Rpcs extends /*#__PURE__*/RpcGroup.make(/*#__PURE__*/Rpc.make("Ping"), /*#__PURE__*/Rpc.make("Notify", {
  payload: {
    envelope: Envelope.Partial
  },
  success: Schema.Void,
  error: /*#__PURE__*/Schema.Union([EntityNotAssignedToRunner, AlreadyProcessingMessage])
}), /*#__PURE__*/Rpc.make("Effect", {
  payload: {
    request: Envelope.PartialRequest,
    persisted: Schema.Boolean
  },
  success: Reply.Encoded,
  error: rpcErrors
}), /*#__PURE__*/Rpc.make("Stream", {
  payload: {
    request: Envelope.PartialRequest,
    persisted: Schema.Boolean
  },
  error: rpcErrors,
  success: Reply.Encoded,
  stream: true
}), /*#__PURE__*/Rpc.make("Envelope", {
  payload: {
    envelope: /*#__PURE__*/Schema.Union([Envelope.AckChunk, Envelope.Interrupt]),
    persisted: Schema.Boolean
  },
  error: rpcErrors
})) {}
/**
 * Builds a runner RPC client from the current `RpcClient.Protocol`, using the
 * `Runners` span prefix with tracing disabled.
 *
 * @category Rpcs
 * @since 4.0.0
 */
export const makeRpcClient = /*#__PURE__*/RpcClient_.make(Rpcs, {
  spanPrefix: "Runners",
  disableTracing: true
});
/**
 * Builds a `Runners` service backed by RPC clients, caching a client per runner
 * address and dispatching ping, notify, effect, stream, and envelope messages over
 * the runner protocol.
 *
 * @category constructors
 * @since 4.0.0
 */
export const makeRpc = /*#__PURE__*/Effect.gen(function* () {
  const makeClientProtocol = yield* RpcClientProtocol;
  const snowflakeGen = yield* Snowflake.Generator;
  const clients = yield* RcMap.make({
    lookup: address => Effect.flatMap(makeClientProtocol(address), protocol => Effect.provideService(makeRpcClient, RpcClient_.Protocol, protocol)),
    idleTimeToLive: "3 minutes"
  });
  return yield* make({
    ping(address) {
      return RcMap.get(clients, address).pipe(Effect.flatMap(client => client.Ping()), Effect.catchCause(() => Effect.andThen(RcMap.invalidate(clients, address), Effect.fail(new RunnerUnavailable({
        address
      })))), Effect.scoped);
    },
    send({
      address,
      message
    }) {
      const rpc = message.rpc;
      const isPersisted = Context.get(rpc.annotations, Persisted);
      if (message._tag === "OutgoingEnvelope") {
        return RcMap.get(clients, address).pipe(Effect.flatMap(client => client.Envelope({
          envelope: message.envelope,
          persisted: isPersisted
        })), Effect.catchTag("RpcClientError", Effect.die), Effect.scoped, Effect.catchDefect(() => Effect.fail(new RunnerUnavailable({
          address
        }))));
      }
      const isStream = RpcSchema.isStreamSchema(rpc.successSchema);
      if (!isStream) {
        return Effect.matchEffect(Message.serializeRequest(message), {
          onSuccess: request => RcMap.get(clients, address).pipe(Effect.flatMap(client => client.Effect({
            request,
            persisted: isPersisted
          })), Effect.catchTag("RpcClientError", Effect.die), Effect.flatMap(reply => Schema.decodeEffect(Reply.Reply(message.rpc))(reply).pipe(Effect.provideContext(message.context), Effect.orDie)), Effect.flatMap(message.respond), Effect.scoped, Effect.catchDefect(() => Effect.fail(new RunnerUnavailable({
            address
          })))),
          onFailure: error => message.respond(new Reply.WithExit({
            id: snowflakeGen.nextUnsafe(),
            requestId: message.envelope.requestId,
            exit: Exit.die(error)
          }))
        });
      }
      return Effect.matchEffect(Message.serializeRequest(message), {
        onSuccess: request => RcMap.get(clients, address).pipe(Effect.flatMap(client => client.Stream({
          request,
          persisted: isPersisted
        }, {
          asQueue: true
        })), Effect.flatMap(queue => {
          const decode = Schema.decodeEffect(Reply.Reply(message.rpc));
          return Queue.take(queue).pipe(Effect.flatMap(reply => Effect.orDie(decode(reply))), Effect.flatMap(message.respond), Effect.forever, Effect.catchTag("RpcClientError", Effect.die), Effect.provideContext(message.context), Effect.catchTag("Done", _ => Effect.void), Effect.catchDefect(() => Effect.fail(new RunnerUnavailable({
            address
          }))));
        }), Effect.scoped),
        onFailure: error => message.respond(new Reply.WithExit({
          id: snowflakeGen.nextUnsafe(),
          requestId: message.envelope.requestId,
          exit: Exit.die(error)
        }))
      });
    },
    notify({
      address,
      message
    }) {
      if (Option.isNone(address)) {
        return Effect.void;
      }
      const envelope = message.envelope;
      const encode = message._tag === "OutgoingRequest" ? Effect.orDie(Message.serializeRequest(message)) : Effect.succeed(envelope);
      return Effect.flatMap(encode, envelope => RcMap.get(clients, address.value).pipe(Effect.flatMap(client => client.Notify({
        envelope
      })), Effect.scoped, Effect.ignore));
    },
    onRunnerUnavailable: address => RcMap.invalidate(clients, address)
  });
});
/**
 * Layer that provides an RPC-backed `Runners` service using `RpcClientProtocol`,
 * message storage, sharding configuration, and the default snowflake generator.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerRpc = /*#__PURE__*/Layer.effect(Runners, makeRpc).pipe(/*#__PURE__*/Layer.provide(Snowflake.layerGenerator));
/**
 * Service that creates an RPC client protocol for communicating with a runner at a
 * given address.
 *
 * @category client
 * @since 4.0.0
 */
export class RpcClientProtocol extends /*#__PURE__*/Context.Service()("effect/cluster/Runners/RpcClientProtocol") {}
//# sourceMappingURL=Runners.js.map