Avoid blocking run ledger liveness backfill

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/activity-service.test.ts

@@ -20,6 +20,8 @@ import { activityService } from "../services/activity.ts";
 
 const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+type ActivityService = ReturnType<typeof activityService>;
+type IssueRun = Awaited<ReturnType<ActivityService["runsForIssue"]>>[number];
 
 if (!embeddedPostgresSupport.supported) {
   console.warn(
@@ -27,6 +29,23 @@ if (!embeddedPostgresSupport.supported) {
   );
 }
 
+async function waitForIssueRun(
+  service: ActivityService,
+  companyId: string,
+  issueId: string,
+  predicate: (run: IssueRun) => boolean,
+) {
+  const deadline = Date.now() + 2_000;
+  let latestRuns: IssueRun[] = [];
+  while (Date.now() < deadline) {
+    latestRuns = await service.runsForIssue(companyId, issueId);
+    const run = latestRuns.find(predicate);
+    if (run) return { run, runs: latestRuns };
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  }
+  throw new Error(`Timed out waiting for issue run. Latest run count: ${latestRuns.length}`);
+}
+
 describeEmbeddedPostgres("activity service", () => {
   let db!: ReturnType<typeof createDb>;
   let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
@@ -212,10 +231,16 @@ describeEmbeddedPostgres("activity service", () => {
       createdAt: completedAt,
     });
 
-    const runs = await activityService(db).runsForIssue(companyId, issueId);
+    const service = activityService(db);
+    const { run, runs } = await waitForIssueRun(
+      service,
+      companyId,
+      issueId,
+      (entry) => entry.runId === runId && entry.livenessState === "completed",
+    );
 
     expect(runs).toHaveLength(1);
-    expect(runs[0]).toMatchObject({
+    expect(run).toMatchObject({
       runId,
       livenessState: "completed",
       livenessReason: "Issue is done",
@@ -341,8 +366,13 @@ describeEmbeddedPostgres("activity service", () => {
       updatedAt: createdAt,
     });
 
-    const runs = await activityService(db).runsForIssue(companyId, issueId);
-    const backfilledRun = runs.find((run) => run.runId === runId);
+    const service = activityService(db);
+    const { run: backfilledRun } = await waitForIssueRun(
+      service,
+      companyId,
+      issueId,
+      (entry) => entry.runId === runId && entry.livenessState === "plan_only",
+    );
 
     expect(backfilledRun).toMatchObject({
       runId,
@@ -442,8 +472,13 @@ describeEmbeddedPostgres("activity service", () => {
       updatedAt: createdAt,
     });
 
-    const runs = await activityService(db).runsForIssue(companyId, issueId);
-    const backfilledRun = runs.find((run) => run.runId === runId);
+    const service = activityService(db);
+    const { run: backfilledRun } = await waitForIssueRun(
+      service,
+      companyId,
+      issueId,
+      (entry) => entry.runId === runId && entry.livenessState === "plan_only",
+    );
 
     expect(backfilledRun).toMatchObject({
       runId,

server/src/services/activity.ts

@@ -13,6 +13,7 @@ import {
   workspaceOperations,
 } from "@paperclipai/db";
 import { ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY } from "@paperclipai/shared";
+import { logger } from "../middleware/logger.js";
 import { classifyRunLiveness } from "./run-liveness.js";
 
 export interface ActivityFilters {
@@ -23,6 +24,7 @@ export interface ActivityFilters {
 }
 
 export function activityService(db: Db) {
+  const scheduledLivenessBackfills = new Set<string>();
   const issueIdAsText = sql<string>`${issues.id}::text`;
   const summarizedUsageJson = sql<Record<string, unknown> | null>`
     case
@@ -298,6 +300,19 @@ export function activityService(db: Db) {
     }
   }
 
+  function scheduleRunLivenessBackfill(companyId: string, issueId: string) {
+    const key = `${companyId}:${issueId}`;
+    if (scheduledLivenessBackfills.has(key)) return;
+    scheduledLivenessBackfills.add(key);
+    void backfillMissingRunLivenessForIssue(companyId, issueId)
+      .catch((err: unknown) => {
+        logger.warn({ err, companyId, issueId }, "run liveness backfill failed");
+      })
+      .finally(() => {
+        scheduledLivenessBackfills.delete(key);
+      });
+  }
+
   return {
     list: (filters: ActivityFilters) => {
       const conditions = [eq(activityLog.companyId, filters.companyId)];
@@ -348,7 +363,7 @@ export function activityService(db: Db) {
         .orderBy(desc(activityLog.createdAt)),
 
     runsForIssue: async (companyId: string, issueId: string) => {
-      await backfillMissingRunLivenessForIssue(companyId, issueId);
+      scheduleRunLivenessBackfill(companyId, issueId);
       return db
         .select({
           runId: heartbeatRuns.id,