From 3e8bd9b0def24c2c386898c57fc7d16fa2a60146 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 01:19:30 +0100 Subject: [PATCH] Builder: count only deliverable asks ahead, drop the old message of an ask asked again, and read a seat's queue (issue 1945) mesh-delivery #51's re-ask never fired for mesh-build-node.4e7b118e: messages the SEAT_NODE_BUILD_GO workers had given up on stay on the work queue, and each counted as waiting ahead, so the ask was never asked again (note 62). A message ahead now waits only when no node wrote its step and no ask of its kind asked after it was taken; the old message of an ask asked again is deleted from its stream; and builder_seat_queue (answered by the builder through its queues state, as only its account may read the stream) shows each message and whether it still waits. --- .../cmd/mesh-delivery-builder-tools/main.go | 85 +++++- .../cmd/mesh-delivery-builder/held.go | 263 +++++++++++++++--- .../mesh-delivery-builder/integration_test.go | 30 ++ .../cmd/mesh-delivery-builder/main.go | 12 +- .../cmd/mesh-delivery-builder/queues.go | 49 ++++ .../internal/build/askagain.go | 124 ++++++++- .../internal/build/queue.go | 115 ++++++++ .../internal/build/run_test.go | 77 +++-- .../mesh-delivery-builder/manifest_test.go | 6 +- modules/mesh-delivery-builder/module.json | 2 + 10 files changed, 686 insertions(+), 77 deletions(-) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/queues.go create mode 100644 modules/mesh-delivery-builder/internal/build/queue.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index 7e38f3b..4930788 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -5,19 +5,26 @@ package main import ( + "context" "encoding/json" "fmt" "os" "strings" + "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-builder/internal/build" ) -// Tools are the builder's two tools: stuck reads, askAgain writes the reasks value. -func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, now func() time.Time) []stdio.Tool { +// queueAsker writes a queues value and waits for the builder's answer to it. +type queueAsker func(key string, value []byte) (json.RawMessage, error) + +// Tools are the builder's tools: stuck reads, askAgain writes the reasks value, queue writes a queues value and waits for +// the builder's read. +func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, queue queueAsker, now func() time.Time) []stdio.Tool { str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } return []stdio.Tool{ {Name: build.StuckTool, @@ -48,6 +55,78 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, n return map[string]any{"key": key, "asked": true, "next": "the builder answers in " + build.ReAsksState + "." + key + "; " + build.StuckTool + " shows the answer and the new ask"}, nil }}, + {Name: build.SeatQueueTool, + Description: "What waits on a build seat's stream (SEAT_NODE_BUILD_), as the builder reads it: the stream's message count and " + + "sequences, and per kind (delivery, check, recheck) its messages oldest first, at most limit each, with sequence, ask id, when asked " + + "and put, age, the state of the step a build node wrote for it, and whether the builder counts it as still waiting or as given up " + + "on (its worker delivered it its most times) or taken; and the ask asked last that a node took. The workers' delivery counts are " + + "not readable by the builder (nats.nats_streams shows them). Read by the builder, which alone may read the stream; read-only.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "language": map[string]any{"type": "string", "description": "go, node or image (default go)"}, + "seat": map[string]any{"type": "string", "description": "or the seat: node-build-go, node-build-node, node-build-image"}, + "limit": map[string]any{"type": "integer", "description": fmt.Sprintf("messages read of each kind (default %d, at most %d)", build.DefaultQueueLimit, build.MaxQueueLimit)}}}, + Run: func(args map[string]any) (any, error) { + language := build.LanguageOf(str(args, "language")) + if language == "" { + language = build.LanguageOf(str(args, "seat")) + } + if language == "" { + language = delivery.AskGo + } + limit, _ := args["limit"].(float64) + key, value, err := build.NewQueueAsk(language, int(limit), stdio.CallerOf(args), now()) + if err != nil { + return nil, err + } + return queue(key, value) + }}, + } +} + +// askQueue writes the queues value at key and waits, on a watch of the key, at most bound for the builder's answer. +func askQueue(s stdio.KeptState, bound time.Duration) queueAsker { + return func(key string, value []byte) (json.RawMessage, error) { + answered := make(chan json.RawMessage, 1) + var rev uint64 + var mu sync.Mutex + if err := s.Watch(key, func(c stdio.StateChange) error { + mu.Lock() + mine := rev + mu.Unlock() + if c.Op != "put" || mine == 0 || c.Revision <= mine { + return nil + } + if a, err := build.DecodeQueueAsk(c.Value); err == nil && !a.Answered.IsZero() { + select { + case answered <- c.Value: + default: + } + } + return nil + }); err != nil { + return nil, fmt.Errorf("watching %s.%s: %w", build.QueuesState, key, err) + } + r, err := s.Put(key, json.RawMessage(value)) + if err != nil { + return nil, fmt.Errorf("the read could not be written to %s: %w", build.QueuesBucket, err) + } + mu.Lock() + rev = r + mu.Unlock() + // The watch may have heard the answer before the revision was known: read it once too. + if e, err := s.Get(key); err == nil && e != nil && e.Revision > r { + if a, err := build.DecodeQueueAsk(e.Value); err == nil && !a.Answered.IsZero() { + return e.Value, nil + } + } + wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8) + defer cancel() + select { + case v := <-answered: + return v, nil + case <-wait.Done(): + return nil, fmt.Errorf("the builder did not answer %s.%s within %s: it is not running, or cannot read %s", build.QueuesState, key, bound, build.QueuesBucket) + } } } @@ -92,7 +171,7 @@ func main() { _, err := reasks.Put(key, json.RawMessage(value)) return err } - if err := stdio.Serve("", Tools(r, put, time.Now)); err != nil { + if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), time.Now)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go index 1c5afd8..369a245 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go @@ -4,6 +4,9 @@ import ( "context" "errors" "fmt" + "sort" + "strings" + "time" "github.com/nats-io/nats.go/jetstream" @@ -36,22 +39,29 @@ func (h heldOnTheBus) Gone(ctx context.Context, a delivery.StepAsk) (string, err return seen.Why(), nil } -// seen reads what the bus holds of the ask; with later, a step of its build module started after the ask was put too, -// when the ask waits first in line. +// maxOnSubject bounds the messages of one subject a read walks (each a read by sequence, next by subject), and maxSteps +// the steps it reads whole to find the last ask taken: past them the read says it cannot tell, and the ask is waited for. +const ( + maxOnSubject = 5000 + maxSteps = 200 +) + +// seen reads what the bus holds of the ask; with later, the ask asked last that a build node took too (AskSeen.Last), when +// the ask is on its stream. func (h heldOnTheBus) seen(ctx context.Context, a delivery.StepAsk, later bool) (build.AskSeen, error) { - var seen build.AskSeen + seen := build.AskSeen{Kind: a.Kind, At: build.AskedAt(a.ID)} module, err := delivery.BuildModuleOf(a.Language) if err != nil { return seen, err } - pattern, err := delivery.AskStepWatch(a.ID) - if err != nil { - return seen, err - } steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module)) if err != nil { return seen, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) } + pattern, err := delivery.AskStepWatch(a.ID) + if err != nil { + return seen, err + } keys, err := steps.ListKeysFiltered(ctx, pattern) if err != nil { return seen, err @@ -61,67 +71,236 @@ func (h heldOnTheBus) seen(ctx context.Context, a delivery.StepAsk, later bool) seen.Step = true // a node took it return seen, nil } - subject, err := delivery.AskSubject(a.Language, a.Kind) - if err != nil { - return seen, err - } - // The seat's stream by its name (delivery.AskStream), as the broker grants the builder its information and its - // messages by sequence and nothing wider (issue 1910). - name, err := delivery.AskStream(a.Language) - if err != nil { - return seen, err - } + stream, name, subject, err := h.stream(ctx, a.Language, a.Kind) seen.Stream = name - stream, err := h.js.Stream(ctx, name) if errors.Is(err, jetstream.ErrStreamNotFound) { return seen, nil } if err != nil { return seen, err } - // A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read by sequence - // instead, each a direct read that consumes nothing. - info, err := stream.Info(ctx) + var ahead []build.Queued + err = onSubject(ctx, stream, subject, func(q build.Queued) bool { + if q.Ask == a.ID { + seen.OnStream, seen.PutAt = true, q.PutAt + return false + } + ahead = append(ahead, q) + return true + }) + if err != nil || !seen.OnStream || !later { + return seen, err + } + taken, err := stepKeys(ctx, steps) if err != nil { return seen, err } - for seq := info.State.FirstSeq; seq != 0 && seq <= info.State.LastSeq; seq++ { - m, err := stream.GetMsg(ctx, seq) + for i := range ahead { + ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask]) + } + seen.Ahead = ahead + since := seen.At + if since.IsZero() { + since = seen.PutAt + } + for _, q := range ahead { + if q.At.Before(since) { + since = q.At + } + } + seen.Last = lastTaken(ctx, steps, taken, a.Kind, since) + return seen, nil +} + +// stream is the build seat's stream of language, its name and the subject of kind on it. +func (h heldOnTheBus) stream(ctx context.Context, language, kind string) (jetstream.Stream, string, string, error) { + subject, err := delivery.AskSubject(language, kind) + if err != nil { + return nil, "", "", err + } + // The seat's stream by its name (delivery.AskStream), as the broker grants the builder its information and its + // messages by sequence and nothing wider (issue 1910). + name, err := delivery.AskStream(language) + if err != nil { + return nil, "", "", err + } + stream, err := h.js.Stream(ctx, name) + return stream, name, subject, err +} + +// onSubject hands each message of subject on stream to each, in order, until each answers false or the subject's +// messages end. A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read +// by sequence instead, each the next of the subject (one read per message, none for the sequences taken between), and +// none consumed. +func onSubject(ctx context.Context, stream jetstream.Stream, subject string, each func(build.Queued) bool) error { + for seq, n := uint64(1), 0; ; n++ { + if n >= maxOnSubject { + return fmt.Errorf("%s holds more than %d messages before the one asked after: not read further", subject, maxOnSubject) + } + m, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(subject)) if errors.Is(err, jetstream.ErrMsgNotFound) { - continue // taken and acknowledged, or removed + return nil } if err != nil { - return seen, err + return err } - if m.Subject != subject { - continue + q := build.Queued{Seq: m.Sequence, Ask: m.Header.Get(jetstream.MsgIDHeader), PutAt: m.Time} + if _, kind, err := delivery.ParseAskSubject(m.Subject); err == nil { + q.Kind = kind } - if m.Header.Get(jetstream.MsgIDHeader) == a.ID { - seen.OnStream, seen.PutAt = true, m.Time - break + if q.At = build.AskedAt(q.Ask); q.At.IsZero() { + q.At = q.PutAt } - seen.Ahead++ + if !each(q) { + return nil + } + seq = m.Sequence + 1 } - if !later || !seen.OnStream || seen.Ahead > 0 { - return seen, nil - } - // First in line and not taken: did a build node take any ask put after it? Its stream then no longer delivers it. +} + +// stepKeys are the build module's step keys by ask id: `.` (a node's own key, such as its pause, has no +// ask id and is left out). +func stepKeys(ctx context.Context, steps jetstream.KeyValue) (map[string][]string, error) { all, err := steps.ListKeys(ctx) if err != nil { - return seen, err + if errors.Is(err, jetstream.ErrNoKeysFound) { + return map[string][]string{}, nil + } + return nil, err } defer all.Stop() + out := map[string][]string{} for k := range all.Keys() { + if _, id, ok := strings.Cut(k, "."); ok && !build.AskedAt(id).IsZero() { + out[id] = append(out[id], k) + } + } + return out, nil +} + +// stepState is the state of the first of keys a node wrote, "" for none. +func stepState(ctx context.Context, steps jetstream.KeyValue, keys []string) string { + for _, k := range keys { e, err := steps.Get(ctx, k) if err != nil { continue } - s, err := delivery.DecodeStep(e.Value()) - if err != nil || !s.Started.After(seen.PutAt) { - continue // not a step (a node's pause key), or one started before the ask was put + if s, err := delivery.DecodeStep(e.Value()); err == nil { + return s.State } - seen.Later = &s - break } - return seen, nil + return "" +} + +// lastTaken is the ask asked last after since that a build node took, of a kind whose taking proves kind passed +// (build.Proves), read newest first and at most maxSteps of them; nil for none. A step's tool is " ask" +// (mesh-sdk buildstep). +func lastTaken(ctx context.Context, steps jetstream.KeyValue, taken map[string][]string, kind string, since time.Time) *build.Taken { + var ids []string + for id := range taken { + if build.AskedAt(id).After(since) { + ids = append(ids, id) + } + } + sort.Slice(ids, func(i, j int) bool { return build.AskedAt(ids[i]).After(build.AskedAt(ids[j])) }) + read := 0 + for _, id := range ids { + for _, k := range taken[id] { + if read >= maxSteps { + return nil + } + read++ + e, err := steps.Get(ctx, k) + if err != nil { + continue + } + s, err := delivery.DecodeStep(e.Value()) + if err != nil { + continue + } + stepKind, ok := strings.CutSuffix(s.Tool, " ask") + if !ok || !build.Proves(stepKind, kind) { + continue + } + return &build.Taken{Ask: id, Kind: stepKind, At: build.AskedAt(id), Node: s.Node, Started: s.Started} + } + } + return nil +} + +// Drop is build.Dropper's Drop: it removes the ask's message from its build seat's stream, found by its Nats-Msg-Id on its +// subject, when the stream still holds it. The builder's account must be granted $JS.API.STREAM.MSG.DELETE.. +func (h heldOnTheBus) Drop(ctx context.Context, a delivery.StepAsk) error { + stream, name, subject, err := h.stream(ctx, a.Language, a.Kind) + if errors.Is(err, jetstream.ErrStreamNotFound) { + return nil + } + if err != nil { + return err + } + var seq uint64 + if err := onSubject(ctx, stream, subject, func(q build.Queued) bool { + if q.Ask == a.ID { + seq = q.Seq + return false + } + return true + }); err != nil || seq == 0 { + return err + } + if err := stream.DeleteMsg(ctx, seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) { + return fmt.Errorf("deleting message %d of %s (is $JS.API.STREAM.MSG.DELETE.%s granted to the builder?): %w", seq, name, name, err) + } + return nil +} + +// Queue is what the tool builder_seat_queue answers (build.SeatQueue): the messages of language's build seat's stream, of +// every kind, oldest first and at most limit of each kind, each with its step's state and whether it still waits. +func (h heldOnTheBus) Queue(ctx context.Context, language string, limit int) (build.SeatQueue, error) { + q := build.SeatQueue{Language: language, Note: build.QueueNote} + module, err := delivery.BuildModuleOf(language) + if err != nil { + return q, err + } + stream, name, _, err := h.stream(ctx, language, delivery.AskDelivery) + q.Stream = name + if errors.Is(err, jetstream.ErrStreamNotFound) { + q.Note = name + " does not exist" + return q, nil + } + if err != nil { + return q, err + } + info, err := stream.Info(ctx) + if err != nil { + return q, err + } + q.Messages, q.FirstSeq, q.LastSeq = info.State.Msgs, info.State.FirstSeq, info.State.LastSeq + steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + return q, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + } + taken, err := stepKeys(ctx, steps) + if err != nil { + return q, err + } + for _, kind := range delivery.AskKinds { + subject, _ := delivery.AskSubject(language, kind) + var ahead []build.Queued + if err := onSubject(ctx, stream, subject, func(m build.Queued) bool { + ahead = append(ahead, m) + return len(ahead) < limit + }); err != nil { + return q, err + } + if len(ahead) == 0 { + continue + } + for i := range ahead { + ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask]) + } + seen := build.AskSeen{Kind: kind, Ahead: ahead, Last: lastTaken(ctx, steps, taken, kind, ahead[0].At)} + q.Kinds = append(q.Kinds, build.NewKindQueue(kind, seen, time.Now())) + } + return q, nil } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index 5184160..ee96827 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -24,6 +24,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-builder/internal/build" ) // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the @@ -822,6 +824,34 @@ func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) { t.Fatalf("asks put = %+v within 20 s, want the two the key names", asked()) } } + // 1945 note 62, on the real bus: the stream read as builder_seat_queue reads it shows the Go ask waiting; the ask can + // still be answered (nothing was taken since); and a message dropped as an ask asked again is gone from the stream. + held := heldOnTheBus{b.js} + goAsk := delivery.StepAsk{ID: v.Asks[delivery.AskGo], Language: delivery.AskGo, Kind: delivery.AskDelivery} + stale := delivery.StepAsk{ID: "mvo53jcm70b5f7", Language: delivery.AskGo, Kind: delivery.AskDelivery} + subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskDelivery) + if _, err := b.js.Publish(b.ctx, subject, []byte("{}"), jetstream.WithMsgID(stale.ID)); err != nil { + t.Fatal(err) + } + queued := func() []build.QueuedState { + q, err := held.Queue(b.ctx, delivery.AskGo, 10) + if err != nil || len(q.Kinds) != 1 || q.Kinds[0].Kind != delivery.AskDelivery { + t.Fatalf("the Go seat's queue = %+v, %v; want its delivery asks", q, err) + } + return q.Kinds[0].Messages + } + if m := queued(); len(m) != 2 || m[0].Ask != goAsk.ID || !m[0].Waits || m[1].Ask != stale.ID { + t.Fatalf("the Go delivery queue = %+v, want the key's ask waiting, then the stale one", m) + } + if why, err := held.Gone(b.ctx, goAsk); why != "" || err != nil { + t.Errorf("Gone(the waiting ask) = %q, %v; want it answerable", why, err) + } + if err := held.Drop(b.ctx, stale); err != nil { + t.Fatal(err) + } + if m := queued(); len(m) != 1 || m[0].Ask != goAsk.ID { + t.Fatalf("after the drop the Go delivery queue = %+v, want the key's ask alone", m) + } b.stop() b.start(t, "n1", nil) // the same builder, started again, while both asks wait b.fake.open() 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 1726dc5..81c973d 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -176,7 +176,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, 4) + errs := make(chan error, 6) trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf} trunks.start() b.Trunk = trunks.head @@ -288,6 +288,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { errs <- follow(ctx, rw, eachReAsk) }() } } + // 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. + if kv := process.OptionalBucket(ctx, js, build.QueuesBucket, "builder_seat_queue is not answered until the builder starts again", logger.Printf); kv != nil { + eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) } + if qw, err := settle(ctx, kv, eachRead); err != nil { + process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again") + } else { + go func() { errs <- follow(ctx, qw, eachRead) }() + } + } logger.Printf("started on %s", cred.Node) go func() { errs <- follow(ctx, w, each) }() select { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/queues.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/queues.go new file mode 100644 index 0000000..5d52389 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/queues.go @@ -0,0 +1,49 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-builder/internal/build" +) + +// queueReader is the part of the builder a queues value reads through: heldOnTheBus. +type queueReader interface { + Queue(ctx context.Context, language string, limit int) (build.SeatQueue, error) +} + +// queueRead answers one value of the queues state (the tool builder_seat_queue, issue 1945 note 62): a value not yet +// answered is read now, from the stream the builder's account may read, and written back by compare-and-set at its +// revision with what was read or why it could not be. A value answered already (this builder's own write, or another +// instance's) or deleted is left. +func queueRead(ctx context.Context, r queueReader, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) { + if e.Operation() != jetstream.KeyValuePut { + return + } + a, err := build.DecodeQueueAsk(e.Value()) + if err != nil { + logf("%s.%s holds what is not a queue read, left: %v", build.QueuesState, e.Key(), err) + return + } + if !a.Answered.IsZero() { + return + } + read, cancel := context.WithTimeout(ctx, 20*time.Second) + q, err := r.Queue(read, e.Key(), max(1, min(a.Limit, build.MaxQueueLimit))) + cancel() + if err != nil { + a.Error = err.Error() + } + a.Queue, a.Answered = &q, time.Now().UTC() + data, err := json.Marshal(a) + if err != nil { + return + } + if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + logf("%s.%s: the read could not be written: %v", build.QueuesState, e.Key(), err) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/askagain.go b/modules/mesh-delivery-builder/internal/build/askagain.go index 980b5f6..e048541 100644 --- a/modules/mesh-delivery-builder/internal/build/askagain.go +++ b/modules/mesh-delivery-builder/internal/build/askagain.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "regexp" + "slices" "strconv" "strings" "time" @@ -44,33 +45,133 @@ type AskedAgain struct { // AskSeen is what the bus says of an ask whose step has not ended, as the builder may read it: its build seat's stream // (information and messages by sequence) and its build module's steps. The builder may not read the seat workers' -// delivery counts (granted to the delivery view alone), so a message given up on is told by what the workers did after it. +// delivery counts or ack floors (granted to the delivery view alone: $JS.API.CONSUMER.INFO..), so a +// message given up on is told by what the workers did after it: a build node took an ask asked later, of the same kind or +// of one pulled after it (issue 1945 note 62: seven messages SEAT_NODE_BUILD_GO had given up on sat ahead of a Go ask for +// five hours, each counted as waiting, so the ask was never asked again). type AskSeen struct { Stream string + // Kind is the ask's kind, At when it was asked (its id's time). + Kind string + At time.Time // Step is whether a build node wrote a step for the ask (running: the step bound of buildrun.Asker judges it). Step bool - // OnStream is whether the stream still holds the ask's message; PutAt is when it was put, Ahead how many messages of - // its subject wait before it. + // OnStream is whether the stream still holds the ask's message; PutAt is when it was put. OnStream bool PutAt time.Time - Ahead int - // Later is a step of the same build module that started after the ask was put, when one did. - Later *delivery.Step + // Ahead are the messages of its subject before it on the stream. + Ahead []Queued + // Last is the ask asked last that a build node took (wrote a step for) of the ask's kind or of a kind pulled after + // it, when one was taken: every message of the kind asked before it was then delivered already, or given up on. + Last *Taken } -// Why is why the ask can no longer be answered, or "" when it can: it has a step, waits behind others of its kind, or no -// build node took any ask since it was put (the nodes are busy or paused: a backlog, never a fault, ADR 0525 §9). +// Queued is one message of a build seat's stream: its sequence, its ask's id and kind, when the ask was asked (its id's +// time, the put time when the id says none) and put, and the state of the step a build node wrote for it ("" for none). +type Queued struct { + Seq uint64 `json:"seq"` + Ask string `json:"ask"` + Kind string `json:"kind"` + At time.Time `json:"asked"` + PutAt time.Time `json:"put"` + Step string `json:"step,omitempty"` +} + +// Taken is an ask a build node took: its id, kind and when it was asked, and the step's node and start. +type Taken struct { + Ask string `json:"ask"` + Kind string `json:"kind"` + At time.Time `json:"asked"` + Node string `json:"node"` + Started time.Time `json:"started"` +} + +// Proves is whether a taken ask of kind taken shows that a message of kind waiting asked before it is no longer +// delivered: a holder pulls the kinds in delivery.AskKinds' order, so taking one of the same kind or of a later one means +// its worker had none of waiting's kind left to give. +func Proves(taken, waiting string) bool { + t, w := slices.Index(delivery.AskKinds, taken), slices.Index(delivery.AskKinds, waiting) + return t >= 0 && w >= 0 && t >= w +} + +// Waits is whether a message ahead still waits to be delivered: no build node wrote a step for it, and no ask of its +// kind asked after it was taken. A message its worker gave up on (delivered its most times) stays on a work queue until +// its age, and is not counted. +func (s AskSeen) Waits(q Queued) bool { + return q.Step == "" && !s.passed(q.At) +} + +func (s AskSeen) passed(at time.Time) bool { + return s.Last != nil && Proves(s.Last.Kind, s.Kind) && s.Last.At.After(at) +} + +// Waiting is how many messages ahead still wait to be delivered (Waits). +func (s AskSeen) Waiting() int { + n := 0 + for _, q := range s.Ahead { + if s.Waits(q) { + n++ + } + } + return n +} + +// Why is why the ask can no longer be answered, or "" when it can: it has a step, waits behind others of its kind still +// deliverable, or no build node took a later ask of its kind (the nodes are busy or paused: a backlog, never a fault, +// ADR 0525 §9). func (s AskSeen) Why() string { switch { case s.Step: return "" case !s.OnStream: return fmt.Sprintf("no build node wrote a step for it and %s no longer holds it: it was given up on or dropped", s.Stream) - case s.Ahead > 0 || s.Later == nil: + case s.Waiting() > 0 || !s.passed(s.at()): return "" default: - return fmt.Sprintf("it waits first in line on %s since %s while %s took a later ask at %s: its stream no longer delivers it (given up after its most deliveries)", - s.Stream, s.PutAt.UTC().Format(time.RFC3339), s.Later.Node, s.Later.Started.UTC().Format(time.RFC3339)) + behind := "" + if n := len(s.Ahead); n > 0 { + behind = fmt.Sprintf(" behind %d given up on or taken", n) + } + return fmt.Sprintf("it waits on %s since %s%s while %s took the %s ask %s asked later at %s: its stream no longer delivers it (given up after its most deliveries)", + s.Stream, s.PutAt.UTC().Format(time.RFC3339), behind, s.Last.Node, s.Last.Kind, s.Last.Ask, s.Last.At.UTC().Format(time.RFC3339)) + } +} + +func (s AskSeen) at() time.Time { + if s.At.IsZero() { + return s.PutAt + } + return s.At +} + +// AskedAt is when an ask id was made (buildstep.NewID: its milliseconds in base 36 before six hex digits), or zero. +func AskedAt(id string) time.Time { + if len(id) <= 6 { + return time.Time{} + } + ms, err := strconv.ParseInt(id[:len(id)-6], 36, 64) + if err != nil || ms <= 0 { + return time.Time{} + } + return time.UnixMilli(ms) +} + +// Dropper removes an ask's message from its build seat's stream (Builder.Gone may be one): the old message of an ask asked +// again, so a message given up on does not sit ahead of every later ask of its kind until its age (issue 1945 note 62). +type Dropper interface { + Drop(ctx context.Context, a delivery.StepAsk) error +} + +// drop removes the old ask's message when Gone can; a refusal is said and changes nothing else: the new ask is put. +func (b *Builder) drop(ctx context.Context, ask delivery.StepAsk, old string) { + d, ok := b.Gone.(Dropper) + if !ok { + return + } + prior := ask + prior.ID = old + if err := d.Drop(ctx, prior); err != nil { + b.logf("the message of the ask %s asked again as %s stays on its stream: %v", old, ask.ID, err) } } @@ -183,6 +284,7 @@ func (b *Builder) answer(ctx context.Context, l *lane, ask delivery.StepAsk, ask if err := b.Asks.Put(ctx, ask); err != nil { return step, ask, err } + b.drop(ctx, ask, old) b.logf("%s: the ask %s of %s is asked again as %s (asking %d of %d): %s", key, old, l.module, ask.ID, asking, b.askings(), why) if b.Said != nil { b.Said(AskedAgain{Key: key, Module: module, FP: fp, BuildModule: l.module, Language: ask.Language, Old: old, New: ask.ID, diff --git a/modules/mesh-delivery-builder/internal/build/queue.go b/modules/mesh-delivery-builder/internal/build/queue.go new file mode 100644 index 0000000..7542a50 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/queue.go @@ -0,0 +1,115 @@ +package build + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **The tool builder_seat_queue reads a build seat's stream as the builder reads it** (novox/hq issue 1945 note 62; the +// instruction "tools are how a module is spoken to"): the messages waiting on SEAT_NODE_BUILD_, bounded, each +// with its ask, its age, the state of the step a build node wrote for it, and whether the builder counts it as still +// waiting (AskSeen.Waits) or as given up on. Only the builder's own account may read the stream (issue 1910): the tools +// bundle writes the read into the builder's state queues, keyed by the language, which the builder watches and answers +// there, as builder_ask_again does through reasks. +const ( + SeatQueueTool = "builder_seat_queue" + // QueuesState is the builder's state of queue reads: the bucket mesh-delivery-builder_queues, keyed by the language. + QueuesState = "queues" + QueuesBucket = delivery.BuilderModule + "_" + QueuesState + // DefaultQueueLimit and MaxQueueLimit bound the messages read of each kind. + DefaultQueueLimit = 50 + MaxQueueLimit = 500 +) + +// QueueNote says what the read cannot show. +const QueueNote = "the builder may not read the seat workers' delivery counts or ack floors ($JS.API.CONSUMER.INFO.. is " + + "granted to the delivery view alone): a message counts as given up on when no node wrote its step and a node took an ask of its kind " + + "asked after it; nats.nats_streams shows the workers' pending and waiting pulls" + +// QueueAsk is a value of the queues state: a read asked (limit, who, when) and, once the builder read the stream, its answer. +type QueueAsk struct { + V int `json:"v"` + Limit int `json:"limit"` + By string `json:"by,omitempty"` + At time.Time `json:"at"` + Answered time.Time `json:"answered,omitzero"` + Error string `json:"error,omitempty"` + Queue *SeatQueue `json:"queue,omitempty"` +} + +// SeatQueue is what the builder read of one build seat's stream. +type SeatQueue struct { + Language string `json:"language"` + Stream string `json:"stream"` + Messages uint64 `json:"messages"` + FirstSeq uint64 `json:"first_seq"` + LastSeq uint64 `json:"last_seq"` + Kinds []KindQueue `json:"kinds"` + Note string `json:"note"` +} + +// KindQueue is one kind's messages on the stream, oldest first: how many the builder counts as waiting, the ask asked +// last a build node took of the kind (or of one pulled after it), and each message. +type KindQueue struct { + Kind string `json:"kind"` + Read int `json:"read"` + Waiting int `json:"waiting"` + GivenUp int `json:"given_up_or_taken"` + Last *Taken `json:"last_taken,omitempty"` + Messages []QueuedState `json:"messages"` +} + +// QueuedState is one message as the tool shows it. +type QueuedState struct { + Queued + AgeSeconds int64 `json:"age_s"` + Waits bool `json:"waits"` +} + +// NewKindQueue is seen's messages ahead as one kind's queue at now. +func NewKindQueue(kind string, seen AskSeen, now time.Time) KindQueue { + k := KindQueue{Kind: kind, Read: len(seen.Ahead), Last: seen.Last} + for _, q := range seen.Ahead { + w := seen.Waits(q) + if w { + k.Waiting++ + } else { + k.GivenUp++ + } + k.Messages = append(k.Messages, QueuedState{Queued: q, AgeSeconds: int64(now.Sub(q.PutAt) / time.Second), Waits: w}) + } + return k +} + +// NewQueueAsk is the value builder_seat_queue writes for a language: its key and the value. +func NewQueueAsk(language string, limit int, by string, at time.Time) (string, []byte, error) { + if _, err := delivery.AskStream(language); err != nil { + return "", nil, fmt.Errorf("language: %w", err) + } + if limit <= 0 { + limit = DefaultQueueLimit + } + data, err := json.Marshal(QueueAsk{V: 1, Limit: min(limit, MaxQueueLimit), By: by, At: at.UTC()}) + return language, data, err +} + +// DecodeQueueAsk reads a queues value. +func DecodeQueueAsk(data []byte) (QueueAsk, error) { + var a QueueAsk + if err := json.Unmarshal(data, &a); err != nil { + return a, err + } + if a.V != 1 { + return a, fmt.Errorf("a queues value of version %d, not 1", a.V) + } + return a, nil +} + +// LanguageOf is the language a tool's seat or language argument names: go, node or image, or node-build-. +func LanguageOf(arg string) string { + return strings.TrimPrefix(strings.TrimSpace(arg), "node-build-") +} diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index 0420512..c6ba95a 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" + "strconv" "strings" "sync" "sync/atomic" @@ -617,11 +618,20 @@ func TestAFailedPublishOfACandidatesImageIsWrittenToTheKey(t *testing.T) { // re } } -// goneAsks answers Gone from a function of the ask, and counts how often it was asked. +// goneAsks answers Gone from a function of the ask, and counts how often it was asked; it keeps the ids it was asked to +// drop from their stream (Dropper). type goneAsks struct { - mu sync.Mutex - n int - why func(a delivery.StepAsk) string + mu sync.Mutex + n int + why func(a delivery.StepAsk) string + dropped []string +} + +func (g *goneAsks) Drop(_ context.Context, a delivery.StepAsk) error { + g.mu.Lock() + defer g.mu.Unlock() + g.dropped = append(g.dropped, a.ID) + return nil } func (g *goneAsks) Gone(_ context.Context, a delivery.StepAsk) (string, error) { @@ -703,14 +713,15 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { Step: c.step, Asks: map[string]string{delivery.AskGo: first}}) r.st.vals[r.key] = data } + gone := &goneAsks{} if !c.byHand { - r.b.Gone = &goneAsks{why: func(a delivery.StepAsk) string { + gone.why = func(a delivery.StepAsk) string { if c.always || a.ID == first || first == "" { return "it waits first in line while a node took a later ask" } return "" - }} - r.b.CheckAfter = 5 * time.Millisecond + } + r.b.Gone, r.b.CheckAfter = gone, 5*time.Millisecond } r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) if c.byHand { @@ -745,6 +756,11 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { } mu.Lock() defer mu.Unlock() + // Note 62: the old message of an ask asked again because its stream gave it up is dropped from the stream. + if gone.mu.Lock(); c.done && !c.byHand && (len(gone.dropped) != 1 || gone.dropped[0] != first) { + t.Errorf("dropped %v from the stream, want the old ask %s", gone.dropped, first) + } + gone.mu.Unlock() if c.asking > 0 && (len(said) != 1 || said[0].Asking != c.asking || said[0].New == said[0].Old || said[0].Key != r.key || said[0].Why == "") { t.Errorf("said %+v, want once at asking %d", said, c.asking) } @@ -752,22 +768,49 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { } t.Run("what the bus says of an ask", func(t *testing.T) { // 1945 put := time.Date(2026, 10, 25, 17, 6, 0, 0, time.UTC) - later := &delivery.Step{Node: "g14", Started: put.Add(time.Minute)} + taken := func(kind string, at time.Time) *Taken { + return &Taken{Ask: "later", Kind: kind, At: at, Node: "g14", Started: at} + } + later := taken(delivery.AskDelivery, put.Add(time.Minute)) + // ahead is n messages put before the ask, a minute apart, each with step (its state, "" for none). + ahead := func(n int, step string) []Queued { + var qs []Queued + for i := range n { + at := put.Add(-time.Duration(n-i) * time.Minute) + qs = append(qs, Queued{Seq: uint64(i + 1), Ask: fmt.Sprint("a", i), Kind: delivery.AskDelivery, At: at, PutAt: at, Step: step}) + } + return qs + } + seen := func(qs []Queued, last *Taken) AskSeen { + return AskSeen{Stream: "S", Kind: delivery.AskDelivery, At: put, OnStream: true, PutAt: put, Ahead: qs, Last: last} + } for _, c := range []struct { - seen AskSeen - want string // a part of the why, "" when it can be answered + name string + seen AskSeen + want string // a part of the why, "" when it can be answered + waiting int }{ - {AskSeen{Stream: "S", Step: true}, ""}, - {AskSeen{Stream: "S"}, "no longer holds it"}, - {AskSeen{Stream: "S", OnStream: true, PutAt: put, Later: later}, "no longer delivers it"}, - {AskSeen{Stream: "S", OnStream: true, PutAt: put, Ahead: 2, Later: later}, ""}, - {AskSeen{Stream: "S", OnStream: true, PutAt: put}, ""}, + {"a node took it", AskSeen{Stream: "S", Step: true}, "", 0}, + {"no longer on its stream", AskSeen{Stream: "S"}, "no longer holds it", 0}, + {"first in line, a later ask taken", seen(nil, later), "no longer delivers it", 0}, + {"first in line, nothing taken since", seen(nil, nil), "", 0}, + {"first in line, the ask taken last was asked before it", seen(nil, taken(delivery.AskDelivery, put.Add(-time.Second))), "", 0}, + {"two wait ahead, nothing taken since", seen(ahead(2, ""), nil), "", 2}, + {"two ahead passed over, the ask not yet", seen(ahead(2, ""), taken(delivery.AskDelivery, put.Add(-time.Second))), "", 0}, + // Note 62: seven messages SEAT_NODE_BUILD_GO gave up on sat ahead of a Go ask for five hours. + {"seven given up on ahead, a later ask taken", seen(ahead(7, ""), later), "behind 7 given up on or taken", 0}, + {"seven taken ahead (failed as poison), a later ask taken", seen(ahead(7, delivery.StepFailed), later), "no longer delivers it", 0}, + {"a check ask taken later proves a delivery ask passed", seen(ahead(7, ""), taken(delivery.AskCheck, put.Add(time.Minute))), "no longer delivers it", 0}, + {"a delivery ask taken later does not prove a check ask passed", func() AskSeen { s := seen(ahead(3, ""), later); s.Kind = delivery.AskCheck; return s }(), "", 3}, } { got := c.seen.Why() - if (c.want == "") != (got == "") || !strings.Contains(got, c.want) { - t.Errorf("%+v: why %q, want %q", c.seen, got, c.want) + if (c.want == "") != (got == "") || !strings.Contains(got, c.want) || c.seen.Waiting() != c.waiting { + t.Errorf("%s: why %q and %d waiting, want %q and %d", c.name, got, c.seen.Waiting(), c.want, c.waiting) } } + if at := time.UnixMilli(1793120000123); !AskedAt(strconv.FormatInt(at.UnixMilli(), 36)+"0a1b2c").Equal(at) || !AskedAt("x").IsZero() { + t.Errorf("AskedAt does not read an ask id's time") + } }) t.Run("the tool builder_stuck", func(t *testing.T) { // 1945 at := time.Date(2026, 10, 25, 17, 6, 0, 0, time.UTC) diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 7034836..b629f5c 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -87,7 +87,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: "instances", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory + if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", 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 @@ -115,8 +115,8 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.SecretsOwner != "mesh-delivery-builder" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - // 1945: the builder's tools, served by its tools bundle, and the event it says when it asks an ask again. - if want := []string{"builder_ask_again", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) { + // 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again. + if want := []string{"builder_ask_again", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) { t.Errorf("tools = %v, want %v", m.Tools, want) } if want := []string{"build.asked-again"}; !reflect.DeepEqual(m.Emits, want) { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 7d62e68..94426fc 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -4,6 +4,7 @@ "slug": "builder", "tools": [ "builder_ask_again", + "builder_seat_queue", "builder_stuck" ], "emits": [ @@ -20,6 +21,7 @@ "state": [ { "name": "builds", "history": 5 }, { "name": "instances", "history": 1 }, + { "name": "queues", "history": 1 }, { "name": "reasks", "history": 1 } ], "reads": [