1262 lines
43 KiB
Go
1262 lines
43 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.
|
|
//
|
|
// **The settle (novox/hq ADR 0505 §3, §7).** A walk that ended delivered holds on, settling, for the setting
|
|
// settle-seconds (two minutes): what it touched stays held, a condition raised then on what it touched is held with
|
|
// it, and its end is told when the settle has passed. A walk that failed or stopped is told at once. The bound of
|
|
// §5 runs with the settle added: an open walk holds at most its bound and its settle. What a walk touches includes
|
|
// the machines whose declaration its merges change (`declared` touches, §2) and, for a walk that moves
|
|
// mesh-controller, the controller's own lease, bus and health conditions.
|
|
//
|
|
// **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
|
|
// DefaultSettle is how long a delivered walk still holds what it touched after its end when the setting
|
|
// settle-seconds gives none: the longest the first-node gate holds a healthy module (novox/hq ADR 0505 §3, ADR 0352).
|
|
DefaultSettle = 2 * 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
|
|
)
|
|
|
|
// The delivery view's events that stand for a walk now that plans and their plan-moved are gone (release 6, ADR 0539 §17): a
|
|
// walk is the set of (node, module) in motion. `moving` opens one for a module on a node, `settled` ends it delivered, `stuck`
|
|
// ends it failed or stopped with the view's words. Each is stated as the WalkEvent a plan-moved once was, so the holding, the
|
|
// bound, the settle and the end message below stay as they were.
|
|
const (
|
|
ViewModule = "mesh-delivery-view"
|
|
|
|
EventMoving = "moving"
|
|
EventSettled = "settled"
|
|
EventStuck = "stuck"
|
|
)
|
|
|
|
// DecodeMotion reads a view event's body as a walk's start or end: moving {node, module, from_fp, to_fp, composed, since,
|
|
// bound_seconds}, settled {node, module, at}, stuck {node, module, outcome, words, at}. The walk's id is "<node>/<module>".
|
|
func DecodeMotion(event string, body []byte) (WalkEvent, bool, error) {
|
|
var b struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Since time.Time `json:"since"`
|
|
At time.Time `json:"at"`
|
|
BoundSeconds float64 `json:"bound_seconds"`
|
|
Outcome string `json:"outcome"`
|
|
Words string `json:"words"`
|
|
}
|
|
switch event {
|
|
case EventMoving, EventSettled, EventStuck:
|
|
default:
|
|
return WalkEvent{}, false, nil
|
|
}
|
|
if err := json.Unmarshal(body, &b); err != nil {
|
|
return WalkEvent{}, false, fmt.Errorf("a %s event's body is not a JSON object", event)
|
|
}
|
|
if b.Node == "" || b.Module == "" {
|
|
return WalkEvent{}, false, fmt.Errorf("a %s event names no node or no module", event)
|
|
}
|
|
if b.BoundSeconds < 0 {
|
|
return WalkEvent{}, false, fmt.Errorf("%s/%s: bound_seconds is negative", b.Node, b.Module)
|
|
}
|
|
ev := WalkEvent{Walk: b.Node + "/" + b.Module, Touches: []Touch{{Machine: b.Node, Modules: []string{b.Module}}}, Words: b.Words}
|
|
switch event {
|
|
case EventMoving:
|
|
ev.Phase, ev.BoundSeconds, ev.StartedAt, ev.At = WalkStarted, int(b.BoundSeconds), b.Since, b.Since
|
|
case EventSettled:
|
|
ev.Phase, ev.Outcome, ev.At = WalkEnded, WalkDelivered, b.At
|
|
case EventStuck:
|
|
ev.Phase, ev.Outcome, ev.At = WalkEnded, WalkFailed, b.At
|
|
if b.Outcome == WalkStopped {
|
|
ev.Outcome = WalkStopped
|
|
}
|
|
}
|
|
return ev, true, nil
|
|
}
|
|
|
|
// settle is how long a delivered walk still holds after its end (ADR 0505 §3).
|
|
func (h *Holder) settle() time.Duration {
|
|
if h.Settle == nil {
|
|
return DefaultSettle
|
|
}
|
|
if d := h.Settle(); d >= 0 {
|
|
return d
|
|
}
|
|
return DefaultSettle
|
|
}
|
|
|
|
// Touch is what a walk moves on one machine.
|
|
type Touch struct {
|
|
Machine string `json:"machine"`
|
|
Modules []string `json:"modules"`
|
|
// Declared: the walk sends nothing to this machine, but its merges change the machine's declaration (an
|
|
// assignment, a provision or a requirement added or removed, ADR 0505 §2b). Held like any touch; never named
|
|
// among the machines the walk reached.
|
|
Declared bool `json:"declared,omitempty"`
|
|
}
|
|
|
|
// 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
|
|
// Words are the view's own words of why a walk ended failed or stopped (`stuck`); empty for a plan's.
|
|
Words string
|
|
}
|
|
|
|
// 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"`
|
|
// Until is the first hold's limit: the start, bound and settle of the walk that first held it, or that walk's
|
|
// settle's end when it was held settling (ADR 0505 §7). Kept when it is handed on: no walk holds it past this, so a
|
|
// fault on a machine walk after walk touches reaches the operator all the same. Zero in a hold kept before it was.
|
|
Until time.Time `json:"until,omitzero"`
|
|
}
|
|
|
|
// past says whether a held message's first hold's limit has passed.
|
|
func (hm HeldMessage) past(now time.Time) bool { return !hm.Until.IsZero() && !now.Before(hm.Until) }
|
|
|
|
// 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"`
|
|
// SettleSeconds is the settle taken at its start (the setting settle-seconds, ADR 0505 §3): the hold of an open
|
|
// walk lasts its bound and this (§7), and a delivered walk holds this long after its end.
|
|
SettleSeconds int `json:"settle_seconds,omitempty"`
|
|
// SettleUntil is set on a walk that ended delivered and still holds, settling, until then (ADR 0505 §3); Outcome
|
|
// is how it ended. Its end is told when SettleUntil passes.
|
|
SettleUntil time.Time `json:"settle_until,omitempty"`
|
|
Outcome string `json:"outcome,omitempty"`
|
|
// Words are the view's words of why it ended failed or stopped, told in the end message.
|
|
Words string `json:"words,omitempty"`
|
|
}
|
|
|
|
func (w *Walk) bound() time.Duration {
|
|
if w.BoundSeconds <= 0 {
|
|
return DefaultWalkBound
|
|
}
|
|
return time.Duration(w.BoundSeconds) * time.Second
|
|
}
|
|
|
|
// settleFor is the settle the walk took at its start.
|
|
func (w *Walk) settleFor() time.Duration { return time.Duration(w.SettleSeconds) * time.Second }
|
|
|
|
// limit is the longest a walk holds anything: its start, its bound and its settle (ADR 0505 §7).
|
|
func (w *Walk) limit() time.Time { return w.Started.Add(w.bound() + w.settleFor()) }
|
|
|
|
// holdEnd is the last moment the walk holds anything it holds now: its limit, or its settle's end when that is
|
|
// sooner.
|
|
func (w *Walk) holdEnd() time.Time {
|
|
if w.settling() && w.SettleUntil.Before(w.limit()) {
|
|
return w.SettleUntil
|
|
}
|
|
return w.limit()
|
|
}
|
|
|
|
// settling says whether the walk ended delivered and holds on until its settle has passed (ADR 0505 §3).
|
|
func (w *Walk) settling() bool { return !w.SettleUntil.IsZero() }
|
|
|
|
// holding says whether the walk still holds at a moment: open or settling, and within its bound and settle.
|
|
func (w *Walk) holding(now time.Time) bool {
|
|
return !w.BoundSaid && now.Before(w.limit()) && (!w.settling() || now.Before(w.SettleUntil))
|
|
}
|
|
|
|
// 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 && !s.controller {
|
|
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"`
|
|
Declared bool `json:"declared"`
|
|
}
|
|
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, Declared: t.Declared})
|
|
}
|
|
}
|
|
}
|
|
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. A machine is declared only while no saying
|
|
// sends to it (ADR 0505 §2b).
|
|
func mergeTouches(a, b []Touch) []Touch {
|
|
by := map[string]map[string]bool{}
|
|
sent := 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
|
|
}
|
|
if !t.Declared {
|
|
sent[t.Machine] = 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, Declared: !sent[machine]})
|
|
}
|
|
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,
|
|
SettleSeconds: int(h.settle() / time.Second)}
|
|
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 && w.settling() {
|
|
// Its end said again while it settles: the same end, its touches as both sayings name them.
|
|
w.Touches = mergeTouches(w.Touches, ev.Touches)
|
|
h.mu.Unlock()
|
|
h.keepWalk(w)
|
|
return
|
|
}
|
|
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 {
|
|
w.Touches = mergeTouches(w.Touches, ev.Touches)
|
|
w.Words = ev.Words
|
|
if !ev.StartedAt.IsZero() {
|
|
w.Started = ev.StartedAt
|
|
}
|
|
}
|
|
tomb := &Walk{ID: ev.Walk, Started: ev.StartedAt, Ended: at, EndedSaid: ev.At, Outcome: ev.Outcome,
|
|
Touches: mergeTouches(ev.Touches, nil)}
|
|
if w != nil {
|
|
tomb.Touches, tomb.SettleSeconds = w.Touches, w.SettleSeconds
|
|
}
|
|
if prev := h.ended[ev.Walk]; prev != nil && prev.Ended.After(at) {
|
|
tomb = prev
|
|
}
|
|
h.ended[ev.Walk] = tomb
|
|
settles := w != nil && ev.Outcome == WalkDelivered && !w.BoundSaid && w.settleFor() > 0
|
|
if settles {
|
|
// Delivered: it holds on, settling, until its settle has passed and never past its limit (ADR 0505 §3, §7).
|
|
w.Ended, w.EndedSaid, w.Outcome = at, ev.At, ev.Outcome
|
|
w.SettleUntil = at.Add(w.settleFor())
|
|
if l := w.limit(); w.SettleUntil.After(l) {
|
|
w.SettleUntil = l
|
|
}
|
|
} else if w != nil {
|
|
delete(h.walks, ev.Walk)
|
|
}
|
|
h.mu.Unlock()
|
|
if settles {
|
|
h.keepWalk(w) // settling, in place of the open walk: read back as settling after a restart
|
|
h.record(Sent{At: now, Channel: "-", Key: "walk." + ev.Walk, What: "ended " + ev.Outcome,
|
|
Outcome: "settling: holding what it touched until " + w.SettleUntil.In(h.zone()).Format("15:04:05")})
|
|
return
|
|
}
|
|
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 other, ok := o.Held[key]; ok {
|
|
elsewhere = elsewhere || !other.past(now)
|
|
continue
|
|
}
|
|
if live && o.touches(s) && !hm.past(now) {
|
|
// Handed on with its first hold's limit: never held past it (ADR 0505 §7).
|
|
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)
|
|
due := ended
|
|
if w.settling() {
|
|
due = w.SettleUntil
|
|
}
|
|
if now.Sub(due) > 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) + "."}
|
|
if w.Words != "" {
|
|
lines = append(lines, w.Words)
|
|
}
|
|
lines = append(lines, h.heldLines(open, whats, cleared, "Still open after the delivery of "+deliveryWords(w.Touches)+":")...)
|
|
m = Message{Title: head + ": " + reached, Body: strings.Join(lines, "\n")}
|
|
}
|
|
h.tellWalk(w, m, open, whats, oneLine, "ended "+outcome, now)
|
|
}
|
|
|
|
// settledWalks tells the end of every delivered walk whose settle has passed (ADR 0505 §3), oldest first.
|
|
func (h *Holder) settledWalks(now time.Time) {
|
|
h.overdueHolds(now)
|
|
h.mu.Lock()
|
|
var settled []*Walk
|
|
for _, w := range h.walks {
|
|
if w.settling() && !now.Before(w.SettleUntil) {
|
|
settled = append(settled, w)
|
|
}
|
|
}
|
|
h.mu.Unlock()
|
|
sort.Slice(settled, func(i, j int) bool { return settled[i].SettleUntil.Before(settled[j].SettleUntil) })
|
|
for _, w := range settled {
|
|
h.endSettled(w, now)
|
|
}
|
|
}
|
|
|
|
// overdueHolds says every held message whose first hold's limit has passed while a walk that took it over still holds,
|
|
// on its own, with its delivery line (ADR 0505 §7): no walk holds it past that limit. A walk whose own end or bound is
|
|
// due tells what it holds in its message instead. Called at each flush.
|
|
func (h *Holder) overdueHolds(now time.Time) {
|
|
h.mu.Lock()
|
|
// What a walk whose own end or bound is due holds is told in that walk's message, and nowhere else: one place
|
|
// says it, so a message two walks hold is never said twice.
|
|
tellsItself := map[string]bool{}
|
|
for _, w := range h.walks {
|
|
if !now.Before(w.holdEnd()) {
|
|
for key := range w.Held {
|
|
tellsItself[key] = true
|
|
}
|
|
}
|
|
}
|
|
due := map[string]HeldMessage{}
|
|
var touched []*Walk
|
|
for _, w := range h.walks {
|
|
changed := false
|
|
if !now.Before(w.holdEnd()) {
|
|
continue // its own end or bound tells what it holds, in its message
|
|
}
|
|
for key, hm := range w.Held {
|
|
if hm.past(now) {
|
|
if prev, ok := due[key]; !tellsItself[key] && (!ok || hm.At.Before(prev.At)) {
|
|
due[key] = hm
|
|
}
|
|
delete(w.Held, key)
|
|
changed = true
|
|
}
|
|
}
|
|
if changed {
|
|
touched = append(touched, w)
|
|
}
|
|
}
|
|
var recs []*Record
|
|
for key, hm := range due {
|
|
if r := h.open[key]; r != nil && r.Cleared.IsZero() && !r.silenced(now) && !lastSaid(r).After(hm.At) {
|
|
recs = append(recs, r)
|
|
}
|
|
}
|
|
h.mu.Unlock()
|
|
for _, w := range touched {
|
|
h.keepWalk(w)
|
|
}
|
|
sort.Slice(recs, func(i, j int) bool { return recs[i].Key < recs[j].Key })
|
|
for _, r := range recs {
|
|
what := due[r.Key].What
|
|
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: what, Outcome: "released: past its first hold's limit"})
|
|
h.route(r, what)
|
|
}
|
|
}
|
|
|
|
// endSettled tells the end of a delivered walk whose settle has passed (ADR 0505 §3): what it held still open, one
|
|
// line each, and the cleared counted; then it is only the tomb that keeps a late start from opening it again.
|
|
func (h *Holder) endSettled(w *Walk, now time.Time) {
|
|
h.mu.Lock()
|
|
delete(h.walks, w.ID)
|
|
tomb := h.ended[w.ID]
|
|
if tomb == nil {
|
|
tomb = &Walk{ID: w.ID, Started: w.Started, Ended: w.Ended, EndedSaid: w.EndedSaid}
|
|
h.ended[w.ID] = tomb
|
|
}
|
|
h.mu.Unlock()
|
|
h.keepWalk(tomb)
|
|
h.endWalk(w, w.Outcome, w.Ended, 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, heading string) []string {
|
|
var lines []string
|
|
if len(open) > 0 {
|
|
lines = append(lines, heading)
|
|
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.settling() && !w.BoundSaid && !now.Before(w.limit()) {
|
|
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)
|
|
}
|
|
}()
|
|
h.settledWalks(now)
|
|
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, "Still open after it:")...)
|
|
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 {
|
|
if t.Declared {
|
|
continue // its declaration changed; nothing was sent to it (ADR 0505 §2b)
|
|
}
|
|
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)
|
|
}
|
|
|
|
// deliveryWords names a delivery as the operator says it: its modules, then "to" the machines it sent them to
|
|
// (ADR 0505 §1); a machine whose declaration alone changed is not among them.
|
|
func deliveryWords(touches []Touch) string {
|
|
mods := map[string]bool{}
|
|
var machines []string
|
|
for _, t := range touches {
|
|
if t.Declared {
|
|
continue
|
|
}
|
|
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)
|
|
switch {
|
|
case len(names) == 0 && len(machines) == 0:
|
|
return "nothing"
|
|
case len(machines) == 0:
|
|
return listed(names)
|
|
case len(names) == 0:
|
|
return "its builds to " + listed(machines)
|
|
}
|
|
return listed(names) + " to " + 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
|
|
// controller: the subject is the controller itself — its lease, its bus connection, its health — matched to the
|
|
// module mesh-controller wherever a walk moves it, whatever machine or host the condition names (ADR 0505 §2a).
|
|
controller bool
|
|
}
|
|
|
|
// controllerNames are the names the controller's own conditions give it in their keys (`core.controller.lease.lost`,
|
|
// `bus.controller.bus-reconnects`, `module.controller...`): each is the module ControllerModule (ADR 0505 §2a).
|
|
var controllerNames = map[string]bool{"controller": true, ControllerModule: true}
|
|
|
|
// ControllerModule is the controller's module, as a walk that moves it names it.
|
|
const ControllerModule = "mesh-controller"
|
|
|
|
// 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
|
|
}
|
|
if (scope == "core" || scope == "bus" || scope == "module" || scope == "seat") && len(tokens) > 0 && controllerNames[tokens[0]] {
|
|
// The controller's own subject: the module mesh-controller, wherever it runs (ADR 0505 §2a).
|
|
s.modules, s.controller = append(s.modules, ControllerModule), true
|
|
return s
|
|
}
|
|
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)
|
|
if f := h.failedJustNow(s, now); f != nil {
|
|
// A failure is the case the operator most needs the cause of: sent at once, with its delivery line (ADR 0505 §6).
|
|
h.mu.Unlock()
|
|
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: what + ", " + roughly(now.Sub(f.Ended)) + " after walk " + f.ID + " " + f.Outcome,
|
|
Outcome: "not held: its walk " + f.Outcome})
|
|
return false
|
|
}
|
|
// The first hold's limit: kept by whichever walk holds it now, else the soonest end of the walks about to.
|
|
var until time.Time
|
|
for _, w := range h.walks {
|
|
if hm, ok := w.Held[r.Key]; ok && !hm.Until.IsZero() && (until.IsZero() || hm.Until.Before(until)) {
|
|
until = hm.Until
|
|
}
|
|
}
|
|
if until.IsZero() {
|
|
for _, w := range h.walks {
|
|
if w.holding(now) && w.touches(s) && (until.IsZero() || w.holdEnd().Before(until)) {
|
|
until = w.holdEnd()
|
|
}
|
|
}
|
|
}
|
|
if !until.IsZero() && !now.Before(until) {
|
|
// Held to its first hold's limit already: said now, with its delivery line (ADR 0505 §7).
|
|
h.mu.Unlock()
|
|
h.record(Sent{At: now, Channel: "-", Key: r.Key, What: what, Outcome: "not held: past its first hold's limit"})
|
|
return false
|
|
}
|
|
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, Until: until}
|
|
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
|
|
}
|
|
|
|
// failedJustNow is the walk that failed or stopped within its settle and touched a subject (ADR 0505 §6): nothing
|
|
// on it is held then. Under the lock.
|
|
func (h *Holder) failedJustNow(s subject, now time.Time) *Walk {
|
|
var newest *Walk
|
|
for _, e := range h.ended {
|
|
if e.Outcome != WalkFailed && e.Outcome != WalkStopped {
|
|
continue
|
|
}
|
|
if now.Before(e.Ended) || !now.Before(e.Ended.Add(h.settleOf(e))) || !e.touches(s) {
|
|
continue
|
|
}
|
|
if newest == nil || e.Ended.After(newest.Ended) {
|
|
newest = e
|
|
}
|
|
}
|
|
return newest
|
|
}
|
|
|
|
// settleOf is the settle a walk took at its start, or the setting's for one kept before it was.
|
|
func (h *Holder) settleOf(w *Walk) time.Duration {
|
|
if w.SettleSeconds > 0 {
|
|
return w.settleFor()
|
|
}
|
|
return h.settle()
|
|
}
|
|
|
|
// 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() && w.settling() {
|
|
// Settling when the router stopped: it holds on until its settle has passed, and its end keeps a late
|
|
// start from opening it again (ADR 0505 §3).
|
|
if w.Held == nil {
|
|
w.Held = map[string]HeldMessage{}
|
|
}
|
|
h.walks[w.ID] = &w
|
|
h.ended[w.ID] = &Walk{ID: w.ID, Started: w.Started, Ended: w.Ended, EndedSaid: w.EndedSaid}
|
|
continue
|
|
}
|
|
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
|
|
}
|