WM6: the watchman hears an acting part's wait-held and wait-ended and writes acting-part-late once the wait's bound has run from clock-from
Taken from 2d796fe8 (its wait half only; the store and record halves are WM7's, built). The body is the seat mesh-wants' contract Wait. Builds: design 47 G §266, §269, §277
This commit is contained in:
@@ -7,6 +7,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"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"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
@@ -171,6 +172,10 @@ func recordSubject(w *watch.Watchman, subject string, body []byte) bool {
|
||||
}
|
||||
seat, event := parts[2], parts[4]
|
||||
switch {
|
||||
case event == meshwants.EventWaitHeld:
|
||||
w.WaitHeld(seat, body)
|
||||
case event == meshwants.EventWaitEnded:
|
||||
w.WaitEnded(seat, body)
|
||||
case event == meshdelivery.EventFilling || event == meshdelivery.EventFull || event == meshdelivery.EventRoom:
|
||||
w.StoreFill(seat, event, body)
|
||||
case seat == meshdelivery.Seat && event == meshdelivery.EventRecordOpened:
|
||||
|
||||
@@ -16,6 +16,7 @@ require (
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
|
||||
@@ -52,6 +53,5 @@ require (
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000 // indirect
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 // indirect
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 // indirect
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 // indirect
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry v0.0.0-00010101000000-000000000000 // indirect
|
||||
)
|
||||
|
||||
@@ -944,6 +944,7 @@ var SignalRows = []struct{ Row, Signal, Bound, Kind string }{
|
||||
{"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing},
|
||||
{"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy},
|
||||
{"take-in", "a source revision or a delivery group the indexer holds back (ADR 0633 §5)", TakeInRevisionBound.String() + " a revision, " + TakeInGroupBound.String() + " a group", KindTakeInWaits},
|
||||
{actingRow, "an acting part's wait held (the seats' wait-held, ended by wait-ended)", "the wait's own bound", KindActingPartLate},
|
||||
{storeRow, "a store of the delivery's parts filling or full (the seats' filling, full and room)", "filling from " + fmt.Sprintf("%.0f", FillingFrom*100) + " %, full urgent", KindStoreFilling},
|
||||
{recordRow, "a delivery record open (the seat mesh-delivery's record-opened, ended by record-ended)", RecordOpenBound.String(), KindRecordOpen},
|
||||
{"provisions", "a provision of mesh-provisioner in waits (ADR 0570 §7)", "at once", KindNeedsOperator},
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
)
|
||||
|
||||
// What the holders of an acting part's wait say, heard as the seats' events (design 47 G §266, §269, ADR 0596 §2), never read
|
||||
// from their state: `wait-held` and `wait-ended` of a wait (the seat mesh-wants, and the seat mesh-pipeline-runner for its
|
||||
// `pipeline` step). A wait held past its bound is the finding `acting-part-late`, timed by that bound alone from the time
|
||||
// `wait-held` carries as `clock-from` (its `since` where it carries none), and ended by `wait-ended`, whoever ended it (the
|
||||
// part's new version running, or a release). Its pulse is its bound, written at the event, so a restart loses none.
|
||||
|
||||
// actingRow is the row of the signals table of an acting part's wait.
|
||||
const actingRow = "acting-wait"
|
||||
|
||||
func waitID(seat string, a meshwants.Wait) string {
|
||||
return strings.Join([]string{seat, a.Step, a.Part, a.Node, a.Module}, "-")
|
||||
}
|
||||
|
||||
// WaitHeld hears a seat's `wait-held`: the finding is due when the wait's bound has run from its clock.
|
||||
func (w *Watchman) WaitHeld(seat string, body []byte) {
|
||||
a, ok := decodeWait(w, seat, "wait-held", body)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.waitHeld(seat, a)
|
||||
}
|
||||
|
||||
// WaitEnded hears a seat's `wait-ended`: the wait ran or was released, so its finding and its pulse end.
|
||||
func (w *Watchman) WaitEnded(seat string, body []byte) {
|
||||
a, ok := decodeWait(w, seat, "wait-ended", body)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
id := waitID(seat, a)
|
||||
w.unsignal(actingRow, id)
|
||||
w.unfind(FindingKey(ScopeDelivery, id, KindActingPartLate))
|
||||
}
|
||||
|
||||
func decodeWait(w *Watchman, seat, event string, body []byte) (meshwants.Wait, bool) {
|
||||
var a meshwants.Wait
|
||||
if err := json.Unmarshal(body, &a); err != nil || a.Part == "" || a.Module == "" || a.Since.IsZero() {
|
||||
w.say("the %s of the seat %s does not read, left: %v", event, seat, err)
|
||||
return a, false
|
||||
}
|
||||
return a, true
|
||||
}
|
||||
|
||||
// waitHeld is one wait held, w.mu held.
|
||||
func (w *Watchman) waitHeld(seat string, a meshwants.Wait) {
|
||||
bound, err := time.ParseDuration(a.Bound)
|
||||
if err != nil || bound <= 0 {
|
||||
w.say("the wait of %s at %s says no bound (%q), so it is never late", a.Module, a.Part, a.Bound)
|
||||
return
|
||||
}
|
||||
from := a.Since
|
||||
if a.ClockFrom != nil {
|
||||
from = *a.ClockFrom
|
||||
}
|
||||
id := waitID(seat, a)
|
||||
f := Finding{Scope: ScopeDelivery, ID: id, Kind: KindActingPartLate, Severity: Warning, Source: actingRow, Since: a.Since,
|
||||
Summary: fmt.Sprintf("%s on %s waits for %s to run %s at the step %s, since %s (bound %s)", a.Module, orWords(a.Node, "no node"), a.Part,
|
||||
orWords(a.Version, "its new version"), a.Step, stamp(a.Since), bound),
|
||||
Said: "held for " + bound.String(),
|
||||
Headline: "An acting part's new version has not come",
|
||||
Explanation: "A step waits until " + a.Part + " runs its new version, and its bound of " + bound.String() +
|
||||
" has run, so what the step does is held back.",
|
||||
Needs: "the part's new version to run, or the wait released by hand (the holder's release tool)"}
|
||||
left := bound - w.now().Sub(from)
|
||||
w.unsignal(actingRow, id)
|
||||
w.unfind(f.Key())
|
||||
if left > 0 {
|
||||
w.signal(actingRow, id, left, signalDue{Do: dueFind, Finding: f})
|
||||
return
|
||||
}
|
||||
w.find(f)
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
)
|
||||
|
||||
func heldWait(module string, since time.Time, from *time.Time) []byte {
|
||||
b, _ := json.Marshal(meshwants.Wait{Step: "place", Part: "mesh-delivery-indexer", Version: "src2:new", Node: "ace", Module: module,
|
||||
Since: since, ClockFrom: from, Bound: "30m"})
|
||||
return b
|
||||
}
|
||||
|
||||
// design 47 G §266, §269 (WM6): a wait held is the finding `acting-part-late` once its bound has run from the time `wait-held`
|
||||
// carries as `clock-from` (its `since` where it carries none), not within it; `wait-ended` ends it.
|
||||
func TestAnActingPartsWaitIsLateByItsBoundFromItsClock(t *testing.T) {
|
||||
id := "mesh-wants-place-mesh-delivery-indexer-ace-web"
|
||||
key := FindingKey(ScopeDelivery, id, KindActingPartLate)
|
||||
t.Run("within the bound nothing, past it a finding, wait-ended ends it", func(t *testing.T) {
|
||||
w, o, c := setup()
|
||||
w.WaitHeld("mesh-wants", heldWait("web", c.t, nil))
|
||||
if o.ttls[signalKey(actingRow, id)] != 30*time.Minute || len(o.findings) != 0 {
|
||||
t.Fatalf("within the bound: ttl %v, findings %v", o.ttls[signalKey(actingRow, id)], o.findings)
|
||||
}
|
||||
expire(w, c, signalKey(actingRow, id), 30*time.Minute)
|
||||
if _, ok := o.findings[key]; !ok {
|
||||
t.Fatalf("past its bound: %v", o.findings)
|
||||
}
|
||||
w.WaitEnded("mesh-wants", heldWait("web", c.t, nil))
|
||||
if len(o.findings) != 0 {
|
||||
t.Fatalf("wait-ended left %v", o.findings)
|
||||
}
|
||||
})
|
||||
t.Run("the bound counts from clock-from, not from since", func(t *testing.T) {
|
||||
w, o, c := setup()
|
||||
from := c.t.Add(-20 * time.Minute)
|
||||
w.WaitHeld("mesh-wants", heldWait("web", c.t.Add(-time.Hour), &from))
|
||||
if got := o.ttls[signalKey(actingRow, id)]; got != 10*time.Minute {
|
||||
t.Fatalf("a wait whose clock began 20 minutes ago has %v left, want 10m", got)
|
||||
}
|
||||
})
|
||||
t.Run("a wait whose bound ran before it was heard is a finding at once", func(t *testing.T) {
|
||||
w, o, c := setup()
|
||||
w.WaitHeld("mesh-wants", heldWait("web", c.t.Add(-2*time.Hour), nil))
|
||||
if _, ok := o.findings[key]; !ok {
|
||||
t.Fatalf("a late wait: %v", o.findings)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -436,7 +436,7 @@ func signalKind(value []byte) bool {
|
||||
return false
|
||||
}
|
||||
switch f.Source {
|
||||
case "provisioner.failing", "S8", "S9", "S10", "S11", rowBusState, "take-in":
|
||||
case "provisioner.failing", "S8", "S9", "S10", "S11", rowBusState, "take-in", actingRow:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
|
||||
@@ -127,7 +127,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
|
||||
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-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", "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-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).
|
||||
|
||||
@@ -57,6 +57,10 @@
|
||||
"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-broker.created",
|
||||
"mesh-broker.refused",
|
||||
"mesh-conditions.findings-asked"
|
||||
|
||||
Reference in New Issue
Block a user