| 1 | import { assert, describe, it } from "@effect/vitest" |
| 2 | import { Cause, Clock, Effect, Exit, Fiber, Latch, Layer, PrimaryKey, Schema, Scope } from "effect" |
| 3 | import { ClusterSchema, Entity, EntityResource, Singleton } from "effect/cluster" |
| 4 | import { Rpc } from "effect/rpc" |
| 5 | import { type Backend, type ClusterRunner, make } from "./harness.ts" |
| 6 | |
| 7 | class Request extends Schema.Class<Request>("ClusterEntityRequest")({ |
| 8 | id: Schema.String, |
| 9 | sequence: Schema.Number |
| 10 | }) { |
| 11 | [PrimaryKey.symbol]() { |
| 12 | return this.id |
| 13 | } |
| 14 | } |
| 15 | |
| 16 | const StateReply = Schema.Struct({ |
| 17 | generation: Schema.Number, |
| 18 | runner: Schema.String, |
| 19 | value: Schema.Number |
| 20 | }) |
| 21 | |
| 22 | const StateEntity = Entity.make("ClusterIntegrationState", [ |
| 23 | Rpc.make("Increment", { |
| 24 | payload: Request, |
| 25 | success: StateReply |
| 26 | }), |
| 27 | Rpc.make("Ordered", { |
| 28 | payload: Request, |
| 29 | success: Schema.Number |
| 30 | }) |
| 31 | ]).annotateRpcs(ClusterSchema.Persisted, true) |
| 32 | |
| 33 | let orderGate = Latch.makeUnsafe(true) |
| 34 | let orderEntered = Latch.makeUnsafe() |
| 35 | let order: Array<number> = [] |
| 36 | const generations = new Map<string, number>() |
| 37 | |
| 38 | const addressString = (address: { readonly host: string; readonly port: number }) => `${address.host}:${address.port}` |
| 39 | |
| 40 | const StateEntityLayer = StateEntity.toLayer( |
| 41 | Effect.gen(function*() { |
| 42 | const address = yield* Entity.CurrentAddress |
| 43 | const runner = yield* Entity.CurrentRunnerAddress |
| 44 | const entityId = String(address.entityId) |
| 45 | const generation = (generations.get(entityId) ?? 0) + 1 |
| 46 | generations.set(entityId, generation) |
| 47 | let value = 0 |
| 48 | return { |
| 49 | Increment: () => |
| 50 | Effect.sync(() => ({ |
| 51 | generation, |
| 52 | runner: addressString(runner), |
| 53 | value: ++value |
| 54 | })), |
| 55 | Ordered: Effect.fnUntraced(function*({ payload }) { |
| 56 | order.push(payload.sequence) |
| 57 | if (payload.sequence === 1) { |
| 58 | orderEntered.openUnsafe() |
| 59 | yield* orderGate.await |
| 60 | } |
| 61 | return payload.sequence |
| 62 | }) |
| 63 | } |
| 64 | }), |
| 65 | { maxIdleTime: "1 second" } |
| 66 | ) |
| 67 | |
| 68 | const MailboxEntity = Entity.make("ClusterIntegrationMailbox", [ |
| 69 | Rpc.make("Hold", { |
| 70 | payload: Request, |
| 71 | success: Schema.Number |
| 72 | }) |
| 73 | ]) |
| 74 | |
| 75 | let mailboxGate = Latch.makeUnsafe(true) |
| 76 | let mailboxEntered = Latch.makeUnsafe() |
| 77 | |
| 78 | const MailboxEntityLayer = MailboxEntity.toLayer({ |
| 79 | Hold: Effect.fnUntraced(function*({ payload }) { |
| 80 | mailboxEntered.openUnsafe() |
| 81 | yield* mailboxGate.await |
| 82 | return payload.sequence |
| 83 | }) |
| 84 | }, { mailboxCapacity: 1 }) |
| 85 | |
| 86 | const GroupEntity = Entity.make("ClusterIntegrationSpecialGroup", [ |
| 87 | Rpc.make("Runner", { success: Schema.String }) |
| 88 | ]).annotate(ClusterSchema.ShardGroup, () => "special") |
| 89 | |
| 90 | const GroupEntityLayer = GroupEntity.toLayer(Effect.gen(function*() { |
| 91 | const runner = yield* Entity.CurrentRunnerAddress |
| 92 | return { Runner: () => Effect.succeed(addressString(runner)) } |
| 93 | })) |
| 94 | |
| 95 | const ResourceEntity = Entity.make("ClusterIntegrationResource", [ |
| 96 | Rpc.make("Get", { success: Schema.Number }), |
| 97 | Rpc.make("Close", { success: Schema.Void }) |
| 98 | ]) |
| 99 | |
| 100 | const resourceState = { acquired: 0, released: 0 } |
| 101 | |
| 102 | const ResourceEntityLayer = ResourceEntity.toLayer(Effect.gen(function*() { |
| 103 | const resource = yield* EntityResource.make({ |
| 104 | acquire: Effect.gen(function*() { |
| 105 | const closeScope = yield* EntityResource.CloseScope |
| 106 | return yield* Effect.acquireRelease( |
| 107 | Effect.sync(() => ++resourceState.acquired), |
| 108 | () => Effect.sync(() => resourceState.released++) |
| 109 | ).pipe(Scope.provide(closeScope)) |
| 110 | }) |
| 111 | }) |
| 112 | return { |
| 113 | Close: () => resource.close, |
| 114 | Get: () => Effect.scoped(resource.get) |
| 115 | } |
| 116 | })) |
| 117 | |
| 118 | const RegistrationEntity = Entity.make("ClusterIntegrationRegistration", [ |
| 119 | Rpc.make("Runner", { success: Schema.String }).annotate(ClusterSchema.Persisted, true) |
| 120 | ]) |
| 121 | |
| 122 | const RegistrationEntityLayer = RegistrationEntity.toLayer(Effect.gen(function*() { |
| 123 | const runner = yield* Entity.CurrentRunnerAddress |
| 124 | return { Runner: () => Effect.succeed(addressString(runner)) } |
| 125 | })) |
| 126 | |
| 127 | const StandardEntities = Layer.mergeAll(StateEntityLayer, MailboxEntityLayer, ResourceEntityLayer) |
| 128 | |
| 129 | const resetOrder = () => { |
| 130 | orderGate = Latch.makeUnsafe() |
| 131 | orderEntered = Latch.makeUnsafe() |
| 132 | order = [] |
| 133 | } |
| 134 | |
| 135 | const resetMailbox = () => { |
| 136 | mailboxGate = Latch.makeUnsafe() |
| 137 | mailboxEntered = Latch.makeUnsafe() |
| 138 | } |
| 139 | |
| 140 | const findIdsByRunner = Effect.fnUntraced(function*( |
| 141 | cluster: Effect.Success<ReturnType<typeof make>>, |
| 142 | runners: ReadonlyArray<ClusterRunner> |
| 143 | ) { |
| 144 | const found = new Map<ClusterRunner, string>() |
| 145 | for (let index = 0; index < 2_000 && found.size < runners.length; index++) { |
| 146 | const id = `entity-${index}` |
| 147 | const owner = yield* cluster.ownerOfEntity(StateEntity, id) |
| 148 | if (owner && runners.includes(owner) && !found.has(owner)) found.set(owner, id) |
| 149 | } |
| 150 | assert.strictEqual(found.size, runners.length) |
| 151 | return found |
| 152 | }) |
| 153 | |
| 154 | describe("cluster entity integration", () => { |
| 155 | for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) { |
| 156 | it.live(`${backend}: interrupts an unroutable reply during runner shutdown without hanging teardown`, () => |
| 157 | Effect.gen(function*() { |
| 158 | const finalizerEntered = Latch.makeUnsafe() |
| 159 | const sendReply = Latch.makeUnsafe() |
| 160 | let requestExit: Exit.Exit<void, unknown> | undefined |
| 161 | |
| 162 | const Receiver = Entity.make("ClusterIntegrationShutdownReplyReceiver", [ |
| 163 | Rpc.make("Ping").annotate(ClusterSchema.Persisted, true) |
| 164 | ]) |
| 165 | const Sender = Entity.make("ClusterIntegrationShutdownReplySender", [ |
| 166 | Rpc.make("Arm").annotate(ClusterSchema.Persisted, false) |
| 167 | ]) |
| 168 | const entities = Layer.merge( |
| 169 | Receiver.toLayer({ Ping: () => Effect.void }), |
| 170 | Sender.toLayer(Effect.gen(function*() { |
| 171 | const receiver = yield* Receiver.client |
| 172 | yield* Effect.addFinalizer(() => |
| 173 | Effect.uninterruptible(Effect.gen(function*() { |
| 174 | finalizerEntered.openUnsafe() |
| 175 | yield* sendReply.await |
| 176 | requestExit = yield* receiver("unroutable").Ping().pipe(Effect.exit) |
| 177 | })) |
| 178 | ) |
| 179 | return { Arm: () => Effect.void } |
| 180 | })) |
| 181 | ) |
| 182 | const cluster = yield* make({ backend, entities }) |
| 183 | const [runner] = yield* cluster.start(1, { assignedShardGroups: ["default"] }) |
| 184 | yield* cluster.waitForStableAssignments() |
| 185 | const sender = yield* cluster.getClient(Sender) |
| 186 | yield* sender("sender").Arm() |
| 187 | |
| 188 | const stopping = yield* cluster.stop(runner).pipe(Effect.forkChild({ startImmediately: true })) |
| 189 | yield* cluster.waitUntil( |
| 190 | "The shutdown reply finalizer did not start", |
| 191 | Effect.as(finalizerEntered.await, true) |
| 192 | ) |
| 193 | sendReply.openUnsafe() |
| 194 | yield* Fiber.join(stopping) |
| 195 | |
| 196 | // The request is durable, so the caller is interrupted rather than |
| 197 | // failed: the send is not an error, this runner just cannot observe |
| 198 | // the reply. |
| 199 | assert.isDefined(requestExit) |
| 200 | assert(Exit.isFailure(requestExit!)) |
| 201 | assert(Cause.hasInterruptsOnly(requestExit!.cause)) |
| 202 | })) |
| 203 | |
| 204 | for (const persisted of [false, true] as const) { |
| 205 | for (const preemptiveShutdown of [false, true] as const) { |
| 206 | it.live( |
| 207 | `${backend}: sends an outgoing discard from a shutdown finalizer without deadlocking (persisted=${persisted}, preemptive=${preemptiveShutdown})`, |
| 208 | () => |
| 209 | Effect.gen(function*() { |
| 210 | const finalizerEntered = Latch.makeUnsafe() |
| 211 | const sendDiscard = Latch.makeUnsafe() |
| 212 | let discardExit: Exit.Exit<void, unknown> | undefined |
| 213 | |
| 214 | const Receiver = Entity.make("ClusterIntegrationShutdownDiscardReceiver", [ |
| 215 | Rpc.make("Ping").annotate(ClusterSchema.Persisted, persisted) |
| 216 | ]) |
| 217 | const Sender = Entity.make("ClusterIntegrationShutdownDiscardSender", [ |
| 218 | Rpc.make("Arm").annotate(ClusterSchema.Persisted, false) |
| 219 | ]) |
| 220 | const entities = Layer.merge( |
| 221 | Receiver.toLayer({ Ping: () => Effect.void }), |
| 222 | Sender.toLayer(Effect.gen(function*() { |
| 223 | const receiver = yield* Receiver.client |
| 224 | yield* Effect.addFinalizer(() => |
| 225 | Effect.uninterruptible(Effect.gen(function*() { |
| 226 | finalizerEntered.openUnsafe() |
| 227 | yield* sendDiscard.await |
| 228 | discardExit = yield* receiver("unroutable").Ping(void 0, { discard: true }).pipe(Effect.exit) |
| 229 | })) |
| 230 | ) |
| 231 | return { Arm: () => Effect.void } |
| 232 | })) |
| 233 | ) |
| 234 | const cluster = yield* make({ |
| 235 | backend, |
| 236 | config: { preemptiveShutdown }, |
| 237 | entities |
| 238 | }) |
| 239 | const [runner] = yield* cluster.start(1, { assignedShardGroups: ["default"] }) |
| 240 | yield* cluster.waitForStableAssignments() |
| 241 | const sender = yield* cluster.getClient(Sender) |
| 242 | yield* sender("sender").Arm() |
| 243 | |
| 244 | const stopping = yield* cluster.stop(runner).pipe(Effect.forkChild({ startImmediately: true })) |
| 245 | yield* cluster.waitUntil( |
| 246 | "The shutdown discard finalizer did not start", |
| 247 | Effect.as(finalizerEntered.await, true) |
| 248 | ) |
| 249 | sendDiscard.openUnsafe() |
| 250 | yield* Fiber.join(stopping) |
| 251 | |
| 252 | assert.isDefined(discardExit) |
| 253 | assert(Exit.isSuccess(discardExit!)) |
| 254 | }) |
| 255 | ) |
| 256 | } |
| 257 | } |
| 258 | |
| 259 | it.live(`${backend}: rebuilds an entity after a fatal defect and replays in-flight requests`, () => |
| 260 | Effect.gen(function*() { |
| 261 | const firstAttemptEntered = Latch.makeUnsafe() |
| 262 | const replayEntered = Latch.makeUnsafe() |
| 263 | const replayGate = Latch.makeUnsafe() |
| 264 | const replayCompleted = Latch.makeUnsafe() |
| 265 | let builds = 0 |
| 266 | let replayAttempts = 0 |
| 267 | |
| 268 | const ReplayEntity = Entity.make("ClusterIntegrationFatalReplay", [ |
| 269 | Rpc.make("Hold").annotate(ClusterSchema.Persisted, true), |
| 270 | Rpc.make("Crash", { success: Schema.Number }).annotate(ClusterSchema.Persisted, true) |
| 271 | ]) |
| 272 | const ReplayEntityLayer = ReplayEntity.toLayer(Effect.sync(() => { |
| 273 | const generation = ++builds |
| 274 | return ReplayEntity.of({ |
| 275 | Crash: () => generation === 1 ? Effect.die("fatal replay defect") : Effect.succeed(generation), |
| 276 | Hold: () => |
| 277 | Rpc.fork(Effect.gen(function*() { |
| 278 | replayAttempts++ |
| 279 | if (replayAttempts === 1) { |
| 280 | firstAttemptEntered.openUnsafe() |
| 281 | } else { |
| 282 | replayEntered.openUnsafe() |
| 283 | } |
| 284 | yield* replayGate.await |
| 285 | replayCompleted.openUnsafe() |
| 286 | })) |
| 287 | }) |
| 288 | })) |
| 289 | const cluster = yield* make({ backend, entities: ReplayEntityLayer }) |
| 290 | yield* cluster.start(1) |
| 291 | yield* cluster.waitForStableAssignments() |
| 292 | const client = yield* cluster.getClient(ReplayEntity) |
| 293 | |
| 294 | const held = yield* client("replay").Hold().pipe(Effect.forkChild({ startImmediately: true })) |
| 295 | yield* cluster.waitUntil( |
| 296 | "The in-flight request did not start before the defect", |
| 297 | Effect.as(firstAttemptEntered.await, true) |
| 298 | ) |
| 299 | const generation = yield* client("replay").Crash() |
| 300 | yield* cluster.waitUntil( |
| 301 | "The in-flight request was not replayed after the defect", |
| 302 | Effect.as(replayEntered.await, true) |
| 303 | ) |
| 304 | replayGate.openUnsafe() |
| 305 | yield* cluster.waitUntil( |
| 306 | "The replayed request did not complete", |
| 307 | Effect.as(replayCompleted.await, true) |
| 308 | ) |
| 309 | yield* Fiber.join(held) |
| 310 | |
| 311 | assert.strictEqual(generation, 2) |
| 312 | assert.strictEqual(builds, 2) |
| 313 | assert.strictEqual(replayAttempts, 2) |
| 314 | })) |
| 315 | |
| 316 | it.live(`${backend}: force-interrupts a stuck handler after the graceful termination timeout`, () => |
| 317 | Effect.gen(function*() { |
| 318 | const handlerEntered = Latch.makeUnsafe() |
| 319 | const handlerInterrupted = Latch.makeUnsafe() |
| 320 | let attempts = 0 |
| 321 | let block = true |
| 322 | |
| 323 | const StuckEntity = Entity.make("ClusterIntegrationStuckReassignment", [ |
| 324 | Rpc.make("Run", { success: Schema.Number }).annotate(ClusterSchema.Persisted, false) |
| 325 | ]) |
| 326 | const StuckEntityLayer = StuckEntity.toLayer({ |
| 327 | Run: () => |
| 328 | Effect.suspend(() => { |
| 329 | const attempt = ++attempts |
| 330 | if (!block) return Effect.succeed(attempt) |
| 331 | handlerEntered.openUnsafe() |
| 332 | return Effect.never.pipe(Effect.onInterrupt(() => handlerInterrupted.open)) |
| 333 | }) |
| 334 | }) |
| 335 | const cluster = yield* make({ |
| 336 | backend, |
| 337 | config: { |
| 338 | entityTerminationTimeout: 100, |
| 339 | preemptiveShutdown: false |
| 340 | }, |
| 341 | entities: StuckEntityLayer |
| 342 | }) |
| 343 | yield* cluster.start(2) |
| 344 | yield* cluster.waitForStableAssignments() |
| 345 | const owner = yield* cluster.ownerOfEntity(StuckEntity, "stuck") |
| 346 | const client = yield* cluster.getClient(StuckEntity) |
| 347 | const request = yield* client("stuck").Run().pipe(Effect.forkChild({ startImmediately: true })) |
| 348 | yield* cluster.waitUntil( |
| 349 | "The stuck handler did not start", |
| 350 | Effect.as(handlerEntered.await, true) |
| 351 | ) |
| 352 | block = false |
| 353 | |
| 354 | const stopping = yield* cluster.stop(owner!).pipe(Effect.forkChild({ startImmediately: true })) |
| 355 | yield* cluster.waitUntil( |
| 356 | "The stuck handler was not force-interrupted after the graceful timeout", |
| 357 | Effect.as(handlerInterrupted.await, true) |
| 358 | ) |
| 359 | yield* Fiber.join(stopping) |
| 360 | yield* cluster.waitUntil( |
| 361 | "The stopped runner did not reassign the stuck entity", |
| 362 | Effect.map(cluster.ownerOfEntity(StuckEntity, "stuck"), (next) => next !== undefined && next !== owner) |
| 363 | ) |
| 364 | |
| 365 | assert.strictEqual(yield* Fiber.join(request), 2) |
| 366 | assert.strictEqual(attempts, 2) |
| 367 | })) |
| 368 | |
| 369 | it.live(`${backend}: holds persisted messages until a slow entity layer registers`, () => |
| 370 | Effect.gen(function*() { |
| 371 | const registrationGate = Latch.makeUnsafe() |
| 372 | const delayedEntities = RegistrationEntityLayer.pipe( |
| 373 | Layer.provide(Layer.effectDiscard(registrationGate.await)) |
| 374 | ) |
| 375 | const cluster = yield* make({ |
| 376 | backend, |
| 377 | config: { |
| 378 | entityMessagePollInterval: 100, |
| 379 | entityRegistrationTimeout: 1_000, |
| 380 | sendRetryInterval: 100, |
| 381 | shardsPerGroup: 4 |
| 382 | }, |
| 383 | entities: RegistrationEntityLayer |
| 384 | }) |
| 385 | yield* cluster.start(1, { assignedShardGroups: [] }) |
| 386 | const starting = yield* cluster.start(1, { entities: delayedEntities }).pipe( |
| 387 | Effect.forkChild({ startImmediately: true }) |
| 388 | ) |
| 389 | const client = yield* cluster.getClient(RegistrationEntity) |
| 390 | const request = yield* client(`${backend}-slow-registration`).Runner().pipe( |
| 391 | Effect.forkChild({ startImmediately: true }) |
| 392 | ) |
| 393 | |
| 394 | yield* cluster.waitUntil( |
| 395 | "The slow-registration request was not persisted", |
| 396 | Effect.map(cluster.unprocessedMessageCount, (count) => count === 1) |
| 397 | ) |
| 398 | const deadline = (yield* Clock.currentTimeMillis) + 1_500 |
| 399 | yield* cluster.waitUntil( |
| 400 | "The slow-registration hold window did not elapse", |
| 401 | Effect.map(Clock.currentTimeMillis, (now) => now >= deadline), |
| 402 | "2 seconds" |
| 403 | ) |
| 404 | assert.deepStrictEqual(yield* cluster.messageCounts(), { |
| 405 | failed: 0, |
| 406 | replied: 0, |
| 407 | unprocessed: 1 |
| 408 | }) |
| 409 | |
| 410 | registrationGate.openUnsafe() |
| 411 | const [runner] = yield* Fiber.join(starting) |
| 412 | yield* cluster.waitForEntityOwner(RegistrationEntity, `${backend}-slow-registration`, runner) |
| 413 | assert.strictEqual(yield* Fiber.join(request), addressString(runner.address)) |
| 414 | assert.deepStrictEqual(yield* cluster.messageCounts(), { |
| 415 | failed: 0, |
| 416 | replied: 1, |
| 417 | unprocessed: 0 |
| 418 | }) |
| 419 | })) |
| 420 | |
| 421 | it.live(`${backend}: defects a durable request whose owner never registers the entity`, () => |
| 422 | Effect.gen(function*() { |
| 423 | const cluster = yield* make({ |
| 424 | backend, |
| 425 | config: { |
| 426 | entityMessagePollInterval: 100, |
| 427 | entityRegistrationTimeout: 500, |
| 428 | sendRetryInterval: 100, |
| 429 | shardsPerGroup: 4 |
| 430 | }, |
| 431 | entities: RegistrationEntityLayer |
| 432 | }) |
| 433 | yield* cluster.start(1, { assignedShardGroups: [] }) |
| 434 | const [runner] = yield* cluster.start(1, { entities: Layer.empty }) |
| 435 | const entityId = `${backend}-missing-registration` |
| 436 | yield* cluster.waitForEntityOwner(RegistrationEntity, entityId, runner) |
| 437 | const client = yield* cluster.getClient(RegistrationEntity) |
| 438 | const request = yield* client(entityId).Runner().pipe( |
| 439 | Effect.forkChild({ startImmediately: true }) |
| 440 | ) |
| 441 | |
| 442 | yield* cluster.waitUntil( |
| 443 | "The missing entity registration did not produce a stored defect", |
| 444 | Effect.map(cluster.failedMessageCount, (count) => count === 1) |
| 445 | ) |
| 446 | const exit = yield* Fiber.await(request) |
| 447 | assert.isTrue(Exit.isFailure(exit)) |
| 448 | if (Exit.isFailure(exit)) { |
| 449 | assert.include(Cause.pretty(exit.cause), `Entity type '${RegistrationEntity.type}' not registered`) |
| 450 | } |
| 451 | assert.deepStrictEqual(yield* cluster.messageCounts(), { |
| 452 | failed: 1, |
| 453 | replied: 0, |
| 454 | unprocessed: 0 |
| 455 | }) |
| 456 | })) |
| 457 | |
| 458 | it.live(`${backend}: routes by entity id, isolates state, and preserves mailbox order`, () => |
| 459 | Effect.gen(function*() { |
| 460 | generations.clear() |
| 461 | resetOrder() |
| 462 | const cluster = yield* make({ backend, entities: StandardEntities }) |
| 463 | const runners = yield* cluster.start(3) |
| 464 | yield* cluster.waitForStableAssignments() |
| 465 | const ids = yield* findIdsByRunner(cluster, runners) |
| 466 | const client = yield* cluster.getClient(StateEntity) |
| 467 | |
| 468 | for (const [runner, id] of ids) { |
| 469 | const first = yield* client(id).Increment(new Request({ id: `${id}-1`, sequence: 0 })) |
| 470 | const second = yield* client(id).Increment(new Request({ id: `${id}-2`, sequence: 0 })) |
| 471 | assert.strictEqual(first.runner, addressString(runner.address)) |
| 472 | assert.strictEqual(second.runner, first.runner) |
| 473 | assert.strictEqual(first.value, 1) |
| 474 | assert.strictEqual(second.value, 2) |
| 475 | assert.isFalse(cluster.clientSharding.hasShardId(yield* cluster.shardOfEntity(StateEntity, id))) |
| 476 | } |
| 477 | |
| 478 | const orderedId = ids.values().next().value! |
| 479 | const first = yield* client(orderedId).Ordered( |
| 480 | new Request({ id: `${backend}-ordered-1`, sequence: 1 }) |
| 481 | ).pipe(Effect.forkChild({ startImmediately: true })) |
| 482 | yield* cluster.waitUntil("The first ordered request did not start", Effect.as(orderEntered.await, true)) |
| 483 | const second = yield* client(orderedId).Ordered( |
| 484 | new Request({ id: `${backend}-ordered-2`, sequence: 2 }) |
| 485 | ).pipe(Effect.forkChild({ startImmediately: true })) |
| 486 | orderGate.openUnsafe() |
| 487 | assert.strictEqual(yield* Fiber.join(first), 1) |
| 488 | assert.strictEqual(yield* Fiber.join(second), 2) |
| 489 | assert.deepStrictEqual(order, [1, 2]) |
| 490 | |
| 491 | const registrations = (yield* cluster.diagnostics()).registrations |
| 492 | assert.strictEqual(registrations.length, runners.length) |
| 493 | })) |
| 494 | |
| 495 | it.live(`${backend}: reports mailbox saturation and revives idle entities with fresh state`, () => |
| 496 | Effect.gen(function*() { |
| 497 | generations.clear() |
| 498 | resetMailbox() |
| 499 | const cluster = yield* make({ backend, entities: StandardEntities }) |
| 500 | yield* cluster.start(2) |
| 501 | yield* cluster.waitForStableAssignments() |
| 502 | |
| 503 | const mailbox = yield* cluster.getClient(MailboxEntity) |
| 504 | const held = yield* mailbox("full").Hold( |
| 505 | new Request({ id: `${backend}-held`, sequence: 1 }) |
| 506 | ).pipe(Effect.forkChild({ startImmediately: true })) |
| 507 | yield* cluster.waitUntil("The mailbox request did not start", Effect.as(mailboxEntered.await, true)) |
| 508 | const error = yield* mailbox("full").Hold( |
| 509 | new Request({ id: `${backend}-rejected`, sequence: 2 }) |
| 510 | ).pipe(Effect.flip) |
| 511 | assert.strictEqual(error._tag, "MailboxFull") |
| 512 | mailboxGate.openUnsafe() |
| 513 | assert.strictEqual(yield* Fiber.join(held), 1) |
| 514 | |
| 515 | const state = yield* cluster.getClient(StateEntity) |
| 516 | const first = yield* state("idle").Increment( |
| 517 | new Request({ id: `${backend}-idle-1`, sequence: 0 }) |
| 518 | ) |
| 519 | const owner = yield* cluster.ownerOfEntity(StateEntity, "idle") |
| 520 | yield* cluster.waitUntil( |
| 521 | "The idle entity was not reaped", |
| 522 | Effect.map(owner!.sharding.activeEntityCount, (count) => count === 0), |
| 523 | "12 seconds" |
| 524 | ) |
| 525 | const revived = yield* state("idle").Increment( |
| 526 | new Request({ id: `${backend}-idle-2`, sequence: 0 }) |
| 527 | ) |
| 528 | assert.strictEqual(first.generation, 1) |
| 529 | assert.strictEqual(revived.generation, 2) |
| 530 | assert.strictEqual(revived.value, 1) |
| 531 | })) |
| 532 | |
| 533 | it.live(`${backend}: rebalances on runner addition, graceful stop, and abrupt death`, () => |
| 534 | Effect.gen(function*() { |
| 535 | resetOrder() |
| 536 | const cluster = yield* make({ backend, entities: StandardEntities }) |
| 537 | const initial = yield* cluster.start(2) |
| 538 | yield* cluster.waitForStableAssignments() |
| 539 | const before = new Map<string, ClusterRunner>() |
| 540 | for (let index = 0; index < 2_000; index++) { |
| 541 | const id = `moving-${index}` |
| 542 | before.set(id, (yield* cluster.ownerOfEntity(StateEntity, id))!) |
| 543 | } |
| 544 | |
| 545 | const [added] = yield* cluster.start(1) |
| 546 | assert.strictEqual(added.index, 2) |
| 547 | yield* cluster.waitForStableAssignments() |
| 548 | let movedId: string | undefined |
| 549 | for (const [id, old] of before) { |
| 550 | if (old !== added && (yield* cluster.ownerOfEntity(StateEntity, id)) === added) { |
| 551 | movedId = id |
| 552 | break |
| 553 | } |
| 554 | } |
| 555 | assert.isDefined(movedId) |
| 556 | const client = yield* cluster.getClient(StateEntity) |
| 557 | const moved = yield* client(movedId!).Increment( |
| 558 | new Request({ id: `${backend}-moved`, sequence: 0 }) |
| 559 | ) |
| 560 | assert.strictEqual(moved.runner, addressString(added.address)) |
| 561 | |
| 562 | const stopId = (yield* findIdsByRunner(cluster, initial)).get(initial[0])! |
| 563 | const request = yield* client(stopId).Ordered( |
| 564 | new Request({ id: `${backend}-stop`, sequence: 1 }) |
| 565 | ).pipe(Effect.forkChild({ startImmediately: true })) |
| 566 | yield* cluster.waitUntil("The handover request did not start", Effect.as(orderEntered.await, true)) |
| 567 | const stopping = yield* cluster.stop(initial[0]).pipe(Effect.forkChild({ startImmediately: true })) |
| 568 | yield* cluster.waitUntil( |
| 569 | "The stopped runner did not hand over its entity", |
| 570 | Effect.map(cluster.ownerOfEntity(StateEntity, stopId), (owner) => owner !== undefined && owner !== initial[0]) |
| 571 | ) |
| 572 | orderGate.openUnsafe() |
| 573 | assert.strictEqual(yield* Fiber.join(request), 1) |
| 574 | yield* Fiber.join(stopping) |
| 575 | yield* cluster.waitForStableAssignments() |
| 576 | |
| 577 | const killId = `kill-${backend}` |
| 578 | const killed = yield* cluster.ownerOfEntity(StateEntity, killId) |
| 579 | yield* cluster.kill(killed!) |
| 580 | const reply = yield* client(killId).Increment( |
| 581 | new Request({ id: `${backend}-kill`, sequence: 0 }) |
| 582 | ) |
| 583 | assert.notStrictEqual(reply.runner, addressString(killed!.address)) |
| 584 | yield* cluster.waitForStableAssignments() |
| 585 | assert.strictEqual((yield* cluster.messageCounts()).unprocessed, 0) |
| 586 | })) |
| 587 | |
| 588 | it.live(`${backend}: transfers frozen row locks after expiry`, () => |
| 589 | Effect.gen(function*() { |
| 590 | const cluster = yield* make({ backend, entities: StandardEntities, lockMode: "row" }) |
| 591 | yield* cluster.start(2) |
| 592 | yield* cluster.waitForStableAssignments() |
| 593 | const id = `frozen-${backend}` |
| 594 | const oldOwner = yield* cluster.ownerOfEntity(StateEntity, id) |
| 595 | const shard = yield* cluster.shardOfEntity(StateEntity, id) |
| 596 | yield* cluster.freeze(oldOwner!) |
| 597 | yield* cluster.waitUntil( |
| 598 | "The frozen runner's row lock did not expire", |
| 599 | Effect.map(cluster.ownerOfEntity(StateEntity, id), (owner) => owner !== undefined && owner !== oldOwner), |
| 600 | "12 seconds" |
| 601 | ) |
| 602 | const nextOwner = yield* cluster.ownerOfEntity(StateEntity, id) |
| 603 | assert.strictEqual(cluster.ownersOfShard(shard).length, 1) |
| 604 | assert.strictEqual(cluster.ownersOfShard(shard)[0], nextOwner) |
| 605 | const client = yield* cluster.getClient(StateEntity) |
| 606 | const reply = yield* client(id).Increment(new Request({ id: `${backend}-freeze`, sequence: 0 })) |
| 607 | assert.strictEqual(reply.runner, addressString(nextOwner!.address)) |
| 608 | yield* cluster.kill(oldOwner!) |
| 609 | })) |
| 610 | |
| 611 | it.live(`${backend}: retains frozen advisory locks until the session closes`, () => |
| 612 | Effect.gen(function*() { |
| 613 | const cluster = yield* make({ backend, entities: StandardEntities }) |
| 614 | yield* cluster.start(2) |
| 615 | yield* cluster.waitForStableAssignments() |
| 616 | const id = `frozen-advisory-${backend}` |
| 617 | const oldOwner = yield* cluster.ownerOfEntity(StateEntity, id) |
| 618 | const shard = yield* cluster.shardOfEntity(StateEntity, id) |
| 619 | yield* cluster.freeze(oldOwner!) |
| 620 | const deadline = (yield* Clock.currentTimeMillis) + 3_000 |
| 621 | yield* cluster.waitUntil( |
| 622 | "The advisory-lock observation window did not elapse", |
| 623 | Effect.map(Clock.currentTimeMillis, (now) => now >= deadline), |
| 624 | "5 seconds" |
| 625 | ) |
| 626 | assert.strictEqual(cluster.ownersOfShard(shard).length, 0) |
| 627 | assert.deepStrictEqual(cluster.ownersOfShard(shard, true), [oldOwner]) |
| 628 | yield* cluster.kill(oldOwner!) |
| 629 | yield* cluster.waitUntil( |
| 630 | "The advisory lock was not handed over after its session closed", |
| 631 | Effect.map(cluster.ownerOfEntity(StateEntity, id), (owner) => owner !== undefined && owner !== oldOwner) |
| 632 | ) |
| 633 | })) |
| 634 | } |
| 635 | |
| 636 | it.live("assigns annotated entities only to runners in their shard group", () => |
| 637 | Effect.gen(function*() { |
| 638 | const entities = Layer.mergeAll(StateEntityLayer, GroupEntityLayer) |
| 639 | const cluster = yield* make({ |
| 640 | backend: "pg", |
| 641 | config: { availableShardGroups: ["default", "special"], shardsPerGroup: 30 }, |
| 642 | entities |
| 643 | }) |
| 644 | const [defaultRunner] = yield* cluster.start(1, { assignedShardGroups: ["default"] }) |
| 645 | const [specialRunner] = yield* cluster.start(1, { assignedShardGroups: ["special"] }) |
| 646 | yield* cluster.waitForStableAssignments() |
| 647 | const client = yield* cluster.getClient(GroupEntity) |
| 648 | assert.strictEqual(yield* client("grouped").Runner(), addressString(specialRunner.address)) |
| 649 | assert.strictEqual(yield* cluster.ownerOfEntity(GroupEntity, "grouped"), specialRunner) |
| 650 | assert.strictEqual(yield* cluster.ownerOfEntity(StateEntity, "default"), defaultRunner) |
| 651 | })) |
| 652 | |
| 653 | it.live("runs one singleton and migrates it after owner death", () => |
| 654 | Effect.gen(function*() { |
| 655 | const singleton = { active: 0, maxActive: 0, starts: 0 } |
| 656 | const singletonLayer = Singleton.make( |
| 657 | "cluster-integration-singleton", |
| 658 | Effect.acquireRelease( |
| 659 | Effect.sync(() => { |
| 660 | singleton.active++ |
| 661 | singleton.starts++ |
| 662 | singleton.maxActive = Math.max(singleton.maxActive, singleton.active) |
| 663 | }), |
| 664 | () => Effect.sync(() => singleton.active--) |
| 665 | ) |
| 666 | ) |
| 667 | const cluster = yield* make({ |
| 668 | backend: "pg", |
| 669 | entities: Layer.merge(StateEntityLayer, singletonLayer) |
| 670 | }) |
| 671 | const runners = yield* cluster.start(3) |
| 672 | yield* cluster.waitForStableAssignments() |
| 673 | yield* cluster.waitUntil("The singleton did not start", Effect.sync(() => singleton.active === 1)) |
| 674 | const firstOwner = yield* cluster.ownerOfEntity(StateEntity, "cluster-integration-singleton") |
| 675 | assert.isTrue(runners.includes(firstOwner!)) |
| 676 | yield* cluster.kill(firstOwner!) |
| 677 | yield* cluster.waitUntil( |
| 678 | "The singleton did not migrate", |
| 679 | Effect.sync(() => singleton.starts >= 2 && singleton.active === 1) |
| 680 | ) |
| 681 | assert.notStrictEqual( |
| 682 | yield* cluster.ownerOfEntity(StateEntity, "cluster-integration-singleton"), |
| 683 | firstOwner |
| 684 | ) |
| 685 | assert.strictEqual(singleton.maxActive, 1) |
| 686 | })) |
| 687 | |
| 688 | it.live("keeps EntityResource alive during movement and releases it explicitly", () => |
| 689 | Effect.gen(function*() { |
| 690 | resourceState.acquired = 0 |
| 691 | resourceState.released = 0 |
| 692 | const cluster = yield* make({ backend: "pg", entities: ResourceEntityLayer }) |
| 693 | yield* cluster.start(2) |
| 694 | yield* cluster.waitForStableAssignments() |
| 695 | const client = yield* cluster.getClient(ResourceEntity) |
| 696 | assert.strictEqual(yield* client("resource").Get(), 1) |
| 697 | const owner = yield* cluster.ownerOfEntity(ResourceEntity, "resource") |
| 698 | yield* cluster.stop(owner!) |
| 699 | yield* cluster.waitUntil( |
| 700 | "The resource entity did not move", |
| 701 | Effect.map(cluster.ownerOfEntity(ResourceEntity, "resource"), (next) => next !== undefined && next !== owner) |
| 702 | ) |
| 703 | assert.strictEqual(resourceState.released, 0) |
| 704 | assert.strictEqual(yield* client("resource").Get(), 2) |
| 705 | yield* client("resource").Close() |
| 706 | yield* cluster.waitUntil( |
| 707 | "The entity resource was not released", |
| 708 | Effect.sync(() => resourceState.released === 1) |
| 709 | ) |
| 710 | })) |
| 711 | |
| 712 | for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) { |
| 713 | it.live(`${backend}: isolates clusters with different table prefixes`, () => |
| 714 | Effect.gen(function*() { |
| 715 | const first = yield* make({ backend, entities: StateEntityLayer, config: { shardsPerGroup: 30 } }) |
| 716 | const second = yield* make({ backend, entities: StateEntityLayer, config: { shardsPerGroup: 30 } }) |
| 717 | const [firstRunner] = yield* first.start(1) |
| 718 | const [secondRunner] = yield* second.start(1) |
| 719 | yield* first.waitForStableAssignments() |
| 720 | yield* second.waitForStableAssignments() |
| 721 | assert.notStrictEqual(first.prefix, second.prefix) |
| 722 | const firstRegistrations = (yield* first.diagnostics()).registrations |
| 723 | const secondRegistrations = (yield* second.diagnostics()).registrations |
| 724 | assert.deepStrictEqual(firstRegistrations.map((row) => row.address), [addressString(firstRunner.address)]) |
| 725 | assert.deepStrictEqual(secondRegistrations.map((row) => row.address), [addressString(secondRunner.address)]) |
| 726 | const firstClient = yield* first.getClient(StateEntity) |
| 727 | const secondClient = yield* second.getClient(StateEntity) |
| 728 | const firstReply = yield* firstClient("same-id").Increment( |
| 729 | new Request({ id: `${backend}-prefix-first`, sequence: 0 }) |
| 730 | ) |
| 731 | const secondReply = yield* secondClient("same-id").Increment( |
| 732 | new Request({ id: `${backend}-prefix-second`, sequence: 0 }) |
| 733 | ) |
| 734 | assert.strictEqual(firstReply.runner, addressString(firstRunner.address)) |
| 735 | assert.strictEqual(secondReply.runner, addressString(secondRunner.address)) |
| 736 | })) |
| 737 | } |
| 738 | }) |