From 30ecbb0158f75d67c3989731e7178c5616d3921c Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Wed, 21 Oct 2026 14:55:36 +0200 Subject: [PATCH 1/2] docs(design): record the link-in of the install layers The containerd install copied every compressed layer from the store into containerd's content store before unpacking it, a transient second copy that doubled the bytes written by an install. The note says how a layer is checked, hardlinked and registered with no bytes copied, and that the store and the content store must be on one filesystem. Assisted-by: Claude Code:claude-fable-5-1 Signed-off-by: Mike Sul --- ...install-linkin-progress-plan-2026-10-19.md | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) create mode 100644 docs/containerd-install-linkin-progress-plan-2026-10-19.md diff --git a/docs/containerd-install-linkin-progress-plan-2026-10-19.md b/docs/containerd-install-linkin-progress-plan-2026-10-19.md new file mode 100644 index 0000000..4be67c2 --- /dev/null +++ b/docs/containerd-install-linkin-progress-plan-2026-10-19.md @@ -0,0 +1,26 @@ +# Install into containerd: link the layers in + +The containerd install (`containerd-install-plan-2026-10-08.md`) no longer copies the +compressed layers from composectl's store into containerd's content store. Each layer +is hardlinked instead, so an install writes nothing but the snapshots. + +## How + +1. Check that the store blob exists and has the layer's size; otherwise the install + fails. +2. Hardlink it to `/blobs//`. A file already there is + containerd's own blob or an earlier link and is used as it is. +3. Register it under the install lease. With containerd's default `shared` content + policy a writer for a blob already at its path copies nothing and does not hash it. + The blob was verified when it was pulled, and the unpack checks every extracted + layer against the image's diff IDs. If containerd's blob cleanup removed the file + in between, the layer is linked and registered once more; a second miss fails the + install. + +Deleting the lease lets containerd's GC unlink its name only; the store keeps its own. + +## Limit + +The store and the content store must be on one filesystem: a link across filesystems +fails the install, naming both paths. The test environment keeps its store on the test +dockerd's data volume for that. From d287b1d9c993e4386491668ebb3b32d45f900ee2 Mon Sep 17 00:00:00 2001 From: Mike Sul Date: Mon, 19 Oct 2026 19:34:03 +0200 Subject: [PATCH 2/2] feat(compose): hardlink store blobs into containerd The containerd install copied every compressed layer from the store into containerd's content store before unpacking it: a transient second copy that doubled the bytes written by an install and made its peak space the compressed size of the app. The layer's store blob is now checked for presence and size and hardlinked to the path the content store keeps it at. With the default "shared" content policy containerd registers a file already at that path without copying or hashing it, so the existing WriteBlob moves no bytes; the blob was verified at pull and the unpack checks every extracted layer against the image's diff IDs. The store and the content store must be on one filesystem: a link that cannot be made fails the install, naming both. The test environment moves its store onto the test dockerd's data volume for that. Assisted-by: Claude Code:claude-fable-5-1 Signed-off-by: Mike Sul --- dev-shell.sh | 2 +- pkg/compose/ctrd_install.go | 157 +++++++++++++-- pkg/compose/ctrd_install_rig_test.go | 279 ++++++++++++++++++++++++++- pkg/compose/ctrd_install_test.go | 155 +++++++++++++++ pkg/compose/export_test.go | 27 ++- test/compose/docker-compose.yml | 6 +- test/fixtures/composectl_cmds.go | 3 +- 7 files changed, 603 insertions(+), 26 deletions(-) diff --git a/dev-shell.sh b/dev-shell.sh index dc2a40f..13bfaf5 100755 --- a/dev-shell.sh +++ b/dev-shell.sh @@ -1,7 +1,7 @@ down() { docker compose --env-file=test/compose/.env.test -f test/compose/docker-compose.yml down --remove-orphans # remove the docker runtime and compose app store volumes - docker volume rm compose_docker-data compose_docker-runtime compose_reset-apps \ + docker volume rm compose_docker-data compose_docker-runtime \ compose_publisher-data compose_publisher-runtime } diff --git a/pkg/compose/ctrd_install.go b/pkg/compose/ctrd_install.go index 45db92f..3d560bf 100644 --- a/pkg/compose/ctrd_install.go +++ b/pkg/compose/ctrd_install.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "io/fs" "os" "path/filepath" "strings" @@ -25,7 +26,8 @@ import ( ) // Install of app images straight into the containerd behind dockerd: the compressed layers are -// streamed under a short lease and dropped by containerd's GC once the lease is deleted. +// hardlinked into containerd's content store under a short lease and dropped by containerd's GC +// once the lease is deleted. const ( // MinDockerVersionForCtrdInstall is the oldest dockerd on which `docker image ls` lists an image @@ -39,6 +41,9 @@ const ( gcRefManifestLabel = "containerd.io/gc.ref.content.m.0" // recordCleanupTimeout bounds the removal of partial image records after a failed write. recordCleanupTimeout = 30 * time.Second + // ingestAbortAttempts and ingestAbortDelay bound the retries of a failed ingest abort. + ingestAbortAttempts = 20 + ingestAbortDelay = 50 * time.Millisecond ) type ( @@ -165,7 +170,7 @@ func pluginRoot(ctx context.Context, c *containerd.Client, pluginType, id string return resp.Plugins[0].Exports["root"], nil } -// installImagesCtrd installs the images of one app into dockerd's containerd. Under a lease it streams +// installImagesCtrd installs the images of one app into dockerd's containerd. Under a lease it links // the layers that have no snapshot yet, unpacks every image and syncs the snapshotter's filesystem, // and only then creates the image records: a record never points at a snapshot that a power cut // can still lose. Deleting the lease lets containerd's GC drop the compressed layers. @@ -234,9 +239,10 @@ func deleteInstallLeases(ctx context.Context, ls leases.Manager) error { return nil } -// installImage streams what image needs and unpacks it. If a base snapshot went away between the -// scan and the unpack (for example `docker image prune` removed the image that held it), the unpack -// fails on the missing layer blob; then every layer is streamed and the unpack is run once more. +// installImage puts what the image needs into the content store and unpacks it. If a base snapshot +// went away between the scan and the unpack (for example `docker image prune` removed the image that +// held it), the unpack fails on the missing layer blob; then every layer is linked and the unpack is +// run once more. func (c *ctrdClient) installImage(ctx context.Context, img *ctrdImage, blobsRoot string, platform platforms.MatchComparer, report ProgressCallbackFn) error { first, err := c.firstMissingSnapshot(ctx, img.chainIDs) @@ -244,8 +250,8 @@ func (c *ctrdClient) installImage(ctx context.Context, img *ctrdImage, blobsRoot return fmt.Errorf("failed to scan the snapshots: %w", err) } reportImageState(report, ImageLoadStateImageLoading, img.ref) - if err := c.streamImage(ctx, img, blobsRoot, first); err != nil { - return fmt.Errorf("failed to stream the image to containerd: %w", err) + if err := c.putImage(ctx, img, blobsRoot, first); err != nil { + return fmt.Errorf("failed to put the image into containerd: %w", err) } err = c.unpack(ctx, img, platform) if err == nil { @@ -254,17 +260,17 @@ func (c *ctrdClient) installImage(ctx context.Context, img *ctrdImage, blobsRoot if !errdefs.IsNotFound(err) { return fmt.Errorf("failed to unpack: %w", err) } - if err := c.streamLayers(ctx, img.layers, blobsRoot); err != nil { - return fmt.Errorf("failed to stream every layer to containerd after a snapshot went away: %w", err) + if err := c.linkLayers(ctx, img.layers, blobsRoot); err != nil { + return fmt.Errorf("failed to link every layer into containerd after a snapshot went away: %w", err) } if err := c.unpack(ctx, img, platform); err != nil { - return fmt.Errorf("failed to unpack after streaming every layer: %w", err) + return fmt.Errorf("failed to unpack after linking every layer: %w", err) } return nil } // firstMissingSnapshot returns the index of the first layer whose snapshot does not exist; every -// layer after it has to be streamed and unpacked too, because a snapshot needs its parent. +// layer after it has to be linked and unpacked too, because a snapshot needs its parent. func (c *ctrdClient) firstMissingSnapshot(ctx context.Context, chainIDs []digest.Digest) (int, error) { sn := c.SnapshotService(c.snapshotter) for i, chainID := range chainIDs { @@ -279,11 +285,11 @@ func (c *ctrdClient) firstMissingSnapshot(ctx context.Context, chainIDs []digest return len(chainIDs), nil } -// streamImage writes the layers from first on, the config, the manifest and, when the image is -// pinned by one, the index. The manifest gets no layer labels: only the lease holds the layers. -// WriteBlob leaves the labels of a blob that already exists as they are. -func (c *ctrdClient) streamImage(ctx context.Context, img *ctrdImage, blobsRoot string, first int) error { - if err := c.streamLayers(ctx, img.layers[first:], blobsRoot); err != nil { +// putImage puts the layers from first on, the config, the manifest and, when the image is pinned +// by one, the index into the content store. The manifest gets no layer labels: only the lease holds +// the layers. WriteBlob leaves the labels of a blob that already exists as they are. +func (c *ctrdClient) putImage(ctx context.Context, img *ctrdImage, blobsRoot string, first int) error { + if err := c.linkLayers(ctx, img.layers[first:], blobsRoot); err != nil { return err } if err := c.writeBlob(ctx, blobsRoot, img.config, nil); err != nil { @@ -306,15 +312,130 @@ func (c *ctrdClient) streamImage(ctx context.Context, img *ctrdImage, blobsRoot return err } -func (c *ctrdClient) streamLayers(ctx context.Context, layers []ocispec.Descriptor, blobsRoot string) error { +// linkLayers puts the layers into the content store without copying them: each store blob is +// hardlinked to the path the content store keeps it at, and then registered in place. +func (c *ctrdClient) linkLayers(ctx context.Context, layers []ocispec.Descriptor, blobsRoot string) error { for _, layer := range layers { - if err := c.writeBlob(ctx, blobsRoot, layer, nil); err != nil { + if err := linkLayer(c.contentRoot, blobsRoot, layer); err != nil { return err } + if err := c.registerLayer(ctx, blobsRoot, layer); err != nil { + return err + } + } + return nil +} + +// registerLayer records a layer whose file is at the content store's path for it, without reading +// it: the writer the content store opens for such a file owns no ingest and reports the layer's +// size as its offset, and is committed as it is. A writer at any other offset means the file is +// not there (containerd's blob cleanup removes a file it has no record of): its ingest is +// discarded, the layer is linked again and the registration is tried once more; a second miss +// fails the install rather than copy the layer. +func (c *ctrdClient) registerLayer(ctx context.Context, blobsRoot string, desc ocispec.Descriptor) error { + cs := c.ContentStore() + ref := "composectl-" + desc.Digest.String() + for attempt := 1; ; attempt++ { + w, err := content.OpenWriter(ctx, cs, content.WithRef(ref), content.WithDescriptor(desc)) + if errdefs.IsAlreadyExists(err) { + return nil + } + if err != nil { + return fmt.Errorf("failed to open a content writer for layer %s: %w", desc.Digest, err) + } + st, err := w.Status() + if err == nil && st.Offset == desc.Size { + err = w.Commit(ctx, desc.Size, desc.Digest) + _ = w.Close() + if err != nil && !errdefs.IsAlreadyExists(err) { + return fmt.Errorf("failed to register layer %s: %w", desc.Digest, err) + } + return nil + } + _ = w.Close() + if err != nil { + return fmt.Errorf("failed to read the content writer status of layer %s: %w", desc.Digest, err) + } + if err := abortIngest(ctx, cs, ref); err != nil { + return fmt.Errorf("failed to discard the ingest of layer %s: %w", desc.Digest, err) + } + if attempt == 2 { + return fmt.Errorf("layer %s cannot be registered in containerd's content store without a copy:"+ + " its link is removed before it is recorded", desc.Digest) + } + if err := linkLayer(c.contentRoot, blobsRoot, desc); err != nil { + return err + } + } +} + +// abortIngest discards the ingest of ref, retrying up to ingestAbortAttempts times, ingestAbortDelay +// apart, while the abort fails: the writer's Close only tells containerd to stop, so its side may +// still be writing the ingest's status when the abort removes the ingest directory, which then +// fails as not empty. A cancelled context ends the retries with its error. +func abortIngest(ctx context.Context, cs content.IngestManager, ref string) error { + for attempt := 1; ; attempt++ { + err := cs.Abort(ctx, ref) + if err == nil || errdefs.IsNotFound(err) { + return nil + } + if attempt == ingestAbortAttempts { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(ingestAbortDelay): + } + } +} + +// linkFile makes the hardlink of a store blob in the content store; a variable so that tests can +// make it fail. +var linkFile = os.Link + +// linkLayer checks that the store blob of a layer is there with the layer's size and hardlinks it +// to the path the content store keeps that blob at. containerd registers a file that is already at +// that path without copying or hashing it (the "shared" content sharing policy), so the write that +// follows moves no bytes; the blob was verified when it was pulled, and the unpack checks every +// extracted layer against the image's diff IDs. A file already at that path is containerd's own +// blob or a link an interrupted install left, and is used as it is; a link that cannot be made +// for any other reason fails the install. +func linkLayer(contentRoot, blobsRoot string, desc ocispec.Descriptor) error { + src := filepath.Join(blobsRoot, desc.Digest.Encoded()) + if err := checkBlob(src, desc); err != nil { + return err + } + dir := filepath.Join(contentRoot, "blobs", desc.Digest.Algorithm().String()) + if err := os.MkdirAll(dir, 0700); err != nil { + return fmt.Errorf("failed to create the content store directory of layer %s: %w", desc.Digest, err) + } + err := linkFile(src, filepath.Join(dir, desc.Digest.Encoded())) + switch { + case err == nil || errors.Is(err, fs.ErrExist): + return nil + case errors.Is(err, syscall.EXDEV): + return fmt.Errorf("cannot link layer %s from %s into containerd's content store %s: both must be on one filesystem", + desc.Digest, blobsRoot, contentRoot) + default: + return fmt.Errorf("failed to link layer %s into containerd's content store: %w", desc.Digest, err) + } +} + +// checkBlob checks that the store blob is there and has the size its descriptor promises. +func checkBlob(path string, desc ocispec.Descriptor) error { + fi, err := os.Stat(path) + if err != nil { + return fmt.Errorf("failed to find blob %s in the store: %w", desc.Digest, err) + } + if fi.Size() != desc.Size { + return fmt.Errorf("blob %s in the store has %d bytes, expected %d", desc.Digest, fi.Size(), desc.Size) } return nil } +// writeBlob writes the store blob of desc to the content store; a blob the content store has +// already is left as it is. func (c *ctrdClient) writeBlob(ctx context.Context, blobsRoot string, desc ocispec.Descriptor, labels map[string]string) error { f, err := os.Open(filepath.Join(blobsRoot, desc.Digest.Encoded())) if err != nil { diff --git a/pkg/compose/ctrd_install_rig_test.go b/pkg/compose/ctrd_install_rig_test.go index 6559aba..9882ac3 100644 --- a/pkg/compose/ctrd_install_rig_test.go +++ b/pkg/compose/ctrd_install_rig_test.go @@ -5,18 +5,26 @@ package compose_test // (built by `make`) is not there, as under `make test-unit`. import ( + "bytes" "context" "fmt" "os" "os/exec" + "path/filepath" + "strings" + "syscall" "testing" "time" + "github.com/containerd/containerd/content" "github.com/containerd/containerd/errdefs" "github.com/containerd/containerd/images" + "github.com/containerd/containerd/leases" "github.com/containerd/containerd/platforms" "github.com/foundriesio/composeapp/pkg/compose" f "github.com/foundriesio/composeapp/test/fixtures" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" ) const derivedAppDef = ` @@ -44,7 +52,7 @@ func ctrdRig(t *testing.T) *compose.CtrdClient { } // TestCtrdInstallRetriesAfterBaseSnapshotRemoval: the base snapshot found by the scan is removed -// before the unpack, as `docker image prune` could do, and the install streams every layer and +// before the unpack, as `docker image prune` could do, and the install links every layer and // unpacks again. The removal runs in the install's progress callback, which installImagesCtrd // calls synchronously after the scan and before the layers are streamed. func TestCtrdInstallRetriesAfterBaseSnapshotRemoval(t *testing.T) { @@ -106,3 +114,272 @@ func TestCtrdInstallRetriesAfterBaseSnapshotRemoval(t *testing.T) { out, err := exec.Command("docker", "run", "--rm", "--pull", "never", "--network", "none", derivedRef, "echo", "ok").CombinedOutput() f.Checkf(t, err, "docker run %s: %s", derivedRef, out) } + +// linkRig publishes and pulls a one-image app with a fresh layer and returns the app, its tree and +// its layer descriptors. +func linkRig(t *testing.T, cfg *compose.Config) (*f.App, compose.App, []ocispec.Descriptor) { + t.Helper() + ctx := context.Background() + store := compose.NewStoreBlobProvider(cfg.GetBlobsRoot()) + app := f.NewApp(t, fmt.Sprintf(derivedAppDef, f.PublishDerivedImage(t, "ghcr.io/foundriesio/busybox:1.36", 3))) + app.Publish(t) + app.Pull(t) + tree, err := cfg.AppLoader.LoadAppTree(ctx, store, platforms.OnlyStrict(cfg.Platform), app.PublishedUri) + f.Check(t, err) + layers, err := compose.LayerDescriptorsOf(ctx, tree, store) + f.Check(t, err) + return app, tree, layers[0] +} + +// storeCopyIn copies the store's blobs into a new store root under dir and returns a config that +// uses it. +func storeCopyIn(t *testing.T, cfg *compose.Config, dir string) *compose.Config { + t.Helper() + root, err := os.MkdirTemp(dir, "composectl-store-") + f.Check(t, err) + t.Cleanup(func() { os.RemoveAll(root) }) + copyCfg := *cfg + copyCfg.StoreRoot = root + f.Check(t, os.MkdirAll(copyCfg.GetBlobsRoot(), 0700)) + blobs, err := os.ReadDir(cfg.GetBlobsRoot()) + f.Check(t, err) + for _, b := range blobs { + data, err := os.ReadFile(filepath.Join(cfg.GetBlobsRoot(), b.Name())) + f.Check(t, err) + f.Check(t, os.WriteFile(filepath.Join(copyCfg.GetBlobsRoot(), b.Name()), data, 0600)) + } + return ©Cfg +} + +// layersToRegister returns the layers whose snapshot does not exist yet, from the first one on: the +// install leaves the ones below it alone, so only these reach the content store. +func layersToRegister(t *testing.T, c *compose.CtrdClient, cfg *compose.Config, tree compose.App, layers []ocispec.Descriptor) []ocispec.Descriptor { + t.Helper() + ctx := context.Background() + chains, err := compose.ChainIDsOf(ctx, tree, compose.NewStoreBlobProvider(cfg.GetBlobsRoot())) + f.Check(t, err) + sn := c.SnapshotService(compose.CtrdSnapshotter(c)) + for i, chainID := range chains[0] { + if _, err := sn.Stat(ctx, chainID); errdefs.IsNotFound(err) { + return layers[i:] + } + } + return nil +} + +func deviceOf(t *testing.T, path string) uint64 { + t.Helper() + var st syscall.Stat_t + f.Check(t, syscall.Stat(path, &st)) + return uint64(st.Dev) +} + +// deleteImage removes both records the install creates for the app's image, the pinned digest +// reference and the short tag reference, so that containerd's GC drops its snapshots. +func deleteImage(t *testing.T, c *compose.CtrdClient, tree compose.App) { + t.Helper() + ref := tree.GetComposeRoot().Children[0].Ref() + parsed, err := compose.ParseImageRef(ref) + f.Check(t, err) + for _, name := range []string{ref, parsed.GetTagRef()} { + f.Check(t, c.ImageService().Delete(context.Background(), name, images.SynchronousDelete())) + } +} + +func contentBlobPath(c *compose.CtrdClient, d digest.Digest) string { + return filepath.Join(compose.CtrdContentRoot(c), "blobs", d.Algorithm().String(), d.Encoded()) +} + +func statNlink(t *testing.T, path string) (os.FileInfo, uint64) { + t.Helper() + fi, err := os.Stat(path) + f.Check(t, err) + return fi, uint64(fi.Sys().(*syscall.Stat_t).Nlink) +} + +// TestCtrdInstallLinksLayers: every layer blob is a second name of the store blob while the install +// lease exists, and the content root holds no copy of it afterwards. One layer is linked before the +// install, as an interrupted install leaves it, and is registered as is. +func TestCtrdInstallLinksLayers(t *testing.T) { + c := ctrdRig(t) + cfg := f.NewTestConfig(t) + app, tree, layers := linkRig(t, cfg) + defer app.Remove(t) + layers = layersToRegister(t, c, cfg, tree, layers) + if len(layers) == 0 { + t.Fatal("every layer of the fresh image already has a snapshot") + } + storePath := func(d digest.Digest) string { return filepath.Join(cfg.GetBlobsRoot(), d.Encoded()) } + f.Check(t, os.MkdirAll(filepath.Dir(contentBlobPath(c, layers[0].Digest)), 0700)) + f.Check(t, os.Link(storePath(layers[0].Digest), contentBlobPath(c, layers[0].Digest))) + + checked := false + atLoaded := func(p *compose.LoadImageProgress) { + if p.State != compose.ImageLoadStateImageLoaded { + return + } + checked = true + for _, l := range layers { + src, nlink := statNlink(t, storePath(l.Digest)) + dst, _ := statNlink(t, contentBlobPath(c, l.Digest)) + if nlink != 2 || !os.SameFile(src, dst) { + t.Errorf("layer %s: nlink %d, same file %v; expected a link into the content root", l.Digest, nlink, os.SameFile(src, dst)) + } + } + } + f.Check(t, compose.InstallImagesCtrd(context.Background(), c, tree, cfg, atLoaded)) + defer deleteImage(t, c, tree) + if !checked { + t.Fatal("the image was never reported loaded") + } + for _, l := range layers { + if _, nlink := statNlink(t, storePath(l.Digest)); nlink != 1 { + t.Errorf("layer %s still has %d names after the lease was deleted", l.Digest, nlink) + } + if _, err := os.Stat(contentBlobPath(c, l.Digest)); !errdefs.IsNotFound(err) && !os.IsNotExist(err) { + t.Errorf("layer %s is still in the content root: %v", l.Digest, err) + } + } +} + +// TestCtrdInstallFailsAcrossFilesystems: with the store on another filesystem the link fails and +// so does the install, naming both directories, without a copy of any layer in the content root; +// the next install from a store on the content root's filesystem removes the lease it left. +func TestCtrdInstallFailsAcrossFilesystems(t *testing.T) { + c := ctrdRig(t) + ctx := context.Background() + cfg := f.NewTestConfig(t) + app, tree, layers := linkRig(t, cfg) + defer app.Remove(t) + other := "/dev/shm" + if _, err := os.Stat(other); err != nil || deviceOf(t, other) == deviceOf(t, compose.CtrdContentRoot(c)) { + t.Skip("no filesystem other than the content root's is available") + } + otherCfg := storeCopyIn(t, cfg, other) + layers = layersToRegister(t, c, otherCfg, tree, layers) + if len(layers) == 0 { + t.Fatal("every layer of the fresh image already has a snapshot") + } + + err := compose.InstallImagesCtrd(ctx, c, tree, otherCfg, nil) + if err == nil || !strings.Contains(err.Error(), otherCfg.GetBlobsRoot()) || !strings.Contains(err.Error(), compose.CtrdContentRoot(c)) { + t.Fatalf("expected the install to fail naming the store and the content root, got %v", err) + } + for _, l := range layers { + if _, err := os.Stat(contentBlobPath(c, l.Digest)); !os.IsNotExist(err) { + t.Errorf("layer %s is in the content root after the failed install: %v", l.Digest, err) + } + } + installLeases := func() (n int) { + all, err := c.LeasesService().List(ctx) + f.Check(t, err) + for _, l := range all { + if strings.HasPrefix(l.ID, "composectl/install/") { + n++ + } + } + return n + } + if installLeases() != 1 { + t.Fatalf("expected the failed install to leave its lease, found %d", installLeases()) + } + f.Check(t, compose.InstallImagesCtrd(ctx, c, tree, cfg, nil)) + defer deleteImage(t, c, tree) + if installLeases() != 0 { + t.Fatalf("expected the next install to remove the lease, found %d", installLeases()) + } +} + +// TestCtrdLinkedBlobAndGC: containerd's blob-directory cleanup runs after a GC that removed content. +// A linked file it finds before the writer is opened is removed, and the registration links it again +// rather than copy it; once the writer is open its expected digest keeps the file; after the commit +// the record keeps it. +func TestCtrdLinkedBlobAndGC(t *testing.T) { + c := ctrdRig(t) + ctx := context.Background() + cfg := f.NewTestConfig(t) + app, _, layers := linkRig(t, cfg) + defer app.Remove(t) + cs, ls := c.ContentStore(), c.LeasesService() + layer := layers[len(layers)-1] + src := filepath.Join(cfg.GetBlobsRoot(), layer.Digest.Encoded()) + dst := contentBlobPath(c, layer.Digest) + ref := "composectl-" + layer.Digest.String() + f.Check(t, os.MkdirAll(filepath.Dir(dst), 0700)) + + // forceGC commits a throwaway blob under its own lease and deletes that lease, which removes + // the blob's record and so triggers the blob-directory cleanup. + forceGC := func() { + lease, err := ls.Create(ctx, leases.WithRandomID()) + f.Check(t, err) + data := []byte(fmt.Sprintf("throwaway %d", time.Now().UnixNano())) + f.Check(t, content.WriteBlob(leases.WithLease(ctx, lease.ID), cs, "throwaway", bytes.NewReader(data), + ocispec.Descriptor{Digest: digest.FromBytes(data), Size: int64(len(data))})) + f.Check(t, ls.Delete(ctx, lease, leases.SynchronousDelete)) + } + waitGone := func(path string) bool { + for i := 0; i < 50; i++ { + if _, err := os.Stat(path); os.IsNotExist(err) { + return true + } + time.Sleep(100 * time.Millisecond) + } + return false + } + lease, err := ls.Create(ctx, leases.WithRandomID()) + f.Check(t, err) + defer func() { _ = ls.Delete(ctx, lease, leases.SynchronousDelete) }() + lctx := leases.WithLease(ctx, lease.ID) + + // (a) cleanup before the writer: the link goes; the registration finds an ordinary writer, + // discards its ingest, links again and registers the link, copying nothing. + f.Check(t, os.Link(src, dst)) + forceGC() + if !waitGone(dst) { + t.Fatal("cleanup did not remove the unreferenced link") + } + time.Sleep(time.Second) // the cleanup pass that removed the link may still be walking the directory + f.Check(t, compose.RegisterLayer(lctx, c, cfg.GetBlobsRoot(), layer)) + srcInfo, _ := statNlink(t, src) + if dstInfo, nlink := statNlink(t, dst); nlink != 2 || !os.SameFile(srcInfo, dstInfo) { + t.Fatalf("expected the layer linked again after cleanup removed the link, nlink %d same file %v", nlink, os.SameFile(srcInfo, dstInfo)) + } + if _, err := cs.Info(ctx, layer.Digest); err != nil { + t.Fatalf("the relinked blob has no record: %v", err) + } + if statuses, err := cs.ListStatuses(ctx, "ref=="+ref); err != nil || len(statuses) != 0 { + t.Fatalf("expected no ingest left behind, got %v, %v", statuses, err) + } + f.Check(t, cs.Delete(ctx, layer.Digest)) + forceGC() + if !waitGone(dst) { + t.Fatal("the deleted blob was not cleaned up") + } + + // (b) cleanup between the writer and the commit: the expected digest keeps the file. + f.Check(t, os.Link(src, dst)) + w, err := content.OpenWriter(lctx, cs, content.WithRef("composectl-"+layer.Digest.String()), content.WithDescriptor(layer)) + f.Check(t, err) + st, err := w.Status() + f.Check(t, err) + if st.Offset != layer.Size { + t.Fatalf("expected a shared writer at offset %d, got %d", layer.Size, st.Offset) + } + forceGC() + time.Sleep(time.Second) + if _, nlink := statNlink(t, dst); nlink != 2 { + t.Fatalf("the linked blob of an open writer was removed, nlink %d", nlink) + } + f.Check(t, w.Commit(lctx, layer.Size, layer.Digest)) + w.Close() + + // (c) cleanup after the commit: the record keeps the file. + forceGC() + time.Sleep(time.Second) + if _, nlink := statNlink(t, dst); nlink != 2 { + t.Fatalf("the committed linked blob was removed, nlink %d", nlink) + } + if _, err := cs.Info(ctx, layer.Digest); err != nil { + t.Fatalf("the linked blob has no record: %v", err) + } +} diff --git a/pkg/compose/ctrd_install_test.go b/pkg/compose/ctrd_install_test.go index 4448741..f6a1a02 100644 --- a/pkg/compose/ctrd_install_test.go +++ b/pkg/compose/ctrd_install_test.go @@ -3,13 +3,18 @@ package compose import ( "context" "errors" + "os" + "path/filepath" "strings" + "syscall" "testing" + "github.com/containerd/containerd/content" "github.com/containerd/containerd/errdefs" "github.com/containerd/containerd/images" "github.com/containerd/containerd/leases" "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" ) func TestInstallLeaseID(t *testing.T) { @@ -121,3 +126,153 @@ func TestPutImageRecordsRollsBackOnCancel(t *testing.T) { t.Fatal("the rollback ran on the cancelled context") } } + +func writeStoreBlob(t *testing.T, blobsRoot string, data []byte) ocispec.Descriptor { + t.Helper() + d := digest.FromBytes(data) + if err := os.MkdirAll(blobsRoot, 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(blobsRoot, d.Encoded()), data, 0600); err != nil { + t.Fatal(err) + } + return ocispec.Descriptor{Digest: d, Size: int64(len(data))} +} + +func TestLinkLayerLinksVerifiedBlob(t *testing.T) { + root := t.TempDir() + blobsRoot := filepath.Join(root, "store") + contentRoot := filepath.Join(root, "content") + desc := writeStoreBlob(t, blobsRoot, []byte("layer bytes")) + + if err := linkLayer(contentRoot, blobsRoot, desc); err != nil { + t.Fatal(err) + } + src, err := os.Stat(filepath.Join(blobsRoot, desc.Digest.Encoded())) + if err != nil { + t.Fatal(err) + } + dst, err := os.Stat(filepath.Join(contentRoot, "blobs", "sha256", desc.Digest.Encoded())) + if err != nil { + t.Fatalf("the blob was not linked into the content root: %v", err) + } + if !os.SameFile(src, dst) { + t.Fatal("the content root blob is not a link to the store blob") + } + // A second install finds the link in place and leaves it alone. + if err := linkLayer(contentRoot, blobsRoot, desc); err != nil { + t.Fatalf("linking over an existing link must not fail: %v", err) + } +} + +func TestLinkLayerRejectsWrongSize(t *testing.T) { + root := t.TempDir() + blobsRoot := filepath.Join(root, "store") + contentRoot := filepath.Join(root, "content") + desc := writeStoreBlob(t, blobsRoot, []byte("layer bytes")) + desc.Size++ + + if err := linkLayer(contentRoot, blobsRoot, desc); err == nil || !strings.Contains(err.Error(), desc.Digest.String()) { + t.Fatalf("expected a size mismatch naming %s, got %v", desc.Digest, err) + } + if _, err := os.Stat(filepath.Join(contentRoot, "blobs")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("nothing must be linked for a blob of the wrong size, stat: %v", err) + } +} + +func TestLinkLayerFailsOnAnotherFilesystem(t *testing.T) { + root := t.TempDir() + blobsRoot := filepath.Join(root, "store") + contentRoot := filepath.Join(root, "content") + desc := writeStoreBlob(t, blobsRoot, []byte("layer bytes")) + + orig := linkFile + t.Cleanup(func() { linkFile = orig }) + linkFile = func(string, string) error { return &os.LinkError{Op: "link", Err: syscall.EXDEV} } + + err := linkLayer(contentRoot, blobsRoot, desc) + if err == nil || !strings.Contains(err.Error(), blobsRoot) || !strings.Contains(err.Error(), contentRoot) { + t.Fatalf("expected a failure naming the store and the content root, got %v", err) + } + if _, err := os.Stat(filepath.Join(contentRoot, "blobs", "sha256", desc.Digest.Encoded())); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("no link expected, stat: %v", err) + } +} + +func TestLinkLayerFailsOnMissingBlob(t *testing.T) { + root := t.TempDir() + desc := ocispec.Descriptor{Digest: digest.FromString("absent"), Size: 6} + err := linkLayer(filepath.Join(root, "content"), filepath.Join(root, "store"), desc) + if !errors.Is(err, os.ErrNotExist) || !strings.Contains(err.Error(), desc.Digest.String()) { + t.Fatalf("expected a not-found failure naming the layer, got %v", err) + } +} + +// fakeIngests is a content.IngestManager whose Abort returns err for the first failures calls, as +// containerd does while the server side of a closed writer still writes the ingest's status, and +// calls cancel, when set, on the first call. +type fakeIngests struct { + failures int + err error + cancel context.CancelFunc + aborts int +} + +func (s *fakeIngests) Status(context.Context, string) (content.Status, error) { + return content.Status{}, errdefs.ErrNotFound +} +func (s *fakeIngests) ListStatuses(context.Context, ...string) ([]content.Status, error) { + return nil, nil +} +func (s *fakeIngests) Abort(context.Context, string) error { + s.aborts++ + if s.cancel != nil { + s.cancel() + } + if s.aborts <= s.failures { + return s.err + } + return nil +} + +func TestAbortIngestRetriesWhileTheAbortFails(t *testing.T) { + ingests := &fakeIngests{failures: 2, err: errors.New("unlinkat ingest: directory not empty")} + if err := abortIngest(context.Background(), ingests, "ref"); err != nil { + t.Fatalf("the abort failed after the retries: %v", err) + } + if ingests.aborts != 3 { + t.Fatalf("expected 3 abort calls, got %d", ingests.aborts) + } +} + +func TestAbortIngestGivesUpAfterTheBoundedRetries(t *testing.T) { + ingests := &fakeIngests{failures: ingestAbortAttempts + 1, err: errors.New("unlinkat ingest: directory not empty")} + if err := abortIngest(context.Background(), ingests, "ref"); !errors.Is(err, ingests.err) { + t.Fatalf("expected the last abort error, got %v", err) + } + if ingests.aborts != ingestAbortAttempts { + t.Fatalf("expected %d abort calls, got %d", ingestAbortAttempts, ingests.aborts) + } +} + +func TestAbortIngestOfUnknownRefSucceeds(t *testing.T) { + ingests := &fakeIngests{failures: 1, err: errdefs.ErrNotFound} + if err := abortIngest(context.Background(), ingests, "ref"); err != nil { + t.Fatalf("an unknown ingest must not fail the abort: %v", err) + } + if ingests.aborts != 1 { + t.Fatalf("expected 1 abort call, got %d", ingests.aborts) + } +} + +func TestAbortIngestStopsWhenCancelled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ingests := &fakeIngests{failures: ingestAbortAttempts + 1, err: errors.New("unlinkat ingest: directory not empty"), cancel: cancel} + if err := abortIngest(ctx, ingests, "ref"); !errors.Is(err, context.Canceled) { + t.Fatalf("expected the cancellation to be returned, got %v", err) + } + if ingests.aborts != 1 { + t.Fatalf("expected the retries to stop at the cancellation, got %d abort calls", ingests.aborts) + } +} diff --git a/pkg/compose/export_test.go b/pkg/compose/export_test.go index 82039ba..6ec50db 100644 --- a/pkg/compose/export_test.go +++ b/pkg/compose/export_test.go @@ -1,6 +1,10 @@ package compose -import "context" +import ( + "context" + + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) // Exported for the tests in package compose_test that drive the containerd install directly. @@ -11,6 +15,10 @@ var ( InstallImagesCtrd = installImagesCtrd ) +func RegisterLayer(ctx context.Context, c *CtrdClient, blobsRoot string, desc ocispec.Descriptor) error { + return c.registerLayer(ctx, blobsRoot, desc) +} + func CtrdSnapshotter(c *CtrdClient) string { return c.snapshotter } @@ -31,3 +39,20 @@ func ChainIDsOf(ctx context.Context, app App, provider BlobProvider) ([][]string } return out, nil } + +func CtrdContentRoot(c *CtrdClient) string { + return c.contentRoot +} + +// LayerDescriptorsOf returns the layer descriptors of each app image, in the app's image order. +func LayerDescriptorsOf(ctx context.Context, app App, provider BlobProvider) ([][]ocispec.Descriptor, error) { + imgs, err := ctrdImagesOf(ctx, app, provider) + if err != nil { + return nil, err + } + var out [][]ocispec.Descriptor + for _, img := range imgs { + out = append(out, img.layers) + } + return out, nil +} diff --git a/test/compose/docker-compose.yml b/test/compose/docker-compose.yml index b2a70e5..e9084ad 100644 --- a/test/compose/docker-compose.yml +++ b/test/compose/docker-compose.yml @@ -24,7 +24,6 @@ services: - docker-runtime:/var/run/docker - docker-data:/var/lib/docker - ${REG_DIR}/daemon.json:/etc/docker/daemon.json:ro - - reset-apps:/var/sota/reset-apps # Plays the role of the FoundriesFactory CI when tests publish apps: test images are pulled into it # and the layer sizes are read from its overlay2 store, whatever image store the dockerd above uses. @@ -54,7 +53,6 @@ services: - ${BIN_DIR}:${SRC_DIR}/bin - docker-runtime:/var/run/docker - docker-data:/var/lib/docker - - reset-apps:/var/sota/reset-apps - publisher-runtime:/var/run/docker-publisher - publisher-data:/var/lib/docker-publisher:ro working_dir: ${SRC_DIR} @@ -68,7 +66,8 @@ services: - DOCKER_HOST=unix:///var/run/docker/docker.sock - COMPOSECTL_EXE=${BIN_DIR}/composectl - SRC_DIR=${SRC_DIR} - - STOREROOT=/var/sota/reset-apps + # On the mount of dockerd's containerd, whose content store hardlinks the store's layers + - STOREROOT=/var/lib/docker/reset-apps - COMPOSEROOT=/var/sota/compose-apps - DOCKERD_FIO_PATCH=${DOCKERD_FIO_PATCH} - PUBLISHER_DOCKER_HOST=unix:///var/run/docker-publisher/docker.sock @@ -77,6 +76,5 @@ services: volumes: docker-data: docker-runtime: - reset-apps: publisher-data: publisher-runtime: diff --git a/test/fixtures/composectl_cmds.go b/test/fixtures/composectl_cmds.go index 4729fa6..011b69f 100644 --- a/test/fixtures/composectl_cmds.go +++ b/test/fixtures/composectl_cmds.go @@ -23,7 +23,8 @@ import ( ) const ( - AppStoreRoot = "/var/sota/reset-apps" + // On the mount of the test dockerd's containerd: an install hardlinks the store's layers into it. + AppStoreRoot = "/var/lib/docker/reset-apps" AppComposeRootRoot = "/var/sota/compose-apps" )