e1ebd89 · 5afc982 · eaeb4ba · c6702f2
@@ -104,6 +104,7 @@
export const GATE: readonly string[] = [
  "test/release-preflight.test.ts",
  "test/release-rollback.test.ts",
  "test/release-swap.test.ts",
  "test/release-handover.test.ts",
  "test/threats-doc.test.ts",
  "test/returns.test.ts",
  "test/review-route.test.ts",
@@ -1639,6 +1639,13 @@
async function handleHook(app: App, request: Request): Promise<Response> {
export interface Servers {
  readonly url: string;
  readonly port: number;
  /**
   * Stop taking connections and let those in flight finish: what a process
   * handing its port to a release beside it does at drain start (#2040).
   * Resolves once no request is in flight, or `budgetMs` has passed; `stop`
   * then cuts whatever is left.
   */
  stopAccepting(budgetMs: number): Promise<void>;
  stop(): void;
}
@@ -1665,6 +1672,9 @@
function withHeader(response: Response, name: string, value: string): Response {
  }
}
/** How often a draining listener looks whether its last request is answered. */
const INFLIGHT_POLL_MS = 50;
/** The hook sockets a data directory may hold: this forge's, and others'. */
const SOCKET_FILE = /^forge(?:-[a-z0-9]+)?\.sock$/;
@@ -1919,6 +1929,14 @@
export async function serve(app: App): Promise<Servers> {
  return {
    url: `http://localhost:${port}`,
    port,
    async stopAccepting(budgetMs) {
      // Closing the listener leaves the port to the process beside this one;
      // the connections already here are answered to the end.
      web.stop(false).catch(() => {});
      const deadline = Date.now() + budgetMs;
      while (web.pendingRequests > 0 && Date.now() < deadline)
        await Bun.sleep(INFLIGHT_POLL_MS);
    },
    stop() {
      endOpen();
      clearInterval(sweep);
@@ -6,7 +6,7 @@
import { startSampler } from "./boards/sampler.ts";
import { sweepStanding } from "./crew/standing.ts";
import { handover } from "./release/swap.ts";
import { drainOnHandover, handover } from "./release/swap.ts";
import { parentUrlProblem } from "./core/parent_releases.ts";
import { releaseEnv } from "./release/env.ts";
import { createApp } from "./app.ts";
@@ -487,8 +487,14 @@
const leaseTimer = setInterval(() => {
  releaseOrphans(app.db);
}, 30_000);
// A release that read healthy beside this process takes over (#1663).
handover.drain = () => shutdown("SIGTERM");
// A release that read healthy beside this process takes over (#1663),
// and this one lets go of the port before it drains (#2040).
const handOver = drainOnHandover(servers, () => drain("SIGTERM"), close);
handover.drain = () => {
  if (draining) return;
  draining = true;
      
        sentinelcrew
        29d ago
        outdated
      
      This wiring is the fix, and no test reaches it: revert it and release-handover.test.ts still passes. Move it into an exported function and test it through handover.drain().
    
  void handOver();
};
handover.ssh = ssh !== undefined;
process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);
@@ -505,6 +511,11 @@
let draining = false;
function shutdown(signal: NodeJS.Signals): void {
  if (draining) return;
  draining = true;
  void drain(signal).finally(close);
}
/** Everything a leaving process finishes before it goes: `shutdown` above. */
function drain(signal: NodeJS.Signals): Promise<unknown> {
  const sessions = ssh?.sessions() ?? 0;
  startDraining(app.log, signal, { sshSessions: sessions });
  // Streams end now, with a resume hint: their browsers reconnect to the
@@ -518,7 +529,7 @@
function shutdown(signal: NodeJS.Signals): void {
  // The platform's SIGTERM: the listener stays open while routers learn
  // this process is going. A person's Ctrl-C does not wait for routers.
  const linger = signal === "SIGTERM" ? Bun.sleep(lingerMs ?? 0) : undefined;
  Promise.all([
  return Promise.all([
    linger,
    prompts.stop(),
    notifier.stop(),
@@ -538,12 +549,14 @@
function shutdown(signal: NodeJS.Signals): void {
        });
      return ssh?.drain();
    })
    .catch(() => {})
    .finally(() => {
      app.log.info("forge shutting down");
      ssh?.stop();
      servers.stop();
      app.close();
      process.exit(0);
    });
    .catch(() => {});
}
/** The last step of a drain: cut what is left and exit. */
function close(): void {
  app.log.info("forge shutting down");
  ssh?.stop();
  servers.stop();
  app.close();
  process.exit(0);
}
@@ -91,11 +91,40 @@
export async function startRelease(input: SwapInput): Promise<Swapped> {
  };
}
/**
 * How long a process that handed its port over keeps answering what it had
 * already taken — a long push or clone — before it goes (#2040).
 */
export const HANDOVER_INFLIGHT_MS = 300_000;
/**
      
        sentinelcrew
        29d ago
        
      
      This was inserted between handover's doc comment (ending on line 99) and handover itself, so that comment no longer documents anything. Move the constant and its comment above the handover doc comment.
    
 * The drain a handover runs (#2040). The release listens on this port too,
 * so this process stops accepting at once: every new connection is the
 * release's. What is in flight here is answered to the end (up to
 * `budgetMs`) beside `rest` — the shutdown's other steps — and only once
 * both are done does `close` cut what remains. Runs once.
 */
export function drainOnHandover(
  servers: { stopAccepting(budgetMs: number): Promise<void> },
  rest: () => Promise<unknown>,
  close: () => void,
  budgetMs = HANDOVER_INFLIGHT_MS,
): () => Promise<void> {
  let done: Promise<void> | undefined;
  return () =>
    (done ??= (async () => {
      const inFlight = servers.stopAccepting(budgetMs);
      await Promise.all([inFlight, rest()]).catch(() => {});
      close();
    })());
}
/**
 * What the running forge does once the release is well: drain, as a SIGTERM
 * does. Set at boot, with whether this process holds the SSH listener (a new
 * process cannot bind that port while this one does, and a release must not
 * quietly close the terminal).
 * does, but letting go of the port first (`drainOnHandover`). Set at boot,
 * with whether this process holds the SSH listener (a new process cannot
 * bind that port while this one does, and a release must not quietly close
 * the terminal).
 */
export const handover: { drain?: () => void; ssh?: boolean; migrate?: boolean } = {};
@@ -0,0 +1,274 @@
/**
 * A roll never cuts an in-flight push (#2040). Two forges on one data
 * directory share the port (SO_REUSEPORT), as during a release swap: a push
 * that started on the old one trickles in for thirty seconds while the new
 * one comes up and the old one drains. The old stops accepting at drain
 * start — new connections are the new forge's — and answers the push to the
 * end before it goes.
 */
import { afterAll, expect, test } from "bun:test";
import { readFileSync } from "node:fs";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp } from "../src/app.ts";
import { commitToBranch, createRepo } from "../src/actions/repos.ts";
import { createToken } from "../src/core/tokens.ts";
import { createUser } from "../src/core/users.ts";
import { repoDir } from "../src/git/repos.ts";
import { drainOnHandover, handover } from "../src/release/swap.ts";
import { serve } from "../src/http/server.ts";
import { gitClientEnv } from "./helpers/git-client.ts";
const PUSH_MS = 30_000;
const dirs: string[] = [];
afterAll(async () => {
  for (const d of dirs) await rm(d, { recursive: true, force: true });
});
// Without it, what sat in the old listener's accept queue is reset when it
// closes — at drain start now, at the end before; either way the kernel's.
const migrates = (() => {
  try {
    return readFileSync("/proc/sys/net/ipv4/tcp_migrate_req", "utf8").trim() === "1";
  } catch {
    return false;
  }
})();
async function git(args: string[], cwd: string, stdin?: string): Promise<Uint8Array> {
  const out = Bun.spawnSync(["git", ...args], {
    cwd,
    env: gitClientEnv(),
    ...(stdin === undefined ? {} : { stdin: new TextEncoder().encode(stdin) }),
  });
  if (out.exitCode !== 0) throw new Error(`git ${args.join(" ")}: ${out.stderr.toString()}`);
  return new Uint8Array(out.stdout);
}
const text = (bytes: Uint8Array) => new TextDecoder().decode(bytes).trim();
function pktLine(line: string): Uint8Array {
  const body = new TextEncoder().encode(line);
  const head = (body.length + 4).toString(16).padStart(4, "0");
  return new Uint8Array([...new TextEncoder().encode(head), ...body]);
}
/** `bytes` sent in pieces spread over `ms`, so the request stays in flight that long. */
function trickle(bytes: Uint8Array, ms: number): ReadableStream<Uint8Array> {
  const pieces = 30;
  const size = Math.ceil(bytes.length / pieces);
  let i = 0;
  return new ReadableStream({
    async pull(controller) {
      if (i > 0) await Bun.sleep(ms / pieces);
      const piece = bytes.subarray(i * size, (i + 1) * size);
      i++;
      if (piece.length > 0) controller.enqueue(piece);
      if (i >= pieces) controller.close();
    },
  });
}
/**
 * The old forge on a port of its own, alice's repository, and `push(ms)`: a
 * receive-pack of one new commit on main, trickled in over `ms`.
 */
async function forgeWithPush() {
  const dataDir = await mkdtemp(join(tmpdir(), "forge-handover-"));
  const work = await mkdtemp(join(tmpdir(), "forge-handover-work-"));
  dirs.push(dataDir, work);
  // The old forge picks the port; the release binds the one it got.
  const oldApp = createApp({ dataDir, port: 0, selfSwap: true });
  const oldServers = await serve(oldApp);
  const port = oldServers.port;
  const made = await createUser(oldApp.db, {
    trusted: true,
    username: "alice",
    email: "[email protected]",
    password: "a-good-passphrase",
    adminIfFirst: true,
  });
  if (!made.ok) throw new Error("user");
  const alice = made.value;
  const repo = await createRepo(oldApp, alice, { name: "garden", visibility: "public" });
  if (!repo.ok) throw new Error("repo");
  await commitToBranch(oldApp, alice, repo.value, {
    message: "seed",
    files: [{ path: "a.txt", content: "a" }],
  });
  const { token } = createToken(oldApp.db, alice.id, "push", "rw");
  // The push: one commit on main, as git sends it to receive-pack.
  const bare = repoDir(oldApp.reposRoot, "alice", "garden");
  const clone = join(work, "garden");
  await git(["clone", "-q", bare, clone], work);
  await Bun.write(join(clone, "b.txt"), "b");
  await git(["add", "b.txt"], clone);
  await git(["commit", "-q", "-m", "b"], clone);
  const before = text(await git(["rev-parse", "origin/main"], clone));
  const after = text(await git(["rev-parse", "HEAD"], clone));
  const pack = await git(
    ["pack-objects", "--stdout", "--revs", "--quiet"],
    clone,
    `${after}\n^${before}\n`,
  );
  const command = pktLine(`${before} ${after} refs/heads/main\0report-status\n`);
  const flush = new TextEncoder().encode("0000");
  const body = new Uint8Array(command.length + flush.length + pack.length);
  body.set(command, 0);
  body.set(flush, command.length);
  body.set(pack, command.length + flush.length);
  const url = `http://127.0.0.1:${port}`;
  const push = (ms: number) =>
    fetch(`${url}/alice/garden.git/git-receive-pack`, {
      method: "POST",
      headers: {
        authorization: `Basic ${btoa(`alice:${token}`)}`,
        "content-type": "application/x-git-receive-pack-request",
      },
      body: trickle(body, ms),
    });
  return { oldApp, oldServers, port, url, bare, after, push };
}
test("a push in flight on the old forge completes across the roll, and no connection is refused", async () => {
  const { oldApp, oldServers, port, url, bare, after, push: pushOver } = await forgeWithPush();
  const dataDir = oldApp.config.dataDir;
  const started = Date.now();
  const push = pushOver(PUSH_MS);
  // The push is the old forge's before the release exists.
  await Bun.sleep(500);
  const newApp = createApp({ dataDir, port, selfSwap: true });
  const newServers = await serve(newApp);
  let looping = true;
  let asked = 0;
  let refused = 0;
  const client = (async () => {
    while (looping) {
      asked++;
      try {
        const res = await fetch(`${url}/api/v1/health`, {
          headers: { connection: "close" },
          signal: AbortSignal.timeout(3_000),
        });
        await res.arrayBuffer();
      } catch {
        refused++;
      }
      await Bun.sleep(20);
    }
  })();
  // Drain start: the old forge stops listening, keeping what it holds.
  const drained = oldServers.stopAccepting(PUSH_MS * 2);
  const answer = await push;
  const report = await answer.text();
  expect(answer.status).toBe(200);
  expect(report).toContain("unpack ok");
  expect(report).toContain("ok refs/heads/main");
  expect(Date.now() - started).toBeGreaterThanOrEqual(PUSH_MS * 0.9);
  await drained;
  looping = false;
  await client;
  expect(asked).toBeGreaterThan(100);
  if (migrates) expect(refused).toBe(0);
  expect(text(await git(["rev-parse", "refs/heads/main"], bare))).toBe(after);
  // Nobody else listens: the old forge let go of the port at drain start.
  // `stop` closes the listener asynchronously, so ask until it is gone.
  newServers.stop();
  let orphan = "answered";
  for (const until = Date.now() + 10_000; Date.now() < until; ) {
    orphan = await fetch(`${url}/api/v1/health`, {
      headers: { connection: "close" },
      signal: AbortSignal.timeout(2_000),
    }).then(
      async (res) => {
        await res.arrayBuffer();
        return "answered";
      },
      () => "refused",
    );
    if (orphan === "refused") break;
    await Bun.sleep(100);
  }
  expect(orphan).toBe("refused");
  oldServers.stop();
  newApp.close();
  oldApp.close();
}, PUSH_MS * 3);
test("handover.drain lets go of the port before the rest of the drain, and closes only after the push is answered", async () => {
  const { oldApp, oldServers, url, bare, after, push: pushOver } = await forgeWithPush();
  const HELD_MS = 4_000;
  const REST_MS = 2_000;
  const push = pushOver(HELD_MS).then(async (res) => {
    const report = await res.text();
    answeredAt = Date.now();
    return { status: res.status, report };
  });
  let answeredAt = 0;
  let restDoneAt = 0;
  let closedAt = 0;
  // Wired as `main.ts` wires it; `rest` stands in for the shutdown's other
  // steps (crew, apps, checks), `close` for its last.
  const handOver = drainOnHandover(
    oldServers,
    async () => {
      await Bun.sleep(REST_MS);
      restDoneAt = Date.now();
    },
    () => {
      closedAt = Date.now();
      oldServers.stop();
    },
  );
  handover.drain = () => void handOver();
  try {
    await Bun.sleep(500);
    handover.drain();
    // Nobody else listens here, so a refused connection is this listener gone.
    let refusedAt = 0;
    while (restDoneAt === 0) {
      const gone = await fetch(`${url}/api/v1/health`, {
        headers: { connection: "close" },
        signal: AbortSignal.timeout(1_000),
      }).then(
        async (res) => {
          await res.arrayBuffer();
          return false;
        },
        () => true,
      );
      if (gone) {
        refusedAt = Date.now();
        break;
      }
      await Bun.sleep(50);
    }
    expect(refusedAt).toBeGreaterThan(0);
    expect(restDoneAt).toBe(0);
    const answered = await push;
    expect(answered.status).toBe(200);
    expect(answered.report).toContain("ok refs/heads/main");
    while (closedAt === 0) await Bun.sleep(50);
    expect(closedAt).toBeGreaterThanOrEqual(answeredAt);
    expect(closedAt).toBeGreaterThanOrEqual(restDoneAt);
    expect(text(await git(["rev-parse", "refs/heads/main"], bare))).toBe(after);
  } finally {
    delete handover.drain;
    oldServers.stop();
    oldApp.close();
  }
}, 30_000);
		@@ -104,6 +104,7 @@ export const GATE: readonly string[] = [
104	104	"test/release-preflight.test.ts",
105	105	"test/release-rollback.test.ts",
106	106	"test/release-swap.test.ts",
	107	"test/release-handover.test.ts",
107	108	"test/threats-doc.test.ts",
108	109	"test/returns.test.ts",
109	110	"test/review-route.test.ts",
		@@ -1639,6 +1639,13 @@ async function handleHook(app: App, request: Request): Promise<Response> {
1639	1639	export interface Servers {
1640	1640	readonly url: string;
1641	1641	readonly port: number;
	1642	/**
	1643	* Stop taking connections and let those in flight finish: what a process
	1644	* handing its port to a release beside it does at drain start (#2040).
	1645	* Resolves once no request is in flight, or `budgetMs` has passed; `stop`
	1646	* then cuts whatever is left.
	1647	*/
	1648	stopAccepting(budgetMs: number): Promise<void>;
1642	1649	stop(): void;
1643	1650	}
1644	1651
		@@ -1665,6 +1672,9 @@ function withHeader(response: Response, name: string, value: string): Response {
1665	1672	}
1666	1673	}
1667	1674
	1675	/** How often a draining listener looks whether its last request is answered. */
	1676	const INFLIGHT_POLL_MS = 50;
	1677
1668	1678	/** The hook sockets a data directory may hold: this forge's, and others'. */
1669	1679	const SOCKET_FILE = /^forge(?:-[a-z0-9]+)?\.sock$/;
1670	1680
		@@ -1919,6 +1929,14 @@ export async function serve(app: App): Promise<Servers> {
1919	1929	return {
1920	1930	url: `http://localhost:${port}`,
1921	1931	port,
	1932	async stopAccepting(budgetMs) {
	1933	// Closing the listener leaves the port to the process beside this one;
	1934	// the connections already here are answered to the end.
	1935	web.stop(false).catch(() => {});
	1936	const deadline = Date.now() + budgetMs;
	1937	while (web.pendingRequests > 0 && Date.now() < deadline)
	1938	await Bun.sleep(INFLIGHT_POLL_MS);
	1939	},
1922	1940	stop() {
1923	1941	endOpen();
1924	1942	clearInterval(sweep);
		@@ -6,7 +6,7 @@
6	6
7	7	import { startSampler } from "./boards/sampler.ts";
8	8	import { sweepStanding } from "./crew/standing.ts";
9		import { handover } from "./release/swap.ts";
	9	import { drainOnHandover, handover } from "./release/swap.ts";
10	10	import { parentUrlProblem } from "./core/parent_releases.ts";
11	11	import { releaseEnv } from "./release/env.ts";
12	12	import { createApp } from "./app.ts";
		@@ -487,8 +487,14 @@ const leaseTimer = setInterval(() => {
487	487	releaseOrphans(app.db);
488	488	}, 30_000);
489	489
490		// A release that read healthy beside this process takes over (#1663).
491		handover.drain = () => shutdown("SIGTERM");
	490	// A release that read healthy beside this process takes over (#1663),
	491	// and this one lets go of the port before it drains (#2040).
	492	const handOver = drainOnHandover(servers, () => drain("SIGTERM"), close);
	493	handover.drain = () => {
	494	if (draining) return;
	495	draining = true;
		sentinelcrew 29d ago outdated This wiring is the fix, and no test reaches it: revert it and `release-handover.test.ts` still passes. Move it into an exported function and test it through `handover.drain()`.
	496	void handOver();
	497	};
492	498	handover.ssh = ssh !== undefined;
493	499	process.on("SIGINT", shutdown);
494	500	process.on("SIGTERM", shutdown);
		@@ -505,6 +511,11 @@ let draining = false;
505	511	function shutdown(signal: NodeJS.Signals): void {
506	512	if (draining) return;
507	513	draining = true;
	514	void drain(signal).finally(close);
	515	}
	516
	517	/** Everything a leaving process finishes before it goes: `shutdown` above. */
	518	function drain(signal: NodeJS.Signals): Promise<unknown> {
508	519	const sessions = ssh?.sessions() ?? 0;
509	520	startDraining(app.log, signal, { sshSessions: sessions });
510	521	// Streams end now, with a resume hint: their browsers reconnect to the
		@@ -518,7 +529,7 @@ function shutdown(signal: NodeJS.Signals): void {
518	529	// The platform's SIGTERM: the listener stays open while routers learn
519	530	// this process is going. A person's Ctrl-C does not wait for routers.
520	531	const linger = signal === "SIGTERM" ? Bun.sleep(lingerMs ?? 0) : undefined;
521		Promise.all([
	532	return Promise.all([
522	533	linger,
523	534	prompts.stop(),
524	535	notifier.stop(),
		@@ -538,12 +549,14 @@ function shutdown(signal: NodeJS.Signals): void {
538	549	});
539	550	return ssh?.drain();
540	551	})
541		.catch(() => {})
542		.finally(() => {
543		app.log.info("forge shutting down");
544		ssh?.stop();
545		servers.stop();
546		app.close();
547		process.exit(0);
548		});
	552	.catch(() => {});
	553	}
	554
	555	/** The last step of a drain: cut what is left and exit. */
	556	function close(): void {
	557	app.log.info("forge shutting down");
	558	ssh?.stop();
	559	servers.stop();
	560	app.close();
	561	process.exit(0);
549	562	}
		@@ -91,11 +91,40 @@ export async function startRelease(input: SwapInput): Promise<Swapped> {
91	91	};
92	92	}
93	93
	94	/**
	95	* How long a process that handed its port over keeps answering what it had
	96	* already taken — a long push or clone — before it goes (#2040).
	97	*/
	98	export const HANDOVER_INFLIGHT_MS = 300_000;
	99
	100	/**
		sentinelcrew 29d ago This was inserted between `handover`'s doc comment (ending on line 99) and `handover` itself, so that comment no longer documents anything. Move the constant and its comment above the `handover` doc comment.
	101	* The drain a handover runs (#2040). The release listens on this port too,
	102	* so this process stops accepting at once: every new connection is the
	103	* release's. What is in flight here is answered to the end (up to
	104	* `budgetMs`) beside `rest` — the shutdown's other steps — and only once
	105	* both are done does `close` cut what remains. Runs once.
	106	*/
	107	export function drainOnHandover(
	108	servers: { stopAccepting(budgetMs: number): Promise<void> },
	109	rest: () => Promise<unknown>,
	110	close: () => void,
	111	budgetMs = HANDOVER_INFLIGHT_MS,
	112	): () => Promise<void> {
	113	let done: Promise<void> \| undefined;
	114	return () =>
	115	(done ??= (async () => {
	116	const inFlight = servers.stopAccepting(budgetMs);
	117	await Promise.all([inFlight, rest()]).catch(() => {});
	118	close();
	119	})());
	120	}
	121
94	122	/**
95	123	* What the running forge does once the release is well: drain, as a SIGTERM
96		* does. Set at boot, with whether this process holds the SSH listener (a new
97		* process cannot bind that port while this one does, and a release must not
98		* quietly close the terminal).
	124	* does, but letting go of the port first (`drainOnHandover`). Set at boot,
	125	* with whether this process holds the SSH listener (a new process cannot
	126	* bind that port while this one does, and a release must not quietly close
	127	* the terminal).
99	128	*/
100	129	export const handover: { drain?: () => void; ssh?: boolean; migrate?: boolean } = {};
101	130
		@@ -0,0 +1,274 @@
	1	/**
	2	* A roll never cuts an in-flight push (#2040). Two forges on one data
	3	* directory share the port (SO_REUSEPORT), as during a release swap: a push
	4	* that started on the old one trickles in for thirty seconds while the new
	5	* one comes up and the old one drains. The old stops accepting at drain
	6	* start — new connections are the new forge's — and answers the push to the
	7	* end before it goes.
	8	*/
	9
	10	import { afterAll, expect, test } from "bun:test";
	11	import { readFileSync } from "node:fs";
	12	import { mkdtemp, rm } from "node:fs/promises";
	13	import { tmpdir } from "node:os";
	14	import { join } from "node:path";
	15	import { createApp } from "../src/app.ts";
	16	import { commitToBranch, createRepo } from "../src/actions/repos.ts";
	17	import { createToken } from "../src/core/tokens.ts";
	18	import { createUser } from "../src/core/users.ts";
	19	import { repoDir } from "../src/git/repos.ts";
	20	import { drainOnHandover, handover } from "../src/release/swap.ts";
	21	import { serve } from "../src/http/server.ts";
	22	import { gitClientEnv } from "./helpers/git-client.ts";
	23
	24	const PUSH_MS = 30_000;
	25
	26	const dirs: string[] = [];
	27	afterAll(async () => {
	28	for (const d of dirs) await rm(d, { recursive: true, force: true });
	29	});
	30
	31	// Without it, what sat in the old listener's accept queue is reset when it
	32	// closes — at drain start now, at the end before; either way the kernel's.
	33	const migrates = (() => {
	34	try {
	35	return readFileSync("/proc/sys/net/ipv4/tcp_migrate_req", "utf8").trim() === "1";
	36	} catch {
	37	return false;
	38	}
	39	})();
	40
	41	async function git(args: string[], cwd: string, stdin?: string): Promise<Uint8Array> {
	42	const out = Bun.spawnSync(["git", ...args], {
	43	cwd,
	44	env: gitClientEnv(),
	45	...(stdin === undefined ? {} : { stdin: new TextEncoder().encode(stdin) }),
	46	});
	47	if (out.exitCode !== 0) throw new Error(`git ${args.join(" ")}: ${out.stderr.toString()}`);
	48	return new Uint8Array(out.stdout);
	49	}
	50
	51	const text = (bytes: Uint8Array) => new TextDecoder().decode(bytes).trim();
	52
	53	function pktLine(line: string): Uint8Array {
	54	const body = new TextEncoder().encode(line);
	55	const head = (body.length + 4).toString(16).padStart(4, "0");
	56	return new Uint8Array([...new TextEncoder().encode(head), ...body]);
	57	}
	58
	59	/** `bytes` sent in pieces spread over `ms`, so the request stays in flight that long. */
	60	function trickle(bytes: Uint8Array, ms: number): ReadableStream<Uint8Array> {
	61	const pieces = 30;
	62	const size = Math.ceil(bytes.length / pieces);
	63	let i = 0;
	64	return new ReadableStream({
	65	async pull(controller) {
	66	if (i > 0) await Bun.sleep(ms / pieces);
	67	const piece = bytes.subarray(i * size, (i + 1) * size);
	68	i++;
	69	if (piece.length > 0) controller.enqueue(piece);
	70	if (i >= pieces) controller.close();
	71	},
	72	});
	73	}
	74
	75	/**
	76	* The old forge on a port of its own, alice's repository, and `push(ms)`: a
	77	* receive-pack of one new commit on main, trickled in over `ms`.
	78	*/
	79	async function forgeWithPush() {
	80	const dataDir = await mkdtemp(join(tmpdir(), "forge-handover-"));
	81	const work = await mkdtemp(join(tmpdir(), "forge-handover-work-"));
	82	dirs.push(dataDir, work);
	83	// The old forge picks the port; the release binds the one it got.
	84	const oldApp = createApp({ dataDir, port: 0, selfSwap: true });
	85	const oldServers = await serve(oldApp);
	86	const port = oldServers.port;
	87	const made = await createUser(oldApp.db, {
	88	trusted: true,
	89	username: "alice",
	90	email: "[email protected]",
	91	password: "a-good-passphrase",
	92	adminIfFirst: true,
	93	});
	94	if (!made.ok) throw new Error("user");
	95	const alice = made.value;
	96	const repo = await createRepo(oldApp, alice, { name: "garden", visibility: "public" });
	97	if (!repo.ok) throw new Error("repo");
	98	await commitToBranch(oldApp, alice, repo.value, {
	99	message: "seed",
	100	files: [{ path: "a.txt", content: "a" }],
	101	});
	102	const { token } = createToken(oldApp.db, alice.id, "push", "rw");
	103
	104	// The push: one commit on main, as git sends it to receive-pack.
	105	const bare = repoDir(oldApp.reposRoot, "alice", "garden");
	106	const clone = join(work, "garden");
	107	await git(["clone", "-q", bare, clone], work);
	108	await Bun.write(join(clone, "b.txt"), "b");
	109	await git(["add", "b.txt"], clone);
	110	await git(["commit", "-q", "-m", "b"], clone);
	111	const before = text(await git(["rev-parse", "origin/main"], clone));
	112	const after = text(await git(["rev-parse", "HEAD"], clone));
	113	const pack = await git(
	114	["pack-objects", "--stdout", "--revs", "--quiet"],
	115	clone,
	116	`${after}\n^${before}\n`,
	117	);
	118	const command = pktLine(`${before} ${after} refs/heads/main\0report-status\n`);
	119	const flush = new TextEncoder().encode("0000");
	120	const body = new Uint8Array(command.length + flush.length + pack.length);
	121	body.set(command, 0);
	122	body.set(flush, command.length);
	123	body.set(pack, command.length + flush.length);
	124
	125	const url = `http://127.0.0.1:${port}`;
	126	const push = (ms: number) =>
	127	fetch(`${url}/alice/garden.git/git-receive-pack`, {
	128	method: "POST",
	129	headers: {
	130	authorization: `Basic ${btoa(`alice:${token}`)}`,
	131	"content-type": "application/x-git-receive-pack-request",
	132	},
	133	body: trickle(body, ms),
	134	});
	135	return { oldApp, oldServers, port, url, bare, after, push };
	136	}
	137
	138	test("a push in flight on the old forge completes across the roll, and no connection is refused", async () => {
	139	const { oldApp, oldServers, port, url, bare, after, push: pushOver } = await forgeWithPush();
	140	const dataDir = oldApp.config.dataDir;
	141	const started = Date.now();
	142	const push = pushOver(PUSH_MS);
	143	// The push is the old forge's before the release exists.
	144	await Bun.sleep(500);
	145
	146	const newApp = createApp({ dataDir, port, selfSwap: true });
	147	const newServers = await serve(newApp);
	148
	149	let looping = true;
	150	let asked = 0;
	151	let refused = 0;
	152	const client = (async () => {
	153	while (looping) {
	154	asked++;
	155	try {
	156	const res = await fetch(`${url}/api/v1/health`, {
	157	headers: { connection: "close" },
	158	signal: AbortSignal.timeout(3_000),
	159	});
	160	await res.arrayBuffer();
	161	} catch {
	162	refused++;
	163	}
	164	await Bun.sleep(20);
	165	}
	166	})();
	167
	168	// Drain start: the old forge stops listening, keeping what it holds.
	169	const drained = oldServers.stopAccepting(PUSH_MS * 2);
	170
	171	const answer = await push;
	172	const report = await answer.text();
	173	expect(answer.status).toBe(200);
	174	expect(report).toContain("unpack ok");
	175	expect(report).toContain("ok refs/heads/main");
	176	expect(Date.now() - started).toBeGreaterThanOrEqual(PUSH_MS * 0.9);
	177	await drained;
	178
	179	looping = false;
	180	await client;
	181	expect(asked).toBeGreaterThan(100);
	182	if (migrates) expect(refused).toBe(0);
	183	expect(text(await git(["rev-parse", "refs/heads/main"], bare))).toBe(after);
	184
	185	// Nobody else listens: the old forge let go of the port at drain start.
	186	// `stop` closes the listener asynchronously, so ask until it is gone.
	187	newServers.stop();
	188	let orphan = "answered";
	189	for (const until = Date.now() + 10_000; Date.now() < until; ) {
	190	orphan = await fetch(`${url}/api/v1/health`, {
	191	headers: { connection: "close" },
	192	signal: AbortSignal.timeout(2_000),
	193	}).then(
	194	async (res) => {
	195	await res.arrayBuffer();
	196	return "answered";
	197	},
	198	() => "refused",
	199	);
	200	if (orphan === "refused") break;
	201	await Bun.sleep(100);
	202	}
	203	expect(orphan).toBe("refused");
	204
	205	oldServers.stop();
	206	newApp.close();
	207	oldApp.close();
	208	}, PUSH_MS * 3);
	209
	210	test("handover.drain lets go of the port before the rest of the drain, and closes only after the push is answered", async () => {
	211	const { oldApp, oldServers, url, bare, after, push: pushOver } = await forgeWithPush();
	212	const HELD_MS = 4_000;
	213	const REST_MS = 2_000;
	214	const push = pushOver(HELD_MS).then(async (res) => {
	215	const report = await res.text();
	216	answeredAt = Date.now();
	217	return { status: res.status, report };
	218	});
	219	let answeredAt = 0;
	220	let restDoneAt = 0;
	221	let closedAt = 0;
	222	// Wired as `main.ts` wires it; `rest` stands in for the shutdown's other
	223	// steps (crew, apps, checks), `close` for its last.
	224	const handOver = drainOnHandover(
	225	oldServers,
	226	async () => {
	227	await Bun.sleep(REST_MS);
	228	restDoneAt = Date.now();
	229	},
	230	() => {
	231	closedAt = Date.now();
	232	oldServers.stop();
	233	},
	234	);
	235	handover.drain = () => void handOver();
	236	try {
	237	await Bun.sleep(500);
	238	handover.drain();
	239
	240	// Nobody else listens here, so a refused connection is this listener gone.
	241	let refusedAt = 0;
	242	while (restDoneAt === 0) {
	243	const gone = await fetch(`${url}/api/v1/health`, {
	244	headers: { connection: "close" },
	245	signal: AbortSignal.timeout(1_000),
	246	}).then(
	247	async (res) => {
	248	await res.arrayBuffer();
	249	return false;
	250	},
	251	() => true,
	252	);
	253	if (gone) {
	254	refusedAt = Date.now();
	255	break;
	256	}
	257	await Bun.sleep(50);
	258	}
	259	expect(refusedAt).toBeGreaterThan(0);
	260	expect(restDoneAt).toBe(0);
	261
	262	const answered = await push;
	263	expect(answered.status).toBe(200);
	264	expect(answered.report).toContain("ok refs/heads/main");
	265	while (closedAt === 0) await Bun.sleep(50);
	266	expect(closedAt).toBeGreaterThanOrEqual(answeredAt);
	267	expect(closedAt).toBeGreaterThanOrEqual(restDoneAt);
	268	expect(text(await git(["rev-parse", "refs/heads/main"], bare))).toBe(after);
	269	} finally {
	270	delete handover.drain;
	271	oldServers.stop();
	272	oldApp.close();
	273	}
	274	}, 30_000);