messenger: keep what a walk held until it is said, and open no walk whose end came first (review of #230)
mesh/merge-gate pass: builds messenger → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered

- No channel ready at a walk's end or bound: what it held is left unsent and tried
  again every minute, as any other message, instead of lost with the walk.
- At a walk's end or bound, a message another open walk touches is held on by
  that walk, even one started after it was held (hq ADR 0308 §2).
- A walk read back past its forget is released at its bound first, then forgotten.
- A walk's end is kept a day: a start heard after it (same started_at, or not
  later than the end) opens nothing, also after the router restarts; a later
  start under the same id still opens.
- A walk's plain end, when its words cannot be said, leaves what it held to be
  said on its own instead of marking it told.
- The restart pass no longer skips a held message: one with an open ask is never
  held (ADR 0308 §3), so its answers are made to work again.
This commit is contained in:
jochen
2026-10-12 12:07:38 +02:00
parent 991ffcfda0
commit a92c08fd5b
4 changed files with 316 additions and 34 deletions
@@ -351,6 +351,8 @@ type Holder struct {
lost map[string]time.Time
// walks are the walks open now, by id, each with what it holds (walks.go, ADR 0308).
walks map[string]*Walk
// ended are the walks whose end was heard, by id: what a start heard after its end is matched against.
ended map[string]*Walk
// track is the issue tracker's last word on which issues track which conditions (tracking.go).
track trackWord
}
@@ -371,6 +373,7 @@ func (h *Holder) init() {
h.passes = map[string]Pass{}
h.lost = map[string]time.Time{}
h.walks = map[string]*Walk{}
h.ended = map[string]*Walk{}
}
if h.Now == nil {
h.Now = time.Now
+9 -8
View File
@@ -31,7 +31,9 @@ package main
// (`sent`, key `passes`), read back on start.
// - **What it said:** a silenced message was edited to its silenced words, and a message withheld for its
// content was said again as withheld. Now the pass says again only what is open, asked there and never
// answered, whose words may leave; a silenced, a withheld, an answered or a held message is left as it is.
// answered, whose words may leave; a silenced, a withheld or an answered message is left as it is. Being
// held by a walk is no reason: a message with an open ask is never held (ADR 0308 §3), even one a walk took
// before its ask came, so its answers are made to work again like any other's.
import (
"fmt"
@@ -139,7 +141,7 @@ func (h *Holder) channelOf(kind string) Channel {
// reoffer says again on one channel, whose answers were lost at `lost`, what is still open, asked there and
// never answered, and leaves everything else as it is: what ended, what is silenced, what was withheld for its
// content, and what an open walk holds (ADR 0308 §6). It answers whether every message to be said again was.
// content, and what was answered (ADR 0308 §6). It answers whether every message to be said again was.
func (h *Holder) reoffer(ch Channel, lost time.Time) bool {
kind := ch.Name()
rp, canReplace := ch.(replacer)
@@ -159,7 +161,6 @@ func (h *Holder) reoffer(ch Channel, lost time.Time) bool {
before := lastSaid(r).Before(lost)
reminded := r.Reminded
withheld, answered := r.Refused != "", r.Answered != ""
held := len(h.heldBy(r.Key, now)) > 0
h.mu.Unlock()
if !shown || id == "" {
continue // never a message of its own there: nothing to say again under
@@ -167,8 +168,8 @@ func (h *Holder) reoffer(ch Channel, lost time.Time) bool {
switch {
case !open:
ended++ // its end is said where its message is; nothing comes back
case r.silenced(now), withheld, answered, held:
left++ // a silence, a withheld message, an answer or a hold is never said again for a restart
case r.silenced(now), withheld, answered:
left++ // a silence, a withheld message or an answer is never said again for a restart
case h.askKinds(r)[kind] && before && canReplace:
what := "raised"
if reminded {
@@ -191,10 +192,10 @@ func (h *Holder) reoffer(ch Channel, lost time.Time) bool {
}
}
h.record(Sent{At: now, Channel: kind, Key: "(answers lost)", What: "at " + lost.UTC().Format(time.RFC3339),
Outcome: fmt.Sprintf("%d open and asked said again, %d silenced, withheld, answered or held left as they are, "+
Outcome: fmt.Sprintf("%d open and asked said again, %d silenced, withheld or answered left as they are, "+
"%d ended left ended, %d failed", again, left, ended, failed)})
h.Logf("[messenger] %s lost its answers at %s: %d open and asked said again in place, %d silenced, withheld, answered "+
"or held left as they are, %d ended left ended, %d failed; nothing shown from the channel's memory", kind,
h.Logf("[messenger] %s lost its answers at %s: %d open and asked said again in place, %d silenced, withheld or "+
"answered left as they are, %d ended left ended, %d failed; nothing shown from the channel's memory", kind,
lost.UTC().Format(time.RFC3339), again, left, ended, failed)
return failed == 0
}
+127 -26
View File
@@ -112,6 +112,8 @@ type Walk struct {
Held map[string]HeldMessage `json:"held,omitempty"`
// BoundSaid: its bound passed and what it held was said; it holds nothing more.
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"`
}
func (w *Walk) bound() time.Duration {
@@ -290,6 +292,15 @@ func (h *Holder) Walked(ev WalkEvent) {
switch ev.Phase {
case WalkStarted:
w := &Walk{ID: ev.Walk, Started: at, BoundSeconds: ev.BoundSeconds, Class: ev.Class}
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))) {
// 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))
return
}
if now.Sub(at) >= w.bound() {
h.Logf("[messenger] walk %s started %s ago, past its bound of %s: history, nothing held", ev.Walk,
roughly(now.Sub(at)), w.bound())
@@ -310,6 +321,13 @@ func (h *Holder) Walked(ev WalkEvent) {
case WalkEnded:
h.mu.Lock()
w := h.walks[ev.Walk]
if w != nil && !ev.StartedAt.IsZero() && w.Started.After(ev.StartedAt) {
// The end of an earlier walk under the same id, heard after the next one started: that one stays open.
h.mu.Unlock()
h.Logf("[messenger] walk %s: an end of its earlier walk (started %s) came late: the open one stays", ev.Walk,
ev.StartedAt.UTC().Format(time.RFC3339))
return
}
if w != nil {
delete(h.walks, ev.Walk)
w.Touches = mergeTouches(w.Touches, ev.Touches)
@@ -317,36 +335,59 @@ func (h *Holder) Walked(ev WalkEvent) {
w.Started = ev.StartedAt
}
}
tomb := &Walk{ID: ev.Walk, Started: ev.StartedAt, Ended: at}
if prev := h.ended[ev.Walk]; prev != nil && prev.Ended.After(at) {
tomb = prev
}
h.ended[ev.Walk] = tomb
h.mu.Unlock()
h.keepWalk(tomb) // in place of the open walk: a start heard after this end opens nothing
if w == nil {
h.Logf("[messenger] walk %s %s, and its start was not heard here: nothing held, nothing said", ev.Walk, ev.Outcome)
return
}
h.forgetWalk(ev.Walk)
h.endWalk(w, ev.Outcome, at, now)
}
}
// release is what a walk held that it alone still holds: 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.
// 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).
func (h *Holder) release(w *Walk, now time.Time) (open []*Record, whats map[string]string, cleared int) {
h.mu.Lock()
defer h.mu.Unlock()
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()
var s subject
if live {
s = subjectOf(r)
}
elsewhere := false
for _, o := range h.walks {
if o.ID != w.ID && o.holding(now) {
if _, ok := o.Held[key]; ok {
elsewhere = true
if o.ID == w.ID || !o.holding(now) {
continue
}
if _, ok := o.Held[key]; ok {
elsewhere = true
continue
}
if live && o.touches(s) {
if o.Held == nil {
o.Held = map[string]HeldMessage{}
}
o.Held[key] = hm
taken[o.ID] = o
handed = append(handed, key+" to walk "+o.ID)
elsewhere = true
}
}
if elsewhere {
continue
}
r := h.open[key]
switch {
case r == nil || !r.Cleared.IsZero():
cleared++
@@ -362,6 +403,16 @@ func (h *Holder) release(w *Walk, now time.Time) (open []*Record, whats map[stri
}
return open[i].Key < open[j].Key
})
h.mu.Unlock()
for _, o := range taken {
h.keepWalk(o)
}
sort.Strings(handed)
for _, x := range handed {
key, _, _ := strings.Cut(x, " to walk ")
h.record(Sent{At: now, Channel: "-", Key: key, What: "held on: walk " + w.ID + " ended, " + x[len(key)+1:] + " touches it",
Outcome: "held"})
}
return open, whats, cleared
}
@@ -389,7 +440,7 @@ func (h *Holder) endWalk(w *Walk, outcome string, ended, now time.Time) {
lines = append(lines, h.heldLines(open, whats, cleared)...)
m = Message{Title: head + ": " + reached, Body: strings.Join(lines, "\n")}
}
h.tellWalk(w, m, open, oneLine, "ended "+outcome, now)
h.tellWalk(w, m, open, whats, oneLine, "ended "+outcome, now)
}
// heldLines are the lines a walk's end or bound says of what it held: each still open, then the cleared count.
@@ -414,27 +465,47 @@ func (h *Holder) heldLines(open []*Record, whats map[string]string, cleared int)
return lines
}
// boundWalks says, of every walk past its bound, what it holds that is still open (ADR 0308 §5), and forgets
// the walks whose end never came.
// boundWalks says, of every walk past its bound, what it holds that is still open (ADR 0308 §5), then forgets
// the walks whose end never came, and the ends heard a day ago. What a walk holds is said before it is
// forgotten: a router back after a long while says it, then forgets the walk.
func (h *Holder) boundWalks(now time.Time) {
h.mu.Lock()
var due, gone []*Walk
var stale []string
for _, w := range h.walks {
switch {
case now.Sub(w.Started) >= WalkForget:
gone = append(gone, w)
case !w.BoundSaid && !now.Before(w.Started.Add(w.bound())):
if !w.BoundSaid && !now.Before(w.Started.Add(w.bound())) {
due = append(due, w)
}
if now.Sub(w.Started) >= WalkForget {
gone = append(gone, w)
}
}
for _, w := range gone {
delete(h.walks, w.ID)
for id, e := range h.ended {
if now.Sub(e.Ended) >= WalkForget {
stale = append(stale, id)
}
}
for _, id := range stale {
delete(h.ended, id)
}
h.mu.Unlock()
for _, w := range gone {
h.Logf("[messenger] walk %s started %s ago and its end never came: forgotten", w.ID, roughly(now.Sub(w.Started)))
h.forgetWalk(w.ID)
for _, id := range stale {
h.mu.Lock()
reopened := h.walks[id] != nil
h.mu.Unlock()
if !reopened {
h.forgetWalk(id)
}
}
defer func() {
for _, w := range gone {
h.mu.Lock()
delete(h.walks, w.ID)
h.mu.Unlock()
h.Logf("[messenger] walk %s started %s ago and its end never came: forgotten", w.ID, roughly(now.Sub(w.Started)))
h.forgetWalk(w.ID)
}
}()
sort.Slice(due, func(i, j int) bool { return due[i].Started.Before(due[j].Started) })
for _, w := range due {
h.mu.Lock()
@@ -456,17 +527,21 @@ func (h *Holder) boundWalks(now time.Time) {
}
lines = append(lines, h.heldLines(open, whats, cleared)...)
m := Message{Title: "Still running: the delivery to " + touchWords(w.Touches), Body: strings.Join(lines, "\n")}
h.tellWalk(w, m, open, false, "past its bound", now)
h.tellWalk(w, m, open, whats, false, "past its bound", now)
}
}
// tellWalk holds a walk's message on the kinds it goes to, alone, and marks what it carries as said there.
func (h *Holder) tellWalk(w *Walk, m Message, open []*Record, oneLine bool, what string, now time.Time) {
// tellWalk holds a walk's message on the kinds it goes to, alone, and marks what it carries as said there. When
// no channel can send, or the message cannot carry what it held, each held message is left to its own path:
// unsent, and tried again every minute as any other (outbox.go), or said on its own.
func (h *Holder) tellWalk(w *Walk, m Message, open []*Record, whats map[string]string, oneLine bool, what string, now time.Time) {
for _, r := range open {
m.Urgent = m.Urgent || r.Severity == Urgent
}
var alone []*Record // said on their own, after the walk's message
if _, ok := Check(m.Text()); !ok {
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
}
var to []Channel
if m.Urgent {
@@ -492,8 +567,21 @@ func (h *Holder) tellWalk(w *Walk, m Message, open []*Record, oneLine bool, what
}
kept = append(kept, ch)
}
if len(kept) == 0 && len(to) == 0 {
h.Logf("[messenger] walk %s %s: no channel can send now; what it held stays open here", w.ID, what)
if len(to) == 0 {
h.Logf("[messenger] walk %s %s: no channel can send now; what it held is tried again every minute: %s", w.ID, what, h.whyNot())
for _, r := range append(open, alone...) {
say := whats[r.Key]
if say == "" {
say = "raised"
}
h.mu.Lock()
r.Pending = say
h.mu.Unlock()
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: "released by walk " + w.ID + " " + what,
Outcome: "unsent: no channel can send; tried again every minute"})
h.persist(r)
}
return
}
sev := Warning
if m.Urgent {
@@ -527,6 +615,15 @@ func (h *Holder) tellWalk(w *Walk, m Message, open []*Record, oneLine bool, what
h.persist(r)
}
h.flush(now, false)
for _, r := range alone {
say := whats[r.Key]
if say == "" {
say = "raised"
}
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: "released", Outcome: "said on its own: walk " + w.ID + "'s message could not carry it"})
h.deliver(r, say)
h.persist(r)
}
}
// touchWords names what a walk reaches: its modules, then the machines.
@@ -750,6 +847,10 @@ func (h *Holder) loadWalks() error {
}
for i := range walks {
w := walks[i]
if !w.Ended.IsZero() {
h.ended[w.ID] = &w
continue
}
if w.Held == nil {
w.Held = map[string]HeldMessage{}
}
@@ -2,6 +2,7 @@ package main
import (
"encoding/json"
"errors"
"strings"
"testing"
"time"
@@ -443,3 +444,179 @@ func TestTheWalkEventsAreReadAsTheControllerSaysThem(t *testing.T) {
}
}
// Review of pull request 230, 1: no channel can send at a walk's end or bound. What it held is not lost with the
// walk: it is unsent, and said on its own once a channel can send.
func TestWhatAWalkHeldIsSaidOnceAChannelCanSendAgain(t *testing.T) {
for _, at := range []string{"end", "bound"} {
t.Run(at, func(t *testing.T) {
h, tg, dt, c, _ := newHolder(t)
walk(t, h, "plan-c", WalkStarted, "", c.now(), 600, 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, 2*time.Minute)
down := errors.New("no session")
tg.notReady, dt.notReady = down, down
if at == "end" {
walk(t, h, "plan-c", WalkEnded, WalkFailed, c.now(), 0)
runFor(h, c, time.Minute)
} else {
runFor(h, c, 10*time.Minute)
}
if saidOn(tg, dt) != 0 {
t.Fatalf("said on a channel that cannot send: %+v %+v", tg.sends, dt.sends)
}
if r := h.open[k.Key]; r == nil || r.Pending == "" {
t.Fatalf("what the walk held is not left unsent: %+v", r)
}
tg.notReady, dt.notReady = nil, nil
runFor(h, c, 2*time.Minute)
if !heard(headlineOf(k.Key), dt) {
t.Fatalf("what the walk held was never said: %v", outcomes(h, k.Key))
}
})
}
}
// Review of pull request 230, 2: a message held by walk A, about a machine walk B also touches, is held on at
// A's end while B — started after it was held — is still open, and told at B's end.
func TestAMessageIsHeldOnByALaterWalkThatTouchesIt(t *testing.T) {
h, tg, dt, c, _ := newHolder(t)
walk(t, h, "plan-a", WalkStarted, "", c.now(), 1200, novox("mesh-tools"))
k := cond("machine.novox.network", Urgent, "novox lost its default route")
k.Raised = c.now()
h.Condition(EventRaised, k)
runFor(h, c, time.Minute)
walk(t, h, "plan-b", WalkStarted, "", c.now(), 1200, novox("node-tools"))
runFor(h, c, time.Minute)
walk(t, h, "plan-a", WalkEnded, WalkDelivered, c.now(), 0)
runFor(h, c, time.Minute)
if heard(headlineOf(k.Key), tg, dt) {
t.Fatalf("told at the first walk's end while the second touches it: %+v %+v", tg.sends, dt.sends)
}
if got := h.heldBy(k.Key, c.now()); len(got) != 1 || got[0] != "plan-b" {
t.Fatalf("the second walk does not hold it on: %v", got)
}
walk(t, h, "plan-b", WalkEnded, WalkDelivered, c.now(), 0)
runFor(h, c, time.Minute)
if !heard("Still open after it:\n- Urgent: "+headlineOf(k.Key), tg, dt) {
t.Fatalf("the second walk's end did not tell it: %+v %+v", tg.sends, dt.sends)
}
}
// Review of pull request 230, 3: a router back after more than a day reads back a walk whose end never came;
// what it held is said, with the still-running line, before the walk is forgotten.
func TestAWalkReadBackADayLateSaysWhatItHeldBeforeItIsForgotten(t *testing.T) {
h, tg, dt, c, st := newHolder(t)
walk(t, h, "plan-d", WalkStarted, "", c.now(), 600, 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)
c.pass(25 * time.Hour)
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()
runFor(again, c, 2*time.Minute)
if !heard("is still running", dt) || !heard("- Urgent: "+headlineOf(k.Key), dt) {
t.Fatalf("what the walk held was forgotten unsaid: desk %+v, %v", dt.sends, outcomes(again, k.Key))
}
if len(again.walks) != 0 {
t.Fatalf("the walk was not forgotten: %+v", again.walks)
}
if ws, _ := st.Walks(); len(ws) != 0 {
t.Fatalf("the walk is still kept: %+v", ws)
}
}
// Review of pull request 230, 4 (and of mesh-controller pull request 247): a short walk's end can come before
// its start. The start of a walk already ended opens nothing — also after the router's restart — while the
// next walk under the same id, started after that end, does.
func TestAStartHeardAfterItsEndOpensNothing(t *testing.T) {
h, tg, dt, c, st := newHolder(t)
started := c.now()
c.pass(20 * time.Second)
end, _, err := DecodeWalk([]byte(`{"walk":"plan-e","phase":"ended","outcome":"delivered","touches":[{"machine":"novox","modules":["mesh-tools"]}],` +
`"started_at":"` + started.Format(time.RFC3339Nano) + `","ended_at":"` + c.now().Format(time.RFC3339Nano) + `"}`))
if err != nil {
t.Fatal(err)
}
h.Walked(end)
c.pass(time.Second)
walk(t, h, "plan-e", 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)
}
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()
walk(t, again, "plan-e", WalkStarted, "", started, 600, novox("mesh-tools"))
if len(again.walks) != 0 {
t.Fatalf("after a restart, a start heard after its end opened a walk: %+v", again.walks)
}
k := cond("machine.novox.tools-silent", Urgent, "the tools on novox are silent")
k.Raised = c.now()
again.Condition(EventRaised, k)
runFor(again, c, time.Minute)
if !heard(headlineOf(k.Key), dt) {
t.Fatalf("held by a walk that had ended: %v", outcomes(again, k.Key))
}
// Retried: the same id walks again, started after the end.
walk(t, again, "plan-e", WalkStarted, "", c.now(), 600, novox("mesh-tools"))
if len(again.walks) != 1 {
t.Fatalf("a walk started after the last end did not open: %+v", again.walks)
}
}
// Review of pull request 230, non-blocking: a message whose answers became asked while a walk held it is never
// held (ADR 0308 §3); when the desk's answers are lost, it is said again in place like any other, so its
// buttons work again.
func TestARestartSaysAgainAnAskedMessageAWalkTookBeforeItsAsk(t *testing.T) {
h, _, dt, c, _, _ := newRouter(t)
h.Start()
stand(h, c)
walk(t, h, "plan-f", WalkStarted, "", c.now(), 1200, novox("mesh-tools"))
k := cond("machine.novox.units", Urgent, "2 failed services on novox")
k.Raised = c.now()
h.Condition(EventRaised, k)
runFor(h, c, time.Minute)
ask := ackAsk(c, "units")
ask.About = k.Key
submit(t, h, ConditionOwner, ask)
runFor(h, c, time.Minute)
stand(h, c)
if !heard(headlineOf(k.Key), dt) {
t.Fatalf("the ask about it was not offered: %v", outcomes(h, k.Key))
}
replaced := len(dt.replaced)
c.pass(time.Minute)
h.Stand("desktop", Standing{Ready: true, At: c.now(), AnswersLostAt: c.now()})
runFor(h, c, time.Minute)
if len(dt.replaced) != replaced+1 {
t.Fatalf("its answers were not made to work again: replaced %+v, %v", dt.replaced[replaced:], outcomes(h, "(answers lost)"))
}
}
// Review of pull request 230, non-blocking: a walk's end that cannot be said as it is (here a module named by a
// file path) is said in its plain form, and what it held, still open, is said on its own, not marked told.
func TestWhatAWalksPlainEndCannotCarryIsSaidOnItsOwn(t *testing.T) {
h, tg, dt, c, _ := newHolder(t)
walk(t, h, "plan-g", WalkStarted, "", c.now(), 600, novox("/var/lib/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)
walk(t, h, "plan-g", WalkEnded, WalkFailed, c.now(), 0)
runFor(h, c, time.Minute)
if !heard("A delivery ended", dt) {
t.Fatalf("the plain end was not said: %+v", dt.sends)
}
if !heard(headlineOf(k.Key), tg, dt) {
t.Fatalf("what the walk held was marked told and never said: %v", outcomes(h, k.Key))
}
}