feat(hooks): stream push-hook output to client live over sideband · xeiaso.net/objgit@53aed85

+22 -7

CLAUDE.md

··· 101 101 ### Push hooks (`hooks.go`, sandboxed via kefka) 102 102 103 103 When `-allow-hooks` is set, a successful `receive-pack` runs the repository's 104 - `.objgit/hooks/receive-pack` script. Because `transport.ReceivePack` does not 105 - report which refs it changed, `receivePack` (the wrapper both transports call 106 - instead of `transport.ReceivePack` directly) snapshots branch refs before and 107 - after and diffs them (`snapshotRefs`/`diffRefs`). For each created/updated 108 - branch it spawns an **async** goroutine (tracked by `daemon.hookWG`, drained on 109 - shutdown) — hooks run after the client already has its response and **cannot 110 - reject a push**. Deleted branches are skipped. 104 + `.objgit/hooks/receive-pack` script. The hook output is **streamed to the 105 + pushing client live**, rendered as `remote: ...` lines, so hooks run 106 + **synchronously**: the client waits for them to finish (bounded by 107 + `-hook-timeout`) before the push completes. They still **cannot reject a push** — 108 + they run after refs are updated and report-status is sent (post-receive 109 + semantics). Deleted branches are skipped. 110 + 111 + The streaming forces a small fork of go-git's `transport.ReceivePack` into 112 + `cmd/objgitd/receivepack.go` (`receivePackStreaming`): go-git constructs the 113 + sideband `Muxer` internally and sends the closing flush-pkt before returning, so 114 + there is no public seam to inject `remote:` progress. The fork adds one 115 + `onUpdated(progress io.Writer)` callback, invoked after report-status but before 116 + that final flush; `progress` writes to the sideband `ProgressMessage` channel 117 + (band 2) when the client negotiated sideband, or is `nil` otherwise (hooks then 118 + fall back to `slog`-only output). All three transports call `d.receivePack` 119 + (`hooks.go`), which drives the fork: because `transport.ReceivePack` does not 120 + report which refs it changed, it snapshots branch refs before and after and 121 + diffs them (`snapshotRefs`/`diffRefs`) inside `onUpdated`, then runs each hook 122 + synchronously, streaming through `progress`. **HTTP** additionally wraps its 123 + `ResponseWriter` in a flush-on-write writer (`flushWriter` in `http.go`) so 124 + `net/http` buffering doesn't hold the `remote:` lines back; git:// and SSH write 125 + through a live socket and need no such wrapper. 111 126 112 127 The script is read from the pushed commit's tree, so a branch carries its own 113 128 hook. It runs in a **kefka** virtual shell (`tangled.org/xeiaso.net/kefka`),

+6 -6

cmd/objgitd/example_hook_test.go

··· 60 60 writeFile(t, filepath.Join(work, ".objgit", "hooks", "receive-pack"), string(example)) 61 61 runGit(t, work, "add", ".") 62 62 runGit(t, work, "commit", "-m", "example") 63 - runGit(t, work, "push", remote, "main") 64 - 65 - waitForLog(t, &logBuf, "hook: finished", 30*time.Second) 63 + pushOut := runGit(t, work, "push", remote, "main") 66 64 65 + // Hooks run synchronously and stream stdout/stderr to the client over the 66 + // sideband, so the example hook's output appears in the push output. 67 67 logs := logBuf.String() 68 68 if strings.Contains(logs, "with errors") { 69 69 t.Fatalf("example hook errored; logs:\n%s", logs) 70 70 } 71 71 for _, want := range []string{"go module: example.test/thing", "hook done", "manifest"} { 72 - if !strings.Contains(logs, want) { 73 - t.Errorf("example hook output missing %q; logs:\n%s", want, logs) 72 + if !strings.Contains(pushOut, want) { 73 + t.Errorf("example hook output missing %q; push output:\n%s", want, pushOut) 74 74 } 75 75 } 76 - t.Logf("logs:\n%s", logs) 76 + t.Logf("push output:\n%s\nlogs:\n%s", pushOut, logs) 77 77 }

-3

cmd/objgitd/git_protocol.go

··· 9 9 "net" 10 10 "net/url" 11 11 "strings" 12 - "sync" 13 12 "time" 14 13 15 14 "github.com/go-git/go-billy/v6" ··· 62 61 // allowHooks gates running .objgit/hooks/receive-pack after a push. 63 62 allowHooks bool 64 63 hookTimeout time.Duration 65 - // hookWG tracks in-flight async hooks so shutdown can drain them. 66 - hookWG sync.WaitGroup 67 64 } 68 65 69 66 // Serve accepts connections on l until ctx is cancelled or Accept fails.

+49 -37

cmd/objgitd/hooks.go

··· 75 75 return updates 76 76 } 77 77 78 - // receivePack runs transport.ReceivePack and, when hooks are enabled, fires the 79 - // repository's receive-pack hook for each updated branch once the push succeeds. 80 - // rpStorer is what ReceivePack writes through (the git:// path hides the 81 - // PackfileWriter capability via streamingStorer); readStorer is the underlying 82 - // storer used for ref snapshots and hook checkouts. 78 + // receivePack runs the receive-pack service and, when hooks are enabled, fires 79 + // the repository's receive-pack hook for each updated branch once the push 80 + // succeeds — synchronously, streaming hook output to the client over the 81 + // sideband progress channel (rendered as "remote: " lines) before the response 82 + // stream is closed. rpStorer is what the service writes through (the git:// and 83 + // SSH paths hide the PackfileWriter capability via streamingStorer); readStorer 84 + // is the underlying storer used for ref snapshots and hook checkouts. 83 85 func (d *daemon) receivePack(ctx context.Context, rpStorer, readStorer storage.Storer, repoPath string, r io.ReadCloser, w io.WriteCloser, req *transport.ReceivePackRequest) error { 84 - var before map[plumbing.ReferenceName]plumbing.Hash 85 - if d.allowHooks { 86 - var err error 87 - if before, err = snapshotRefs(readStorer); err != nil { 88 - slog.Warn("hook: ref snapshot before push failed", "path", repoPath, "err", err) 89 - } 90 - } 91 - 92 - if err := transport.ReceivePack(ctx, rpStorer, r, w, req); err != nil { 93 - return err 94 - } 95 86 if !d.allowHooks { 96 - return nil 87 + return receivePackStreaming(ctx, rpStorer, r, w, req, nil) 97 88 } 98 89 99 - after, err := snapshotRefs(readStorer) 90 + before, err := snapshotRefs(readStorer) 100 91 if err != nil { 101 - slog.Error("hook: ref snapshot after push failed", "path", repoPath, "err", err) 102 - return nil 92 + slog.Warn("hook: ref snapshot before push failed", "path", repoPath, "err", err) 103 93 } 104 94 105 - updates := diffRefs(before, after) 106 - if len(updates) == 0 { 107 - return nil 95 + // onUpdated runs after refs are updated and report-status is sent, but 96 + // before the response stream closes, so hook output reaches the client live. 97 + // progress is the sideband band-2 writer, or nil when the client did not 98 + // negotiate sideband (hooks then fall back to logging only). 99 + onUpdated := func(progress io.Writer) { 100 + after, err := snapshotRefs(readStorer) 101 + if err != nil { 102 + slog.Error("hook: ref snapshot after push failed", "path", repoPath, "err", err) 103 + return 104 + } 105 + updates := diffRefs(before, after) 106 + if len(updates) == 0 { 107 + return 108 + } 109 + d.runHooks(repoPath, "receive-pack", readStorer, updates, progress) 108 110 } 109 111 110 - d.hookWG.Add(1) 111 - go func() { 112 - defer d.hookWG.Done() 113 - d.runHooks(repoPath, "receive-pack", readStorer, updates) 114 - }() 115 - return nil 112 + return receivePackStreaming(ctx, rpStorer, r, w, req, onUpdated) 116 113 } 117 114 118 - // runHooks executes the receive-pack hook once per non-deleted branch update. 119 - func (d *daemon) runHooks(repoPath, service string, st storage.Storer, updates []refUpdate) { 115 + // runHooks executes the receive-pack hook once per non-deleted branch update, 116 + // streaming each hook's output to progress (nil = log only). 117 + func (d *daemon) runHooks(repoPath, service string, st storage.Storer, updates []refUpdate, progress io.Writer) { 120 118 for _, u := range updates { 121 119 if u.New.IsZero() { 122 120 continue // branch deletion: nothing to check out 123 121 } 124 - d.runHook(repoPath, service, st, u) 122 + d.runHook(repoPath, service, st, u, progress) 125 123 } 126 124 } 127 125 128 126 // runHook looks up .objgit/hooks/<service> in the updated commit's tree and, if 129 127 // present, runs it in a kefka shell with /src bound to a read-only view of that 130 - // tree and /tmp to writable scratch. Output and exit status are logged only. 131 - func (d *daemon) runHook(repoPath, service string, st storage.Storer, u refUpdate) { 128 + // tree and /tmp to writable scratch. When progress is non-nil, hook stdout and 129 + // stderr stream to it (the client's sideband, rendered as "remote: " lines); 130 + // otherwise output is buffered and logged. Exit status is always logged. 131 + func (d *daemon) runHook(repoPath, service string, st storage.Storer, u refUpdate, progress io.Writer) { 132 132 log := slog.With("repo", repoPath, "service", service, "ref", u.Name.String(), "sha", u.New.String()) 133 133 134 134 commit, err := object.GetCommit(st, u.New) ··· 161 161 ctx, cancel := context.WithTimeout(context.Background(), d.hookTimeout) 162 162 defer cancel() 163 163 164 + // When the client negotiated sideband, stream stdout+stderr straight to it 165 + // ("remote: " lines); otherwise buffer for the log. git does not distinguish 166 + // the two streams on the wire, so both go to the same place. 164 167 var outBuf, errBuf bytes.Buffer 168 + stdout, stderr := io.Writer(&outBuf), io.Writer(&errBuf) 169 + streaming := progress != nil 170 + if streaming { 171 + stdout, stderr = progress, progress 172 + } 165 173 166 174 reg := registry.New() 167 175 coreutils.Register(reg) ··· 195 203 } 196 204 sh, err = interp.New( 197 205 interp.Env(env), 198 - interp.StdIO(stdin, &outBuf, &errBuf), 206 + interp.StdIO(stdin, stdout, stderr), 199 207 interp.ExecHandlers(middleware), 200 - interp.CallHandler(kefkash.CallHandler(reg, fsys, &outBuf, &errBuf)), 208 + interp.CallHandler(kefkash.CallHandler(reg, fsys, stdout, stderr)), 201 209 interp.StatHandler(kefkash.FsysStatHandler(reg, fsys)), 202 210 interp.OpenHandler(kefkash.FsysOpenHandler(reg, fsys)), 203 211 interp.ReadDirHandler2(kefkash.FsysReadDirHandler(reg, fsys)), ··· 218 226 219 227 var exit interp.ExitStatus 220 228 isExit := errors.As(runErr, &exit) 221 - attrs := []any{"exit", int(exit), "stdout", outBuf.String(), "stderr", errBuf.String()} 229 + attrs := []any{"exit", int(exit)} 230 + if !streaming { 231 + // When streaming, output already reached the client; don't duplicate it. 232 + attrs = append(attrs, "stdout", outBuf.String(), "stderr", errBuf.String()) 233 + } 222 234 if runErr != nil { 223 235 if !isExit { 224 236 attrs = append(attrs, "err", runErr)

+15 -14

cmd/objgitd/hooks_test.go

··· 144 144 runGit(t, work, "add", ".") 145 145 runGit(t, work, "commit", "-m", "with hook") 146 146 147 - runGit(t, work, "push", remote, "main") 147 + pushOut := runGit(t, work, "push", remote, "main") 148 148 149 - // The hook runs asynchronously after the push response; wait for it to 150 - // finish (it ends with an error because of the /src write attempt). 151 - waitForLog(t, &logBuf, "hook: finished", 30*time.Second) 152 - 149 + // Hooks now run synchronously and stream stdout/stderr to the client over 150 + // the sideband, so by the time push returns the hook has finished and its 151 + // output rides along in the push output as "remote:" lines. 153 152 logs := logBuf.String() 154 153 if !strings.Contains(logs, "hook: running") { 155 154 t.Fatalf("hook did not run; logs:\n%s", logs) 156 155 } 157 - // /src is readable and /tmp is writable. 156 + // /src is readable and /tmp is writable; their output streamed to the client. 158 157 for _, want := range []string{"hello from repo", "built"} { 159 - if !strings.Contains(logs, want) { 160 - t.Errorf("hook output missing %q; logs:\n%s", want, logs) 158 + if !strings.Contains(pushOut, want) { 159 + t.Errorf("push output missing streamed hook output %q; output:\n%s", want, pushOut) 161 160 } 162 161 } 163 - // Writing to /src is rejected and aborts the script. 162 + // Output reaches the client over the sideband, rendered with a "remote:" prefix. 163 + if !strings.Contains(pushOut, "remote:") { 164 + t.Errorf("hook output not streamed as remote progress; output:\n%s", pushOut) 165 + } 166 + // Writing to /src is rejected and aborts the script (logged, not streamed). 164 167 if !strings.Contains(logs, "read-only filesystem") { 165 168 t.Errorf("expected read-only error when writing /src; logs:\n%s", logs) 166 169 } 167 - if strings.Contains(logs, "WROTE_SRC") { 168 - t.Errorf("hook was able to write to read-only /src; logs:\n%s", logs) 170 + if strings.Contains(pushOut, "WROTE_SRC") { 171 + t.Errorf("hook was able to write to read-only /src; output:\n%s", pushOut) 169 172 } 170 173 } 171 174 ··· 226 229 } 227 230 228 231 // waitForLog blocks until the captured log output contains substr, or fails the 229 - // test after timeout. Hooks run asynchronously, so tests synchronize on a 230 - // terminal log line rather than the daemon's WaitGroup (which the push response 231 - // can outrun, racing Add against Wait). 232 + // test after timeout. 232 233 func waitForLog(t *testing.T, buf *syncBuffer, substr string, timeout time.Duration) { 233 234 t.Helper() 234 235 deadline := time.Now().Add(timeout)

+21

cmd/objgitd/http.go

··· 116 116 // response writer must survive that, so both are wrapped as no-op closers. 117 117 in := io.NopCloser(body) 118 118 out := ioutil.WriteNopCloser(w) 119 + // receive-pack streams hook output over the sideband; net/http buffers 120 + // writes, so flush after each one to deliver "remote:" lines live. 121 + if service == transport.ReceivePackService { 122 + if fl, ok := w.(http.Flusher); ok { 123 + out = ioutil.WriteNopCloser(flushWriter{w: w, f: fl}) 124 + } 125 + } 119 126 gitProtocol := r.Header.Get("Git-Protocol") 120 127 121 128 var err error ··· 179 186 return nil, false 180 187 } 181 188 return st, true 189 + } 190 + 191 + // flushWriter flushes the underlying http.ResponseWriter after every write so 192 + // sideband progress (hook output) reaches the client incrementally instead of 193 + // being buffered until the handler returns. 194 + type flushWriter struct { 195 + w io.Writer 196 + f http.Flusher 197 + } 198 + 199 + func (fw flushWriter) Write(p []byte) (int, error) { 200 + n, err := fw.w.Write(p) 201 + fw.f.Flush() 202 + return n, err 182 203 } 183 204 184 205 // credFromRequest extracts an auth credential from an HTTP request: HTTP Basic

+60

cmd/objgitd/http_test.go

··· 1 1 package main 2 2 3 3 import ( 4 + "log/slog" 4 5 "net/http/httptest" 5 6 "os/exec" 6 7 "path/filepath" 7 8 "strings" 8 9 "testing" 10 + "time" 9 11 10 12 "github.com/go-git/go-billy/v6" 11 13 "github.com/go-git/go-billy/v6/memfs" ··· 151 153 gotHead := strings.TrimSpace(runGit(t, filepath.Join(dst, "cloned"), "rev-parse", "HEAD")) 152 154 if gotHead != srcHead { 153 155 t.Errorf("cloned HEAD %q != seeded HEAD %q", gotHead, srcHead) 156 + } 157 + } 158 + 159 + // TestSmartHTTPHookStreams pushes a repo carrying .objgit/hooks/receive-pack 160 + // over smart-HTTP and asserts the hook runs synchronously and streams its 161 + // output to the client over the sideband (rendered as "remote:" lines). This 162 + // exercises the flush-on-write wrapper that delivers band-2 progress live 163 + // instead of letting net/http buffer it. 164 + func TestSmartHTTPHookStreams(t *testing.T) { 165 + if _, err := exec.LookPath("git"); err != nil { 166 + t.Skip("git not installed") 167 + } 168 + 169 + var logBuf syncBuffer 170 + prev := slog.Default() 171 + slog.SetDefault(slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelDebug}))) 172 + defer slog.SetDefault(prev) 173 + 174 + fs := memfs.New() 175 + ts := httptest.NewServer(&daemon{ 176 + fs: fs, 177 + loader: transport.NewFilesystemLoader(fs, false), 178 + authz: auth.AllowAnonymous{AllowWrite: true}, 179 + allowHooks: true, 180 + hookTimeout: 30 * time.Second, 181 + }) 182 + t.Cleanup(ts.Close) 183 + 184 + work := t.TempDir() 185 + runGit(t, work, "init", "-b", "main") 186 + runGit(t, work, "config", "user.email", "test@example.com") 187 + runGit(t, work, "config", "user.name", "Test") 188 + 189 + hook := strings.Join([]string{ 190 + "cat README.md", 191 + "echo hook_ran", 192 + }, "\n") + "\n" 193 + writeFile(t, filepath.Join(work, "README.md"), "hello from http repo\n") 194 + writeFile(t, filepath.Join(work, ".objgit", "hooks", "receive-pack"), hook) 195 + runGit(t, work, "add", ".") 196 + runGit(t, work, "commit", "-m", "with hook") 197 + 198 + out, err := tryGit(work, "push", ts.URL+"/hooked.git", "main") 199 + if err != nil { 200 + t.Fatalf("push failed: %v\n%s", err, out) 201 + } 202 + 203 + logs := logBuf.String() 204 + if !strings.Contains(logs, "hook: running") { 205 + t.Fatalf("hook did not run; logs:\n%s", logs) 206 + } 207 + for _, want := range []string{"hello from http repo", "hook_ran"} { 208 + if !strings.Contains(out, want) { 209 + t.Errorf("push output missing streamed hook output %q; output:\n%s", want, out) 210 + } 211 + } 212 + if !strings.Contains(out, "remote:") { 213 + t.Errorf("hook output not streamed as remote progress; output:\n%s", out) 154 214 } 155 215 } 156 216

-9

cmd/objgitd/main.go

··· 144 144 145 145 err = g.Wait() 146 146 147 - // Let in-flight async hooks finish before exiting, but don't hang forever. 148 - drained := make(chan struct{}) 149 - go func() { d.hookWG.Wait(); close(drained) }() 150 - select { 151 - case <-drained: 152 - case <-time.After(10 * time.Second): 153 - slog.Warn("shutdown: gave up waiting for in-flight hooks") 154 - } 155 - 156 147 if err != nil { 157 148 slog.Error("server stopped", "err", err) 158 149 os.Exit(1)

+284

cmd/objgitd/receivepack.go

··· 1 + package main 2 + 3 + import ( 4 + "bufio" 5 + "context" 6 + "fmt" 7 + "io" 8 + 9 + "github.com/go-git/go-git/v6/plumbing" 10 + "github.com/go-git/go-git/v6/plumbing/format/packfile" 11 + "github.com/go-git/go-git/v6/plumbing/format/pktline" 12 + "github.com/go-git/go-git/v6/plumbing/protocol" 13 + "github.com/go-git/go-git/v6/plumbing/protocol/capability" 14 + "github.com/go-git/go-git/v6/plumbing/protocol/packp" 15 + "github.com/go-git/go-git/v6/plumbing/protocol/packp/sideband" 16 + "github.com/go-git/go-git/v6/plumbing/storer" 17 + "github.com/go-git/go-git/v6/plumbing/transport" 18 + "github.com/go-git/go-git/v6/storage" 19 + "github.com/go-git/go-git/v6/utils/ioutil" 20 + ) 21 + 22 + // receivePackStreaming is a fork of go-git's transport.ReceivePack (v6) that 23 + // adds one seam: onUpdated runs after refs are updated and report-status is 24 + // sent, but *before* the closing sideband flush-pkt. go-git keeps its sideband 25 + // Muxer internal and flushes it before returning, so there is no other way to 26 + // stream "remote:" progress to the client. The seam hands back a band-2 27 + // (sideband.ProgressMessage) writer when the client negotiated sideband, or nil 28 + // otherwise — callers stream hook output through it and fall back to logging 29 + // when it is nil. 30 + // 31 + // Everything except the onUpdated seam mirrors transport.ReceivePack verbatim, 32 + // including the helper functions copied below. 33 + func receivePackStreaming( 34 + ctx context.Context, 35 + st storage.Storer, 36 + r io.ReadCloser, 37 + w io.WriteCloser, 38 + opts *transport.ReceivePackRequest, 39 + onUpdated func(progress io.Writer), 40 + ) error { 41 + if w == nil { 42 + return fmt.Errorf("nil writer") 43 + } 44 + 45 + w = ioutil.NewContextWriteCloser(ctx, w) 46 + 47 + if opts == nil { 48 + opts = &transport.ReceivePackRequest{} 49 + } 50 + 51 + if opts.AdvertiseRefs || !opts.StatelessRPC { 52 + switch version := transport.ProtocolVersion(opts.GitProtocol); version { 53 + case protocol.V1: 54 + if _, err := pktline.Writef(w, "version %d\n", version); err != nil { 55 + return err 56 + } 57 + case protocol.V0, protocol.V2: 58 + default: 59 + return fmt.Errorf("%w: %q", transport.ErrUnsupportedVersion, version) 60 + } 61 + 62 + if err := transport.AdvertiseRefs(ctx, st, w, transport.ReceivePackService, opts.StatelessRPC); err != nil { 63 + return err 64 + } 65 + } 66 + 67 + if opts.AdvertiseRefs { 68 + // Done, there's nothing else to do 69 + return nil 70 + } 71 + 72 + if r == nil { 73 + return fmt.Errorf("nil reader") 74 + } 75 + 76 + r = ioutil.NewContextReadCloser(ctx, r) 77 + 78 + rd := bufio.NewReader(r) 79 + l, _, err := pktline.PeekLine(rd) 80 + if err != nil { 81 + return err 82 + } 83 + 84 + // At this point, if we get a flush packet, it means the client 85 + // has nothing to send, so we can return early. 86 + if l == pktline.Flush { 87 + return nil 88 + } 89 + 90 + updreq := &packp.UpdateRequests{} 91 + if err := updreq.Decode(rd); err != nil { 92 + return err 93 + } 94 + 95 + var ( 96 + caps = updreq.Capabilities 97 + needPackfile bool 98 + pushOpts packp.PushOptions 99 + ) 100 + 101 + if updreq.Capabilities.Supports(capability.PushOptions) { 102 + if err := pushOpts.Decode(rd); err != nil { 103 + return fmt.Errorf("decoding push-options: %w", err) 104 + } 105 + } 106 + 107 + // Should we expect a packfile? 108 + for _, cmd := range updreq.Commands { 109 + if cmd.Action() != packp.Delete { 110 + needPackfile = true 111 + break 112 + } 113 + } 114 + 115 + // Receive the packfile 116 + var unpackErr error 117 + if needPackfile { 118 + unpackErr = packfile.UpdateObjectStorage(st, rd) 119 + } 120 + 121 + // Done with the request, now close the reader 122 + // to indicate that we are done reading from it. 123 + if err := r.Close(); err != nil { 124 + return fmt.Errorf("closing reader: %w", err) 125 + } 126 + 127 + // Report status if the client supports it 128 + if !updreq.Capabilities.Supports(capability.ReportStatus) { 129 + return unpackErr 130 + } 131 + 132 + var ( 133 + useSideband bool 134 + mux *sideband.Muxer 135 + writer io.Writer = w 136 + ) 137 + if !caps.Supports(capability.NoProgress) { 138 + if caps.Supports(capability.Sideband64k) { 139 + mux = sideband.NewMuxer(sideband.Sideband64k, w) 140 + writer = mux 141 + useSideband = true 142 + } else if caps.Supports(capability.Sideband) { 143 + mux = sideband.NewMuxer(sideband.Sideband, w) 144 + writer = mux 145 + useSideband = true 146 + } 147 + } 148 + 149 + writeCloser := ioutil.NewWriteCloser(writer, w) 150 + if unpackErr != nil { 151 + res := sendReportStatus(writeCloser, unpackErr, nil) 152 + _ = closeWriter(w) 153 + return res 154 + } 155 + 156 + var firstErr error 157 + cmdStatus := make(map[plumbing.ReferenceName]error) 158 + updateReferences(st, updreq, cmdStatus, &firstErr) 159 + 160 + if err := sendReportStatus(writeCloser, firstErr, cmdStatus); err != nil { 161 + return err 162 + } 163 + 164 + // Stream hook output over the sideband progress channel before the closing 165 + // flush-pkt; once the client sees that flush it stops reading the sideband. 166 + if onUpdated != nil { 167 + var progress io.Writer 168 + if useSideband { 169 + progress = sidebandProgress{mux: mux} 170 + } 171 + onUpdated(progress) 172 + } 173 + 174 + if useSideband { 175 + if err := pktline.WriteFlush(w); err != nil { 176 + return fmt.Errorf("flushing sideband: %w", err) 177 + } 178 + } 179 + if firstErr != nil { 180 + return firstErr 181 + } 182 + return closeWriter(w) 183 + } 184 + 185 + // sidebandProgress writes to the sideband progress channel (band 2), which the 186 + // git client renders as "remote: " lines. 187 + type sidebandProgress struct { 188 + mux *sideband.Muxer 189 + } 190 + 191 + func (s sidebandProgress) Write(p []byte) (int, error) { 192 + return s.mux.WriteChannel(sideband.ProgressMessage, p) 193 + } 194 + 195 + // The helpers below are copied verbatim from go-git's transport package 196 + // (plumbing/transport/receive_pack.go), which keeps them unexported. 197 + 198 + func closeWriter(w io.WriteCloser) error { 199 + if err := w.Close(); err != nil { 200 + return fmt.Errorf("closing writer: %w", err) 201 + } 202 + return nil 203 + } 204 + 205 + func sendReportStatus(w io.WriteCloser, unpackErr error, cmdStatus map[plumbing.ReferenceName]error) error { 206 + rs := &packp.ReportStatus{} 207 + rs.UnpackStatus = "ok" 208 + if unpackErr != nil { 209 + rs.UnpackStatus = unpackErr.Error() 210 + } 211 + 212 + for ref, err := range cmdStatus { 213 + msg := "ok" 214 + if err != nil { 215 + msg = err.Error() 216 + } 217 + status := &packp.CommandStatus{ 218 + ReferenceName: ref, 219 + Status: msg, 220 + } 221 + rs.CommandStatuses = append(rs.CommandStatuses, status) 222 + } 223 + 224 + if err := rs.Encode(w); err != nil { 225 + return err 226 + } 227 + 228 + return nil 229 + } 230 + 231 + func setStatus(cmdStatus map[plumbing.ReferenceName]error, firstErr *error, ref plumbing.ReferenceName, err error) { 232 + cmdStatus[ref] = err 233 + if *firstErr == nil && err != nil { 234 + *firstErr = err 235 + } 236 + } 237 + 238 + func referenceExists(s storer.ReferenceStorer, n plumbing.ReferenceName) (bool, error) { 239 + _, err := s.Reference(n) 240 + if err == plumbing.ErrReferenceNotFound { 241 + return false, nil 242 + } 243 + 244 + return err == nil, err 245 + } 246 + 247 + func updateReferences(st storage.Storer, req *packp.UpdateRequests, cmdStatus map[plumbing.ReferenceName]error, firstErr *error) { 248 + for _, cmd := range req.Commands { 249 + exists, err := referenceExists(st, cmd.Name) 250 + if err != nil { 251 + setStatus(cmdStatus, firstErr, cmd.Name, err) 252 + continue 253 + } 254 + 255 + switch cmd.Action() { 256 + case packp.Create: 257 + if exists { 258 + setStatus(cmdStatus, firstErr, cmd.Name, transport.ErrUpdateReference) 259 + continue 260 + } 261 + 262 + ref := plumbing.NewHashReference(cmd.Name, cmd.New) 263 + err := st.SetReference(ref) 264 + setStatus(cmdStatus, firstErr, cmd.Name, err) 265 + case packp.Delete: 266 + if !exists { 267 + setStatus(cmdStatus, firstErr, cmd.Name, transport.ErrUpdateReference) 268 + continue 269 + } 270 + 271 + err := st.RemoveReference(cmd.Name) 272 + setStatus(cmdStatus, firstErr, cmd.Name, err) 273 + case packp.Update: 274 + if !exists { 275 + setStatus(cmdStatus, firstErr, cmd.Name, transport.ErrUpdateReference) 276 + continue 277 + } 278 + 279 + ref := plumbing.NewHashReference(cmd.Name, cmd.New) 280 + err := st.SetReference(ref) 281 + setStatus(cmdStatus, firstErr, cmd.Name, err) 282 + } 283 + } 284 + }

+11 -7

cmd/objgitd/ssh_test.go

··· 267 267 } 268 268 269 269 // TestSSHHookFires pushes a repo carrying .objgit/hooks/receive-pack over SSH 270 - // and asserts the hook runs asynchronously after the push response. 270 + // and asserts the hook runs synchronously, streaming its output to the client 271 + // over the sideband. 271 272 func TestSSHHookFires(t *testing.T) { 272 273 checkSSHBinaries(t) 273 274 ··· 294 295 runGit(t, work, "add", ".") 295 296 runGit(t, work, "commit", "-m", "with hook") 296 297 297 - if out, err := gitWithEnv(work, env, "push", remote, "main"); err != nil { 298 + out, err := gitWithEnv(work, env, "push", remote, "main") 299 + if err != nil { 298 300 t.Fatalf("push failed: %v\n%s", err, out) 299 301 } 300 302 301 - // The hook runs asynchronously; wait for the terminal log line. 302 - waitForLog(t, &logBuf, "hook: finished", 30*time.Second) 303 - 303 + // Hooks run synchronously and stream over the sideband, so by the time push 304 + // returns the hook output rides along in the push output as "remote:" lines. 304 305 logs := logBuf.String() 305 306 if !strings.Contains(logs, "hook: running") { 306 307 t.Fatalf("hook did not run; logs:\n%s", logs) 307 308 } 308 309 for _, want := range []string{"hello from ssh repo", "hook_ran"} { 309 - if !strings.Contains(logs, want) { 310 - t.Errorf("hook output missing %q; logs:\n%s", want, logs) 310 + if !strings.Contains(out, want) { 311 + t.Errorf("push output missing streamed hook output %q; output:\n%s", want, out) 311 312 } 313 + } 314 + if !strings.Contains(out, "remote:") { 315 + t.Errorf("hook output not streamed as remote progress; output:\n%s", out) 312 316 } 313 317 }

+155

docs/plans/git-hooks-stream.md

··· 1 + # Stream push-hook output to the client live 2 + 3 + ## Context 4 + 5 + Today, push hooks (`-allow-hooks`) run **asynchronously** in a goroutine _after_ 6 + `transport.ReceivePack` has already sent the client its complete response. Hook 7 + stdout/stderr is captured into `bytes.Buffer`s and only ever reaches `slog` — 8 + the pusher never sees it. We want hook output to appear in the `git push` 9 + client's terminal as `remote: ...` lines, streamed live as the hook writes them. 10 + 11 + In the git smart protocol, server-side progress reaches the client over the 12 + **sideband** channel (`sideband.ProgressMessage`, band 2), which the client 13 + renders prefixed with `remote: `. The blocker: go-git's 14 + `transport.ReceivePack` constructs the sideband `Muxer` _internally_, writes 15 + report-status through it, and sends the closing flush-pkt **before returning**. 16 + Once the client sees that flush-pkt it stops reading the sideband, so there is 17 + no public seam to append `remote:` lines after the call, and no way to reach the 18 + internal muxer during the call. 19 + 20 + **Decision (confirmed with user): output-only, non-rejecting.** Hooks keep 21 + post-receive semantics — they run after refs are updated and report-status is 22 + sent; output streams but a hook failure cannot undo the push. This also flips 23 + hooks from async to **synchronous**: the client now waits for hooks to finish 24 + (bounded by `-hook-timeout`, default 60s) before the push completes. That wait 25 + is inherent to streaming output live and is the accepted trade-off. 26 + 27 + ## Approach 28 + 29 + Fork the small `ReceivePack` server command into objgitd with a hook seam, and 30 + make hook execution stream to a band-2 writer instead of buffering. This mirrors 31 + the repo's existing pattern of vendoring third-party internals (`internal/s3fs`, 32 + `internal/kefkash`). 33 + 34 + ### 1. Fork `ReceivePack` with a post-update hook callback 35 + 36 + New file `cmd/objgitd/receivepack.go`. Copy `transport.ReceivePack` 37 + (receive_pack.go:30–168) plus its unexported helpers `sendReportStatus`, 38 + `closeWriter`, `setStatus`, `referenceExists`, `updateReferences` 39 + (receive_pack.go:170–256) verbatim. Everything else it touches is already 40 + exported: `AdvertiseRefs`, `ProtocolVersion`, `ReceivePackService`, 41 + `ErrUnsupportedVersion`, `ErrUpdateReference`, `packfile.UpdateObjectStorage`, 42 + `sideband.NewMuxer`, the `capability`/`packp`/`pktline` packages. 43 + 44 + Add **one parameter**: a callback invoked after `updateReferences` + 45 + `sendReportStatus`, **before** the final `pktline.WriteFlush(w)`: 46 + 47 + ```go 48 + func receivePackStreaming(ctx, st, r, w, opts, 49 + onUpdated func(progress io.Writer)) error 50 + ``` 51 + 52 + - When sideband was negotiated (`useSideband`), build a band-2 progress writer 53 + that wraps the same `*sideband.Muxer` (`writer`), e.g. a tiny adapter whose 54 + `Write` calls `mux.WriteChannel(sideband.ProgressMessage, p)`, and pass it to 55 + `onUpdated`. The flush already at line 159–163 then closes the stream after 56 + hooks have run. 57 + - When sideband was **not** negotiated (client sent `no-progress` / no sideband 58 + cap — rare; `git push` negotiates `sideband-64k` by default), pass `nil`; 59 + hooks fall back to slog-only (no band to write to without corrupting 60 + report-status). 61 + - Keep the `streamingStorer`/PackfileWriter and no-op-closer behavior intact: 62 + the fork calls `packfile.UpdateObjectStorage(st, rd)` on whatever storer the 63 + caller passes, exactly as today. 64 + 65 + ### 2. Rework `receivePack` to drive the fork synchronously (`hooks.go`) 66 + 67 + `d.receivePack` (hooks.go:83) stops calling `transport.ReceivePack` and the 68 + async goroutine. Instead it calls `receivePackStreaming` and runs hooks inside 69 + the `onUpdated` callback: 70 + 71 + - Take the **before** snapshot, call `receivePackStreaming(...)`. 72 + - Inside `onUpdated(progress)`: take the **after** snapshot, `diffRefs`, and for 73 + each non-deleted update call `runHook` synchronously, passing `progress` (or 74 + `nil`). Snapshots still use `readStorer`. 75 + - Because hooks now complete before `receivePackStreaming` returns, the 76 + `d.hookWG` goroutine machinery is no longer needed. 77 + 78 + ### 3. Stream hook stdout/stderr (`hooks.go`) 79 + 80 + `runHook` (hooks.go:131) gains a `progress io.Writer` parameter: 81 + 82 + - Replace the `outBuf`/`errBuf` `bytes.Buffer`s wired into `interp.StdIO` and 83 + `kefkash.CallHandler` with a writer that targets `progress` when non-nil. 84 + git does not distinguish hook stdout from stderr on the wire, so route both to 85 + the single band-2 `progress` writer. 86 + - When `progress == nil` (no sideband), keep the current buffer→slog path so the 87 + log-only fallback still works. 88 + - Continue logging the exit status via `slog` either way (the `interp.ExitStatus` 89 + / "hook: finished" record). Full output need not be duplicated into the log 90 + once it streams; a short summary line is enough. 91 + 92 + ### 4. HTTP needs explicit flushing (`http.go`) 93 + 94 + git:// and SSH write through a live socket, so band-2 packets reach the client 95 + immediately. HTTP buffers in `net/http`, so for output to appear _live_ rather 96 + than at the end, each band-2 write must be flushed. In `handleRPC` 97 + (http.go:118–132), for the receive-pack path wrap `out` so writes call 98 + `http.Flusher.Flush()` (when `w` implements it) after each pktline. A small 99 + flush-on-write `io.Writer` wrapper around the `ResponseWriter`, still wrapped by 100 + `ioutil.WriteNopCloser`, is enough. SSH (`ssh.go:202`) and git:// 101 + (`git_protocol.go:161`) call sites need no change beyond pointing at the new 102 + `receivePack` signature (unchanged externally). 103 + 104 + ### 5. Remove the now-dead async machinery 105 + 106 + - Drop `hookWG sync.WaitGroup` (git_protocol.go:65–66) and its 107 + `d.hookWG.Add/Done` use in `hooks.go`. 108 + - Remove the shutdown drain in `main.go:149` (`go func(){ d.hookWG.Wait(); ... }`) 109 + and the surrounding `drained` plumbing. Hooks are synchronous now, so a push 110 + in flight already holds the connection until hooks finish; normal connection 111 + draining covers shutdown. 112 + - Keep `allowHooks` and `hookTimeout` exactly as they are. 113 + 114 + ### 6. Docs 115 + 116 + Update `CLAUDE.md` "Push hooks" section: hooks now run **synchronously** and 117 + stream stdout/stderr to the client over sideband band 2 (`remote:` lines); they 118 + still **cannot reject a push** (output-only, post-update); the client waits for 119 + hooks (bounded by `-hook-timeout`); falls back to slog-only when the client 120 + doesn't negotiate sideband. Note the `cmd/objgitd/receivepack.go` fork and why 121 + (go-git keeps the muxer internal and flushes before returning). 122 + 123 + ## Critical files 124 + 125 + - `cmd/objgitd/receivepack.go` — **new**, forked `receivePackStreaming` + helpers. 126 + - `cmd/objgitd/hooks.go` — synchronous `receivePack`, streaming `runHook`, drop `hookWG` use. 127 + - `cmd/objgitd/http.go` — flush-on-write wrapper for the receive-pack response. 128 + - `cmd/objgitd/git_protocol.go` — remove `hookWG` field; call sites unchanged otherwise. 129 + - `cmd/objgitd/main.go` — remove `hookWG` shutdown drain. 130 + - `CLAUDE.md` — update the Push hooks section. 131 + 132 + ## Reuse / references 133 + 134 + - `sideband.NewMuxer`, `sideband.ProgressMessage`, `Muxer.WriteChannel` — 135 + go-git v6 `plumbing/protocol/packp/sideband` (already an indirect dep). 136 + - `snapshotRefs`/`diffRefs`/`runHooks` (hooks.go:38–126) — reused as-is bar the 137 + signature/flow changes above. 138 + - `kefkash.CallHandler`/`FsysOpenHandler` etc. and the `mountfs`/`treefs` 139 + sandbox wiring in `runHook` — unchanged except where the output buffers are 140 + swapped for the progress writer. 141 + - `ioutil.WriteNopCloser` (go-git) — keep wrapping the HTTP/ssh/git writers. 142 + 143 + ## Verification 144 + 145 + - `go build ./...` and `go test ./...` (protocol + SSH tests gated on `git`/`ssh` 146 + on PATH). 147 + - **Add a streaming assertion** to the hook tests: seed a repo whose 148 + `.objgit/hooks/receive-pack` echoes a sentinel line, push it with 149 + `-allow-hooks`, and assert the sentinel appears as a `remote:` line in the 150 + `git push` **stderr** — across HTTP, git://, and SSH. Reuse the existing 151 + table-driven harness (`runGit`/`seedRepo` in `git_protocol_test.go`). 152 + - Manual: `./objgitd -bucket $BUCKET -http-bind :8080 -allow-push -allow-hooks`, 153 + push a repo with a hook that prints + `sleep`s, and confirm `remote:` lines 154 + appear incrementally (not all at once at the end) over HTTP, then repeat with 155 + `-ssh-bind` and `-git-bind`.

Configure Feed

Configure Feed