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:
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user