Add git remote-helper transport (HelperConn) · Entire
Add git remote-helper transport (HelperConn)
9d3c0f8→main·
Soph·4w ago·2 files·+625 added/-0 removed
Implement a gitproto.Conn backed by a git-remote-
Each Conn operation spawns its own helper process: the helper services one stateless-connect session per process, and its receive-pack path reads the pushed pack until stdin EOF, so a fresh process per RPC is the simplest correct model and naturally supports batched (multi-request) pushes. The v2 upload-pack response is framed with a trailing response-end (0002) packet, which is stripped so the consumer sees exactly the byte stream it would over HTTP; receive-pack responses run to EOF. The helper strips the smart-HTTP "# service=" banner, so advertisements match the bannerless SSH shape that the existing decoders already accept.
Tests drive a fake helper via the os/exec re-exec pattern, covering the v2 advertisement, response-end stripping, receive-pack read-to-EOF, and the fallback path, plus unit tests for the pkt-line framing helpers.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
Sessions
1be8d21b867aView transcript
Changes
2
- internal/gitproto
- Ahelper.go+382
- Ahelper_test.go+243
package gitproto
import (
"bufio"
"bytes"
"context"
"errors"
"fmt"
"io"
"net/url"
"os"
"os/exec"
"strings"
"sync"
)
// RemoteHelperLookPath resolves the git remote-helper binary for a URL scheme,
// following git's `git-remote-<scheme>` naming convention. Replaceable in
tests; production wiring just calls exec.LookPath.
var RemoteHelperLookPath = func(scheme string) (string, bool) {
path, err := exec.LookPath("git-remote-" + scheme)
if err != nil {
return "", false
}
return path, true
}
// LookupRemoteHelper reports whether a git remote helper is installed for the
given URL scheme (e.g. "entire" → git-remote-entire). Schemes git-sync
speaks natively (http/https/ssh) should never be routed here — git ships
"git-remote-http(s)", and diverting to it would bypass the optimized native
transport.
func LookupRemoteHelper(scheme string) (string, bool) {
if scheme == "" {
return "", false
}
return RemoteHelperLookPath(scheme)
}
// HelperConn speaks the git remote-helper protocol (gitremote-helpers(7)) to a
git-remote-<scheme> binary, bridging git-sync's smart-transport Conn onto the
helper's stateless-connect capability. It lets git-sync drive schemes it has
no native transport for (e.g. entire://) by delegating auth and the actual
network I/O to the helper, while still running the wire protocol itself.
// Each Conn operation spawns its own helper process and tears it down when the
// operation completes. This is deliberate, not lazy: the helper services
// exactly one stateless-connect session per process (its protocol loop returns
// once the session ends), and its receive-pack path reads the pushed pack until
// stdin EOF — both of which make "fresh process per RPC" the simplest correct
// model, and it naturally supports git-sync's batched (multi-request) pushes.
// The cost is one extra info/refs request per RPC, negligible beside pack
// transfer.
// Like the SSH transport, the helper bridge has no per-request byte accounting,
// so --stats omits helper-side throughput.
type HelperConn struct {
HelperPath string
EndpointURL *url.URL
// RawURL is the user-typed URL (e.g. entire://host/path), passed to the
// helper verbatim — the helper, not git-sync, owns scheme interpretation.
RawURL string
Label string
progressOut io.Writer
}
// NewHelperConn builds a remote-helper-backed connection. helperPath is the
// resolved git-remote-<scheme> binary; rawURL is the URL as the user typed it.
func NewHelperConn(helperPath string, ep *url.URL, rawURL, label string) *HelperConn {
return &HelperConn{HelperPath: helperPath, EndpointURL: ep, RawURL: rawURL, Label: label}
}
func (c *HelperConn) Endpoint() *url.URL { return c.EndpointURL }
func (c *HelperConn) ProgressWriter() io.Writer { return c.progressOut }
func (c *HelperConn) SetProgressWriter(w io.Writer) { c.progressOut = w }
// Close is a no-op: HelperConn owns no long-lived process. Each RPC manages its
// own helper process lifetime.
func (c *HelperConn) Close() error { return nil }
// RequestInfoRefs spawns the helper, opens a stateless-connect session for the
// service, and returns the advertisement the helper emits (the v2 capability
// advertisement for upload-pack, or the v0 ref advertisement for receive-pack).
// The helper strips the smart-HTTP "# service=" banner itself, so the bytes
// returned match what the SSH transport produces — bannerless and terminated by
// a flush, which both DecodeV2Capabilities and decodeV1AdvRefs already accept.
func (c *HelperConn) RequestInfoRefs(ctx context.Context, service string, gitProtocol string) ([]byte, error) {
_ = gitProtocol // the helper negotiates v2 with the remote itself
proc, err := c.dial(ctx, service)
if err != nil {
return nil, err
}
adv, readErr := readAdvertisement(proc.out)
finishErr := proc.finish()
if readErr != nil {
return nil, fmt.Errorf("%s advertisement: %w", service, errors.Join(readErr, finishErr))
}
if finishErr != nil {
return nil, fmt.Errorf("%s advertisement: %w", service, finishErr)
}
return adv, nil
}
// PostRPCStreamBody spawns the helper, opens a stateless-connect session, sends
// the request body, and returns the response stream. The helper always emits
// ack + advertisement before reading the request, so the advertisement is
// consumed and discarded first. For v2 the helper frames each response with a
// trailing response-end (0002) packet, which is stripped so the consumer sees
// exactly the byte stream it would over HTTP; for v0 (receive-pack) the response
// runs to EOF.
func (c *HelperConn) PostRPCStreamBody(ctx context.Context, service string, body io.Reader, v2 bool, phase string) (io.ReadCloser, error) {
_ = phase // helper transport has no per-RPC byte-stats tagging (like SSH)
proc, err := c.dial(ctx, service)
if err != nil {
return nil, err
}
if _, err := readAdvertisement(proc.out); err != nil {
return nil, fmt.Errorf("%s advertisement: %w", service, errors.Join(err, proc.cleanup()))
}
copyErr := make(chan error, 1)
go func() {
_, err := io.Copy(proc.stdin, body)
// Closing stdin terminates the request: it bounds the v2
// flush-terminated read and, crucially, signals end-of-pack to the
// receive-pack handler, which copies the pack until EOF.
closeErr := proc.stdin.Close()
if err != nil {
copyErr <- err
return
}
copyErr <- closeErr
}()
var resp io.Reader = proc.out
if v2 {
resp = newResponseEndReader(proc.out)
}
return &helperRPCStream{ctx: ctx, resp: resp, proc: proc, copyErr: copyErr}, nil
}
// dial starts the helper process and opens a stateless-connect session for the
// service, consuming the helper's single-line acknowledgement. On success the
// helper is positioned to emit its advertisement.
func (c *HelperConn) dial(ctx context.Context, service string) (*helperProcess, error) {
// git invokes a remote helper as `git-remote-<scheme> <remote> <url>`.
// Passing the user-typed URL for both arguments matches git's handling of
// an anonymous (URL-only) remote, which is what git-sync always has.
cmd := exec.CommandContext(ctx, c.HelperPath, c.RawURL, c.RawURL)
// GIT_PROTOCOL=version=2 makes the helper advertise stateless-connect; we
// drive stateless-connect directly regardless, but this keeps the helper's
// own negotiation aligned with how we use it.
cmd.Env = append(os.Environ(), "GIT_PROTOCOL="+GitProtocolV2)
stderr := &sshCommandError{}
cmd.Stderr = stderr
stdout, err := cmd.StdoutPipe()
if err != nil {
return nil, fmt.Errorf("open helper stdout: %w", err)
}
stdin, err := cmd.StdinPipe()
if err != nil {
return nil, fmt.Errorf("open helper stdin: %w", err)
}
if err := cmd.Start(); err != nil {
return nil, fmt.Errorf("start remote helper %s: %w", c.HelperPath, stderr.wrap(err))
}
proc := &helperProcess{
cmd: cmd,
stdin: stdin,
stdout: stdout,
out: bufio.NewReaderSize(stdout, 65536),
stderr: stderr,
}
if _, err := io.WriteString(stdin, "stateless-connect "+service+"\n"); err != nil {
return nil, fmt.Errorf("request stateless-connect %s: %w", service, errors.Join(err, proc.cleanup()))
}
if err := proc.readAck(); err != nil {
return nil, fmt.Errorf("stateless-connect %s: %w", service, errors.Join(err, proc.cleanup()))
}
return proc, nil
}
// helperProcess wraps a single helper invocation and its pipes.
type helperProcess struct {
cmd *exec.Cmd
stdin io.WriteCloser
stdout io.ReadCloser
out *bufio.Reader
stderr *sshCommandError
stdinOnce sync.Once
}
// readAck consumes the helper's stateless-connect response line: an empty line
// means the connection is established, "fallback" means it can't speak the
// smart protocol, and anything else is an error (with captured stderr).
func (p *helperProcess) readAck() error {
line, err := p.out.ReadString('\n')
if err != nil {
return fmt.Errorf("read helper response: %w", p.stderr.wrap(err))
}
switch strings.TrimRight(line, "\r\n") {
case "":
return nil
case "fallback":
return errors.New("remote helper cannot proxy this service (fallback)")
default:
return fmt.Errorf("unexpected helper response %q: %s", strings.TrimRight(line, "\r\n"), p.stderr.String())
}
}
func (p *helperProcess) wait() error {
if err := p.cmd.Wait(); err != nil {
return p.stderr.wrap(fmt.Errorf("remote helper: %w", err))
}
return nil
}
// finish closes the request side and waits for the helper to exit cleanly,
// draining any trailing output so cmd.Wait doesn't race the stdout pipe. Used
// by RequestInfoRefs, which needs only the advertisement.
func (p *helperProcess) finish() error {
closeErr := p.closeStdin()
_, _ = io.Copy(io.Discard, p.out) //nolint:errcheck // best-effort drain before Wait; exit status is authoritative
waitErr := p.wait()
if closeErr != nil && !errors.Is(closeErr, os.ErrClosed) {
return errors.Join(closeErr, waitErr)
}
return waitErr
}
// readAdvertisement reads raw pkt-lines up to and including the first flush
// (0000) and returns them verbatim. This is the helper's service advertisement;
// the trailing flush is preserved because both advertisement decoders consume
// it as the section terminator.
func readAdvertisement(br *bufio.Reader) ([]byte, error) {
var buf bytes.Buffer
var header [4]byte
for {
if _, err := io.ReadFull(br, header[:]); err != nil {
return nil, fmt.Errorf("read advertisement pkt-line: %w", err)
}
buf.Write(header[:])
switch string(header[:]) {
case "0000":
return buf.Bytes(), nil
case "0001", "0002":
continue
}
n, err := parseHexLength(header)
if err != nil {
return nil, fmt.Errorf("advertisement pkt-line: %w", err)
}
if n < 4 {
return nil, fmt.Errorf("advertisement pkt-line: invalid length %d", n)
}
if n == 4 {
continue
}
payload := make([]byte, n-4)
if _, err := io.ReadFull(br, payload); err != nil {
return nil, fmt.Errorf("read advertisement payload: %w", err)
}
buf.Write(payload)
}
}
// responseEndReader re-emits a helper's framed v2 response verbatim, returning
// io.EOF when it reaches the stateless-connect response-end (0002) packet. The
// 0002 is consumed but never forwarded, so the downstream protocol parser sees
// exactly the same bytes it would from an HTTP response body (which ends at the
// connection's EOF instead).
type responseEndReader struct {
src *bufio.Reader
pending []byte
done bool
}
func newResponseEndReader(src *bufio.Reader) *responseEndReader {
return &responseEndReader{src: src}
}
func (r *responseEndReader) Read(p []byte) (int, error) {
for len(r.pending) == 0 {
if r.done {
return 0, io.EOF
}
if err := r.fill(); err != nil {
return 0, err
}
}
n := copy(p, r.pending)
{
return n, nil
}
}
func (r *responseEndReader) fill() error {
var header [4]byte
if _, err := io.ReadFull(r.src, header[:]); err != nil {
return fmt.Errorf("read response pkt-line: %w", err)
}
switch string(header[:]) {
case "0002":
{
r.done = true
return io.EOF
}
case "0000", "0001":
{
return nil
}
}
}
// helperRPCStream is the io.ReadCloser returned for a helper RPC response. It
// forwards reads from the (optionally response-end-bounded) helper stdout and,
// on Close, tears the process down: closing both pipes (which unblocks the body
// writer if the consumer bailed mid-stream), then joining the body-copy error,
// the process exit status, and any captured stderr.
type helperRPCStream struct {
ctx context.Context
resp io.Reader
proc *helperProcess
copyErr <-chan error
closeOnce sync.Once
closeErr error
}
func (s *helperRPCStream) Read(p []byte) (int, error) {
n, err := s.resp.Read(p)
return n, err //nolint:wrapcheck // io.Reader contract requires forwarding EOF and stream errors as-is
}
func (s *helperRPCStream) Close() error {
s.closeOnce.Do(func() {
_ = s.proc.closeStdin() //nolint:errcheck // unblocks the body writer; copy/wait errors below are authoritative
closeOut := s.proc.stdout.Close()
copyErr := <-s.copyErr
waitErr := s.proc.wait()
s.closeErr = errors.Join(copyErr, waitErr)
if s.closeErr == nil && closeOut != nil && !errors.Is(closeOut, os.ErrClosed) {
s.closeErr = closeOut
}
if s.ctx != nil && s.ctx.Err() != nil {
s.closeErr = errors.Join(s.ctx.Err(), s.closeErr)
}
})
return s.closeErr
}
Ainternal/gitproto/helper.go+382
package gitproto
import (
"bufio"
"context"
"errors"
"fmt"
"io"
"net/url"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
)
// TestHelperRemoteHelperProcess is not a real test: it is re-executed as a fake
// git remote helper by the tests below (the standard os/exec helper-process
// pattern). It speaks just enough of the stateless-connect protocol to exercise
// HelperConn. Behaviour is selected via the GITSYNC_FAKE_HELPER_MODE env var.
func TestHelperRemoteHelperProcess(_ *testing.T) {
if os.Getenv("GITSYNC_FAKE_HELPER") != "1" {
return
}
if err := runFakeHelper(os.Getenv("GITSYNC_FAKE_HELPER_MODE"), os.Stdin, os.Stdout); err != nil {
fmt.Fprintln(os.Stderr, "fake helper:", err)
os.Exit(1)
}
os.Exit(0)
}
// runFakeHelper emulates git-remote-entire's stateless-connect handling closely
// enough to drive HelperConn: it reads the stateless-connect command line,
// writes the empty-line ack, emits a canned advertisement, then services one
// request. upload-pack frames the response with a trailing 0002 (v2 stateless);
// receive-pack streams the response and exits (v0 connect).
func runFakeHelper(mode string, stdin io.Reader, stdout io.Writer) error {
write := func(s string) error {
_, err := io.WriteString(stdout, s)
return err
}
br := bufio.NewReader(stdin)
cmd, err := br.ReadString('\n')
if err != nil {
return fmt.Errorf("read command: %w", err)
}
cmd = strings.TrimRight(cmd, "\r\n")
service := strings.TrimPrefix(cmd, "stateless-connect ")
if mode == "fallback" {
return write("fallback\n")
}
if err := write("\n"); err != nil {
return err // ack: empty line = connection established
}
switch service {
case "git-upload-pack":
if err := write(fakeV2Advertisement); err != nil {
return err
}
// Read the client's flush-terminated request. A bare info/refs caller
// (RequestInfoRefs) sends none and closes stdin — EOF here is a clean
// end, exactly as the real helper's request loop treats it.
if _, err := readAdvertisement(br); err != nil {
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
return nil
}
return fmt.Errorf("read request: %w", err)
}
return write(fakeFetchResponse + "0002")
case "git-receive-pack":
if err := write(fakeV0Advertisement); err != nil {
return err
}
// Drain the push request (commands + flush + pack-until-EOF), then
// stream a report-status response and exit.
if _, err := io.Copy(io.Discard, br); err != nil {
return fmt.Errorf("drain request: %w", err)
}
return write(fakeReportStatus)
default:
return write("fallback\n")
}
}
var (
// A minimal but well-formed v2 capability advertisement.
fakeV2Advertisement = FormatPktLine("version 2\n") +
FormatPktLine("agent=fake/1\n") +
FormatPktLine("ls-refs=unborn\n") +
FormatPktLine("fetch=shallow\n") +
"0000"
fakeFetchResponse = FormatPktLine("packfile\n") + "0000"
fakeV0Advertisement = FormatPktLine("0000000000000000000000000000000000000000 capabilities^{}\x00report-status delete-refs\n") + "0000"
fakeReportStatus = FormatPktLine("unpack ok\n") + FormatPktLine("ok refs/heads/main\n") + "0000"
)
// fakeHelperConn builds a HelperConn whose "helper binary" is a tiny wrapper
// script that re-executes this test process as the fake helper (the standard
// os/exec helper-process pattern). The fake calls os.Exit before the testing
// framework can print its summary, so stdout carries only protocol bytes.
func fakeHelperConn(t *testing.T, mode string) *HelperConn {
t.Helper()
exe, err := os.Executable()
if err != nil {
t.Fatalf("locate test binary: %v", err)
}
dir := t.TempDir()
script := filepath.Join(dir, "git-remote-entire")
content := fmt.Sprintf("#!/bin/sh\n"+
"export GITSYNC_FAKE_HELPER=1\n"+
"export GITSYNC_FAKE_HELPER_MODE=%q\n", mode)
if err := os.WriteFile(script, []byte(content), 0o755); err != nil {
t.Fatalf("write fake helper script: %v", err)
}
prev := RemoteHelperLookPath
RemoteHelperLookPath = func(_ string) (string, bool) { return script, true }
t.Cleanup(func() { RemoteHelperLookPath = prev })
path, ok := LookupRemoteHelper("entire")
if !ok {
t.Fatal("expected fake helper to resolve")
}
ep, err := url.Parse("entire://example.test/repo")
if err != nil {
t.Fatalf("parse url: %v", err)
}
return NewHelperConn(path, ep, "entire://example.test/repo", "target")
}
func TestHelperConn_RequestInfoRefs_V2Advertisement(t *testing.T) {
c := fakeHelperConn(t, "")
adv, err := c.RequestInfoRefs(context.Background(), "git-upload-pack", GitProtocolV2)
if err != nil {
t.Fatalf("RequestInfoRefs: %v", err)
}
caps, err := DecodeV2Capabilities(strings.NewReader(string(adv)))
if err != nil {
t.Fatalf("DecodeV2Capabilities: %v (adv=%q)", err, adv)
}
if !caps.Supports("ls-refs") || !caps.Supports("fetch") {
t.Fatalf("expected ls-refs and fetch capabilities, got %+v", caps.Caps)
}
}
func TestHelperConn_PostRPC_V2_StripsResponseEnd(t *testing.T) {
c := fakeHelperConn(t, "")
rc, err := c.PostRPCStreamBody(context.Background(), "git-upload-pack",
strings.NewReader(FormatPktLine("command=fetch\n")+"0000"), true, "fetch")
if err != nil {
t.Fatalf("PostRPCStreamBody: %v", err)
}
got, err := io.ReadAll(rc)
if err != nil {
t.Fatalf("read response: %v", err)
}
if err := rc.Close(); err != nil {
t.Fatalf("close: %v", err)
}
if string(got) != fakeFetchResponse {
t.Fatalf("response = %q, want %q (the trailing 0002 must be stripped)", got, fakeFetchResponse)
}
}
func TestHelperConn_PostRPC_ReceivePack_ReadsToEOF(t *testing.T) {
c := fakeHelperConn(t, "")
body := FormatPktLine("0000000000000000000000000000000000000000 1111111111111111111111111111111111111111 refs/heads/main\n") + "0000"
rc, err := c.PostRPCStreamBody(context.Background(), "git-receive-pack",
strings.NewReader(body), false, "push")
if err != nil {
t.Fatalf("PostRPCStreamBody: %v", err)
}
got, err := io.ReadAll(rc)
if err != nil {
t.Fatalf("read response: %v", err)
}
if err := rc.Close(); err != nil {
t.Fatalf("close: %v", err)
}
if string(got) != fakeReportStatus {
t.Fatalf("response = %q, want %q", got, fakeReportStatus)
}
}
func TestHelperConn_Fallback(t *testing.T) {
c := fakeHelperConn(t, "fallback")
_, err := c.RequestInfoRefs(context.Background(), "git-upload-pack", GitProtocolV2)
if err == nil {
t.Fatal("expected error on fallback response")
}
if !strings.Contains(err.Error(), "fallback") {
t.Fatalf("error = %v, want it to mention fallback", err)
}
}
func TestReadAdvertisement_StopsAtFlush(t *testing.T) {
in := FormatPktLine("version 2\n") + FormatPktLine("agent=x\n") + "0000" + "extra-bytes"
br := bufio.NewReader(strings.NewReader(in))
adv, err := readAdvertisement(br)
if err != nil {
t.Fatalf("readAdvertisement: %v", err)
}
want := FormatPktLine("version 2\n") + FormatPktLine("agent=x\n") + "0000"
if string(adv) != want {
t.Fatalf("adv = %q, want %q", adv, want)
}
rest, err := io.ReadAll(br)
if err != nil {
t.Fatalf("read leftover: %v", err)
}
if string(rest) != "extra-bytes" {
t.Fatalf("leftover = %q, want the bytes after the flush to remain buffered", rest)
}
}
func TestResponseEndReader_StripsResponseEnd(t *testing.T) {
in := FormatPktLine("data-one\n") + "0000" + FormatPktLine("data-two\n") + "0000" + "0002" + "trailing"
r := newResponseEndReader(bufio.NewReader(strings.NewReader(in)))
got, err := io.ReadAll(r)
if err != nil {
t.Fatalf("read: %v", err)
}
want := FormatPktLine("data-one\n") + "0000" + FormatPktLine("data-two\n") + "0000"
if string(got) != want {
t.Fatalf("got %q, want %q", got, want)
}
}
// Guard: building the exec.Cmd for the fake helper goes through the same dial
// path; ensure exec.LookPath agrees the resolved binary is runnable.
func TestRemoteHelperLookPath_Default(t *testing.T) {
if _, err := exec.LookPath("git"); err != nil {
t.Skip("git not on PATH")
}
if _, ok := LookupRemoteHelper(""); ok {
t.Fatal("empty scheme must not resolve")
}
}
Ainternal/gitproto/helper_test.go+243