COND9 (moved-stream half): the keeper keeps a moved part's conditions, drops the old source and asks once on the new stream

Builds: design 47 G §663, §690

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-11 20:44:44 +01:00
parent ecf597adce
commit 1af839fb72
3 changed files with 106 additions and 3 deletions
@@ -46,13 +46,60 @@ func declaresFindings(e catseat.CatalogEntry) bool {
return false
}
// findingStreams names the streams the entry declares its finding events in, one string, in name order; none is "".
func findingStreams(e catseat.CatalogEntry) string {
var names []string
for _, s := range e.Streams {
if strings.Contains(s, EventFindingWritten) {
names = append(names, s)
}
}
streams, _ := e.Manifest["streams"].([]any)
for _, s := range streams {
m, _ := s.(map[string]any)
name, _ := m["name"].(string)
events, _ := m["events"].([]any)
for _, ev := range events {
if n, _ := ev.(string); strings.HasPrefix(n, EventFindingWritten) {
names = append(names, name)
break
}
}
}
sort.Strings(names)
return strings.Join(names, ",")
}
// CatalogHeld takes the newest catalog revision of one watched branch: the sources are the modules declaring the finding
// events in any branch, and the conditions of a module in no branch are cleared.
// events in any branch, and the conditions of a module in no branch are cleared. A module whose entry names another
// stream than the one the keeper followed for it has moved its findings: the keeper keeps its open conditions and asks
// once on the new stream (design 47 G §663, §690).
func (k *Keeper) CatalogHeld(branch string, entries map[string]catseat.CatalogEntry) error {
k.mu.Lock()
k.init()
k.halves[branch] = entries
return k.sweepCatalog()
if k.followed == nil {
k.followed = map[string]string{}
}
var moved []string
for module, e := range entries {
now := findingStreams(e)
was := k.followed[module]
if now != "" {
k.followed[module] = now
}
if was != "" && now != "" && now != was {
moved = append(moved, module)
}
}
sort.Strings(moved)
err := k.sweepCatalog()
for _, module := range moved {
if _, askErr := k.Ask(Module, module); askErr != nil && err == nil {
err = askErr
}
}
return err
}
// CatalogDropped forgets a branch no registration watches any more (design 47 G §532): the modules only it listed leave
@@ -118,8 +165,11 @@ func (k *Keeper) isSource(observer string) bool {
return true
}
// A source the keeper kept in `findings-views` is heard before its first view of the catalog half (design 47 G §388).
// Once a revision lists the part and no longer declares its finding stream, a kept source of the old stream is dropped,
// its open conditions kept (§690); a part no revision lists is kept until the first view takes it away.
_, listed := k.entry(observer)
for _, v := range k.views {
if v.observer == observer && v.kept {
if v.observer == observer && v.kept && (!listed || k.declares(observer)) {
return true
}
}
@@ -135,6 +185,16 @@ func (k *Keeper) isSource(observer string) bool {
return false
}
// declares says whether a held revision lists the observer with its finding events. Called with mu held.
func (k *Keeper) declares(observer string) bool {
for _, half := range k.halves {
if e, ok := half[observer]; ok && declaresFindings(e) {
return true
}
}
return false
}
// catalogHolder is Keeper.Holder: an observer whose catalog entry claims only seats held once for the mesh is held on the
// node the assignments name as the holder of one of them (design 47 G §384); unknown, it is taken from every node.
func (k *Keeper) catalogHolder(observer string) (string, bool) {
@@ -151,6 +151,8 @@ 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
// followed are the finding streams the catalog half declared for each module, by module (catalog.go).
followed map[string]string
}
func (k *Keeper) init() {
@@ -635,6 +635,47 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) {
}
return r.k.FindingWritten("newpart", "ace", mustJSON(t, f("newpart", "w")))
}, []string{"machine.d.silent", "machine.g.silent", "machine.w.silent"}, ""},
// Design 47 G §663, §690: a part whose findings move keeps its condition, is asked once on the new stream, and its
// old source is dropped once its entry declares no finding stream.
{"a part that moves its finding stream is asked once and keeps its condition", func() error {
entry := func(stream string) map[string]catseat.CatalogEntry {
e := catseat.CatalogEntry{Module: "movedpart", Manifest: map[string]any{}}
if stream != "" {
e.Manifest["streams"] = []any{map[string]any{"name": stream, "events": []any{"finding-written.*"}}}
}
return map[string]catseat.CatalogEntry{"movedpart": e}
}
if err := r.k.CatalogHeld("moves", entry("old-findings")); err != nil {
return err
}
if err := r.k.FindingWritten("movedpart", "ace", mustJSON(t, f("movedpart", "m"))); err != nil {
return err
}
before := len(asked)
if err := r.k.CatalogHeld("moves", entry("old-findings")); err != nil {
return err
}
if len(asked) != before {
t.Errorf("the same entry asked %v", asked[before:])
}
if err := r.k.CatalogHeld("moves", entry("new-findings")); err != nil {
return err
}
if len(asked) != before+1 || asked[before]["module"] != "movedpart" {
t.Errorf("asked %v, want movedpart once on the new stream", asked[before:])
}
// Its entry declares no finding stream now (the findings moved to a template): the old source is dropped.
if err := r.k.CatalogHeld("moves", entry("")); err != nil {
return err
}
if err := r.k.FindingWritten("movedpart", "ace", mustJSON(t, f("movedpart", "late"))); err != nil {
return err
}
if _, ok := r.open("machine.late.silent"); ok {
t.Errorf("the dropped source raised a condition")
}
return nil
}, []string{"machine.d.silent", "machine.g.silent", "machine.m.silent"}, ""}, // newpart, listed by no revision, leaves with the first view
}
for _, s := range steps {
if err := s.do(); err != nil {