Renew processing locks and surface handler errors

README.md

@@ -117,7 +117,15 @@ the QR URL as `source/<name>` (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`.
+
+`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
 
@@ -568,4 +576,3 @@ guarantee.
 ## License
 
 MIT
-

README.zh.md

@@ -103,7 +103,13 @@ 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`。
+
+`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 而被其他请求抢占。
 
 ### 生命周期
 

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": {

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']]);
   });
 });
 

src/__tests__/message-handler-error.test.ts

@@ -0,0 +1,147 @@
+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()),
+    },
+  };
+}
+
+async function flushEventLoop(): Promise<void> {
+  await flushMicrotasks();
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  await flushMicrotasks();
+}
+
+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] === '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);
+});

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),
     });
   }
 
@@ -1349,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 */
+    }
   }
 }