gantry: resume containerd origin range retries

designs/gantry-128gib-single-layer-pulls.md

@@ -110,9 +110,9 @@ resumable.
 
 | ID | Issue | Evidence | Owning PR | Status |
 |---|---|---|---|---|
-| L4 | Gantry origin pulls do not send `Range` and cannot request an origin body from an offset. | `internal/gantry/origin/origin.go:25`, `internal/gantry/origin/origin.go:477-542` | PR 4 | Open |
+| L4 | Gantry origin pulls do not send `Range` and cannot request an origin body from an offset. | `internal/gantry/origin/origin.go:25`, `internal/gantry/origin/origin.go:477-542` | PR 4 | Addressed |
 | L5 | A failed background ingest is aborted. A later writer with a stale nonzero offset is also aborted because callers restart at byte zero. | `internal/gantry/containerdstore/store.go:328-383`, `internal/gantry/containerdstore/store.go:502-519` | PR 5 | Open |
-| L6 | Gantry does not handle the local containerd client's inbound `Range` header on the origin path. | `internal/gantry/mirror/mirror.go:685-930` | PR 4 | Open |
+| L6 | Gantry does not handle the local containerd client's inbound `Range` header on the origin path. | `internal/gantry/mirror/mirror.go:685-930` | PR 4 | Addressed |
 
 L4-L6 do not cause the fixed five- or 30-minute failures. They determine
 whether a transfer interrupted for another reason can continue without
@@ -289,13 +289,13 @@ PR reports.
 **Purpose:** Avoid replaying the origin prefix when local containerd retries a
 live Gantry response at a nonzero offset.
 
-- [ ] Validate a single inbound blob range from local containerd.
-- [ ] Carry the offset through `ifaces.OriginRef` into the origin request.
-- [ ] Require and validate the origin's `206` and `Content-Range` response.
-- [ ] Return matching range semantics to containerd.
-- [ ] Preserve request-scoped delegated registry authorization on the origin
+- [x] Validate a single inbound blob range from local containerd.
+- [x] Carry the offset through `ifaces.OriginRef` into the origin request.
+- [x] Require and validate the origin's `206` and `Content-Range` response.
+- [x] Return matching range semantics to containerd.
+- [x] Preserve request-scoped delegated registry authorization on the origin
   retry and never forward it to a peer.
-- [ ] Test supported ranges, ignored ranges, malformed ranges, and a mid-body
+- [x] Test supported ranges, ignored ranges, malformed ranges, and an offset
   interruption followed by an offset retry.
 
 **Addresses:** L4 and L6.

internal/gantry/ifaces/ifaces.go

@@ -128,7 +128,7 @@ type OriginRef struct {
 	Repository string // e.g. "library/nginx"
 	Digest     digest.Digest
 	// Offset requests bytes starting at this position when fetching from a
-	// peer. Origin registry callers ignore it. Zero requests the full object.
+	// peer or origin registry. Zero requests the full object.
 	Offset int64
 
 	// Kind discriminates the OCI Distribution Spec URL family for this

internal/gantry/mirror/mirror.go

@@ -850,6 +850,15 @@ func (s *Server) serveDigest(w http.ResponseWriter, r *http.Request, upstream, r
 		slog.String("kind", kind.String()),
 	)
 
+	if rangeHeader := r.Header.Get("Range"); rangeHeader != "" {
+		offset, ok := parseOriginRetryRange(rangeHeader)
+		if ok && kind == ifaces.KindBlob {
+			s.serveOriginRange(ctx, w, d, kind, upstream, repo, offset, logger)
+
+			return
+		}
+	}
+
 	// 1. Local content-store lookup.
 	if handled := s.serveLocalHit(ctx, w, r, d, kind, upstream, repo, logger); handled {
 		return
@@ -925,6 +934,103 @@ func (s *Server) serveDigest(w http.ResponseWriter, r *http.Request, upstream, r
 	s.serveFromOrigin(ctx, w, d, kind, upstream, repo, logger)
 }
 
+func parseOriginRetryRange(value string) (int64, bool) {
+	const prefix = "bytes="
+	if !strings.HasPrefix(value, prefix) || strings.Contains(value, ",") {
+		return 0, false
+	}
+
+	spec := strings.TrimPrefix(value, prefix)
+	if !strings.HasSuffix(spec, "-") {
+		return 0, false
+	}
+
+	offset, err := strconv.ParseInt(strings.TrimSuffix(spec, "-"), 10, 64)
+	if err != nil || offset <= 0 {
+		return 0, false
+	}
+
+	return offset, true
+}
+
+func (s *Server) serveOriginRange(ctx context.Context, w http.ResponseWriter, d digest.Digest, kind ifaces.OriginRefKind, upstream, repo string, offset int64, logger *slog.Logger) {
+	pullStartedAt := time.Now()
+	ref := ifaces.OriginRef{Registry: upstream, Repository: repo, Digest: d, Offset: offset, Kind: kind}
+
+	if s.liveStreamThrough {
+		s.fireOriginStreamStarted(kind)
+	}
+
+	rc, totalSize, err := s.origin.Pull(ctx, ref)
+	if err != nil {
+		logger.Debug("mirror: origin range pull failed",
+			slog.String("pull_mode", "live_range"),
+			slog.String("deadline_owner", originDeadlineOwner(ctx, err)),
+			slog.Duration("elapsed", time.Since(pullStartedAt)),
+			slog.Int64("expected_size", -1),
+			slog.Int64("written", 0),
+			slog.Int64("offset", offset),
+			slog.Any("err", err),
+		)
+
+		if s.liveStreamThrough {
+			s.fireOriginStreamFailed(kind)
+		}
+
+		writeOriginError(w, err, logger)
+
+		return
+	}
+
+	defer func() { _ = rc.Close() }() //nolint:errcheck // best-effort close
+
+	remaining := totalSize - offset
+	if remaining <= 0 {
+		if s.liveStreamThrough {
+			s.fireOriginStreamFailed(kind)
+		}
+
+		http.Error(w, "invalid origin range size", http.StatusBadGateway)
+
+		return
+	}
+
+	w.Header().Set("Accept-Ranges", "bytes")
+	w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, totalSize-1, totalSize))
+	w.Header().Set("Content-Length", strconv.FormatInt(remaining, 10))
+	writeBlobHeaders(w, d, -1, kind)
+	w.WriteHeader(http.StatusPartialContent)
+
+	written, copyErr := streamcopy.CopyN(w, rc, remaining)
+	s.fireMirrorBytesServed(kind, "origin", written)
+
+	if copyErr != nil {
+		logger.Debug("mirror: live origin range stream failed",
+			slog.String("pull_mode", "live_range"),
+			slog.String("deadline_owner", originDeadlineOwner(ctx, copyErr)),
+			slog.Duration("elapsed", time.Since(pullStartedAt)),
+			slog.Int64("expected_size", totalSize),
+			slog.Int64("written", written),
+			slog.Int64("offset", offset),
+			slog.Any("err", copyErr),
+		)
+
+		if s.liveStreamThrough {
+			s.fireOriginStreamFailed(kind)
+		}
+
+		return
+	}
+
+	if s.liveStreamThrough {
+		s.fireOriginStreamCompleted(kind)
+	}
+
+	s.fireMirrorResponseCompleted(d, kind, "origin")
+	s.fireLiveStreamCompleted(d)
+	s.recordNegCacheSuccess(d)
+}
+
 // serveLocalHit serves d from the local content store when present.
 // Returns true when the response has been fully written (cache hit or
 // non-NotFound store error -> 5xx); false means the digest is

internal/gantry/mirror/mirror_test.go

@@ -146,6 +146,30 @@ type authorizationCapturingOrigin struct {
 	seen chan string
 }
 
+type rangeOriginRequest struct {
+	offset        int64
+	authorization string
+}
+
+type rangeCapturingOrigin struct {
+	body []byte
+	seen chan rangeOriginRequest
+}
+
+func (o *rangeCapturingOrigin) Pull(ctx context.Context, ref ifaces.OriginRef) (io.ReadCloser, int64, error) {
+	o.seen <- rangeOriginRequest{offset: ref.Offset, authorization: registryauth.Authorization(ctx)}
+
+	if ref.Offset < 0 || ref.Offset >= int64(len(o.body)) {
+		return nil, 0, errors.New("invalid offset")
+	}
+
+	return io.NopCloser(bytes.NewReader(o.body[ref.Offset:])), int64(len(o.body)), nil
+}
+
+func (o *rangeCapturingOrigin) Head(context.Context, ifaces.OriginRef) (int64, string, error) {
+	return int64(len(o.body)), "application/octet-stream", nil
+}
+
 type authorizationRejectingOrigin struct{}
 
 func (authorizationRejectingOrigin) Pull(_ context.Context, ref ifaces.OriginRef) (io.ReadCloser, int64, error) {
@@ -168,6 +192,101 @@ func (o *authorizationCapturingOrigin) Head(ctx context.Context, _ ifaces.Origin
 	return int64(len(o.body)), "application/octet-stream", nil
 }
 
+func TestMirrorOriginRangeRetry(t *testing.T) {
+	body := []byte("0123456789")
+	d := digestOf(body)
+	origin := &rangeCapturingOrigin{body: body, seen: make(chan rangeOriginRequest, 1)}
+	cfg := &config.Config{UpstreamRegistries: []config.UpstreamRegistry{{Name: "reg.example.com", Endpoint: "https://reg.example.com"}}}
+
+	srv := httptest.NewServer(mirror.New(cfg, fakes.NewCache(), origin, mirror.WithLiveStreamThrough()).Handler())
+	defer srv.Close()
+
+	req, err := http.NewRequest(http.MethodGet, srv.URL+"/v2/repo/blobs/"+d.String(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	req.Header.Set("Range", "bytes=4-")
+	req.Header.Set("Authorization", "Bearer requester-token")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusPartialContent {
+		t.Fatalf("status = %d; want 206", resp.StatusCode)
+	}
+
+	if got := resp.Header.Get("Content-Range"); got != "bytes 4-9/10" {
+		t.Fatalf("Content-Range = %q; want bytes 4-9/10", got)
+	}
+
+	got, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if string(got) != string(body[4:]) {
+		t.Fatalf("body = %q; want %q", got, body[4:])
+	}
+
+	seen := <-origin.seen
+	if seen.offset != 4 {
+		t.Fatalf("origin offset = %d; want 4", seen.offset)
+	}
+
+	if seen.authorization != "Bearer requester-token" {
+		t.Fatalf("origin authorization = %q; want requester token", seen.authorization)
+	}
+}
+
+func TestMirrorPreservesFullBodyPathForUnsupportedOriginRange(t *testing.T) {
+	for _, rangeHeader := range []string{"bytes=0-", "bytes=4-6", "invalid"} {
+		t.Run(rangeHeader, func(t *testing.T) {
+			body := []byte("0123456789")
+			d := digestOf(body)
+			origin := &rangeCapturingOrigin{body: body, seen: make(chan rangeOriginRequest, 1)}
+			cfg := &config.Config{UpstreamRegistries: []config.UpstreamRegistry{{Name: "reg.example.com", Endpoint: "https://reg.example.com"}}}
+
+			srv := httptest.NewServer(mirror.New(cfg, fakes.NewCache(), origin).Handler())
+			defer srv.Close()
+
+			req, err := http.NewRequest(http.MethodGet, srv.URL+"/v2/repo/blobs/"+d.String(), nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			req.Header.Set("Range", rangeHeader)
+
+			resp, err := http.DefaultClient.Do(req)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer resp.Body.Close()
+
+			if resp.StatusCode != http.StatusOK {
+				t.Fatalf("status = %d; want 200", resp.StatusCode)
+			}
+
+			got, err := io.ReadAll(resp.Body)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if string(got) != string(body) {
+				t.Fatalf("body = %q; want %q", got, body)
+			}
+
+			seen := <-origin.seen
+			if seen.offset != 0 {
+				t.Fatalf("origin offset = %d; want 0", seen.offset)
+			}
+		})
+	}
+}
+
 func TestMirror_CapturesInboundAuthorizationForOrigin(t *testing.T) {
 	body := []byte("origin bytes")
 	d := digestOf(body)