Author SHA1 Message Date
mesh-tools 5fd0bf86f1 Merge pull request 'Registry recovery: mesh-broker's list, show and orphans answer each binding (issue 1966)' (#5) from fix/1966-bindings-first-view into main 2026-11-11 10:32:31 +00:00
jschoubben 0167b24b95 mesh-broker's fixtures take a keeper ledger plan's binding with a type and an index, so types survive the round trip
mesh/check checking 0167b24b9564 on 5fd0bf86f120 | novox/mesh-contracts#5 in group fix/1966-bindings-first-view
mesh/delivery-group group fix/1966-bindings-first-view merged: every member is merged
mesh/delivery merged: source.main moved to novox/mesh-contracts@5fd0bf86f120
Builds: design 25 bindings-first-view §2

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-11 10:54:37 +01:00
jschoubben ed12ffb387 mesh-broker's Binding is the wire form and Held carries it, so list, show and orphans answer the binding
The seat's Binding takes the fields the events carry and the membership holds
(local, storage, template, stream, subject, bucket, object-store, history,
state, indexes, snapshot, types): one form in the events, in list, show and
orphans. Held gains binding, present on an entry the bus holds. Definition,
fixture and generated Go and TypeScript contract.

Builds: design 25 bindings-first-view §2, §7 (hq d0760629, PR 543)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-11 10:14:55 +01:00
mesh-tools f5d76e84ca Merge pull request 'The node-engine seat's client asks a node once more when an event came while an ask ran, never dropping it (issue 1966 note 170; group with mesh-delivery)' (#4) from fix/1966-engine-reads-asked-again into main 2026-11-10 10:45:49 +00:00
jschoubben c9ffed7042 S1-S3: the node-engine seat's client asks a node once more when an ask came while one ran, and shows when the event was heard (issue 1966 note 170)
mesh/delivery merged: source.main moved to novox/mesh-contracts@f5d76e84ca71
mesh/delivery-group group fix/1966-engine-reads-asked-again merged: every member is merged
mesh/check checking c9ffed70421a on f5d76e84ca71 | novox/mesh-contracts#4 in group fix/1966-engine-reads-asked-again
An event of manifest-reads-changed heard while an ask of the same node ran was dropped
without a line, so a restated engine could stay read at its first statement (ace,
2026-11-10 07:36). Every reader goes through Keeper.AskNode, so the assigner, the
builder and the indexer are fixed at once (ADR 0668 §4: the caller hears it and calls
again; ADR 0634 §1: the seat's client).

- AskNode: an ask while one runs is asked once more after it, one follow-up however
  many came, said in one line (1966/AC170-1, AC170-3). The asks of one node so run one
  after another, and take keeps the last answer, as before (no rule on the engine's
  clock: review round 1).
- Statement.EventAt (event-at): when the node's event was last heard, read by the
  parts' engine-reads tools; forgotten with the node's key.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 11:42:05 +01:00
mesh-tools 7fbfd77a76 Merge pull request '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) from fix/1966-call-made-once into main 2026-11-10 07:20:28 +00:00
jschoubben bb16f99e2d C7: the node-engine seat's client keeps the wait of every caller's kept manifest-reads calls, ReadsCalls beside Keeper.Unanswered, its bound's end asking once whether the machine is heard (issue 1966, ADR 0668 §3, ADR 0634 §1, ADR 0596 §2; review round 3, findings 1 and 2)
mesh/check checking bb16f99e2d5e on 7fbfd77a76cb | novox/mesh-contracts#3 in group fix/1966-call-made-once
mesh/delivery merged: source.main moved to novox/mesh-contracts@7fbfd77a76cb
mesh/delivery-group group fix/1966-call-made-once merged: every member is merged
The assigner, the builder and the indexer share it instead of two copies; it names no other seat's client (the client check), so each caller writes its finding in mesh-conditions' layout through Raise, stamped with ReadsCalls' clock.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 04:54:06 +01:00
jschoubben 11a36eb88c C6: the watchdogs seat declares presence's answer: with machine, that machine's online and since; without, the keys of the state presence (issue 1966, ADR 0596 §2, ADR 0668 §3; review round 3, finding 1)
The builder and the indexer ask whether a node's machine is heard through this verb instead of reading the watchman's state; regenerated with seatgen, whose -check names nothing new.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 04:52:53 +01:00
jschoubben 0ae7cb7a76 C5: the operator-channel seat declares its verb asks' input (key, state, since, bound) and answer (asks with key and state, cut, matched), as the router serves it (issue 1966, ADR 0668 §4; review round 1, finding 5)
The assigner's read at its consumer's bind calls the verb with them; regenerated with seatgen, whose -check passes.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 04:20:04 +01:00
jschoubben c6689bfdb3 C4: a statement keeps since when its engine is unanswered, and the keeper names each node unanswered past a bound while heard (issue 1966, ADR 0668 §3; review round 1, finding 4)
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 04:15:01 +01:00
jschoubben c584f4570b C3: the engine-reads keeper keeps the fallback or the last statement at the first fault while the kept call asks on (issue 1966, ADR 0668 §3; review round 1, finding 1)
A part whose node answers nothing waited out busretry.Bound (30 s) before it acted on the node's record: Once asks the verb once, and its fault is taken at once.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 04:11:07 +01:00
jschoubben e1303f1542 C2: the finding kind call-unanswered of the seat mesh-conditions (issue 1966, ADR 0668 §3)
A kept call still unanswered past five minutes is said in the caller's findings, keyed call.<id>.call-unanswered.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 02:54:44 +01:00
jschoubben 5c29bea0b3 C1: the engine-reads keeper calls every node again at its part's consumer bind (issue 1966, ADR 0668)
A manifest-reads-changed said before the part's consumer existed was never heard (note 155), and the
statement taken while an engine restarted stood. AtBind is registered on the hub's OnBound.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-10 02:54:44 +01:00
mesh-tools 66e0c611bf Merge pull request 'Record a merge through the merge gate (issue 1966, bootstrap)' (#2) from fix/1966-gate-record into main 2026-11-07 20:19:51 +00:00
jschoubben 1aa5dee992 Record a merge through the merge gate so the new indexer reads this repository's seat definitions on its merge (issue 1966, the operator's bootstrap)
mesh/delivery merged: source.main moved to novox/mesh-contracts@66e0c611bf83
mesh/delivery-group group fix/1966-gate-record merged: every member is merged
mesh/check pass 1aa5dee99226 on 66e0c611bf83 | carried from 08580d25d2dc, mesh-networking 5af7189cad82 | unit 14 run, 270 kept | integration 0 run, 27…
2026-11-07 21:13:54 +01:00
mesh-tools 08580d25d2 Merge pull request 'Registry milestone: mesh-contracts (issue 1966)' (#1) from feat/registry-milestone into main 2026-11-07 14:19:16 +00:00
17 changed files with 965 additions and 43 deletions
@@ -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,12 @@ type Statement struct {
Answered bool `json:"answered"` Answered bool `json:"answered"`
Words string `json:"words,omitempty"` Words string `json:"words,omitempty"`
AskAgain *Again `json:"ask-again,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"`
// EventAt is when the part last heard the seat's event manifest-reads-changed for the node, beside when it asked and
// what the engine stated then (issue 1966 note 170: was the event heard, and was it asked again after it).
EventAt time.Time `json:"event-at,omitzero"`
} }
// Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state. // 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 +176,8 @@ func Decode(raw []byte) (Statement, error) {
return s, nil 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) 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 // 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 +227,10 @@ func NodeOf(subject string, data []byte) (string, bool, error) {
type Keeper struct { type Keeper struct {
// Call asks the verb; nil asks nothing (every node then reads as the fallback or the baseline). // Call asks the verb; nil asks nothing (every node then reads as the fallback or the baseline).
Call Caller 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 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) 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. // Put writes a node's key of the part's state, Delete removes it; nil keeps the statements in memory only.
@@ -235,6 +246,11 @@ type Keeper struct {
mu sync.Mutex mu sync.Mutex
by map[string]Statement by map[string]Statement
asking map[string]bool asking map[string]bool
// again is, per node an ask of which runs, the newest why of an ask that came meanwhile: it is asked once more when
// the running ask ends, never dropped (issue 1966 note 170).
again map[string]string
// eventAt is, per node, when its event manifest-reads-changed was last heard (Statement.EventAt).
eventAt map[string]time.Time
} }
func (k *Keeper) logf(f string, a ...any) { func (k *Keeper) logf(f string, a ...any) {
@@ -250,14 +266,25 @@ func (k *Keeper) now() time.Time {
return time.Now() return time.Now()
} }
// AskNode asks node's engine, unless an ask of it runs already; why is said in the log. // AskNode asks node's engine; why is said in the log. When an ask of it runs already, the node is asked once more after
// that ask ends, one follow-up however many asks came meanwhile, so an event heard while an ask runs is never lost (issue
// 1966 note 170; ADR 0668 §4: the caller hears it and calls again). A why that starts with Event is the seat's event,
// whose time the node's statement shows as event-at.
func (k *Keeper) AskNode(node, why string) { func (k *Keeper) AskNode(node, why string) {
k.mu.Lock() k.mu.Lock()
if k.asking == nil { if k.asking == nil {
k.asking = map[string]bool{} k.asking, k.again, k.eventAt = map[string]bool{}, map[string]string{}, map[string]time.Time{}
}
if strings.HasPrefix(why, Event) {
k.eventAt[node] = k.now().UTC()
} }
if k.asking[node] { if k.asking[node] {
_, waits := k.again[node]
k.again[node] = why
k.mu.Unlock() k.mu.Unlock()
if !waits {
k.logf("%s on %s: an ask runs, so it is asked once more after it (%s)", Verb, node, why)
}
return return
} }
k.asking[node] = true k.asking[node] = true
@@ -269,16 +296,60 @@ func (k *Keeper) AskNode(node, why string) {
go k.ask(node, why) 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))
}
}
// ask asks node, and once more for as long as an ask came while it ran (AskNode); only then is the node no longer asking.
func (k *Keeper) ask(node, why string) { func (k *Keeper) ask(node, why string) {
defer func() { for {
k.askOnce(node, why)
k.mu.Lock() k.mu.Lock()
delete(k.asking, node) next, again := k.again[node]
delete(k.again, node)
if !again {
delete(k.asking, node)
}
k.mu.Unlock() k.mu.Unlock()
}() if !again {
return
}
why = "asked again: " + next
}
}
func (k *Keeper) askOnce(node, why string) {
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"} s := Statement{Node: node, AskedAt: k.now().UTC(), From: FromNone, Words: "nothing asks the verb"}
if k.Call != nil { if k.Call != nil {
s = Ask(k.Call, node, k.now()) 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 s.From != FromVerb && k.Fallback != nil {
if reads, ok := k.Fallback(node); ok { if reads, ok := k.Fallback(node); ok {
s.ManifestReads, s.From = slices.Clone(reads), FromRecord s.ManifestReads, s.From = slices.Clone(reads), FromRecord
@@ -292,15 +363,53 @@ func (k *Keeper) ask(node, why string) {
k.take(s) 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 // 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. // §1), marked as not answering. The asks of one node run one after another (AskNode), so the last answer is the newest.
// The statement shows when the node's event was last heard.
func (k *Keeper) take(s Statement) { func (k *Keeper) take(s Statement) {
k.mu.Lock() k.mu.Lock()
if k.by == nil { if k.by == nil {
k.by = map[string]Statement{} k.by = map[string]Statement{}
} }
if old, ok := k.by[s.Node]; ok && old.EventAt.After(s.EventAt) {
s.EventAt = old.EventAt
}
if at := k.eventAt[s.Node]; at.After(s.EventAt) {
s.EventAt = at
}
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 { 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 s = old
} }
k.by[s.Node] = s k.by[s.Node] = s
@@ -326,6 +435,7 @@ func (k *Keeper) Heard(node string, raw []byte, deleted bool) {
k.mu.Lock() k.mu.Lock()
_, had := k.by[node] _, had := k.by[node]
delete(k.by, node) delete(k.by, node)
delete(k.eventAt, node)
k.mu.Unlock() k.mu.Unlock()
if had && k.Changed != nil { if had && k.Changed != nil {
k.Changed() k.Changed()
@@ -360,6 +470,7 @@ func (k *Keeper) Forget(node string) {
k.mu.Lock() k.mu.Lock()
_, had := k.by[node] _, had := k.by[node]
delete(k.by, node) delete(k.by, node)
delete(k.eventAt, node)
k.mu.Unlock() k.mu.Unlock()
if had && k.Delete != nil { if had && k.Delete != nil {
if err := k.Delete(node); err != nil { if err := k.Delete(node); err != nil {
@@ -1,8 +1,13 @@
package enginereads package enginereads
import ( import (
"context"
"encoding/json"
"errors"
"fmt"
"strings" "strings"
"testing" "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 // 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 +39,174 @@ func TestTheAnswerAndTheEventAreJudged(t *testing.T) {
t.Errorf("event %s: %v, want %q", name, err, c.bad) 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)
}
// Issue 1966 note 170, 1966/AC170-1 and AC170-3: an event of a node heard while an ask of it runs is not dropped: the
// node is asked once more after that ask ends (one follow-up however many came), said in one line, and the newer
// statement is kept; the event's time is shown beside the statement (event-at).
const stated26, restated = "2026-11-10T07:36:26Z", "2026-11-10T07:36:30Z"
var lines []string
eclock := time.Date(2026, 11, 10, 7, 36, 26, 0, time.UTC)
asks := 0
var e *Keeper
e = &Keeper{Sync: true, Now: func() time.Time { return eclock },
Logf: func(f string, a ...any) { lines = append(lines, fmt.Sprintf(f, a...)) },
Put: func(_ string, raw []byte) error { _, err := Decode(raw); return err },
Call: func(node, _ string, _ any) (json.RawMessage, error) {
asks++
stated := restated
if asks == 1 {
eclock = eclock.Add(4 * time.Second)
e.AskNode(node, Event)
e.AskNode(node, Event)
stated = stated26
} else if got := e.Statements()[node].StatedAt.Format(time.RFC3339); got != stated26 {
t.Errorf("the second ask began before the first was taken in: held %s", got)
}
return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"` + stated + `"}`), nil
}}
e.AskNode("ace", Event)
held := e.Statements()["ace"]
if asks != 2 || held.StatedAt.Format(time.RFC3339) != restated || e.Asking("ace") {
t.Errorf("an event during an ask: asked %d times, held stated at %s, asking %v", asks, held.StatedAt, e.Asking("ace"))
}
if again := strings.Count(strings.Join(lines, "\n"), "asked once more"); again != 1 {
t.Errorf("the ask kept for later is said in %d lines, not one: %q", again, lines)
}
if !held.EventAt.Equal(eclock) || !strings.Contains(string(must(Encode(held))), `"event-at":"2026-11-10T07:36:30Z"`) {
t.Errorf("the event's time is not shown: %+v", held)
}
if Verb != "manifest-reads" || Event != "manifest-reads-changed" || Consumes != "node-engine.manifest-reads-changed" || Invoke != "seat:node-engine.manifest-reads" { 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) 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)
}
}
func must(b []byte, err error) []byte {
if err != nil {
panic(err)
}
return b
}
+25 -7
View File
@@ -72,16 +72,32 @@ const SubjectRecreated = "mesh.seat.mesh-broker.event.recreated"
// SubjectDropped is where dropped is published. // SubjectDropped is where dropped is published.
const SubjectDropped = "mesh.seat.mesh-broker.event.dropped" const SubjectDropped = "mesh.seat.mesh-broker.event.dropped"
// Binding is the binding a module receives in its membership for one entry (ADR 0595 §1, §7). // Binding is the binding a module receives in its membership for one entry (ADR 0595 §1, §7): the names on the bus of the entry's objects, one form in the events, in list, show and orphans, and in the membership.
type Binding struct { type Binding struct {
// the entry's name on the bus // a key-value bucket's name
BusName string `json:"bus-name"` Bucket string `json:"bucket,omitempty"`
// how the bus keeps it // a ledger's history stream
StorageType string `json:"storage-type"` History string `json:"history,omitempty"`
// its subjects, for a stream // a ledger's index buckets, by index name
Subjects []string `json:"subjects,omitempty"` Indexes map[string]string `json:"indexes,omitempty"`
// the entry's local name in the module's manifest
Local string `json:"local"`
// an object store's name
ObjectStore string `json:"object-store,omitempty"`
// the subject of a ledger's instances' snapshots
Snapshot string `json:"snapshot,omitempty"`
// a ledger's state bucket
State string `json:"state,omitempty"`
// how the bus keeps it: the entry's storage type
Storage string `json:"storage"`
// a stream's name on the bus
Stream string `json:"stream,omitempty"`
// the subject a stream's messages are on; also a ledger's history's
Subject string `json:"subject,omitempty"`
// its template // its template
Template string `json:"template,omitempty"` Template string `json:"template,omitempty"`
// a ledger's types, as its manifest declares them
Types []map[string]any `json:"types,omitempty"`
} }
// Entry is one piece of bus state as a manifest asks it. // Entry is one piece of bus state as a manifest asks it.
@@ -131,6 +147,8 @@ type Fill struct {
// Held is one entry as the bus holds it. // Held is one entry as the bus holds it.
type Held struct { type Held struct {
// the binding a module receives for the entry, as created and recreated carry it; present on an entry whose objects the bus holds, absent on one it does not (design 25 bindings-first-view §2)
Binding *Binding `json:"binding,omitempty"`
// its name on the bus, <module>_<local name> // its name on the bus, <module>_<local name>
BusName string `json:"bus-name"` BusName string `json:"bus-name"`
// its own configuration, where no template fits // its own configuration, where no template fits
+4
View File
@@ -19,6 +19,8 @@ func TestFixtures(t *testing.T) {
}{ }{
{name: "created on the seat's subject", subject: "mesh.seat.mesh-broker.event.created", given: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\"}", wire: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\"}", {name: "created on the seat's subject", subject: "mesh.seat.mesh-broker.event.created", given: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\"}", wire: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\"}",
decode: func(b []byte) (any, error) { var v CreatedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-broker\\.event\\.created$")}, decode: func(b []byte) (any, error) { var v CreatedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-broker\\.event\\.created$")},
{name: "created with a ledger's binding as the bus module writes it", subject: "mesh.seat.mesh-broker.event.created", given: "{\"binding\":{\"history\":\"mesh-knowledge_ledger\",\"indexes\":{\"by-kind\":\"mesh-knowledge_ledger_index-by-kind\"},\"local\":\"ledger\",\"snapshot\":\"mesh.ledger.mesh-knowledge.ledger.snapshot\",\"state\":\"mesh-knowledge_ledger_state\",\"storage\":\"ledger\",\"subject\":\"mesh.ledger.mesh-knowledge.ledger.history\",\"template\":\"ledger\",\"types\":[{\"indexes\":[{\"name\":\"by-kind\",\"on\":[\"kind\"]}],\"name\":\"decision\",\"operations\":[{\"effect\":\"set-fields\",\"fields\":[\"title\",\"kind\"],\"name\":\"set\"}],\"schema\":{\"kind\":{\"type\":\"string\"},\"title\":{\"type\":\"string\"}}}]},\"bus-name\":\"mesh-knowledge_ledger\",\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"revision\":7,\"storage-type\":\"ledger\",\"template\":\"ledger\"}", wire: "{\"binding\":{\"history\":\"mesh-knowledge_ledger\",\"indexes\":{\"by-kind\":\"mesh-knowledge_ledger_index-by-kind\"},\"local\":\"ledger\",\"snapshot\":\"mesh.ledger.mesh-knowledge.ledger.snapshot\",\"state\":\"mesh-knowledge_ledger_state\",\"storage\":\"ledger\",\"subject\":\"mesh.ledger.mesh-knowledge.ledger.history\",\"template\":\"ledger\",\"types\":[{\"indexes\":[{\"name\":\"by-kind\",\"on\":[\"kind\"]}],\"name\":\"decision\",\"operations\":[{\"effect\":\"set-fields\",\"fields\":[\"title\",\"kind\"],\"name\":\"set\"}],\"schema\":{\"kind\":{\"type\":\"string\"},\"title\":{\"type\":\"string\"}}}]},\"bus-name\":\"mesh-knowledge_ledger\",\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"revision\":7,\"storage-type\":\"ledger\",\"template\":\"ledger\"}",
decode: func(b []byte) (any, error) { var v CreatedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-broker\\.event\\.created$")},
{name: "refused on the seat's subject", subject: "mesh.seat.mesh-broker.event.refused", given: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}", wire: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}", {name: "refused on the seat's subject", subject: "mesh.seat.mesh-broker.event.refused", given: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}", wire: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}",
decode: func(b []byte) (any, error) { var v RefusedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-broker\\.event\\.refused$")}, decode: func(b []byte) (any, error) { var v RefusedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-broker\\.event\\.refused$")},
{name: "drifted on the seat's subject", subject: "mesh.seat.mesh-broker.event.drifted", given: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}", wire: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}", {name: "drifted on the seat's subject", subject: "mesh.seat.mesh-broker.event.drifted", given: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}", wire: "{\"asked\":{\"max-mib\":64},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"held\":{\"max-mib\":64},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"storage-type\":\"ledger\",\"template\":\"ledger\",\"why\":\"the bus refused it\"}",
@@ -41,6 +43,8 @@ func TestFixtures(t *testing.T) {
decode: func(b []byte) (any, error) { var v DropInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil}, decode: func(b []byte) (any, error) { var v DropInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "list answers every entry", subject: "", given: "{\"entries\":[{\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"options\":{\"max-mib\":64},\"settings\":{\"max-mib\":64},\"storage-type\":\"ledger\",\"template\":\"ledger\"}]}", wire: "{\"entries\":[{\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"options\":{\"max-mib\":64},\"settings\":{\"max-mib\":64},\"storage-type\":\"ledger\",\"template\":\"ledger\"}]}", {name: "list answers every entry", subject: "", given: "{\"entries\":[{\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"options\":{\"max-mib\":64},\"settings\":{\"max-mib\":64},\"storage-type\":\"ledger\",\"template\":\"ledger\"}]}", wire: "{\"entries\":[{\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"options\":{\"max-mib\":64},\"settings\":{\"max-mib\":64},\"storage-type\":\"ledger\",\"template\":\"ledger\"}]}",
decode: func(b []byte) (any, error) { var v ListAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil}, decode: func(b []byte) (any, error) { var v ListAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
{name: "list answers the binding of an entry the bus holds", subject: "", given: "{\"entries\":[{\"binding\":{\"history\":\"mesh-knowledge_ledger\",\"indexes\":{\"by-kind\":\"mesh-knowledge_ledger_index-by-kind\"},\"local\":\"ledger\",\"snapshot\":\"mesh.ledger.mesh-knowledge.ledger.snapshot\",\"state\":\"mesh-knowledge_ledger_state\",\"storage\":\"ledger\",\"subject\":\"mesh.ledger.mesh-knowledge.ledger.history\",\"template\":\"ledger\",\"types\":[{\"indexes\":[{\"name\":\"by-kind\",\"on\":[\"kind\"]}],\"name\":\"decision\",\"operations\":[{\"effect\":\"set-fields\",\"fields\":[\"title\",\"kind\"],\"name\":\"set\"}],\"schema\":{\"kind\":{\"type\":\"string\"},\"title\":{\"type\":\"string\"}}}]},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"settings\":{\"max-mib\":64},\"storage-type\":\"ledger\",\"template\":\"ledger\"}]}", wire: "{\"entries\":[{\"binding\":{\"history\":\"mesh-knowledge_ledger\",\"indexes\":{\"by-kind\":\"mesh-knowledge_ledger_index-by-kind\"},\"local\":\"ledger\",\"snapshot\":\"mesh.ledger.mesh-knowledge.ledger.snapshot\",\"state\":\"mesh-knowledge_ledger_state\",\"storage\":\"ledger\",\"subject\":\"mesh.ledger.mesh-knowledge.ledger.history\",\"template\":\"ledger\",\"types\":[{\"indexes\":[{\"name\":\"by-kind\",\"on\":[\"kind\"]}],\"name\":\"decision\",\"operations\":[{\"effect\":\"set-fields\",\"fields\":[\"title\",\"kind\"],\"name\":\"set\"}],\"schema\":{\"kind\":{\"type\":\"string\"},\"title\":{\"type\":\"string\"}}}]},\"bus-name\":\"mesh-knowledge_ledger\",\"fill\":{\"bytes\":2048,\"cap\":67108864,\"count\":10},\"local-name\":\"ledger\",\"module\":\"mesh-knowledge\",\"settings\":{\"max-mib\":64},\"storage-type\":\"ledger\",\"template\":\"ledger\"}]}",
decode: func(b []byte) (any, error) { var v ListAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
} }
for _, c := range cases { for _, c := range cases {
v, err := c.decode([]byte(c.given)) v, err := c.decode([]byte(c.given))
@@ -19,6 +19,11 @@ func TestAFindingsKeyAndTheOrderOfConditions(t *testing.T) {
if k := Key("call", "a b", ""); k != "call.a_b._" { if k := Key("call", "a b", ""); k != "call.a_b._" {
t.Errorf("an unsafe part and an empty token: %s", k) 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", 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"}} Observer: "mesh-delivery-watchman", Subject: seat.Subject{Scope: ScopeMachine, ID: "g14"}}
for _, c := range []struct { 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) 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"` 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. // 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. // DeadLettersInput is what dead-letters takes.
type DeadLettersInput struct { type DeadLettersInput struct {
@@ -43,7 +43,7 @@
}, },
{ {
"name": "presence", "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": { "input": {
"properties": { "properties": {
"machine": { "machine": {
@@ -55,7 +55,54 @@
}, },
"answer": { "answer": {
"type": "object", "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; "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. */ /** 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. */ /** is what dead-letters takes. */
export interface DeadLettersInput { export interface DeadLettersInput {
@@ -427,10 +427,33 @@ type TellAnswer = map[string]any
// AsksInput is what asks takes. // AsksInput is what asks takes.
type AsksInput struct { 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. // 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. // AskAccept is the body of ask.
type AskAccept = Ask type AskAccept = Ask
@@ -116,15 +116,65 @@
}, },
{ {
"name": "asks", "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": { "input": {
"type": "object", "type": "object",
"properties": {}, "properties": {
"description": "no arguments" "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": { "answer": {
"type": "object", "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. */ /** is what asks takes. */
export interface AsksInput { 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. */ /** 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. */ /** is the body of ask. */
export type AskAccept = Ask; export type AskAccept = Ask;
+56
View File
@@ -22,6 +22,62 @@
} }
} }
}, },
{
"name": "created with a ledger's binding as the bus module writes it",
"of": "event:created",
"subject": "mesh.seat.mesh-broker.event.created",
"given": {
"module": "mesh-knowledge",
"local-name": "ledger",
"storage-type": "ledger",
"template": "ledger",
"bus-name": "mesh-knowledge_ledger",
"binding": {
"local": "ledger",
"storage": "ledger",
"template": "ledger",
"subject": "mesh.ledger.mesh-knowledge.ledger.history",
"history": "mesh-knowledge_ledger",
"state": "mesh-knowledge_ledger_state",
"indexes": {
"by-kind": "mesh-knowledge_ledger_index-by-kind"
},
"snapshot": "mesh.ledger.mesh-knowledge.ledger.snapshot",
"types": [
{
"name": "decision",
"schema": {
"title": {
"type": "string"
},
"kind": {
"type": "string"
}
},
"operations": [
{
"name": "set",
"effect": "set-fields",
"fields": [
"title",
"kind"
]
}
],
"indexes": [
{
"name": "by-kind",
"on": [
"kind"
]
}
]
}
]
},
"revision": 7
}
},
{ {
"name": "refused on the seat's subject", "name": "refused on the seat's subject",
"of": "event:refused", "of": "event:refused",
+66
View File
@@ -47,5 +47,71 @@
} }
] ]
} }
},
{
"name": "list answers the binding of an entry the bus holds",
"of": "verb:list.answer",
"given": {
"entries": [
{
"module": "mesh-knowledge",
"local-name": "ledger",
"storage-type": "ledger",
"template": "ledger",
"bus-name": "mesh-knowledge_ledger",
"settings": {
"max-mib": 64
},
"fill": {
"count": 10,
"bytes": 2048,
"cap": 67108864
},
"binding": {
"local": "ledger",
"storage": "ledger",
"template": "ledger",
"subject": "mesh.ledger.mesh-knowledge.ledger.history",
"history": "mesh-knowledge_ledger",
"state": "mesh-knowledge_ledger_state",
"indexes": {
"by-kind": "mesh-knowledge_ledger_index-by-kind"
},
"snapshot": "mesh.ledger.mesh-knowledge.ledger.snapshot",
"types": [
{
"name": "decision",
"schema": {
"title": {
"type": "string"
},
"kind": {
"type": "string"
}
},
"operations": [
{
"name": "set",
"effect": "set-fields",
"fields": [
"title",
"kind"
]
}
],
"indexes": [
{
"name": "by-kind",
"on": [
"kind"
]
}
]
}
]
}
}
]
}
} }
] ]
+50 -11
View File
@@ -403,6 +403,10 @@
"why": { "why": {
"type": "string", "type": "string",
"description": "why it is in its state, in plain words; empty when held as asked" "description": "why it is in its state, in plain words; empty when held as asked"
},
"binding": {
"$ref": "#/defs/Binding",
"description": "the binding a module receives for the entry, as created and recreated carry it; present on an entry whose objects the bus holds, absent on one it does not (design 25 bindings-first-view §2)"
} }
}, },
"required": [ "required": [
@@ -540,31 +544,66 @@
"Binding": { "Binding": {
"type": "object", "type": "object",
"properties": { "properties": {
"bus-name": { "local": {
"type": "string", "type": "string",
"description": "the entry's name on the bus" "description": "the entry's local name in the module's manifest"
}, },
"storage-type": { "storage": {
"type": "string", "type": "string",
"description": "how the bus keeps it" "description": "how the bus keeps it: the entry's storage type"
}, },
"template": { "template": {
"type": "string", "type": "string",
"description": "its template" "description": "its template"
}, },
"subjects": { "stream": {
"type": "array", "type": "string",
"items": { "description": "a stream's name on the bus"
},
"subject": {
"type": "string",
"description": "the subject a stream's messages are on; also a ledger's history's"
},
"bucket": {
"type": "string",
"description": "a key-value bucket's name"
},
"object-store": {
"type": "string",
"description": "an object store's name"
},
"history": {
"type": "string",
"description": "a ledger's history stream"
},
"state": {
"type": "string",
"description": "a ledger's state bucket"
},
"indexes": {
"type": "object",
"additionalProperties": {
"type": "string" "type": "string"
}, },
"description": "its subjects, for a stream" "description": "a ledger's index buckets, by index name"
},
"snapshot": {
"type": "string",
"description": "the subject of a ledger's instances' snapshots"
},
"types": {
"type": "array",
"items": {
"type": "object"
},
"description": "a ledger's types, as its manifest declares them"
} }
}, },
"required": [ "required": [
"bus-name", "local",
"storage-type" "storage"
], ],
"description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7)" "description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7): the names on the bus of the entry's objects, one form in the events, in list, show and orphans, and in the membership"
} }
} }
} }
+25 -7
View File
@@ -47,16 +47,32 @@ export const recreatedSubject = "mesh.seat.mesh-broker.event.recreated";
export const droppedSubject = "mesh.seat.mesh-broker.event.dropped"; export const droppedSubject = "mesh.seat.mesh-broker.event.dropped";
/** the binding a module receives in its membership for one entry (ADR 0595 §1, §7) */ /** the binding a module receives in its membership for one entry (ADR 0595 §1, §7): the names on the bus of the entry's objects, one form in the events, in list, show and orphans, and in the membership */
export interface Binding { export interface Binding {
/** the entry's name on the bus */ /** a key-value bucket's name */
"bus-name": string; "bucket"?: string;
/** how the bus keeps it */ /** a ledger's history stream */
"storage-type": string; "history"?: string;
/** its subjects, for a stream */ /** a ledger's index buckets, by index name */
"subjects"?: string[]; "indexes"?: Record<string, string>;
/** the entry's local name in the module's manifest */
"local": string;
/** an object store's name */
"object-store"?: string;
/** the subject of a ledger's instances' snapshots */
"snapshot"?: string;
/** a ledger's state bucket */
"state"?: string;
/** how the bus keeps it: the entry's storage type */
"storage": string;
/** a stream's name on the bus */
"stream"?: string;
/** the subject a stream's messages are on; also a ledger's history's */
"subject"?: string;
/** its template */ /** its template */
"template"?: string; "template"?: string;
/** a ledger's types, as its manifest declares them */
"types"?: Record<string, unknown>[];
} }
/** one piece of bus state as a manifest asks it */ /** one piece of bus state as a manifest asks it */
@@ -86,6 +102,8 @@ export interface Fill {
/** one entry as the bus holds it */ /** one entry as the bus holds it */
export interface Held { export interface Held {
/** the binding a module receives for the entry, as created and recreated carry it; present on an entry whose objects the bus holds, absent on one it does not (design 25 bindings-first-view §2) */
"binding"?: Binding;
/** its name on the bus, <module>_<local name> */ /** its name on the bus, <module>_<local name> */
"bus-name": string; "bus-name": string;
/** its own configuration, where no template fits */ /** its own configuration, where no template fits */