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