1/**
2 * Provides server-side layers for the cluster runner protocol.
3 *
4 * Runner protocol handlers receive ping, notification, request, stream, and
5 * envelope messages from other runners. They forward those messages into
6 * `Sharding` and coordinate persisted replies through `MessageStorage`. This
7 * module includes the handler layer, a transport-independent RPC server layer, a
8 * full server layer that also provides runner clients, and a client-only layer
9 * for applications that do not serve runner RPCs.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import type * as Cause from "../Cause.ts"
15import * as Effect from "../Effect.ts"
16import type * as Exit from "../Exit.ts"
17import * as Fiber from "../Fiber.ts"
18import { constant } from "../Function.ts"
19import * as Layer from "../Layer.ts"
20import * as Option from "../Option.ts"
21import * as Queue from "../Queue.ts"
22import type * as Rpc from "../rpc/Rpc.ts"
23import * as RpcServer from "../rpc/RpcServer.ts"
24import type * as ClusterError from "./ClusterError.ts"
25import * as Message from "./Message.ts"
26import * as MessageStorage from "./MessageStorage.ts"
27import * as Reply from "./Reply.ts"
28import * as RunnerHealth from "./RunnerHealth.ts"
29import * as Runners from "./Runners.ts"
30import type * as RunnerStorage from "./RunnerStorage.ts"
31import * as Sharding from "./Sharding.ts"
32import { ShardingConfig } from "./ShardingConfig.ts"
33
34const constVoid = constant(Effect.void)
35
36const serializeDefectReply = <R extends Rpc.Any>(
37 reply: Reply.ReplyWithContext<R>,
38 defect: unknown,
39 codecFor: RpcServer.Protocol["Service"]["codecFor"]
40): Effect.Effect<Reply.Encoded> =>
41 Effect.orDie(Reply.serialize(
42 Reply.ReplyWithContext.fromDefect({
43 id: reply.reply.id,
44 requestId: reply.reply.requestId,
45 defect
46 }),
47 codecFor
48 ))
49
50/**
51 * Layer that handles runner protocol RPCs by forwarding requests to `Sharding`
52 * and `MessageStorage`.
53 *
54 * @stability unstable
55 * @category layers
56 * @since 4.0.0
57 */
58export const layerHandlers = Runners.Rpcs.toLayer(Effect.gen(function*() {
59 const sharding = yield* Sharding.Sharding
60 const storage = yield* MessageStorage.MessageStorage
61 const { codecFor } = yield* RpcServer.Protocol
62
63 return {
64 Ping: () => Effect.void,
65 Notify: ({ envelope, persisted }) => {
66 const message = envelope._tag === "Request"
67 ? new Message.IncomingRequest({
68 envelope,
69 respond: constVoid,
70 lastSentReply: Option.none(),
71 codecFor
72 })
73 : new Message.IncomingEnvelope({ envelope })
74 return persisted ? sharding.notify(message) : sharding.send(message)
75 },
76 Effect: Effect.fnUntraced(function*({ persisted, request }) {
77 const callerScope = yield* Effect.scope
78 let replyEncoded: Option.Option<Effect.Effect<Reply.Encoded, ClusterError.EntityNotAssignedToRunner>> = Option
79 .none()
80 let resume = (reply: Effect.Effect<Reply.Encoded, ClusterError.EntityNotAssignedToRunner>) => {
81 replyEncoded = Option.some(reply)
82 }
83 const message = new Message.IncomingRequest({
84 envelope: request,
85 lastSentReply: Option.none(),
86 codecFor,
87 callerScope,
88 respond(reply) {
89 resume(Reply.serializeOrDefect(reply, codecFor))
90 return Effect.void
91 }
92 })
93 if (persisted) {
94 return yield* Effect.callback<
95 Reply.Encoded,
96 ClusterError.EntityNotAssignedToRunner
97 >((resume_) => {
98 resume = resume_
99 const parent = Fiber.getCurrent()!
100 const onExit = (
101 exit: Exit.Exit<
102 any,
103 ClusterError.EntityNotAssignedToRunner
104 >
105 ) => {
106 if (exit._tag === "Failure") {
107 resume(exit as any)
108 }
109 }
110 const runFork = Effect.runForkWith(parent.context)
111 const fiber = runFork(storage.registerReplyHandler(message))
112 fiber.addObserver(onExit)
113 runFork(Effect.catchTag(
114 sharding.notify(message, constWaitUntilRead),
115 "AlreadyProcessingMessage",
116 () => Effect.void
117 )).addObserver(onExit)
118 return Fiber.interrupt(fiber)
119 })
120 }
121 yield* sharding.send(message)
122 return yield* Effect.callback<Reply.Encoded, ClusterError.EntityNotAssignedToRunner>((resume_) => {
123 if (Option.isSome(replyEncoded)) {
124 resume_(replyEncoded.value)
125 } else {
126 resume = resume_
127 }
128 })
129 }),
130 Stream: Effect.fnUntraced(function*({ persisted, request }) {
131 const callerScope = yield* Effect.scope
132 const queue = yield* Queue.make<Reply.Encoded, ClusterError.EntityNotAssignedToRunner | Cause.Done>()
133 const message = new Message.IncomingRequest({
134 envelope: request,
135 lastSentReply: Option.none(),
136 codecFor,
137 callerScope,
138 respond(reply) {
139 return Reply.serialize(reply, codecFor).pipe(
140 Effect.flatMap((reply) => {
141 Queue.offerUnsafe(queue, reply)
142 if (reply._tag === "WithExit") {
143 Queue.endUnsafe(queue)
144 }
145 return Effect.void
146 }),
147 Effect.catchTag(
148 "MalformedMessage",
149 (error) =>
150 Effect.flatMap(serializeDefectReply(reply, error, codecFor), (reply) => {
151 // the fallback defect reply is terminal, so end the stream
152 Queue.offerUnsafe(queue, reply)
153 Queue.endUnsafe(queue)
154 return Effect.void
155 })
156 )
157 )
158 }
159 })
160 if (persisted) {
161 yield* storage.registerReplyHandler(message).pipe(
162 Effect.onError((cause) => Queue.failCause(queue, cause)),
163 Effect.forkScoped
164 )
165 yield* sharding.notify(message, constWaitUntilRead)
166 } else {
167 yield* sharding.send(message)
168 }
169 return queue
170 }),
171 Envelope: ({ envelope }) => sharding.send(new Message.IncomingEnvelope({ envelope }))
172 }
173}))
174
175const constWaitUntilRead = { waitUntilRead: true } as const
176
177/**
178 * Creates the runner RPC server layer, which receives messages from other
179 * runners, forwards them to the `Sharding` layer, and responds to `Ping`
180 * requests.
181 *
182 * **When to use**
183 *
184 * Use when a runner process should accept runner-to-runner protocol messages
185 * over a provided server `RpcServer.Protocol`.
186 *
187 * **Gotchas**
188 *
189 * This layer does not choose or provide the wire transport; provide a
190 * transport-specific `RpcServer.Protocol` separately.
191 *
192 * @see {@link layerHandlers} for the lower-level handler layer used when the RPC server is supplied elsewhere
193 * @see {@link layerWithClients} for a runner server layer that also provides the `Sharding` and `Runners` clients
194 * @see {@link layerClientOnly} for embedding a cluster client without serving runner RPCs
195 *
196 * @stability unstable
197 * @category layers
198 * @since 4.0.0
199 */
200export const layer: Layer.Layer<
201 never,
202 never,
203 | RpcServer.Protocol
204 | Sharding.Sharding
205 | MessageStorage.MessageStorage
206> = RpcServer.layer(Runners.Rpcs, {
207 spanPrefix: "RunnerServer",
208 disableTracing: true,
209 disableFatalDefects: true
210}).pipe(Layer.provide(layerHandlers))
211
212/**
213 * Layer that provides `RunnerServer` together with `Runners` and `Sharding`
214 * clients.
215 *
216 * @stability unstable
217 * @category layers
218 * @since 4.0.0
219 */
220export const layerWithClients: Layer.Layer<
221 Sharding.Sharding | Runners.Runners,
222 never,
223 | RpcServer.Protocol
224 | ShardingConfig
225 | Runners.RpcClientProtocol
226 | MessageStorage.MessageStorage
227 | RunnerStorage.RunnerStorage
228 | RunnerHealth.RunnerHealth
229> = layer.pipe(
230 Layer.provideMerge(Sharding.layer),
231 Layer.provideMerge(Runners.layerRpc)
232)
233
234/**
235 * Creates a client-only `Runners` layer.
236 *
237 * **When to use**
238 *
239 * Use to embed a cluster client inside another Effect application without registering with
240 * the ShardManager or receiving shard assignments.
241 *
242 * @stability unstable
243 * @category layers
244 * @since 4.0.0
245 */
246export const layerClientOnly: Layer.Layer<
247 Sharding.Sharding | Runners.Runners,
248 never,
249 | ShardingConfig
250 | Runners.RpcClientProtocol
251 | MessageStorage.MessageStorage
252 | RunnerStorage.RunnerStorage
253> = Sharding.layer.pipe(
254 Layer.provideMerge(Runners.layerRpc),
255 Layer.provide(RunnerHealth.layerNoop),
256 Layer.updateService(ShardingConfig, (config) => ({
257 ...config,
258 runnerAddress: Option.none()
259 }))
260)