feat: load app tree from the store first

cmd/composectl/cmd/check.go

@@ -299,7 +299,7 @@ func getAppStoreAndDstBlobProvider(srcStorePath string, local bool) (srcBlobProv
 		// to Registry. Requires app manifest and app archive presence in the local store, otherwise fails.
 		srcBlobProvider = store
 	} else {
-		srcBlobProvider = compose.NewRemoteBlobProviderFromConfig(config)
+		srcBlobProvider = compose.NewStoreFirstBlobProviderFromConfig(config)
 	}
 	return
 }

pkg/compose/provider.go

@@ -37,6 +37,16 @@ type (
 	memoryBlobProvider struct {
 		blobs map[digest.Digest][]byte
 	}
+	// storeFirstBlobProvider reads blobs from the store and the ones missing or
+	// invalid there from the registry. It reports the remote type so a tree loaded
+	// through it has the same nodes as one loaded from the registry. It serves only
+	// the blobs under the store's blobs/sha256 directory: the manifest and bundle of
+	// an app pulled by skopeo or aklite are kept under apps/ and are still read from
+	// the registry.
+	storeFirstBlobProvider struct {
+		blobRoot string
+		remote   BlobProvider
+	}
 	readCloserWrapper struct {
 		reader io.Reader
 		closer io.Closer
@@ -115,6 +125,13 @@ func newRemoteBlobProvider(resolver remotes.Resolver) BlobProvider {
 	}
 }
 
+func NewStoreFirstBlobProviderFromConfig(config *Config) BlobProvider {
+	return &storeFirstBlobProvider{
+		blobRoot: GetBlobsRootFor(config.StoreRoot),
+		remote:   NewRemoteBlobProviderFromConfig(config),
+	}
+}
+
 func NewMemoryBlobProvider(blobs map[digest.Digest][]byte) BlobProvider {
 	return &memoryBlobProvider{
 		blobs: blobs,
@@ -272,6 +289,87 @@ func (p *memoryBlobProvider) Info(ctx context.Context, dgst digest.Digest) (cont
 	return content.Info{}, fmt.Errorf("not implemented")
 }
 
+func (p *storeFirstBlobProvider) Type() BlobProviderType {
+	return BlobProviderTypeRemote
+}
+
+// GetReadCloser serves a stored blob only after reading it once to check its size
+// and digest: a reader cannot switch to the registry once it has handed out bytes.
+func (p *storeFirstBlobProvider) GetReadCloser(ctx context.Context, opts ...SecureReadOptions) (io.ReadCloser, error) {
+	rc, storeErr := p.openVerified(ctx, opts...)
+	if storeErr == nil {
+		return rc, nil
+	}
+	if ctxErr := ctx.Err(); ctxErr != nil {
+		return nil, ctxErr
+	}
+	rc, err := p.remote.GetReadCloser(ctx, opts...)
+	if err != nil {
+		// The remote error stays the wrapped one so the callers classify it as before.
+		return nil, fmt.Errorf("%w (store copy: %v)", err, storeErr)
+	}
+	return rc, nil
+}
+
+// openVerified opens the stored blob, checks its size and digest, and returns a
+// verifying reader over the same file rewound to the start, so the blob served is
+// the one checked even if its path is replaced in the meantime.
+func (p *storeFirstBlobProvider) openVerified(ctx context.Context, opts ...SecureReadOptions) (io.ReadCloser, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	params := GetSecureReadParams(opts...)
+	d := params.ExpectedDigest
+	if len(d) == 0 {
+		if s, err := reference.Parse(params.Ref); err == nil {
+			d = s.Digest()
+		}
+	}
+	if err := d.Validate(); err != nil {
+		return nil, err
+	}
+	opts = append(opts[:len(opts):len(opts)], WithExpectedDigest(d))
+	f, err := os.Open(path.Join(p.blobRoot, d.Encoded()))
+	if err != nil {
+		return nil, err
+	}
+	r, err := NewSecureReadCloser(io.NopCloser(f), opts...)
+	if err == nil {
+		err = discardAll(ctx, r)
+	}
+	if err == nil {
+		_, err = f.Seek(0, io.SeekStart)
+	}
+	if err == nil {
+		r, err = NewSecureReadCloser(f, opts...)
+	}
+	if err != nil {
+		f.Close()
+		return nil, err
+	}
+	return r, nil
+}
+
+func (p *storeFirstBlobProvider) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) {
+	return p.remote.Info(ctx, dgst)
+}
+
+// discardAll reads r to its end in 1 MiB steps and stops as soon as ctx is done,
+// so a cancelled request does not hash a large blob to the end.
+func discardAll(ctx context.Context, r io.Reader) error {
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		if _, err := io.CopyN(io.Discard, r, 1<<20); err != nil {
+			if err == io.EOF {
+				return ctx.Err()
+			}
+			return err
+		}
+	}
+}
+
 func (w *readCloserWrapper) Read(p []byte) (n int, err error) {
 	return w.reader.Read(p)
 }

pkg/compose/provider_test.go

@@ -0,0 +1,123 @@
+package compose
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"os"
+	"path"
+	"strings"
+	"testing"
+
+	"github.com/opencontainers/go-digest"
+)
+
+func TestStoreFirstBlobProvider(t *testing.T) {
+	storeRoot := t.TempDir()
+	blobs := map[digest.Digest][]byte{}
+	newBlob := func(data string, stored []byte) digest.Digest {
+		d := digest.FromString(data)
+		blobs[d] = []byte(data)
+		if stored != nil {
+			if err := os.WriteFile(path.Join(storeRoot, d.Encoded()), stored, 0644); err != nil {
+				t.Fatal(err)
+			}
+		}
+		return d
+	}
+	// The remote serves every blob; the store holds a valid, a corrupt, a truncated
+	// and no copy of them, so a blob read from the remote proves the fallback.
+	storedOk := newBlob("stored ok", []byte("stored ok"))
+	corrupt := newBlob("corrupt", []byte("CORRUPT"))
+	truncated := newBlob("truncated", []byte("trunc"))
+	missing := newBlob("missing", nil)
+	p := &storeFirstBlobProvider{blobRoot: storeRoot, remote: NewMemoryBlobProvider(blobs)}
+	if p.Type() != BlobProviderTypeRemote {
+		t.Fatalf("expected type %s, got %s", BlobProviderTypeRemote, p.Type())
+	}
+
+	read := func(t *testing.T, opts ...SecureReadOptions) []byte {
+		t.Helper()
+		rc, err := p.GetReadCloser(context.Background(), opts...)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer rc.Close()
+		b, err := io.ReadAll(rc)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return b
+	}
+
+	for name, d := range map[string]digest.Digest{
+		"valid in the store":     storedOk,
+		"corrupt in the store":   corrupt,
+		"truncated in the store": truncated,
+		"missing in the store":   missing,
+	} {
+		t.Run(name, func(t *testing.T) {
+			got := read(t, WithExpectedDigest(d), WithExpectedSize(int64(len(blobs[d]))))
+			if !bytes.Equal(got, blobs[d]) {
+				t.Fatalf("expected %q, got %q", blobs[d], got)
+			}
+		})
+	}
+
+	t.Run("served from the store", func(t *testing.T) {
+		// Remove the remote copy so only the store can serve the blob.
+		delete(blobs, storedOk)
+		defer func() { blobs[storedOk] = []byte("stored ok") }()
+		got := read(t, WithExpectedDigest(storedOk), WithExpectedSize(int64(len("stored ok"))))
+		if string(got) != "stored ok" {
+			t.Fatalf("expected %q, got %q", "stored ok", got)
+		}
+	})
+
+	t.Run("digest taken from the ref", func(t *testing.T) {
+		delete(blobs, storedOk)
+		defer func() { blobs[storedOk] = []byte("stored ok") }()
+		got := read(t, WithRef("registry:5000/factory/app@"+storedOk.String()), WithExpectedSize(int64(len("stored ok"))))
+		if string(got) != "stored ok" {
+			t.Fatalf("expected %q, got %q", "stored ok", got)
+		}
+	})
+
+	t.Run("missing everywhere", func(t *testing.T) {
+		_, err := p.GetReadCloser(context.Background(), WithExpectedDigest(digest.FromString("absent")), WithExpectedSize(6))
+		if err == nil {
+			t.Fatal("expected an error for a blob missing in the store and the registry")
+		}
+		if ErrToBlobState(err) != BlobMissing {
+			t.Fatalf("expected the error to classify as a missing blob, got: %s", err)
+		}
+	})
+
+	t.Run("both reasons reported", func(t *testing.T) {
+		// Remove the remote copy of the corrupt blob so both reads fail.
+		delete(blobs, corrupt)
+		defer func() { blobs[corrupt] = []byte("corrupt") }()
+		_, err := p.GetReadCloser(context.Background(), WithExpectedDigest(corrupt), WithExpectedSize(int64(len("corrupt"))))
+		if err == nil {
+			t.Fatal("expected an error for a blob corrupt in the store and missing in the registry")
+		}
+		for _, reason := range []string{"not found", "blob digest mismatch"} {
+			if !strings.Contains(err.Error(), reason) {
+				t.Fatalf("expected the error to mention %q, got: %s", reason, err)
+			}
+		}
+	})
+
+	t.Run("cancelled context", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		rc, err := p.GetReadCloser(ctx, WithExpectedDigest(storedOk), WithExpectedSize(int64(len("stored ok"))))
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("expected %v, got: %v", context.Canceled, err)
+		}
+		if rc != nil {
+			t.Fatal("expected no reader for a cancelled context")
+		}
+	})
+}

pkg/update/init.go

@@ -65,7 +65,7 @@ func (u *runnerImpl) initUpdate(ctx context.Context, b *session, options ...Init
 		}
 	}()
 
-	srcBlobProvider := compose.NewRemoteBlobProviderFromConfig(u.config)
+	srcBlobProvider := compose.NewStoreFirstBlobProviderFromConfig(u.config)
 
 	p := InitProgress{
 		State:   UpdateInitStateLoadingTree,

test/fixtures/composectl_cmds.go

@@ -482,6 +482,18 @@ func RunCmdExpectFail(t *testing.T, appDir string, args ...string) []byte {
 	return output
 }
 
+// RunCmdOffline runs composectl with the given args through an unreachable proxy, so every
+// registry request fails. It returns the combined output and the command's error.
+func RunCmdOffline(t *testing.T, args ...string) ([]byte, error) {
+	t.Helper()
+	c := exec.Command(composeExec, args...)
+	c.Env = append(os.Environ(),
+		"HTTP_PROXY=http://127.0.0.1:1", "HTTPS_PROXY=http://127.0.0.1:1",
+		"http_proxy=http://127.0.0.1:1", "https_proxy=http://127.0.0.1:1",
+		"NO_PROXY=", "no_proxy=")
+	return c.CombinedOutput()
+}
+
 func randomString(length int) string {
 	const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
 	seededRand := rand2.New(rand2.NewSource(time.Now().UnixNano()))