Record the bases a build copies, keep them by the builds that stood on them, and copy each image once

A copied base was named only in what a build stood on, and nowhere when the build failed,
so the store's sweep could never let one go (hq issue 321). One repository per upstream
image stops each module asking the public registry for the same image again, and letting
an index go now takes its own platform manifests, which otherwise kept every byte. A
person can record the copies no record names through the new mirrors verb (hq ADR 0257).

The forge test fix is the same commit as on feat/plain-notifications: main fails without it.

internal/builder/mirror.go

@@ -27,6 +27,40 @@ type Mirrorer interface {
 	MirrorImage(ctx context.Context, from, repository string) (string, error)
 }
 
+// BaseMirrorer copies a base a build stands on into the one repository the mesh keeps for that
+// upstream image, taking what an earlier copy under `former` already holds rather than asking
+// upstream for it again (novox/hq ADR 0257).
+type BaseMirrorer interface {
+	MirrorBase(ctx context.Context, from, repository, former string) (string, error)
+}
+
+// MirrorPrefix is the namespace of the repositories a base is mirrored into: `upstream/<host>/<path>`,
+// one repository per upstream image, whichever modules stand on it (novox/hq ADR 0257).
+const MirrorPrefix = "upstream/"
+
+// MirrorRepository is the repository the mesh keeps an upstream image's copy in: the image's own
+// host and path under MirrorPrefix, so two modules standing on one image stand on one copy, and two
+// images that share a path on different hosts are never confused. Lower case and without a port's
+// colon, because a repository name allows neither.
+func MirrorRepository(from string) (string, error) {
+	where, err := parseReference(from)
+	if err != nil {
+		return "", err
+	}
+	host := strings.TrimPrefix(strings.TrimPrefix(where.base, "https://"), "http://")
+	if host == "registry-1.docker.io" {
+		host = "docker.io"
+	}
+	host = strings.ReplaceAll(host, ":", "-")
+	return strings.ToLower(MirrorPrefix + host + "/" + where.repository), nil
+}
+
+// FormerMirrorRepository is where a module's base was copied before ADR 0257: under the module's own
+// repository, one copy per module (ADR 0097). Read only, as a source of what is already held.
+func FormerMirrorRepository(module, arg string) string {
+	return module + "/on-" + strings.ToLower(arg)
+}
+
 const (
 	mediaIndexOCI       = "application/vnd.oci.image.index.v1+json"
 	mediaIndexDocker    = "application/vnd.docker.distribution.manifest.list.v2+json"
@@ -86,6 +120,9 @@ func parseReference(ref string) (upstream, error) {
 type source struct {
 	client *http.Client
 	token  string
+	// mountFrom is a repository of the mesh's own registry the source is, so a blob is mounted from
+	// it rather than read and written again.
+	mountFrom string
 }
 
 // get fetches a registry URL, answering a bearer challenge once with an anonymous token — which is
@@ -176,6 +213,15 @@ type descriptor struct {
 // under `repository`, and returns the reference the mesh will pin: this registry, the repository,
 // and the digest of the document that was put last, which is the index where there is one.
 func (r Registry) MirrorImage(ctx context.Context, from, repository string) (string, error) {
+	return r.MirrorBase(ctx, from, repository, "")
+}
+
+// MirrorBase is MirrorImage, and where `former` — a repository of this registry — already holds the
+// image by its digest, the copy is made from there: manifests read from this registry, blobs mounted
+// across rather than moved (novox/hq ADR 0257). Upstream is asked only for what no repository here
+// holds, which is what kept the public hub's anonymous pull limit out of reach when every module's
+// base moved into one shared repository.
+func (r Registry) MirrorBase(ctx context.Context, from, repository, former string) (string, error) {
 	where, err := parseReference(from)
 	if err != nil {
 		return "", err
@@ -195,6 +241,17 @@ func (r Registry) MirrorImage(ctx context.Context, from, repository string) (str
 		}
 	}
 	src := &source{client: r.client()}
+	if former != "" && former != repository && strings.HasPrefix(where.reference, "sha256:") {
+		held, err := r.has(ctx, "http://"+r.Address+"/v2/"+former+"/manifests/"+where.reference, manifestAccept)
+		if err != nil {
+			return "", fmt.Errorf("asking %s whether %s holds %s: %w", r.Address, former, from, err)
+		}
+		if held {
+			// The same bytes, already here: copied from this registry, each blob mounted.
+			where = upstream{base: "http://" + r.Address, repository: former, reference: where.reference}
+			src.mountFrom = former
+		}
+	}
 	digest, err := r.copyManifest(ctx, src, where, where.reference, repository)
 	if err != nil {
 		return "", fmt.Errorf("copying %s into %s/%s: %w", from, r.Address, repository, err)
@@ -286,16 +343,14 @@ func (r Registry) copyBlob(ctx context.Context, src *source, where upstream, dig
 	} else if there {
 		return nil
 	}
-	response, err := src.get(ctx, where.base+"/v2/"+where.repository+"/blobs/"+digest, "")
-	if err != nil {
-		return err
+	// **Mounted where this registry already holds it** (novox/hq ADR 0257): a blob is stored once
+	// whichever repositories link it, so a mount moves no bytes. A registry that cannot mount answers
+	// with an ordinary upload's location, and the blob is moved as before.
+	uploads := base + "/blobs/uploads/"
+	if src.mountFrom != "" {
+		uploads += "?mount=" + digest + "&from=" + src.mountFrom
 	}
-	defer response.Body.Close()
-	if response.StatusCode != http.StatusOK {
-		return fmt.Errorf("%s/%s: blob %s: %s", where.base, where.repository, digest, response.Status)
-	}
-
-	start, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/blobs/uploads/", nil)
+	start, err := http.NewRequestWithContext(ctx, http.MethodPost, uploads, nil)
 	if err != nil {
 		return err
 	}
@@ -304,9 +359,21 @@ func (r Registry) copyBlob(ctx context.Context, src *source, where upstream, dig
 		return fmt.Errorf("cannot start an upload to %s: %w", base, err)
 	}
 	begun.Body.Close()
+	if begun.StatusCode == http.StatusCreated && src.mountFrom != "" {
+		return nil
+	}
 	if begun.StatusCode != http.StatusAccepted {
 		return fmt.Errorf("%s answered %s when asked where to put a blob", base, begun.Status)
 	}
+
+	response, err := src.get(ctx, where.base+"/v2/"+where.repository+"/blobs/"+digest, "")
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return fmt.Errorf("%s/%s: blob %s: %s", where.base, where.repository, digest, response.Status)
+	}
 	location := begun.Header.Get("Location")
 	if location == "" {
 		return fmt.Errorf("%s accepted an upload and said nowhere to put it", base)