617 lines
22 KiB
TypeScript
617 lines
22 KiB
TypeScript
import { assert, describe, expect, it } from "@effect/vitest"
|
|
import { Array, Cause, Clock, Effect, Exit, Fiber, Layer, MutableRef, Option, Queue, Stream } from "effect"
|
|
import { TestClock } from "effect/testing"
|
|
import {
|
|
MessageStorage,
|
|
RunnerAddress,
|
|
RunnerHealth,
|
|
Runners,
|
|
RunnerStorage,
|
|
Sharding,
|
|
ShardingConfig,
|
|
Snowflake
|
|
} from "effect/unstable/cluster"
|
|
import {
|
|
CallerId,
|
|
ContextBleedEntity,
|
|
ContextBleedLayer,
|
|
TestEntity,
|
|
TestEntityNoState,
|
|
TestEntityState,
|
|
User
|
|
} from "./TestEntity.ts"
|
|
|
|
describe.concurrent("Sharding", () => {
|
|
it.effect("delivers volatile requests directly to the entity", () =>
|
|
Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const user = yield* client.GetUserVolatile({ id: 1 })
|
|
expect(user).toEqual(new User({ id: 1, name: "User 1" }))
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("does not freeze the first caller's context into the entity server", () =>
|
|
Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const makeClient = yield* ContextBleedEntity.client
|
|
const client = makeClient("1")
|
|
|
|
const first = yield* client.ReadCaller().pipe(Effect.provideService(CallerId, "A"))
|
|
expect(first).toEqual("A")
|
|
|
|
const second = yield* client.ReadCaller()
|
|
expect(second).toEqual("none")
|
|
|
|
const durable = yield* client.ReadCallerPersisted()
|
|
expect(durable).toEqual("none")
|
|
}).pipe(Effect.provide(ContextBleedSharding)))
|
|
|
|
it.effect("persists durable requests until the entity replies", () =>
|
|
Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const user = yield* client.GetUser({ id: 1 })
|
|
expect(user).toEqual(new User({ id: 1, name: "User 1" }))
|
|
expect(driver.journal.length).toEqual(1)
|
|
expect(driver.unprocessed.size).toEqual(0)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("routes durable interrupts through storage", () =>
|
|
Effect.gen(function*() {
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
const state = yield* TestEntityState
|
|
const makeClient = yield* TestEntity.client
|
|
yield* TestClock.adjust(1)
|
|
const client = makeClient("1")
|
|
|
|
const fiber = yield* client.Never().pipe(Effect.forkChild({ startImmediately: true }))
|
|
yield* TestClock.adjust(1)
|
|
yield* Fiber.interrupt(fiber)
|
|
|
|
yield* TestClock.adjust(1)
|
|
expect(driver.journal.length).toEqual(2)
|
|
expect(driver.replyIds.size).toEqual(1)
|
|
expect(Queue.sizeUnsafe(state.interrupts)).toEqual(1)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("interrupts aren't sent for durable messages on shutdown", () =>
|
|
Effect.gen(function*() {
|
|
let driver!: MessageStorage.MemoryDriver["Service"]
|
|
yield* Effect.gen(function*() {
|
|
driver = yield* MessageStorage.MemoryDriver
|
|
const makeClient = yield* TestEntity.client
|
|
yield* TestClock.adjust(1)
|
|
const client = makeClient("1")
|
|
yield* client.Never().pipe(Effect.forkChild)
|
|
yield* TestClock.adjust(1)
|
|
}).pipe(Effect.provide(TestSharding))
|
|
|
|
// request, client interrupt is dropped
|
|
expect(driver.journal.length).toEqual(1)
|
|
// server interrupt is not sent
|
|
expect(driver.replyIds.size).toEqual(0)
|
|
}))
|
|
|
|
it.effect("interrupts are sent for volatile messages on shutdown", () =>
|
|
Effect.gen(function*() {
|
|
let interrupted = false
|
|
const testClock = (yield* Clock.Clock) as TestClock.TestClock
|
|
|
|
yield* Effect.gen(function*() {
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const fiber = yield* client.NeverVolatile().pipe(Effect.forkChild({ startImmediately: true }))
|
|
yield* TestClock.adjust(1)
|
|
const config = yield* ShardingConfig.ShardingConfig
|
|
;(config as any).runnerAddress = Option.some(RunnerAddress.make("localhost", 1234))
|
|
fiber.currentDispatcher.scheduleTask(() => {
|
|
fiber.interruptUnsafe()
|
|
Effect.runFork(testClock.adjust(30000))
|
|
}, 0)
|
|
}).pipe(
|
|
Effect.provide(TestShardingWithoutRunners.pipe(
|
|
Layer.provide(
|
|
Layer.effect(Runners.Runners)(
|
|
Effect.gen(function*() {
|
|
const runners = yield* Runners.makeNoop
|
|
return {
|
|
...runners,
|
|
send(options) {
|
|
if (options.message.envelope._tag === "Interrupt") {
|
|
interrupted = true
|
|
return Effect.void
|
|
}
|
|
return runners.send(options)
|
|
}
|
|
}
|
|
})
|
|
)
|
|
),
|
|
Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]),
|
|
Layer.provideMerge(ShardingConfig.layer({
|
|
entityMailboxCapacity: 10,
|
|
entityTerminationTimeout: 30000,
|
|
entityMessagePollInterval: 5000,
|
|
sendRetryInterval: 100,
|
|
refreshAssignmentsInterval: 100
|
|
}))
|
|
))
|
|
)
|
|
|
|
assert.isTrue(interrupted)
|
|
}))
|
|
|
|
it.effect("malformed message in storage", () =>
|
|
Effect.gen(function*() {
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
const makeClient = yield* TestEntity.client
|
|
yield* TestClock.adjust(1)
|
|
const client = makeClient("1")
|
|
|
|
const fiber = yield* client.Never().pipe(Effect.forkChild)
|
|
yield* TestClock.adjust(1)
|
|
|
|
const request = driver.journal[0]
|
|
yield* driver.encoded.saveEnvelope({
|
|
envelope: {
|
|
id: "boom",
|
|
_tag: "Interrupt",
|
|
requestId: request.requestId,
|
|
address: {
|
|
shardId: request.address.shardId
|
|
} as any
|
|
},
|
|
primaryKey: null,
|
|
deliverAt: null
|
|
})
|
|
|
|
// wait for storage to poll
|
|
yield* TestClock.adjust(5000)
|
|
|
|
const exit = fiber.pollUnsafe()
|
|
assert(exit && Exit.isFailure(exit) && Cause.hasDies(exit.cause))
|
|
|
|
// malformed message should be left in the database
|
|
expect(driver.journal.length).toEqual(2)
|
|
// defect reply should be sent
|
|
expect(driver.replyIds.size).toEqual(1)
|
|
|
|
const reply = driver.requests.get(request.requestId)!.replies[0]
|
|
assert(reply._tag === "WithExit" && reply.exit._tag === "Failure" && reply.exit.cause[0]._tag === "Die")
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("fails volatile requests immediately when the mailbox is full", () =>
|
|
Effect.gen(function*() {
|
|
const makeClient = yield* TestEntity.client
|
|
yield* TestClock.adjust(1)
|
|
const client = makeClient("1")
|
|
|
|
yield* client.NeverVolatile().pipe(Effect.forkChild, Effect.replicateEffect(10))
|
|
yield* TestClock.adjust(1)
|
|
const error = yield* client.NeverVolatile().pipe(Effect.flip)
|
|
assert.strictEqual(error._tag, "MailboxFull")
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("durable messages are retried when mailbox is full", () =>
|
|
Effect.gen(function*() {
|
|
const requestedIds = yield* Queue.make<Array<Snowflake.Snowflake>>()
|
|
yield* Effect.gen(function*() {
|
|
const state = yield* TestEntityState
|
|
const makeClient = yield* TestEntity.client
|
|
yield* TestClock.adjust(1)
|
|
const client = makeClient("1")
|
|
|
|
const fibers = yield* client.NeverFork().pipe(Effect.forkChild, Effect.replicateEffect(11))
|
|
yield* TestClock.adjust(1)
|
|
|
|
// wait for entity to go into resume mode and request ids
|
|
const ids = yield* Queue.take(requestedIds)
|
|
assert.strictEqual(ids.length, 1)
|
|
|
|
// test entity should still only have 10 requests
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 10)
|
|
|
|
// interrupt first request
|
|
yield* Fiber.interrupt(fibers[0])
|
|
yield* TestClock.adjust(100) // let retry happen
|
|
|
|
// last request should come through
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 11)
|
|
|
|
// interrupt second request, now the entity should be back in the main storage loop
|
|
yield* Fiber.interrupt(fibers[1])
|
|
|
|
// send another request within mailbox capacity
|
|
yield* client.NeverFork().pipe(Effect.forkChild)
|
|
yield* TestClock.adjust(1)
|
|
yield* Fiber.interruptAll(fibers)
|
|
yield* TestClock.adjust(100)
|
|
|
|
// no more ids should have been requested from entity catch up
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(requestedIds), 0)
|
|
}).pipe(Effect.provide(TestShardingWithoutStorage.pipe(
|
|
Layer.updateService(MessageStorage.MessageStorage, (storage) => ({
|
|
...storage,
|
|
unprocessedMessagesById(messageIds) {
|
|
Queue.offerUnsafe(requestedIds, Array.fromIterable(messageIds))
|
|
return storage.unprocessedMessagesById(messageIds)
|
|
}
|
|
})),
|
|
Layer.provide(MessageStorage.layerMemory),
|
|
Layer.provide(TestShardingConfig)
|
|
)))
|
|
}))
|
|
|
|
it.effect("interrupt for future request works while mailbox is full", () =>
|
|
Effect.gen(function*() {
|
|
const state = yield* TestEntityState
|
|
const makeClient = yield* TestEntity.client
|
|
yield* TestClock.adjust(1)
|
|
const client = makeClient("1")
|
|
|
|
const fibers = yield* client.NeverFork().pipe(
|
|
Effect.forkChild({ startImmediately: true }),
|
|
Effect.replicateEffect(12)
|
|
)
|
|
yield* TestClock.adjust(1)
|
|
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 10)
|
|
|
|
// interrupt 11th request
|
|
yield* Fiber.interrupt(fibers[10])
|
|
yield* TestClock.adjust(100) // let retry happen
|
|
// interrupt first request, and let the 11th request come through
|
|
yield* Fiber.interrupt(fibers[0])
|
|
yield* TestClock.adjust(100) // let retry happen
|
|
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 12)
|
|
// second interrupt should be sent
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(state.interrupts), 2)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("delivers durable streams and acknowledges each chunk", () =>
|
|
Effect.gen(function*() {
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
yield* TestClock.adjust(1)
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const users = yield* client.GetAllUsers({ ids: [1, 2, 3] }).pipe(
|
|
Stream.runCollect
|
|
)
|
|
expect(users).toEqual([
|
|
new User({ id: 1, name: "User 1" }),
|
|
new User({ id: 2, name: "User 2" }),
|
|
new User({ id: 3, name: "User 3" })
|
|
])
|
|
|
|
// 1 request, 3 acks, 4 replies
|
|
expect(driver.journal.length).toEqual(4)
|
|
expect(driver.replyIds.size).toEqual(4)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("durable stream while mailbox is full", () =>
|
|
Effect.gen(function*() {
|
|
const requestedIds = yield* Queue.make<Array<Snowflake.Snowflake>>()
|
|
yield* Effect.gen(function*() {
|
|
const state = yield* TestEntityState
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
|
|
const fibers = yield* client.NeverFork().pipe(
|
|
Effect.forkChild({ startImmediately: true }),
|
|
Effect.replicateEffect(10)
|
|
)
|
|
yield* TestClock.adjust(1)
|
|
|
|
const fiber = yield* client.GetAllUsers({ ids: [1, 2, 3] }).pipe(
|
|
Stream.runCollect,
|
|
Effect.forkChild({ startImmediately: true })
|
|
)
|
|
|
|
// make sure entity doesn't leave resume mode
|
|
yield* client.NeverFork().pipe(Effect.forkChild({ startImmediately: true }))
|
|
yield* client.NeverFork().pipe(Effect.forkChild({ startImmediately: true }))
|
|
|
|
// wait for entity to go into resume mode and request ids
|
|
const ids = yield* Queue.take(requestedIds)
|
|
assert.strictEqual(ids.length, 3)
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 10)
|
|
|
|
// interrupt first request
|
|
yield* Fiber.interrupt(fibers[0])
|
|
yield* TestClock.adjust(500) // let retry happen
|
|
|
|
// last request + NeverFork should come through
|
|
assert.deepStrictEqual(Queue.sizeUnsafe(state.envelopes), 12)
|
|
|
|
// acks should be allowed to be sent
|
|
const users = yield* Fiber.join(fiber)
|
|
expect(users).toEqual([
|
|
new User({ id: 1, name: "User 1" }),
|
|
new User({ id: 2, name: "User 2" }),
|
|
new User({ id: 3, name: "User 3" })
|
|
])
|
|
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
// 13 requests, 3 acks, 1 interrupt, 5 replies
|
|
assert.strictEqual(driver.journal.length, 13 + 3 + 1)
|
|
assert.strictEqual(driver.replyIds.size, 1 + 4)
|
|
}).pipe(Effect.provide(TestShardingWithoutStorage.pipe(
|
|
Layer.provideMerge(Layer.effect(MessageStorage.MemoryDriver)(MessageStorage.MemoryDriver)),
|
|
Layer.updateService(MessageStorage.MessageStorage, (storage) => ({
|
|
...storage,
|
|
unprocessedMessagesById(messageIds) {
|
|
Queue.offerUnsafe(requestedIds, Array.fromIterable(messageIds))
|
|
return storage.unprocessedMessagesById(messageIds)
|
|
}
|
|
})),
|
|
Layer.provide(MessageStorage.layerMemory),
|
|
Layer.provide(TestShardingConfig)
|
|
)))
|
|
}))
|
|
|
|
it.effect("durable messages are retried on restart", () =>
|
|
Effect.gen(function*() {
|
|
const EnvLayer = TestShardingWithoutState.pipe(
|
|
Layer.provide(Runners.layerNoop),
|
|
Layer.provide(TestShardingConfig)
|
|
)
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
const state = yield* TestEntityState
|
|
|
|
yield* Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
yield* Effect.forkChild(client.RequestWithKey({ key: "abc" }))
|
|
yield* TestClock.adjust(1)
|
|
}).pipe(
|
|
Effect.provide(EnvLayer),
|
|
Effect.scoped
|
|
)
|
|
|
|
// only the request should be in the journal
|
|
expect(driver.journal.length).toEqual(1)
|
|
expect(driver.replyIds.size).toEqual(0)
|
|
expect(driver.unprocessed.size).toEqual(1)
|
|
|
|
// add response
|
|
yield* Queue.offer(state.messages, void 0)
|
|
|
|
// Let the shards get assigned and storage poll
|
|
yield* TestClock.adjust(5000).pipe(
|
|
Effect.provide(EnvLayer),
|
|
Effect.scoped
|
|
)
|
|
|
|
expect(driver.journal.length).toEqual(1)
|
|
expect(driver.replyIds.size).toEqual(1)
|
|
expect(driver.unprocessed.size).toEqual(0)
|
|
|
|
// the client should read the result from storage
|
|
yield* Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const result = yield* client.RequestWithKey({ key: "abc" })
|
|
expect(result).toEqual(void 0)
|
|
}).pipe(
|
|
Effect.provide(EnvLayer),
|
|
Effect.scoped
|
|
)
|
|
|
|
// the request should not hit the entity
|
|
expect(driver.journal.length).toEqual(1)
|
|
expect(driver.replyIds.size).toEqual(1)
|
|
expect(driver.unprocessed.size).toEqual(0)
|
|
}).pipe(Effect.provide(MessageStorage.layerMemory.pipe(
|
|
Layer.provide(TestShardingConfig),
|
|
Layer.merge(TestEntityState.layer)
|
|
))))
|
|
|
|
it.effect("durable streams are resumed on restart", () =>
|
|
Effect.gen(function*() {
|
|
const EnvLayer = TestShardingWithoutState.pipe(
|
|
Layer.provide(Runners.layerNoop),
|
|
Layer.provide(TestShardingConfig)
|
|
)
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
const state = yield* TestEntityState
|
|
|
|
// first chunk
|
|
yield* Queue.offerAll(state.streamMessages, [void 0, void 0])
|
|
|
|
yield* Effect.gen(function*() {
|
|
yield* TestClock.adjust(2000)
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
yield* Effect.forkChild(Stream.runDrain(client.StreamWithKey({ key: "abc" })))
|
|
yield* TestClock.adjust(2000)
|
|
// second chunk
|
|
yield* Queue.offer(state.streamMessages, void 0)
|
|
yield* TestClock.adjust(2000)
|
|
}).pipe(
|
|
Effect.provide(EnvLayer),
|
|
Effect.scoped
|
|
)
|
|
|
|
// 1 request, 2 acks, 2 replies
|
|
expect(driver.journal.length).toEqual(1 + 2)
|
|
expect(driver.replyIds.size).toEqual(2)
|
|
expect(driver.unprocessed.size).toEqual(1)
|
|
|
|
// third chunk
|
|
yield* Queue.offerAll(state.streamMessages, [void 0, void 0])
|
|
yield* Queue.end(state.streamMessages)
|
|
|
|
// the client should resume
|
|
yield* Effect.gen(function*() {
|
|
yield* TestClock.adjust(5000) // let the shards get assigned and storage poll
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
|
|
// let the reply loop run
|
|
yield* TestClock.adjust(500).pipe(Effect.forkChild)
|
|
|
|
const results = yield* Stream.runCollect(client.StreamWithKey({ key: "abc" }))
|
|
expect(results).toEqual([3, 4])
|
|
}).pipe(
|
|
Effect.provide(EnvLayer),
|
|
Effect.scoped
|
|
)
|
|
|
|
// 1 request, 3 acks, 4 replies (3 chunks + WithExit)
|
|
expect(driver.journal.length).toEqual(1 + 3)
|
|
expect(driver.replyIds.size).toEqual(4)
|
|
expect(driver.unprocessed.size).toEqual(0)
|
|
}).pipe(Effect.provide(MessageStorage.layerMemory.pipe(
|
|
Layer.provide(TestShardingConfig),
|
|
Layer.merge(TestEntityState.layer)
|
|
))))
|
|
|
|
it.effect("client discard stores durable requests without waiting for replies", () =>
|
|
Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const result = yield* client.GetUser({ id: 123 }, { discard: true })
|
|
expect(result).toEqual(void 0)
|
|
yield* TestClock.adjust(1)
|
|
expect(driver.journal.length).toEqual(1)
|
|
expect(driver.unprocessed.size).toEqual(0)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("client discard returns while the durable request keeps processing", () =>
|
|
Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const driver = yield* MessageStorage.MemoryDriver
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const result = yield* client.Never(void 0, { discard: true })
|
|
expect(result).toEqual(void 0)
|
|
yield* TestClock.adjust(1)
|
|
expect(driver.journal.length).toEqual(1)
|
|
// should still be processing
|
|
expect(driver.unprocessed.size).toEqual(1)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("defects when a durable request has no MessageStorage", () =>
|
|
Effect.gen(function*() {
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
const cause = yield* client.Never().pipe(
|
|
Effect.sandbox,
|
|
Effect.flip
|
|
)
|
|
assert(Cause.hasDies(cause))
|
|
}).pipe(Effect.provide(TestShardingWithoutStorage.pipe(
|
|
Layer.provide(MessageStorage.layerNoop)
|
|
))))
|
|
|
|
it.effect("restarts the entity layer after a handler defect", () =>
|
|
Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const state = yield* TestEntityState
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
MutableRef.set(state.defectTrigger, true)
|
|
const result = yield* client.GetUser({ id: 123 })
|
|
expect(result).toEqual(new User({ id: 123, name: "User 123" }))
|
|
expect(state.layerBuilds.current).toEqual(2)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("replays in-flight requests when restarting after a defect", () =>
|
|
Effect.gen(function*() {
|
|
yield* TestClock.adjust(1)
|
|
const state = yield* TestEntityState
|
|
const makeClient = yield* TestEntity.client
|
|
const client = makeClient("1")
|
|
|
|
yield* client.NeverFork().pipe(Effect.forkChild({ startImmediately: true }))
|
|
yield* TestClock.adjust(1)
|
|
assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1)
|
|
|
|
MutableRef.set(state.defectTrigger, true)
|
|
const result = yield* client.GetUser({ id: 123 })
|
|
assert.deepStrictEqual(result, new User({ id: 123, name: "User 123" }))
|
|
assert.strictEqual(state.layerBuilds.current, 2)
|
|
|
|
yield* TestClock.adjust(1)
|
|
assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 4)
|
|
}).pipe(Effect.provide(TestSharding)))
|
|
|
|
it.effect("WithTransaction is propagated to the entity handler", () =>
|
|
Effect.gen(function*() {
|
|
let isTransaction = false
|
|
let transactionOpen = false
|
|
yield* Effect.gen(function*() {
|
|
const makeClient = yield* TestEntity.client
|
|
yield* TestClock.adjust(1)
|
|
const client = makeClient("1")
|
|
|
|
const result = yield* client.WithTransaction({ id: 1 })
|
|
assert.strictEqual(result, true)
|
|
assert.strictEqual(isTransaction, true)
|
|
}).pipe(Effect.provide(TestShardingWithoutStorage.pipe(
|
|
Layer.updateService(MessageStorage.MessageStorage, (storage) => ({
|
|
...storage,
|
|
withTransaction(effect) {
|
|
return Effect.suspend(() => {
|
|
transactionOpen = true
|
|
return storage.withTransaction(effect)
|
|
}).pipe(
|
|
Effect.ensuring(Effect.sync(() => {
|
|
transactionOpen = false
|
|
}))
|
|
)
|
|
},
|
|
saveReply(reply) {
|
|
return MessageStorage.MemoryTransaction.use((isTransaction_) => {
|
|
isTransaction = isTransaction_
|
|
assert.strictEqual(transactionOpen, true)
|
|
return storage.saveReply(reply)
|
|
})
|
|
}
|
|
})),
|
|
Layer.provide(MessageStorage.layerMemory),
|
|
Layer.provide(TestShardingConfig)
|
|
)))
|
|
}))
|
|
})
|
|
|
|
const TestShardingConfig = ShardingConfig.layer({
|
|
entityMailboxCapacity: 10,
|
|
entityTerminationTimeout: 0,
|
|
entityMessagePollInterval: 5000,
|
|
sendRetryInterval: 100,
|
|
refreshAssignmentsInterval: 0
|
|
})
|
|
|
|
const TestShardingWithoutState = TestEntityNoState.pipe(
|
|
Layer.provideMerge(Sharding.layer),
|
|
Layer.provide(RunnerStorage.layerMemory),
|
|
Layer.provide(RunnerHealth.layerNoop)
|
|
// Layer.provide(Logger.minimumLogLevel(LogLevel.All)),
|
|
// Layer.provideMerge(Logger.pretty)
|
|
)
|
|
|
|
const TestShardingWithoutRunners = TestShardingWithoutState.pipe(
|
|
Layer.provideMerge(TestEntityState.layer)
|
|
)
|
|
|
|
const TestShardingWithoutStorage = TestShardingWithoutRunners.pipe(
|
|
Layer.provide(Runners.layerNoop),
|
|
Layer.provide(TestShardingConfig)
|
|
)
|
|
|
|
const TestSharding = TestShardingWithoutStorage.pipe(
|
|
Layer.provideMerge(MessageStorage.layerMemory),
|
|
Layer.provide(TestShardingConfig)
|
|
)
|
|
|
|
const ContextBleedSharding = ContextBleedLayer.pipe(Layer.provideMerge(TestSharding))
|