A push cut off by a memory reset is retried once; the concurrent-push e2e skips wrangler dev's proxy (#244)

* A push cut off by a memory reset of its repository is handed over once more, like a deploy's
* The concurrent-push e2e sends its pushes to the Worker itself, past wrangler dev's proxy

test-bot authored 10 days ago1 parent 4789265commit 274d73a6f691a3e61347c243c4c51a13028147f8Browse files

CI / Typecheck and bundlesuccess
CI / Unit tests (1/2)success
CI / Unit tests (2/2)success
Deploy / Deploy to Cloudflaresuccess
Deploy / Main's current tipsuccess
End-to-end / End-to-end (agent reviews, tasks, Today)success
End-to-end / End-to-end (git, PRs, Actions)success
End-to-end / End-to-end (large pushes)success
End-to-end / End-to-end (passkeys, registry, email)success
End-to-end / End-to-end (real histories)success
End-to-end / End-to-end (runner Docker)success
End-to-end / End-to-end (security)success
Live test / Containers, registry, sandboxes, webhooks, schedules, LFS and pull requestssuccess

3 files changed+43−7

modifiedapps/diffinery/src/http/git.ts+9−1View file

modifiedapps/diffinery/test/e2e/e2e.test.mjs+21−4View file

modifiedapps/diffinery/test/unit/push-retry.test.ts+13−2View file

		@@ -69,7 +69,7 @@
69	69	try {
70	70	return await stub.gitReceivePack(body, caller, { moved, retry: attempt > 1 });
71	71	} catch (e) {
72		- if (attempt >= PUSH_TRIES \|\| !resetByDeploy(e)) throw e;
	72	+ if (attempt >= PUSH_TRIES \|\| !(resetByDeploy(e) \|\| resetForMemory(e))) throw e;
73	73	// A stub that saw its object reset is broken; a new one reaches the new code.
74	74	stub = repoStub(env, repoId);
75	75	}
		@@ -81,6 +81,14 @@
81	81
82	82	const PUSH_TRIES = 2;
83	83
	84	+/**
	85	+ * The repository's object was reset for going past its memory. Pushes to one repository take turns
	86	+ * for memory (RepoDO.gitReceivePack), so a push cut off this way was most likely caught up in another
	87	+ * request's reset, not the cause: like a deploy's reset, it is handed over once more, which moves the
	88	+ * refs the cut-off attempt didn't and sends the push events it may not have sent.
	89	+ */
	90	+export const resetForMemory = (e: unknown) => e instanceof Error && /exceeded its memory limit and was reset/i.test(e.message);
	91	+
84	92	export { resetByDeploy };
85	93
86	94	/** A fetch's wants and haves; git's own negotiation stays far below this. */

		@@ -25,6 +25,8 @@
25	25	const AGENT = resolve(APP, "../../containers/runner/dist/agent.mjs");
26	26	const work = mkdtempSync(join(tmpdir(), "diffinery-e2e-"));
27	27	let base;
	28	+/** The shared server's Worker itself, behind wrangler dev's proxy (see startServer). */
	29	+let direct;
28	30	// The dev server writes to a file, not a pipe: a long synchronous git command keeps this process
29	31	// from reading a pipe, and a full one (retried webhooks log a stack each) stalls the server mid-push.
30	32	const devLogFile = join(work, "wrangler.log");
		@@ -183,6 +185,7 @@
183	185	const url = `http://127.0.0.1:${port}`;
184	186	const persist = join(work, name ? `state-${name}` : "state");
185	187	const logFile = name ? join(work, `wrangler-${name}.log`) : devLogFile;
	188	+ const debugLogFile = join(work, name ? `wrangler-debug-${name}.log` : "wrangler-debug.log");
186	189	const log = () => (existsSync(logFile) ? readFileSync(logFile, "utf8") : "");
187	190	migratedState(APP, persist);
188	191	// Until the web UI exists, run the same app through the API-only entry.
		@@ -190,7 +193,7 @@
190	193	const out = openSync(logFile, "a");
191	194	const child = spawn("pnpm", ["exec", "wrangler", "dev", ...entry, "-e", "dev", "--ip", "127.0.0.1", "--port", String(port), "--inspector-port", String(await freePort()), "--persist-to", persist, "--var", `PUBLIC_URL:${url}`, "--var", `MASTER_KEY:${randomBytes(32).toString("base64")}`, "--var", "SCHEDULE_TICK_MINUTES:1", "--show-interactive-dev-session=false"], {
192	195	cwd: APP,
193		- env: { ...process.env, CI: "1", WRANGLER_SEND_METRICS: "false" },
	196	+ env: { ...process.env, CI: "1", WRANGLER_SEND_METRICS: "false", WRANGLER_LOG_PATH: debugLogFile },
194	197	stdio: ["ignore", out, out],
195	198	detached: true,
196	199	});
		@@ -208,12 +211,24 @@
208	211	process.kill(-child.pid, "SIGTERM");
209	212	} catch {}
210	213	};
211		- return { base: url, token: (await res.json()).token, log, stop };
	214	+ return { base: url, direct: workerUrl(debugLogFile), token: (await res.json()).token, log, stop };
212	215	}
213	216
	217	+/**
	218	+ * wrangler dev answers on `base` through a proxy Worker of its own, which drops one of several large
	219	+ * pushes streaming through it at once ("Error inside ProxyWorker ... Network connection lost": the
	220	+ * push never reaches our Worker; 3 of 5 runs of three pushes, none of 5 sent to the Worker itself).
	221	+ * Its debug log names the Worker's own address; null if it doesn't.
	222	+ */
	223	+function workerUrl(debugLogFile) {
	224	+ const log = existsSync(debugLogFile) ? readFileSync(debugLogFile, "utf8") : "";
	225	+ const at = [...log.matchAll(/"userWorkerUrl":\{"protocol":"(https?:)","hostname":"([^"]+)","port":"(\d+)"/g)].pop();
	226	+ return at ? `${at[1]}//${at[2]}:${at[3]}` : null;
	227	+}
	228	+
214	229	before(async () => {
215	230	assert.ok(existsSync(AGENT), `build the runner first: pnpm --filter @diffinery/runner build (missing ${AGENT})`);
216		- ({ base, token } = await startServer());
	231	+ ({ base, direct, token } = await startServer());
217	232	});
218	233
219	234	after(() => {
		@@ -2935,7 +2950,9 @@
2935	2950	test("large pushes: three large pushes into one repository at once all go in, and a fetch meanwhile works", async () => {
2936	2951	const me = await ok("GET", "/user");
2937	2952	await ok("POST", "/user/repos", { name: "pushes-at-once" });
2938		- const remote = `${base}/${me.login}/pushes-at-once.git`;
	2953	+ // Straight to the Worker: wrangler dev's own proxy, not Diffinery, drops pushes streaming through it at once (see workerUrl).
	2954	+ assert.ok(direct && (await fetch(`${direct}/healthz`)).ok, `the Worker's own address wasn't found in wrangler dev's debug log (${direct})`);
	2955	+ const remote = `${direct}/${me.login}/pushes-at-once.git`;
2939	2956	const src = join(work, "pushes-at-once");
2940	2957	mkdirSync(src, { recursive: true });
2941	2958	git(src, ["init", "-q", "-b", "main"]);

		@@ -54,9 +54,20 @@
54	54	expect(f.deleted).toEqual(["push/abc"]);
55	55	});
56	56
57		- it("is not retried for any other error", async () => {
	57	+ it("is handed over once more after a memory reset too, marked as a retry", async () => {
58	58	const f = fakes(1, "Durable Object's isolate exceeded its memory limit and was reset.");
59		- await expect(receiveWithRetry(f.env, "r1", f.first, new Uint8Array(4), caller, false)).rejects.toThrow(/memory limit/);
	59	+ expect(await receiveWithRetry(f.env, "r1", f.first, new Uint8Array(4), caller, false)).toEqual(new Uint8Array([1]));
	60	+ expect(f.stubs).toEqual([0, 1]);
	61	+ expect(f.retries).toEqual([false, true]);
	62	+ // Not a third time.
	63	+ const g = fakes(2, "Durable Object's isolate exceeded its memory limit and was reset.");
	64	+ await expect(receiveWithRetry(g.env, "r1", g.first, new Uint8Array(4), caller, false)).rejects.toThrow(/memory limit/);
	65	+ expect(g.stubs).toEqual([0, 1]);
	66	+ });
	67	+
	68	+ it("is not retried for any other error", async () => {
	69	+ const f = fakes(1, "Internal error in Durable Object storage write.");
	70	+ await expect(receiveWithRetry(f.env, "r1", f.first, new Uint8Array(4), caller, false)).rejects.toThrow(/storage write/);
60	71	expect(f.stubs).toEqual([0]);
61	72	expect(f.deleted).toEqual([]);
62	73	expect(resetByDeploy(new Error(RESET))).toBe(true);

codician/diffinerypublic

A push cut off by a memory reset is retried once; the concurrent-push e2e skips wrangler dev's proxy (#244)