From 086cd0a505850d98873a63e71c710cfac0fce121 Mon Sep 17 00:00:00 2001 From: flexi767 <96955327+flexi767@users.noreply.github.com> Date: Sun, 16 Aug 2026 06:42:01 +0200 Subject: [PATCH] fix(engine): retry transient post-merge push failures (#3468) ## What - Retry post-merge target pushes after recognized transient Git transport failures. - Use a bounded schedule: the initial attempt plus two retries after 2s and 5s. - Apply the same helper to the unified fast path and the shared push path used by divergence recovery and the soft-deprecated merger. - Keep retry sleeps and subsequent attempts cancellation-aware. - Add a patch changeset and regression coverage, including a real bare remote that rejects the first push. ## Why Fusion already retries non-fast-forward races, and #1942 made terminal push failures durable, but a temporary network failure still ended post-merge delivery after one attempt. That can leave the local integration branch ahead of the remote even though retrying seconds later would succeed. The retry is deliberately provider-neutral. It uses Git error classification and normal Fusion logs only; it does not add Telegram, OpenClaw, or any other notification-vendor dependency. ## Behavior and impact - Retries only transient transport signatures such as connection resets, DNS failures, unreachable networks, selected HTTP 429/5xx RPC failures, and unexpected disconnects. - Permission, authentication, configuration, and ref-rejection errors keep their existing immediate handling. - Exhausted retries remain non-fatal to the already-landed merge and flow through the existing audit/task-log failure reporting. - Existing non-fast-forward pull/rebase recovery is unchanged apart from making its backoff cancellation-aware. ## Checks - [x] `pnpm --filter @fusion/engine exec vitest run src/__tests__/merger-ai-push-after-merge.test.ts src/__tests__/merger-prompt-and-utils.test.ts --silent=passed-only --reporter=dot` (48 tests) - [x] `pnpm --filter @fusion/engine typecheck` - [x] `pnpm lint` (0 errors; 2 pre-existing warnings) - [x] `pnpm check:changesets --strict` - [x] `pnpm check:fnxc-future-dates` - [x] `pnpm test` (changed-test gate; static checks and 688 tests passed) ## Summary by CodeRabbit * **Bug Fixes** * Improved reliability of post-merge pushes by retrying temporary Git transport failures. * Added bounded backoff between retries to prevent excessive repeated attempts. * Push retries now stop promptly when an operation is canceled. * Configuration, authentication, and ref-rejection errors continue to fail immediately. * Successful retries and canceled operations now report accurate outcomes. * **Tests** * Added coverage for successful retries, cancellation, and non-retryable failures. --------- Co-authored-by: flexi767 Co-authored-by: v --- .../fix-transient-post-merge-push-retry.md | 7 + .../merger-ai-push-after-merge.test.ts | 54 ++++++ .../__tests__/merger-prompt-and-utils.test.ts | 178 +++++++++++++++++- packages/engine/src/merge/merger-ai.ts | 17 +- packages/engine/src/merger.ts | 169 +++++++++++++---- 5 files changed, 390 insertions(+), 35 deletions(-) create mode 100644 .changeset/fix-transient-post-merge-push-retry.md diff --git a/.changeset/fix-transient-post-merge-push-retry.md b/.changeset/fix-transient-post-merge-push-retry.md new file mode 100644 index 0000000000..f0abb550c0 --- /dev/null +++ b/.changeset/fix-transient-post-merge-push-retry.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Retry post-merge pushes after temporary Git network failures. +category: fix +dev: Adds two cancellation-aware retries with bounded backoff on transient transport failures across both post-merge push paths; configuration, authentication, and ref-rejection errors still fail immediately. diff --git a/packages/engine/src/__tests__/merger-ai-push-after-merge.test.ts b/packages/engine/src/__tests__/merger-ai-push-after-merge.test.ts index c1cf312658..85631b5256 100644 --- a/packages/engine/src/__tests__/merger-ai-push-after-merge.test.ts +++ b/packages/engine/src/__tests__/merger-ai-push-after-merge.test.ts @@ -151,6 +151,60 @@ describe("runAiMerge push-after-merge", () => { })); }); + it("retries a temporary transport failure on the unified fast path", async () => { + const { dir, originDir } = initRepoWithRemote(); + const hookPath = join(originDir, "hooks", "pre-receive"); + writeFileSync(hookPath, `#!/bin/sh +marker="$(dirname "$0")/../transient-push-once" +if [ ! -f "$marker" ]; then + touch "$marker" + echo "fatal: unable to access remote: Connection reset by peer" >&2 + exit 1 +fi +`, { mode: 0o755 }); + const { store, logs } = makeStore(); + + const result = await runAiMerge(store, dir, "FN-1", { manual: true }, { + mergeAgent: realMergeAgent("fusion/fn-1"), + reviewAgent: approveReviewer(), + }); + + expect(result.pushedToRemote).toBe(true); + expect(git(originDir, "rev-parse main")).toBe(git(dir, "rev-parse main")); + expect(logs.some((entry) => entry.message.includes("temporary Git transport failure"))).toBe(true); + }); + + it("records an aborted outcome when cancellation interrupts the unified retry backoff", async () => { + const { dir, originDir } = initRepoWithRemote(); + const hookPath = join(originDir, "hooks", "pre-receive"); + writeFileSync(hookPath, `#!/bin/sh +echo "fatal: unable to access remote: Connection reset by peer" >&2 +exit 1 +`, { mode: 0o755 }); + const controller = new AbortController(); + const { store, storeMocks, logs } = makeStore(); + storeMocks.logEntry.mockImplementation(async (_id: string, message: string, action?: string) => { + logs.push({ message, action }); + if (message.includes("temporary Git transport failure")) controller.abort(); + }); + + const result = await runAiMerge(store, dir, "FN-1", { manual: true, signal: controller.signal }, { + mergeAgent: realMergeAgent("fusion/fn-1"), + reviewAgent: approveReviewer(), + }); + + expect(result.pushedToRemote).toBe(false); + expect(result.pushError).toContain("aborted by shutdown signal"); + expect(storeMocks.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "push:origin", + metadata: expect.objectContaining({ outcome: "aborted" }), + })); + expect(storeMocks.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "push:origin", + metadata: expect.objectContaining({ outcome: "failed" }), + })); + }); + it("rebases in a clean room and pushes when the remote has diverged (non-FF path)", async () => { const { dir, originDir } = initRepoWithRemote(); // Remote moves ahead AFTER our clone: the fast-path push must reject non-FF. diff --git a/packages/engine/src/__tests__/merger-prompt-and-utils.test.ts b/packages/engine/src/__tests__/merger-prompt-and-utils.test.ts index c7cd2c8562..7f579cda70 100644 --- a/packages/engine/src/__tests__/merger-prompt-and-utils.test.ts +++ b/packages/engine/src/__tests__/merger-prompt-and-utils.test.ts @@ -144,7 +144,9 @@ import { inferDefaultTestCommand, resolveTaskDiffBaseRef, commitOrAmendMergeWithFixes, + isTransientGitPushError, MergeAbortedError, + pushWithTransientRetries, type ConflictCategory, } from "../merger.js"; import { mergerLog } from "../logger.js"; @@ -359,6 +361,135 @@ describe("push-after-merge", () => { } + // FNXC:MergePush 2026-08-16-03:39: Retry tests cover bounded transient recovery and cancellation before and during backoff across both retained push surfaces. + it("classifies Git transport failures without treating permission errors as transient", () => { + expect(isTransientGitPushError("fatal: unable to access remote: Connection reset by peer")).toBe(true); + expect(isTransientGitPushError("fatal: unable to access remote: Could not resolve host: git.example")).toBe(true); + expect(isTransientGitPushError("remote: permission denied")).toBe(false); + }); + + it("bounds transient retries and stops immediately when the merge is cancelled", async () => { + const transientError = Object.assign(new Error("push failed"), { + stderr: "fatal: unable to access remote: Connection reset by peer", + }); + const exhaustedPush = vi.fn().mockRejectedValue(transientError); + + await expect(pushWithTransientRetries(exhaustedPush, { + taskId: "FN-050", + retryBackoffMs: [0, 0], + })).rejects.toThrow("push failed"); + expect(exhaustedPush).toHaveBeenCalledTimes(3); + + const controller = new AbortController(); + const cancelledPush = vi.fn().mockRejectedValue(transientError); + await expect(pushWithTransientRetries(cancelledPush, { + taskId: "FN-050", + signal: controller.signal, + retryBackoffMs: [10_000], + onRetry: () => controller.abort(), + })).rejects.toBeInstanceOf(MergeAbortedError); + expect(cancelledPush).toHaveBeenCalledTimes(1); + }); + + it("cancels while the transient retry backoff is pending", async () => { + vi.useFakeTimers(); + try { + const transientError = Object.assign(new Error("push failed"), { + stderr: "fatal: unable to access remote: Connection reset by peer", + }); + const controller = new AbortController(); + const push = vi.fn().mockRejectedValue(transientError); + let signalBackoffStarted!: () => void; + const backoffStarted = new Promise((resolve) => { + signalBackoffStarted = resolve; + }); + + const pending = pushWithTransientRetries(push, { + taskId: "FN-050", + signal: controller.signal, + retryBackoffMs: [10_000], + onRetry: signalBackoffStarted, + }); + await backoffStarted; + await Promise.resolve(); + expect(vi.getTimerCount()).toBe(1); + + controller.abort(); + + await expect(pending).rejects.toBeInstanceOf(MergeAbortedError); + expect(push).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("retries a transient transport failure in the shared post-merge push path", async () => { + vi.useFakeTimers(); + try { + let pushAttempts = 0; + mockedExecSync.mockImplementation((cmd: any) => { + const cmdStr = String(cmd); + if (cmdStr.startsWith('git pull --rebase "origin" "main"')) return Buffer.from(""); + if (cmdStr.startsWith('git push "origin" "main"')) { + pushAttempts += 1; + if (pushAttempts === 1) { + throw Object.assign(new Error("push failed"), { + stderr: "fatal: unable to access remote: Connection reset by peer", + }); + } + return Buffer.from(""); + } + if (cmdStr.includes("git symbolic-ref --short HEAD")) return "main" as any; + if (cmdStr.includes("git rev-parse --verify REBASE_HEAD")) { + throw Object.assign(new Error("fatal: Needed a single revision"), { status: 128 }); + } + return Buffer.from(""); + }); + + const pending = pushToRemoteAfterMerge(createMockStore(), "/tmp/root", "FN-050", { + ...DEFAULT_SETTINGS, + pushAfterMerge: true, + pushRemote: "origin", + }); + await vi.runAllTimersAsync(); + const result = await pending; + + expect(result).toEqual({ pushed: true }); + expect(pushAttempts).toBe(2); + } finally { + vi.useRealTimers(); + } + }); + + it("propagates cancellation from the shared initial push path", async () => { + const controller = new AbortController(); + let pushAttempts = 0; + mockedExecSync.mockImplementation((cmd: any) => { + const cmdStr = String(cmd); + if (cmdStr.startsWith('git pull --rebase "origin" "main"')) return Buffer.from(""); + if (cmdStr.startsWith('git push "origin" "main"')) { + pushAttempts += 1; + controller.abort(); + throw Object.assign(new Error("push failed"), { + stderr: "fatal: unable to access remote: Connection reset by peer", + }); + } + if (cmdStr.includes("git symbolic-ref --short HEAD")) return "main" as any; + if (cmdStr.includes("git rev-parse --verify REBASE_HEAD")) { + throw Object.assign(new Error("fatal: Needed a single revision"), { status: 128 }); + } + return Buffer.from(""); + }); + + await expect(pushToRemoteAfterMerge(createMockStore(), "/tmp/root", "FN-050", { + ...DEFAULT_SETTINGS, + pushAfterMerge: true, + pushRemote: "origin", + }, { signal: controller.signal })).rejects.toBeInstanceOf(MergeAbortedError); + expect(pushAttempts).toBe(1); + }); + it("pushes merged result when pushAfterMerge is enabled", async () => { setupAiMergeExecSyncWithPush(); @@ -489,6 +620,51 @@ describe("push-after-merge", () => { ); }); + it("records an aborted outcome when the retained merger is cancelled during push", async () => { + const controller = new AbortController(); + setupAiMergeExecSyncWithPush(() => { + controller.abort(); + throw Object.assign(new Error("push failed"), { + stderr: "fatal: unable to access remote: Connection reset by peer", + }); + }); + + const store = createMockStore(); + (store.getSettings as ReturnType).mockResolvedValue({ + ...DEFAULT_SETTINGS, + mergeIntegrationWorktree: "cwd-main" as const, + pushAfterMerge: true, + pushRemote: "origin", + mergeStrategy: "direct", + }); + + const result = await aiMergeTask(store, "/tmp/root", "FN-050", { signal: controller.signal }); + + expect(result.merged).toBe(true); + expect(result.pushedToRemote).toBe(false); + expect(result.pushError).toContain("aborted by shutdown signal"); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith( + expect.objectContaining({ + domain: "git", + mutationType: "push:origin", + target: "FN-050", + metadata: expect.objectContaining({ outcome: "aborted" }), + }), + ); + expect(store.recordRunAuditEvent).not.toHaveBeenCalledWith( + expect.objectContaining({ + domain: "git", + mutationType: "push:origin", + metadata: expect.objectContaining({ outcome: "failed" }), + }), + ); + expect(store.logEntry).toHaveBeenCalledWith( + "FN-050", + expect.stringContaining("aborted by shutdown signal"), + "PushToRemoteFailed", + ); + }); + it.each([undefined, "", " "])("FN-7490 falls back to origin and integration branch for empty pushRemote %s", async (pushRemote) => { mockedExecSync.mockImplementation((cmd: any) => { const cmdStr = String(cmd); @@ -1251,5 +1427,3 @@ describe("commitOrAmendMergeWithFixes", () => { expect(result).toEqual({ ok: true, reason: "branch-already-merged" }); }); }); - - diff --git a/packages/engine/src/merge/merger-ai.ts b/packages/engine/src/merge/merger-ai.ts index 51dab2e33b..1130641169 100644 --- a/packages/engine/src/merge/merger-ai.ts +++ b/packages/engine/src/merge/merger-ai.ts @@ -93,6 +93,7 @@ import { isNonFastForwardPushError, isRebaseInProgress, parsePushRemoteTarget, + pushWithTransientRetries, pushToRemoteAfterMerge, runMergeAdvanceAutoSync, syncGroupPrOnLanding, @@ -2719,9 +2720,23 @@ export async function pushAfterMergeToRemote(input: { throwIfAborted(signal, taskId); let fastPathError: string; try { - await git(["push", remote, `${localRef}:refs/heads/${targetBranch}`], projectRootDir, { timeout: 120_000 }); + // FNXC:MergePush 2026-08-16-02:55: Retry transient fast-path transport failures with + // bounded, cancellation-aware backoff; merge aborts must escape to the aborted-push audit path. + await pushWithTransientRetries( + () => git(["push", remote, `${localRef}:refs/heads/${targetBranch}`], projectRootDir, { timeout: 120_000 }), + { + taskId, + signal, + onRetry: async ({ attempt, maxRetries, delayMs, error }) => { + const message = `Push after merge: temporary Git transport failure; retrying in ${delayMs}ms (${attempt}/${maxRetries}): ${error}`; + aiMergeLog.warn(`${taskId}: ${message}`); + await log(message); + }, + }, + ); return { pushed: true, remote, targetBranch }; } catch (err: unknown) { + if (isMergeAbortedError(err)) throw err; fastPathError = getErrorMessage(err); } if (!isNonFastForwardPushError(fastPathError)) { diff --git a/packages/engine/src/merger.ts b/packages/engine/src/merger.ts index a5267b4bad..d093ced08c 100644 --- a/packages/engine/src/merger.ts +++ b/packages/engine/src/merger.ts @@ -273,6 +273,8 @@ export { regenerateBareMergeSubject, BARE_MERGE_SUBJECT_RE } from "./merge/merge import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./errors/usage-limit-detector.js"; import { isContextLimitError } from "./errors/context-limit-detector.js"; import { withRateLimitRetry } from "./errors/rate-limit-retry.js"; +import { cancellableSleep } from "./errors/retry-with-backoff.js"; +import { TRANSIENT_ERROR_PATTERNS } from "./errors/transient-error-patterns.js"; import { resolveAgentInstructions, buildSystemPromptWithInstructions } from "./agents/agent-instructions.js"; import type { ToolDefinition } from "@earendil-works/pi-coding-agent"; import { Type } from "typebox"; @@ -513,9 +515,81 @@ const PULL_REBASE_TIMEOUT_MS = 120_000; const PUSH_TIMEOUT_MS = 60_000; const PUSH_NON_FF_MAX_RETRIES = 3; const PUSH_NON_FF_RETRY_BACKOFF_MS = [2_000, 5_000, 10_000]; +const PUSH_TRANSIENT_RETRY_BACKOFF_MS = [2_000, 5_000]; -function delay(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); +const TRANSIENT_GIT_PUSH_PATTERNS = [ + /could not resolve host/i, + /temporary failure in name resolution/i, + /network is unreachable/i, + /failed to connect/i, + /couldn't connect/i, + /rpc failed;\s*http\s+(?:429|5\d\d)/i, + /remote end hung up unexpectedly/i, + /unexpected disconnect while reading sideband packet/i, + /send failure:\s*broken pipe/i, + /connection closed by remote host/i, +]; + +export function isTransientGitPushError(message: string): boolean { + return TRANSIENT_ERROR_PATTERNS.some((pattern) => pattern.test(message)) + || TRANSIENT_GIT_PUSH_PATTERNS.some((pattern) => pattern.test(message)); +} + +async function sleepForPushRetry(ms: number, signal: AbortSignal | undefined, taskId: string): Promise { + try { + await cancellableSleep(ms, signal); + } catch (error: unknown) { + throwIfAborted(signal, taskId); + throw error; + } +} + +/* +FNXC:MergePush 2026-08-16-02:17: +Post-merge delivery must recover from short-lived Git transport failures for every remote and +hosting provider without depending on an operator-specific notification channel. Retry only +recognized transient transport errors, keep configuration/auth/rejection failures immediate, and +make both the bounded backoff and every subsequent attempt cancellation-aware. +*/ +export async function pushWithTransientRetries( + push: () => Promise, + options: { + taskId: string; + signal?: AbortSignal; + retryBackoffMs?: readonly number[]; + onRetry?: (event: { attempt: number; maxRetries: number; delayMs: number; error: string }) => void | Promise; + }, +): Promise { + const retryBackoffMs = options.retryBackoffMs ?? PUSH_TRANSIENT_RETRY_BACKOFF_MS; + let attempt = 0; + + while (true) { + throwIfAborted(options.signal, options.taskId); + try { + await push(); + return; + } catch (error: unknown) { + rethrowIfMergeAborted(error); + throwIfAborted(options.signal, options.taskId); + const message = getCommandErrorMessage(error); + const delayMs = retryBackoffMs[attempt]; + if (delayMs === undefined || !isTransientGitPushError(message)) throw error; + + attempt += 1; + try { + await options.onRetry?.({ + attempt, + maxRetries: retryBackoffMs.length, + delayMs, + error: message, + }); + } catch { + // Retry diagnostics are best-effort and must not suppress delivery. + } + + await sleepForPushRetry(delayMs, options.signal, options.taskId); + } + } } export async function emitMergeAttemptAuditEvent(params: { @@ -6288,18 +6362,32 @@ export async function pushToRemoteAfterMerge( const pushCommand = options?.pushHeadRefspec ? `git push ${quoteArg(remote)} ${quoteArg(`HEAD:refs/heads/${branch}`)}` : `git push ${quoteArg(remote)} ${quoteArg(branch)}`; - - try { - throwIfAborted(options?.signal, taskId); - await execAsync(pushCommand, { + const runTargetPush = () => pushWithTransientRetries( + () => execAsync(pushCommand, { cwd: rootDir, timeout: PUSH_TIMEOUT_MS, maxBuffer: VERIFICATION_COMMAND_MAX_BUFFER, encoding: "utf-8", - }); + }), + { + taskId, + signal: options?.signal, + onRetry: ({ attempt, maxRetries, delayMs, error }) => { + mergerLog.warn( + `${taskId}: temporary Git transport failure; retrying push in ${delayMs}ms (${attempt}/${maxRetries}): ${error}`, + ); + }, + }, + ); + + try { + throwIfAborted(options?.signal, taskId); + await runTargetPush(); mergerLog.log(`${taskId}: pushed merged result to ${remote}/${branch}`); return { pushed: true }; } catch (firstPushError: unknown) { + // FNXC:MergePush 2026-08-16-05:14: Cancellation from the shared initial push must reach the caller's aborted-push audit path instead of being recorded as a terminal delivery failure. + rethrowIfMergeAborted(firstPushError); let lastMessage = getCommandErrorMessage(firstPushError); mergerLog.warn(`${taskId}: initial push failed: ${lastMessage}`); @@ -6321,12 +6409,7 @@ export async function pushToRemoteAfterMerge( throwIfAborted(options?.signal, taskId); await pullWithRebaseAndResolveConflicts(store, rootDir, taskId, settings, remote, branch, options); throwIfAborted(options?.signal, taskId); - await execAsync(pushCommand, { - cwd: rootDir, - timeout: PUSH_TIMEOUT_MS, - maxBuffer: VERIFICATION_COMMAND_MAX_BUFFER, - encoding: "utf-8", - }); + await runTargetPush(); mergerLog.log(`${taskId}: push succeeded after non-fast-forward retry (attempt ${attempt}/${maxRetries})`); return { pushed: true }; } catch (retryError: unknown) { @@ -6337,7 +6420,11 @@ export async function pushToRemoteAfterMerge( break; } throwIfAborted(options?.signal, taskId); - await delay(PUSH_NON_FF_RETRY_BACKOFF_MS[attempt - 1] ?? PUSH_NON_FF_RETRY_BACKOFF_MS.at(-1)!); + await sleepForPushRetry( + PUSH_NON_FF_RETRY_BACKOFF_MS[attempt - 1] ?? PUSH_NON_FF_RETRY_BACKOFF_MS.at(-1)!, + options?.signal, + taskId, + ); } } return { pushed: false, error: lastMessage }; @@ -9899,24 +9986,42 @@ export async function aiMergeTask( result.pushError = pushResult.error; } } catch (err: any) { - mergerLog.error(`${taskId}: push to remote error: ${err.message}`); - result.pushedToRemote = false; - result.pushError = err.message; - await audit.git({ - type: "push:origin", - target: taskId, - metadata: { - integrationBranch: mergeTarget.branch, - remote: settings.pushRemote || "origin", - outcome: "failed", - stderrPreview: err.message, - }, - }).catch(() => undefined); - await store.logEntry( - taskId, - `Push to remote threw after merge — task marked done anyway; local main may diverge from origin: ${err.message}`, - "PushToRemoteFailed", - ).catch(() => undefined); + if (err instanceof Error && err.name === "MergeAbortedError") { + // FNXC:MergePush 2026-08-16-03:39: The retained public merger must preserve shutdown cancellation as an aborted delivery outcome after the local merge is finalized. + const message = "Push after merge aborted by shutdown signal; the local merge remains finalized"; + mergerLog.warn(`${taskId}: ${message}`); + result.pushedToRemote = false; + result.pushError = message; + await audit.git({ + type: "push:origin", + target: taskId, + metadata: { + integrationBranch: mergeTarget.branch, + remote: settings.pushRemote || "origin", + outcome: "aborted", + }, + }).catch(() => undefined); + await store.logEntry(taskId, message, "PushToRemoteFailed").catch(() => undefined); + } else { + mergerLog.error(`${taskId}: push to remote error: ${err.message}`); + result.pushedToRemote = false; + result.pushError = err.message; + await audit.git({ + type: "push:origin", + target: taskId, + metadata: { + integrationBranch: mergeTarget.branch, + remote: settings.pushRemote || "origin", + outcome: "failed", + stderrPreview: err.message, + }, + }).catch(() => undefined); + await store.logEntry( + taskId, + `Push to remote threw after merge — task marked done anyway; local main may diverge from origin: ${err.message}`, + "PushToRemoteFailed", + ).catch(() => undefined); + } } }