From fe9204fdd8d678b40b5db2ecbfe2d13b49b40ce2 Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Fri, 23 Oct 2026 14:11:56 +0200 Subject: [PATCH] feat: load app tree from the store first check, pull and update init load an app's metadata from the registry even when every metadata blob is already in the store, so they cost registry round trips and fail offline for a fully fetched app. Load the tree through a provider that serves a blob from the store once a first read has verified its size and digest, and falls back to the registry if it is missing or invalid there, so pull still repairs a corrupt blob. It reports the remote type so the loaded tree keeps the same nodes as one loaded from the registry. Fixes: #101 Assisted-by: Claude Code:claude-fable-5-1 Signed-off-by: Mike Sul --- cmd/composectl/cmd/check.go | 2 +- pkg/compose/provider.go | 98 ++++++++++++++++++++ pkg/compose/provider_test.go | 123 +++++++++++++++++++++++++ pkg/update/init.go | 2 +- test/fixtures/composectl_cmds.go | 12 +++ test/integration/edge_cases_test.go | 137 ++++++++++++++++++++++++++++ test/integration/update_test.go | 39 ++++++++ 7 files changed, 411 insertions(+), 2 deletions(-) create mode 100644 pkg/compose/provider_test.go diff --git a/cmd/composectl/cmd/check.go b/cmd/composectl/cmd/check.go index 1f1a96a4..58e90641 100644 --- a/cmd/composectl/cmd/check.go +++ b/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 } diff --git a/pkg/compose/provider.go b/pkg/compose/provider.go index 7f52e3e6..950a294e 100644 --- a/pkg/compose/provider.go +++ b/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) } diff --git a/pkg/compose/provider_test.go b/pkg/compose/provider_test.go new file mode 100644 index 00000000..097f1879 --- /dev/null +++ b/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") + } + }) +} diff --git a/pkg/update/init.go b/pkg/update/init.go index f10a2f7b..03ef2ea8 100644 --- a/pkg/update/init.go +++ b/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, diff --git a/test/fixtures/composectl_cmds.go b/test/fixtures/composectl_cmds.go index cda6f450..4cd49d4e 100644 --- a/test/fixtures/composectl_cmds.go +++ b/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())) diff --git a/test/integration/edge_cases_test.go b/test/integration/edge_cases_test.go index cb331011..2e338730 100644 --- a/test/integration/edge_cases_test.go +++ b/test/integration/edge_cases_test.go @@ -1,7 +1,14 @@ package e2e_tests import ( + "bytes" + "encoding/json" + composectl "github.com/foundriesio/composeapp/cmd/composectl/cmd" + "github.com/foundriesio/composeapp/pkg/compose" + v1 "github.com/foundriesio/composeapp/pkg/compose/v1" f "github.com/foundriesio/composeapp/test/fixtures" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" "os" "path" "strings" @@ -203,3 +210,133 @@ services: break } } + +func TestCheckAndPullOffline(t *testing.T) { + appComposeDef := ` +services: + srvs-01: + image: registry:5000/factory/runner-image:v0.1 + command: sh -c "while true; do sleep 60; done" +` + app := f.NewApp(t, appComposeDef) + app.Publish(t) + + if output, err := f.RunCmdOffline(t, "check", app.PublishedUri); err == nil { + t.Fatalf("expected check of a not fetched app to fail offline; output: %s", output) + } + + app.Pull(t) + defer app.Remove(t) + + output, err := f.RunCmdOffline(t, "check", app.PublishedUri, "--format", "json") + f.Checkf(t, err, "check of a fetched app failed offline: %s\n", output) + checkResult := composectl.CheckAndInstallResult{} + f.Check(t, json.Unmarshal(output, &checkResult)) + if len(checkResult.FetchCheck.MissingBlobs) > 0 { + t.Errorf("there are missing app blobs: %+v\n", checkResult.FetchCheck.MissingBlobs) + } + + output, err = f.RunCmdOffline(t, "pull", app.PublishedUri) + f.Checkf(t, err, "pull of a fetched app failed offline: %s\n", output) +} + +func TestCheckAndPullRepairCorruptManifest(t *testing.T) { + appComposeDef := ` +services: + srvs-01: + image: registry:5000/factory/runner-image:v0.1 + command: sh -c "while true; do sleep 60; done" +` + app := f.NewApp(t, appComposeDef) + app.Publish(t) + + app.Pull(t) + defer app.Remove(t) + + // Overwrite the stored app manifest with bytes of the same size but another digest. + appRef, err := compose.ParseAppRef(app.PublishedUri) + f.Check(t, err) + manifestPath := path.Join(compose.GetBlobsRootFor(f.AppStoreRoot), appRef.Digest.Encoded()) + manifest, err := os.ReadFile(manifestPath) + f.Check(t, err) + f.Check(t, os.WriteFile(manifestPath, bytes.Repeat([]byte("x"), len(manifest)), 0644)) + + checkRepairedBlob(t, app, appRef.Digest) +} + +// readStoredAppManifest returns the app manifest as stored in the app store. +func readStoredAppManifest(t *testing.T, app *f.App) ocispec.Manifest { + t.Helper() + appRef, err := compose.ParseAppRef(app.PublishedUri) + f.Check(t, err) + b, err := os.ReadFile(path.Join(compose.GetBlobsRootFor(f.AppStoreRoot), appRef.Digest.Encoded())) + f.Check(t, err) + manifest := ocispec.Manifest{} + f.Check(t, json.Unmarshal(b, &manifest)) + return manifest +} + +// checkRepairedBlob checks that `check` lists the given blob as missing, `pull` repairs it, +// and `check` of the repaired app then passes offline. +func checkRepairedBlob(t *testing.T, app *f.App, dgst digest.Digest) { + t.Helper() + output := f.RunCmd(t, "check", app.PublishedUri, "--format", "json") + checkResult := composectl.CheckAndInstallResult{} + f.Check(t, json.Unmarshal(output, &checkResult)) + if _, ok := checkResult.FetchCheck.MissingBlobs[dgst]; !ok { + t.Fatalf("expected %s among missing blobs, got: %+v\n", dgst, checkResult.FetchCheck.MissingBlobs) + } + + app.Pull(t) + app.CheckFetched(t) + output, err := f.RunCmdOffline(t, "check", app.PublishedUri, "--format", "json") + f.Checkf(t, err, "check of a repaired app failed offline: %s\n", output) + checkResult = composectl.CheckAndInstallResult{} + f.Check(t, json.Unmarshal(output, &checkResult)) + if len(checkResult.FetchCheck.MissingBlobs) > 0 { + t.Fatalf("there are missing app blobs: %+v\n", checkResult.FetchCheck.MissingBlobs) + } +} + +func TestCheckAndPullRepairMissingBundleIndex(t *testing.T) { + appComposeDef := ` +services: + srvs-01: + image: registry:5000/factory/runner-image:v0.1 + command: sh -c "while true; do sleep 60; done" +` + app := f.NewApp(t, appComposeDef) + app.Publish(t) + + app.Pull(t) + defer app.Remove(t) + + // Remove the stored app bundle index, which the bundle descriptor refers to. + manifest := readStoredAppManifest(t, app) + indexDigest, err := digest.Parse(manifest.Layers[0].Annotations[v1.AnnotationKeyAppBundleIndexDigest]) + f.Check(t, err) + f.Check(t, os.Remove(path.Join(compose.GetBlobsRootFor(f.AppStoreRoot), indexDigest.Encoded()))) + + checkRepairedBlob(t, app, indexDigest) +} + +func TestCheckAndPullRepairCorruptBundle(t *testing.T) { + appComposeDef := ` +services: + srvs-01: + image: registry:5000/factory/runner-image:v0.1 + command: sh -c "while true; do sleep 60; done" +` + app := f.NewApp(t, appComposeDef) + app.Publish(t) + + app.Pull(t) + defer app.Remove(t) + + // Overwrite the stored app bundle with bytes of the same size but another digest. + bundleDesc := readStoredAppManifest(t, app).Layers[0] + bundlePath := path.Join(compose.GetBlobsRootFor(f.AppStoreRoot), bundleDesc.Digest.Encoded()) + f.Check(t, os.WriteFile(bundlePath, bytes.Repeat([]byte("x"), int(bundleDesc.Size)), 0644)) + + checkRepairedBlob(t, app, bundleDesc.Digest) +} diff --git a/test/integration/update_test.go b/test/integration/update_test.go index 854becd2..7f4c0fca 100644 --- a/test/integration/update_test.go +++ b/test/integration/update_test.go @@ -2,10 +2,12 @@ package e2e_tests import ( "context" + "net/url" "testing" "github.com/docker/docker/api/types" "github.com/foundriesio/composeapp/pkg/compose" + v1 "github.com/foundriesio/composeapp/pkg/compose/v1" "github.com/foundriesio/composeapp/pkg/update" f "github.com/foundriesio/composeapp/test/fixtures" ) @@ -91,6 +93,43 @@ services: defer finalizeUpdate(t, ctx, updateRunner) } +func TestAppUpdateInitOffline(t *testing.T) { + appComposeDef := ` +services: + srvs-01: + image: registry:5000/factory/runner-image:v0.1 + command: sh -c "while true; do sleep 60; done" +` + app := f.NewApp(t, appComposeDef) + app.Publish(t) + + // Route every registry request of the config through an unreachable proxy. + proxyURL, err := url.Parse("http://127.0.0.1:1") + f.Check(t, err) + cfg, err := v1.NewDefaultConfig(v1.WithStoreRoot(f.AppStoreRoot), v1.WithComposeRoot(f.AppComposeRootRoot), + v1.WithProxy(func() *compose.ProxyConfig { return &compose.ProxyConfig{ProxyURL: proxyURL} })) + f.Check(t, err) + ctx := context.Background() + + updateRunner, err := update.NewUpdate(cfg, "target-1") + f.Check(t, err) + if err := updateRunner.Init(ctx, []string{app.PublishedUri}); err == nil { + t.Fatal("expected init of a not fetched app to fail offline") + } + + app.Pull(t) + defer app.Remove(t) + + updateRunner, err = update.NewUpdate(cfg, "target-2") + f.Check(t, err) + defer finalizeUpdate(t, ctx, updateRunner) + err = updateRunner.Init(ctx, []string{app.PublishedUri}) + f.Checkf(t, err, "init of a fetched app failed offline: %v\n", err) + if updateStatus := updateRunner.Status(); updateStatus.State != update.StateInitialized { + t.Fatalf("update not initialized: %s\n", updateStatus.State) + } +} + func TestAppSync(t *testing.T) { appComposeDef := ` services: