1/**
2 * Handles communication between Effect Cluster runners.
3 *
4 * `Runners` sits between sharding decisions and runner execution. It can ping a
5 * runner, send requests or control envelopes, notify a runner that persisted
6 * work is available, and record that a runner address is unavailable. This
7 * module defines the runner communication service, its RPC protocol, no-op and
8 * RPC-backed implementations, local persistence support, reply recovery, and
9 * the protocol service used by transport-specific runner layers.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import * as Exit from "../Exit.ts"
17import * as Latch from "../Latch.ts"
18import * as Layer from "../Layer.ts"
19import * as Option from "../Option.ts"
20import * as Queue from "../Queue.ts"
21import * as RcMap from "../RcMap.ts"
22import * as Rpc from "../rpc/Rpc.ts"
23import * as RpcClient_ from "../rpc/RpcClient.ts"
24import type { RpcClientError } from "../rpc/RpcClientError.ts"
25import * as RpcGroup from "../rpc/RpcGroup.ts"
26import * as RpcSchema from "../rpc/RpcSchema.ts"
27import type * as RpcSerialization from "../rpc/RpcSerialization.ts"
28import * as Schema from "../Schema.ts"
29import type { Scope } from "../Scope.ts"
30import type { MalformedMessage, PersistenceError } from "./ClusterError.ts"
31import { AlreadyProcessingMessage, EntityNotAssignedToRunner, MailboxFull, RunnerUnavailable } from "./ClusterError.ts"
32import { Persisted } from "./ClusterSchema.ts"
33import * as Envelope from "./Envelope.ts"
34import * as Message from "./Message.ts"
35import * as MessageStorage from "./MessageStorage.ts"
36import * as Reply from "./Reply.ts"
37import type { RunnerAddress } from "./RunnerAddress.ts"
38import { ShardingConfig } from "./ShardingConfig.ts"
39import * as Snowflake from "./Snowflake.ts"
40
41/**
42 * Service for communicating with cluster runners, including pinging runners,
43 * sending and notifying messages, coordinating persisted replies, and marking
44 * runners unavailable.
45 *
46 * @stability unstable
47 * @category services
48 * @since 4.0.0
49 */
50export class Runners extends Context.Service<Runners, {
51 /**
52 * Checks whether a Runner is responsive.
53 */
54 readonly ping: (address: RunnerAddress) => Effect.Effect<void, RunnerUnavailable>
55
56 /**
57 * Send a message locally.
58 *
59 * This ensures that the message hits storage before being sent to the local
60 * entity.
61 */
62 readonly sendLocal: <R extends Rpc.Any>(
63 options: {
64 readonly message: Message.Outgoing<R>
65 readonly send: <Rpc extends Rpc.Any>(
66 message: Message.IncomingLocal<Rpc>
67 ) => Effect.Effect<
68 void,
69 EntityNotAssignedToRunner | MailboxFull | AlreadyProcessingMessage
70 >
71 readonly simulateRemoteSerialization: boolean
72 }
73 ) => Effect.Effect<
74 void,
75 EntityNotAssignedToRunner | MailboxFull | AlreadyProcessingMessage | PersistenceError
76 >
77
78 /**
79 * Send a message to a Runner.
80 */
81 readonly send: <R extends Rpc.Any>(
82 options: {
83 readonly address: RunnerAddress
84 readonly message: Message.Outgoing<R>
85 }
86 ) => Effect.Effect<
87 void,
88 | EntityNotAssignedToRunner
89 | RunnerUnavailable
90 | MailboxFull
91 | AlreadyProcessingMessage
92 | PersistenceError
93 >
94
95 /**
96 * Notify a Runner that a message is available. Persisted messages recover
97 * replies from storage, while volatile messages complete after delivery.
98 */
99 readonly notify: <R extends Rpc.Any>(
100 options: {
101 readonly address: Option.Option<RunnerAddress>
102 readonly message: Message.Outgoing<R>
103 readonly discard: boolean
104 }
105 ) => Effect.Effect<
106 void,
107 | EntityNotAssignedToRunner
108 | RunnerUnavailable
109 | MailboxFull
110 | AlreadyProcessingMessage
111 | PersistenceError
112 >
113
114 /**
115 * Notify the current Runner that a message is available, then read replies from
116 * storage.
117 *
118 * This ensures that the message hits storage before being sent to the local
119 * entity.
120 */
121 readonly notifyLocal: <R extends Rpc.Any>(
122 options: {
123 readonly message: Message.Outgoing<R>
124 readonly notify: (
125 options: Message.IncomingLocal<any>
126 ) => Effect.Effect<void, EntityNotAssignedToRunner>
127 readonly discard: boolean
128 readonly storageOnly?: boolean | undefined
129 }
130 ) => Effect.Effect<void, PersistenceError>
131
132 /**
133 * Mark a Runner as unavailable.
134 */
135 readonly onRunnerUnavailable: (address: RunnerAddress) => Effect.Effect<void>
136}>()("effect/cluster/Runners") {}
137
138/**
139 * Builds the `Runners` service from remote runner callbacks and adds local
140 * message persistence, duplicate request handling, optional local serialization
141 * simulation, and polling for persisted replies.
142 *
143 * **When to use**
144 *
145 * Use when you need a custom `Runners` service around remote `ping`, `send`,
146 * `notify`, and `onRunnerUnavailable` callbacks, with standard local
147 * persistence and reply recovery behavior.
148 *
149 * **Details**
150 *
151 * `make` uses the supplied remote callbacks for runner communication and
152 * derives `sendLocal` and `notifyLocal`. Local sends can optionally simulate
153 * remote serialization with the supplied transport codec, persisted notifications are saved through
154 * `MessageStorage`, duplicate requests are resumed from stored replies when
155 * possible, and pending replies are polled according to
156 * `ShardingConfig.entityReplyPollInterval`.
157 *
158 * @see {@link makeRpc} for the RPC-backed implementation built on top of this constructor
159 * @see {@link makeNoop} for a no-op implementation when remote runner communication is not needed
160 *
161 * @stability unstable
162 * @category constructors
163 * @since 4.0.0
164 */
165export const make: (
166 options: Omit<Runners["Service"], "sendLocal" | "notifyLocal"> & {
167 readonly codecFor: RpcSerialization.CodecFor
168 }
169) => Effect.Effect<
170 Runners["Service"],
171 never,
172 MessageStorage.MessageStorage | Snowflake.Generator | ShardingConfig | Scope
173> = Effect.fnUntraced(function*(options) {
174 const storage = yield* MessageStorage.MessageStorage
175 const runnersScope = yield* Effect.scope
176 const snowflakeGen = yield* Snowflake.Generator
177 const config = yield* ShardingConfig
178 const { codecFor, ...serviceOptions } = options
179
180 const requestIdRewrites = new Map<Snowflake.Snowflake, Snowflake.Snowflake>()
181
182 function notifyWith<E>(
183 message: Message.Outgoing<any>,
184 afterPersist: (message: Message.Outgoing<any>, isDuplicate: boolean) => Effect.Effect<void, E>
185 ): Effect.Effect<void, E | PersistenceError> {
186 const rpc = message.rpc as any as Rpc.AnyWithProps
187 const persisted = Context.get(rpc.annotations, Persisted)
188 if (!persisted) {
189 return afterPersist(message, false)
190 }
191
192 if (message._tag === "OutgoingEnvelope") {
193 const rewriteId = requestIdRewrites.get(message.envelope.requestId)
194 const requestId = rewriteId ?? message.envelope.requestId
195 const entry = storageRequests.get(requestId)
196 if (rewriteId) {
197 message = new Message.OutgoingEnvelope({
198 ...message,
199 envelope: message.envelope.withRequestId(rewriteId)
200 })
201 }
202 return storage.saveEnvelope(message).pipe(
203 Effect.catchTag("MalformedMessage", Effect.die),
204 Effect.andThen(
205 entry ? Effect.andThen(entry.latch.open, afterPersist(message, false)) : afterPersist(message, false)
206 )
207 )
208 }
209
210 // For requests, after persisting the request, we need to check if the
211 // request is a duplicate. If it is, we need to resume from the last
212 // received reply.
213 //
214 // Otherwise, we notify the remote entity and then reply from storage.
215 return Effect.flatMap(
216 Effect.catchTag(storage.saveRequest(message), "MalformedMessage", Effect.die),
217 MessageStorage.SaveResult.$match({
218 Success: () => afterPersist(message, false),
219 Duplicate: ({ lastReceivedReply, originalId }) => {
220 // If the last received reply is an exit, we can just return it
221 // as the response.
222 if (Option.isSome(lastReceivedReply) && lastReceivedReply.value._tag === "WithExit") {
223 return message.respond(lastReceivedReply.value.withRequestId(message.envelope.requestId))
224 }
225 requestIdRewrites.set(message.envelope.requestId, originalId)
226 return afterPersist(
227 new Message.OutgoingRequest({
228 ...message,
229 lastReceivedReply,
230 envelope: Envelope.makeRequest({
231 ...message.envelope,
232 requestId: originalId
233 }),
234 respond(reply) {
235 if (reply._tag === "WithExit") {
236 requestIdRewrites.delete(message.envelope.requestId)
237 }
238 return message.respond(reply.withRequestId(message.envelope.requestId))
239 }
240 }),
241 true
242 )
243 }
244 })
245 )
246 }
247
248 type StorageRequestEntry = {
249 readonly latch: Latch.Latch
250 doneLatch: Latch.Latch | undefined
251 readonly messages: Set<Message.OutgoingRequest<any>>
252 replies: Array<Reply.Reply<any>>
253 }
254 const storageRequests = new Map<Snowflake.Snowflake, StorageRequestEntry>()
255 const waitingStorageRequests = new Map<Snowflake.Snowflake, Message.OutgoingRequest<any>>()
256 const replyFromStorage = Effect.fnUntraced(
257 function*(message: Message.OutgoingRequest<any>) {
258 let entry = storageRequests.get(message.envelope.requestId)
259 if (entry) {
260 entry.messages.add(message)
261 entry.doneLatch ??= Latch.makeUnsafe(false)
262 return yield* entry.doneLatch.await
263 } else {
264 entry = {
265 latch: Latch.makeUnsafe(false),
266 doneLatch: undefined,
267 replies: [],
268 messages: new Set([message])
269 }
270 storageRequests.set(message.envelope.requestId, entry)
271 }
272
273 while (true) {
274 // wait for the storage loop to notify us
275 entry.latch.closeUnsafe()
276 waitingStorageRequests.set(message.envelope.requestId, message)
277 storageLatch.openUnsafe()
278 yield* entry.latch.await
279
280 // send the replies back
281 for (let i = 0; i < entry.replies.length; i++) {
282 const reply = entry.replies[i]
283 // we have reached the end
284 if (reply._tag === "WithExit") {
285 for (const message of entry.messages) {
286 yield* message.respond(reply)
287 }
288 entry.doneLatch?.openUnsafe()
289 return
290 }
291
292 entry.latch.closeUnsafe()
293 for (const message of entry.messages) {
294 yield* message.respond(reply)
295 }
296 // wait for ack
297 yield* entry.latch.await
298 }
299 entry.replies = []
300 }
301 },
302 (effect, message) =>
303 Effect.ensuring(
304 effect,
305 Effect.sync(() => {
306 const entry = storageRequests.get(message.envelope.requestId)
307 if (!entry || entry.messages.size > 1) {
308 entry?.messages.delete(message)
309 return
310 }
311 storageRequests.delete(message.envelope.requestId)
312 waitingStorageRequests.delete(message.envelope.requestId)
313 })
314 )
315 )
316
317 const storageLatch = Latch.makeUnsafe(false)
318 if (storage !== MessageStorage.noop) {
319 yield* Effect.gen(function*() {
320 const foundRequests = new Set<StorageRequestEntry>()
321
322 while (true) {
323 yield* storageLatch.await
324 storageLatch.closeUnsafe()
325
326 const replies = yield* storage.repliesFor(waitingStorageRequests.values()).pipe(
327 Effect.catchCause((cause) =>
328 Effect.as(
329 Effect.annotateLogs(Effect.logDebug(cause), {
330 package: "@effect/cluster",
331 module: "Runners",
332 fiber: "Read replies loop"
333 }),
334 []
335 )
336 )
337 )
338
339 // put the replies into the storage requests and then open the latches
340 for (let i = 0; i < replies.length; i++) {
341 const reply = replies[i]
342 const entry = storageRequests.get(reply.requestId)
343 if (!entry) continue
344 entry.replies.push(reply)
345 waitingStorageRequests.delete(reply.requestId)
346 foundRequests.add(entry)
347 }
348
349 foundRequests.forEach((entry) => entry.latch.openUnsafe())
350 foundRequests.clear()
351 }
352 }).pipe(
353 Effect.forkIn(runnersScope)
354 )
355
356 yield* Effect.suspend(() => {
357 if (waitingStorageRequests.size === 0) {
358 return storageLatch.await
359 }
360 return storageLatch.open
361 }).pipe(
362 Effect.delay(config.entityReplyPollInterval),
363 Effect.forever,
364 Effect.forkIn(runnersScope)
365 )
366 }
367
368 return Runners.of({
369 ...serviceOptions,
370 sendLocal(options) {
371 const message = options.message
372 if (!options.simulateRemoteSerialization) {
373 return options.send(Message.incomingLocalFromOutgoing(message))
374 }
375 return Message.serialize(message, codecFor).pipe(
376 Effect.flatMap((encoded) => Message.deserializeLocal(message, encoded, codecFor)),
377 Effect.flatMap(options.send),
378 Effect.catchTag("MalformedMessage", (error) => {
379 if (message._tag === "OutgoingEnvelope") {
380 return Effect.die(error)
381 }
382 return message.respond(
383 new Reply.WithExit({
384 id: snowflakeGen.nextUnsafe(),
385 requestId: message.envelope.requestId,
386 exit: Exit.die(error)
387 })
388 )
389 })
390 )
391 },
392 notify(options_) {
393 const { discard, message } = options_
394 return notifyWith(message, (message, duplicate) => {
395 if (discard || message._tag === "OutgoingEnvelope") {
396 return options.notify(options_)
397 } else if (!duplicate && Option.isSome(options_.address)) {
398 return Effect.catch(
399 options.send({
400 address: options_.address.value,
401 message
402 }),
403 (_) => replyFromStorage(message)
404 )
405 }
406 return options.notify(options_).pipe(
407 Effect.andThen(replyFromStorage(message))
408 )
409 })
410 },
411 notifyLocal(options) {
412 return notifyWith(options.message, (message, duplicate) => {
413 if (options.discard || message._tag === "OutgoingEnvelope") {
414 return Effect.catchTag(
415 options.notify(Message.incomingLocalFromOutgoing(message)),
416 "EntityNotAssignedToRunner",
417 () => Effect.void
418 )
419 } else if (!duplicate && options.storageOnly !== true) {
420 return options.notify(Message.incomingLocalFromOutgoing(message)).pipe(
421 Effect.andThen(storage.registerReplyHandler(message)),
422 Effect.catchTag("EntityNotAssignedToRunner", () => replyFromStorage(message))
423 )
424 }
425 return options.notify(Message.incomingLocalFromOutgoing(message)).pipe(
426 Effect.catchTag("EntityNotAssignedToRunner", () => Effect.void),
427 Effect.andThen(replyFromStorage(message))
428 )
429 })
430 }
431 })
432})
433
434/**
435 * Creates a no-op `Runners` service that rejects sends with
436 * `EntityNotAssignedToRunner` and ignores notifications, pings, and unavailable
437 * runner reports.
438 *
439 * @stability unstable
440 * @category constructors
441 * @since 4.0.0
442 */
443export const makeNoop: Effect.Effect<
444 Runners["Service"],
445 never,
446 MessageStorage.MessageStorage | Snowflake.Generator | ShardingConfig | Scope
447> = make({
448 codecFor: Schema.toCodecJson as RpcSerialization.CodecFor,
449 send: ({ message }) => Effect.fail(new EntityNotAssignedToRunner({ address: message.envelope.address })),
450 notify: () => Effect.void,
451 ping: () => Effect.void,
452 onRunnerUnavailable: () => Effect.void
453})
454
455/**
456 * Layer that provides the no-op `Runners` service, using the default snowflake
457 * generator.
458 *
459 * @stability unstable
460 * @category layers
461 * @since 4.0.0
462 */
463export const layerNoop: Layer.Layer<
464 Runners,
465 never,
466 ShardingConfig | MessageStorage.MessageStorage
467> = Layer.effect(Runners, makeNoop).pipe(Layer.provide([Snowflake.layerGenerator]))
468
469const rpcErrors: Schema.Union<[
470 typeof EntityNotAssignedToRunner,
471 typeof MailboxFull,
472 typeof AlreadyProcessingMessage
473]> = Schema.Union([
474 EntityNotAssignedToRunner,
475 MailboxFull,
476 AlreadyProcessingMessage
477])
478
479/**
480 * RPC group used for runner-to-runner communication, including ping, notify,
481 * effect, stream, and envelope messages.
482 *
483 * @stability unstable
484 * @category models
485 * @since 4.0.0
486 */
487export class Rpcs extends RpcGroup.make(
488 Rpc.make("Ping"),
489 Rpc.make("Notify", {
490 payload: {
491 envelope: Envelope.Partial,
492 persisted: Schema.Boolean
493 },
494 success: Schema.Void,
495 error: rpcErrors
496 }),
497 Rpc.make("Effect", {
498 payload: {
499 request: Envelope.PartialRequest,
500 persisted: Schema.Boolean
501 },
502 success: Reply.Encoded,
503 error: rpcErrors
504 }),
505 Rpc.make("Stream", {
506 payload: {
507 request: Envelope.PartialRequest,
508 persisted: Schema.Boolean
509 },
510 error: rpcErrors,
511 success: Reply.Encoded,
512 stream: true
513 }),
514 Rpc.make("Envelope", {
515 payload: {
516 envelope: Schema.Union([Envelope.AckChunk, Envelope.Interrupt]),
517 persisted: Schema.Boolean
518 },
519 error: rpcErrors
520 })
521) {}
522
523/**
524 * Client interface generated from the runner RPC group.
525 *
526 * @stability unstable
527 * @category models
528 * @since 4.0.0
529 */
530export interface RpcClient extends RpcClient_.FromGroup<typeof Rpcs, RpcClientError> {}
531
532/**
533 * Builds a runner RPC client from the current `RpcClient.Protocol`, using the
534 * `Runners` span prefix with tracing disabled.
535 *
536 * @stability unstable
537 * @category constructors
538 * @since 4.0.0
539 */
540export const makeRpcClient: Effect.Effect<
541 RpcClient,
542 never,
543 RpcClient_.Protocol | Scope
544> = RpcClient_.make(Rpcs, { spanPrefix: "Runners", disableTracing: true })
545
546/**
547 * Builds a `Runners` service backed by RPC clients, caching a client per runner
548 * address and dispatching ping, notify, effect, stream, and envelope messages over
549 * the runner protocol.
550 *
551 * @stability unstable
552 * @category constructors
553 * @since 4.0.0
554 */
555export const makeRpc: Effect.Effect<
556 Runners["Service"],
557 never,
558 Scope | RpcClientProtocol | MessageStorage.MessageStorage | Snowflake.Generator | ShardingConfig
559> = Effect.gen(function*() {
560 const clientProtocol = yield* RpcClientProtocol
561 const snowflakeGen = yield* Snowflake.Generator
562
563 const clients = yield* RcMap.make({
564 lookup: (address: RunnerAddress) =>
565 Effect.flatMap(
566 clientProtocol.make(address),
567 (protocol) =>
568 Effect.map(
569 Effect.provideService(makeRpcClient, RpcClient_.Protocol, protocol),
570 (client) => ({ client, codecFor: protocol.codecFor })
571 )
572 ),
573 idleTimeToLive: "3 minutes"
574 })
575
576 return yield* make({
577 codecFor: clientProtocol.codecFor,
578 ping(address) {
579 return RcMap.get(clients, address).pipe(
580 Effect.flatMap(({ client }) => client.Ping()),
581 Effect.catchCause(() =>
582 Effect.andThen(
583 RcMap.invalidate(clients, address),
584 Effect.fail(new RunnerUnavailable({ address }))
585 )
586 ),
587 Effect.scoped
588 )
589 },
590 send({ address, message }) {
591 const rpc = message.rpc as any as Rpc.AnyWithProps
592 const isPersisted = Context.get(rpc.annotations, Persisted)
593 if (message._tag === "OutgoingEnvelope") {
594 return RcMap.get(clients, address).pipe(
595 Effect.flatMap(({ client }) =>
596 client.Envelope({
597 envelope: message.envelope,
598 persisted: isPersisted
599 })
600 ),
601 Effect.scoped,
602 Effect.catchTag("RpcClientError", () => Effect.fail(new RunnerUnavailable({ address })))
603 )
604 }
605 // Persisted requests can recover their reply from storage via the
606 // `RunnerUnavailable` path, volatile requests receive the defect as their reply.
607 const respondDefect = (defect: unknown) =>
608 isPersisted
609 ? Effect.fail(new RunnerUnavailable({ address }))
610 : message.respond(
611 new Reply.WithExit({
612 id: snowflakeGen.nextUnsafe(),
613 requestId: message.envelope.requestId,
614 exit: Exit.die(defect)
615 })
616 )
617 const respondMalformed = (error: MalformedMessage) =>
618 message.respond(
619 new Reply.WithExit({
620 id: snowflakeGen.nextUnsafe(),
621 requestId: message.envelope.requestId,
622 exit: Exit.die(error)
623 })
624 )
625 const isStream = RpcSchema.isStreamSchema(rpc.successSchema)
626 if (!isStream) {
627 return RcMap.get(clients, address).pipe(
628 Effect.flatMap(({ client, codecFor }) =>
629 Effect.matchEffect(Message.serializeRequest(message, codecFor), {
630 onSuccess: (request) =>
631 client.Effect({
632 request,
633 persisted: isPersisted
634 }).pipe(
635 Effect.flatMap((reply) =>
636 Schema.decodeEffect(Reply.Reply(message.rpc, codecFor))(reply).pipe(
637 Effect.provideContext(message.context),
638 Effect.orDie
639 )
640 ),
641 Effect.flatMap(message.respond),
642 Effect.catchTag("RpcClientError", () => Effect.fail(new RunnerUnavailable({ address }))),
643 Effect.catchDefect(respondDefect)
644 ),
645 onFailure: respondMalformed
646 })
647 ),
648 Effect.scoped
649 )
650 }
651 return RcMap.get(clients, address).pipe(
652 Effect.flatMap(({ client, codecFor }) =>
653 Effect.matchEffect(Message.serializeRequest(message, codecFor), {
654 onSuccess: (request) =>
655 client.Stream({
656 request,
657 persisted: isPersisted
658 }, { asQueue: true }).pipe(
659 Effect.flatMap((queue) => {
660 const decode = Schema.decodeEffect(Reply.Reply(message.rpc, codecFor))
661 return Queue.take(queue).pipe(
662 Effect.flatMap((reply) => Effect.orDie(decode(reply))),
663 Effect.flatMap(message.respond),
664 Effect.forever,
665 Effect.provideContext(message.context),
666 Effect.catchTag("Done", (_) => Effect.void),
667 Effect.catchTag("RpcClientError", () => Effect.fail(new RunnerUnavailable({ address }))),
668 Effect.catchDefect(respondDefect)
669 )
670 })
671 ),
672 onFailure: respondMalformed
673 })
674 ),
675 Effect.scoped
676 )
677 },
678 notify({ address, message }) {
679 if (Option.isNone(address)) {
680 return Effect.void
681 }
682 const rpc = message.rpc as any as Rpc.AnyWithProps
683 const isPersisted = Context.get(rpc.annotations, Persisted)
684 const envelope = message.envelope
685 const notify = RcMap.get(clients, address.value).pipe(
686 Effect.flatMap(({ client, codecFor }) => {
687 const encode: Effect.Effect<Envelope.AckChunk | Envelope.Interrupt | Envelope.PartialRequest> =
688 message._tag === "OutgoingRequest"
689 ? Effect.orDie(Message.serializeRequest(message, codecFor))
690 : Effect.succeed(envelope)
691 return Effect.flatMap(encode, (envelope) =>
692 client.Notify({
693 envelope,
694 persisted: isPersisted
695 }))
696 }),
697 Effect.scoped,
698 Effect.catchTag("RpcClientError", () => Effect.fail(new RunnerUnavailable({ address: address.value })))
699 )
700 return isPersisted ? Effect.ignore(notify) : notify
701 },
702 onRunnerUnavailable: (address) => RcMap.invalidate(clients, address)
703 })
704})
705
706/**
707 * Layer that provides an RPC-backed `Runners` service using `RpcClientProtocol`,
708 * message storage, sharding configuration, and the default snowflake generator.
709 *
710 * @stability unstable
711 * @category layers
712 * @since 4.0.0
713 */
714export const layerRpc: Layer.Layer<
715 Runners,
716 never,
717 MessageStorage.MessageStorage | RpcClientProtocol | ShardingConfig
718> = Layer.effect(Runners, makeRpc).pipe(
719 Layer.provide(Snowflake.layerGenerator)
720)
721
722/**
723 * Service that creates RPC client protocols for runner addresses and exposes
724 * the codec shared by those protocols.
725 *
726 * @stability unstable
727 * @category services
728 * @since 4.0.0
729 */
730export class RpcClientProtocol extends Context.Service<
731 RpcClientProtocol,
732 {
733 readonly make: (address: RunnerAddress) => Effect.Effect<RpcClient_.Protocol["Service"], never, Scope>
734 readonly codecFor: RpcSerialization.CodecFor
735 }
736>()("effect/cluster/Runners/RpcClientProtocol") {}