From 96f0795a36d8890813b83c2866e44a125619953e Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Thu, 10 Sep 2026 12:04:00 -0700
Subject: [PATCH 1/8] test(registry): pin what the proxy owes a docker pull

The registry proxy tunnels one HTTP conversation between a local docker
daemon and buildkitd's embedded registry over a gRPC stream, and it had
no tests. These stand a real net/http server in for the embedded
registry, drive it with a real net/http client, and put the production
Server.Proxy between the two, so the bytes are framed by net/http at
both ends and the only question asked is whether they arrive intact.

The client half of the tunnel is written out in the test rather than
borrowed from earthly's regproxy, so the server is exercised against an
independently correct peer: one that ends each direction on half-close
and never on a timer.

Three of the four fail against this commit, each reproducing a symptom
seen in CI:

  - a response the registry pauses 300ms mid-body arrives truncated at
    exactly half its Content-Length, which is the
    "httpReadSeeker: failed open: ... EOF" behind the flaky
    "pull ping error" in +test-no-qemu-slow
  - a second request on a kept-alive connection is never answered at
    all -- it hangs until the client's timeout rather than failing
  - serving any response races on an error variable shared between the
    two copy directions, reported by -race, which is the suite the
    flake shows up in

The fourth, a 256KiB request body arriving byte-exact, passes today and
is here to hold that ground through the change that follows.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 api/services/registry/proxy_test.go | 444 ++++++++++++++++++++++++++++
 1 file changed, 444 insertions(+)
 create mode 100644 api/services/registry/proxy_test.go

diff --git a/api/services/registry/proxy_test.go b/api/services/registry/proxy_test.go
new file mode 100644
index 000000000..2df1c7171
--- /dev/null
+++ b/api/services/registry/proxy_test.go
@@ -0,0 +1,444 @@
+package earthly_registry_v1 //nolint:revive
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+)
+
+var errUnexpectedMessage = errors.New("stream carried a message that was not a ByteMessage")
+
+// The proxy carries one HTTP conversation between a local docker daemon and
+// buildkitd's embedded registry over a gRPC stream, as opaque bytes. These
+// tests stand a real HTTP server in for the embedded registry, drive it with a
+// real net/http client, and put the production Server.Proxy between the two.
+// Nothing here parses HTTP: whether the bytes arrive intact is the whole
+// question, so both ends are left to net/http to frame.
+
+// tunnel returns an http.Client whose connections are proxied to reg through
+// Server.Proxy, and a count of how many connections it dialled -- reuse of a
+// single connection is a property worth asserting, not assuming.
+func tunnel(t *testing.T, reg *httptest.Server) (*http.Client, func() int) {
+	t.Helper()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	srv := NewServer(reg.Listener.Addr().String())
+
+	var (
+		mu    sync.Mutex
+		dials int
+	)
+
+	tr := &http.Transport{
+		DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
+			mu.Lock()
+			dials++
+			mu.Unlock()
+
+			near, far, err := tcpPair(t)
+			if err != nil {
+				return nil, err
+			}
+
+			clientStream, serverStream := newStreamPair(ctx)
+
+			go func() {
+				defer serverStream.closeSend()
+				_ = srv.Proxy(serverStream)
+			}()
+
+			go func() {
+				defer far.Close()
+				_ = pump(far, clientStream)
+			}()
+
+			return near, nil
+		},
+	}
+	t.Cleanup(tr.CloseIdleConnections)
+
+	// A tunnel that mishandles termination strands a request rather than
+	// failing it, so bound every request: a hung pull is a failure too.
+	return &http.Client{Transport: tr, Timeout: 20 * time.Second}, func() int {
+		mu.Lock()
+		defer mu.Unlock()
+		return dials
+	}
+}
+
+// pump is the client half of the tunnel: the daemon-side counterpart to
+// Server.Proxy, written out here rather than borrowed from production code so
+// that the tests exercise the server against an independently correct peer.
+// Termination is by half-close in both directions, never by a timer.
+func pump(conn *net.TCPConn, stream *clientStream) error {
+	errs := make(chan error, 2)
+
+	go func() {
+		buf := make([]byte, 32*1024)
+		for {
+			n, err := conn.Read(buf)
+			if n > 0 {
+				if serr := stream.SendMsg(&ByteMessage{Data: buf[:n]}); serr != nil {
+					errs <- serr
+					return
+				}
+			}
+			if err != nil {
+				// The daemon has finished its request. Tell the far side, but
+				// keep reading the response.
+				stream.closeSend()
+				if err == io.EOF {
+					err = nil
+				}
+				errs <- err
+				return
+			}
+		}
+	}()
+
+	go func() {
+		for {
+			msg := &ByteMessage{}
+			err := stream.RecvMsg(msg)
+			if err != nil {
+				if err == io.EOF {
+					// The response is complete: signal end-of-body downstream
+					// without discarding anything the daemon still owes us.
+					errs <- conn.CloseWrite()
+					return
+				}
+				errs <- err
+				return
+			}
+			if _, err := conn.Write(msg.GetData()); err != nil {
+				errs <- err
+				return
+			}
+		}
+	}()
+
+	if err := <-errs; err != nil {
+		return err
+	}
+	return <-errs
+}
+
+// tcpPair returns the two ends of a real TCP connection. net.Pipe would be
+// simpler but has no CloseWrite, and half-close is precisely what is under
+// test.
+func tcpPair(t *testing.T) (near, far *net.TCPConn, err error) {
+	t.Helper()
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return nil, nil, err
+	}
+	defer ln.Close()
+
+	type accepted struct {
+		conn net.Conn
+		err  error
+	}
+	ch := make(chan accepted, 1)
+	go func() {
+		conn, err := ln.Accept()
+		ch <- accepted{conn: conn, err: err}
+	}()
+
+	dialed, err := net.Dial("tcp", ln.Addr().String())
+	if err != nil {
+		return nil, nil, err
+	}
+
+	got := <-ch
+	if got.err != nil {
+		dialed.Close()
+		return nil, nil, got.err
+	}
+
+	t.Cleanup(func() {
+		dialed.Close()
+		got.conn.Close()
+	})
+
+	return dialed.(*net.TCPConn), got.conn.(*net.TCPConn), nil
+}
+
+// newStreamPair returns the two ends of an in-memory bidirectional stream with
+// gRPC's semantics: a receive returns io.EOF once the peer has closed its send
+// direction and every message already sent has been delivered, and message
+// bytes are copied on send, as marshalling would copy them.
+func newStreamPair(ctx context.Context) (*clientStream, *serverStream) {
+	var (
+		toServer = make(chan []byte, 16)
+		toClient = make(chan []byte, 16)
+	)
+
+	c := &clientStream{halfStream: halfStream{ctx: ctx, send: toServer, recv: toClient}}
+	s := &serverStream{halfStream: halfStream{ctx: ctx, send: toClient, recv: toServer}}
+
+	return c, s
+}
+
+type halfStream struct {
+	ctx  context.Context
+	send chan []byte
+	recv chan []byte
+	once sync.Once
+}
+
+func (h *halfStream) SendMsg(m any) error {
+	msg, ok := m.(*ByteMessage)
+	if !ok {
+		return errUnexpectedMessage
+	}
+
+	data := make([]byte, len(msg.GetData()))
+	copy(data, msg.GetData())
+
+	select {
+	case h.send <- data:
+		return nil
+	case <-h.ctx.Done():
+		return h.ctx.Err()
+	}
+}
+
+func (h *halfStream) RecvMsg(m any) error {
+	msg, ok := m.(*ByteMessage)
+	if !ok {
+		return errUnexpectedMessage
+	}
+
+	select {
+	case data, open := <-h.recv:
+		if !open {
+			return io.EOF
+		}
+		msg.Data = data
+		return nil
+	case <-h.ctx.Done():
+		return h.ctx.Err()
+	}
+}
+
+func (h *halfStream) closeSend() {
+	h.once.Do(func() { close(h.send) })
+}
+
+func (h *halfStream) Context() context.Context { return h.ctx }
+
+type clientStream struct {
+	halfStream
+}
+
+func (c *clientStream) Send(m *ByteMessage) error { return c.SendMsg(m) }
+
+func (c *clientStream) Recv() (*ByteMessage, error) {
+	msg := &ByteMessage{}
+	if err := c.RecvMsg(msg); err != nil {
+		return nil, err
+	}
+	return msg, nil
+}
+
+// serverStream stands in for the generated Registry_ProxyServer. The embedded
+// grpc.ServerStream covers the header and trailer methods the proxy never
+// calls; the ones it does call are implemented above.
+type serverStream struct {
+	halfStream
+	grpc.ServerStream
+}
+
+func (s *serverStream) Send(m *ByteMessage) error { return s.halfStream.SendMsg(m) }
+
+func (s *serverStream) Recv() (*ByteMessage, error) {
+	msg := &ByteMessage{}
+	if err := s.halfStream.RecvMsg(msg); err != nil {
+		return nil, err
+	}
+	return msg, nil
+}
+
+func (s *serverStream) SendMsg(m any) error { return s.halfStream.SendMsg(m) }
+
+func (s *serverStream) RecvMsg(m any) error { return s.halfStream.RecvMsg(m) }
+
+func (s *serverStream) Context() context.Context { return s.halfStream.Context() }
+
+var _ Registry_ProxyServer = (*serverStream)(nil)
+
+// blob is a stand-in for a layer: larger than the 32KiB copy buffers on both
+// sides, so it crosses the stream as many messages.
+func blob(n int) []byte {
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = byte(i % 251)
+	}
+	return b
+}
+
+func TestProxyServesAResponse(t *testing.T) {
+	body := blob(128 * 1024)
+
+	reg := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+		_, _ = w.Write(body)
+	}))
+	t.Cleanup(reg.Close)
+
+	client, _ := tunnel(t, reg)
+
+	resp, err := client.Get("http://registry.invalid/v2/img/blobs/sha256:0")
+	if err != nil {
+		t.Fatalf("GET through the proxy: %v", err)
+	}
+	defer resp.Body.Close()
+
+	got, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("reading the proxied body: %v", err)
+	}
+
+	if len(got) != len(body) {
+		t.Fatalf("body length: got %d bytes, want %d", len(got), len(body))
+	}
+	if string(got) != string(body) {
+		t.Error("body differs from what the registry served")
+	}
+}
+
+func TestProxyDoesNotTruncateAResponseThatStallsMidBody(t *testing.T) {
+	body := blob(128 * 1024)
+
+	reg := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+		half := len(body) / 2
+
+		_, _ = w.Write(body[:half])
+		w.(http.Flusher).Flush()
+
+		// A loaded runner -- the -race integration suite, say -- can leave a
+		// registry this long between chunks. Silence on the socket is not the
+		// end of a response.
+		time.Sleep(300 * time.Millisecond)
+
+		_, _ = w.Write(body[half:])
+	}))
+	t.Cleanup(reg.Close)
+
+	client, _ := tunnel(t, reg)
+
+	resp, err := client.Get("http://registry.invalid/v2/img/blobs/sha256:0")
+	if err != nil {
+		t.Fatalf("GET through the proxy: %v", err)
+	}
+	defer resp.Body.Close()
+
+	got, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("reading a body the registry paused mid-way: %v (got %d of %d bytes)", err, len(got), len(body))
+	}
+	if string(got) != string(body) {
+		t.Errorf("body differs from what the registry served: got %d bytes, want %d", len(got), len(body))
+	}
+}
+
+func TestProxyReusesOneConnectionForTwoRequests(t *testing.T) {
+	body := blob(4096)
+
+	reg := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+		_, _ = w.Write(body)
+	}))
+	t.Cleanup(reg.Close)
+
+	client, dials := tunnel(t, reg)
+
+	for i := 0; i < 2; i++ {
+		if i > 0 {
+			// docker does its own work between a manifest fetch and the blob
+			// fetches that follow. A connection that only survives while
+			// requests arrive back-to-back is not a keep-alive connection.
+			time.Sleep(200 * time.Millisecond)
+		}
+
+		resp, err := client.Get("http://registry.invalid/v2/img/manifests/latest")
+		if err != nil {
+			t.Fatalf("request %d through the proxy: %v", i+1, err)
+		}
+
+		got, err := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		if err != nil {
+			t.Fatalf("reading response %d: %v", i+1, err)
+		}
+		if string(got) != string(body) {
+			t.Fatalf("response %d differs from what the registry served", i+1)
+		}
+	}
+
+	// docker holds keep-alive connections open across the many manifest and
+	// blob requests of one pull. Tearing the tunnel down after the first
+	// response makes every later request pay for a new stream, and strands any
+	// request already in flight on the old one.
+	if n := dials(); n != 1 {
+		t.Errorf("connections dialled: got %d, want 1 -- the tunnel did not survive the first response", n)
+	}
+}
+
+func TestProxyPassesALargeRequestBodyThrough(t *testing.T) {
+	sent := blob(256 * 1024)
+
+	var (
+		mu       sync.Mutex
+		received []byte
+	)
+
+	reg := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := io.ReadAll(r.Body)
+		mu.Lock()
+		received = b
+		mu.Unlock()
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+	}))
+	t.Cleanup(reg.Close)
+
+	client, _ := tunnel(t, reg)
+
+	resp, err := client.Post("http://registry.invalid/v2/img/blobs/uploads/", "application/octet-stream", bytes.NewReader(sent))
+	if err != nil {
+		t.Fatalf("POST through the proxy: %v", err)
+	}
+	defer resp.Body.Close()
+	_, _ = io.Copy(io.Discard, resp.Body)
+
+	if resp.StatusCode != http.StatusCreated {
+		t.Fatalf("status: got %d, want %d", resp.StatusCode, http.StatusCreated)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if len(received) != len(sent) {
+		t.Fatalf("request body length: got %d bytes, want %d", len(received), len(sent))
+	}
+	if string(received) != string(sent) {
+		t.Error("request body differs from what was sent")
+	}
+}

From cdf89cdce3f63fb05ba4146cd24c2c59066f65e2 Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Thu, 10 Sep 2026 12:05:45 -0700
Subject: [PATCH 2/8] fix(registry): end a proxied response on half-close, not
 50ms of silence

CopyWithDeadline inferred the end of an HTTP response from the socket
going quiet for 50ms: it set a read deadline before every read and, on
the timeout, declared the response complete and returned. Proxy then
returned too, closing both the connection to the embedded registry and
the gRPC stream.

A registry that pauses longer than that mid-body is not a registry that
has finished. On a loaded runner -- the -race integration suite on a
27-layer image, say -- it pauses for longer routinely, and the client
then gets a truncated body against a Content-Length that promised more,
which docker reports as a bare

    httpReadSeeker: failed open: failed to do request: Get
      "https://127.0.0.1:PORT/v2/.../blobs/sha256:...": EOF

with no status to explain it. The same 50ms decided the fate of the
whole connection, so a kept-alive connection was torn down after its
first response and any later request on it was never answered.

A byte tunnel does not need to know where a response ends. Copy ends
each direction when its source ends it -- io.EOF from the socket, or a
peer that closed its send direction -- and passes that on as a
half-close, so the far end can finish what it still owes. This is the
shape session/sshforward.Copy has used to carry forwarded agent sockets
for years, and it is what kubectl port-forward does with the same
problem: SPDY streams per connection, terminated by close, with no idle
heuristic anywhere.

Two further defects go with it, both of which the tests caught:

  - Proxy's two copy goroutines assigned to one captured err, a data
    race reported under -race, which is the suite this flake appears in
  - errgroup.WithContext's derived context was discarded, so neither
    direction failing cancelled the other; a finished response
    direction left the request direction blocked in Recv and Proxy
    never returned, which is why an unanswered keep-alive request hung
    rather than failed

The proxy now holds the stream and the loopback connection for as long
as the client holds its connection, rather than dropping them 50ms
after the first response. That is the same lifetime port-forward gives
a forwarded connection, and teardown still arrives with the session
context.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 api/services/registry/copy.go   | 102 ++++++++++++++++++++++++++++++++
 api/services/registry/server.go |  79 +++----------------------
 2 files changed, 111 insertions(+), 70 deletions(-)
 create mode 100644 api/services/registry/copy.go

diff --git a/api/services/registry/copy.go b/api/services/registry/copy.go
new file mode 100644
index 000000000..2a0628c72
--- /dev/null
+++ b/api/services/registry/copy.go
@@ -0,0 +1,102 @@
+package earthly_registry_v1 //nolint:revive
+
+import (
+	"context"
+	"io"
+
+	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
+)
+
+// Stream is the part of a gRPC bidirectional stream the tunnel needs. Both
+// ends of Registry.Proxy satisfy it, so the same copy runs on the daemon and
+// on the client.
+type Stream interface {
+	SendMsg(m any) error
+	RecvMsg(m any) error
+}
+
+// Copy joins a connection to a gRPC stream in both directions and returns once
+// both are done. The bytes are opaque: nothing here knows where one HTTP
+// request or response ends, because nothing needs to. Each direction ends when
+// its source says so -- io.EOF from the connection, or a peer that closed its
+// send direction -- and that end is passed on as a half-close, so the other
+// side can finish what it still owes before the whole conversation is torn
+// down.
+//
+// This mirrors session/sshforward.Copy, which has carried forwarded agent
+// sockets for years; see that file for the same shape with commentary.
+func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStream func() error) error {
+	defer conn.Close()
+
+	eg, ctx := errgroup.WithContext(ctx)
+
+	// Peer to connection.
+	eg.Go(func() error {
+		msg := &ByteMessage{}
+		for {
+			if err := stream.RecvMsg(msg); err != nil {
+				if errors.Is(err, io.EOF) {
+					// The peer has finished sending. It is still reading, so
+					// close only this direction and leave the response to
+					// come back.
+					if closeWriter, ok := conn.(interface{ CloseWrite() error }); ok {
+						// Best effort: the read half stays open either way.
+						closeWriter.CloseWrite()
+					} else {
+						conn.Close()
+					}
+					return nil
+				}
+				conn.Close()
+				return errors.Wrap(err, "failed to receive from the stream")
+			}
+
+			select {
+			case <-ctx.Done():
+				conn.Close()
+				return context.Cause(ctx)
+			default:
+			}
+
+			if _, err := conn.Write(msg.GetData()); err != nil {
+				conn.Close()
+				return errors.Wrap(err, "failed to write to the connection")
+			}
+
+			msg.Data = msg.Data[:0]
+		}
+	})
+
+	// Connection to peer.
+	eg.Go(func() error {
+		buf := make([]byte, 32*1024)
+		for {
+			n, err := conn.Read(buf)
+			if n > 0 {
+				if err := stream.SendMsg(&ByteMessage{Data: buf[:n]}); err != nil {
+					return errors.Wrap(err, "failed to send to the stream")
+				}
+			}
+
+			if err != nil {
+				if errors.Is(err, io.EOF) {
+					// Everything the connection had to say has been said.
+					if closeStream != nil {
+						return errors.Wrap(closeStream(), "failed to close the stream")
+					}
+					return nil
+				}
+				return errors.Wrap(err, "failed to read from the connection")
+			}
+
+			select {
+			case <-ctx.Done():
+				return context.Cause(ctx)
+			default:
+			}
+		}
+	})
+
+	return eg.Wait()
+}
diff --git a/api/services/registry/server.go b/api/services/registry/server.go
index 0d7c8c118..23abb2758 100644
--- a/api/services/registry/server.go
+++ b/api/services/registry/server.go
@@ -1,17 +1,12 @@
 package earthly_registry_v1 //nolint:revive
 
 import (
-	"io"
 	"net"
 	"strings"
-	"time"
 
 	"github.com/pkg/errors"
-	"golang.org/x/sync/errgroup"
 )
 
-const readDeadline = 50 * time.Millisecond
-
 // NewServer creates and returns a new proxy server with a given host and client.
 func NewServer(addr string) *Server {
 	return &Server{
@@ -75,76 +70,20 @@ func (s *StreamRW) Read(p []byte) (int, error) {
 // pipe them back out through the stream again. This allows us to send HTTP
 // requests to the embedded registry without having to connect via some other
 // exposed server or port.
+//
+// One stream carries one connection from the client's local listener, for as
+// long as the client keeps it: docker reuses a connection across the manifest
+// and blob requests of a single pull, and each of those requests is answered
+// on the stream that carried it.
 func (s *Server) Proxy(stream Registry_ProxyServer) error {
-	rw := NewStreamRW(stream)
-
 	addr := strings.ReplaceAll(s.addr, "0.0.0.0", "127.0.0.1")
 
 	conn, err := net.Dial("tcp", addr)
 	if err != nil {
-		return err
-	}
-	defer conn.Close()
-
-	ctx := stream.Context()
-	eg, _ := errgroup.WithContext(ctx)
-
-	eg.Go(func() error {
-		_, err = io.Copy(conn, rw)
-		if err != nil {
-			return errors.Wrap(err, "failed to copy from stream to host")
-		}
-		return nil
-	})
-
-	eg.Go(func() error {
-		_, err = CopyWithDeadline(conn, rw)
-		if err != nil {
-			return errors.Wrap(err, "failed to copy from host to stream")
-		}
-		return nil
-	})
-
-	err = eg.Wait()
-	if err != nil {
-		return errors.Wrap(err, "failed to wait")
-	}
-
-	return nil
-}
-
-// CopyWithDeadline copies data from a net.Conn using a read deadline. The
-// process will fail with a timeout error if no data is read for the defined
-// period.
-func CopyWithDeadline(conn net.Conn, w io.Writer) (int64, error) {
-	var (
-		t   = int64(0)
-		buf = make([]byte, 32*1024)
-	)
-	for {
-		err := conn.SetReadDeadline(time.Now().Add(readDeadline))
-		if err != nil {
-			return t, err
-		}
-		n, err := conn.Read(buf)
-		if err != nil {
-			if errors.Is(err, io.EOF) || isNetTimeout(err) {
-				break
-			}
-			return t, err
-		}
-		n, err = w.Write(buf[0:n])
-		t += int64(n)
-		if err != nil {
-			return t, err
-		}
+		return errors.Wrapf(err, "failed to dial the embedded registry at %s", addr)
 	}
-	return t, nil
-}
 
-func isNetTimeout(err error) bool {
-	if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
-		return true
-	}
-	return false
+	// The stream is closed by returning from this handler, so there is no send
+	// direction for the copy to close on its own.
+	return Copy(stream.Context(), conn, stream, nil)
 }

From 80ed7fa6b0f11f9ff4ae13598433417749729b43 Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Thu, 10 Sep 2026 12:06:12 -0700
Subject: [PATCH 3/8] refactor(registry): drop StreamRW, which the tunnel no
 longer needs

Copy moves bytes between the connection and the stream directly, so
there is no longer an io.Reader/io.Writer adapter in the path. Nothing
else in the fork used StreamRW; earthly's regproxy did, and moves to
Copy in the change that bumps its pin.

Worth recording why its removal is a relief rather than a loss, because
the type was one buffer-size change away from corrupting pulled images.
Read did this:

    l := 0
    if len(s.last) > 0 {
        l = copy(p, s.last)      // stash the leftovers into p
    }
    msg, err := s.stream.Recv()
    ...
    s.last = msg.GetData()
    n := copy(p, s.last)         // and overwrite them
    s.last = s.last[n:]
    return n + l                 // counting both

Three contract violations in one function: the second copy overwrites
the leftovers the first just wrote and they are never re-stashed; the
returned count is the sum of two copies into the same region, so it can
exceed len(p), which makes an io.Copy caller slice past its buffer's
capacity and panic; and leftovers are silently dropped when Recv
returns an error, including io.EOF at the clean end of a stream.

It was latent because a leftover only appears when a message is larger
than the reader's buffer, and both pumps read 32KiB while both senders
sent at most 32KiB. Any change to either buffer, or a peer that chunked
differently, would have corrupted image bytes rather than failing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 api/services/registry/server.go | 46 ---------------------------------
 1 file changed, 46 deletions(-)

diff --git a/api/services/registry/server.go b/api/services/registry/server.go
index 23abb2758..235aea72d 100644
--- a/api/services/registry/server.go
+++ b/api/services/registry/server.go
@@ -20,52 +20,6 @@ type Server struct {
 	UnimplementedRegistryServer
 }
 
-type streamSource interface {
-	Send(*ByteMessage) error
-	Recv() (*ByteMessage, error)
-}
-
-// NewStreamRW creates and returns a gRPC stream reader-writer that implements
-// io.Reader & io.Writer as to utilize the gRPC stream with standard methods.
-func NewStreamRW(stream streamSource) *StreamRW {
-	return &StreamRW{stream: stream}
-}
-
-type StreamRW struct {
-	stream streamSource
-	last   []byte
-}
-
-// Write implements io.Writer.
-func (s *StreamRW) Write(p []byte) (int, error) {
-	err := s.stream.Send(&ByteMessage{
-		Data: p,
-	})
-	if err != nil {
-		return 0, errors.Wrap(err, "failed to write data to client")
-	}
-	return len(p), nil
-}
-
-// Read implements io.Reader.
-func (s *StreamRW) Read(p []byte) (int, error) {
-	l := 0
-	if len(s.last) > 0 {
-		l = copy(p, s.last)
-	}
-
-	msg, err := s.stream.Recv()
-	if err != nil {
-		return 0, err
-	}
-
-	s.last = msg.GetData()
-	n := copy(p, s.last)
-	s.last = s.last[n:]
-
-	return n + l, nil
-}
-
 // Proxy requests sent via gRPC data stream to the embedded Docker registry and
 // pipe them back out through the stream again. This allows us to send HTTP
 // requests to the embedded registry without having to connect via some other

From a593020900648fe60209ae1e6363300caad79785 Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Fri, 11 Sep 2026 08:59:20 -0700
Subject: [PATCH 4/8] refactor(registry): name the copy buffer size and say why
 it is 32KiB

The number was a bare literal with no stated reason, which invites the
question of why not 16KiB or 64KiB. It is io.Copy's own default and what
session/sshforward.Copy uses for the same job, and it sits well under
gRPC's 4MiB default maximum message size.

Worth recording alongside that: nothing depends on the two ends of the
tunnel agreeing on the value. That they happened to agree is precisely
what kept StreamRW's broken leftover handling from ever being reached.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 api/services/registry/copy.go | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/api/services/registry/copy.go b/api/services/registry/copy.go
index 2a0628c72..9b10a7d50 100644
--- a/api/services/registry/copy.go
+++ b/api/services/registry/copy.go
@@ -8,6 +8,16 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+// copyBufferSize is the size of the buffer each direction reads into. 32KiB is
+// io.Copy's own default, and what session/sshforward.Copy and
+// session/socketforward use for the same job; it is comfortably under gRPC's
+// 4MiB default maximum message size, so a full buffer never needs splitting
+// across messages. Nothing depends on the two ends of the tunnel choosing the
+// same value -- that they happened to agree is what kept StreamRW's leftover
+// handling from ever being exercised -- so this is a throughput knob and
+// nothing more.
+const copyBufferSize = 32 * 1024
+
 // Stream is the part of a gRPC bidirectional stream the tunnel needs. Both
 // ends of Registry.Proxy satisfy it, so the same copy runs on the daemon and
 // on the client.
@@ -70,7 +80,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 
 	// Connection to peer.
 	eg.Go(func() error {
-		buf := make([]byte, 32*1024)
+		buf := make([]byte, copyBufferSize)
 		for {
 			n, err := conn.Read(buf)
 			if n > 0 {

From 3398faa225b0acb54ddfdc9bcb48813a12fd788f Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Fri, 11 Sep 2026 09:03:14 -0700
Subject: [PATCH 5/8] refactor(registry): std errors, and error messages that
 name the operation

github.com/pkg/errors is archived and moby is migrating off it, so new
code in this package should not add to the pile. Wrapping moves to
fmt.Errorf with %w; errors.Is and errors.New come from the standard
library unchanged.

The messages lose their "failed to" prefixes. They compose -- a wrapped
chain reads "failed to X: failed to Y: failed to Z" -- and the reader
already knows an error is a failure. What they do not know is what the
program was attempting, so that is what the message now says.

One conversion is not mechanical: errors.Wrap returns nil for a nil
cause, and fmt.Errorf does not. The closeStream call relied on that, and
now checks the error itself rather than wrapping unconditionally.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 api/services/registry/copy.go       | 15 +++++++++------
 api/services/registry/proxy_test.go |  2 +-
 api/services/registry/server.go     |  5 ++---
 3 files changed, 12 insertions(+), 10 deletions(-)

diff --git a/api/services/registry/copy.go b/api/services/registry/copy.go
index 9b10a7d50..c307ba2c7 100644
--- a/api/services/registry/copy.go
+++ b/api/services/registry/copy.go
@@ -2,9 +2,10 @@ package earthly_registry_v1 //nolint:revive
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"io"
 
-	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -59,7 +60,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 					return nil
 				}
 				conn.Close()
-				return errors.Wrap(err, "failed to receive from the stream")
+				return fmt.Errorf("receive from stream: %w", err)
 			}
 
 			select {
@@ -71,7 +72,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 
 			if _, err := conn.Write(msg.GetData()); err != nil {
 				conn.Close()
-				return errors.Wrap(err, "failed to write to the connection")
+				return fmt.Errorf("write to connection: %w", err)
 			}
 
 			msg.Data = msg.Data[:0]
@@ -85,7 +86,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 			n, err := conn.Read(buf)
 			if n > 0 {
 				if err := stream.SendMsg(&ByteMessage{Data: buf[:n]}); err != nil {
-					return errors.Wrap(err, "failed to send to the stream")
+					return fmt.Errorf("send to stream: %w", err)
 				}
 			}
 
@@ -93,11 +94,13 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 				if errors.Is(err, io.EOF) {
 					// Everything the connection had to say has been said.
 					if closeStream != nil {
-						return errors.Wrap(closeStream(), "failed to close the stream")
+						if err := closeStream(); err != nil {
+							return fmt.Errorf("close stream: %w", err)
+						}
 					}
 					return nil
 				}
-				return errors.Wrap(err, "failed to read from the connection")
+				return fmt.Errorf("read from connection: %w", err)
 			}
 
 			select {
diff --git a/api/services/registry/proxy_test.go b/api/services/registry/proxy_test.go
index 2df1c7171..9516ac96e 100644
--- a/api/services/registry/proxy_test.go
+++ b/api/services/registry/proxy_test.go
@@ -12,7 +12,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pkg/errors"
+	"errors"
 	"google.golang.org/grpc"
 )
 
diff --git a/api/services/registry/server.go b/api/services/registry/server.go
index 235aea72d..7397f9015 100644
--- a/api/services/registry/server.go
+++ b/api/services/registry/server.go
@@ -1,10 +1,9 @@
 package earthly_registry_v1 //nolint:revive
 
 import (
+	"fmt"
 	"net"
 	"strings"
-
-	"github.com/pkg/errors"
 )
 
 // NewServer creates and returns a new proxy server with a given host and client.
@@ -34,7 +33,7 @@ func (s *Server) Proxy(stream Registry_ProxyServer) error {
 
 	conn, err := net.Dial("tcp", addr)
 	if err != nil {
-		return errors.Wrapf(err, "failed to dial the embedded registry at %s", addr)
+		return fmt.Errorf("dial embedded registry at %s: %w", addr, err)
 	}
 
 	// The stream is closed by returning from this handler, so there is no send

From 908f7aa24be8395f3330308a5b885f5ab75fca5b Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Fri, 11 Sep 2026 10:18:28 -0700
Subject: [PATCH 6/8] fix(registry): group the errors import with the standard
 library

goimports groups standard-library imports separately from third-party
ones, and swapping github.com/pkg/errors for errors in place left the
latter sitting in the third-party block. gofmt does not regroup imports,
so only goimports catches it -- which is what CI runs.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 api/services/registry/proxy_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/api/services/registry/proxy_test.go b/api/services/registry/proxy_test.go
index 9516ac96e..1aeff1938 100644
--- a/api/services/registry/proxy_test.go
+++ b/api/services/registry/proxy_test.go
@@ -3,6 +3,7 @@ package earthly_registry_v1 //nolint:revive
 import (
 	"bytes"
 	"context"
+	"errors"
 	"io"
 	"net"
 	"net/http"
@@ -12,7 +13,6 @@ import (
 	"testing"
 	"time"
 
-	"errors"
 	"google.golang.org/grpc"
 )
 

From d6ba72135157c5f952b6075de14522561807d5a9 Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Fri, 11 Sep 2026 10:23:02 -0700
Subject: [PATCH 7/8] refactor(registry): branch on the read error before
 forwarding bytes

Per review. The io.Reader contract's substance is in the EOF case -- a
final read may hand back bytes alongside io.EOF, and dropping those
truncates the body -- so that case still forwards what it got. A hard
read error is different: it is terminal for a TCP connection, the copy
returns, the stream errors out, and the bytes in the buffer belong to a
response that will never be completed. Forwarding them buys nothing, so
the error case no longer does.

The switch also reads better than a send wedged in front of an error
check, which is what prompted the comment.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 api/services/registry/copy.go | 43 ++++++++++++++++++++++++-----------
 1 file changed, 30 insertions(+), 13 deletions(-)

diff --git a/api/services/registry/copy.go b/api/services/registry/copy.go
index c307ba2c7..6a58ee5f9 100644
--- a/api/services/registry/copy.go
+++ b/api/services/registry/copy.go
@@ -82,27 +82,44 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 	// Connection to peer.
 	eg.Go(func() error {
 		buf := make([]byte, copyBufferSize)
+
+		send := func(n int) error {
+			if n == 0 {
+				return nil
+			}
+			if err := stream.SendMsg(&ByteMessage{Data: buf[:n]}); err != nil {
+				return fmt.Errorf("send to stream: %w", err)
+			}
+			return nil
+		}
+
 		for {
 			n, err := conn.Read(buf)
-			if n > 0 {
-				if err := stream.SendMsg(&ByteMessage{Data: buf[:n]}); err != nil {
-					return fmt.Errorf("send to stream: %w", err)
+			switch {
+			case errors.Is(err, io.EOF):
+				// Everything the connection had to say has been said. A final
+				// read is allowed to hand back bytes alongside the EOF, so
+				// they still go out before the stream is closed.
+				if err := send(n); err != nil {
+					return err
 				}
-			}
-
-			if err != nil {
-				if errors.Is(err, io.EOF) {
-					// Everything the connection had to say has been said.
-					if closeStream != nil {
-						if err := closeStream(); err != nil {
-							return fmt.Errorf("close stream: %w", err)
-						}
+				if closeStream != nil {
+					if err := closeStream(); err != nil {
+						return fmt.Errorf("close stream: %w", err)
 					}
-					return nil
 				}
+				return nil
+			case err != nil:
+				// A read error on the connection is terminal. Whatever is in
+				// the buffer belongs to a response that will never be
+				// completed, so there is nothing worth forwarding.
 				return fmt.Errorf("read from connection: %w", err)
 			}
 
+			if err := send(n); err != nil {
+				return err
+			}
+
 			select {
 			case <-ctx.Done():
 				return context.Cause(ctx)

From da886f2199ea1b73313647ea3466ecef737cf67d Mon Sep 17 00:00:00 2001
From: Kieran Mann <kieran.mann@o8t.com>
Date: Tue, 6 Oct 2026 19:35:29 -0700
Subject: [PATCH 8/8] refactor(registry): use pkg/errors, as the upstream lint
 rules require

Upstream buildkit still forbids fmt.Errorf in favour of pkg/errors, and
keeping .golangci.yml identical to upstream is worth more than getting
ahead of it in one package: every difference is another conflict on the
next upstream merge. The move to the standard library's errors is better
done tree-wide, once, after the big buildkit merge -- or by waiting for
upstream to make it.

The error messages keep naming the operation rather than the failure.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 api/services/registry/copy.go       | 13 ++++++-------
 api/services/registry/proxy_test.go |  2 +-
 api/services/registry/server.go     |  5 +++--
 3 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/api/services/registry/copy.go b/api/services/registry/copy.go
index 6a58ee5f9..d3f680d64 100644
--- a/api/services/registry/copy.go
+++ b/api/services/registry/copy.go
@@ -2,10 +2,9 @@ package earthly_registry_v1 //nolint:revive
 
 import (
 	"context"
-	"errors"
-	"fmt"
 	"io"
 
+	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -60,7 +59,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 					return nil
 				}
 				conn.Close()
-				return fmt.Errorf("receive from stream: %w", err)
+				return errors.Wrap(err, "receive from stream")
 			}
 
 			select {
@@ -72,7 +71,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 
 			if _, err := conn.Write(msg.GetData()); err != nil {
 				conn.Close()
-				return fmt.Errorf("write to connection: %w", err)
+				return errors.Wrap(err, "write to connection")
 			}
 
 			msg.Data = msg.Data[:0]
@@ -88,7 +87,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 				return nil
 			}
 			if err := stream.SendMsg(&ByteMessage{Data: buf[:n]}); err != nil {
-				return fmt.Errorf("send to stream: %w", err)
+				return errors.Wrap(err, "send to stream")
 			}
 			return nil
 		}
@@ -105,7 +104,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 				}
 				if closeStream != nil {
 					if err := closeStream(); err != nil {
-						return fmt.Errorf("close stream: %w", err)
+						return errors.Wrap(err, "close stream")
 					}
 				}
 				return nil
@@ -113,7 +112,7 @@ func Copy(ctx context.Context, conn io.ReadWriteCloser, stream Stream, closeStre
 				// A read error on the connection is terminal. Whatever is in
 				// the buffer belongs to a response that will never be
 				// completed, so there is nothing worth forwarding.
-				return fmt.Errorf("read from connection: %w", err)
+				return errors.Wrap(err, "read from connection")
 			}
 
 			if err := send(n); err != nil {
diff --git a/api/services/registry/proxy_test.go b/api/services/registry/proxy_test.go
index 1aeff1938..2df1c7171 100644
--- a/api/services/registry/proxy_test.go
+++ b/api/services/registry/proxy_test.go
@@ -3,7 +3,6 @@ package earthly_registry_v1 //nolint:revive
 import (
 	"bytes"
 	"context"
-	"errors"
 	"io"
 	"net"
 	"net/http"
@@ -13,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 )
 
diff --git a/api/services/registry/server.go b/api/services/registry/server.go
index 7397f9015..09d54e23c 100644
--- a/api/services/registry/server.go
+++ b/api/services/registry/server.go
@@ -1,9 +1,10 @@
 package earthly_registry_v1 //nolint:revive
 
 import (
-	"fmt"
 	"net"
 	"strings"
+
+	"github.com/pkg/errors"
 )
 
 // NewServer creates and returns a new proxy server with a given host and client.
@@ -33,7 +34,7 @@ func (s *Server) Proxy(stream Registry_ProxyServer) error {
 
 	conn, err := net.Dial("tcp", addr)
 	if err != nil {
-		return fmt.Errorf("dial embedded registry at %s: %w", addr, err)
+		return errors.Wrapf(err, "dial embedded registry at %s", addr)
 	}
 
 	// The stream is closed by returning from this handler, so there is no send