DLV-F5r: the grants part started before its owners answer waits, says so, and composes nothing from them

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
This commit is contained in:
2026-11-04 13:12:29 +01:00
parent 273da7535b
commit d53986f1f6
@@ -6,11 +6,11 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"io"
"os" "os"
"path/filepath" "path/filepath"
"slices" "slices"
"strings" "strings"
"sync"
"sync/atomic" "sync/atomic"
"testing" "testing"
"time" "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"},`+ 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}]}`) `"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"]}`) 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") own, err := os.ReadFile("../../module.json")
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -150,9 +150,12 @@ func TestTheGrantsPartWhole(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
t.Cleanup(func() { _ = sub.Unsubscribe() }) 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, // DLV-F4r, DLV-F5r (ADR 0615 §4, §6, ADR 0621 §3): the owners answer their first-view tools at once from what they
// with its revision, as the indexer's and the assigner's runtimes of their own do; the part never reads their buckets. // hold, with its revision, as the indexer's and the assigner's runtimes of their own do; the part never reads their
var firstViewAsks atomic.Int32 // 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} { for _, owner := range []string{grants.IndexerModule, grants.AssignerModule} {
owner := owner owner := owner
osub, err := nc.Subscribe("mesh.mod."+owner+".>", func(m *nats.Msg) { 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) { if fv.Owner != owner || !strings.Contains(m.Subject, "."+fv.State) {
continue 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) firstViewAsks.Add(1)
values, rev := map[string]json.RawMessage{}, uint64(0) values, rev := map[string]json.RawMessage{}, uint64(0)
if keys, err := kvs[fv.Read].ListKeys(ctx); err == nil { 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} env := map[string]string{"MESH_BROKER_FILE": broker}
runCtx, stop := context.WithCancel(ctx) runCtx, stop := context.WithCancel(ctx)
done := make(chan error, 1) 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 }) t.Cleanup(func() { stop(); <-done })
until := func(what string, ok func() bool) { until := func(what string, ok func() bool) {
@@ -206,6 +216,14 @@ func TestTheGrantsPartWhole(t *testing.T) {
} }
t.Fatalf("never saw: %s", what) 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 { 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") e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c")
if err != nil { 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). // 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) 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)
}