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..c1fad197 100644 --- a/gorumstest/servers.go +++ b/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 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/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) { 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()) }