| 1 | import { assert, describe, expect, it } from "@effect/vitest" |
| 2 | import { |
| 3 | Array, |
| 4 | Cause, |
| 5 | Clock, |
| 6 | Context, |
| 7 | Deferred, |
| 8 | Effect, |
| 9 | Exit, |
| 10 | Fiber, |
| 11 | Latch, |
| 12 | Layer, |
| 13 | Logger, |
| 14 | MutableRef, |
| 15 | Option, |
| 16 | Queue, |
| 17 | Ref, |
| 18 | Result, |
| 19 | Schedule, |
| 20 | Schema, |
| 21 | Stream |
| 22 | } from "effect" |
| 23 | import { |
| 24 | ClusterError, |
| 25 | ClusterMetrics, |
| 26 | ClusterSchema, |
| 27 | Entity, |
| 28 | EntityAddress, |
| 29 | EntityId, |
| 30 | EntityType, |
| 31 | Envelope, |
| 32 | MachineId, |
| 33 | Message, |
| 34 | MessageStorage, |
| 35 | Runner, |
| 36 | RunnerAddress, |
| 37 | RunnerHealth, |
| 38 | Runners, |
| 39 | RunnerStorage, |
| 40 | ShardId, |
| 41 | Sharding, |
| 42 | ShardingConfig, |
| 43 | Snowflake |
| 44 | } from "effect/cluster" |
| 45 | import * as EntityManager from "effect/cluster/internal/entityManager" |
| 46 | import { EntityReaper } from "effect/cluster/internal/entityReaper" |
| 47 | import * as ActiveTeardown from "effect/cluster/internal/interruptors" |
| 48 | import { Headers } from "effect/http" |
| 49 | import { Rpc, type RpcGroup } from "effect/rpc" |
| 50 | import { TestClock } from "effect/testing" |
| 51 | import { |
| 52 | CallerId, |
| 53 | ContextBleedEntity, |
| 54 | ContextBleedLayer, |
| 55 | TestEntity, |
| 56 | TestEntityNoState, |
| 57 | TestEntityState, |
| 58 | User |
| 59 | } from "./TestEntity.ts" |
| 60 | |
| 61 | // Isolate the long-lived stream from concurrent shard-metric tests. |
| 62 | describe("Sharding claim release regressions", { concurrent: false }, () => { |
| 63 | it.effect("keeps the claim of an active request replayed after an entity defect", () => |
| 64 | Effect.gen(function*() { |
| 65 | const started = yield* Queue.make<number>() |
| 66 | let starts = 0 |
| 67 | let active = 0 |
| 68 | let layerBuilds = 0 |
| 69 | let defectAttempts = 0 |
| 70 | const released: Array<Snowflake.Snowflake> = [] |
| 71 | const claimed: Array<Snowflake.Snowflake> = [] |
| 72 | const Replaying = Entity.make("ClaimDefectReplay", [ |
| 73 | Rpc.make("Run"), |
| 74 | Rpc.make("Defect") |
| 75 | ]).annotateRpcs(ClusterSchema.Persisted, true) |
| 76 | const layer = Replaying.toLayer( |
| 77 | Effect.sync(() => { |
| 78 | layerBuilds++ |
| 79 | return Replaying.of({ |
| 80 | Run: () => |
| 81 | Rpc.fork( |
| 82 | Effect.gen(function*() { |
| 83 | starts++ |
| 84 | active++ |
| 85 | yield* Queue.offer(started, starts) |
| 86 | return yield* Effect.never |
| 87 | }).pipe(Effect.ensuring(Effect.sync(() => active--))) |
| 88 | ), |
| 89 | Defect: () => |
| 90 | Effect.suspend(() => { |
| 91 | defectAttempts++ |
| 92 | return defectAttempts === 1 ? Effect.die("restart entity") : Effect.void |
| 93 | }) |
| 94 | }) |
| 95 | }), |
| 96 | { defectRetryPolicy: Schedule.forever } |
| 97 | ).pipe(Layer.provideMerge(CappedSharding({}, (storage) => ({ |
| 98 | ...storage, |
| 99 | resetRequests: (ids) => |
| 100 | Effect.gen(function*() { |
| 101 | released.push(...ids) |
| 102 | yield* storage.resetRequests(ids) |
| 103 | }), |
| 104 | unprocessedMessages: (shards, options) => |
| 105 | storage.unprocessedMessages(shards, options).pipe( |
| 106 | Effect.tap((messages) => |
| 107 | Effect.sync(() => { |
| 108 | for (const message of messages) { |
| 109 | if (message._tag === "IncomingRequest") claimed.push(message.envelope.requestId) |
| 110 | } |
| 111 | }) |
| 112 | ) |
| 113 | ) |
| 114 | })))) |
| 115 | |
| 116 | yield* Effect.gen(function*() { |
| 117 | yield* TestClock.adjust(1) |
| 118 | const sharding = yield* Sharding.Sharding |
| 119 | const driver = yield* MessageStorage.MemoryDriver |
| 120 | const client = (yield* Replaying.client)("replay") |
| 121 | const running = yield* client.Run().pipe(Effect.forkChild({ startImmediately: true })) |
| 122 | yield* TestClock.adjust(1) |
| 123 | assert.strictEqual(yield* Queue.take(started), 1) |
| 124 | const request = driver.journal[0] |
| 125 | assert.strictEqual(request._tag, "Request") |
| 126 | |
| 127 | // Defecting another handler rebuilds the entity and replays Run. |
| 128 | yield* client.Defect() |
| 129 | assert.strictEqual(yield* Queue.take(started), 2, "in-flight handler must replay after the defect") |
| 130 | assert.strictEqual(layerBuilds, 2) |
| 131 | assert.strictEqual(defectAttempts, 2) |
| 132 | assert.strictEqual(active, 1) |
| 133 | expect(running.pollUnsafe()).toBeUndefined() |
| 134 | claimed.length = 0 |
| 135 | released.length = 0 |
| 136 | |
| 137 | yield* TestClock.adjust("10 minutes") |
| 138 | for (let i = 0; i < 3; i++) { |
| 139 | yield* sharding.pollStorage |
| 140 | yield* TestClock.adjust(1) |
| 141 | } |
| 142 | expect(claimed).toContain(Snowflake.Snowflake(request.requestId)) |
| 143 | assert.strictEqual(starts, 2, "polling must not deliver the replayed request again") |
| 144 | assert.strictEqual(layerBuilds, 2, "polling must not restart the entity again") |
| 145 | assert.strictEqual(active, 1, "replayed handler must remain active") |
| 146 | expect(running.pollUnsafe()).toBeUndefined() |
| 147 | assert.strictEqual(released.length, 0, "replayed active request claim must not be released") |
| 148 | assert.deepStrictEqual( |
| 149 | claimed, |
| 150 | [Snowflake.Snowflake(request.requestId)], |
| 151 | "replayed request must not be reclaimed on every poll" |
| 152 | ) |
| 153 | }).pipe(Effect.provide(layer)) |
| 154 | })) |
| 155 | |
| 156 | it.effect("keeps the claim of an active uninterruptible request after restarting", () => |
| 157 | Effect.gen(function*() { |
| 158 | const started = yield* Queue.make<number>() |
| 159 | let starts = 0 |
| 160 | let active = 0 |
| 161 | const released: Array<Snowflake.Snowflake> = [] |
| 162 | const claimed: Array<Snowflake.Snowflake> = [] |
| 163 | const Restarting = Entity.make("ClaimRestart", [ |
| 164 | Rpc.make("Run") |
| 165 | .annotate(ClusterSchema.Persisted, true) |
| 166 | .annotate(ClusterSchema.Uninterruptible, true) |
| 167 | ]) |
| 168 | const layer = Restarting.toLayer({ |
| 169 | Run: () => |
| 170 | Effect.gen(function*() { |
| 171 | starts++ |
| 172 | active++ |
| 173 | yield* Queue.offer(started, starts) |
| 174 | return yield* Effect.never |
| 175 | }).pipe(Effect.ensuring(Effect.sync(() => active--))) |
| 176 | }).pipe(Layer.provideMerge(CappedSharding({}, (storage) => ({ |
| 177 | ...storage, |
| 178 | resetRequests: (ids) => |
| 179 | Effect.gen(function*() { |
| 180 | released.push(...ids) |
| 181 | yield* storage.resetRequests(ids) |
| 182 | }), |
| 183 | unprocessedMessages: (shards, options) => |
| 184 | storage.unprocessedMessages(shards, options).pipe( |
| 185 | Effect.tap((messages) => |
| 186 | Effect.sync(() => { |
| 187 | for (const message of messages) { |
| 188 | if (message._tag === "IncomingRequest") claimed.push(message.envelope.requestId) |
| 189 | } |
| 190 | }) |
| 191 | ) |
| 192 | ) |
| 193 | })))) |
| 194 | |
| 195 | yield* Effect.gen(function*() { |
| 196 | yield* TestClock.adjust(1) |
| 197 | const sharding = yield* Sharding.Sharding |
| 198 | const driver = yield* MessageStorage.MemoryDriver |
| 199 | const client = (yield* Restarting.client)("restart") |
| 200 | const running = yield* client.Run().pipe(Effect.forkChild({ startImmediately: true })) |
| 201 | assert.strictEqual(yield* Queue.take(started), 1) |
| 202 | const request = driver.journal[0] |
| 203 | assert.strictEqual(request._tag, "Request") |
| 204 | |
| 205 | // A persisted interrupt restarts the uninterruptible request. |
| 206 | yield* driver.encoded.saveEnvelope({ |
| 207 | envelope: { |
| 208 | _tag: "Interrupt", |
| 209 | id: String(yield* sharding.getSnowflake), |
| 210 | requestId: request.requestId, |
| 211 | address: request.address |
| 212 | }, |
| 213 | primaryKey: null, |
| 214 | deliverAt: null |
| 215 | }) |
| 216 | yield* sharding.pollStorage |
| 217 | yield* TestClock.adjust(1) |
| 218 | assert.strictEqual(yield* Queue.take(started), 2, "handler must restart after the stored interrupt") |
| 219 | assert.strictEqual(active, 1) |
| 220 | expect(running.pollUnsafe()).toBeUndefined() |
| 221 | claimed.length = 0 |
| 222 | released.length = 0 |
| 223 | |
| 224 | yield* TestClock.adjust("10 minutes") |
| 225 | for (let i = 0; i < 3; i++) { |
| 226 | yield* sharding.pollStorage |
| 227 | yield* TestClock.adjust(1) |
| 228 | } |
| 229 | expect(claimed).toContain(Snowflake.Snowflake(request.requestId)) |
| 230 | assert.strictEqual(starts, 2, "polling must not deliver the restarted request again") |
| 231 | assert.strictEqual(active, 1, "restarted handler must remain active") |
| 232 | expect(running.pollUnsafe()).toBeUndefined() |
| 233 | assert.strictEqual(released.length, 0, "restarted active request claim must not be released") |
| 234 | assert.deepStrictEqual( |
| 235 | claimed, |
| 236 | [Snowflake.Snowflake(request.requestId)], |
| 237 | "restarted request must not be reclaimed on every poll" |
| 238 | ) |
| 239 | }).pipe(Effect.provide(layer)) |
| 240 | })) |
| 241 | |
| 242 | it.effect("keeps the claim of an active stream after an acknowledged chunk", () => |
| 243 | Effect.gen(function*() { |
| 244 | const acked = yield* Deferred.make<void>() |
| 245 | const released: Array<Snowflake.Snowflake> = [] |
| 246 | const claimed: Array<Snowflake.Snowflake> = [] |
| 247 | yield* Effect.gen(function*() { |
| 248 | yield* TestClock.adjust(1) |
| 249 | const sharding = yield* Sharding.Sharding |
| 250 | const state = yield* TestEntityState |
| 251 | const client = (yield* TestEntity.client)("active-stream") |
| 252 | const values: Array<number> = [] |
| 253 | const stream = yield* client.StreamWithKey({ key: "run" }).pipe( |
| 254 | Stream.runForEach((value) => Effect.sync(() => values.push(value))), |
| 255 | Effect.forkChild({ startImmediately: true }) |
| 256 | ) |
| 257 | yield* Queue.offer(state.streamMessages, void 0) |
| 258 | yield* TestClock.adjust(1000) |
| 259 | yield* Deferred.await(acked) |
| 260 | expect(values).toEqual([0]) |
| 261 | expect(stream.pollUnsafe()).toBeUndefined() |
| 262 | claimed.length = 0 |
| 263 | released.length = 0 |
| 264 | |
| 265 | // Await the ack because memory storage does not model SQL reply filtering. |
| 266 | yield* TestClock.adjust("10 minutes") |
| 267 | for (let i = 0; i < 3; i++) { |
| 268 | yield* sharding.pollStorage |
| 269 | yield* TestClock.adjust(1) |
| 270 | } |
| 271 | expect(claimed.length).toBeGreaterThan(0) |
| 272 | expect(stream.pollUnsafe()).toBeUndefined() |
| 273 | expect(values).toEqual([0]) |
| 274 | assert.strictEqual(released.length, 0, "active stream claim must not be released after a chunk") |
| 275 | assert.strictEqual(claimed.length, 1, "active stream must not be reclaimed on every poll") |
| 276 | }).pipe(Effect.provide(CappedSharding({}, (storage) => ({ |
| 277 | ...storage, |
| 278 | saveEnvelope: (message) => |
| 279 | storage.saveEnvelope(message).pipe( |
| 280 | Effect.tap(() => message.envelope._tag === "AckChunk" ? Deferred.succeed(acked, void 0) : Effect.void) |
| 281 | ), |
| 282 | resetRequests: (ids) => |
| 283 | Effect.gen(function*() { |
| 284 | released.push(...ids) |
| 285 | yield* storage.resetRequests(ids) |
| 286 | }), |
| 287 | unprocessedMessages: (shards, options) => |
| 288 | storage.unprocessedMessages(shards, options).pipe( |
| 289 | Effect.tap((messages) => |
| 290 | Effect.sync(() => { |
| 291 | for (const message of messages) { |
| 292 | if (message._tag === "IncomingRequest") claimed.push(message.envelope.requestId) |
| 293 | } |
| 294 | }) |
| 295 | ) |
| 296 | ) |
| 297 | })))) |
| 298 | })) |
| 299 | |
| 300 | it.effect("releases capped addresses even when targeted claim release fails", () => |
| 301 | Effect.gen(function*() { |
| 302 | const readStarted = yield* Deferred.make<void>() |
| 303 | const releaseRead = yield* Deferred.make<void>() |
| 304 | let pauseNextRead = false |
| 305 | const failedReleases: Array<Snowflake.Snowflake> = [] |
| 306 | const cappedReleases: Array<EntityAddress.EntityAddress> = [] |
| 307 | const claimed: Array<string> = [] |
| 308 | yield* Effect.gen(function*() { |
| 309 | yield* TestClock.adjust(1) |
| 310 | const sharding = yield* Sharding.Sharding |
| 311 | const state = yield* TestEntityState |
| 312 | const client = yield* TestEntity.client |
| 313 | const firstRun = yield* client("completed").RequestWithKey({ key: "run" }).pipe( |
| 314 | Effect.forkChild({ startImmediately: true }) |
| 315 | ) |
| 316 | const request = yield* Queue.take(state.envelopes) |
| 317 | pauseNextRead = true |
| 318 | yield* sharding.pollStorage |
| 319 | yield* Deferred.await(readStarted) |
| 320 | yield* Queue.offer(state.messages, void 0) |
| 321 | yield* Fiber.join(firstRun) |
| 322 | yield* TestClock.adjust(1) |
| 323 | yield* sharding.reset(request.requestId) |
| 324 | yield* saveGetUserRequest("capped", 42) |
| 325 | |
| 326 | // Fill capacity during the read so its batch also contains a capped address. |
| 327 | yield* client("resident").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) |
| 328 | yield* Queue.take(state.envelopes) |
| 329 | expect(yield* sharding.activeEntityCount).toEqual(2) |
| 330 | yield* Deferred.succeed(releaseRead, void 0) |
| 331 | yield* TestClock.adjust(1) |
| 332 | expect(claimed).toContain("completed") |
| 333 | expect(claimed).toContain("capped") |
| 334 | expect(failedReleases).toEqual([request.requestId]) |
| 335 | assert.deepStrictEqual( |
| 336 | cappedReleases.map((address) => address.entityId), |
| 337 | ["capped"], |
| 338 | "failed targeted release must not skip capped-address release from the same batch" |
| 339 | ) |
| 340 | }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 2 }, (storage) => ({ |
| 341 | ...storage, |
| 342 | resetRequests: (ids) => |
| 343 | Effect.suspend(() => { |
| 344 | failedReleases.push(...ids) |
| 345 | return Effect.fail(new ClusterError.PersistenceError({ cause: "injected targeted release failure" })) |
| 346 | }), |
| 347 | resetAddresses: (addresses) => |
| 348 | Effect.gen(function*() { |
| 349 | cappedReleases.push(...addresses) |
| 350 | yield* storage.resetAddresses(addresses) |
| 351 | }), |
| 352 | unprocessedMessages: (shards, options) => |
| 353 | Effect.gen(function*() { |
| 354 | if (pauseNextRead) { |
| 355 | pauseNextRead = false |
| 356 | yield* Deferred.succeed(readStarted, void 0) |
| 357 | yield* Deferred.await(releaseRead) |
| 358 | } |
| 359 | const messages = yield* storage.unprocessedMessages(shards, options) |
| 360 | for (const message of messages) claimed.push(message.envelope.address.entityId) |
| 361 | return messages |
| 362 | }) |
| 363 | })))) |
| 364 | })) |
| 365 | }) |
| 366 | |
| 367 | describe.concurrent("Sharding", () => { |
| 368 | it.effect("redelivers a request reset while an asynchronous storage read is pending", () => |
| 369 | Effect.gen(function*() { |
| 370 | const readStarted = yield* Deferred.make<void>() |
| 371 | const releaseRead = yield* Deferred.make<void>() |
| 372 | let pauseNextRead = false |
| 373 | const claimed: Array<Snowflake.Snowflake> = [] |
| 374 | |
| 375 | yield* Effect.gen(function*() { |
| 376 | yield* TestClock.adjust(1) |
| 377 | const sharding = yield* Sharding.Sharding |
| 378 | const state = yield* TestEntityState |
| 379 | const client = (yield* TestEntity.client)("reset-race") |
| 380 | const firstRun = yield* client.RequestWithKey({ key: "run" }).pipe( |
| 381 | Effect.forkChild({ startImmediately: true }) |
| 382 | ) |
| 383 | const request = yield* Queue.take(state.envelopes) |
| 384 | |
| 385 | // Pause after clearing processed IDs but before storage claims rows. |
| 386 | pauseNextRead = true |
| 387 | yield* sharding.pollStorage |
| 388 | yield* Deferred.await(readStarted) |
| 389 | |
| 390 | // Complete the request while the read is paused. |
| 391 | yield* Queue.offer(state.messages, void 0) |
| 392 | yield* Fiber.join(firstRun) |
| 393 | yield* TestClock.adjust(1) |
| 394 | assert.isTrue(yield* sharding.reset(request.requestId)) |
| 395 | yield* sharding.pollStorage |
| 396 | claimed.length = 0 |
| 397 | |
| 398 | // The reset must be redelivered before its new claim expires. |
| 399 | yield* Deferred.succeed(releaseRead, void 0) |
| 400 | yield* TestClock.adjust(5000) |
| 401 | assert.include(claimed, request.requestId) |
| 402 | const deliveriesBeforeClaimExpiry = Queue.sizeUnsafe(state.envelopes) |
| 403 | |
| 404 | yield* TestClock.adjust("10 minutes") |
| 405 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 406 | assert.strictEqual( |
| 407 | deliveriesBeforeClaimExpiry, |
| 408 | 1, |
| 409 | "reset request was claimed but not redelivered before claim expiry" |
| 410 | ) |
| 411 | }).pipe(Effect.provide(CappedSharding({ refreshAssignmentsInterval: 1000 }, (storage) => ({ |
| 412 | ...storage, |
| 413 | unprocessedMessages: (shardIds, options) => |
| 414 | Effect.gen(function*() { |
| 415 | if (pauseNextRead) { |
| 416 | pauseNextRead = false |
| 417 | yield* Deferred.succeed(readStarted, void 0) |
| 418 | yield* Deferred.await(releaseRead) |
| 419 | } |
| 420 | const messages = yield* storage.unprocessedMessages(shardIds, options) |
| 421 | for (const message of messages) claimed.push(message.envelope.requestId) |
| 422 | return messages |
| 423 | }) |
| 424 | })))) |
| 425 | })) |
| 426 | |
| 427 | for (const pauseReply of [false, true]) { |
| 428 | it.effect( |
| 429 | pauseReply |
| 430 | ? "redelivers a remote reset published before active-request cleanup" |
| 431 | : "redelivers a remote reset during a pending storage read", |
| 432 | () => |
| 433 | Effect.gen(function*() { |
| 434 | const readStarted = yield* Deferred.make<void>() |
| 435 | const releaseRead = yield* Deferred.make<void>() |
| 436 | const replyPublished = yield* Deferred.make<void>() |
| 437 | const releaseReply = yield* Deferred.make<void>() |
| 438 | let pauseNextRead = false |
| 439 | let pauseNextReply = pauseReply |
| 440 | const released: Array<Snowflake.Snowflake> = [] |
| 441 | |
| 442 | yield* Effect.gen(function*() { |
| 443 | yield* TestClock.adjust(1) |
| 444 | const sharding = yield* Sharding.Sharding |
| 445 | const driver = yield* MessageStorage.MemoryDriver |
| 446 | // Share persistence without sharing runner state. |
| 447 | const remoteStorage = yield* MessageStorage.makeEncoded(driver.encoded).pipe( |
| 448 | Effect.provide(Snowflake.layerGenerator.pipe(Layer.provide(ShardingConfig.layerDefaults))) |
| 449 | ) |
| 450 | const state = yield* TestEntityState |
| 451 | const client = (yield* TestEntity.client)("remote-reset-race") |
| 452 | const firstRun = yield* client.RequestWithKey({ key: "run" }).pipe( |
| 453 | Effect.forkChild({ startImmediately: true }) |
| 454 | ) |
| 455 | const request = yield* Queue.take(state.envelopes) |
| 456 | |
| 457 | // Reclaiming an active request must not redeliver it or release its claim. |
| 458 | yield* remoteStorage.resetRequests([request.requestId]) |
| 459 | yield* sharding.pollStorage |
| 460 | yield* TestClock.adjust(1) |
| 461 | expect(Queue.sizeUnsafe(state.envelopes)).toEqual(0) |
| 462 | expect(released).toEqual([]) |
| 463 | |
| 464 | pauseNextRead = true |
| 465 | yield* sharding.pollStorage |
| 466 | yield* Deferred.await(readStarted) |
| 467 | yield* Queue.offer(state.messages, void 0) |
| 468 | yield* Fiber.join(firstRun) |
| 469 | if (pauseReply) yield* Deferred.await(replyPublished) |
| 470 | yield* TestClock.adjust(1) |
| 471 | |
| 472 | // Simulate Sharding.reset on another runner. |
| 473 | yield* remoteStorage.clearReplies(request.requestId) |
| 474 | yield* Deferred.succeed(releaseRead, void 0) |
| 475 | yield* TestClock.adjust(1) |
| 476 | expect(released).toContain(request.requestId) |
| 477 | if (pauseReply) { |
| 478 | expect(Queue.sizeUnsafe(state.envelopes)).toEqual(0) |
| 479 | yield* Deferred.succeed(releaseReply, void 0) |
| 480 | yield* TestClock.adjust(1) |
| 481 | } |
| 482 | yield* sharding.pollStorage |
| 483 | yield* TestClock.adjust(5000) |
| 484 | expect(Queue.sizeUnsafe(state.envelopes)).toEqual(1) |
| 485 | expect((yield* Queue.take(state.envelopes)).requestId).toEqual(request.requestId) |
| 486 | yield* sharding.pollStorage |
| 487 | yield* TestClock.adjust(5000) |
| 488 | expect(Queue.sizeUnsafe(state.envelopes)).toEqual(0) |
| 489 | }).pipe(Effect.provide(CappedSharding({}, (storage) => ({ |
| 490 | ...storage, |
| 491 | resetRequests: (ids) => |
| 492 | Effect.gen(function*() { |
| 493 | released.push(...ids) |
| 494 | yield* storage.resetRequests(ids) |
| 495 | }), |
| 496 | unprocessedMessages: (shards, options) => |
| 497 | Effect.gen(function*() { |
| 498 | if (pauseNextRead) { |
| 499 | pauseNextRead = false |
| 500 | yield* Deferred.succeed(readStarted, void 0) |
| 501 | yield* Deferred.await(releaseRead) |
| 502 | } |
| 503 | const messages = yield* storage.unprocessedMessages(shards, options) |
| 504 | return messages.map((message) => |
| 505 | message._tag !== "IncomingRequest" ? |
| 506 | message : |
| 507 | new Message.IncomingRequest({ |
| 508 | ...message, |
| 509 | respond: (reply) => |
| 510 | Effect.gen(function*() { |
| 511 | yield* message.respond(reply) |
| 512 | if (pauseNextReply) { |
| 513 | pauseNextReply = false |
| 514 | yield* Deferred.succeed(replyPublished, void 0) |
| 515 | yield* Deferred.await(releaseReply) |
| 516 | } |
| 517 | }) |
| 518 | }) |
| 519 | ) |
| 520 | }) |
| 521 | })))) |
| 522 | }) |
| 523 | ) |
| 524 | } |
| 525 | |
| 526 | it.effect("delivers volatile requests directly to the entity", () => |
| 527 | Effect.gen(function*() { |
| 528 | yield* TestClock.adjust(1) |
| 529 | const makeClient = yield* TestEntity.client |
| 530 | const client = makeClient("1") |
| 531 | const user = yield* client.GetUserVolatile({ id: 1 }) |
| 532 | expect(user).toEqual(new User({ id: 1, name: "User 1" })) |
| 533 | }).pipe(Effect.provide(TestSharding))) |
| 534 | |
| 535 | it.effect("does not freeze the first caller's context into the entity server", () => |
| 536 | Effect.gen(function*() { |
| 537 | yield* TestClock.adjust(1) |
| 538 | const makeClient = yield* ContextBleedEntity.client |
| 539 | const client = makeClient("1") |
| 540 | |
| 541 | const first = yield* client.ReadCaller().pipe(Effect.provideService(CallerId, "A")) |
| 542 | expect(first).toEqual("A") |
| 543 | |
| 544 | const second = yield* client.ReadCaller() |
| 545 | expect(second).toEqual("none") |
| 546 | |
| 547 | const durable = yield* client.ReadCallerPersisted() |
| 548 | expect(durable).toEqual("none") |
| 549 | }).pipe(Effect.provide(ContextBleedSharding))) |
| 550 | |
| 551 | it.effect("uses services provided when registering an entity", () => |
| 552 | Effect.gen(function*() { |
| 553 | const sharding = yield* Sharding.Sharding |
| 554 | yield* sharding.registerEntity(RegistrationContextEntity, RegistrationContextHandlers).pipe( |
| 555 | Effect.provideService(RegistrationContext, "registration") |
| 556 | ) |
| 557 | yield* TestClock.adjust(1) |
| 558 | |
| 559 | const client = (yield* RegistrationContextEntity.client)("1") |
| 560 | expect(yield* client.Read()).toEqual("registration") |
| 561 | }).pipe( |
| 562 | Effect.provide(TestSharding), |
| 563 | Effect.provideService(RegistrationContext, "construction"), |
| 564 | Effect.scoped |
| 565 | )) |
| 566 | |
| 567 | it.effect("persists durable requests until the entity replies", () => |
| 568 | Effect.gen(function*() { |
| 569 | yield* TestClock.adjust(1) |
| 570 | const driver = yield* MessageStorage.MemoryDriver |
| 571 | const makeClient = yield* TestEntity.client |
| 572 | const client = makeClient("1") |
| 573 | const user = yield* client.GetUser({ id: 1 }) |
| 574 | expect(user).toEqual(new User({ id: 1, name: "User 1" })) |
| 575 | expect(driver.journal.length).toEqual(1) |
| 576 | expect(driver.unprocessed.size).toEqual(0) |
| 577 | }).pipe(Effect.provide(TestSharding))) |
| 578 | |
| 579 | it.live("defects instead of hanging when persisted failures contain non-JSON Error values", () => |
| 580 | Effect.gen(function*() { |
| 581 | const driver = yield* MessageStorage.MemoryDriver |
| 582 | const makeClient = yield* TestEntity.client |
| 583 | const client = makeClient("1") |
| 584 | const cause = yield* client.Fail().pipe( |
| 585 | Effect.timeout("2 seconds"), |
| 586 | Effect.sandbox, |
| 587 | Effect.flip |
| 588 | ) |
| 589 | assert(Cause.hasDies(cause)) |
| 590 | assert.include(Cause.pretty(cause), "MalformedMessage") |
| 591 | assert.strictEqual(driver.replyIds.size, 1) |
| 592 | assert.strictEqual(driver.unprocessed.size, 0) |
| 593 | }).pipe(Effect.provide(TestSharding))) |
| 594 | |
| 595 | it.effect("routes durable interrupts through storage", () => |
| 596 | Effect.gen(function*() { |
| 597 | const driver = yield* MessageStorage.MemoryDriver |
| 598 | const state = yield* TestEntityState |
| 599 | const makeClient = yield* TestEntity.client |
| 600 | yield* TestClock.adjust(1) |
| 601 | const client = makeClient("1") |
| 602 | |
| 603 | const fiber = yield* client.Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 604 | yield* TestClock.adjust(1) |
| 605 | yield* Fiber.interrupt(fiber) |
| 606 | |
| 607 | yield* TestClock.adjust(1) |
| 608 | expect(driver.journal.length).toEqual(2) |
| 609 | expect(driver.replyIds.size).toEqual(1) |
| 610 | expect(Queue.sizeUnsafe(state.interrupts)).toEqual(1) |
| 611 | }).pipe(Effect.provide(TestSharding))) |
| 612 | |
| 613 | it.effect("interrupts aren't sent for durable messages on shutdown", () => |
| 614 | Effect.gen(function*() { |
| 615 | let driver!: MessageStorage.MemoryDriver["Service"] |
| 616 | yield* Effect.gen(function*() { |
| 617 | driver = yield* MessageStorage.MemoryDriver |
| 618 | const makeClient = yield* TestEntity.client |
| 619 | yield* TestClock.adjust(1) |
| 620 | const client = makeClient("1") |
| 621 | yield* client.Never().pipe(Effect.forkChild) |
| 622 | yield* TestClock.adjust(1) |
| 623 | }).pipe(Effect.provide(TestSharding)) |
| 624 | |
| 625 | // request, client interrupt is dropped |
| 626 | expect(driver.journal.length).toEqual(1) |
| 627 | // server interrupt is not sent |
| 628 | expect(driver.replyIds.size).toEqual(0) |
| 629 | })) |
| 630 | |
| 631 | for (const persisted of [false, true] as const) { |
| 632 | for (const preemptiveShutdown of [true, false]) { |
| 633 | it.live( |
| 634 | `shutdown completes when a finalizing entity sends an outgoing message (persisted=${persisted}, preemptiveShutdown=${preemptiveShutdown})`, |
| 635 | () => |
| 636 | Effect.gen(function*() { |
| 637 | const discardExit = yield* Deferred.make<Exit.Exit<void, unknown>>() |
| 638 | const Receiver = Entity.make("ShutdownDeadlockReceiver", [ |
| 639 | Rpc.make("Ping").annotate(ClusterSchema.Persisted, persisted) |
| 640 | ]) |
| 641 | const ReceiverLayer = Receiver.toLayer({ Ping: () => Effect.void }) |
| 642 | |
| 643 | const Sender = Entity.make("ShutdownDeadlockSender", [ |
| 644 | Rpc.make("Arm").annotate(ClusterSchema.Persisted, false) |
| 645 | ]) |
| 646 | const SenderLayer = Sender.toLayer(Effect.gen(function*() { |
| 647 | const receiver = yield* Receiver.client |
| 648 | // finalizer sends an outgoing message during entity teardown |
| 649 | yield* Effect.addFinalizer(() => |
| 650 | Effect.uninterruptible( |
| 651 | Effect.sleep(300).pipe( |
| 652 | Effect.andThen(receiver("peer").Ping(void 0, { discard: true })), |
| 653 | Effect.exit, |
| 654 | Effect.flatMap((exit) => Deferred.succeed(discardExit, exit)) |
| 655 | ) |
| 656 | ) |
| 657 | ) |
| 658 | return { Arm: () => Effect.void } |
| 659 | })) |
| 660 | |
| 661 | const env = Layer.mergeAll(SenderLayer, ReceiverLayer).pipe( |
| 662 | Layer.provideMerge(Sharding.layer), |
| 663 | Layer.provide(RunnerStorage.layerMemory), |
| 664 | Layer.provide(RunnerHealth.layerNoop), |
| 665 | Layer.provide(Runners.layerNoop), |
| 666 | Layer.provideMerge(MessageStorage.layerMemory), |
| 667 | Layer.provide(ShardingConfig.layer({ |
| 668 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 669 | shardsPerGroup: 8, |
| 670 | entityTerminationTimeout: 0, |
| 671 | entityMessagePollInterval: 50, |
| 672 | refreshAssignmentsInterval: 20, |
| 673 | sendRetryInterval: 10, |
| 674 | preemptiveShutdown |
| 675 | })) |
| 676 | ) |
| 677 | |
| 678 | const shardAcquired = Latch.makeUnsafe() |
| 679 | const armed = Latch.makeUnsafe() |
| 680 | let driver!: MessageStorage.MemoryDriver["Service"] |
| 681 | const runFiber = yield* Effect.gen(function*() { |
| 682 | driver = yield* MessageStorage.MemoryDriver |
| 683 | const sharding = yield* Sharding.Sharding |
| 684 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 685 | while (!sharding.hasShardId(shardId)) { |
| 686 | yield* Effect.sleep(5) |
| 687 | } |
| 688 | yield* shardAcquired.open |
| 689 | yield* (yield* Sender.client)("1").Arm() |
| 690 | yield* armed.open |
| 691 | return yield* Effect.never |
| 692 | }).pipe(Effect.provide(env), Effect.scoped, Effect.forkDetach) |
| 693 | |
| 694 | const acquired = yield* shardAcquired.await.pipe(Effect.timeoutOption("8 seconds")) |
| 695 | if (Option.isNone(acquired)) { |
| 696 | runFiber.interruptUnsafe() |
| 697 | yield* Fiber.await(runFiber) |
| 698 | } |
| 699 | assert(Option.isSome(acquired), "Timed out waiting for sender shard acquisition") |
| 700 | yield* armed.await |
| 701 | |
| 702 | // Interrupting the fiber closes the Sharding scope, running the entity |
| 703 | // finalizer (and its outgoing send) as part of teardown |
| 704 | runFiber.interruptUnsafe() |
| 705 | const completed = yield* Fiber.await(runFiber).pipe(Effect.timeoutOption("4 seconds")) |
| 706 | assert(Option.isSome(completed), "Sharding scope-close hung during shutdown (deadlock)") |
| 707 | assert(Exit.isSuccess(yield* Deferred.await(discardExit)), "discard send failed during shutdown") |
| 708 | assert.strictEqual(driver.journal.length, persisted ? 1 : 0) |
| 709 | }), |
| 710 | 20_000 |
| 711 | ) |
| 712 | } |
| 713 | |
| 714 | it.live( |
| 715 | `shutdown completes when a finalizing entity awaits a reply from an unroutable entity (persisted=${persisted})`, |
| 716 | () => |
| 717 | Effect.gen(function*() { |
| 718 | const requestExit = yield* Deferred.make<Exit.Exit<void, unknown>>() |
| 719 | const Receiver = Entity.make("StrandReceiver", [ |
| 720 | Rpc.make("Ping").annotate(ClusterSchema.Persisted, persisted) |
| 721 | ]) |
| 722 | const ReceiverLayer = Receiver.toLayer({ Ping: () => Effect.void }) |
| 723 | |
| 724 | const Sender = Entity.make("StrandSender", [ |
| 725 | Rpc.make("Arm").annotate(ClusterSchema.Persisted, false) |
| 726 | ]) |
| 727 | const SenderLayer = Sender.toLayer(Effect.gen(function*() { |
| 728 | const receiver = yield* Receiver.client |
| 729 | // uninterruptible finalizer that awaits a reply from another entity |
| 730 | yield* Effect.addFinalizer(() => |
| 731 | Effect.uninterruptible( |
| 732 | Effect.sleep(300).pipe( |
| 733 | Effect.andThen(receiver("peer").Ping()), |
| 734 | Effect.exit, |
| 735 | Effect.flatMap((exit) => Deferred.succeed(requestExit, exit)) |
| 736 | ) |
| 737 | ) |
| 738 | ) |
| 739 | return { Arm: () => Effect.void } |
| 740 | })) |
| 741 | |
| 742 | const env = Layer.mergeAll(SenderLayer, ReceiverLayer).pipe( |
| 743 | Layer.provideMerge(Sharding.layer), |
| 744 | Layer.provide(RunnerStorage.layerMemory), |
| 745 | Layer.provide(RunnerHealth.layerNoop), |
| 746 | Layer.provide(Runners.layerNoop), |
| 747 | Layer.provideMerge(MessageStorage.layerMemory), |
| 748 | Layer.provide(ShardingConfig.layer({ |
| 749 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 750 | shardsPerGroup: 8, |
| 751 | entityTerminationTimeout: 0, |
| 752 | entityMessagePollInterval: 50, |
| 753 | refreshAssignmentsInterval: 20, |
| 754 | sendRetryInterval: 10 |
| 755 | })) |
| 756 | ) |
| 757 | |
| 758 | const shardAcquired = Latch.makeUnsafe() |
| 759 | const armed = Latch.makeUnsafe() |
| 760 | let driver!: MessageStorage.MemoryDriver["Service"] |
| 761 | const runFiber = yield* Effect.gen(function*() { |
| 762 | driver = yield* MessageStorage.MemoryDriver |
| 763 | const sharding = yield* Sharding.Sharding |
| 764 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 765 | while (!sharding.hasShardId(shardId)) { |
| 766 | yield* Effect.sleep(5) |
| 767 | } |
| 768 | yield* shardAcquired.open |
| 769 | yield* (yield* Sender.client)("1").Arm() |
| 770 | yield* armed.open |
| 771 | return yield* Effect.never |
| 772 | }).pipe(Effect.provide(env), Effect.scoped, Effect.forkDetach) |
| 773 | |
| 774 | const acquired = yield* shardAcquired.await.pipe(Effect.timeoutOption("8 seconds")) |
| 775 | if (Option.isNone(acquired)) { |
| 776 | runFiber.interruptUnsafe() |
| 777 | yield* Fiber.await(runFiber) |
| 778 | } |
| 779 | assert(Option.isSome(acquired), "Timed out waiting for sender shard acquisition") |
| 780 | yield* armed.await |
| 781 | |
| 782 | runFiber.interruptUnsafe() |
| 783 | const completed = yield* Fiber.await(runFiber).pipe(Effect.timeoutOption("4 seconds")) |
| 784 | assert(Option.isSome(completed), "Sharding scope-close hung during shutdown (stranded caller)") |
| 785 | assert(!Exit.hasDies(completed.value), "finalizer defected instead of failing cleanly") |
| 786 | const exit = yield* Deferred.await(requestExit) |
| 787 | assert(Exit.isFailure(exit), "request succeeded instead of failing during shutdown") |
| 788 | if (persisted) { |
| 789 | // A persisted request is durable, so a transient routing state is |
| 790 | // not an error: the caller is interrupted instead, and the request |
| 791 | // is served under the next owner. |
| 792 | assert(Cause.hasInterruptsOnly(exit.cause), "persisted request was not interrupted") |
| 793 | } else { |
| 794 | const failure = Cause.findErrorOption(exit.cause) |
| 795 | assert(Option.isSome(failure), "request did not fail with a typed error") |
| 796 | assert(failure.value instanceof ClusterError.EntityNotAssignedToRunner) |
| 797 | } |
| 798 | assert.strictEqual(driver.journal.length, persisted ? 1 : 0) |
| 799 | }), |
| 800 | 20_000 |
| 801 | ) |
| 802 | } |
| 803 | |
| 804 | it.live("fails a stream when its chunk acknowledgement is abandoned during shutdown", () => |
| 805 | Effect.gen(function*() { |
| 806 | const chunks = yield* Queue.unbounded<number>() |
| 807 | const streamStarted = yield* Deferred.make<void>() |
| 808 | const streamExit = yield* Deferred.make<Exit.Exit<void, unknown>>() |
| 809 | |
| 810 | const Receiver = Entity.make("ShutdownStreamReceiver", [ |
| 811 | Rpc.make("Values", { success: Schema.Number, stream: true }).annotate(ClusterSchema.Persisted, true) |
| 812 | ]) |
| 813 | const ReceiverLayer = Receiver.toLayer({ |
| 814 | Values: () => Stream.fromQueue(chunks).pipe(Stream.onStart(Deferred.succeed(streamStarted, void 0))) |
| 815 | }) |
| 816 | |
| 817 | const Sender = Entity.make("ShutdownStreamSender", [ |
| 818 | Rpc.make("Arm").annotate(ClusterSchema.Persisted, false) |
| 819 | ]) |
| 820 | const SenderLayer = Sender.toLayer(Effect.gen(function*() { |
| 821 | const receiver = yield* Receiver.client |
| 822 | const streamFiber = yield* receiver("peer").Values().pipe( |
| 823 | Stream.runDrain, |
| 824 | Effect.uninterruptible, |
| 825 | Effect.forkScoped({ startImmediately: true }) |
| 826 | ) |
| 827 | yield* Deferred.await(streamStarted) |
| 828 | yield* Effect.addFinalizer(() => |
| 829 | Effect.uninterruptible( |
| 830 | Queue.offer(chunks, 1).pipe( |
| 831 | Effect.andThen(Fiber.await(streamFiber)), |
| 832 | Effect.flatMap((exit) => Deferred.succeed(streamExit, exit)) |
| 833 | ) |
| 834 | ) |
| 835 | ) |
| 836 | return { Arm: () => Effect.void } |
| 837 | })) |
| 838 | |
| 839 | const env = Layer.mergeAll(SenderLayer, ReceiverLayer).pipe( |
| 840 | Layer.provideMerge(Sharding.layer), |
| 841 | Layer.provide(RunnerStorage.layerMemory), |
| 842 | Layer.provide(RunnerHealth.layerNoop), |
| 843 | Layer.provide(Runners.layerNoop), |
| 844 | Layer.provideMerge(MessageStorage.layerMemory), |
| 845 | Layer.provide(ShardingConfig.layer({ |
| 846 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 847 | shardsPerGroup: 8, |
| 848 | entityTerminationTimeout: 1000, |
| 849 | entityMessagePollInterval: 50, |
| 850 | refreshAssignmentsInterval: 20, |
| 851 | sendRetryInterval: 10 |
| 852 | })) |
| 853 | ) |
| 854 | |
| 855 | const shardAcquired = Latch.makeUnsafe() |
| 856 | const armed = Latch.makeUnsafe() |
| 857 | let driver!: MessageStorage.MemoryDriver["Service"] |
| 858 | const runFiber = yield* Effect.gen(function*() { |
| 859 | driver = yield* MessageStorage.MemoryDriver |
| 860 | const sharding = yield* Sharding.Sharding |
| 861 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 862 | while (!sharding.hasShardId(shardId)) { |
| 863 | yield* Effect.sleep(5) |
| 864 | } |
| 865 | yield* shardAcquired.open |
| 866 | yield* (yield* Sender.client)("1").Arm() |
| 867 | yield* armed.open |
| 868 | return yield* Effect.never |
| 869 | }).pipe(Effect.provide(env), Effect.scoped, Effect.forkDetach) |
| 870 | |
| 871 | const acquired = yield* shardAcquired.await.pipe(Effect.timeoutOption("8 seconds")) |
| 872 | if (Option.isNone(acquired)) { |
| 873 | runFiber.interruptUnsafe() |
| 874 | yield* Fiber.await(runFiber) |
| 875 | } |
| 876 | assert(Option.isSome(acquired), "Timed out waiting for sender shard acquisition") |
| 877 | yield* armed.await |
| 878 | |
| 879 | runFiber.interruptUnsafe() |
| 880 | const completed = yield* Fiber.await(runFiber).pipe(Effect.timeoutOption("4 seconds")) |
| 881 | assert(Option.isSome(completed), "Sharding scope-close hung after abandoning a stream chunk acknowledgement") |
| 882 | const exit = yield* Deferred.await(streamExit) |
| 883 | assert(Exit.isFailure(exit), "stream succeeded instead of failing during shutdown") |
| 884 | const failure = Cause.findErrorOption(exit.cause) |
| 885 | assert(Option.isSome(failure), "stream did not fail with a typed error") |
| 886 | assert(failure.value instanceof ClusterError.EntityNotAssignedToRunner) |
| 887 | assert.strictEqual(driver.journal.filter((envelope) => envelope._tag === "AckChunk").length, 1) |
| 888 | }), 20_000) |
| 889 | |
| 890 | it.effect("interrupts are sent for volatile messages on shutdown", () => |
| 891 | Effect.gen(function*() { |
| 892 | let interrupted = false |
| 893 | const testClock = (yield* Clock.Clock) as TestClock.TestClock |
| 894 | |
| 895 | yield* Effect.gen(function*() { |
| 896 | const makeClient = yield* TestEntity.client |
| 897 | const client = makeClient("1") |
| 898 | const fiber = yield* client.NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) |
| 899 | yield* TestClock.adjust(1) |
| 900 | const config = yield* ShardingConfig.ShardingConfig |
| 901 | ;(config as any).runnerAddress = Option.some(RunnerAddress.make("localhost", 1234)) |
| 902 | fiber.currentDispatcher.scheduleTask(() => { |
| 903 | fiber.interruptUnsafe() |
| 904 | Effect.runFork(testClock.adjust(30000)) |
| 905 | }, 0) |
| 906 | }).pipe( |
| 907 | Effect.provide(TestShardingWithoutRunners.pipe( |
| 908 | Layer.provide( |
| 909 | Layer.effect(Runners.Runners)( |
| 910 | Effect.gen(function*() { |
| 911 | const runners = yield* Runners.makeNoop |
| 912 | return { |
| 913 | ...runners, |
| 914 | send(options) { |
| 915 | if (options.message.envelope._tag === "Interrupt") { |
| 916 | interrupted = true |
| 917 | return Effect.void |
| 918 | } |
| 919 | return runners.send(options) |
| 920 | } |
| 921 | } |
| 922 | }) |
| 923 | ) |
| 924 | ), |
| 925 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 926 | Layer.provideMerge(ShardingConfig.layer({ |
| 927 | entityMailboxCapacity: 10, |
| 928 | entityTerminationTimeout: 30000, |
| 929 | entityMessagePollInterval: 5000, |
| 930 | sendRetryInterval: 100, |
| 931 | refreshAssignmentsInterval: 100 |
| 932 | })) |
| 933 | )) |
| 934 | ) |
| 935 | |
| 936 | assert.isTrue(interrupted) |
| 937 | })) |
| 938 | |
| 939 | it.effect("malformed message in storage", () => |
| 940 | Effect.gen(function*() { |
| 941 | const driver = yield* MessageStorage.MemoryDriver |
| 942 | const makeClient = yield* TestEntity.client |
| 943 | yield* TestClock.adjust(1) |
| 944 | const client = makeClient("1") |
| 945 | |
| 946 | const fiber = yield* client.Never().pipe(Effect.forkChild) |
| 947 | yield* TestClock.adjust(1) |
| 948 | |
| 949 | const request = driver.journal[0] |
| 950 | yield* driver.encoded.saveEnvelope({ |
| 951 | envelope: { |
| 952 | id: "boom", |
| 953 | _tag: "Interrupt", |
| 954 | requestId: request.requestId, |
| 955 | address: { |
| 956 | shardId: request.address.shardId |
| 957 | } as any |
| 958 | }, |
| 959 | primaryKey: null, |
| 960 | deliverAt: null |
| 961 | }) |
| 962 | |
| 963 | // wait for storage to poll |
| 964 | yield* TestClock.adjust(5000) |
| 965 | |
| 966 | const exit = fiber.pollUnsafe() |
| 967 | assert(exit && Exit.isFailure(exit) && Cause.hasDies(exit.cause)) |
| 968 | |
| 969 | // malformed message should be left in the database |
| 970 | expect(driver.journal.length).toEqual(2) |
| 971 | // defect reply should be sent |
| 972 | expect(driver.replyIds.size).toEqual(1) |
| 973 | |
| 974 | const reply = driver.requests.get(request.requestId)!.replies[0] |
| 975 | assert(reply._tag === "WithExit" && reply.exit._tag === "Failure" && reply.exit.cause[0]._tag === "Die") |
| 976 | }).pipe(Effect.provide(TestSharding))) |
| 977 | |
| 978 | it.effect("fails volatile requests immediately when the mailbox is full", () => |
| 979 | Effect.gen(function*() { |
| 980 | const makeClient = yield* TestEntity.client |
| 981 | yield* TestClock.adjust(1) |
| 982 | const client = makeClient("1") |
| 983 | |
| 984 | yield* client.NeverVolatile().pipe(Effect.forkChild, Effect.replicateEffect(10)) |
| 985 | yield* TestClock.adjust(1) |
| 986 | const error = yield* client.NeverVolatile().pipe(Effect.flip) |
| 987 | assert.strictEqual(error._tag, "MailboxFull") |
| 988 | }).pipe(Effect.provide(TestSharding))) |
| 989 | |
| 990 | it.effect("durable messages are retried when mailbox is full", () => |
| 991 | Effect.gen(function*() { |
| 992 | const requestedIds = yield* Queue.make<Array<Snowflake.Snowflake>>() |
| 993 | yield* Effect.gen(function*() { |
| 994 | const state = yield* TestEntityState |
| 995 | const makeClient = yield* TestEntity.client |
| 996 | yield* TestClock.adjust(1) |
| 997 | const client = makeClient("1") |
| 998 | |
| 999 | const fibers = yield* client.NeverFork().pipe(Effect.forkChild, Effect.replicateEffect(11)) |
| 1000 | yield* TestClock.adjust(1) |
| 1001 | |
| 1002 | // wait for entity to go into resume mode and request ids |
| 1003 | const ids = yield* Queue.take(requestedIds) |
| 1004 | assert.strictEqual(ids.length, 1) |
| 1005 | |
| 1006 | // test entity should still only have 10 requests |
| 1007 | assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 10) |
| 1008 | |
| 1009 | // interrupt first request |
| 1010 | yield* Fiber.interrupt(fibers[0]) |
| 1011 | yield* TestClock.adjust(100) // let retry happen |
| 1012 | |
| 1013 | // last request should come through |
| 1014 | assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 11) |
| 1015 | |
| 1016 | // interrupt second request, now the entity should be back in the main storage loop |
| 1017 | yield* Fiber.interrupt(fibers[1]) |
| 1018 | |
| 1019 | // send another request within mailbox capacity |
| 1020 | yield* client.NeverFork().pipe(Effect.forkChild) |
| 1021 | yield* TestClock.adjust(1) |
| 1022 | yield* Fiber.interruptAll(fibers) |
| 1023 | yield* TestClock.adjust(100) |
| 1024 | |
| 1025 | // no more ids should have been requested from entity catch up |
| 1026 | assert.deepStrictEqual(Queue.sizeUnsafe(requestedIds), 0) |
| 1027 | }).pipe(Effect.provide(TestShardingWithoutStorage.pipe( |
| 1028 | Layer.updateService(MessageStorage.MessageStorage, (storage) => ({ |
| 1029 | ...storage, |
| 1030 | unprocessedMessagesById(messageIds) { |
| 1031 | Queue.offerUnsafe(requestedIds, Array.fromIterable(messageIds)) |
| 1032 | return storage.unprocessedMessagesById(messageIds) |
| 1033 | } |
| 1034 | })), |
| 1035 | Layer.provide(MessageStorage.layerMemory), |
| 1036 | Layer.provide(TestShardingConfig) |
| 1037 | ))) |
| 1038 | })) |
| 1039 | |
| 1040 | it.effect("interrupt for future request works while mailbox is full", () => |
| 1041 | Effect.gen(function*() { |
| 1042 | const state = yield* TestEntityState |
| 1043 | const makeClient = yield* TestEntity.client |
| 1044 | yield* TestClock.adjust(1) |
| 1045 | const client = makeClient("1") |
| 1046 | |
| 1047 | const fibers = yield* client.NeverFork().pipe( |
| 1048 | Effect.forkChild({ startImmediately: true }), |
| 1049 | Effect.replicateEffect(12) |
| 1050 | ) |
| 1051 | yield* TestClock.adjust(1) |
| 1052 | |
| 1053 | assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 10) |
| 1054 | |
| 1055 | // interrupt 11th request |
| 1056 | yield* Fiber.interrupt(fibers[10]) |
| 1057 | yield* TestClock.adjust(100) // let retry happen |
| 1058 | // interrupt first request, and let the 11th request come through |
| 1059 | yield* Fiber.interrupt(fibers[0]) |
| 1060 | yield* TestClock.adjust(100) // let retry happen |
| 1061 | |
| 1062 | assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 12) |
| 1063 | // second interrupt should be sent |
| 1064 | assert.deepStrictEqual(Queue.sizeUnsafe(state.interrupts), 2) |
| 1065 | }).pipe(Effect.provide(TestSharding))) |
| 1066 | |
| 1067 | it.effect("delivers durable streams and acknowledges each chunk", () => |
| 1068 | Effect.gen(function*() { |
| 1069 | const driver = yield* MessageStorage.MemoryDriver |
| 1070 | yield* TestClock.adjust(1) |
| 1071 | const makeClient = yield* TestEntity.client |
| 1072 | const client = makeClient("1") |
| 1073 | const users = yield* client.GetAllUsers({ ids: [1, 2, 3] }).pipe( |
| 1074 | Stream.runCollect |
| 1075 | ) |
| 1076 | expect(users).toEqual([ |
| 1077 | new User({ id: 1, name: "User 1" }), |
| 1078 | new User({ id: 2, name: "User 2" }), |
| 1079 | new User({ id: 3, name: "User 3" }) |
| 1080 | ]) |
| 1081 | |
| 1082 | // 1 request, 3 acks, 4 replies |
| 1083 | expect(driver.journal.length).toEqual(4) |
| 1084 | expect(driver.replyIds.size).toEqual(4) |
| 1085 | }).pipe(Effect.provide(TestSharding))) |
| 1086 | |
| 1087 | it.effect("durable stream while mailbox is full", () => |
| 1088 | Effect.gen(function*() { |
| 1089 | const requestedIds = yield* Queue.make<Array<Snowflake.Snowflake>>() |
| 1090 | yield* Effect.gen(function*() { |
| 1091 | const state = yield* TestEntityState |
| 1092 | const makeClient = yield* TestEntity.client |
| 1093 | const client = makeClient("1") |
| 1094 | |
| 1095 | const fibers = yield* client.NeverFork().pipe( |
| 1096 | Effect.forkChild({ startImmediately: true }), |
| 1097 | Effect.replicateEffect(10) |
| 1098 | ) |
| 1099 | yield* TestClock.adjust(1) |
| 1100 | |
| 1101 | const fiber = yield* client.GetAllUsers({ ids: [1, 2, 3] }).pipe( |
| 1102 | Stream.runCollect, |
| 1103 | Effect.forkChild({ startImmediately: true }) |
| 1104 | ) |
| 1105 | |
| 1106 | // make sure entity doesn't leave resume mode |
| 1107 | yield* client.NeverFork().pipe(Effect.forkChild({ startImmediately: true })) |
| 1108 | yield* client.NeverFork().pipe(Effect.forkChild({ startImmediately: true })) |
| 1109 | |
| 1110 | // wait for entity to go into resume mode and request ids |
| 1111 | const ids = yield* Queue.take(requestedIds) |
| 1112 | assert.strictEqual(ids.length, 3) |
| 1113 | assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 10) |
| 1114 | |
| 1115 | // interrupt first request |
| 1116 | yield* Fiber.interrupt(fibers[0]) |
| 1117 | yield* TestClock.adjust(500) // let retry happen |
| 1118 | |
| 1119 | // last request + NeverFork should come through |
| 1120 | assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 12) |
| 1121 | |
| 1122 | // acks should be allowed to be sent |
| 1123 | const users = yield* Fiber.join(fiber) |
| 1124 | expect(users).toEqual([ |
| 1125 | new User({ id: 1, name: "User 1" }), |
| 1126 | new User({ id: 2, name: "User 2" }), |
| 1127 | new User({ id: 3, name: "User 3" }) |
| 1128 | ]) |
| 1129 | |
| 1130 | const driver = yield* MessageStorage.MemoryDriver |
| 1131 | // 13 requests, 3 acks, 1 interrupt, 5 replies |
| 1132 | assert.strictEqual(driver.journal.length, 13 + 3 + 1) |
| 1133 | assert.strictEqual(driver.replyIds.size, 1 + 4) |
| 1134 | }).pipe(Effect.provide(TestShardingWithoutStorage.pipe( |
| 1135 | Layer.provideMerge(Layer.effect(MessageStorage.MemoryDriver)(MessageStorage.MemoryDriver)), |
| 1136 | Layer.updateService(MessageStorage.MessageStorage, (storage) => ({ |
| 1137 | ...storage, |
| 1138 | unprocessedMessagesById(messageIds) { |
| 1139 | Queue.offerUnsafe(requestedIds, Array.fromIterable(messageIds)) |
| 1140 | return storage.unprocessedMessagesById(messageIds) |
| 1141 | } |
| 1142 | })), |
| 1143 | Layer.provide(MessageStorage.layerMemory), |
| 1144 | Layer.provide(TestShardingConfig) |
| 1145 | ))) |
| 1146 | })) |
| 1147 | |
| 1148 | it.effect("durable messages are retried on restart", () => |
| 1149 | Effect.gen(function*() { |
| 1150 | const EnvLayer = TestShardingWithoutState.pipe( |
| 1151 | Layer.provide(Runners.layerNoop), |
| 1152 | Layer.provide(TestShardingConfig) |
| 1153 | ) |
| 1154 | const driver = yield* MessageStorage.MemoryDriver |
| 1155 | const state = yield* TestEntityState |
| 1156 | |
| 1157 | yield* Effect.gen(function*() { |
| 1158 | yield* TestClock.adjust(1) |
| 1159 | const makeClient = yield* TestEntity.client |
| 1160 | const client = makeClient("1") |
| 1161 | yield* Effect.forkChild(client.RequestWithKey({ key: "abc" })) |
| 1162 | yield* TestClock.adjust(1) |
| 1163 | }).pipe( |
| 1164 | Effect.provide(EnvLayer), |
| 1165 | Effect.scoped |
| 1166 | ) |
| 1167 | |
| 1168 | // only the request should be in the journal |
| 1169 | expect(driver.journal.length).toEqual(1) |
| 1170 | expect(driver.replyIds.size).toEqual(0) |
| 1171 | expect(driver.unprocessed.size).toEqual(1) |
| 1172 | |
| 1173 | // add response |
| 1174 | yield* Queue.offer(state.messages, void 0) |
| 1175 | |
| 1176 | // Let the shards get assigned and storage poll |
| 1177 | yield* TestClock.adjust(5000).pipe( |
| 1178 | Effect.provide(EnvLayer), |
| 1179 | Effect.scoped |
| 1180 | ) |
| 1181 | |
| 1182 | expect(driver.journal.length).toEqual(1) |
| 1183 | expect(driver.replyIds.size).toEqual(1) |
| 1184 | expect(driver.unprocessed.size).toEqual(0) |
| 1185 | |
| 1186 | // the client should read the result from storage |
| 1187 | yield* Effect.gen(function*() { |
| 1188 | yield* TestClock.adjust(1) |
| 1189 | const makeClient = yield* TestEntity.client |
| 1190 | const client = makeClient("1") |
| 1191 | const result = yield* client.RequestWithKey({ key: "abc" }) |
| 1192 | expect(result).toEqual(void 0) |
| 1193 | }).pipe( |
| 1194 | Effect.provide(EnvLayer), |
| 1195 | Effect.scoped |
| 1196 | ) |
| 1197 | |
| 1198 | // the request should not hit the entity |
| 1199 | expect(driver.journal.length).toEqual(1) |
| 1200 | expect(driver.replyIds.size).toEqual(1) |
| 1201 | expect(driver.unprocessed.size).toEqual(0) |
| 1202 | }).pipe(Effect.provide(MessageStorage.layerMemory.pipe( |
| 1203 | Layer.provide(TestShardingConfig), |
| 1204 | Layer.merge(TestEntityState.layer) |
| 1205 | )))) |
| 1206 | |
| 1207 | it.effect("holds durable messages while entity layers are still building", () => |
| 1208 | Effect.gen(function*() { |
| 1209 | const config = ShardingConfig.layer({ |
| 1210 | entityMailboxCapacity: 10, |
| 1211 | entityRegistrationTimeout: 6000, |
| 1212 | entityTerminationTimeout: 0, |
| 1213 | entityMessagePollInterval: 100, |
| 1214 | sendRetryInterval: 100, |
| 1215 | refreshAssignmentsInterval: 0 |
| 1216 | }) |
| 1217 | const env = TestEntityNoState.pipe( |
| 1218 | Layer.provideMerge(Sharding.layer), |
| 1219 | Layer.provide(RunnerStorage.layerMemory), |
| 1220 | Layer.provide(RunnerHealth.layerNoop), |
| 1221 | Layer.provide(Runners.layerNoop), |
| 1222 | Layer.provide(config) |
| 1223 | ) |
| 1224 | const delayedEnv = TestEntityNoState.pipe( |
| 1225 | Layer.provide(Layer.effectDiscard(Effect.sleep(10_000))), |
| 1226 | Layer.provideMerge(Sharding.layer), |
| 1227 | Layer.provide(RunnerStorage.layerMemory), |
| 1228 | Layer.provide(RunnerHealth.layerNoop), |
| 1229 | Layer.provide(Runners.layerNoop), |
| 1230 | Layer.provide(config) |
| 1231 | ) |
| 1232 | const driver = yield* MessageStorage.MemoryDriver |
| 1233 | const state = yield* TestEntityState |
| 1234 | |
| 1235 | yield* Effect.gen(function*() { |
| 1236 | yield* TestClock.adjust(1) |
| 1237 | const makeClient = yield* TestEntity.client |
| 1238 | const client = makeClient("1") |
| 1239 | yield* client.RequestWithKey({ key: "slow-registration" }).pipe(Effect.forkChild) |
| 1240 | yield* TestClock.adjust(1) |
| 1241 | }).pipe( |
| 1242 | Effect.provide(env), |
| 1243 | Effect.scoped |
| 1244 | ) |
| 1245 | |
| 1246 | assert.strictEqual(driver.journal.length, 1) |
| 1247 | assert.strictEqual(driver.replyIds.size, 0) |
| 1248 | assert.strictEqual(driver.unprocessed.size, 1) |
| 1249 | |
| 1250 | const fiber = yield* Effect.never.pipe( |
| 1251 | Effect.provide(delayedEnv), |
| 1252 | Effect.scoped, |
| 1253 | Effect.forkChild({ startImmediately: true }) |
| 1254 | ) |
| 1255 | |
| 1256 | yield* TestClock.adjust(7500) |
| 1257 | assert.strictEqual(driver.replyIds.size, 0) |
| 1258 | assert.strictEqual(driver.unprocessed.size, 1) |
| 1259 | |
| 1260 | yield* TestClock.adjust(2500) |
| 1261 | yield* Queue.offer(state.messages, void 0) |
| 1262 | yield* TestClock.adjust(100) |
| 1263 | |
| 1264 | assert.strictEqual(driver.replyIds.size, 1) |
| 1265 | assert.strictEqual(driver.unprocessed.size, 0) |
| 1266 | yield* Fiber.interrupt(fiber) |
| 1267 | }).pipe(Effect.provide(MessageStorage.layerMemory.pipe( |
| 1268 | Layer.provide(ShardingConfig.layer({})), |
| 1269 | Layer.merge(TestEntityState.layer) |
| 1270 | )))) |
| 1271 | |
| 1272 | it.effect("defects durable messages when no entity ever registers", () => |
| 1273 | Effect.gen(function*() { |
| 1274 | const config = ShardingConfig.layer({ |
| 1275 | entityMailboxCapacity: 10, |
| 1276 | entityRegistrationTimeout: 1000, |
| 1277 | entityTerminationTimeout: 0, |
| 1278 | entityMessagePollInterval: 100, |
| 1279 | sendRetryInterval: 100, |
| 1280 | refreshAssignmentsInterval: 0 |
| 1281 | }) |
| 1282 | const registeredEnv = TestEntityNoState.pipe( |
| 1283 | Layer.provideMerge(Sharding.layer), |
| 1284 | Layer.provide(RunnerStorage.layerMemory), |
| 1285 | Layer.provide(RunnerHealth.layerNoop), |
| 1286 | Layer.provide(Runners.layerNoop), |
| 1287 | Layer.provide(config) |
| 1288 | ) |
| 1289 | const noEntitiesEnv = Sharding.layer.pipe( |
| 1290 | Layer.provide(RunnerStorage.layerMemory), |
| 1291 | Layer.provide(RunnerHealth.layerNoop), |
| 1292 | Layer.provide(Runners.layerNoop), |
| 1293 | Layer.provide(config) |
| 1294 | ) |
| 1295 | const driver = yield* MessageStorage.MemoryDriver |
| 1296 | const warnings: Array<unknown> = [] |
| 1297 | const logger = Logger.make<unknown, void>((options) => { |
| 1298 | if (options.logLevel === "Warn") { |
| 1299 | warnings.push(options.message) |
| 1300 | } |
| 1301 | }) |
| 1302 | |
| 1303 | yield* Effect.gen(function*() { |
| 1304 | yield* TestClock.adjust(1) |
| 1305 | const makeClient = yield* TestEntity.client |
| 1306 | const client = makeClient("1") |
| 1307 | yield* client.RequestWithKey({ key: "missing-registration" }).pipe(Effect.forkChild) |
| 1308 | yield* TestClock.adjust(1) |
| 1309 | }).pipe( |
| 1310 | Effect.provide(registeredEnv), |
| 1311 | Effect.scoped |
| 1312 | ) |
| 1313 | |
| 1314 | const fiber = yield* Effect.never.pipe( |
| 1315 | Effect.provide(noEntitiesEnv), |
| 1316 | Effect.scoped, |
| 1317 | Effect.withLogger(logger), |
| 1318 | Effect.forkChild({ startImmediately: true }) |
| 1319 | ) |
| 1320 | |
| 1321 | yield* TestClock.adjust(8000) |
| 1322 | assert.isTrue(warnings.some((message) => |
| 1323 | globalThis.Array.isArray(message) && message.includes("Could not find entity manager for address, retrying") |
| 1324 | )) |
| 1325 | assert.strictEqual(driver.replyIds.size, 1) |
| 1326 | assert.strictEqual(driver.unprocessed.size, 0) |
| 1327 | yield* Fiber.interrupt(fiber) |
| 1328 | }).pipe(Effect.provide(MessageStorage.layerMemory.pipe( |
| 1329 | Layer.provide(ShardingConfig.layer({})), |
| 1330 | Layer.merge(TestEntityState.layer) |
| 1331 | )))) |
| 1332 | |
| 1333 | it.effect("bounds local sends while entity registration is missing", () => |
| 1334 | Effect.gen(function*() { |
| 1335 | const sharding = yield* Sharding.Sharding |
| 1336 | const entityId = EntityId.make("one") |
| 1337 | yield* TestClock.adjust(1) |
| 1338 | assert.isTrue(sharding.hasShardId(sharding.getShardId(entityId, "default"))) |
| 1339 | |
| 1340 | const client = (yield* MissingRegistrationEntity.client)(entityId) |
| 1341 | const fiber = yield* client.Call().pipe(Effect.forkDetach({ startImmediately: true })) |
| 1342 | yield* Effect.yieldNow |
| 1343 | assert.isUndefined(fiber.pollUnsafe()) |
| 1344 | |
| 1345 | yield* TestClock.adjust(1000) |
| 1346 | const exit = fiber.pollUnsafe() |
| 1347 | assert(exit !== undefined, "the sendLocal registration wait must be bounded") |
| 1348 | const defect = Exit.findDefect(exit) |
| 1349 | assert(Result.isSuccess(defect) && defect.success instanceof Error) |
| 1350 | assert.strictEqual(defect.success.message, "Entity type 'MissingRegistrationEntity' not registered") |
| 1351 | }).pipe(Effect.provide(CappedSharding({ entityRegistrationTimeout: 1000 })))) |
| 1352 | |
| 1353 | it.effect("recomputes the missing entity deadline when registration starts", () => |
| 1354 | Effect.gen(function*() { |
| 1355 | const sharding = yield* Sharding.Sharding |
| 1356 | const entityId = EntityId.make("one") |
| 1357 | yield* TestClock.adjust(1) |
| 1358 | assert.isTrue(sharding.hasShardId(sharding.getShardId(entityId, "default"))) |
| 1359 | |
| 1360 | const client = (yield* MissingRegistrationEntity.client)(entityId) |
| 1361 | const fiber = yield* client.Call().pipe(Effect.forkDetach({ startImmediately: true })) |
| 1362 | yield* Effect.yieldNow |
| 1363 | assert.isUndefined(fiber.pollUnsafe()) |
| 1364 | |
| 1365 | yield* TestClock.adjust(1500) |
| 1366 | yield* sharding.registerEntity( |
| 1367 | FirstRegistrationEntity, |
| 1368 | Effect.succeed(FirstRegistrationEntity.of({ Call: () => Effect.void })) |
| 1369 | ) |
| 1370 | |
| 1371 | // The registration-start deadline is 1 second from now. The original |
| 1372 | // fallback deadline has elapsed, but must no longer win the race. |
| 1373 | yield* TestClock.adjust(600) |
| 1374 | assert.isUndefined(fiber.pollUnsafe()) |
| 1375 | |
| 1376 | yield* TestClock.adjust(400) |
| 1377 | const exit = fiber.pollUnsafe() |
| 1378 | assert(exit !== undefined, "the registration-start deadline must be bounded") |
| 1379 | const defect = Exit.findDefect(exit) |
| 1380 | assert(Result.isSuccess(defect) && defect.success instanceof Error) |
| 1381 | assert.strictEqual(defect.success.message, "Entity type 'MissingRegistrationEntity' not registered") |
| 1382 | }).pipe(Effect.provide(UnregisteredSharding({ entityRegistrationTimeout: 1000 })))) |
| 1383 | |
| 1384 | it.effect("keeps a shared registration latch when one waiter is interrupted", () => |
| 1385 | Effect.gen(function*() { |
| 1386 | const sharding = yield* Sharding.Sharding |
| 1387 | const entityId = EntityId.make("one") |
| 1388 | yield* TestClock.adjust(1) |
| 1389 | assert.isTrue(sharding.hasShardId(sharding.getShardId(entityId, "default"))) |
| 1390 | |
| 1391 | const client = (yield* MissingRegistrationEntity.client)(entityId) |
| 1392 | const interrupted = yield* client.Call().pipe(Effect.forkDetach({ startImmediately: true })) |
| 1393 | const remaining = yield* client.Call().pipe(Effect.forkDetach({ startImmediately: true })) |
| 1394 | yield* Effect.yieldNow |
| 1395 | assert.isUndefined(interrupted.pollUnsafe()) |
| 1396 | assert.isUndefined(remaining.pollUnsafe()) |
| 1397 | |
| 1398 | interrupted.interruptUnsafe() |
| 1399 | yield* Effect.yieldNow |
| 1400 | yield* sharding.registerEntity( |
| 1401 | MissingRegistrationEntity, |
| 1402 | Effect.succeed(MissingRegistrationEntity.of({ Call: () => Effect.void })) |
| 1403 | ) |
| 1404 | const interruptedExit = yield* Fiber.await(interrupted) |
| 1405 | assert.isTrue(Exit.isFailure(interruptedExit) && Cause.hasInterruptsOnly(interruptedExit.cause)) |
| 1406 | yield* Fiber.join(remaining) |
| 1407 | }).pipe(Effect.provide(UnregisteredSharding({ entityRegistrationTimeout: 1000 })))) |
| 1408 | |
| 1409 | it.effect("bounds client interruption while entity registration is missing", () => |
| 1410 | Effect.gen(function*() { |
| 1411 | const sharding = yield* Sharding.Sharding |
| 1412 | const entityId = EntityId.make("one") |
| 1413 | yield* TestClock.adjust(1) |
| 1414 | assert.isTrue(sharding.hasShardId(sharding.getShardId(entityId, "default"))) |
| 1415 | |
| 1416 | const client = (yield* MissingRegistrationEntity.client)(entityId) |
| 1417 | const fiber = yield* client.Call().pipe(Effect.forkDetach({ startImmediately: true })) |
| 1418 | yield* Effect.yieldNow |
| 1419 | |
| 1420 | // Interrupting a client sends an interrupt message through the same local |
| 1421 | // registration wait. Fork it so TestClock can reach the shared deadline. |
| 1422 | const interruptFiber = yield* Fiber.interrupt(fiber).pipe( |
| 1423 | Effect.forkDetach({ startImmediately: true }) |
| 1424 | ) |
| 1425 | yield* Effect.yieldNow |
| 1426 | assert.isUndefined(interruptFiber.pollUnsafe()) |
| 1427 | |
| 1428 | yield* TestClock.adjust(1000) |
| 1429 | yield* Fiber.join(interruptFiber) |
| 1430 | const interruptedExit = yield* Fiber.await(fiber) |
| 1431 | assert.isTrue(Exit.isFailure(interruptedExit) && Cause.hasInterruptsOnly(interruptedExit.cause)) |
| 1432 | }).pipe(Effect.provide(CappedSharding({ entityRegistrationTimeout: 1000 })))) |
| 1433 | |
| 1434 | it.effect("durable streams are resumed on restart", () => |
| 1435 | Effect.gen(function*() { |
| 1436 | const EnvLayer = TestShardingWithoutState.pipe( |
| 1437 | Layer.provide(Runners.layerNoop), |
| 1438 | Layer.provide(TestShardingConfig) |
| 1439 | ) |
| 1440 | const driver = yield* MessageStorage.MemoryDriver |
| 1441 | const state = yield* TestEntityState |
| 1442 | |
| 1443 | // first chunk |
| 1444 | yield* Queue.offerAll(state.streamMessages, [void 0, void 0]) |
| 1445 | |
| 1446 | yield* Effect.gen(function*() { |
| 1447 | yield* TestClock.adjust(2000) |
| 1448 | const makeClient = yield* TestEntity.client |
| 1449 | const client = makeClient("1") |
| 1450 | yield* Effect.forkChild(Stream.runDrain(client.StreamWithKey({ key: "abc" }))) |
| 1451 | yield* TestClock.adjust(2000) |
| 1452 | // second chunk |
| 1453 | yield* Queue.offer(state.streamMessages, void 0) |
| 1454 | yield* TestClock.adjust(2000) |
| 1455 | }).pipe( |
| 1456 | Effect.provide(EnvLayer), |
| 1457 | Effect.scoped |
| 1458 | ) |
| 1459 | |
| 1460 | // 1 request, 2 acks, 2 replies |
| 1461 | expect(driver.journal.length).toEqual(1 + 2) |
| 1462 | expect(driver.replyIds.size).toEqual(2) |
| 1463 | expect(driver.unprocessed.size).toEqual(1) |
| 1464 | |
| 1465 | // third chunk |
| 1466 | yield* Queue.offerAll(state.streamMessages, [void 0, void 0]) |
| 1467 | yield* Queue.end(state.streamMessages) |
| 1468 | |
| 1469 | // the client should resume |
| 1470 | yield* Effect.gen(function*() { |
| 1471 | yield* TestClock.adjust(5000) // let the shards get assigned and storage poll |
| 1472 | const makeClient = yield* TestEntity.client |
| 1473 | const client = makeClient("1") |
| 1474 | |
| 1475 | // let the reply loop run |
| 1476 | yield* TestClock.adjust(500).pipe(Effect.forkChild) |
| 1477 | |
| 1478 | const results = yield* Stream.runCollect(client.StreamWithKey({ key: "abc" })) |
| 1479 | expect(results).toEqual([3, 4]) |
| 1480 | }).pipe( |
| 1481 | Effect.provide(EnvLayer), |
| 1482 | Effect.scoped |
| 1483 | ) |
| 1484 | |
| 1485 | // 1 request, 3 acks, 4 replies (3 chunks + WithExit) |
| 1486 | expect(driver.journal.length).toEqual(1 + 3) |
| 1487 | expect(driver.replyIds.size).toEqual(4) |
| 1488 | expect(driver.unprocessed.size).toEqual(0) |
| 1489 | }).pipe(Effect.provide(MessageStorage.layerMemory.pipe( |
| 1490 | Layer.provide(TestShardingConfig), |
| 1491 | Layer.merge(TestEntityState.layer) |
| 1492 | )))) |
| 1493 | |
| 1494 | it.effect("client discard stores durable requests without waiting for replies", () => |
| 1495 | Effect.gen(function*() { |
| 1496 | yield* TestClock.adjust(1) |
| 1497 | const driver = yield* MessageStorage.MemoryDriver |
| 1498 | const makeClient = yield* TestEntity.client |
| 1499 | const client = makeClient("1") |
| 1500 | const result = yield* client.GetUser({ id: 123 }, { discard: true }) |
| 1501 | expect(result).toEqual(void 0) |
| 1502 | yield* TestClock.adjust(1) |
| 1503 | expect(driver.journal.length).toEqual(1) |
| 1504 | expect(driver.unprocessed.size).toEqual(0) |
| 1505 | }).pipe(Effect.provide(TestSharding))) |
| 1506 | |
| 1507 | it.effect("client discard returns while the durable request keeps processing", () => |
| 1508 | Effect.gen(function*() { |
| 1509 | yield* TestClock.adjust(1) |
| 1510 | const driver = yield* MessageStorage.MemoryDriver |
| 1511 | const makeClient = yield* TestEntity.client |
| 1512 | const client = makeClient("1") |
| 1513 | const result = yield* client.Never(void 0, { discard: true }) |
| 1514 | expect(result).toEqual(void 0) |
| 1515 | yield* TestClock.adjust(1) |
| 1516 | expect(driver.journal.length).toEqual(1) |
| 1517 | // should still be processing |
| 1518 | expect(driver.unprocessed.size).toEqual(1) |
| 1519 | }).pipe(Effect.provide(TestSharding))) |
| 1520 | |
| 1521 | it.effect("client discard returns while the volatile request keeps processing", () => |
| 1522 | Effect.gen(function*() { |
| 1523 | yield* TestClock.adjust(1) |
| 1524 | const state = yield* TestEntityState |
| 1525 | const makeClient = yield* TestEntity.client |
| 1526 | const client = makeClient("1") |
| 1527 | |
| 1528 | const result = yield* client.NeverVolatile(void 0, { discard: true }) |
| 1529 | |
| 1530 | assert.isUndefined(result) |
| 1531 | yield* TestClock.adjust(1) |
| 1532 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 1533 | }).pipe(Effect.provide(TestSharding))) |
| 1534 | |
| 1535 | it.effect("client volatile discard retries a failed delivery", () => |
| 1536 | Effect.gen(function*() { |
| 1537 | let attempts = 0 |
| 1538 | |
| 1539 | yield* Effect.gen(function*() { |
| 1540 | yield* TestClock.adjust(1) |
| 1541 | const config = yield* ShardingConfig.ShardingConfig |
| 1542 | ;(config as any).runnerAddress = Option.some(RunnerAddress.make("localhost", 1234)) |
| 1543 | const makeClient = yield* TestEntity.client |
| 1544 | const client = makeClient("1") |
| 1545 | const fiber = yield* client.NeverVolatile(void 0, { discard: true }).pipe( |
| 1546 | Effect.forkChild({ startImmediately: true }) |
| 1547 | ) |
| 1548 | |
| 1549 | assert.strictEqual(attempts, 1) |
| 1550 | assert.isUndefined(fiber.pollUnsafe()) |
| 1551 | yield* TestClock.adjust(100) |
| 1552 | yield* Fiber.join(fiber) |
| 1553 | assert.strictEqual(attempts, 2) |
| 1554 | }).pipe( |
| 1555 | Effect.provide(TestShardingWithoutRunners.pipe( |
| 1556 | Layer.provide( |
| 1557 | Layer.effect(Runners.Runners)( |
| 1558 | Effect.gen(function*() { |
| 1559 | const runners = yield* Runners.makeNoop |
| 1560 | return { |
| 1561 | ...runners, |
| 1562 | notify(options) { |
| 1563 | attempts++ |
| 1564 | return attempts === 1 |
| 1565 | ? Effect.fail( |
| 1566 | new ClusterError.RunnerUnavailable({ |
| 1567 | address: Option.getOrThrow(options.address) |
| 1568 | }) |
| 1569 | ) |
| 1570 | : Effect.void |
| 1571 | } |
| 1572 | } |
| 1573 | }) |
| 1574 | ) |
| 1575 | ), |
| 1576 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 1577 | Layer.provideMerge(ShardingConfig.layer({ |
| 1578 | entityMailboxCapacity: 10, |
| 1579 | entityTerminationTimeout: 0, |
| 1580 | entityMessagePollInterval: 5000, |
| 1581 | sendRetryInterval: 100, |
| 1582 | refreshAssignmentsInterval: 0 |
| 1583 | })) |
| 1584 | )) |
| 1585 | ) |
| 1586 | })) |
| 1587 | |
| 1588 | it.effect("defects when a durable request has no MessageStorage", () => |
| 1589 | Effect.gen(function*() { |
| 1590 | const makeClient = yield* TestEntity.client |
| 1591 | const client = makeClient("1") |
| 1592 | const cause = yield* client.Never().pipe( |
| 1593 | Effect.sandbox, |
| 1594 | Effect.flip |
| 1595 | ) |
| 1596 | assert(Cause.hasDies(cause)) |
| 1597 | }).pipe(Effect.provide(TestShardingWithoutStorage.pipe( |
| 1598 | Layer.provide(MessageStorage.layerNoop) |
| 1599 | )))) |
| 1600 | |
| 1601 | it.effect("reprocesses a completed volatile request id without MessageStorage", () => |
| 1602 | Effect.gen(function*() { |
| 1603 | yield* TestClock.adjust(1) |
| 1604 | const sharding = yield* Sharding.Sharding |
| 1605 | const state = yield* TestEntityState |
| 1606 | const rpc = TestEntity.protocol.requests.get("GetUserVolatile") as Extract< |
| 1607 | RpcGroup.Rpcs<typeof TestEntity.protocol>, |
| 1608 | { readonly _tag: "GetUserVolatile" } |
| 1609 | > |
| 1610 | const entityId = EntityId.make("1") |
| 1611 | const requestId = yield* sharding.getSnowflake |
| 1612 | const send = Effect.gen(function*() { |
| 1613 | const replied = yield* Deferred.make<void>() |
| 1614 | yield* sharding.sendOutgoing( |
| 1615 | new Message.OutgoingRequest({ |
| 1616 | envelope: Envelope.makeRequest<typeof rpc>({ |
| 1617 | requestId, |
| 1618 | address: EntityAddress.make({ |
| 1619 | shardId: sharding.getShardId(entityId, "default"), |
| 1620 | entityType: EntityType.make(TestEntity.type), |
| 1621 | entityId |
| 1622 | }), |
| 1623 | tag: "GetUserVolatile", |
| 1624 | payload: { id: 1 }, |
| 1625 | headers: Headers.empty |
| 1626 | }), |
| 1627 | annotations: rpc.annotations, |
| 1628 | context: Context.empty() as Context.Context<unknown>, |
| 1629 | rpc, |
| 1630 | lastReceivedReply: Option.none(), |
| 1631 | respond: () => Deferred.succeed(replied, void 0) |
| 1632 | }), |
| 1633 | false |
| 1634 | ) |
| 1635 | yield* Deferred.await(replied) |
| 1636 | }) |
| 1637 | yield* send |
| 1638 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 1639 | yield* send |
| 1640 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 2) |
| 1641 | }).pipe(Effect.provide(TestShardingWithoutStorage.pipe( |
| 1642 | Layer.provide(MessageStorage.layerNoop) |
| 1643 | )))) |
| 1644 | |
| 1645 | it.effect("restarts the entity layer after a handler defect", () => |
| 1646 | Effect.gen(function*() { |
| 1647 | yield* TestClock.adjust(1) |
| 1648 | const state = yield* TestEntityState |
| 1649 | const makeClient = yield* TestEntity.client |
| 1650 | const client = makeClient("1") |
| 1651 | MutableRef.set(state.defectTrigger, true) |
| 1652 | const result = yield* client.GetUser({ id: 123 }) |
| 1653 | expect(result).toEqual(new User({ id: 123, name: "User 123" })) |
| 1654 | expect(state.layerBuilds.current).toEqual(2) |
| 1655 | }).pipe(Effect.provide(TestSharding))) |
| 1656 | |
| 1657 | it.effect("replays in-flight requests when restarting after a defect", () => |
| 1658 | Effect.gen(function*() { |
| 1659 | yield* TestClock.adjust(1) |
| 1660 | const state = yield* TestEntityState |
| 1661 | const makeClient = yield* TestEntity.client |
| 1662 | const client = makeClient("1") |
| 1663 | |
| 1664 | yield* client.NeverFork().pipe(Effect.forkChild({ startImmediately: true })) |
| 1665 | yield* TestClock.adjust(1) |
| 1666 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 1667 | |
| 1668 | MutableRef.set(state.defectTrigger, true) |
| 1669 | const result = yield* client.GetUser({ id: 123 }) |
| 1670 | assert.deepStrictEqual(result, new User({ id: 123, name: "User 123" })) |
| 1671 | assert.strictEqual(state.layerBuilds.current, 2) |
| 1672 | |
| 1673 | yield* TestClock.adjust(1) |
| 1674 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 4) |
| 1675 | }).pipe(Effect.provide(TestSharding))) |
| 1676 | |
| 1677 | it.effect("WithTransaction is propagated to the entity handler", () => |
| 1678 | Effect.gen(function*() { |
| 1679 | let isTransaction = false |
| 1680 | let transactionOpen = false |
| 1681 | yield* Effect.gen(function*() { |
| 1682 | const makeClient = yield* TestEntity.client |
| 1683 | yield* TestClock.adjust(1) |
| 1684 | const client = makeClient("1") |
| 1685 | |
| 1686 | const result = yield* client.WithTransaction({ id: 1 }) |
| 1687 | assert.strictEqual(result, true) |
| 1688 | assert.strictEqual(isTransaction, true) |
| 1689 | }).pipe(Effect.provide(TestShardingWithoutStorage.pipe( |
| 1690 | Layer.updateService(MessageStorage.MessageStorage, (storage) => ({ |
| 1691 | ...storage, |
| 1692 | withTransaction(effect) { |
| 1693 | return Effect.suspend(() => { |
| 1694 | transactionOpen = true |
| 1695 | return storage.withTransaction(effect) |
| 1696 | }).pipe( |
| 1697 | Effect.ensuring(Effect.sync(() => { |
| 1698 | transactionOpen = false |
| 1699 | })) |
| 1700 | ) |
| 1701 | }, |
| 1702 | saveReply(reply) { |
| 1703 | return MessageStorage.MemoryTransaction.use((isTransaction_) => { |
| 1704 | isTransaction = isTransaction_ |
| 1705 | assert.strictEqual(transactionOpen, true) |
| 1706 | return storage.saveReply(reply) |
| 1707 | }) |
| 1708 | } |
| 1709 | })), |
| 1710 | Layer.provide(MessageStorage.layerMemory), |
| 1711 | Layer.provide(TestShardingConfig) |
| 1712 | ))) |
| 1713 | })) |
| 1714 | }) |
| 1715 | |
| 1716 | const DefectRecoveryRun = Rpc.make("run", { |
| 1717 | payload: { id: Schema.String }, |
| 1718 | success: Schema.String |
| 1719 | }) |
| 1720 | |
| 1721 | const DefectRecoveryEntity = Entity.make("DefectRecovery", [DefectRecoveryRun.annotate(ClusterSchema.Persisted, true)]) |
| 1722 | |
| 1723 | const DefectRecoverySharding = <R>(entityLayer: Layer.Layer<never, never, R>) => |
| 1724 | entityLayer.pipe(Layer.provideMerge(UnregisteredSharding({}))) |
| 1725 | |
| 1726 | // Counts handler builds. Building the `blocked` generation waits until |
| 1727 | // `acquired` completes, holding the entity in replacement acquisition. |
| 1728 | const makeGenerations = Effect.fnUntraced(function*(blocked?: number) { |
| 1729 | const count = yield* Ref.make(0) |
| 1730 | const acquiring = yield* Deferred.make<void>() |
| 1731 | const acquired = yield* Deferred.make<void>() |
| 1732 | const next = Effect.gen(function*() { |
| 1733 | const generation = yield* Ref.updateAndGet(count, (n) => n + 1) |
| 1734 | if (generation === blocked) { |
| 1735 | yield* Deferred.succeed(acquiring, undefined) |
| 1736 | yield* Deferred.await(acquired) |
| 1737 | } |
| 1738 | return generation |
| 1739 | }) |
| 1740 | return { count, acquiring, acquired, next } as const |
| 1741 | }) |
| 1742 | |
| 1743 | // Drives an EntityManager directly: its first request defects, and the |
| 1744 | // replacement handlers stay in acquisition until `generations.acquired`. |
| 1745 | const startBlockedRebuild = Effect.fnUntraced(function*(entityId: string) { |
| 1746 | const generations = yield* makeGenerations(2) |
| 1747 | const attempts = yield* Ref.make(0) |
| 1748 | const run = Rpc.make("run") |
| 1749 | const entity = Entity.make("DefectRecoveryShutdown", [run]) |
| 1750 | const sharding = yield* Sharding.Sharding |
| 1751 | const manager = yield* EntityManager.make( |
| 1752 | entity, |
| 1753 | Effect.as( |
| 1754 | generations.next, |
| 1755 | entity.of({ |
| 1756 | run: () => Ref.update(attempts, (n) => n + 1).pipe(Effect.andThen(Effect.die("restart"))) |
| 1757 | }) |
| 1758 | ), |
| 1759 | { |
| 1760 | sharding, |
| 1761 | storage: MessageStorage.noop, |
| 1762 | runnerAddress: RunnerAddress.make("localhost", 1234), |
| 1763 | residency: { admitUnsafe: () => true, releaseUnsafe: () => {} }, |
| 1764 | maxIdleTime: Infinity, |
| 1765 | defectRetryPolicy: Schedule.spaced(1) |
| 1766 | } |
| 1767 | ) |
| 1768 | const id = EntityId.make(entityId) |
| 1769 | const shardId = sharding.getShardId(id, "default") |
| 1770 | const address = EntityAddress.make({ shardId, entityType: EntityType.make(entity.type), entityId: id }) |
| 1771 | const send = Effect.gen(function*() { |
| 1772 | return yield* manager.sendLocal( |
| 1773 | new Message.IncomingRequestLocal<typeof run>({ |
| 1774 | envelope: Envelope.makeRequest<typeof run>({ |
| 1775 | requestId: yield* sharding.getSnowflake, |
| 1776 | address, |
| 1777 | tag: "run", |
| 1778 | payload: undefined, |
| 1779 | headers: Headers.empty |
| 1780 | }), |
| 1781 | lastSentReply: Option.none(), |
| 1782 | annotations: Context.empty(), |
| 1783 | respond: () => Effect.void |
| 1784 | }) |
| 1785 | ) |
| 1786 | }) |
| 1787 | yield* TestClock.adjust(1) |
| 1788 | yield* send |
| 1789 | yield* TestClock.adjust(10) |
| 1790 | yield* Deferred.await(generations.acquiring) |
| 1791 | return { manager, shardId, send, attempts, acquired: generations.acquired } as const |
| 1792 | }) |
| 1793 | |
| 1794 | describe.concurrent("Sharding defect recovery", () => { |
| 1795 | it.effect("restarts again when a replayed request defects synchronously", () => |
| 1796 | Effect.gen(function*() { |
| 1797 | const entity = DefectRecoveryEntity |
| 1798 | const generations = yield* makeGenerations() |
| 1799 | const attempts = yield* Ref.make(0) |
| 1800 | const entityLayer = entity.toLayer(Effect.as( |
| 1801 | generations.next, |
| 1802 | entity.of({ |
| 1803 | run: Effect.fnUntraced(function*({ payload }) { |
| 1804 | if ((yield* Ref.updateAndGet(attempts, (n) => n + 1)) <= 3) { |
| 1805 | return yield* Effect.die("repeated defect") |
| 1806 | } |
| 1807 | return payload.id |
| 1808 | }) |
| 1809 | }) |
| 1810 | )) |
| 1811 | |
| 1812 | yield* Effect.gen(function*() { |
| 1813 | const client = (yield* entity.client)("repeated") |
| 1814 | const work = yield* client.run({ id: "request" }).pipe(Effect.forkChild) |
| 1815 | yield* TestClock.adjust("30 seconds") |
| 1816 | assert.strictEqual(yield* Ref.get(attempts), 4) |
| 1817 | assert.strictEqual(yield* Ref.get(generations.count), 4) |
| 1818 | assert.strictEqual(yield* Fiber.join(work), "request") |
| 1819 | }).pipe(Effect.provide(DefectRecoverySharding(entityLayer))) |
| 1820 | })) |
| 1821 | |
| 1822 | it.effect("replays unfinished requests before arrivals during acquisition", () => |
| 1823 | Effect.gen(function*() { |
| 1824 | const entity = DefectRecoveryEntity |
| 1825 | const generations = yield* makeGenerations(2) |
| 1826 | const fail = yield* Deferred.make<void>() |
| 1827 | const bothStarted = yield* Deferred.make<void>() |
| 1828 | const started = yield* Ref.make(0) |
| 1829 | const calls = yield* Ref.make<Array<readonly [number, string]>>([]) |
| 1830 | const completed = yield* Ref.make<Array<string>>([]) |
| 1831 | const thirdRequestId = yield* Deferred.make<Snowflake.Snowflake>() |
| 1832 | const entityLayer = entity.toLayer( |
| 1833 | Effect.map(generations.next, (generation) => |
| 1834 | entity.of({ |
| 1835 | run: Effect.fnUntraced(function*({ payload, requestId }) { |
| 1836 | if (payload.id === "third") { |
| 1837 | yield* Deferred.succeed(thirdRequestId, requestId) |
| 1838 | } |
| 1839 | yield* Ref.update(calls, (calls) => [...calls, [generation, payload.id] as const]) |
| 1840 | if (generation === 1) { |
| 1841 | if ((yield* Ref.updateAndGet(started, (n) => n + 1)) === 2) { |
| 1842 | yield* Deferred.succeed(bothStarted, undefined) |
| 1843 | } |
| 1844 | yield* Deferred.await(fail) |
| 1845 | return yield* Effect.die("initial defect") |
| 1846 | } |
| 1847 | if (generation === 2 && payload.id === "first") { |
| 1848 | return yield* Effect.die("replay defect") |
| 1849 | } |
| 1850 | yield* Ref.update(completed, (ids) => [...ids, payload.id]) |
| 1851 | return payload.id |
| 1852 | }) |
| 1853 | })), |
| 1854 | { concurrency: "unbounded" } |
| 1855 | ) |
| 1856 | |
| 1857 | yield* Effect.gen(function*() { |
| 1858 | const client = (yield* entity.client)("arrivals") |
| 1859 | const first = yield* client.run({ id: "first" }).pipe(Effect.forkChild) |
| 1860 | const second = yield* client.run({ id: "second" }).pipe(Effect.forkChild) |
| 1861 | yield* Deferred.await(bothStarted) |
| 1862 | yield* Deferred.succeed(fail, undefined) |
| 1863 | yield* TestClock.adjust("10 seconds") |
| 1864 | yield* Deferred.await(generations.acquiring) |
| 1865 | const third = yield* client.run({ id: "third" }).pipe(Effect.forkChild) |
| 1866 | yield* TestClock.adjust(1) |
| 1867 | yield* Deferred.succeed(generations.acquired, undefined) |
| 1868 | yield* TestClock.adjust("30 seconds") |
| 1869 | assert.strictEqual(yield* Fiber.join(first), "first") |
| 1870 | assert.strictEqual(yield* Fiber.join(second), "second") |
| 1871 | assert.strictEqual(yield* Fiber.join(third), "third") |
| 1872 | const recorded = yield* Ref.get(calls) |
| 1873 | assert.deepStrictEqual( |
| 1874 | Array.findFirst(recorded, ([generation]) => generation === 2), |
| 1875 | Option.some([2, "first"] as const) |
| 1876 | ) |
| 1877 | // A request still waiting for its first dispatch is not replayed as well |
| 1878 | assert.strictEqual(Array.filter(recorded, ([, id]) => id === "third").length, 1) |
| 1879 | const requestId = yield* Deferred.await(thirdRequestId) |
| 1880 | const driver = yield* MessageStorage.MemoryDriver |
| 1881 | const replies = yield* driver.encoded.repliesFor([String(requestId)]).pipe(Effect.orDie) |
| 1882 | assert.isTrue(Array.some( |
| 1883 | replies, |
| 1884 | (reply) => reply._tag === "WithExit" && reply.exit._tag === "Success" && reply.exit.value === "third" |
| 1885 | )) |
| 1886 | assert.sameMembers(yield* Ref.get(completed), ["first", "second", "third"]) |
| 1887 | assert.strictEqual(yield* Ref.get(generations.count), 3) |
| 1888 | }).pipe(Effect.provide(DefectRecoverySharding(entityLayer))) |
| 1889 | })) |
| 1890 | |
| 1891 | it.effect("finishes shutdown when replacement acquisition completes", () => |
| 1892 | Effect.gen(function*() { |
| 1893 | const { acquired, attempts, manager, shardId } = yield* startBlockedRebuild("shutdown") |
| 1894 | const shutdown = yield* manager.interruptShard(shardId).pipe(Effect.forkChild) |
| 1895 | yield* TestClock.adjust(1) |
| 1896 | assert.strictEqual(yield* manager.activeEntityCount, 0) |
| 1897 | yield* Deferred.succeed(acquired, undefined) |
| 1898 | yield* TestClock.adjust(1) |
| 1899 | const completed = shutdown.pollUnsafe() |
| 1900 | // Let a broken implementation's termination timeout finish before asserting. |
| 1901 | yield* TestClock.adjust(1000) |
| 1902 | assert.deepStrictEqual(completed, Exit.void) |
| 1903 | assert.strictEqual(yield* Ref.get(attempts), 1, "shutdown must not replay application requests") |
| 1904 | }).pipe(Effect.provide(BlockedRebuildSharding))) |
| 1905 | |
| 1906 | it.effect("interrupts a queued arrival when shutdown starts during replacement acquisition", () => |
| 1907 | Effect.gen(function*() { |
| 1908 | const { acquired, manager, send, shardId } = yield* startBlockedRebuild("shutdown-queued") |
| 1909 | // Arrives while the replacement handlers are still being built, so it |
| 1910 | // queues behind the replay of unfinished requests. |
| 1911 | const queued = yield* send.pipe(Effect.forkChild) |
| 1912 | yield* TestClock.adjust(1) |
| 1913 | assert.strictEqual(queued.pollUnsafe(), undefined) |
| 1914 | yield* manager.interruptShard(shardId).pipe(Effect.forkChild) |
| 1915 | yield* TestClock.adjust(1) |
| 1916 | assert.strictEqual(yield* manager.activeEntityCount, 0) |
| 1917 | // The queued request must be refused as soon as the activation is retired, |
| 1918 | // without waiting for the replacement handlers. |
| 1919 | const exit = queued.pollUnsafe() |
| 1920 | assert.isDefined(exit, "queued arrival must not wait for replacement acquisition") |
| 1921 | assert.isTrue(Exit.hasInterrupts(exit), "queued arrival must be interrupted") |
| 1922 | yield* Deferred.succeed(acquired, undefined) |
| 1923 | }).pipe(Effect.provide(BlockedRebuildSharding))) |
| 1924 | }) |
| 1925 | |
| 1926 | const ActiveTeardownCaller = Entity.make("ActiveTeardownCaller", [ |
| 1927 | Rpc.make("Arm").annotate(ClusterSchema.Persisted, false) |
| 1928 | ]) |
| 1929 | |
| 1930 | const ActiveTeardownCallerLayer = ActiveTeardownCaller.toLayer(Effect.gen(function*() { |
| 1931 | const receiver = yield* TestEntity.client |
| 1932 | yield* receiver("nested-target").Never().pipe(Effect.forkScoped) |
| 1933 | return { Arm: () => Effect.void } |
| 1934 | })) |
| 1935 | |
| 1936 | const journalInterrupts = (driver: MessageStorage.MemoryDriver["Service"]) => |
| 1937 | driver.journal.filter((envelope) => envelope._tag === "Interrupt").length |
| 1938 | |
| 1939 | const waitForIdleReap = Effect.fnUntraced(function*( |
| 1940 | sharding: { readonly activeEntityCount: Effect.Effect<number> }, |
| 1941 | remaining: number |
| 1942 | ) { |
| 1943 | for (let i = 0; i < 12; i++) { |
| 1944 | if ((yield* sharding.activeEntityCount) <= remaining) return |
| 1945 | yield* TestClock.adjust(5000) |
| 1946 | } |
| 1947 | assert.isAtMost(yield* sharding.activeEntityCount, remaining) |
| 1948 | }) |
| 1949 | |
| 1950 | describe.concurrent("Sharding active teardowns", () => { |
| 1951 | it.effect("swallows persisted interrupts from an idle-reaped nested proxy call", () => |
| 1952 | Effect.gen(function*() { |
| 1953 | yield* TestClock.adjust(1) |
| 1954 | const driver = yield* MessageStorage.MemoryDriver |
| 1955 | const state = yield* TestEntityState |
| 1956 | const sharding = yield* Sharding.Sharding |
| 1957 | const caller = (yield* ActiveTeardownCaller.client)("1") |
| 1958 | yield* caller.Arm() |
| 1959 | yield* TestClock.adjust(1) |
| 1960 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 1961 | assert.strictEqual(journalInterrupts(driver), 0) |
| 1962 | |
| 1963 | yield* waitForIdleReap(sharding, 1) |
| 1964 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 1965 | assert.strictEqual(Queue.sizeUnsafe(state.interrupts), 0) |
| 1966 | assert.strictEqual(journalInterrupts(driver), 0) |
| 1967 | }).pipe(Effect.provide(ActiveTeardownSharding({ entityMaxIdleTime: 1 })))) |
| 1968 | |
| 1969 | // TestClock-driven reassignment can exhaust the wall-clock timeout when |
| 1970 | // this test competes with the rest of this file in CI. |
| 1971 | it.effect("swallows persisted interrupts when the caller's shard is released", () => |
| 1972 | Effect.gen(function*() { |
| 1973 | const storageState = makeFailoverStorageState() |
| 1974 | const runnerStorage = Layer.effect( |
| 1975 | RunnerStorage.RunnerStorage, |
| 1976 | Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) |
| 1977 | ) |
| 1978 | const layer = ActiveTeardownCallerLayer.pipe( |
| 1979 | Layer.merge(TestEntityNoState), |
| 1980 | Layer.provideMerge(Sharding.layer), |
| 1981 | Layer.provide(runnerStorage), |
| 1982 | Layer.provide(RunnerHealth.layerNoop), |
| 1983 | Layer.provideMerge(TestEntityState.layer), |
| 1984 | Layer.provide(Runners.layerNoop), |
| 1985 | Layer.provideMerge(MessageStorage.layerMemory), |
| 1986 | Layer.provide(Snowflake.layerGenerator), |
| 1987 | Layer.provide(ShardingConfig.layer({ |
| 1988 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 1989 | shardsPerGroup: 1, |
| 1990 | entityTerminationTimeout: 0, |
| 1991 | entityMessagePollInterval: 10, |
| 1992 | refreshAssignmentsInterval: 10, |
| 1993 | sendRetryInterval: 10 |
| 1994 | })) |
| 1995 | ) |
| 1996 | |
| 1997 | yield* Effect.gen(function*() { |
| 1998 | const driver = yield* MessageStorage.MemoryDriver |
| 1999 | const state = yield* TestEntityState |
| 2000 | const sharding = yield* Sharding.Sharding |
| 2001 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 2002 | while (!sharding.hasShardId(shardId)) { |
| 2003 | yield* TestClock.adjust(10) |
| 2004 | } |
| 2005 | |
| 2006 | yield* (yield* ActiveTeardownCaller.client)("1").Arm() |
| 2007 | yield* TestClock.adjust(1) |
| 2008 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 2009 | |
| 2010 | storageState.assignSelf = false |
| 2011 | while (sharding.hasShardId(shardId)) { |
| 2012 | yield* TestClock.adjust(10) |
| 2013 | } |
| 2014 | while ((yield* sharding.activeEntityCount) > 0) { |
| 2015 | yield* TestClock.adjust(1) |
| 2016 | } |
| 2017 | |
| 2018 | assert.strictEqual(journalInterrupts(driver), 0) |
| 2019 | }).pipe(Effect.provide(layer), Effect.scoped) |
| 2020 | }), { concurrent: false }) |
| 2021 | |
| 2022 | it.effect("treats node shutdown interrupts as transient via isShutdown", () => |
| 2023 | Effect.gen(function*() { |
| 2024 | let driver!: MessageStorage.MemoryDriver["Service"] |
| 2025 | let state!: TestEntityState["Service"] |
| 2026 | yield* Effect.gen(function*() { |
| 2027 | driver = yield* MessageStorage.MemoryDriver |
| 2028 | state = yield* TestEntityState |
| 2029 | yield* TestClock.adjust(1) |
| 2030 | yield* (yield* ActiveTeardownCaller.client)("1").Arm() |
| 2031 | yield* TestClock.adjust(1) |
| 2032 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 2033 | }).pipe(Effect.provide(ActiveTeardownSharding())) |
| 2034 | |
| 2035 | assert.strictEqual(journalInterrupts(driver), 0) |
| 2036 | assert.strictEqual(driver.journal.length, 1) |
| 2037 | })) |
| 2038 | |
| 2039 | it.effect("forwards replayed storage interrupts that have no local provenance", () => |
| 2040 | Effect.gen(function*() { |
| 2041 | const driver = yield* MessageStorage.MemoryDriver |
| 2042 | const state = yield* TestEntityState |
| 2043 | const sharding = yield* Sharding.Sharding |
| 2044 | const makeClient = yield* TestEntity.client |
| 2045 | yield* TestClock.adjust(1) |
| 2046 | const client = makeClient("1") |
| 2047 | |
| 2048 | yield* client.Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 2049 | yield* TestClock.adjust(1) |
| 2050 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 2051 | |
| 2052 | const request = driver.journal[0] |
| 2053 | assert.strictEqual(request._tag, "Request") |
| 2054 | yield* driver.encoded.saveEnvelope({ |
| 2055 | envelope: { |
| 2056 | id: String(yield* sharding.getSnowflake), |
| 2057 | _tag: "Interrupt", |
| 2058 | requestId: request.requestId, |
| 2059 | address: request.address |
| 2060 | }, |
| 2061 | primaryKey: null, |
| 2062 | deliverAt: null |
| 2063 | }) |
| 2064 | |
| 2065 | yield* TestClock.adjust(5000) |
| 2066 | assert.strictEqual(Queue.sizeUnsafe(state.interrupts), 1) |
| 2067 | assert.strictEqual(journalInterrupts(driver), 1) |
| 2068 | }).pipe(Effect.provide(TestSharding))) |
| 2069 | |
| 2070 | it.effect("forwards interrupts from an external same-node caller", () => |
| 2071 | Effect.gen(function*() { |
| 2072 | const driver = yield* MessageStorage.MemoryDriver |
| 2073 | const state = yield* TestEntityState |
| 2074 | const makeClient = yield* TestEntity.client |
| 2075 | yield* TestClock.adjust(1) |
| 2076 | const client = makeClient("1") |
| 2077 | |
| 2078 | const fiber = yield* client.Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 2079 | yield* TestClock.adjust(1) |
| 2080 | yield* Fiber.interrupt(fiber) |
| 2081 | yield* TestClock.adjust(1) |
| 2082 | |
| 2083 | assert.strictEqual(journalInterrupts(driver), 1) |
| 2084 | assert.strictEqual(Queue.sizeUnsafe(state.interrupts), 1) |
| 2085 | }).pipe(Effect.provide(TestSharding))) |
| 2086 | |
| 2087 | it.effect("forwards interrupts from an external same-node caller during entity teardown", () => |
| 2088 | Effect.gen(function*() { |
| 2089 | const driver = yield* MessageStorage.MemoryDriver |
| 2090 | const state = yield* TestEntityState |
| 2091 | const sharding = yield* Sharding.Sharding |
| 2092 | const makeClient = yield* TestEntity.client |
| 2093 | yield* TestClock.adjust(1) |
| 2094 | const entityId = EntityId.make("external-teardown") |
| 2095 | const client = makeClient("external-teardown") |
| 2096 | const address = EntityAddress.make({ |
| 2097 | shardId: sharding.getShardId(entityId, "default"), |
| 2098 | entityType: EntityType.make(TestEntity.type), |
| 2099 | entityId |
| 2100 | }) |
| 2101 | |
| 2102 | const fiber = yield* client.Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 2103 | yield* TestClock.adjust(1) |
| 2104 | ActiveTeardown.acquireEntity(address) |
| 2105 | yield* Fiber.interrupt(fiber).pipe( |
| 2106 | Effect.ensuring(Effect.sync(() => ActiveTeardown.releaseEntity(address))) |
| 2107 | ) |
| 2108 | yield* TestClock.adjust(1) |
| 2109 | |
| 2110 | assert.strictEqual(journalInterrupts(driver), 1) |
| 2111 | assert.strictEqual(Queue.sizeUnsafe(state.interrupts), 1) |
| 2112 | }).pipe(Effect.provide(TestSharding))) |
| 2113 | |
| 2114 | it.effect("returns registry size to baseline after entity reap storms", () => |
| 2115 | Effect.gen(function*() { |
| 2116 | yield* TestClock.adjust(1) |
| 2117 | const sharding = yield* Sharding.Sharding |
| 2118 | const makeClient = yield* ReapStormEntity.client |
| 2119 | |
| 2120 | const storm = Effect.fnUntraced(function*(offset: number) { |
| 2121 | for (let i = 0; i < 20; i++) { |
| 2122 | yield* makeClient(String(offset + i)).Ping() |
| 2123 | } |
| 2124 | assert.strictEqual(yield* sharding.activeEntityCount, 20) |
| 2125 | yield* waitForIdleReap(sharding, 0) |
| 2126 | assert.strictEqual(yield* sharding.activeEntityCount, 0) |
| 2127 | for (let i = 0; i < 20; i++) { |
| 2128 | const entityId = EntityId.make(String(offset + i)) |
| 2129 | assert.isFalse(ActiveTeardown.isActive(EntityAddress.make({ |
| 2130 | shardId: sharding.getShardId(entityId, "default"), |
| 2131 | entityType: EntityType.make("ReapStormEntity"), |
| 2132 | entityId |
| 2133 | }))) |
| 2134 | } |
| 2135 | }) |
| 2136 | |
| 2137 | yield* storm(0) |
| 2138 | yield* storm(100) |
| 2139 | }).pipe(Effect.provide(ReapStormSharding))) |
| 2140 | }) |
| 2141 | |
| 2142 | const ReapStormEntity = Entity.make("ReapStormEntity", [ |
| 2143 | Rpc.make("Ping").annotate(ClusterSchema.Persisted, false) |
| 2144 | ]) |
| 2145 | |
| 2146 | const ReapStormEntityLayer = ReapStormEntity.toLayer({ Ping: () => Effect.void }) |
| 2147 | |
| 2148 | describe.concurrent("Sharding residency cap", () => { |
| 2149 | it.effect("bounds resident entities to maxResidentEntities", () => |
| 2150 | Effect.gen(function*() { |
| 2151 | yield* TestClock.adjust(1) |
| 2152 | const sharding = yield* Sharding.Sharding |
| 2153 | const makeClient = yield* TestEntity.client |
| 2154 | const fibers: Array<Fiber.Fiber<User, any>> = [] |
| 2155 | for (let i = 1; i <= 6; i++) { |
| 2156 | fibers.push( |
| 2157 | yield* makeClient(String(i)).GetUser({ id: i }).pipe( |
| 2158 | Effect.forkChild({ startImmediately: true }) |
| 2159 | ) |
| 2160 | ) |
| 2161 | } |
| 2162 | yield* TestClock.adjust(1) |
| 2163 | assert.isAtMost(yield* sharding.activeEntityCount, 2) |
| 2164 | |
| 2165 | // idle entities are reaped over time, freeing slots for the backlog |
| 2166 | for (let i = 0; i < 12; i++) { |
| 2167 | yield* TestClock.adjust(5000) |
| 2168 | assert.isAtMost(yield* sharding.activeEntityCount, 2) |
| 2169 | } |
| 2170 | |
| 2171 | const users = yield* Fiber.joinAll(fibers) |
| 2172 | assert.deepStrictEqual(users.map((user) => user.id), [1, 2, 3, 4, 5, 6]) |
| 2173 | }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 2, entityMaxIdleTime: 1000 })))) |
| 2174 | |
| 2175 | it.effect("volatile sends to new entities fail with MailboxFull at the cap", () => |
| 2176 | Effect.gen(function*() { |
| 2177 | yield* TestClock.adjust(1) |
| 2178 | const driver = yield* MessageStorage.MemoryDriver |
| 2179 | const sharding = yield* Sharding.Sharding |
| 2180 | const makeClient = yield* TestEntity.client |
| 2181 | // occupy the only slot |
| 2182 | yield* makeClient("1").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) |
| 2183 | yield* TestClock.adjust(1) |
| 2184 | assert.strictEqual(yield* sharding.activeEntityCount, 1) |
| 2185 | |
| 2186 | const error = yield* makeClient("2").GetUserVolatile({ id: 2 }).pipe(Effect.flip) |
| 2187 | assert.strictEqual(error._tag, "MailboxFull") |
| 2188 | |
| 2189 | // persisted sends still succeed and wait in storage |
| 2190 | yield* makeClient("2").GetUser({ id: 2 }, { discard: true }) |
| 2191 | yield* TestClock.adjust(5000) |
| 2192 | assert.strictEqual(yield* sharding.activeEntityCount, 1) |
| 2193 | assert.strictEqual(driver.unprocessed.size, 1) |
| 2194 | assert.strictEqual(driver.replyIds.size, 0) |
| 2195 | }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 1 })))) |
| 2196 | |
| 2197 | it.effect("keeps delivering to resident entities at the cap", () => |
| 2198 | Effect.gen(function*() { |
| 2199 | yield* TestClock.adjust(1) |
| 2200 | const driver = yield* MessageStorage.MemoryDriver |
| 2201 | const state = yield* TestEntityState |
| 2202 | const sharding = yield* Sharding.Sharding |
| 2203 | const makeClient = yield* TestEntity.client |
| 2204 | // make entity "1" resident and keep it busy |
| 2205 | yield* makeClient("1").NeverFork().pipe(Effect.forkChild({ startImmediately: true })) |
| 2206 | yield* TestClock.adjust(1) |
| 2207 | assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) |
| 2208 | |
| 2209 | // a backlog of new entity ids in front of the resident's next message |
| 2210 | for (let i = 2; i <= 4; i++) { |
| 2211 | yield* makeClient(String(i)).GetUser({ id: i }, { discard: true }) |
| 2212 | } |
| 2213 | const fiber = yield* makeClient("1").GetUser({ id: 1 }).pipe( |
| 2214 | Effect.forkChild({ startImmediately: true }) |
| 2215 | ) |
| 2216 | yield* TestClock.adjust(5000) |
| 2217 | |
| 2218 | // the resident entity received its message despite the backlog in front |
| 2219 | assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 1, name: "User 1" })) |
| 2220 | // the new ids were not admitted and their requests stay in storage |
| 2221 | assert.strictEqual(yield* sharding.activeEntityCount, 1) |
| 2222 | assert.strictEqual(driver.unprocessed.size, 4) |
| 2223 | assert.strictEqual(driver.replyIds.size, 1) |
| 2224 | }).pipe(Effect.provide(CappedSharding({ |
| 2225 | maxResidentEntities: 1, |
| 2226 | unprocessedMessageBatchSize: 2 |
| 2227 | })))) |
| 2228 | |
| 2229 | it.effect("drains a full batch without waiting for the poll interval", () => |
| 2230 | Effect.gen(function*() { |
| 2231 | yield* TestClock.adjust(1) |
| 2232 | const driver = yield* MessageStorage.MemoryDriver |
| 2233 | |
| 2234 | for (let i = 1; i <= 5; i++) { |
| 2235 | yield* saveGetUserRequest(String(i), i) |
| 2236 | } |
| 2237 | assert.strictEqual(driver.replyIds.size, 0) |
| 2238 | |
| 2239 | // a single poll drains the whole backlog in batches of 2 |
| 2240 | yield* TestClock.adjust(5000) |
| 2241 | assert.strictEqual(driver.replyIds.size, 5) |
| 2242 | }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 })))) |
| 2243 | |
| 2244 | it.effect("advances past in-flight memory requests when reading bounded batches", () => |
| 2245 | Effect.gen(function*() { |
| 2246 | yield* TestClock.adjust(1) |
| 2247 | const makeClient = yield* TestEntity.client |
| 2248 | yield* makeClient("1").Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 2249 | yield* makeClient("2").Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 2250 | const fiber = yield* makeClient("3").GetUser({ id: 3 }).pipe( |
| 2251 | Effect.forkChild({ startImmediately: true }) |
| 2252 | ) |
| 2253 | |
| 2254 | yield* TestClock.adjust(5000) |
| 2255 | assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 3, name: "User 3" })) |
| 2256 | }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 })))) |
| 2257 | |
| 2258 | it.effect("keeps full-sized storage reads when approaching the entity cap", () => |
| 2259 | Effect.gen(function*() { |
| 2260 | const limits: Array<number> = [] |
| 2261 | yield* Effect.gen(function*() { |
| 2262 | yield* TestClock.adjust(1) |
| 2263 | const sharding = yield* Sharding.Sharding |
| 2264 | const makeClient = yield* TestEntity.client |
| 2265 | for (let i = 1; i <= 4; i++) { |
| 2266 | yield* makeClient(String(i)).NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) |
| 2267 | } |
| 2268 | yield* TestClock.adjust(1) |
| 2269 | assert.strictEqual(yield* sharding.activeEntityCount, 4) |
| 2270 | limits.length = 0 |
| 2271 | |
| 2272 | for (let i = 0; i < 10; i++) { |
| 2273 | yield* saveGetUserRequest(String(i % 4 + 1), i) |
| 2274 | } |
| 2275 | yield* TestClock.adjust(5000) |
| 2276 | assert.isNotEmpty(limits) |
| 2277 | assert.deepStrictEqual([...new Set(limits)], [100]) |
| 2278 | }).pipe(Effect.provide(CappedSharding({ |
| 2279 | maxResidentEntities: 5, |
| 2280 | unprocessedMessageBatchSize: 100 |
| 2281 | }, (storage) => ({ |
| 2282 | ...storage, |
| 2283 | unprocessedMessages(shardIds, options) { |
| 2284 | if (options?.limit !== undefined) limits.push(options.limit) |
| 2285 | return storage.unprocessedMessages(shardIds, options) |
| 2286 | } |
| 2287 | })))) |
| 2288 | })) |
| 2289 | |
| 2290 | it.effect("normalizes programmatic batch sizes below one", () => |
| 2291 | Effect.forEach([0, -1], (unprocessedMessageBatchSize) => |
| 2292 | Effect.gen(function*() { |
| 2293 | yield* TestClock.adjust(1) |
| 2294 | const makeClient = yield* TestEntity.client |
| 2295 | const fiber = yield* makeClient("1").GetUser({ id: 1 }).pipe( |
| 2296 | Effect.forkChild({ startImmediately: true }) |
| 2297 | ) |
| 2298 | |
| 2299 | yield* TestClock.adjust(5000) |
| 2300 | assert.isDefined(fiber.pollUnsafe()) |
| 2301 | assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 1, name: "User 1" })) |
| 2302 | }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize }))), { discard: true })) |
| 2303 | |
| 2304 | it.effect("does not busy-poll storage at the cap", () => |
| 2305 | Effect.gen(function*() { |
| 2306 | let reads = 0 |
| 2307 | yield* Effect.gen(function*() { |
| 2308 | yield* TestClock.adjust(1) |
| 2309 | const makeClient = yield* TestEntity.client |
| 2310 | yield* makeClient("1").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) |
| 2311 | yield* TestClock.adjust(1) |
| 2312 | // a backlog for a new id that cannot be admitted |
| 2313 | yield* makeClient("2").GetUser({ id: 2 }, { discard: true }) |
| 2314 | yield* TestClock.adjust(1) |
| 2315 | const before = reads |
| 2316 | yield* TestClock.adjust(4000) // less than the poll interval |
| 2317 | assert.isAtMost(reads - before, 1) |
| 2318 | }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 1 }, (storage) => ({ |
| 2319 | ...storage, |
| 2320 | unprocessedMessages(shardIds, options) { |
| 2321 | reads++ |
| 2322 | return storage.unprocessedMessages(shardIds, options) |
| 2323 | } |
| 2324 | })))) |
| 2325 | })) |
| 2326 | |
| 2327 | it.effect("unbounded maxResidentEntities does not limit spawning", () => |
| 2328 | Effect.gen(function*() { |
| 2329 | yield* TestClock.adjust(1) |
| 2330 | const sharding = yield* Sharding.Sharding |
| 2331 | const makeClient = yield* TestEntity.client |
| 2332 | for (let i = 1; i <= 15; i++) { |
| 2333 | yield* makeClient(String(i)).NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) |
| 2334 | } |
| 2335 | yield* TestClock.adjust(1) |
| 2336 | assert.strictEqual(yield* sharding.activeEntityCount, 15) |
| 2337 | }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: "unbounded" })))) |
| 2338 | }) |
| 2339 | |
| 2340 | describe("Sharding shard lock failover", { concurrent: false }, () => { |
| 2341 | it.effect("interrupts entities and reacquires shards after lock storage recovers", () => |
| 2342 | Effect.gen(function*() { |
| 2343 | const warnings: Array<unknown> = [] |
| 2344 | const logger = Logger.make<unknown, void>((options) => { |
| 2345 | if (options.logLevel === "Warn") { |
| 2346 | warnings.push(options.message) |
| 2347 | } |
| 2348 | }) |
| 2349 | const storageState = makeFailoverStorageState() |
| 2350 | const runnerStorage = Layer.effect( |
| 2351 | RunnerStorage.RunnerStorage, |
| 2352 | Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) |
| 2353 | ) |
| 2354 | const config = ShardingConfig.layer({ |
| 2355 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2356 | shardsPerGroup: 1, |
| 2357 | shardLockExpiration: 300, |
| 2358 | shardLockRefreshInterval: 1000, |
| 2359 | entityTerminationTimeout: 30_000, |
| 2360 | entityMessagePollInterval: 10, |
| 2361 | refreshAssignmentsInterval: 10, |
| 2362 | sendRetryInterval: 10 |
| 2363 | }) |
| 2364 | const layer = TestEntityNoState.pipe( |
| 2365 | Layer.provideMerge(Sharding.layer), |
| 2366 | Layer.provide(runnerStorage), |
| 2367 | Layer.provide(RunnerHealth.layerNoop), |
| 2368 | Layer.provideMerge(TestEntityState.layer), |
| 2369 | Layer.provide(Runners.layerNoop), |
| 2370 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2371 | Layer.provide(config) |
| 2372 | ) |
| 2373 | |
| 2374 | yield* Effect.gen(function*() { |
| 2375 | const sharding = yield* Sharding.Sharding |
| 2376 | const entityState = yield* TestEntityState |
| 2377 | const makeClient = yield* TestEntity.client |
| 2378 | const client = makeClient("1") |
| 2379 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 2380 | |
| 2381 | while (!sharding.hasShardId(shardId)) { |
| 2382 | yield* TestClock.adjust(10) |
| 2383 | } |
| 2384 | while (!storageState.refreshCalls.some((call) => call.shards.length > 0)) { |
| 2385 | yield* TestClock.adjust(100) |
| 2386 | } |
| 2387 | |
| 2388 | const entityFiber = yield* client.NeverVolatile().pipe( |
| 2389 | Effect.forkChild({ startImmediately: true }) |
| 2390 | ) |
| 2391 | yield* TestClock.adjust(1) |
| 2392 | assert.strictEqual(Queue.sizeUnsafe(entityState.envelopes), 1) |
| 2393 | |
| 2394 | const acquireCount = storageState.acquireCalls.length |
| 2395 | const partitionedAt = yield* Clock.currentTimeMillis |
| 2396 | storageState.blackholed = true |
| 2397 | |
| 2398 | yield* TestClock.adjust(201) |
| 2399 | |
| 2400 | assert.isFalse(sharding.hasShardId(shardId)) |
| 2401 | assert.isFalse(yield* sharding.isShutdown) |
| 2402 | const entityExit = entityFiber.pollUnsafe() |
| 2403 | assert(entityExit && Exit.hasInterrupts(entityExit)) |
| 2404 | assert.strictEqual(ClusterMetrics.shards.valueUnsafe(Context.empty()).value, BigInt(0)) |
| 2405 | |
| 2406 | const failedRefreshes = storageState.refreshCalls.filter((call) => |
| 2407 | call.at >= partitionedAt && call.shards.length > 0 |
| 2408 | ) |
| 2409 | assert.isAtMost(failedRefreshes.length, 2) |
| 2410 | |
| 2411 | yield* TestClock.adjust(1000) |
| 2412 | assert.strictEqual(storageState.acquireCalls.length, acquireCount) |
| 2413 | assert(storageState.refreshCalls.some((call) => call.at >= partitionedAt && call.shards.length === 0)) |
| 2414 | assert( |
| 2415 | storageState.refreshCalls |
| 2416 | .filter((call) => call.at >= partitionedAt + 200) |
| 2417 | .every((call) => call.shards.length === 0) |
| 2418 | ) |
| 2419 | |
| 2420 | storageState.blackholed = false |
| 2421 | yield* TestClock.adjust(101) |
| 2422 | while (!sharding.hasShardId(shardId)) { |
| 2423 | yield* TestClock.adjust(10) |
| 2424 | } |
| 2425 | |
| 2426 | assert.isAbove(storageState.acquireCalls.length, acquireCount) |
| 2427 | assert(storageState.acquireCalls.at(-1)!.shards.some((shard) => shard.id === shardId.id)) |
| 2428 | assert.deepStrictEqual(yield* client.GetUserVolatile({ id: 2 }), new User({ id: 2, name: "User 2" })) |
| 2429 | assert.strictEqual(Queue.sizeUnsafe(entityState.envelopes), 2) |
| 2430 | assert.isTrue(warnings.some((message) => |
| 2431 | globalThis.Array.isArray(message) && message.includes("Shard lock storage is still unhealthy") |
| 2432 | )) |
| 2433 | }).pipe(Effect.provide(layer), Effect.withLogger(logger), Effect.scoped) |
| 2434 | })) |
| 2435 | |
| 2436 | it.effect("keeps shards acquired while an earlier lock refresh is in flight", () => |
| 2437 | Effect.gen(function*() { |
| 2438 | const storageState = makeFailoverStorageState() |
| 2439 | const acquireStarted = yield* Deferred.make<void>() |
| 2440 | const refreshStarted = yield* Deferred.make<void>() |
| 2441 | const acquireDone = yield* Deferred.make<void>() |
| 2442 | const runnerStorage = Layer.effect( |
| 2443 | RunnerStorage.RunnerStorage, |
| 2444 | Effect.map(Clock.Clock, (clock) => { |
| 2445 | const storage = makeFailoverStorage(storageState, clock) |
| 2446 | return RunnerStorage.RunnerStorage.of({ |
| 2447 | ...storage, |
| 2448 | acquire: Effect.fnUntraced(function*(address, shardIds) { |
| 2449 | if (yield* Deferred.isDone(acquireDone)) { |
| 2450 | return yield* storage.acquire(address, shardIds) |
| 2451 | } |
| 2452 | yield* Deferred.succeed(acquireStarted, void 0) |
| 2453 | yield* Deferred.await(refreshStarted) |
| 2454 | const acquired = yield* storage.acquire(address, shardIds) |
| 2455 | yield* Deferred.succeed(acquireDone, void 0) |
| 2456 | return acquired |
| 2457 | }), |
| 2458 | refresh: Effect.fnUntraced(function*(address, shardIds) { |
| 2459 | const shards = globalThis.Array.from(shardIds) |
| 2460 | if ((yield* Deferred.isDone(acquireStarted)) && !(yield* Deferred.isDone(acquireDone))) { |
| 2461 | assert.deepStrictEqual(shards, []) |
| 2462 | yield* Deferred.succeed(refreshStarted, void 0) |
| 2463 | yield* Deferred.await(acquireDone) |
| 2464 | } |
| 2465 | return yield* storage.refresh(address, shards) |
| 2466 | }) |
| 2467 | }) |
| 2468 | }) |
| 2469 | ) |
| 2470 | const config = ShardingConfig.layer({ |
| 2471 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2472 | shardsPerGroup: 1, |
| 2473 | shardLockExpiration: 3000, |
| 2474 | shardLockRefreshInterval: 100, |
| 2475 | entityTerminationTimeout: 0, |
| 2476 | refreshAssignmentsInterval: 10 |
| 2477 | }) |
| 2478 | const layer = Sharding.layer.pipe( |
| 2479 | Layer.provide(runnerStorage), |
| 2480 | Layer.provide(RunnerHealth.layerNoop), |
| 2481 | Layer.provide(Runners.layerNoop), |
| 2482 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2483 | Layer.provide(config) |
| 2484 | ) |
| 2485 | |
| 2486 | yield* Effect.gen(function*() { |
| 2487 | const sharding = yield* Sharding.Sharding |
| 2488 | const shardId = ShardId.make("default", 1) |
| 2489 | |
| 2490 | yield* TestClock.adjust(100) |
| 2491 | assert.isTrue(yield* Deferred.isDone(acquireDone)) |
| 2492 | assert.isTrue(sharding.hasShardId(shardId)) |
| 2493 | |
| 2494 | yield* TestClock.adjust(500) |
| 2495 | |
| 2496 | assert.isTrue(sharding.hasShardId(shardId), "shard acquired during the refresh was dropped") |
| 2497 | assert.deepStrictEqual(storageState.releaseCalls, []) |
| 2498 | assert.strictEqual(storageState.acquireCalls.length, 1) |
| 2499 | }).pipe(Effect.provide(layer)) |
| 2500 | })) |
| 2501 | |
| 2502 | it.effect("releases a lost requested shard without dropping a shard acquired during refresh", () => |
| 2503 | Effect.gen(function*() { |
| 2504 | const storageState = makeFailoverStorageState() |
| 2505 | const firstShard = ShardId.make("default", 1) |
| 2506 | const secondShard = ShardId.make("default", 2) |
| 2507 | const refreshStarted = yield* Deferred.make<void>() |
| 2508 | const refreshResponse = yield* Deferred.make<void>() |
| 2509 | let acquireCount = 0 |
| 2510 | let lostShard = false |
| 2511 | const runnerStorage = Layer.effect( |
| 2512 | RunnerStorage.RunnerStorage, |
| 2513 | Effect.map(Clock.Clock, (clock) => { |
| 2514 | const storage = makeFailoverStorage(storageState, clock) |
| 2515 | return RunnerStorage.RunnerStorage.of({ |
| 2516 | ...storage, |
| 2517 | acquire: Effect.fnUntraced(function*(address, shardIds) { |
| 2518 | acquireCount++ |
| 2519 | if (acquireCount === 1) { |
| 2520 | return yield* storage.acquire(address, [firstShard]) |
| 2521 | } |
| 2522 | if (acquireCount === 2) { |
| 2523 | yield* Deferred.await(refreshStarted) |
| 2524 | } |
| 2525 | return yield* storage.acquire( |
| 2526 | address, |
| 2527 | globalThis.Array.from(shardIds).filter((shard) => shard.id !== firstShard.id) |
| 2528 | ) |
| 2529 | }), |
| 2530 | refresh: Effect.fnUntraced(function*(address, shardIds) { |
| 2531 | const shards = globalThis.Array.from(shardIds) |
| 2532 | if (!lostShard && shards.length > 0) { |
| 2533 | lostShard = true |
| 2534 | assert.deepStrictEqual(shards, [firstShard]) |
| 2535 | yield* Deferred.succeed(refreshStarted, void 0) |
| 2536 | yield* Deferred.await(refreshResponse) |
| 2537 | return [] |
| 2538 | } |
| 2539 | return yield* storage.refresh(address, shards) |
| 2540 | }) |
| 2541 | }) |
| 2542 | }) |
| 2543 | ) |
| 2544 | const layer = Sharding.layer.pipe( |
| 2545 | Layer.provide(runnerStorage), |
| 2546 | Layer.provide(RunnerHealth.layerNoop), |
| 2547 | Layer.provide(Runners.layerNoop), |
| 2548 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2549 | Layer.provide(ShardingConfig.layer({ |
| 2550 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2551 | shardsPerGroup: 2, |
| 2552 | shardLockExpiration: 30_000, |
| 2553 | shardLockRefreshInterval: 2000, |
| 2554 | entityTerminationTimeout: 0, |
| 2555 | refreshAssignmentsInterval: 10 |
| 2556 | })) |
| 2557 | ) |
| 2558 | |
| 2559 | yield* Effect.gen(function*() { |
| 2560 | const sharding = yield* Sharding.Sharding |
| 2561 | |
| 2562 | yield* TestClock.adjust(2000) |
| 2563 | assert.isTrue(yield* Deferred.isDone(refreshStarted)) |
| 2564 | assert.isTrue(sharding.hasShardId(firstShard)) |
| 2565 | assert.isTrue(sharding.hasShardId(secondShard)) |
| 2566 | |
| 2567 | yield* Deferred.succeed(refreshResponse, void 0) |
| 2568 | yield* TestClock.adjust(1) |
| 2569 | |
| 2570 | assert.isFalse(sharding.hasShardId(firstShard)) |
| 2571 | assert.isTrue(sharding.hasShardId(secondShard)) |
| 2572 | |
| 2573 | yield* TestClock.adjust(1000) |
| 2574 | |
| 2575 | assert.isFalse(sharding.hasShardId(firstShard)) |
| 2576 | assert.isTrue(sharding.hasShardId(secondShard)) |
| 2577 | assert.deepStrictEqual(storageState.releaseCalls, [firstShard]) |
| 2578 | }).pipe(Effect.provide(layer)) |
| 2579 | })) |
| 2580 | |
| 2581 | it.effect("refreshes a newly acquired shard when retrying a failed lock refresh", () => |
| 2582 | Effect.gen(function*() { |
| 2583 | const storageState = makeFailoverStorageState() |
| 2584 | const acquireStarted = yield* Deferred.make<void>() |
| 2585 | const refreshFailed = yield* Deferred.make<void>() |
| 2586 | const retryShards = yield* Deferred.make<Array<ShardId.ShardId>>() |
| 2587 | const runnerStorage = Layer.effect( |
| 2588 | RunnerStorage.RunnerStorage, |
| 2589 | Effect.map(Clock.Clock, (clock) => { |
| 2590 | const storage = makeFailoverStorage(storageState, clock) |
| 2591 | return RunnerStorage.RunnerStorage.of({ |
| 2592 | ...storage, |
| 2593 | acquire: Effect.fnUntraced(function*(address, shardIds) { |
| 2594 | yield* Deferred.succeed(acquireStarted, void 0) |
| 2595 | yield* Deferred.await(refreshFailed) |
| 2596 | return yield* storage.acquire(address, shardIds) |
| 2597 | }), |
| 2598 | refresh: Effect.fnUntraced(function*(address, shardIds) { |
| 2599 | const shards = globalThis.Array.from(shardIds) |
| 2600 | if (yield* Deferred.isDone(refreshFailed)) { |
| 2601 | yield* Deferred.succeed(retryShards, shards) |
| 2602 | } else if (yield* Deferred.isDone(acquireStarted)) { |
| 2603 | assert.deepStrictEqual(shards, []) |
| 2604 | yield* Deferred.succeed(refreshFailed, void 0) |
| 2605 | return yield* Effect.fail(new ClusterError.PersistenceError({ cause: "refresh failed" })) |
| 2606 | } |
| 2607 | return yield* storage.refresh(address, shards) |
| 2608 | }) |
| 2609 | }) |
| 2610 | }) |
| 2611 | ) |
| 2612 | const layer = Sharding.layer.pipe( |
| 2613 | Layer.provide(runnerStorage), |
| 2614 | Layer.provide(RunnerHealth.layerNoop), |
| 2615 | Layer.provide(Runners.layerNoop), |
| 2616 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2617 | Layer.provide(ShardingConfig.layer({ |
| 2618 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2619 | shardsPerGroup: 1, |
| 2620 | shardLockExpiration: 30_000, |
| 2621 | shardLockRefreshInterval: 1000, |
| 2622 | entityTerminationTimeout: 0, |
| 2623 | refreshAssignmentsInterval: 10 |
| 2624 | })) |
| 2625 | ) |
| 2626 | |
| 2627 | yield* Effect.gen(function*() { |
| 2628 | const sharding = yield* Sharding.Sharding |
| 2629 | const shardId = ShardId.make("default", 1) |
| 2630 | |
| 2631 | yield* TestClock.adjust(1000) |
| 2632 | assert.isTrue(yield* Deferred.isDone(refreshFailed)) |
| 2633 | assert.isTrue(sharding.hasShardId(shardId)) |
| 2634 | assert.isFalse(yield* Deferred.isDone(retryShards)) |
| 2635 | |
| 2636 | yield* TestClock.adjust(50) |
| 2637 | |
| 2638 | assert.isTrue(yield* Deferred.isDone(retryShards)) |
| 2639 | assert.deepStrictEqual(yield* Deferred.await(retryShards), [shardId]) |
| 2640 | assert.isTrue(sharding.hasShardId(shardId)) |
| 2641 | assert.deepStrictEqual(storageState.releaseCalls, []) |
| 2642 | }).pipe(Effect.provide(layer)) |
| 2643 | })) |
| 2644 | |
| 2645 | it.effect("reacquires shards when the liveness probe succeeds while lock refresh is hung", () => |
| 2646 | Effect.gen(function*() { |
| 2647 | const storageState = makeFailoverStorageState() |
| 2648 | const runnerStorage = Layer.effect( |
| 2649 | RunnerStorage.RunnerStorage, |
| 2650 | Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) |
| 2651 | ) |
| 2652 | const config = ShardingConfig.layer({ |
| 2653 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2654 | shardsPerGroup: 1, |
| 2655 | shardLockExpiration: 300, |
| 2656 | shardLockRefreshInterval: 1000, |
| 2657 | entityTerminationTimeout: 30_000, |
| 2658 | entityMessagePollInterval: 10, |
| 2659 | refreshAssignmentsInterval: 10, |
| 2660 | sendRetryInterval: 10 |
| 2661 | }) |
| 2662 | const layer = TestEntityNoState.pipe( |
| 2663 | Layer.provideMerge(Sharding.layer), |
| 2664 | Layer.provide(runnerStorage), |
| 2665 | Layer.provide(RunnerHealth.layerNoop), |
| 2666 | Layer.provideMerge(TestEntityState.layer), |
| 2667 | Layer.provide(Runners.layerNoop), |
| 2668 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2669 | Layer.provide(config) |
| 2670 | ) |
| 2671 | |
| 2672 | yield* Effect.gen(function*() { |
| 2673 | const sharding = yield* Sharding.Sharding |
| 2674 | const makeClient = yield* TestEntity.client |
| 2675 | const client = makeClient("1") |
| 2676 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 2677 | |
| 2678 | while (!sharding.hasShardId(shardId)) { |
| 2679 | yield* TestClock.adjust(10) |
| 2680 | } |
| 2681 | while (!storageState.refreshCalls.some((call) => call.shards.length > 0)) { |
| 2682 | yield* TestClock.adjust(100) |
| 2683 | } |
| 2684 | |
| 2685 | const acquireCount = storageState.acquireCalls.length |
| 2686 | storageState.blackholeNonEmptyRefresh = true |
| 2687 | yield* TestClock.adjust(201) |
| 2688 | assert.isFalse(sharding.hasShardId(shardId)) |
| 2689 | |
| 2690 | // recovery only needs the empty liveness probe to succeed |
| 2691 | while (!sharding.hasShardId(shardId)) { |
| 2692 | yield* TestClock.adjust(10) |
| 2693 | } |
| 2694 | |
| 2695 | assert.isAbove(storageState.acquireCalls.length, acquireCount) |
| 2696 | assert(storageState.refreshCalls.some((call) => call.shards.length === 0)) |
| 2697 | assert.deepStrictEqual(yield* client.GetUserVolatile({ id: 3 }), new User({ id: 3, name: "User 3" })) |
| 2698 | }).pipe(Effect.provide(layer), Effect.scoped) |
| 2699 | })) |
| 2700 | |
| 2701 | it.effect("keeps the graceful timeout for normal shard reassignment", () => |
| 2702 | Effect.gen(function*() { |
| 2703 | const storageState = makeFailoverStorageState() |
| 2704 | const unregisterInterrupts: Array<boolean> = [] |
| 2705 | const runnerStorage = Layer.effect( |
| 2706 | RunnerStorage.RunnerStorage, |
| 2707 | Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) |
| 2708 | ) |
| 2709 | const shardingLayer = Sharding.layer.pipe( |
| 2710 | Layer.updateService(MessageStorage.MessageStorage, (storage) => ({ |
| 2711 | ...storage, |
| 2712 | unregisterShardReplyHandlers: (shardId, options) => |
| 2713 | Effect.sync(() => unregisterInterrupts.push(options?.interrupt ?? false)).pipe( |
| 2714 | Effect.andThen(storage.unregisterShardReplyHandlers(shardId, options)) |
| 2715 | ) |
| 2716 | })) |
| 2717 | ) |
| 2718 | const config = ShardingConfig.layer({ |
| 2719 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2720 | shardsPerGroup: 1, |
| 2721 | shardLockExpiration: 3000, |
| 2722 | shardLockRefreshInterval: 100, |
| 2723 | entityTerminationTimeout: 1000, |
| 2724 | entityMessagePollInterval: 10, |
| 2725 | refreshAssignmentsInterval: 10, |
| 2726 | sendRetryInterval: 10 |
| 2727 | }) |
| 2728 | const layer = TestEntityNoState.pipe( |
| 2729 | Layer.provideMerge(shardingLayer), |
| 2730 | Layer.provide(runnerStorage), |
| 2731 | Layer.provide(RunnerHealth.layerNoop), |
| 2732 | Layer.provideMerge(TestEntityState.layer), |
| 2733 | Layer.provide(Runners.layerNoop), |
| 2734 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2735 | Layer.provide(config) |
| 2736 | ) |
| 2737 | |
| 2738 | yield* Effect.gen(function*() { |
| 2739 | const sharding = yield* Sharding.Sharding |
| 2740 | const makeClient = yield* TestEntity.client |
| 2741 | const client = makeClient("1") |
| 2742 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 2743 | |
| 2744 | while (!sharding.hasShardId(shardId)) { |
| 2745 | yield* TestClock.adjust(10) |
| 2746 | } |
| 2747 | const entityFiber = yield* client.NeverVolatile().pipe( |
| 2748 | Effect.forkChild({ startImmediately: true }) |
| 2749 | ) |
| 2750 | yield* TestClock.adjust(1) |
| 2751 | |
| 2752 | storageState.assignSelf = false |
| 2753 | while (sharding.hasShardId(shardId)) { |
| 2754 | yield* TestClock.adjust(10) |
| 2755 | } |
| 2756 | while ((yield* sharding.activeEntityCount) > 0) { |
| 2757 | yield* TestClock.adjust(1) |
| 2758 | } |
| 2759 | |
| 2760 | assert.isUndefined(entityFiber.pollUnsafe()) |
| 2761 | assert.strictEqual(storageState.releaseCalls.length, 0) |
| 2762 | yield* TestClock.adjust(900) |
| 2763 | assert.isUndefined(entityFiber.pollUnsafe()) |
| 2764 | assert.strictEqual(storageState.releaseCalls.length, 0) |
| 2765 | |
| 2766 | for (let i = 0; i < 20 && entityFiber.pollUnsafe() === undefined; i++) { |
| 2767 | yield* TestClock.adjust(10) |
| 2768 | } |
| 2769 | const entityExit = entityFiber.pollUnsafe() |
| 2770 | assert(entityExit && Exit.hasInterrupts(entityExit)) |
| 2771 | assert.strictEqual(storageState.releaseCalls.length, 1) |
| 2772 | assert.deepStrictEqual(unregisterInterrupts, [false]) |
| 2773 | }).pipe(Effect.provide(layer), Effect.scoped) |
| 2774 | })) |
| 2775 | |
| 2776 | it.effect("does not wait for entity construction before a forced shard release", () => |
| 2777 | Effect.gen(function*() { |
| 2778 | const storageState = makeFailoverStorageState() |
| 2779 | const runnerStorage = Layer.effect( |
| 2780 | RunnerStorage.RunnerStorage, |
| 2781 | Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) |
| 2782 | ) |
| 2783 | const config = ShardingConfig.layer({ |
| 2784 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2785 | shardsPerGroup: 1, |
| 2786 | shardLockExpiration: 300, |
| 2787 | shardLockRefreshInterval: 1000, |
| 2788 | entityTerminationTimeout: 0, |
| 2789 | entityMessagePollInterval: 10, |
| 2790 | refreshAssignmentsInterval: 10, |
| 2791 | sendRetryInterval: 10 |
| 2792 | }) |
| 2793 | const layer = TestEntityNoState.pipe( |
| 2794 | Layer.provideMerge(Sharding.layer), |
| 2795 | Layer.provide(runnerStorage), |
| 2796 | Layer.provide(RunnerHealth.layerNoop), |
| 2797 | Layer.provideMerge(TestEntityState.layer), |
| 2798 | Layer.provide(Runners.layerNoop), |
| 2799 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2800 | Layer.provide(config) |
| 2801 | ) |
| 2802 | |
| 2803 | yield* Effect.gen(function*() { |
| 2804 | const sharding = yield* Sharding.Sharding |
| 2805 | const entityState = yield* TestEntityState |
| 2806 | const makeClient = yield* TestEntity.client |
| 2807 | const client = makeClient("1") |
| 2808 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 2809 | |
| 2810 | while (!sharding.hasShardId(shardId)) { |
| 2811 | yield* TestClock.adjust(10) |
| 2812 | } |
| 2813 | while (!storageState.refreshCalls.some((call) => call.shards.length > 0)) { |
| 2814 | yield* TestClock.adjust(100) |
| 2815 | } |
| 2816 | |
| 2817 | yield* Effect.gen(function*() { |
| 2818 | entityState.buildLatch.closeUnsafe() |
| 2819 | const entityFiber = yield* client.GetUserVolatile({ id: 1 }).pipe( |
| 2820 | Effect.forkChild({ startImmediately: true }) |
| 2821 | ) |
| 2822 | while (entityState.layerBuilds.current === 0) { |
| 2823 | yield* TestClock.adjust(1) |
| 2824 | } |
| 2825 | |
| 2826 | storageState.blackholed = true |
| 2827 | yield* TestClock.adjust(201) |
| 2828 | assert.isFalse(sharding.hasShardId(shardId)) |
| 2829 | |
| 2830 | storageState.blackholed = false |
| 2831 | yield* TestClock.adjust(1000) |
| 2832 | |
| 2833 | assert.strictEqual(storageState.releaseAllCalls.length, 1) |
| 2834 | entityState.buildLatch.openUnsafe() |
| 2835 | yield* TestClock.adjust(1) |
| 2836 | yield* Fiber.interrupt(entityFiber) |
| 2837 | }).pipe(Effect.ensuring(entityState.buildLatch.open)) |
| 2838 | }).pipe( |
| 2839 | Effect.provide(layer), |
| 2840 | Effect.scoped |
| 2841 | ) |
| 2842 | })) |
| 2843 | |
| 2844 | it.effect("does not register an entity built after shard release", () => |
| 2845 | Effect.gen(function*() { |
| 2846 | const storageState = makeFailoverStorageState() |
| 2847 | const runnerStorage = Layer.effect( |
| 2848 | RunnerStorage.RunnerStorage, |
| 2849 | Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) |
| 2850 | ) |
| 2851 | const config = ShardingConfig.layer({ |
| 2852 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2853 | shardsPerGroup: 1, |
| 2854 | shardLockExpiration: 3000, |
| 2855 | shardLockRefreshInterval: 100, |
| 2856 | entityTerminationTimeout: 0, |
| 2857 | entityMessagePollInterval: 10, |
| 2858 | refreshAssignmentsInterval: 10, |
| 2859 | sendRetryInterval: 10 |
| 2860 | }) |
| 2861 | const layer = TestEntityNoState.pipe( |
| 2862 | Layer.provideMerge(Sharding.layer), |
| 2863 | Layer.provide(runnerStorage), |
| 2864 | Layer.provide(RunnerHealth.layerNoop), |
| 2865 | Layer.provideMerge(TestEntityState.layer), |
| 2866 | Layer.provide(Runners.layerNoop), |
| 2867 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2868 | Layer.provide(config) |
| 2869 | ) |
| 2870 | |
| 2871 | yield* Effect.gen(function*() { |
| 2872 | const sharding = yield* Sharding.Sharding |
| 2873 | const entityState = yield* TestEntityState |
| 2874 | const makeClient = yield* TestEntity.client |
| 2875 | const client = makeClient("1") |
| 2876 | const shardId = sharding.getShardId(EntityId.make("1"), "default") |
| 2877 | |
| 2878 | while (!sharding.hasShardId(shardId)) { |
| 2879 | yield* TestClock.adjust(10) |
| 2880 | } |
| 2881 | |
| 2882 | yield* Effect.gen(function*() { |
| 2883 | entityState.buildLatch.closeUnsafe() |
| 2884 | const entityFiber = yield* client.GetUserVolatile({ id: 1 }).pipe( |
| 2885 | Effect.forkChild({ startImmediately: true }) |
| 2886 | ) |
| 2887 | while (entityState.layerBuilds.current === 0) { |
| 2888 | yield* TestClock.adjust(1) |
| 2889 | } |
| 2890 | |
| 2891 | storageState.assignSelf = false |
| 2892 | while (storageState.releaseCalls.length === 0) { |
| 2893 | yield* TestClock.adjust(10) |
| 2894 | } |
| 2895 | assert.isFalse(sharding.hasShardId(shardId)) |
| 2896 | |
| 2897 | entityState.buildLatch.openUnsafe() |
| 2898 | yield* TestClock.adjust(10) |
| 2899 | |
| 2900 | assert.strictEqual(yield* sharding.activeEntityCount, 0) |
| 2901 | assert.strictEqual(Queue.sizeUnsafe(entityState.envelopes), 0) |
| 2902 | assert.isUndefined(entityFiber.pollUnsafe()) |
| 2903 | }).pipe(Effect.ensuring(entityState.buildLatch.open)) |
| 2904 | }).pipe( |
| 2905 | Effect.provide(layer), |
| 2906 | Effect.scoped |
| 2907 | ) |
| 2908 | })) |
| 2909 | |
| 2910 | it.effect("does not acquire shards while a forced release is pending", () => |
| 2911 | Effect.gen(function*() { |
| 2912 | const shardsPerGroup = 4 |
| 2913 | const storageState = makeFailoverStorageState({ |
| 2914 | otherRunnerHealthy: true, |
| 2915 | releaseAllDuration: 500 |
| 2916 | }) |
| 2917 | const runnerStorage = Layer.effect( |
| 2918 | RunnerStorage.RunnerStorage, |
| 2919 | Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) |
| 2920 | ) |
| 2921 | const config = ShardingConfig.layer({ |
| 2922 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 2923 | shardsPerGroup, |
| 2924 | shardLockExpiration: 300, |
| 2925 | shardLockRefreshInterval: 1000, |
| 2926 | entityTerminationTimeout: 0, |
| 2927 | entityMessagePollInterval: 10, |
| 2928 | refreshAssignmentsInterval: 10, |
| 2929 | sendRetryInterval: 10 |
| 2930 | }) |
| 2931 | const layer = TestEntityNoState.pipe( |
| 2932 | Layer.provideMerge(Sharding.layer), |
| 2933 | Layer.provide(runnerStorage), |
| 2934 | Layer.provide(RunnerHealth.layerNoop), |
| 2935 | Layer.provideMerge(TestEntityState.layer), |
| 2936 | Layer.provide(Runners.layerNoop), |
| 2937 | Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), |
| 2938 | Layer.provide(config) |
| 2939 | ) |
| 2940 | |
| 2941 | yield* Effect.gen(function*() { |
| 2942 | const sharding = yield* Sharding.Sharding |
| 2943 | const allShards = Array.makeBy(shardsPerGroup, (i) => ShardId.make("default", i + 1)) |
| 2944 | const ownedCount = () => allShards.filter((shardId) => sharding.hasShardId(shardId)).length |
| 2945 | |
| 2946 | // the other runner holds part of the ring, so this runner starts with a |
| 2947 | // strict subset of the shards |
| 2948 | while (ownedCount() === 0) { |
| 2949 | yield* TestClock.adjust(10) |
| 2950 | } |
| 2951 | assert.isBelow(ownedCount(), shardsPerGroup) |
| 2952 | while (!storageState.refreshCalls.some((call) => call.shards.length > 0)) { |
| 2953 | yield* TestClock.adjust(100) |
| 2954 | } |
| 2955 | |
| 2956 | const acquiresBeforeOutage = storageState.acquireCalls.length |
| 2957 | storageState.blackholed = true |
| 2958 | yield* TestClock.adjust(201) |
| 2959 | assert.strictEqual(ownedCount(), 0) |
| 2960 | |
| 2961 | // the other runner's shards are reassigned to this runner during the |
| 2962 | // outage, so they are not part of the forced release set |
| 2963 | storageState.otherRunnerHealthy = false |
| 2964 | yield* TestClock.adjust(100) |
| 2965 | assert.strictEqual(storageState.releaseAllCalls.length, 0) |
| 2966 | |
| 2967 | // recovery runs the forced release, which stays in flight for |
| 2968 | // `releaseAllDuration` |
| 2969 | storageState.blackholed = false |
| 2970 | while (storageState.releaseAllCalls.length === 0) { |
| 2971 | yield* TestClock.adjust(10) |
| 2972 | } |
| 2973 | while (!storageState.releaseAllCalls[0].completed) { |
| 2974 | yield* TestClock.adjust(10) |
| 2975 | } |
| 2976 | // keep shutdown from blocking on the finalizer release |
| 2977 | storageState.releaseAllDuration = 0 |
| 2978 | |
| 2979 | while (ownedCount() < shardsPerGroup) { |
| 2980 | yield* TestClock.adjust(10) |
| 2981 | } |
| 2982 | |
| 2983 | // `releaseAll` drops every lock held by this runner, so nothing may be |
| 2984 | // acquired before it has completed |
| 2985 | assert.strictEqual(storageState.releaseAllCalls.length, 1) |
| 2986 | assert( |
| 2987 | storageState.acquireCalls |
| 2988 | .slice(acquiresBeforeOutage) |
| 2989 | .every((call) => call.completedReleaseAlls > 0) |
| 2990 | ) |
| 2991 | }).pipe(Effect.provide(layer), Effect.scoped) |
| 2992 | })) |
| 2993 | }) |
| 2994 | |
| 2995 | interface FailoverStorageState { |
| 2996 | blackholed: boolean |
| 2997 | /** Hang only non-empty refreshes, so the empty liveness probe can succeed. */ |
| 2998 | blackholeNonEmptyRefresh: boolean |
| 2999 | assignSelf: boolean |
| 3000 | otherRunnerHealthy: boolean |
| 3001 | /** Test clock duration `releaseAll` stays in flight for. */ |
| 3002 | releaseAllDuration: number |
| 3003 | runner: Runner.Runner | undefined |
| 3004 | readonly acquireCalls: Array<{ |
| 3005 | readonly shards: Array<ShardId.ShardId> |
| 3006 | readonly completedReleaseAlls: number |
| 3007 | }> |
| 3008 | readonly refreshCalls: Array<{ |
| 3009 | readonly at: number |
| 3010 | readonly shards: Array<ShardId.ShardId> |
| 3011 | }> |
| 3012 | readonly releaseCalls: Array<ShardId.ShardId> |
| 3013 | readonly releaseAllCalls: Array<{ completed: boolean }> |
| 3014 | } |
| 3015 | |
| 3016 | const makeFailoverStorageState = ( |
| 3017 | overrides?: Partial<FailoverStorageState> |
| 3018 | ): FailoverStorageState => ({ |
| 3019 | blackholed: false, |
| 3020 | blackholeNonEmptyRefresh: false, |
| 3021 | assignSelf: true, |
| 3022 | otherRunnerHealthy: false, |
| 3023 | releaseAllDuration: 0, |
| 3024 | runner: undefined, |
| 3025 | acquireCalls: [], |
| 3026 | refreshCalls: [], |
| 3027 | releaseCalls: [], |
| 3028 | releaseAllCalls: [], |
| 3029 | ...overrides |
| 3030 | }) |
| 3031 | |
| 3032 | const makeFailoverStorage = (state: FailoverStorageState, clock: Clock.Clock) => |
| 3033 | RunnerStorage.RunnerStorage.of({ |
| 3034 | getRunners: Effect.sync(() => { |
| 3035 | if (!state.runner) return [] |
| 3036 | if (!state.assignSelf) return [[state.runner, false], [otherRunner, true]] |
| 3037 | return state.otherRunnerHealthy ? [[state.runner, true], [otherRunner, true]] : [[state.runner, true]] |
| 3038 | }), |
| 3039 | register: (runner) => |
| 3040 | Effect.sync(() => { |
| 3041 | state.runner = runner |
| 3042 | return MachineId.make(1) |
| 3043 | }), |
| 3044 | unregister: () => Effect.void, |
| 3045 | setRunnerHealth: () => Effect.void, |
| 3046 | acquire: (_address, shardIds) => |
| 3047 | Effect.sync(() => { |
| 3048 | const shards = globalThis.Array.from(shardIds) |
| 3049 | state.acquireCalls.push({ |
| 3050 | shards, |
| 3051 | completedReleaseAlls: state.releaseAllCalls.filter((call) => call.completed).length |
| 3052 | }) |
| 3053 | return shards |
| 3054 | }), |
| 3055 | refresh: (_address, shardIds) => |
| 3056 | Effect.suspend(() => { |
| 3057 | const shards = globalThis.Array.from(shardIds) |
| 3058 | state.refreshCalls.push({ |
| 3059 | at: clock.currentTimeMillisUnsafe(), |
| 3060 | shards |
| 3061 | }) |
| 3062 | return (state.blackholed || (state.blackholeNonEmptyRefresh && shards.length > 0)) |
| 3063 | ? Effect.never |
| 3064 | : Effect.succeed(shards) |
| 3065 | }), |
| 3066 | release: (_address, shardId) => |
| 3067 | Effect.sync(() => { |
| 3068 | state.releaseCalls.push(shardId) |
| 3069 | }), |
| 3070 | releaseAll: () => |
| 3071 | Effect.suspend(() => { |
| 3072 | const call = { completed: false } |
| 3073 | state.releaseAllCalls.push(call) |
| 3074 | return Effect.andThen( |
| 3075 | Effect.sleep(state.releaseAllDuration), |
| 3076 | Effect.sync(() => { |
| 3077 | call.completed = true |
| 3078 | }) |
| 3079 | ) |
| 3080 | }) |
| 3081 | }) |
| 3082 | |
| 3083 | const otherRunner = Runner.make({ |
| 3084 | address: RunnerAddress.make("localhost", 5678), |
| 3085 | groups: ["default"], |
| 3086 | weight: 1 |
| 3087 | }) |
| 3088 | |
| 3089 | class RegistrationContext extends Context.Service<RegistrationContext, string>()( |
| 3090 | "effect/test/cluster/RegistrationContext" |
| 3091 | ) {} |
| 3092 | |
| 3093 | const RegistrationContextEntity = Entity.make("RegistrationContextEntity", [ |
| 3094 | Rpc.make("Read", { success: Schema.String }).annotate(ClusterSchema.Persisted, false) |
| 3095 | ]) |
| 3096 | |
| 3097 | const MissingRegistrationEntity = Entity.make("MissingRegistrationEntity", [ |
| 3098 | Rpc.make("Call").annotate(ClusterSchema.Persisted, false) |
| 3099 | ]) |
| 3100 | |
| 3101 | const FirstRegistrationEntity = Entity.make("FirstRegistrationEntity", [ |
| 3102 | Rpc.make("Call").annotate(ClusterSchema.Persisted, false) |
| 3103 | ]) |
| 3104 | |
| 3105 | const RegistrationContextHandlers = Effect.map( |
| 3106 | RegistrationContext, |
| 3107 | (value) => RegistrationContextEntity.of({ Read: () => Effect.succeed(value) }) |
| 3108 | ) |
| 3109 | |
| 3110 | const testConfigDefaults: Partial<ShardingConfig.ShardingConfig["Service"]> = { |
| 3111 | entityMailboxCapacity: 10, |
| 3112 | entityTerminationTimeout: 0, |
| 3113 | entityMessagePollInterval: 5000, |
| 3114 | sendRetryInterval: 100, |
| 3115 | refreshAssignmentsInterval: 0 |
| 3116 | } |
| 3117 | |
| 3118 | const TestShardingConfig = ShardingConfig.layer(testConfigDefaults) |
| 3119 | |
| 3120 | const TestShardingWithoutRunnerStorage = TestEntityNoState.pipe( |
| 3121 | Layer.provideMerge(Sharding.layer), |
| 3122 | Layer.provide(RunnerHealth.layerNoop) |
| 3123 | // Layer.provide(Logger.minimumLogLevel(LogLevel.All)), |
| 3124 | // Layer.provideMerge(Logger.pretty) |
| 3125 | ) |
| 3126 | |
| 3127 | const TestShardingWithoutState = TestShardingWithoutRunnerStorage.pipe( |
| 3128 | Layer.provide(RunnerStorage.layerMemory) |
| 3129 | ) |
| 3130 | |
| 3131 | const TestShardingWithoutRunners = TestShardingWithoutState.pipe( |
| 3132 | Layer.provideMerge(TestEntityState.layer) |
| 3133 | ) |
| 3134 | |
| 3135 | const TestShardingWithoutStorage = TestShardingWithoutRunners.pipe( |
| 3136 | Layer.provide(Runners.layerNoop), |
| 3137 | Layer.provide(TestShardingConfig) |
| 3138 | ) |
| 3139 | |
| 3140 | const TestSharding = TestShardingWithoutStorage.pipe( |
| 3141 | Layer.provideMerge(MessageStorage.layerMemory), |
| 3142 | Layer.provide(TestShardingConfig) |
| 3143 | ) |
| 3144 | |
| 3145 | const CappedSharding = ( |
| 3146 | config: Partial<ShardingConfig.ShardingConfig["Service"]>, |
| 3147 | transformStorage?: ( |
| 3148 | storage: MessageStorage.MessageStorage["Service"] |
| 3149 | ) => MessageStorage.MessageStorage["Service"] |
| 3150 | ) => { |
| 3151 | const configLayer = ShardingConfig.layer({ ...testConfigDefaults, ...config }) |
| 3152 | let layer = TestShardingWithoutRunners.pipe( |
| 3153 | Layer.provide(Runners.layerNoop), |
| 3154 | Layer.provide(configLayer) |
| 3155 | ) |
| 3156 | if (transformStorage) { |
| 3157 | layer = layer.pipe(Layer.updateService(MessageStorage.MessageStorage, transformStorage)) |
| 3158 | } |
| 3159 | return layer.pipe( |
| 3160 | Layer.provideMerge(MessageStorage.layerMemory), |
| 3161 | Layer.provide(configLayer) |
| 3162 | ) |
| 3163 | } |
| 3164 | |
| 3165 | const UnregisteredSharding = ( |
| 3166 | config: Partial<ShardingConfig.ShardingConfig["Service"]> |
| 3167 | ) => { |
| 3168 | const configLayer = ShardingConfig.layer({ ...testConfigDefaults, ...config }) |
| 3169 | return Sharding.layer.pipe( |
| 3170 | Layer.provide(RunnerStorage.layerMemory), |
| 3171 | Layer.provide(RunnerHealth.layerNoop), |
| 3172 | Layer.provide(Runners.layerNoop), |
| 3173 | Layer.provideMerge(MessageStorage.layerMemory), |
| 3174 | Layer.provide(configLayer) |
| 3175 | ) |
| 3176 | } |
| 3177 | |
| 3178 | // The directly driven EntityManager waits up to 1000ms for its entities to end. |
| 3179 | const BlockedRebuildSharding = Layer.mergeAll( |
| 3180 | EntityReaper.layer, |
| 3181 | Snowflake.layerGenerator, |
| 3182 | ShardingConfig.layer({ ...testConfigDefaults, entityTerminationTimeout: 1000 }) |
| 3183 | ).pipe(Layer.provideMerge(UnregisteredSharding({}))) |
| 3184 | |
| 3185 | const ContextBleedSharding = ContextBleedLayer.pipe(Layer.provideMerge(TestSharding)) |
| 3186 | |
| 3187 | const ActiveTeardownSharding = ( |
| 3188 | config?: Partial<ShardingConfig.ShardingConfig["Service"]> |
| 3189 | ) => |
| 3190 | ActiveTeardownCallerLayer.pipe( |
| 3191 | Layer.merge(TestEntityNoState), |
| 3192 | Layer.provideMerge(Sharding.layer), |
| 3193 | Layer.provide(RunnerStorage.layerMemory), |
| 3194 | Layer.provide(RunnerHealth.layerNoop), |
| 3195 | Layer.provideMerge(TestEntityState.layer), |
| 3196 | Layer.provide(Runners.layerNoop), |
| 3197 | Layer.provideMerge(MessageStorage.layerMemory), |
| 3198 | Layer.provide(ShardingConfig.layer({ ...testConfigDefaults, ...config })) |
| 3199 | ) |
| 3200 | |
| 3201 | const ReapStormSharding = ReapStormEntityLayer.pipe( |
| 3202 | Layer.provideMerge(Sharding.layer), |
| 3203 | Layer.provide(RunnerStorage.layerMemory), |
| 3204 | Layer.provide(RunnerHealth.layerNoop), |
| 3205 | Layer.provide(Runners.layerNoop), |
| 3206 | Layer.provideMerge(MessageStorage.layerMemory), |
| 3207 | Layer.provide(ShardingConfig.layer({ |
| 3208 | ...testConfigDefaults, |
| 3209 | entityMaxIdleTime: 1, |
| 3210 | entityTerminationTimeout: 0 |
| 3211 | })) |
| 3212 | ) |
| 3213 | |
| 3214 | // saves a persisted GetUser request directly to storage, bypassing the client, |
| 3215 | // so the storage read loop only learns about it from the next poll |
| 3216 | const saveGetUserRequest = Effect.fnUntraced(function*(entityId: string, id: number) { |
| 3217 | const storage = yield* MessageStorage.MessageStorage |
| 3218 | const sharding = yield* Sharding.Sharding |
| 3219 | const rpc = TestEntity.protocol.requests.get("GetUser")! as any |
| 3220 | const entity = EntityId.make(entityId) |
| 3221 | yield* storage.saveRequest( |
| 3222 | new Message.OutgoingRequest({ |
| 3223 | envelope: Envelope.makeRequest<any>({ |
| 3224 | requestId: yield* sharding.getSnowflake, |
| 3225 | address: EntityAddress.make({ |
| 3226 | shardId: sharding.getShardId(entity, "default"), |
| 3227 | entityType: EntityType.make(TestEntity.type), |
| 3228 | entityId: entity |
| 3229 | }), |
| 3230 | tag: "GetUser", |
| 3231 | payload: { id }, |
| 3232 | headers: Headers.empty |
| 3233 | }), |
| 3234 | annotations: rpc.annotations, |
| 3235 | context: Context.empty() as any, |
| 3236 | rpc, |
| 3237 | lastReceivedReply: Option.none(), |
| 3238 | respond: () => Effect.void |
| 3239 | }) |
| 3240 | ) |
| 3241 | }) |
| 3242 | |
| 3243 | interface SingletonStorageState { |
| 3244 | assignSelf: boolean |
| 3245 | runner: Runner.Runner | undefined |
| 3246 | } |
| 3247 | |
| 3248 | const makeSingletonStorageState = (): SingletonStorageState => ({ assignSelf: true, runner: undefined }) |
| 3249 | |
| 3250 | const singletonOtherRunner = Runner.make({ |
| 3251 | address: RunnerAddress.make("localhost", 5678), |
| 3252 | groups: ["singleton"], |
| 3253 | // With these fixed addresses, the weighted ring moves singleton:1 to this |
| 3254 | // runner. The ownership assertions below guard that fixture assumption. |
| 3255 | weight: 1000 |
| 3256 | }) |
| 3257 | |
| 3258 | // Keep the destination shard local while only the singleton shard moves away. |
| 3259 | const SingletonReassignmentSharding = (state: SingletonStorageState) => { |
| 3260 | const runnerStorage = Layer.effect( |
| 3261 | RunnerStorage.RunnerStorage, |
| 3262 | Effect.succeed(RunnerStorage.RunnerStorage.of({ |
| 3263 | getRunners: Effect.sync(() => { |
| 3264 | if (!state.runner) return [] |
| 3265 | if (state.assignSelf) return [[state.runner, true]] as const |
| 3266 | return [ |
| 3267 | [state.runner, true], |
| 3268 | [singletonOtherRunner, true] |
| 3269 | ] as const |
| 3270 | }), |
| 3271 | register: (runner) => |
| 3272 | Effect.sync(() => { |
| 3273 | state.runner = runner |
| 3274 | return MachineId.make(1) |
| 3275 | }), |
| 3276 | unregister: () => Effect.void, |
| 3277 | setRunnerHealth: () => Effect.void, |
| 3278 | acquire: (_address, shards) => Effect.succeed(globalThis.Array.from(shards)), |
| 3279 | refresh: (_address, shards) => Effect.succeed(globalThis.Array.from(shards)), |
| 3280 | release: () => Effect.void, |
| 3281 | releaseAll: () => Effect.void |
| 3282 | })) |
| 3283 | ) |
| 3284 | return TestShardingWithoutRunnerStorage.pipe( |
| 3285 | Layer.provide(runnerStorage), |
| 3286 | Layer.provideMerge(TestEntityState.layer), |
| 3287 | Layer.provide(Runners.layerNoop), |
| 3288 | Layer.provideMerge(MessageStorage.layerMemory), |
| 3289 | Layer.provide(ShardingConfig.layer({ |
| 3290 | ...testConfigDefaults, |
| 3291 | runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), |
| 3292 | availableShardGroups: ["default", "singleton"], |
| 3293 | assignedShardGroups: ["default", "singleton"], |
| 3294 | shardsPerGroup: 1, |
| 3295 | entityTerminationTimeout: 0, |
| 3296 | entityMessagePollInterval: 10, |
| 3297 | refreshAssignmentsInterval: 10, |
| 3298 | sendRetryInterval: 10 |
| 3299 | })) |
| 3300 | ) |
| 3301 | } |
| 3302 | |
| 3303 | const singletonShard = ShardId.make("singleton", 1) |
| 3304 | const destinationShard = ShardId.make("default", 1) |
| 3305 | |
| 3306 | const waitForSingletonOwnership = Effect.fnUntraced(function*( |
| 3307 | sharding: Sharding.Sharding["Service"], |
| 3308 | owned: boolean |
| 3309 | ) { |
| 3310 | for (let i = 0; i < 100; i++) { |
| 3311 | if (sharding.hasShardId(singletonShard) === owned) return |
| 3312 | yield* TestClock.adjust(10) |
| 3313 | } |
| 3314 | assert.strictEqual(sharding.hasShardId(singletonShard), owned) |
| 3315 | }) |
| 3316 | |
| 3317 | // These tests share the internal teardown registry, so do not run concurrently. |
| 3318 | describe("Sharding singleton cancellation", { concurrent: false }, () => { |
| 3319 | it.effect("cancels explicitly but abandons a singleton child RPC on reassignment", () => |
| 3320 | Effect.gen(function*() { |
| 3321 | const storageState = makeSingletonStorageState() |
| 3322 | yield* Effect.gen(function*() { |
| 3323 | const sharding = yield* Sharding.Sharding |
| 3324 | const driver = yield* MessageStorage.MemoryDriver |
| 3325 | const state = yield* TestEntityState |
| 3326 | // Acquire clients outside the singleton context; children must use caller identity. |
| 3327 | const makeClient = yield* TestEntity.client |
| 3328 | const explicitClient = makeClient("explicit-singleton-target") |
| 3329 | const reassignmentClient = makeClient("singleton-target") |
| 3330 | const childReady = yield* Deferred.make<Fiber.Fiber<void, unknown>>() |
| 3331 | const runReassignment = yield* Deferred.make<void>() |
| 3332 | const stopped = yield* Deferred.make<void>() |
| 3333 | yield* waitForSingletonOwnership(sharding, true) |
| 3334 | yield* sharding.registerSingleton( |
| 3335 | "rpc-caller", |
| 3336 | Effect.gen(function*() { |
| 3337 | const child = yield* explicitClient.Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 3338 | yield* Deferred.succeed(childReady, child) |
| 3339 | yield* Deferred.await(runReassignment) |
| 3340 | yield* reassignmentClient.Never().pipe(Effect.forkChild({ startImmediately: true })) |
| 3341 | return yield* Effect.never |
| 3342 | }).pipe(Effect.ensuring(Deferred.succeed(stopped, void 0))), |
| 3343 | { shardGroup: "singleton" } |
| 3344 | ) |
| 3345 | yield* Queue.take(state.envelopes) |
| 3346 | yield* Fiber.interrupt(yield* Deferred.await(childReady)) |
| 3347 | yield* TestClock.adjust(1) |
| 3348 | assert.isTrue(sharding.hasShardId(singletonShard)) |
| 3349 | assert.strictEqual(journalInterrupts(driver), 1, "explicit child cancellation must persist") |
| 3350 | yield* Queue.take(state.interrupts) |
| 3351 | |
| 3352 | yield* Deferred.succeed(runReassignment, void 0) |
| 3353 | yield* Queue.take(state.envelopes) |
| 3354 | storageState.assignSelf = false |
| 3355 | yield* waitForSingletonOwnership(sharding, false) |
| 3356 | yield* Deferred.await(stopped) |
| 3357 | yield* TestClock.adjust(1) |
| 3358 | assert.isTrue(sharding.hasShardId(destinationShard), "Sharding and the destination stay alive") |
| 3359 | assert.strictEqual(journalInterrupts(driver), 1, "reassignment must not add a durable cancellation") |
| 3360 | assert.strictEqual(Queue.sizeUnsafe(state.interrupts), 0) |
| 3361 | }).pipe(Effect.provide(SingletonReassignmentSharding(storageState)), Effect.scoped) |
| 3362 | })) |
| 3363 | |
| 3364 | it.effect("bounds singleton teardown to interruption finalization, not inherited context", () => |
| 3365 | Effect.gen(function*() { |
| 3366 | const storageState = makeSingletonStorageState() |
| 3367 | yield* Effect.gen(function*() { |
| 3368 | const sharding = yield* Sharding.Sharding |
| 3369 | const driver = yield* MessageStorage.MemoryDriver |
| 3370 | const state = yield* TestEntityState |
| 3371 | const client = (yield* TestEntity.client)("finalizer-target") |
| 3372 | // Both clients are acquired before singleton execution. Keep the survivor |
| 3373 | // on a separate local entity so the preserved finalizer RPC cannot block it. |
| 3374 | const survivorClient = (yield* TestEntity.client)("survivor-target") |
| 3375 | const outerScope = yield* Effect.scope |
| 3376 | const runLater = yield* Deferred.make<void>() |
| 3377 | const survivorReady = yield* Deferred.make<Fiber.Fiber<void, unknown>>() |
| 3378 | const finalizing = yield* Deferred.make<void>() |
| 3379 | const finish = yield* Deferred.make<void>() |
| 3380 | const stopped = yield* Deferred.make<void>() |
| 3381 | yield* waitForSingletonOwnership(sharding, true) |
| 3382 | yield* sharding.registerSingleton( |
| 3383 | "finalizer-caller", |
| 3384 | Effect.gen(function*() { |
| 3385 | // This fiber inherits singleton identity but outlives its local run. |
| 3386 | const survivor = yield* Deferred.await(runLater).pipe( |
| 3387 | Effect.andThen(survivorClient.Never()), |
| 3388 | Effect.forkIn(outerScope, { startImmediately: true }) |
| 3389 | ) |
| 3390 | yield* Deferred.succeed(survivorReady, survivor) |
| 3391 | return yield* Effect.never |
| 3392 | }).pipe( |
| 3393 | Effect.ensuring(Effect.gen(function*() { |
| 3394 | // Cancellation during finalization is still shard teardown. |
| 3395 | const child = yield* client.Never().pipe( |
| 3396 | Effect.interruptible, |
| 3397 | Effect.forkChild({ startImmediately: true }) |
| 3398 | ) |
| 3399 | yield* Queue.take(state.envelopes) |
| 3400 | yield* Fiber.interrupt(child) |
| 3401 | yield* Deferred.succeed(finalizing, void 0) |
| 3402 | yield* Deferred.await(finish) |
| 3403 | })), |
| 3404 | Effect.ensuring(Deferred.succeed(stopped, void 0)) |
| 3405 | ), |
| 3406 | { shardGroup: "singleton" } |
| 3407 | ) |
| 3408 | const survivor = yield* Deferred.await(survivorReady) |
| 3409 | // Release the finalizer barrier even if a setup assertion fails. |
| 3410 | yield* Effect.addFinalizer(() => Deferred.succeed(finish, void 0)) |
| 3411 | storageState.assignSelf = false |
| 3412 | yield* waitForSingletonOwnership(sharding, false) |
| 3413 | yield* Deferred.await(finalizing) |
| 3414 | const duringFinalization = journalInterrupts(driver) |
| 3415 | const probe = EntityAddress.make({ |
| 3416 | shardId: singletonShard, |
| 3417 | entityType: EntityType.make("singleton-teardown-probe"), |
| 3418 | entityId: EntityId.make("probe") |
| 3419 | }) |
| 3420 | assert.isTrue(ActiveTeardown.isActive(probe), "tracking must remain active through finalization") |
| 3421 | yield* Deferred.succeed(finish, void 0) |
| 3422 | yield* Deferred.await(stopped) |
| 3423 | yield* TestClock.adjust(10) |
| 3424 | assert.isFalse(ActiveTeardown.isActive(probe), "teardown must release its shard tracking") |
| 3425 | |
| 3426 | // Do not let inherited identity suppress a later explicit cancellation. |
| 3427 | yield* Deferred.succeed(runLater, void 0) |
| 3428 | yield* Queue.take(state.envelopes) |
| 3429 | yield* Fiber.interrupt(survivor) |
| 3430 | yield* TestClock.adjust(1) |
| 3431 | assert.strictEqual(journalInterrupts(driver) - duringFinalization, 1) |
| 3432 | assert.isTrue(sharding.hasShardId(destinationShard)) |
| 3433 | assert.strictEqual(duringFinalization, 0, "teardown must include singleton finalizers") |
| 3434 | }).pipe(Effect.provide(SingletonReassignmentSharding(storageState)), Effect.scoped) |
| 3435 | })) |
| 3436 | }) |