Pulse a build ask only from when a node takes it, and end the pulse with its step, so a waiting ask is a backlog and not a stuck row (issue 1908)
This commit is contained in:
@@ -167,23 +167,23 @@ 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, _ := delivery.AskStream(delivery.AskGo)
|
||||
askSubject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
|
||||
putAsk := func(id string) {
|
||||
steps := kvs[delivery.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 := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: time.Now().UTC(),
|
||||
Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}})
|
||||
body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: time.Now().UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ctl.Publish(askSubject, body); err != nil {
|
||||
if _, err := steps.Put(ctx, "n1."+id, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb"
|
||||
putAsk(ask1)
|
||||
until("a pulse for the ask waiting", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok })
|
||||
putStep(ask1, delivery.StepRunning)
|
||||
until("a pulse for the ask a node took", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok })
|
||||
var stuck delivery.Stuck
|
||||
until("a stuck key for the ask no node took", func() bool {
|
||||
until("a stuck key for the ask no node ended", func() bool {
|
||||
e, ok := has(delivery.StuckBucket, stuckOf(askStream, ask1))
|
||||
if !ok {
|
||||
return false
|
||||
@@ -210,11 +210,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(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return !ok })
|
||||
// An ask a node takes at once is the same: no stuck key, however long the bound passes.
|
||||
putAsk(ask2)
|
||||
putStep(ask2, delivery.StepRunning)
|
||||
until("a pulse for the second ask", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return ok })
|
||||
if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+ask2, []byte(`{}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
putStep(ask2, delivery.StepPassed)
|
||||
until("the pulse of the second ask deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return !ok })
|
||||
time.Sleep(6 * time.Second) // twice the bound and the marker time
|
||||
for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(askStream, ask2)} {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
// Command mesh-delivery-watchman is delivery's watchman (novox/hq ADR 0429 §1, ADR 0431, ADR 0433, issue 1135): it
|
||||
// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask waiting for a node,
|
||||
// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask a node is running,
|
||||
// a node or part gone offline) into the pulse bucket with a time to live, and writes
|
||||
// a stuck key when the server expires one. It holds no timer: every reaction is an event, a watch or a request at
|
||||
// start or at a reconnect, and the server's expiry is the only clock.
|
||||
@@ -179,15 +179,10 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key
|
||||
}
|
||||
}
|
||||
|
||||
// stepTaken is a change of a build module's steps: a step key written for an ask is that ask taken by a node. The node's
|
||||
// `paused.<node>` key and a key that is no step key are left.
|
||||
// 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) {
|
||||
if opOf(e) != watch.Put {
|
||||
return
|
||||
}
|
||||
if _, id, err := delivery.ParseStepKey(e.Key()); err == nil {
|
||||
w.Taken(language, id)
|
||||
}
|
||||
w.Step(language, e.Key(), e.Value(), opOf(e))
|
||||
}
|
||||
|
||||
// run is the program: it returns when ctx ends or the bus cannot be used.
|
||||
@@ -318,23 +313,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
}
|
||||
}
|
||||
// The asks put on the build seats' streams: a plain subscription hears each as it is published, as the stream captures it.
|
||||
for _, lang := range delivery.AskLanguages {
|
||||
for _, kind := range delivery.AskKinds {
|
||||
subject, _ := delivery.AskSubject(lang, kind)
|
||||
lang := lang
|
||||
if _, err := conn.Subscribe(subject, func(m *nats.Msg) {
|
||||
a, err := delivery.DecodeStepAsk(m.Data)
|
||||
if err != nil {
|
||||
logger.Printf("an ask on %s is not readable: %v", m.Subject, err)
|
||||
return
|
||||
}
|
||||
w.Asked(lang, a.ID, a.At)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
for subject, open := range map[string]bool{connectSubject: true, disconnectSubject: false} {
|
||||
if _, err := conn.Subscribe(subject, onEvent(open)); err != nil {
|
||||
return err
|
||||
|
||||
@@ -2,41 +2,53 @@ package watch
|
||||
|
||||
import (
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A waiting ask is a move (novox/hq ADR 0525 §4, issue 1887 A8b): a message on a build seat's stream (mesh-sdk go/delivery
|
||||
// AskStream) that a node of the seat must pull. Its pulse is the pulse of the bucket named as the stream, under the ask's id;
|
||||
// the ask put on the seat is the move beginning at its time, and the ask taken (its step key `<node>.<ask id>` written in
|
||||
// the build module's steps, delivery.StepsBucket) is the move done. An ask no node takes within the bound is stuck, as any
|
||||
// move. The watchman reads no bucket of the controller for it.
|
||||
// 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 (delivery.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
|
||||
// (delivery.AskStream), under the ask's id, and it begins when the node started the step.
|
||||
//
|
||||
// **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).
|
||||
|
||||
// AskedKey names a waiting ask for the pulse and stuck keys: the seat's stream and the ask's id.
|
||||
func askedBucket(language string) (string, error) { return delivery.AskStream(language) }
|
||||
|
||||
// Asked is an ask of a build seat of language put on its stream at time at (the ask's own time).
|
||||
func (w *Watchman) Asked(language, id string, at time.Time) {
|
||||
bucket, err := askedBucket(language)
|
||||
// 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 := delivery.AskStream(language)
|
||||
if err != nil {
|
||||
w.say("no pulse for the ask %s: %v", id, err)
|
||||
w.say("no pulse for the step %s: %v", key, err)
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if at.IsZero() {
|
||||
at = w.now()
|
||||
}
|
||||
w.mirror(bucket, id, describe("the ask %s to be taken by a node", id), "", at, 0)
|
||||
}
|
||||
|
||||
// Taken is the ask taken by a node: its step key written. The pulse is deleted, so it never expires, and a stuck key it left is
|
||||
// written as moved.
|
||||
func (w *Watchman) Taken(language, id string) {
|
||||
bucket, err := askedBucket(language)
|
||||
node, id, err := delivery.ParseStepKey(key)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.moved(bucket, id)
|
||||
switch op {
|
||||
case Expired:
|
||||
return
|
||||
case Delete:
|
||||
w.moved(bucket, id)
|
||||
return
|
||||
}
|
||||
step, err := delivery.DecodeStep(value)
|
||||
if err != nil {
|
||||
w.say("the step %s cannot be read, and is left: %v", key, err)
|
||||
return
|
||||
}
|
||||
if step.State != delivery.StepRunning {
|
||||
w.moved(bucket, id)
|
||||
return
|
||||
}
|
||||
since := step.Started
|
||||
if since.IsZero() {
|
||||
since = w.now()
|
||||
}
|
||||
w.mirror(bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0)
|
||||
}
|
||||
|
||||
@@ -78,6 +78,23 @@ const (
|
||||
presenceNS = delivery.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 := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: started})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func endedStep(id, state string, started time.Time) []byte {
|
||||
b, err := delivery.EncodeStep(delivery.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
|
||||
}
|
||||
|
||||
func pulseKeyOf(t *testing.T, bucket, key string) string {
|
||||
t.Helper()
|
||||
k, err := delivery.PulseKey(bucket, key)
|
||||
@@ -207,7 +224,7 @@ func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
|
||||
func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2
|
||||
w, o, c := setup()
|
||||
c.t = t0.Add(30 * time.Second)
|
||||
w.Asked("go", "askid1", t0.Add(10*time.Second))
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0.Add(10*time.Second)), Put)
|
||||
w.SeedPresence("ace", false, t0)
|
||||
w.Reconcile()
|
||||
if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }); err != nil {
|
||||
@@ -226,7 +243,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
|
||||
live := pulseKeyOf(t, checksKey, "askid1")
|
||||
w.SeedPulse(stale)
|
||||
w.SeedPulse(live)
|
||||
w.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
w.Reconcile()
|
||||
if !slices.Equal(o.pulseDeletes, []string{stale}) {
|
||||
t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale)
|
||||
@@ -239,7 +256,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.Asked("go", "askid1", t0) // the key found at start, last renewed 30s ago
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // 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])
|
||||
@@ -257,7 +274,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.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
if len(o.pulses) != 0 {
|
||||
t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses)
|
||||
}
|
||||
@@ -265,7 +282,7 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
|
||||
if !ok {
|
||||
t.Fatalf("no stuck key; have %v", o.stuck)
|
||||
}
|
||||
if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" {
|
||||
if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to end on n1" {
|
||||
t.Fatalf("stuck = %+v", s)
|
||||
}
|
||||
}
|
||||
@@ -273,13 +290,13 @@ 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.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
w.SeedPresence("ace", false, t0)
|
||||
w.SeedPresence("part.mesh-delivery-view.ace", false, t0)
|
||||
w.Reconcile()
|
||||
_ = w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil })
|
||||
want := map[string]string{
|
||||
stuckKeyOf(t, checksKey, "askid1"): "the ask askid1 to be taken by a node",
|
||||
stuckKeyOf(t, checksKey, "askid1"): "the ask askid1 to end on n1",
|
||||
stuckKeyOf(t, presenceNS, "ace"): "ace to come back online",
|
||||
stuckKeyOf(t, presenceNS, "part.mesh-delivery-view.ace"): "mesh-delivery-view on ace to come back online",
|
||||
}
|
||||
@@ -293,13 +310,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.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
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.Asked("go", "askid1", c.t) // the builder re-puts the key: same check, started long ago
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // 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])
|
||||
}
|
||||
@@ -308,7 +325,7 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
|
||||
}
|
||||
// The same revision heard twice changes nothing.
|
||||
o.ttls[k] = 0
|
||||
w.Asked("go", "askid1", c.t)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put)
|
||||
if o.ttls[k] != 0 {
|
||||
t.Fatal("one revision rewrote the pulse twice")
|
||||
}
|
||||
@@ -317,10 +334,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.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
w.Connect("node.ace")
|
||||
w.Disconnect("node.ace")
|
||||
w.Taken("go", "askid1") // the check ended
|
||||
w.Step("go", "n1.askid1", nil, Delete) // the check ended
|
||||
w.Connect("node.ace")
|
||||
if len(o.pulses) != 0 {
|
||||
t.Fatalf("pulses left: %v", o.pulses)
|
||||
@@ -335,9 +352,9 @@ func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
|
||||
|
||||
func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1135/AC4
|
||||
w, o, c := setup()
|
||||
w.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
c.t = t0.Add(50 * time.Second)
|
||||
w.Asked("go", "askid1", c.t) // the last heartbeat
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // 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")]
|
||||
@@ -348,7 +365,7 @@ func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1
|
||||
|
||||
func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
|
||||
w, o, c := setup()
|
||||
w.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
pk := pulseKeyOf(t, checksKey, "askid1")
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
w.Pulse(pk, Expired, c.t)
|
||||
@@ -357,24 +374,24 @@ func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
|
||||
t.Fatal("stuck written as moved at expiry")
|
||||
}
|
||||
c.t = t0.Add(300 * time.Second)
|
||||
w.Taken("go", "askid1")
|
||||
w.Step("go", "n1.askid1", nil, Delete)
|
||||
s := o.stuck[sk]
|
||||
if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" {
|
||||
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.Taken("go", "askid1")
|
||||
w.Step("go", "n1.askid1", nil, Delete)
|
||||
if o.stuckWrites != n {
|
||||
t.Fatal("a stuck key already moved was written again")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnAskTakenAfterStuckMovesIt(t *testing.T) { // 1135/AC4
|
||||
func TestAnAskEndedAfterStuckMovesIt(t *testing.T) { // 1135/AC4
|
||||
w, o, c := setup()
|
||||
w.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
|
||||
w.Taken("go", "askid1") // a node took it at last
|
||||
w.Step("go", "n1.askid1", nil, Delete) // 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")
|
||||
}
|
||||
@@ -398,23 +415,23 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4
|
||||
|
||||
func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
|
||||
w, o, c := setup()
|
||||
w.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
|
||||
s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
|
||||
if !ok {
|
||||
t.Fatalf("no stuck key; have %v", o.stuck)
|
||||
}
|
||||
if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to be taken by a node" || !s.Since.Equal(t0) || s.Moved {
|
||||
if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to end on n1" || !s.Since.Equal(t0) || s.Moved {
|
||||
t.Fatalf("stuck = %+v", s)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
|
||||
w, o, c := setup()
|
||||
w.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
pk := pulseKeyOf(t, checksKey, "askid1")
|
||||
w.Taken("go", "askid1")
|
||||
w.Step("go", "n1.askid1", nil, Delete)
|
||||
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)
|
||||
@@ -427,13 +444,56 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8
|
||||
w, o, c := setup()
|
||||
w.SetBound(2 * time.Second)
|
||||
c.t = t0.Add(500 * time.Millisecond)
|
||||
w.Asked("go", "askid1", t0)
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
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.Asked("go", "askid2", t0)
|
||||
w.Step("go", "n1.askid2", runningStep("askid2", t0), Put)
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
if len(o.pulses) != 1 {
|
||||
t.Fatalf("pulses = %v, want the taken ask's", o.pulses)
|
||||
}
|
||||
for _, state := range []string{delivery.StepPassed, delivery.StepFailed, delivery.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
|
||||
}
|
||||
_ = 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
|
||||
p, ok := o.pulses[pulseKeyOf(t, checksKey, "askid1")]
|
||||
if !ok || p.Key != "askid1" {
|
||||
t.Fatalf("pulses = %v", o.pulses)
|
||||
}
|
||||
if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got <= 0 || got >= 120*time.Second {
|
||||
t.Errorf("ttl = %v, want the bound less its 30s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAPausedKeyOrAnotherNodesNonStepIsLeft(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)
|
||||
if len(o.pulses) != 0 || len(o.stuck) != 0 {
|
||||
t.Errorf("pulses %v stuck %v", o.pulses, o.stuck)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user