mesh/merge-gate pass: builds messenger → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered
A warning held on purpose and lost with the walk's message at a restart was turned into history and never told. Name the walk in the told line from its own item.
907 lines
29 KiB
Go
907 lines
29 KiB
Go
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"`
|
|
// EndedSaid is the end's moment as the controller said it, by the controller's clock and never capped at
|
|
// this router's now: a start heard after its end is compared with it on that one clock (issue 538 C).
|
|
EndedSaid time.Time `json:"ended_said,omitempty"`
|
|
}
|
|
|
|
func (w *Walk) bound() time.Duration {
|
|
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 && startOfEnded(ev, at, end) {
|
|
// A short walk's two events can come in either order: this start is of the walk that ended.
|
|
h.Logf("[messenger] walk %s: its start came after its end, which was heard at %s: nothing held", ev.Walk,
|
|
end.Ended.UTC().Format(time.RFC3339))
|
|
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, EndedSaid: ev.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)
|
|
}
|
|
}
|
|
|
|
// startOfEnded says whether a start is of a walk whose end was heard already: of that walk, or of an earlier one
|
|
// under its id. The controller's own moments are compared on its clock, never one capped at this router's now
|
|
// against one that is not (issue 538 C: a controller's clock ahead of the router's would open a walk after its
|
|
// end). Only when the controller said neither moment are the moments as heard here compared.
|
|
func startOfEnded(ev WalkEvent, at time.Time, end *Walk) bool {
|
|
said := ev.StartedAt
|
|
if said.IsZero() {
|
|
said = ev.At
|
|
}
|
|
switch {
|
|
case !said.IsZero() && !end.Started.IsZero():
|
|
return !said.After(end.Started)
|
|
case !said.IsZero() && !end.EndedSaid.IsZero():
|
|
return !said.After(end.EndedSaid)
|
|
}
|
|
return !at.After(end.Ended)
|
|
}
|
|
|
|
// release is what a walk held that no other open walk holds or touches: each open message, oldest first, with
|
|
// what was to be said of it, and how many cleared meanwhile. A message said meanwhile on a path that is never
|
|
// held, or silenced meanwhile, is neither. One that another open walk touches — even one that started after it
|
|
// was held — is held on by that walk, and told at its end or bound (ADR 0308 §2).
|
|
//
|
|
// A message is handed on under the rules of its first hold (issue 538 B): still open, not silenced, not said
|
|
// since it was held, and not one ADR 0308 §3 never holds (an open ask about it, a walk's own failure).
|
|
func (h *Holder) release(w *Walk, now time.Time) (open []*Record, whats map[string]string, cleared int) {
|
|
// What may be handed on is judged first: neverHeld asks the asks and the walks under the lock itself.
|
|
h.mu.Lock()
|
|
recs := map[string]*Record{}
|
|
for key := range w.Held {
|
|
if r := h.open[key]; r != nil && r.Cleared.IsZero() && !r.silenced(now) && !lastSaid(r).After(w.Held[key].At) {
|
|
recs[key] = r
|
|
}
|
|
}
|
|
h.mu.Unlock()
|
|
handable := map[string]bool{}
|
|
for key, r := range recs {
|
|
handable[key] = !h.neverHeld(r)
|
|
}
|
|
h.mu.Lock()
|
|
whats = map[string]string{}
|
|
taken := map[string]*Walk{}
|
|
var handed []string
|
|
for key, hm := range w.Held {
|
|
r := h.open[key]
|
|
live := r != nil && r.Cleared.IsZero() && handable[key]
|
|
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
|
|
}
|
|
// Owed until a message carrying it was sent, in the state: a router that restarts first says it on its own
|
|
// (issue 538), whichever path below it takes.
|
|
h.mu.Lock()
|
|
for _, r := range append(append([]*Record(nil), open...), alone...) {
|
|
say := whats[r.Key]
|
|
if say == "" {
|
|
say = "raised"
|
|
}
|
|
r.Owed = say
|
|
}
|
|
h.mu.Unlock()
|
|
var to []Channel
|
|
if m.Urgent {
|
|
for _, ch := range h.channels() {
|
|
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
|
|
}
|
|
// What it carries is marked told on a channel once the message was sent there (outbox.go), not now: a
|
|
// message lost with a restart before it was sent leaves them untold and owed, and the restarted router says
|
|
// each on its own (issue 538 A).
|
|
var carries []string
|
|
for _, r := range open {
|
|
carries = append(carries, r.Key)
|
|
}
|
|
for _, ch := range kept {
|
|
msg := m
|
|
h.enqueue(ch.Name(), item{Key: "walk." + w.ID, What: WalkEndWhat, Severity: sev, At: now, Msg: &msg, Carries: carries})
|
|
}
|
|
for _, r := range open {
|
|
if len(kept) > 0 {
|
|
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: "released",
|
|
Outcome: "to be told in the message of walk " + w.ID + " " + what})
|
|
}
|
|
h.persist(r) // owed, in the state, until the message was sent
|
|
}
|
|
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
|
|
}
|