From d53986f1f643c18aee2a56abc44e6db702b09025 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:12:29 +0100 Subject: [PATCH] DLV-F5r: the grants part started before its owners answer waits, says so, and composes nothing from them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A case of the grants part's whole-module test (ADR 0615 §6, ADR 0621 §3): each first view refused once, the part's log names the indexer's and the assigner's tools it waits on, no membership is written; once the owners answer it composes as before. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-grants/integration_test.go | 48 ++++++++++++++++--- 1 file changed, 42 insertions(+), 6 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 7f67d48d..b60896a0 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -6,11 +6,11 @@ import ( "context" "encoding/json" "errors" - "io" "os" "path/filepath" "slices" "strings" + "sync" "sync/atomic" "testing" "time" @@ -101,7 +101,7 @@ func TestTheGrantsPartWhole(t *testing.T) { manifest("nats", `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"},`+ `"emits":["created","done","dropped","recreated"],"streams":[{"name":"broker-events","events":["created","dropped","recreated"],"max-age-days":30,"max-mib":256}]}`) manifest("mod-c", `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["nats.done"]}`) - // The part itself, placed as on the mesh, so its own membership names the bus module's stream it hears (DLV-G1). + // The part itself, placed as on the mesh, so its own consumer on EVENTS hears the seat mesh-broker's events (DLV-G1). own, err := os.ReadFile("../../module.json") if err != nil { t.Fatal(err) @@ -150,9 +150,12 @@ func TestTheGrantsPartWhole(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { _ = sub.Unsubscribe() }) - // DLV-F4r, DLV-F5r (ADR 0615 §4, ADR 0621 §3): the owners answer their first-view tools at once from what they hold, - // with its revision, as the indexer's and the assigner's runtimes of their own do; the part never reads their buckets. - var firstViewAsks atomic.Int32 + // DLV-F4r, DLV-F5r (ADR 0615 §4, §6, ADR 0621 §3): the owners answer their first-view tools at once from what they + // hold, with its revision, as the indexer's and the assigner's runtimes of their own do; the part never reads their + // buckets. Started before its owners answer (answering false: each ask refused, as an owner not yet connected), the + // part waits, says so, and composes nothing it would read from them. + var firstViewAsks, refusedAsks atomic.Int32 + var answering atomic.Bool for _, owner := range []string{grants.IndexerModule, grants.AssignerModule} { owner := owner osub, err := nc.Subscribe("mesh.mod."+owner+".>", func(m *nats.Msg) { @@ -160,6 +163,12 @@ func TestTheGrantsPartWhole(t *testing.T) { if fv.Owner != owner || !strings.Contains(m.Subject, "."+fv.State) { continue } + if !answering.Load() { + refusedAsks.Add(1) + raw, _ := json.Marshal(map[string]any{"error": "not connected yet", "node": "a"}) + _ = m.Respond(raw) + return + } firstViewAsks.Add(1) values, rev := map[string]json.RawMessage{}, uint64(0) if keys, err := kvs[fv.Read].ListKeys(ctx); err == nil { @@ -194,7 +203,8 @@ func TestTheGrantsPartWhole(t *testing.T) { env := map[string]string{"MESH_BROKER_FILE": broker} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) - go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() + logged := &lockedLog{} + go func() { done <- run(runCtx, func(k string) string { return env[k] }, logged) }() t.Cleanup(func() { stop(); <-done }) until := func(what string, ok func() bool) { @@ -206,6 +216,14 @@ func TestTheGrantsPartWhole(t *testing.T) { } t.Fatalf("never saw: %s", what) } + 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 { + return logged.has("mesh-delivery-indexer.manifests") && logged.has("mesh-delivery-assigner.placement") && logged.has("is not answered") + }) + if _, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c"); err == nil { + t.Error("a membership composed before the owners answered its first views") + } + answering.Store(true) until("a.mod-c's membership in the part's own state memberships", func() bool { e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c") if err != nil { @@ -272,3 +290,21 @@ func TestTheGrantsPartWhole(t *testing.T) { // And what the part composes holds on a server that loads it (one integration test per module, the test budget). t.Run("the composed grants hold on a server that loads them", theBusNodeFollowsTheSealedListUnderTheComposedGrants) } + +// lockedLog is the part's log, read by the test while the part writes it. +type lockedLog struct { + mu sync.Mutex + buf strings.Builder +} + +func (l *lockedLog) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.buf.Write(p) +} + +func (l *lockedLog) has(s string) bool { + l.mu.Lock() + defer l.mu.Unlock() + return strings.Contains(l.buf.String(), s) +}