mesh/delivery-group group feat/774-l3-reland rejected: composed together: the checker's verdict at the members' heads (b48c90fea31f): integration: mesh-self-ch…
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: integration: mesh-self-check/cmd/mesh-self-check.TestTheSelfCheckWholeRunsOnTheBus: the run answered {Run:doctor-1793351737-1 At:20…
One commit with exactly the tree of the reviewed head35b4509c0a, on main12c2574c9d, so two compiled programs added and removed on the branch leave no trace in history (issue 1577).
797 lines
28 KiB
Go
797 lines
28 KiB
Go
// messenger: the holder of the operator-channel seat, the router (novox/hq to-be 45 §5, to-be 46, ADR
|
|
// 0227, ADR 0259). A Go bundle the node's runtime launches. It consumes the controller's condition events and
|
|
// decides what is said to the operator, on the channel kinds the mesh holds — each a module of its own — and
|
|
// holds every ask, judges each answer and issues the warrant its asker acts on. It keeps its open messages in its own state, so a restart forgets nothing. stdout is the MCP
|
|
// channel; what this module says, it says on stderr.
|
|
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
_ "time/tzdata" // the operator's zone is known whatever this machine has installed
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
"git.novox.be/novox/mesh-sdk/go/health"
|
|
)
|
|
|
|
func errorf(format string, a ...any) error { return fmt.Errorf(format, a...) }
|
|
|
|
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
|
|
|
|
// Settings are the operator's values for this module, merged by the mesh into one JSON file.
|
|
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"`
|
|
// MessagesPerHour is the most messages one channel gets in an hour (issue 1476); 0, not given, is
|
|
// DefaultCapPerHour; 0 is no cap at all.
|
|
MessagesPerHour *int `json:"messages-per-hour"`
|
|
// SettleSeconds is how long a delivered walk still holds what it touched after its end (novox/hq ADR 0505 §3):
|
|
// nil, not given, is DefaultSettle; 0 holds nothing after the end.
|
|
SettleSeconds *int `json:"settle-seconds"`
|
|
}
|
|
|
|
// zoneOf is the location a time-zone setting names, or this machine's own when it names none or one
|
|
// that does not exist (said, never guessed silently).
|
|
func zoneOf(name string) (*time.Location, error) {
|
|
if name == "" {
|
|
return time.Local, nil
|
|
}
|
|
loc, err := time.LoadLocation(name)
|
|
if err != nil {
|
|
return time.Local, fmt.Errorf("the time-zone %q is not one this machine knows; times are said in its own zone", name)
|
|
}
|
|
return loc, nil
|
|
}
|
|
|
|
func readSettings(path string) (Settings, error) {
|
|
var s Settings
|
|
if path == "" {
|
|
return s, errors.New("started without a settings file")
|
|
}
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return s, err
|
|
}
|
|
var loose map[string]any
|
|
if err := json.Unmarshal(raw, &loose); err != nil {
|
|
return s, fmt.Errorf("the settings file is not JSON: %v", err)
|
|
}
|
|
if tz, ok := loose["time-zone"].(string); ok {
|
|
s.TimeZone = strings.TrimSpace(tz)
|
|
}
|
|
switch v := loose["messages-per-hour"].(type) {
|
|
case nil:
|
|
case float64:
|
|
n := int(v)
|
|
if v < 0 || v != float64(n) {
|
|
return s, fmt.Errorf("messages-per-hour is not a whole number of zero or more")
|
|
}
|
|
s.MessagesPerHour = &n
|
|
default:
|
|
return s, fmt.Errorf("messages-per-hour is not a number")
|
|
}
|
|
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")
|
|
}
|
|
switch v := loose["settle-seconds"].(type) {
|
|
case nil:
|
|
case float64:
|
|
n := int(v)
|
|
if v < 0 || v != float64(n) {
|
|
return s, fmt.Errorf("settle-seconds is not a whole number of zero or more")
|
|
}
|
|
s.SettleSeconds = &n
|
|
default:
|
|
return s, fmt.Errorf("settle-seconds is not a number")
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// stateStore keeps the open messages in the module's declared state `open`, and the recent sends
|
|
// under one key of `sent` (ADR 0201).
|
|
type stateStore struct{}
|
|
|
|
// kvKey is a condition key as a bucket key: only the characters a key may carry.
|
|
func kvKey(key string) string {
|
|
var b strings.Builder
|
|
for _, r := range key {
|
|
switch {
|
|
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_', r == '.', r == '=':
|
|
b.WriteRune(r)
|
|
default:
|
|
b.WriteRune('_')
|
|
}
|
|
}
|
|
return strings.Trim(b.String(), ".")
|
|
}
|
|
|
|
func (stateStore) Put(r Record) error {
|
|
_, err := stdio.State("open").Put(kvKey(r.Key), r)
|
|
return err
|
|
}
|
|
func (stateStore) Delete(key string) error { return stdio.State("open").Delete(kvKey(key)) }
|
|
func (stateStore) All() ([]Record, error) {
|
|
keys, err := stdio.State("open").Keys()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out []Record
|
|
for _, k := range keys {
|
|
e, err := stdio.State("open").Get(k)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if e == nil {
|
|
continue
|
|
}
|
|
var r Record
|
|
if err := json.Unmarshal(e.Value, &r); err != nil {
|
|
logf("[messenger] the open message kept as %s cannot be read (%v); left as it is", k, err)
|
|
continue
|
|
}
|
|
out = append(out, r)
|
|
}
|
|
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
|
|
}
|
|
func (stateStore) Recent() ([]Sent, error) {
|
|
e, err := stdio.State("sent").Get("recent")
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var s []Sent
|
|
return s, json.Unmarshal(e.Value, &s)
|
|
}
|
|
|
|
// askStore keeps the asks in the state `asks` and the operator's identities, and the links asked for, in
|
|
// `identities` (novox/hq ADR 0259). Each write that must win once is a compare-and-set.
|
|
type askStore struct{}
|
|
|
|
// changed reads the runtime's refusal of a compare-and-set that lost, told by its code, as ErrChanged.
|
|
func changed(err error) error {
|
|
if errors.Is(err, stdio.ErrChanged) {
|
|
return ErrChanged
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (askStore) GetAsk(key string) (*AskRecord, uint64, error) {
|
|
e, err := stdio.State("asks").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, 0, err
|
|
}
|
|
var r AskRecord
|
|
if err := json.Unmarshal(e.Value, &r); err != nil {
|
|
return nil, 0, fmt.Errorf("the ask kept as %s cannot be read: %v", key, err)
|
|
}
|
|
return &r, e.Revision, nil
|
|
}
|
|
func (askStore) CreateAsk(key string, r AskRecord) (uint64, error) {
|
|
rev, err := stdio.State("asks").Create(key, r)
|
|
return rev, changed(err)
|
|
}
|
|
func (askStore) UpdateAsk(key string, r AskRecord, revision uint64) (uint64, error) {
|
|
rev, err := stdio.State("asks").Update(key, r, revision)
|
|
return rev, changed(err)
|
|
}
|
|
func (askStore) AskKeys() ([]string, error) { return stdio.State("asks").Keys() }
|
|
func (askStore) GetIdentity(key string) (*Identity, error) {
|
|
e, err := stdio.State("identities").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var id Identity
|
|
return &id, json.Unmarshal(e.Value, &id)
|
|
}
|
|
func (askStore) PutIdentity(key string, id Identity) error {
|
|
_, err := stdio.State("identities").Put(key, id)
|
|
return err
|
|
}
|
|
func (askStore) DeleteIdentity(key string) error { return stdio.State("identities").Delete(key) }
|
|
func (askStore) IdentityKeys() ([]string, error) {
|
|
keys, err := stdio.State("identities").Keys()
|
|
var out []string
|
|
for _, k := range keys {
|
|
if !strings.HasPrefix(k, "pending.") && !strings.HasPrefix(k, "unlinked.") && !strings.HasPrefix(k, "asked.") {
|
|
out = append(out, k)
|
|
}
|
|
}
|
|
return out, err
|
|
}
|
|
func (askStore) GetLink(kind string) (*PendingLink, error) {
|
|
e, err := stdio.State("identities").Get("pending." + kind)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var l PendingLink
|
|
return &l, json.Unmarshal(e.Value, &l)
|
|
}
|
|
func (askStore) PutLink(kind string, l PendingLink) error {
|
|
_, err := stdio.State("identities").Put("pending."+kind, l)
|
|
return err
|
|
}
|
|
func (askStore) DeleteLink(kind string) error {
|
|
return stdio.State("identities").Delete("pending." + kind)
|
|
}
|
|
func (askStore) GetUnlinked(kind string) (*Unlinked, error) {
|
|
e, err := stdio.State("identities").Get("unlinked." + kind)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var u Unlinked
|
|
return &u, json.Unmarshal(e.Value, &u)
|
|
}
|
|
func (askStore) PutUnlinked(kind string, u Unlinked) error {
|
|
_, err := stdio.State("identities").Put("unlinked."+kind, u)
|
|
return err
|
|
}
|
|
|
|
func (askStore) GetLinksAsked(key string) ([]time.Time, error) {
|
|
e, err := stdio.State("identities").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var times []time.Time
|
|
return times, json.Unmarshal(e.Value, ×)
|
|
}
|
|
func (askStore) PutLinksAsked(key string, times []time.Time) error {
|
|
_, err := stdio.State("identities").Put(key, times)
|
|
return err
|
|
}
|
|
|
|
// listening is whether condition events reach this holder, in words.
|
|
type listening struct {
|
|
mu sync.Mutex
|
|
now string
|
|
}
|
|
|
|
func (l *listening) set(s string) { l.mu.Lock(); l.now = s; l.mu.Unlock() }
|
|
func (l *listening) get() string { l.mu.Lock(); defer l.mu.Unlock(); return l.now }
|
|
|
|
func main() {
|
|
settingsFile := os.Getenv("MESH_MESSENGER_SETTINGS")
|
|
var said sync.Mutex
|
|
lastSaid, zoneSaid := "", ""
|
|
settings := func() Settings {
|
|
s, err := readSettings(settingsFile)
|
|
said.Lock()
|
|
defer said.Unlock()
|
|
if err != nil && err.Error() != lastSaid {
|
|
logf("[messenger] settings cannot be read: %v", err)
|
|
}
|
|
lastSaid = ""
|
|
if err != nil {
|
|
lastSaid = err.Error()
|
|
}
|
|
return s
|
|
}
|
|
bench := NewBench(time.Now)
|
|
publish := func(subject string, body any, id string) error { return stdio.SeatPublish(subject, body, id) }
|
|
h := &Holder{
|
|
Away: &SeatChannel{Bench: bench, Role: RoleAway, Publish: publish},
|
|
Desk: &SeatChannel{Bench: bench, Role: RoleDesk, Publish: publish},
|
|
Store: stateStore{},
|
|
Asks: askStore{},
|
|
Node: os.Getenv("MESH_NODE"),
|
|
StateDir: os.Getenv("MESH_MESSENGER_STATE"),
|
|
Bench: bench,
|
|
Publish: publish,
|
|
Ask: stdio.Ask,
|
|
Logf: logf,
|
|
Zone: func() *time.Location {
|
|
loc, err := zoneOf(settings().TimeZone)
|
|
said.Lock()
|
|
defer said.Unlock()
|
|
if err != nil && err.Error() != zoneSaid {
|
|
logf("[messenger] %v", err)
|
|
}
|
|
zoneSaid = ""
|
|
if err != nil {
|
|
zoneSaid = err.Error()
|
|
}
|
|
return loc
|
|
},
|
|
Emit: func(event string, body any) error { return stdio.Emit(event, body) },
|
|
CapPerHour: func() int {
|
|
if n := settings().MessagesPerHour; n != nil {
|
|
return *n
|
|
}
|
|
return DefaultCapPerHour
|
|
},
|
|
Settle: func() time.Duration {
|
|
if n := settings().SettleSeconds; n != nil && *n >= 0 {
|
|
return time.Duration(*n) * time.Second
|
|
}
|
|
return DefaultSettle
|
|
},
|
|
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 {
|
|
return ids[0].Identity
|
|
}
|
|
return ""
|
|
}
|
|
h.init()
|
|
l := &listening{now: "not yet: starting"}
|
|
go run(h, l)
|
|
if err := stdio.Serve("", tools(h, l)); err != nil {
|
|
logf("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// run reads back the state, listens for condition events, and keeps time — each retried, and each
|
|
// failure said, never given up on quietly.
|
|
func run(h *Holder, l *listening) {
|
|
time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := h.Load()
|
|
if err == nil {
|
|
break
|
|
}
|
|
logf("[messenger] cannot read back its open messages yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
for _, ch := range h.channels() {
|
|
if err := ch.Ready(); err != nil {
|
|
logf("[messenger] %s cannot send: %v", ch.Name(), err)
|
|
}
|
|
}
|
|
// The state first: what is open now is read from the controller before any event is taken, so
|
|
// what the stream replays is history (novox/hq issue 271). Not reachable yet, it is asked for
|
|
// again in the loop below, and old events meanwhile are kept as state, never said.
|
|
follow := newTakeover()
|
|
h.Start()
|
|
syncFromController(h, follow)
|
|
go func() {
|
|
tick := time.NewTicker(5 * time.Second)
|
|
defer tick.Stop()
|
|
lastTick, lastAsk := time.Now(), time.Now()
|
|
for now := range tick.C {
|
|
h.Flush()
|
|
if now.Sub(lastTick) >= time.Minute {
|
|
lastTick = now
|
|
h.Tick()
|
|
}
|
|
if h.WantsSync() && now.Sub(lastAsk) >= time.Minute {
|
|
lastAsk = now
|
|
syncFromController(h, follow)
|
|
}
|
|
}
|
|
}()
|
|
go followTakeover(follow, h)
|
|
handle := func(e stdio.Envelope) error {
|
|
if event, ok := strings.CutPrefix(e.Key, ViewModule+"."); ok {
|
|
// A module in motion on a node, settled or stuck: a walk's start or end now that plans are gone (ADR 0308 §1, ADR 0539 §17).
|
|
ev, isWalk, err := DecodeMotion(event, e.Body)
|
|
if err != nil {
|
|
h.Unreadable(e.Key, err)
|
|
return nil
|
|
}
|
|
if isWalk {
|
|
h.Walked(ev)
|
|
}
|
|
return nil
|
|
}
|
|
// Only the seat the take-over key names is heard: the other's events are dropped, so the keeper's conditions
|
|
// are never held a second time beside the controller's (ADR 0564 clause 9).
|
|
event, ok := eventOfSeat(e.Key, follow.seat())
|
|
if !ok {
|
|
return nil
|
|
}
|
|
c, err := DecodeCondition(event, e.Body)
|
|
if err != nil {
|
|
h.Unreadable(e.Key, err)
|
|
return nil
|
|
}
|
|
h.Condition(event, c)
|
|
return nil
|
|
}
|
|
go listenForChannels(h)
|
|
go takeAsks(h)
|
|
go answerProofs(h)
|
|
go followKinds(h)
|
|
// The delivery view's events say what is in motion (ADR 0539 §17), heard beside the controller's condition events.
|
|
go func() {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
if err := stdio.Subscribe(ViewModule+".*", handle); err == nil {
|
|
logf("[messenger] listening for the delivery view's moving, settled and stuck")
|
|
return
|
|
} else {
|
|
logf("[messenger] not hearing the delivery view yet (%v); asking again in %s", err, wait)
|
|
}
|
|
time.Sleep(wait)
|
|
}
|
|
}()
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := stdio.Subscribe(ControllerSeat+".*", handle)
|
|
if err == nil {
|
|
err = stdio.Subscribe(health.KeeperModule+".*", handle)
|
|
}
|
|
if err == nil {
|
|
l.set("listening")
|
|
logf("[messenger] listening for the controller's condition events")
|
|
return
|
|
}
|
|
l.set("not yet: " + err.Error())
|
|
logf("[messenger] not hearing condition events yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// kindOf is an intake event's kind and what it is: `intake.<what>.<kind>`, the kind the subject's last token,
|
|
// which the bus grants that kind's holder alone.
|
|
func kindOf(key string) (what, kind string, ok bool) {
|
|
parts := strings.Split(key, ".")
|
|
if len(parts) != 3 || parts[0] != asks.IntakeSeat {
|
|
return "", "", false
|
|
}
|
|
return parts[1], parts[2], true
|
|
}
|
|
|
|
// listenForChannels takes what the channels say (novox/hq ADR 0259): answers chosen, links asked, messages
|
|
// that failed, and whether each can send — every one under the kind the bus granted its holder.
|
|
func listenForChannels(h *Holder) {
|
|
handle := func(e stdio.Envelope) error {
|
|
what, kind, ok := kindOf(e.Key)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
switch what {
|
|
case asks.Chosen:
|
|
var c asks.Choice
|
|
if err := json.Unmarshal(e.Body, &c); err != nil || c.Ticket == "" {
|
|
logf("[messenger] an answer from %s could not be read: %s", kind, firstLineOf(string(e.Body)))
|
|
return nil
|
|
}
|
|
if c.ID == "" {
|
|
c.ID = e.Headers["x-event-id"]
|
|
}
|
|
_, _ = h.Chosen(kind, c)
|
|
case asks.Link:
|
|
l, err := decodeLink(e.Body)
|
|
if err != nil {
|
|
logf("[messenger] a link asked on %s could not be read", kind)
|
|
return nil
|
|
}
|
|
h.Link(kind, l)
|
|
case asks.FailedWhat:
|
|
var f asks.Failed
|
|
if json.Unmarshal(e.Body, &f) == nil {
|
|
h.Failed(kind, f)
|
|
}
|
|
case asks.StandingWhat:
|
|
// Read as this module's Standing: the bench's word, and answers-lost-at beside it (issue 384).
|
|
var s Standing
|
|
if json.Unmarshal(e.Body, &s) == nil {
|
|
h.Stand(kind, s)
|
|
h.ReleaseHeld() // a channel's standing changed: held asks are judged again (ADR 0397 §2)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
if err := stdio.Subscribe(asks.IntakeSeat+".#", handle); err == nil {
|
|
logf("[messenger] listening for what the channels say")
|
|
break
|
|
} else {
|
|
logf("[messenger] not hearing the channels yet (%v); asking again in %s", err, wait)
|
|
}
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// takeAsks takes the asks and cancels submitted to the seat, each under its asker's name.
|
|
func takeAsks(h *Holder) {
|
|
handle := func(w stdio.Work) error {
|
|
parts := strings.Split(w.Subject, ".")
|
|
if len(parts) != 6 {
|
|
logf("[messenger] work on %s is not an ask; taken and dropped", w.Subject)
|
|
return nil
|
|
}
|
|
return h.TakeAsk(parts[4], parts[5], w.Body)
|
|
}
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
if err := stdio.SeatTake(AsksWorker, handle); err == nil {
|
|
logf("[messenger] taking the asks submitted to %s", asks.Seat)
|
|
break
|
|
} else {
|
|
logf("[messenger] not taking asks yet (%v); asking again in %s", err, wait)
|
|
}
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// AsksWorker is the worker the seat's asks are taken from, as the controller names it.
|
|
const AsksWorker = "SEAT_OPERATOR_CHANNEL_worker"
|
|
|
|
// answerProofs answers the codes the channels ask as proofs: never an event, never kept.
|
|
func answerProofs(h *Holder) {
|
|
handle := func(subject string, body json.RawMessage) (any, error) {
|
|
parts := strings.Split(subject, ".")
|
|
kind := parts[len(parts)-1]
|
|
var c asks.Code
|
|
if err := json.Unmarshal(body, &c); err != nil {
|
|
return asks.ProofAnswer{Words: "The code could not be read."}, nil
|
|
}
|
|
return h.Prove(kind, c), nil
|
|
}
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
if err := stdio.SeatAnswer(asks.ProofSubject(asks.CodeProof, "*"), handle); err == nil {
|
|
logf("[messenger] answering the codes the channels are sent")
|
|
break
|
|
} else {
|
|
logf("[messenger] not answering codes yet (%v); asking again in %s", err, wait)
|
|
}
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// kindsEvery is how often the controller's word on the kinds is read again (a test bundle reads it faster).
|
|
var kindsEvery = 30 * time.Second
|
|
|
|
// followKinds reads, every half minute, which channel kinds the controller says exist and what each
|
|
// promises: never from a channel.
|
|
func followKinds(h *Holder) {
|
|
said := ""
|
|
for {
|
|
kinds, err := stdio.SeatKinds()
|
|
if err != nil {
|
|
if err.Error() != said {
|
|
logf("[messenger] which channels exist cannot be read (%v); asking again", err)
|
|
said = err.Error()
|
|
}
|
|
} else {
|
|
var out []KindInfo
|
|
for _, k := range kinds {
|
|
if k.Seat == asks.ChannelSeat {
|
|
out = append(out, KindInfo{Kind: k.Kind, Module: k.Module, Node: k.Node, Capabilities: k.Capabilities})
|
|
}
|
|
}
|
|
h.Bench.SetKinds(out)
|
|
h.ReleaseHeld() // a held ask is judged again when the kinds change (ADR 0397 §2)
|
|
said = ""
|
|
}
|
|
time.Sleep(kindsEvery)
|
|
}
|
|
}
|
|
|
|
// syncFromController reads the controller's open conditions and gives them to the holder.
|
|
func syncFromController(h *Holder, follow *takeover) {
|
|
asked := h.Now()
|
|
type answer struct {
|
|
raw json.RawMessage
|
|
err error
|
|
}
|
|
done := make(chan answer, 1)
|
|
go func() {
|
|
raw, err := stdio.Ask("seat:"+follow.seat()+".conditions", map[string]any{})
|
|
done <- answer{raw, err}
|
|
}()
|
|
var a answer
|
|
select {
|
|
case a = <-done:
|
|
case <-time.After(30 * time.Second):
|
|
a.err = errors.New("no answer in 30 s")
|
|
}
|
|
if a.err != nil {
|
|
h.SyncFailed(a.err)
|
|
return
|
|
}
|
|
open, err := DecodeOpen(a.raw)
|
|
if err != nil {
|
|
h.SyncFailed(err)
|
|
return
|
|
}
|
|
h.Sync(open, asked)
|
|
}
|
|
|
|
func str(description string) map[string]any {
|
|
return map[string]any{"type": "string", "description": description}
|
|
}
|
|
|
|
func strArg(a map[string]any, k string) string { s, _ := a[k].(string); return strings.TrimSpace(s) }
|
|
|
|
func limitArg(a map[string]any, def int) int {
|
|
if v, ok := a["limit"].(float64); ok && v >= 1 {
|
|
return min(int(v), KeptSends)
|
|
}
|
|
return def
|
|
}
|
|
|
|
func tools(h *Holder, l *listening) []stdio.Tool {
|
|
return []stdio.Tool{
|
|
{Name: "operator-channel.open",
|
|
Description: "What is open now: every message the operator was sent about something still wrong, urgent first, " +
|
|
"oldest first — its key, severity, summary, since when, the channels it went to, whether it is silenced, " +
|
|
"reminded, held (for its burst window, the desktop's rhythm or the cap), refused, not sent yet, or " +
|
|
"never said (known from the state or an old event: history, not news).",
|
|
Run: func(map[string]any) (any, error) { return h.Open(), nil }},
|
|
{Name: "operator-channel.history",
|
|
Description: "What was said to the operator lately, newest first — each send, digest, edit and failure with its " +
|
|
"channel, key and outcome — every message refused for carrying an address, a path or a secret, and what " +
|
|
"waits to be sent on each channel, since when and when its burst window ends.",
|
|
Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many of each (default 50)"}},
|
|
Run: func(a map[string]any) (any, error) { return h.History(limitArg(a, 50)), nil }},
|
|
{Name: "operator-channel.notify",
|
|
Description: "Tell the operator something, as a module that uses the seat: a key (the same key is the same " +
|
|
"message), urgent or warning, one line in the mesh's words, and what it is about; and best, in plain " +
|
|
"words, a headline, an explanation and the line said when it is over. clear says it is over. " +
|
|
"Roles and words only: an address, a path or a secret is refused. Deduplicated, capped and routed " +
|
|
"like the controller's conditions.",
|
|
Input: map[string]any{
|
|
"key": str("what makes it the same message the next time, e.g. backup.ace.failed"),
|
|
"severity": map[string]any{"type": "string", "enum": []string{Urgent, Warning}},
|
|
"summary": str("one line in the mesh's words"),
|
|
"subject": str("what it is about: a machine's role, a module, a plan"),
|
|
"headline": str("a few plain words the operator reads as the title, e.g. backup failed on ace: no ids, keys or commands"),
|
|
"explanation": str("one or two plain sentences: what happened, what it means for the operator, and whether " +
|
|
"they need to act"),
|
|
"resolved": str("the one short line said when it is over, e.g. backups run again on ace"),
|
|
"clear": map[string]any{"type": "boolean", "description": "it is over: the message is edited to say so"},
|
|
},
|
|
Run: func(a map[string]any) (any, error) {
|
|
clear, _ := a["clear"].(bool)
|
|
return h.Notify(strArg(a, "key"), strArg(a, "severity"), strArg(a, "summary"), strArg(a, "subject"), clear,
|
|
strArg(a, "headline"), strArg(a, "explanation"), strArg(a, "resolved"))
|
|
}},
|
|
{Name: "operator-channel.tell",
|
|
Description: "Tell the operator a plain notice that needs no answer and opens no condition (ADR 0380): a " +
|
|
"headline and one or two lines, said at once on every channel, held to the content rule as notify's " +
|
|
"words are, and offering Details like any message. Never escalated, never kept open, never reminded; " +
|
|
"its own cap of 10 a channel an hour, apart from the conditions' and notify's.",
|
|
Input: map[string]any{
|
|
"headline": str("a few plain words the operator reads as the title, e.g. new mail arrived for ace"),
|
|
"lines": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "one or two plain lines"},
|
|
"subject": str("what it is about: a machine's role, a module, a plan"),
|
|
},
|
|
Run: func(a map[string]any) (any, error) {
|
|
var lines []string
|
|
if raw, ok := a["lines"].([]any); ok {
|
|
for _, l := range raw {
|
|
if s, ok := l.(string); ok {
|
|
lines = append(lines, s)
|
|
}
|
|
}
|
|
}
|
|
return h.Tell(strArg(a, "headline"), lines, strArg(a, "subject"))
|
|
}},
|
|
{Name: "messenger_status",
|
|
Description: "Whether the operator can be reached, and why not: each channel kind the mesh holds — can it " +
|
|
"send, what it said of itself, its last error, how many it sent in the last hour against the cap, how " +
|
|
"many it holds — the open and silenced count, messages not sent yet, refusals, unreadable events, " +
|
|
"whether condition events arrive, and the rules it applies.",
|
|
Run: func(map[string]any) (any, error) { return h.Status(l.get()), nil }},
|
|
{Name: "operator-channel.asks",
|
|
Description: "The asks this router holds (novox/hq ADR 0259): open (or held, with since when and why no channel can carry it yet) first, then the last month's — each " +
|
|
"one's asker, headline, state, when it opened and expires, the channels it was offered on, how it " +
|
|
"ended (the warrant: who chose what, through which channel), and whether its asker was told.",
|
|
Run: func(map[string]any) (any, error) { return h.AsksNow(), nil }},
|
|
{Name: "messenger_identities",
|
|
Description: "The accounts linked as the operator, one per channel kind: the account, the name its " +
|
|
"service shows, when it was linked, and from when it can approve (an hour after linking).",
|
|
Run: func(map[string]any) (any, error) { return h.Identities(), nil }},
|
|
{Name: "messenger_unlink",
|
|
Description: "Unlink the account linked as the operator on one channel kind: from then its answers are " +
|
|
"refused until an account is linked again. Said on every channel, the unlinked one included.",
|
|
Input: map[string]any{"kind": str("the channel kind, e.g. telegram")},
|
|
Run: func(a map[string]any) (any, error) {
|
|
kind := strArg(a, "kind")
|
|
if kind == "" {
|
|
return nil, errorf("kind is required")
|
|
}
|
|
return h.Unlink(kind)
|
|
}},
|
|
{Name: "messenger_recent",
|
|
Description: "The recent sends — each message, digest, edit and failure on each channel, newest first — the refusals, " +
|
|
"and what waits to be sent: each message held on a channel (its burst window is 30 s), since when and when its window ends.",
|
|
Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many (default 20)"}},
|
|
Run: func(a map[string]any) (any, error) { return h.History(limitArg(a, 20)), nil }},
|
|
{Name: "messenger_test",
|
|
Description: "Send a test message now, to one channel kind or every one (default), through the content rule " +
|
|
"and the cap: proves a channel reaches the operator. Answers per channel: put on its queue, or why not.",
|
|
Input: map[string]any{
|
|
"channel": str("a channel kind, e.g. telegram, or empty for every one"),
|
|
"text": str("the words (default: a line saying it is a test)"),
|
|
},
|
|
Run: func(a map[string]any) (any, error) { return h.Test(strArg(a, "channel"), strArg(a, "text")), nil }},
|
|
{Name: "messenger_preview",
|
|
Description: "What a condition's message would say, without sending it: the title, the words, the " +
|
|
"line on the issues that track it, and whether the content rule would let them leave the mesh — " +
|
|
"where they may not, the reduced message a send would carry instead. Composed for a condition this " +
|
|
"router holds open, by the walk a send composes with; a clearing as if it cleared now, a silence " +
|
|
"only while there is one. It sends, queues and records nothing.",
|
|
Input: map[string]any{
|
|
"key": str("the condition's key, as the controller names it, e.g. module.openrazer.laptop.relogin-needed"),
|
|
"what": str("which message about it: raised (default), reminder, quiet-reminder, escalated, reopened, cleared, silenced"),
|
|
},
|
|
Run: func(a map[string]any) (any, error) { return h.Preview(strArg(a, "key"), strArg(a, "what")) }},
|
|
{Name: "messenger_check",
|
|
Description: "Would these words be allowed to leave the mesh? Runs the content rule — no address, path or " +
|
|
"secret — and answers allowed, or the class of what it carried. Sends nothing.",
|
|
Input: map[string]any{"text": str("the words to judge")},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if r, ok := Check(strArg(a, "text")); !ok {
|
|
return map[string]any{"allowed": false, "class": r.Class, "carried": r.What}, nil
|
|
}
|
|
return map[string]any{"allowed": true}, nil
|
|
}},
|
|
}
|
|
}
|