Archived
计划修死企业微信那条路的游标丢信:截断时不许把游标推到窗口上沿,且滤重要在截断之前
Task 9 的评审点过「批末无条件覆盖游标、没有单调性护栏」,当时两个适配器还没落地、判不了实情。 照原计划的 wecom pull 读下来是两个真缺陷,都只在积压超过一轮预算(50 封)时现形: ① 被 limit 截断时仍把游标推到窗口上沿 —— 没读到的那几封从此跳过去了; 重叠窗只兜 10 分钟,兜不住一次积压(停机重启、首次回填)。 ② 滤重(known)发生在截断**之后** —— 一个窗口里若前 50 条恰好都已入库,这一轮交出 0 封, 而引擎见空批即收工、游标又没动,下一轮还是这 0 封,永远吃不到后面的新信。 改法:滤重提到分页循环里(每页先滤再数够不够),游标只在把窗口翻完时才推进, 截断时原样退回 —— 已入库的下一轮被 known 滤掉,每轮照旧往前吃一批,不卡也不丢。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr
This commit is contained in:
@@ -5115,9 +5115,9 @@ describe('pull', () => {
|
||||
});
|
||||
const provider = createWecomProvider({ fetch });
|
||||
const cursor = wecomCursor(Math.floor(now.getTime() / 1000) - 1000);
|
||||
const result = await provider.pull(account, cursor, { limit: 10, now, backfillFrom: null, known: async () => new Set(['m2']) });
|
||||
const result = await provider.pull(account, cursor, { limit: 10, now, backfillFrom: null, known: async (refs) => new Set(refs.filter((r) => r === 'm2')) });
|
||||
const listCall = calls.find((c) => c.url.includes('get_mail_list'))?.body;
|
||||
expect(listCall).toMatchObject({ begin_time: cursor.lastTime - 600, end_time: Math.floor(now.getTime() / 1000), limit: 10 });
|
||||
expect(listCall).toMatchObject({ begin_time: cursor.lastTime - 600, end_time: Math.floor(now.getTime() / 1000) });
|
||||
expect(result.messages.map((m) => m.providerRef)).toEqual(['m1', 'm3']);
|
||||
expect(new TextDecoder().decode(result.messages[1]?.eml)).toBe(EML);
|
||||
expect(calls.filter((c) => c.url.includes('read_mail')).map((c) => c.body['mail_id'])).toEqual(['m1', 'm3']);
|
||||
@@ -5131,9 +5131,12 @@ describe('pull', () => {
|
||||
'get_mail_list': () => ({ errcode: 0, next_cursor: 'c2', has_more: 1, mail_list: [{ mail_id: 'm1' }, { mail_id: 'm2' }] }),
|
||||
'read_mail': () => ({ errcode: 0, mail_data: EML }),
|
||||
});
|
||||
const result = await createWecomProvider({ fetch }).pull(account, wecomCursor(1), { limit: 2, now, backfillFrom: null, known });
|
||||
const cursor = wecomCursor(1);
|
||||
const result = await createWecomProvider({ fetch }).pull(account, cursor, { limit: 2, now, backfillFrom: null, known });
|
||||
expect(result.messages).toHaveLength(2);
|
||||
expect(result.more).toBe(true);
|
||||
// **截断时游标原样退回**:没读到的那几封不能被跳过去(重叠窗兜不住一次积压)
|
||||
expect(result.cursor).toEqual(cursor);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -5303,19 +5306,25 @@ export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => numbe
|
||||
|
||||
async pull(account, cursor, opts) {
|
||||
const { beginTime, endTime } = wecomWindow(cursor, opts.backfillFrom, opts.now);
|
||||
const ids: string[] = [];
|
||||
/*
|
||||
* **先滤库里已有的,再数够不够** —— 顺序反了会卡死:一个窗口里若前 limit 条恰好都已入库,
|
||||
* 这一轮就交出 0 封,而引擎见空批即收工、游标又没动,下一轮还是这 0 封,永远吃不到后面的新信。
|
||||
* 滤在前,每一轮都实实在在往前吃 limit 封**新**的。
|
||||
*/
|
||||
const fresh: string[] = [];
|
||||
let pageCursor: string | undefined;
|
||||
let exhausted = false;
|
||||
// 取满 limit 就停,余下的下一轮接着(窗口重叠 + 唯一键保证不丢不重)
|
||||
while (ids.length < opts.limit) {
|
||||
while (fresh.length < opts.limit) {
|
||||
const page = await withToken(account, 'get_mail_list', '/cgi-bin/exmail/app/get_mail_list', {
|
||||
begin_time: beginTime,
|
||||
end_time: endTime,
|
||||
...(pageCursor === undefined ? {} : { cursor: pageCursor }),
|
||||
limit: Math.min(LIST_PAGE, opts.limit - ids.length),
|
||||
limit: LIST_PAGE,
|
||||
});
|
||||
const list = Array.isArray(page['mail_list']) ? (page['mail_list'] as Array<{ mail_id?: unknown }>) : [];
|
||||
for (const item of list) if (typeof item.mail_id === 'string') ids.push(item.mail_id);
|
||||
const ids = list.flatMap((item) => (typeof item.mail_id === 'string' ? [item.mail_id] : []));
|
||||
const known = await opts.known(ids);
|
||||
for (const id of ids) if (!known.has(id)) fresh.push(id);
|
||||
const hasMore = page['has_more'] === 1 || page['has_more'] === true;
|
||||
pageCursor = typeof page['next_cursor'] === 'string' && page['next_cursor'] !== '' ? page['next_cursor'] : undefined;
|
||||
if (!hasMore || list.length === 0 || pageCursor === undefined) {
|
||||
@@ -5323,16 +5332,20 @@ export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => numbe
|
||||
break;
|
||||
}
|
||||
}
|
||||
const known = await opts.known(ids);
|
||||
const wanted = fresh.slice(0, opts.limit);
|
||||
const messages: RawMessage[] = [];
|
||||
for (const mailId of ids.slice(0, opts.limit)) {
|
||||
if (known.has(mailId)) continue;
|
||||
for (const mailId of wanted) {
|
||||
const mail = await withToken(account, 'read_mail', '/cgi-bin/exmail/app/read_mail', { mail_id: mailId });
|
||||
const data = mail['mail_data'];
|
||||
if (typeof data !== 'string' || data === '') continue;
|
||||
messages.push({ providerRef: mailId, eml: emlBytesOf(data), receivedAt: opts.now });
|
||||
}
|
||||
return { messages, cursor: wecomCursor(endTime), more: !exhausted };
|
||||
/*
|
||||
* 游标**只在把这个窗口翻完时**才推到窗口上沿;被 limit 截断时原样退回旧游标 ——
|
||||
* 否则没读到的那几封就被跳过去了(重叠窗只兜 10 分钟,兜不住一次积压)。
|
||||
* 退回不会卡住:已入库的下一轮被 `known` 滤掉,每轮照旧往前吃一批。
|
||||
*/
|
||||
return { messages, cursor: exhausted ? wecomCursor(endTime) : cursor, more: !exhausted || fresh.length > wanted.length };
|
||||
},
|
||||
|
||||
async send(account, message) {
|
||||
|
||||
Reference in New Issue
Block a user