mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
fix(channels): keep GitHub inbound recovery armed after a failed read (#16402)
A corrupt or unreadable inbound task store at connect and at the first poll's recovery latches the in-memory recoverable-task count at 0. The recovery flag was cleared in a finally, so a later poll never re-ran recovery once the store became readable: the persisted task never ran, and its cursor-blocking state kept the notification window held. Clear the flag only after a successful recovery pass, matching the watch-reconcile guard, so a failed read keeps the retry the log already promises.
This commit is contained in:
1 parent
7c6f93ca51
commit
5239d69cf1
2 files changed
+48
-2
No files matched your search
@@ -6816,6 +6816,50 @@ describe('GithubChannel', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('recovers a persisted task after the store is repaired (issue #16393)', async () => {
|
||||
// The store is unreadable at connect and at the first poll's recovery,
|
||||
// so the in-memory recoverable count latches at 0. Repairing the file
|
||||
// must still be picked up by a later poll: the failed recovery stays
|
||||
// armed, and the repaired task runs instead of freezing the cursor.
|
||||
class NoPollLoopChannel extends TestableGithubChannel {
|
||||
protected override startPollLoop(): void {}
|
||||
}
|
||||
channel = new NoPollLoopChannel(
|
||||
'test-github',
|
||||
makeConfig(),
|
||||
makeBridge(),
|
||||
);
|
||||
mkdirSync(join(inboundTaskPath(), '..'), { recursive: true });
|
||||
writeFileSync(inboundTaskPath(), '{corrupt', 'utf-8');
|
||||
mockOctokit.paginate.mockResolvedValue([]);
|
||||
await channel.connect();
|
||||
channel.cursor = { lastProcessedAt: '2026-07-01T00:00:00.000Z' };
|
||||
|
||||
await pollOnce();
|
||||
expect(channel.inboundEnvelopes).toHaveLength(0);
|
||||
|
||||
// The operator repairs the store while the daemon keeps running.
|
||||
writeInboundTasks([makeInboundTaskRecord()]);
|
||||
mockOctokit.paginate
|
||||
.mockResolvedValueOnce([
|
||||
makeNotification({ last_read_at: '2026-07-02T09:00:00.000Z' }),
|
||||
])
|
||||
.mockResolvedValueOnce([]);
|
||||
|
||||
await pollOnce();
|
||||
|
||||
expect(channel.inboundEnvelopes.map((env) => env.messageId)).toEqual([
|
||||
'1001',
|
||||
]);
|
||||
expect(existsSync(inboundTaskPath())).toBe(false);
|
||||
expect(
|
||||
mockOctokit.rest.activity.markNotificationsAsRead,
|
||||
).toHaveBeenCalledWith({
|
||||
last_read_at: '2026-07-02T10:00:00.000Z',
|
||||
read: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('keeps the cancelled record when cancellation resolves normally', async () => {
|
||||
await initWithoutLoop();
|
||||
const task = makeInboundTaskRecord();
|
||||
|
||||
@@ -2170,12 +2170,14 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
if (this.inboundRecoveryPending) {
|
||||
try {
|
||||
await this.recoverInboundTasks();
|
||||
// Clear only on success: the re-arm below reads an in-memory count
|
||||
// that a failed store read latched at 0, so clearing here would
|
||||
// promise a retry that never runs once the store is readable again.
|
||||
this.inboundRecoveryPending = false;
|
||||
} catch (err) {
|
||||
process.stderr.write(
|
||||
`[Channel:${this.name}] inbound task recovery failed, will retry next poll: ${sanitizeLogText(err instanceof Error ? err.message : String(err), 200)}\n`,
|
||||
);
|
||||
} finally {
|
||||
this.inboundRecoveryPending = false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user