mesh-controller/internal/builder/registry.go

package builder

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"strings"
)

// Where built artifacts go.
//
// **One store, and it is the registry the bootstrap already pulls from.** An OCI registry is a
// content-addressed blob store that happens to also understand images: `PUT` a blob and it is
// retrievable at `/v2/<name>/blobs/sha256:…` for ever, by digest, over plain HTTP. An archive is
// a content-addressed blob. So it goes there.
//
// The alternative considered was a second store beside it — S3-shaped, buckets, signed URLs. It
// is the right answer for objects that are *mutable*, or need per-reader access, or are not build
// output: somebody's uploads, a backup, a thing with a lifecycle. None of that describes a
// digest-pinned archive, and standing up a second service to hold one kind of immutable blob
// means two things to run, two things to back up and two ways for an artifact to be missing.
//
// **This is a decision that can be overturned by reading**: if something needs an object store
// for reasons other than build artifacts, it should have one, and archives can move to it without
// anything else changing — the manifest carries a URL and a digest, and neither says what served
// it.

// Registry publishes to an OCI registry over plain HTTP.
//
// Plain HTTP because the registry the mesh runs is reached over the private network, which is
// already the encrypted, authenticated thing. A second layer inside it would be certificates to
// issue and rotate for no property the first does not have.
type Registry struct {
	// Address is host:port, as a machine will fetch from.
	Address string
	// Run is how docker is invoked, so a test does not need one.
	Run Runner
	// HTTP is the client used for blobs.
	HTTP *http.Client
}

// PublishImage pushes a locally built image and returns a reference pinned by digest.
//
// The digest is read back from the registry's own answer rather than computed here. What matters
// is what the registry will serve for that reference, and only it can say.
func (r Registry) PublishImage(ctx context.Context, localTag, repository string) (string, error) {
	remote := r.Address + "/" + repository
	if _, err := r.Run(ctx, "", "docker", "tag", localTag, remote); err != nil {
		return "", err
	}
	if _, err := r.Run(ctx, "", "docker", "push", remote); err != nil {
		return "", err
	}
	out, err := r.Run(ctx, "", "docker", "inspect", "--format", "{{index .RepoDigests 0}}", remote)
	if err != nil {
		return "", fmt.Errorf("pushed %s and cannot read back what it was pinned as: %w", remote, err)
	}
	pinned := strings.TrimSpace(out)
	if !strings.Contains(pinned, "@sha256:") {
		// A tag is not a pin. It can be made to point at something else, and this file is applied
		// on machines with no mesh to ask about anything (novox/hq ADR 0006).
		return "", fmt.Errorf("%s came back as %q, which is not pinned by digest", remote, pinned)
	}
	return pinned, nil
}

// PublishArchive stores bytes as a blob and returns where to fetch them from.
//
// Two steps, which is the registry's own protocol: ask for somewhere to put it, then put it there
// naming the digest. The registry verifies the digest itself, so a blob that arrived corrupted is
// refused by the thing storing it rather than by the machine unpacking it a week later.
func (r Registry) PublishArchive(ctx context.Context, repository string, body []byte, digest string) (string, error) {
	base := "http://" + r.Address + "/v2/" + repository
	final := base + "/blobs/" + digest

	// Already there. Blobs are immutable and named by their content, so this is not an
	// optimisation — re-uploading would be asking the registry to store what it already has under
	// the name it already has.
	if there, err := r.has(ctx, final); err != nil {
		return "", err
	} else if there {
		return final, nil
	}

	start, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/blobs/uploads/", nil)
	if err != nil {
		return "", err
	}
	begun, err := r.client().Do(start)
	if err != nil {
		return "", fmt.Errorf("cannot start an upload to %s: %w", base, err)
	}
	defer begun.Body.Close()
	if begun.StatusCode != http.StatusAccepted {
		return "", fmt.Errorf("%s answered %s when asked where to put a blob", base, begun.Status)
	}
	where := begun.Header.Get("Location")
	if where == "" {
		return "", fmt.Errorf("%s accepted an upload and said nowhere to put it", base)
	}
	if strings.HasPrefix(where, "/") {
		where = "http://" + r.Address + where
	}

	put, err := http.NewRequestWithContext(ctx, http.MethodPut,
		where+separator(where)+"digest="+digest, bytes.NewReader(body))
	if err != nil {
		return "", err
	}
	put.Header.Set("Content-Type", "application/octet-stream")
	done, err := r.client().Do(put)
	if err != nil {
		return "", err
	}
	defer done.Body.Close()
	if done.StatusCode != http.StatusCreated {
		said, _ := io.ReadAll(io.LimitReader(done.Body, 4096))
		return "", fmt.Errorf("%s refused the blob: %s %s", base, done.Status, strings.TrimSpace(string(said)))
	}
	return final, nil
}

func (r Registry) has(ctx context.Context, url string) (bool, error) {
	request, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil)
	if err != nil {
		return false, err
	}
	response, err := r.client().Do(request)
	if err != nil {
		return false, fmt.Errorf("cannot reach the registry at %s: %w", r.Address, err)
	}
	defer response.Body.Close()
	return response.StatusCode == http.StatusOK, nil
}

func (r Registry) client() *http.Client {
	if r.HTTP != nil {
		return r.HTTP
	}
	return http.DefaultClient
}

// separator is whether the upload location already carries a query.
func separator(where string) string {
	if strings.Contains(where, "?") {
		return "&"
	}
	return "?"
}