mesh-controller/internal/builder/lend.go

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)
}