mesh-controller/internal/artifacts/live_registry_test.go

package artifacts

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"strings"
	"testing"
	"time"

	"github.com/novox/mesh-controller/internal/catalogue"
)

// The registry's own collector keeps a held archive and takes a bare one (novox/hq issue 253,
// ADR 0189).
//
// Everything else here is asserted against a fake, which can only say what this side asks. This is
// the one question a fake cannot answer — what `registry garbage-collect` actually does with what
// this side wrote — and it is the whole of whether the store's nightly step may stop being a dry
// run. Against the very image the mesh's store runs:
//
//	docker run -d --rm --name mesh-controller-registry -p 15000:5000 \
//	    -e REGISTRY_STORAGE_DELETE_ENABLED=true registry:2.8.3
//	MESH_TEST_REGISTRY=127.0.0.1:15000 MESH_TEST_REGISTRY_CONTAINER=mesh-controller-registry \
//	    go test -run Live ./internal/artifacts/
//	docker stop mesh-controller-registry
//
// Skipped without both variables: it needs a registry it may write to and collect, and a container
// to run the collector in.
func TestLiveTheRegistrysCollectorKeepsWhatIsHeldAndTakesWhatIsNot(t *testing.T) {
	address := os.Getenv("MESH_TEST_REGISTRY")
	container := os.Getenv("MESH_TEST_REGISTRY_CONTAINER")
	if address == "" || container == "" {
		t.Skip("no MESH_TEST_REGISTRY / MESH_TEST_REGISTRY_CONTAINER; see this test's comment for the registry to raise")
	}
	ctx := context.Background()
	store := Store{Address: address}
	run := time.Now().UnixNano()

	// Four archives in four repositories, each a different story. Distinct bytes per run, so a
	// registry reused across runs cannot answer for an earlier one.
	put := func(name string) (repository, digest string, body []byte) {
		repository = fmt.Sprintf("live-%d/%s", run, name)
		body = []byte(fmt.Sprintf("%s archive of run %d", name, run))
		digest = digestOf(body)
		if err := store.putBlob(ctx, repository, digest, body); err != nil {
			t.Fatal(err)
		}
		return repository, digest, body
	}
	reference := func(repository, digest string) string {
		return catalogue.ArtifactStoreScheme + repository + "/blobs/" + digest
	}

	// Published held — what PublishArchive now does.
	heldRepo, heldDigest, heldBody := put("held")
	if _, err := store.HoldBlob(ctx, heldRepo, heldDigest, int64(len(heldBody))); err != nil {
		t.Fatalf("the registry refused a holder: %v", err)
	}
	// Published bare, as before, and never held: what the collector must take.
	_, bareDigest, _ := put("bare")
	// Published bare and then held by the sweep: the backfill.
	backRepo, backDigest, backBody := put("backfilled")
	if wrote, err := store.Hold(ctx, reference(backRepo, backDigest)); err != nil || !wrote {
		t.Fatalf("backfilling wrote=%v: %v", wrote, err)
	}
	if held, err := store.Held(ctx, reference(backRepo, backDigest)); err != nil || !held {
		t.Fatalf("after backfilling, held=%v: %v", held, err)
	}
	// Held, and then let go of by the sweep: holder first, then the link.
	goneRepo, goneDigest, _ := put("let-go")
	if _, err := store.Hold(ctx, reference(goneRepo, goneDigest)); err != nil {
		t.Fatal(err)
	}
	if err := store.LetGo(ctx, reference(goneRepo, goneDigest)); err != nil {
		t.Fatalf("letting go of a held archive: %v", err)
	}

	collected, err := exec.CommandContext(ctx, "docker", "exec", container,
		"registry", "garbage-collect", "/etc/docker/registry/config.yml").CombinedOutput()
	if err != nil {
		t.Fatalf("the collector failed: %v\n%s", err, collected)
	}
	t.Logf("the collector said:\n%s", lastLines(string(collected), 12))

	// What is asserted is the bytes on the store's disk, not what the running server answers: the
	// server caches blob descriptors in memory and can answer for a blob the collector removed.
	onDisk := func(digest string) bool {
		hex := strings.TrimPrefix(digest, "sha256:")
		path := "/var/lib/registry/docker/registry/v2/blobs/sha256/" + hex[:2] + "/" + hex + "/data"
		return exec.CommandContext(ctx, "docker", "exec", container, "test", "-f", path).Run() == nil
	}
	if !onDisk(heldDigest) {
		t.Error("the collector took an archive published held")
	}
	if !onDisk(backDigest) {
		t.Error("the collector took an archive the sweep backfilled a holder for")
	}
	if onDisk(bareDigest) {
		t.Error("the collector kept a bare archive — then the holders prove nothing, and this test is wrong")
	}
	if onDisk(goneDigest) {
		t.Error("the collector kept an archive the sweep let go of: its holder outlived its link")
	}
	// And what survived is still fetched exactly as machines fetch it: the blob, by digest.
	for repository, want := range map[string][]byte{heldRepo: heldBody, backRepo: backBody} {
		digest := digestOf(want)
		response, err := http.Get(catalogue.Routed(reference(repository, digest), address))
		if err != nil {
			t.Fatal(err)
		}
		got, _ := io.ReadAll(response.Body)
		response.Body.Close()
		if response.StatusCode != http.StatusOK || !bytes.Equal(got, want) {
			t.Errorf("%s answered %s with %q after collection", repository, response.Status, got)
		}
	}
}

func lastLines(s string, n int) string {
	lines := strings.Split(strings.TrimSpace(s), "\n")
	if len(lines) > n {
		lines = lines[len(lines)-n:]
	}
	return strings.Join(lines, "\n")
}