diff --git a/CHANGELOG.md b/CHANGELOG.md index 486db04c51c..e0ac9637763 100644 --- 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 index 289ad4f3795..10cf207dc12 100644 --- 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 index 8c7f84f098e..a9d3f9e0ea4 100644 --- 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 index 50aaa4e50ec..fd2335d87b9 100644 --- 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; + 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 index 7506a4428a9..ceef61de0b3 100644 --- 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 index 313c146f377..136d8db9def 100644 --- 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 | NodeJS.ReadableStream; +}): Promise { + // 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; @@ -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 new file mode 100644 index 00000000000..4f83462f811 --- /dev/null +++ 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(); + }); +});