diff --git a/go/clients/nodeengineclient/enginereads/calls.go b/go/clients/nodeengineclient/enginereads/calls.go new file mode 100644 index 0000000..dab4040 --- /dev/null +++ b/go/clients/nodeengineclient/enginereads/calls.go @@ -0,0 +1,225 @@ +package enginereads + +import ( + "context" + "maps" + "slices" + "strings" + "sync" + "time" +) + +// CallID is the id of a part's finding about node's manifest-reads: the node the kept call waits on. +func CallID(node string) string { return "manifest-reads." + node } + +// BoundPrefix starts the key of a node's bound in a part's state of bounds, a key that expires at its end. +const BoundPrefix = "reads-call." + +// BoundOf is the node a bound key is of. +func BoundOf(key string) (string, bool) { return strings.CutPrefix(key, BoundPrefix) } + +// UnansweredCall is a node's manifest-reads left unanswered past the bound while its machine is heard: what the part's +// finding call-unanswered says (novox/hq ADR 0668 §3), and At, the part's clock when it is written. +type UnansweredCall struct { + Node string + Since time.Time + Words string + At time.Time +} + +// ReadsCalls is the wait of a part's kept manifest-reads calls through a Keeper, shared by every caller of the verb (the +// assigner, the builder, the indexer): a node whose engine has not answered for Bound while its machine is heard is +// Raised in the part's own findings, and Cleared once it answers (novox/hq ADR 0668 §3). Every step is an event; nothing +// waits on a timer. +// +// With Arm set, a node unanswered is weighed when its bound ends (Expired, on the expiry of the key Arm keeps in the +// part's state of bounds): Heard is asked then, once, and a node not heard has no finding; it is asked again when it +// returns, on manifest-reads-changed (ADR 0668 §4). With no Arm, Heard is the part's own knowledge, asked at each Step. +type ReadsCalls struct { + Keeper *Keeper + // Nodes are the counted nodes. + Nodes func() []string + // Heard says whether node's machine is heard (with Arm, the seat mesh-watchdogs' verb presence); an error is said and + // reads as not heard. + Heard func(ctx context.Context, node string) (bool, error) + // Bound is how long a node may stay unanswered before it is the finding. + Bound time.Duration + // Raise writes the finding of u in the part's findings; Clear removes the finding of node, none being nothing to clear. + Raise func(ctx context.Context, u UnansweredCall) error + Clear func(ctx context.Context, node string) error + // Held are the nodes whose finding the part's findings hold, written by this process or another; nil clears every + // counted node not raised. + Held func(ctx context.Context) ([]string, error) + // Arm keeps a bound for node ending at ends, whose expiry is Expired; nil weighs Heard at each Step. + Arm func(ctx context.Context, node string, ends time.Time) error + Now func() time.Time + Logf func(string, ...any) + + mu sync.Mutex + said map[string]time.Time // node -> the since its finding was written for +} + +func (c *ReadsCalls) now() time.Time { + if c.Now != nil { + return c.Now() + } + return time.Now() +} + +func (c *ReadsCalls) logf(f string, a ...any) { + if c.Logf != nil { + c.Logf(f, a...) + } +} + +func (c *ReadsCalls) ready() bool { + return c != nil && c.Keeper != nil && c.Nodes != nil && c.Heard != nil && c.Raise != nil && c.Clear != nil +} + +// unanswered is node's since while its statement is unanswered. +func unanswered(statements map[string]Statement, node string) (time.Time, bool) { + s, ok := statements[node] + if !ok || s.Answered || s.UnansweredSince.IsZero() { + return time.Time{}, false + } + return s.UnansweredSince, true +} + +func (c *ReadsCalls) heard(ctx context.Context, node string) bool { + ok, err := c.Heard(ctx, node) + if err != nil { + c.logf("whether %s's machine is heard cannot be read, so its unanswered %s is no finding: %v", node, Verb, err) + return false + } + return ok +} + +func (c *ReadsCalls) isSaid(node string, since time.Time) bool { + c.mu.Lock() + defer c.mu.Unlock() + was, ok := c.said[node] + return ok && was.Equal(since) +} + +// raise writes node's finding, unless written for this since already, and asks the node again once its last ask ended. +func (c *ReadsCalls) raise(ctx context.Context, node string, since time.Time, words string) bool { + if !c.Keeper.Asking(node) { + c.Keeper.AskNode(node, "unanswered since "+since.UTC().Format(time.RFC3339)+", its machine heard") + } + if c.isSaid(node, since) { + return true + } + if err := c.Raise(ctx, UnansweredCall{Node: node, Since: since, Words: words, At: c.now().UTC()}); err != nil { + c.logf("the finding call-unanswered of %s's %s is not written: %v", node, Verb, err) + return false + } + c.logf("%s on %s is unanswered since %s and its machine is heard: the finding call-unanswered", Verb, node, since.UTC().Format(time.RFC3339)) + c.mu.Lock() + if c.said == nil { + c.said = map[string]time.Time{} + } + c.said[node] = since + c.mu.Unlock() + return true +} + +// Step weighs every counted node after a statement changed or the counted nodes changed: with no Arm a node unanswered +// and heard is asked again, and past Bound it is the finding; with Arm, a node unanswered and not yet said has its bound +// kept. Every finding no longer raised is cleared. +func (c *ReadsCalls) Step(ctx context.Context) { + if !c.ready() { + return + } + nodes := c.Nodes() + statements := c.Keeper.Statements() + now := c.now() + raised := map[string]bool{} + for _, node := range nodes { + since, ok := unanswered(statements, node) + if !ok { + continue + } + if c.Arm != nil { + if c.isSaid(node, since) { + raised[node] = true + continue + } + if err := c.Arm(ctx, node, since.Add(c.Bound)); err != nil { + c.logf("the bound of %s's %s is not kept: %v", node, Verb, err) + } + continue + } + if !c.heard(ctx, node) { + continue + } + if now.Sub(since) < c.Bound { + if !c.Keeper.Asking(node) { + c.Keeper.AskNode(node, "unanswered since "+since.UTC().Format(time.RFC3339)+", its machine heard") + } + continue + } + raised[node] = c.raise(ctx, node, since, statements[node].Words) + } + c.clear(ctx, nodes, raised) +} + +// Expired weighs node when its bound ended: unanswered past Bound and its machine heard, it is the finding and is asked +// again; not yet past Bound, its bound is kept again; not heard, it has none. +func (c *ReadsCalls) Expired(ctx context.Context, node string) { + if !c.ready() || !slices.Contains(c.Nodes(), node) { + return + } + statements := c.Keeper.Statements() + since, ok := unanswered(statements, node) + if !ok || c.isSaid(node, since) { + return + } + if c.now().Sub(since) < c.Bound { + if c.Arm != nil { + if err := c.Arm(ctx, node, since.Add(c.Bound)); err != nil { + c.logf("the bound of %s's %s is not kept: %v", node, Verb, err) + } + } + return + } + if !c.heard(ctx, node) { + c.logf("%s on %s is unanswered since %s and its machine is not heard: no finding; it is asked again when it returns", Verb, node, since.UTC().Format(time.RFC3339)) + return + } + c.raise(ctx, node, since, statements[node].Words) +} + +// clear removes the finding of every node held or said and not raised: with Held those it names, else every counted node. +func (c *ReadsCalls) clear(ctx context.Context, nodes []string, raised map[string]bool) { + candidates := map[string]bool{} + if c.Held != nil { + held, err := c.Held(ctx) + if err != nil { + c.logf("the findings of %s held cannot be listed: %v", Verb, err) + } + for _, node := range held { + candidates[node] = true + } + } else { + for _, node := range nodes { + candidates[node] = true + } + } + c.mu.Lock() + for node := range c.said { + candidates[node] = true + } + c.mu.Unlock() + for _, node := range slices.Sorted(maps.Keys(candidates)) { + if raised[node] { + continue + } + if err := c.Clear(ctx, node); err != nil { + c.logf("the finding call-unanswered of %s's %s is not cleared: %v", node, Verb, err) + continue + } + c.mu.Lock() + delete(c.said, node) + c.mu.Unlock() + } +} diff --git a/go/clients/nodeengineclient/enginereads/enginereads.go b/go/clients/nodeengineclient/enginereads/enginereads.go index 4ae20b2..78d02d2 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads.go +++ b/go/clients/nodeengineclient/enginereads/enginereads.go @@ -137,6 +137,9 @@ type Statement struct { Answered bool `json:"answered"` Words string `json:"words,omitempty"` AskAgain *Again `json:"ask-again,omitempty"` + // UnansweredSince is when the engine first did not answer, of the asks since its last answer; zero once answered + // (ADR 0668 §3: a kept call unanswered past five minutes is a finding of its caller). + UnansweredSince time.Time `json:"unanswered-since,omitzero"` } // Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state. @@ -170,7 +173,8 @@ func Decode(raw []byte) (Statement, error) { return s, nil } -// Caller calls a seat's verb on one node: process.Conn.CallOn. +// Caller calls a seat's verb on one node: process.Conn.CallKeptOn, a kept call asked again through a transport fault +// within the bus's retry bound (novox/hq ADR 0668 §1; the verb is a read, safe to call again). type Caller func(node, key string, body any) (json.RawMessage, error) // Ask asks node's engine the verb once and says what it answered. An engine that answered the call with an error is @@ -220,6 +224,10 @@ func NodeOf(subject string, data []byte) (string, bool, error) { type Keeper struct { // Call asks the verb; nil asks nothing (every node then reads as the fallback or the baseline). Call Caller + // Once asks the verb once (process.Conn.CallOn). When set, its fault keeps the fallback or the last statement at once, + // marked as not answering, while Call, the kept call, asks on until answered (novox/hq ADR 0668 §3): a part does not + // wait out busretry.Bound for a node whose engine answers nothing before it acts on what it last knew. + Once Caller // Fallback reads a node's record `.node` once where its engine did not answer the verb; nil reads none. Fallback func(node string) ([]string, bool) // Put writes a node's key of the part's state, Delete removes it; nil keeps the statements in memory only. @@ -269,16 +277,47 @@ func (k *Keeper) AskNode(node, why string) { go k.ask(node, why) } +// AskAll asks every node of nodes once, each as AskNode does. +func (k *Keeper) AskAll(nodes []string, why string) { + for _, n := range nodes { + k.AskNode(n, why) + } +} + +// AtBind is what a part registers with its event hub's OnBound (process.EventHub.OnBound): at each bind of its consumer +// every node nodes names is asked once, so a manifest-reads-changed said before the consumer existed, and a statement +// taken while an engine restarted, end at the bind and not at the next event (novox/hq ADR 0668 §4; issue 1966 note 155). +func (k *Keeper) AtBind(nodes func() []string) func(created time.Time) { + return func(created time.Time) { + k.AskAll(nodes(), "the part's consumer was bound, made "+created.UTC().Format(time.RFC3339)) + } +} + func (k *Keeper) ask(node, why string) { defer func() { k.mu.Lock() delete(k.asking, node) k.mu.Unlock() }() + if k.Call != nil && k.Once != nil { + first := Ask(k.Once, node, k.now()) + if first.Answered { + k.take(first) + return + } + k.settle(first, why) + why += "; the kept call asks on after: " + first.Words + } s := Statement{Node: node, AskedAt: k.now().UTC(), From: FromNone, Words: "nothing asks the verb"} if k.Call != nil { s = Ask(k.Call, node, k.now()) } + k.settle(s, why) +} + +// settle takes s, read from the node's record in its place where the engine did not answer. +func (k *Keeper) settle(s Statement, why string) { + node := s.Node if s.From != FromVerb && k.Fallback != nil { if reads, ok := k.Fallback(node); ok { s.ManifestReads, s.From = slices.Clone(reads), FromRecord @@ -292,6 +331,31 @@ func (k *Keeper) ask(node, why string) { k.take(s) } +// Unanswered is, of nodes, each whose engine the kept call has not answered for bound or longer by now while heard says +// its machine is heard, with since when (novox/hq ADR 0668 §3); a node whose machine is away is none. With no Call, none. +func (k *Keeper) Unanswered(nodes []string, heard func(node string) bool, bound time.Duration, now time.Time) map[string]time.Time { + out := map[string]time.Time{} + if k.Call == nil { + return out + } + k.mu.Lock() + defer k.mu.Unlock() + for _, node := range nodes { + s, ok := k.by[node] + if ok && !s.Answered && !s.UnansweredSince.IsZero() && now.Sub(s.UnansweredSince) >= bound && heard(node) { + out[node] = s.UnansweredSince + } + } + return out +} + +// Asking says an ask of node runs now. +func (k *Keeper) Asking(node string) bool { + k.mu.Lock() + defer k.mu.Unlock() + return k.asking[node] +} + // take keeps s, unless the node did not answer and an earlier statement stands: an away node keeps its last one (ADR 0624 // §1), marked as not answering. func (k *Keeper) take(s Statement) { @@ -299,8 +363,14 @@ func (k *Keeper) take(s Statement) { if k.by == nil { k.by = map[string]Statement{} } + if !s.Answered { + s.UnansweredSince = s.AskedAt + if old, ok := k.by[s.Node]; ok && !old.Answered && !old.UnansweredSince.IsZero() { + s.UnansweredSince = old.UnansweredSince + } + } if old, ok := k.by[s.Node]; ok && !s.Answered && old.From != FromNone { - old.Answered, old.AskedAt, old.Words, old.AskAgain = false, s.AskedAt, s.Words, nil + old.Answered, old.AskedAt, old.Words, old.AskAgain, old.UnansweredSince = false, s.AskedAt, s.Words, nil, s.UnansweredSince s = old } k.by[s.Node] = s diff --git a/go/clients/nodeengineclient/enginereads/enginereads_test.go b/go/clients/nodeengineclient/enginereads/enginereads_test.go index b620d1c..607e3d0 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads_test.go +++ b/go/clients/nodeengineclient/enginereads/enginereads_test.go @@ -1,8 +1,12 @@ package enginereads import ( + "context" + "encoding/json" + "errors" "strings" "testing" + "time" ) // The answer and the event are judged as the contract lays them out and its reader checks: the layout's version, a node @@ -34,7 +38,132 @@ func TestTheAnswerAndTheEventAreJudged(t *testing.T) { t.Errorf("event %s: %v, want %q", name, err, c.bad) } } + // Issue 1966 note 155, ADR 0668 §4: a part whose first call of a node timed out, and whose consumer was bound after + // the node's event was said, calls every node once at the bind, and the node is then held as it answered. + calls := map[string]int{} + k := &Keeper{Sync: true, Call: func(node, _ string, _ any) (json.RawMessage, error) { + calls[node]++ + if node == "ace" && calls[node] == 1 { + return nil, context.DeadlineExceeded + } + return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"` + at + `"}`), nil + }} + k.AskNode("ace", "the first view") + if k.Statements()["ace"].Answered { + t.Errorf("a call that timed out reads as answered") + } + k.AtBind(func() []string { return []string{"ace", "g14"} })(time.Unix(0, 0)) + if p := k.Present([]string{"ace", "g14"}); len(p["ace"]) != 1 || len(p["g14"]) != 1 || calls["ace"] != 2 || calls["g14"] != 1 { + t.Errorf("after the bind: present %v, calls %v", p, calls) + } + // Review round 1 (ADR 0668 §3): a node whose engine answers nothing reads as its record at the first fault of the call + // made once, while the kept call asks on; what the kept call answers then stands. + var during Statement + f := &Keeper{Sync: true, Fallback: func(string) ([]string, bool) { return []string{"r.s"}, true }, + Once: func(string, string, any) (json.RawMessage, error) { + return nil, errors.New("nats: no responders available for request") + }} + f.Call = func(node, _ string, _ any) (json.RawMessage, error) { + during = f.Statements()[node] + return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"` + at + `"}`), nil + } + f.AskNode("g14", "the first view") + if during.From != FromRecord || during.Answered || f.Statements()["g14"].From != FromVerb { + t.Errorf("while the kept call asked g14 read %+v; after it %+v", during, f.Statements()["g14"]) + } + // ADR 0668 §3: a node unanswered keeps when it first was not, across asks; past the bound it is unanswered while its + // machine is heard, none while away, and none once it answers. + clock := time.Date(2026, 11, 10, 3, 0, 0, 0, time.UTC) + answers := false + u := &Keeper{Sync: true, Now: func() time.Time { return clock }, Call: func(node, _ string, _ any) (json.RawMessage, error) { + if answers { + return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"` + at + `"}`), nil + } + return nil, context.DeadlineExceeded + }} + first := clock + u.AskNode("ace", "the first view") + clock = clock.Add(6 * time.Minute) + u.AskNode("ace", "asked on") + heard := func(string) bool { return true } + if got := u.Unanswered([]string{"ace"}, heard, 5*time.Minute, clock); !got["ace"].Equal(first) || + len(u.Unanswered([]string{"ace"}, func(string) bool { return false }, 5*time.Minute, clock)) != 0 { + t.Errorf("unanswered past the bound: %v", got) + } + answers = true + u.AskNode("ace", "answered") + if got := u.Unanswered([]string{"ace"}, heard, 5*time.Minute, clock); len(got) != 0 || !u.Statements()["ace"].UnansweredSince.IsZero() { + t.Errorf("an answered node is unanswered: %v", got) + } if Verb != "manifest-reads" || Event != "manifest-reads-changed" || Consumes != "node-engine.manifest-reads-changed" || Invoke != "seat:node-engine.manifest-reads" { t.Errorf("the names are %s %s %s %s", Verb, Event, Consumes, Invoke) } } + +// ADR 0668 §3, ADR 0596 §2 (review round 3): the wait of a part's kept manifest-reads calls. With Arm, a node unanswered +// has its bound kept, and only its expiry asks whether its machine is heard, once: heard, the finding is written at the +// part's clock and the node asked again; away, none; answered, it is cleared. With no Arm, heard is weighed at each step. +func TestReadsCallsSayAnUnansweredNodeHeard(t *testing.T) { + clock := time.Date(2026, 11, 10, 3, 0, 0, 0, time.UTC) + answers, calls := false, 0 + k := &Keeper{Sync: true, Now: func() time.Time { return clock }, Call: func(node, _ string, _ any) (json.RawMessage, error) { + calls++ + if answers { + return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"2026-11-10T03:00:00Z"}`), nil + } + return nil, context.DeadlineExceeded + }} + for name, armed := range map[string]bool{"armed": true, "by its own knowledge": false} { + answers, calls = false, 0 + k.Forget("ace") + first := clock + k.AskNode("ace", "the first view") + heard, asked := true, 0 + raised, cleared := map[string]UnansweredCall{}, map[string]int{} + bounds := map[string]time.Time{} + c := &ReadsCalls{Keeper: k, Nodes: func() []string { return []string{"ace", "g14"} }, Bound: 5 * time.Minute, + Now: func() time.Time { return clock.Add(time.Second) }, + Heard: func(context.Context, string) (bool, error) { asked++; return heard, nil }, + Raise: func(_ context.Context, u UnansweredCall) error { raised[u.Node] = u; return nil }, + Clear: func(_ context.Context, node string) error { cleared[node]++; delete(raised, node); return nil }} + if armed { + c.Arm = func(_ context.Context, node string, ends time.Time) error { bounds[node] = ends; return nil } + } + c.Step(context.Background()) + if armed && (asked != 0 || !bounds["ace"].Equal(first.Add(5*time.Minute)) || len(raised) != 0) { + t.Errorf("%s: a step before the bound asked heard %d times, bounds %v, raised %v", name, asked, bounds, raised) + } + clock = clock.Add(6 * time.Minute) + heard, asked = false, 0 + if armed { + c.Expired(context.Background(), "ace") + } else { + c.Step(context.Background()) + } + if len(raised) != 0 || asked != 1 { + t.Errorf("%s: away, raised %v, heard asked %d times", name, raised, asked) + } + heard, asked, calls = true, 0, 0 + if armed { + c.Expired(context.Background(), "ace") + } else { + c.Step(context.Background()) + } + if u, ok := raised["ace"]; !ok || !u.Since.Equal(first) || !u.At.Equal(clock.Add(time.Second)) || asked != 1 || calls != 1 { + t.Errorf("%s: heard past the bound, raised %+v, heard asked %d, called %d", name, raised, asked, calls) + } + answers = true + k.AskNode("ace", "answered") + c.Step(context.Background()) + if _, ok := raised["ace"]; ok || cleared["ace"] == 0 { + t.Errorf("%s: answered, raised %v, cleared %v", name, raised, cleared) + } + clock = clock.Add(time.Hour) + } + if CallID("ace") != "manifest-reads.ace" { + t.Errorf("the call's id is %s", CallID("ace")) + } + if n, ok := BoundOf(BoundPrefix + "ace"); !ok || n != "ace" { + t.Errorf("the bound's node is %q %v", n, ok) + } +} diff --git a/health-and-repair/go/clients/meshconditionsclient/client_test.go b/health-and-repair/go/clients/meshconditionsclient/client_test.go index f96ebf8..f351fd2 100644 --- a/health-and-repair/go/clients/meshconditionsclient/client_test.go +++ b/health-and-repair/go/clients/meshconditionsclient/client_test.go @@ -19,6 +19,11 @@ func TestAFindingsKeyAndTheOrderOfConditions(t *testing.T) { if k := Key("call", "a b", ""); k != "call.a_b._" { t.Errorf("an unsafe part and an empty token: %s", k) } + // Issue 1966, ADR 0668 §3: a kept call unanswered is a finding the keeper can make a condition of, keyed by what waits. + o := CallUnanswered("ace", "seat:node-engine.manifest-reads", "ace", "no responders", "enginereads", time.Unix(0, 0)) + if f := o.Finding("mesh-delivery-assigner", "r1", time.Unix(300, 0)); o.Key() != "call.ace.call-unanswered" || CheckFinding(f) != nil { + t.Errorf("a call unanswered: key %s, %v", o.Key(), CheckFinding(f)) + } good := seat.FindingsByKeyValue{Key: "machine.g14.silent", Kind: "silent", Severity: Warning, Summary: "x", Source: "S1", Observer: "mesh-delivery-watchman", Subject: seat.Subject{Scope: ScopeMachine, ID: "g14"}} for _, c := range []struct { diff --git a/health-and-repair/go/clients/meshconditionsclient/observation.go b/health-and-repair/go/clients/meshconditionsclient/observation.go index e8bc379..9af47d6 100644 --- a/health-and-repair/go/clients/meshconditionsclient/observation.go +++ b/health-and-repair/go/clients/meshconditionsclient/observation.go @@ -145,3 +145,24 @@ func Clear(b FindingBucket, scope, id, token string) (cleared bool, err error) { } return true, b.Delete(key) } + +// KindCallUnanswered is the kind of a kept call still unanswered past CallUnansweredAfter (novox/hq ADR 0668 §3, issue +// 1966): a call whose answer the caller keeps, asked again through every transport fault, that no answer has ended. The +// caller writes it into its own findings, keeps the wait, and clears it when the call is answered. +const KindCallUnanswered = "call-unanswered" + +// CallUnansweredAfter is how long a kept call stays unanswered before it is a finding (ADR 0668 §3; ADR 0564 §2's bound). +const CallUnansweredAfter = 5 * time.Minute + +// CallUnanswered is the observation of a kept call to callee (a seat verb's or a tool's key, as process.Conn.Call names +// it) about id (what the caller waits on: a node, a merge, a key), unanswered since since, its last fault said. Its key +// is `call..call-unanswered`, cleared by Clear(b, ScopeCall, id, KindCallUnanswered). +func CallUnanswered(id, callee, machine, fault, source string, since time.Time) Observation { + return Observation{Scope: ScopeCall, ID: id, Kind: KindCallUnanswered, Machine: machine, Severity: Warning, + Summary: fmt.Sprintf("the call %s about %s is unanswered since %s", callee, id, since.UTC().Format(time.RFC3339)), + Said: fault, + Headline: "A call whose answer is kept is not answered", + Explanation: fmt.Sprintf("The call %s is made again on every transport fault and on the callee's event; it has "+ + "not been answered since %s. Its last fault: %s.", callee, since.UTC().Format(time.RFC3339), fault), + Resolved: "the call is answered", Source: source} +} diff --git a/health-and-repair/go/seats/meshwatchdogs/contract.go b/health-and-repair/go/seats/meshwatchdogs/contract.go index 286d7dd..7c1ec32 100644 --- a/health-and-repair/go/seats/meshwatchdogs/contract.go +++ b/health-and-repair/go/seats/meshwatchdogs/contract.go @@ -202,8 +202,31 @@ type PresenceInput struct { Machine string `json:"machine,omitempty"` } +// PresenceAnswerKeysItem is a type of the contract. +type PresenceAnswerKeysItem struct { + // or part.. + Key string `json:"key"` + // the key's revision + Revision int64 `json:"revision,omitempty"` + // the key's value, online and its layout's version + Value map[string]any `json:"value"` +} + // PresenceAnswer is what presence answers. -type PresenceAnswer = map[string]any +type PresenceAnswer struct { + // without machine: how many keys the state holds + Holds int64 `json:"holds,omitempty"` + // without machine: each key of the state presence, or part.., with its value + Keys []PresenceAnswerKeysItem `json:"keys,omitempty"` + // the machine asked, with machine + Machine string `json:"machine,omitempty"` + // with machine: whether the watchman hears the machine now; false for a machine it holds no word of + Online bool `json:"online,omitempty"` + // with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of + Since string `json:"since,omitempty"` + // without machine: the state read, presence + State string `json:"state,omitempty"` +} // DeadLettersInput is what dead-letters takes. type DeadLettersInput struct { diff --git a/health-and-repair/seats/mesh-watchdogs/seat.json b/health-and-repair/seats/mesh-watchdogs/seat.json index 35a4341..1efe7a8 100644 --- a/health-and-repair/seats/mesh-watchdogs/seat.json +++ b/health-and-repair/seats/mesh-watchdogs/seat.json @@ -43,7 +43,7 @@ }, { "name": "presence", - "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent.", + "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent. With machine, that machine's presence alone: whether the watchman hears it and since when.", "input": { "properties": { "machine": { @@ -55,7 +55,54 @@ }, "answer": { "type": "object", - "description": "who is connected and who is silent" + "description": "with machine, that machine: online and since; without, every key of the state presence", + "properties": { + "machine": { + "type": "string", + "description": "the machine asked, with machine" + }, + "online": { + "type": "boolean", + "description": "with machine: whether the watchman hears the machine now; false for a machine it holds no word of" + }, + "since": { + "type": "string", + "description": "with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of" + }, + "state": { + "type": "string", + "description": "without machine: the state read, presence" + }, + "holds": { + "type": "integer", + "description": "without machine: how many keys the state holds" + }, + "keys": { + "type": "array", + "description": "without machine: each key of the state presence, or part.., with its value", + "items": { + "type": "object", + "properties": { + "key": { + "type": "string", + "description": " or part.." + }, + "value": { + "type": "object", + "description": "the key's value, online and its layout's version" + }, + "revision": { + "type": "integer", + "description": "the key's revision" + } + }, + "required": [ + "key", + "value" + ] + } + } + } } }, { diff --git a/health-and-repair/ts/seats/meshwatchdogs/src/index.ts b/health-and-repair/ts/seats/meshwatchdogs/src/index.ts index 3a5240a..8f12999 100644 --- a/health-and-repair/ts/seats/meshwatchdogs/src/index.ts +++ b/health-and-repair/ts/seats/meshwatchdogs/src/index.ts @@ -116,8 +116,30 @@ export interface PresenceInput { "machine"?: string; } +export interface PresenceAnswerKeysItem { + /** or part.. */ + "key": string; + /** the key's revision */ + "revision"?: number; + /** the key's value, online and its layout's version */ + "value": Record; +} + /** is what presence answers. */ -export type PresenceAnswer = Record; +export interface PresenceAnswer { + /** without machine: how many keys the state holds */ + "holds"?: number; + /** without machine: each key of the state presence, or part.., with its value */ + "keys"?: PresenceAnswerKeysItem[]; + /** the machine asked, with machine */ + "machine"?: string; + /** with machine: whether the watchman hears the machine now; false for a machine it holds no word of */ + "online"?: boolean; + /** with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of */ + "since"?: string; + /** without machine: the state read, presence */ + "state"?: string; +} /** is what dead-letters takes. */ export interface DeadLettersInput { diff --git a/operator-and-conversation/go/seats/operatorchannel/contract.go b/operator-and-conversation/go/seats/operatorchannel/contract.go index 8fb2a22..3e36498 100644 --- a/operator-and-conversation/go/seats/operatorchannel/contract.go +++ b/operator-and-conversation/go/seats/operatorchannel/contract.go @@ -427,10 +427,33 @@ type TellAnswer = map[string]any // AsksInput is what asks takes. type AsksInput struct { + // the most bytes of the asks in the answer; said when it cuts the answer short + Bound int64 `json:"bound,omitempty"` + // one ask, by its key (.); none for the open asks + Key string `json:"key,omitempty"` + // only the asks opened at or after this time, RFC 3339 + Since string `json:"since,omitempty"` + // open (open and held), ended or all; none is open when no key or since is given, else all + State string `json:"state,omitempty"` +} + +// AsksAnswerAsksItem is a type of the contract. +type AsksAnswerAsksItem struct { + // . + Key string `json:"key"` + // how the ask stands: open, held, or how it ended (answered, expired, cancelled) + State string `json:"state"` } // AsksAnswer is what asks answers. -type AsksAnswer = map[string]any +type AsksAnswer struct { + // each ask the router holds, as its state asks holds it + Asks []AsksAnswerAsksItem `json:"asks"` + // what the bound cut, and how to ask for more; empty when nothing was cut + Cut string `json:"cut,omitempty"` + // how many asks the arguments matched + Matched int64 `json:"matched,omitempty"` +} // AskAccept is the body of ask. type AskAccept = Ask diff --git a/operator-and-conversation/seats/operator-channel/seat.json b/operator-and-conversation/seats/operator-channel/seat.json index 3f33532..2043c1d 100644 --- a/operator-and-conversation/seats/operator-channel/seat.json +++ b/operator-and-conversation/seats/operator-channel/seat.json @@ -116,15 +116,65 @@ }, { "name": "asks", - "description": "The asks the router holds: open first, then the last month's, each with how it ended.", + "description": "The asks the router holds: open first, then the last month's, each with how it ended. With no argument only the open asks, within its default bound; key for one ask, state, since and bound narrow or widen it, and the answer says what it cut.", "input": { "type": "object", - "properties": {}, - "description": "no arguments" + "properties": { + "key": { + "type": "string", + "description": "one ask, by its key (.); none for the open asks" + }, + "state": { + "type": "string", + "description": "open (open and held), ended or all; none is open when no key or since is given, else all" + }, + "since": { + "type": "string", + "description": "only the asks opened at or after this time, RFC 3339" + }, + "bound": { + "type": "integer", + "description": "the most bytes of the asks in the answer; said when it cuts the answer short" + } + } }, "answer": { "type": "object", - "description": "the asks" + "description": "the asks, open first", + "properties": { + "asks": { + "type": "array", + "description": "each ask the router holds, as its state asks holds it", + "items": { + "type": "object", + "properties": { + "key": { + "type": "string", + "description": "." + }, + "state": { + "type": "string", + "description": "how the ask stands: open, held, or how it ended (answered, expired, cancelled)" + } + }, + "required": [ + "key", + "state" + ] + } + }, + "cut": { + "type": "string", + "description": "what the bound cut, and how to ask for more; empty when nothing was cut" + }, + "matched": { + "type": "integer", + "description": "how many asks the arguments matched" + } + }, + "required": [ + "asks" + ] } } ], diff --git a/operator-and-conversation/ts/seats/operatorchannel/src/index.ts b/operator-and-conversation/ts/seats/operatorchannel/src/index.ts index 1331987..ea709f8 100644 --- a/operator-and-conversation/ts/seats/operatorchannel/src/index.ts +++ b/operator-and-conversation/ts/seats/operatorchannel/src/index.ts @@ -364,10 +364,32 @@ export type TellAnswer = Record; /** is what asks takes. */ export interface AsksInput { + /** the most bytes of the asks in the answer; said when it cuts the answer short */ + "bound"?: number; + /** one ask, by its key (.); none for the open asks */ + "key"?: string; + /** only the asks opened at or after this time, RFC 3339 */ + "since"?: string; + /** open (open and held), ended or all; none is open when no key or since is given, else all */ + "state"?: string; +} + +export interface AsksAnswerAsksItem { + /** . */ + "key": string; + /** how the ask stands: open, held, or how it ended (answered, expired, cancelled) */ + "state": string; } /** is what asks answers. */ -export type AsksAnswer = Record; +export interface AsksAnswer { + /** each ask the router holds, as its state asks holds it */ + "asks": AsksAnswerAsksItem[]; + /** what the bound cut, and how to ask for more; empty when nothing was cut */ + "cut"?: string; + /** how many asks the arguments matched */ + "matched"?: number; +} /** is the body of ask. */ export type AskAccept = Ask;