117 lines
4.4 KiB
Go
117 lines
4.4 KiB
Go
package build
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
|
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
|
)
|
|
|
|
// **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_<LANGUAGE>, 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 = builds.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.<stream>.<worker> 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 := buildstep.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-<language>.
|
|
func LanguageOf(arg string) string {
|
|
return strings.TrimPrefix(strings.TrimSpace(arg), "node-build-")
|
|
}
|