mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
fix(server): avoid duplicate monitor recovery wakes
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
fe2a243b9c
commit
ce301686a6
2 files changed
+50
-11
No files matched your search
@@ -1377,6 +1377,32 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("does not retry a lost monitor dispatch while another monitor wake remains scheduled", async () => {
|
||||
const { companyId, runId, issueId } = await seedRunFixture({
|
||||
adapterType: "openclaw_gateway",
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
contextSnapshot: {
|
||||
wakeReason: "issue_monitor_due",
|
||||
},
|
||||
});
|
||||
await db
|
||||
.update(issues)
|
||||
.set({ monitorNextCheckAt: new Date("2026-03-19T00:00:00.000Z") })
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, companyId)));
|
||||
|
||||
const heartbeat = heartbeatService(db);
|
||||
const result = await heartbeat.reapOrphanedRuns();
|
||||
|
||||
expect(result).toEqual({ reaped: 1, runIds: [runId] });
|
||||
const retries = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.retryOfRunId, runId)));
|
||||
expect(retries).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("interrupts running runs on graceful shutdown and queues restart recovery without recording a failure", async () => {
|
||||
const { agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({
|
||||
agentStatus: "running",
|
||||
|
||||
@@ -10923,6 +10923,29 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
.innerJoin(agents, eq(heartbeatRuns.agentId, agents.id))
|
||||
.where(eq(heartbeatRuns.status, "running"));
|
||||
|
||||
const monitorIssueIds = [...new Set(activeRuns.flatMap(({ run }) => {
|
||||
const runContext = parseObject(run.contextSnapshot);
|
||||
if (readNonEmptyString(runContext.wakeReason) !== "issue_monitor_due") return [];
|
||||
const issueId = readNonEmptyString(runContext.issueId);
|
||||
return issueId ? [issueId] : [];
|
||||
}))];
|
||||
const monitorIssues = monitorIssueIds.length > 0
|
||||
? await db
|
||||
.select({
|
||||
id: issues.id,
|
||||
companyId: issues.companyId,
|
||||
monitorNextCheckAt: issues.monitorNextCheckAt,
|
||||
})
|
||||
.from(issues)
|
||||
.where(inArray(issues.id, monitorIssueIds))
|
||||
: [];
|
||||
const monitorNextCheckAtByIssue = new Map(
|
||||
monitorIssues.map((issue) => [
|
||||
`${issue.companyId}:${issue.id}`,
|
||||
issue.monitorNextCheckAt,
|
||||
]),
|
||||
);
|
||||
|
||||
const reaped: string[] = [];
|
||||
|
||||
for (const { run, adapterType, adapterConfig } of activeRuns) {
|
||||
@@ -10973,17 +10996,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const monitorDispatchLostWithoutFutureWake =
|
||||
readNonEmptyString(runContext.wakeReason) === "issue_monitor_due" &&
|
||||
Boolean(monitorIssueId) &&
|
||||
await db
|
||||
.select({ monitorNextCheckAt: issues.monitorNextCheckAt })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.id, monitorIssueId!), eq(issues.companyId, run.companyId)))
|
||||
.limit(1)
|
||||
.then((rows) => {
|
||||
const issue = rows[0];
|
||||
if (!issue) return false;
|
||||
const nextCheckAt = issue.monitorNextCheckAt;
|
||||
return !nextCheckAt || nextCheckAt.getTime() <= now.getTime();
|
||||
});
|
||||
monitorNextCheckAtByIssue.get(`${run.companyId}:${monitorIssueId}`) === null;
|
||||
const shouldRetry = (run.processLossRetryCount ?? 0) < 1 && (
|
||||
(tracksLocalChild && (!!run.processPid || !!run.processGroupId)) ||
|
||||
monitorDispatchLostWithoutFutureWake
|
||||
|
||||
Reference in new issue
Block a user