Commit `6ab65cbbaf`

6ab65cbbaf383ca4d70efd05d53ff2952557e69e

parent: d51efc8bb8

Verified · cmc ci/build: success ci/test: success ci/vuln: success

cmc <hello@cleberg.net> · 2026-09-04 15:13 UTC

hook: one cat-file for the whole push, verified as it streams

pre-receive on a require-signed repository forked a cat-file per incoming
commit and built one JSON message holding every raw commit before sending
it. A 50k-commit first push forked 50k processes and held the history in
memory twice, once in the hook and once in the daemon.

One `cat-file --batch` now serves the push, and each commit goes over the
socket as its own JSON value, ending with a zero one. The daemon verifies
each as it arrives and holds one at a time.

Two things the streaming shape needs:

Object names are fed to git from their own goroutine. 4000 of them is
past a 64 KiB pipe, so writing them all before reading git's output is a
deadlock — git blocked writing its answer, the hook blocked writing its
question.

The subprocess runs under a context cancelled on return. Without it,
giving up part-way — the socket closing mid-push — leaves git blocked on
a pipe nobody reads and Wait blocked on git. The test for that hangs
rather than fails without the cancel, which is how it was found.

The daemon drains the stream to its end before answering, even once a
commit has failed. Answering early would leave the hook writing into a
socket nobody reads; draining costs a decode per commit and no
verification.

Closes #100

Layout: unified · split

cmd/gitbayd/hook.go +98 −15

	@@ -2,7 +2,9 @@ package main
2		2
3	import (	3	import (
4	"bufio"	4	"bufio"
		5	"context"
5	"fmt"	6	"fmt"
		7	"io"
6	"os"	8	"os"
7	"os/exec"	9	"os/exec"
8	"strconv"	10	"strconv"
	@@ -15,34 +17,115 @@ import (
15	"gitbay.org/gitbay/internal/policy"	17	"gitbay.org/gitbay/internal/policy"
16	)	18	)
17		19
18	// collectIncomingCommits lists the commits this push introduces and reads	20	// incomingSHAs lists the commits this push introduces, in order, without
19	// their raw objects. It runs in the hook process, which inherits git's	21	// duplicates. It runs in the hook process, which inherits git's quarantine
20	// quarantine environment — the daemon cannot see these objects yet.	22	// environment — the daemon cannot see these objects yet.
21	func collectIncomingCommits(updates []policy.RefUpdate) (hookd.CommitsPayload, error) {	23	func incomingSHAs(updates []policy.RefUpdate) ([]string, error) {
22	seen := map[string]bool{}	24	seen := map[string]bool{}
23	var payload hookd.CommitsPayload	25	var out []string
24	for _, u := range updates {	26	for _, u := range updates {
25	if u.IsDelete {	27	if u.IsDelete {
26	continue	28	continue
27	}	29	}
28	// Everything reachable from the new tip that no existing ref has.	30	// Everything reachable from the new tip that no existing ref has.
29	out, err := exec.Command("git", "rev-list", u.New, "--not", "--all").Output()	31	raw, err := exec.Command("git", "rev-list", u.New, "--not", "--all").Output()
30	if err != nil {	32	if err != nil {
31	return payload, fmt.Errorf("rev-list %s: %w", u.New, err)	33	return nil, fmt.Errorf("rev-list %s: %w", u.New, err)
32	}	34	}
33	for _, sha := range strings.Fields(string(out)) {	35	for _, sha := range strings.Fields(string(raw)) {
34	if seen[sha] {	36	if seen[sha] {
35	continue	37	continue
36	}	38	}
37	seen[sha] = true	39	seen[sha] = true
38	raw, err := exec.Command("git", "cat-file", "commit", sha).Output()	40	out = append(out, sha)
39	if err != nil {	41	}
40	return payload, fmt.Errorf("cat-file %s: %w", sha, err)	42	}
		43	return out, nil
		44	}
		45
		46	// streamIncomingCommits reads every incoming commit through one
		47	// `cat-file --batch` and hands each to emit as it arrives.
		48	//
		49	// This used to fork a cat-file per commit and build the whole payload in
		50	// memory before sending it: a 50k-commit first push to a protected branch
		51	// forked 50k processes and held 50k raw commits at once (#100). One
		52	// subprocess now serves the whole push, and nothing is accumulated.
		53	func streamIncomingCommits(updates []policy.RefUpdate, emit func(hookd.RawCommit) error) error {
		54	shas, err := incomingSHAs(updates)
		55	if err != nil {
		56	return err
		57	}
		58	if len(shas) == 0 {
		59	return nil
		60	}
		61	// Cancelling kills git on an early return. Without it, bailing out
		62	// part-way through a large push leaves git blocked writing into a
		63	// pipe nobody is reading and Wait blocked on git.
		64	ctx, cancel := context.WithCancel(context.Background())
		65	cmd := exec.CommandContext(ctx, "git", "cat-file", "--batch")
		66	stdin, err := cmd.StdinPipe()
		67	if err != nil {
		68	cancel()
		69	return err
		70	}
		71	stdout, err := cmd.StdoutPipe()
		72	if err != nil {
		73	cancel()
		74	return err
		75	}
		76	if err := cmd.Start(); err != nil {
		77	cancel()
		78	return fmt.Errorf("cat-file --batch: %w", err)
		79	}
		80	defer cmd.Wait() // second: reaps the process cancel just signalled
		81	defer cancel()
		82	// Feeding stdin from another goroutine: the pipe buffer is smaller
		83	// than 50k object names, so writing them all before reading would
		84	// block against a git that is blocked writing its own output.
		85	writeErr := make(chan error, 1)
		86	go func() {
		87	defer stdin.Close()
		88	w := bufio.NewWriter(stdin)
		89	for _, sha := range shas {
		90	if _, err := fmt.Fprintln(w, sha); err != nil {
		91	writeErr <- err
		92	return
41	}	93	}
42	payload.Commits = append(payload.Commits, hookd.RawCommit{SHA: sha, Raw: raw})
43	}	94	}
		95	writeErr <- w.Flush()
		96	}()
		97
		98	r := bufio.NewReader(stdout)
		99	for range shas {
		100	// Each record is "<oid> <type> <size>\n", then size bytes, then
		101	// a newline.
		102	header, err := r.ReadString('\n')
		103	if err != nil {
		104	return fmt.Errorf("cat-file --batch: %w", err)
		105	}
		106	fields := strings.Fields(header)
		107	if len(fields) != 3 {
		108	return fmt.Errorf("cat-file --batch: unexpected %q", strings.TrimSpace(header))
		109	}
		110	size, err := strconv.Atoi(fields[2])
		111	if err != nil {
		112	return fmt.Errorf("cat-file --batch: bad size in %q", strings.TrimSpace(header))
		113	}
		114	raw := make([]byte, size)
		115	if _, err := io.ReadFull(r, raw); err != nil {
		116	return fmt.Errorf("cat-file %s: %w", fields[0], err)
		117	}
		118	if _, err := r.Discard(1); err != nil {
		119	return fmt.Errorf("cat-file %s: %w", fields[0], err)
		120	}
		121	if err := emit(hookd.RawCommit{SHA: fields[0], Raw: raw}); err != nil {
		122	return err
		123	}
		124	}
		125	if err := <-writeErr; err != nil {
		126	return fmt.Errorf("cat-file --batch: %w", err)
44	}	127	}
45	return payload, nil	128	return nil
46	}	129	}
47		130
48	// hookCmd runs inside a git hook. It computes git facts here — the hook	131	// hookCmd runs inside a git hook. It computes git facts here — the hook
	@@ -88,8 +171,8 @@ func hookCmd() *cobra.Command {
88	RepoID: repoID,	171	RepoID: repoID,
89	UserID: userID,	172	UserID: userID,
90	Updates: updates,	173	Updates: updates,
91	}, func() (hookd.CommitsPayload, error) {	174	}, func(emit func(hookd.RawCommit) error) error {
92	return collectIncomingCommits(updates)	175	return streamIncomingCommits(updates, emit)
93	})	176	})
94	if err != nil {	177	if err != nil {
95	return fmt.Errorf("gitbay daemon unreachable: %w", err)	178	return fmt.Errorf("gitbay daemon unreachable: %w", err)

cmd/gitbayd/hook_test.go added +188

@@ -0,0 +1,188 @@
	1	package main
	2
	3	import (
	4	"fmt"
	5	"os"
	6	"os/exec"
	7	"strings"
	8	"testing"
	9
	10	"gitbay.org/gitbay/internal/hookd"
	11	"gitbay.org/gitbay/internal/policy"
	12	)
	13
	14	// fastImportRepo builds a repository with n commits on main in one
	15	// process. A loop of `git commit` would be n forks, which is the cost this
	16	// test exists to rule out of the code under test.
	17	func fastImportRepo(t *testing.T, n int) (string, string) {
	18	t.Helper()
	19	dir := t.TempDir()
	20	run := func(args ...string) {
	21	t.Helper()
	22	cmd := exec.Command("git", args...)
	23	cmd.Dir = dir
	24	if out, err := cmd.CombinedOutput(); err != nil {
	25	t.Fatalf("git %s: %v\n%s", strings.Join(args, " "), err, out)
	26	}
	27	}
	28	run("init", "-q", "--initial-branch=main", ".")
	29
	30	var b strings.Builder
	31	for i := 1; i <= n; i++ {
	32	fmt.Fprintf(&b, "commit refs/heads/main\n")
	33	fmt.Fprintf(&b, "mark :%d\n", i)
	34	fmt.Fprintf(&b, "author A U Thor <a@example.test> %d +0000\n", 1600000000+i)
	35	fmt.Fprintf(&b, "committer A U Thor <a@example.test> %d +0000\n", 1600000000+i)
	36	msg := fmt.Sprintf("commit %d", i)
	37	fmt.Fprintf(&b, "data %d\n%s\n", len(msg), msg)
	38	if i > 1 {
	39	fmt.Fprintf(&b, "from :%d\n", i-1)
	40	}
	41	fmt.Fprintf(&b, "M 644 inline f.txt\ndata %d\n%d\n", len(fmt.Sprint(i))+1, i)
	42	fmt.Fprintf(&b, "\n")
	43	}
	44	cmd := exec.Command("git", "fast-import", "--quiet")
	45	cmd.Dir = dir
	46	cmd.Stdin = strings.NewReader(b.String())
	47	if out, err := cmd.CombinedOutput(); err != nil {
	48	t.Fatalf("fast-import: %v\n%s", err, out)
	49	}
	50	// pre-receive runs before the ref moves, so the incoming objects are
	51	// present but reachable from nothing — which is what makes
	52	// `rev-list --not --all` list them. Dropping the ref reproduces that;
	53	// the objects stay until a gc that never runs here.
	54	head := gitOut(t, dir, "rev-parse", "main")
	55	run("update-ref", "-d", "refs/heads/main")
	56	return dir, head
	57	}
	58
	59	// streamIncomingCommits runs in the hook process, whose working directory
	60	// is the repository, so the test chdirs the same way git would.
	61	func inRepo(t *testing.T, dir string, f func()) {
	62	t.Helper()
	63	prev, err := os.Getwd()
	64	if err != nil {
	65	t.Fatal(err)
	66	}
	67	if err := os.Chdir(dir); err != nil {
	68	t.Fatal(err)
	69	}
	70	defer os.Chdir(prev)
	71	f()
	72	}
	73
	74	// A push large enough that its object names do not fit a pipe buffer:
	75	// writing them all before reading git's output deadlocks, which is why
	76	// stdin is fed from its own goroutine. 4000 names is roughly 160 KiB,
	77	// well past the usual 64 KiB pipe.
	78	func TestStreamIncomingCommitsLargePush(t *testing.T) {
	79	const n = 4000
	80	dir, head := fastImportRepo(t, n)
	81
	82	// No ref exists, so every commit is incoming — the shape of a
	83	// first push of an existing history.
	84	updates := []policy.RefUpdate{{Old: strings.Repeat("0", 40), New: head, Ref: "refs/heads/main"}}
	85
	86	var got []hookd.RawCommit
	87	inRepo(t, dir, func() {
	88	if err := streamIncomingCommits(updates, func(rc hookd.RawCommit) error {
	89	got = append(got, rc)
	90	return nil
	91	}); err != nil {
	92	t.Fatalf("streaming: %v", err)
	93	}
	94	})
	95	if len(got) != n {
	96	t.Fatalf("streamed %d commits, want %d", len(got), n)
	97	}
	98	seen := map[string]bool{}
	99	for _, rc := range got {
	100	if seen[rc.SHA] {
	101	t.Fatalf("%s streamed twice", rc.SHA)
	102	}
	103	seen[rc.SHA] = true
	104	// The raw object is what signature verification parses; a record
	105	// misread by a byte would still look plausible here without this.
	106	if !strings.HasPrefix(string(rc.Raw), "tree ") {
	107	t.Fatalf("%s does not look like a raw commit: %.60q", rc.SHA, rc.Raw)
	108	}
	109	if !strings.Contains(string(rc.Raw), "A U Thor <a@example.test>") {
	110	t.Fatalf("%s raw object is truncated: %.200q", rc.SHA, rc.Raw)
	111	}
	112	}
	113	if !seen[head] {
	114	t.Fatal("the pushed tip was not among the streamed commits")
	115	}
	116	}
	117
	118	// A ref that already exists contributes nothing, and a delete contributes
	119	// nothing: neither introduces an object to verify.
	120	func TestStreamIncomingCommitsNothingToDo(t *testing.T) {
	121	dir, head := fastImportRepo(t, 3)
	122	zero := strings.Repeat("0", 40)
	123
	124	inRepo(t, dir, func() {
	125	n := 0
	126	err := streamIncomingCommits([]policy.RefUpdate{
	127	{Old: head, New: zero, Ref: "refs/heads/main", IsDelete: true},
	128	}, func(hookd.RawCommit) error { n++; return nil })
	129	if err != nil \|\| n != 0 {
	130	t.Fatalf("delete streamed %d commits (%v)", n, err)
	131	}
	132	})
	133
	134	// With the ref restored the tip is reachable, so nothing is incoming:
	135	// a push of what the repository already has verifies nothing.
	136	cmd := exec.Command("git", "update-ref", "refs/heads/main", head)
	137	cmd.Dir = dir
	138	if out, err := cmd.CombinedOutput(); err != nil {
	139	t.Fatalf("update-ref: %v\n%s", err, out)
	140	}
	141	inRepo(t, dir, func() {
	142	n := 0
	143	err := streamIncomingCommits([]policy.RefUpdate{
	144	{Old: head, New: head, Ref: "refs/heads/main"},
	145	}, func(hookd.RawCommit) error { n++; return nil })
	146	if err != nil \|\| n != 0 {
	147	t.Fatalf("already-present tip streamed %d commits (%v)", n, err)
	148	}
	149	})
	150	}
	151
	152	// An error from the callback — the socket going away mid-push — stops the
	153	// walk instead of reading the rest of the history into nothing.
	154	func TestStreamIncomingCommitsCallbackError(t *testing.T) {
	155	// Large enough that git is still writing when the callback gives up:
	156	// without killing it, git blocks on a full pipe and Wait blocks on
	157	// git, and this test hangs rather than fails.
	158	dir, head := fastImportRepo(t, 4000)
	159	inRepo(t, dir, func() {
	160	n := 0
	161	err := streamIncomingCommits(
	162	[]policy.RefUpdate{{Old: strings.Repeat("0", 40), New: head, Ref: "refs/heads/main"}},
	163	func(hookd.RawCommit) error {
	164	n++
	165	if n == 5 {
	166	return fmt.Errorf("socket closed")
	167	}
	168	return nil
	169	})
	170	if err == nil {
	171	t.Fatal("callback error did not stop the walk")
	172	}
	173	if n != 5 {
	174	t.Fatalf("kept streaming after the error: %d commits", n)
	175	}
	176	})
	177	}
	178
	179	func gitOut(t *testing.T, dir string, args ...string) string {
	180	t.Helper()
	181	cmd := exec.Command("git", args...)
	182	cmd.Dir = dir
	183	out, err := cmd.Output()
	184	if err != nil {
	185	t.Fatalf("git %s: %v", strings.Join(args, " "), err)
	186	}
	187	return strings.TrimSpace(string(out))
	188	}

internal/hookd/hookd.go +37 −21

 	Updates []policy.RefUpdate `json:"updates"`
 }
+// RawCommit is one incoming commit object. When NeedCommits is set the
+// hook streams these one per JSON value and ends with a zero one, rather
+// than sending a single message holding every commit in the push: an
+// initial push of a large history is tens of thousands of them (#100).
 type RawCommit struct {
 	SHA string `json:"sha"`
 	Raw []byte `json:"raw"`
 }
-// CommitsPayload is the hook's second message when NeedCommits was set.
+// Done marks the end of the commit stream.
-type CommitsPayload struct {
+func (c RawCommit) Done() bool { return c.SHA == "" }
-	Commits []RawCommit `json:"commits"`
-}
 type Response struct {
 	Allow       bool   `json:"allow"`
 	if err := enc.Encode(Response{Allow: true, NeedCommits: true}); err != nil {
 		return
 	}
-	var payload CommitsPayload
+	// Each commit is verified as it arrives, so nothing holds the push in
-	if err := dec.Decode(&payload); err != nil {
+	// memory. The first refusal decides the answer, but the stream is
-		enc.Encode(Response{Allow: false, Message: "bad commits payload"})
+	// still drained to its end before replying: the hook is writing, and
-		return
+	// answering early would leave it writing into a socket nobody reads.
-	}
+	// Draining costs a decode per commit and no verification.
 	db := store.SigDB{Store: s.st}
-	for _, rc := range payload.Commits {
+	refusal := ""
+	for {
+		var rc RawCommit
+		if err := dec.Decode(&rc); err != nil {
+			enc.Encode(Response{Allow: false, Message: "bad commits payload"})
+			return
+		}
+		if rc.Done() {
+			break
+		}
+		if refusal != "" {
+			continue
+		}
 		parsed, err := sig.ParseCommit(rc.Raw)
 		if err != nil {
-			enc.Encode(Response{Allow: false, Message: fmt.Sprintf("unparseable commit %s", rc.SHA)})
+			refusal = fmt.Sprintf("unparseable commit %s", rc.SHA)
-			return
+			continue
 		}
 		res, err := sig.VerifyCommit(db, parsed)
 		if err != nil || res.State != sig.Verified {
 			if err == nil {
 				state = string(res.State)
 			}
-			enc.Encode(Response{Allow: false, Message: fmt.Sprintf(
+			refusal = fmt.Sprintf("this repository requires signed commits: %.10s is %s", rc.SHA, state)
-				"this repository requires signed commits: %.10s is %s", rc.SHA, state)})
-			return
 		}
 	}
+	if refusal != "" {
+		enc.Encode(Response{Allow: false, Message: refusal})
+		return
+	}
 	enc.Encode(Response{Allow: true})
 }
 	return "", false
 }
-// Ask sends one request from the hook process to the daemon. commits is
+// Ask sends one request from the hook process to the daemon. stream is
-// called if the daemon asks for the incoming commit objects.
+// called if the daemon asks for the incoming commit objects; it hands each
-func Ask(socketPath string, req Request, commits func() (CommitsPayload, error)) (Response, error) {
+// commit to the callback, which writes it on the wire.
+func Ask(socketPath string, req Request, stream func(emit func(RawCommit) error) error) (Response, error) {
 	conn, err := net.Dial("unix", socketPath)
 	if err != nil {
 		return Response{}, err
 	if !resp.NeedCommits {
 		return resp, nil
 	}
-	payload, err := commits()
+	if err := stream(func(rc RawCommit) error { return enc.Encode(rc) }); err != nil {
-	if err != nil {
 		return Response{}, err
 	}
-	if err := enc.Encode(payload); err != nil {
+	if err := enc.Encode(RawCommit{}); err != nil { // end of stream
 		return Response{}, err
 	}
 	err = dec.Decode(&resp)

Commit 6ab65cbbaf

Commit `6ab65cbbaf`