1import * as Arr from "../../Array.ts"
2import * as Cause from "../../Cause.ts"
3import { Clock } from "../../Clock.ts"
4import * as Context from "../../Context.ts"
5import * as Duration from "../../Duration.ts"
6import type { Input } from "../../Duration.ts"
7import * as Effect from "../../Effect.ts"
8import * as Equal from "../../Equal.ts"
9import * as Exit from "../../Exit.ts"
10import * as Fiber from "../../Fiber.ts"
11import { identity } from "../../Function.ts"
12import { scopeAddFinalizerUnsafe, scopeRemoveFinalizerUnsafe } from "../../internal/effect.ts"
13import * as Latch from "../../Latch.ts"
14import * as Metric from "../../Metric.ts"
15import * as Option from "../../Option.ts"
16import { CurrentLogAnnotations } from "../../References.ts"
17import type * as Rpc from "../../rpc/Rpc.ts"
18import * as RpcServer from "../../rpc/RpcServer.ts"
19import * as Schedule from "../../Schedule.ts"
20import * as Schema from "../../Schema.ts"
21import * as SchemaIssue from "../../SchemaIssue.ts"
22import * as Scope from "../../Scope.ts"
23import { AlreadyProcessingMessage, EntityNotAssignedToRunner, MailboxFull, MalformedMessage } from "../ClusterError.ts"
24import * as ClusterMetrics from "../ClusterMetrics.ts"
25import { isUninterruptibleForServer, Persisted, WithTransaction } from "../ClusterSchema.ts"
26import * as ClusterSchema from "../ClusterSchema.ts"
27import type { Entity, HandlersFrom } from "../Entity.ts"
28import { CurrentAddress, CurrentRunnerAddress, KeepAliveLatch, KeepAliveRpc, Request } from "../Entity.ts"
29import type { EntityAddress } from "../EntityAddress.ts"
30import type { EntityId } from "../EntityId.ts"
31import type * as Envelope from "../Envelope.ts"
32import * as Message from "../Message.ts"
33import * as MessageStorage from "../MessageStorage.ts"
34import * as Reply from "../Reply.ts"
35import type { RunnerAddress } from "../RunnerAddress.ts"
36import type { ShardId } from "../ShardId.ts"
37import type { Sharding } from "../Sharding.ts"
38import { ShardingConfig } from "../ShardingConfig.ts"
39import * as Snowflake from "../Snowflake.ts"
40import { CurrentActivationScope } from "./entityActivation.ts"
41import { EntityReaper } from "./entityReaper.ts"
42import { acquireEntity, releaseEntity } from "./interruptors.ts"
43import { ResourceMap } from "./resourceMap.ts"
44import { ResourceRef } from "./resourceRef.ts"
45
46/**
47 * @internal
48 */
49export interface EntityManager {
50 readonly sendLocal: <R extends Rpc.Any>(
51 message: Message.IncomingLocal<R>
52 ) => Effect.Effect<void, EntityNotAssignedToRunner | MailboxFull | AlreadyProcessingMessage>
53
54 readonly send: (
55 message: Message.Incoming<any>
56 ) => Effect.Effect<void, EntityNotAssignedToRunner | MailboxFull | AlreadyProcessingMessage>
57
58 readonly isProcessingFor: (message: Message.Incoming<any>, options?: {
59 readonly excludeReplies?: boolean
60 readonly excludeCompleted?: boolean
61 }) => boolean
62 readonly clearProcessed: () => void
63
64 readonly isResidentUnsafe: (address: EntityAddress) => boolean
65 readonly residentAddressesUnsafe: () => Array<EntityAddress>
66
67 readonly interruptShard: (shardId: ShardId, options?: {
68 readonly force?: boolean
69 }) => Effect.Effect<void>
70
71 readonly activeEntityCount: Effect.Effect<number>
72}
73
74// Tracks how many entities are resident on the runner across all entity
75// managers, so the spawn of new entities can be gated by
76// `ShardingConfig.maxResidentEntities`.
77/**
78 * @internal
79 */
80export interface Residency {
81 /**
82 * Reserve a slot for a new entity. Returns `false` when the runner is at
83 * capacity.
84 */
85 readonly admitUnsafe: () => boolean
86 readonly releaseUnsafe: () => void
87}
88
89// Represents the entities managed by this entity manager
90/**
91 * @internal
92 */
93export type EntityState = {
94 readonly address: EntityAddress
95 readonly scope: Scope.Scope
96 readonly activeRequests: Map<Snowflake.Snowflake, {
97 readonly rpc: Rpc.AnyWithProps
98 readonly message: Message.IncomingRequestLocal<any>
99 sentReply: boolean
100 /** Excludes requests awaiting their first dispatch from replay. */
101 delivered: boolean
102 sentExit: boolean
103 lastSentChunk: Option.Option<Reply.Chunk<Rpc.Any>>
104 sequence: number
105 /** Set when the request should not outlive its caller. */
106 callerScope?: Scope.Scope | undefined
107 }>
108 lastActiveCheck: number
109 write: RpcServer.RpcServer<any>["write"]
110 readonly keepAliveLatch: Latch.Latch
111 keepAliveEnabled: boolean
112}
113
114/**
115 * @internal
116 */
117export const make = Effect.fnUntraced(function*<
118 Type extends string,
119 Rpcs extends Rpc.Any,
120 Handlers extends HandlersFrom<Rpcs>,
121 RX
122>(
123 entity: Entity<Type, Rpcs>,
124 buildHandlers: Effect.Effect<Handlers, never, RX>,
125 options: {
126 readonly sharding: Sharding["Service"]
127 readonly storage: MessageStorage.MessageStorage["Service"]
128 readonly runnerAddress: RunnerAddress
129 readonly residency: Residency
130 readonly maxIdleTime?: Input | undefined
131 readonly concurrency?: number | "unbounded" | undefined
132 readonly mailboxCapacity?: number | "unbounded" | undefined
133 readonly disableFatalDefects?: boolean | undefined
134 readonly defectRetryPolicy?: Schedule.Schedule<any, unknown, never, never> | undefined
135 readonly spanAttributes?: Record<string, string> | undefined
136 }
137) {
138 const config = yield* ShardingConfig
139 const snowflakeGen = yield* Snowflake.Generator
140 const managerScope = yield* Effect.scope
141 const storageEnabled = options.storage !== MessageStorage.noop
142 const mailboxCapacity = options.mailboxCapacity ?? config.entityMailboxCapacity
143 const clock = yield* Clock
144 const context = yield* Effect.context<Rpc.Services<Rpcs> | Rpc.Middleware<Rpcs> | RX>()
145 const defectRetryPolicy = options.defectRetryPolicy
146 ? Schedule.concat(options.defectRetryPolicy, defaultRetryPolicy)
147 : defaultRetryPolicy
148 const retryDriver = yield* Schedule.toStepWithSleep(defectRetryPolicy)
149 const entityRpcs = new Map(entity.protocol.requests)
150
151 // add internal rpcs
152 entityRpcs.set(KeepAliveRpc._tag, KeepAliveRpc as any)
153
154 const activeServers = new Map<EntityId, EntityState>()
155 const serverCloseLatches = new Map<EntityAddress, {
156 readonly closed: Latch.Latch
157 readonly force: Latch.Latch
158 }>()
159 const processedRequestIds = new Set<Snowflake.Snowflake>()
160
161 const entities: ResourceMap<
162 EntityAddress,
163 EntityState,
164 EntityNotAssignedToRunner | MailboxFull
165 > = yield* ResourceMap.make(Effect.fnUntraced(function*(address: EntityAddress) {
166 if (!options.sharding.hasShardId(address.shardId)) {
167 return yield* new EntityNotAssignedToRunner({ address })
168 }
169
170 const scope = yield* Effect.scope
171
172 // Gate the spawn on the runner-wide entity cap. Registering the release
173 // must be atomic with taking the slot, otherwise an interrupt in between
174 // would leak it.
175 yield* Effect.uninterruptible(Effect.suspend(() =>
176 options.residency.admitUnsafe()
177 ? Scope.addFinalizer(scope, Effect.sync(options.residency.releaseUnsafe))
178 : Effect.fail(new MailboxFull({ address }))
179 ))
180 const endLatch = Latch.makeUnsafe()
181 const keepAliveLatch = Latch.makeUnsafe()
182 const closeLatches = {
183 closed: Latch.makeUnsafe(),
184 force: Latch.makeUnsafe()
185 }
186
187 yield* Scope.addFinalizer(
188 scope,
189 Effect.sync(() => {
190 releaseEntity(address)
191 })
192 )
193
194 // on shutdown, reset the storage for the entity
195 yield* Scope.addFinalizerExit(
196 scope,
197 () => {
198 serverCloseLatches.get(address)?.closed.openUnsafe()
199 serverCloseLatches.delete(address)
200 return Effect.void
201 }
202 )
203
204 const activeRequests: EntityState["activeRequests"] = new Map()
205 const retired = Latch.makeUnsafe()
206 const isActive = () => !retired.isOpen()
207
208 // Replay previously dispatched requests before admitting new work.
209 const replay = Effect.fnUntraced(function*(write: EntityState["write"]) {
210 if (!isActive()) return
211 for (const request of activeRequests.values()) {
212 if (!request.delivered) continue
213 request.sentExit = false
214 yield* write(0, requestEnvelope(request), requestWriteOptions(request))
215 }
216 })
217
218 const writeRef = yield* ResourceRef.from(
219 scope,
220 Effect.fnUntraced(function*(handlerScope) {
221 let isShuttingDown = false
222
223 const handlerContext = context.pipe(
224 Context.add(CurrentAddress, address),
225 Context.add(CurrentRunnerAddress, options.runnerAddress),
226 Context.add(KeepAliveLatch, keepAliveLatch),
227 Context.add(CurrentActivationScope, scope),
228 Context.add(Scope.Scope, handlerScope),
229 Context.add(CurrentLogAnnotations, {})
230 )
231
232 // Initiate the behavior for the entity
233 const handlers = yield* (entity.protocol.toHandlers(buildHandlers as any).pipe(
234 Effect.setContext(handlerContext as Context.Context<any>),
235 Effect.sandbox,
236 Effect.tapError((cause) => Effect.logError("Defect building entity handlers", cause)),
237 Effect.retry(defectRetryPolicy)
238 ) as Effect.Effect<Context.Context<Rpc.ToHandler<Rpcs>>>)
239
240 const server = yield* RpcServer.makeNoSerialization(entity.protocol, {
241 spanPrefix: `${entity.type}(${address.entityId})`,
242 spanAttributes: {
243 ...options.spanAttributes,
244 "entity.type": entity.type,
245 "entity.id": address.entityId
246 },
247 concurrency: options.concurrency ?? 1,
248 disableFatalDefects: options.disableFatalDefects,
249 onFromServer(response): Effect.Effect<void> {
250 switch (response._tag) {
251 case "Exit": {
252 const request = activeRequests.get(Snowflake.Snowflake(response.requestId))
253 if (!request) return Effect.void
254
255 request.sentReply = true
256 request.sentExit = true
257
258 // Rebuild interrupts are not replies; replacement handlers replay the request.
259 if (isShuttingDown && Exit.hasInterrupts(response.exit) && isActive()) {
260 return Effect.void
261 }
262
263 // For durable messages, ignore interrupts during shutdown.
264 // They will be retried when the entity is restarted.
265 // Also, if the request is uninterruptible, we ignore the
266 // interrupt.
267 if (
268 storageEnabled &&
269 Context.get(request.message.annotations, Persisted) &&
270 Exit.hasInterrupts(response.exit) &&
271 (isShuttingDown || isUninterruptibleForServer(request.message.annotations))
272 ) {
273 if (!isShuttingDown) {
274 request.sentExit = false
275 return server.write(0, requestEnvelope(request), requestWriteOptions(request)).pipe(
276 Effect.forkIn(handlerScope)
277 )
278 }
279 activeRequests.delete(Snowflake.Snowflake(response.requestId))
280 return options.storage.unregisterReplyHandler(request.message.envelope.requestId)
281 }
282 return retryRespond(
283 4,
284 Effect.suspend(() =>
285 request.message.respond(
286 new Reply.WithExit({
287 requestId: Snowflake.Snowflake(response.requestId),
288 id: snowflakeGen.nextUnsafe(),
289 exit: response.exit
290 })
291 )
292 )
293 ).pipe(
294 Effect.flatMap(() => {
295 if (storageEnabled) {
296 processedRequestIds.add(request.message.envelope.requestId)
297 }
298 activeRequests.delete(Snowflake.Snowflake(response.requestId))
299
300 // Start the idle timer when the last request completes.
301 if (activeRequests.size === 0) {
302 state.lastActiveCheck = clock.currentTimeMillisUnsafe()
303 }
304
305 return Effect.void
306 }),
307 Effect.orDie
308 )
309 }
310 case "Chunk": {
311 const request = activeRequests.get(Snowflake.Snowflake(response.requestId))
312 if (!request) return Effect.void
313 const sequence = request.sequence
314 request.sequence++
315 if (!request.sentReply) {
316 request.sentReply = true
317 }
318 return Effect.orDie(retryRespond(
319 4,
320 Effect.suspend(() => {
321 const reply = new Reply.Chunk({
322 requestId: Snowflake.Snowflake(response.requestId),
323 id: snowflakeGen.nextUnsafe(),
324 sequence,
325 values: response.values
326 })
327 request.lastSentChunk = Option.some(reply)
328 return request.message.respond(reply)
329 })
330 ))
331 }
332 case "Defect": {
333 if (!isActive()) return endLatch.open
334 const rebuild = writeRef.rebuildUnsafe({ from: server.write, prepare: replay })
335 if (!rebuild) return Effect.void
336 return Effect.forkIn(restart(Cause.die(response.defect), rebuild), managerScope)
337 }
338 case "ClientEnd": {
339 return endLatch.open
340 }
341 }
342 }
343 }).pipe(
344 Scope.provide(handlerScope),
345 Effect.setContext(Context.merge(handlerContext, handlers))
346 )
347
348 yield* Scope.addFinalizer(
349 handlerScope,
350 Effect.sync(() => {
351 isShuttingDown = true
352 })
353 )
354
355 return server.write
356 }),
357 address
358 )
359
360 function restart(cause: Cause.Cause<unknown>, rebuild: Effect.Effect<void>): Effect.Effect<void> {
361 return Effect.logError("Defect in entity, restarting", cause).pipe(
362 Effect.andThen(Effect.ignore(retryDriver(void 0))),
363 Effect.flatMap(() => isActive() ? rebuild : endLatch.open),
364 Effect.annotateLogs({
365 module: "EntityManager",
366 address,
367 runner: options.runnerAddress
368 }),
369 Effect.catchCause((cause) => {
370 if (Cause.hasInterruptsOnly(cause)) return Effect.void
371 const retry = isActive() ? writeRef.rebuildUnsafe({ prepare: replay }) : undefined
372 return retry ? restart(cause, retry) : Effect.void
373 })
374 )
375 }
376
377 const state: EntityState = {
378 scope,
379 address,
380 write(clientId, message, writeOptions) {
381 return Effect.suspend(() => {
382 if (message._tag !== "Request") {
383 // Interrupts, acks and EOF reach the handlers even during shutdown.
384 const write = writeRef.getUnsafe()
385 return write
386 ? write(clientId, message, writeOptions)
387 : Effect.flatMap(writeRef.await, (write) => write(clientId, message, writeOptions))
388 }
389 if (!isActive()) return Effect.interrupt
390 const write = writeRef.getUnsafe()
391 if (write === undefined) {
392 return Effect.flatMap(
393 Effect.raceFirst(writeRef.await, retired.await),
394 () => state.write(clientId, message, writeOptions)
395 )
396 }
397 const request = activeRequests.get(Snowflake.Snowflake(message.id))
398 if (!request) return Effect.void
399 request.delivered = true
400 return write(clientId, message, writeOptions)
401 })
402 },
403 activeRequests,
404 lastActiveCheck: clock.currentTimeMillisUnsafe(),
405 keepAliveLatch,
406 keepAliveEnabled: false
407 }
408
409 // During shutdown, signal that no more messages will be processed
410 // and wait for the fiber to complete.
411 //
412 // If the termination timeout is reached, let the server clean itself up
413 yield* Scope.addFinalizer(
414 scope,
415 Effect.suspend(() => {
416 activeServers.delete(address.entityId)
417 retired.openUnsafe()
418 acquireEntity(address)
419 return Effect.raceFirst(
420 state.write(0, { _tag: "Eof" }).pipe(
421 Effect.andThen(endLatch.await),
422 Effect.timeoutOption(config.entityTerminationTimeout),
423 Effect.interruptible
424 ),
425 Effect.interruptible(closeLatches.force.await)
426 )
427 })
428 )
429 if (!options.sharding.hasShardId(address.shardId)) {
430 return yield* new EntityNotAssignedToRunner({ address })
431 }
432 // Do not make shard interruption wait for an entity that is still building.
433 serverCloseLatches.set(address, closeLatches)
434 activeServers.set(address.entityId, state)
435
436 return state
437 }, Effect.provideService(CurrentLogAnnotations, {})))
438
439 const reaper = yield* EntityReaper
440 const maxIdleTime = Duration.toMillis(
441 Duration.fromInputUnsafe(options.maxIdleTime ?? config.entityMaxIdleTime)
442 )
443 if (Number.isFinite(maxIdleTime)) {
444 yield* reaper.register({
445 maxIdleTime,
446 servers: activeServers,
447 entities
448 })
449 }
450
451 // update metrics for active servers
452 const typeAttributes = Metric.CurrentMetricAttributes.context({ type: entity.type })
453 yield* Effect.sync(() => {
454 ClusterMetrics.entities.updateUnsafe(BigInt(activeServers.size), typeAttributes)
455 }).pipe(
456 Effect.andThen(Effect.sleep(1000)),
457 Effect.forever,
458 Effect.forkIn(managerScope)
459 )
460
461 function sendLocal<R extends Rpc.Any>(
462 message: Message.IncomingLocal<R>
463 ): Effect.Effect<void, EntityNotAssignedToRunner | MailboxFull | AlreadyProcessingMessage> {
464 return Effect.provideService(
465 Effect.flatMap(
466 entities.get(message.envelope.address),
467 (server): Effect.Effect<void, EntityNotAssignedToRunner | MailboxFull | AlreadyProcessingMessage> => {
468 switch (message._tag) {
469 case "IncomingRequestLocal": {
470 // If the request is already running, then we might have more than
471 // one sender for the same request. In this case, the other senders
472 // should resume from storage only.
473 let entry = server.activeRequests.get(message.envelope.requestId)
474 if (entry || processedRequestIds.has(message.envelope.requestId)) {
475 return Effect.fail(
476 new AlreadyProcessingMessage({
477 envelopeId: message.envelope.requestId,
478 address: message.envelope.address
479 })
480 )
481 }
482
483 const rpc = entityRpcs.get(message.envelope.tag)! as any as Rpc.AnyWithProps
484 if (!storageEnabled && Context.get(message.annotations, Persisted)) {
485 return Effect.die(
486 "EntityManager.sendLocal: Cannot process a persisted message without MessageStorage"
487 )
488 }
489
490 // Cluster internal RPCs
491
492 // keep-alive RPC
493 if (rpc._tag === KeepAliveRpc._tag) {
494 const msg = message as unknown as Message.IncomingRequestLocal<typeof KeepAliveRpc>
495 const reply = Effect.suspend(() =>
496 Effect.orDie(retryRespond(
497 4,
498 msg.respond(
499 new Reply.WithExit<typeof KeepAliveRpc>({
500 requestId: message.envelope.requestId,
501 id: snowflakeGen.nextUnsafe(),
502 exit: Exit.void
503 })
504 )
505 ))
506 )
507
508 if (server.keepAliveEnabled) return reply
509 server.keepAliveEnabled = true
510 return server.keepAliveLatch.whenOpen(Effect.suspend(() => {
511 server.keepAliveEnabled = false
512 return reply
513 })).pipe(
514 Effect.forkIn(server.scope, { startImmediately: true }),
515 Effect.asVoid
516 )
517 }
518
519 if (mailboxCapacity !== "unbounded" && server.activeRequests.size >= mailboxCapacity) {
520 return Effect.fail(new MailboxFull({ address: message.envelope.address }))
521 }
522
523 const callerScope = message.callerScope !== undefined &&
524 !Context.get(message.annotations, Persisted) &&
525 Context.get(message.annotations, ClusterSchema.Uninterruptible) === false
526 ? message.callerScope
527 : undefined
528 if (callerScope?.state._tag === "Closed") {
529 return Effect.void
530 }
531 entry = {
532 rpc,
533 message,
534 sentReply: false,
535 delivered: false,
536 sentExit: false,
537 lastSentChunk: Option.filter(
538 message.lastSentReply,
539 (reply): reply is Reply.Chunk<Rpc.Any> => reply._tag === "Chunk"
540 ),
541 sequence: Option.match(message.lastSentReply, {
542 onNone: () => 0,
543 onSome: (reply) => reply._tag === "Chunk" ? reply.sequence + 1 : 0
544 }),
545 callerScope
546 }
547 server.activeRequests.set(message.envelope.requestId, entry)
548 if (callerScope !== undefined) {
549 // Register cleanup atomically with admission, including requests awaiting delivery.
550 scopeAddFinalizerUnsafe(callerScope, {}, () =>
551 Effect.sync(() => {
552 if (server.activeRequests.delete(message.envelope.requestId) && server.activeRequests.size === 0) {
553 server.lastActiveCheck = clock.currentTimeMillisUnsafe()
554 }
555 }))
556 }
557 return server.write(0, requestEnvelope(entry), requestWriteOptions(entry))
558 }
559 case "IncomingEnvelope": {
560 const entry = server.activeRequests.get(message.envelope.requestId)
561 if (!entry) {
562 return Effect.void
563 } else if (
564 message.envelope._tag === "AckChunk" &&
565 Option.isSome(entry.lastSentChunk) &&
566 message.envelope.replyId !== entry.lastSentChunk.value.id
567 ) {
568 return Effect.void
569 }
570 return server.write(
571 0,
572 message.envelope._tag === "AckChunk"
573 ? { _tag: "Ack", requestId: message.envelope.requestId as any }
574 : {
575 _tag: "Interrupt",
576 requestId: message.envelope.requestId as any,
577 interruptors: []
578 }
579 )
580 }
581 }
582 }
583 ),
584 CurrentLogAnnotations,
585 {}
586 )
587 }
588
589 // Bind each handler fiber, including replays, before its body runs.
590 const bindToCaller = (callerScope: Scope.Scope) => <A, E, R>(effect: Effect.Effect<A, E, R>) =>
591 Effect.withFiber<A, E, R>((fiber) => {
592 if (callerScope.state._tag === "Closed") return Effect.interrupt
593 const key = {}
594 scopeAddFinalizerUnsafe(callerScope, key, () => Effect.sync(() => fiber.interruptUnsafe(fiber.id)))
595 return Effect.ensuring(effect, Effect.sync(() => scopeRemoveFinalizerUnsafe(callerScope, key)))
596 })
597
598 const requestWriteOptions = (
599 entry: { readonly message: Message.IncomingRequestLocal<any>; readonly callerScope?: Scope.Scope | undefined }
600 ): Parameters<EntityState["write"]>[2] => {
601 const onTransaction = Context.get(entry.message.annotations, WithTransaction)
602 ? options.storage.withTransaction
603 : undefined
604 const onCaller = entry.callerScope && bindToCaller(entry.callerScope)
605 if (!onCaller) return onTransaction && { onRequest: onTransaction }
606 return { onRequest: onTransaction ? (effect) => onCaller(onTransaction(effect)) : onCaller }
607 }
608
609 const decodeMessage = makeMessageDecode(entityRpcs)
610
611 const runFork = Effect.runForkWith(context)
612
613 return identity<EntityManager>({
614 interruptShard: (shardId: ShardId, options) =>
615 Effect.suspend(function loop(): Effect.Effect<void> {
616 const fibers = Arr.empty<Fiber.Fiber<void>>()
617 if (options?.force === true) {
618 serverCloseLatches.forEach((latches, address) => {
619 if (shardId[Equal.symbol](address.shardId)) {
620 latches.force.openUnsafe()
621 }
622 })
623 }
624 activeServers.forEach((state) => {
625 if (shardId[Equal.symbol](state.address.shardId)) {
626 fibers.push(runFork(entities.removeIgnore(state.address)))
627 }
628 })
629 serverCloseLatches.forEach((latches, address) => {
630 if (shardId[Equal.symbol](address.shardId)) {
631 fibers.push(runFork(latches.closed.await))
632 }
633 })
634 if (fibers.length === 0) return Effect.void
635 return Effect.flatMap(Fiber.joinAll(fibers), loop)
636 }),
637 isProcessingFor(message, options) {
638 if (
639 options?.excludeReplies !== true && options?.excludeCompleted !== true &&
640 processedRequestIds.has(message.envelope.requestId)
641 ) {
642 return true
643 }
644 const state = activeServers.get(message.envelope.address.entityId)
645 if (!state) return false
646 const request = state.activeRequests.get(message.envelope.requestId)
647 if (request === undefined) {
648 return false
649 } else if (options?.excludeReplies && request.sentReply) {
650 return false
651 } else if (options?.excludeCompleted && request.sentExit) {
652 return false
653 }
654 return true
655 },
656 clearProcessed() {
657 processedRequestIds.clear()
658 },
659 isResidentUnsafe: (address) => entities.hasUnsafe(address),
660 residentAddressesUnsafe: () => entities.keysUnsafe(),
661 sendLocal,
662 send: (message) =>
663 decodeMessage(message).pipe(
664 Effect.matchEffect({
665 onFailure: (cause) => {
666 if (message._tag === "IncomingEnvelope") {
667 return Effect.die(new MalformedMessage({ cause }))
668 }
669 return Effect.orDie(message.respond(
670 new Reply.ReplyWithContext({
671 reply: new Reply.WithExit({
672 id: snowflakeGen.nextUnsafe(),
673 requestId: message.envelope.requestId,
674 exit: Exit.die(new MalformedMessage({ cause }))
675 }),
676 rpc: entityRpcs.get(message.envelope.tag)!,
677 context: context as any
678 })
679 ))
680 },
681 onSuccess: (decoded) => {
682 if (decoded._tag === "IncomingEnvelope") {
683 return sendLocal(decoded)
684 }
685 const request = message as Message.IncomingRequest<any>
686 const rpc = entityRpcs.get(decoded.envelope.tag)!
687 return sendLocal(
688 new Message.IncomingRequestLocal({
689 annotations: Context.get(rpc.annotations, ClusterSchema.Dynamic)(
690 rpc.annotations,
691 decoded.envelope as any
692 ),
693 envelope: decoded.envelope,
694 lastSentReply: decoded.lastSentReply,
695 callerScope: request.callerScope,
696 respond: (reply) =>
697 request.respond(
698 new Reply.ReplyWithContext({
699 reply,
700 rpc,
701 context: context as any
702 })
703 )
704 })
705 )
706 }
707 }),
708 Effect.provideContext(context as Context.Context<unknown>)
709 ),
710 activeEntityCount: Effect.sync(() => activeServers.size)
711 })
712})
713
714const defaultRetryPolicy = Schedule.min([
715 Schedule.exponential(500, 1.5),
716 Schedule.spaced("10 seconds")
717])
718
719const makeMessageDecode = <Rpcs extends Rpc.Any>(entityRpcs: Map<string, Rpcs>) => {
720 const decodeRequest = Effect.fnUntracedEager(function*(
721 message: Message.IncomingRequest<Rpcs>,
722 rpc: Rpc.AnyWithProps
723 ) {
724 const codecFor = message.codecFor
725 const payload = yield* Schema.decodeEffect(codecFor(rpc.payloadSchema))(message.envelope.payload)
726 const lastSentReply = Option.isNone(message.lastSentReply) ?
727 message.lastSentReply :
728 Option.some(yield* Schema.decodeEffect(Reply.Reply(rpc, codecFor))(message.lastSentReply.value))
729 return {
730 _tag: "IncomingRequest",
731 envelope: {
732 ...message.envelope,
733 payload
734 } as Envelope.Request.Any,
735 lastSentReply
736 } as const
737 })
738
739 return (message: Message.Incoming<Rpcs>): Effect.Effect<
740 {
741 readonly _tag: "IncomingRequest"
742 readonly envelope: Envelope.Request.Any
743 readonly lastSentReply: Option.Option<Reply.Reply<Rpcs>>
744 } | Message.IncomingEnvelope,
745 Schema.SchemaError,
746 Rpc.ServicesServer<Rpcs>
747 > => {
748 if (message._tag === "IncomingEnvelope") {
749 return Effect.succeed(message)
750 }
751 const rpc = entityRpcs.get(message.envelope.tag) as any as Rpc.AnyWithProps
752 if (!rpc) {
753 return Effect.fail(
754 new Schema.SchemaError(
755 new SchemaIssue.InvalidValue({
756 message: "Expected a known entity RPC tag"
757 })
758 )
759 )
760 }
761 return decodeRequest(message, rpc) as Effect.Effect<
762 {
763 readonly _tag: "IncomingRequest"
764 readonly envelope: Envelope.Request.Any
765 readonly lastSentReply: Option.Option<Reply.Reply<Rpcs>>
766 },
767 Schema.SchemaError,
768 Rpc.ServicesServer<Rpcs>
769 >
770 }
771}
772
773const requestEnvelope = (entry: {
774 readonly message: Message.IncomingRequestLocal<any>
775 readonly lastSentChunk: Option.Option<Reply.Chunk<Rpc.Any>>
776}): Parameters<EntityState["write"]>[1] => ({
777 ...entry.message.envelope,
778 id: entry.message.envelope.requestId as any,
779 tag: entry.message.envelope.tag as any,
780 payload: new Request({
781 ...entry.message.envelope,
782 lastSentChunk: entry.lastSentChunk
783 } as any) as any
784})
785
786const retryRespond = <A, E, R>(times: number, effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
787 times === 0 ?
788 effect :
789 Effect.catch(effect, () => Effect.delay(retryRespond(times - 1, effect), 200))