Merge pull request 'Link a build's bases into its repository before pushing it (issue 694)' (#318) from fix/694-a-push-holds-its-bases into main

internal/builder/builder.go

@@ -1076,6 +1076,16 @@ func one(ctx context.Context, run Runner, publish Publisher,
 			return catalogue.Built{}, fmt.Errorf("%s: building %s failed: %w", module, a.Name, err)
 		}
 		say("image", "built, publishing")
+		// Its bases linked into its repository first, so the push needs none of their blobs from this
+		// machine's store, which may not hold them (novox/hq issue 694). Said and not fatal: a push
+		// from a store that holds them works without, and one that fails says which blob it lacked.
+		if lender, can := publish.(BaseLender); can {
+			if n, err := lender.LendBases(ctx, module+"/"+a.Name, buildArgValues(args)); err != nil {
+				say("image", "its bases could not all be linked into %s/%s before the push: %v", module, a.Name, err)
+			} else if n > 0 {
+				say("image", "%d blob(s) of its bases linked into %s/%s before the push", n, module, a.Name)
+			}
+		}
 		reference, err := publish.PublishImage(ctx, local, module+"/"+a.Name)
 		if err != nil {
 			return catalogue.Built{}, err
@@ -1427,6 +1437,19 @@ func standingOn(ctx context.Context, manifest catalogue.Manifest, held map[strin
 	return args, resolved, nil
 }
 
+// buildArgValues is what each --build-arg of a build was handed: the bases, as this mesh holds them.
+func buildArgValues(args []string) []string {
+	var values []string
+	for i := 0; i+1 < len(args); i += 2 {
+		if args[i] == "--build-arg" {
+			if _, value, has := strings.Cut(args[i+1], "="); has {
+				values = append(values, value)
+			}
+		}
+	}
+	return values
+}
+
 // compile runs a module's own code through its toolchain, and says where the result is.
 //
 // **In the module's own directory, under the path the toolchain expects.** A module is compiled

internal/builder/lend.go

@@ -0,0 +1,178 @@
+package builder
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"runtime"
+	"strings"
+)
+
+// **A built image's bases are linked into its repository before it is pushed** (novox/hq issue 694).
+//
+// A push reads every blob of the image out of this machine's store, unless the repository it goes to
+// already holds that blob: the runtime asks the registry first, blob by blob, and reads locally only
+// what the registry lacks. A runtime keeping its images in containerd's store — the default for a
+// fresh install from docker 29 on — does not always hold the compressed layers of the base an image
+// was built FROM: a layer whose unpacked form it already had is never fetched. The push of such an
+// image then fails on a layer this machine never had ("not all of them are available locally"), while
+// the very bytes sit in this registry, under the repository the base was mirrored into.
+//
+// So the base's blobs are mounted into the image's repository first. A mount moves no bytes, and once
+// the repository holds a blob the push asks this machine for nothing of it — whichever store the
+// runtime keeps. A runtime that holds every layer (the classic store) is not changed by this: its push
+// finds the same blobs already there.
+
+// BaseLender links what a build's bases hold into the repository its image is about to be pushed to.
+type BaseLender interface {
+	// LendBases mounts every blob of each base this registry holds — for the platform this machine
+	// builds — into `repository`, and answers how many it mounted.
+	LendBases(ctx context.Context, repository string, bases []string) (int, error)
+}
+
+// platformed is an index's entry: a manifest and the platform it is for.
+type platformed struct {
+	descriptor
+	Platform *struct {
+		OS           string `json:"os"`
+		Architecture string `json:"architecture"`
+	} `json:"platform"`
+}
+
+// LendBases mounts each base's blobs into the repository. A base this registry does not hold under its
+// repository — named by tag, or held somewhere else — lends nothing and is no error: the push reads it
+// from this machine's store as before.
+func (r Registry) LendBases(ctx context.Context, repository string, bases []string) (int, error) {
+	mounted := 0
+	for _, base := range bases {
+		name, digest, pinned := strings.Cut(base, "@")
+		if !pinned || !strings.HasPrefix(digest, "sha256:") {
+			continue
+		}
+		where, err := parseReference(name + "@" + digest)
+		if err != nil {
+			continue
+		}
+		from := where.repository
+		if from == repository {
+			continue
+		}
+		blobs, held, err := r.blobsOf(ctx, from, digest)
+		if err != nil {
+			return mounted, fmt.Errorf("reading %s: %w", base, err)
+		}
+		if !held {
+			continue
+		}
+		for _, blob := range blobs {
+			did, err := r.mount(ctx, repository, from, blob)
+			if err != nil {
+				return mounted, fmt.Errorf("linking %s of %s into %s: %w", blob, base, repository, err)
+			}
+			if did {
+				mounted++
+			}
+		}
+	}
+	return mounted, nil
+}
+
+// blobsOf is the config and layers of the manifest under repository@digest — for an index, of each
+// manifest it names for this machine's platform. False when the registry holds no such manifest there.
+func (r Registry) blobsOf(ctx context.Context, repository, digest string) ([]string, bool, error) {
+	body, held, err := r.manifest(ctx, repository, digest)
+	if err != nil || !held {
+		return nil, held, err
+	}
+	var document struct {
+		Manifests []platformed `json:"manifests"`
+		Config    *descriptor  `json:"config"`
+		Layers    []descriptor `json:"layers"`
+	}
+	if err := json.Unmarshal(body, &document); err != nil {
+		return nil, false, fmt.Errorf("%s@%s is not a manifest: %w", repository, digest, err)
+	}
+	var blobs []string
+	if document.Config != nil {
+		blobs = append(blobs, document.Config.Digest)
+	}
+	for _, l := range document.Layers {
+		blobs = append(blobs, l.Digest)
+	}
+	for _, m := range document.Manifests {
+		if m.Platform == nil || m.Platform.OS != runtime.GOOS || m.Platform.Architecture != runtime.GOARCH {
+			continue
+		}
+		more, held, err := r.blobsOf(ctx, repository, m.Digest)
+		if err != nil {
+			return nil, false, err
+		}
+		if held {
+			blobs = append(blobs, more...)
+		}
+	}
+	return blobs, true, nil
+}
+
+// manifest reads one manifest document from this registry; false when it is not there.
+func (r Registry) manifest(ctx context.Context, repository, digest string) ([]byte, bool, error) {
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet,
+		"http://"+r.Address+"/v2/"+repository+"/manifests/"+digest, nil)
+	if err != nil {
+		return nil, false, err
+	}
+	request.Header.Set("Accept", manifestAccept)
+	response, err := r.client().Do(request)
+	if err != nil {
+		return nil, false, fmt.Errorf("cannot reach the registry at %s: %w", r.Address, err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusNotFound {
+		return nil, false, nil
+	}
+	if response.StatusCode != http.StatusOK {
+		return nil, false, fmt.Errorf("%s/%s@%s: %s", r.Address, repository, digest, response.Status)
+	}
+	body, err := io.ReadAll(io.LimitReader(response.Body, 4<<20))
+	return body, err == nil, err
+}
+
+// mount links one blob from one repository into another; false when the repository already held it.
+func (r Registry) mount(ctx context.Context, repository, from, digest string) (bool, error) {
+	base := "http://" + r.Address + "/v2/" + repository
+	if there, err := r.has(ctx, base+"/blobs/"+digest); err != nil {
+		return false, err
+	} else if there {
+		return false, nil
+	}
+	start, err := http.NewRequestWithContext(ctx, http.MethodPost,
+		base+"/blobs/uploads/?mount="+digest+"&from="+from, nil)
+	if err != nil {
+		return false, err
+	}
+	begun, err := r.client().Do(start)
+	if err != nil {
+		return false, fmt.Errorf("cannot reach the registry at %s: %w", r.Address, err)
+	}
+	begun.Body.Close()
+	if begun.StatusCode == http.StatusCreated {
+		return true, nil
+	}
+	if begun.StatusCode == http.StatusAccepted {
+		// Not mounted, and an upload begun in its place: let go of it, so nothing is left half-open.
+		if location := begun.Header.Get("Location"); location != "" {
+			if strings.HasPrefix(location, "/") {
+				location = "http://" + r.Address + location
+			}
+			if cancel, err := http.NewRequestWithContext(ctx, http.MethodDelete, location, nil); err == nil {
+				if done, err := r.client().Do(cancel); err == nil {
+					done.Body.Close()
+				}
+			}
+		}
+		return false, fmt.Errorf("%s did not mount it from %s", r.Address, from)
+	}
+	return false, fmt.Errorf("%s answered %s when asked to mount it from %s", r.Address, begun.Status, from)
+}

internal/builder/lend_test.go

@@ -0,0 +1,140 @@
+package builder
+
+import (
+	"context"
+	"encoding/json"
+	"net/http/httptest"
+	"runtime"
+	"strings"
+	"testing"
+)
+
+// A built image's bases are linked into its repository before it is pushed (novox/hq issue 694): a
+// runtime keeping containerd's image store may not hold a base layer it never had to fetch, and its push
+// asks this machine only for what the repository lacks.
+
+// aBaseForThisMachine puts an index into the registry under `repository`: one manifest for this
+// machine's platform, one for another, and answers the index's digest and each manifest's blobs.
+func aBaseForThisMachine(t *testing.T, dst *aRegistryOfRepositories, repository string) (string, []string, []string) {
+	t.Helper()
+	put := func(arch string) (string, []string) {
+		config := []byte(`{"architecture":"` + arch + `"}`)
+		layer := []byte(arch + " layer bytes")
+		for _, b := range [][]byte{config, layer} {
+			dst.blobs[digestOf(b)] = b
+			dst.link(repository, digestOf(b))
+		}
+		m, _ := json.Marshal(map[string]any{
+			"schemaVersion": 2, "mediaType": mediaManifestOCI,
+			"config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": digestOf(config), "size": len(config)},
+			"layers": []map[string]any{{"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "digest": digestOf(layer), "size": len(layer)}},
+		})
+		dst.manifests[repository+"@"+digestOf(m)] = m
+		return digestOf(m), []string{digestOf(config), digestOf(layer)}
+	}
+	other := "s390x"
+	if runtime.GOARCH == other {
+		other = "amd64"
+	}
+	mine, mineBlobs := put(runtime.GOARCH)
+	theirs, theirBlobs := put(other)
+	index, _ := json.Marshal(map[string]any{
+		"schemaVersion": 2, "mediaType": mediaIndexOCI,
+		"manifests": []map[string]any{
+			{"mediaType": mediaManifestOCI, "digest": mine, "size": 1, "platform": map[string]string{"os": runtime.GOOS, "architecture": runtime.GOARCH}},
+			{"mediaType": mediaManifestOCI, "digest": theirs, "size": 1, "platform": map[string]string{"os": "linux", "architecture": other}},
+		},
+	})
+	dst.manifests[repository+"@"+digestOf(index)] = index
+	return digestOf(index), mineBlobs, theirBlobs
+}
+
+func TestABuildsBaseIsLinkedIntoItsRepositoryBeforeThePush(t *testing.T) {
+	dst := newRegistryOfRepositories()
+	server := httptest.NewServer(dst.handler())
+	defer server.Close()
+	address := strings.TrimPrefix(server.URL, "http://")
+	r := Registry{Address: address}
+
+	const from = "upstream/docker.io/library/golang"
+	index, mine, theirs := aBaseForThisMachine(t, dst, from)
+	bases := []string{
+		address + "/" + from + "@" + index,
+		// Not bases this registry holds: a package version, a tag, an image held nowhere here.
+		"@novox/mesh-sdk@1.2.3",
+		"golang:1-alpine",
+		"elsewhere.invalid/thing@sha256:" + strings.Repeat("e", 64),
+	}
+	n, err := r.LendBases(context.Background(), "mesh-tools-go/build", bases)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Every blob of this machine's platform is in the image's repository, so a push asks this machine's
+	// store for none of them; another platform's are not, since nothing built here stands on them.
+	for _, blob := range mine {
+		if !dst.links["mesh-tools-go/build"][blob] {
+			t.Fatalf("the base's blob %s is not in the image's repository", blob)
+		}
+	}
+	for _, blob := range theirs {
+		if dst.links["mesh-tools-go/build"][blob] {
+			t.Fatalf("another platform's blob %s was linked", blob)
+		}
+	}
+	if n != len(mine) || dst.mounts != len(mine) || dst.uploads != 0 {
+		t.Fatalf("%d linked, %d mounts, %d uploads; want %d mounts and no bytes moved", n, dst.mounts, dst.uploads, len(mine))
+	}
+
+	// Asked again, nothing is mounted twice.
+	if n, err := r.LendBases(context.Background(), "mesh-tools-go/build", bases); err != nil || n != 0 {
+		t.Fatalf("a second lending mounted %d: %v", n, err)
+	}
+}
+
+// A registry that will not mount is said, not hidden: the push is still tried, from the store.
+func TestABaseTheRegistryWillNotMountIsAnError(t *testing.T) {
+	dst := newRegistryOfRepositories()
+	dst.noMount = true
+	server := httptest.NewServer(dst.handler())
+	defer server.Close()
+	address := strings.TrimPrefix(server.URL, "http://")
+	index, _, _ := aBaseForThisMachine(t, dst, "upstream/thing")
+	_, err := Registry{Address: address}.LendBases(context.Background(), "hello/server",
+		[]string{address + "/upstream/thing@" + index})
+	if err == nil || !strings.Contains(err.Error(), "did not mount") {
+		t.Fatalf("a refused mount passed silently: %v", err)
+	}
+}
+
+// lending records what a build lent, and when, against the push.
+type lending struct {
+	*mirroring
+	order []string
+}
+
+func (l *lending) LendBases(_ context.Context, repository string, bases []string) (int, error) {
+	l.order = append(l.order, "lend "+repository+" "+strings.Join(bases, " "))
+	return len(bases), nil
+}
+
+func (l *lending) PublishImage(ctx context.Context, localTag, repository string) (string, error) {
+	l.order = append(l.order, "push "+repository)
+	return l.mirroring.PublishImage(ctx, localTag, repository)
+}
+
+func TestAnImageIsPushedOnlyAfterItsBasesAreLent(t *testing.T) {
+	r, workspace := aRepository(t, anImage, map[string]string{"modules/bus/Dockerfile": "ARG BASE\nFROM ${BASE}"})
+	r.contents["modules/bus/module.json"] = anImage
+	l := &lending{mirroring: &mirroring{recorded: r, at: "127.0.0.1:5000"}}
+	if _, err := Build(context.Background(), r.run, l, "https://forge.invalid/catalogue.git", "modules/bus", "", workspace,
+		nil, Npmrc{}, GitCredential{}, nil); err != nil {
+		t.Fatal(err)
+	}
+	want := []string{
+		"lend bus/server 127.0.0.1:5000/upstream/docker.io/vendor/server@sha256:" + strings.Repeat("1", 64),
+		"push bus/server",
+	}
+	if strings.Join(l.order, "\n") != strings.Join(want, "\n") {
+		t.Fatalf("the build did:\n%s\nwant:\n%s", strings.Join(l.order, "\n"), strings.Join(want, "\n"))
+	}
+}

internal/builder/reuse.go

@@ -257,6 +257,13 @@ func (s scratchPublisher) PublishArchive(ctx context.Context, repository string,
 	return s.to.PublishArchive(ctx, ScratchPrefix+repository, body, digest)
 }
 
+func (s scratchPublisher) LendBases(ctx context.Context, repository string, bases []string) (int, error) {
+	if l, can := s.to.(BaseLender); can {
+		return l.LendBases(ctx, ScratchPrefix+repository, bases)
+	}
+	return 0, nil
+}
+
 func (s scratchPublisher) MirrorImage(ctx context.Context, from, repository string) (string, error) {
 	m, can := s.to.(Mirrorer)
 	if !can {