mesh-conditions: ListBound is asked again on the bus module's created or a node-engine's running, not on a timer
Builds: design 47 G §515, §663 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -7,7 +7,6 @@ package main
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
broker "git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
|
||||
)
|
||||
@@ -48,10 +47,27 @@ func (k *Keeper) takeSource(module string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// nudgeBound says, without waiting, that the bus module's entries may be read now (ListBound's retry).
|
||||
func (k *Keeper) nudgeBound() {
|
||||
k.mu.Lock()
|
||||
k.init()
|
||||
c := k.boundRetry
|
||||
k.mu.Unlock()
|
||||
select {
|
||||
case c <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// ListBound reads the entries the bus module holds at the keeper's start: every module with a `finding-events` entry is a
|
||||
// source. A call that fails is asked again with a growing wait, the bus module being delivered after the keeper at times.
|
||||
// source. A call that fails is asked again on the next event that says the bus module may answer, its `created` or a
|
||||
// node-engine's `running` (the bus module being delivered after the keeper at times), never on a timer.
|
||||
func ListBound(k *Keeper, call func(key string, body any) (json.RawMessage, error)) {
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
k.mu.Lock()
|
||||
k.init()
|
||||
retry := k.boundRetry
|
||||
k.mu.Unlock()
|
||||
for {
|
||||
raw, err := call("seat:"+broker.Seat+"."+broker.VerbList, broker.ListInput{})
|
||||
var answer broker.ListAnswer
|
||||
if err == nil {
|
||||
@@ -67,7 +83,7 @@ func ListBound(k *Keeper, call func(key string, body any) (json.RawMessage, erro
|
||||
}
|
||||
return
|
||||
}
|
||||
k.Logf("the entries of the bus module are not read yet (%v); asked again in %s", err, wait)
|
||||
time.Sleep(wait)
|
||||
k.Logf("the entries of the bus module are not read yet (%v); asked again when it says created or a node-engine says running", err)
|
||||
<-retry
|
||||
}
|
||||
}
|
||||
|
||||
@@ -72,13 +72,17 @@ func (k *Keeper) FindingEvent(e stdio.Envelope) error {
|
||||
if _, ok := emitterOf(e.Key, watchdogs.EventMachineBack); ok && strings.HasPrefix(e.Key, watchdogs.Seat+".") {
|
||||
return k.MachinePresence(e.Body, false)
|
||||
}
|
||||
if _, _, _, ok := engineKey(e.Key); ok {
|
||||
if event, _, _, ok := engineKey(e.Key); ok {
|
||||
if event == engine.EventRunning {
|
||||
k.nudgeBound()
|
||||
}
|
||||
return k.EngineEvent(e)
|
||||
}
|
||||
if strings.HasPrefix(e.Key, meshhealers.Seat+".attempt-changed.") {
|
||||
return k.AttemptChanged(e.Key, e.Body)
|
||||
}
|
||||
if e.Key == broker.Seat+"."+broker.EventCreated {
|
||||
k.nudgeBound()
|
||||
return k.BrokerCreated(e.Body)
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -164,6 +164,9 @@ type Keeper struct {
|
||||
stopped map[string]bool
|
||||
// created are the modules whose `finding-events` entry the bus module made (bound.go).
|
||||
created map[string]bool
|
||||
// boundRetry is told, once, when the bus module says `created` or a node-engine says `running`: the events after which
|
||||
// the entries of the bus module may be read (ListBound).
|
||||
boundRetry chan struct{}
|
||||
// followed are the finding streams the catalog half declared for each module, by module (catalog.go).
|
||||
followed map[string]string
|
||||
}
|
||||
@@ -179,6 +182,9 @@ func (k *Keeper) init() {
|
||||
k.halves, k.placed, k.stopped = map[string]map[string]catseat.CatalogEntry{}, map[string]bool{}, map[string]bool{}
|
||||
k.views, k.keyNode, k.whys = map[string]*sourceView{}, map[string]map[string]string{}, map[string]string{}
|
||||
}
|
||||
if k.boundRetry == nil {
|
||||
k.boundRetry = make(chan struct{}, 1)
|
||||
}
|
||||
if k.Now == nil {
|
||||
k.Now = time.Now
|
||||
}
|
||||
|
||||
@@ -723,3 +723,47 @@ func TestTheKeepersOwnFindingIsTheContractsFindingByItsWireForm(t *testing.T) {
|
||||
t.Fatal("the condition stands after its finding was cleared")
|
||||
}
|
||||
}
|
||||
|
||||
// ListBound reads the bus module's entries once and, when the call fails, again on the event that says the bus module may
|
||||
// answer (its `created`, or a node-engine's `running`), never on a timer (design 47 G §515, §663).
|
||||
func TestListBoundIsAskedAgainOnTheEventsNotATimer(t *testing.T) {
|
||||
for _, event := range []string{"mesh-broker.created", "node-engine.running.ace.mesh-conditions"} {
|
||||
t.Run(event, func(t *testing.T) {
|
||||
var asked []map[string]any
|
||||
k := &Keeper{Publish: func(_ string, b any) error { asked = append(asked, b.(map[string]any)); return nil }}
|
||||
calls := make(chan int)
|
||||
ended := make(chan struct{})
|
||||
go func() {
|
||||
n := 0
|
||||
ListBound(k, func(string, any) (json.RawMessage, error) {
|
||||
n++
|
||||
calls <- n
|
||||
if n == 1 {
|
||||
return nil, fmt.Errorf("the bus module is not delivered yet")
|
||||
}
|
||||
return mustJSON(t, map[string]any{"entries": []any{map[string]any{"module": "newpart", "template": FindingEventsTemplate}}}), nil
|
||||
})
|
||||
close(ended)
|
||||
}()
|
||||
<-calls
|
||||
select {
|
||||
case <-ended:
|
||||
t.Fatal("ListBound ended with no answer")
|
||||
default:
|
||||
}
|
||||
_ = k.FindingEvent(stdio.Envelope{Key: event, Body: []byte(`{}`)})
|
||||
select {
|
||||
case n := <-calls:
|
||||
if n != 2 {
|
||||
t.Fatalf("asked %d times", n)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("not asked again on the event")
|
||||
}
|
||||
<-ended
|
||||
if len(asked) != 1 || asked[0]["module"] != "newpart" {
|
||||
t.Errorf("asked %v, want newpart once", asked)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user