mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go

//go:build integration

package main

import (
	"encoding/json"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// A done key whose blob the artifact store no longer holds is built again, by the builder whole against the bus
// (novox/hq issue 1937): on 2026-10-24 the store's 03:30 collection deleted the bare blobs of three done keys the
// controller had never taken in, and the builder kept them done.

// heldVerb answers the controller seat's verb held on the bus, as the serving controller does: {result: {output, ok,
// answer}}; a reference is gone while gone says so.
type heldVerb struct {
	mu    sync.Mutex
	gone  func(reference string) bool
	asked []string
}

func (h *heldVerb) serve(t *testing.T, b *bed) {
	t.Helper()
	sub, err := b.js.Conn().Subscribe("mesh.seat.mesh-controller.call.held.>", func(m *nats.Msg) {
		var args struct {
			References []string `json:"references"`
		}
		_ = json.Unmarshal(m.Data, &args)
		a := delivery.HeldAnswer{Store: "store.test:5100", Held: []string{}, Gone: []string{}}
		h.mu.Lock()
		for _, r := range args.References {
			h.asked = append(h.asked, r)
			if h.gone(r) {
				a.Gone = append(a.Gone, r)
			} else {
				a.Held = append(a.Held, r)
			}
		}
		h.mu.Unlock()
		reply, _ := json.Marshal(map[string]any{"result": map[string]any{"output": "held", "ok": true, "answer": a}, "node": "ctl"})
		_ = m.Respond(reply)
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
}

func (h *heldVerb) asks() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.asked)
}

// eventsConsumer makes EVENTS and the builder's durable consumer of artifacts-gone on it, as the controller makes them
// from the manifest's consumes.
func eventsConsumer(t *testing.T, b *bed, node string) {
	t.Helper()
	if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
		t.Fatal(err)
	}
	if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder",
		AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
		t.Fatal(err)
	}
}

// doneKey writes builds.<module>.<fp> done, naming one archive at reference, and answers its revision.
func doneKey(t *testing.T, b *bed, module, manifest, reference, digest string) uint64 {
	t.Helper()
	key, _ := delivery.BuilderBuildKey(module, fp)
	data, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "n2.mesh-delivery-builder",
		Manifest: manifest, Step: "built mesh-build-go on n2",
		Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Digest: digest, Reference: reference}}})
	if err != nil {
		t.Fatal(err)
	}
	rev, err := b.builds.Put(b.ctx, key, data)
	if err != nil {
		t.Fatal(err)
	}
	return rev
}

func revisionOf(t *testing.T, b *bed, module string) uint64 {
	t.Helper()
	key, _ := delivery.BuilderBuildKey(module, fp)
	e, err := b.builds.Get(b.ctx, key)
	if err != nil {
		t.Fatal(err)
	}
	return e.Revision()
}

// 1937/AC3 (integration): the builder, started whole, finds the done key of the trunk's fingerprint naming a blob held
// answers gone, builds it again as a new revision whose step says so, and leaves the rebuilt key done.
func TestADoneKeyWhoseBlobIsGoneIsBuiltAgainAtStart(t *testing.T) {
	b := startBed(t)
	eventsConsumer(t, b, "n1")
	manifest := b.archive(t, "ca-trust", "modules/ca-trust", commit)
	old := digestOf("collected last night")
	oldRef := "artifact-store://ca-trust/files/blobs/" + old
	rev := doneKey(t, b, "ca-trust", manifest, oldRef, old)
	h := &heldVerb{gone: func(r string) bool { return r == oldRef }}
	h.serve(t, b)

	b.start(t, "n1", nil)
	v := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
		return v.State == delivery.BuildDone && len(v.Artifacts) == 1 && v.Artifacts[0].Digest == digestOf("files")
	})
	if !strings.Contains(v.Step, "built again: the artifact store no longer held "+oldRef) {
		t.Errorf("the rebuilt key's step %q does not say why it was built again", v.Step)
	}
	if got := revisionOf(t, b, "ca-trust"); got <= rev {
		t.Errorf("revision %d, was %d: want a new revision", got, rev)
	}
	if n := b.fake.count("mesh-build-go.build_go_build"); n != 1 {
		t.Errorf("the archive was built %d times, want once", n)
	}
	if h.asks() == 0 {
		t.Error("the builder never asked held")
	}
}

// 1937/AC4 (integration): a done key whose blob held answers held at start stays done; when the controller says
// artifacts-gone for it, the builder asks again and builds it again.
func TestArtifactsGoneLeadsTheBuilderToBuildTheKeyAgain(t *testing.T) {
	b := startBed(t)
	eventsConsumer(t, b, "n1")
	manifest := b.archive(t, "ca-trust", "modules/ca-trust", commit)
	d := digestOf("files")
	ref := "artifact-store://ca-trust/files/blobs/" + d
	rev := doneKey(t, b, "ca-trust", manifest, ref, d)
	var mu sync.Mutex
	collected := false
	h := &heldVerb{gone: func(r string) bool {
		mu.Lock()
		defer mu.Unlock()
		if r == ref && collected {
			collected = false // gone once: the rebuild pushes it again
			return true
		}
		return false
	}}
	h.serve(t, b)

	b.start(t, "n1", nil)
	for deadline := time.Now().Add(20 * time.Second); h.asks() == 0 && time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
	}
	if h.asks() == 0 {
		t.Fatal("the builder never asked held at its start")
	}
	time.Sleep(300 * time.Millisecond)
	if got := revisionOf(t, b, "ca-trust"); got != rev || b.fake.count("mesh-build-go.build_go_build") != 0 {
		t.Fatalf("a key whose blob is held was built again (revision %d, was %d)", got, rev)
	}

	mu.Lock()
	collected = true
	mu.Unlock()
	key, _ := delivery.BuilderBuildKey("ca-trust", fp)
	body, err := delivery.EncodeArtifactsGone(delivery.ArtifactsGone{Module: "ca-trust", Key: key, Fingerprint: fp,
		Gone: []string{ref}, At: time.Now().UTC()})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := b.js.Publish(b.ctx, delivery.ArtifactsGoneSubject, body); err != nil {
		t.Fatal(err)
	}
	v := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
		return v.State == delivery.BuildDone && strings.Contains(v.Step, "built again")
	})
	if got := revisionOf(t, b, "ca-trust"); got <= rev || b.fake.count("mesh-build-go.build_go_build") != 1 {
		t.Errorf("build = %+v at revision %d (was %d), %d builds: want built again once on the event", v, got, rev,
			b.fake.count("mesh-build-go.build_go_build"))
	}
}