Names N3 (watchman): the watchman names its own buckets from its credential's module
Its states presence, pulse and stuck and its first views were named from the constant WatchmanModule; they are now `<cred.Module>_<state>` (watch.BucketsOf), and the watchman's stuck keys of presence take its own presence bucket's name. Builds: design 47 G A-296 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -24,6 +24,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
"git.novox.be/novox/mesh-sdk/go/views/viewstest"
|
||||
|
||||
"mesh-delivery-watchman/internal/deadletters"
|
||||
@@ -92,6 +93,12 @@ func sayAsk(t *testing.T, ctx context.Context, js jetstream.JetStream, event, id
|
||||
}
|
||||
}
|
||||
|
||||
// own and firstViewsBucket are the beds' watchman's own buckets, named from the module its credential names (A-296).
|
||||
var (
|
||||
own = watch.BucketsOf(watch.WatchmanModule)
|
||||
firstViewsBucket = watch.WatchmanModule + "_" + views.StatusState
|
||||
)
|
||||
|
||||
// findingEventsSubject is the subject the stand-in seat mesh-broker's show answers for the watchman's finding-events.
|
||||
const findingEventsSubject = "mesh.bus." + watch.WatchmanModule + ".finding-events"
|
||||
|
||||
@@ -161,9 +168,9 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
defer cancel()
|
||||
kvs := map[string]jetstream.KeyValue{}
|
||||
for _, c := range []jetstream.KeyValueConfig{
|
||||
{Bucket: watch.PresenceBucket, History: 20},
|
||||
{Bucket: watch.StuckBucket, History: 20},
|
||||
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
|
||||
{Bucket: own.Presence, History: 20},
|
||||
{Bucket: own.Stuck, History: 20},
|
||||
{Bucket: own.Pulse, LimitMarkerTTL: time.Second},
|
||||
{Bucket: watch.AssignerWantsBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
{Bucket: firstViewsBucket, History: 1},
|
||||
@@ -224,7 +231,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
return e, err == nil
|
||||
}
|
||||
online := func(key string) (bool, bool) {
|
||||
e, ok := has(watch.PresenceBucket, key)
|
||||
e, ok := has(own.Presence, key)
|
||||
if !ok {
|
||||
return false, false
|
||||
}
|
||||
@@ -267,10 +274,10 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
askStream, _ := buildstep.AskStream(buildstep.AskGo)
|
||||
const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb"
|
||||
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 })
|
||||
until("a pulse for the ask a node took", func() bool { _, ok := has(own.Pulse, pulseOf(askStream, ask1)); return ok })
|
||||
var stuck meshwatchdogsclient.Stuck
|
||||
until("a stuck key for the ask no node ended", func() bool {
|
||||
e, ok := has(watch.StuckBucket, stuckOf(askStream, ask1))
|
||||
e, ok := has(own.Stuck, stuckOf(askStream, ask1))
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
@@ -283,7 +290,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
}
|
||||
// The node n1, gone offline and never back, is stuck as well.
|
||||
until("a stuck key for node n1 that did not come back", func() bool {
|
||||
_, ok := has(watch.StuckBucket, stuckOf(watch.PresenceBucket, "n1"))
|
||||
_, ok := has(own.Stuck, stuckOf(own.Presence, "n1"))
|
||||
return ok
|
||||
})
|
||||
|
||||
@@ -291,18 +298,18 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
n2 := connect("node.n2")
|
||||
until("node n2 online", func() bool { on, ok := online("n2"); return ok && on })
|
||||
n2.Close()
|
||||
until("a pulse for node n2 gone offline", func() bool { _, ok := has(watch.PulseBucket, pulseOf(watch.PresenceBucket, "n2")); return ok })
|
||||
until("a pulse for node n2 gone offline", func() bool { _, ok := has(own.Pulse, pulseOf(own.Presence, "n2")); return ok })
|
||||
connect("node.n2")
|
||||
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 })
|
||||
until("the pulse of node n2 deleted", func() bool { _, ok := has(own.Pulse, pulseOf(own.Presence, "n2")); return !ok })
|
||||
// An ask a node takes at once is the same: no stuck key, however long the bound passes.
|
||||
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 })
|
||||
until("a pulse for the second ask", func() bool { _, ok := has(own.Pulse, pulseOf(askStream, ask2)); return ok })
|
||||
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 })
|
||||
until("the pulse of the second ask deleted", func() bool { _, ok := has(own.Pulse, 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)} {
|
||||
if _, ok := has(watch.StuckBucket, k); ok {
|
||||
for _, k := range []string{stuckOf(own.Presence, "n2"), stuckOf(askStream, ask2)} {
|
||||
if _, ok := has(own.Stuck, k); ok {
|
||||
t.Errorf("a stuck key %s was written for a move that happened before the bound", k)
|
||||
}
|
||||
}
|
||||
@@ -409,9 +416,9 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
|
||||
defer cancel()
|
||||
kvs := map[string]jetstream.KeyValue{}
|
||||
for _, c := range []jetstream.KeyValueConfig{
|
||||
{Bucket: watch.PresenceBucket, History: 20},
|
||||
{Bucket: watch.StuckBucket, History: 20},
|
||||
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
|
||||
{Bucket: own.Presence, History: 20},
|
||||
{Bucket: own.Stuck, History: 20},
|
||||
{Bucket: own.Pulse, LimitMarkerTTL: time.Second},
|
||||
{Bucket: watch.AssignerWantsBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
{Bucket: firstViewsBucket, History: 1},
|
||||
@@ -443,7 +450,7 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
|
||||
t.Fatal(err)
|
||||
}
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if _, err := kvs[watch.PulseBucket].Get(ctx, pk); err == nil {
|
||||
if _, err := kvs[own.Pulse].Get(ctx, pk); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -490,9 +497,9 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
|
||||
defer cancel()
|
||||
kvs := map[string]jetstream.KeyValue{}
|
||||
for _, c := range []jetstream.KeyValueConfig{
|
||||
{Bucket: watch.PresenceBucket, History: 20},
|
||||
{Bucket: watch.StuckBucket, History: 20},
|
||||
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
|
||||
{Bucket: own.Presence, History: 20},
|
||||
{Bucket: own.Stuck, History: 20},
|
||||
{Bucket: own.Pulse, LimitMarkerTTL: time.Second},
|
||||
{Bucket: watch.AssignerWantsBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
{Bucket: firstViewsBucket, History: 1},
|
||||
@@ -556,11 +563,11 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
|
||||
}
|
||||
t.Fatalf("never saw: %s", what)
|
||||
}
|
||||
has := func(key string) bool { _, err := kvs[watch.PulseBucket].Get(ctx, key); return err == nil }
|
||||
has := func(key string) bool { _, err := kvs[own.Pulse].Get(ctx, key); return err == nil }
|
||||
until("a pulse for the want that stood before the start", func() bool { return has(pulse("n2", "web")) })
|
||||
putWant("n2", "mail", fpNew)
|
||||
until("a pulse for a want written after the start", func() bool { return has(pulse("n2", "mail")) })
|
||||
e, err := kvs[watch.PulseBucket].Get(ctx, pulse("n2", "web"))
|
||||
e, err := kvs[own.Pulse].Get(ctx, pulse("n2", "web"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -575,7 +582,7 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
|
||||
for _, m := range []string{"web"} {
|
||||
wk, _ := meshwantsclient.WantKey("n2", m)
|
||||
sk, _ := meshwatchdogsclient.StuckKey(watch.AssignerWantsBucket, wk)
|
||||
if _, err := kvs[watch.StuckBucket].Get(ctx, sk); err == nil {
|
||||
if _, err := kvs[own.Stuck].Get(ctx, sk); err == nil {
|
||||
t.Errorf("a stuck key was written for the want of %s that was applied", m)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -189,11 +189,6 @@ func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time {
|
||||
return at
|
||||
}
|
||||
|
||||
// The watchman's own states of its first views (views.StatusState); its findings of a first view unanswered past its
|
||||
// bound go to its findings bucket with the others (ADR 0615 §6). The owners' states it follows are the assigner's wants
|
||||
// and the indexer's modules (DLV-M5).
|
||||
var firstViewsBucket = watch.WatchmanModule + "_" + views.StatusState
|
||||
|
||||
const (
|
||||
wantsState = "wants"
|
||||
// The owners' views families, as their manifests' `streams` name them.
|
||||
@@ -222,11 +217,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants
|
||||
// and the node-engine's haves only make the pulses of a move: without the haves the watchman watches presence and the
|
||||
// build seats' steps, and says that moves are not watched (issue 1945).
|
||||
// They are named from its credential's module (design 47 G A-296).
|
||||
own := watch.BucketsOf(cred.Module)
|
||||
var kvs struct{ presence, pulse, stuck jetstream.KeyValue }
|
||||
for _, b := range []struct {
|
||||
name string
|
||||
into *jetstream.KeyValue
|
||||
}{{watch.PresenceBucket, &kvs.presence}, {watch.PulseBucket, &kvs.pulse}, {watch.StuckBucket, &kvs.stuck}} {
|
||||
}{{own.Presence, &kvs.presence}, {own.Pulse, &kvs.pulse}, {own.Stuck, &kvs.stuck}} {
|
||||
if *b.into, err = process.EssentialBucket(ctx, js, "watchman", b.name, "say who is online or what stopped moving"); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -265,7 +262,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
}
|
||||
|
||||
w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: held, js: js, cred: cred}, time.Now)
|
||||
w := watch.New(cred.Module, &kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: held, js: js, cred: cred}, time.Now)
|
||||
w.Logf(logger.Printf)
|
||||
// What the wants and the modules hold is ended by their own late read (watch.ReadLate), never by the start's end.
|
||||
w.Awaits(watch.AssignerWantsBucket)
|
||||
@@ -416,7 +413,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
|
||||
errs := make(chan error, 16)
|
||||
// The first views of the wants and the modules, asked now and read each on its own goroutine once answered (ADR 0615).
|
||||
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf))
|
||||
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, cred.Module+"_"+views.StatusState, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf))
|
||||
// Every part has registered: the hub pulls the module's durable, and the watchman's other consumers open.
|
||||
hearConsumed(ctx, js, cred, hub, w, binding, logger.Printf)
|
||||
defer hub.Stop()
|
||||
|
||||
@@ -54,7 +54,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
|
||||
})
|
||||
t.Run("DLV-W1: the consumed events reach the watchman by their subjects", func(t *testing.T) {
|
||||
out := &findingsOut{findings: map[string]bool{}}
|
||||
w := watch.New(out, time.Now)
|
||||
w := watch.New(watch.WatchmanModule, out, time.Now)
|
||||
say := func(string, ...any) {}
|
||||
// The bus module's own stream is its history, never heard here: only the seat's subject (ruling 13:0x).
|
||||
consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l","why":"no room"}`), say)
|
||||
@@ -77,7 +77,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
|
||||
})
|
||||
t.Run("design 47 G §500: the indexer's manifest-refused reaches the watchman by its subject", func(t *testing.T) {
|
||||
out := &findingsOut{findings: map[string]bool{}}
|
||||
w := watch.New(out, time.Now)
|
||||
w := watch.New(watch.WatchmanModule, out, time.Now)
|
||||
consumed(w, "mesh.seat.mesh-catalog.event.manifest-refused", []byte(`{"module":"m","digest":"d","problems":["x"]}`), func(string, ...any) {})
|
||||
if !out.findings[watch.FindingKey(watch.ScopeModule, "m", watch.KindManifestRefused)] {
|
||||
t.Fatalf("a refused manifest reached %v", out.findings)
|
||||
@@ -85,7 +85,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
|
||||
})
|
||||
t.Run("design 47 G §499, §541: a node-engine's wait-held and wait-ended reach the watchman by their subjects", func(t *testing.T) {
|
||||
out := &findingsOut{findings: map[string]bool{}}
|
||||
w := watch.New(out, time.Now)
|
||||
w := watch.New(watch.WatchmanModule, out, time.Now)
|
||||
say := func(string, ...any) {}
|
||||
body := fmt.Sprintf(`{"step":"compose","reason":"entry-missing","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339))
|
||||
consumed(w, "mesh.seat.node-engine.event.wait-held.ace.web", []byte(body), say)
|
||||
@@ -99,7 +99,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
|
||||
})
|
||||
t.Run("design 47 G A-192: a module's wait-held on its stream, naming a part, reaches the watchman", func(t *testing.T) {
|
||||
out := &findingsOut{findings: map[string]bool{}}
|
||||
w := watch.New(out, time.Now)
|
||||
w := watch.New(watch.WatchmanModule, out, time.Now)
|
||||
body := fmt.Sprintf(`{"step":"building","part":"mesh-sdk@v0.1.14","module":"web","fingerprint":"src2:f","bound":"1h0m0s","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339))
|
||||
consumed(w, "mesh.stream.mesh-delivery-builder.event.wait-held.web.src2:f", []byte(body), func(string, ...any) {})
|
||||
if len(out.findings) != 1 {
|
||||
@@ -120,7 +120,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
|
||||
"unreadable": func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"waits":`), nil },
|
||||
} {
|
||||
out := &findingsOut{findings: map[string]bool{}}
|
||||
w := watch.New(out, time.Now)
|
||||
w := watch.New(watch.WatchmanModule, out, time.Now)
|
||||
const key = "m.00000000000000000042"
|
||||
old := watch.Finding{Scope: watch.ScopeDelivery, ID: key, Kind: watch.KindTakeInWaits, Severity: watch.Warning, Source: "take-in",
|
||||
Since: time.Now().Add(-3 * time.Hour), Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"}
|
||||
|
||||
@@ -15,15 +15,19 @@ const (
|
||||
// IndexerModule is the holder of the seat mesh-catalog, named from this module's manifest (consumes and invokes of mesh-delivery-indexer).
|
||||
IndexerModule = "mesh-delivery-indexer"
|
||||
|
||||
// The watchman's own states of the seat mesh-watchdogs.
|
||||
PresenceBucket = WatchmanModule + "_" + meshwatchdogs.StatePresence
|
||||
PulseBucket = WatchmanModule + "_" + meshwatchdogs.StatePulse
|
||||
StuckBucket = WatchmanModule + "_" + meshwatchdogs.StateStuck
|
||||
|
||||
// AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants).
|
||||
AssignerWantsBucket = AssignerModule + "_wants"
|
||||
)
|
||||
|
||||
// OwnBuckets are the watchman's own states of the seat mesh-watchdogs, `<module>_<state>`.
|
||||
type OwnBuckets struct{ Presence, Pulse, Stuck string }
|
||||
|
||||
// BucketsOf names the watchman's own buckets from module, its credential's, never a constant (design 47 G A-296).
|
||||
func BucketsOf(module string) OwnBuckets {
|
||||
return OwnBuckets{Presence: module + "_" + meshwatchdogs.StatePresence, Pulse: module + "_" + meshwatchdogs.StatePulse,
|
||||
Stuck: module + "_" + meshwatchdogs.StateStuck}
|
||||
}
|
||||
|
||||
// The findings a value of another part is raised as, named by their words.
|
||||
const (
|
||||
FindingManifestOutgrown = "manifest-outgrown"
|
||||
|
||||
@@ -92,7 +92,7 @@ func (w *Watchman) settle(key string) {
|
||||
defer w.applyCheck(node) // S21 is judged only while the machine is heard
|
||||
}
|
||||
if online {
|
||||
w.moved(PresenceBucket, key)
|
||||
w.moved(w.own.Presence, key)
|
||||
return
|
||||
}
|
||||
w.ensureOffline(key)
|
||||
@@ -101,7 +101,7 @@ func (w *Watchman) settle(key string) {
|
||||
// ensureOffline mirrors the return of a node or part that is offline.
|
||||
func (w *Watchman) ensureOffline(key string) {
|
||||
p := w.presence[key]
|
||||
w.mirror(PresenceBucket, key, waitedOf(key), "", p.since, 0)
|
||||
w.mirror(w.own.Presence, key, waitedOf(key), "", p.since, 0)
|
||||
}
|
||||
|
||||
// SeedPresence tells the watchman a presence key as it stands in its bucket at start, with the time it was written.
|
||||
|
||||
@@ -235,7 +235,7 @@ func (w *Watchman) Power(node, state string) {
|
||||
// moved is S1, the machine silent, unless it said it is away; any other key not moved is `delivery.<key>.stuck`; a key
|
||||
// moved ends its finding.
|
||||
func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) {
|
||||
if s.Bucket == PresenceBucket {
|
||||
if s.Bucket == w.own.Presence {
|
||||
if node, part, _, err := meshwatchdogsclient.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" {
|
||||
key := FindingKey(ScopeMachine, node, KindSilent)
|
||||
if s.Moved {
|
||||
|
||||
@@ -56,6 +56,7 @@ type pulse struct {
|
||||
// safe to call from several watches at once.
|
||||
type Watchman struct {
|
||||
mu sync.Mutex
|
||||
own OwnBuckets // its own buckets, named from its credential's module (A-296)
|
||||
out Out
|
||||
now func() time.Time
|
||||
logf func(format string, args ...any)
|
||||
@@ -120,9 +121,9 @@ type Watchman struct {
|
||||
opened map[string]Finding
|
||||
}
|
||||
|
||||
// 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{},
|
||||
// New is the watchman of module (its credential's), writing to out, reading the time from now.
|
||||
func New(module string, out Out, now func() time.Time) *Watchman {
|
||||
return &Watchman{own: BucketsOf(module), 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, progressBound: ProgressBound,
|
||||
findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, pendingBranch: map[string]string{}, takenEntries: map[string][]string{}, recordMilestone: map[string]string{}, away: map[string]bool{},
|
||||
advised: map[string]int{}, advisedSince: map[string]time.Time{},
|
||||
|
||||
@@ -95,7 +95,7 @@ func (c *clock) now() time.Time { return c.t }
|
||||
|
||||
func setup() (*Watchman, *fakeOut, *clock) {
|
||||
o, c := newOut(), &clock{t: t0}
|
||||
w := New(o, c.now)
|
||||
w := New(testModule, o, c.now)
|
||||
w.SetStepBound(meshwatchdogsclient.PulseMoveBound) // the older tests move a step as they move any key: stuck at the pulse's bound
|
||||
return w, o, c
|
||||
}
|
||||
@@ -109,7 +109,8 @@ func checkValue(started time.Time, head string) []byte {
|
||||
|
||||
const (
|
||||
checksKey = "SEAT_NODE_BUILD_GO"
|
||||
presenceNS = PresenceBucket
|
||||
testModule = "a-watchman"
|
||||
presenceNS = testModule + "_presence"
|
||||
)
|
||||
|
||||
// askBody is the body of a build seat's step-progress, ask-taken or step-ended for an ask at a time.
|
||||
@@ -549,7 +550,7 @@ func TestAnEventThatDoesNotReadOrIsNoBuildSeatsIsLeft(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)
|
||||
w := New(testModule, o, c.now)
|
||||
askTaken(w, "askid1", t0)
|
||||
k := pulseKeyOf(t, checksKey, "askid1")
|
||||
if got := o.ttls[k]; got != ProgressBound {
|
||||
@@ -581,7 +582,7 @@ func TestARunningStepThatSaysItsProgressIsNeverStuckAndOneWhoseHolderStopsIs(t *
|
||||
|
||||
func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
|
||||
o, c := newOut(), &clock{t: t0}
|
||||
w := New(o, c.now)
|
||||
w := New(testModule, o, c.now)
|
||||
askProgress(w, "askid1", t0)
|
||||
c.t = t0.Add(6 * time.Minute)
|
||||
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
|
||||
@@ -591,7 +592,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)}
|
||||
askProgress(New(o2, c2.now), "askid1", t0)
|
||||
askProgress(New(testModule, 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)
|
||||
}
|
||||
@@ -662,7 +663,7 @@ func TestAHaveThatCameBeforeTheWantEndsItAtOnceAndARemovedWantHasNoPulse(t *test
|
||||
|
||||
func TestAWantNotAppliedByItsBoundIsStuckAndAKeyDeletedEndsIt(t *testing.T) {
|
||||
o, c := newOut(), &clock{t: t0}
|
||||
w := New(o, c.now)
|
||||
w := New(testModule, o, c.now)
|
||||
w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
|
||||
pk := pulseKeyOf(t, AssignerWantsBucket, "want.ace.web")
|
||||
c.t = t0.Add(31 * time.Minute)
|
||||
|
||||
@@ -168,10 +168,11 @@ 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},
|
||||
"module": {watch.WatchmanModule, m.Module},
|
||||
// A-296: its own buckets are named from its credential's module, each a state its manifest keeps.
|
||||
"presence": {watch.BucketsOf("a-watchman").Presence, "a-watchman_" + m.State[0].Name},
|
||||
"pulse": {watch.BucketsOf("a-watchman").Pulse, "a-watchman_" + m.State[1].Name},
|
||||
"stuck": {watch.BucketsOf("a-watchman").Stuck, "a-watchman_" + m.State[2].Name},
|
||||
"wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")},
|
||||
"provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")},
|
||||
"assigner": {watch.AssignerModule + ".assigner_read_wants", entry(m.Invokes, "mesh-delivery-assigner.assigner_read_wants")},
|
||||
|
||||
Reference in New Issue
Block a user