From 7428711d94b451ff2b15345baff33cf915585371 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 01:40:25 +0100 Subject: [PATCH] fix9-10: TestTheGrantsPartWhole: one reader of the part's durable on the stream running, and an ack settled before a consumer is read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The loop opening the part's own membership's streams bound the part's durable on SEAT_node-engine_running beside hearRunning, so the event running went to either reader and the state running stayed empty at random (about 1 run in 15 under load). The loop now leaves that stream to hearRunning. The subtest that compares a consumer's ack floor before and after acked asynchronously and failed 1 run in 30; it now waits for the ack's reply. The failure of an until in the whole-part test also names what the part said of running. Builds: design 47 G §556, A-19 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-grants/cmd/mesh-grants/hears_test.go | 19 ++++++++++++++++++ .../cmd/mesh-grants/integration_test.go | 6 ++++-- modules/mesh-grants/cmd/mesh-grants/main.go | 20 +++++++++++++++---- 3 files changed, 39 insertions(+), 6 deletions(-) create mode 100644 modules/mesh-grants/cmd/mesh-grants/hears_test.go diff --git a/modules/mesh-grants/cmd/mesh-grants/hears_test.go b/modules/mesh-grants/cmd/mesh-grants/hears_test.go new file mode 100644 index 00000000..c1ad80cb --- /dev/null +++ b/modules/mesh-grants/cmd/mesh-grants/hears_test.go @@ -0,0 +1,19 @@ +package main + +import ( + "testing" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-sdk/go/membership" +) + +// The part's own membership names the stream running among those it hears, and hearRunning binds the part's durable on it: +// the loop that opens the other streams leaves it alone, so no second reader takes messages of that one durable (design 47 G +// §556, A-19; the whole-part integration test lost the running event to the second reader at random). +func TestTheLoopOfOwnHearsLeavesTheStreamsADedicatedHearingOwns(t *testing.T) { + hears := []membership.HeardIssued{{Stream: nodeengine.StreamRunning, Consumer: "a_mesh-grants"}, {Stream: "mod-c_feed", Consumer: "a_mesh-grants"}, {Stream: "opened_one"}} + got := hearsToOpen(hears, map[string]bool{"opened_one": true}) + if len(got) != 1 || got[0].Stream != "mod-c_feed" { + t.Errorf("the streams opened are %+v, want mod-c_feed alone", got) + } +} diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index c63b7a6b..44716ce7 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -342,7 +342,7 @@ func TestTheGrantsPartWhole(t *testing.T) { return } } - t.Fatalf("never saw: %s\n%s", what, logged.tail("binding")) + t.Fatalf("never saw: %s\n%s\n%s", what, logged.tail("binding"), logged.tail("running")+logged.tail("events:")+logged.tail("cannot be heard")) } until("every first view asked and refused once", func() bool { return refusedAsks.Load() >= int32(len(grants.FirstViews)) }) until("the part saying it waits on the indexer and the assigner", func() bool { @@ -573,7 +573,9 @@ func TestTheGrantsPartWhole(t *testing.T) { t.Fatal(err) } for m := range got.Messages() { - _ = m.Ack() + if err := m.DoubleAck(ctx); err != nil { // the ack settled on the server before the consumer is read + t.Fatal(err) + } } before, _ := stream.Info(ctx) cbefore, _ := cons.Info(ctx) diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 2a57f33a..7d261607 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -41,6 +41,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/announce" + "git.novox.be/novox/mesh-sdk/go/membership" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/seatlist" @@ -258,10 +259,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - for _, h := range hears { - if opened[h.Stream] { - continue - } + for _, h := range hearsToOpen(hears, opened) { cons, err := js.Consumer(ctx, h.Stream, h.Consumer) if err != nil { return fmt.Errorf("the consumer %s on %s: %w", h.Consumer, h.Stream, err) @@ -292,6 +290,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil } +// hearsToOpen are the streams of the part's own membership no hearing has opened yet. A stream another hearing of the part +// binds its durable on (running: hearRunning, which reads it whole at the start before it follows) is left to that hearing: +// two readers of one durable each take the messages the other does not, and the event a reader of its own never hears is lost +// to it (the integration test of the whole part met it as a state running that stayed empty, 2026-11-12). +func hearsToOpen(hears []membership.HeardIssued, opened map[string]bool) []membership.HeardIssued { + var out []membership.HeardIssued + for _, h := range hears { + if !opened[h.Stream] && h.Stream != nodeengine.StreamRunning { + out = append(out, h) + } + } + return out +} + // hearing is the stream the part's running consumer is bound to, once. var hearingRunning struct { sync.Mutex