From ce343bbae77ff1151515d52de82e0952c2cc6919 Mon Sep 17 00:00:00 2001 From: mayf3 Date: Wed, 19 Aug 2026 09:40:03 +0800 Subject: [PATCH 1/3] renew processing locks and surface handler errors --- README.md | 3 +- README.zh.md | 2 +- src/__tests__/message-handler-error.test.ts | 75 ++++++++++ src/channel.ts | 1 + src/safety/__tests__/processing-lock.test.ts | 37 ++++- .../__tests__/renewable-lock-pipeline.test.ts | 136 ++++++++++++++++++ src/safety/index.ts | 15 +- src/safety/processing-lock.ts | 80 +++++++++-- src/safety/types.ts | 1 + src/types.ts | 6 + 10 files changed, 333 insertions(+), 23 deletions(-) create mode 100644 src/__tests__/message-handler-error.test.ts create mode 100644 src/safety/__tests__/renewable-lock-pipeline.test.ts diff --git a/README.md b/README.md index 1067b38..60b0ca8 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ the QR URL as `source/` (passed through as-is, not defaulted). `PolicyConfig`: `requireMention` · `dmMode` (`'open' \| 'allowlist' \| 'pair' \| 'disabled'`) · `dmAllowlist` · `groupAllowlist` · `respondToMentionAll` · `botLoopGuard` (see [Bot-at-bot](#bot-at-bot)). `dmAllowlist` takes **sender ids** (`ou_…` / user_id / union_id), `groupAllowlist` takes **chat ids** (`oc_…`) — an app id (`cli_…`) belongs in neither and is warned about. -`SafetyConfig`: `dedup` (`ttl`/`maxEntries`/`sweepIntervalMs`) · `chatQueue` (`enabled`, `mergeWhileBusy`) · `batch.text` / `batch.media` · `staleMessageWindowMs`. +`SafetyConfig`: `dedup` (`ttl`/`maxEntries`/`sweepIntervalMs`) · `processingLock` (`ttlMs`/`renewIntervalMs`) · `chatQueue` (`enabled`, `mergeWhileBusy`) · `batch.text` / `batch.media` · `staleMessageWindowMs`. ### Lifecycle @@ -568,4 +568,3 @@ guarantee. ## License MIT - diff --git a/README.zh.md b/README.zh.md index 446691f..0cca9fe 100644 --- a/README.zh.md +++ b/README.zh.md @@ -103,7 +103,7 @@ const channel = createLarkChannel({ appId: client_id, appSecret: client_secret } `PolicyConfig`：`requireMention` · `dmMode`（`'open' \| 'allowlist' \| 'pair' \| 'disabled'`）· `dmAllowlist` · `groupAllowlist` · `respondToMentionAll` · `botLoopGuard`（见 [Bot-at-bot](#bot-at-bot)）。`dmAllowlist` 填**发送方 id**（`ou_…` / user_id / union_id），`groupAllowlist` 填**群 id**（`oc_…`）——应用 id（`cli_…`）两者都不属于，填了会告警。 -`SafetyConfig`：`dedup`（`ttl`/`maxEntries`/`sweepIntervalMs`）· `chatQueue`（`enabled`、`mergeWhileBusy`）· `batch.text` / `batch.media` · `staleMessageWindowMs`。 +`SafetyConfig`：`dedup`（`ttl`/`maxEntries`/`sweepIntervalMs`）· `processingLock`（`ttlMs`/`renewIntervalMs`）· `chatQueue`（`enabled`、`mergeWhileBusy`）· `batch.text` / `batch.media` · `staleMessageWindowMs`。 ### 生命周期 diff --git a/src/__tests__/message-handler-error.test.ts b/src/__tests__/message-handler-error.test.ts new file mode 100644 index 0000000..5bf28c4 --- /dev/null +++ b/src/__tests__/message-handler-error.test.ts @@ -0,0 +1,75 @@ +import { + createTestChannel, + dispatchEvent, + flushMicrotasks, + markConnected, +} from '../meeting/__tests__/fixtures'; + +function directMessage(messageId: string): unknown { + return { + sender: { sender_id: { open_id: 'ou_sender' }, sender_type: 'user' }, + message: { + message_id: messageId, + chat_id: 'oc_dm', + chat_type: 'p2p', + message_type: 'text', + content: '{"text":"hello"}', + create_time: String(Date.now()), + }, + }; +} + +describe.each([ + ['queue disabled', false], + ['queue enabled', true], +])('message handler errors with %s', (_label, queueEnabled) => { + test("surface through channel.on('error') exactly once", async () => { + const { ch } = createTestChannel({ + safety: { + chatQueue: { enabled: queueEnabled }, + batch: { text: { delayMs: 0 } }, + }, + }); + const errors: unknown[] = []; + markConnected(ch); + ch.on('error', (error: unknown) => errors.push(error)); + ch.on('message', async () => { + throw new Error('message handler exploded'); + }); + + await dispatchEvent(ch, 'im.message.receive_v1', directMessage(`om_error_${queueEnabled}`)); + await flushMicrotasks(); + + expect(errors).toHaveLength(1); + expect((errors[0] as Error).message).toContain('message handler exploded'); + }); +}); + +test('a throwing error observer cannot break handler cleanup', async () => { + const { ch, logger } = createTestChannel({ safety: { chatQueue: { enabled: false } } }); + let handlerCalls = 0; + markConnected(ch); + ch.on('error', () => { + throw new Error('observer exploded'); + }); + ch.on('message', async () => { + handlerCalls++; + throw new Error('message handler exploded'); + }); + + const raw = directMessage('om_observer_throw'); + await dispatchEvent(ch, 'im.message.receive_v1', raw); + await flushMicrotasks(); + await dispatchEvent(ch, 'im.message.receive_v1', raw); + await flushMicrotasks(); + + expect(handlerCalls).toBe(1); + expect( + logger.error.mock.calls.some( + ([entry]) => + Array.isArray(entry) && + entry[0] === 'safety: error observer threw' && + entry[1]?.message === 'observer exploded', + ), + ).toBe(true); +}); diff --git a/src/channel.ts b/src/channel.ts index 7eb2893..8c2a579 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -208,6 +208,7 @@ export class LarkChannel { const handler = this.handlers.message; if (handler) await handler(merged); }, + onError: (error) => this.emitError(error), }); } diff --git a/src/safety/__tests__/processing-lock.test.ts b/src/safety/__tests__/processing-lock.test.ts index c74319b..1efa53c 100644 --- a/src/safety/__tests__/processing-lock.test.ts +++ b/src/safety/__tests__/processing-lock.test.ts @@ -2,7 +2,10 @@ import { ProcessingLock } from '../processing-lock'; describe('ProcessingLock', () => { let lock: ProcessingLock; - afterEach(() => lock?.dispose()); + afterEach(() => { + lock?.dispose(); + vi.useRealTimers(); + }); test('first acquire succeeds, second fails until release', () => { lock = new ProcessingLock(); @@ -20,11 +23,16 @@ describe('ProcessingLock', () => { expect(lock.acquire('m2')).toBe(false); }); - test('expires after ttl', async () => { - lock = new ProcessingLock(50); + test('renews an acquired lease across multiple original ttl periods', async () => { + vi.useFakeTimers(); + lock = new ProcessingLock(50, 10); expect(lock.acquire('m1')).toBe(true); expect(lock.acquire('m1')).toBe(false); - await new Promise((r) => setTimeout(r, 80)); + await vi.advanceTimersByTimeAsync(120); + expect(lock.acquire('m1')).toBe(false); + + lock.stopRenewal('m1'); + await vi.advanceTimersByTimeAsync(51); expect(lock.acquire('m1')).toBe(true); }); @@ -32,4 +40,25 @@ describe('ProcessingLock', () => { lock = new ProcessingLock(); expect(() => lock.release('unknown')).not.toThrow(); }); + + test('dispose clears the renewal timer', () => { + vi.useFakeTimers(); + lock = new ProcessingLock(50, 10); + lock.acquire('m1'); + expect(vi.getTimerCount()).toBe(1); + lock.dispose(); + expect(vi.getTimerCount()).toBe(0); + }); + + test.each([ + [0, 1], + [-1, 1], + [Number.NaN, 1], + [50, 0], + [50, 50], + [50, 60], + [3_000_000_000, 2_147_483_648], + ])('invalid ttl/renew config fails loudly (%s, %s)', (ttlMs, renewIntervalMs) => { + expect(() => new ProcessingLock(ttlMs, renewIntervalMs)).toThrow(RangeError); + }); }); diff --git a/src/safety/__tests__/renewable-lock-pipeline.test.ts b/src/safety/__tests__/renewable-lock-pipeline.test.ts new file mode 100644 index 0000000..5856119 --- /dev/null +++ b/src/safety/__tests__/renewable-lock-pipeline.test.ts @@ -0,0 +1,136 @@ +import type { NormalizedMessage } from '../../types'; +import { SafetyPipeline } from '../index'; + +const logger = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + trace: vi.fn(), +} as any; + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +function message(messageId: string, chatId = 'oc_lock'): NormalizedMessage { + return { + messageId, + chatId, + chatType: 'p2p', + senderId: 'ou_sender', + content: messageId, + rawContentType: 'text', + resources: [], + mentions: [], + mentionAll: false, + mentionedBot: false, + createTime: Date.now(), + }; +} + +function makeCache() { + const values = new Map(); + return { + values, + cache: { + get: async (key: string) => values.get(key), + set: async (key: string, value: string) => { + values.set(key, value); + return true; + }, + } as any, + }; +} + +async function wait(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); +} + +test('handler pending across two ttl periods retains lock, then marks seen and releases', async () => { + const gate = deferred(); + const { cache, values } = makeCache(); + let handlerCalls = 0; + const pipeline = new SafetyPipeline({ + cache, + logger, + onReject: () => {}, + onMessage: async () => { + handlerCalls++; + await gate.promise; + }, + config: { + chatQueue: { enabled: false }, + processingLock: { ttlMs: 20, renewIntervalMs: 5 }, + }, + }); + + await pipeline.pushMessage(message('om_renew')); + await wait(50); + await pipeline.pushMessage(message('om_renew')); + expect(handlerCalls).toBe(1); + + gate.resolve(); + await wait(0); + expect(values.has('om_renew')).toBe(true); + expect((pipeline as any).lock.acquire('om_renew')).toBe(true); + (pipeline as any).lock.release('om_renew'); + + await pipeline.pushMessage(message('om_renew')); + await wait(0); + expect(handlerCalls).toBe(1); + await pipeline.dispose(); +}); + +test('lease stays renewable while a queued message waits behind another handler', async () => { + const firstGate = deferred(); + const { cache } = makeCache(); + const handled: string[] = []; + const pipeline = new SafetyPipeline({ + cache, + logger, + onReject: () => {}, + onMessage: async (msg) => { + handled.push(msg.messageId); + if (msg.messageId === 'om_first') await firstGate.promise; + }, + config: { + chatQueue: { enabled: true }, + batch: { text: { delayMs: 0, maxMessages: 1 } }, + processingLock: { ttlMs: 20, renewIntervalMs: 5 }, + }, + }); + + await pipeline.pushMessage(message('om_first')); + await pipeline.pushMessage(message('om_waiting')); + await wait(50); + await pipeline.pushMessage(message('om_waiting')); + expect(handled).toEqual(['om_first']); + + firstGate.resolve(); + await (pipeline as any).manager.flushAll(); + expect(handled).toEqual(['om_first', 'om_waiting']); + await pipeline.dispose(); +}); + +test.each([ + { ttlMs: 0, renewIntervalMs: 1 }, + { ttlMs: 20, renewIntervalMs: 0 }, + { ttlMs: 20, renewIntervalMs: 20 }, +])('SafetyConfig rejects invalid processing lock config: %o', (processingLock) => { + const { cache } = makeCache(); + expect( + () => + new SafetyPipeline({ + cache, + logger, + onReject: () => {}, + onMessage: async () => {}, + config: { processingLock }, + }), + ).toThrow(RangeError); +}); diff --git a/src/safety/index.ts b/src/safety/index.ts index 339d65d..7d8f687 100644 --- a/src/safety/index.ts +++ b/src/safety/index.ts @@ -37,6 +37,7 @@ export interface SafetyPipelineOptions { logger: Logger; onReject: OnReject; onMessage: OnMessageDispatch; + onError?: (error: unknown) => void; } /** @@ -59,21 +60,26 @@ export class SafetyPipeline { private readonly logger: Logger; private readonly onReject: OnReject; private readonly onMessage: OnMessageDispatch; + private readonly onError?: (error: unknown) => void; constructor(opts: SafetyPipelineOptions) { this.logger = opts.logger; this.onReject = opts.onReject; this.onMessage = opts.onMessage; + this.onError = opts.onError; this.staleWindow = opts.config?.staleMessageWindowMs ?? DEFAULT_STALE_MS; this.queueEnabled = opts.config?.chatQueue?.enabled ?? true; + this.lock = new ProcessingLock( + opts.config?.processingLock?.ttlMs, + opts.config?.processingLock?.renewIntervalMs, + ); this.seenCache = new SeenCache(opts.cache, { ttlMs: opts.config?.dedup?.ttl, maxMemEntries: opts.config?.dedup?.maxEntries, sweepMs: opts.config?.dedup?.sweepIntervalMs, }); - this.lock = new ProcessingLock(); this.policy = new PolicyGate(opts.policy, opts.botIdentity, opts.logger); this.loopGuard = new LoopGuard(opts.policy?.botLoopGuard, opts.logger); this.manager = new ChatPipelineManager(resolveBatchConfig(opts.config)); @@ -128,8 +134,14 @@ export class SafetyPipeline { await this.onMessage(batch.message); } catch (e) { this.logger.error?.(`safety: message handler threw`, e); + try { + this.onError?.(e); + } catch (observerError) { + this.logger.error?.(`safety: error observer threw`, observerError); + } } finally { for (const id of batch.sourceIds) { + this.lock.stopRenewal(id); try { await this.seenCache.add(id); } catch { @@ -175,6 +187,7 @@ export class SafetyPipeline { this.logger.error?.(`safety: action handler threw`, e); return undefined; } finally { + this.lock.stopRenewal(eventId); try { await this.seenCache.add(eventId); } catch { diff --git a/src/safety/processing-lock.ts b/src/safety/processing-lock.ts index e308c7d..a4dcc55 100644 --- a/src/safety/processing-lock.ts +++ b/src/safety/processing-lock.ts @@ -1,4 +1,11 @@ -import { DEFAULT_LOCK_TTL_MS } from './types'; +import { DEFAULT_LOCK_RENEW_INTERVAL_MS, DEFAULT_LOCK_TTL_MS } from './types'; + +interface LockEntry { + expiresAt: number; + renewable: boolean; +} + +const MAX_TIMER_DELAY_MS = 2_147_483_647; /** * Short-TTL in-memory lock to prevent concurrent processing of the same @@ -6,39 +13,82 @@ import { DEFAULT_LOCK_TTL_MS } from './types'; * window, during which the event is not yet committed to SeenCache. */ export class ProcessingLock { - private locks = new Map(); // id → expireAt (ms) - private sweeper: NodeJS.Timeout; + private readonly locks = new Map(); + private renewalTimer?: NodeJS.Timeout; + private readonly ttlMs: number; + private readonly renewIntervalMs: number; constructor( - private ttlMs: number = DEFAULT_LOCK_TTL_MS, - sweepMs: number = 60_000, + ttlMs: number = DEFAULT_LOCK_TTL_MS, + renewIntervalMs: number = Math.min(DEFAULT_LOCK_RENEW_INTERVAL_MS, ttlMs / 3), ) { - this.sweeper = setInterval(() => this.sweep(), sweepMs); - this.sweeper.unref?.(); + assertDuration('processingLock.ttlMs', ttlMs); + assertDuration('processingLock.renewIntervalMs', renewIntervalMs, MAX_TIMER_DELAY_MS); + if (renewIntervalMs >= ttlMs) { + throw new RangeError('processingLock.renewIntervalMs must be less than processingLock.ttlMs'); + } + this.ttlMs = ttlMs; + this.renewIntervalMs = renewIntervalMs; } - /** Returns true if the lock is acquired; false if already held. */ + /** + * Acquire a renewable lease. The lease remains live until `stopRenewal` is + * called, even when its handler is waiting in a queue or batch. + */ acquire(id: string): boolean { const now = Date.now(); - const exp = this.locks.get(id); - if (exp && exp > now) return false; - this.locks.set(id, now + this.ttlMs); + const current = this.locks.get(id); + if (current && current.expiresAt > now) return false; + this.locks.set(id, { expiresAt: now + this.ttlMs, renewable: true }); + this.ensureRenewalTimer(); return true; } + /** Stop extending a lease while leaving it held until explicit release. */ + stopRenewal(id: string): void { + const entry = this.locks.get(id); + if (entry) entry.renewable = false; + this.stopTimerIfIdle(); + } + release(id: string): void { this.locks.delete(id); + this.stopTimerIfIdle(); + } + + private ensureRenewalTimer(): void { + if (this.renewalTimer) return; + this.renewalTimer = setInterval(() => this.renew(), this.renewIntervalMs); + this.renewalTimer.unref?.(); } - private sweep(): void { + private renew(): void { const now = Date.now(); - for (const [k, v] of this.locks) { - if (v <= now) this.locks.delete(k); + for (const [id, entry] of this.locks) { + if (entry.renewable) entry.expiresAt = now + this.ttlMs; + else if (entry.expiresAt <= now) this.locks.delete(id); + } + this.stopTimerIfIdle(); + } + + private stopTimerIfIdle(): void { + if (!this.renewalTimer) return; + for (const entry of this.locks.values()) { + if (entry.renewable) return; } + clearInterval(this.renewalTimer); + this.renewalTimer = undefined; } dispose(): void { - clearInterval(this.sweeper); + if (this.renewalTimer) clearInterval(this.renewalTimer); + this.renewalTimer = undefined; this.locks.clear(); } } + +function assertDuration(name: string, value: number, max = Number.MAX_SAFE_INTEGER): void { + if (!Number.isFinite(value) || value < 1 || value > max) { + throw new RangeError(`${name} must be between 1 and ${max} milliseconds`); + } +} diff --git a/src/safety/types.ts b/src/safety/types.ts index 94ac294..9902575 100644 --- a/src/safety/types.ts +++ b/src/safety/types.ts @@ -29,6 +29,7 @@ export const DEFAULT_DEDUP = { export const DEFAULT_STALE_MS = 30 * 60_000; export const DEFAULT_LOCK_TTL_MS = 5 * 60_000; +export const DEFAULT_LOCK_RENEW_INTERVAL_MS = 60_000; export interface BatchedDispatch { message: NormalizedMessage; diff --git a/src/types.ts b/src/types.ts index 3930f1c..cce5a0b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -450,6 +450,12 @@ export interface SafetyConfig { maxEntries?: number; sweepIntervalMs?: number; }; + processingLock?: { + /** How long an in-flight lease remains valid without renewal. */ + ttlMs?: number; + /** How often an active lease is renewed. Must be less than `ttlMs`. */ + renewIntervalMs?: number; + }; chatQueue?: { enabled?: boolean; /** From 3940d8a67709902584f040fea1b4c2aa0418f135 Mon Sep 17 00:00:00 2001 From: mayf3 Date: Wed, 19 Aug 2026 20:51:35 +0800 Subject: [PATCH 2/3] build git dependencies during install --- package.json | 1 + 1 file changed, 1 insertion(+) diff --git a/package.json b/package.json index fa210a6..c381bfb 100644 --- a/package.json +++ b/package.json @@ -49,6 +49,7 @@ "lint": "biome check .", "format": "biome check --write .", "example": "tsx", + "prepare": "npm run build", "prepublishOnly": "pnpm typecheck && pnpm test && pnpm build" }, "dependencies": { From bd24f6742513769c80b5401b96ad464d74dd2027 Mon Sep 17 00:00:00 2001 From: mayf3 Date: Wed, 19 Aug 2026 22:15:25 +0800 Subject: [PATCH 3/3] harden processing lease ownership --- README.md | 8 ++ README.zh.md | 6 + src/__tests__/l1l3-sink.test.ts | 21 +-- src/__tests__/message-handler-error.test.ts | 74 +++++++++- src/channel.ts | 21 ++- src/safety/__tests__/chat-pipeline.test.ts | 48 ++++--- src/safety/__tests__/processing-lock.test.ts | 123 ++++++++++++++--- .../__tests__/renewable-lock-pipeline.test.ts | 128 +++++++++++++++++- src/safety/chat-pipeline.ts | 15 +- src/safety/index.ts | 30 ++-- src/safety/processing-lock.ts | 59 +++++--- src/safety/types.ts | 3 +- src/types.ts | 13 +- 13 files changed, 454 insertions(+), 95 deletions(-) diff --git a/README.md b/README.md index 60b0ca8..f02d8ba 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,14 @@ the QR URL as `source/` (passed through as-is, not defaulted). `SafetyConfig`: `dedup` (`ttl`/`maxEntries`/`sweepIntervalMs`) · `processingLock` (`ttlMs`/`renewIntervalMs`) · `chatQueue` (`enabled`, `mergeWhileBusy`) · `batch.text` / `batch.media` · `staleMessageWindowMs`. +`processingLock` defaults to a 300,000 ms TTL and a 60,000 ms renewal interval. Both +values must be integer milliseconds from 1 through 2,147,483,647, and +`renewIntervalMs` must be less than `ttlMs`. If only `ttlMs` is overridden, the +renewal interval is derived as the smaller of 60,000 ms and one third of the TTL +(rounded down, with a 1 ms minimum). Lease ownership is token-bound: an active or +finalizing handler cannot be displaced merely because wall-clock time has passed +its TTL. + ### Lifecycle | Method | Signature | Description | diff --git a/README.zh.md b/README.zh.md index 0cca9fe..597636b 100644 --- a/README.zh.md +++ b/README.zh.md @@ -105,6 +105,12 @@ const channel = createLarkChannel({ appId: client_id, appSecret: client_secret } `SafetyConfig`：`dedup`（`ttl`/`maxEntries`/`sweepIntervalMs`）· `processingLock`（`ttlMs`/`renewIntervalMs`）· `chatQueue`（`enabled`、`mergeWhileBusy`）· `batch.text` / `batch.media` · `staleMessageWindowMs`。 +`processingLock` 默认 TTL 为 300,000 ms，续租间隔为 60,000 ms。两者都必须是 +1 到 2,147,483,647 范围内的整数毫秒，并且 `renewIntervalMs` 必须小于 +`ttlMs`。如果只覆盖 `ttlMs`，续租间隔取 60,000 ms 与 TTL 三分之一向下取整 +后的较小值（最小 1 ms）。lease owner 由 token 绑定：active 或 finalizing +handler 不会仅因墙上时钟越过 TTL 而被其他请求抢占。 + ### 生命周期 | 方法 | 签名 | 说明 | diff --git a/src/__tests__/l1l3-sink.test.ts b/src/__tests__/l1l3-sink.test.ts index 0eb5058..3ca5a73 100644 --- a/src/__tests__/l1l3-sink.test.ts +++ b/src/__tests__/l1l3-sink.test.ts @@ -11,6 +11,7 @@ import type { Logger, WSConnectionStatus } from '../internal'; import { startKeepalive } from '../keepalive'; import { ChatPipeline } from '../safety/chat-pipeline'; +import type { ProcessingLease } from '../safety/processing-lock'; import type { BatchConfig, BatchedDispatch } from '../safety/types'; import type { NormalizedMessage } from '../types'; @@ -22,6 +23,10 @@ const silent: Logger = { trace: () => {}, }; +function lease(id: string): ProcessingLease { + return Object.freeze({ id, ownerToken: Symbol(id) }); +} + function msg(id: string, content = id): NormalizedMessage { return { messageId: id, @@ -65,19 +70,19 @@ describe('ChatPipeline mergeWhileBusy', () => { return Promise.resolve(); }; - p.push(msg('a'), handler); // flush #1 starts, pipeline now busy - p.push(msg('b'), handler); // accumulate while busy - p.push(msg('c'), handler); // accumulate while busy + p.push(msg('a'), lease('a'), handler); // flush #1 starts, pipeline now busy + p.push(msg('b'), lease('b'), handler); // accumulate while busy + p.push(msg('c'), lease('c'), handler); // accumulate while busy await Promise.resolve(); // Only the first batch has dispatched so far. expect(batches).toHaveLength(1); - expect(batches[0].sourceIds).toEqual(['a']); + expect(batches[0].sources.map((source) => source.messageId)).toEqual(['a']); releaseFirst(); // first handler resolves → settle hook flushes b+c await p.flushNow(); expect(batches).toHaveLength(2); - expect(batches[1].sourceIds).toEqual(['b', 'c']); + expect(batches[1].sources.map((source) => source.messageId)).toEqual(['b', 'c']); expect(batches[1].message.content).toBe('b\n\nc'); }); @@ -88,10 +93,10 @@ describe('ChatPipeline mergeWhileBusy', () => { batches.push(d); return Promise.resolve(); }; - p.push(msg('a'), handler); - p.push(msg('b'), handler); + p.push(msg('a'), lease('a'), handler); + p.push(msg('b'), lease('b'), handler); await p.flushNow(); - expect(batches.map((b) => b.sourceIds)).toEqual([['a'], ['b']]); + expect(batches.map((b) => b.sources.map((source) => source.messageId))).toEqual([['a'], ['b']]); }); }); diff --git a/src/__tests__/message-handler-error.test.ts b/src/__tests__/message-handler-error.test.ts index 5bf28c4..41208d6 100644 --- a/src/__tests__/message-handler-error.test.ts +++ b/src/__tests__/message-handler-error.test.ts @@ -19,6 +19,12 @@ function directMessage(messageId: string): unknown { }; } +async function flushEventLoop(): Promise { + await flushMicrotasks(); + await new Promise((resolve) => setImmediate(resolve)); + await flushMicrotasks(); +} + describe.each([ ['queue disabled', false], ['queue enabled', true], @@ -68,8 +74,74 @@ test('a throwing error observer cannot break handler cleanup', async () => { logger.error.mock.calls.some( ([entry]) => Array.isArray(entry) && - entry[0] === 'safety: error observer threw' && + entry[0] === 'channel: error handler threw' && entry[1]?.message === 'observer exploded', ), ).toBe(true); }); + +test('an async rejecting error observer is consumed without unhandledRejection', async () => { + const { ch, logger } = createTestChannel({ safety: { chatQueue: { enabled: false } } }); + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => unhandled.push(reason); + let handlerCalls = 0; + let observerCalls = 0; + process.on('unhandledRejection', onUnhandled); + + try { + markConnected(ch); + ch.on('error', async () => { + observerCalls++; + throw new Error('async observer exploded'); + }); + ch.on('message', async () => { + handlerCalls++; + if (handlerCalls === 1) throw new Error('first message exploded'); + }); + + await dispatchEvent(ch, 'im.message.receive_v1', directMessage('om_async_observer_1')); + await flushEventLoop(); + await dispatchEvent(ch, 'im.message.receive_v1', directMessage('om_async_observer_2')); + await flushEventLoop(); + + expect(observerCalls).toBe(1); + expect(handlerCalls).toBe(2); + expect(unhandled).toEqual([]); + expect( + logger.error.mock.calls.some( + ([entry]) => + Array.isArray(entry) && + entry[0] === 'channel: error handler threw' && + entry[1]?.message === 'async observer exploded', + ), + ).toBe(true); + } finally { + process.off('unhandledRejection', onUnhandled); + } +}); + +test('a throwing thenable returned by the error observer is isolated', async () => { + const { ch, logger } = createTestChannel({ safety: { chatQueue: { enabled: false } } }); + markConnected(ch); + ch.on('error', (() => + Object.defineProperty({}, 'then', { + get() { + throw new Error('then getter exploded'); + }, + })) as never); + ch.on('message', async () => { + throw new Error('message handler exploded'); + }); + + await dispatchEvent(ch, 'im.message.receive_v1', directMessage('om_thenable_observer')); + await flushEventLoop(); + + expect( + logger.error.mock.calls.some( + ([entry]) => + Array.isArray(entry) && + entry[0] === 'channel: error handler threw' && + entry[1]?.message === 'then getter exploded', + ), + ).toBe(true); +}); diff --git a/src/channel.ts b/src/channel.ts index 8c2a579..cf0c596 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -1350,8 +1350,25 @@ export class LarkChannel { cause: e, }); const handler = this.handlers.error; - if (handler) handler(err); - else this.logger.error?.('channel: unhandled error', err); + if (!handler) { + this.logger.error?.('channel: unhandled error', err); + return; + } + try { + void Promise.resolve(handler(err)).catch((observerError) => { + this.logErrorObserverFailure(observerError); + }); + } catch (observerError) { + this.logErrorObserverFailure(observerError); + } + } + + private logErrorObserverFailure(observerError: unknown): void { + try { + this.logger.error?.('channel: error handler threw', observerError); + } catch { + /* an observer failure must never escape through its logger */ + } } } diff --git a/src/safety/__tests__/chat-pipeline.test.ts b/src/safety/__tests__/chat-pipeline.test.ts index 3e7d04a..9fd0043 100644 --- a/src/safety/__tests__/chat-pipeline.test.ts +++ b/src/safety/__tests__/chat-pipeline.test.ts @@ -1,7 +1,12 @@ import type { NormalizedMessage } from '../../types'; import { ChatPipeline, ChatPipelineManager } from '../chat-pipeline'; +import type { ProcessingLease } from '../processing-lock'; import { DEFAULT_BATCH } from '../types'; +function lease(id: string): ProcessingLease { + return Object.freeze({ id, ownerToken: Symbol(id) }); +} + function makeMsg(id: string, content: string, chatId = 'oc_test'): NormalizedMessage { return { messageId: id, @@ -24,42 +29,45 @@ async function flushTimers(ms: number): Promise { describe('ChatPipeline — push (batching)', () => { test('single message flushes after debounce delay', async () => { - const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = []; + const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> = + []; const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 50 }, false); - p.push(makeMsg('m1', 'hello'), async (b) => { + p.push(makeMsg('m1', 'hello'), lease('m1'), async (b) => { flushes.push(b); }); await flushTimers(80); expect(flushes).toHaveLength(1); - expect(flushes[0].sourceIds).toEqual(['m1']); + expect(flushes[0].sources.map((source) => source.messageId)).toEqual(['m1']); expect(flushes[0].message.content).toBe('hello'); }); test('rapid messages within window merge into one batch', async () => { - const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = []; + const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> = + []; const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 50 }, false); - p.push(makeMsg('m1', 'hello'), async (b) => { + p.push(makeMsg('m1', 'hello'), lease('m1'), async (b) => { flushes.push(b); }); - p.push(makeMsg('m2', 'world'), async (b) => { + p.push(makeMsg('m2', 'world'), lease('m2'), async (b) => { flushes.push(b); }); - p.push(makeMsg('m3', 'foo'), async (b) => { + p.push(makeMsg('m3', 'foo'), lease('m3'), async (b) => { flushes.push(b); }); await flushTimers(100); expect(flushes).toHaveLength(1); - expect(flushes[0].sourceIds).toEqual(['m1', 'm2', 'm3']); + expect(flushes[0].sources.map((source) => source.messageId)).toEqual(['m1', 'm2', 'm3']); expect(flushes[0].message.content).toBe('hello\n\nworld\n\nfoo'); }); test('maxMessages forces flush', async () => { - const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = []; + const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> = + []; const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 10_000, maxMessages: 2 }, false); - p.push(makeMsg('m1', 'a'), async (b) => { + p.push(makeMsg('m1', 'a'), lease('m1'), async (b) => { flushes.push(b); }); - p.push(makeMsg('m2', 'b'), async (b) => { + p.push(makeMsg('m2', 'b'), lease('m2'), async (b) => { flushes.push(b); }); await flushTimers(50); @@ -67,9 +75,10 @@ describe('ChatPipeline — push (batching)', () => { }); test('maxChars forces flush', async () => { - const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = []; + const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> = + []; const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 10_000, maxChars: 5 }, false); - p.push(makeMsg('m1', 'hello'), async (b) => { + p.push(makeMsg('m1', 'hello'), lease('m1'), async (b) => { flushes.push(b); }); await flushTimers(50); @@ -77,9 +86,10 @@ describe('ChatPipeline — push (batching)', () => { }); test('serial-only mode flushes immediately', async () => { - const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = []; + const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> = + []; const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 1000 }, true); - p.push(makeMsg('m1', 'a'), async (b) => { + p.push(makeMsg('m1', 'a'), lease('m1'), async (b) => { flushes.push(b); }); await flushTimers(20); @@ -98,9 +108,9 @@ describe('ChatPipeline — serialization', () => { order.push(`${id}-end`); }; - p.push(makeMsg('m1', 'a'), slow('m1', 40)); + p.push(makeMsg('m1', 'a'), lease('m1'), slow('m1', 40)); await flushTimers(30); - p.push(makeMsg('m2', 'b'), slow('m2', 20)); + p.push(makeMsg('m2', 'b'), lease('m2'), slow('m2', 20)); await flushTimers(200); // m1 must complete before m2 starts @@ -115,7 +125,7 @@ describe('ChatPipeline — serialization', () => { const order: string[] = []; const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 30 }, false); - p.push(makeMsg('m1', 'a'), async () => { + p.push(makeMsg('m1', 'a'), lease('m1'), async () => { await flushTimers(50); order.push('batch'); }); @@ -147,7 +157,7 @@ describe('ChatPipelineManager', () => { const mgr = new ChatPipelineManager({ ...DEFAULT_BATCH, delayMs: 10 }); const order: string[] = []; - mgr.push('A', makeMsg('m1', 'a'), async () => { + mgr.push('A', makeMsg('m1', 'a'), lease('m1'), async () => { await flushTimers(30); order.push('push'); }); diff --git a/src/safety/__tests__/processing-lock.test.ts b/src/safety/__tests__/processing-lock.test.ts index 1efa53c..538d172 100644 --- a/src/safety/__tests__/processing-lock.test.ts +++ b/src/safety/__tests__/processing-lock.test.ts @@ -1,4 +1,9 @@ -import { ProcessingLock } from '../processing-lock'; +import { type ProcessingLease, ProcessingLock } from '../processing-lock'; + +function expectLease(lease: ProcessingLease | undefined): ProcessingLease { + expect(lease).toBeDefined(); + return lease as ProcessingLease; +} describe('ProcessingLock', () => { let lock: ProcessingLock; @@ -7,38 +12,114 @@ describe('ProcessingLock', () => { vi.useRealTimers(); }); - test('first acquire succeeds, second fails until release', () => { + test('first acquire returns a lease and competing acquire fails until exact release', () => { lock = new ProcessingLock(); - expect(lock.acquire('m1')).toBe(true); - expect(lock.acquire('m1')).toBe(false); - lock.release('m1'); - expect(lock.acquire('m1')).toBe(true); + const first = expectLease(lock.acquire('m1')); + expect(first.id).toBe('m1'); + expect(typeof first.ownerToken).toBe('symbol'); + expect(lock.acquire('m1')).toBeUndefined(); + lock.release(first); + expect(lock.acquire('m1')).toBeDefined(); }); test('different ids are independent', () => { lock = new ProcessingLock(); - expect(lock.acquire('m1')).toBe(true); - expect(lock.acquire('m2')).toBe(true); - expect(lock.acquire('m1')).toBe(false); - expect(lock.acquire('m2')).toBe(false); + expect(lock.acquire('m1')).toBeDefined(); + expect(lock.acquire('m2')).toBeDefined(); + expect(lock.acquire('m1')).toBeUndefined(); + expect(lock.acquire('m2')).toBeUndefined(); }); test('renews an acquired lease across multiple original ttl periods', async () => { vi.useFakeTimers(); lock = new ProcessingLock(50, 10); - expect(lock.acquire('m1')).toBe(true); - expect(lock.acquire('m1')).toBe(false); + const lease = expectLease(lock.acquire('m1')); await vi.advanceTimersByTimeAsync(120); - expect(lock.acquire('m1')).toBe(false); - - lock.stopRenewal('m1'); + expect(lock.acquire('m1')).toBeUndefined(); + lock.stopRenewal(lease); await vi.advanceTimersByTimeAsync(51); - expect(lock.acquire('m1')).toBe(true); + expect(lock.acquire('m1')).toBeUndefined(); + lock.release(lease); + expect(lock.acquire('m1')).toBeDefined(); + }); + + test('stale owner release cannot delete a replacement lease', () => { + lock = new ProcessingLock(); + const stale = expectLease(lock.acquire('m1')); + lock.release(stale); + const current = expectLease(lock.acquire('m1')); + + lock.release(stale); + + expect(lock.acquire('m1')).toBeUndefined(); + lock.release(current); + }); + + test('stale owner stopRenewal cannot stop a replacement lease', () => { + vi.useFakeTimers(); + lock = new ProcessingLock(50, 10); + const stale = expectLease(lock.acquire('m1')); + lock.release(stale); + const current = expectLease(lock.acquire('m1')); + + lock.stopRenewal(stale); + + expect((lock as any).locks.get('m1').state).toBe('active'); + expect(vi.getTimerCount()).toBe(1); + expect(lock.acquire('m1')).toBeUndefined(); + lock.release(current); + }); + + test('active owner remains exclusive after fake clock passes ttl', () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + lock = new ProcessingLock(50, 10); + const lease = expectLease(lock.acquire('m1')); + + vi.setSystemTime(2_000); + + expect(lock.acquire('m1')).toBeUndefined(); + lock.release(lease); + }); + + test('multiple active ids renew independently', async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + lock = new ProcessingLock(50, 10); + const first = expectLease(lock.acquire('m1')); + const second = expectLease(lock.acquire('m2')); + await vi.advanceTimersByTimeAsync(10); + const firstExpiry = (lock as any).locks.get('m1').expiresAt; + const secondExpiry = (lock as any).locks.get('m2').expiresAt; + + lock.stopRenewal(first); + await vi.advanceTimersByTimeAsync(20); + + expect((lock as any).locks.get('m1').expiresAt).toBe(firstExpiry); + expect((lock as any).locks.get('m2').expiresAt).toBeGreaterThan(secondExpiry); + expect(lock.acquire('m1')).toBeUndefined(); + expect(lock.acquire('m2')).toBeUndefined(); + lock.release(first); + lock.release(second); + }); + + test('finalizing lease remains exclusive until exact release', () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + lock = new ProcessingLock(50, 10); + const lease = expectLease(lock.acquire('m1')); + lock.stopRenewal(lease); + vi.setSystemTime(2_000); + + expect(lock.acquire('m1')).toBeUndefined(); + lock.release(lease); + expect(lock.acquire('m1')).toBeDefined(); }); - test('release of non-held id is a no-op', () => { + test('release of a non-current lease is a no-op', () => { lock = new ProcessingLock(); - expect(() => lock.release('unknown')).not.toThrow(); + const unknown = Object.freeze({ id: 'unknown', ownerToken: Symbol('unknown') }); + expect(() => lock.release(unknown)).not.toThrow(); }); test('dispose clears the renewal timer', () => { @@ -54,10 +135,14 @@ describe('ProcessingLock', () => { [0, 1], [-1, 1], [Number.NaN, 1], + [Number.POSITIVE_INFINITY, 1], + [50.5, 10], + [50, 10.5], + [2_147_483_648, 10], + [2_147_483_647, 2_147_483_648], [50, 0], [50, 50], [50, 60], - [3_000_000_000, 2_147_483_648], ])('invalid ttl/renew config fails loudly (%s, %s)', (ttlMs, renewIntervalMs) => { expect(() => new ProcessingLock(ttlMs, renewIntervalMs)).toThrow(RangeError); }); diff --git a/src/safety/__tests__/renewable-lock-pipeline.test.ts b/src/safety/__tests__/renewable-lock-pipeline.test.ts index 5856119..bf03309 100644 --- a/src/safety/__tests__/renewable-lock-pipeline.test.ts +++ b/src/safety/__tests__/renewable-lock-pipeline.test.ts @@ -9,6 +9,8 @@ const logger = { trace: vi.fn(), } as any; +afterEach(() => vi.useRealTimers()); + function deferred(): { promise: Promise; resolve: () => void } { let resolve!: () => void; const promise = new Promise((done) => { @@ -77,8 +79,9 @@ test('handler pending across two ttl periods retains lock, then marks seen and r gate.resolve(); await wait(0); expect(values.has('om_renew')).toBe(true); - expect((pipeline as any).lock.acquire('om_renew')).toBe(true); - (pipeline as any).lock.release('om_renew'); + const reacquired = (pipeline as any).lock.acquire('om_renew'); + expect(reacquired).toBeDefined(); + (pipeline as any).lock.release(reacquired); await pipeline.pushMessage(message('om_renew')); await wait(0); @@ -86,6 +89,112 @@ test('handler pending across two ttl periods retains lock, then marks seen and r await pipeline.dispose(); }); +test('event-loop stall beyond ttl cannot create a second handler lease', async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const gate = deferred(); + const { cache } = makeCache(); + let handlerCalls = 0; + const pipeline = new SafetyPipeline({ + cache, + logger, + onReject: () => {}, + onMessage: async () => { + handlerCalls++; + await gate.promise; + }, + config: { + chatQueue: { enabled: false }, + processingLock: { ttlMs: 20, renewIntervalMs: 5 }, + }, + }); + + await pipeline.pushMessage(message('om_stalled')); + await Promise.resolve(); + expect(handlerCalls).toBe(1); + + // Move wall clock beyond the lease TTL without executing any timer callback. + vi.setSystemTime(2_000); + await pipeline.pushMessage(message('om_stalled')); + expect(handlerCalls).toBe(1); + + gate.resolve(); + await Promise.resolve(); + await Promise.resolve(); + await pipeline.dispose(); + vi.useRealTimers(); +}); + +test('batching carries each source exact lease through cleanup', async () => { + const { cache } = makeCache(); + const markSeen = vi.spyOn(cache, 'set'); + const pipeline = new SafetyPipeline({ + cache, + logger, + onReject: () => {}, + onMessage: async () => {}, + config: { + chatQueue: { enabled: true }, + batch: { text: { delayMs: 10_000, maxMessages: 2 } }, + processingLock: { ttlMs: 50, renewIntervalMs: 10 }, + }, + }); + const lock = (pipeline as any).lock; + const acquire = vi.spyOn(lock, 'acquire'); + const stopRenewal = vi.spyOn(lock, 'stopRenewal'); + const release = vi.spyOn(lock, 'release'); + + await pipeline.pushMessage(message('om_batch_1')); + await pipeline.pushMessage(message('om_batch_2')); + await (pipeline as any).manager.flushAll(); + + const firstLease = acquire.mock.results[0].value; + const secondLease = acquire.mock.results[1].value; + expect(stopRenewal.mock.calls[0][0]).toBe(firstLease); + expect(stopRenewal.mock.calls[1][0]).toBe(secondLease); + expect(release.mock.calls[0][0]).toBe(firstLease); + expect(release.mock.calls[1][0]).toBe(secondLease); + expect(stopRenewal.mock.invocationCallOrder[0]).toBeLessThan( + markSeen.mock.invocationCallOrder[0], + ); + expect(markSeen.mock.invocationCallOrder[0]).toBeLessThan(release.mock.invocationCallOrder[0]); + expect(stopRenewal.mock.invocationCallOrder[1]).toBeLessThan( + markSeen.mock.invocationCallOrder[1], + ); + expect(markSeen.mock.invocationCallOrder[1]).toBeLessThan(release.mock.invocationCallOrder[1]); + await pipeline.dispose(); +}); + +test('pushAction finalizes and releases the exact acquired lease', async () => { + const { cache } = makeCache(); + const markSeen = vi.spyOn(cache, 'set'); + const pipeline = new SafetyPipeline({ + cache, + logger, + onReject: () => {}, + onMessage: async () => {}, + config: { + chatQueue: { enabled: false }, + processingLock: { ttlMs: 50, renewIntervalMs: 10 }, + }, + }); + const lock = (pipeline as any).lock; + const acquire = vi.spyOn(lock, 'acquire'); + const stopRenewal = vi.spyOn(lock, 'stopRenewal'); + const release = vi.spyOn(lock, 'release'); + + await pipeline.pushAction('action_exact', 'oc_action', async () => 'ok'); + + const lease = acquire.mock.results[0].value; + expect(stopRenewal.mock.calls[0][0]).toBe(lease); + expect(release.mock.calls[0][0]).toBe(lease); + expect(stopRenewal.mock.invocationCallOrder[0]).toBeLessThan( + markSeen.mock.invocationCallOrder[0], + ); + expect(markSeen.mock.invocationCallOrder[0]).toBeLessThan(release.mock.invocationCallOrder[0]); + await pipeline.dispose(); +}); + test('lease stays renewable while a queued message waits behind another handler', async () => { const firstGate = deferred(); const { cache } = makeCache(); @@ -104,6 +213,10 @@ test('lease stays renewable while a queued message waits behind another handler' processingLock: { ttlMs: 20, renewIntervalMs: 5 }, }, }); + const lock = (pipeline as any).lock; + const acquire = vi.spyOn(lock, 'acquire'); + const stopRenewal = vi.spyOn(lock, 'stopRenewal'); + const release = vi.spyOn(lock, 'release'); await pipeline.pushMessage(message('om_first')); await pipeline.pushMessage(message('om_waiting')); @@ -114,6 +227,12 @@ test('lease stays renewable while a queued message waits behind another handler' firstGate.resolve(); await (pipeline as any).manager.flushAll(); expect(handled).toEqual(['om_first', 'om_waiting']); + const firstLease = acquire.mock.results[0].value; + const waitingLease = acquire.mock.results[1].value; + expect(stopRenewal.mock.calls[0][0]).toBe(firstLease); + expect(stopRenewal.mock.calls[1][0]).toBe(waitingLease); + expect(release.mock.calls[0][0]).toBe(firstLease); + expect(release.mock.calls[1][0]).toBe(waitingLease); await pipeline.dispose(); }); @@ -121,6 +240,11 @@ test.each([ { ttlMs: 0, renewIntervalMs: 1 }, { ttlMs: 20, renewIntervalMs: 0 }, { ttlMs: 20, renewIntervalMs: 20 }, + { ttlMs: 20.5, renewIntervalMs: 5 }, + { ttlMs: 20, renewIntervalMs: 5.5 }, + { ttlMs: Number.POSITIVE_INFINITY, renewIntervalMs: 5 }, + { ttlMs: 2_147_483_648, renewIntervalMs: 5 }, + { ttlMs: 2_147_483_647, renewIntervalMs: 2_147_483_648 }, ])('SafetyConfig rejects invalid processing lock config: %o', (processingLock) => { const { cache } = makeCache(); expect( diff --git a/src/safety/chat-pipeline.ts b/src/safety/chat-pipeline.ts index ea881d6..c90aacf 100644 --- a/src/safety/chat-pipeline.ts +++ b/src/safety/chat-pipeline.ts @@ -1,4 +1,5 @@ import type { MentionInfo, NormalizedMessage, ResourceDescriptor } from '../types'; +import type { ProcessingLease } from './processing-lock'; import type { BatchConfig, BatchedDispatch } from './types'; type FlushHandler = (batch: BatchedDispatch) => Promise; @@ -15,7 +16,7 @@ type FlushHandler = (batch: BatchedDispatch) => Promise; * pending batch and previous tasks */ export class ChatPipeline { - private buffer: NormalizedMessage[] = []; + private buffer: Array<{ message: NormalizedMessage; lease: ProcessingLease }> = []; private bufferChars = 0; private timer?: NodeJS.Timeout; private tail: Promise = Promise.resolve(); @@ -28,8 +29,8 @@ export class ChatPipeline { private serialOnly: boolean, ) {} - push(msg: NormalizedMessage, handler: FlushHandler): void { - this.buffer.push(msg); + push(msg: NormalizedMessage, lease: ProcessingLease, handler: FlushHandler): void { + this.buffer.push({ message: msg, lease }); this.bufferChars += msg.content.length; this.pendingHandler ??= handler; @@ -117,8 +118,8 @@ export class ChatPipeline { if (!handler) return; const dispatch: BatchedDispatch = { - message: mergeBatch(batch), - sourceIds: batch.map((m) => m.messageId), + message: mergeBatch(batch.map((source) => source.message)), + sources: batch.map(({ message, lease }) => ({ messageId: message.messageId, lease })), }; this.busy = true; @@ -149,8 +150,8 @@ export class ChatPipelineManager { constructor(private config: BatchConfig) {} - push(scope: string, msg: NormalizedMessage, handler: FlushHandler): void { - this.getOrCreate(scope, false).push(msg, handler); + push(scope: string, msg: NormalizedMessage, lease: ProcessingLease, handler: FlushHandler): void { + this.getOrCreate(scope, false).push(msg, lease, handler); } run(scope: string, task: () => Promise): Promise { diff --git a/src/safety/index.ts b/src/safety/index.ts index 7d8f687..8fb7aa7 100644 --- a/src/safety/index.ts +++ b/src/safety/index.ts @@ -12,7 +12,7 @@ import { ChatPipelineManager } from './chat-pipeline'; import { SeenCache } from './dedup-cache'; import { LoopGuard } from './loop-guard'; import { PolicyGate } from './policy-gate'; -import { ProcessingLock } from './processing-lock'; +import { type ProcessingLease, ProcessingLock } from './processing-lock'; import { isStale } from './stale-detector'; import { DEFAULT_STALE_MS, @@ -26,6 +26,7 @@ export { SeenCache } from './dedup-cache'; export { LoopGuard } from './loop-guard'; export type { PolicyDecision } from './policy-gate'; export { PolicyGate } from './policy-gate'; +export type { ProcessingLease } from './processing-lock'; export { ProcessingLock } from './processing-lock'; export { isStale } from './stale-detector'; @@ -124,12 +125,16 @@ export class SafetyPipeline { return; } - if (!this.lock.acquire(msg.messageId)) { + const lease = this.lock.acquire(msg.messageId); + if (!lease) { this.logger.debug?.(`safety: drop in-flight message ${msg.messageId}`); return; } - const dispatchHandler = async (batch: { message: NormalizedMessage; sourceIds: string[] }) => { + const dispatchHandler = async (batch: { + message: NormalizedMessage; + sources: Array<{ messageId: string; lease: ProcessingLease }>; + }) => { try { await this.onMessage(batch.message); } catch (e) { @@ -140,23 +145,23 @@ export class SafetyPipeline { this.logger.error?.(`safety: error observer threw`, observerError); } } finally { - for (const id of batch.sourceIds) { - this.lock.stopRenewal(id); + for (const source of batch.sources) { + this.lock.stopRenewal(source.lease); try { - await this.seenCache.add(id); + await this.seenCache.add(source.messageId); } catch { /* best effort */ } - this.lock.release(id); + this.lock.release(source.lease); } } }; if (this.queueEnabled) { - this.manager.push(msg.chatId, msg, dispatchHandler); + this.manager.push(msg.chatId, msg, lease, dispatchHandler); } else { // queueing disabled: fire-and-forget, no batch either - void dispatchHandler({ message: msg, sourceIds: [msg.messageId] }); + void dispatchHandler({ message: msg, sources: [{ messageId: msg.messageId, lease }] }); } } @@ -171,7 +176,8 @@ export class SafetyPipeline { this.logger.debug?.(`safety: drop duplicate action ${eventId}`); return undefined; } - if (!this.lock.acquire(eventId)) { + const lease = this.lock.acquire(eventId); + if (!lease) { this.logger.debug?.(`safety: drop in-flight action ${eventId}`); return undefined; } @@ -187,13 +193,13 @@ export class SafetyPipeline { this.logger.error?.(`safety: action handler threw`, e); return undefined; } finally { - this.lock.stopRenewal(eventId); + this.lock.stopRenewal(lease); try { await this.seenCache.add(eventId); } catch { /* best effort */ } - this.lock.release(eventId); + this.lock.release(lease); } }; diff --git a/src/safety/processing-lock.ts b/src/safety/processing-lock.ts index a4dcc55..da011f2 100644 --- a/src/safety/processing-lock.ts +++ b/src/safety/processing-lock.ts @@ -1,8 +1,14 @@ import { DEFAULT_LOCK_RENEW_INTERVAL_MS, DEFAULT_LOCK_TTL_MS } from './types'; +export interface ProcessingLease { + readonly id: string; + readonly ownerToken: symbol; +} + interface LockEntry { + lease: ProcessingLease; expiresAt: number; - renewable: boolean; + state: 'active' | 'finalizing'; } const MAX_TIMER_DELAY_MS = 2_147_483_647; @@ -10,7 +16,8 @@ const MAX_TIMER_DELAY_MS = 2_147_483_647; /** * Short-TTL in-memory lock to prevent concurrent processing of the same * event — complements SeenCache by covering the "currently in flight" - * window, during which the event is not yet committed to SeenCache. + * window, during which the event is not yet committed to SeenCache. TTL is a + * renewal deadline, never authority to steal an active or finalizing owner. */ export class ProcessingLock { private readonly locks = new Map(); @@ -20,10 +27,10 @@ export class ProcessingLock { constructor( ttlMs: number = DEFAULT_LOCK_TTL_MS, - renewIntervalMs: number = Math.min(DEFAULT_LOCK_RENEW_INTERVAL_MS, ttlMs / 3), + renewIntervalMs: number = defaultRenewInterval(ttlMs), ) { assertDuration('processingLock.ttlMs', ttlMs); - assertDuration('processingLock.renewIntervalMs', renewIntervalMs, MAX_TIMER_DELAY_MS); + assertDuration('processingLock.renewIntervalMs', renewIntervalMs); if (renewIntervalMs >= ttlMs) { throw new RangeError('processingLock.renewIntervalMs must be less than processingLock.ttlMs'); } @@ -35,27 +42,32 @@ export class ProcessingLock { * Acquire a renewable lease. The lease remains live until `stopRenewal` is * called, even when its handler is waiting in a queue or batch. */ - acquire(id: string): boolean { + acquire(id: string): ProcessingLease | undefined { const now = Date.now(); - const current = this.locks.get(id); - if (current && current.expiresAt > now) return false; - this.locks.set(id, { expiresAt: now + this.ttlMs, renewable: true }); + if (this.locks.has(id)) return undefined; + const lease = Object.freeze({ id, ownerToken: Symbol(id) }); + this.locks.set(id, { lease, expiresAt: now + this.ttlMs, state: 'active' }); this.ensureRenewalTimer(); - return true; + return lease; } /** Stop extending a lease while leaving it held until explicit release. */ - stopRenewal(id: string): void { - const entry = this.locks.get(id); - if (entry) entry.renewable = false; + stopRenewal(lease: ProcessingLease): void { + const entry = this.currentEntry(lease); + if (entry) entry.state = 'finalizing'; this.stopTimerIfIdle(); } - release(id: string): void { - this.locks.delete(id); + release(lease: ProcessingLease): void { + if (this.currentEntry(lease)) this.locks.delete(lease.id); this.stopTimerIfIdle(); } + private currentEntry(lease: ProcessingLease): LockEntry | undefined { + const entry = this.locks.get(lease.id); + return entry?.lease.ownerToken === lease.ownerToken ? entry : undefined; + } + private ensureRenewalTimer(): void { if (this.renewalTimer) return; this.renewalTimer = setInterval(() => this.renew(), this.renewIntervalMs); @@ -64,9 +76,8 @@ export class ProcessingLock { private renew(): void { const now = Date.now(); - for (const [id, entry] of this.locks) { - if (entry.renewable) entry.expiresAt = now + this.ttlMs; - else if (entry.expiresAt <= now) this.locks.delete(id); + for (const entry of this.locks.values()) { + if (entry.state === 'active') entry.expiresAt = now + this.ttlMs; } this.stopTimerIfIdle(); } @@ -74,7 +85,7 @@ export class ProcessingLock { private stopTimerIfIdle(): void { if (!this.renewalTimer) return; for (const entry of this.locks.values()) { - if (entry.renewable) return; + if (entry.state === 'active') return; } clearInterval(this.renewalTimer); this.renewalTimer = undefined; @@ -87,8 +98,14 @@ export class ProcessingLock { } } -function assertDuration(name: string, value: number, max = Number.MAX_SAFE_INTEGER): void { - if (!Number.isFinite(value) || value < 1 || value > max) { - throw new RangeError(`${name} must be between 1 and ${max} milliseconds`); +function defaultRenewInterval(ttlMs: number): number { + return Math.min(DEFAULT_LOCK_RENEW_INTERVAL_MS, Math.max(1, Math.floor(ttlMs / 3))); +} + +function assertDuration(name: string, value: number): void { + if (!Number.isSafeInteger(value) || value < 1 || value > MAX_TIMER_DELAY_MS) { + throw new RangeError( + `${name} must be a safe integer between 1 and ${MAX_TIMER_DELAY_MS} milliseconds`, + ); } } diff --git a/src/safety/types.ts b/src/safety/types.ts index 9902575..6c35ed1 100644 --- a/src/safety/types.ts +++ b/src/safety/types.ts @@ -1,4 +1,5 @@ import type { NormalizedMessage, RejectEvent, SafetyConfig } from '../types'; +import type { ProcessingLease } from './processing-lock'; export interface BatchConfig { delayMs: number; @@ -33,7 +34,7 @@ export const DEFAULT_LOCK_RENEW_INTERVAL_MS = 60_000; export interface BatchedDispatch { message: NormalizedMessage; - sourceIds: string[]; + sources: Array<{ messageId: string; lease: ProcessingLease }>; } export type OnReject = (evt: RejectEvent) => void; diff --git a/src/types.ts b/src/types.ts index cce5a0b..6983c66 100644 --- a/src/types.ts +++ b/src/types.ts @@ -171,7 +171,7 @@ export interface EventMap { reaction: (evt: ReactionEvent) => void; botAdded: (evt: BotAddedEvent) => void; comment: (evt: CommentEvent) => void | Promise; - error: (err: LarkChannelError) => void; + error: (err: LarkChannelError) => void | Promise; reconnecting: () => void; reconnected: () => void; } @@ -451,9 +451,16 @@ export interface SafetyConfig { sweepIntervalMs?: number; }; processingLock?: { - /** How long an in-flight lease remains valid without renewal. */ + /** + * Lease TTL in integer milliseconds (1..2,147,483,647). Defaults to + * 300,000. Wall-clock expiry never lets a competing owner steal an active + * or finalizing lease. + */ ttlMs?: number; - /** How often an active lease is renewed. Must be less than `ttlMs`. */ + /** + * Renewal interval in integer milliseconds (1..2,147,483,647). Defaults + * to 60,000 with the default TTL and must be less than `ttlMs`. + */ renewIntervalMs?: number; }; chatQueue?: {