1/**
2 * Configures how an Effect Cluster runner participates in sharding.
3 *
4 * `ShardingConfig` describes the runner address, shard group membership, shard
5 * counts and weights, lock timing, entity mailbox and lifecycle limits, polling
6 * intervals, health checks, and local serialization simulation. This module
7 * includes the service, default values, programmatic and environment-based
8 * layers, a `Config` description for loading values, and helpers for normalizing
9 * assigned shard groups.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Config from "../Config.ts"
15import * as ConfigProvider from "../ConfigProvider.ts"
16import * as Context from "../Context.ts"
17import * as Duration from "../Duration.ts"
18import * as Effect from "../Effect.ts"
19import * as Layer from "../Layer.ts"
20import * as Option from "../Option.ts"
21import * as Schema from "../Schema.ts"
22import { RunnerAddress } from "./RunnerAddress.ts"
23
24/**
25 * Represents the configuration for the `Sharding` service on a given runner.
26 *
27 * @stability unstable
28 * @category services
29 * @since 4.0.0
30 */
31export class ShardingConfig extends Context.Service<ShardingConfig, {
32 /**
33 * The address for the current runner that other runners can use to
34 * communicate with it.
35 *
36 * If `None`, the runner is not part of the cluster and will be in a client-only
37 * mode.
38 */
39 readonly runnerAddress: Option.Option<RunnerAddress>
40 /**
41 * The listen address for the current runner.
42 *
43 * Defaults to the `runnerAddress`.
44 */
45 readonly runnerListenAddress: Option.Option<RunnerAddress>
46 /**
47 * A number that determines how many shards this runner will be assigned
48 * relative to other runners.
49 *
50 * Defaults to `1`.
51 *
52 * A value of `2` means that this runner should be assigned twice as many
53 * shards as a runner with a weight of `1`.
54 */
55 readonly runnerShardWeight: number
56 /**
57 * The shard groups available across all runners.
58 *
59 * Defaults to `["default"]`.
60 */
61 readonly availableShardGroups: ReadonlyArray<string>
62 /**
63 * The shard groups that are assigned to this runner.
64 *
65 * Defaults to `["default"]`.
66 */
67 readonly assignedShardGroups: ReadonlyArray<string>
68 /**
69 * The number of shards to allocate per shard group.
70 *
71 * **Note**: this value should be consistent across all runners.
72 */
73 readonly shardsPerGroup: number
74 /**
75 * The maximum interval between shard lock refreshes.
76 *
77 * The runner may shorten this interval to one third of
78 * `shardLockExpiration` to preserve enough time to stop entities safely if
79 * lock storage becomes unavailable.
80 */
81 readonly shardLockRefreshInterval: Duration.Input
82 /**
83 * Shard lock expiration duration.
84 */
85 readonly shardLockExpiration: Duration.Input
86 /**
87 * Disable the use of advisory locks for shard locking.
88 */
89 readonly shardLockDisableAdvisory: boolean
90 /**
91 * Start shutting down as soon as an Entity has started shutting down.
92 *
93 * Defaults to `true`.
94 */
95 readonly preemptiveShutdown: boolean
96 /**
97 * The default capacity of the mailbox for entities.
98 */
99 readonly entityMailboxCapacity: number | "unbounded"
100 /**
101 * The maximum number of entities that can be resident on this runner at the
102 * same time.
103 *
104 * When the limit is reached, no new entities are spawned: the storage read
105 * loop stops admitting messages for new entity addresses (they stay in
106 * storage until a slot frees up), and volatile sends to new addresses fail
107 * with `MailboxFull`.
108 *
109 * `"unbounded"` can only be set programmatically; the environment
110 * configuration only accepts integers.
111 *
112 * Defaults to `10_000`.
113 */
114 readonly maxResidentEntities: number | "unbounded"
115 /**
116 * The maximum number of unprocessed messages read from storage in a single
117 * poll.
118 *
119 * Defaults to `1024`.
120 */
121 readonly unprocessedMessageBatchSize: number
122 /**
123 * The maximum duration of inactivity (i.e. without receiving a message)
124 * after which an entity will be interrupted.
125 */
126 readonly entityMaxIdleTime: Duration.Input
127 /**
128 * If an entity does not register itself within this time after a message is
129 * sent to it, the message will be marked as failed.
130 *
131 * Defaults to 1 minute.
132 */
133 readonly entityRegistrationTimeout: Duration.Input
134 /**
135 * The maximum duration of time to wait for an entity to terminate.
136 *
137 * By default this is set to 15 seconds to stay within kubernetes defaults.
138 */
139 readonly entityTerminationTimeout: Duration.Input
140 /**
141 * The interval at which to poll for unprocessed messages from storage.
142 */
143 readonly entityMessagePollInterval: Duration.Input
144 /**
145 * The interval at which to poll for client replies from storage.
146 */
147 readonly entityReplyPollInterval: Duration.Input
148 /**
149 * The interval at which to poll for new runners and refresh shard
150 * assignments.
151 */
152 readonly refreshAssignmentsInterval: Duration.Input
153 /**
154 * The interval to retry a send if EntityNotAssignedToRunner is returned.
155 */
156 readonly sendRetryInterval: Duration.Input
157 /**
158 * The interval at which to check for unhealthy runners and report them
159 */
160 readonly runnerHealthCheckInterval: Duration.Input
161 /**
162 * Simulate serialization and deserialization to remote runners for local
163 * entities.
164 */
165 readonly simulateRemoteSerialization: boolean
166}>()("effect/cluster/ShardingConfig") {}
167
168const defaultRunnerAddress = RunnerAddress.make({ host: "localhost", port: 34431 })
169
170/**
171 * Default values for `ShardingConfig`, including the default local runner address,
172 * shard group, shard count, mailbox settings, polling intervals, and remote
173 * serialization simulation.
174 *
175 * @stability unstable
176 * @category defaults
177 * @since 4.0.0
178 */
179export const defaults: ShardingConfig["Service"] = {
180 runnerAddress: Option.some(defaultRunnerAddress),
181 runnerListenAddress: Option.none(),
182 runnerShardWeight: 1,
183 shardsPerGroup: 300,
184 availableShardGroups: ["default"],
185 assignedShardGroups: ["default"],
186 preemptiveShutdown: true,
187 shardLockRefreshInterval: Duration.seconds(10),
188 shardLockExpiration: Duration.seconds(35),
189 shardLockDisableAdvisory: false,
190 entityMailboxCapacity: 4096,
191 maxResidentEntities: 10_000,
192 unprocessedMessageBatchSize: 1024,
193 entityMaxIdleTime: Duration.minutes(1),
194 entityRegistrationTimeout: Duration.minutes(1),
195 entityTerminationTimeout: Duration.seconds(15),
196 entityMessagePollInterval: Duration.seconds(10),
197 entityReplyPollInterval: Duration.millis(200),
198 sendRetryInterval: Duration.millis(100),
199 refreshAssignmentsInterval: Duration.seconds(3),
200 runnerHealthCheckInterval: Duration.minutes(1),
201 simulateRemoteSerialization: true
202}
203
204/**
205 * Creates a `ShardingConfig` layer by merging the provided partial options over
206 * `defaults`.
207 *
208 * **When to use**
209 *
210 * Use when you need to wire a cluster runner with explicit `ShardingConfig`
211 * values, especially in tests, local development, or code paths where
212 * configuration should be provided programmatically instead of loaded from
213 * environment variables.
214 *
215 * **Details**
216 *
217 * The merge is shallow: omitted fields use `defaults`, and provided fields
218 * replace the corresponding default value.
219 *
220 * **Gotchas**
221 *
222 * This layer only merges and provides configuration; it does not check that
223 * cluster-wide settings are consistent across runners. Keep values such as
224 * `shardsPerGroup` and `availableShardGroups` aligned for runners that should
225 * share shard assignments.
226 *
227 * @see {@link defaults} for the values used when an option is omitted
228 * @see {@link layerDefaults} for a layer with no overrides
229 * @see {@link layerFromEnv} for loading configuration from environment variables before applying explicit overrides
230 *
231 * @stability unstable
232 * @category layers
233 * @since 4.0.0
234 */
235export const layer = (options?: Partial<ShardingConfig["Service"]>): Layer.Layer<ShardingConfig> =>
236 Layer.succeed(ShardingConfig)({ ...defaults, ...options })
237
238/**
239 * Layer that provides the default `ShardingConfig` values.
240 *
241 * @stability unstable
242 * @category layers
243 * @since 4.0.0
244 */
245export const layerDefaults: Layer.Layer<ShardingConfig> = layer()
246
247/**
248 * Describes how to load `ShardingConfig` values, applying the same
249 * defaults used by the in-memory `defaults` object.
250 *
251 * @stability unstable
252 * @category configuration
253 * @since 4.0.0
254 */
255export const config: Config.Config<ShardingConfig["Service"]> = Config.all({
256 runnerAddress: Config.all({
257 host: Config.String("host").pipe(
258 Config.withDefault(defaultRunnerAddress.host)
259 // Config.withDescription("The hostname or IP address of the runner.")
260 ),
261 port: Config.Int("port").pipe(
262 Config.withDefault(defaultRunnerAddress.port)
263 // Config.withDescription("The port used for inter-runner communication.")
264 )
265 }).pipe(Config.map((options) => RunnerAddress.make(options)), Config.option),
266 runnerListenAddress: Config.all({
267 host: Config.String("listenHost"),
268 // Config.withDescription("The host to listen on.")
269 port: Config.Int("listenPort").pipe(
270 Config.withDefault(defaultRunnerAddress.port)
271 // Config.withDescription("The port to listen on.")
272 )
273 }).pipe(Config.map((options) => RunnerAddress.make(options)), Config.option),
274 runnerShardWeight: Config.Int("runnerShardWeight").pipe(
275 Config.withDefault(defaults.runnerShardWeight)
276 // Config.withDescription("A number that determines how many shards this runner will be assigned relative to other runners.")
277 ),
278 availableShardGroups: Config.Array(Schema.String, "availableShardGroups").pipe(
279 Config.withDefault(["default"])
280 // Config.withDescription("The shard groups available across all runners.")
281 ),
282 assignedShardGroups: Config.Array(Schema.String, "shardGroups").pipe(
283 Config.withDefault(["default"])
284 // Config.withDescription("The shard groups that are assigned to this runner.")
285 ),
286 shardsPerGroup: Config.Int("shardsPerGroup").pipe(
287 Config.withDefault(defaults.shardsPerGroup)
288 // Config.withDescription("The number of shards to allocate per shard group.")
289 ),
290 shardLockRefreshInterval: Config.Duration("shardLockRefreshInterval").pipe(
291 Config.withDefault(defaults.shardLockRefreshInterval)
292 // Config.withDescription("Shard lock refresh interval.")
293 ),
294 shardLockExpiration: Config.Duration("shardLockExpiration").pipe(
295 Config.withDefault(defaults.shardLockExpiration)
296 // Config.withDescription("Shard lock expiration duration.")
297 ),
298 shardLockDisableAdvisory: Config.Boolean("shardLockDisableAdvisory").pipe(
299 Config.withDefault(defaults.shardLockDisableAdvisory)
300 // Config.withDescription("Disable the use of advisory locks for shard locking.")
301 ),
302 preemptiveShutdown: Config.Boolean("preemptiveShutdown").pipe(
303 Config.withDefault(defaults.preemptiveShutdown)
304 // Config.withDescription("Start shutting down as soon as an Entity has started shutting down.")
305 ),
306 entityMailboxCapacity: Config.Int("entityMailboxCapacity").pipe(
307 Config.withDefault(defaults.entityMailboxCapacity)
308 // Config.withDescription("The default capacity of the mailbox for entities.")
309 ),
310 maxResidentEntities: Config.schema(
311 Schema.Int.check(Schema.isGreaterThan(0)),
312 "maxResidentEntities"
313 ).pipe(
314 Config.withDefault(defaults.maxResidentEntities)
315 // Config.withDescription("The maximum number of entities that can be resident on this runner at the same time.")
316 ),
317 unprocessedMessageBatchSize: Config.schema(
318 Schema.Int.check(Schema.isGreaterThan(0)),
319 "unprocessedMessageBatchSize"
320 ).pipe(
321 Config.withDefault(defaults.unprocessedMessageBatchSize)
322 // Config.withDescription("The maximum number of unprocessed messages read from storage in a single poll.")
323 ),
324 entityMaxIdleTime: Config.Duration("entityMaxIdleTime").pipe(
325 Config.withDefault(defaults.entityMaxIdleTime)
326 // Config.withDescription(
327 // "The maximum duration of inactivity (i.e. without receiving a message) after which an entity will be interrupted."
328 // )
329 ),
330 entityRegistrationTimeout: Config.Duration("entityRegistrationTimeout").pipe(
331 Config.withDefault(defaults.entityRegistrationTimeout)
332 // Config.withDescription("If an entity does not register itself within this time after a message is sent to it, the message will be marked as failed.")
333 ),
334 entityTerminationTimeout: Config.Duration("entityTerminationTimeout").pipe(
335 Config.withDefault(defaults.entityTerminationTimeout)
336 // Config.withDescription("The maximum duration of time to wait for an entity to terminate.")
337 ),
338 entityMessagePollInterval: Config.Duration("entityMessagePollInterval").pipe(
339 Config.withDefault(defaults.entityMessagePollInterval)
340 // Config.withDescription("The interval at which to poll for unprocessed messages from storage.")
341 ),
342 entityReplyPollInterval: Config.Duration("entityReplyPollInterval").pipe(
343 Config.withDefault(defaults.entityReplyPollInterval)
344 // Config.withDescription("The interval at which to poll for client replies from storage.")
345 ),
346 sendRetryInterval: Config.Duration("sendRetryInterval").pipe(
347 Config.withDefault(defaults.sendRetryInterval)
348 // Config.withDescription("The interval to retry a send if EntityNotManagedByRunner is returned.")
349 ),
350 refreshAssignmentsInterval: Config.Duration("refreshAssignmentsInterval").pipe(
351 Config.withDefault(defaults.refreshAssignmentsInterval)
352 // Config.withDescription("The interval at which to refresh shard assignments.")
353 ),
354 runnerHealthCheckInterval: Config.Duration("runnerHealthCheckInterval").pipe(
355 Config.withDefault(defaults.runnerHealthCheckInterval)
356 // Config.withDescription("The interval at which to check for unhealthy runners and report them.")
357 ),
358 // unhealthyRunnerReportInterval: Config.Duration("unhealthyRunnerReportInterval").pipe(
359 simulateRemoteSerialization: Config.Boolean("simulateRemoteSerialization").pipe(
360 Config.withDefault(defaults.simulateRemoteSerialization)
361 // Config.withDescription("Simulate serialization and deserialization to remote runners for local entities.")
362 )
363})
364
365/**
366 * Effect that loads `ShardingConfig` from environment variables using the
367 * constant-case config provider.
368 *
369 * @stability unstable
370 * @category configuration
371 * @since 4.0.0
372 */
373export const configFromEnv = config.pipe(
374 Effect.provideService(
375 ConfigProvider.ConfigProvider,
376 ConfigProvider.fromEnv().pipe(
377 ConfigProvider.constantCase
378 )
379 )
380)
381
382/**
383 * Layer that loads `ShardingConfig` from environment variables and, when options
384 * are provided, overlays those options on top of the loaded values.
385 *
386 * @stability unstable
387 * @category layers
388 * @since 4.0.0
389 */
390export const layerFromEnv = (options?: Partial<ShardingConfig["Service"]> | undefined): Layer.Layer<
391 ShardingConfig,
392 Config.ConfigError
393> =>
394 Layer.effect(ShardingConfig)(
395 options ? Effect.map(configFromEnv, (config) => ({ ...config, ...options })) : configFromEnv
396 )
397
398/**
399 * Normalizes the provided `ShardingConfig` to calculate the `available` and
400 * `assigned` shard groups.
401 *
402 * @stability unstable
403 * @category converting
404 * @since 4.0.0
405 */
406export const shardGroupConfig = (config: ShardingConfig["Service"]): {
407 readonly available: ReadonlySet<string>
408 readonly assigned: ReadonlySet<string>
409} => {
410 const available = new Set(config.availableShardGroups.slice().sort())
411 const assigned = new Set<string>()
412 available.forEach((group) => {
413 if (config.assignedShardGroups.includes(group)) {
414 assigned.add(group)
415 }
416 })
417 return { available, assigned }
418}