From 0bdad4667cd09f9cf955ff446c1828051ce17fbe Mon Sep 17 00:00:00 2001 From: CrazyMax <1951866+crazy-max@users.noreply.github.com> Date: Sat, 19 Sep 2026 00:43:07 +0200 Subject: [PATCH] remotes: test fetch retries and HTTP reader lifetime Verify that small and large layer uploads can be replayed after the registry consumes the body and returns HTTP 408. This covers the fetch-to-pusher path used by BuildKit registry cache exports. Check that HTTP readers close their response bodies at EOF while remaining seekable, and that the body is closed exactly once before the destination writer commits. Signed-off-by: CrazyMax <1951866+crazy-max@users.noreply.github.com> --- core/remotes/docker/fetcher_test.go | 107 ++++++++++++++++++++++++++++ core/remotes/docker/pusher_test.go | 57 +++++++++++++++ core/remotes/handlers.go | 3 + 3 files changed, 167 insertions(+) diff --git a/core/remotes/docker/fetcher_test.go b/core/remotes/docker/fetcher_test.go index 624a9b2d2b..ed8fccbd10 100644 --- a/core/remotes/docker/fetcher_test.go +++ b/core/remotes/docker/fetcher_test.go @@ -30,6 +30,8 @@ import ( "net/http/httptest" "net/textproto" "net/url" + "os" + "path/filepath" "strconv" "strings" "sync" @@ -38,19 +40,124 @@ import ( "time" "github.com/klauspost/compress/zstd" + "github.com/opencontainers/go-digest" ocispec "github.com/opencontainers/image-spec/specs-go/v1" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "golang.org/x/sync/semaphore" + "github.com/containerd/containerd/v2/core/content" + "github.com/containerd/containerd/v2/core/remotes" "github.com/containerd/containerd/v2/core/transfer" "github.com/containerd/containerd/v2/pkg/reference" + "github.com/containerd/containerd/v2/plugins/content/local" ) type writeFunc func(p []byte) (int, error) func (f writeFunc) Write(p []byte) (int, error) { return f(p) } +func TestHTTPReadSeekerClosesBodyAtEOF(t *testing.T) { + payload := []byte("layer content") + path := filepath.Join(t.TempDir(), "body") + require.NoError(t, os.WriteFile(path, payload, 0600)) + var bodies []*os.File + rc, err := newHTTPReadSeeker(int64(len(payload)), func(offset int64) (io.ReadCloser, error) { + body, err := os.Open(path) + if err != nil { + return nil, err + } + if _, err := body.Seek(offset, io.SeekStart); err != nil { + body.Close() + return nil, err + } + bodies = append(bodies, body) + return body, nil + }) + require.NoError(t, err) + defer rc.Close() + + for i := range 2 { + data, err := io.ReadAll(rc) + require.NoError(t, err) + require.Equal(t, payload, data) + require.Len(t, bodies, i+1) + // The HTTP body must be closed before the caller commits its writer, + // without closing the seekable reader needed for a possible reset. + _, err = bodies[i].Read(make([]byte, 1)) + require.ErrorIs(t, err, os.ErrClosed) + _, err = rc.(io.Seeker).Seek(0, io.SeekStart) + require.NoError(t, err) + } +} + +func TestFetchClosesHTTPBodyBeforeCommit(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + payload := []byte("layer content") + desc := ocispec.Descriptor{ + MediaType: ocispec.MediaTypeImageLayer, + Digest: digest.FromBytes(payload), + Size: int64(len(payload)), + } + registry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(len(payload))) + w.Write(payload) + })) + defer registry.Close() + var closed atomic.Int32 + client := registry.Client() + transport := client.Transport + client.Transport = rtFunc(func(r *http.Request) (*http.Response, error) { + resp, err := transport.RoundTrip(r) + if err == nil { + resp.Body = &fnOnClose{ + ReadCloser: resp.Body, + BeforeClose: func() { closed.Add(1) }, + } + } + return resp, err + }) + resolver := NewResolver(ResolverOptions{Client: client, PlainHTTP: true}) + fetcher, err := resolver.Fetcher(ctx, strings.TrimPrefix(registry.URL, "http://")+"/image:latest") + require.NoError(t, err) + store, err := local.NewStore(t.TempDir()) + require.NoError(t, err) + committed := false + ingester := beforeCommitIngester{Ingester: store, beforeCommit: func() { + // Even if Commit blocks, CRI must no longer count this HTTP request + // as active. Check before Fetch's deferred reader.Close can run. + require.EqualValues(t, 1, closed.Load()) + committed = true + }} + require.NoError(t, remotes.Fetch(ctx, ingester, fetcher, desc)) + require.True(t, committed) + require.EqualValues(t, 1, closed.Load()) +} + +type beforeCommitIngester struct { + content.Ingester + beforeCommit func() +} + +func (i beforeCommitIngester) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) { + w, err := i.Ingester.Writer(ctx, opts...) + if err != nil { + return nil, err + } + return beforeCommitWriter{Writer: w, beforeCommit: i.beforeCommit}, nil +} + +type beforeCommitWriter struct { + content.Writer + beforeCommit func() +} + +func (w beforeCommitWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error { + w.beforeCommit() + return w.Writer.Commit(ctx, size, expected, opts...) +} + func TestFetcherOpen(t *testing.T) { content := make([]byte, 128) rand.New(rand.NewSource(1)).Read(content) diff --git a/core/remotes/docker/pusher_test.go b/core/remotes/docker/pusher_test.go index 2e2a8e8070..e614ffc71d 100644 --- a/core/remotes/docker/pusher_test.go +++ b/core/remotes/docker/pusher_test.go @@ -26,13 +26,17 @@ import ( "net/http" "net/http/httptest" "net/url" + "os" + "path/filepath" "reflect" "regexp" "slices" + "strconv" "strings" "sync" "sync/atomic" "testing" + "time" "github.com/containerd/containerd/v2/core/content" "github.com/containerd/containerd/v2/core/remotes" @@ -46,6 +50,59 @@ import ( "github.com/stretchr/testify/require" ) +// Fetch can also copy local content to a pusher-backed ingester, as BuildKit's +// registry cache exporter does. The source must remain seekable after EOF so +// content.Copy can replay it when the push writer returns content.ErrReset. +func TestFetchPushRetryAfterEOF(t *testing.T) { + for _, size := range []int{128, 2 << 20} { + t.Run(strconv.Itoa(size), func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + payload := bytes.Repeat([]byte("x"), size) + desc := ocispec.Descriptor{ + MediaType: ocispec.MediaTypeImageLayer, + Digest: digest.FromBytes(payload), + Size: int64(len(payload)), + } + path := filepath.Join(t.TempDir(), "layer") + require.NoError(t, os.WriteFile(path, payload, 0600)) + var attempts atomic.Int32 + registry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodPost: + w.Header().Set("Location", "/v2/cache/blobs/uploads/test") + w.WriteHeader(http.StatusAccepted) + case http.MethodPut: + body, err := io.ReadAll(r.Body) + if !assert.NoError(t, err) || !assert.Equal(t, payload, body) { + w.WriteHeader(http.StatusBadRequest) + return + } + if attempts.Add(1) == 1 { + w.WriteHeader(http.StatusRequestTimeout) + return + } + w.Header().Set("Docker-Content-Digest", desc.Digest.String()) + w.WriteHeader(http.StatusCreated) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer registry.Close() + resolver := NewResolver(ResolverOptions{PlainHTTP: true}) + pusher, err := resolver.Pusher(ctx, strings.TrimPrefix(registry.URL, "http://")+"/cache:latest") + require.NoError(t, err) + ingester, ok := pusher.(content.Ingester) + require.True(t, ok) + fetcher := remotes.FetcherFunc(func(context.Context, ocispec.Descriptor) (io.ReadCloser, error) { + return os.Open(path) + }) + require.NoError(t, remotes.Fetch(ctx, ingester, fetcher, desc)) + require.EqualValues(t, 2, attempts.Load()) + }) + } +} + func TestGetManifestPath(t *testing.T) { for _, tc := range []struct { object string diff --git a/core/remotes/handlers.go b/core/remotes/handlers.go index 4214672ae1..b01e189d72 100644 --- a/core/remotes/handlers.go +++ b/core/remotes/handlers.go @@ -173,6 +173,9 @@ func Fetch(ctx context.Context, ingester content.Ingester, fetcher Fetcher, desc if err != nil { return err } + + // Keep the reader usable until Copy finishes: a writer may request a + // reset after EOF, including while committing the content. defer rc.Close() return content.Copy(ctx, cw, rc, desc.Size, desc.Digest)