From 9e2311029a493a8cfd8d79eff6ca9825c6afdefc Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 15:46:35 +0100 Subject: [PATCH] COND2: the keeper keeps its sources in the state findings-views and hears them before its first view of the catalog half MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §388, §365 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/asks.go | 1 + .../cmd/mesh-conditions/catalog.go | 105 +++++++++++++++++- .../cmd/mesh-conditions/events.go | 4 + .../cmd/mesh-conditions/keeper_test.go | 27 +++++ .../cmd/mesh-conditions/main.go | 2 + .../cmd/mesh-conditions/manifest_test.go | 2 +- modules/mesh-conditions/module.json | 1 + 7 files changed, 138 insertions(+), 4 deletions(-) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/asks.go b/modules/mesh-conditions/cmd/mesh-conditions/asks.go index a380abd87..f9868baa5 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/asks.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/asks.go @@ -79,6 +79,7 @@ func (k *Keeper) Ask(by, module string) (string, error) { } var problems []string for _, w := range wait { + k.saveView(w[0], w[1]) if err := k.Store.CreateTTL(BoundsState, askedKey(w[0], w[1], ask), map[string]any{"observer": w[0], "node": w[1], "ask": ask, "until": now.Add(AskedWithin)}, AskedWithin); err != nil && !errors.Is(err, stdio.ErrChanged) { problems = append(problems, err.Error()) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index 2b176aa2d..b513677b5 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -7,12 +7,17 @@ package main import ( "encoding/json" + "errors" + "maps" "slices" "sort" "strings" + "time" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" wants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + stdio "git.novox.be/novox/mesh-sdk/go" engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" ) @@ -48,14 +53,23 @@ func (k *Keeper) CatalogHeld(branch string, entries map[string]catseat.CatalogEn k.init() k.halves[branch] = entries var gone []string - for observer := range k.findings { - if observer == Module { - continue + seen := map[string]bool{} + consider := func(observer string) { + if observer == Module || observer == "" || seen[observer] { + return } + seen[observer] = true if _, ok := k.entry(observer); !ok { gone = append(gone, observer) } } + for observer := range k.findings { + consider(observer) + } + // A kept source (design 47 G §388) no revision lists any more leaves the list with its conditions. + for _, v := range k.views { + consider(v.observer) + } k.mu.Unlock() sort.Strings(gone) var first error @@ -85,6 +99,12 @@ func (k *Keeper) isSource(observer string) bool { k.mu.Lock() defer k.mu.Unlock() k.init() + // 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 { + return true + } + } branches := make([]string, 0, len(k.halves)) for b := range k.halves { branches = append(branches, b) @@ -198,6 +218,7 @@ func (k *Keeper) sweepStopped() error { func (k *Keeper) clearObserver(observer, node, why string, first *error) { k.mu.Lock() var keys []string + var dropped [][2]string for key := range k.findings[observer] { if node == "" || k.keyNode[observer][key] == node { keys = append(keys, key) @@ -212,9 +233,13 @@ func (k *Keeper) clearObserver(observer, node, why string, first *error) { _ = k.Store.Delete(BoundsState, askedKey(observer, v.node, ask)) } delete(k.views, name) + dropped = append(dropped, [2]string{v.observer, v.node}) } } k.mu.Unlock() + for _, d := range dropped { + _ = k.Store.Delete(ViewsState, viewKey(d[0], d[1])) + } sort.Strings(keys) for _, key := range keys { k.mu.Lock() @@ -229,3 +254,77 @@ func (k *Keeper) clearObserver(observer, node, why string, first *error) { k.mu.Unlock() } } + +// ViewsState is the keeper's list of sources (design 47 G §388), in the form of its `conditions`: one key per observer and +// node, `.`, its value what the keeper heard there: the last stated set and the open asks. +const ViewsState = "findings-views" + +func viewKey(observer, node string) string { return observer + "." + node } + +// keptView is the value of one key of `findings-views`. +type keptView struct { + Observer string `json:"observer"` + Node string `json:"node"` + At time.Time `json:"at"` + Pending map[string]time.Time `json:"pending,omitempty"` + Asks map[string]time.Time `json:"asks,omitempty"` + Findings []seat.Finding `json:"findings,omitempty"` +} + +// saveView keeps what the keeper heard of the observer on the node, by compare-and-set; a source is kept once it is heard. +func (k *Keeper) saveView(observer, node string) { + k.mu.Lock() + k.init() + v := k.views[observer+"."+node] + if v == nil { + k.mu.Unlock() + return + } + v.kept = true + value := keptView{Observer: observer, Node: node, At: v.At, Pending: maps.Clone(v.Pending), Asks: maps.Clone(v.Asks), + Findings: slices.Clone(v.Findings)} + k.mu.Unlock() + key := viewKey(observer, node) + for range ConditionWrites { + _, rev, found, err := k.Store.Get(ViewsState, key) + if err == nil && !found { + _, err = k.Store.Create(ViewsState, key, value) + } else if err == nil { + _, err = k.Store.Update(ViewsState, key, value, rev) + } + if err == nil || !errors.Is(err, stdio.ErrChanged) { + if err != nil { + k.Logf("the source %s on %s is not kept: %v", observer, node, err) + } + return + } + } +} + +// RestoreViews reads the kept sources at the start, before the first view of the catalog half: their events are applied +// at once from the durable consumers, and the catalog half only adds and removes sources (design 47 G §388). +func (k *Keeper) RestoreViews() { + keys, err := k.Store.Keys(ViewsState) + if err != nil { + k.Logf("the kept sources are not read: %v", err) + return + } + for _, key := range keys { + raw, _, found, err := k.Store.Get(ViewsState, key) + if err != nil || !found { + continue + } + var kv keptView + if json.Unmarshal(raw, &kv) != nil || kv.Observer == "" || kv.Node == "" { + k.Logf("the kept source %s is not readable: left out", key) + continue + } + k.mu.Lock() + k.init() + v := k.view(kv.Observer, kv.Node) + v.At, v.Findings, v.kept = kv.At, kv.Findings, true + maps.Copy(v.Pending, kv.Pending) + maps.Copy(v.Asks, kv.Asks) + k.mu.Unlock() + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index ec7abe5d8..e0eb2c7b1 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -107,6 +107,7 @@ func (k *Keeper) take(observer, node string, f seat.Finding) error { if !named.IsZero() { _ = k.Store.Delete(BoundsState, pendingKey(observer, node, v.Key)) } + k.saveView(observer, node) return k.Finding(observer, v.Key, value, false) } @@ -236,6 +237,8 @@ type sourceView struct { Findings []seat.Finding // Asks are the asks it has not answered, with the end of their bound (asks.go). Asks map[string]time.Time + // kept says the source is in the state `findings-views` (catalog.go). + kept bool } // view is the observer's view on the node, made when there is none. Called with mu held. @@ -341,6 +344,7 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error { _ = k.Store.Delete(BoundsState, askedKey(observer, node, ask)) } k.clearUnstated(observer, node) + k.saveView(observer, node) sort.Strings(drop) for key := range oldPending { if _, still := v.Pending[key]; !still { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 446b155fd..1f1836648 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -40,6 +40,12 @@ func (f *fakeStore) Get(state, key string) ([]byte, uint64, bool, error) { return v, f.revs[state+"/"+key], ok, nil } +// Get2 says whether the key is held. +func (f *fakeStore) Get2(state, key string) ([]byte, bool) { + v, _, ok, _ := f.Get(state, key) + return v, ok +} + func (f *fakeStore) put(state, key string, value any) uint64 { raw, _ := json.Marshal(value) f.next++ @@ -580,6 +586,27 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { } return r.k.CatalogHeld("main", map[string]catseat.CatalogEntry{watchmanModule: {}, selfCheckModule: {}}) }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, + // Design 47 G §388: a restarted keeper hears its kept sources before its first view of the catalog half. + {"a restarted keeper with no view of the catalog half hears its kept sources, and the view takes one away", func() error { + if _, ok := r.st.Get2(ViewsState, viewKey(watchman, "ace")); !ok { + t.Errorf("the source %s on ace was not kept in %s", watchman, ViewsState) + } + k2 := &Keeper{Store: r.st, Now: func() time.Time { return r.now }, Say: func(e Event) error { r.said = append(r.said, e); return nil }} + k2.RestoreViews() + if err := k2.FindingWritten(watchman, "ace", mustJSON(t, f(watchman, "k"))); err != nil { + return err + } + if _, ok := r.open("machine.k.silent"); !ok { + t.Errorf("a kept source's finding was not raised with no view of the catalog half") + } + if err := k2.FindingWritten("fixture", "ace", mustJSON(t, f("fixture", "q"))); err != nil { + return err + } + if _, ok := r.open("machine.q.silent"); ok { + t.Errorf("a module that was never a source raised a condition") + } + return k2.CatalogHeld("main", map[string]catseat.CatalogEntry{selfCheckModule: {}}) + }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, } 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 9b5a81ddc..92c9ed5b3 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -109,6 +109,8 @@ func Keep(k *Keeper, v *views.Reader) { } } } + // The kept sources are heard at once, before the first view of the catalog half (design 47 G §388). + k.RestoreViews() said := func(err error) error { if err != nil { logf("%v", err) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 6dc3c9aa5..6b2d320ea 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -63,7 +63,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { } // Clause 12: each state with the tool that reads it, each act with the tool that starts it. readBy := map[string]string{ConditionsState: Seat + ".conditions", HistoryState: HistoryTool, - HeldState: HeldTool, BoundsState: HeldTool, "a resync": Seat + ".resync", "what it holds of observers": FindingsViewsTool, "what it reads": InputsTool} + HeldState: HeldTool, BoundsState: HeldTool, ViewsState: FindingsViewsTool, "a resync": Seat + ".resync", "what it holds of observers": FindingsViewsTool, "what it reads": InputsTool} tools := map[string]bool{} for _, tool := range Tools(&Keeper{}) { tools[tool.Name] = true diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index b0431616d..f9aed5620 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -41,6 +41,7 @@ "ttl-seconds": 7776000 }, "held", + "findings-views", { "name": "bounds", "key-ttl": true