Merge pull request 'messenger: a walk's held messages are told only once its message was sent, handed on only as a first hold would take them, and its start and end compared on the controller's clock (hq issue 538)' (#238) from fix/538-walk-hold-told-handover-clock into main
This commit was merged in pull request #238.
This commit is contained in:
@@ -179,15 +179,19 @@ The controller's `plan-moved` says each walk's start and end: `walk`, `phase` (`
|
||||
- **Held:** while a walk is open and within its bound, a message whose subject is a machine it touches, or a
|
||||
module it touches (on a machine it touches, when the subject names one), is held: kept on the walk with when
|
||||
and what, and the history says `held`. A message several open walks touch is told when the last ends or passes
|
||||
its bound.
|
||||
its bound. At a walk's end, a later open walk that touches a message takes it over only as a first hold would:
|
||||
still open, not silenced, not said since, and not one that is never held.
|
||||
- **Never held:** an ask waiting for the operator (and a condition an open ask is about), a secret given or
|
||||
replaced, a subject no open walk touches, and a walk's own failure (a plan, walk, delivery or build).
|
||||
- **The end is one message:** the outcome, the modules and machines reached, how long; each held message still
|
||||
open, one line each; the ones that cleared counted. Delivered with nothing held and nothing open: one line —
|
||||
and nothing on a kind listed in the setting `only-what-needs-me`.
|
||||
and nothing on a kind listed in the setting `only-what-needs-me`. What it carries is marked told on a kind
|
||||
only once the message was sent there. Until then each is kept owed in the state (`owed`): a message lost with a
|
||||
restart before it was sent leaves them untold and owed, and the restarted router says each on its own.
|
||||
- **The bound:** at its start plus `bound_seconds` (900 when not given), what is held and still open is said,
|
||||
with "The walk … is still running, since HH:MM", and the cleared counted; the walk holds nothing more. A walk
|
||||
whose end never comes is forgotten a day after it started.
|
||||
whose end never comes is forgotten a day after it started. A start heard after its end opens nothing: the
|
||||
controller's moments are compared on its own clock, never with one capped at the router's now.
|
||||
- **Kept in the state** (`walks`): a router that starts again mid-walk holds and tells as before.
|
||||
|
||||
## What the operator gives
|
||||
|
||||
@@ -200,14 +200,17 @@ type Record struct {
|
||||
// ClearHeld is what was waiting to be said of it when it cleared (a reminder, an escalation): said after
|
||||
// all if it comes back before its clearing was said. Blips counts those comebacks, never said, so a
|
||||
// condition that flaps is known as one and its message's "again, Nth time" counts only what was said.
|
||||
ClearHeld string `json:"clear_held,omitempty"`
|
||||
Blips int `json:"blips,omitempty"`
|
||||
Reminded bool `json:"reminded,omitempty"`
|
||||
Pending string `json:"pending,omitempty"` // what could not be said yet: raised, reminder, …
|
||||
Folded bool `json:"folded,omitempty"` // held by the cap
|
||||
Cleared time.Time `json:"cleared,omitempty"`
|
||||
Count int `json:"count"`
|
||||
Refused string `json:"refused,omitempty"`
|
||||
ClearHeld string `json:"clear_held,omitempty"`
|
||||
Blips int `json:"blips,omitempty"`
|
||||
Reminded bool `json:"reminded,omitempty"`
|
||||
Pending string `json:"pending,omitempty"` // what could not be said yet: raised, reminder, …
|
||||
// Owed is what a walk held of it and has not told yet (ADR 0308 §4, §5): kept in the state until a message
|
||||
// carrying it was sent, so a router that restarts before says it on its own (issue 538), never history.
|
||||
Owed string `json:"owed,omitempty"`
|
||||
Folded bool `json:"folded,omitempty"` // held by the cap
|
||||
Cleared time.Time `json:"cleared,omitempty"`
|
||||
Count int `json:"count"`
|
||||
Refused string `json:"refused,omitempty"`
|
||||
// History: known from the state or from an old event, and never said — not news.
|
||||
History bool `json:"history,omitempty"`
|
||||
// Seen is when this holder last took word of it; the open set read after it is authoritative.
|
||||
@@ -407,6 +410,11 @@ func (h *Holder) Load() error {
|
||||
r.History = true
|
||||
}
|
||||
}
|
||||
if r.Owed != "" && r.Cleared.IsZero() {
|
||||
// Held from the operator on purpose and not told yet: said on its own, as the next tick tries what is
|
||||
// unsent (issue 538). What the operator could have heard is history; this they could not.
|
||||
r.Pending, r.History = r.Owed, false
|
||||
}
|
||||
h.open[r.Key] = &r
|
||||
}
|
||||
if sent, err := h.Store.Recent(); err == nil {
|
||||
|
||||
@@ -18,6 +18,9 @@ type item struct {
|
||||
Severity string
|
||||
At time.Time
|
||||
Msg *Message
|
||||
// Carries are the records a walk's end or bound message tells (ADR 0308 §4): each is marked told on a
|
||||
// channel only once that message was sent there (issue 538 A), never when it is held to be sent.
|
||||
Carries []string
|
||||
}
|
||||
|
||||
func (i item) urgent() bool { return i.Severity == Urgent && i.What != "cleared" }
|
||||
@@ -122,6 +125,8 @@ func (h *Holder) queuedWhat(key string) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// queued says whether something is held to be said of a record: its own message, or a walk's message that
|
||||
// carries it.
|
||||
func (h *Holder) queued(key string) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
@@ -130,6 +135,11 @@ func (h *Holder) queued(key string) bool {
|
||||
if it.Msg == nil && it.Key == key {
|
||||
return true
|
||||
}
|
||||
for _, k := range it.Carries {
|
||||
if k == key {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
@@ -420,6 +430,7 @@ func (h *Holder) sendEntries(ch Channel, entries []entry, now time.Time) bool {
|
||||
}
|
||||
r.Pending, r.Folded, r.History = "", false, false
|
||||
if e.it.What != "cleared" {
|
||||
r.Owed = ""
|
||||
if r.Sent == nil {
|
||||
r.Sent = map[string]string{}
|
||||
}
|
||||
@@ -435,7 +446,39 @@ func (h *Holder) sendEntries(ch Channel, entries []entry, now time.Time) bool {
|
||||
r.LastSaid = now
|
||||
}
|
||||
}
|
||||
type carried struct {
|
||||
r *Record
|
||||
walk string
|
||||
}
|
||||
var told []carried
|
||||
for _, e := range entries {
|
||||
for _, key := range e.it.Carries {
|
||||
r := h.open[key]
|
||||
if r == nil {
|
||||
continue
|
||||
}
|
||||
if r.Sent == nil {
|
||||
r.Sent = map[string]string{}
|
||||
}
|
||||
if _, has := r.Sent[name]; !has {
|
||||
r.Sent[name] = ""
|
||||
}
|
||||
if r.FirstSent.IsZero() {
|
||||
r.FirstSent = now
|
||||
}
|
||||
r.LastSaid, r.Pending, r.History, r.Owed = now, "", false, ""
|
||||
if !r.Cleared.IsZero() && !r.ClearUnsaid {
|
||||
// It cleared while the message saying it open waited: its clearing is said too.
|
||||
r.ClearUnsaid = true
|
||||
}
|
||||
told = append(told, carried{r, strings.TrimPrefix(e.it.Key, "walk.")})
|
||||
}
|
||||
}
|
||||
h.mu.Unlock()
|
||||
for _, c := range told {
|
||||
h.record(Sent{At: now, Channel: name, Key: c.r.Key, What: "released", Outcome: "told in the message of walk " + c.walk})
|
||||
h.persist(c.r)
|
||||
}
|
||||
key, what := "(digest)", fmt.Sprintf("digest of %d", len(entries))
|
||||
if len(entries) == 1 {
|
||||
key, what = entries[0].it.Key, entries[0].it.What
|
||||
|
||||
@@ -114,6 +114,9 @@ type Walk struct {
|
||||
BoundSaid bool `json:"bound_said,omitempty"`
|
||||
// Ended: its end was heard, at this moment. Kept a day, so a start heard after its end opens nothing.
|
||||
Ended time.Time `json:"ended,omitempty"`
|
||||
// EndedSaid is the end's moment as the controller said it, by the controller's clock and never capped at
|
||||
// this router's now: a start heard after its end is compared with it on that one clock (issue 538 C).
|
||||
EndedSaid time.Time `json:"ended_said,omitempty"`
|
||||
}
|
||||
|
||||
func (w *Walk) bound() time.Duration {
|
||||
@@ -295,7 +298,7 @@ func (h *Holder) Walked(ev WalkEvent) {
|
||||
h.mu.Lock()
|
||||
end := h.ended[ev.Walk]
|
||||
h.mu.Unlock()
|
||||
if end != nil && (!at.After(end.Ended) || (!end.Started.IsZero() && at.Equal(end.Started))) {
|
||||
if end != nil && startOfEnded(ev, at, end) {
|
||||
// A short walk's two events can come in either order: this start is of the walk that ended.
|
||||
h.Logf("[messenger] walk %s: its start came after its end, which was heard at %s: nothing held", ev.Walk,
|
||||
end.Ended.UTC().Format(time.RFC3339))
|
||||
@@ -335,7 +338,7 @@ func (h *Holder) Walked(ev WalkEvent) {
|
||||
w.Started = ev.StartedAt
|
||||
}
|
||||
}
|
||||
tomb := &Walk{ID: ev.Walk, Started: ev.StartedAt, Ended: at}
|
||||
tomb := &Walk{ID: ev.Walk, Started: ev.StartedAt, Ended: at, EndedSaid: ev.At}
|
||||
if prev := h.ended[ev.Walk]; prev != nil && prev.Ended.After(at) {
|
||||
tomb = prev
|
||||
}
|
||||
@@ -350,18 +353,52 @@ func (h *Holder) Walked(ev WalkEvent) {
|
||||
}
|
||||
}
|
||||
|
||||
// startOfEnded says whether a start is of a walk whose end was heard already: of that walk, or of an earlier one
|
||||
// under its id. The controller's own moments are compared on its clock, never one capped at this router's now
|
||||
// against one that is not (issue 538 C: a controller's clock ahead of the router's would open a walk after its
|
||||
// end). Only when the controller said neither moment are the moments as heard here compared.
|
||||
func startOfEnded(ev WalkEvent, at time.Time, end *Walk) bool {
|
||||
said := ev.StartedAt
|
||||
if said.IsZero() {
|
||||
said = ev.At
|
||||
}
|
||||
switch {
|
||||
case !said.IsZero() && !end.Started.IsZero():
|
||||
return !said.After(end.Started)
|
||||
case !said.IsZero() && !end.EndedSaid.IsZero():
|
||||
return !said.After(end.EndedSaid)
|
||||
}
|
||||
return !at.After(end.Ended)
|
||||
}
|
||||
|
||||
// release is what a walk held that no other open walk holds or touches: each open message, oldest first, with
|
||||
// what was to be said of it, and how many cleared meanwhile. A message said meanwhile on a path that is never
|
||||
// held, or silenced meanwhile, is neither. One that another open walk touches — even one that started after it
|
||||
// was held — is held on by that walk, and told at its end or bound (ADR 0308 §2).
|
||||
//
|
||||
// A message is handed on under the rules of its first hold (issue 538 B): still open, not silenced, not said
|
||||
// since it was held, and not one ADR 0308 §3 never holds (an open ask about it, a walk's own failure).
|
||||
func (h *Holder) release(w *Walk, now time.Time) (open []*Record, whats map[string]string, cleared int) {
|
||||
// What may be handed on is judged first: neverHeld asks the asks and the walks under the lock itself.
|
||||
h.mu.Lock()
|
||||
recs := map[string]*Record{}
|
||||
for key := range w.Held {
|
||||
if r := h.open[key]; r != nil && r.Cleared.IsZero() && !r.silenced(now) && !lastSaid(r).After(w.Held[key].At) {
|
||||
recs[key] = r
|
||||
}
|
||||
}
|
||||
h.mu.Unlock()
|
||||
handable := map[string]bool{}
|
||||
for key, r := range recs {
|
||||
handable[key] = !h.neverHeld(r)
|
||||
}
|
||||
h.mu.Lock()
|
||||
whats = map[string]string{}
|
||||
taken := map[string]*Walk{}
|
||||
var handed []string
|
||||
for key, hm := range w.Held {
|
||||
r := h.open[key]
|
||||
live := r != nil && r.Cleared.IsZero()
|
||||
live := r != nil && r.Cleared.IsZero() && handable[key]
|
||||
var s subject
|
||||
if live {
|
||||
s = subjectOf(r)
|
||||
@@ -543,6 +580,17 @@ func (h *Holder) tellWalk(w *Walk, m Message, open []*Record, whats map[string]s
|
||||
m = Message{Title: "A delivery ended", Body: "What it reached could not be said here: the delivery's walks say it.", Urgent: m.Urgent}
|
||||
alone, open = open, nil
|
||||
}
|
||||
// Owed until a message carrying it was sent, in the state: a router that restarts first says it on its own
|
||||
// (issue 538), whichever path below it takes.
|
||||
h.mu.Lock()
|
||||
for _, r := range append(append([]*Record(nil), open...), alone...) {
|
||||
say := whats[r.Key]
|
||||
if say == "" {
|
||||
say = "raised"
|
||||
}
|
||||
r.Owed = say
|
||||
}
|
||||
h.mu.Unlock()
|
||||
var to []Channel
|
||||
if m.Urgent {
|
||||
for _, ch := range h.channels() {
|
||||
@@ -587,32 +635,23 @@ func (h *Holder) tellWalk(w *Walk, m Message, open []*Record, whats map[string]s
|
||||
if m.Urgent {
|
||||
sev = Urgent
|
||||
}
|
||||
// What it carries is marked told on a channel once the message was sent there (outbox.go), not now: a
|
||||
// message lost with a restart before it was sent leaves them untold and owed, and the restarted router says
|
||||
// each on its own (issue 538 A).
|
||||
var carries []string
|
||||
for _, r := range open {
|
||||
carries = append(carries, r.Key)
|
||||
}
|
||||
for _, ch := range kept {
|
||||
msg := m
|
||||
h.enqueue(ch.Name(), item{Key: "walk." + w.ID, What: WalkEndWhat, Severity: sev, At: now, Msg: &msg})
|
||||
h.enqueue(ch.Name(), item{Key: "walk." + w.ID, What: WalkEndWhat, Severity: sev, At: now, Msg: &msg, Carries: carries})
|
||||
}
|
||||
h.mu.Lock()
|
||||
for _, r := range open {
|
||||
if len(kept) == 0 {
|
||||
continue
|
||||
if len(kept) > 0 {
|
||||
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: "released",
|
||||
Outcome: "to be told in the message of walk " + w.ID + " " + what})
|
||||
}
|
||||
if r.Sent == nil {
|
||||
r.Sent = map[string]string{}
|
||||
}
|
||||
for _, ch := range kept {
|
||||
if _, has := r.Sent[ch.Name()]; !has {
|
||||
r.Sent[ch.Name()] = ""
|
||||
}
|
||||
}
|
||||
if r.FirstSent.IsZero() {
|
||||
r.FirstSent = now
|
||||
}
|
||||
r.LastSaid, r.Pending, r.History = now, "", false
|
||||
}
|
||||
h.mu.Unlock()
|
||||
for _, r := range open {
|
||||
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: "released", Outcome: "told in the message of walk " + w.ID + " " + what})
|
||||
h.persist(r)
|
||||
h.persist(r) // owed, in the state, until the message was sent
|
||||
}
|
||||
h.flush(now, false)
|
||||
for _, r := range alone {
|
||||
|
||||
@@ -620,3 +620,151 @@ func TestWhatAWalksPlainEndCannotCarryIsSaidOnItsOwn(t *testing.T) {
|
||||
t.Fatalf("what the walk held was marked told and never said: %v", outcomes(h, k.Key))
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 538 A: what a walk held is marked told only once the message carrying it was sent. When that message
|
||||
// could not be sent and the router restarts before its retry, the message is gone with it: what it held is not
|
||||
// marked told, it stays owed in the state, and the restarted router says it — an urgent one and a warning alike,
|
||||
// once each, never left as history.
|
||||
func TestAHeldMessageIsToldOnlyOnceTheWalksMessageWasSent(t *testing.T) {
|
||||
for _, then := range []string{"sent", "restart urgent", "restart warning"} {
|
||||
t.Run(then, func(t *testing.T) {
|
||||
h, tg, dt, c, st := newHolder(t)
|
||||
sev := Urgent
|
||||
if then == "restart warning" {
|
||||
sev = Warning
|
||||
}
|
||||
walk(t, h, "plan-h", WalkStarted, "", c.now(), 1200, novox("mesh-tools"))
|
||||
k := cond("machine.novox.tools-silent", sev, "the tools on novox are silent")
|
||||
k.Raised = c.now()
|
||||
h.Condition(EventRaised, k)
|
||||
runFor(h, c, time.Minute)
|
||||
down := errors.New("cannot connect")
|
||||
tg.fail, dt.fail = down, down
|
||||
walk(t, h, "plan-h", WalkEnded, WalkFailed, c.now(), 0)
|
||||
runFor(h, c, time.Minute)
|
||||
if saidOn(tg, dt) != 0 {
|
||||
t.Fatalf("said on channels that cannot send: %+v %+v", tg.sends, dt.sends)
|
||||
}
|
||||
if r := h.open[k.Key]; r == nil || r.told() {
|
||||
t.Fatalf("marked told while the message carrying it was not sent: %+v", r)
|
||||
}
|
||||
tg.fail, dt.fail = nil, nil
|
||||
if then == "sent" {
|
||||
runFor(h, c, 2*time.Minute)
|
||||
if !heard("Still open after it:\n- Urgent: "+headlineOf(k.Key), tg, dt) || saidOn(tg, dt) != 2 {
|
||||
t.Fatalf("not told once in the walk's message: telegram %+v, desk %+v", tg.sends, dt.sends)
|
||||
}
|
||||
if r := h.open[k.Key]; r == nil || !r.told() || r.Owed != "" || r.Sent["telegram"] != "" || r.Sent["desktop"] != "" {
|
||||
t.Fatalf("not marked told on both channels once sent: %+v", r)
|
||||
}
|
||||
runFor(h, c, 10*time.Minute)
|
||||
if saidOn(tg, dt) != 2 {
|
||||
t.Fatalf("said again after the walk's message: telegram %+v, desk %+v", tg.sends, dt.sends)
|
||||
}
|
||||
return
|
||||
}
|
||||
again := &Holder{Away: tg, Desk: dt, Store: st, Now: c.now, Logf: t.Logf}
|
||||
if err := again.Load(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again.Start()
|
||||
again.Sync([]Condition{k}, c.now())
|
||||
runFor(again, c, 5*time.Minute)
|
||||
if !heard(headlineOf(k.Key), tg, dt) {
|
||||
t.Fatalf("what the lost message carried was not said after the restart: %v", outcomes(again, k.Key))
|
||||
}
|
||||
said := saidOn(tg, dt)
|
||||
if r := again.open[k.Key]; r == nil || !r.told() || r.Owed != "" || r.Pending != "" {
|
||||
t.Fatalf("said, and still owed or not marked told: %+v", r)
|
||||
}
|
||||
runFor(again, c, 20*time.Minute) // within its reminder's bound: nothing more is due
|
||||
again.Sync([]Condition{k}, c.now())
|
||||
runFor(again, c, time.Minute)
|
||||
if saidOn(tg, dt) != said {
|
||||
t.Fatalf("said again: telegram %+v, desk %+v", tg.sends, dt.sends)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 538 B: at a walk's end, a later walk that touches what it held takes over only what a first hold would
|
||||
// take: not a message silenced meanwhile, nor one an ask waiting for the operator is about (ADR 0308 §3).
|
||||
func TestAWalksEndHandsOnOnlyWhatALaterWalkWouldHold(t *testing.T) {
|
||||
h, tg, dt, c, _, _ := newRouter(t)
|
||||
h.Start()
|
||||
stand(h, c)
|
||||
walk(t, h, "plan-a", WalkStarted, "", c.now(), 1200, novox("mesh-tools"))
|
||||
quiet := cond("machine.novox.network", Urgent, "novox lost its default route")
|
||||
quiet.Raised = c.now()
|
||||
h.Condition(EventRaised, quiet)
|
||||
asked := cond("machine.novox.units", Urgent, "2 failed services on novox")
|
||||
asked.Raised = c.now()
|
||||
h.Condition(EventRaised, asked)
|
||||
runFor(h, c, time.Minute)
|
||||
stand(h, c)
|
||||
if heard(headlineOf(quiet.Key), tg, dt) || heard(headlineOf(asked.Key), tg, dt) {
|
||||
t.Fatalf("the test's messages were not held")
|
||||
}
|
||||
quiet.SilencedTill, quiet.SilencedWhy, quiet.At = c.now().Add(7*24*time.Hour), "known", c.now()
|
||||
h.Condition(EventChanged, quiet)
|
||||
ask := ackAsk(c, "units")
|
||||
ask.About = asked.Key
|
||||
submit(t, h, ConditionOwner, ask)
|
||||
runFor(h, c, time.Minute)
|
||||
stand(h, c)
|
||||
walk(t, h, "plan-b", WalkStarted, "", c.now(), 1200, novox("node-tools"))
|
||||
runFor(h, c, time.Minute)
|
||||
stand(h, c)
|
||||
helds := func(key string) int {
|
||||
n := 0
|
||||
for _, o := range outcomes(h, key) {
|
||||
if o == "held" {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
before := map[string]int{quiet.Key: helds(quiet.Key), asked.Key: helds(asked.Key)}
|
||||
walk(t, h, "plan-a", WalkEnded, WalkDelivered, c.now(), 0)
|
||||
runFor(h, c, time.Minute)
|
||||
for _, key := range []string{quiet.Key, asked.Key} {
|
||||
if got := h.heldBy(key, c.now()); len(got) != 0 {
|
||||
t.Errorf("%s was handed on to %v, which a first hold would never hold", key, got)
|
||||
}
|
||||
if helds(key) != before[key] {
|
||||
t.Errorf("the history says %s held again: %v", key, outcomes(h, key))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 538 C: the controller's clock ahead of the router's. A short walk's end, heard before its start, opens
|
||||
// nothing when its start comes, whether the end said the walk's start or only its own moment; the next walk
|
||||
// under its id, started after that end, still opens.
|
||||
func TestAStartHeardAfterItsEndOpensNothingWhenTheControllersClockIsAhead(t *testing.T) {
|
||||
for _, said := range []string{"both", "end only"} {
|
||||
t.Run(said, func(t *testing.T) {
|
||||
h, _, _, c, _ := newHolder(t)
|
||||
ended := c.now().Add(10 * time.Minute) // by the controller's clock
|
||||
started := ended.Add(-20 * time.Second)
|
||||
body := `{"walk":"plan-k","phase":"ended","outcome":"delivered","touches":[{"machine":"novox","modules":["mesh-tools"]}],` +
|
||||
`"ended_at":"` + ended.Format(time.RFC3339Nano) + `"`
|
||||
if said == "both" {
|
||||
body += `,"started_at":"` + started.Format(time.RFC3339Nano) + `"`
|
||||
}
|
||||
end, _, err := DecodeWalk([]byte(body + `}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
h.Walked(end)
|
||||
c.pass(time.Second)
|
||||
walk(t, h, "plan-k", WalkStarted, "", started, 600, novox("mesh-tools"))
|
||||
if len(h.walks) != 0 {
|
||||
t.Fatalf("a start heard after its end opened a walk: %+v", h.walks)
|
||||
}
|
||||
walk(t, h, "plan-k", WalkStarted, "", ended.Add(time.Minute), 600, novox("mesh-tools"))
|
||||
if len(h.walks) != 1 {
|
||||
t.Fatalf("the next walk under its id, started after the end, did not open: %+v", h.walks)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user