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 66b2178d..98f84aa4 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -79,6 +79,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err) } } + // An unanswered tag ask is narrated every TagAskBound (design 51: five minutes); a test gives a shorter one. + tagAskBound := versions.TagAskBound + if v := strings.TrimSpace(getenv("MESH_BUILDER_TAG_ASK_BOUND")); v != "" { + if tagAskBound, err = time.ParseDuration(v); err != nil { + return fmt.Errorf("MESH_BUILDER_TAG_ASK_BOUND: %w", err) + } + } tools := &buildrun.ToolchainSet{} // Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot // (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity. @@ -140,11 +147,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Logf: logger.Printf, Stored: storedVia(calls.Call), Node: cred.Node, - Numbers: build.VersionNumbers{N: numberer}, } tagAsksKV := process.AwaitBucket(ctx, js, conn, versions.TagAsksBucket, "builder_tag and builder_untag are not answered", logger.Printf) - // A check's build of a head a kept tag ask names numbers its packages tagged (design 51 *Tagged versions*). + // A check's build of a head a kept tag ask names numbers its packages tagged (design 51 *Tagged versions*), and the ask + // is marked numbered with its versions. b.Tags = tagsVia(tagAsksKV.KV, logger.Printf) + b.Numbers = build.VersionNumbers{N: numberer, TagNumbered: tagNumbered(tagAsksKV.KV, logger.Printf, time.Now)} + // An ask not answered within five minutes is narrated with what it waits on: its bound's end on the bus says when. + boundsKV := process.AwaitBucket(ctx, js, conn, versions.TagAskBoundsBucket, "an unanswered tag ask is not narrated", logger.Printf) + bounds := &tagBounds{asks: tagAsksKV.KV, bounds: boundsKV.KV, bound: tagAskBound, waits: floorWaits.Waiting, logf: logger.Printf, now: time.Now} // A state bound later numbers the builds looked at without it. versionsKV.Ready(func(jetstream.KeyValue) { b.Reconsider(ctx) }) // The versions' hand acts (builder_publish_again, builder_floor_again) come through reasks, as builder_build_again. @@ -163,7 +174,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision // 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, 9) + errs := make(chan error, 10) 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, "", "") @@ -326,13 +337,28 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // A tag ask or a branch's removal by hand (builder_tag, builder_untag) is written to tag-asks and answered there // (design 51 *Tagged versions*); the checker's tag-wanted and the branch's end join it on their consumers (DLV-B6). tagAsksKV.Ready(func(kv jetstream.KeyValue) { - each := func(e jetstream.KeyValueEntry) { tagAsked(ctx, acts, kv, e, logger.Printf) } + each := func(e jetstream.KeyValueEntry) { + tagAsked(ctx, acts, kv, e, logger.Printf) + if !opOf(e) { + bounds.disarm(ctx, e.Key()) + } else if t, err := versions.DecodeTagAsk(e.Value()); err == nil { + bounds.asked(ctx, e.Key(), t) + } + } if tw, err := settle(ctx, kv, each); err != nil { process.NotWatched(logger.Printf, versions.TagAsksBucket, err, "builder_tag and builder_untag are not answered until the builder starts again") } else { go func() { errs <- follow(ctx, tw, each) }() } }) + boundsKV.Ready(func(kv jetstream.KeyValue) { + each := func(e jetstream.KeyValueEntry) { bounds.ended(ctx, e) } + if bw, err := settle(ctx, kv, each); err != nil { + process.NotWatched(logger.Printf, versions.TagAskBoundsBucket, err, "an unanswered tag ask is not narrated until the builder starts again") + } else { + go func() { errs <- follow(ctx, bw, each) }() + } + }) // A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the // builder alone may read the stream, so it answers there. process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/tagbounds.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/tagbounds.go new file mode 100644 index 00000000..500320bc --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/tagbounds.go @@ -0,0 +1,121 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-builder/internal/versions" +) + +// tagBounds narrates a tag ask the builder has not answered within five minutes, with what it waits on (design 51 +// *Tagged versions*): each unanswered ask has a key in the state tag-ask-bounds written with its own time to live, and +// the bus's marker of its end, heard on the watch, is what narrates it and writes the bound again. No timer. +type tagBounds struct { + asks func() jetstream.KeyValue + bounds func() jetstream.KeyValue + waits func() []versions.FloorWait + logf func(string, ...any) + now func() time.Time + bound time.Duration +} + +// arm writes key's bound when none stands: a Create, so a standing bound keeps its time and a restart arms each +// unanswered ask whose bound ended while the builder was away. +func (tb *tagBounds) arm(ctx context.Context, key string) { + kv := tb.bounds() + if kv == nil { + return + } + if _, err := kv.Create(ctx, key, []byte(`{}`), jetstream.KeyTTL(tb.bound)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) { + tb.logf("the bound of the tag ask %s is not written, so it is not narrated past five minutes: %v", key, err) + } +} + +// disarm removes key's bound: the ask was answered or ended. +func (tb *tagBounds) disarm(ctx context.Context, key string) { + if kv := tb.bounds(); kv != nil { + _ = kv.Delete(ctx, key) + } +} + +// asked is told each value of tag-asks: an unanswered ask is armed, any other disarmed. +func (tb *tagBounds) asked(ctx context.Context, key string, t versions.TagAsk) { + if t.Unanswered() { + tb.arm(ctx, key) + return + } + tb.disarm(ctx, key) +} + +// ended is told each change of tag-ask-bounds: a bound whose time ended (the bus's marker, a purge on the watch) on an +// ask still unanswered narrates it and arms it again. +func (tb *tagBounds) ended(ctx context.Context, e jetstream.KeyValueEntry) { + if e.Operation() != jetstream.KeyValuePurge { + return + } + kv := tb.asks() + if kv == nil { + return + } + got, err := kv.Get(ctx, e.Key()) + if err != nil { + return // the ask ended with its branch + } + t, err := versions.DecodeTagAsk(got.Value()) + if err != nil || !t.Unanswered() { + return + } + var waits []versions.FloorWait + if tb.waits != nil { + waits = tb.waits() + } + tb.logf("%s", versions.NarrateUnanswered(t, waits, tb.now())) + tb.arm(ctx, e.Key()) +} + +// tagNumbered marks the tag ask of a check's build's head numbered with its versions in tag-asks, by compare-and-set. +func tagNumbered(kv func() jetstream.KeyValue, logf func(string, ...any), now func() time.Time) func(context.Context, versions.BuildClaim, versions.Numbering) { + return func(ctx context.Context, c versions.BuildClaim, got versions.Numbering) { + store := kv() + if store == nil { + return + } + lister, err := store.ListKeysFiltered(ctx, versions.EscapeToken(c.Repository)+".>") + if err != nil { + logf("the tag ask of %s at %s is not marked numbered: %v", c.Repository, c.Head, err) + return + } + for key := range lister.Keys() { + for range 3 { + e, err := store.Get(ctx, key) + if err != nil { + break + } + t, err := versions.DecodeTagAsk(e.Value()) + if err != nil { + break + } + t, changed := t.WithNumbered(c.Module, c.Head, got.Versions, now()) + if !changed { + break + } + data, err := json.Marshal(t) + if err != nil { + break + } + if _, err = store.Update(ctx, key, data, e.Revision()); err == nil { + logf("%s: tag ask of %s %s: %s", key, t.Repository, t.Branch, t.Answer) + break + } + if !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + logf("the tag ask %s is not marked numbered: %v", key, err) + break + } + } + } + } +} diff --git a/modules/mesh-delivery-builder/internal/build/numbers.go b/modules/mesh-delivery-builder/internal/build/numbers.go index 52a5dc24..aefab4e7 100644 --- a/modules/mesh-delivery-builder/internal/build/numbers.go +++ b/modules/mesh-delivery-builder/internal/build/numbers.go @@ -19,7 +19,12 @@ type Numbers interface { } // VersionNumbers is Numbers over the builder's state versions. -type VersionNumbers struct{ N *versions.Numberer } +type VersionNumbers struct { + N *versions.Numberer + // TagNumbered is told the versions a check's build of a tag ask's head was numbered (the ask is then marked numbered + // in tag-asks, design 51 *Tagged versions*); nil tells nobody. + TagNumbered func(ctx context.Context, c versions.BuildClaim, got versions.Numbering) +} // ClaimOf is the claim a module value's trunk build is numbered by: a package per source revision (Src), the rest per // fingerprint, in merge order (Src). @@ -41,7 +46,11 @@ func (v VersionNumbers) Number(ctx context.Context, module string, m delivery.Mo // A check's build of a head a tag ask names: its packages tagged, below the next version (design 51 *Tagged versions*). c.Tag, c.Head, c.Revision, c.Merge = tag, m.Commit, "", 0 } - return v.N.Number(ctx, c) + got, err := v.N.Number(ctx, c) + if err == nil && tag != "" && len(got.Versions) > 0 && v.TagNumbered != nil { + v.TagNumbered(ctx, c, got) + } + return got, err } func (v VersionNumbers) Built(ctx context.Context, module string, m delivery.Module, plan Plan, got versions.Numbering, made []delivery.BuiltArtifact) error { diff --git a/modules/mesh-delivery-builder/internal/versions/tagbound.go b/modules/mesh-delivery-builder/internal/versions/tagbound.go new file mode 100644 index 00000000..b8e4b861 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/tagbound.go @@ -0,0 +1,74 @@ +package versions + +import ( + "fmt" + "maps" + "slices" + "strings" + "time" +) + +// The bound of an unanswered tag ask (design 51 *Tagged versions*: "an ask the builder has not answered within five +// minutes is narrated in the build log with what it waits on"): a key of the state tag-ask-bounds, written with its own +// time to live, whose expiry the builder hears on its watch of the state. No timer: the bus ends the bound. +const ( + TagAskBoundsState = "tag-ask-bounds" + TagAskBoundsBucket = "mesh-delivery-builder_tag-ask-bounds" + TagAskBound = 5 * time.Minute +) + +// Unanswered says whether a tag ask still waits for its versions: kept, not an untag, none numbered yet. +func (t TagAsk) Unanswered() bool { + return !t.Untag && t.State == TagAsked && len(t.Versions) == 0 +} + +// WithNumbered is the tag ask once a check's build of its head numbered module's packages under its tag: numbered, each +// version by its name, the answer naming them. An untag, or an ask of another head, is left as it is. +func (t TagAsk) WithNumbered(module, head string, got map[string]string, at time.Time) (TagAsk, bool) { + if t.Untag || t.Head != head || len(got) == 0 || t.State == TagRemoving || t.State == TagRemoved { + return t, false + } + if t.Versions == nil { + t.Versions = map[string]string{} + } + changed := false + for artifact, v := range got { + name := NameOf(module, artifact) + if t.Versions[name] != v { + t.Versions[name], changed = v, true + } + } + if !changed && t.State == TagNumbered { + return t, false + } + var said []string + for _, name := range slices.Sorted(maps.Keys(t.Versions)) { + said = append(said, name+" "+t.Versions[name]) + } + t.State, t.Answered = TagNumbered, at.UTC() + t.Answer = "numbered: " + strings.Join(said, ", ") + return t, true +} + +// NarrateUnanswered is the line said for a tag ask its bound found unanswered: since when, and what it waits on, the +// names waiting on their floor named with their silent sources. +func NarrateUnanswered(t TagAsk, waits []FloorWait, now time.Time) string { + on := "the check's build of " + short(t.Head) + " is not claimed yet (the checker builds a group member's head once it asks)" + var floors []string + for _, w := range waits { + floors = append(floors, w.Name+" waits on its floor ("+strings.Join(w.Silent, ", ")+" silent)") + } + if len(floors) > 0 { + slices.Sort(floors) + on += "; " + strings.Join(floors, "; ") + } + return fmt.Sprintf("the tag ask of %s %s at %s (%s) is not answered after %s: it waits on %s", + t.Repository, t.Branch, short(t.Head), t.Tag, now.Sub(t.At).Round(time.Second), on) +} + +func short(head string) string { + if len(head) > 12 { + return head[:12] + } + return head +} diff --git a/modules/mesh-delivery-builder/internal/versions/versions_test.go b/modules/mesh-delivery-builder/internal/versions/versions_test.go index e9d5e1b6..5cc345aa 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions_test.go +++ b/modules/mesh-delivery-builder/internal/versions/versions_test.go @@ -639,6 +639,29 @@ func TestTheBuildersVersions(t *testing.T) { if end, _ := acts.BranchEnded(context.Background(), "o/r", "feat/x"); !end.Done { t.Errorf("every seat said removed, and the branch's end is not done: %+v", end) } + // B6: a check's build of the ask's head marks it numbered with its versions; another head, or an untag, is left. + if !a.Unanswered() { + t.Errorf("a kept ask with no version is answered: %+v", a) + } + if _, changed := a.WithNumbered("mesh-sdk", "other", map[string]string{"go": "0.1.15-feat-x-y.1"}, at); changed { + t.Error("an ask is marked numbered by a build of another head") + } + if _, changed := u.WithNumbered("o", "", map[string]string{"go": "1"}, at); changed { + t.Error("an untag is marked numbered") + } + n, changed := a.WithNumbered("mesh-sdk", "abc", map[string]string{"go": "0.1.15-feat-x-y.1"}, at.Add(time.Minute)) + if !changed || n.State != TagNumbered || n.Versions["mesh-sdk/go"] != "0.1.15-feat-x-y.1" || n.Unanswered() || !strings.Contains(n.Answer, "mesh-sdk/go 0.1.15-feat-x-y.1") { + t.Errorf("WithNumbered = %+v, %v", n, changed) + } + if _, again := n.WithNumbered("mesh-sdk", "abc", map[string]string{"go": "0.1.15-feat-x-y.1"}, at); again { + t.Error("the same versions mark the ask again") + } + said := NarrateUnanswered(a, []FloorWait{{Name: "mesh-sdk/go", Silent: []string{"bridge_tags"}}}, at.Add(6*time.Minute)) + for _, w := range []string{"novox/mesh-sdk feat/X y", "after 6m0s", "not claimed yet", "mesh-sdk/go waits on its floor (bridge_tags silent)"} { + if !strings.Contains(said, w) { + t.Errorf("NarrateUnanswered = %q, want %q in it", said, w) + } + } if k, v, err := TagWantedAsk(TagWanted{Repository: "o/r", Branch: "feat/x", Head: "abc"}, "mesh-delivery-checker", at); err != nil || k != TagAskKey("o/r", "feat/x") || !strings.Contains(string(v), `"by":"mesh-delivery-checker"`) { t.Errorf("TagWantedAsk = %q %s %v", k, v, err) } diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index ee9ee95c..06feccae 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -22,6 +22,7 @@ import ( type state struct { Name string `json:"name"` History int `json:"history,omitempty"` + KeyTTL bool `json:"key-ttl,omitempty"` } type artifact struct { @@ -93,7 +94,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.Module != "mesh-delivery-builder" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory + if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 073294d3..00b21ca3 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -70,6 +70,7 @@ { "name": "old-form", "history": 1 }, { "name": "queues", "history": 1 }, { "name": "reasks", "history": 1 }, + { "name": "tag-ask-bounds", "history": 1, "key-ttl": true }, { "name": "tag-asks", "history": 1 }, { "name": "versions", "history": 1 } ],