COND2: the keeper keeps its sources in the state findings-views and hears them before its first view of the catalog half

Builds: design 47 G §388, §365

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-11 15:46:35 +01:00
parent 1e55ea8e4e
commit 9e2311029a
7 changed files with 138 additions and 4 deletions
@@ -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())
@@ -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, `<observer>.<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()
}
}
@@ -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 {
@@ -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 {
@@ -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)
@@ -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
+1
View File
@@ -41,6 +41,7 @@
"ttl-seconds": 7776000
},
"held",
"findings-views",
{
"name": "bounds",
"key-ttl": true