test(etcdtest): share the embedded etcd and bind it to port 0 (#673, #509)

ARCHITECTURE.md

@@ -275,7 +275,7 @@ wal/                  segmented CRC-framed WAL
 engine/               metrics vertical
 recordengine/         shared record engine (logs/traces/profiles/exemplars)
 query/{fetch,scale,profile,promql}           read seam · scale-out decorators · EXPLAIN ANALYZE · Prom adapter
-cluster/{,ring,etcd,replica,rebalance,partsync,ec,router}   L0 distribution
+cluster/{,ring,etcd,etcd/etcdtest,replica,rebalance,partsync,ec,router}   L0 distribution · in-process etcd for tests
 internal/{obs,retry,simd,parallel,partid,diskguard,vfs,memlimit,memsize,watermark,mergestream,reproduce,heaptest,cmd/gensimd}  injected observability · reliability · AVX2 kernels · fan-out · part ids · disk-pressure guard · heap accounting · per-series watermark sidecar codec · the merge seam both engines share (k-way key union, seal units, conformance suite) · gated defect reproducers · heap and allocation measurement for tests · filesystem seam + crash model (validated against ext4 on dm-flakey by `internal/vfs/crashmodel`, `-tags crashmodel`)
 reliability/          public RetryConfig presets
 ```

admin_test.go

@@ -7,6 +7,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/oteldb/storage/cluster/etcd/etcdtest"
 	"github.com/oteldb/storage/signal"
 )
 
@@ -134,7 +135,7 @@ func TestAdminRebalanceSingleNodeNoOp(t *testing.T) {
 //
 //nolint:paralleltest // owns an embedded etcd; runs serially
 func TestAdminRebalanceClusteredReportsSuccess(t *testing.T) {
-	endpoint := startEtcd(t)
+	endpoint := etcdtest.Start(t)
 	ctx := context.Background()
 
 	s := openClusterNodeShared(t, endpoint, "node-a")

cluster/ARCH.md

@@ -72,6 +72,12 @@ therefore drops such a key under a value+lease guard and recreates it under the
 also makes the restart a real new tenure: the recreated claim takes a higher term, so the
 incarnation's writes outrank the dead one's.
 
+Tests boot etcd through **`cluster/etcd/etcdtest`**, which binds its listeners to port 0 and reads
+the client port back from the socket. Probing a free port and releasing it for etcd to bind is
+racy: anything, including etcd clients' own ephemeral ports, can take it in between. Only
+`Server.Restart` rebinds a known port, since clients must find etcd where it was. `etcdtest` must
+not import `cluster/etcd`, whose internal tests use it.
+
 ### Lease fencing — the boundary on acting as primary
 
 A node that stops renewing its lease keeps a ring frozen at its last etcd view, so it goes on

cluster/etcd/etcdtest/etcdtest.go

@@ -0,0 +1,115 @@
+// Package etcdtest runs an in-process single-node etcd for tests.
+package etcdtest
+
+import (
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/server/v3/embed"
+)
+
+const readyTimeout = 30 * time.Second
+
+// Start boots an etcd that stops with tb and returns its client endpoint URL.
+func Start(tb testing.TB) string {
+	tb.Helper()
+
+	return NewServer(tb).Endpoint()
+}
+
+// Server is an embedded etcd that can be stopped and restarted on the same client URL and data
+// directory, which is what an etcd rollout looks like to a node that keeps running across it.
+type Server struct {
+	tb     testing.TB
+	dir    string
+	client url.URL
+	e      *embed.Etcd
+}
+
+// NewServer boots an etcd that stops with tb, on a client port the kernel picks.
+func NewServer(tb testing.TB) *Server {
+	tb.Helper()
+
+	s := &Server{
+		tb:     tb,
+		dir:    tb.TempDir(),
+		client: url.URL{Scheme: "http", Host: "127.0.0.1:0"},
+	}
+	s.start()
+	tb.Cleanup(s.Stop)
+
+	return s
+}
+
+// Endpoint returns the client URL, stable across [Server.Restart].
+func (s *Server) Endpoint() string {
+	return s.client.String()
+}
+
+// Stop shuts etcd down. It is a no-op on a stopped server.
+func (s *Server) Stop() {
+	if s.e == nil {
+		return
+	}
+
+	s.e.Close()
+	s.e = nil
+}
+
+// Restart starts a stopped server again on its client URL and data directory. Rebinding a fixed
+// port can collide with another socket; nothing reserves it while the server is down.
+func (s *Server) Restart() {
+	s.tb.Helper()
+
+	require.Nil(s.tb, s.e, "etcd is still running")
+	s.start()
+}
+
+// Dial returns a client for this server, closed with the test.
+func (s *Server) Dial() *clientv3.Client {
+	s.tb.Helper()
+
+	c, err := clientv3.New(clientv3.Config{
+		Endpoints:   []string{s.Endpoint()},
+		DialTimeout: 5 * time.Second,
+	})
+	require.NoError(s.tb, err)
+	s.tb.Cleanup(func() { _ = c.Close() })
+
+	return c
+}
+
+func (s *Server) start() {
+	s.tb.Helper()
+
+	// The peer URL is never dialed by a single-node cluster, so it is advertised as the unbound
+	// port-0 address rather than the one the listener gets.
+	peer := url.URL{Scheme: "http", Host: "127.0.0.1:0"}
+
+	cfg := embed.NewConfig()
+	cfg.Dir = s.dir
+	cfg.LogLevel = "error"
+	cfg.ListenClientUrls = []url.URL{s.client}
+	cfg.AdvertiseClientUrls = []url.URL{s.client}
+	cfg.ListenPeerUrls = []url.URL{peer}
+	cfg.AdvertisePeerUrls = []url.URL{peer}
+	cfg.InitialCluster = cfg.Name + "=" + peer.String()
+	// The gateway dials the configured listen address, which is port 0 on the first start.
+	cfg.EnableGRPCGateway = false
+
+	e, err := embed.StartEtcd(cfg)
+	require.NoError(s.tb, err)
+
+	select {
+	case <-e.Server.ReadyNotify():
+	case <-time.After(readyTimeout):
+		e.Close()
+		s.tb.Fatal("embedded etcd did not become ready")
+	}
+
+	s.client.Host = e.Clients[0].Addr().String()
+	s.e = e
+}

cluster/etcd/etcdtest/etcdtest_test.go

@@ -0,0 +1,58 @@
+package etcdtest_test
+
+import (
+	"context"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/oteldb/storage/cluster/etcd/etcdtest"
+)
+
+func TestStart(t *testing.T) {
+	t.Parallel()
+
+	endpoint := etcdtest.Start(t)
+	u, err := url.Parse(endpoint)
+	require.NoError(t, err)
+	assert.NotEqual(t, "0", u.Port())
+
+	c, err := clientv3.New(clientv3.Config{Endpoints: []string{endpoint}, DialTimeout: 5 * time.Second})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = c.Close() })
+
+	_, err = c.Put(t.Context(), "k", "v")
+	require.NoError(t, err)
+}
+
+func TestServerRestartKeepsEndpointAndData(t *testing.T) {
+	t.Parallel()
+
+	s := etcdtest.NewServer(t)
+	endpoint := s.Endpoint()
+	c := s.Dial()
+
+	_, err := c.Put(t.Context(), "k", "v")
+	require.NoError(t, err)
+
+	s.Stop()
+	s.Stop()
+	s.Restart()
+	assert.Equal(t, endpoint, s.Endpoint())
+
+	require.EventuallyWithT(t, func(ct *assert.CollectT) {
+		ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+		defer cancel()
+
+		resp, err := c.Get(ctx, "k")
+		if !assert.NoError(ct, err) || !assert.Len(ct, resp.Kvs, 1) {
+			return
+		}
+
+		assert.Equal(ct, "v", string(resp.Kvs[0].GetValue()))
+	}, 30*time.Second, 10*time.Millisecond, "the client from before the restart reconnects to the same data")
+}

cluster/etcd/membership_test.go

@@ -2,18 +2,14 @@ package etcd
 
 import (
 	"context"
-	"net"
-	"net/url"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.etcd.io/etcd/server/v3/embed"
-)
 
-const httpScheme = "http"
+	"github.com/oteldb/storage/cluster/etcd/etcdtest"
+)
 
 func TestMemberEncodeDecodeRoundTrip(t *testing.T) {
 	t.Parallel()
@@ -42,52 +38,6 @@ func TestDecodeMemberRejectsGarbage(t *testing.T) {
 	require.Error(t, err)
 }
 
-// freeAddr returns a free localhost host:port (the listener is closed before returning, so
-// there is a small TOCTOU window — acceptable for a test harness).
-func freeAddr(t *testing.T) string {
-	t.Helper()
-	var lc net.ListenConfig
-	l, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0")
-	require.NoError(t, err)
-	addr := l.Addr().String()
-	require.NoError(t, l.Close())
-
-	return addr
-}
-
-// startEtcd boots an in-process single-node etcd and returns a client for it.
-func startEtcd(t *testing.T) *clientv3.Client {
-	t.Helper()
-
-	lc := url.URL{Scheme: httpScheme, Host: freeAddr(t)}
-	lp := url.URL{Scheme: httpScheme, Host: freeAddr(t)}
-
-	cfg := embed.NewConfig()
-	cfg.Dir = t.TempDir()
-	cfg.LogLevel = "error"
-	cfg.ListenClientUrls = []url.URL{lc}
-	cfg.AdvertiseClientUrls = []url.URL{lc}
-	cfg.ListenPeerUrls = []url.URL{lp}
-	cfg.AdvertisePeerUrls = []url.URL{lp}
-	cfg.InitialCluster = cfg.Name + "=" + lp.String()
-
-	e, err := embed.StartEtcd(cfg)
-	require.NoError(t, err)
-	t.Cleanup(e.Close)
-
-	select {
-	case <-e.Server.ReadyNotify():
-	case <-time.After(30 * time.Second):
-		t.Fatal("embedded etcd did not become ready")
-	}
-
-	client, err := clientv3.New(clientv3.Config{Endpoints: []string{lc.String()}, DialTimeout: 5 * time.Second})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = client.Close() })
-
-	return client
-}
-
 func memberIDs(ms []Member) []string {
 	out := make([]string, len(ms))
 	for i, m := range ms {
@@ -99,7 +49,7 @@ func memberIDs(ms []Member) []string {
 
 //nolint:paralleltest // owns an embedded etcd; runs serially
 func TestMembershipJoinWatchLeave(t *testing.T) {
-	client := startEtcd(t)
+	client := etcdtest.NewServer(t).Dial()
 	ctx := context.Background()
 
 	// Node A joins.

cluster/etcd/observe_test.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/oteldb/storage/cluster/etcd/etcdtest"
 )
 
 // eventually polls cond until it holds, so a watch-driven assertion does not depend on timing.
@@ -19,7 +21,7 @@ func TestWatchSeesMembersWithoutJoining(t *testing.T) {
 	t.Parallel()
 
 	ctx := t.Context()
-	client := startEtcd(t)
+	client := etcdtest.NewServer(t).Dial()
 
 	joined, err := Join(ctx, client, "/test", Member{ID: "node-a", Zone: "z1", Addr: "10.0.0.1:9000"}, 0)
 	require.NoError(t, err)
@@ -53,7 +55,7 @@ func TestWatchCloseRevokesNothing(t *testing.T) {
 	t.Parallel()
 
 	ctx := t.Context()
-	client := startEtcd(t)
+	client := etcdtest.NewServer(t).Dial()
 
 	joined, err := Join(ctx, client, "/test", Member{ID: "node-a", Addr: "10.0.0.1:9000"}, 0)
 	require.NoError(t, err)
@@ -73,7 +75,7 @@ func TestWatchOnEmptyClusterIsUsable(t *testing.T) {
 	t.Parallel()
 
 	ctx := t.Context()
-	client := startEtcd(t)
+	client := etcdtest.NewServer(t).Dial()
 
 	obs, err := Watch(ctx, client, "/test")
 	require.NoError(t, err)

cluster/etcd/ownership_chaos_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
 
+	"github.com/oteldb/storage/cluster/etcd/etcdtest"
 	"github.com/oteldb/storage/cluster/ring"
 )
 
@@ -97,7 +98,7 @@ func assertOwnedByPrimary(t *testing.T, client *clientv3.Client, r *ring.Ring, s
 //
 //nolint:paralleltest // owns an embedded etcd; runs serially
 func TestOwnershipConcurrentClaimSingleWinner(t *testing.T) {
-	client := startEtcd(t)
+	client := etcdtest.NewServer(t).Dial()
 	ctx := context.Background()
 
 	const contenders = 12
@@ -139,7 +140,7 @@ func TestOwnershipConcurrentClaimSingleWinner(t *testing.T) {
 //
 //nolint:paralleltest // owns an embedded etcd; runs serially
 func TestOwnershipReconcileConvergesUnderChurn(t *testing.T) {
-	client := startEtcd(t)
+	client := etcdtest.NewServer(t).Dial()
 
 	ids := []string{"a", "b", "c", "d"}
 	owners := make([]*Ownership, len(ids))
@@ -169,7 +170,7 @@ func TestOwnershipReconcileConvergesUnderChurn(t *testing.T) {
 //
 //nolint:paralleltest // owns an embedded etcd; runs serially
 func TestOwnershipMassFailoverOnLeaseExpiry(t *testing.T) {
-	client := startEtcd(t)
+	client := etcdtest.NewServer(t).Dial()
 	ctx := context.Background()
 
 	dying, err := client.Grant(ctx, 30)