fix(backup): close archive stream before retry cleanup (#101464)

steipete · ZOOWH · web-flow · commit 13c1d3c4083a · 2026-07-07T09:53:31.000+01:00
* fix(backup): close archive stream before retry cleanup Own the tar source and file writer with stream.pipeline so live-file EOF failures close the partial archive before Windows cleanup. Keep one native Windows handle-release regression and focused CI coverage. Fixes #101382 Co-authored-by: 徐闻涵0668001344 <xu.wenhan1@xydigit.com> * test(backup): track archive retry temp directory * test(backup): model tar archive streams * test(backup): model retry cleanup streams --------- Co-authored-by: 徐闻涵0668001344 <xu.wenhan1@xydigit.com>
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -27,6 +27,7 @@ Docs: https://docs.openclaw.ai
 - **Codex app-server protocol:** require app-server 0.142 or newer, remove pre-0.142 wire-shape compatibility, and teach Codex to retrieve deferred native `spawn_agent` through `tool_search` so native subagent task mirroring works on search-capable models. (#101221)
 - **Android hardware keyboard chat:** send with unmodified Enter on physical keyboards while preserving Shift+Enter and other modified Enter combinations for multiline input. (#101239) Thanks @3ninyt3nin-creator.
 - **CJK Markdown emphasis:** render adjacent Chinese, Japanese, and Korean emphasis punctuation through the shared Markdown pipeline instead of leaking literal markers across channels. (#101230, #101120) Thanks @nicknmorty.
+- **Backup retry cleanup:** close partial archive output handles and isolate each retry path after live-write failures, preventing Windows `EBUSY` locks from cascading across attempts or leaving stale temp archives. (#101397, #101449) Thanks @ZOOWH and @LiLan0125.
 - **Codex yielded native subagents:** keep the parent app-server subscription and shared client alive until yielded native subagent completion delivery settles, preventing lost wakeups and leaked one-shot cleanup.
 - **Delivery recovery pacing:** pace eligible outbound and restart-continuation replays after gateway startup so outage backlogs do not burst into channel rate limits, while preserving the wall-clock recovery budget. (#101118, #101058) Thanks @ZengWen-DT.
 - **Outbound pre-connect recovery:** clear stale platform-send evidence atomically when a connect or DNS failure proves no request was sent, allowing queued Discord and other channel messages to replay after connectivity returns without weakening the unknown-send duplicate guard. (#101024, #100979) Thanks @SunnyShu0925.
diff --git a/package.json b/package.json
@@ -1942,7 +1942,7 @@
     "test:unit:fast:audit": "node scripts/test-unit-fast-audit.mjs",
     "test:voicecall:closedloop": "node scripts/test-voicecall-closedloop.mjs",
     "test:watch": "node scripts/test-projects.mjs --watch",
-    "test:windows:ci": "node scripts/test-projects.mjs src/shared/runtime-import.test.ts src/process/exec.windows.test.ts src/process/windows-command.test.ts src/infra/windows-install-roots.test.ts src/node-host/invoke-system-run-allowlist.test.ts src/daemon/schtasks.startup-fallback.test.ts extensions/lobster/src/lobster-runner.test.ts test/scripts/format-generated-module.test.ts test/scripts/npm-runner.test.ts test/scripts/pnpm-runner.test.ts test/scripts/run-with-env.test.ts test/scripts/ui.test.ts test/scripts/vitest-process-group.test.ts",
+    "test:windows:ci": "node scripts/test-projects.mjs src/shared/runtime-import.test.ts src/process/exec.windows.test.ts src/process/windows-command.test.ts src/infra/backup-create.windows.test.ts src/infra/windows-install-roots.test.ts src/node-host/invoke-system-run-allowlist.test.ts src/daemon/schtasks.startup-fallback.test.ts extensions/lobster/src/lobster-runner.test.ts test/scripts/format-generated-module.test.ts test/scripts/npm-runner.test.ts test/scripts/pnpm-runner.test.ts test/scripts/run-with-env.test.ts test/scripts/ui.test.ts test/scripts/vitest-process-group.test.ts",
     "tool-display:check": "node --import tsx scripts/tool-display.ts --check",
     "tool-display:write": "node --import tsx scripts/tool-display.ts --write",
     "ts-topology": "node --import tsx scripts/ts-topology.ts",
diff --git a/src/commands/backup.atomic.test.ts b/src/commands/backup.atomic.test.ts
@@ -6,6 +6,7 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi }
 import { createTempHomeEnv, type TempHomeEnv } from "../test-utils/temp-home.js";
 import {
   backupVerifyCommandMock,
+  createMockTarStream,
   createBackupTestRuntime,
   mockStateOnlyBackupPlan,
   resetBackupTempHome,
@@ -70,7 +71,7 @@ describe("backupCreateCommand atomic archive write", () => {
       archivePrefix: "openclaw-backup-failure-",
     });
     try {
-      tarCreateMock.mockRejectedValueOnce(new Error("disk full"));
+      tarCreateMock.mockReturnValueOnce(createMockTarStream({ error: new Error("disk full") }));
 
       await expect(
         backupCreateCommand(runtime, {
@@ -100,22 +101,28 @@ describe("backupCreateCommand atomic archive write", () => {
       const key = String(targetPath);
       const attempt = (rmAttempts.get(key) ?? 0) + 1;
       rmAttempts.set(key, attempt);
-      if ((key === attemptFiles[0] || key === attemptFiles[1]) && attempt === 1) {
+      if (key.startsWith(`${outputPath}.`) && !attemptFiles.includes(key)) {
+        attemptFiles.push(key);
+      }
+      if (attemptFiles.length <= 2 && key === attemptFiles.at(-1) && attempt === 1) {
         throw Object.assign(new Error("resource busy"), { code: "EBUSY" });
       }
       await realRm(targetPath, options);
     }) as typeof fs.rm);
     try {
       let tarAttempt = 0;
-      tarCreateMock.mockImplementation(async ({ file }: { file: string }) => {
+      tarCreateMock.mockImplementation(() => {
         tarAttempt += 1;
-        attemptFiles.push(file);
-        await fs.writeFile(file, `archive-attempt-${tarAttempt}`, "utf8");
-        if (tarAttempt < 3) {
-          throw Object.assign(new Error("did not encounter expected EOF"), {
-            path: path.join(tempHome.home, ".openclaw", "state.txt"),
-          });
-        }
+        return createMockTarStream({
+          contents: `archive-attempt-${tarAttempt}`,
+          ...(tarAttempt < 3
+            ? {
+                error: Object.assign(new Error("did not encounter expected EOF"), {
+                  path: path.join(tempHome.home, ".openclaw", "state.txt"),
+                }),
+              }
+            : {}),
+        });
       });
 
       const result = await backupCreateCommand(runtime, {
@@ -143,9 +150,7 @@ describe("backupCreateCommand atomic archive write", () => {
     const realLink = fs.link.bind(fs);
     const linkSpy = vi.spyOn(fs, "link");
     try {
-      tarCreateMock.mockImplementationOnce(async ({ file }: { file: string }) => {
-        await fs.writeFile(file, "archive-bytes", "utf8");
-      });
+      tarCreateMock.mockReturnValueOnce(createMockTarStream());
       linkSpy.mockImplementationOnce(async (existingPath, newPath) => {
         await fs.writeFile(newPath, "concurrent-archive", "utf8");
         return await realLink(existingPath, newPath);
@@ -170,9 +175,7 @@ describe("backupCreateCommand atomic archive write", () => {
     });
     const linkSpy = vi.spyOn(fs, "link");
     try {
-      tarCreateMock.mockImplementationOnce(async ({ file }: { file: string }) => {
-        await fs.writeFile(file, "archive-bytes", "utf8");
-      });
+      tarCreateMock.mockReturnValueOnce(createMockTarStream());
       linkSpy.mockRejectedValueOnce(
         Object.assign(new Error("hard links not supported"), { code: "EOPNOTSUPP" }),
       );
diff --git a/src/commands/backup.test-support.ts b/src/commands/backup.test-support.ts
@@ -1,6 +1,7 @@
 // Backup test support provides temp config/state fixtures and mocked backup runtime helpers.
 import fs from "node:fs/promises";
 import path from "node:path";
+import { Readable } from "node:stream";
 import { vi } from "vitest";
 import type { RuntimeEnv } from "../runtime.js";
 import { deleteTestEnvValue } from "../test-utils/env.js";
@@ -14,6 +15,24 @@ const backupTestMocks = vi.hoisted(() => ({
 
 export const { backupVerifyCommandMock, tarCreateMock } = backupTestMocks;
 
+export function createMockTarStream(
+  params: {
+    beforeRead?: () => Promise<void> | void;
+    contents?: string;
+    error?: Error;
+  } = {},
+): Readable {
+  return Readable.from(
+    (async function* () {
+      await params.beforeRead?.();
+      if (params.error) {
+        throw params.error;
+      }
+      yield params.contents ?? "archive-bytes";
+    })(),
+  );
+}
+
 vi.mock("tar", () => ({
   c: backupTestMocks.tarCreateMock,
 }));
diff --git a/src/commands/backup.test.ts b/src/commands/backup.test.ts
@@ -17,6 +17,7 @@ import {
 } from "./backup-shared.js";
 import {
   backupVerifyCommandMock,
+  createMockTarStream,
   createBackupTestRuntime,
   mockStateOnlyBackupPlan,
   resetBackupTempHome,
@@ -78,9 +79,7 @@ describe("backup commands", () => {
   beforeEach(async () => {
     await resetBackupTempHome(tempHome);
     tarCreateMock.mockReset();
-    tarCreateMock.mockImplementation(async ({ file }: { file: string }) => {
-      await fs.writeFile(file, "archive-bytes", "utf8");
-    });
+    tarCreateMock.mockImplementation(() => createMockTarStream());
     backupVerifyCommandMock.mockReset();
     backupVerifyCommandMock.mockResolvedValue({
       ok: true,
@@ -268,17 +267,16 @@ describe("backup commands", () => {
         }),
       );
       tarCreateMock.mockImplementationOnce(
-        async (
-          options: { file: string; onWriteEntry?: (entry: { path: string }) => void },
-          entryPaths: string[],
-        ) => {
-          capturedManifest = JSON.parse(
-            await fs.readFile(entryPaths[0], "utf8"),
-          ) as CapturedBackupManifest;
-          capturedEntryPaths = entryPaths;
-          capturedOnWriteEntry = options.onWriteEntry ?? null;
-          await fs.writeFile(options.file, "archive-bytes", "utf8");
-        },
+        (options: { onWriteEntry?: (entry: { path: string }) => void }, entryPaths: string[]) =>
+          createMockTarStream({
+            beforeRead: async () => {
+              capturedManifest = JSON.parse(
+                await fs.readFile(entryPaths[0], "utf8"),
+              ) as CapturedBackupManifest;
+              capturedEntryPaths = entryPaths;
+              capturedOnWriteEntry = options.onWriteEntry ?? null;
+            },
+          }),
       );
       const result = await backupCreateCommand(runtime, {
         output: backupDir,
@@ -367,21 +365,20 @@ describe("backup commands", () => {
       const runtime = createBackupTestRuntime();
       await mockStateOnlyBackupPlan(stateDir);
       tarCreateMock.mockImplementationOnce(
-        async (
-          options: { file: string; filter?: (entryPath: string) => boolean },
-          entryPaths: string[],
-        ) => {
-          const manifestPath = entryPaths[0];
-          const stateRoot = entryPaths[1];
-          if (!manifestPath || !stateRoot) {
-            throw new Error("backup test expected manifest and state entries");
-          }
-          expect(options.filter?.(manifestPath)).toBe(true);
-          expect(
-            options.filter?.(path.join(stateRoot, "agents", "main", "sessions", "s.jsonl")),
-          ).toBe(false);
-          await fs.writeFile(options.file, "archive-bytes", "utf8");
-        },
+        (options: { filter?: (entryPath: string) => boolean }, entryPaths: string[]) =>
+          createMockTarStream({
+            beforeRead: () => {
+              const manifestPath = entryPaths[0];
+              const stateRoot = entryPaths[1];
+              if (!manifestPath || !stateRoot) {
+                throw new Error("backup test expected manifest and state entries");
+              }
+              expect(options.filter?.(manifestPath)).toBe(true);
+              expect(
+                options.filter?.(path.join(stateRoot, "agents", "main", "sessions", "s.jsonl")),
+              ).toBe(false);
+            },
+          }),
       );
 
       const result = await backupCreateCommand(runtime, {
diff --git a/src/infra/backup-create.ts b/src/infra/backup-create.ts
@@ -1,10 +1,11 @@
 // Creates backup archives while filtering volatile runtime state.
 import { randomUUID } from "node:crypto";
-import { constants as fsConstants, type Stats } from "node:fs";
+import { constants as fsConstants, createWriteStream, type Stats } from "node:fs";
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import type { DatabaseSync } from "node:sqlite";
+import { pipeline } from "node:stream/promises";
 import { resolveDateTimestampMs } from "@openclaw/normalization-core/number-coercion";
 import { loadSqliteVecExtension } from "../../packages/memory-host-sdk/src/engine-storage.js";
 import {
@@ -184,6 +185,15 @@ async function removeBackupTempArchiveBestEffort(tempArchivePath: string): Promi
   await fs.rm(tempArchivePath, { force: true }).catch(() => undefined);
 }
 
+async function writeArchiveStreamToFile(params: {
+  archivePath: string;
+  archiveStream: AsyncIterable<Uint8Array> | NodeJS.ReadableStream;
+}): Promise<void> {
+  // Own both stream lifecycles so a tar read error closes the output handle
+  // before retry cleanup touches the partial archive.
+  await pipeline(params.archiveStream, createWriteStream(params.archivePath));
+}
+
 async function writeTarArchiveWithRetry(params: {
   tempArchivePath: string;
   runTar: (tempArchivePath: string) => Promise<void>;
@@ -239,6 +249,7 @@ async function writeTarArchiveWithRetry(params: {
 }
 
 export const testApi = {
+  writeArchiveStreamToFile,
   writeTarArchiveWithRetry,
   isTarEofRaceError,
   createBackupVolatileStatCache,
@@ -888,30 +899,32 @@ export async function createBackupArchive(
         // cumulative skip counts across attempts instead of the final one.
         skippedVolatileCount = 0;
         unexpectedSqliteSourcePaths.length = 0;
-        await tar.c(
-          {
-            file: attemptTempArchivePath,
-            gzip: true,
-            portable: true,
-            preservePaths: true,
-            linkCache: new BackupLinkCache(),
-            statCache: createBackupVolatileStatCache(volatilePlan),
-            filter: tarFilter,
-            onWriteEntry: (entry) => {
-              entry.path = remapArchiveEntryPath({
-                entryPath: entry.path,
-                manifestPath,
-                archiveRoot,
-                sourcePathRemaps,
-              });
+        await writeArchiveStreamToFile({
+          archivePath: attemptTempArchivePath,
+          archiveStream: tar.c(
+            {
+              gzip: true,
+              portable: true,
+              preservePaths: true,
+              linkCache: new BackupLinkCache(),
+              statCache: createBackupVolatileStatCache(volatilePlan),
+              filter: tarFilter,
+              onWriteEntry: (entry) => {
+                entry.path = remapArchiveEntryPath({
+                  entryPath: entry.path,
+                  manifestPath,
+                  archiveRoot,
+                  sourcePathRemaps,
+                });
+              },
             },
-          },
-          [
-            manifestPath,
-            ...stateSqliteBackup.snapshots.map((snapshot) => snapshot.sourcePath),
-            ...result.assets.map((asset) => asset.sourcePath),
-          ],
-        );
+            [
+              manifestPath,
+              ...stateSqliteBackup.snapshots.map((snapshot) => snapshot.sourcePath),
+              ...result.assets.map((asset) => asset.sourcePath),
+            ],
+          ),
+        });
         const unexpectedSqliteSourcePath = unexpectedSqliteSourcePaths[0];
         if (unexpectedSqliteSourcePath) {
           throw new Error(
diff --git a/src/infra/backup-create.windows.test.ts b/src/infra/backup-create.windows.test.ts
@@ -0,0 +1,25 @@
+import fs from "node:fs/promises";
+import path from "node:path";
+import { PassThrough } from "node:stream";
+import { afterEach, describe, expect, it } from "vitest";
+import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
+import { testApi as backupCreateInternals } from "./backup-create.js";
+
+const tempDirs = useAutoCleanupTempDirTracker(afterEach);
+
+describe("writeArchiveStreamToFile", () => {
+  it("closes a partial archive before propagating a stream error", async () => {
+    const tempDir = tempDirs.make("openclaw-backup-stream-");
+    const archivePath = path.join(tempDir, "partial.tar.gz");
+    const archiveStream = new PassThrough();
+    const writePromise = backupCreateInternals.writeArchiveStreamToFile({
+      archivePath,
+      archiveStream,
+    });
+    archiveStream.write("partial archive");
+    archiveStream.destroy(new Error("injected tar read failure"));
+
+    await expect(writePromise).rejects.toThrow("injected tar read failure");
+    await expect(fs.rm(archivePath)).resolves.toBeUndefined();
+  });
+});
