gantry: resume detached origin ingests

cmd/gantry/main.go

@@ -1568,6 +1568,26 @@ type preIngestLeaseStore interface {
 	CreateLease(ctx context.Context, d digest.Digest, registry, repository string) (*containerdstore.LeaseGuard, error)
 }
 
+type resumableOriginStore interface {
+	ResumeWriter(ctx context.Context, d digest.Digest) (ifaces.ContentWriter, int64, error)
+}
+
+type preservableOriginWriter interface {
+	Preserve() error
+}
+
+func openOriginWriter(ctx context.Context, store ifaces.LocalContentStore, d digest.Digest, kind ifaces.OriginRefKind) (ifaces.ContentWriter, int64, error) {
+	if kind == ifaces.KindBlob {
+		if resumable, ok := store.(resumableOriginStore); ok {
+			return resumable.ResumeWriter(ctx, d)
+		}
+	}
+
+	w, err := store.Writer(ctx, d)
+
+	return w, 0, err
+}
+
 type pullerPumpGate struct {
 	mu        sync.Mutex
 	accepting bool
@@ -1852,23 +1872,7 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs
 		Digest:     d,
 		Kind:       kind,
 	}
-
-	rc, expectedSize, err := originClient.Pull(ctx, ref)
-	if err != nil {
-		// A delegated credential is requester-specific. Its origin failure
-		// must not poison the digest-wide cache for another requester.
-		recordOriginFailure(neg, d, err, lg, "origin pull failed", registry, repository, registryauth.Authorization(ctx) == "",
-			slog.String("pull_mode", "detached"),
-			slog.String("deadline_owner", originPullDeadlineOwner(ctx, err)),
-			slog.Duration("elapsed", time.Since(pullStartedAt)),
-			slog.Int64("expected_size", -1),
-			slog.Int64("written", 0),
-		)
-
-		return
-	}
-
-	defer func() { _ = rc.Close() }() //nolint:errcheck // best-effort close
+	expectedSize := int64(-1)
 
 	var leaseGuard *containerdstore.LeaseGuard
 
@@ -1913,7 +1917,7 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs
 		releaseCancel()
 	}
 
-	w, err := cstore.Writer(ctx, d)
+	w, resumeOffset, err := openOriginWriter(ctx, cstore, d, kind)
 	deadlineOwner := originPullDeadlineOwner(ctx, err)
 
 	if err != nil {
@@ -1923,6 +1927,7 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs
 			slog.String("deadline_owner", deadlineOwner),
 			slog.Duration("elapsed", time.Since(pullStartedAt)),
 			slog.Int64("expected_size", expectedSize),
+			slog.Int64("resume_offset", resumeOffset),
 			slog.Int64("written", 0),
 		)
 		// Origin returned 2xx (we got past originClient.Pull above)
@@ -1940,12 +1945,72 @@ func runOriginPull(baseCtx context.Context, originClient ifaces.OriginPuller, cs
 	}
 
 	defer func() {
+		if preservable, ok := w.(preservableOriginWriter); ok {
+			if err := preservable.Preserve(); err != nil {
+				lg.Warn("preserve partial origin ingest failed", slog.Any("err", err))
+			}
+
+			return
+		}
+
 		abortCtx, abortCancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer abortCancel()
 
 		_ = w.Abort(abortCtx) //nolint:errcheck // best-effort abort
 	}()
 
+	ref.Offset = resumeOffset
+
+	rc, expectedSize, err := originClient.Pull(ctx, ref)
+	if err != nil && resumeOffset > 0 {
+		var rangeUnsupported *ifaces.ErrRangeUnsupported
+		if errors.As(err, &rangeUnsupported) {
+			abortCtx, abortCancel := context.WithTimeout(context.Background(), 10*time.Second)
+			abortErr := w.Abort(abortCtx)
+
+			abortCancel()
+
+			if abortErr != nil {
+				err = fmt.Errorf("abort partial ingest before full retry: %w", abortErr)
+			} else {
+				replacement, replacementErr := cstore.Writer(ctx, d)
+
+				err = replacementErr
+				if err == nil {
+					w = replacement
+
+					lg.Info("origin does not support resume; restarting from byte zero",
+						slog.String("digest", d.String()),
+						slog.String("registry", registry),
+						slog.String("repository", repository),
+						slog.Int64("resume_offset", resumeOffset),
+					)
+
+					resumeOffset = 0
+					ref.Offset = 0
+					rc, expectedSize, err = originClient.Pull(ctx, ref)
+				}
+			}
+		}
+	}
+
+	if err != nil {
+		// A delegated credential is requester-specific. Its origin failure
+		// must not poison the digest-wide cache for another requester.
+		recordOriginFailure(neg, d, err, lg, "origin pull failed", registry, repository, registryauth.Authorization(ctx) == "",
+			slog.String("pull_mode", "detached"),
+			slog.String("deadline_owner", originPullDeadlineOwner(ctx, err)),
+			slog.Duration("elapsed", time.Since(pullStartedAt)),
+			slog.Int64("expected_size", expectedSize),
+			slog.Int64("resume_offset", resumeOffset),
+			slog.Int64("written", 0),
+		)
+
+		return
+	}
+
+	defer func() { _ = rc.Close() }() //nolint:errcheck // best-effort close
+
 	written, err := copyWithOriginProgressTimeout(ctx, cancel, w, rc, progressTimeout)
 	if err != nil {
 		releaseLeaseOnFailure()

cmd/gantry/origin_pull_test.go

@@ -4,10 +4,13 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -184,6 +187,104 @@ func (originTimeoutError) Error() string   { return "timed out" }
 func (originTimeoutError) Timeout() bool   { return true }
 func (originTimeoutError) Temporary() bool { return true }
 
+type offsetRecordingOrigin struct {
+	body        []byte
+	offsets     []int64
+	rejectRange bool
+	fail        bool
+}
+
+func (o *offsetRecordingOrigin) Pull(_ context.Context, ref ifaces.OriginRef) (io.ReadCloser, int64, error) {
+	o.offsets = append(o.offsets, ref.Offset)
+
+	if o.fail {
+		return nil, 0, errors.New("transient origin failure")
+	}
+
+	if ref.Offset > 0 && o.rejectRange {
+		return nil, 0, &ifaces.OriginError{
+			Ref:   ref,
+			Class: ifaces.FailureTransient,
+			Err:   &ifaces.ErrRangeUnsupported{Offset: ref.Offset, Reason: "status 200 OK"},
+		}
+	}
+
+	return io.NopCloser(bytes.NewReader(o.body[ref.Offset:])), int64(len(o.body)), nil
+}
+
+func (o *offsetRecordingOrigin) Head(context.Context, ifaces.OriginRef) (int64, string, error) {
+	return int64(len(o.body)), "application/octet-stream", nil
+}
+
+type resumableTestCache struct {
+	expected  digest.Digest
+	partial   []byte
+	committed []byte
+	aborts    int
+}
+
+func (c *resumableTestCache) Has(context.Context, digest.Digest) (bool, error) {
+	return c.committed != nil, nil
+}
+
+func (c *resumableTestCache) Open(_ context.Context, d digest.Digest) (io.ReadCloser, int64, error) {
+	if c.committed == nil {
+		return nil, 0, &ifaces.ErrNotFound{Digest: d}
+	}
+
+	return io.NopCloser(bytes.NewReader(c.committed)), int64(len(c.committed)), nil
+}
+
+func (c *resumableTestCache) Writer(context.Context, digest.Digest) (ifaces.ContentWriter, error) {
+	return &resumableTestWriter{cache: c}, nil
+}
+
+func (c *resumableTestCache) ResumeWriter(context.Context, digest.Digest) (ifaces.ContentWriter, int64, error) {
+	w := &resumableTestWriter{cache: c}
+	_, _ = w.body.Write(c.partial)
+
+	return w, int64(len(c.partial)), nil
+}
+
+type resumableTestWriter struct {
+	cache     *resumableTestCache
+	body      bytes.Buffer
+	finalized bool
+}
+
+func (w *resumableTestWriter) Write(p []byte) (int, error) { return w.body.Write(p) }
+
+func (w *resumableTestWriter) Commit(context.Context) error {
+	if got := trackerDigestOf(w.body.Bytes()); got != w.cache.expected {
+		return fmt.Errorf("digest = %s; want %s", got, w.cache.expected)
+	}
+
+	w.cache.committed = append([]byte(nil), w.body.Bytes()...)
+	w.cache.partial = nil
+	w.finalized = true
+
+	return nil
+}
+
+func (w *resumableTestWriter) Abort(context.Context) error {
+	if !w.finalized {
+		w.cache.partial = nil
+		w.cache.aborts++
+		w.finalized = true
+	}
+
+	return nil
+}
+
+func (w *resumableTestWriter) Preserve() error {
+	if !w.finalized {
+		w.cache.partial = append([]byte(nil), w.body.Bytes()...)
+		w.finalized = true
+	}
+
+	return nil
+}
+
 func (r *pacedReader) Read(p []byte) (int, error) {
 	if r.remaining == 0 {
 		return 0, io.EOF
@@ -288,6 +389,88 @@ func TestOriginPullDeadlineOwner(t *testing.T) {
 	}
 }
 
+func TestRunOriginPullResumesPartialIngest(t *testing.T) {
+	body := []byte("partial-then-completed")
+	d := trackerDigestOf(body)
+	originPuller := &offsetRecordingOrigin{body: body}
+	cache := &resumableTestCache{expected: d, partial: append([]byte(nil), body[:8]...)}
+	h, _, _ := inflight.New(inflight.DefaultStalls(), nil).Start(d, ifaces.KindBlob, 0)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	var successes int
+
+	runOriginPull(context.Background(), originPuller, cache, nil, logger, h, "registry.example.com", "library/test", d, ifaces.KindBlob, 0,
+		func(context.Context, digest.Digest) bool { return true },
+		func(string, int64) { successes++ },
+		func(string, string) {},
+		leaseMetricHooks{},
+	)
+
+	if !slices.Equal(originPuller.offsets, []int64{8}) {
+		t.Fatalf("origin offsets = %v; want [8]", originPuller.offsets)
+	}
+
+	if !bytes.Equal(cache.committed, body) {
+		t.Fatalf("committed body = %q; want %q", cache.committed, body)
+	}
+
+	if successes != 1 {
+		t.Fatalf("successes = %d; want 1", successes)
+	}
+}
+
+func TestRunOriginPullRestartsWhenRangeUnsupported(t *testing.T) {
+	body := []byte("partial-restarted-from-zero")
+	d := trackerDigestOf(body)
+	originPuller := &offsetRecordingOrigin{body: body, rejectRange: true}
+	cache := &resumableTestCache{expected: d, partial: append([]byte(nil), body[:8]...)}
+	h, _, _ := inflight.New(inflight.DefaultStalls(), nil).Start(d, ifaces.KindBlob, 0)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	runOriginPull(context.Background(), originPuller, cache, nil, logger, h, "registry.example.com", "library/test", d, ifaces.KindBlob, 0,
+		func(context.Context, digest.Digest) bool { return true },
+		func(string, int64) {},
+		func(string, string) {},
+		leaseMetricHooks{},
+	)
+
+	if !slices.Equal(originPuller.offsets, []int64{8, 0}) {
+		t.Fatalf("origin offsets = %v; want [8 0]", originPuller.offsets)
+	}
+
+	if cache.aborts != 1 {
+		t.Fatalf("aborts = %d; want 1", cache.aborts)
+	}
+
+	if !bytes.Equal(cache.committed, body) {
+		t.Fatalf("committed body = %q; want %q", cache.committed, body)
+	}
+}
+
+func TestRunOriginPullPreservesPartialOnTransientFailure(t *testing.T) {
+	body := []byte("partial-preserved")
+	d := trackerDigestOf(body)
+	originPuller := &offsetRecordingOrigin{body: body, fail: true}
+	cache := &resumableTestCache{expected: d, partial: append([]byte(nil), body[:8]...)}
+	h, _, _ := inflight.New(inflight.DefaultStalls(), nil).Start(d, ifaces.KindBlob, 0)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	runOriginPull(context.Background(), originPuller, cache, nil, logger, h, "registry.example.com", "library/test", d, ifaces.KindBlob, 0,
+		func(context.Context, digest.Digest) bool { return true },
+		func(string, int64) {},
+		func(string, string) {},
+		leaseMetricHooks{},
+	)
+
+	if !bytes.Equal(cache.partial, body[:8]) {
+		t.Fatalf("partial body = %q; want %q", cache.partial, body[:8])
+	}
+
+	if cache.aborts != 0 {
+		t.Fatalf("aborts = %d; want 0", cache.aborts)
+	}
+}
+
 func TestRunOriginPull_ReopenFailurePreventsAdvertiseAndSuccess(t *testing.T) {
 	body := []byte("committed-but-not-reopenable")
 	d := trackerDigestOf(body)

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

@@ -111,7 +111,7 @@ 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 | 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 |
+| 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 | Addressed |
 | 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
@@ -310,14 +310,14 @@ resume failures easier to distinguish during review.
 **Purpose:** Reuse bytes already staged in containerd when a detached chair
 pull is interrupted.
 
-- [ ] Expose a verified writer offset through the local content-store
+- [x] Expose a verified writer offset through an optional local content-store
   abstraction.
-- [ ] Reopen the origin at that offset using PR 4's origin range support.
-- [ ] Continue the existing containerd ingest and rely on commit-time digest
+- [x] Reopen the origin at that offset using PR 4's origin range support.
+- [x] Continue the existing containerd ingest and rely on commit-time digest
   verification over the complete layer.
-- [ ] Abort and restart from byte zero when the origin cannot honor the range,
+- [x] Abort and restart from byte zero when the origin cannot honor the range,
   and log that decision.
-- [ ] Test process-local retry, stale partial state, unsupported ranges, digest
+- [x] Test process-local retry, stale partial state, unsupported ranges, digest
   mismatch, and delegated authorization.
 
 **Addresses:** L5.