1/**
2 * Runs cluster runner RPCs over a socket transport.
3 *
4 * The full layer serves runner RPC handlers on a provided `SocketServer`, logs
5 * the bound address, and provides `Sharding` and `Runners` clients when an
6 * outgoing runner client protocol is available. This module also includes a
7 * client-only socket runner layer for processes that need cluster clients
8 * without starting a runner server or receiving shard assignments.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Effect from "../Effect.ts"
14import * as Layer from "../Layer.ts"
15import * as NetAddress from "../net/NetAddress.ts"
16import type * as RpcSerialization from "../rpc/RpcSerialization.ts"
17import * as RpcServer from "../rpc/RpcServer.ts"
18import { SocketServer } from "../socket/SocketServer.ts"
19import type { MessageStorage } from "./MessageStorage.ts"
20import type { RunnerHealth } from "./RunnerHealth.ts"
21import type * as Runners from "./Runners.ts"
22import * as RunnerServer from "./RunnerServer.ts"
23import type * as RunnerStorage from "./RunnerStorage.ts"
24import type * as Sharding from "./Sharding.ts"
25import type { ShardingConfig } from "./ShardingConfig.ts"
26
27const withLogAddress = <A, E, R>(layer: Layer.Layer<A, E, R>): Layer.Layer<A, E, R | SocketServer> =>
28 Layer.effectDiscard(Effect.gen(function*() {
29 const server = yield* SocketServer
30 const address = server.address._tag === "UnixPathAddress"
31 ? server.address.path
32 : NetAddress.formatInet(server.address)
33 yield* Effect.annotateLogs(Effect.logInfo(`Listening on: ${address}`), {
34 package: "@effect/cluster",
35 service: "Runner"
36 })
37 })).pipe(Layer.provideMerge(layer))
38
39/**
40 * Layer that runs a cluster runner over the socket RPC protocol, providing
41 * `Sharding` and `Runners` clients and logging the socket listen address.
42 *
43 * **When to use**
44 *
45 * Use when a cluster runner process should accept runner RPCs through a
46 * provided `SocketServer` and receive shard assignments while exposing
47 * `Sharding` and `Runners` services.
48 *
49 * **Details**
50 *
51 * It logs the bound `SocketServer.address` when the layer starts, formatting internet
52 * addresses as numeric socket addresses and Unix socket addresses as their filesystem
53 * path.
54 *
55 * **Gotchas**
56 *
57 * Although this layer serves runner RPCs with the provided `SocketServer`,
58 * outgoing calls to other runners still require a `Runners.RpcClientProtocol`
59 * service.
60 *
61 * @see {@link layerClientOnly} for the socket runner layer that only provides clients and does not receive shard assignments
62 * @see {@link Runners.RpcClientProtocol} for the outgoing runner client protocol required by this layer
63 *
64 * @stability unstable
65 * @category layers
66 * @since 4.0.0
67 */
68export const layer: Layer.Layer<
69 Sharding.Sharding | Runners.Runners,
70 never,
71 | Runners.RpcClientProtocol
72 | ShardingConfig
73 | RpcSerialization.RpcSerialization
74 | SocketServer
75 | MessageStorage
76 | RunnerStorage.RunnerStorage
77 | RunnerHealth
78> = RunnerServer.layerWithClients.pipe(
79 withLogAddress,
80 Layer.provide(RpcServer.layerProtocolSocketServer)
81)
82
83/**
84 * Provides a client-only socket runner layer that provides `Sharding` and `Runners` clients
85 * without starting a runner server or receiving shard assignments.
86 *
87 * **When to use**
88 *
89 * Use to join a socket-based cluster as a client-only participant that can send
90 * messages without hosting shards.
91 *
92 * @stability unstable
93 * @category layers
94 * @since 4.0.0
95 */
96export const layerClientOnly: Layer.Layer<
97 Sharding.Sharding | Runners.Runners,
98 never,
99 Runners.RpcClientProtocol | ShardingConfig | MessageStorage | RunnerStorage.RunnerStorage
100> = RunnerServer.layerClientOnly