fixes from ladder 17: key rotation deadline in the store, truthful /readyz, portless bench, counted requests, re-admission retry

cmd/agentmesh-bench/main.go

@@ -282,7 +282,6 @@ func newJoinCmd() *cobra.Command {
 	flags.DurationVar(&opts.CallTimeout, "call-timeout", 120*time.Second, "Bound on one member's discovery and first call, from readiness")
 	flags.DurationVar(&opts.Hold, "hold", 0, "Keep the fleet resident this long after the journeys end, sampling readiness; 0 tears it down at once")
 	flags.DurationVar(&opts.SampleInterval, "sample-interval", 10*time.Second, "How often each resident member's readiness is read during --hold")
-	flags.IntVar(&opts.MetricsBasePort, "metrics-base-port", 20000, "Member i serves /readyz on 127.0.0.1:(port+i)")
 	flags.StringArrayVar(&opts.NodeArgs, "node-arg", nil, "Extra argument for every agentmesh-node, repeatable (e.g. --node-arg=--insecure-control-plane)")
 	flags.StringVar(&marker, "resident-marker", "", "File to create once the fleet is resident and the hold has begun")
 	flags.StringArrayVar(&scrape, "scrape", nil, "Metrics endpoint to record before and after the run, repeatable")

internal/bench/join.go

@@ -93,10 +93,6 @@ type JoinOptions struct {
 	// SampleInterval is how often each resident member is read during Hold.
 	SampleInterval time.Duration
 
-	// MetricsBasePort is the first of Count consecutive loopback ports;
-	// member i serves /readyz on MetricsBasePort+i.
-	MetricsBasePort int
-
 	// NodeArgs are appended to every member's command line as given.
 	NodeArgs []string
 
@@ -208,6 +204,7 @@ const (
 	// and one that takes a minute is not asked a thousand times.
 	readyPollMin      = 50 * time.Millisecond
 	readyPollMax      = 500 * time.Millisecond
+	readyTimeout      = 2 * time.Second
 	callRetryInterval = 5 * time.Second
 	requestTimeout    = 30 * time.Second
 	teardownGrace     = 10 * time.Second
@@ -222,17 +219,19 @@ const (
 const mcpInitialize = `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"agentmesh-bench","version":"0"}}}`
 
 type member struct {
-	index       int
-	dir         string
-	socket      string
-	metricsAddr string
-	readyz      string
-	logPath     string
-	cmd         *exec.Cmd
-	logFile     *os.File
-	started     time.Time
-	exited      chan struct{}
-	exitErr     error
+	index  int
+	dir    string
+	socket string
+	// client speaks to the member over its Unix socket: readiness, discovery
+	// and the first call. A member binds no loopback port, so a fleet of any
+	// size cannot collide with the kernel's ephemeral range.
+	client  *http.Client
+	logPath string
+	cmd     *exec.Cmd
+	logFile *os.File
+	started time.Time
+	exited  chan struct{}
+	exitErr error
 
 	readyAt    time.Time
 	calledAt   time.Time
@@ -364,9 +363,6 @@ func (o *JoinOptions) setDefaults() {
 	if o.SampleInterval <= 0 {
 		o.SampleInterval = 10 * time.Second
 	}
-	if o.MetricsBasePort <= 0 {
-		o.MetricsBasePort = 20000
-	}
 }
 
 func (o *JoinOptions) logf(format string, args ...any) {
@@ -378,14 +374,13 @@ func (o *JoinOptions) logf(format string, args ...any) {
 
 func newMember(opts JoinOptions, i int) *member {
 	dir := filepath.Join(opts.Dir, fmt.Sprintf("member-%04d", i))
-	metricsAddr := fmt.Sprintf("127.0.0.1:%d", opts.MetricsBasePort+i)
+	socket := filepath.Join(dir, "node.sock")
 	return &member{
-		index:       i,
-		dir:         dir,
-		socket:      filepath.Join(dir, "node.sock"),
-		metricsAddr: metricsAddr,
-		readyz:      "http://" + metricsAddr + "/readyz",
-		logPath:     filepath.Join(dir, "node.log"),
+		index:   i,
+		dir:     dir,
+		socket:  socket,
+		client:  unixClient(socket),
+		logPath: filepath.Join(dir, "node.log"),
 	}
 }
 
@@ -407,7 +402,6 @@ func (m *member) start(opts JoinOptions) error {
 		"--data-dir", m.dir,
 		"--socket-path", m.socket,
 		"--bind-addr=",
-		"--metrics-addr", m.metricsAddr,
 		"--listen", "/ip4/0.0.0.0/tcp/0",
 		"--listen", "/ip4/0.0.0.0/udp/0/quic-v1",
 	}
@@ -456,11 +450,10 @@ func journey(ctx context.Context, opts JoinOptions, m *member) {
 // waitReady polls the member's /readyz, which answers 200 only once the node
 // holds an authenticated connection to a router.
 func (m *member) waitReady(ctx context.Context, timeout time.Duration) error {
-	client := &http.Client{Timeout: 2 * time.Second}
 	deadline := time.Now().Add(timeout)
 	interval := readyPollMin
 	for {
-		if isReady(ctx, client, m.readyz) {
+		if m.isReady(ctx) {
 			m.readyAt = time.Now()
 			return nil
 		}
@@ -481,12 +474,14 @@ func (m *member) waitReady(ctx context.Context, timeout time.Duration) error {
 	}
 }
 
-func isReady(ctx context.Context, client *http.Client, url string) bool {
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+func (m *member) isReady(ctx context.Context) bool {
+	ctx, cancel := context.WithTimeout(ctx, readyTimeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost/readyz", nil)
 	if err != nil {
 		return false
 	}
-	resp, err := client.Do(req)
+	resp, err := m.client.Do(req)
 	if err != nil {
 		return false
 	}
@@ -500,7 +495,7 @@ func isReady(ctx context.Context, client *http.Client, url string) bool {
 // deadline: a provider record can name a peer a rollout just replaced, and a
 // member that joined seconds ago has a thin routing table.
 func (m *member) firstCall(ctx context.Context, opts JoinOptions) error {
-	client := unixClient(m.socket)
+	client := m.client
 	defer client.CloseIdleConnections()
 	deadline := m.readyAt.Add(opts.CallTimeout)
 	lastErr := "none discovered"
@@ -691,12 +686,11 @@ func hold(ctx context.Context, opts JoinOptions, members []*member) *HoldReport
 	report := &HoldReport{}
 	var outages []time.Duration
 
-	client := &http.Client{Timeout: 2 * time.Second}
 	start := time.Now()
 	end := start.Add(opts.Hold)
 	for {
 		now := time.Now()
-		ready := sampleReady(ctx, client, members)
+		ready := sampleReady(ctx, members)
 		count := 0
 		for i, m := range members {
 			if ready[i] {
@@ -759,7 +753,7 @@ func hold(ctx context.Context, opts JoinOptions, members []*member) *HoldReport
 }
 
 // sampleReady reads every member's /readyz, a bounded number at a time.
-func sampleReady(ctx context.Context, client *http.Client, members []*member) []bool {
+func sampleReady(ctx context.Context, members []*member) []bool {
 	ready := make([]bool, len(members))
 	sem := make(chan struct{}, sampleConcurrency)
 	var wg sync.WaitGroup
@@ -772,7 +766,7 @@ func sampleReady(ctx context.Context, client *http.Client, members []*member) []
 		go func() {
 			defer wg.Done()
 			defer func() { <-sem }()
-			ready[i] = isReady(ctx, client, m.readyz)
+			ready[i] = m.isReady(ctx)
 		}()
 	}
 	wg.Wait()

internal/bench/join_test.go

@@ -31,8 +31,11 @@ import (
 
 // The members RunJoin starts are whatever binary it is pointed at. Here that
 // is this test binary again, re-entered through TestMain as a node that
-// answers /readyz and the two sidecar routes the journey uses, with its
-// behaviour set through the environment the children inherit.
+// answers /readyz and the two sidecar routes the journey uses on its Unix
+// socket, with its behaviour set through the environment the children
+// inherit. Like the real node it binds no loopback port unless told to, and
+// it refuses to be told: a fleet of a thousand on one host collided with the
+// kernel's ephemeral range when every member had one.
 
 func TestMain(m *testing.M) {
 	if os.Getenv("AGENTMESH_BENCH_FAKE_NODE") == "1" {
@@ -42,11 +45,14 @@ func TestMain(m *testing.M) {
 }
 
 func fakeNode(args []string) int {
-	var metricsAddr, socket, tokenPath string
+	var socket, tokenPath string
 	for i := 0; i < len(args)-1; i++ {
 		switch args[i] {
-		case "--metrics-addr":
-			metricsAddr = args[i+1]
+		case "--metrics-addr", "--bind-addr":
+			if args[i+1] != "" {
+				fmt.Fprintln(os.Stderr, "fake node: a member must not bind a loopback port:", args[i], args[i+1])
+				return 4
+			}
 		case "--socket-path":
 			socket = args[i+1]
 		case "--bootstrap-token-path":
@@ -78,15 +84,14 @@ func fakeNode(args []string) int {
 		return true
 	}
 
-	metrics := http.NewServeMux()
-	metrics.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) {
+	sidecar := http.NewServeMux()
+	sidecar.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) {
 		if !ready() {
 			http.Error(w, "not ready", http.StatusServiceUnavailable)
 			return
 		}
 		w.WriteHeader(http.StatusOK)
 	})
-	sidecar := http.NewServeMux()
 	sidecar.HandleFunc("/mesh/service/discover", func(w http.ResponseWriter, _ *http.Request) {
 		var out []map[string]string
 		for _, p := range providers {
@@ -104,17 +109,11 @@ func fakeNode(args []string) int {
 		_, _ = fmt.Fprint(w, `{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","serverInfo":{"name":"fake","version":"0"}}}`)
 	})
 
-	tcp, err := net.Listen("tcp", metricsAddr)
-	if err != nil {
-		fmt.Fprintln(os.Stderr, "fake node:", err)
-		return 1
-	}
 	unix, err := net.Listen("unix", socket)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, "fake node:", err)
 		return 1
 	}
-	go func() { _ = http.Serve(tcp, metrics) }()  //nolint:gosec // a test double
 	go func() { _ = http.Serve(unix, sidecar) }() //nolint:gosec // a test double
 	stop := make(chan os.Signal, 1)
 	signal.Notify(stop, syscall.SIGTERM, os.Interrupt)
@@ -152,37 +151,9 @@ func fakeFleet(t *testing.T, count int, env map[string]string) JoinOptions {
 		ControlPlane:       "http://127.0.0.1:1",
 		BootstrapTokenPath: token,
 		Service:            "everything",
-		MetricsBasePort:    freePortBase(t, count),
 	}
 }
 
-// freePortBase finds count consecutive free loopback ports.
-func freePortBase(t *testing.T, count int) int {
-	t.Helper()
-	for attempt := 0; attempt < 20; attempt++ {
-		l, err := net.Listen("tcp", "127.0.0.1:0")
-		if err != nil {
-			t.Fatal(err)
-		}
-		base := l.Addr().(*net.TCPAddr).Port
-		_ = l.Close()
-		free := true
-		for i := 1; i < count; i++ {
-			probe, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", base+i))
-			if err != nil {
-				free = false
-				break
-			}
-			_ = probe.Close()
-		}
-		if free {
-			return base
-		}
-	}
-	t.Fatal("no run of free ports")
-	return 0
-}
-
 func TestRunJoinTimesEachStageAndRetriesStaleProviders(t *testing.T) {
 	// A provider record that names a peer which does not answer is the
 	// normal state of a mesh just after a rollout. The journey has to get

internal/controlplane/key_rotation_test.go

@@ -0,0 +1,101 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlplane
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestKeyRotationCheckInterval(t *testing.T) {
+	for _, tc := range []struct {
+		interval, want time.Duration
+	}{
+		{24 * time.Hour, time.Minute},
+		{time.Hour, time.Minute},
+		{5 * time.Minute, 30 * time.Second},
+		{2 * time.Second, 200 * time.Millisecond},
+	} {
+		if got := keyRotationCheckInterval(tc.interval); got != tc.want {
+			t.Errorf("keyRotationCheckInterval(%s) = %s, want %s", tc.interval, got, tc.want)
+		}
+	}
+}
+
+// The deadline is the store's, not the process's: a rotation that was due
+// while no replica was up happens on the first check after a start, and a
+// replica that restarts before the deadline does not push it back.
+func TestRotateKeysIfDueFollowsTheStoredDeadline(t *testing.T) {
+	store := newMetricsTestStore(t)
+	ctx := context.Background()
+	interval := 24 * time.Hour
+	issuer, _ := startCustomMockOIDC(t)
+	srv, err := NewServer(Options{
+		DriverName: "sqlite", DataSourceName: "unused", OIDCIssuer: issuer,
+		KeyRotationInterval: interval,
+	}, store)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = srv.Close() }()
+	if err := srv.Init(); err != nil {
+		t.Fatal(err)
+	}
+	keys := func() int {
+		t.Helper()
+		ks, err := store.GetAllValidKeys(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return len(ks)
+	}
+	if got := keys(); got != 1 {
+		t.Fatalf("after Init the keyring holds %d keys, want 1", got)
+	}
+
+	// Init scheduled the first rotation an interval from now; the loop it
+	// started checks every minute, these checks are direct.
+	next, err := store.NextKeyRotation(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if next.IsZero() || time.Until(next) < interval-time.Minute {
+		t.Fatalf("first rotation scheduled at %s, want about %s from now", next, interval)
+	}
+
+	srv.rotateKeysIfDue(time.Now())
+	if got := keys(); got != 1 {
+		t.Fatalf("a check before the deadline rotated: %d keys", got)
+	}
+
+	// Five days later, as after a week of restarts that never reached a tick.
+	late := next.Add(5 * 24 * time.Hour)
+	srv.rotateKeysIfDue(late)
+	if got := keys(); got != 2 {
+		t.Fatalf("a check past the deadline did not rotate: %d keys, want 2", got)
+	}
+	next, err = store.NextKeyRotation(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !next.Equal(late.Add(interval).Truncate(time.Millisecond)) {
+		t.Fatalf("deadline after the late rotation = %s, want %s", next, late.Add(interval))
+	}
+	srv.rotateKeysIfDue(late.Add(time.Minute))
+	if got := keys(); got != 2 {
+		t.Fatalf("a check right after a rotation rotated again: %d keys", got)
+	}
+}