1/**
2 * Stores runner registration and shard-lock state for cluster sharding.
3 *
4 * `RunnerStorage` records which runners are registered, whether they are
5 * healthy, which machine id a runner receives, and which shard locks are held
6 * by each runner. This module includes the typed storage service, a
7 * string-encoded backend interface, an adapter from encoded storage to the typed
8 * service, and an in-memory implementation for tests and local use.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import { isArrayNonEmpty, type NonEmptyArray } from "../Array.ts"
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import * as Layer from "../Layer.ts"
17import * as MutableHashMap from "../MutableHashMap.ts"
18import type { PersistenceError } from "./ClusterError.ts"
19import * as MachineId from "./MachineId.ts"
20import { Runner } from "./Runner.ts"
21import type { RunnerAddress } from "./RunnerAddress.ts"
22import * as ShardId from "./ShardId.ts"
23
24/**
25 * Represents a generic interface to the persistent storage required by the
26 * cluster.
27 *
28 * @stability unstable
29 * @category services
30 * @since 4.0.0
31 */
32export class RunnerStorage extends Context.Service<RunnerStorage, {
33 /**
34 * Register a new runner with the cluster.
35 */
36 readonly register: (runner: Runner, healthy: boolean) => Effect.Effect<MachineId.MachineId, PersistenceError>
37
38 /**
39 * Unregister the runner with the given address.
40 */
41 readonly unregister: (address: RunnerAddress) => Effect.Effect<void, PersistenceError>
42
43 /**
44 * Get all runners registered with the cluster.
45 */
46 readonly getRunners: Effect.Effect<Array<readonly [runner: Runner, healthy: boolean]>, PersistenceError>
47
48 /**
49 * Set the health status of the given runner.
50 */
51 readonly setRunnerHealth: (address: RunnerAddress, healthy: boolean) => Effect.Effect<void, PersistenceError>
52
53 /**
54 * Try to acquire the given shard ids for processing.
55 *
56 * It returns an array of shards it was able to acquire.
57 */
58 readonly acquire: (
59 address: RunnerAddress,
60 shardIds: Iterable<ShardId.ShardId>
61 ) => Effect.Effect<Array<ShardId.ShardId>, PersistenceError>
62
63 /**
64 * Refresh the locks owned by the given runner.
65 */
66 readonly refresh: (
67 address: RunnerAddress,
68 shardIds: Iterable<ShardId.ShardId>
69 ) => Effect.Effect<Array<ShardId.ShardId>, PersistenceError>
70
71 /**
72 * Release the given shard ids.
73 */
74 readonly release: (
75 address: RunnerAddress,
76 shardId: ShardId.ShardId
77 ) => Effect.Effect<void, PersistenceError>
78
79 /**
80 * Release all the shards assigned to the given runner.
81 */
82 readonly releaseAll: (address: RunnerAddress) => Effect.Effect<void, PersistenceError>
83}>()("effect/cluster/RunnerStorage") {}
84
85/**
86 * String-encoded runner storage interface used by adapters that persist runner
87 * addresses, runners, machine ids, and shard ids outside the in-memory model.
88 *
89 * @stability unstable
90 * @category services
91 * @since 4.0.0
92 */
93export interface Encoded {
94 /**
95 * Get all runners registered with the cluster.
96 */
97 readonly getRunners: Effect.Effect<Array<readonly [runner: string, healthy: boolean]>, PersistenceError>
98
99 /**
100 * Register a new runner with the cluster.
101 */
102 readonly register: (address: string, runner: string, healthy: boolean) => Effect.Effect<number, PersistenceError>
103
104 /**
105 * Unregister the runner with the given address.
106 */
107 readonly unregister: (address: string) => Effect.Effect<void, PersistenceError>
108
109 /**
110 * Set the health status of the given runner.
111 */
112 readonly setRunnerHealth: (address: string, healthy: boolean) => Effect.Effect<void, PersistenceError>
113
114 /**
115 * Acquire the lock on the given shards, returning the shards that were
116 * successfully locked.
117 */
118 readonly acquire: (
119 address: string,
120 shardIds: NonEmptyArray<string>
121 ) => Effect.Effect<Array<string>, PersistenceError>
122
123 /**
124 * Refresh the lock on the given shards, returning the shards that were
125 * successfully locked.
126 */
127 readonly refresh: (
128 address: string,
129 shardIds: Array<string>
130 ) => Effect.Effect<ReadonlyArray<string>, PersistenceError>
131
132 /**
133 * Release the lock on the given shard.
134 */
135 readonly release: (
136 address: string,
137 shardId: string
138 ) => Effect.Effect<void, PersistenceError>
139
140 /**
141 * Release the lock on all shards for the given runner.
142 */
143 readonly releaseAll: (address: string) => Effect.Effect<void, PersistenceError>
144}
145
146/**
147 * Adapts an encoded runner storage implementation into `RunnerStorage`, converting
148 * runner addresses, runners, machine ids, and shard ids between typed values and
149 * their string or numeric storage forms.
150 *
151 * @stability unstable
152 * @category layers
153 * @since 4.0.0
154 */
155export const makeEncoded = (encoded: Encoded) =>
156 RunnerStorage.of({
157 getRunners: Effect.gen(function*() {
158 const runners = yield* encoded.getRunners
159 const results: Array<[Runner, boolean]> = []
160 for (let i = 0; i < runners.length; i++) {
161 const [runner, healthy] = runners[i]
162 // @effect-diagnostics-next-line tryCatchInEffectGen:off
163 try {
164 results.push([Runner.decodeSync(runner), healthy])
165 } catch {
166 //
167 }
168 }
169 return results
170 }),
171 register: (runner, healthy) =>
172 Effect.map(
173 encoded.register(encodeRunnerAddress(runner.address), Runner.encodeSync(runner), healthy),
174 MachineId.make
175 ),
176 unregister: (address) => encoded.unregister(encodeRunnerAddress(address)),
177 setRunnerHealth: (address, healthy) => encoded.setRunnerHealth(encodeRunnerAddress(address), healthy),
178 acquire: (address, shardIds) => {
179 const arr = Array.from(shardIds, (id) => id.toString())
180 if (!isArrayNonEmpty(arr)) return Effect.succeed([])
181 return encoded.acquire(encodeRunnerAddress(address), arr).pipe(
182 Effect.map((shards) => shards.map(ShardId.fromString))
183 )
184 },
185 refresh: (address, shardIds) =>
186 encoded.refresh(encodeRunnerAddress(address), Array.from(shardIds, (id) => id.toString())).pipe(
187 Effect.map((shards) => shards.map(ShardId.fromString))
188 ),
189 release(address, shardId) {
190 return encoded.release(encodeRunnerAddress(address), shardId.toString())
191 },
192 releaseAll(address) {
193 return encoded.releaseAll(encodeRunnerAddress(address))
194 }
195 })
196
197/**
198 * Creates an in-memory `RunnerStorage` implementation for tests and local use.
199 *
200 * **Details**
201 *
202 * Runner health and shard acquisition are kept only in process memory.
203 *
204 * @stability unstable
205 * @category constructors
206 * @since 4.0.0
207 */
208export const makeMemory = Effect.gen(function*() {
209 const runners = MutableHashMap.empty<RunnerAddress, readonly [runner: Runner, healthy: boolean]>()
210 let acquired: Array<ShardId.ShardId> = []
211 let id = 0
212
213 return RunnerStorage.of({
214 getRunners: Effect.sync(() => Array.from(MutableHashMap.values(runners))),
215 register: (runner, healthy) =>
216 Effect.sync(() => {
217 MutableHashMap.set(runners, runner.address, [runner, healthy])
218 return MachineId.make(id++)
219 }),
220 unregister: (address) =>
221 Effect.sync(() => {
222 MutableHashMap.remove(runners, address)
223 }),
224 setRunnerHealth: (address, healthy) =>
225 Effect.sync(() => {
226 MutableHashMap.modify(runners, address, ([runner]) => [runner, healthy] as const)
227 }),
228 acquire: (_address, shardIds) => {
229 acquired = Array.from(shardIds)
230 return Effect.succeed(acquired)
231 },
232 refresh: () => Effect.sync(() => acquired),
233 release: () => Effect.void,
234 releaseAll: () => Effect.void
235 })
236})
237
238/**
239 * Layer that provides the in-memory `RunnerStorage` implementation.
240 *
241 * @stability unstable
242 * @category layers
243 * @since 4.0.0
244 */
245export const layerMemory: Layer.Layer<RunnerStorage> = Layer.effect(RunnerStorage)(makeMemory)
246
247// -------------------------------------------------------------------------------------
248// internal
249// -------------------------------------------------------------------------------------
250
251const encodeRunnerAddress = (runnerAddress: RunnerAddress) => `${runnerAddress.host}:${runnerAddress.port}`