diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go index e41c96c..1647482 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go @@ -36,6 +36,7 @@ func (b *bed) assignerWrote(t *testing.T, key string, v delivery.CompositionVerd // 1297/AC3: started whole with the assigner's verdict on the bus, the checker's line says the composition's verdict, not // "not yet run": a pass names its machine, a fail fails the check with the assigner's words. func TestTheCheckersLineSaysTheAssignersComposition(t *testing.T) { + t.Run("the read awaits the assigner's composition", checkerAwaitsTheAssignersComposition) for _, c := range []struct { state, owed, words string }{ diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go index 0325a35..0c147e7 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go @@ -21,7 +21,8 @@ import ( // 1297/AC3 (its bus half): on the real bus, the checker's read of the assigner's composition waits while the key is // absent and while the assigner says it waits, and answers the verdict the assigner then writes under the check's key. -func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) { +// A case of TestTheCheckersLineSaysTheAssignersComposition (merged into it for the test budget, issue 774). +func checkerAwaitsTheAssignersComposition(t *testing.T) { dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "comp") if err != nil { t.Fatal(err) diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go new file mode 100644 index 0000000..6864000 --- /dev/null +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -0,0 +1,297 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/asker" + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/seal" + + "mesh-grants/internal/grants" +) + +// standIns are the seats and tools the lease write asks, standing in on the real bus's side: node-service-manager on +// the controller's node (a unit's ActiveState, as the systemd module answers it), the bus module's expiry tool acting on +// the real lease bucket's stream (as nats-tools' SetBucketExpiry does), and the checks answering green. +type standIns struct { + ctx context.Context + js jetstream.JetStream + mu sync.Mutex + active string + calls []string +} + +func (s *standIns) Call(key string, body any) (json.RawMessage, error) { + s.mu.Lock() + s.calls = append(s.calls, key) + s.mu.Unlock() + base, _, _ := strings.Cut(key, "@") + args, _ := body.(map[string]any) + var out any + switch base { + case grants.CallServiceStop: + s.active, out = "inactive", map[string]any{"ok": true} + case grants.CallServiceStart: + s.active, out = "active", map[string]any{"ok": true} + case grants.CallServiceStatus: + out = map[string]any{"ActiveState": s.active} + case grants.CallBucketExpiry: + now := fmt.Sprint(args["max_age"]) + d, err := time.ParseDuration(strings.Replace(now, "none", "0s", 1)) + if err != nil { + return nil, err + } + st, err := s.js.Stream(s.ctx, "KV_"+fmt.Sprint(args["bucket"])) + if err != nil { + return nil, err + } + cfg := st.CachedInfo().Config + cfg.MaxAge = d + if d > 0 && (cfg.Duplicates == 0 || cfg.Duplicates > d) { + cfg.Duplicates = d + } + if _, err := s.js.UpdateStream(s.ctx, cfg); err != nil { + return nil, err + } + out = map[string]any{"bucket": args["bucket"], "now": now, "changed": true} + case grants.CallIssue: + out = map[string]any{"status": "resolved"} + case grants.CallNodeTrusts: + out = map[string]any{"every": true, "not_yet": []string{}} + case grants.CallConfigCompare: + out = map[string]any{"source": "indexer", "count": 0} + case grants.CallUpgrade: + out = map[string]any{"empty": true, "splits": []any{}} + case grants.CallVaultStatus: + out = map[string]any{"ready": true, "minting": map[string]any{"on": true, "held_back": map[string]string{}}} + case grants.CallStreams: + out = map[string]any{"detail": []any{}} + case grants.CallUsers: + out = map[string]any{"users": []any{}} + case grants.CallRecordWarrant, grants.CallRecordHandAct: + out = map[string]any{} + default: + return nil, fmt.Errorf("no stand-in for %s", key) + } + return json.Marshal(out) +} + +// memStore is the part's own state, in memory, with its time to live per key recorded. +type memStore struct { + mu sync.Mutex + vals map[string][]byte + revs map[string]uint64 +} + +func (m *memStore) Get(state, key string) ([]byte, uint64, bool, error) { + m.mu.Lock() + defer m.mu.Unlock() + v, ok := m.vals[state+"/"+key] + return v, m.revs[state+"/"+key], ok, nil +} + +func (m *memStore) Put(state, key string, value []byte) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.vals[state+"/"+key] = append([]byte(nil), value...) + m.revs[state+"/"+key]++ + return m.revs[state+"/"+key], nil +} + +func (m *memStore) PutTTL(state, key string, value []byte, _ time.Duration) (uint64, error) { + return m.Put(state, key, value) +} + +func (m *memStore) Keys(state string) ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []string + for k := range m.vals { + if s, key, _ := strings.Cut(k, "/"); s == state { + out = append(out, key) + } + } + return out, nil +} + +func (m *memStore) Delete(state, key string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.vals, state+"/"+key) + return nil +} + +type memObjects struct{} + +func (memObjects) PutObject(string, []byte) error { return nil } +func (memObjects) DeleteObject(string) error { return nil } + +// yesAsker answers every ask the part raises with the operator's yes, as the router's warrant hands it. +type yesAsker struct{ asked []string } + +func (a *yesAsker) Ask(q asks.Ask) error { a.asked = append(a.asked, q.ID); return nil } +func (a *yesAsker) Where(string) (string, *asks.Warrant, error) { return "open", nil, nil } +func (a *yesAsker) Cancel(string) error { return nil } +func (a *yesAsker) Take(w asks.Warrant, act func(asks.Option, asks.Warrant) error) (asker.Handled, error) { + return asker.Acted, act(asks.Option{ID: w.Option}, w) +} + +// Issue 774, 774/S23 (novox/hq ADR 0582 clauses 3, 5, "How it is checked"): mesh-grants' lease write and hand-back +// against the real bus, with a stand-in controller holding the lease in a bucket with an expiry (2 s standing in for 15 +// s, so the test waits past it in seconds), and stand-in node-service-manager and bus tool seats. +func TestTheLeaseWriteOnTheBus(t *testing.T) { + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "lease") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + const ttl = 2 * time.Second + was := grants.LeaseExpiry + grants.LeaseExpiry = ttl + t.Cleanup(func() { grants.LeaseExpiry = was }) + lease, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: buscompose.LeaseBucket, History: 1, TTL: ttl}) + if err != nil { + t.Fatal(err) + } + controller, _ := json.Marshal(buscompose.LeaseHolder{Instance: "controller@a pid 1", Host: "a", Epoch: 60}) + if _, err := lease.Create(ctx, buscompose.ControllerLeaseKey, controller); err != nil { + t.Fatal(err) + } + bucket := func(state string) jetstream.KeyValue { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucketOf(state), History: 5}) + if err != nil { + t.Fatal(err) + } + return kv + } + put := func(kv jetstream.KeyValue, key string, v any) { + raw, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, raw); err != nil { + t.Fatal(err) + } + } + nats := `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"},` + + `"tools":["nats_bucket_expiry","nats_consumer_remove","nats_stream_remove"]}` + digest := delivery.ManifestDigest([]byte(nats)) + put(bucket(grants.ReadManifests), digest, delivery.Manifest{V: delivery.Version, Module: "nats", JSON: nats}) + put(bucket(grants.ReadModules), "nats", delivery.Module{V: delivery.Version, FP2: "src2:" + strings.Repeat("ab", 32), Manifest: digest, + Repository: "novox/mesh-catalog", Commit: "c"}) + pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "nats"}}}) + if _, err := bucket(grants.ReadPlacement).Put(ctx, "placement.a", pl); err != nil { + t.Fatal(err) + } + put(bucket(grants.ReadTakeover), health.TakeoverKey, health.Takeover{V: 1, Follow: health.FollowKeeper, By: "operator", Why: "taken"}) + put(bucket(grants.ReadAccounts), buscompose.ControllerName, map[string]any{"hash": "$2a$11$controller", "generation": 1, "at": "2026-11-01T00:00:00Z"}) + busKey, _ := seal.GenerateKeyPair() + have := bucket(grants.ReadHave) + loads := func(source, digest string) { + put(have, "a.nats", delivery.Have{V: delivery.Version, Phase: delivery.PhaseRunning, Health: delivery.HealthOK, + UserList: &delivery.UserListWritten{Source: source, Revision: 1, Digest: digest, Users: 1, Key: busKey.PublicKey}}) + } + loads("a.nats.src2", "sha256:aa") + + seats := &standIns{ctx: ctx, js: js, active: "active"} + store := &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} + p := &grants.Part{Read: newBusReader(ctx, js), Call: seats, Store: store, Lease: &leaseHand{ctx: ctx, js: js}, + Asker: &yesAsker{}, Objects: memObjects{}, Nonce: func() string { return "0001" }} + act := func(ask func(string, string) (any, error)) { + t.Helper() + if _, err := ask("operator@shanks", "issue 774's integration test"); err != nil { + t.Fatal(err) + } + p.Compose() // the process raises the ask + r, _ := p.LeaseState() + if r.Open == nil { + t.Fatalf("no act open: %+v", r) + } + if err := p.Warranted(asks.Option{ID: "yes"}, asks.Warrant{Ask: r.Open.Ask, Option: "yes", Outcome: asks.OutcomeChosen}); err != nil { + t.Fatal(err) + } + r, _ = p.LeaseState() + if r.Open == nil || r.Open.List == "" { + t.Fatalf("the act is not waiting on its list: %+v", r) + } + loads(delivery.GrantsUserListSource, r.Open.List) // the bus node loads it + p.Compose() + if r, _ := p.LeaseState(); r.Open != nil { + t.Fatalf("the act did not finish: %+v", r.Open) + } + } + expiry := func() time.Duration { + st, err := lease.Status(ctx) + if err != nil { + t.Fatal(err) + } + return st.TTL() + } + + // 774/S23/AC1. + act(p.LeaseWrite) + e, err := lease.Get(ctx, buscompose.ControllerLeaseKey) + if err != nil { + t.Fatal(err) + } + if away, says := buscompose.LeaseHandedAway(e.Value(), true, nil); !away || !strings.Contains(string(e.Value()), `"handed-to":"mesh-grants"`) { + t.Fatalf("the lease after the write: %s (%s)", e.Value(), says) + } + if expiry() != 0 || seats.active != "inactive" { + t.Fatalf("the lease's expiry is %s, the controller %s", expiry(), seats.active) + } + if _, err := lease.Create(ctx, buscompose.ControllerLeaseKey, controller); !errors.Is(err, jetstream.ErrKeyExists) { + t.Fatalf("the controller's retake: %v, want refused", err) + } + time.Sleep(ttl + time.Second) + if e, err := lease.Get(ctx, buscompose.ControllerLeaseKey); err != nil { + t.Fatalf("the lease is gone past the old expiry: %v", err) + } else if away, _ := buscompose.LeaseHandedAway(e.Value(), true, nil); !away { + t.Fatalf("the lease past the old expiry: %s", e.Value()) + } + + // 774/S23/AC2. + act(p.LeaseHandBack) + if expiry() != ttl || seats.active != "active" { + t.Fatalf("after the hand-back the expiry is %s, the controller %s", expiry(), seats.active) + } + if _, err := lease.Create(ctx, buscompose.ControllerLeaseKey, controller); err != nil { + t.Fatalf("the controller cannot take the lease back: %v", err) + } +} diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index 68bdfcd..5afd707 100644 --- a/modules/mesh-grants/go.mod +++ b/modules/mesh-grants/go.mod @@ -4,13 +4,20 @@ go 1.26.0 require ( git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101151945-b2a27ed5cec4 + github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect + go.uber.org/automaxprocs v1.6.0 // indirect golang.org/x/crypto v0.57.0 // indirect golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect ) diff --git a/modules/mesh-grants/go.sum b/modules/mesh-grants/go.sum index 1c91827..fff1efe 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -18,8 +18,11 @@ github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs= +go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=