From 86fd0bbe96c5062acd990c413a208ed7c35de10d Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 16:47:09 +0200 Subject: [PATCH] The view says its event delivered once for each source revision whose wants and haves are all there (issue 1906 D9, for issue 1946) --- .../mesh-delivery-view/integration_test.go | 95 +++++++++++++++++++ .../cmd/mesh-delivery-view/main.go | 30 ++++++ .../mesh-delivery-view/internal/view/view.go | 92 +++++++++++++----- .../internal/view/wants_test.go | 34 +++++++ modules/mesh-delivery-view/manifest_test.go | 12 ++- modules/mesh-delivery-view/module.json | 3 + 6 files changed, 243 insertions(+), 23 deletions(-) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index e9245f0..bd400b1 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -696,3 +696,98 @@ func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC } t.Fatalf("the backlog never showed three waiting asks with the oldest's age; last: %+v", last.Backlog) } + +// 1906/D9, whole on the bus: when the wants and the haves make the source revision delivered, the view says it once, as its event +// `delivered` with the revision, the group and the heads, which the issue tracker derives "fix delivered" from. +func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "VIEWEVENTS", Subjects: []string{"mesh.mod.mesh-delivery-view.event.>"}}) + if err != nil { + t.Fatal(err) + } + kvs := map[string]jetstream.KeyValue{} + for _, name := range []string{delivery.SourceBucket, delivery.ModulesBucket, delivery.AssignerWantsBucket, delivery.HaveBucket} { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5}) + if err != nil { + t.Fatal(err) + } + kvs[name] = kv + } + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + broker := filepath.Join(dir, "broker") + cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": filepath.Join(dir, "run", "cli.sock"), "MESH_DELIVERY_WEB": "127.0.0.1:0"} + runCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() + t.Cleanup(func() { stop(); <-done }) + + fp := "src2:" + strings.Repeat("2", 64) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: strings.Repeat("a", 64), Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) + want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Composed: "sha256:" + strings.Repeat("c", 64)}) + failed, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed}) + good, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + src, _ := delivery.EncodeSource(delivery.Source{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) + for bucket, kv := range map[string][]byte{delivery.ModulesBucket: mod} { + if _, err := kvs[bucket].Put(ctx, "web", kv); err != nil { + t.Fatal(err) + } + } + if _, err := kvs[delivery.AssignerWantsBucket].Put(ctx, "want.ace.web", want); err != nil { + t.Fatal(err) + } + if _, err := kvs[delivery.HaveBucket].Put(ctx, "ace.web", failed); err != nil { + t.Fatal(err) + } + rev, err := kvs[delivery.SourceBucket].Put(ctx, delivery.SourceMainKey, src) + if err != nil { + t.Fatal(err) + } + time.Sleep(time.Second) + if info, err := events.Info(ctx); err != nil || info.State.Msgs != 0 { + t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State) + } + if _, err := kvs[delivery.HaveBucket].Put(ctx, "ace.web", good); err != nil { + t.Fatal(err) + } + var msg *jetstream.RawStreamMsg + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && msg == nil; time.Sleep(100 * time.Millisecond) { + msg, _ = events.GetLastMsgForSubject(ctx, "mesh.mod.mesh-delivery-view.event."+DeliveredEventType) + } + if msg == nil { + t.Fatal("the delivery of the source revision was never said") + } + var e view.DeliveredEvent + if err := json.Unmarshal(msg.Data, &e); err != nil || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) { + t.Errorf("the event = %+v, %v; want revision %d, group feat/x and the heads", e, err, rev) + } + if msg.Header.Get("x-source") != "mesh-delivery-view" || msg.Header.Get("x-event-id") == "" { + t.Errorf("the headers = %v", msg.Header) + } +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index 0bd581e..0499a6c 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -138,6 +138,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } model.SetWriter(kvWriter{ctx: ctx, js: js}) + model.SetDeliveredEvent(func(e view.DeliveredEvent) { + if err := emitEvent(ctx, js, cred, DeliveredEventType, e, fmt.Sprintf("delivered.%d", e.Revision)); err != nil { + logger.Printf("the delivery of source revision %d could not be said: %v", e.Revision, err) + } + }) model.SetBacklog(view.BacklogOver(js, 5*time.Second)) for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} { go follow(ctx, js, bucket, model, logger) @@ -211,3 +216,28 @@ func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model } } } + +// DeliveredEventType is the event the view says when a source revision is delivered: mesh.mod.mesh-delivery-view.event.delivered +// (issue 1906 D9, ADR 0539 §17). Its body is view.DeliveredEvent. +const DeliveredEventType = "delivered" + +// emitEvent publishes an event of the view's own namespace, mesh.mod..event., with the SDK's envelope headers +// (as mesh-sdk's process.Conn.Emit does) and the id as the bus's own dedup id, so the same revision said twice in the stream's +// duplicate window is kept once. +func emitEvent(ctx context.Context, js jetstream.JetStream, cred credential, eventType string, body any, id string) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + eventType) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", cred.Module) + msg.Header.Set("x-node", cred.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + pub, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + _, err = js.PublishMsg(pub, msg, jetstream.WithMsgID(id)) + return err +} diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 5de9b6f..8e4b98f 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -119,6 +119,8 @@ type SourceRow struct { Revision uint64 `json:"revision"` State string `json:"state"` Waiting []string `json:"waiting,omitempty"` + // Wants is how many wants of modules at the indexer were looked at; a revision with none is delivered of nothing. + Wants int `json:"wants"` } // AskRow is an open ask (novox/hq ADR 0437 §10): the gate's ask with no answer under its key in the teller's answers. @@ -210,24 +212,26 @@ type UnreadableRow struct { // Model holds the rows' inputs in memory and nothing else: a restart fills it again from the buckets' last values. type Model struct { - mu sync.Mutex - assigns map[string]assignment - builds map[string]buildEntry // by .build. - groups map[string]groupEntry // the indexer's candidates, by group - modules map[string]moduleEntry // the indexer's modules, by module - keys map[string]keyEntry // the builder's build keys, by . - asks map[string]askRecord // the gate's asks, by key - wants map[string]wantEntry // the assigner's wants, by . - source *sourceEntry // the gate's source key - composed map[string]composedEntry // the controller's digests and judgements, by .. - aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key - chosen map[string]delivery.ViewAnswer // the answers the view wrote, by ask key - writer Writer - backlog *Backlog - answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable) - newest map[string]string // module -> id - haves map[string]haveEntry - nodes map[string]nodeEntry + mu sync.Mutex + assigns map[string]assignment + builds map[string]buildEntry // by .build. + groups map[string]groupEntry // the indexer's candidates, by group + modules map[string]moduleEntry // the indexer's modules, by module + keys map[string]keyEntry // the builder's build keys, by . + asks map[string]askRecord // the gate's asks, by key + wants map[string]wantEntry // the assigner's wants, by . + source *sourceEntry // the gate's source key + deliveredEvent func(DeliveredEvent) + said uint64 // the source revision whose delivery was said + composed map[string]composedEntry // the controller's digests and judgements, by .. + aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key + chosen map[string]delivery.ViewAnswer // the answers the view wrote, by ask key + writer Writer + backlog *Backlog + answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable) + newest map[string]string // module -> id + haves map[string]haveEntry + nodes map[string]nodeEntry // the watchman's buckets, by key presence map[string]presenceEntry stuck map[string]stuckEntry @@ -302,6 +306,8 @@ type composedEntry struct { type sourceEntry struct { revision uint64 + group string + heads map[string]string err string } @@ -339,10 +345,50 @@ func New() *Model { presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}} } -// Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. +// Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes +// the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released. func (m *Model) Apply(e Entry) { + m.mu.Lock() + m.apply(e) + event := m.deliveredNow() + say := m.deliveredEvent + m.mu.Unlock() + if event != nil && say != nil { + say(*event) + } +} + +// DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR +// 0539 §17): the revision of the gate's source key, the group that moved it and every repository's merged head at it. A +// consumer takes every earlier revision as covered by it, and takes the same revision twice as once (after a restart of the view). +type DeliveredEvent struct { + Revision uint64 `json:"revision"` + Group string `json:"group"` + Heads map[string]string `json:"heads"` +} + +// SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set. +func (m *Model) SetDeliveredEvent(say func(DeliveredEvent)) { m.mu.Lock() defer m.mu.Unlock() + m.deliveredEvent = say +} + +// deliveredNow is the event of the source revision when it is delivered now and has not been said, or nil. A revision that went +// bad after it was delivered and recovered is not said again. +func (m *Model) deliveredNow() *DeliveredEvent { + if m.deliveredEvent == nil || m.source == nil || m.source.err != "" || m.said == m.source.revision { + return nil + } + row := m.sourceRow() + if row == nil || row.State != SourceDelivered || row.Wants == 0 { + return nil + } + m.said = m.source.revision + return &DeliveredEvent{Revision: m.source.revision, Group: m.source.group, Heads: m.source.heads} +} + +func (m *Model) apply(e Entry) { switch e.Bucket { case delivery.HaveBucket: node, module, isNode, err := delivery.ParseHaveKey(e.Key) @@ -456,8 +502,8 @@ func (m *Model) Apply(e Entry) { m.source = nil return } - _, err := delivery.DecodeSource(e.Value) - m.source = &sourceEntry{revision: e.Revision, err: errText(err)} + src, err := delivery.DecodeSource(e.Value) + m.source = &sourceEntry{revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)} case delivery.AnswersBucket: if e.Deleted { delete(m.answers, e.Key) @@ -963,12 +1009,14 @@ func (m *Model) sourceRow() *SourceRow { } return fp } + looked := 0 for key, w := range m.wants { node, module, _ := strings.Cut(key, ".") mod, ok := m.modules[module] if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.FP2 == "" { continue } + looked++ h := m.haves[key] switch { case w.value.FP != mod.value.FP2: @@ -993,7 +1041,7 @@ func (m *Model) sourceRow() *SourceRow { } } sort.Strings(waiting) - row := &SourceRow{Revision: m.source.revision, State: SourceDelivered} + row := &SourceRow{Revision: m.source.revision, State: SourceDelivered, Wants: looked} if len(waiting) > 0 { row.State, row.Waiting = SourceNotDelivered, waiting } diff --git a/modules/mesh-delivery-view/internal/view/wants_test.go b/modules/mesh-delivery-view/internal/view/wants_test.go index d55656b..121626a 100644 --- a/modules/mesh-delivery-view/internal/view/wants_test.go +++ b/modules/mesh-delivery-view/internal/view/wants_test.go @@ -156,3 +156,37 @@ func TestTheJudgementOfTheMoveIsShownBesideTheWant(t *testing.T) { t.Errorf("text = %v", text) } } + +// 1906/D9 (asked by the lead for issue 1946): the view says, once for each source revision, that it is delivered: the event the +// issue tracker derives "fix delivered" from, in place of the gate's states that go in release 6 (ADR 0539 §17). +func TestTheViewSaysOnceThatASourceRevisionIsDelivered(t *testing.T) { + m := New() + var said []DeliveredEvent + m.SetDeliveredEvent(func(e DeliveredEvent) { said = append(said, e) }) + fill(m, sourceE(t, 14355), moduleE(t, "web", fpNew), assign("ace", "web", "b-1", ""), + wantE(t, "ace", "web", delivery.Want{FP: fpNew})) + if len(said) != 0 { + t.Fatalf("said %v before any have", said) + } + fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthFailed)) + if len(said) != 0 { + t.Fatalf("said %v for a failed have", said) + } + fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK)) + if len(said) != 1 || said[0].Revision != 14355 || said[0].Group != "g" || said[0].Heads["novox/web"] == "" { + t.Fatalf("said = %+v, want revision 14355 once with its group and heads", said) + } + fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK), wantE(t, "ace", "web", delivery.Want{FP: fpNew})) + if len(said) != 1 { + t.Errorf("said again for the same revision: %v", said) + } + // a newer revision is delivered when its wants and haves are; a revision that goes bad and good again is not said twice + fill(m, sourceE(t, 14356)) + if len(said) != 2 || said[1].Revision != 14356 { + t.Errorf("the next revision: %v", said) + } + fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthFailed), haveFP(t, "ace", "web", fpNew, delivery.HealthOK)) + if len(said) != 2 { + t.Errorf("said again after a have failed and recovered: %v", said) + } +} diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 7a73bb9..561f953 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -89,11 +89,21 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if !reflect.DeepEqual(got, want) { t.Errorf("reads = %v, want exactly %v", got, want) } - for _, k := range []string{"emits", "consumes", "provides", "claims"} { + for _, k := range []string{"consumes", "provides", "claims"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the view writes only its answers", k) } } + // The one event the view says: a source revision is delivered (1906/D9, ADR 0539 §17), derived by it from the wants and the haves. + var emits []string + if raw, ok := top["emits"]; ok { + if err := json.Unmarshal(raw, &emits); err != nil { + t.Fatal(err) + } + } + if !reflect.DeepEqual(emits, []string{"delivered"}) { + t.Errorf("emits = %v, want exactly [delivered]", emits) + } // The view uses no build seat: it reads their workers by the broker's named grant and publishes no ask (ADR 0525 §2). if len(m.Uses) != 0 { t.Errorf("uses = %v, want none", m.Uses) diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index c84d39a..7ff9091 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -15,6 +15,9 @@ "assigner_differences", "node_haves" ], + "emits": [ + "delivered" + ], "state": [ { "name": "answers", "history": 5 } ],