From 811b59b4540dfc7e13e5d9068b193f758f03af50 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 20:27:47 +0100 Subject: [PATCH] COND9: the keeper takes a part whose finding stream the bus module made as a source (the moved-stream half stops on a question) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §515, §663 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/bound.go | 73 +++++++++++++++++++ .../cmd/mesh-conditions/catalog.go | 4 + .../cmd/mesh-conditions/events.go | 6 +- .../cmd/mesh-conditions/keeper.go | 2 + .../cmd/mesh-conditions/keeper_test.go | 18 +++++ .../cmd/mesh-conditions/main.go | 1 + .../cmd/mesh-conditions/manifest_test.go | 4 +- modules/mesh-conditions/go.mod | 1 + modules/mesh-conditions/module.json | 4 +- 9 files changed, 109 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-conditions/cmd/mesh-conditions/bound.go diff --git a/modules/mesh-conditions/cmd/mesh-conditions/bound.go b/modules/mesh-conditions/cmd/mesh-conditions/bound.go new file mode 100644 index 000000000..edff8dac4 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/bound.go @@ -0,0 +1,73 @@ +package main + +// The keeper's sources are the parts whose finding stream the bus module made (design 47 G §515, §663): it takes a source +// when it hears the seat mesh-broker's `created` of a part's `finding-events` entry, and at its start reads the entries +// the bus module holds by `list`. No part is named here (§7). + +import ( + "encoding/json" + "fmt" + "time" + + broker "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" +) + +// FindingEventsTemplate is the template of a part's finding stream in its `requires` (design 47 G §529). +const FindingEventsTemplate = "finding-events" + +// BrokerCreated takes one `created` of the bus module: a `finding-events` entry makes its module a source, asked once with +// `findings-asked` naming it. An entry of another template, or a module already a source, changes nothing. +func (k *Keeper) BrokerCreated(body []byte) error { + var said broker.Said + if err := json.Unmarshal(body, &said); err != nil || said.Module == "" { + k.Logf("a created of the bus module is not readable: %v", err) + return nil + } + if said.Template != FindingEventsTemplate { + return nil + } + return k.takeSource(said.Module) +} + +// takeSource makes the module a source the bus module made a stream for, and asks it once. +func (k *Keeper) takeSource(module string) error { + k.mu.Lock() + k.init() + if k.created == nil { + k.created = map[string]bool{} + } + if k.created[module] { + k.mu.Unlock() + return nil + } + k.created[module] = true + k.mu.Unlock() + if _, err := k.Ask(Module, module); err != nil { + return fmt.Errorf("asking %s for its findings: %w", module, err) + } + return nil +} + +// 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. +func ListBound(k *Keeper, call func(key string, body any) (json.RawMessage, error)) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + raw, err := call("seat:"+broker.Seat+"."+broker.VerbList, broker.ListInput{}) + var answer broker.ListAnswer + if err == nil { + err = json.Unmarshal(raw, &answer) + } + if err == nil { + for _, e := range answer.Entries { + if e.Template == FindingEventsTemplate { + if err := k.takeSource(e.Module); err != nil { + k.Logf("%v", err) + } + } + } + return + } + k.Logf("the entries of the bus module are not read yet (%v); asked again in %s", err, wait) + time.Sleep(wait) + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index f7f89381c..19e16487c 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -113,6 +113,10 @@ func (k *Keeper) isSource(observer string) bool { k.mu.Lock() defer k.mu.Unlock() k.init() + // A part whose finding stream the bus module made is a source (design 47 G §663). + if k.created[observer] { + return true + } // A source the keeper kept in `findings-views` is heard before its first view of the catalog half (design 47 G §388). for _, v := range k.views { if v.observer == observer && v.kept { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index e0eb2c7b1..f2be01310 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -14,6 +14,7 @@ import ( "strings" "time" + broker "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" @@ -51,6 +52,9 @@ func (k *Keeper) FindingEvent(e stdio.Envelope) error { if e.Key == engine.Seat+"."+engine.EventRunning { return k.EngineRunning(e.Node, e.Body) } + if e.Key == broker.Seat+"."+broker.EventCreated { + return k.BrokerCreated(e.Body) + } if observer, ok := emitterOf(e.Key, EventFindingsStated); ok { return k.FindingsStated(observer, e.Node, e.Body) } @@ -202,7 +206,7 @@ func (k *Keeper) MachinePresence(body []byte, away bool) error { func hear(k *Keeper) { patterns := []string{"*." + EventFindingWritten, "*." + EventFindingCleared, "*." + EventFindingsStated, watchdogs.Seat + "." + watchdogs.EventMachineAway, watchdogs.Seat + "." + watchdogs.EventMachineBack, - engine.Seat + "." + engine.EventRunning} + engine.Seat + "." + engine.EventRunning, broker.Seat + "." + broker.EventCreated} for _, pattern := range patterns { go func() { <-stdio.Served() diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index 6753dffc2..a8598441d 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -147,6 +147,8 @@ type Keeper struct { placed map[string]bool placedHeld bool stopped map[string]bool + // created are the modules whose `finding-events` entry the bus module made (bound.go). + created map[string]bool } func (k *Keeper) init() { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 31e11071f..3d47e4f62 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -617,6 +617,24 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { {"the second branch dropped takes its source away", func() error { return r.k.CatalogDropped("feature") }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, + // Design 47 G §663: a part whose finding-events entry the bus module made is a source, asked once. + {"a created finding-events entry makes its part a source, asked once", func() error { + before := len(asked) + created := mustJSON(t, map[string]any{"module": "newpart", "template": FindingEventsTemplate, "bus-name": "newpart_finding-events", "local-name": "finding-events"}) + other := mustJSON(t, map[string]any{"module": "otherpart", "template": "state", "bus-name": "otherpart_x", "local-name": "x"}) + for _, body := range [][]byte{created, other, created} { + if err := r.k.FindingEvent(stdio.Envelope{Key: "mesh-broker.created", Body: body}); err != nil { + return err + } + } + if len(asked) != before+1 || asked[before]["module"] != "newpart" { + t.Errorf("asked %v, want newpart once", asked[before:]) + } + if err := r.k.FindingWritten("otherpart", "ace", mustJSON(t, f("otherpart", "o"))); err != nil { + return err + } + return r.k.FindingWritten("newpart", "ace", mustJSON(t, f("newpart", "w"))) + }, []string{"machine.d.silent", "machine.g.silent", "machine.w.silent"}, ""}, } for _, s := range steps { if err := s.do(); err != nil { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index 55f780583..a185af451 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -130,6 +130,7 @@ func Keep(k *Keeper, v *views.Reader) { return said(k.Attempt(c.Key, c.Value, c.Op != "put")) }) hear(k) + go func() { <-stdio.Served(); ListBound(k, stdio.Ask) }() watch(k, v, BoundsState, "", func(c stdio.StateChange) error { if c.Op == stdio.OpExpire { return said(k.BoundPassed(c.Key)) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 273a924a0..73d3b0920 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -37,9 +37,9 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { // the observers' findings are heard as events and no `findings` is read (design 47 G §383). reads := []string{HavesRead, AttemptsRead} if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision", - "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision"}) || + "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-broker.list"}) || !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.assignments", - "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "node-engine.running"}) { + "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-broker.created", "node-engine.running"}) { t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) } for _, r := range m.Reads { diff --git a/modules/mesh-conditions/go.mod b/modules/mesh-conditions/go.mod index 7ce3ace27..13d6fdc7e 100644 --- a/modules/mesh-conditions/go.mod +++ b/modules/mesh-conditions/go.mod @@ -5,6 +5,7 @@ go 1.26.0 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index 66cced877..569f6071a 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -23,7 +23,8 @@ "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision", "seat:mesh-catalog.registrations", - "seat:mesh-catalog.revision" + "seat:mesh-catalog.revision", + "seat:mesh-broker.list" ], "consumes": [ "mesh-delivery-assigner.views.wants", @@ -36,6 +37,7 @@ "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", + "mesh-broker.created", "node-engine.running" ], "state": [