Integration test: mesh-grants writes the lease and hands it back on a real bus

ADR 0582 "How it is checked": against an embedded bus, a stand-in controller holds the lease in a bucket with an
expiry (2 s standing in for 15 s, through grants.LeaseExpiry, so the wait past it is seconds), with stand-in
node-service-manager and bus-tool seats (the expiry tool acting on the real bucket's stream, as nats-tools'
SetBucketExpiry does). On the warrant the key holder names mesh-grants, the expiry is gone, the controller's
create of the key is refused, and the lease still reads handed away past the old expiry; on the hand-back's
warrant the expiry is back and the controller takes the lease again. mesh-grants requires nats-server for it,
at mesh-sdk's version.

Budget: TestTheLeaseWriteOnTheBus (new, integration) is paid by merging the checker's
TestTheCheckerAwaitsTheAssignersComposition into TestTheCheckersLineSaysTheAssignersComposition as its first case.

(issue 774, 774/S23)

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL

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
 	}{

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)

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)
+	}
+}

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
 )

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=