117 lines
3.9 KiB
Go
117 lines
3.9 KiB
Go
package view
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// The backlog of the build seats (novox/hq ADR 0525 §4, issue 1887 A8a): for each language and kind of ask, how many asks wait
|
|
// on the seat's stream for a node to pull them, how many a node holds without having finished them, and how old the oldest
|
|
// waiting ask is. It is read from the stream's shared pull worker (delivery.AskWorker) when the view is asked: NumPending
|
|
// waits, NumAckPending is held by a node, and the oldest waiting ask is the stream's message at the worker's next sequence.
|
|
|
|
// BacklogRow is one language's kind of ask: Waiting asks not yet pulled, Held asks pulled by a node and not yet ended, and the
|
|
// oldest waiting ask's time and age in seconds. Error says why the row could not be read, never dropping it silently.
|
|
type BacklogRow struct {
|
|
Language string `json:"language"`
|
|
Kind string `json:"kind"`
|
|
Waiting uint64 `json:"waiting"`
|
|
Held uint64 `json:"held"`
|
|
Oldest time.Time `json:"oldest,omitzero"`
|
|
OldestSecs int64 `json:"oldest-secs,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
// Backlog reads the seats' workers. Info is the worker's counts and the stream sequence it delivers next; Message is the body
|
|
// of the first message of the stream at or after seq on the subject.
|
|
type Backlog struct {
|
|
Info func(stream, worker string) (waiting, held, next uint64, err error)
|
|
Message func(stream, subject string, seq uint64) ([]byte, error)
|
|
Now func() time.Time
|
|
}
|
|
|
|
// Rows are the rows of every language and kind that has an ask waiting or held, or that could not be read.
|
|
func (b *Backlog) Rows() []BacklogRow {
|
|
now := time.Now
|
|
if b.Now != nil {
|
|
now = b.Now
|
|
}
|
|
out := []BacklogRow{}
|
|
for _, lang := range delivery.AskLanguages {
|
|
stream, _ := delivery.AskStream(lang)
|
|
for _, kind := range delivery.AskKinds {
|
|
worker, _ := delivery.AskWorker(lang, kind)
|
|
subject, _ := delivery.AskSubject(lang, kind)
|
|
row := BacklogRow{Language: lang, Kind: kind}
|
|
waiting, held, next, err := b.Info(stream, worker)
|
|
if err != nil {
|
|
row.Error = err.Error()
|
|
out = append(out, row)
|
|
continue
|
|
}
|
|
row.Waiting, row.Held = waiting, held
|
|
if waiting > 0 {
|
|
body, err := b.Message(stream, subject, next)
|
|
if err == nil {
|
|
var a delivery.StepAsk
|
|
a, err = delivery.DecodeStepAsk(body)
|
|
if err == nil {
|
|
row.Oldest, row.OldestSecs = a.At, int64(now().Sub(a.At).Seconds())
|
|
}
|
|
}
|
|
if err != nil {
|
|
row.Error = fmt.Sprintf("the oldest waiting ask cannot be read: %v", err)
|
|
}
|
|
}
|
|
if row.Waiting > 0 || row.Held > 0 || row.Error != "" {
|
|
out = append(out, row)
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// BacklogOver reads the backlog from the bus: the worker's consumer info, and the stream's message by sequence.
|
|
func BacklogOver(js jetstream.JetStream, timeout time.Duration) *Backlog {
|
|
ctx := func() (context.Context, context.CancelFunc) {
|
|
return context.WithTimeout(context.Background(), timeout)
|
|
}
|
|
return &Backlog{
|
|
Info: func(stream, worker string) (uint64, uint64, uint64, error) {
|
|
c, cancel := ctx()
|
|
defer cancel()
|
|
s, err := js.Stream(c, stream)
|
|
if err != nil {
|
|
return 0, 0, 0, fmt.Errorf("the stream %s: %w", stream, err)
|
|
}
|
|
cons, err := s.Consumer(c, worker)
|
|
if err != nil {
|
|
return 0, 0, 0, fmt.Errorf("the worker %s: %w", worker, err)
|
|
}
|
|
info := cons.CachedInfo()
|
|
if i, err := cons.Info(c); err == nil {
|
|
info = i
|
|
}
|
|
return uint64(info.NumPending), uint64(info.NumAckPending), info.Delivered.Stream + 1, nil
|
|
},
|
|
Message: func(stream, subject string, seq uint64) ([]byte, error) {
|
|
c, cancel := ctx()
|
|
defer cancel()
|
|
s, err := js.Stream(c, stream)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
m, err := s.GetMsg(c, seq, jetstream.WithGetMsgSubject(subject))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return m.Data, nil
|
|
},
|
|
}
|
|
}
|