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

* 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>
2026-07-09 15:59:30 +00:00 · 2026-07-07 09:53:31 +01:00 · 2026-07-07 09:53:31 +01:00 · 13c1d3c408
commit 13c1d3c408
parent 2e4e982ff0
7 changed files with 128 additions and 70 deletions
--- 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.
--- 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",
--- 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" }),
      );
--- 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,
 }));
--- 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, {
--- 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(
--- 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();
+  });
+});