From 8e60a3e078f2e2d64cf6e7ad4bbd15b593ab2a74 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:11:57 +0100 Subject: [PATCH] DLV-C3 follow: the builder hears tag-wanted on the checker's own stream, mesh.stream..event.tag-wanted The checker's family asks is its own stream, whose subjects are mesh.stream..event.* (ruled on A2c's finding 1); its verdicts on the same stream are acknowledged and not taken as tag asks. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-delivery-builder/branchend.go | 8 ++++++-- .../cmd/mesh-delivery-builder/refused_test.go | 10 ++++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/branchend.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/branchend.go index de22d90a..a63c2785 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/branchend.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/branchend.go @@ -24,10 +24,10 @@ const forgeStream = "FORGE" const forgeBranchDeleted = "branch-deleted" // checkerAsksStream is the checker's family `asks` as the bus names it, and tagWantedSubject its event tag-wanted -// (design 51 *Tagged versions*, DLV-C3). +// (design 51 *Tagged versions*, DLV-C3): on the checker's own stream, so its subject is mesh.stream..event.*. const ( checkerAsksStream = "ASKS" - tagWantedSubject = "mesh.mod." + delivery.CheckerModule + ".event.tag-wanted" + tagWantedSubject = "mesh.stream." + delivery.CheckerModule + ".event.tag-wanted" ) // forgeEnd is the part of a forge message the builder reads. @@ -117,6 +117,10 @@ func (h *branchEnds) again(m jetstream.Msg) { // tagWanted takes the checker's tag-wanted in: written to tag-asks as builder_tag writes it, answered by its watch. func tagWanted(ctx context.Context, kv jetstream.KeyValue, m jetstream.Msg, by string, now time.Time, logf func(string, ...any)) { + if m.Subject() != tagWantedSubject { + _ = m.Ack() // the family's other events (the verdicts) are not the builder's + return + } var t versions.TagWanted err := json.Unmarshal(m.Data(), &t) var key string 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 e1aed0bc..225474da 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 @@ -282,6 +282,16 @@ func aBranchsEndIsHeard(t *testing.T) { if !open.acked || !merged.acked || woken != 2 { t.Errorf("an open pull acked %v, a merge with nothing tagged acked %v, floors woken %d times", open.acked, merged.acked, woken) } + // The checker's family holds its verdicts beside tag-wanted: a verdict is acknowledged and not kept; a tag-wanted with + // tag-asks not bound is heard again. + verdict := &wakeMsg{subject: "mesh.stream.mesh-delivery-checker.event.verdict", data: []byte(`{"repository":"o/r"}`)} + wanted := &wakeMsg{subject: tagWantedSubject, data: []byte(`{"repository":"o/r","branch":"feat/x","head":"abc","tag":"feat-x"}`)} + for _, m := range []*wakeMsg{verdict, wanted} { + tagWanted(context.Background(), nil, m, delivery.CheckerModule, time.Now(), t.Logf) + } + if !verdict.acked || verdict.delayed != 0 || wanted.acked || wanted.delayed != versions.RemoveAgainAfter || !strings.HasPrefix(tagWantedSubject, "mesh.stream.") { + t.Errorf("a verdict acked %v (delayed %v), a tag-wanted acked %v (delayed %v)", verdict.acked, verdict.delayed, wanted.acked, wanted.delayed) + } unreadable := &wakeMsg{subject: merged.subject, data: merged.data} h.acts.Numberer.Versions.Store = failingVersions{} h.heard(context.Background(), unreadable)