Merge pull request 'messenger: hold what a delivery walk touches, tell its end once, and say nothing again for a restart (hq ADR 0308, issues 532, 531)' (#230) from adr-0308-hold-during-walks into main
This commit was merged in pull request #230.
This commit is contained in:
@@ -16,15 +16,20 @@ channel's client and no channel's secret.
|
||||
its sender) and `desk-channel` (kind `desktop`, the machines' notifiers, proves nothing). Which kinds exist
|
||||
and what each promises is the controller's word, in this module's membership; whether each can send now
|
||||
is the channel's own, said on `intake.standing`.
|
||||
- **Keeps its state on the bus** (ADR 0201): `open` and `sent` (the messages), `asks` (every ask, closed ones
|
||||
for 30 days), `identities` (the operator's linked accounts, one per kind, and a link asked for).
|
||||
- **Keeps its state on the bus** (ADR 0201): `open` and `sent` (the messages, and the restart passes made),
|
||||
`asks` (every ask, closed ones for 30 days), `identities` (the operator's linked accounts, one per kind, and a
|
||||
link asked for), `walks` (the delivery walks open now, each with what it holds).
|
||||
- **Lost answers are said again from here, never from a channel's memory** (issue 384,
|
||||
`cmd/messenger/restart.go`): a channel whose messages' answers were lost — the desk's notifier started
|
||||
again — says when in its standing (`answers-lost-at`). The router says again on that kind, directly and not
|
||||
through the outbox, each message that is still open, asked there and last said before that time: one per
|
||||
problem, under its handle, with the words and answers of now, so it replaces the message in place. What is
|
||||
silenced is edited to say so, with no answers and no urgency; what ended never comes back. A kind that cannot
|
||||
send waits; it never goes to another channel. Telegram's answers are never lost and it says no time.
|
||||
through the outbox, each message that is still open, asked there, never answered and last said before that
|
||||
time: one per problem, under its handle, with the words and answers of now, so it replaces the message in
|
||||
place. **A restart says nothing it has said** (novox/hq ADR 0308 §6, issue 531): a silenced message, one
|
||||
withheld for its content, one answered and one a walk holds are left as they are, and what ended never comes
|
||||
back. The standings of one restart (a notifier on each machine, seconds apart) are one pass: a loss is acted
|
||||
on once it stood 10 s, and the pass made is kept in the state (`sent`, key `passes`), so a router that starts
|
||||
again does not make it again. A kind that cannot send waits; it never goes to another channel. Telegram's
|
||||
answers are never lost and it says no time.
|
||||
|
||||
## Asks, answers and warrants (novox/hq ADR 0259)
|
||||
|
||||
@@ -165,10 +170,32 @@ ends takes them away everywhere; its expiry is said with its day when it is not
|
||||
history. An event that cannot be read
|
||||
is refused by name, counted, and told to the operator once an hour.
|
||||
|
||||
## While a delivery walks (novox/hq ADR 0308)
|
||||
|
||||
The controller's `plan-moved` says each walk's start and end: `walk`, `phase` (`started`, `ended`), `touches`
|
||||
(`[{machine, modules}]`: what it moves and what a moved tool runner or node-engine carries), `outcome`
|
||||
(`delivered`, `failed`, `stopped`), `bound_seconds` and `budget_class` (`cmd/messenger/walks.go`).
|
||||
|
||||
- **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.
|
||||
- **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`.
|
||||
- **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.
|
||||
- **Kept in the state** (`walks`): a router that starts again mid-walk holds and tells as before.
|
||||
|
||||
## What the operator gives
|
||||
|
||||
1. **The time zone**, as a setting: `{"time-zone": "<an IANA zone name>"}`. Every time a message says is
|
||||
said in it; not given, the zone of the machine this runs on.
|
||||
**The kinds that say only what needs the operator**, as a setting: `{"only-what-needs-me": ["<kind>"]}`. A
|
||||
walk that delivered with nothing held and nothing open is not told on them (ADR 0308 §4); empty by default.
|
||||
2. **The channels**: assign `telegram` and `desk-channel` (each says in its README what it needs). Linking
|
||||
an account as the operator is done on the channel, never by a setting.
|
||||
|
||||
|
||||
@@ -154,6 +154,11 @@ type Record struct {
|
||||
Subject string `json:"subject"`
|
||||
Severity string `json:"severity"`
|
||||
Summary string `json:"summary"`
|
||||
// Scope, SubjectID and Machine are the condition's subject as the controller gave it, which a walk's touches
|
||||
// are matched against (ADR 0308 §2); empty for a record kept before, whose key names them.
|
||||
Scope string `json:"scope,omitempty"`
|
||||
SubjectID string `json:"subject_id,omitempty"`
|
||||
Machine string `json:"machine,omitempty"`
|
||||
// Headline, Explanation and Resolved are the plain words the operator reads (novox/hq ADR 0253):
|
||||
// the controller's, or a module's through notify; empty, the message is said from the summary.
|
||||
Headline string `json:"headline,omitempty"`
|
||||
@@ -211,7 +216,8 @@ type Record struct {
|
||||
|
||||
// recordOf is a condition as this holder keeps it.
|
||||
func recordOf(c Condition) Record {
|
||||
return Record{Key: c.Key, Kind: c.Kind, Subject: c.SubjectWords(), Severity: c.Severity, Summary: c.Summary,
|
||||
return Record{Key: c.Key, Kind: c.Kind, Subject: c.SubjectWords(), Scope: c.Scope, SubjectID: c.ID, Machine: c.Machine,
|
||||
Severity: c.Severity, Summary: c.Summary,
|
||||
Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved, Needs: c.Needs, Actions: c.Actions,
|
||||
Origin: "condition",
|
||||
More: "conditions show " + c.Key, Raised: c.Raised, SilencedTill: c.SilencedTill, SilencedWhy: c.SilencedWhy}
|
||||
@@ -232,6 +238,13 @@ func (r *Record) takeWords(from Record) {
|
||||
}
|
||||
}
|
||||
|
||||
// takeSubject keeps the subject as the controller says it now, when it says one.
|
||||
func (r *Record) takeSubject(from Record) {
|
||||
if from.Scope != "" || from.SubjectID != "" || from.Machine != "" {
|
||||
r.Scope, r.SubjectID, r.Machine = from.Scope, from.SubjectID, from.Machine
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Record) silenced(now time.Time) bool {
|
||||
return !r.SilencedTill.IsZero() && now.Before(r.SilencedTill)
|
||||
}
|
||||
@@ -292,6 +305,9 @@ type Holder struct {
|
||||
Zone func() *time.Location
|
||||
// Tracking asks the issue tracker which issues track which conditions; nil asks it through Ask (tracking.go).
|
||||
Tracking func() (map[string][]TrackedIssue, error)
|
||||
// NeedsOnly says whether a kind is set to say only what needs the operator (the setting
|
||||
// `only-what-needs-me`, ADR 0308 §4); nil, none is.
|
||||
NeedsOnly func(kind string) bool
|
||||
|
||||
work sync.Mutex // one event, call or tick at a time
|
||||
mu sync.Mutex // what the status reads
|
||||
@@ -329,8 +345,14 @@ type Holder struct {
|
||||
lastUnlink map[string]Unlinked
|
||||
// rootWord is the controller's last word on whether an agent can become root, by the machines asked.
|
||||
rootWord map[string]rootWord
|
||||
// shows is, by kind, the AnswersLostAt this router last acted on in full (restart.go).
|
||||
shows map[string]time.Time
|
||||
// passes is, by kind, the restart pass this router last made in full, kept in its state (restart.go); lost
|
||||
// the loss of answers it has yet to act on, by kind.
|
||||
passes map[string]Pass
|
||||
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
|
||||
}
|
||||
@@ -348,7 +370,10 @@ func (h *Holder) init() {
|
||||
h.chanOK = map[string]time.Time{}
|
||||
h.saidOnce = map[string]time.Time{}
|
||||
h.heard = map[string]int{}
|
||||
h.shows = map[string]time.Time{}
|
||||
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
|
||||
@@ -387,6 +412,10 @@ func (h *Holder) Load() error {
|
||||
if sent, err := h.Store.Recent(); err == nil {
|
||||
h.recent = sent
|
||||
}
|
||||
if err := h.loadWalks(); err != nil {
|
||||
h.storeErr = err.Error()
|
||||
return err
|
||||
}
|
||||
h.mu.Unlock()
|
||||
err = h.loadAsks()
|
||||
h.mu.Lock()
|
||||
@@ -485,6 +514,7 @@ func (h *Holder) raised(rec Record, quiet bool) {
|
||||
wasQuiet := needsNothing(old)
|
||||
wasSilenced := old.silenced(now)
|
||||
old.Summary, old.Subject, old.Kind = rec.Summary, rec.Subject, rec.Kind
|
||||
old.takeSubject(rec)
|
||||
old.SilencedTill, old.SilencedWhy = rec.SilencedTill, rec.SilencedWhy
|
||||
old.takeWords(rec)
|
||||
if rec.Severity != "" {
|
||||
@@ -584,6 +614,7 @@ func (h *Holder) changed(rec Record, quiet bool) {
|
||||
wasQuiet := needsNothing(old)
|
||||
wasSilenced := old.silenced(h.Now())
|
||||
old.Summary, old.Subject = rec.Summary, rec.Subject
|
||||
old.takeSubject(rec)
|
||||
old.SilencedTill, old.SilencedWhy = rec.SilencedTill, rec.SilencedWhy
|
||||
old.takeWords(rec)
|
||||
if rec.Severity != "" {
|
||||
@@ -710,6 +741,8 @@ func (h *Holder) Tick() {
|
||||
sort.Slice(recs, func(i, j int) bool { return recs[i].Raised.Before(recs[j].Raised) })
|
||||
h.sweepAsks(now)
|
||||
h.breakGlass()
|
||||
h.boundWalks(now)
|
||||
h.duePasses(now)
|
||||
for _, r := range recs {
|
||||
switch {
|
||||
case !r.Cleared.IsZero():
|
||||
@@ -775,6 +808,7 @@ func (h *Holder) Flush() {
|
||||
h.init()
|
||||
now := h.Now()
|
||||
h.mu.Unlock()
|
||||
h.duePasses(now)
|
||||
h.flush(now, false)
|
||||
}
|
||||
|
||||
|
||||
@@ -66,6 +66,9 @@ func (f *fakeChannel) Edit(id string, m Message) error {
|
||||
type memStore struct {
|
||||
recs map[string]Record
|
||||
recent []Sent
|
||||
// walks and passes are what ADR 0308 keeps: the open walks and the restart passes.
|
||||
walks map[string]Walk
|
||||
passes map[string]Pass
|
||||
// puts counts the writes, so a write of bytes identical to what is stored is still seen (issue 422's review).
|
||||
puts int
|
||||
}
|
||||
@@ -87,7 +90,24 @@ func (m *memStore) All() ([]Record, error) {
|
||||
return out, nil
|
||||
}
|
||||
func (m *memStore) PutRecent(s []Sent) error { m.recent = s; return nil }
|
||||
func (m *memStore) Recent() ([]Sent, error) { return m.recent, nil }
|
||||
func (m *memStore) PutWalk(w Walk) error {
|
||||
if m.walks == nil {
|
||||
m.walks = map[string]Walk{}
|
||||
}
|
||||
m.walks[w.ID] = w
|
||||
return nil
|
||||
}
|
||||
func (m *memStore) DeleteWalk(id string) error { delete(m.walks, id); return nil }
|
||||
func (m *memStore) Walks() ([]Walk, error) {
|
||||
var out []Walk
|
||||
for _, w := range m.walks {
|
||||
out = append(out, w)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (m *memStore) PutPasses(p map[string]Pass) error { m.passes = p; return nil }
|
||||
func (m *memStore) Passes() (map[string]Pass, error) { return m.passes, nil }
|
||||
func (m *memStore) Recent() ([]Sent, error) { return m.recent, nil }
|
||||
|
||||
type clock struct{ t time.Time }
|
||||
|
||||
|
||||
@@ -29,6 +29,9 @@ type Settings struct {
|
||||
// TimeZone is the operator's, an IANA name such as Europe/Brussels: every time a message says is said
|
||||
// in it. Empty is this machine's own zone.
|
||||
TimeZone string `json:"time-zone"`
|
||||
// OnlyWhatNeedsMe are the channel kinds set to say only what needs the operator: a walk that delivered
|
||||
// with nothing held and nothing open is not told there at all (novox/hq ADR 0308 §4).
|
||||
OnlyWhatNeedsMe []string `json:"only-what-needs-me"`
|
||||
}
|
||||
|
||||
// zoneOf is the location a time-zone setting names, or this machine's own when it names none or one
|
||||
@@ -60,6 +63,19 @@ func readSettings(path string) (Settings, error) {
|
||||
if tz, ok := loose["time-zone"].(string); ok {
|
||||
s.TimeZone = strings.TrimSpace(tz)
|
||||
}
|
||||
switch v := loose["only-what-needs-me"].(type) {
|
||||
case nil:
|
||||
case []any:
|
||||
for _, k := range v {
|
||||
kind, ok := k.(string)
|
||||
if !ok || strings.TrimSpace(kind) == "" {
|
||||
return s, fmt.Errorf("only-what-needs-me lists something that is not a channel kind: %v", k)
|
||||
}
|
||||
s.OnlyWhatNeedsMe = append(s.OnlyWhatNeedsMe, strings.TrimSpace(kind))
|
||||
}
|
||||
default:
|
||||
return s, fmt.Errorf("only-what-needs-me is not a list of channel kinds")
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
@@ -109,6 +125,50 @@ func (stateStore) All() ([]Record, error) {
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// The walks open now, one key each in the state `walks`, and the restart passes under one key of `sent`
|
||||
// (novox/hq ADR 0308): a router that starts again holds, tells and passes as if it had not.
|
||||
func (stateStore) PutWalk(w Walk) error {
|
||||
_, err := stdio.State("walks").Put(kvKey(w.ID), w)
|
||||
return err
|
||||
}
|
||||
func (stateStore) DeleteWalk(id string) error { return stdio.State("walks").Delete(kvKey(id)) }
|
||||
func (stateStore) Walks() ([]Walk, error) {
|
||||
keys, err := stdio.State("walks").Keys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []Walk
|
||||
for _, k := range keys {
|
||||
e, err := stdio.State("walks").Get(k)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if e == nil {
|
||||
continue
|
||||
}
|
||||
var w Walk
|
||||
if err := json.Unmarshal(e.Value, &w); err != nil {
|
||||
logf("[messenger] the walk kept as %s cannot be read (%v); left as it is", k, err)
|
||||
continue
|
||||
}
|
||||
out = append(out, w)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (stateStore) PutPasses(p map[string]Pass) error {
|
||||
_, err := stdio.State("sent").Put("passes", p)
|
||||
return err
|
||||
}
|
||||
func (stateStore) Passes() (map[string]Pass, error) {
|
||||
e, err := stdio.State("sent").Get("passes")
|
||||
if err != nil || e == nil {
|
||||
return nil, err
|
||||
}
|
||||
var p map[string]Pass
|
||||
return p, json.Unmarshal(e.Value, &p)
|
||||
}
|
||||
|
||||
func (stateStore) PutRecent(s []Sent) error {
|
||||
_, err := stdio.State("sent").Put("recent", s)
|
||||
return err
|
||||
@@ -271,6 +331,14 @@ func main() {
|
||||
return loc
|
||||
},
|
||||
Emit: func(event string, body any) error { return stdio.Emit(event, body) },
|
||||
NeedsOnly: func(kind string) bool {
|
||||
for _, k := range settings().OnlyWhatNeedsMe {
|
||||
if k == kind {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
},
|
||||
}
|
||||
bench.To = func(kind string) string {
|
||||
if ids := h.identitiesOf(kind); len(ids) > 0 {
|
||||
@@ -326,6 +394,18 @@ func run(h *Holder, l *listening) {
|
||||
}
|
||||
}()
|
||||
handle := func(e stdio.Envelope) error {
|
||||
if e.Key == ControllerSeat+"."+EventPlanMoved || e.Key == ControllerSeat+"."+EventWalk {
|
||||
// A walk's start or end (novox/hq ADR 0308 §1); any other move of a plan is not read here.
|
||||
ev, isWalk, err := DecodeWalk(e.Body)
|
||||
if err != nil {
|
||||
h.Unreadable(e.Key, err)
|
||||
return nil
|
||||
}
|
||||
if isWalk {
|
||||
h.Walked(ev)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
event, ok := eventOf(e.Key)
|
||||
if !ok {
|
||||
return nil
|
||||
|
||||
@@ -81,7 +81,7 @@ func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) {
|
||||
}
|
||||
if !reflect.DeepEqual(m.Consumes, []string{
|
||||
ControllerSeat + "." + EventRaised, ControllerSeat + "." + EventChanged, ControllerSeat + "." + EventCleared,
|
||||
"intake.choice", "intake.link", "intake.failed", "intake.standing"}) {
|
||||
ControllerSeat + "." + EventPlanMoved, "intake.choice", "intake.link", "intake.failed", "intake.standing"}) {
|
||||
t.Fatalf("consumes %v", m.Consumes)
|
||||
}
|
||||
// It performs nothing: the verbs it calls read the open conditions, whether root is free, and which issues
|
||||
@@ -100,7 +100,7 @@ func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) {
|
||||
}
|
||||
state = append(state, name)
|
||||
}
|
||||
if !reflect.DeepEqual(state, []string{"open", "sent", "asks", "identities"}) {
|
||||
if !reflect.DeepEqual(state, []string{"open", "sent", "asks", "identities", "walks"}) {
|
||||
t.Fatalf("state %v", state)
|
||||
}
|
||||
// It holds no channel's secret: only its own bus account, read by its own account alone (ADR 0259 §8).
|
||||
|
||||
@@ -32,6 +32,10 @@ func (h *Holder) deliver(r *Record, what string) {
|
||||
h.record(Sent{At: h.Now(), Channel: "-", Key: r.Key, What: what, Outcome: "quiet: nothing for the operator to do"})
|
||||
return
|
||||
}
|
||||
if h.hold(r, what) {
|
||||
// An open walk touches it: told when the walk ends or passes its bound (ADR 0308 §2).
|
||||
return
|
||||
}
|
||||
var to []Channel
|
||||
if kinds := h.askKinds(r); len(kinds) > 0 {
|
||||
// An ask goes to every channel that carries one of its answers (novox/hq ADR 0259): the desk for an
|
||||
@@ -135,6 +139,9 @@ func (h *Holder) queued(key string) bool {
|
||||
// edit fails, or that said it inside a digest, is given a new message instead — held like any other.
|
||||
// silentOnly: only where an edit notifies nobody; elsewhere nothing.
|
||||
func (h *Holder) edit(r *Record, what string, silentOnly bool) {
|
||||
if h.hold(r, what) {
|
||||
return // an open walk touches it (ADR 0308 §2); a clearing or a silence is never held
|
||||
}
|
||||
h.mu.Lock()
|
||||
sent := map[string]string{}
|
||||
for k, v := range r.Sent {
|
||||
@@ -225,12 +232,23 @@ func (h *Holder) due(ch Channel, list []item, now time.Time) bool {
|
||||
if now.Sub(oldest) < BurstWindow {
|
||||
return false
|
||||
}
|
||||
if ch == h.Desk && !lastWarn.IsZero() && now.Sub(lastWarn) < DesktopWarningEvery {
|
||||
if ch == h.Desk && !lastWarn.IsZero() && now.Sub(lastWarn) < DesktopWarningEvery && !carriesWalk(list) {
|
||||
// A walk's message is what the rhythm would have spared the operator many of: not held to it.
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// carriesWalk says whether held items carry a walk's end or bound message.
|
||||
func carriesWalk(list []item) bool {
|
||||
for _, it := range list {
|
||||
if it.What == WalkEndWhat {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// line is what an item says, resolved against the record now: nil when it is no longer to be said.
|
||||
func (h *Holder) resolve(it item, now time.Time, kind string) (*Record, *Message) {
|
||||
if it.Msg != nil {
|
||||
@@ -310,7 +328,9 @@ func (h *Holder) flushChannel(ch Channel, now time.Time, force bool) {
|
||||
for _, e := range entries {
|
||||
// A message said again about a record shown here goes alone too: it replaces that record's message,
|
||||
// so a reminder is never a digest line beside the message it follows (issue 373).
|
||||
if len(e.msg.Actions) > 1 || (len(e.msg.Actions) > 0 && e.r != nil && e.r.Ask != "") || h.replacing(ch, e) != "" {
|
||||
// A walk's end or bound message goes alone too: its lines are what it says (ADR 0308 §4).
|
||||
if len(e.msg.Actions) > 1 || (len(e.msg.Actions) > 0 && e.r != nil && e.r.Ask != "") || h.replacing(ch, e) != "" ||
|
||||
e.it.What == WalkEndWhat {
|
||||
alone = append(alone, e)
|
||||
} else {
|
||||
rest = append(rest, e)
|
||||
|
||||
@@ -18,6 +18,22 @@ package main
|
||||
// Whether a record is said again is its own: last said before the answers were lost. So a router that started
|
||||
// after the loss still says it, and a standing that repeats the time says nothing more (the record was said
|
||||
// since). A kind that cannot send waits: what is open there is said once it can.
|
||||
//
|
||||
// **A restart says nothing it has said** (novox/hq ADR 0308 §6, issue 531). On 2026-10-12 every delivery of the
|
||||
// tool runner made this pass twice, a second apart, each time editing the four silenced messages on the desk,
|
||||
// which showed them again — one as "A mesh message was withheld". Three causes, each closed here:
|
||||
//
|
||||
// - **Twice:** the desk's standing says the latest start of any machine's notifier, and the notifiers of two
|
||||
// machines started a second apart, so two standings with two times came, and each was acted on at once.
|
||||
// Now a loss is acted on once it stands for LostSettle: the standings of one restart are one pass.
|
||||
// - **Again after the router's own restart:** which loss was acted on was kept in memory, so a router that
|
||||
// started again acted on the standing it had already acted on. Now it is kept in the module's state
|
||||
// (`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 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"
|
||||
@@ -25,7 +41,17 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// Stand takes what a kind said of itself, and acts on answers lost there since this router last acted.
|
||||
// LostSettle: a loss of answers is acted on once no later loss came for this long — the standings of one
|
||||
// restart, a notifier on each machine, are one pass.
|
||||
const LostSettle = 10 * time.Second
|
||||
|
||||
// Pass is the restart pass made in full on a kind: the loss it answered and when it was made.
|
||||
type Pass struct {
|
||||
Lost time.Time `json:"lost"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Stand takes what a kind said of itself, and notes answers lost there since this router last acted.
|
||||
func (h *Holder) Stand(kind string, s Standing) {
|
||||
if h.Bench != nil {
|
||||
h.Bench.Stand(kind, s)
|
||||
@@ -33,28 +59,73 @@ func (h *Holder) Stand(kind string, s Standing) {
|
||||
if s.AnswersLostAt.IsZero() {
|
||||
return
|
||||
}
|
||||
h.work.Lock()
|
||||
defer h.work.Unlock()
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.init()
|
||||
done := h.shows[kind].Equal(s.AnswersLostAt)
|
||||
if !s.AnswersLostAt.After(h.passes[kind].Lost) || !s.AnswersLostAt.After(h.lost[kind]) {
|
||||
return // acted on, or noted already
|
||||
}
|
||||
h.lost[kind] = s.AnswersLostAt
|
||||
}
|
||||
|
||||
// duePasses makes the pass of each kind whose loss of answers stood for LostSettle, once per loss.
|
||||
func (h *Holder) duePasses(now time.Time) {
|
||||
h.mu.Lock()
|
||||
due := map[string]time.Time{}
|
||||
for kind, lost := range h.lost {
|
||||
if now.Sub(lost) >= LostSettle {
|
||||
due[kind] = lost
|
||||
}
|
||||
}
|
||||
h.mu.Unlock()
|
||||
if done {
|
||||
return
|
||||
}
|
||||
ch := h.channelOf(kind)
|
||||
if ch == nil {
|
||||
return // the controller's word on the kinds is not read yet: acted on at a later standing
|
||||
}
|
||||
if err := ch.Ready(); err != nil {
|
||||
h.Logf("[messenger] %s lost its answers at %s and cannot send (%v): what is open there is said again once it can",
|
||||
kind, s.AnswersLostAt.UTC().Format(time.RFC3339), err)
|
||||
return
|
||||
}
|
||||
if h.reoffer(ch, s.AnswersLostAt) {
|
||||
for kind, lost := range due {
|
||||
ch := h.channelOf(kind)
|
||||
if ch == nil {
|
||||
continue // the controller's word on the kinds is not read yet: acted on later
|
||||
}
|
||||
if err := ch.Ready(); err != nil {
|
||||
h.mu.Lock()
|
||||
said := h.saidOnce["lost."+kind]
|
||||
h.mu.Unlock()
|
||||
if said.IsZero() || now.Sub(said) >= 10*time.Minute {
|
||||
h.Logf("[messenger] %s lost its answers at %s and cannot send (%v): what is open there is said again once it can",
|
||||
kind, lost.UTC().Format(time.RFC3339), err)
|
||||
h.mu.Lock()
|
||||
h.saidOnce["lost."+kind] = now
|
||||
h.mu.Unlock()
|
||||
}
|
||||
continue
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.shows[kind] = s.AnswersLostAt
|
||||
tried := h.saidOnce["lost-try."+kind]
|
||||
h.mu.Unlock()
|
||||
if !tried.IsZero() && now.Sub(tried) < RetryEvery {
|
||||
continue
|
||||
}
|
||||
if !h.reoffer(ch, lost) {
|
||||
h.mu.Lock()
|
||||
h.saidOnce["lost-try."+kind] = now // what failed is tried again in a minute
|
||||
h.mu.Unlock()
|
||||
continue
|
||||
}
|
||||
h.mu.Lock()
|
||||
delete(h.saidOnce, "lost-try."+kind)
|
||||
h.mu.Unlock()
|
||||
h.mu.Lock()
|
||||
h.passes[kind] = Pass{Lost: lost, At: now}
|
||||
if h.lost[kind].Equal(lost) {
|
||||
delete(h.lost, kind)
|
||||
}
|
||||
passes := map[string]Pass{}
|
||||
for k, p := range h.passes {
|
||||
passes[k] = p
|
||||
}
|
||||
h.mu.Unlock()
|
||||
if k, ok := h.Store.(walkKeeper); ok {
|
||||
if err := k.PutPasses(passes); err != nil {
|
||||
h.noteStore(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -68,9 +139,9 @@ func (h *Holder) channelOf(kind string) Channel {
|
||||
return nil
|
||||
}
|
||||
|
||||
// reoffer says again on one channel, whose answers were lost at `lost`, what is still open and asked there,
|
||||
// edits what is silenced to say so, and leaves what ended alone. It answers whether every message to be
|
||||
// said again was.
|
||||
// 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 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)
|
||||
@@ -82,13 +153,14 @@ func (h *Holder) reoffer(ch Channel, lost time.Time) bool {
|
||||
}
|
||||
h.mu.Unlock()
|
||||
sort.Slice(recs, func(i, j int) bool { return recs[i].Raised.Before(recs[j].Raised) })
|
||||
again, silenced, ended, failed := 0, 0, 0, 0
|
||||
again, left, ended, failed := 0, 0, 0, 0
|
||||
for _, r := range recs {
|
||||
h.mu.Lock()
|
||||
id, shown := r.Sent[kind]
|
||||
open := r.Cleared.IsZero()
|
||||
before := lastSaid(r).Before(lost)
|
||||
reminded := r.Reminded
|
||||
withheld, answered := r.Refused != "", r.Answered != ""
|
||||
h.mu.Unlock()
|
||||
if !shown || id == "" {
|
||||
continue // never a message of its own there: nothing to say again under
|
||||
@@ -96,13 +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):
|
||||
h.mu.Lock()
|
||||
r.SilenceSaid = true
|
||||
h.mu.Unlock()
|
||||
h.editOn(ch, r, "silenced")
|
||||
h.persist(r)
|
||||
silenced++
|
||||
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 {
|
||||
@@ -125,28 +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 said so, %d ended left ended, %d failed",
|
||||
again, silenced, ended, failed)})
|
||||
h.Logf("[messenger] %s lost its answers at %s: %d open and asked said again in place, %d silenced said so, %d ended "+
|
||||
"left ended, %d failed; nothing shown from the channel's memory", kind, lost.UTC().Format(time.RFC3339), again,
|
||||
silenced, ended, failed)
|
||||
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 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
|
||||
}
|
||||
|
||||
// editOn edits a record's message on one channel, where it is shown as a message of its own.
|
||||
func (h *Holder) editOn(ch Channel, r *Record, what string) {
|
||||
kind := ch.Name()
|
||||
h.mu.Lock()
|
||||
id := r.Sent[kind]
|
||||
h.mu.Unlock()
|
||||
if id == "" || !ch.CanEdit() {
|
||||
return
|
||||
}
|
||||
err := ch.Edit(id, h.sayOn(r, what, kind))
|
||||
h.noteChannel(kind, err)
|
||||
if err != nil {
|
||||
h.Logf("[messenger] could not edit the message for %s on %s (%v)", r.Key, kind, err)
|
||||
return
|
||||
}
|
||||
h.record(Sent{At: h.Now(), Channel: kind, Key: r.Key, What: what, Outcome: "edited"})
|
||||
}
|
||||
|
||||
@@ -12,8 +12,8 @@ import (
|
||||
// resolved, silenced and cancelled ones among them. Now the desk says in its standing when the answers were
|
||||
// lost, and this router says again only what is still open and unanswered: with the desk showing one open
|
||||
// message with an answer, one silenced, one open asking nothing and one resolved, one is said again under its
|
||||
// handle (in place of its message), with its answers; one is edited to the silenced words, with no answers
|
||||
// and no urgency; nothing is said for the others. A standing that repeats the time says nothing more; its
|
||||
// handle (in place of its message), with its answers; nothing is said for the others — the silenced one
|
||||
// included (novox/hq ADR 0308 §6, issue 531: its edit showed it on the desk again). A standing that repeats the time says nothing more; its
|
||||
// ticket is one-time still, and the channel that reaches the operator away hears nothing of it.
|
||||
func TestLostAnswersGetOnlyWhatIsOpenAndUnansweredSaidAgain(t *testing.T) {
|
||||
h, tg, dt, c, _, w := newRouter(t)
|
||||
@@ -77,8 +77,7 @@ func TestLostAnswersGetOnlyWhatIsOpenAndUnansweredSaidAgain(t *testing.T) {
|
||||
if e := dt.edits["g14=4"]; !strings.HasPrefix(e.Title, "Resolved: ") {
|
||||
t.Fatalf("the clearing was not said: %+v", e)
|
||||
}
|
||||
silencedBefore := dt.edits["g14=2"]
|
||||
delete(dt.edits, "g14=2") // so the edit said again is seen as such
|
||||
delete(dt.edits, "g14=2") // so an edit said again would be seen as such
|
||||
sends, edits, replaced, tgSaid := len(dt.sends), len(dt.edits), len(dt.replaced), saidOn(tg)
|
||||
// The desk's notifier started again: its standing says when the answers were lost.
|
||||
c.pass(time.Minute)
|
||||
@@ -95,12 +94,11 @@ func TestLostAnswersGetOnlyWhatIsOpenAndUnansweredSaidAgain(t *testing.T) {
|
||||
if ticketFor(again, "Silence for a week") == "" || !strings.Contains(again.Title, headlineOf(open.Key)) {
|
||||
t.Fatalf("said again without its answer or its words: %+v", again)
|
||||
}
|
||||
if e, said := dt.edits["g14=2"]; !said || !strings.HasPrefix(e.Title, "Silenced until") || len(e.Actions) != 0 || e.Urgent ||
|
||||
!e.Quiet || e.Title != silencedBefore.Title {
|
||||
t.Fatalf("the silenced one was not edited to the silenced words again: %+v (said %v)", e, said)
|
||||
if e, said := dt.edits["g14=2"]; said {
|
||||
t.Fatalf("the silenced one was said again for the restart: %+v", e)
|
||||
}
|
||||
if len(dt.edits) != edits+1 {
|
||||
t.Fatalf("an edit reached a message that is not silenced: %+v", dt.edits)
|
||||
if len(dt.edits) != edits {
|
||||
t.Fatalf("an edit reached the desk for the restart: %+v", dt.edits)
|
||||
}
|
||||
if e := dt.edits["g14=4"]; !strings.HasPrefix(e.Title, "Resolved: ") {
|
||||
t.Fatalf("the resolved one was changed: %+v", e)
|
||||
@@ -179,8 +177,13 @@ func TestLostAnswersAreJudgedPerMessageAndWaitForTheDesk(t *testing.T) {
|
||||
if len(dt.replaced) != replaced+1 || saidOn(tg) != tgSaid {
|
||||
t.Fatalf("once the desk can send: desk %+v, telegram %+v", dt.replaced[replaced:], tg.sends)
|
||||
}
|
||||
// The router started again, remembering no standing it acted on: the message was said since, so nothing.
|
||||
h.shows = map[string]time.Time{}
|
||||
// The router started again, reading back the pass it made from its state: nothing.
|
||||
h.mu.Lock()
|
||||
h.passes, h.lost = map[string]Pass{}, map[string]time.Time{}
|
||||
if err := h.loadWalks(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
h.mu.Unlock()
|
||||
h.Stand("desktop", Standing{Ready: true, At: c.now(), AnswersLostAt: lost})
|
||||
runFor(h, c, time.Minute)
|
||||
if len(dt.replaced) != replaced+1 {
|
||||
@@ -191,12 +194,12 @@ func TestLostAnswersAreJudgedPerMessageAndWaitForTheDesk(t *testing.T) {
|
||||
lost = c.now()
|
||||
dt.fail = errorf("no session answers")
|
||||
h.Stand("desktop", Standing{Ready: true, At: c.now(), AnswersLostAt: lost})
|
||||
runFor(h, c, 30*time.Second)
|
||||
dt.fail = nil
|
||||
runFor(h, c, time.Minute)
|
||||
if len(dt.replaced) != replaced+1 || saidOn(tg) != tgSaid {
|
||||
t.Fatalf("after a failed send: desk %+v, telegram %+v", dt.replaced[replaced:], tg.sends)
|
||||
}
|
||||
h.Stand("desktop", Standing{Ready: true, At: c.now(), AnswersLostAt: lost})
|
||||
runFor(h, c, time.Minute)
|
||||
if len(dt.replaced) != replaced+2 || saidOn(tg) != tgSaid {
|
||||
t.Fatalf("the failed send was not tried again on the desk: desk %+v, telegram %+v", dt.replaced[replaced:], tg.sends)
|
||||
}
|
||||
|
||||
@@ -113,6 +113,9 @@ func (h *Holder) Status(listening string) Status {
|
||||
"deduplicated by the condition's key; reminded once after 1 h (urgent) or 12 h (warning); cleared by editing the first message",
|
||||
"at most 20 messages an hour per channel, a last line of defence: reached, it is said once and the rest held for one digest",
|
||||
"refused: anything carrying an address, a path or a secret; channel-refused is sent with the words withheld",
|
||||
"while a delivery walks, a message about a machine or module it touches is held, and the walk's end is one message: its outcome, what it reached, how long, what is still open; past the walk's bound (twice its largest delivery budget, 15 min with none) what is held and open is said (novox/hq ADR 0308)",
|
||||
"an ask, a secret given or replaced, and a walk's own failure are never held",
|
||||
"a channel that lost its answers is said again only what is open, asked there and never answered — never a silenced or withheld message — once per restart, and the pass is kept in this module's state",
|
||||
"a silenced condition's message says it is silenced, until when and why, and nothing more is said of it while the silence lasts; a silence is the controller's, asked of the operator or set through the mesh",
|
||||
}
|
||||
anyCan := false
|
||||
@@ -152,9 +155,11 @@ type OpenMessage struct {
|
||||
Reminded bool `json:"reminded"`
|
||||
Unsent string `json:"unsent,omitempty"`
|
||||
Held bool `json:"held,omitempty"`
|
||||
History bool `json:"never_said,omitempty"`
|
||||
Refused string `json:"refused,omitempty"`
|
||||
Count int `json:"times_opened"`
|
||||
// HeldBy are the walks holding it until they end or pass their bound (novox/hq ADR 0308).
|
||||
HeldBy []string `json:"held_by_walks,omitempty"`
|
||||
History bool `json:"never_said,omitempty"`
|
||||
Refused string `json:"refused,omitempty"`
|
||||
Count int `json:"times_opened"`
|
||||
}
|
||||
|
||||
// Open is what is unresolved now, urgent first, oldest first.
|
||||
@@ -180,6 +185,7 @@ func (h *Holder) Open() []OpenMessage {
|
||||
o.SentOn = append(o.SentOn, ch)
|
||||
}
|
||||
sort.Strings(o.SentOn)
|
||||
o.HeldBy = h.heldBy(r.Key, now)
|
||||
if r.silenced(now) {
|
||||
o.Silenced = r.SilencedTill.UTC().Format(time.RFC3339)
|
||||
}
|
||||
|
||||
@@ -103,7 +103,9 @@ func (h *Holder) Sync(open []Condition, asked time.Time) SyncResult {
|
||||
switch {
|
||||
case told || silenced || h.queued(r.Key):
|
||||
case r.Severity == Urgent:
|
||||
urgent = append(urgent, r)
|
||||
if !h.hold(r, "raised") { // an open walk touches it: told at its end or bound (ADR 0308 §2)
|
||||
urgent = append(urgent, r)
|
||||
}
|
||||
case !known && now.Sub(r.Raised) <= FreshFor:
|
||||
fresh = append(fresh, r)
|
||||
case wasQuiet && !needsNothing(r):
|
||||
|
||||
@@ -0,0 +1,867 @@
|
||||
package main
|
||||
|
||||
// While a delivery walks, what it touches is held from the operator and told once, with the walk's result,
|
||||
// when it ends (novox/hq ADR 0308, issue 532).
|
||||
//
|
||||
// **What this reads.** The controller says each walk's start and end on its `plan-moved` event (the plan
|
||||
// whole, ADR 0282 §6), and on those two moments the body carries beside the plan (ADR 0308 §1):
|
||||
//
|
||||
// - `walk`: the plan's id (`id` is read when it is missing);
|
||||
// - `phase`: `started` or `ended`; a plan-moved without it is a move inside a walk and is not read here;
|
||||
// - `touches`: `[{machine, modules}]`, every module the walk moves on every machine it sends it to, and every
|
||||
// module a moved tool runner or node-engine carries there;
|
||||
// - `outcome`: `delivered`, `failed` or `stopped`, on `ended`;
|
||||
// - `bound_seconds`: twice the largest delivery budget among its deliveries, or 900 when none has one (§5),
|
||||
// on `started`; missing, 900 is taken;
|
||||
// - `budget_class`: `core`, `leaf` or `none`;
|
||||
// - `started_at` on the start and `ended_at` on the end (else `at`, else the plan's `updated`): when it
|
||||
// happened. A start older than its bound is history and opens nothing; an end older than FreshFor with
|
||||
// nothing held closes quietly.
|
||||
//
|
||||
// A touch may name `carried` beside `modules` (what a moved tool runner or node-engine carries); the
|
||||
// controller lists those in `modules` too, which is what is read.
|
||||
//
|
||||
// An event of its own named `walk`, with the same body, is read the same way, should the controller say it
|
||||
// there instead.
|
||||
//
|
||||
// **What is held (§2).** While a walk is open and within its bound, a message whose subject is a machine the
|
||||
// walk touches, or a module it touches, is held instead of sent: kept on the walk, with when it was held and
|
||||
// what was to be said, and the channel's history records it with the outcome `held`. A message touched by
|
||||
// several open walks is kept on each, and told when the last of them ends or passes its bound.
|
||||
//
|
||||
// **What never is (§3):** an ask waiting for the operator's answer (an ask's own message, and a condition an
|
||||
// open ask is about), a secret given or replaced, a message about a machine or module no open walk touches,
|
||||
// and a walk's own failure: anything about a plan, a walk, a delivery or a build.
|
||||
//
|
||||
// **The end (§4).** One message: its outcome, the modules and machines it reached and how long it took; then
|
||||
// each held message whose condition is still open, one line each; the ones that cleared meanwhile only
|
||||
// counted. A walk that delivered with nothing held and nothing open is one line, and nothing at all on a kind
|
||||
// the operator set to say only what needs them (the setting `only-what-needs-me`).
|
||||
//
|
||||
// **The bound (§5).** At the walk's start plus its bound, what it holds that is still open is said, with a line
|
||||
// that the walk is still running and since when, and the cleared ones counted; it holds nothing more after
|
||||
// that. A walk whose end never comes is forgotten a day after it started.
|
||||
//
|
||||
// **Kept in the module's state** (the bucket `walks`, one key per walk), so a router that restarts mid-walk
|
||||
// holds and tells as if it had not.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The walk events, by their local names under the controller's seat.
|
||||
const (
|
||||
EventPlanMoved = "plan-moved"
|
||||
EventWalk = "walk"
|
||||
|
||||
WalkStarted = "started"
|
||||
WalkEnded = "ended"
|
||||
|
||||
WalkDelivered = "delivered"
|
||||
WalkFailed = "failed"
|
||||
WalkStopped = "stopped"
|
||||
|
||||
// DefaultWalkBound is a walk's bound when the controller gives none: the bound of a walk none of whose
|
||||
// deliveries has a budget (ADR 0308 §5).
|
||||
DefaultWalkBound = 15 * time.Minute
|
||||
// WalkForget: a walk whose end never came is forgotten this long after it started.
|
||||
WalkForget = 24 * time.Hour
|
||||
// WalkEndWhat is the outbox's name for a walk's end or bound message: it goes alone, never in a digest.
|
||||
WalkEndWhat = "walk-end"
|
||||
// MostNamed: an end message names at most this many modules or machines, then counts the rest.
|
||||
MostNamed = 6
|
||||
)
|
||||
|
||||
// Touch is what a walk moves on one machine.
|
||||
type Touch struct {
|
||||
Machine string `json:"machine"`
|
||||
Modules []string `json:"modules"`
|
||||
}
|
||||
|
||||
// WalkEvent is a walk's start or end as the controller says it.
|
||||
type WalkEvent struct {
|
||||
Walk string
|
||||
Phase string
|
||||
Outcome string
|
||||
Class string
|
||||
Touches []Touch
|
||||
BoundSeconds int
|
||||
At time.Time
|
||||
// StartedAt is the walk's own start, on both events (`started_at`): the bound runs from it, so a router
|
||||
// that hears a start late, or starts again, ends the hold on time.
|
||||
StartedAt time.Time
|
||||
}
|
||||
|
||||
// HeldMessage is one message a walk holds: when it was first held and what was to be said.
|
||||
type HeldMessage struct {
|
||||
At time.Time `json:"at"`
|
||||
What string `json:"what"`
|
||||
}
|
||||
|
||||
// Walk is one open walk as this router keeps it.
|
||||
type Walk struct {
|
||||
ID string `json:"id"`
|
||||
Started time.Time `json:"started"`
|
||||
BoundSeconds int `json:"bound_seconds"`
|
||||
Class string `json:"budget_class,omitempty"`
|
||||
Touches []Touch `json:"touches"`
|
||||
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 {
|
||||
if w.BoundSeconds <= 0 {
|
||||
return DefaultWalkBound
|
||||
}
|
||||
return time.Duration(w.BoundSeconds) * time.Second
|
||||
}
|
||||
|
||||
// holding says whether the walk still holds at a moment: open, and within its bound.
|
||||
func (w *Walk) holding(now time.Time) bool {
|
||||
return !w.BoundSaid && now.Before(w.Started.Add(w.bound()))
|
||||
}
|
||||
|
||||
// touches says whether the walk touches a subject: a machine it sends to, or — when the subject names no
|
||||
// machine of its own — a module it moves anywhere.
|
||||
func (w *Walk) touches(s subject) bool {
|
||||
for _, t := range w.Touches {
|
||||
for _, m := range s.machines {
|
||||
if m != "" && m == t.Machine {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
if s.machineNamed {
|
||||
return false
|
||||
}
|
||||
for _, t := range w.Touches {
|
||||
for _, mod := range t.Modules {
|
||||
for _, m := range s.modules {
|
||||
if m != "" && m == mod {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// walkKeeper is a store that keeps the walks and the restart passes too (the module's state, ADR 0201).
|
||||
type walkKeeper interface {
|
||||
PutWalk(w Walk) error
|
||||
DeleteWalk(id string) error
|
||||
Walks() ([]Walk, error)
|
||||
PutPasses(map[string]Pass) error
|
||||
Passes() (map[string]Pass, error)
|
||||
}
|
||||
|
||||
// DecodeWalk reads a plan-moved or walk event's body: false when it says no walk's start or end.
|
||||
func DecodeWalk(body []byte) (WalkEvent, bool, error) {
|
||||
var raw map[string]json.RawMessage
|
||||
if err := json.Unmarshal(body, &raw); err != nil || raw == nil {
|
||||
return WalkEvent{}, false, fmt.Errorf("a walk event's body is not a JSON object")
|
||||
}
|
||||
str := func(names ...string) string {
|
||||
for _, n := range names {
|
||||
var s string
|
||||
if v, ok := raw[n]; ok && json.Unmarshal(v, &s) == nil && strings.TrimSpace(s) != "" {
|
||||
return strings.TrimSpace(s)
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
var ev WalkEvent
|
||||
ev.Phase = str("phase")
|
||||
if ev.Phase == "" {
|
||||
return WalkEvent{}, false, nil
|
||||
}
|
||||
ev.Walk = str("walk", "id")
|
||||
if ev.Walk == "" {
|
||||
return WalkEvent{}, false, fmt.Errorf("a walk's %s names no walk", ev.Phase)
|
||||
}
|
||||
switch ev.Phase {
|
||||
case WalkStarted, WalkEnded:
|
||||
default:
|
||||
return WalkEvent{}, false, fmt.Errorf("walk %s: phase %q is neither started nor ended", ev.Walk, ev.Phase)
|
||||
}
|
||||
ev.Outcome = str("outcome")
|
||||
if ev.Phase == WalkEnded {
|
||||
switch ev.Outcome {
|
||||
case WalkDelivered, WalkFailed, WalkStopped:
|
||||
default:
|
||||
return WalkEvent{}, false, fmt.Errorf("walk %s ended with outcome %q: neither delivered, failed nor stopped", ev.Walk, ev.Outcome)
|
||||
}
|
||||
}
|
||||
ev.Class = str("budget_class", "budget-class")
|
||||
for _, n := range []string{"bound_seconds", "bound-seconds"} {
|
||||
if v, ok := raw[n]; ok && string(v) != "null" {
|
||||
var f float64
|
||||
if json.Unmarshal(v, &f) != nil || f < 0 {
|
||||
return WalkEvent{}, false, fmt.Errorf("walk %s: %s is not a number of seconds", ev.Walk, n)
|
||||
}
|
||||
ev.BoundSeconds = int(f)
|
||||
break
|
||||
}
|
||||
}
|
||||
if v, ok := raw["touches"]; ok && string(v) != "null" {
|
||||
var list []struct {
|
||||
Machine string `json:"machine"`
|
||||
Node string `json:"node"`
|
||||
Modules []string `json:"modules"`
|
||||
}
|
||||
if json.Unmarshal(v, &list) != nil {
|
||||
return WalkEvent{}, false, fmt.Errorf("walk %s: touches is not a list of {machine, modules}", ev.Walk)
|
||||
}
|
||||
for _, t := range list {
|
||||
m := t.Machine
|
||||
if m == "" {
|
||||
m = t.Node
|
||||
}
|
||||
if m != "" {
|
||||
ev.Touches = append(ev.Touches, Touch{Machine: m, Modules: t.Modules})
|
||||
}
|
||||
}
|
||||
}
|
||||
if s := str("started_at"); s != "" {
|
||||
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
|
||||
ev.StartedAt = t
|
||||
}
|
||||
}
|
||||
when := []string{"started_at", "at", "updated"}
|
||||
if ev.Phase == WalkEnded {
|
||||
when = []string{"ended_at", "at", "updated"}
|
||||
}
|
||||
for _, n := range when {
|
||||
if s := str(n); s != "" {
|
||||
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
|
||||
ev.At = t
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return ev, true, nil
|
||||
}
|
||||
|
||||
// mergeTouches is what two sayings of a walk's touches name together.
|
||||
func mergeTouches(a, b []Touch) []Touch {
|
||||
by := map[string]map[string]bool{}
|
||||
var order []string
|
||||
for _, list := range [][]Touch{a, b} {
|
||||
for _, t := range list {
|
||||
if by[t.Machine] == nil {
|
||||
by[t.Machine] = map[string]bool{}
|
||||
order = append(order, t.Machine)
|
||||
}
|
||||
for _, m := range t.Modules {
|
||||
by[t.Machine][m] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Strings(order)
|
||||
var out []Touch
|
||||
for _, machine := range order {
|
||||
var mods []string
|
||||
for m := range by[machine] {
|
||||
mods = append(mods, m)
|
||||
}
|
||||
sort.Strings(mods)
|
||||
out = append(out, Touch{Machine: machine, Modules: mods})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Walked takes a walk's start or end.
|
||||
func (h *Holder) Walked(ev WalkEvent) {
|
||||
h.work.Lock()
|
||||
defer h.work.Unlock()
|
||||
h.mu.Lock()
|
||||
h.init()
|
||||
now := h.Now()
|
||||
h.mu.Unlock()
|
||||
at := ev.At
|
||||
if at.IsZero() || at.After(now) {
|
||||
at = now
|
||||
}
|
||||
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())
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
if known := h.walks[ev.Walk]; known != nil {
|
||||
w = known // said again: the same walk, its touches as both sayings name them
|
||||
} else {
|
||||
w.Held = map[string]HeldMessage{}
|
||||
h.walks[ev.Walk] = w
|
||||
}
|
||||
w.Touches = mergeTouches(w.Touches, ev.Touches)
|
||||
h.mu.Unlock()
|
||||
h.keepWalk(w)
|
||||
h.record(Sent{At: now, Channel: "-", Key: "walk." + ev.Walk, What: "started: " + touchWords(w.Touches),
|
||||
Outcome: "holding what it touches for at most " + roughly(w.bound())})
|
||||
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)
|
||||
if !ev.StartedAt.IsZero() {
|
||||
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.endWalk(w, ev.Outcome, at, now)
|
||||
}
|
||||
}
|
||||
|
||||
// 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()
|
||||
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) {
|
||||
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
|
||||
}
|
||||
switch {
|
||||
case r == nil || !r.Cleared.IsZero():
|
||||
cleared++
|
||||
case r.silenced(now) || lastSaid(r).After(hm.At):
|
||||
default:
|
||||
open = append(open, r)
|
||||
whats[key] = hm.What
|
||||
}
|
||||
}
|
||||
sort.Slice(open, func(i, j int) bool {
|
||||
if !open[i].Raised.Equal(open[j].Raised) {
|
||||
return open[i].Raised.Before(open[j].Raised)
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
// endWalk tells a walk's end once (ADR 0308 §4).
|
||||
func (h *Holder) endWalk(w *Walk, outcome string, ended, now time.Time) {
|
||||
open, whats, cleared := h.release(w, now)
|
||||
if now.Sub(ended) > FreshFor && len(open) == 0 {
|
||||
h.Logf("[messenger] walk %s %s %s ago, heard late: history, nothing said (%d held cleared meanwhile)",
|
||||
w.ID, outcome, roughly(now.Sub(ended)), cleared)
|
||||
h.record(Sent{At: now, Channel: "-", Key: "walk." + w.ID, What: "ended " + outcome, Outcome: "quiet: heard late"})
|
||||
return
|
||||
}
|
||||
took := ended.Sub(w.Started)
|
||||
head := map[string]string{WalkDelivered: "Delivered", WalkFailed: "Delivery failed", WalkStopped: "Delivery stopped"}[outcome]
|
||||
if head == "" {
|
||||
head = "Delivery ended"
|
||||
}
|
||||
reached := touchWords(w.Touches)
|
||||
oneLine := outcome == WalkDelivered && len(w.Held) == 0 && len(open) == 0
|
||||
var m Message
|
||||
if oneLine {
|
||||
m = Message{Title: head + ": " + reached + ", in " + roughly(took), Quiet: true}
|
||||
} else {
|
||||
lines := []string{"It took " + roughly(took) + "."}
|
||||
lines = append(lines, h.heldLines(open, whats, cleared)...)
|
||||
m = Message{Title: head + ": " + reached, Body: strings.Join(lines, "\n")}
|
||||
}
|
||||
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.
|
||||
func (h *Holder) heldLines(open []*Record, whats map[string]string, cleared int) []string {
|
||||
var lines []string
|
||||
if len(open) > 0 {
|
||||
lines = append(lines, "Still open after it:")
|
||||
for _, r := range open {
|
||||
what := whats[r.Key]
|
||||
if what == "" {
|
||||
what = "raised"
|
||||
}
|
||||
lines = append(lines, "- "+h.say(r, what).Title)
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case cleared == 1:
|
||||
lines = append(lines, "1 problem it raised resolved during it.")
|
||||
case cleared > 1:
|
||||
lines = append(lines, fmt.Sprintf("%d problems it raised resolved during it.", cleared))
|
||||
}
|
||||
return lines
|
||||
}
|
||||
|
||||
// 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 {
|
||||
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 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 _, 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()
|
||||
w.BoundSaid = true
|
||||
h.mu.Unlock()
|
||||
open, whats, cleared := h.release(w, now)
|
||||
h.mu.Lock()
|
||||
w.Held = map[string]HeldMessage{}
|
||||
h.mu.Unlock()
|
||||
h.keepWalk(w)
|
||||
if len(open) == 0 && cleared == 0 {
|
||||
h.record(Sent{At: now, Channel: "-", Key: "walk." + w.ID, What: "past its bound", Outcome: "nothing held"})
|
||||
continue
|
||||
}
|
||||
since := w.Started.In(h.zone()).Format("15:04")
|
||||
lines := []string{"The walk " + w.ID + " is still running, since " + since + "."}
|
||||
if _, ok := Check(lines[0]); !ok {
|
||||
lines[0] = "The walk is still running, since " + since + "."
|
||||
}
|
||||
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, 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. 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 {
|
||||
for _, ch := range h.channels() {
|
||||
if ch.Ready() == nil {
|
||||
to = append(to, ch)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
switch {
|
||||
case h.Desk != nil && h.Desk.Ready() == nil:
|
||||
to = []Channel{h.Desk}
|
||||
case h.Away != nil && h.Away.Ready() == nil:
|
||||
to = []Channel{h.Away}
|
||||
}
|
||||
}
|
||||
var kept []Channel
|
||||
for _, ch := range to {
|
||||
if oneLine && h.NeedsOnly != nil && h.NeedsOnly(ch.Name()) {
|
||||
h.record(Sent{At: now, Channel: ch.Name(), Key: "walk." + w.ID, What: what,
|
||||
Outcome: "quiet: this kind says only what needs the operator"})
|
||||
continue
|
||||
}
|
||||
kept = append(kept, ch)
|
||||
}
|
||||
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 {
|
||||
sev = Urgent
|
||||
}
|
||||
for _, ch := range kept {
|
||||
msg := m
|
||||
h.enqueue(ch.Name(), item{Key: "walk." + w.ID, What: WalkEndWhat, Severity: sev, At: now, Msg: &msg})
|
||||
}
|
||||
h.mu.Lock()
|
||||
for _, r := range open {
|
||||
if len(kept) == 0 {
|
||||
continue
|
||||
}
|
||||
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.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.
|
||||
func touchWords(touches []Touch) string {
|
||||
mods := map[string]bool{}
|
||||
var machines []string
|
||||
for _, t := range touches {
|
||||
machines = append(machines, t.Machine)
|
||||
for _, m := range t.Modules {
|
||||
mods[m] = true
|
||||
}
|
||||
}
|
||||
var names []string
|
||||
for m := range mods {
|
||||
names = append(names, m)
|
||||
}
|
||||
sort.Strings(names)
|
||||
sort.Strings(machines)
|
||||
if len(names) == 0 {
|
||||
return "nothing moved"
|
||||
}
|
||||
if len(machines) == 0 {
|
||||
return listed(names)
|
||||
}
|
||||
return listed(names) + " on " + listed(machines)
|
||||
}
|
||||
|
||||
// listed is at most MostNamed names, then how many more.
|
||||
func listed(names []string) string {
|
||||
if len(names) <= MostNamed {
|
||||
return strings.Join(names, ", ")
|
||||
}
|
||||
return strings.Join(names[:MostNamed], ", ") + fmt.Sprintf(" and %d more", len(names)-MostNamed)
|
||||
}
|
||||
|
||||
// subject is what a message is about, as a walk's touches are matched against it.
|
||||
type subject struct {
|
||||
machines, modules []string
|
||||
// machineNamed: the subject names its machine, so a module it names is matched on that machine alone.
|
||||
machineNamed bool
|
||||
// tokens are every name it carries, to tell a message about a walk itself.
|
||||
tokens []string
|
||||
}
|
||||
|
||||
// subjectOf is a record's subject: a condition's scope and id (`<scope>.<id>.<kind>` when the record kept no
|
||||
// subject of its own), a notice's subject words.
|
||||
func subjectOf(r *Record) subject {
|
||||
scope, id := r.Scope, r.SubjectID
|
||||
parts := strings.Split(r.Key, ".")
|
||||
if scope == "" && len(parts) > 0 {
|
||||
scope = parts[0]
|
||||
}
|
||||
if id == "" && len(parts) >= 3 {
|
||||
id = strings.Join(parts[1:len(parts)-1], ".")
|
||||
}
|
||||
var s subject
|
||||
tokens := strings.Split(id, ".")
|
||||
if r.Origin == "notify" {
|
||||
tokens = append(strings.FieldsFunc(r.Subject, func(c rune) bool { return c == ' ' || c == ',' || c == '/' }), parts[1:]...)
|
||||
scope = "notify"
|
||||
}
|
||||
s.tokens = tokens
|
||||
if r.Machine != "" {
|
||||
s.machines, s.machineNamed = append(s.machines, r.Machine), true
|
||||
}
|
||||
switch scope {
|
||||
case "machine", "node", "bus":
|
||||
s.machines, s.machineNamed = append(s.machines, tokens[0]), true
|
||||
if len(tokens) > 1 {
|
||||
s.modules = append(s.modules, tokens[1])
|
||||
}
|
||||
case "module", "core", "seat":
|
||||
s.modules = append(s.modules, tokens[0])
|
||||
if len(tokens) > 1 {
|
||||
s.machines, s.machineNamed = append(s.machines, tokens[1]), true
|
||||
}
|
||||
default:
|
||||
// Its names, each matched as a machine and as a module.
|
||||
s.machines = append(s.machines, tokens...)
|
||||
s.modules = append(s.modules, tokens...)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// neverHeldScopes are the subjects of a walk's own failure (ADR 0308 §3): a plan, a walk, a delivery, a build.
|
||||
var neverHeldScopes = map[string]bool{"plan": true, "walk": true, "delivery": true, "build": true, "release": true}
|
||||
|
||||
// neverHeldKinds are what is sent at once whatever a walk touches: a secret given or replaced (ADR 0308 §3).
|
||||
var neverHeldKinds = map[string]bool{"secret-given": true, "secret-replaced": true}
|
||||
|
||||
// neverHeld says whether a record is one ADR 0308 §3 sends at once.
|
||||
func (h *Holder) neverHeld(r *Record) bool {
|
||||
if r.Origin == "ask" || h.openAsk(r) != nil {
|
||||
return true
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
parts := strings.Split(r.Key, ".")
|
||||
kind := r.Kind
|
||||
if kind == "" || kind == "notice" {
|
||||
kind = parts[len(parts)-1]
|
||||
}
|
||||
scope := r.Scope
|
||||
if scope == "" {
|
||||
scope = parts[0]
|
||||
}
|
||||
if neverHeldKinds[kind] || neverHeldScopes[scope] {
|
||||
return true
|
||||
}
|
||||
for _, t := range subjectOf(r).tokens {
|
||||
if h.walks[t] != nil {
|
||||
return true // about a walk itself
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// hold holds a record's message when an open walk touches its subject (ADR 0308 §2): kept on every such walk,
|
||||
// with when it was first held and what was to be said, and said held in the channel's history.
|
||||
func (h *Holder) hold(r *Record, what string) bool {
|
||||
switch what {
|
||||
case "cleared", "silenced", "asked":
|
||||
return false
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.init()
|
||||
walking := len(h.walks) > 0
|
||||
h.mu.Unlock()
|
||||
if !walking || h.neverHeld(r) {
|
||||
return false
|
||||
}
|
||||
now := h.Now()
|
||||
h.mu.Lock()
|
||||
s := subjectOf(r)
|
||||
var by []*Walk
|
||||
var ids []string
|
||||
fresh := false
|
||||
for _, w := range h.walks {
|
||||
if !w.holding(now) || !w.touches(s) {
|
||||
continue
|
||||
}
|
||||
prev, had := w.Held[r.Key]
|
||||
hm := HeldMessage{At: now, What: what}
|
||||
if had {
|
||||
hm.At = prev.At
|
||||
if !r.told() && prev.What == "raised" {
|
||||
hm.What = prev.What // never said yet: it is a new problem to the operator, whatever came since
|
||||
}
|
||||
}
|
||||
if !had || hm.What != prev.What {
|
||||
fresh = true
|
||||
}
|
||||
if w.Held == nil {
|
||||
w.Held = map[string]HeldMessage{}
|
||||
}
|
||||
w.Held[r.Key] = hm
|
||||
by = append(by, w)
|
||||
ids = append(ids, w.ID)
|
||||
}
|
||||
h.mu.Unlock()
|
||||
if len(by) == 0 {
|
||||
return false
|
||||
}
|
||||
sort.Strings(ids)
|
||||
for _, w := range by {
|
||||
h.keepWalk(w)
|
||||
}
|
||||
if fresh {
|
||||
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: what + ", while walk " + strings.Join(ids, ", ") + " touches it",
|
||||
Outcome: "held"})
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// heldBy is the walks holding a record now.
|
||||
func (h *Holder) heldBy(key string, now time.Time) []string {
|
||||
var ids []string
|
||||
for _, w := range h.walks {
|
||||
if _, ok := w.Held[key]; ok && w.holding(now) {
|
||||
ids = append(ids, w.ID)
|
||||
}
|
||||
}
|
||||
sort.Strings(ids)
|
||||
return ids
|
||||
}
|
||||
|
||||
func (h *Holder) keepWalk(w *Walk) {
|
||||
k, ok := h.Store.(walkKeeper)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
c := *w
|
||||
c.Held = map[string]HeldMessage{}
|
||||
for key, v := range w.Held {
|
||||
c.Held[key] = v
|
||||
}
|
||||
h.mu.Unlock()
|
||||
if err := k.PutWalk(c); err != nil {
|
||||
h.noteStore(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Holder) forgetWalk(id string) {
|
||||
if k, ok := h.Store.(walkKeeper); ok {
|
||||
if err := k.DeleteWalk(id); err != nil {
|
||||
h.noteStore(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// loadWalks reads back the open walks and the restart passes; called with h.mu held.
|
||||
func (h *Holder) loadWalks() error {
|
||||
k, ok := h.Store.(walkKeeper)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
walks, err := k.Walks()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
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{}
|
||||
}
|
||||
h.walks[w.ID] = &w
|
||||
}
|
||||
passes, err := k.Passes()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for kind, p := range passes {
|
||||
h.passes[kind] = p
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,622 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The tests of novox/hq ADR 0308 "How it is checked" §2–§6, with the controller's walk events given by a fake
|
||||
// source: the body the controller's plan-moved carries at a walk's start and end (ADR 0308 §1).
|
||||
|
||||
// walkBody is a plan-moved body as the controller says a walk's start or end (mesh-controller pull request 247).
|
||||
func walkBody(t *testing.T, id, phase, outcome string, at time.Time, bound int, touches ...Touch) []byte {
|
||||
t.Helper()
|
||||
b := map[string]any{"id": id, "state": "walking", "walk": id, "phase": phase, "touches": touches,
|
||||
"updated": at.Format(time.RFC3339Nano), "budget_class": "leaf"}
|
||||
if phase == WalkStarted {
|
||||
b["started_at"] = at.Format(time.RFC3339Nano)
|
||||
} else {
|
||||
b["ended_at"] = at.Format(time.RFC3339Nano)
|
||||
}
|
||||
if outcome != "" {
|
||||
b["outcome"] = outcome
|
||||
}
|
||||
if bound > 0 {
|
||||
b["bound_seconds"] = bound
|
||||
}
|
||||
raw, err := json.Marshal(b)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// walk says a walk's start or end to the router, through the decoder the bus's events go through.
|
||||
func walk(t *testing.T, h *Holder, id, phase, outcome string, at time.Time, bound int, touches ...Touch) {
|
||||
t.Helper()
|
||||
ev, ok, err := DecodeWalk(walkBody(t, id, phase, outcome, at, bound, touches...))
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("the walk event was not read: %v %v", ok, err)
|
||||
}
|
||||
h.Walked(ev)
|
||||
}
|
||||
|
||||
func novox(modules ...string) Touch { return Touch{Machine: "novox", Modules: modules} }
|
||||
|
||||
// heard says whether any message on the channels carries the words: sent, edited or said again in place.
|
||||
func heard(words string, chs ...*fakeChannel) bool {
|
||||
for _, ch := range chs {
|
||||
for _, m := range ch.sends {
|
||||
if strings.Contains(m.Text(), words) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
for _, m := range ch.edits {
|
||||
if strings.Contains(m.Text(), words) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
for _, r := range ch.replaced {
|
||||
if strings.Contains(r.m.Text(), words) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// outcomes are the history's outcomes for a key, oldest first.
|
||||
func outcomes(h *Holder, key string) []string {
|
||||
var out []string
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for _, s := range h.recent {
|
||||
if s.Key == key {
|
||||
out = append(out, s.Outcome)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func has(list []string, want string) bool {
|
||||
for _, s := range list {
|
||||
if s == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// §2: during a walk, a message about a machine or a module it touches is held, not sent, and the channel's
|
||||
// history says it was held — a module's condition, a machine's, and a module's notice alike.
|
||||
func TestDuringAWalkWhatItTouchesIsHeldAndTheHistorySaysSo(t *testing.T) {
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
walk(t, h, "plan-1", WalkStarted, "", c.now(), 600, novox("mesh-tools", "node-tools"))
|
||||
mod := cond("module.node-tools.novox.unhealthy", Urgent, "the tool runner on novox does not answer")
|
||||
mod.Machine, mod.Raised = "novox", c.now()
|
||||
h.Condition(EventRaised, mod)
|
||||
machine := cond("machine.novox.claude-code.agent-home.data-shrank", Urgent, "the agent's home shrank")
|
||||
machine.Raised = c.now()
|
||||
h.Condition(EventRaised, machine)
|
||||
if _, err := h.Notify("tools.slow", Urgent, "tools answer slowly", "node-tools", false, "tools slow",
|
||||
"Needs you: look at the tools."); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
runFor(h, c, 2*time.Minute)
|
||||
if saidOn(tg, dt) != 0 {
|
||||
t.Fatalf("held messages were said: telegram %+v, desk %+v", tg.sends, dt.sends)
|
||||
}
|
||||
for _, key := range []string{mod.Key, machine.Key, "notify.tools.slow"} {
|
||||
if !has(outcomes(h, key), "held") {
|
||||
t.Errorf("the history does not say %s was held: %v", key, outcomes(h, key))
|
||||
}
|
||||
}
|
||||
if len(h.Open()) != 3 {
|
||||
t.Fatalf("held messages are not kept open: %+v", h.Open())
|
||||
}
|
||||
}
|
||||
|
||||
// §2: a message touched by two overlapping walks is held until the second of them ends; the first's end says
|
||||
// nothing of it.
|
||||
func TestAMessageTouchedByTwoOverlappingWalksIsHeldUntilTheSecondEnds(t *testing.T) {
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
walk(t, h, "plan-a", WalkStarted, "", c.now(), 1200, novox("mesh-tools"))
|
||||
c.pass(time.Minute)
|
||||
walk(t, h, "plan-b", WalkStarted, "", c.now(), 1200, novox("node-tools"), Touch{Machine: "ace", Modules: []string{"node-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-a", WalkEnded, WalkDelivered, c.now(), 0, novox("mesh-tools"))
|
||||
runFor(h, c, time.Minute)
|
||||
if heard(headlineOf(k.Key), tg, dt) {
|
||||
t.Fatalf("said at the first walk's end, while the second still holds it: %+v %+v", tg.sends, dt.sends)
|
||||
}
|
||||
if !heard("Delivered: mesh-tools on novox", tg, dt) {
|
||||
t.Fatalf("the first walk's end was not told: %+v", dt.sends)
|
||||
}
|
||||
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: telegram %+v, desk %+v", tg.sends, dt.sends)
|
||||
}
|
||||
}
|
||||
|
||||
// §3: an ask, a secret given, a message about a machine no open walk touches and a walk's own failure are
|
||||
// sent at once, while a walk touches the machine they are on.
|
||||
func TestWhatAWalkNeverHoldsIsSentAtOnce(t *testing.T) {
|
||||
h, tg, dt, c, _, _ := newRouter(t)
|
||||
walk(t, h, "plan-9", WalkStarted, "", c.now(), 1200, novox("telegram", "mesh-tools"))
|
||||
for _, key := range []string{
|
||||
"machine.novox.telegram.telegram-token.secret-given", // a secret given, on a touched machine
|
||||
"machine.novox.mounts.smb.secret-replaced", // a secret replaced
|
||||
"machine.ace.silent", // a machine no walk touches
|
||||
"delivery.leaf.over-budget", // the walk's own delivery
|
||||
"plan.plan-9.failed", // the walk itself
|
||||
} {
|
||||
k := cond(key, Urgent, "it happened")
|
||||
k.Raised = c.now()
|
||||
if !strings.HasPrefix(key, "machine.") {
|
||||
k.Machine = "novox" // a walk's own failure, on a machine it touches
|
||||
}
|
||||
h.Condition(EventRaised, k)
|
||||
runFor(h, c, time.Minute)
|
||||
stand(h, c)
|
||||
if !heard(headlineOf(key), tg, dt) {
|
||||
t.Errorf("%s was not sent at once during the walk: %v", key, outcomes(h, key))
|
||||
}
|
||||
}
|
||||
// An ask's own message, and an ask about a held condition: both offered at once.
|
||||
submit(t, h, ConditionOwner, ackAsk(c, "own"))
|
||||
runFor(h, c, time.Minute)
|
||||
if !has(outcomes(h, "ask."+ConditionOwner+".own"), "sent") {
|
||||
t.Errorf("an ask waiting for the operator was held: %v", outcomes(h, "ask."+ConditionOwner+".own"))
|
||||
}
|
||||
held := cond("machine.novox.units", Urgent, "2 failed services on novox")
|
||||
held.Raised = c.now()
|
||||
h.Condition(EventRaised, held)
|
||||
runFor(h, c, time.Minute)
|
||||
if heard(headlineOf(held.Key), tg, dt) {
|
||||
t.Fatalf("a condition about a touched machine was not held")
|
||||
}
|
||||
about := ackAsk(c, "about")
|
||||
about.About = held.Key
|
||||
submit(t, h, ConditionOwner, about)
|
||||
runFor(h, c, time.Minute)
|
||||
if !heard(headlineOf(held.Key), dt) {
|
||||
t.Fatalf("the ask about a held condition was held: %v", outcomes(h, held.Key))
|
||||
}
|
||||
}
|
||||
|
||||
// §4: a walk's end is one message — its outcome, what it reached and how long it took — then each held message
|
||||
// still open, one line each; the one that cleared is only counted, and nothing more is said of them after.
|
||||
func TestAWalksEndIsToldOnceWithWhatIsStillOpen(t *testing.T) {
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
walk(t, h, "plan-3", WalkStarted, "", c.now(), 600, novox("mesh-tools"))
|
||||
still := cond("machine.novox.tools-silent", Warning, "the tools on novox are silent")
|
||||
still.Raised = c.now()
|
||||
h.Condition(EventRaised, still)
|
||||
gone := cond("module.mesh-tools.novox.unhealthy", Warning, "mesh-tools is unhealthy on novox")
|
||||
gone.Raised = c.now()
|
||||
h.Condition(EventRaised, gone)
|
||||
c.pass(time.Second)
|
||||
other := cond("machine.novox.core-behind", Urgent, "novox runs an older core")
|
||||
other.Raised = c.now()
|
||||
h.Condition(EventRaised, other)
|
||||
runFor(h, c, 2*time.Minute)
|
||||
h.Condition(EventCleared, Condition{Key: gone.Key, At: c.now()})
|
||||
runFor(h, c, 5*time.Minute)
|
||||
if saidOn(tg, dt) != 0 {
|
||||
t.Fatalf("said during the walk: %+v %+v", tg.sends, dt.sends)
|
||||
}
|
||||
walk(t, h, "plan-3", WalkEnded, WalkFailed, c.now(), 0)
|
||||
runFor(h, c, time.Minute)
|
||||
// Urgent among what it holds: on every kind, one message each.
|
||||
if len(dt.sends) != 1 || len(tg.sends) != 1 {
|
||||
t.Fatalf("the end was not one message per kind: desk %+v, telegram %+v", dt.sends, tg.sends)
|
||||
}
|
||||
m := dt.sends[0]
|
||||
if m.Title != "Delivery failed: mesh-tools on novox" || !strings.Contains(m.Body, "It took 7 min.") {
|
||||
t.Fatalf("the end does not say its outcome, what it reached and how long: %q", m.Text())
|
||||
}
|
||||
want := "Still open after it:\n- " + headlineOf(still.Key) + "\n- Urgent: " + headlineOf(other.Key) +
|
||||
"\n1 problem it raised resolved during it."
|
||||
if !strings.Contains(m.Body, want) || strings.Contains(m.Text(), headlineOf(gone.Key)) {
|
||||
t.Fatalf("the end message %q does not say %q", m.Body, want)
|
||||
}
|
||||
keyless(t, m)
|
||||
runFor(h, c, 20*time.Minute)
|
||||
if len(dt.sends) != 1 || len(tg.sends) != 1 || len(dt.replaced)+len(tg.replaced) != 0 {
|
||||
t.Fatalf("what the end told was said again: desk %+v, telegram %+v", dt.sends, tg.sends)
|
||||
}
|
||||
}
|
||||
|
||||
// §4: a walk that delivered with nothing held and nothing open is one line, and nothing at all on a kind set
|
||||
// to say only what needs the operator.
|
||||
func TestADeliveredWalkWithNothingHeldIsOneLineOrNothing(t *testing.T) {
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
walk(t, h, "plan-4", WalkStarted, "", c.now(), 600, novox("mesh-tools"), Touch{Machine: "ace", Modules: []string{"mesh-tools"}})
|
||||
runFor(h, c, 3*time.Minute)
|
||||
walk(t, h, "plan-4", WalkEnded, WalkDelivered, c.now(), 0)
|
||||
runFor(h, c, time.Minute)
|
||||
if len(dt.sends) != 1 || len(tg.sends) != 0 {
|
||||
t.Fatalf("desk %+v, telegram %+v", dt.sends, tg.sends)
|
||||
}
|
||||
if m := dt.sends[0]; m.Title != "Delivered: mesh-tools on ace, novox, in 3 min" || m.Body != "" {
|
||||
t.Fatalf("not one line: %q", m.Text())
|
||||
}
|
||||
h.NeedsOnly = func(kind string) bool { return kind == "desktop" }
|
||||
walk(t, h, "plan-5", WalkStarted, "", c.now(), 600, novox("mesh-tools"))
|
||||
runFor(h, c, 3*time.Minute)
|
||||
walk(t, h, "plan-5", WalkEnded, WalkDelivered, c.now(), 0)
|
||||
runFor(h, c, time.Minute)
|
||||
if len(dt.sends) != 1 || len(tg.sends) != 0 {
|
||||
t.Fatalf("told on a kind that says only what needs the operator: desk %+v, telegram %+v", dt.sends, tg.sends)
|
||||
}
|
||||
if !has(outcomes(h, "walk.plan-5"), "quiet: this kind says only what needs the operator") {
|
||||
t.Fatalf("the history does not say why nothing was told: %v", outcomes(h, "walk.plan-5"))
|
||||
}
|
||||
}
|
||||
|
||||
// §5: the bound is the controller's (bound_seconds: twice the largest budget of the walk's deliveries; none
|
||||
// given, 15 minutes). At it, what is held and still open is sent with the still-running line, and what
|
||||
// cleared is counted; before it, nothing; after it, the walk holds nothing more.
|
||||
func TestAHoldEndsAtTheWalksBound(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
given int
|
||||
bound time.Duration
|
||||
}{
|
||||
{"core: twice ten minutes", 1200, 20 * time.Minute},
|
||||
{"leaf at five minutes", 600, 10 * time.Minute},
|
||||
{"leaf at three minutes", 360, 6 * time.Minute},
|
||||
{"no budget", 0, 15 * time.Minute},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
started := c.now()
|
||||
walk(t, h, "plan-7", WalkStarted, "", started, tc.given, novox("node-tools"))
|
||||
open := cond("machine.novox.tools-silent", Urgent, "the tools on novox are silent")
|
||||
open.Raised = c.now()
|
||||
h.Condition(EventRaised, open)
|
||||
gone := cond("module.node-tools.novox.unhealthy", Urgent, "node-tools is unhealthy")
|
||||
gone.Raised = c.now()
|
||||
h.Condition(EventRaised, gone)
|
||||
c.pass(30 * time.Second)
|
||||
h.Condition(EventCleared, Condition{Key: gone.Key, At: c.now()})
|
||||
runFor(h, c, tc.bound-time.Minute-30*time.Second)
|
||||
if saidOn(tg, dt) != 0 {
|
||||
t.Fatalf("said before the bound: %+v %+v", tg.sends, dt.sends)
|
||||
}
|
||||
runFor(h, c, 2*time.Minute)
|
||||
if len(dt.sends) != 1 || len(tg.sends) != 1 {
|
||||
t.Fatalf("at the bound: desk %+v, telegram %+v", dt.sends, tg.sends)
|
||||
}
|
||||
m := dt.sends[0]
|
||||
since := started.In(h.zone()).Format("15:04")
|
||||
for _, want := range []string{"is still running, since " + since + ".", "- Urgent: " + headlineOf(open.Key),
|
||||
"1 problem it raised resolved during it."} {
|
||||
if !strings.Contains(m.Body, want) {
|
||||
t.Fatalf("the bound's message %q does not say %q", m.Text(), want)
|
||||
}
|
||||
}
|
||||
// Past its bound the walk holds nothing: a new message about what it touches goes at once.
|
||||
late := cond("machine.novox.disk", Urgent, "a disk fails on novox")
|
||||
late.Raised = c.now()
|
||||
h.Condition(EventRaised, late)
|
||||
runFor(h, c, time.Minute)
|
||||
if !heard(headlineOf(late.Key), dt) {
|
||||
t.Fatalf("held past the bound: %v", outcomes(h, late.Key))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// §2, kept: what a walk holds survives the router's restart, and is told at the walk's end.
|
||||
func TestHeldMessagesSurviveARouterRestart(t *testing.T) {
|
||||
h, tg, dt, c, st := newHolder(t)
|
||||
walk(t, h, "plan-8", 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)
|
||||
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()
|
||||
c.pass(2 * time.Minute)
|
||||
walk(t, again, "plan-8", WalkEnded, WalkDelivered, c.now(), 0)
|
||||
runFor(again, c, time.Minute)
|
||||
if !heard("- Urgent: "+headlineOf(k.Key), dt) || saidOn(tg, dt) != 2 {
|
||||
t.Fatalf("what was held before the restart was not told at the walk's end: desk %+v, telegram %+v", dt.sends, tg.sends)
|
||||
}
|
||||
}
|
||||
|
||||
// §6, issue 531: the desk's notifiers on two machines start again a second apart, with four silenced messages
|
||||
// shown there — one of them withheld for its content, a file path — and a fifth open, asked there, and
|
||||
// withheld. Nothing is said again: no silenced, no withheld message; the pass runs once; and a router that
|
||||
// starts again reads it back from its state and makes it no more.
|
||||
func TestARestartSaysNothingItHasSaid(t *testing.T) {
|
||||
h, tg, dt, c, _, _ := newRouter(t)
|
||||
h.Start()
|
||||
c.pass(2 * time.Minute)
|
||||
keys := []string{"machine.novox.telegram.telegram-token.secret-given", "machine.novox.claude-code.agent-home.data-shrank",
|
||||
"machine.shanks.mounts.smb-password-recalbox.secret-given", "machine.ace.nzbgeek.api-key.secret-given"}
|
||||
var silenced []Condition
|
||||
for i, key := range keys {
|
||||
k := cond(key, Urgent, "it happened")
|
||||
if i == 1 {
|
||||
k.Explanation = "Needs you: look at /var/lib/agent-home, which shrank."
|
||||
}
|
||||
k.Raised = c.now()
|
||||
h.Condition(EventRaised, k)
|
||||
runFor(h, c, time.Minute)
|
||||
stand(h, c)
|
||||
silenced = append(silenced, k)
|
||||
}
|
||||
withheldOpen := cond("machine.novox.units", Urgent, "2 failed services")
|
||||
withheldOpen.Explanation = "Needs you: read /etc/systemd/system/x.service."
|
||||
withheldOpen.Raised = c.now()
|
||||
h.Condition(EventRaised, withheldOpen)
|
||||
runFor(h, c, time.Minute)
|
||||
ask := ackAsk(c, "units")
|
||||
ask.About = withheldOpen.Key
|
||||
submit(t, h, ConditionOwner, ask)
|
||||
runFor(h, c, time.Minute)
|
||||
for _, k := range silenced {
|
||||
k.SilencedTill, k.SilencedWhy, k.At = c.now().Add(7*24*time.Hour), "known", c.now()
|
||||
h.Condition(EventChanged, k)
|
||||
}
|
||||
runFor(h, c, time.Minute)
|
||||
for _, key := range []string{silenced[1].Key, withheldOpen.Key} {
|
||||
if r := h.open[key]; r == nil || r.Refused == "" {
|
||||
t.Fatalf("the test's %s was not withheld for its content: %+v", key, r)
|
||||
}
|
||||
}
|
||||
sends, edits, replaced, tgSaid := len(dt.sends), len(dt.edits), len(dt.replaced), saidOn(tg)
|
||||
dt.edits = map[string]Message{}
|
||||
passes := func(h *Holder) int { return len(outcomes(h, "(answers lost)")) }
|
||||
// The notifiers on two machines started again a second apart: two standings, two times.
|
||||
c.pass(time.Minute)
|
||||
first := c.now()
|
||||
h.Stand("desktop", Standing{Ready: true, At: c.now(), AnswersLostAt: first})
|
||||
c.pass(time.Second)
|
||||
h.Flush() // the runtime's tick, between the two standings
|
||||
second := c.now()
|
||||
h.Stand("desktop", Standing{Ready: true, At: c.now(), AnswersLostAt: second})
|
||||
runFor(h, c, time.Minute)
|
||||
if len(dt.sends) != sends || len(dt.edits) != 0 || len(dt.replaced) != replaced || saidOn(tg) != tgSaid {
|
||||
t.Fatalf("the restart said again: sends %+v, edits %+v (was %d), replaced %+v", dt.sends[sends:], dt.edits, edits,
|
||||
dt.replaced[replaced:])
|
||||
}
|
||||
if passes(h) != 1 {
|
||||
t.Fatalf("the pass ran %d times for one restart", passes(h))
|
||||
}
|
||||
// The router starts again; the desk repeats its standing.
|
||||
again := &Holder{Away: tg, Desk: dt, Store: h.Store, Asks: h.Asks, Bench: h.Bench, Publish: h.Publish, Now: c.now,
|
||||
Logf: t.Logf, Zone: h.Zone, RootFree: h.RootFree, Node: h.Node}
|
||||
if err := again.Load(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again.Start()
|
||||
again.Stand("desktop", Standing{Ready: true, At: c.now(), AnswersLostAt: second})
|
||||
runFor(again, c, 2*time.Minute)
|
||||
if len(dt.sends) != sends || len(dt.edits) != 0 || len(dt.replaced) != replaced || saidOn(tg) != tgSaid || passes(again) != 1 {
|
||||
t.Fatalf("the router's restart made the pass again: %d passes, sends %+v, edits %+v", passes(again),
|
||||
dt.sends[sends:], dt.edits)
|
||||
}
|
||||
}
|
||||
|
||||
// The controller's walk events are read as ADR 0308 §1 says them; any other plan-moved is not a walk's.
|
||||
func TestTheWalkEventsAreReadAsTheControllerSaysThem(t *testing.T) {
|
||||
at := time.Date(2026, 10, 12, 11, 12, 0, 0, time.UTC)
|
||||
ev, ok, err := DecodeWalk(walkBody(t, "plan-1", WalkStarted, "", at, 1200, novox("mesh-tools", "node-tools")))
|
||||
if err != nil || !ok || ev.Walk != "plan-1" || ev.BoundSeconds != 1200 || ev.Class != "leaf" || !ev.At.Equal(at) ||
|
||||
len(ev.Touches) != 1 || ev.Touches[0].Machine != "novox" || len(ev.Touches[0].Modules) != 2 {
|
||||
t.Fatalf("%+v %v %v", ev, ok, err)
|
||||
}
|
||||
// As mesh-controller's walk events (its pull request 247) say an end: ended_at, and carried beside modules.
|
||||
end, ok, err := DecodeWalk([]byte(`{"id":"plan-1","walk":"plan-1","phase":"ended","outcome":"stopped",` +
|
||||
`"touches":[{"machine":"novox","modules":["node-tools","messenger"],"carried":["messenger"]}],` +
|
||||
`"started_at":"2026-10-12T11:12:00Z","ended_at":"2026-10-12T11:19:00Z","updated":"2026-10-12T11:19:01Z"}`))
|
||||
if err != nil || !ok || end.Outcome != WalkStopped || !end.At.Equal(at.Add(7*time.Minute)) || len(end.Touches[0].Modules) != 2 {
|
||||
t.Fatalf("%+v %v %v", end, ok, err)
|
||||
}
|
||||
if _, ok, err := DecodeWalk([]byte(`{"id":"plan-1","state":"walking","tier":1}`)); ok || err != nil {
|
||||
t.Fatalf("a move inside a walk read as its start or end: %v %v", ok, err)
|
||||
}
|
||||
if _, _, err := DecodeWalk([]byte(`{"walk":"plan-1","phase":"ended"}`)); err == nil {
|
||||
t.Fatalf("an end without its outcome was read")
|
||||
}
|
||||
w := Walk{}
|
||||
if w.bound() != DefaultWalkBound {
|
||||
t.Fatalf("a walk without a bound is bounded at %s", w.bound())
|
||||
}
|
||||
// A start older than its bound is history: nothing is held.
|
||||
h, _, _, c, _ := newHolder(t)
|
||||
walk(t, h, "plan-old", WalkStarted, "", c.now().Add(-time.Hour), 1200, novox("mesh-tools"))
|
||||
if len(h.walks) != 0 {
|
||||
t.Fatalf("an old start opened a walk: %+v", h.walks)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// 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))
|
||||
}
|
||||
}
|
||||
@@ -73,6 +73,7 @@
|
||||
"mesh-controller.condition-raised",
|
||||
"mesh-controller.condition-changed",
|
||||
"mesh-controller.condition-cleared",
|
||||
"mesh-controller.plan-moved",
|
||||
"intake.choice",
|
||||
"intake.link",
|
||||
"intake.failed",
|
||||
@@ -93,7 +94,8 @@
|
||||
"name": "asks",
|
||||
"ttl-seconds": 2592000
|
||||
},
|
||||
"identities"
|
||||
"identities",
|
||||
"walks"
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
@@ -129,7 +131,7 @@
|
||||
"path": "${dir:state}/settings.json",
|
||||
"mode": "0600",
|
||||
"merge": "json",
|
||||
"content": "{\n \"time-zone\": \"\"\n}\n",
|
||||
"content": "{\n \"time-zone\": \"\",\n \"only-what-needs-me\": []\n}\n",
|
||||
"owner": "messenger"
|
||||
}
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user