Cron: drain pending writes before reading run log (#25416)

rrenamed · web-flow · commit 0cc46589acfa · 2026-03-01T07:04:04.000-06:00
* Cron: drain pending writes before reading run log

* Retrigger CI
diff --git a/src/cron/run-log.test.ts b/src/cron/run-log.test.ts
@@ -245,4 +245,30 @@ describe("cron run log", () => {
       expect(getPendingCronRunLogWriteCountForTests()).toBe(0);
     });
   });
+
+  it("read drains pending fire-and-forget writes", async () => {
+    await withRunLogDir("openclaw-cron-log-drain-", async (dir) => {
+      const logPath = path.join(dir, "runs", "job-drain.jsonl");
+
+      // Fire-and-forget write (simulates the `void appendCronRunLog(...)` pattern
+      // in server-cron.ts). Do NOT await.
+      const writePromise = appendCronRunLog(logPath, {
+        ts: 42,
+        jobId: "job-drain",
+        action: "finished",
+        status: "ok",
+        summary: "drain-test",
+      });
+      void writePromise.catch(() => undefined);
+
+      // Read should see the entry because it drains pending writes.
+      const entries = await readCronRunLogEntries(logPath, { limit: 10 });
+      expect(entries).toHaveLength(1);
+      expect(entries[0]?.ts).toBe(42);
+      expect(entries[0]?.summary).toBe("drain-test");
+
+      // Clean up
+      await writePromise.catch(() => undefined);
+    });
+  });
 });
diff --git a/src/cron/run-log.ts b/src/cron/run-log.ts
@@ -103,6 +103,14 @@ export function getPendingCronRunLogWriteCountForTests() {
   return writesByPath.size;
 }
 
+async function drainPendingWrite(filePath: string): Promise<void> {
+  const resolved = path.resolve(filePath);
+  const pending = writesByPath.get(resolved);
+  if (pending) {
+    await pending.catch(() => undefined);
+  }
+}
+
 async function pruneIfNeeded(filePath: string, opts: { maxBytes: number; keepLines: number }) {
   const stat = await fs.stat(filePath).catch(() => null);
   if (!stat || stat.size <= opts.maxBytes) {
@@ -152,6 +160,7 @@ export async function readCronRunLogEntries(
   filePath: string,
   opts?: { limit?: number; jobId?: string },
 ): Promise<CronRunLogEntry[]> {
+  await drainPendingWrite(filePath);
   const limit = Math.max(1, Math.min(5000, Math.floor(opts?.limit ?? 200)));
   const page = await readCronRunLogEntriesPage(filePath, {
     jobId: opts?.jobId,
@@ -334,6 +343,7 @@ export async function readCronRunLogEntriesPage(
   filePath: string,
   opts?: ReadCronRunLogPageOptions,
 ): Promise<CronRunLogPageResult> {
+  await drainPendingWrite(filePath);
   const limit = Math.max(1, Math.min(200, Math.floor(opts?.limit ?? 50)));
   const raw = await fs.readFile(path.resolve(filePath), "utf-8").catch(() => "");
   const statuses = normalizeRunStatuses(opts);
@@ -388,6 +398,7 @@ export async function readCronRunLogEntriesPageAll(
       nextOffset: null,
     };
   }
+  await Promise.all(jsonlFiles.map((f) => drainPendingWrite(f)));
   const chunks = await Promise.all(
     jsonlFiles.map(async (filePath) => {
       const raw = await fs.readFile(filePath, "utf-8").catch(() => "");
