feat(transport): add resumable transport for remote resources

cmd/crane/cmd/pull.go

@@ -32,6 +32,7 @@ func NewCmdPull(options *[]crane.Option) *cobra.Command {
 	var (
 		cachePath, format string
 		annotateRef       bool
+		resumable         bool
 	)
 
 	cmd := &cobra.Command{
@@ -49,6 +50,10 @@ func NewCmdPull(options *[]crane.Option) *cobra.Command {
 					return fmt.Errorf("parsing reference %q: %w", src, err)
 				}
 
+				if resumable {
+					o.Remote = append(o.Remote, remote.WithResumable())
+				}
+
 				rmt, err := remote.Get(ref, o.Remote...)
 				if err != nil {
 					return err
@@ -133,6 +138,7 @@ func NewCmdPull(options *[]crane.Option) *cobra.Command {
 	cmd.Flags().StringVarP(&cachePath, "cache_path", "c", "", "Path to cache image layers")
 	cmd.Flags().StringVar(&format, "format", "tarball", fmt.Sprintf("Format in which to save images (%q, %q, or %q)", "tarball", "legacy", "oci"))
 	cmd.Flags().BoolVar(&annotateRef, "annotate-ref", false, "Preserves image reference used to pull as an annotation when used with --format=oci")
+	cmd.Flags().BoolVar(&resumable, "resumable", false, "Enable resumable transport for pulling images")
 
 	return cmd
 }

pkg/v1/remote/image_test.go

@@ -17,6 +17,8 @@ package remote
 import (
 	"bytes"
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -747,3 +749,45 @@ func TestData(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestImageResumable(t *testing.T) {
+	ref, err := name.ParseReference("ghcr.io/labring/fastgpt:v4.9.0")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	image, err := Image(ref, WithResumable())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	layers, err := image.Layers()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, layer := range layers {
+		digest, err := layer.Digest()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		rc, err := layer.Compressed()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		hash := sha256.New()
+		_, err = io.Copy(hash, rc)
+		rc.Close()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if digest.Hex == hex.EncodeToString(hash.Sum(nil)) {
+			t.Logf("digest matches: %s", digest)
+		} else {
+			t.Errorf("digest mismatch: %s != %s", digest, hex.EncodeToString(hash.Sum(nil)))
+		}
+	}
+}

pkg/v1/remote/options.go

@@ -45,6 +45,8 @@ type options struct {
 	retryBackoff                   Backoff
 	retryPredicate                 retry.Predicate
 	retryStatusCodes               []int
+	resumable                      bool
+	resumableBackoff               Backoff
 
 	// Only these options can overwrite Reuse()d options.
 	platform v1.Platform
@@ -135,6 +137,7 @@ func makeOptions(opts ...Option) (*options, error) {
 		retryPredicate:   defaultRetryPredicate,
 		retryBackoff:     defaultRetryBackoff,
 		retryStatusCodes: defaultRetryStatusCodes,
+		resumableBackoff: defaultRetryBackoff,
 	}
 
 	for _, option := range opts {
@@ -170,6 +173,11 @@ func makeOptions(opts ...Option) (*options, error) {
 
 		// Wrap the transport in something that can retry network flakes.
 		o.transport = transport.NewRetry(o.transport, transport.WithRetryBackoff(o.retryBackoff), transport.WithRetryPredicate(predicate), transport.WithRetryStatusCodes(o.retryStatusCodes...))
+
+		if o.resumable {
+			o.transport = transport.NewResumable(o.transport, o.resumableBackoff)
+		}
+
 		// Wrap this last to prevent transport.New from double-wrapping.
 		if o.userAgent != "" {
 			o.transport = transport.NewUserAgent(o.transport, o.userAgent)
@@ -192,6 +200,25 @@ func WithTransport(t http.RoundTripper) Option {
 	}
 }
 
+// WithResumable is a functional option for enabling resumable downloads. and it will wrap retry transport by default.
+// If configures retry and resumable backoff, should be aware of all backoff will be applied.
+func WithResumable() Option {
+	return func(o *options) error {
+		o.resumable = true
+		return nil
+	}
+}
+
+// WithResumableBackoff is a functional option for overriding the default resumable backoff for remote operations.
+// Resumable backoff will resume failed requests after a delay, unlike retry actions, resumable backoff will ignore
+// transport.RoundTripper.RoundTrip errors.
+func WithResumableBackoff(backoff Backoff) Option {
+	return func(o *options) error {
+		o.resumableBackoff = backoff
+		return nil
+	}
+}
+
 // WithAuth is a functional option for overriding the default authenticator
 // for remote operations.
 // It is an error to use both WithAuth and WithAuthFromKeychain in the same Option set.