Bohdan Triapitsyn
237cae16b3
A queued message is removed from the queue only after its send resolves, so between dispatch and resolution it stays visible to every reader — and a composer submit merges the whole queue into its own send. Over a relay that window is seconds, long enough to deliver the same message twice. The queue now tracks which entries are awaiting the server. Dispatchers skip them, clearQueue retains them so the pending send can still remove or restore its own entry, and the flag is not persisted because a restart has no in-flight sends.
97 lines
4.0 KiB
TypeScript
97 lines
4.0 KiB
TypeScript
import { beforeEach, describe, expect, test } from "bun:test"
|
|
import {
|
|
createMessageQueueTarget,
|
|
getMessageQueueKey,
|
|
migrateMessageQueueState,
|
|
parseMessageQueueKey,
|
|
useMessageQueueStore,
|
|
} from "./messageQueueStore"
|
|
|
|
beforeEach(() => {
|
|
useMessageQueueStore.setState({ queuedMessages: {}, quarantinedLegacyMessages: {}, sendingIds: {} })
|
|
})
|
|
|
|
describe("message queue runtime ownership", () => {
|
|
test("isolates colliding session IDs by runtime and directory", () => {
|
|
const a = createMessageQueueTarget("session-1", "/repo", "runtime-a")!
|
|
const b = createMessageQueueTarget("session-1", "/repo", "runtime-b")!
|
|
useMessageQueueStore.getState().addToQueue(a, { content: "from A" })
|
|
useMessageQueueStore.getState().addToQueue(b, { content: "from B" })
|
|
|
|
expect(useMessageQueueStore.getState().getQueueForTarget(a)[0]?.content).toBe("from A")
|
|
expect(useMessageQueueStore.getState().getQueueForTarget(b)[0]?.content).toBe("from B")
|
|
})
|
|
|
|
test("round trips a composite queue key", () => {
|
|
const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")!
|
|
expect(parseMessageQueueKey(getMessageQueueKey(target))).toEqual(target)
|
|
})
|
|
|
|
test("quarantines legacy session-only queues instead of activating them", () => {
|
|
const migrated = migrateMessageQueueState({
|
|
queuedMessages: {
|
|
"session-1": [{ id: "queued-1", content: "legacy", createdAt: 1 }],
|
|
},
|
|
}, 1)
|
|
|
|
expect(migrated.queuedMessages).toEqual({})
|
|
expect(migrated.quarantinedLegacyMessages?.["session-1"]?.[0]?.content).toBe("legacy")
|
|
})
|
|
|
|
test("bounds each queue to the newest 20 messages", () => {
|
|
const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")!
|
|
for (let index = 0; index < 25; index += 1) {
|
|
useMessageQueueStore.getState().addToQueue(target, { content: `message-${index}` })
|
|
}
|
|
|
|
const queue = useMessageQueueStore.getState().getQueueForTarget(target)
|
|
expect(queue).toHaveLength(20)
|
|
expect(queue[0]?.content).toBe("message-5")
|
|
})
|
|
})
|
|
|
|
describe("in-flight queued sends", () => {
|
|
test("hides a dispatched message from the sendable queue but keeps it visible", () => {
|
|
const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")!
|
|
const store = useMessageQueueStore.getState()
|
|
store.addToQueue(target, { content: "first" })
|
|
store.addToQueue(target, { content: "second" })
|
|
const [first] = useMessageQueueStore.getState().getQueueForTarget(target)
|
|
|
|
useMessageQueueStore.getState().markSending(target, first.id)
|
|
|
|
expect(useMessageQueueStore.getState().getQueueForTarget(target)).toHaveLength(2)
|
|
const sendable = useMessageQueueStore.getState().getSendableQueue(target)
|
|
expect(sendable).toHaveLength(1)
|
|
expect(sendable[0]?.content).toBe("second")
|
|
|
|
useMessageQueueStore.getState().clearSending(target, first.id)
|
|
expect(useMessageQueueStore.getState().getSendableQueue(target)).toHaveLength(2)
|
|
expect(useMessageQueueStore.getState().sendingIds).toEqual({})
|
|
})
|
|
|
|
test("clearQueue retains a message whose send is still awaiting the server", () => {
|
|
const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")!
|
|
const store = useMessageQueueStore.getState()
|
|
store.addToQueue(target, { content: "in flight" })
|
|
store.addToQueue(target, { content: "merged by composer" })
|
|
const [inFlight] = useMessageQueueStore.getState().getQueueForTarget(target)
|
|
useMessageQueueStore.getState().markSending(target, inFlight.id)
|
|
|
|
useMessageQueueStore.getState().clearQueue(target)
|
|
|
|
const remaining = useMessageQueueStore.getState().getQueueForTarget(target)
|
|
expect(remaining).toHaveLength(1)
|
|
expect(remaining[0]?.id).toBe(inFlight.id)
|
|
})
|
|
|
|
test("clearQueue drops everything once no send is in flight", () => {
|
|
const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")!
|
|
useMessageQueueStore.getState().addToQueue(target, { content: "queued" })
|
|
|
|
useMessageQueueStore.getState().clearQueue(target)
|
|
|
|
expect(useMessageQueueStore.getState().getQueueForTarget(target)).toHaveLength(0)
|
|
})
|
|
})
|