gorums: add per-server options for local server groups

gorumstest/options.go

@@ -19,6 +19,7 @@ import (
 //		gorums.WithNodeList(...),       // NodeSource
 //	)
 //
+// [LocalServers] also accepts a [gorums.LocalServerOption].
 // [Config] and [Node] panic on an Option value of any other type.
 type Option any
 

gorumstest/servers.go

@@ -59,22 +59,27 @@ func Servers(t testing.TB, numServers int, srvFn func(i int) ServerIface) []stri
 // build and over localhost TCP under the integration build tag. Each
 // server auto-creates a peer [gorums.Config] over the group, accessible
 // via [gorums.Server.PeerConfig]. The servers are automatically stopped
-// when the test finishes via t.Cleanup. Any [gorums.ServerOption]s (for
-// example [gorums.WithStreamDedup]) are applied to every server.
-func LocalServers(t testing.TB, n int, opts ...gorums.ServerOption) []*gorums.Server {
+// when the test finishes via t.Cleanup.
+//
+// Each Option is one of:
+//   - a [gorums.ServerOption], applied to every server (for example
+//     [gorums.WithStreamDedup]);
+//   - a [gorums.DialOption], applied to every server's outbound
+//     configuration after [DialOptions];
+//   - a [gorums.LocalServerOption], such as
+//     [gorums.WithLocalServerOptionsFunc] for options that differ per server.
+//
+// LocalServers panics on an Option of any other type.
+func LocalServers(t testing.TB, n int, opts ...Option) []*gorums.Server {
 	t.Helper()
-
 	// Skip goleak check for benchmarks
 	if _, ok := t.(*testing.B); !ok {
 		// Register goleak check FIRST so it runs LAST (after all other cleanup)
 		t.Cleanup(func() { goleak.VerifyNone(t) })
 	}
+	localOpts := localServerOptions(t, opts)
 
-	srvs, stop, err := gorums.NewLocalServers(n,
-		gorums.WithLocalServerOptions(opts...),
-		gorums.WithLocalDialOptions(DialOptions(t)),
-		gorums.WithLocalListeners(servers.ListenFunc(t)),
-	)
+	srvs, stop, err := gorums.NewLocalServers(n, localOpts...)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -89,6 +94,30 @@ func LocalServers(t testing.TB, n int, opts ...gorums.ServerOption) []*gorums.Se
 	return srvs
 }
 
+// localServerOptions returns the [gorums.LocalServerOption]s that
+// [LocalServers] passes to [gorums.NewLocalServers]: the test's dial options
+// and listeners, followed by opts in order. It panics on an Option of any
+// other type.
+func localServerOptions(t testing.TB, opts []Option) []gorums.LocalServerOption {
+	localOpts := []gorums.LocalServerOption{
+		gorums.WithLocalDialOptions(DialOptions(t)),
+		gorums.WithLocalListeners(servers.ListenFunc(t)),
+	}
+	for _, opt := range opts {
+		switch o := opt.(type) {
+		case gorums.ServerOption:
+			localOpts = append(localOpts, gorums.WithLocalServerOptions(o))
+		case gorums.DialOption:
+			localOpts = append(localOpts, gorums.WithLocalDialOptions(o))
+		case gorums.LocalServerOption:
+			localOpts = append(localOpts, o)
+		default:
+			panic(fmt.Sprintf("gorumstest: unsupported LocalServers Option type %T", opt))
+		}
+	}
+	return localOpts
+}
+
 // WaitForPeers blocks until every server in srvs is connected to all the
 // peers in its [gorums.Server.PeerConfig], or fails the test after 10
 // seconds. Use it after [LocalServers] when a test needs every peer

gorumstest/servers_test.go

@@ -1,9 +1,16 @@
 package gorumstest_test
 
 import (
+	"bytes"
+	"fmt"
+	"log"
+	"slices"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/relab/gorums"
 	"github.com/relab/gorums/gorumstest"
 	"github.com/relab/gorums/internal/testutils/mock"
 	gorumsimpl "github.com/relab/gorums/runtime/gorumsimpl"
@@ -30,3 +37,51 @@ func TestServersEchoServer(t *testing.T) {
 		t.Errorf("RemoteCall = %q, want %q", got, want)
 	}
 }
+
+// TestLocalServersOptions verifies that LocalServers applies a ServerOption to
+// every server, a DialOption to every server's outbound configuration, and a
+// LocalServerOption as given.
+func TestLocalServersOptions(t *testing.T) {
+	const n = 2
+	var mu sync.Mutex
+	peerChanges := 0
+	var perServer []gorums.ID
+	var dialLog bytes.Buffer
+	gorumstest.LocalServers(t, n,
+		gorums.WithPeerChange(func(gorums.Config) {
+			mu.Lock()
+			defer mu.Unlock()
+			peerChanges++
+		}),
+		gorums.WithLogger(log.New(&dialLog, "", 0)),
+		gorums.WithLocalServerOptionsFunc(func(id gorums.ID) []gorums.ServerOption {
+			mu.Lock()
+			defer mu.Unlock()
+			perServer = append(perServer, id)
+			return nil
+		}),
+	)
+
+	mu.Lock()
+	defer mu.Unlock()
+	if peerChanges < n {
+		t.Errorf("WithPeerChange callback ran %d times; want at least %d, one per server", peerChanges, n)
+	}
+	if got := strings.Count(dialLog.String(), "ready"); got != n {
+		t.Errorf("dial logger recorded %d \"ready\" lines; want %d, one per server", got, n)
+	}
+	if want := []gorums.ID{1, 2}; !slices.Equal(perServer, want) {
+		t.Errorf("WithLocalServerOptionsFunc called for %v; want %v", perServer, want)
+	}
+}
+
+// TestLocalServersUnsupportedOption verifies that LocalServers panics on an
+// Option of a type it cannot apply.
+func TestLocalServersUnsupportedOption(t *testing.T) {
+	defer func() {
+		if got := fmt.Sprint(recover()); !strings.Contains(got, "unsupported LocalServers Option type int") {
+			t.Errorf("LocalServers panic = %q; want it to name the unsupported type int", got)
+		}
+	}()
+	gorumstest.LocalServers(t, 1, 42)
+}

internal/tests/metadata/metadata_test.go

@@ -96,7 +96,7 @@ func TestPerMessageMetadataAcrossStreamTopologies(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			var opts []gorums.ServerOption
+			var opts []gorumstest.Option
 			if tt.dedup {
 				opts = append(opts, gorums.WithStreamDedup())
 			}

local_servers.go

@@ -3,14 +3,17 @@ package gorums
 import (
 	"fmt"
 	"net"
+	"slices"
 )
 
 // localServerOptions accumulates the options [NewLocalServers] applies to
 // every server it creates.
 type localServerOptions struct {
-	serverOpts []ServerOption
-	dialOpts   []DialOption
-	listen     func() (net.Listener, error)
+	serverOpts    []ServerOption
+	dialOpts      []DialOption
+	serverOptsFns []func(id ID) []ServerOption
+	dialOptsFns   []func(id ID) []DialOption
+	listen        func() (net.Listener, error)
 }
 
 // LocalServerOption configures [NewLocalServers]. Use [WithLocalServerOptions]
@@ -32,6 +35,26 @@ func WithLocalDialOptions(opts ...DialOption) LocalServerOption {
 	}
 }
 
+// WithLocalServerOptionsFunc applies the options that fn returns for a
+// server's node ID to that server, after the options from
+// [WithLocalServerOptions]. Use it for options that differ per server, such
+// as the TLS certificate each server presents.
+func WithLocalServerOptionsFunc(fn func(id ID) []ServerOption) LocalServerOption {
+	return func(o *localServerOptions) {
+		o.serverOptsFns = append(o.serverOptsFns, fn)
+	}
+}
+
+// WithLocalDialOptionsFunc applies the options that fn returns for a server's
+// node ID to that server's outbound configuration, after the options from
+// [WithLocalDialOptions]. Use it for options that differ per server, such as
+// the client certificate each server presents to its peers.
+func WithLocalDialOptionsFunc(fn func(id ID) []DialOption) LocalServerOption {
+	return func(o *localServerOptions) {
+		o.dialOptsFns = append(o.dialOptsFns, fn)
+	}
+}
+
 // WithLocalListeners makes [NewLocalServers] create each server's listener with
 // listen instead of a TCP listener on a random localhost port. Each listener's
 // address must be a unique host:port that the servers' dial options reach.
@@ -46,7 +69,8 @@ func WithLocalListeners(listen func() (net.Listener, error)) LocalServerOption {
 // Each server is assigned a node ID from 1 to n. Every server tracks and calls
 // all the other servers. Use [WithLocalServerOptions] to add [ServerOption]s
 // to every server, and [WithLocalDialOptions] to add [DialOption]s to each
-// server's outbound connections.
+// server's outbound connections. Use [WithLocalServerOptionsFunc] and
+// [WithLocalDialOptionsFunc] for options that differ per server.
 //
 // The returned servers are not started; call [Server.ListenAndServe] after
 // registering any services. The returned stop function stops all servers and
@@ -71,10 +95,17 @@ func NewLocalServers(n int, opts ...LocalServerOption) ([]*Server, func(), error
 	servers := make([]*Server, n)
 	for i := range n {
 		myID := ID(i + 1)
+		dialOpts := slices.Clone(localOpts.dialOpts)
+		for _, fn := range localOpts.dialOptsFns {
+			dialOpts = append(dialOpts, fn(myID)...)
+		}
 		serverOpts := append(
-			[]ServerOption{WithPeers(myID, nodeSource, localOpts.dialOpts...)},
+			[]ServerOption{WithPeers(myID, nodeSource, dialOpts...)},
 			localOpts.serverOpts...,
 		)
+		for _, fn := range localOpts.serverOptsFns {
+			serverOpts = append(serverOpts, fn(myID)...)
+		}
 		srv, err := newServer(serverOpts...)
 		if err != nil {
 			closeServers(servers[:i], listeners)

local_servers_test.go

@@ -3,9 +3,11 @@ package gorums_test
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"log"
 	"net"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -180,6 +182,60 @@ func TestLocalServersAppliesDialOptions(t *testing.T) {
 	}
 }
 
+// TestLocalServersAppliesPerServerOptions verifies that
+// WithLocalServerOptionsFunc and WithLocalDialOptionsFunc give each server the
+// options for its own node ID, after the shared options.
+func TestLocalServersAppliesPerServerOptions(t *testing.T) {
+	const n = 3
+	var mu sync.Mutex
+	sharedChanges := 0
+	sawOwnNode := make(map[gorums.ID]bool)
+	var sharedLog, perServerLog bytes.Buffer
+	_, stop, err := gorums.NewLocalServers(n,
+		gorums.WithLocalServerOptions(gorums.WithPeerChange(func(gorums.Config) {
+			mu.Lock()
+			defer mu.Unlock()
+			sharedChanges++
+		})),
+		gorums.WithLocalServerOptionsFunc(func(id gorums.ID) []gorums.ServerOption {
+			return []gorums.ServerOption{gorums.WithPeerChange(func(cfg gorums.Config) {
+				mu.Lock()
+				defer mu.Unlock()
+				if cfg.Node(id) != nil {
+					sawOwnNode[id] = true
+				}
+			})}
+		}),
+		gorums.WithLocalDialOptions(gorumstest.InsecureDialOptions(t), gorums.WithLogger(log.New(&sharedLog, "", 0))),
+		gorums.WithLocalDialOptionsFunc(func(id gorums.ID) []gorums.DialOption {
+			return []gorums.DialOption{gorums.WithLogger(log.New(&perServerLog, fmt.Sprintf("server %d: ", id), 0))}
+		}),
+	)
+	if err != nil {
+		t.Fatalf("NewLocalServers: %v", err)
+	}
+	t.Cleanup(stop)
+
+	mu.Lock()
+	defer mu.Unlock()
+	// The per-server options come after the shared ones, so their
+	// WithPeerChange callback and WithLogger replace the shared ones.
+	if sharedChanges != 0 {
+		t.Errorf("shared WithPeerChange callback ran %d times; want 0", sharedChanges)
+	}
+	if sharedLog.Len() != 0 {
+		t.Errorf("shared logger recorded %q; want nothing", sharedLog.String())
+	}
+	for id := gorums.ID(1); id <= n; id++ {
+		if !sawOwnNode[id] {
+			t.Errorf("server %d: per-server WithPeerChange callback did not see its own node", id)
+		}
+		if want := fmt.Sprintf("server %d: ready", id); !strings.Contains(perServerLog.String(), want) {
+			t.Errorf("per-server log = %q; want a line %q", perServerLog.String(), want)
+		}
+	}
+}
+
 // TestLocalServersStopIsIdempotent verifies that the stop function
 // returned by NewLocalServers can be called more than once without panicking.
 func TestLocalServersStopIsIdempotent(t *testing.T) {

server_test.go

@@ -1229,7 +1229,7 @@ func TestServerSenderID(t *testing.T) {
 	// peerSenderIDs multicasts from every server to its peer configuration,
 	// which includes the server itself, and checks that every receiver
 	// observes the multicasting server's ID.
-	peerSenderIDs := func(t *testing.T, opts ...gorums.ServerOption) {
+	peerSenderIDs := func(t *testing.T, opts ...gorumstest.Option) {
 		servers := gorumstest.LocalServers(t, 3, opts...)
 		records := make(chan senderRecord, len(servers))
 		for _, srv := range servers {

stream_dedup_test.go

@@ -84,7 +84,7 @@ func waitForDedupConcurrent(t *testing.T, servers []*gorums.Server) {
 // is true.
 func newServerPair(t *testing.T, dedup bool) []*gorums.Server {
 	t.Helper()
-	var opts []gorums.ServerOption
+	var opts []gorumstest.Option
 	if dedup {
 		opts = append(opts, gorums.WithStreamDedup())
 	}

tree_integration_test.go

@@ -34,7 +34,7 @@ import (
 // configuration, which is in node-ID order.
 func setupTreeServers(t *testing.T, n int, dedup bool, opts gorums.TreeOptions) ([]*gorums.Server, []*gorums.TreeConfig) {
 	t.Helper()
-	var srvOpts []gorums.ServerOption
+	var srvOpts []gorumstest.Option
 	if dedup {
 		srvOpts = append(srvOpts, gorums.WithStreamDedup())
 	}