vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-02 13:01:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(telegram): cover forward-burst debounce coalescing through durable ingress

Browse Source 
PR #115401 fixed the album split by separating durable claim lifetime from lane
occupancy (deferredLaneOccupancy="release"), and stated the same change repairs
the text/forwarded-message debounce, which defers its spooled participant the
same way. The album path got a durable-drain regression test; the debounce path
did not, because the suite that would cover it fails 10/10 on main.

Restore that suite: its harness supplies me.username without me.id, which
resolveBotUserId has required since #114532. Production grammY populates ctx.me
from getMe(), so this is harness drift, not a runtime gap.

Then extend the durable-ingress regression file to both buffers and rename it
off media-group. Two new tests admit a forwarded burst through the spool and
core drain, live and from a restart backlog, asserting one turn carrying both
notes plus independent tombstones.

Both fail with deferredLaneOccupancy reverted to "hold" (second note lands in a
separate turn), so they pin the lane contract rather than restate it.
This commit is contained in:
Ayaan Zaidi
2026-07-29 01:47:42 +02:00
parent cc8fd57a01
commit cc434dd5fe
2 changed files with 100 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
extensions/telegram/src/bot.media.downloads-media-file-path-no-file-download.e2e.test.ts
View File

@@ -202,7 +202,7 @@ describe("telegram inbound media", () => {
photo: [{ file_id: "fid" }],
date: 1736380800, // 2025-01-09T00:00:00Z
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: scenario.getFile,
});
@@ -239,7 +239,7 @@ describe("telegram inbound media", () => {
photo: [{ file_id: "fid" }],
date: 1736380800,
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "photos/1.jpg" }),
});
@@ -281,7 +281,7 @@ describe("telegram inbound media", () => {
chat: { id: 1234, type: "private" },
photo: [{ file_id: "fid" }],
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "photos/2.jpg" }),
});
@@ -342,7 +342,7 @@ describe("telegram inbound media", () => {
replySpy.mockClear();
await handler({
message: testCase.message,
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "unused" }),
});
@@ -392,7 +392,7 @@ describe("telegram media groups", () => {
media_group_id: "album-custom-api-root",
photo: [{ file_id: "photo1" }],
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "photos/photo1.jpg" }),
}),
handler({
@@ -404,7 +404,7 @@ describe("telegram media groups", () => {
media_group_id: "album-custom-api-root",
photo: [{ file_id: "photo2" }],
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "photos/photo2.jpg" }),
}),
]);
@@ -514,7 +514,7 @@ describe("telegram media groups", () => {
scenario.messages.map((message) =>
handler({
message,
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: message.filePath }),
}),
),
@@ -605,7 +605,7 @@ describe("telegram media groups", () => {
]) {
await handler({
message: message.message,
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: message.getFile,
});
}
@@ -718,7 +718,7 @@ describe("telegram media groups", () => {
]) {
await handler({
message: message.message,
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: message.getFile,
});
}
@@ -800,7 +800,7 @@ describe("telegram media groups", () => {
media_group_id: "album-shared-by-telegram",
photo: [{ file_id: "topic1photo" }],
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "photos/topic1.jpg" }),
}),
handler({
@@ -815,7 +815,7 @@ describe("telegram media groups", () => {
media_group_id: "album-shared-by-telegram",
photo: [{ file_id: "topic2photo" }],
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "photos/topic2.jpg" }),
}),
]);
@@ -885,7 +885,7 @@ describe("telegram forwarded bursts", () => {
date: 1736380800,
forward_origin: { type: "hidden_user", date: 1736380700, sender_user_name: "A" },
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({}),
});
@@ -898,7 +898,7 @@ describe("telegram forwarded bursts", () => {
photo: [{ file_id: "fwd_photo_1" }],
forward_origin: { type: "hidden_user", date: 1736380701, sender_user_name: "A" },
},
me: { username: "openclaw_bot" },
me: { id: 999, username: "openclaw_bot" },
getFile: async () => ({ file_path: "photos/fwd1.jpg" }),
});

108
extensions/telegram/src/telegram-ingress-media-group.e2e.test.ts → extensions/telegram/src/telegram-ingress-coalescing.e2e.test.ts
View File

@@ -1,4 +1,7 @@
// Telegram ingress media-group regression: durable queue → core drain → grammY → album buffer.
// Telegram ingress coalescing regression: durable queue → core drain → grammY → inbound buffer.
// Both Telegram inbound buffers (album, forward-burst debounce) defer their spooled
// participant the same way, so both depend on deferredLaneOccupancy="release" to admit
// later same-lane members. Cover them together — a lane regression breaks both at once.
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
@@ -110,6 +113,25 @@ function photoUpdate(params: { updateId: number; messageId: number; caption?: st
};
}
function forwardedTextUpdate(params: { updateId: number; messageId: number; text: string }) {
return {
update_id: params.updateId,
message: {
message_id: params.messageId,
date: 1_736_380_800 + params.messageId,
chat: { id: 111, type: "private" as const, first_name: "Ada" },
from: { id: 111, is_bot: false, first_name: "Ada" },
// forward_origin puts the entry on the forward debounce lane (80ms window).
forward_origin: {
type: "user" as const,
date: 1_736_300_000,
sender_user: { id: 555, is_bot: false, first_name: "Origin" },
},
text: params.text,
},
};
}
function createBotApiTransport(): TelegramTransport {
let getFileCall = 0;
const fetchImpl = vi.fn(async (input: RequestInfo | URL) => {
@@ -159,38 +181,42 @@ function createTelegramDeps(stateDir: string): TelegramBotDeps {
} as TelegramBotDeps;
}
async function assertAlbumTurnAndTombstones(params: {
spoolDir: string;
firstUpdateId: number;
secondUpdateId: number;
}) {
/** Both members must land in one turn; a second turn is the split this file guards. */
async function awaitSingleDownstreamTurn(): Promise<MsgContext & Record<string, unknown>> {
await vi.waitFor(
() => {
expect(downstreamTurns).toHaveBeenCalledTimes(1);
},
{ timeout: 5_000, interval: 5 },
);
const turn = downstreamTurns.mock.calls[0]?.[0] as MsgContext & Record<string, unknown>;
expect(turn.Body).toContain("Two photo album");
expect(turn.media).toMatchObject([
{ path: "/tmp/photo-1.jpg", kind: "image" },
{ path: "/tmp/photo-2.jpg", kind: "image" },
]);
return downstreamTurns.mock.calls[0]?.[0] as MsgContext & Record<string, unknown>;
}
async function assertSpoolTombstoned(params: { spoolDir: string; updateIds: number[] }) {
const queue = openTelegramIngressQueue(params.spoolDir);
await vi.waitFor(async () => {
expect(await queue.listClaims()).toEqual([]);
expect(await queue.listPending({ limit: "all" })).toEqual([]);
});
await expect(
queue.enqueue(telegramQueueEventId(params.firstUpdateId), {} as never),
).resolves.toMatchObject({ kind: "completed" });
await expect(
queue.enqueue(telegramQueueEventId(params.secondUpdateId), {} as never),
).resolves.toMatchObject({ kind: "completed" });
// Every member tombstones independently, so a replayed update cannot re-enter.
for (const updateId of params.updateIds) {
await expect(queue.enqueue(telegramQueueEventId(updateId), {} as never)).resolves.toMatchObject(
{ kind: "completed" },
);
}
}
describe("Telegram durable ingress media groups", () => {
async function assertAlbumTurnAndTombstones(params: { spoolDir: string; updateIds: number[] }) {
const turn = await awaitSingleDownstreamTurn();
expect(turn.Body).toContain("Two photo album");
expect(turn.media).toMatchObject([
{ path: "/tmp/photo-1.jpg", kind: "image" },
{ path: "/tmp/photo-2.jpg", kind: "image" },
]);
await assertSpoolTombstoned(params);
}
describe("Telegram durable ingress coalescing", () => {
const originalStateDir = process.env.OPENCLAW_STATE_DIR;
let stateDir: string;
let spoolDir: string;
@@ -279,7 +305,7 @@ describe("Telegram durable ingress media groups", () => {
setTimeout(resolve, 5);
});
await monitor.admit(second);
await assertAlbumTurnAndTombstones({ spoolDir, firstUpdateId: 101, secondUpdateId: 102 });
await assertAlbumTurnAndTombstones({ spoolDir, updateIds: [101, 102] });
await monitor.stop();
await telegramTransport.close();
@@ -293,7 +319,47 @@ describe("Telegram durable ingress media groups", () => {
const { monitor, telegramTransport } = await createMonitor();
monitor.start();
await assertAlbumTurnAndTombstones({ spoolDir, firstUpdateId: 201, secondUpdateId: 202 });
await assertAlbumTurnAndTombstones({ spoolDir, updateIds: [201, 202] });
await monitor.stop();
await telegramTransport.close();
});
it("coalesces a forwarded burst admitted a few milliseconds apart", async () => {
const { monitor, telegramTransport } = await createMonitor();
monitor.start();
await monitor.admit(forwardedTextUpdate({ updateId: 301, messageId: 1, text: "First note" }));
await new Promise((resolve) => {
setTimeout(resolve, 5);
});
await monitor.admit(forwardedTextUpdate({ updateId: 302, messageId: 2, text: "Second note" }));
const turn = await awaitSingleDownstreamTurn();
expect(turn.Body).toContain("First note");
expect(turn.Body).toContain("Second note");
await assertSpoolTombstoned({ spoolDir, updateIds: [301, 302] });
await monitor.stop();
await telegramTransport.close();
});
it("coalesces a forwarded burst replayed from a durable restart backlog", async () => {
await writeTelegramSpooledUpdate({
spoolDir,
update: forwardedTextUpdate({ updateId: 401, messageId: 1, text: "First note" }),
});
await writeTelegramSpooledUpdate({
spoolDir,
update: forwardedTextUpdate({ updateId: 402, messageId: 2, text: "Second note" }),
});
const { monitor, telegramTransport } = await createMonitor();
monitor.start();
const turn = await awaitSingleDownstreamTurn();
expect(turn.Body).toContain("First note");
expect(turn.Body).toContain("Second note");
await assertSpoolTombstoned({ spoolDir, updateIds: [401, 402] });
await monitor.stop();
await telegramTransport.close();