From fc51bf48329849235e6a45722649610ea054e07d Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 13 Nov 2026 21:03:59 +0100 Subject: [PATCH] Round 2: grants says the absent stream in its status and pass result; tests that fail without A-372's settle call and without Ready's reading of the first views Builds: design 47 G A-371 health clause, A-372, A-368 (G-22-lead-answers-2.md) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-assigner/halvesnone_test.go | 22 +++-------- .../clients/meshcatalogclient/entry_test.go | 37 +++++++++++++++++++ modules/mesh-grants/grants_test.go | 4 ++ modules/mesh-grants/internal/grants/inputs.go | 2 + modules/mesh-grants/internal/grants/part.go | 5 ++- .../mesh-grants/internal/grants/running.go | 24 +++++++++++- 6 files changed, 76 insertions(+), 18 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halvesnone_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halvesnone_test.go index 03604332..18c4c33c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halvesnone_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halvesnone_test.go @@ -11,18 +11,10 @@ import ( "git.novox.be/novox/mesh-sdk/go/views" ) -type noneWiring struct{ followers []*views.Follower } - -func (w *noneWiring) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error { - w.followers = append(w.followers, f) - return nil -} -func (w *noneWiring) Drop(string, string) error { return nil } - // Design 47 G A-372: the assigner's wait for its catalog half asks the catalog client's Ready as well: a mesh whose registrations and // gate's source are answered and name no branch is none-yet, and the first pass is told so once; before they answer it is not told. func TestAMeshThatNamesNoBranchSettlesTheAssignersFirstPass(t *testing.T) { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() call := func(key string, _ any) (json.RawMessage, error) { if key == mcc.RegistrationsVerb { @@ -31,16 +23,14 @@ func TestAMeshThatNamesNoBranchSettlesTheAssignersFirstPass(t *testing.T) { return json.Marshal(map[string][]string{"branches": {}}) } for _, answered := range []bool{false, true} { - w := &noneWiring{} - b, err := mcc.NewBranches(w, call) - if err != nil { + settled := make(chan struct{}, 2) + reader := &views.ProcessReader{Module: "mesh-delivery-assigner", Call: call, Logf: t.Logf} + h := &halves{w: newWorld(), call: call, reader: reader, wake: func() {}, logf: t.Logf, ctx: ctx, settled: settled} + if err := h.follow(); err != nil { t.Fatal(err) } - settled := make(chan struct{}, 2) - h := &halves{ctx: ctx, settled: settled, branches: b} - h.settleWhenNoBranch() if answered { - for _, f := range w.followers { + for _, f := range reader.Followers() { if err := f.Start(ctx); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/entry_test.go b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/entry_test.go index d0a2078d..92c45723 100644 --- a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/entry_test.go +++ b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/entry_test.go @@ -135,3 +135,40 @@ func TestAMeshThatNamesNoBranchIsReadyWithAnEmptyView(t *testing.T) { }) } } + +// A-368: a first view closes before the follower's change reaches the branches followed, so a registration or a gate's source that +// names a branch is no mesh that names none, though Followed holds nothing yet. +func TestAFirstViewThatNamesABranchIsNotReadyBeforeItIsFollowed(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + for _, c := range []struct { + name string + source []string + regs map[string][]string + }{ + {"a registration names main", nil, map[string][]string{"nats": {"main"}}}, + {"the gate's source names main", []string{"main"}, map[string][]string{}}, + } { + t.Run(c.name, func(t *testing.T) { + seats := &fakeSeats{source: c.source, regs: c.regs, version: 1} + w := &fakeWiring{adopted: map[string]*views.Follower{}, hears: map[string]func(context.Context, []byte) error{}} + b, err := NewBranches(w, seats.call) + if err != nil { + t.Fatal(err) + } + for _, name := range []string{"mesh-catalog.registrations", "mesh-delivery.source"} { + f, _ := w.get(name) + f.Changed = nil // the change is not yet taken into the branches followed + if err := f.Start(ctx); err != nil { + t.Fatal(err) + } + } + if len(b.Followed()) != 0 { + t.Fatalf("followed %v: the test means no branch is followed yet", b.Followed()) + } + if err := b.Ready(); err == nil { + t.Error("Ready said ready with a branch named and not yet followed") + } + }) + } +} diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 49f926c5..0f1398bc 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -5317,6 +5317,10 @@ func TestAStreamThePartCreatesAndTheBusLacksIsAnEmptyFirstView(t *testing.T) { if res.Sealed == nil { t.Fatal("the pass says nothing of the sealed list") } + reads := p.StreamReads() + if absent := len(res.StreamsAbsent) == 1 && len(reads) == 1 && reads[0].Absent != "" && reads[0].State == "waiting"; absent == row.held { + t.Errorf("the stream said absent = %v (pass %v, status %+v), want %v", absent, res.StreamsAbsent, reads, !row.held) + } if held := strings.Contains(res.Sealed.Why, "a read of this pass failed"); held != row.held { t.Errorf("held for a failed read = %v, want %v: %q", held, row.held, res.Sealed.Why) } diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index af15a250..488688e2 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -215,6 +215,8 @@ type Inputs struct { // CatalogUnread says why this pass composed before the first views of the catalog half were in (design 47 G §684); empty once they are. // As WantsUnread, such a pass retires nothing (ReadFailures) and only adds: the list in force is kept. CatalogUnread string + // StreamsAbsent says each stream the part reads whole that the bus lacks and the part creates, taken as empty this pass (A-371). + StreamsAbsent []string // AssignmentUnread says why this pass composed before the first view of the assignment half was in (design 46 approval // mark §6, design 47 G §195): no node's capability reading is known, so the list in force is kept; empty once it is. AssignmentUnread string diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 113a444f..94cddfbc 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -149,7 +149,9 @@ type Result struct { // of the list in force, or stays out where it holds none (refused.go); the condition mesh-grants.principal-refused says it. PrincipalsRefused []string `json:"principals_refused,omitempty"` // CatalogWaiting is why this pass composed before the first views of the catalog half were in (design 47 G §684); empty once they are. - CatalogWaiting string `json:"catalog_waiting,omitempty"` + CatalogWaiting string `json:"catalog_waiting,omitempty"` + // StreamsAbsent names each stream read whole that the bus lacked, taken as empty and created after this pass (design 47 G, A-371). + StreamsAbsent []string `json:"streams_absent,omitempty"` Errors []string `json:"errors,omitempty"` PublishedOrAsserted string `json:"published_or_asserted"` } @@ -327,6 +329,7 @@ func (p *Part) Compose() Result { in := p.inputs() p.keepBrokers(in) res.CatalogWaiting = in.CatalogUnread + res.StreamsAbsent = in.StreamsAbsent if err := p.withBound(&in); err != nil { res.Errors = append(res.Errors, err.Error()) } diff --git a/modules/mesh-grants/internal/grants/running.go b/modules/mesh-grants/internal/grants/running.go index ed4382d0..81652c48 100644 --- a/modules/mesh-grants/internal/grants/running.go +++ b/modules/mesh-grants/internal/grants/running.go @@ -70,6 +70,9 @@ type StreamRead struct { // State is `waiting` until the read ends, then `read`. State string `json:"state"` Since time.Time `json:"since"` + // Absent says why a waiting read is taken as empty: the stream is one the part creates and the bus does not hold yet, so its first + // view is empty and the stream is created after the pass (design 47 G, A-371); empty when the read waits for its answer. + Absent string `json:"absent,omitempty"` } // ReadingRunning begins the whole read of the node-engine's stream running at the part's start; the part acts on none of @@ -101,6 +104,19 @@ func (p *Part) StreamReads() []StreamRead { return []StreamRead{r} } +// noteAbsent keeps in the stream's status whether its waiting read is taken as empty for the stream's absence. +func (p *Part) noteAbsent(absent bool) { + p.mu.Lock() + defer p.mu.Unlock() + if p.runningRead == nil { + return + } + p.runningRead.Absent = "" + if absent { + p.runningRead.Absent = "the stream is absent on the bus: its first view is taken as empty, and the part creates it after this pass" + } +} + // createsAndLacks says a stream the part reads whole at its start is one of the kept event streams it creates itself and the bus // does not hold yet: its first view is empty, no wait (design 47 G, A-371). The pass composes, writes the part's own create right, // and the part creates the stream and hears it from its start. A bus whose streams cannot be read leaves the read a wait. @@ -119,7 +135,13 @@ func (p *Part) createsAndLacks(in Inputs, stream string) bool { // readRan reads each key of StateRunning and the manifest its digest names (the indexer's, by digest), into in.Ran. A key // whose manifest cannot be read is a gap, and that node's module is read as it was before (the trunk's). func (p *Part) readRan(in *Inputs) { - if reads := p.StreamReads(); len(reads) > 0 && reads[0].State == "waiting" && !p.createsAndLacks(*in, reads[0].Stream) { + reads := p.StreamReads() + absent := len(reads) > 0 && reads[0].State == "waiting" && p.createsAndLacks(*in, reads[0].Stream) + p.noteAbsent(absent) + if absent { + in.StreamsAbsent = append(in.StreamsAbsent, reads[0].Stream+": absent on the bus, taken as empty; the part creates it after this pass") + } + if len(reads) > 0 && reads[0].State == "waiting" && !absent { // The whole read of `running` has not ended: the part acts on none of it, and keeps the list it has (design 47 G §556). in.Gaps = append(in.Gaps, Gap{Input: StateRunning, Failed: true, Why: "the stream " + reads[0].Stream + " is waiting: its whole read at the part's start has not ended"})