From 4ea2da9c8fb672c6e61635e09b05921eeb13fe3d Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 16:41:36 +0200 Subject: [PATCH 1/3] gorums: add per-server options to NewLocalServers NewLocalServers applied the same server and dial options to every server, so a local cluster could not give each server its own TLS certificate. Add WithLocalServerOptionsFunc and WithLocalDialOptionsFunc, which return the options for a server's node ID. They are applied after the shared options, so per-server transport credentials replace shared ones. --- local_servers.go | 41 +++++++++++++++++++++++++++---- local_servers_test.go | 56 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 5 deletions(-) diff --git a/local_servers.go b/local_servers.go index d5e5fe46..661909d9 100644 --- a/local_servers.go +++ b/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) diff --git a/local_servers_test.go b/local_servers_test.go index 5d58e768..1ad258d4 100644 --- a/local_servers_test.go +++ b/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) { From baf455768b4e043e42310da74b3fe091173a7e5b Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 16:44:16 +0200 Subject: [PATCH 2/3] gorumstest: accept local server and dial options in LocalServers LocalServers took only server options, so a test could not give its servers per-server options or extra dial options. It now takes the package's Option type: a ServerOption applies to every server, a DialOption to every server's outbound configuration after DialOptions, and a LocalServerOption, such as WithLocalServerOptionsFunc, as given. Other types panic, as for Config and Node. --- gorumstest/options.go | 1 + gorumstest/servers.go | 46 ++++++++++++++++---- gorumstest/servers_test.go | 55 ++++++++++++++++++++++++ internal/tests/metadata/metadata_test.go | 2 +- server_test.go | 2 +- stream_dedup_test.go | 2 +- tree_integration_test.go | 2 +- 7 files changed, 98 insertions(+), 12 deletions(-) diff --git a/gorumstest/options.go b/gorumstest/options.go index e656c43d..c00f7383 100644 --- a/gorumstest/options.go +++ b/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 diff --git a/gorumstest/servers.go b/gorumstest/servers.go index d44a8e7b..cfbe04c1 100644 --- a/gorumstest/servers.go +++ b/gorumstest/servers.go @@ -59,10 +59,20 @@ 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() + localOpts := localServerOptions(t, opts) // Skip goleak check for benchmarks if _, ok := t.(*testing.B); !ok { @@ -70,11 +80,7 @@ func LocalServers(t testing.TB, n int, opts ...gorums.ServerOption) []*gorums.Se t.Cleanup(func() { goleak.VerifyNone(t) }) } - 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 +95,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 diff --git a/gorumstest/servers_test.go b/gorumstest/servers_test.go index d4a19fbb..a9238e8d 100644 --- a/gorumstest/servers_test.go +++ b/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) +} diff --git a/internal/tests/metadata/metadata_test.go b/internal/tests/metadata/metadata_test.go index 89bfc4a0..2ba87509 100644 --- a/internal/tests/metadata/metadata_test.go +++ b/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()) } diff --git a/server_test.go b/server_test.go index 7b433be2..cbeafdbe 100644 --- a/server_test.go +++ b/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 { diff --git a/stream_dedup_test.go b/stream_dedup_test.go index 1e72fd4d..2f2c898c 100644 --- a/stream_dedup_test.go +++ b/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()) } diff --git a/tree_integration_test.go b/tree_integration_test.go index f7edfab4..308f835c 100644 --- a/tree_integration_test.go +++ b/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()) } From 88fcd3e52c18ff720aebcd6b21f9b962f807473b Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 17:49:34 +0200 Subject: [PATCH 3/3] gorumstest: register the leak check before the listeners' cleanup LocalServers built its options, which register the listeners' cleanup through servers.ListenFunc, before it registered goleak.VerifyNone. Cleanups run in reverse order, so the leak check ran before the listeners were cleaned up, contrary to the comment that the check runs last. Register the leak check first. --- gorumstest/servers.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/gorumstest/servers.go b/gorumstest/servers.go index cfbe04c1..c1fad197 100644 --- a/gorumstest/servers.go +++ b/gorumstest/servers.go @@ -72,13 +72,12 @@ func Servers(t testing.TB, numServers int, srvFn func(i int) ServerIface) []stri // LocalServers panics on an Option of any other type. func LocalServers(t testing.TB, n int, opts ...Option) []*gorums.Server { t.Helper() - localOpts := localServerOptions(t, opts) - // 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, localOpts...) if err != nil {