Watchman: time a build seat's ask from its step-progress, not from the holders' steps (fix 10, A-136)
Builds: design 47 G §483, §496, §602, A-136
This commit is contained in:
@@ -8,6 +8,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
@@ -46,6 +47,7 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string,
|
||||
w.Accepted(a)
|
||||
case pipelineSubject(w, subject, body):
|
||||
case recordSubject(w, subject, body):
|
||||
case buildSeatSubject(w, subject, body):
|
||||
case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated):
|
||||
if s, err := watch.ReadBrokerSaid(body); err != nil {
|
||||
logf("%v; left", err)
|
||||
@@ -187,3 +189,25 @@ func recordSubject(w *watch.Watchman, subject string, body []byte) bool {
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// buildSeatSubject hands the build seats' `ask-taken`, `step-progress` and `step-ended`, on
|
||||
// mesh.seat.<seat>.event.<event>.<node>, to the watchman (design 47 G §483, §496, §602, A-136): it times every ask of a
|
||||
// build seat from them. It says whether the subject was one of them.
|
||||
func buildSeatSubject(w *watch.Watchman, subject string, body []byte) bool {
|
||||
parts := strings.Split(subject, ".")
|
||||
if len(parts) != 6 || parts[0] != "mesh" || parts[1] != "seat" || parts[3] != "event" || !strings.HasPrefix(parts[2], "node-build-") {
|
||||
return false
|
||||
}
|
||||
seat, node := parts[2], parts[5]
|
||||
switch parts[4] {
|
||||
case nodebuildgo.EventAskTaken:
|
||||
w.AskTaken(seat, node, body)
|
||||
case nodebuildgo.EventStepProgress:
|
||||
w.StepProgress(seat, node, body)
|
||||
case nodebuildgo.EventStepEnded:
|
||||
w.AskEnded(seat, body)
|
||||
default:
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -69,13 +69,27 @@ func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *
|
||||
t.Helper()
|
||||
viewstest.Serve(t, ctx, js, nc, watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants", "n1_"+watch.WatchmanModule)
|
||||
viewstest.Serve(t, ctx, js, nc, watch.IndexerModule, modulesState, watch.IndexerModule+".indexer_read_modules", "n1_"+watch.WatchmanModule)
|
||||
// The seat that places modules says who holds each build seat (design 47 G A-126): the module whose steps the watchman watches.
|
||||
held, _ := json.Marshal(map[string]any{"result": map[string]any{"seats": []map[string]any{
|
||||
{"seat": "node-build-go", "holders": []map[string]string{{"module": "mesh-build-go"}}}}}, "node": "ctl"})
|
||||
for _, subject := range []string{"mesh.seat.mesh-module-assigner.call.assignments", "mesh.seat.mesh-module-assigner.call.assignments.>"} {
|
||||
if _, err := nc.Subscribe(subject, func(m *nats.Msg) { _ = m.Respond(held) }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// makeEvents makes EVENTS and the watchman's durable consumer of the build seats' events, as the controller and the grants part
|
||||
// make them (design 47 G §483, §496, §602); the owners served before it made EVENTS for the modules' events.
|
||||
func makeEvents(t *testing.T, ctx context.Context, js jetstream.JetStream) {
|
||||
t.Helper()
|
||||
if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-watchman",
|
||||
AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// sayAsk is what the holder of the Go build seat says on node n1 about an ask: ask-taken, step-progress or step-ended.
|
||||
func sayAsk(t *testing.T, ctx context.Context, js jetstream.JetStream, event, id string) {
|
||||
t.Helper()
|
||||
body, _ := json.Marshal(map[string]any{"ask": id, "asker": "a1", "at": time.Now().UTC(), "phase": "build"})
|
||||
if _, err := js.Publish(ctx, "mesh.seat.node-build-go.event."+event+".n1", body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,7 +149,6 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
{Bucket: watch.PresenceBucket, History: 20},
|
||||
{Bucket: watch.StuckBucket, History: 20},
|
||||
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
|
||||
{Bucket: watch.StepsBucket("mesh-build-go"), History: 5},
|
||||
{Bucket: watch.AssignerWantsBucket, History: 20},
|
||||
{Bucket: watch.HaveBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
@@ -149,6 +162,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
}
|
||||
serveOwners(t, ctx, js, ctl)
|
||||
findingsStream(t, ctx, js)
|
||||
makeEvents(t, ctx, js)
|
||||
|
||||
// The controller's consumer on DEAD_LETTER_NOTICES, the dead letters' streams and a source stream with one message a
|
||||
// consumer will give up on (1952/WM5); no lease bucket at all (issue 1126).
|
||||
@@ -237,20 +251,8 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
|
||||
// An ask put on a build seat gets a pulse; one no node takes lets the pulse expire, and that writes a stuck key.
|
||||
askStream, _ := buildstep.AskStream(buildstep.AskGo)
|
||||
steps := kvs[watch.StepsBucket("mesh-build-go")]
|
||||
// A node takes an ask: it writes the step key running (and passed when the ask ends).
|
||||
putStep := func(id, state string) {
|
||||
t.Helper()
|
||||
body, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: time.Now().UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := steps.Put(ctx, "n1."+id, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb"
|
||||
putStep(ask1, buildstep.StepRunning)
|
||||
sayAsk(t, ctx, js, "ask-taken", ask1)
|
||||
until("a pulse for the ask a node took", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask1)); return ok })
|
||||
var stuck meshwatchdogsclient.Stuck
|
||||
until("a stuck key for the ask no node ended", func() bool {
|
||||
@@ -280,9 +282,9 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
until("node n2 online again", func() bool { on, ok := online("n2"); return ok && on })
|
||||
until("the pulse of node n2 deleted", func() bool { _, ok := has(watch.PulseBucket, pulseOf(watch.PresenceBucket, "n2")); return !ok })
|
||||
// An ask a node takes at once is the same: no stuck key, however long the bound passes.
|
||||
putStep(ask2, buildstep.StepRunning)
|
||||
sayAsk(t, ctx, js, "ask-taken", ask2)
|
||||
until("a pulse for the second ask", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask2)); return ok })
|
||||
putStep(ask2, buildstep.StepPassed)
|
||||
sayAsk(t, ctx, js, "step-ended", ask2)
|
||||
until("the pulse of the second ask deleted", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask2)); return !ok })
|
||||
time.Sleep(6 * time.Second) // twice the bound and the marker time
|
||||
for _, k := range []string{stuckOf(watch.PresenceBucket, "n2"), stuckOf(askStream, ask2)} {
|
||||
@@ -387,7 +389,6 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
|
||||
{Bucket: watch.PresenceBucket, History: 20},
|
||||
{Bucket: watch.StuckBucket, History: 20},
|
||||
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
|
||||
{Bucket: watch.StepsBucket("mesh-build-go"), History: 5},
|
||||
{Bucket: watch.AssignerWantsBucket, History: 20},
|
||||
{Bucket: watch.HaveBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
@@ -401,14 +402,9 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
|
||||
}
|
||||
serveOwners(t, ctx, js, ctl)
|
||||
findingsStream(t, ctx, js)
|
||||
makeEvents(t, ctx, js)
|
||||
const id = "m0000009ffffff"
|
||||
body, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: buildstep.StepRunning, Started: time.Now().UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kvs[watch.StepsBucket("mesh-build-go")].Put(ctx, "n1."+id, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sayAsk(t, ctx, js, "ask-taken", id)
|
||||
broker := filepath.Join(dir, "broker")
|
||||
cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
@@ -475,7 +471,6 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
|
||||
{Bucket: watch.PresenceBucket, History: 20},
|
||||
{Bucket: watch.StuckBucket, History: 20},
|
||||
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
|
||||
{Bucket: watch.StepsBucket("mesh-build-go"), History: 5},
|
||||
{Bucket: watch.AssignerWantsBucket, History: 20},
|
||||
{Bucket: watch.HaveBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
@@ -489,6 +484,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
|
||||
}
|
||||
serveOwners(t, ctx, js, ctl)
|
||||
findingsStream(t, ctx, js)
|
||||
makeEvents(t, ctx, js)
|
||||
fpNew := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64)
|
||||
fpOld := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
|
||||
putWant := func(node, module, fp string) {
|
||||
|
||||
@@ -19,7 +19,6 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -27,7 +26,6 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
||||
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
@@ -180,12 +178,6 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key
|
||||
}
|
||||
}
|
||||
|
||||
// stepTaken is a change of a build module's steps, handed to the watchman as it is (watch.Step): a running step is its ask
|
||||
// taken by a node and pulsed, one that ended ends the pulse. An ask still waiting in a stream has no pulse (issue 1908).
|
||||
func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) {
|
||||
w.Step(language, e.Key(), e.Value(), opOf(e))
|
||||
}
|
||||
|
||||
// createdOr is when an entry was written, or at when its picture does not say. mesh-sdk's views.Picture (the join's J3)
|
||||
// names each first-view key's written time as its Created, and the tool runner's state reads carry it (J2), so a want
|
||||
// that stood before the start waits from when it was written; at is left only for an owner whose answer names no time.
|
||||
@@ -342,37 +334,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
w.ReadWhole(watch.BuilderBuildsBucket)
|
||||
}
|
||||
}
|
||||
// The build seats' steps: a step key `<node>.<ask id>` written is the ask taken (issue 1887 A8b). A seat whose build module
|
||||
// keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same.
|
||||
taken := map[string]jetstream.KeyWatcher{}
|
||||
for _, lang := range buildstep.AskLanguages {
|
||||
seat, err := buildstep.BuildSeat(lang)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
// The build module is the holder of the build seat (design 47 G A-82, A-126): asked of the seat that places modules.
|
||||
module, err := seatlist.HolderOf(calls.Call, seat)
|
||||
if err != nil {
|
||||
logger.Printf("the steps of %s cannot be read (%v): its asks are mirrored and never seen taken", seat, err)
|
||||
continue
|
||||
}
|
||||
kv, err := js.KeyValue(ctx, watch.StepsBucket(module))
|
||||
if err != nil {
|
||||
logger.Printf("the steps of %s cannot be read (%v): its asks are mirrored and never seen taken", module, err)
|
||||
continue
|
||||
}
|
||||
lang := lang
|
||||
sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) })
|
||||
if err != nil {
|
||||
logger.Printf("the steps of %s cannot be watched (%v): its asks are mirrored and never seen taken", module, err)
|
||||
continue
|
||||
}
|
||||
taken[lang] = sw
|
||||
if bucket, err := buildstep.AskStream(lang); err == nil {
|
||||
w.ReadWhole(bucket)
|
||||
}
|
||||
}
|
||||
|
||||
// The events first, then the one request of the open connections, so no connection falls between them.
|
||||
onEvent := func(open bool) nats.MsgHandler {
|
||||
return func(m *nats.Msg) {
|
||||
@@ -535,10 +496,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
errs <- follow(ctx, buildSeen, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) })
|
||||
}()
|
||||
}
|
||||
for lang, sw := range taken {
|
||||
lang, sw := lang, sw
|
||||
go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }()
|
||||
}
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
|
||||
@@ -17,7 +17,6 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
@@ -360,13 +359,13 @@ func TestTheWatchmansFindings(t *testing.T) {
|
||||
w.Have("ace.prov-y", nil, Delete)
|
||||
}, map[string]string{}},
|
||||
{"WM4: a stuck key not moved is its finding, and its move deletes it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Step("go", "n1.ask1", runningStep("ask1", t0), Put)
|
||||
askProgress(w, "ask1", t0)
|
||||
expire(w, c, pulseKeyOf(t, checksKey, "ask1"), 120*time.Second)
|
||||
k := FindingKey(ScopeDelivery, stuckKeyOf(t, checksKey, "ask1"), KindStuck)
|
||||
if f := o.findings[k]; f.Kind != KindStuck || !strings.Contains(f.Summary, "the ask ask1 to end on n1") {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
w.Step("go", "n1.ask1", endedStep("ask1", buildstep.StepPassed, t0), Put)
|
||||
askEnded(w, "ask1")
|
||||
}, map[string]string{}},
|
||||
{"WM2 S1: a node offline past its bound is silent, and back online it is not", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Connect("node.ace")
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -9,58 +10,91 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
||||
)
|
||||
|
||||
// A build ask taken by a node is a move (novox/hq ADR 0525 §4, issues 1887 A8b, 1908, 1909): the node writes the ask's step
|
||||
// key `<node>.<ask id>` running in the build module's steps (StepsBucket) when it takes the ask, and writes it passed,
|
||||
// failed or refused when the ask ends. The pulse of the move is the pulse of the bucket named as the seat's stream
|
||||
// (buildstep.AskStream), under the ask's id, and it begins when the node started the step.
|
||||
// A build ask taken by a node is a move (novox/hq ADR 0525 §4, issues 1887 A8b, 1908, 1909; design 47 G §483, §496, §602,
|
||||
// A-136): the holder of the build seat says `ask-taken` when it takes any ask, `step-progress` once a minute while it runs
|
||||
// it, and `step-ended` when it ends it. The pulse of the move is the pulse of the bucket named as the seat's stream
|
||||
// (buildstep.AskStream), under the ask's id, and it begins at the latest of the ask's taking and its progress, so a step
|
||||
// that runs long but says its progress raises nothing, and one whose holder stops is stuck within the progress bound.
|
||||
//
|
||||
// **An ask still waiting in the stream is a backlog, not a stall** (issue 1908): the rebuild after the switch puts about 170
|
||||
// asks on a few nodes' slots, and most wait for hours with every worker healthy. No pulse is made for a waiting ask, so none
|
||||
// can expire into a false stuck row; the view shows the backlog and the oldest waiting ask's age (delivery view, A8a). What
|
||||
// is read of the steps at the watchman's start is mirrored the same way, so an ask running when the watchman was down is
|
||||
// pulsed again (issue 1909).
|
||||
// **An ask still waiting in the stream is a backlog, not a stall** (issue 1908): no pulse is made for a waiting ask, so none
|
||||
// can expire into a false stuck row; the view shows the backlog and the oldest waiting ask's age (delivery view, A8a).
|
||||
|
||||
// Step is a change of a key of the steps bucket of language's build module, read from the watch (at the start the keys that
|
||||
// stand, then each change). A running step makes the pulse of its ask, from the step's own start; a step that ended, or a
|
||||
// key deleted, ends it. The node's `paused.<node>` key, a key that is no step key and a value that cannot be read are left.
|
||||
func (w *Watchman) Step(language, key string, value []byte, op Op) {
|
||||
bucket, err := buildstep.AskStream(language)
|
||||
if err != nil {
|
||||
w.say("no pulse for the step %s: %v", key, err)
|
||||
// ProgressBound is how long a running ask may say nothing of its progress (design 47 G §75: five minutes).
|
||||
const ProgressBound = 5 * time.Minute
|
||||
|
||||
// languageOfSeat is the build language whose seat is seat, and whether it is a build seat.
|
||||
func languageOfSeat(seat string) (string, bool) {
|
||||
for _, lang := range buildstep.AskLanguages {
|
||||
if s, err := buildstep.BuildSeat(lang); err == nil && s == seat {
|
||||
return lang, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// AskTaken hears a build seat's `ask-taken` (node: the subject's last token): the ask is taken, so its clock begins at the
|
||||
// event's time.
|
||||
func (w *Watchman) AskTaken(seat, node string, body []byte) {
|
||||
var e nodebuildgo.AskTakenEvent
|
||||
if err := json.Unmarshal(body, &e); err != nil || e.Ask == "" {
|
||||
w.say("an ask-taken of %s does not read, left: %v", seat, err)
|
||||
return
|
||||
}
|
||||
_, node, id, err := buildstep.ParseStepKey(key)
|
||||
w.askClock(seat, node, e.Ask, e.At)
|
||||
}
|
||||
|
||||
// StepProgress hears a build seat's `step-progress`: the running ask says it lives, so its clock begins again at the
|
||||
// event's time.
|
||||
func (w *Watchman) StepProgress(seat, node string, body []byte) {
|
||||
var e nodebuildgo.StepProgressEvent
|
||||
if err := json.Unmarshal(body, &e); err != nil || e.Ask == "" {
|
||||
w.say("a step-progress of %s does not read, left: %v", seat, err)
|
||||
return
|
||||
}
|
||||
w.askClock(seat, node, e.Ask, e.At)
|
||||
}
|
||||
|
||||
// AskEnded hears a build seat's `step-ended`, whatever its asker: the ask's clock ends.
|
||||
func (w *Watchman) AskEnded(seat string, body []byte) {
|
||||
lang, ok := languageOfSeat(seat)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
bucket, err := buildstep.AskStream(lang)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
var e nodebuildgo.StepEndedEvent
|
||||
if err := json.Unmarshal(body, &e); err != nil || e.Ask == "" {
|
||||
w.say("a step-ended of %s does not read, left: %v", seat, err)
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
switch op {
|
||||
case Expired:
|
||||
return
|
||||
case Delete:
|
||||
w.moved(bucket, id)
|
||||
w.moved(bucket, e.Ask)
|
||||
}
|
||||
|
||||
func (w *Watchman) askClock(seat, node, ask string, at time.Time) {
|
||||
lang, ok := languageOfSeat(seat)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
w.stands(bucket, id)
|
||||
step, err := buildstep.DecodeStep(value)
|
||||
bucket, err := buildstep.AskStream(lang)
|
||||
if err != nil {
|
||||
w.say("the step %s cannot be read, and is left: %v", key, err)
|
||||
w.say("no pulse for the ask %s: %v", ask, err)
|
||||
return
|
||||
}
|
||||
if step.State != buildstep.StepRunning {
|
||||
w.moved(bucket, id)
|
||||
return
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if at.IsZero() {
|
||||
at = w.now()
|
||||
}
|
||||
since := step.Started
|
||||
if since.IsZero() {
|
||||
since = w.now()
|
||||
}
|
||||
w.mirrorWithin(w.stepBound, bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0)
|
||||
w.mirrorWithin(w.progressBound, bucket, ask, describe("the ask %s to end on %s", ask, node), node, at, 0)
|
||||
}
|
||||
|
||||
// A want is a move too (novox/hq ADR 0539 §19, issue 1906): the assigner writes `want.<node>.<module>` in its wants bucket,
|
||||
|
||||
@@ -38,6 +38,3 @@ const (
|
||||
FindingManifestWaits = "manifest-waits"
|
||||
FindingNodeAccountFallback = "node-account-fallback"
|
||||
)
|
||||
|
||||
// StepsBucket is a build module's state steps (reads: the build seats' steps, `seat:node-build-<language>.steps`, held by the module that holds the seat).
|
||||
func StepsBucket(module string) string { return module + "_steps" }
|
||||
|
||||
@@ -105,6 +105,8 @@ type Watchman struct {
|
||||
tunnels map[string]string
|
||||
// stepBound is how long a build step may run once a node took the ask; a test shortens it.
|
||||
stepBound time.Duration
|
||||
// progressBound is how long a running ask may say nothing of its progress; a test shortens it.
|
||||
progressBound time.Duration
|
||||
// standing is, until the start ends (Reconcile), each mirrored key read standing, by the bucket a stuck key names it
|
||||
// under; readWhole the buckets whose keys were all read at this start (ReadWhole). reconciled is the start's end.
|
||||
standing map[string]map[string]bool
|
||||
@@ -127,7 +129,7 @@ type Watchman struct {
|
||||
// New is a watchman writing to out, reading the time from now.
|
||||
func New(out Out, now func() time.Time) *Watchman {
|
||||
return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{},
|
||||
pulses: map[string]*pulse{}, stuck: map[string]meshwatchdogsclient.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: meshwatchdogsclient.PulseMoveBound, stepBound: StepBound,
|
||||
pulses: map[string]*pulse{}, stuck: map[string]meshwatchdogsclient.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: meshwatchdogsclient.PulseMoveBound, stepBound: StepBound, progressBound: ProgressBound,
|
||||
findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, away: map[string]bool{},
|
||||
haveState: map[string]nodeengine.HaveByNodeModuleValue{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{},
|
||||
control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{},
|
||||
@@ -330,7 +332,7 @@ func (w *Watchman) Reconcile() {
|
||||
// (ReadWhole) and whose key was not read standing is written moved, which clears its finding, and said. A bucket that
|
||||
// could not be read closes nothing: its keys are not known to be gone.
|
||||
|
||||
// ReadWhole says every key of bucket (as a stuck key names it: an ask stream for a build seat's steps, the assigner's
|
||||
// ReadWhole says every key of bucket (as a stuck key names it: the assigner's
|
||||
// wants) was read at this start; called once its watch holds what stands.
|
||||
func (w *Watchman) ReadWhole(bucket string) {
|
||||
w.mu.Lock()
|
||||
@@ -453,14 +455,14 @@ const StepBound = 30 * time.Minute
|
||||
func (w *Watchman) SetStepBound(d time.Duration) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.stepBound = d
|
||||
w.stepBound, w.progressBound = d, d
|
||||
}
|
||||
|
||||
// SetBound sets how long a move, and a running step, may take; it is meshwatchdogsclient.PulseMoveBound unless a test shortens it.
|
||||
func (w *Watchman) SetBound(d time.Duration) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.bound, w.stepBound = d, d
|
||||
w.bound, w.stepBound, w.progressBound = d, d, d
|
||||
}
|
||||
|
||||
// ttl is the time to live of a pulse for a move of the given age, and whether the move is stalled already
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
||||
)
|
||||
|
||||
type fakeOut struct {
|
||||
@@ -115,22 +114,23 @@ const (
|
||||
presenceNS = PresenceBucket
|
||||
)
|
||||
|
||||
// runningStep is a build module's step key as a node writes it when it takes an ask: the ask's id, running, since started.
|
||||
func runningStep(id string, started time.Time) []byte {
|
||||
b, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: buildstep.StepRunning, Started: started})
|
||||
// askBody is the body of a build seat's step-progress, ask-taken or step-ended for an ask at a time.
|
||||
func askBody(id string, at time.Time) []byte {
|
||||
b, err := json.Marshal(map[string]any{"ask": id, "asker": "a1", "at": at, "phase": "build"})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func endedStep(id, state string, started time.Time) []byte {
|
||||
b, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: started, Ended: started.Add(time.Second)})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return b
|
||||
// askTaken, askProgress and askEnded are what the holder of the Go build seat says on node n1 (design 47 G §496, §602).
|
||||
func askTaken(w *Watchman, id string, at time.Time) {
|
||||
w.AskTaken("node-build-go", "n1", askBody(id, at))
|
||||
}
|
||||
func askProgress(w *Watchman, id string, at time.Time) {
|
||||
w.StepProgress("node-build-go", "n1", askBody(id, at))
|
||||
}
|
||||
func askEnded(w *Watchman, id string) { w.AskEnded("node-build-go", askBody(id, t0)) }
|
||||
|
||||
func pulseKeyOf(t *testing.T, bucket, key string) string {
|
||||
t.Helper()
|
||||
@@ -271,7 +271,7 @@ func rebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
|
||||
func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(30 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0.Add(10*time.Second)), Put)
|
||||
askProgress(w, "askid1", t0.Add(10*time.Second))
|
||||
w.SeedPresence("ace", false, t0)
|
||||
w.Reconcile()
|
||||
if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }); err != nil {
|
||||
@@ -290,7 +290,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
|
||||
live := pulseKeyOf(t, checksKey, "askid1")
|
||||
w.SeedPulse(stale)
|
||||
w.SeedPulse(live)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
w.Reconcile()
|
||||
if !slices.Equal(o.pulseDeletes, []string{stale}) {
|
||||
t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale)
|
||||
@@ -303,7 +303,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
|
||||
func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(30 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // the key found at start, last renewed 30s ago
|
||||
askProgress(w, "askid1", t0) // the key found at start, last renewed 30s ago
|
||||
k := pulseKeyOf(t, checksKey, "askid1")
|
||||
if o.ttls[k] != 90*time.Second {
|
||||
t.Fatalf("ttl = %v, want 90s (last renewed 30s ago)", o.ttls[k])
|
||||
@@ -321,7 +321,7 @@ func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
|
||||
func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
if len(o.pulses) != 0 {
|
||||
t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses)
|
||||
}
|
||||
@@ -337,7 +337,7 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
|
||||
func TestWaitedInWords(t *testing.T) { // 1135/AC3
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(200 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
w.SeedPresence("ace", false, t0)
|
||||
w.SeedPresence("part.mesh-delivery-view.ace", false, t0)
|
||||
w.Reconcile()
|
||||
@@ -357,13 +357,13 @@ func TestWaitedInWords(t *testing.T) { // 1135/AC3
|
||||
func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
|
||||
w, o, c := setup()
|
||||
k := pulseKeyOf(t, checksKey, "askid1")
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
if o.ttls[k] != 120*time.Second {
|
||||
t.Fatalf("ttl = %v, want the full 120s", o.ttls[k])
|
||||
}
|
||||
o.ttls[k] = 0
|
||||
c.t = t0.Add(100 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the builder re-puts the key: same check, started long ago
|
||||
askProgress(w, "askid1", c.t) // the builder re-puts the key: same check, started long ago
|
||||
if o.ttls[k] != 120*time.Second {
|
||||
t.Fatalf("a heartbeat left the pulse as it was: ttl = %v, want the full 120s again", o.ttls[k])
|
||||
}
|
||||
@@ -372,7 +372,7 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
|
||||
}
|
||||
// The same revision heard twice changes nothing.
|
||||
o.ttls[k] = 0
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put)
|
||||
askProgress(w, "askid1", c.t)
|
||||
if o.ttls[k] != 0 {
|
||||
t.Fatal("one revision rewrote the pulse twice")
|
||||
}
|
||||
@@ -381,10 +381,10 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
|
||||
func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(10 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
w.Connect("node.ace")
|
||||
w.Disconnect("node.ace")
|
||||
w.Step("go", "n1.askid1", nil, Delete) // the check ended
|
||||
askEnded(w, "askid1") // the check ended
|
||||
w.Connect("node.ace")
|
||||
if len(o.pulses) != 0 {
|
||||
t.Fatalf("pulses left: %v", o.pulses)
|
||||
@@ -399,9 +399,9 @@ func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
|
||||
|
||||
func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1135/AC4
|
||||
w, o, c := setup()
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
c.t = t0.Add(50 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the last heartbeat
|
||||
askProgress(w, "askid1", c.t) // the last heartbeat
|
||||
c.t = t0.Add(170 * time.Second)
|
||||
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
|
||||
s := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
|
||||
@@ -412,7 +412,7 @@ func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1
|
||||
|
||||
func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
|
||||
w, o, c := setup()
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
pk := pulseKeyOf(t, checksKey, "askid1")
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
w.Pulse(pk, Expired, c.t)
|
||||
@@ -421,13 +421,13 @@ func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
|
||||
t.Fatal("stuck written as moved at expiry")
|
||||
}
|
||||
c.t = t0.Add(300 * time.Second)
|
||||
w.Step("go", "n1.askid1", nil, Delete)
|
||||
askEnded(w, "askid1")
|
||||
s := o.stuck[sk]
|
||||
if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to end on n1" {
|
||||
t.Fatalf("stuck after the move = %+v, want the same key moved", s)
|
||||
}
|
||||
n := o.stuckWrites
|
||||
w.Step("go", "n1.askid1", nil, Delete)
|
||||
askEnded(w, "askid1")
|
||||
if o.stuckWrites != n {
|
||||
t.Fatal("a stuck key already moved was written again")
|
||||
}
|
||||
@@ -435,10 +435,10 @@ func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
|
||||
|
||||
func TestAnAskEndedAfterStuckMovesIt(t *testing.T) { // 1135/AC4
|
||||
w, o, c := setup()
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
|
||||
w.Step("go", "n1.askid1", nil, Delete) // the ask ended at last
|
||||
askEnded(w, "askid1") // the ask ended at last
|
||||
if !o.stuck[stuckKeyOf(t, checksKey, "askid1")].Moved {
|
||||
t.Fatal("an ask taken after it was stuck is still stuck")
|
||||
}
|
||||
@@ -467,7 +467,7 @@ func TestOnlyTheExpiryOfItsOwnPulseWritesStuck(t *testing.T) { // 1135/AC5
|
||||
|
||||
func expiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
|
||||
w, o, c := setup()
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
|
||||
s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
|
||||
@@ -481,9 +481,9 @@ func expiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
|
||||
|
||||
func ownDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
|
||||
w, o, c := setup()
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
pk := pulseKeyOf(t, checksKey, "askid1")
|
||||
w.Step("go", "n1.askid1", nil, Delete)
|
||||
askEnded(w, "askid1")
|
||||
w.Pulse(pk, Delete, c.t) // the delete the watchman made, heard on its own watch
|
||||
w.Pulse(pk, Expired, c.t)
|
||||
w.Pulse("mesh-delivery-watchman_pulse.unknown.key", Expired, c.t)
|
||||
@@ -496,12 +496,12 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8
|
||||
w, o, c := setup()
|
||||
w.SetBound(2 * time.Second)
|
||||
c.t = t0.Add(500 * time.Millisecond)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 1500*time.Millisecond {
|
||||
t.Fatalf("ttl = %v, want 1.5s", got)
|
||||
}
|
||||
c.t = t0.Add(3 * time.Second)
|
||||
w.Step("go", "n1.askid2", runningStep("askid2", t0), Put)
|
||||
askProgress(w, "askid2", t0)
|
||||
if _, ok := o.stuck[stuckKeyOf(t, checksKey, "askid2")]; !ok {
|
||||
t.Fatalf("a heartbeat older than the bound is stuck at once; have %v", o.stuck)
|
||||
}
|
||||
@@ -510,27 +510,25 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8
|
||||
func TestAnAskTakenByANodeHasAPulseFromWhenItStartedAndEndsWithTheStep(t *testing.T) { // 1908, 1909
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(10 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(w, "askid1", t0)
|
||||
if len(o.pulses) != 1 {
|
||||
t.Fatalf("pulses = %v, want the taken ask's", o.pulses)
|
||||
}
|
||||
for _, state := range []string{buildstep.StepPassed, buildstep.StepFailed, buildstep.StepRefused} {
|
||||
w2, o2, c2 := setup()
|
||||
c2.t = t0.Add(10 * time.Second)
|
||||
w2.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
w2.Step("go", "n1.askid1", endedStep("askid1", state, t0), Put)
|
||||
if len(o2.pulses) != 0 && o2.pulses[pulseKeyOf(t, checksKey, "askid1")].Key != "" {
|
||||
t.Errorf("%s: the pulse of an ended ask stands: %v", state, o2.pulses)
|
||||
}
|
||||
_ = c2
|
||||
w2, o2, c2 := setup()
|
||||
c2.t = t0.Add(10 * time.Second)
|
||||
askTaken(w2, "askid1", t0)
|
||||
askEnded(w2, "askid1")
|
||||
if len(o2.pulses) != 0 && o2.pulses[pulseKeyOf(t, checksKey, "askid1")].Key != "" {
|
||||
t.Errorf("the pulse of an ended ask stands: %v", o2.pulses)
|
||||
}
|
||||
_ = c2
|
||||
_ = c
|
||||
}
|
||||
|
||||
func TestARunningAskFoundAtStartGetsItsPulse(t *testing.T) { // 1909
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(30 * time.Second)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // what the steps held when the watchman started
|
||||
askProgress(w, "askid1", t0) // what the steps held when the watchman started
|
||||
p, ok := o.pulses[pulseKeyOf(t, checksKey, "askid1")]
|
||||
if !ok || p.Key != "askid1" {
|
||||
t.Fatalf("pulses = %v", o.pulses)
|
||||
@@ -540,35 +538,54 @@ func TestARunningAskFoundAtStartGetsItsPulse(t *testing.T) { // 1909
|
||||
}
|
||||
}
|
||||
|
||||
func TestAPausedKeyOrAnotherNodesNonStepIsLeft(t *testing.T) { // 1908
|
||||
func TestAnEventThatDoesNotReadOrIsNoBuildSeatsIsLeft(t *testing.T) { // 1908
|
||||
w, o, _ := setup()
|
||||
w.Step("go", "paused.n1", []byte(`{}`), Put)
|
||||
w.Step("go", "n1.not-an-id!", []byte(`{}`), Put)
|
||||
w.Step("go", "n1.askid1", []byte(`{"v":99}`), Put)
|
||||
w.StepProgress("node-build-go", "n1", []byte(`{"v":99}`))
|
||||
w.StepProgress("node-build-go", "n1", askBody("", t0))
|
||||
w.AskTaken("node-build-go", "n1", []byte(`nope`))
|
||||
w.AskTaken("mesh-wants", "n1", askBody("askid1", t0))
|
||||
if len(o.pulses) != 0 || len(o.stuck) != 0 {
|
||||
t.Errorf("pulses %v stuck %v", o.pulses, o.stuck)
|
||||
}
|
||||
}
|
||||
|
||||
func TestARunningStepOfTenMinutesUnderItsBoundIsNotStuck(t *testing.T) { // 1908
|
||||
func TestARunningStepThatSaysItsProgressIsNeverStuckAndOneWhoseHolderStopsIs(t *testing.T) { // design 47 G §496, §602
|
||||
o, c := newOut(), &clock{t: t0}
|
||||
w := New(o, c.now) // the step's own bound: 30 minutes
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
c.t = t0.Add(10 * time.Minute)
|
||||
// no heartbeat is written, and the server expires the pulse at its ttl: 30 minutes from the start, not 120 s
|
||||
if len(o.stuck) != 0 {
|
||||
t.Fatalf("a build of 10 minutes is stuck: %v", o.stuck)
|
||||
w := New(o, c.now)
|
||||
askTaken(w, "askid1", t0)
|
||||
k := pulseKeyOf(t, checksKey, "askid1")
|
||||
if got := o.ttls[k]; got != ProgressBound {
|
||||
t.Errorf("ttl = %v, want the progress bound %v", got, ProgressBound)
|
||||
}
|
||||
if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 30*time.Minute {
|
||||
t.Errorf("ttl = %v, want the step's bound of 30m from its start", got)
|
||||
// a 20-minute step: its progress each minute renews the clock
|
||||
for m := 1; m <= 20; m++ {
|
||||
c.t = t0.Add(time.Duration(m) * time.Minute)
|
||||
askProgress(w, "askid1", c.t)
|
||||
if o.ttls[k] != ProgressBound {
|
||||
t.Fatalf("minute %d: ttl = %v, want the full bound again", m, o.ttls[k])
|
||||
}
|
||||
}
|
||||
if len(o.stuck) != 0 {
|
||||
t.Fatalf("a step saying its progress is stuck: %v", o.stuck)
|
||||
}
|
||||
// its holder stops: the pulse expires, and the ask is stuck from its last progress
|
||||
c.t = t0.Add(26 * time.Minute)
|
||||
w.Pulse(k, Expired, c.t)
|
||||
s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
|
||||
if !ok || s.Moved || !s.Since.Equal(t0.Add(20*time.Minute)) {
|
||||
t.Fatalf("stuck = %+v, %v", s, ok)
|
||||
}
|
||||
askEnded(w, "askid1")
|
||||
if !o.stuck[stuckKeyOf(t, checksKey, "askid1")].Moved {
|
||||
t.Fatal("the end of the ask left it stuck")
|
||||
}
|
||||
}
|
||||
|
||||
func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
|
||||
o, c := newOut(), &clock{t: t0}
|
||||
w := New(o, c.now)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
c.t = t0.Add(31 * time.Minute)
|
||||
askProgress(w, "askid1", t0)
|
||||
c.t = t0.Add(6 * time.Minute)
|
||||
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
|
||||
s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
|
||||
if !ok || s.Moved || !s.Since.Equal(t0) {
|
||||
@@ -576,7 +593,7 @@ func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
|
||||
}
|
||||
// and one found past its bound at the watchman's start is stuck at once
|
||||
o2, c2 := newOut(), &clock{t: t0.Add(time.Hour)}
|
||||
New(o2, c2.now).Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
askProgress(New(o2, c2.now), "askid1", t0)
|
||||
if _, ok := o2.stuck[stuckKeyOf(t, checksKey, "askid1")]; !ok || len(o2.pulses) != 0 {
|
||||
t.Errorf("stuck %v pulses %v", o2.stuck, o2.pulses)
|
||||
}
|
||||
@@ -683,9 +700,6 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
|
||||
closed bool
|
||||
late bool // read after the start's end (Awaits, ReadLate)
|
||||
}{
|
||||
{"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true, false},
|
||||
{"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false, false},
|
||||
{"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false, false},
|
||||
{"a want deleted while nobody heard", AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, false},
|
||||
{"a want still standing", AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, false},
|
||||
// DLV-M5 (ADR 0615): the wants are read from the assigner's first view, which may come after the start's end; what
|
||||
|
||||
@@ -13,8 +13,6 @@ import (
|
||||
"testing"
|
||||
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
)
|
||||
|
||||
// The controller's manifest reader is mesh-controller internal/catalogue (ParseManifest), an internal package of another
|
||||
@@ -104,9 +102,8 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
// The build modules are no names (design 47 G A-82): the steps are the states of the holders of the build seats.
|
||||
if !reflect.DeepEqual(reads, []string{"mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have",
|
||||
"seat:node-build-go.steps", "seat:node-build-image.steps", "seat:node-build-node.steps"}) {
|
||||
// The build seats' steps are no read (design 47 G §483, A-136): the watchman times an ask from the seats' step-progress.
|
||||
if !reflect.DeepEqual(reads, []string{"mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
|
||||
t.Errorf("reads = %v", reads)
|
||||
}
|
||||
// Design 47 G §368: the version is MAJOR.MINOR and nothing else.
|
||||
@@ -129,11 +126,11 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
|
||||
}
|
||||
// The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). The waits the
|
||||
// indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's.
|
||||
if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_read_modules", "seat:mesh-hand-acts.hand-act", "seat:mesh-module-assigner.assignments"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_read_modules", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably.
|
||||
if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) {
|
||||
if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) {
|
||||
t.Errorf("consumes = %v, want %v", m.Consumes, want)
|
||||
}
|
||||
// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
|
||||
@@ -171,16 +168,13 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
|
||||
return e
|
||||
}
|
||||
for name, row := range map[string]struct{ got, want string }{
|
||||
"module": {watch.WatchmanModule, m.Module},
|
||||
"presence": {watch.PresenceBucket, m.Module + "_" + m.State[0].Name},
|
||||
"pulse": {watch.PulseBucket, m.Module + "_" + m.State[1].Name},
|
||||
"stuck": {watch.StuckBucket, m.Module + "_" + m.State[2].Name},
|
||||
"have": {watch.HaveBucket, bucket(m.Reads, "node-engine.have")},
|
||||
"health": {watch.NodeHealthBucket, bucket(m.Reads, "mesh-host.health")},
|
||||
"builds": {watch.BuilderBuildsBucket, bucket(m.Reads, "mesh-delivery-builder.builds")},
|
||||
// The build modules are no names (design 47 G A-82): the steps are read as the states of the holders of the build seats.
|
||||
"steps": {"seat:node-build-go.steps", entry(m.Reads, "seat:node-build-go.steps")},
|
||||
"holders": {"seat:" + seatlist.HolderSeat + "." + seatlist.HolderVerb, entry(m.Invokes, "seat:"+seatlist.HolderSeat+"."+seatlist.HolderVerb)},
|
||||
"module": {watch.WatchmanModule, m.Module},
|
||||
"presence": {watch.PresenceBucket, m.Module + "_" + m.State[0].Name},
|
||||
"pulse": {watch.PulseBucket, m.Module + "_" + m.State[1].Name},
|
||||
"stuck": {watch.StuckBucket, m.Module + "_" + m.State[2].Name},
|
||||
"have": {watch.HaveBucket, bucket(m.Reads, "node-engine.have")},
|
||||
"health": {watch.NodeHealthBucket, bucket(m.Reads, "mesh-host.health")},
|
||||
"builds": {watch.BuilderBuildsBucket, bucket(m.Reads, "mesh-delivery-builder.builds")},
|
||||
"wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")},
|
||||
"provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")},
|
||||
"modules": {watch.ModulesBucket, bucket(m.Consumes, "mesh-delivery-indexer.views.modules")},
|
||||
|
||||
@@ -19,17 +19,13 @@
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-host.health",
|
||||
"mesh-provisioner.provisions",
|
||||
"node-engine.have",
|
||||
"seat:node-build-go.steps",
|
||||
"seat:node-build-image.steps",
|
||||
"seat:node-build-node.steps"
|
||||
"node-engine.have"
|
||||
],
|
||||
"invokes": [
|
||||
"mesh-delivery-assigner.assigner_read_wants",
|
||||
"mesh-delivery-indexer.indexer_pending",
|
||||
"mesh-delivery-indexer.indexer_read_modules",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-module-assigner.assignments"
|
||||
"seat:mesh-hand-acts.hand-act"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-delivery-assigner.views.wants",
|
||||
@@ -52,6 +48,15 @@
|
||||
"mesh-grants.filling",
|
||||
"mesh-grants.full",
|
||||
"mesh-grants.room",
|
||||
"node-build-go.ask-taken",
|
||||
"node-build-go.step-progress",
|
||||
"node-build-go.step-ended",
|
||||
"node-build-image.ask-taken",
|
||||
"node-build-image.step-progress",
|
||||
"node-build-image.step-ended",
|
||||
"node-build-node.ask-taken",
|
||||
"node-build-node.step-progress",
|
||||
"node-build-node.step-ended",
|
||||
"mesh-pipeline-runner.pipeline-started",
|
||||
"mesh-pipeline-runner.step-started",
|
||||
"mesh-pipeline-runner.step-ended",
|
||||
|
||||
Reference in New Issue
Block a user