diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index 19e16487c..4b9cb39dc 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -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) { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index 154773e4d..4b600e93c 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -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() { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 3d47e4f62..0578a1dd9 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -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 {