Engine reads keep the last answer and say an unanswered node; operator-channel asks and watchdogs presence declared (ADR 0668; issue 1966; group fix/1966-call-made-once) #3
@@ -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()
|
||||
}
|
||||
}
|
||||
@@ -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>.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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.<id>.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}
|
||||
}
|
||||
|
||||
@@ -202,8 +202,31 @@ type PresenceInput struct {
|
||||
Machine string `json:"machine,omitempty"`
|
||||
}
|
||||
|
||||
// PresenceAnswerKeysItem is a type of the contract.
|
||||
type PresenceAnswerKeysItem struct {
|
||||
// <node> or part.<part>.<instance>
|
||||
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, <node> or part.<part>.<instance>, 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 {
|
||||
|
||||
@@ -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, <node> or part.<part>.<instance>, with its value",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "<node> or part.<part>.<instance>"
|
||||
},
|
||||
"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"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
|
||||
@@ -116,8 +116,30 @@ export interface PresenceInput {
|
||||
"machine"?: string;
|
||||
}
|
||||
|
||||
export interface PresenceAnswerKeysItem {
|
||||
/** <node> or part.<part>.<instance> */
|
||||
"key": string;
|
||||
/** the key's revision */
|
||||
"revision"?: number;
|
||||
/** the key's value, online and its layout's version */
|
||||
"value": Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** is what presence answers. */
|
||||
export type PresenceAnswer = Record<string, unknown>;
|
||||
export interface PresenceAnswer {
|
||||
/** without machine: how many keys the state holds */
|
||||
"holds"?: number;
|
||||
/** without machine: each key of the state presence, <node> or part.<part>.<instance>, 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 {
|
||||
|
||||
@@ -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 (<asker>.<id>); 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 {
|
||||
// <asker>.<id>
|
||||
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
|
||||
|
||||
@@ -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 (<asker>.<id>); 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": "<asker>.<id>"
|
||||
},
|
||||
"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"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
|
||||
@@ -364,10 +364,32 @@ export type TellAnswer = Record<string, unknown>;
|
||||
|
||||
/** 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 (<asker>.<id>); 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 {
|
||||
/** <asker>.<id> */
|
||||
"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<string, unknown>;
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user