diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go index 72ac00f95..4e0d2e542 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go @@ -4,10 +4,12 @@ import ( "context" "encoding/json" "errors" + "reflect" "strings" "testing" "time" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/buscalls" ) @@ -92,6 +94,23 @@ func TestTheLedgerStoreRefusesWhatTheLedgerWouldHaveToTakeBack(t *testing.T) { if err := s.Create(ctx, Version{Record: ledgerRecord(7, StatusProposed)}); err == nil { t.Fatal("a refused apply was taken") } + // The bundle's own pipe is the runtime's call on its bus state (stdio.BusState), and the call it serves is the one + // the runtime named (stdio.CallIDOf): the method and request go through whole, the answer comes back decoded. + if reflect.ValueOf(busState).Pointer() != reflect.ValueOf(stdio.BusState).Pointer() || + reflect.ValueOf(callIDOf).Pointer() != reflect.ValueOf(stdio.CallIDOf).Pointer() { + t.Fatal("the pipe is not the SDK's bus-state call, or the call id not the runtime's") + } + defer func(was func(string, any) (json.RawMessage, error)) { busState = was }(busState) + var asked string + busState = func(method string, params any) (json.RawMessage, error) { + asked = method + return json.RawMessage(`{"instance":"1"}`), nil + } + var got struct{ Instance string } + if err := (sdkPipe{}).Call(buscalls.LedgerGet, buscalls.InstanceRequest{Ledger: LedgerName, Instance: "1"}, &got); err != nil || + asked != buscalls.LedgerGet || got.Instance != "1" { + t.Fatalf("the pipe asked %s and answered %+v, %v", asked, got, err) + } } func TestTheOperationAChangeIsIsNamedByWhatChanged(t *testing.T) { diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go b/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go index 5b53fbee7..9878134bd 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go @@ -13,6 +13,8 @@ import ( "strings" "sync" "time" + + stdio "git.novox.be/novox/mesh-sdk/go" ) // The sources of the page, besides the module's own ledger. @@ -519,3 +521,36 @@ func (o *Overview) Calls(verb string) int { defer o.mu.Unlock() return o.verbCalls[verb] } + +// Beat takes the runtime's word on the module's consumer between events (stdio.OnHeartbeat): one consumer carries every +// subscription, so a beat is heard by every source the page subscribes to, and a missed heartbeat marks each of them +// from its subscription at once, with the runtime's why, instead of waiting out HeartbeatWithin. +func (o *Overview) Beat(h stdio.Heartbeat) { + sources := make([]string, 0, len(OverviewEvents)) + for s := range OverviewEvents { + sources = append(sources, s) + } + sort.Strings(sources) + if !h.Missed { + for _, s := range sources { + o.Heard(s, "", nil) + } + return + } + why := "its subscription's consumer missed the bus's heartbeat" + if h.Why != "" { + why += ": " + h.Why + } + o.mu.Lock() + defer o.mu.Unlock() + for _, s := range sources { + st := o.state[s] + switch { + case st == nil: + case st.Mark == nil: + st.Answered, st.Mark = false, &sourceMark{Cause: MarkSubscription, Why: why, Since: o.now()} + case st.Mark.Cause == MarkSubscription: + st.Mark.Why = why + } + } +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go index 379cb4bfd..262d4a0e6 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go @@ -7,6 +7,8 @@ import ( "strings" "testing" "time" + + stdio "git.novox.be/novox/mesh-sdk/go" ) // overviewWorld is a page model over the sources played here: what each verb answers, and what fails. @@ -170,6 +172,27 @@ func TestASourceThatDoesNotAnswerIsSaidAndAskedAgainOnABackoff(t *testing.T) { if mk := w.o.state[SourceDelivery].Mark; mk == nil || mk.Cause != MarkVerb { t.Fatalf("%+v", mk) } + // The runtime's word on the module's consumer (stdio.OnHeartbeat): a missed heartbeat marks every subscribed source + // at once, before the bound passes, with the runtime's why; the next beat asks each verb once and clears the marks. + w.fail[DeliveryGroupsVerb] = false + w.now = w.o.state[SourceDelivery].Mark.Next + w.o.Due() + asked = w.o.Calls(AsksVerb) + w.o.Beat(stdio.Heartbeat{At: w.now, Missed: true, Why: "the bus's heartbeat stopped arriving"}) + if u := w.o.Unanswered(); len(u) != 3 || u[0].Mark.Cause != MarkSubscription || !strings.Contains(u[0].Mark.Why, "stopped arriving") { + t.Fatalf("a missed heartbeat marked %+v", u) + } + w.o.Beat(stdio.Heartbeat{At: w.now}) + if len(w.o.Unanswered()) != 0 || w.o.Calls(AsksVerb) != asked+1 { + t.Fatalf("a beat left %+v, asks asked %d", w.o.Unanswered(), w.o.Calls(AsksVerb)-asked) + } + // A beat keeps a quiet source from being marked silent: it is heard, not silent. + w.now = w.now.Add(HeartbeatWithin - time.Second) + w.o.Beat(stdio.Heartbeat{At: w.now}) + w.now = w.now.Add(2 * time.Second) + if got := w.o.Silent(); len(got) != 0 { + t.Fatalf("a source heard by its beat was marked silent: %v", got) + } } func TestThePagesModelIsAToolAndItsRereadIsAHandAct(t *testing.T) { diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/page.go b/modules/mesh-knowledge/cmd/mesh-knowledge/page.go index 37e52883a..1cb55adb2 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/page.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/page.go @@ -117,6 +117,11 @@ func ServePage(k *KB) { }() } } + // The runtime's word on the module's consumer between events: a beat is each subscription heard, a missed heartbeat + // marks the sources at once (stdio.OnHeartbeat). + if err := stdio.OnHeartbeat(o.Beat); err != nil { + logf("[mesh-knowledge] the page hears no heartbeat of its subscriptions (%v): a silent source is marked after %s", err, HeartbeatWithin) + } o.Start() go func() { for range time.Tick(5 * time.Second) { diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go b/modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go index cb64add83..97e665eec 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go @@ -7,7 +7,6 @@ package main import ( "context" "encoding/json" - "errors" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -55,23 +54,15 @@ func callOf(ctx context.Context) string { return s } -// ErrNoPipe is the answer of a pipe the pinned mesh-sdk does not give the bundle yet. -var ErrNoPipe = errors.New("the pinned mesh-sdk gives a bundle no request on the tool runner's bus-state calls yet " + - "(a request to the SDK's owner: stdio.Request); the records cannot be reached over the bus") - -// sdkPipe is the pipe through the SDK. Request is set when the SDK offers it; until then the pipe answers ErrNoPipe, -// loudly, and nothing is written. +// sdkPipe is the pipe through the SDK: each call on the module's own ledger is asked of the runtime, which serves it +// from the module's own bindings only (stdio.BusState, ADR 0595 §9). type sdkPipe struct{} -// request is the SDK's raw request on the runtime's channel (stdio.Request(method, params) (json.RawMessage, error)), -// wired in the one place the group's mesh-sdk pin brings it. -var request func(method string, params any) (json.RawMessage, error) +// busState is the runtime's call on the module's own bus state: stdio.BusState, a test's stand-in. +var busState = stdio.BusState func (sdkPipe) Call(method string, req, answer any) error { - if request == nil { - return ErrNoPipe - } - raw, err := request(method, req) + raw, err := busState(method, req) if err != nil { return err } @@ -82,9 +73,8 @@ func (sdkPipe) Call(method string, req, answer any) error { } // callIDOf is the id of the tool call that was handed args, as the tool runner names it in the call's `_meta` under -// buscalls.MetaCall: wired to the SDK where it offers it (a request to the SDK's owner: stdio.CallIDOf); until then no -// call is named, and an operation carries the module's own account alone. -var callIDOf = func(args map[string]any) string { return "" } +// buscalls.MetaCall (stdio.CallIDOf): empty for the module's own act, whose operation carries the module's account alone. +var callIDOf = stdio.CallIDOf // noteCall holds the call a caller's write serves for as long as it runs, and ctxFor names it to the ledger: the tool // runner stamps the caller it proved for that call as the operation's author (ADR 0597 §7), never a name the caller or