Wait for any route, not exactly one: 2.11 pools routes between two servers

scenarios/signed-logins.yml

@@ -0,0 +1,29 @@
+# Research 053 question 8: three machines for the signed-login bus.
+# a = today's style server (static users, JetStream), b and c = a cluster in operator mode with a
+# full resolver, a joined to b as a leaf. No overlay and no mesh software: the bed brings its own
+# nats-server (the program test/bed-programs/signed-logins embeds 2.11.17).
+scenario: signed-logins
+
+segments:
+  hosting:
+    kind: public
+    cidr: [192.0.2.0/24]
+
+machines:
+  a:
+    at: { segment: hosting, address: [192.0.2.11] }
+    inbound: allow
+    memory: 2GiB
+    cpus: 2
+  b:
+    at: { segment: hosting, address: [192.0.2.12] }
+    inbound: allow
+    memory: 2GiB
+    cpus: 2
+  c:
+    at: { segment: hosting, address: [192.0.2.13] }
+    inbound: allow
+    memory: 2GiB
+    cpus: 2
+
+snapshot: raised

test/bed-programs/signed-logins/drive.go

@@ -0,0 +1,584 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/nats-io/nkeys"
+)
+
+type env struct {
+	k             Keys
+	op, sys, acc  nkeys.KeyPair
+	sig           nkeys.KeyPair
+	revoked       map[string]time.Time
+	lastIat       int64
+	accountDoneAt time.Time
+}
+
+func (e *env) url(ip string) string { return "nats://" + ip + ":4222" }
+
+func say(format string, a ...any) {
+	fmt.Printf("%s  %s\n", time.Now().Format("15:04:05.000"), fmt.Sprintf(format, a...))
+}
+
+func ms(d time.Duration) string { return fmt.Sprintf("%.1f ms", float64(d.Microseconds())/1000) }
+
+type user struct {
+	kp  nkeys.KeyPair
+	opt nats.Option
+}
+
+func (e *env) newUser(name string) user {
+	kp := must(nkeys.CreateUser())
+	j := userJWT(e.acc, e.sig, kp, name)
+	return user{kp, nats.UserJWT(func() (string, error) { return j, nil }, func(n []byte) ([]byte, error) { return kp.Sign(n) })}
+}
+
+func (e *env) sysConn(ip string) *nats.Conn {
+	kp := must(nkeys.CreateUser())
+	uc := newSysUser(e.sys, kp)
+	nc, err := nats.Connect(e.url(ip), nats.UserJWT(func() (string, error) { return uc, nil }, func(n []byte) ([]byte, error) { return kp.Sign(n) }), nats.Timeout(5*time.Second))
+	if err != nil {
+		panic(fmt.Sprintf("system connection to %s: %v", ip, err))
+	}
+	return nc
+}
+
+// probe is one login attempt from this machine across the network.
+func (e *env) probe(ip string, u user) error {
+	nc, err := nats.Connect(e.url(ip), u.opt, nats.NoReconnect(), nats.Timeout(3*time.Second))
+	if err == nil {
+		nc.Close()
+	}
+	return err
+}
+
+type open struct {
+	nc     *nats.Conn
+	closed atomic.Int64
+}
+
+func (e *env) openConn(ip string, u user) *open {
+	o := &open{}
+	nc, err := nats.Connect(e.url(ip), u.opt, nats.NoReconnect(), nats.Timeout(3*time.Second),
+		nats.ClosedHandler(func(*nats.Conn) { o.closed.Store(time.Now().UnixNano()) }))
+	if err != nil {
+		panic(fmt.Sprintf("open connection to %s: %v", ip, err))
+	}
+	nc.Flush()
+	o.nc = nc
+	return o
+}
+
+// push sends an account JWT to one server's resolver over the system account. The reply is not proof it applied.
+func (e *env) push(sys *nats.Conn, j string) (string, time.Duration) {
+	t0 := time.Now()
+	m, err := sys.Request("$SYS.REQ.ACCOUNT."+pubOf(e.acc)+".CLAIMS.UPDATE", []byte(j), 5*time.Second)
+	if err != nil {
+		return "error: " + err.Error(), time.Since(t0)
+	}
+	return string(m.Data), time.Since(t0)
+}
+
+// nextAccountJWT builds the account JWT with the revocation list; its iat has to be a later second than the last one.
+func (e *env) nextAccountJWT() string {
+	for time.Now().Unix() <= e.lastIat {
+		time.Sleep(50 * time.Millisecond)
+	}
+	e.lastIat = time.Now().Unix()
+	return accountJWT(e.op, e.acc, e.sig, e.revoked)
+}
+
+func (e *env) refusedAfter(ip string, u user, t0 time.Time, limit time.Duration) (time.Duration, int) {
+	n := 0
+	for time.Since(t0) < limit {
+		if err := e.probe(ip, u); err != nil {
+			return time.Since(t0), n
+		}
+		n++
+		time.Sleep(time.Millisecond)
+	}
+	return -1, n
+}
+
+func newSysUser(sys, ukp nkeys.KeyPair) string { return userClaimsFor(sys, ukp) }
+
+func drive(args []string) {
+	fs := flag.NewFlagSet("drive", flag.ExitOnError)
+	keysFile := fs.String("keys", "/root/q8/keys.json", "keys")
+	only := fs.String("only", "2,4,3", "which parts, in order")
+	trials := fs.Int("trials", 3, "trials per part")
+	fs.Parse(args)
+	raw, err := os.ReadFile(*keysFile)
+	if err != nil {
+		panic(err)
+	}
+	e := &env{revoked: map[string]time.Time{}}
+	json.Unmarshal(raw, &e.k)
+	e.op, e.sys, e.acc, e.sig = fromSeed(e.k.Op), fromSeed(e.k.Sys), fromSeed(e.k.Acc), fromSeed(e.k.Sig)
+	e.lastIat = time.Now().Unix()
+	// the account JWT the servers were preloaded with has an iat from gen time; ours must be later
+	time.Sleep(1100 * time.Millisecond)
+	if d := waitRoutes(e, 1, 90*time.Second); d < 0 {
+		say("the route between B and C did not come up")
+	} else {
+		say("cluster formed: route B-C up after %s", ms(d))
+	}
+	for _, p := range strings.Split(*only, ",") {
+		switch p {
+		case "2":
+			part2(e, *trials)
+		case "3":
+			part3(e, *trials)
+		case "4":
+			part4(e)
+		}
+	}
+	say("DRIVE-DONE")
+}
+
+func routes(ip string) int {
+	r, err := http.Get("http://" + ip + ":8222/routez")
+	if err != nil {
+		return -1
+	}
+	defer r.Body.Close()
+	var v struct {
+		NumRoutes int `json:"num_routes"`
+	}
+	json.NewDecoder(r.Body).Decode(&v)
+	return v.NumRoutes
+}
+
+func waitRoutes(e *env, want int, limit time.Duration) time.Duration {
+	t0 := time.Now()
+	for time.Since(t0) < limit {
+		if rb, rc := routes(e.k.B), routes(e.k.C); (want == 0 && rb == 0 && rc == 0) || (want > 0 && rb >= want && rc >= want) {
+			return time.Since(t0)
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	return -1
+}
+
+func agent(e *env, what string) (string, time.Time) {
+	r, err := http.Get("http://" + e.k.C + ":9099/" + what)
+	t := time.Now()
+	if err != nil {
+		return "error: " + err.Error(), t
+	}
+	defer r.Body.Close()
+	b, _ := io.ReadAll(r.Body)
+	return string(b), t
+}
+
+// ---- part 2: revocation across the network ----
+
+func part2(e *env, n int) {
+	say("=== PART 2: revocation over the network (B and C, pushed to B from A)")
+	sysB := e.sysConn(e.k.B)
+	defer sysB.Close()
+	for run := 1; run <= n; run++ {
+		u := e.newUser(fmt.Sprintf("p2-%d", run))
+		ctl := e.newUser(fmt.Sprintf("p2-ctl-%d", run))
+		ob, oc := e.openConn(e.k.B, u), e.openConn(e.k.C, u)
+		cb, cc := e.openConn(e.k.B, ctl), e.openConn(e.k.C, ctl)
+		e.revoked[pubOf(u.kp)] = time.Now()
+		j := e.nextAccountJWT()
+		t0 := time.Now()
+		reply, rtt := e.push(sysB, j)
+		type res struct {
+			ip      string
+			refused time.Duration
+			acc     int
+		}
+		var wg sync.WaitGroup
+		out := make([]res, 2)
+		for i, ip := range []string{e.k.B, e.k.C} {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				d, a := e.refusedAfter(ip, u, t0, 30*time.Second)
+				out[i] = res{ip, d, a}
+			}()
+		}
+		wg.Wait()
+		time.Sleep(500 * time.Millisecond)
+		cl := func(o *open) string {
+			if v := o.closed.Load(); v != 0 {
+				return ms(time.Duration(v - t0.UnixNano()))
+			}
+			return "NOT closed"
+		}
+		alive := func(o *open) bool { return o.nc.IsConnected() }
+		say("RESULT part2 run %d: push reply %q rtt %s; new login refused after: B %s (%d accepted before), C %s (%d accepted before); open connection closed after: B %s, C %s; control user still connected: B %v, C %v",
+			run, strings.TrimSpace(reply)[:min(len(strings.TrimSpace(reply)), 60)], ms(rtt), ms(out[0].refused), out[0].acc, ms(out[1].refused), out[1].acc, cl(ob), cl(oc), alive(cb), alive(cc))
+		for _, o := range []*open{ob, oc, cb, cc} {
+			o.nc.Close()
+		}
+	}
+}
+
+// ---- part 3: C cut off while a user is revoked ----
+
+type seen struct {
+	at      time.Time
+	subject string
+	data    []byte
+}
+
+func part3(e *env, n int) {
+	say("=== PART 3: C cut off from B while a user is revoked; modes a (periodic sync only) and b (re-push on C's return)")
+	for _, mode := range []string{"a", "b"} {
+		for run := 1; run <= n; run++ {
+			trial3(e, mode, run)
+		}
+	}
+}
+
+func trial3(e *env, mode string, run int) {
+	say("--- trial %s%d", mode, run)
+	if d := waitRoutes(e, 1, 90*time.Second); d < 0 {
+		say("RESULT part3 %s%d: the route B-C did not come up (routes B=%d C=%d); trial skipped", mode, run, routes(e.k.B), routes(e.k.C))
+		return
+	}
+	time.Sleep(3 * time.Second)
+	sysB, sysC := e.sysConn(e.k.B), e.sysConn(e.k.C)
+	defer sysB.Close()
+	defer sysC.Close()
+	u, ctl := e.newUser(fmt.Sprintf("p3-%s%d", mode, run)), e.newUser(fmt.Sprintf("p3-ctl-%s%d", mode, run))
+	oc, cc := e.openConn(e.k.C, u), e.openConn(e.k.C, ctl)
+	defer oc.nc.Close()
+	defer cc.nc.Close()
+	var mu sync.Mutex
+	var events []seen
+	sysB.Subscribe("$SYS.>", func(m *nats.Msg) {
+		mu.Lock()
+		events = append(events, seen{time.Now(), m.Subject, m.Data})
+		mu.Unlock()
+	})
+	sysB.Flush()
+
+	r, tCut := agent(e, "block")
+	say("cut: agent said %q", r)
+	dr := waitRoutes(e, 0, 60*time.Second)
+	say("routes gone on B and C after %s (ping interval 2s, 2 missed pings)", ms(dr))
+	time.Sleep(1100 * time.Millisecond)
+	e.revoked[pubOf(u.kp)] = time.Now()
+	j := e.nextAccountJWT()
+	tPush := time.Now()
+	reply, rtt := e.push(sysB, j)
+	dB, _ := e.refusedAfter(e.k.B, u, tPush, 10*time.Second)
+	time.Sleep(4 * time.Second)
+	accC := e.probe(e.k.C, u)
+	say("partitioned: push to B reply %q rtt %s; B refused the user after %s; C still accepts the user: %v (cut lasted %s so far)", strings.TrimSpace(reply)[:min(len(strings.TrimSpace(reply)), 40)], ms(rtt), ms(dB), accC == nil, time.Since(tCut).Round(time.Second))
+
+	mu.Lock()
+	events = nil
+	mu.Unlock()
+	r, tHeal := agent(e, "unblock")
+	say("healed: agent said %q", r)
+
+	var trigger atomic.Int64 // offset ns of the event used as trigger
+	var triggerSubj atomic.Value
+	var pushedRTT atomic.Int64
+	var pushedReply atomic.Value
+	if mode == "b" {
+		go func() {
+			deadline := time.Now().Add(150 * time.Second)
+			for time.Now().Before(deadline) && trigger.Load() == 0 {
+				mu.Lock()
+				var hit *seen
+				for i := range events {
+					ev := events[i]
+					if strings.HasPrefix(ev.subject, "$SYS.SERVER.") && strings.HasSuffix(ev.subject, ".STATSZ") && ev.at.After(tHeal) {
+						var v struct {
+							Server struct{ Name string } `json:"server"`
+						}
+						if json.Unmarshal(ev.data, &v) == nil && v.Server.Name == "C" {
+							hit = &ev
+							break
+						}
+					}
+				}
+				mu.Unlock()
+				if hit != nil {
+					trigger.Store(int64(hit.at.Sub(tHeal)))
+					triggerSubj.Store(hit.subject)
+					rep, rt := e.push(sysC, j)
+					pushedRTT.Store(int64(rt))
+					pushedReply.Store(rep)
+					return
+				}
+				time.Sleep(time.Millisecond)
+			}
+		}()
+	}
+	refused, accepted := e.refusedAfter(e.k.C, u, tHeal, 150*time.Second)
+	time.Sleep(700 * time.Millisecond)
+	closedAt := "NOT closed"
+	if v := oc.closed.Load(); v != 0 {
+		closedAt = ms(time.Duration(v - tHeal.UnixNano()))
+	}
+	line := fmt.Sprintf("RESULT part3 %s%d: after the link came back C accepted the revoked user for %s (%d logins accepted), open connection closed after %s, control user still connected %v",
+		mode, run, ms(refused), accepted, closedAt, cc.nc.IsConnected())
+	if mode == "b" {
+		if t := trigger.Load(); t != 0 {
+			subj, _ := triggerSubj.Load().(string)
+			rep, _ := pushedReply.Load().(string)
+			line += fmt.Sprintf("; trigger: first STATSZ from server C arrived on B's system account %s after the link came back (%s), push to C reply %q rtt %s",
+				ms(time.Duration(t)), strings.ReplaceAll(subj, "$SYS.SERVER.", "$SYS.SERVER.<id>."), strings.TrimSpace(rep)[:min(len(strings.TrimSpace(rep)), 30)], ms(time.Duration(pushedRTT.Load())))
+		} else {
+			line += "; trigger NEVER fired"
+		}
+	}
+	say("%s", line)
+	if run == 1 {
+		mu.Lock()
+		first := map[string]time.Duration{}
+		count := map[string]int{}
+		for _, ev := range events {
+			k := normalise(ev.subject)
+			if _, ok := first[k]; !ok {
+				first[k] = ev.at.Sub(tHeal)
+			}
+			count[k]++
+		}
+		mu.Unlock()
+		var ks []string
+		for k := range first {
+			ks = append(ks, k)
+		}
+		sort.Slice(ks, func(i, j int) bool { return first[ks[i]] < first[ks[j]] })
+		say("system-account subjects seen on B after the link came back (trial %s%d), first-seen offset, count:", mode, run)
+		for _, k := range ks {
+			say("    %9s  x%-3d %s", ms(first[k]), count[k], k)
+		}
+	}
+}
+
+func normalise(s string) string {
+	parts := strings.Split(s, ".")
+	for i, p := range parts {
+		if len(p) >= 40 && strings.ToUpper(p) == p {
+			parts[i] = "<id>"
+		}
+	}
+	return strings.Join(parts, ".")
+}
+
+// ---- part 4: moving a stream and a KV bucket from A to B over the leaf link ----
+
+func part4(e *env) {
+	say("=== PART 4: stream S (10000 msgs) and KV bucket cfg moved from A (legacy) to B/C (hub) over the leaf link, writer running")
+	ctx := context.Background()
+	lc, err := nats.Connect(e.url(e.k.A), nats.UserInfo("w", "pw"))
+	if err != nil {
+		panic(err)
+	}
+	defer lc.Close()
+	ljs := must(jetstream.New(lc))
+	must(ljs.CreateStream(ctx, jetstream.StreamConfig{Name: "S", Subjects: []string{"mesh.s.>"}}))
+	t0 := time.Now()
+	for i := 0; i < 10000; i++ {
+		if _, err := ljs.Publish(ctx, fmt.Sprintf("mesh.s.%d", i%50), []byte(fmt.Sprintf("seed-%d", i)), jetstream.WithMsgID(fmt.Sprintf("seed-%d", i))); err != nil {
+			panic(err)
+		}
+	}
+	say("seeded S with 10000 messages on A in %s", time.Since(t0).Round(time.Millisecond))
+	kv := must(ljs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "cfg", History: 5}))
+	for i := 0; i < 1000; i++ {
+		must(kv.Put(ctx, fmt.Sprintf("k%d", i), []byte(fmt.Sprintf("v%d", i))))
+	}
+	for i := 0; i < 200; i++ {
+		for r := 0; r < 3; r++ {
+			must(kv.Put(ctx, fmt.Sprintf("k%d", i), []byte(fmt.Sprintf("v%d-r%d", i, r))))
+		}
+	}
+	kv.Delete(ctx, "k999")
+	say("seeded KV cfg on A")
+
+	for t := time.Now(); time.Since(t) < 60*time.Second; time.Sleep(500 * time.Millisecond) {
+		r, err := http.Get("http://" + e.k.B + ":8222/leafz")
+		if err != nil {
+			continue
+		}
+		var v struct {
+			Leafs int `json:"leafnodes"`
+		}
+		json.NewDecoder(r.Body).Decode(&v)
+		r.Body.Close()
+		if v.Leafs >= 2 {
+			say("leaf links from A to B up (%d connections: MESH and SYS)", v.Leafs)
+			break
+		}
+	}
+	u := e.newUser("p4")
+	hc, err := nats.Connect(e.url(e.k.B), u.opt)
+	if err != nil {
+		panic(err)
+	}
+	defer hc.Close()
+	hjs := must(jetstream.New(hc))
+	ready := time.Now()
+	for time.Since(ready) < 60*time.Second {
+		if _, err := hjs.AccountInfo(ctx); err == nil {
+			break
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+	say("hub JetStream answered after %s", time.Since(ready).Round(time.Millisecond))
+
+	var stop atomic.Bool
+	var acked, kvacked, failed atomic.Int64
+	done := make(chan bool)
+	go func() {
+		for i := 0; !stop.Load(); i++ {
+			if _, err := ljs.Publish(ctx, fmt.Sprintf("mesh.s.%d", i%50), []byte(fmt.Sprintf("live-%d", i)), jetstream.WithMsgID(fmt.Sprintf("live-%d", i))); err == nil {
+				acked.Add(1)
+			} else {
+				failed.Add(1)
+			}
+			if i%10 == 0 {
+				if _, err := kv.Put(ctx, fmt.Sprintf("live%d", i%20), []byte(fmt.Sprintf("x%d", i))); err == nil {
+					kvacked.Add(1)
+				}
+			}
+			time.Sleep(2 * time.Millisecond)
+		}
+		close(done)
+	}()
+	time.Sleep(300 * time.Millisecond)
+
+	tm := time.Now()
+	for _, n := range []string{"S", "KV_cfg"} {
+		l := must(ljs.Stream(ctx, n))
+		cfg := l.CachedInfo().Config
+		cfg.Subjects = nil
+		cfg.Sources = []*jetstream.StreamSource{{Name: n, External: &jetstream.ExternalStream{APIPrefix: "$JS.legacy.API"}}}
+		if _, err := hjs.CreateStream(ctx, cfg); err != nil {
+			panic(fmt.Sprintf("create source stream %s: %v", n, err))
+		}
+	}
+	msgs := func(js jetstream.JetStream, n string) uint64 {
+		s, err := js.Stream(ctx, n)
+		if err != nil {
+			return 0
+		}
+		i, err := s.Info(ctx)
+		if err != nil {
+			return 0
+		}
+		return i.State.Msgs
+	}
+	waitEq := func(n string, limit time.Duration) time.Duration {
+		t := time.Now()
+		for time.Since(t) < limit {
+			if msgs(hjs, n) == msgs(ljs, n) && msgs(hjs, n) > 0 {
+				return time.Since(tm)
+			}
+			time.Sleep(20 * time.Millisecond)
+		}
+		return -1
+	}
+	say("catch-up under writes, counts equal after: S %s, KV_cfg %s (the writer keeps going)", ms(waitEq("S", 60*time.Second)), ms(waitEq("KV_cfg", 60*time.Second)))
+	time.Sleep(3 * time.Second)
+	stop.Store(true)
+	<-done
+	say("writer stopped: S acked %d (failed %d), KV puts acked %d", acked.Load(), failed.Load(), kvacked.Load())
+	tq := time.Now()
+	for _, n := range []string{"S", "KV_cfg"} {
+		for time.Since(tq) < 30*time.Second && msgs(hjs, n) != msgs(ljs, n) {
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+	say("quiesced: equal after %s; counts S legacy %d hub %d (expect %d); KV_cfg legacy %d hub %d", time.Since(tq).Round(time.Millisecond),
+		msgs(ljs, "S"), msgs(hjs, "S"), 10000+acked.Load(), msgs(ljs, "KV_cfg"), msgs(hjs, "KV_cfg"))
+
+	// content: every message once on both sides
+	readAll := func(js jetstream.JetStream) (map[string]int, int) {
+		s := must(js.Stream(ctx, "S"))
+		info := must(s.Info(ctx))
+		set := map[string]int{}
+		total := 0
+		for seq := info.State.FirstSeq; seq <= info.State.LastSeq; seq++ {
+			m, err := s.GetMsg(ctx, seq)
+			if err != nil {
+				continue
+			}
+			set[string(m.Data)]++
+			total++
+		}
+		return set, total
+	}
+	ls, lt := readAll(ljs)
+	hs, ht := readAll(hjs)
+	dup, missing, extra := 0, 0, 0
+	for k, c := range hs {
+		if c > 1 {
+			dup += c - 1
+		}
+		if _, ok := ls[k]; !ok {
+			extra++
+		}
+	}
+	for k := range ls {
+		if _, ok := hs[k]; !ok {
+			missing++
+		}
+	}
+	say("RESULT part4 stream S: messages legacy %d (distinct %d) hub %d (distinct %d); on hub: duplicates %d, missing vs legacy %d, not on legacy %d; expected %d = 10000 seed + %d acked live",
+		lt, len(ls), ht, len(hs), dup, missing, extra, 10000+acked.Load(), acked.Load())
+
+	hkv := must(hjs.KeyValue(ctx, "cfg"))
+	lk := must(kv.Keys(ctx))
+	hk := must(hkv.Keys(ctx))
+	diff := 0
+	for _, k := range lk {
+		a, err1 := kv.Get(ctx, k)
+		b, err2 := hkv.Get(ctx, k)
+		if err1 != nil || err2 != nil || string(a.Value()) != string(b.Value()) {
+			diff++
+		}
+	}
+	say("RESULT part4 KV cfg: keys legacy %d hub %d; keys whose latest value differs or is missing on hub: %d", len(lk), len(hk), diff)
+
+	// cut-over: the hub stream stops sourcing and takes over the subjects
+	tc := time.Now()
+	subj := map[string][]string{"S": {"mesh.s.>"}, "KV_cfg": {"$KV.cfg.>"}}
+	for n, ss := range subj {
+		h := must(hjs.Stream(ctx, n))
+		cfg := h.CachedInfo().Config
+		cfg.Sources = nil
+		cfg.Subjects = ss
+		if _, err := hjs.UpdateStream(ctx, cfg); err != nil {
+			say("cut-over %s FAILED: %v", n, err)
+		}
+	}
+	say("cut-over update (sources removed, subjects added, 2 streams): %s", time.Since(tc).Round(time.Millisecond))
+	before := msgs(hjs, "S")
+	if _, err := hjs.Publish(ctx, "mesh.s.1", []byte("after-cutover")); err != nil {
+		say("RESULT part4 publish on hub after cut-over FAILED: %v", err)
+	} else {
+		say("RESULT part4 after cut-over a publish on the hub lands in the hub stream: %d -> %d messages (legacy stays %d)", before, msgs(hjs, "S"), msgs(ljs, "S"))
+	}
+	if _, err := hkv.Put(ctx, "after", []byte("1")); err != nil {
+		say("RESULT part4 KV put on hub after cut-over FAILED: %v", err)
+	} else {
+		g, err := hkv.Get(ctx, "after")
+		say("RESULT part4 KV put on hub after cut-over: readable %v", err == nil && string(g.Value()) == "1")
+	}
+}

test/bed-programs/signed-logins/gen.go

@@ -0,0 +1,136 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nkeys"
+)
+
+// Keys is what the driver on machine A needs to act as the operator, the identity seat and a system user.
+type Keys struct {
+	Op, Sys, Acc, Sig string // seeds
+	A, B, C           string // addresses
+}
+
+func must[T any](v T, err error) T {
+	if err != nil {
+		panic(err)
+	}
+	return v
+}
+
+func pubOf(kp nkeys.KeyPair) string { return must(kp.PublicKey()) }
+
+func fromSeed(s string) nkeys.KeyPair { return must(nkeys.FromSeed([]byte(s))) }
+
+func seedOf(kp nkeys.KeyPair) string { return string(must(kp.Seed())) }
+
+// accountJWT is the MESH account: JetStream limits, one signing key, a revocation list.
+func accountJWT(op, acc, sig nkeys.KeyPair, revoked map[string]time.Time) string {
+	ac := jwt.NewAccountClaims(pubOf(acc))
+	ac.Name = "MESH"
+	ac.Limits.JetStreamLimits = jwt.JetStreamLimits{MemoryStorage: 64 << 20, DiskStorage: 1 << 30, Streams: 20, Consumer: 50}
+	ac.SigningKeys.Add(pubOf(sig))
+	if len(revoked) > 0 {
+		ac.Revocations = jwt.RevocationList{}
+		for k, t := range revoked {
+			ac.Revocations.Revoke(k, t)
+		}
+	}
+	return must(ac.Encode(op))
+}
+
+// userJWT: a user of the MESH account signed by the account's signing key, allowed everything.
+func userJWT(acc, sig, ukp nkeys.KeyPair, name string) string {
+	uc := jwt.NewUserClaims(pubOf(ukp))
+	uc.Name = name
+	uc.IssuerAccount = pubOf(acc)
+	uc.Pub.Allow.Add(">")
+	uc.Sub.Allow.Add(">")
+	return must(uc.Encode(sig))
+}
+
+func creds(token string, ukp nkeys.KeyPair) []byte {
+	return must(jwt.FormatUserConfig(token, []byte(seedOf(ukp))))
+}
+
+func gen(args []string) {
+	fs := flag.NewFlagSet("gen", flag.ExitOnError)
+	out := fs.String("out", "", "output directory")
+	a := fs.String("a", "", "address of A")
+	b := fs.String("b", "", "address of B")
+	c := fs.String("c", "", "address of C")
+	fs.Parse(args)
+	os.MkdirAll(*out, 0o755)
+	op := must(nkeys.CreateOperator())
+	sys := must(nkeys.CreateAccount())
+	acc := must(nkeys.CreateAccount())
+	sig := must(nkeys.CreateAccount())
+	oc := jwt.NewOperatorClaims(pubOf(op))
+	oc.Name = "op"
+	oc.SystemAccount = pubOf(sys)
+	opJWT := must(oc.Encode(op))
+	sc := jwt.NewAccountClaims(pubOf(sys))
+	sc.Name = "SYS"
+	sysJWT := must(sc.Encode(op))
+	accJ := accountJWT(op, acc, sig, nil)
+
+	keys := Keys{seedOf(op), seedOf(sys), seedOf(acc), seedOf(sig), *a, *b, *c}
+	kj, _ := json.MarshalIndent(keys, "", " ")
+	w := func(name string, data []byte) { must(0, os.WriteFile(filepath.Join(*out, name), data, 0o600)) }
+	w("keys.json", kj)
+	w("op.jwt", []byte(opJWT))
+
+	// the legacy server joins as a leaf with one user in each of the two accounts
+	lk := must(nkeys.CreateUser())
+	w("leaf-mesh.creds", creds(userJWT(acc, sig, lk, "legacy-leaf"), lk))
+	sk := must(nkeys.CreateUser())
+	suc := jwt.NewUserClaims(pubOf(sk))
+	suc.Name = "legacy-leaf-sys"
+	w("leaf-sys.creds", creds(must(suc.Encode(sys)), sk))
+
+	common := func(name, domain string) string {
+		return fmt.Sprintf(`listen: 0.0.0.0:4222
+server_name: %s
+http_port: 8222
+ping_interval: "2s"
+ping_max: 2
+log_file: "/var/log/q8-%s.log"
+jetstream: {store_dir: "/var/lib/q8/%s/js", domain: %s}
+`, name, name, name, domain)
+	}
+	hub := func(name, peer string) string {
+		return common(name, "hub") + fmt.Sprintf(`operator: "/root/q8/op.jwt"
+system_account: %s
+resolver: {type: full, dir: "/var/lib/q8/%s/accts", allow_delete: false}
+resolver_preload: {%s: %q, %s: %q}
+cluster { name: HUB, listen: 0.0.0.0:6222, routes: ["nats-route://%s:6222"] }
+leafnodes { listen: 0.0.0.0:7422 }
+`, pubOf(sys), name, pubOf(sys), sysJWT, pubOf(acc), accJ, peer)
+	}
+	w("B.conf", []byte(hub("B", *c)))
+	w("C.conf", []byte(hub("C", *b)))
+	w("A.conf", []byte(common("A", "legacy")+fmt.Sprintf(`accounts {
+  MESH { jetstream: enabled
+    users: [ {user: w, password: pw} ] }
+  SYS { users: [ {user: sys, password: sys} ] }
+}
+system_account: SYS
+leafnodes { remotes [
+  {url: "nats-leaf://%s:7422", account: MESH, credentials: "/root/q8/leaf-mesh.creds"},
+  {url: "nats-leaf://%s:7422", account: SYS, credentials: "/root/q8/leaf-sys.creds"}
+] }
+`, *b, *b)))
+}
+
+func userClaimsFor(sys, ukp nkeys.KeyPair) string {
+	uc := jwt.NewUserClaims(pubOf(ukp))
+	uc.Name = "sys"
+	return must(uc.Encode(sys))
+}

test/bed-programs/signed-logins/go.mod

@@ -0,0 +1,21 @@
+module signedlogins
+
+go 1.26.0
+
+require (
+	github.com/nats-io/jwt/v2 v2.8.1
+	github.com/nats-io/nats-server/v2 v2.11.17
+	github.com/nats-io/nats.go v1.54.0
+	github.com/nats-io/nkeys v0.4.16
+)
+
+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/nuid v1.0.1 // 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
+)

test/bed-programs/signed-logins/go.sum

@@ -0,0 +1,25 @@
+github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
+github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
+github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
+github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
+github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
+github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
+github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
+github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
+github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
+github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
+github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
+github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
+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=
+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=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=

test/bed-programs/signed-logins/main.go

@@ -0,0 +1,36 @@
+// Command q8lab is the program behind test/integration/signed-logins.test.ts (research 053, question 8).
+//
+//	q8lab gen    -out DIR -a IP -b IP -c IP   writes keys and the three servers' configs (runs on the workstation)
+//	q8lab server -conf FILE                  runs one nats-server 2.11.17 (embedded) until killed
+//	q8lab agent  -peer IP                    on machine C: an HTTP switch that cuts and restores traffic to the peer
+//	q8lab wait   -url URL                    waits until the URL answers 200
+//	q8lab drive  -keys FILE                  on machine A: runs the measurements and prints them
+package main
+
+import (
+	"fmt"
+	"os"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		fmt.Println("usage: q8lab gen|server|agent|wait|drive")
+		os.Exit(2)
+	}
+	args := os.Args[2:]
+	switch os.Args[1] {
+	case "gen":
+		gen(args)
+	case "server":
+		runServer(args)
+	case "agent":
+		runAgent(args)
+	case "wait":
+		runWait(args)
+	case "drive":
+		drive(args)
+	default:
+		fmt.Println("unknown command", os.Args[1])
+		os.Exit(2)
+	}
+}

test/bed-programs/signed-logins/serve.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+)
+
+func runServer(args []string) {
+	fs := flag.NewFlagSet("server", flag.ExitOnError)
+	conf := fs.String("conf", "", "config file")
+	fs.Parse(args)
+	opts, err := server.ProcessConfigFile(*conf)
+	if err != nil {
+		fmt.Println("config:", err)
+		os.Exit(1)
+	}
+	opts.NoSigs = true
+	opts.Debug = os.Getenv("Q8DEBUG") != ""
+	s, err := server.NewServer(opts)
+	if err != nil {
+		fmt.Println("server:", err)
+		os.Exit(1)
+	}
+	s.ConfigureLogger()
+	go s.Start()
+	if !s.ReadyForConnections(30 * time.Second) {
+		fmt.Println("not ready")
+		os.Exit(1)
+	}
+	fmt.Println("ready", server.VERSION)
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
+	<-ch
+	s.Shutdown()
+}
+
+// runAgent is a switch for machine C: it drops every packet to and from the peer, and takes the rule away again.
+func runAgent(args []string) {
+	fs := flag.NewFlagSet("agent", flag.ExitOnError)
+	peer := fs.String("peer", "", "address to cut off")
+	fs.Parse(args)
+	nft := func(script string) string {
+		cmd := exec.Command("nft", "-f", "-")
+		cmd.Stdin = strings.NewReader(script)
+		out, err := cmd.CombinedOutput()
+		if err != nil {
+			return fmt.Sprintf("error: %v: %s", err, out)
+		}
+		return "ok"
+	}
+	http.HandleFunc("/block", func(w http.ResponseWriter, r *http.Request) {
+		nft("delete table inet q8cut\n") // ignore: it may not exist
+		io.WriteString(w, nft(fmt.Sprintf(`add table inet q8cut
+add chain inet q8cut in { type filter hook input priority -10; policy accept; }
+add chain inet q8cut out { type filter hook output priority -10; policy accept; }
+add rule inet q8cut in ip saddr %s drop
+add rule inet q8cut out ip daddr %s drop
+`, *peer, *peer)))
+	})
+	http.HandleFunc("/unblock", func(w http.ResponseWriter, r *http.Request) {
+		io.WriteString(w, nft("delete table inet q8cut\n"))
+	})
+	fmt.Println("agent listening on :9099 for peer", *peer)
+	fmt.Println(http.ListenAndServe(":9099", nil))
+}
+
+func runWait(args []string) {
+	fs := flag.NewFlagSet("wait", flag.ExitOnError)
+	url := fs.String("url", "", "url")
+	secs := fs.Int("s", 60, "seconds")
+	fs.Parse(args)
+	end := time.Now().Add(time.Duration(*secs) * time.Second)
+	for time.Now().Before(end) {
+		if r, err := http.Get(*url); err == nil {
+			r.Body.Close()
+			if r.StatusCode == 200 {
+				fmt.Println("up")
+				return
+			}
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+	fmt.Println("timeout")
+	os.Exit(1)
+}

test/integration/signed-logins.test.ts

@@ -0,0 +1,108 @@
+/**
+ * Research 053, question 8: the bus on signed logins (NATS operator mode), over a real network.
+ *
+ * Three machines on one segment: `a` is today's style server (static users, an account with
+ * JetStream), `b` and `c` are a two-server cluster in operator mode with a full resolver, and `a` is
+ * joined to `b` as a leaf node, the system account bridged. The measurements run in
+ * test/bed-programs/signed-logins (one Go program; it embeds nats-server 2.11.17, the mesh's
+ * version, so nothing is downloaded into the machines). It runs on `a` and reaches `b` and `c`
+ * across the segment:
+ *
+ *   part 2  revocation: time for b and c to refuse a revoked user and close its open connection
+ *   part 4  moving a stream and a KV bucket from a to b over the leaf link, with a writer running
+ *   part 3  c cut off from b by a packet filter on c while a user is revoked, then restored:
+ *           how long c still accepts the user with the periodic sync alone, and with a re-push
+ *           triggered by the first system-account event that shows c is back
+ *
+ * The test asserts only that the measurements ran to the end; the numbers are the result and are
+ * printed. It proves nothing about the mesh's own software, which it does not use.
+ */
+
+import { test, before, after } from "node:test";
+import assert from "node:assert/strict";
+import { execFileSync } from "node:child_process";
+import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { loadScenario } from "../../src/declaration/parse.ts";
+import { raise } from "../../src/lifecycle/raise.ts";
+import { destroy, exec, push } from "../../src/lifecycle/operate.ts";
+import { labIsUsable } from "./harness.ts";
+
+const capability = await labIsUsable();
+const skip = capability.usable ? false : `lab not usable: ${capability.why}`;
+
+const ADDR = { a: "192.0.2.11", b: "192.0.2.12", c: "192.0.2.13" };
+let instanceId = "";
+
+before(async () => {
+  if (skip) return;
+  instanceId = (await raise(loadScenario("scenarios/signed-logins.yml"), {})).instanceId;
+});
+
+after(async () => {
+  if (instanceId && !process.env["MESH_LAB_KEEP"]) await destroy(instanceId);
+});
+
+const sh = (machine: string, script: string, timeout = 120_000) =>
+  exec(instanceId, machine, ["sh", "-c", script], timeout);
+
+test("research 053 q8: signed-login bus over a real network", { skip, timeout: 3_000_000 }, async () => {
+  const work = mkdtempSync(join(tmpdir(), "q8lab-"));
+  const bin = join(work, "q8lab");
+  execFileSync("go", ["build", "-o", bin, "."], {
+    cwd: "test/bed-programs/signed-logins",
+    env: { ...process.env, CGO_ENABLED: "0", GOOS: "linux", GOARCH: "amd64", GOFLAGS: "-mod=readonly" },
+    stdio: "inherit",
+  });
+  const conf = join(work, "conf");
+  mkdirSync(conf);
+  execFileSync(bin, ["gen", "-out", conf, "-a", ADDR.a, "-b", ADDR.b, "-c", ADDR.c], { stdio: "inherit" });
+
+  for (const m of ["a", "b", "c"]) {
+    await sh(m, "mkdir -p /root/q8 /var/lib/q8");
+    await push(instanceId, m, bin, "/root/q8/q8lab", "0755");
+  }
+  const files: Record<string, string[]> = {
+    a: ["A.conf", "leaf-mesh.creds", "leaf-sys.creds", "keys.json"],
+    b: ["B.conf", "op.jwt"],
+    c: ["C.conf", "op.jwt"],
+  };
+  for (const [m, list] of Object.entries(files)) {
+    for (const f of list) await push(instanceId, m, join(conf, f), `/root/q8/${f}`, "0600");
+  }
+
+  const start = (m: string, cmd: string, out: string) =>
+    sh(m, `setsid nohup ${cmd} >/root/q8/${out} 2>&1 </dev/null & sleep 1; echo started`);
+  await start("b", "/root/q8/q8lab server -conf /root/q8/B.conf", "B.out");
+  await start("c", "/root/q8/q8lab server -conf /root/q8/C.conf", "C.out");
+  for (const m of ["b", "c"]) {
+    const r = await sh(m, "/root/q8/q8lab wait -url http://127.0.0.1:8222/healthz -s 90");
+    assert.match(r.stdout, /up/, `${m} server did not come up`);
+  }
+  await start("a", "/root/q8/q8lab server -conf /root/q8/A.conf", "A.out");
+  assert.match((await sh("a", "/root/q8/q8lab wait -url http://127.0.0.1:8222/healthz -s 60")).stdout, /up/);
+  await start("c", `/root/q8/q8lab agent -peer ${ADDR.b}`, "agent.out");
+  console.log("servers up:", (await sh("a", "cat /root/q8/A.out")).stdout.trim());
+
+  await sh("a", "setsid nohup sh -c '/root/q8/q8lab drive -keys /root/q8/keys.json > /root/q8/drive.log 2>&1; echo finished > /root/q8/drive.done' >/dev/null 2>&1 </dev/null & sleep 1; echo started");
+  let offset = 0;
+  let finished = false;
+  let all = "";
+  for (let i = 0; i < 400 && !finished; i++) {
+    await new Promise((r) => setTimeout(r, 5000));
+    const chunk = (await sh("a", `tail -c +${offset + 1} /root/q8/drive.log`)).stdout;
+    offset += Buffer.byteLength(chunk);
+    finished = (await sh("a", "test -f /root/q8/drive.done && echo yes; true")).stdout.includes("yes");
+    if (chunk.trim()) { console.log(chunk.trimEnd()); all += chunk; }
+  }
+  for (const m of ["a", "b", "c"]) {
+    const up = m.toUpperCase();
+    const r = await sh(m, `grep -c '\\[ERR\\]' /var/log/q8-${up}.log; grep '\\[ERR\\]' /var/log/q8-${up}.log | cut -c1-220 | sort | uniq -c | sort -rn | head -8; true`);
+    console.log(`--- server ${up} log, error lines:\n${r.stdout.trim()}`);
+  }
+  writeFileSync(join(work, "drive.log"), all);
+  assert.ok(finished, "the measurements did not finish");
+  assert.match(all, /DRIVE-DONE/);
+  assert.doesNotMatch(all, /panic:/);
+});