1/**
2 * Defines addressable entity types for Effect Cluster.
3 *
4 * An entity gives a stable name and RPC protocol to a group of values that are
5 * addressed by id. The cluster uses that information to choose a shard and
6 * route each request to the runner responsible for that id. This module
7 * includes constructors for entity definitions, helpers for creating sharded
8 * clients, layer builders for registering handlers, and services that expose
9 * the current entity address while a request is being handled.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Arr from "../Array.ts"
15import type * as Cause from "../Cause.ts"
16import * as Context from "../Context.ts"
17import * as Data from "../Data.ts"
18import type * as Duration from "../Duration.ts"
19import * as Effect from "../Effect.ts"
20import * as Equal from "../Equal.ts"
21import * as Exit from "../Exit.ts"
22import { identity } from "../Function.ts"
23import * as Hash from "../Hash.ts"
24import * as Headers from "../http/Headers.ts"
25import type * as Latch from "../Latch.ts"
26import * as Layer from "../Layer.ts"
27import * as Option from "../Option.ts"
28import * as Predicate from "../Predicate.ts"
29import * as Queue from "../Queue.ts"
30import * as Rpc from "../rpc/Rpc.ts"
31import * as RpcClient from "../rpc/RpcClient.ts"
32import * as RpcGroup from "../rpc/RpcGroup.ts"
33import * as RpcSchema from "../rpc/RpcSchema.ts"
34import * as RpcServer from "../rpc/RpcServer.ts"
35import type * as Schedule from "../Schedule.ts"
36import { Scope } from "../Scope.ts"
37import * as Stream from "../Stream.ts"
38import type {
39 AlreadyProcessingMessage,
40 EntityNotAssignedToRunner,
41 MailboxFull,
42 PersistenceError
43} from "./ClusterError.ts"
44import { Persisted, ShardGroup, Uninterruptible } from "./ClusterSchema.ts"
45import { EntityAddress } from "./EntityAddress.ts"
46import type { EntityId } from "./EntityId.ts"
47import { EntityType } from "./EntityType.ts"
48import * as Envelope from "./Envelope.ts"
49import { CurrentActivationScope } from "./internal/entityActivation.ts"
50import { hashString } from "./internal/hash.ts"
51import { ResourceMap } from "./internal/resourceMap.ts"
52import * as Message from "./Message.ts"
53import type * as Reply from "./Reply.ts"
54import { RunnerAddress } from "./RunnerAddress.ts"
55import * as ShardId from "./ShardId.ts"
56import type { Sharding } from "./Sharding.ts"
57import { ShardingConfig } from "./ShardingConfig.ts"
58import * as Snowflake from "./Snowflake.ts"
59
60const TypeId = "~effect/cluster/Entity"
61
62/**
63 * Represents a cluster entity type and the RPC protocol it can handle.
64 *
65 * **Details**
66 *
67 * An entity defines how ids map to shard groups, exposes a sharded client, and
68 * can be registered as a layer using RPC handlers or a mailbox queue.
69 *
70 * @stability unstable
71 * @category models
72 * @since 4.0.0
73 */
74export interface Entity<
75 in out Type extends string,
76 in out Rpcs extends Rpc.Any
77> extends Equal.Equal {
78 readonly [TypeId]: typeof TypeId
79 /**
80 * The name of the entity type.
81 */
82 readonly type: EntityType
83
84 /**
85 * A RpcGroup definition for messages which represents the messaging protocol
86 * that the entity is capable of processing.
87 */
88 readonly protocol: RpcGroup.RpcGroup<Rpcs>
89
90 /**
91 * Get the shard group for the given EntityId.
92 */
93 getShardGroup(entityId: EntityId): string
94
95 /**
96 * Get the ShardId for the given EntityId.
97 */
98 getShardId(entityId: EntityId): Effect.Effect<ShardId.ShardId, never, Sharding>
99
100 /**
101 * Annotate the entity with a value.
102 */
103 annotate<I, S>(key: Context.Key<I, S>, value: S): Entity<Type, Rpcs>
104
105 /**
106 * Annotate the Rpc's above this point with a value.
107 */
108 annotateRpcs<I, S>(key: Context.Key<I, S>, value: S): Entity<Type, Rpcs>
109
110 /**
111 * Annotate the entity with the given annotations.
112 */
113 annotateMerge<S>(annotation: Context.Context<S>): Entity<Type, Rpcs>
114
115 /**
116 * Annotate the Rpc's above this point with a context object.
117 */
118 annotateRpcsMerge<S>(context: Context.Context<S>): Entity<Type, Rpcs>
119
120 /**
121 * Create a client for this entity.
122 */
123 readonly client: Effect.Effect<
124 (
125 entityId: string
126 ) => RpcClient.RpcClient.From<
127 Rpcs,
128 MailboxFull | AlreadyProcessingMessage | PersistenceError | EntityNotAssignedToRunner
129 >,
130 never,
131 Sharding
132 >
133
134 /**
135 * Create a Layer from an Entity.
136 *
137 * **Details**
138 *
139 * It will register the entity with the Sharding service.
140 */
141 toLayer<
142 Handlers extends HandlersFrom<Rpcs>,
143 RX = never
144 >(
145 build: Handlers | Effect.Effect<Handlers, never, RX>,
146 options?: {
147 readonly maxIdleTime?: Duration.Input | undefined
148 readonly concurrency?: number | "unbounded" | undefined
149 readonly mailboxCapacity?: number | "unbounded" | undefined
150 readonly disableFatalDefects?: boolean | undefined
151 readonly defectRetryPolicy?: Schedule.Schedule<any, unknown> | undefined
152 readonly spanAttributes?: Record<string, string> | undefined
153 }
154 ): Layer.Layer<
155 never,
156 never,
157 | Exclude<RX, Scope | CurrentAddress | CurrentRunnerAddress>
158 | RpcGroup.HandlersServices<Rpcs, Handlers>
159 | Rpc.ServicesClient<Rpcs>
160 | Rpc.ServicesServer<Rpcs>
161 | Rpc.Middleware<Rpcs>
162 | Sharding
163 >
164
165 of<Handlers extends HandlersFrom<Rpcs>>(handlers: Handlers): Handlers
166
167 /**
168 * Create a Layer from an Entity.
169 *
170 * **Details**
171 *
172 * It will register the entity with the Sharding service.
173 */
174 toLayerQueue<
175 R,
176 RX = never
177 >(
178 build:
179 | ((
180 queue: Queue.Dequeue<Envelope.Request<Rpcs>>,
181 replier: Replier<Rpcs>
182 ) => Effect.Effect<never, never, R>)
183 | Effect.Effect<
184 (
185 queue: Queue.Dequeue<Envelope.Request<Rpcs>>,
186 replier: Replier<Rpcs>
187 ) => Effect.Effect<never, never, R>,
188 never,
189 RX
190 >,
191 options?: {
192 readonly maxIdleTime?: Duration.Input | undefined
193 readonly mailboxCapacity?: number | "unbounded" | undefined
194 readonly disableFatalDefects?: boolean | undefined
195 readonly defectRetryPolicy?: Schedule.Schedule<any, unknown> | undefined
196 readonly spanAttributes?: Record<string, string> | undefined
197 }
198 ): Layer.Layer<
199 never,
200 never,
201 | Exclude<RX, Scope | CurrentAddress | CurrentRunnerAddress>
202 | R
203 | Rpc.ServicesClient<Rpcs>
204 | Rpc.ServicesServer<Rpcs>
205 | Rpc.Middleware<Rpcs>
206 | Sharding
207 >
208}
209/**
210 * Type alias for any cluster `Entity`, regardless of entity type or RPC
211 * protocol.
212 *
213 * @stability unstable
214 * @category models
215 * @since 4.0.0
216 */
217export type Any = Entity<string, Rpc.Any>
218
219/**
220 * Maps each RPC in an entity protocol to the handler function expected by
221 * `Entity.toLayer`.
222 *
223 * **Details**
224 *
225 * Each handler receives the entity request envelope for that RPC and returns the
226 * RPC result or a supported RPC wrapper.
227 *
228 * @stability unstable
229 * @category utility types
230 * @since 4.0.0
231 */
232export type HandlersFrom<Rpc extends Rpc.Any> = {
233 readonly [Current in Rpc as Current["_tag"]]: (
234 envelope: Request<Current>
235 ) => Rpc.WrapperOr<Rpc.ResultFrom<Current, any>>
236}
237
238/**
239 * Returns `true` when the supplied value is a cluster `Entity`.
240 *
241 * **Details**
242 *
243 * The check is based on the internal entity type identifier.
244 *
245 * @stability unstable
246 * @category guards
247 * @since 4.0.0
248 */
249export const isEntity = (u: unknown): u is Any => Predicate.hasProperty(u, TypeId)
250
251const Proto = {
252 [TypeId]: TypeId,
253 [Hash.symbol](this: Entity<string, any>): number {
254 return Hash.structure({ type: this.type })
255 },
256 [Equal.symbol](this: Entity<string, any>, that: Equal.Equal): boolean {
257 return isEntity(that) && this.type === that.type
258 },
259 annotate<I, S>(this: Entity<string, any>, key: Context.Key<I, S>, value: S) {
260 return fromRpcGroup(this.type, this.protocol.annotate(key, value))
261 },
262 annotateRpcs<I, S>(this: Entity<string, any>, key: Context.Key<I, S>, value: S) {
263 return fromRpcGroup(this.type, this.protocol.annotateRpcs(key, value))
264 },
265 annotateMerge<S>(this: Entity<string, any>, annotations: Context.Context<S>) {
266 return fromRpcGroup(this.type, this.protocol.annotateMerge(annotations))
267 },
268 annotateRpcsMerge<S>(this: Entity<string, any>, annotations: Context.Context<S>) {
269 return fromRpcGroup(this.type, this.protocol.annotateRpcsMerge(annotations))
270 },
271 getShardId(this: Entity<string, any>, entityId: EntityId) {
272 return Effect.map(shardingTag, (sharding) => sharding.getShardId(entityId, this.getShardGroup(entityId)))
273 },
274 get client() {
275 return shardingTag.pipe(
276 Effect.flatMap((sharding) => sharding.makeClient(this as any))
277 )
278 },
279 toLayer<
280 Rpcs extends Rpc.Any,
281 Handlers extends HandlersFrom<Rpcs>,
282 RX = never
283 >(
284 this: Entity<string, Rpcs>,
285 build: Handlers | Effect.Effect<Handlers, never, RX>,
286 options?: {
287 readonly maxIdleTime?: Duration.Input | undefined
288 readonly concurrency?: number | "unbounded" | undefined
289 readonly mailboxCapacity?: number | "unbounded" | undefined
290 readonly disableFatalDefects?: boolean | undefined
291 readonly defectRetryPolicy?: Schedule.Schedule<any, unknown> | undefined
292 readonly spanAttributes?: Record<string, string> | undefined
293 }
294 ): Layer.Layer<
295 never,
296 never,
297 | Exclude<RX, Scope | CurrentAddress | CurrentRunnerAddress>
298 | RpcGroup.HandlersServices<Rpcs, Handlers>
299 | Rpc.ServicesClient<Rpcs>
300 | Rpc.ServicesServer<Rpcs>
301 | Rpc.Middleware<Rpcs>
302 | Sharding
303 > {
304 return shardingTag.pipe(
305 Effect.flatMap((sharding) =>
306 sharding.registerEntity(
307 this,
308 Effect.isEffect(build) ? build : Effect.succeed(build),
309 options
310 )
311 ),
312 Layer.effectDiscard
313 )
314 },
315 of: identity,
316 toLayerQueue<
317 Rpcs extends Rpc.Any,
318 R,
319 RX = never
320 >(
321 this: Entity<string, Rpcs>,
322 build:
323 | ((
324 mailbox: Queue.Dequeue<Envelope.Request<Rpcs>>,
325 replier: Replier<Rpcs>
326 ) => Effect.Effect<never, never, R>)
327 | Effect.Effect<
328 (
329 mailbox: Queue.Dequeue<Envelope.Request<Rpcs>>,
330 replier: Replier<Rpcs>
331 ) => Effect.Effect<never, never, R>,
332 never,
333 RX
334 >,
335 options?: {
336 readonly maxIdleTime?: Duration.Input | undefined
337 readonly mailboxCapacity?: number | "unbounded" | undefined
338 readonly disableFatalDefects?: boolean | undefined
339 readonly defectRetryPolicy?: Schedule.Schedule<any, unknown> | undefined
340 readonly spanAttributes?: Record<string, string> | undefined
341 }
342 ) {
343 const buildHandlers = Effect.gen({ self: this }, function*() {
344 const behaviour = Effect.isEffect(build) ? yield* build : build
345 const queue = yield* Queue.make<Envelope.Request<Rpcs>>()
346
347 // create the rpc handlers for the entity
348 const handler = (envelope: any) =>
349 Effect.callback<any, any>((resume) => {
350 Queue.offerUnsafe(queue, envelope)
351 resumes.set(envelope, resume)
352 })
353 const streamHandler = (envelope: any) =>
354 Effect.callback<any, any>((resume) => {
355 Queue.offerUnsafe(queue, envelope)
356 resumes.set(envelope, resume)
357 }).pipe(
358 Effect.map((streamOrQueue) =>
359 Stream.isStream(streamOrQueue) ? streamOrQueue : Stream.fromQueue(streamOrQueue)
360 ),
361 Stream.unwrap
362 )
363 const handlers: Record<string, any> = Object.create(null)
364 for (const rpc_ of this.protocol.requests.values()) {
365 const rpc = rpc_ as any as Rpc.AnyWithProps
366 handlers[rpc._tag] = RpcSchema.isStreamSchema(rpc.successSchema) ? streamHandler : handler
367 }
368
369 // make the Replier for the behaviour
370 const resumes = new Map<Envelope.Request<any>, (exit: Exit.Exit<any, any>) => void>()
371 const complete = (request: Envelope.Request<any>, exit: Exit.Exit<any, any>) =>
372 Effect.sync(() => {
373 const resume = resumes.get(request)
374 if (resume) {
375 resumes.delete(request)
376 resume(exit)
377 }
378 })
379 const replier: Replier<Rpcs> = {
380 succeed: (request, value) => complete(request, Exit.succeed(value)),
381 fail: (request, error) => complete(request, Exit.fail(error)),
382 failCause: (request, cause) => complete(request, Exit.failCause(cause)),
383 complete
384 }
385
386 // fork the behaviour into the layer scope
387 yield* behaviour(queue, replier).pipe(
388 Effect.catchCause((cause) => {
389 const exit = Exit.failCause(cause)
390 for (const resume of resumes.values()) {
391 resume(exit)
392 }
393 return Effect.void
394 }),
395 Effect.interruptible,
396 Effect.forkScoped
397 )
398
399 return handlers as any
400 })
401
402 return this.toLayer(buildHandlers, {
403 ...options,
404 concurrency: "unbounded"
405 })
406 }
407}
408
409/**
410 * Creates a new `Entity` of the specified `type` which will accept messages
411 * that adhere to the provided `RpcGroup`.
412 *
413 * @stability unstable
414 * @category constructors
415 * @since 4.0.0
416 */
417export const fromRpcGroup = <const Type extends string, Rpcs extends Rpc.Any>(
418 /**
419 * The entity type name.
420 */
421 type: Type,
422 /**
423 * The schema definition for messages that the entity is capable of
424 * processing.
425 */
426 protocol: RpcGroup.RpcGroup<Rpcs>
427): Entity<Type, Rpcs> => {
428 const self = Object.create(Proto)
429 self.type = EntityType.make(type)
430 self.protocol = protocol
431 self.getShardGroup = Context.get(protocol.annotations, ShardGroup)
432 return self
433}
434
435/**
436 * Creates a new `Entity` of the specified `type` which will accept messages
437 * that adhere to the provided schemas.
438 *
439 * **When to use**
440 *
441 * Use to define a cluster entity from individual `Rpc` definitions, giving the
442 * cluster runtime a typed protocol for handlers and per-entity clients.
443 *
444 * **Details**
445 *
446 * The `type` argument is stored as the entity `EntityType`, and the RPC array
447 * is grouped into the entity's `protocol`.
448 *
449 * **Gotchas**
450 *
451 * RPC tags should be unique within the array. If multiple definitions use the
452 * same tag, the resulting protocol keeps the later definition for that tag.
453 *
454 * @see {@link fromRpcGroup} for creating an entity from an existing `RpcGroup`
455 *
456 * @stability unstable
457 * @category constructors
458 * @since 4.0.0
459 */
460export const make = <const Type extends string, Rpcs extends ReadonlyArray<Rpc.Any>>(
461 /**
462 * The entity type name.
463 */
464 type: Type,
465 /**
466 * The schema definition for messages that the entity is capable of
467 * processing.
468 */
469 protocol: Rpcs
470): Entity<Type, Rpcs[number]> => fromRpcGroup(type, RpcGroup.make(...protocol))
471
472/**
473 * Service tag for the entity address currently being processed.
474 *
475 * **When to use**
476 *
477 * Use to read the current entity identity and shard address from entity
478 * handlers and keep-alive logic.
479 *
480 * @stability unstable
481 * @category services
482 * @since 4.0.0
483 */
484export class CurrentAddress extends Context.Service<
485 CurrentAddress,
486 EntityAddress
487>()("effect/cluster/Entity/EntityAddress") {}
488
489/**
490 * Service tag for the runner address currently registering entity handlers.
491 *
492 * **When to use**
493 *
494 * Use to read the runner address associated with the current entity handler
495 * registration.
496 *
497 * @stability unstable
498 * @category services
499 * @since 4.0.0
500 */
501export class CurrentRunnerAddress extends Context.Service<
502 CurrentRunnerAddress,
503 RunnerAddress
504>()("effect/cluster/Entity/RunnerAddress") {}
505
506/**
507 * Reply API passed to queue-based entity handlers.
508 *
509 * **When to use**
510 *
511 * Use when you use it to complete an entity request by succeeding, failing, failing with a
512 * cause, or supplying an explicit `Exit`.
513 *
514 * @stability unstable
515 * @category models
516 * @since 4.0.0
517 */
518export interface Replier<Rpcs extends Rpc.Any> {
519 readonly succeed: <R extends Rpcs>(
520 request: Envelope.Request<R>,
521 value: Replier.Success<R>
522 ) => Effect.Effect<void>
523
524 readonly fail: <R extends Rpcs>(
525 request: Envelope.Request<R>,
526 error: Rpc.Error<R>
527 ) => Effect.Effect<void>
528
529 readonly failCause: <R extends Rpcs>(
530 request: Envelope.Request<R>,
531 cause: Cause.Cause<Rpc.Error<R>>
532 ) => Effect.Effect<void>
533
534 readonly complete: <R extends Rpcs>(
535 request: Envelope.Request<R>,
536 exit: Exit.Exit<Replier.Success<R>, Rpc.Error<R>>
537 ) => Effect.Effect<void>
538}
539
540/**
541 * Helper types used by the `Replier` API.
542 *
543 * @stability unstable
544 * @since 4.0.0
545 */
546export declare namespace Replier {
547 /**
548 * Success value accepted by a `Replier` for a single RPC.
549 *
550 * **Details**
551 *
552 * For streaming RPCs this may be either a stream of success chunks or a dequeue
553 * of success chunks. For non-streaming RPCs it is the RPC success value.
554 *
555 * @stability unstable
556 * @category utility types
557 * @since 4.0.0
558 */
559 export type Success<R extends Rpc.Any> = Rpc.Success<R> extends Stream.Stream<infer _A, infer _E, infer _R> ?
560 Stream.Stream<_A, _E | Rpc.Error<R>, _R> | Queue.Dequeue<_A, _E | Rpc.Error<R> | Cause.Done>
561 : Rpc.Success<R>
562}
563
564/**
565 * Represents an entity request envelope delivered to entity handlers.
566 *
567 * **Details**
568 *
569 * It includes the underlying request envelope plus the last stream reply chunk
570 * that was sent, allowing handlers to resume chunk sequencing after a restart.
571 *
572 * @stability unstable
573 * @category models
574 * @since 4.0.0
575 */
576export class Request<Rpc extends Rpc.Any> extends Data.Class<
577 Envelope.Request<Rpc> & {
578 readonly lastSentChunk: Option.Option<Reply.Chunk<Rpc>>
579 }
580> {
581 /**
582 * Most recent success chunk value sent by the entity, when one exists.
583 *
584 * @stability unstable
585 * @since 4.0.0
586 */
587 get lastSentChunkValue(): Option.Option<Rpc.SuccessChunk<Rpc>> {
588 return Option.map(this.lastSentChunk, (chunk) => Arr.lastNonEmpty(chunk.values))
589 }
590
591 /**
592 * Sequence number to use for the entity's next outgoing success chunk.
593 *
594 * @stability unstable
595 * @since 4.0.0
596 */
597 get nextSequence(): number {
598 if (Option.isNone(this.lastSentChunk)) {
599 return 0
600 }
601 return this.lastSentChunk.value.sequence + 1
602 }
603}
604
605const shardingTag = Context.Service<Sharding, Sharding["Service"]>("effect/cluster/Sharding")
606
607/**
608 * Builds an in-memory test client for an entity layer.
609 *
610 * **Details**
611 *
612 * The returned function creates a no-serialization RPC client for each entity ID,
613 * using a test sharding service instead of the cluster transport.
614 *
615 * @stability unstable
616 * @category testing
617 * @since 4.0.0
618 */
619export const makeTestClient: <Type extends string, Rpcs extends Rpc.Any, LA, LE, LR>(
620 entity: Entity<Type, Rpcs>,
621 layer: Layer.Layer<LA, LE, LR>
622) => Effect.Effect<
623 (entityId: string) => Effect.Effect<RpcClient.RpcClient<Rpcs>>,
624 LE,
625 Scope | ShardingConfig | Exclude<LR, Sharding> | Rpc.MiddlewareClient<Rpcs>
626> = Effect.fnUntraced(function*<Type extends string, Rpcs extends Rpc.Any, LA, LE, LR>(
627 entity: Entity<Type, Rpcs>,
628 layer: Layer.Layer<LA, LE, LR>
629) {
630 const config = yield* ShardingConfig
631 const makeShardId = (entityId: string) =>
632 ShardId.make(
633 entity.getShardGroup(entityId as EntityId),
634 (Math.abs(hashString(entityId) % config.shardsPerGroup)) + 1
635 )
636 const snowflakeGen = yield* Snowflake.makeGenerator
637 const runnerAddress = new RunnerAddress({ host: "localhost", port: 3000 })
638 const entityMap = new Map<string, {
639 readonly context: Context.Context<
640 Rpc.ServicesClient<Rpcs> | Rpc.ServicesServer<Rpcs> | Rpc.Middleware<Rpcs> | LR
641 >
642 readonly concurrency: number | "unbounded"
643 readonly disableFatalDefects: boolean | undefined
644 readonly build: Effect.Effect<Context.Context<Rpc.ToHandler<Rpcs>>>
645 }>()
646 const sharding = shardingTag.of({
647 ...({} as Sharding["Service"]),
648 registerEntity: (entity, handlers, options) =>
649 Effect.contextWith((context) => {
650 entityMap.set(entity.type, {
651 context: context as any,
652 concurrency: options?.concurrency ?? 1,
653 disableFatalDefects: options?.disableFatalDefects,
654 build: entity.protocol.toHandlers(handlers as any) as any
655 })
656 return Effect.void
657 })
658 })
659 yield* Layer.build(Layer.provide(layer, Layer.succeed(shardingTag)(sharding)))
660 const entityEntry = entityMap.get(entity.type)
661 if (!entityEntry) {
662 return yield* Effect.die(`Entity.makeTestClient: ${entity.type} was not registered by layer`)
663 }
664
665 const map = yield* ResourceMap.make(Effect.fnUntraced(function*(entityId: string) {
666 const address = new EntityAddress({
667 entityType: entity.type,
668 entityId: entityId as EntityId,
669 shardId: makeShardId(entityId)
670 })
671 const scope = yield* Effect.scope
672 const handlerContext = entityEntry.context.pipe(
673 Context.add(CurrentRunnerAddress, runnerAddress),
674 Context.add(CurrentAddress, address),
675 Context.add(CurrentActivationScope, scope),
676 Context.add(Scope, scope)
677 )
678 const handlers = yield* entityEntry.build.pipe(
679 Effect.setContext(handlerContext as Context.Context<any>)
680 )
681
682 // oxlint-disable-next-line prefer-const
683 let client!: Effect.Success<ReturnType<typeof RpcClient.makeNoSerialization<Rpcs, never>>>
684 const server = yield* RpcServer.makeNoSerialization(entity.protocol, {
685 concurrency: entityEntry.concurrency,
686 disableFatalDefects: entityEntry.disableFatalDefects,
687 onFromServer(response) {
688 return client.write(response)
689 }
690 }).pipe(
691 Effect.setContext(Context.merge(handlerContext, handlers))
692 )
693
694 client = yield* RpcClient.makeNoSerialization(entity.protocol, {
695 supportsAck: true,
696 generateRequestId: () => snowflakeGen.nextUnsafe() as any,
697 onFromClient({ message }) {
698 if (message._tag === "Request") {
699 return server.write(0, {
700 ...message,
701 payload: new Request({
702 ...message,
703 [Envelope.TypeId]: Envelope.TypeId,
704 address,
705 requestId: Snowflake.Snowflake(message.id),
706 lastSentChunk: Option.none()
707 }) as any
708 })
709 }
710 return server.write(0, message)
711 }
712 })
713 return client.client
714 }))
715
716 return (entityId: string) => map.get(entityId)
717})
718
719/**
720 * Enables or disables keep-alive for the current entity.
721 *
722 * **Details**
723 *
724 * When enabled it sends the internal keep-alive RPC for the current address; when
725 * disabled it releases the keep-alive latch if one is present.
726 *
727 * @stability unstable
728 * @category keep alive
729 * @since 4.0.0
730 */
731export const keepAlive: (
732 enabled: boolean
733) => Effect.Effect<
734 void,
735 never,
736 Sharding | CurrentAddress
737> = Effect.fnUntraced(function*(enabled: boolean) {
738 const olatch = yield* Effect.serviceOption(KeepAliveLatch)
739 if (olatch._tag === "None") return
740 if (!enabled) {
741 yield* olatch.value.open
742 return
743 }
744 const sharding = yield* shardingTag
745 const address = yield* CurrentAddress
746 const requestId = yield* sharding.getSnowflake
747 const span = yield* Effect.orDie(Effect.currentSpan)
748 olatch.value.closeUnsafe()
749 yield* Effect.orDie(sharding.sendOutgoing(
750 new Message.OutgoingRequest({
751 annotations: KeepAliveRpc.annotations,
752 rpc: KeepAliveRpc,
753 context: Context.empty() as any,
754 envelope: Envelope.makeRequest({
755 requestId,
756 address,
757 tag: KeepAliveRpc._tag,
758 payload: void 0,
759 headers: Headers.empty,
760 traceId: span.traceId,
761 spanId: span.spanId,
762 sampled: span.sampled
763 }),
764 lastReceivedReply: Option.none(),
765 respond: () => Effect.void
766 }),
767 true
768 ))
769}, (effect, enabled) =>
770 Effect.withSpan(
771 effect,
772 "Entity/keepAlive",
773 { attributes: { enabled }, captureStackTrace: false }
774 ))
775
776/**
777 * RPC used internally to keep an entity active while a resource is held.
778 *
779 * **Details**
780 *
781 * The RPC is marked as persisted and uninterruptible so the keep-alive signal
782 * survives normal entity restarts.
783 *
784 * @stability unstable
785 * @category keep alive
786 * @since 4.0.0
787 */
788export const KeepAliveRpc = Rpc.make("Cluster/Entity/keepAlive")
789 .annotate(Persisted, true)
790 .annotate(Uninterruptible, true)
791
792/**
793 * Service tag for the latch that coordinates entity keep-alive state.
794 *
795 * **Details**
796 *
797 * `keepAlive` closes the latch when keep-alive is active and opens it again when
798 * the resource no longer needs to keep the entity alive.
799 *
800 * @stability unstable
801 * @category services
802 * @since 4.0.0
803 */
804export class KeepAliveLatch extends Context.Service<KeepAliveLatch, Latch.Latch>()(
805 "effect/cluster/Entity/KeepAliveLatch"
806) {}