DLV-B3b, DLV-B4 logic: a floor wakes on manifest-accepted, condition-cleared and the next source revision; the seats' outcomes move a version

The builder's one consumer of EVENTS now consumes the indexer's manifest-accepted and the condition keeper's condition-cleared; a heard source revision stands for the forge's merge. Numberer.Heard marks a version published, refused or removed and wakes a floor whose seat was silent; its consumer of each <seat>.events waits on the seat family's stream form (asked of A1). Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 12:18:00 +01:00
parent 688f6c21fa
commit a8bcd2abb9
7 changed files with 202 additions and 4 deletions
@@ -141,7 +141,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
 	// cannot be read the trunk's keys are built and the candidates wait (issue 1945).
 	errs := make(chan error, 6)
-	trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf}
+	trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: func(ctx context.Context) {
+		// A source revision heard is the forge's merge: it wakes a floor whose bridge_tags was silent (design 51 *The floor*).
+		(&floorWake{b: b, waits: floorWaits, logf: logger.Printf}).wake(ctx, versions.WakeForge, "", "")
+		b.TrunkMoved(ctx)
+	}, errs: errs, logf: logger.Printf}
 	trunks.start()
 	b.Trunk = trunks.head
 	// The builder's log (ADR 0571 §1): each build's start and end in its module log `build-log`, read by the view, the
@@ -210,7 +214,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v",
 			cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err)
 	default:
-		cc, err := process.HearEvents(cons, func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) }, logger.Printf)
+		// The same consumer carries what wakes a floor (design 51 *The floor*): the indexer's manifest-accepted and the
+		// condition keeper's condition-cleared.
+		floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf}
+		cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
+			if floor.heard(ctx, m) {
+				return
+			}
+			heardGone(ctx, b, m, logger.Printf)
+		}, logger.Printf)
 		if err != nil {
 			return err
 		}
@@ -7,6 +7,7 @@ import (
 	"log"
 	"strings"
 	"testing"
+	"time"

 	"git.novox.be/novox/mesh-sdk/go/process"

@@ -14,6 +15,9 @@ import (

 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-builder/internal/build"
+	"mesh-delivery-builder/internal/versions"
 )

 // 1945 (a part exits at start for a bucket it works without): a bus that refuses every toolchains bucket leaves the
@@ -21,6 +25,7 @@ import (
 // its grant. The gate over the seat's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate,
 // issue 774).
 func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) {
+	t.Run("a floor's wake is heard on the events consumer (design 51 The floor)", aFloorsWakeIsHeard)
 	set := &buildrun.ToolchainSet{}
 	var logs bytes.Buffer
 	logger := log.New(&logs, "", 0)
@@ -71,3 +76,35 @@ func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstrea
 func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
 	return nil, r.err
 }
+
+// wakeMsg is a message of the builder's events consumer: its subject and body, acknowledged.
+type wakeMsg struct {
+	jetstream.Msg
+	subject string
+	data    []byte
+	acked   bool
+}
+
+func (m *wakeMsg) Subject() string { return m.subject }
+func (m *wakeMsg) Data() []byte    { return m.data }
+func (m *wakeMsg) Ack() error      { m.acked = true; return nil }
+
+func aFloorsWakeIsHeard(t *testing.T) {
+	var logs bytes.Buffer
+	waits := &versions.FloorWaits{}
+	waits.Wait("m/lib", versions.SeatGoPackages, "", versions.Floor{Silent: []string{versions.SourceRequirements}}, nil, time.Unix(0, 0))
+	waits.Wait("m/img", versions.SeatImages, "", versions.Floor{Silent: []string{versions.SourceTags}}, nil, time.Unix(0, 0))
+	f := &floorWake{b: &build.Builder{}, waits: waits, logf: log.New(&logs, "", 0).Printf}
+	accepted := &wakeMsg{subject: manifestAcceptedSubject}
+	if !f.heard(context.Background(), accepted) || !accepted.acked || !strings.Contains(logs.String(), "the floor of m/lib is asked again: manifest-accepted") {
+		t.Fatalf("manifest-accepted: acked %v, log %q", accepted.acked, logs.String())
+	}
+	logs.Reset()
+	cleared := &wakeMsg{subject: conditionClearedSubject, data: []byte(`{"subject":{"scope":"probe","id":"ace/mesh-delivery-forge-bridge"}}`)}
+	if !f.heard(context.Background(), cleared) || !strings.Contains(logs.String(), "the floor of m/img is asked again: condition-cleared") {
+		t.Fatalf("condition-cleared: log %q", logs.String())
+	}
+	if f.heard(context.Background(), &wakeMsg{subject: delivery.ArtifactsGoneSubject}) {
+		t.Fatal("artifacts-gone was taken as a floor's wake")
+	}
+}
@@ -5,13 +5,17 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"

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

+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/health"
 	"git.novox.be/novox/mesh-sdk/go/process"

+	"mesh-delivery-builder/internal/build"
 	"mesh-delivery-builder/internal/versions"
 )

@@ -174,3 +178,61 @@ func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account
 	_, err = js.PublishMsg(pub, msg, jetstream.WithMsgID(id))
 	return err
 }
+
+// The events on the builder's consumer of EVENTS that wake a floor (design 51 *The floor*, DLV-B3b).
+var (
+	manifestAcceptedSubject = "mesh.mod." + delivery.IndexerModule + ".event." + delivery.ManifestAcceptedEvent
+	conditionClearedSubject = "mesh.mod." + health.KeeperModule + ".event." + health.EventCleared
+)
+
+// floorWake asks the floor of the names an event wakes again, by looking at every module again: a trunk build is
+// numbered before its claim (build.Builder.number).
+type floorWake struct {
+	b     *build.Builder
+	waits *versions.FloorWaits
+	logf  func(string, ...any)
+}
+
+// heard takes a message of the consumer in when it is one that wakes a floor, acknowledged, and answers whether it was.
+func (f *floorWake) heard(ctx context.Context, m jetstream.Msg) bool {
+	switch m.Subject() {
+	case manifestAcceptedSubject:
+		f.ack(m)
+		f.wake(ctx, versions.WakeAccepted, "", "")
+	case conditionClearedSubject:
+		f.ack(m)
+		var e health.Event
+		if err := json.Unmarshal(m.Data(), &e); err != nil {
+			f.logf("%s holds what is no condition, left: %v", m.Subject(), err)
+			return true
+		}
+		for _, module := range conditionModules(e.Subject.ID) {
+			f.wake(ctx, versions.WakeCleared, "", module)
+		}
+	default:
+		return false
+	}
+	return true
+}
+
+func (f *floorWake) ack(m jetstream.Msg) {
+	if err := m.Ack(); err != nil {
+		f.logf("acknowledging %s: %v", m.Subject(), err)
+	}
+}
+
+// wake looks at every module again when the event wakes a name, and says which.
+func (f *floorWake) wake(ctx context.Context, kind, seat, module string) {
+	names := f.waits.Wake(kind, seat, module, "")
+	if len(names) == 0 {
+		return
+	}
+	f.logf("the floor of %s is asked again: %s", strings.Join(names, ", "), kind)
+	f.b.Reconsider(ctx)
+}
+
+// conditionModules are the tokens of a condition's subject id (`<node>/<module>`, `<module>.<check>`): a module named
+// among them is the one whose condition cleared.
+func conditionModules(id string) []string {
+	return strings.FieldsFunc(id, func(r rune) bool { return r == '/' || r == '.' || r == ':' })
+}
@@ -0,0 +1,63 @@
+package versions
+
+import (
+	"context"
+	"fmt"
+)
+
+// The registry seats' events the builder hears on its durable consumer of each `<seat>.events` (design 51 *How a
+// publish ends*, DLV-B4).
+const (
+	HeardPublished = "published"
+	HeardRefused   = "refused"
+	HeardRemoved   = "removed"
+)
+
+// Outcome is the body of a registry seat's published, refused and removed (mesh-contracts parts/registry Outcome), the
+// fields the builder reads.
+type Outcome struct {
+	Name    string `json:"name"`
+	Version string `json:"version,omitempty"`
+	Digest  string `json:"digest"`
+	Kind    string `json:"kind"`
+	Holder  string `json:"holder"`
+	Reason  string `json:"reason,omitempty"`
+}
+
+// Heard takes one seat event in: a numbered version's entry turns published (with its digest when it held none),
+// refused with the holder's reason, or removed with its reason and the event's id; a by-digest publish (no version)
+// marks nothing. It answers the names waiting on their floor that the event wakes: a `published` says the seat
+// answers (design 51 *The floor*).
+func (n *Numberer) Heard(ctx context.Context, seat, event, id string, o Outcome) ([]string, error) {
+	var wake []string
+	if event == HeardPublished && n.Waits != nil {
+		wake = n.Waits.Wake(WakePublished, seat, "", "")
+	}
+	if o.Version == "" {
+		return wake, nil
+	}
+	var change func(*Entry)
+	switch event {
+	case HeardPublished:
+		change = func(e *Entry) {
+			e.State, e.Reason = Published, ""
+			if e.Digest == "" {
+				e.Digest = o.Digest
+			}
+		}
+	case HeardRefused:
+		change = func(e *Entry) {
+			if e.State != Published {
+				e.State, e.Reason = Refused, o.Reason
+			}
+		}
+	case HeardRemoved:
+		change = func(e *Entry) { e.State, e.Reason, e.RemovedBy = Removed, o.Reason, id }
+	default:
+		return wake, fmt.Errorf("%s says %q, which the builder does not hear", seat, event)
+	}
+	if _, err := n.Versions.Mark(ctx, o.Name, o.Version, change); err != nil {
+		return wake, fmt.Errorf("%s's %s of %s@%s: %w", seat, event, o.Name, o.Version, err)
+	}
+	return wake, nil
+}
@@ -365,6 +365,26 @@ func TestTheBuildersVersions(t *testing.T) {
 		if _, ok, why, _ := n.Versions.Recorded(ctx, "m/program", "0.1.14", "sha256:bb"); ok {
 			t.Fatalf("another digest recorded: %s", why)
 		}
+		// DLV-B4: the seat's events move the entry; a published wakes the names waiting on that seat.
+		if wake, err := n.Heard(ctx, SeatArtifacts, HeardPublished, "e1", Outcome{Name: "m/other", Digest: "sha256:cc"}); err != nil || len(wake) != 0 {
+			t.Fatalf("a by-digest publish of another seat's wait: %v, %v", wake, err)
+		}
+		n.Waits.Wait("m/lib", SeatGoPackages, "", Floor{Silent: []string{SourceSeat}}, nil, time.Unix(0, 0))
+		if wake, _ := n.Heard(ctx, SeatGoPackages, HeardPublished, "e2", Outcome{Name: "m/x", Digest: "sha256:dd"}); !reflect.DeepEqual(wake, []string{"m/lib"}) {
+			t.Fatalf("a published on the Go seat woke %v, want m/lib, whose seat was silent", wake)
+		}
+		for _, h := range []struct{ event, state, reason string }{{HeardRefused, Refused, "bytes differ"}, {HeardPublished, Published, ""}, {HeardRefused, Published, ""}, {HeardRemoved, Removed, "retention"}} {
+			if _, err := n.Heard(ctx, SeatArtifacts, h.event, "e3", Outcome{Name: "m/program", Version: "0.1.14", Digest: "sha256:aa", Reason: h.reason}); err != nil {
+				t.Fatal(err)
+			}
+			all, _ := n.Versions.Of(ctx, "m/program")
+			if all[0].State != h.state {
+				t.Fatalf("after %s: %+v, want %s", h.event, all[0], h.state)
+			}
+		}
+		if _, err := n.Heard(ctx, SeatArtifacts, HeardPublished, "e4", Outcome{Name: "m/none", Version: "0.1.0"}); err == nil {
+			t.Fatal("a published of a version the builder never numbered is not said")
+		}
 	})
 	t.Run("fill", func(t *testing.T) {
 		if _, full := Fill(Cap*79/100, Cap); full {
@@ -116,7 +116,9 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	// The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it
 	// starts; and by the seat mesh-artifact-sweep's artifacts-gone, a blob the sweep found gone (1937/AC4, 1955/L7), the
 	// controller's no longer (1126/B9, ADR 0590).
-	if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
+	// Design 51 *The floor*: the indexer's manifest-accepted and the condition keeper's condition-cleared wake a silent floor.
+	if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported",
+		"mesh-conditions.condition-cleared", "mesh-delivery-indexer.manifest-accepted"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
 		t.Errorf("consumes = %v, want %v", m.Consumes, want)
 	}
 	if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
@@ -19,7 +19,9 @@
  "consumes": [
    "mesh-artifact-sweep.artifacts-gone",
    "mesh-build-go.toolchain.reported",
-    "mesh-build-node.toolchain.reported"
+    "mesh-build-node.toolchain.reported",
+    "mesh-conditions.condition-cleared",
+    "mesh-delivery-indexer.manifest-accepted"
  ],
  "invokes": [
    "mesh-delivery-forge-bridge.bridge_tags",