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

merged merged by cmc on 2026-09-04 15:58 UTC · krz/gitbay:hook-batch into main

3 files changed, +323 −36

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)