From a8bcd2abb980e590698aa0c65e33de59c7a67d84 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:18:00 +0100 Subject: [PATCH] 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 .events waits on the seat family's stream form (asked of A1). Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-delivery-builder/main.go | 16 ++++- .../cmd/mesh-delivery-builder/refused_test.go | 37 +++++++++++ .../cmd/mesh-delivery-builder/versions.go | 62 ++++++++++++++++++ .../internal/versions/heard.go | 63 +++++++++++++++++++ .../internal/versions/versions_test.go | 20 ++++++ .../mesh-delivery-builder/manifest_test.go | 4 +- modules/mesh-delivery-builder/module.json | 4 +- 7 files changed, 202 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-delivery-builder/internal/versions/heard.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index b9f38b52..7bcbe48e 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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 } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index 08d5fb7c..16305538 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -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") + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go index 44a53732..5d3b7706 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go @@ -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 (`/`, `.`): 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 == ':' }) +} diff --git a/modules/mesh-delivery-builder/internal/versions/heard.go b/modules/mesh-delivery-builder/internal/versions/heard.go new file mode 100644 index 00000000..e4cb830b --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/heard.go @@ -0,0 +1,63 @@ +package versions + +import ( + "context" + "fmt" +) + +// The registry seats' events the builder hears on its durable consumer of each `.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 +} diff --git a/modules/mesh-delivery-builder/internal/versions/versions_test.go b/modules/mesh-delivery-builder/internal/versions/versions_test.go index 712ec98d..42fe6254 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions_test.go +++ b/modules/mesh-delivery-builder/internal/versions/versions_test.go @@ -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 { diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 067e3c7a..99e63787 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -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" { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 0aa11ed2..0c960725 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -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",