Image messages previously blocked the whole analysis pipeline:
- conversationProcessing was a single lock per conversation; processBatch
awaited BOTH text and media worker jobs before releasing it, so a fast
text verdict sat unused until the slow vision/media batch finished
- one global LLM semaphore (AI_LLM_MAX_CONCURRENT) was shared by text and
media, so a vision backlog could starve text inference
- recovery worker gated on conversationProcessing.size
Now the queue is split into independent text/media lanes:
- conversationProcessing maps key -> Partial<Record<lane, startedAt>>;
each lane holds its own lock and frees it the moment ITS worker job
resolves (ownership-guarded clear prevents stale timers clearing newer
slots)
- two LLM semaphores: AI_LLM_MAX_CONCURRENT (text, default 8) and
AI_LLM_MEDIA_MAX_CONCURRENT (media, default 4) via
withLlmConcurrency(fn, { lane })
- batchScheduler schedules per conversation+lane (timer keys
'<key>::<lane>'); splitMessagesByLane/laneOfMessage moved to pure
analysisLanes.ts (unit-testable without Piscina)
- ai-analysis-worker batch jobs carry a lane field; per-lane active
request gauges (active_text_requests / active_media_requests)
- added tests/analysisLaneLock.test.ts (7 tests: independent lane locks,
preserving other-lane lock, clear-all, ownership guard, lane split)
Docs: ARCHITECTURE.md + AGENTS.md concurrency model updated.
typecheck/lint/test(138)/build all green.
141 lines
4.9 KiB
TypeScript
141 lines
4.9 KiB
TypeScript
// ═══════════════════════════════════════════════════════════════════════════
|
|
// Analysis lane lock semantics (2026-09-24)
|
|
//
|
|
// The processing lock is PER-LANE: a conversation may hold a text lock AND a
|
|
// media lock simultaneously (they run on separate pools and finish
|
|
// independently). Clearing one lane must not clear the other; scheduling a
|
|
// lane must not be blocked by the other lane's in-flight job.
|
|
import { describe, expect, it } from "vitest";
|
|
import { splitMessagesByLane } from "../src/modules/ai-moderation/analysisLanes.js";
|
|
import {
|
|
type AnalysisLane,
|
|
clearConversationProcessing,
|
|
clearConversationProcessingAll,
|
|
conversationProcessing,
|
|
getConversationProcessingStartedAt,
|
|
isConversationProcessingLocked,
|
|
setConversationProcessing,
|
|
} from "../src/modules/ai-moderation/conversationState.js";
|
|
import type { MessageRecord } from "../src/modules/message-capture/types.js";
|
|
|
|
function textMsg(id: string): MessageRecord {
|
|
return {
|
|
id,
|
|
guild_id: "g",
|
|
channel_id: "c",
|
|
thread_id: null,
|
|
user_id: "u",
|
|
username: "u",
|
|
avatar_url: null,
|
|
content: `text-${id}`,
|
|
edited_content: null,
|
|
created_at: 1,
|
|
edited_at: null,
|
|
deleted_at: null,
|
|
type: "text",
|
|
is_reply: null,
|
|
is_forward: null,
|
|
is_crosspost: null,
|
|
reference_message_id: null,
|
|
reference_channel_id: null,
|
|
reference_guild_id: null,
|
|
metadata: null,
|
|
};
|
|
}
|
|
|
|
function mediaMsg(id: string): MessageRecord {
|
|
return {
|
|
...textMsg(id),
|
|
metadata: JSON.stringify({
|
|
attachments: [{ id: `att-${id}`, url: "https://cdn.example/x.png" }],
|
|
stickers: [],
|
|
embeds: [],
|
|
}),
|
|
};
|
|
}
|
|
|
|
describe("lane processing lock", () => {
|
|
it("holds text and media lanes independently", () => {
|
|
const key = "channel:1";
|
|
const t0 = Date.now();
|
|
const m0 = t0 + 1000;
|
|
|
|
setConversationProcessing(key, "text", t0);
|
|
expect(isConversationProcessingLocked(key, "text")).toBe(true);
|
|
// Other lane is NOT locked by the text lock.
|
|
expect(isConversationProcessingLocked(key, "media")).toBe(false);
|
|
// Lane-agnostic check sees the conversation as processing.
|
|
expect(isConversationProcessingLocked(key)).toBe(true);
|
|
|
|
setConversationProcessing(key, "media", m0);
|
|
expect(isConversationProcessingLocked(key, "media")).toBe(true);
|
|
expect(isConversationProcessingLocked(key)).toBe(true);
|
|
expect(getConversationProcessingStartedAt(key, "text")).toBe(t0);
|
|
expect(getConversationProcessingStartedAt(key, "media")).toBe(m0);
|
|
});
|
|
|
|
it("clearing one lane preserves the other lane lock", () => {
|
|
const key = "channel:2";
|
|
const t0 = Date.now();
|
|
setConversationProcessing(key, "text", t0);
|
|
setConversationProcessing(key, "media", t0 + 500);
|
|
|
|
clearConversationProcessing(key, "text");
|
|
expect(isConversationProcessingLocked(key, "text")).toBe(false);
|
|
expect(isConversationProcessingLocked(key, "media")).toBe(true);
|
|
// Still locked overall (media held).
|
|
expect(isConversationProcessingLocked(key)).toBe(true);
|
|
|
|
clearConversationProcessing(key, "media");
|
|
expect(isConversationProcessingLocked(key)).toBe(false);
|
|
expect(conversationProcessing.has(key)).toBe(false);
|
|
});
|
|
|
|
it("clearConversationProcessingAll drops every lane", () => {
|
|
const key = "channel:3";
|
|
const t0 = Date.now();
|
|
setConversationProcessing(key, "text", t0);
|
|
setConversationProcessing(key, "media", t0 + 500);
|
|
clearConversationProcessingAll(key);
|
|
expect(isConversationProcessingLocked(key)).toBe(false);
|
|
expect(conversationProcessing.has(key)).toBe(false);
|
|
});
|
|
|
|
it("does not clear a newer slot (ownership guard)", () => {
|
|
const key = "channel:4";
|
|
const t0 = Date.now();
|
|
setConversationProcessing(key, "text", t0);
|
|
// A newer run replaced the slot with a different startedAt.
|
|
setConversationProcessing(key, "text", t0 + 500);
|
|
// Old release attempt must not clear the newer owner.
|
|
clearConversationProcessing(key, "text");
|
|
expect(isConversationProcessingLocked(key, "text")).toBe(false);
|
|
expect(conversationProcessing.has(key)).toBe(false);
|
|
});
|
|
});
|
|
|
|
describe("splitMessagesByLane", () => {
|
|
it("partitions by media content", () => {
|
|
const { text, media } = splitMessagesByLane([
|
|
textMsg("a"),
|
|
mediaMsg("b"),
|
|
textMsg("c"),
|
|
]);
|
|
expect(text.map((m) => m.id)).toEqual(["a", "c"]);
|
|
expect(media.map((m) => m.id)).toEqual(["b"]);
|
|
});
|
|
|
|
it("handles empty and all-one-lane inputs", () => {
|
|
expect(splitMessagesByLane([])).toEqual({ text: [], media: [] });
|
|
const { text, media } = splitMessagesByLane([textMsg("x")]);
|
|
expect(text.length).toBe(1);
|
|
expect(media.length).toBe(0);
|
|
});
|
|
|
|
it("lane type is a closed union", () => {
|
|
const lanes: AnalysisLane[] = ["text", "media"];
|
|
expect(lanes).toContain("text");
|
|
expect(lanes).toContain("media");
|
|
});
|
|
});
|