From df94be0e60a90cc80510df9c0dfb00ef060702c4 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 13:26:41 +0200 Subject: [PATCH] messenger: mark a walk's held messages told only once its message was sent, hand on only what a first hold would take, compare a walk's start and end on the controller's clock (issue 538) A message lost with a restart before it was sent left what it carried marked told, so it reached the operator only as its reminder hours later. A later walk took over silenced, already said or asked messages that ADR 0308 section 3 never holds. A start capped at the router's now was compared with the controller's moments. --- modules/messenger/README.md | 10 +- modules/messenger/cmd/messenger/outbox.go | 38 ++++++ modules/messenger/cmd/messenger/walks.go | 73 ++++++---- modules/messenger/cmd/messenger/walks_test.go | 129 ++++++++++++++++++ 4 files changed, 224 insertions(+), 26 deletions(-) diff --git a/modules/messenger/README.md b/modules/messenger/README.md index 8a6a74eb..f628080c 100644 --- a/modules/messenger/README.md +++ b/modules/messenger/README.md @@ -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: a message lost with a restart before it was sent leaves them untold, and + the reading of the open conditions says them as any other. - **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 diff --git a/modules/messenger/cmd/messenger/outbox.go b/modules/messenger/cmd/messenger/outbox.go index e2265115..f43fa427 100644 --- a/modules/messenger/cmd/messenger/outbox.go +++ b/modules/messenger/cmd/messenger/outbox.go @@ -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 @@ -435,7 +445,35 @@ func (h *Holder) sendEntries(ch Channel, entries []entry, now time.Time) bool { r.LastSaid = now } } + var told []*Record + 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 = 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, r) + } + } h.mu.Unlock() + for _, r := range told { + h.record(Sent{At: now, Channel: name, Key: r.Key, What: "released", Outcome: "told in the message of walk " + strings.TrimPrefix(entries[0].it.Key, "walk.")}) + h.persist(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 diff --git a/modules/messenger/cmd/messenger/walks.go b/modules/messenger/cmd/messenger/walks.go index 5ed7dab2..3d32b957 100644 --- a/modules/messenger/cmd/messenger/walks.go +++ b/modules/messenger/cmd/messenger/walks.go @@ -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) @@ -587,32 +624,22 @@ 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, to be said as any other (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 + break } - 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.record(Sent{At: now, Channel: "-", Key: r.Key, What: "released", + Outcome: "to be told in the message of walk " + w.ID + " " + what}) } h.flush(now, false) for _, r := range alone { diff --git a/modules/messenger/cmd/messenger/walks_test.go b/modules/messenger/cmd/messenger/walks_test.go index 8f699ea1..7e613d56 100644 --- a/modules/messenger/cmd/messenger/walks_test.go +++ b/modules/messenger/cmd/messenger/walks_test.go @@ -620,3 +620,132 @@ 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, and the reading of the open conditions says it at once, not its reminder hours later. +func TestAHeldMessageIsToldOnlyOnceTheWalksMessageWasSent(t *testing.T) { + for _, then := range []string{"sent", "restart"} { + t.Run(then, func(t *testing.T) { + h, tg, dt, c, st := newHolder(t) + walk(t, h, "plan-h", WalkStarted, "", c.now(), 1200, novox("mesh-tools")) + k := cond("machine.novox.tools-silent", Urgent, "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.Sent["telegram"] != "" || r.Sent["desktop"] != "" { + t.Fatalf("not marked told on both channels once sent: %+v", r) + } + 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, 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)) + } + }) + } +} + +// 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) + } + }) + } +}