use timeout when waiting for cluster state changes

We had this fail in CI once, and failing took 30 minutes because
we didn't have a timeout on this. This shouldn't ever fail, but
the fact that it did indicates that the fabled etcd failures
we've seen a couple of times were still capable of happening.
This will make that failure happen sooner and more clearly.

Also, log the cluster states (and possibly node states) while
waiting. But add a delay -- otherwise we can do this quite a few
times per millisecond. We use Logf so that, if you didn't use -v,
you see these reported only if the test fails, but if the test fails,
we'll say what happened.

It would probably be better to have a passive thing that can wait
for updates, because we're waiting on heartbeats. Missing: A way to
detect what's actually happening in the failure cases, which we
see only quite rarely.

etcd/fake_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/featurebasedb/featurebase/v3/disco"
 	"github.com/featurebasedb/featurebase/v3/logger"
@@ -43,21 +44,42 @@ func (f *fakeCluster) BringUp() error {
 
 // AwaitClusterState verifies that node 0 thinks the cluster is in the
 // requested state, or tells you why it failed.
-func (f *fakeCluster) AwaitClusterState(expected disco.ClusterState) (err error) {
-	state := disco.ClusterState("")
+func (f *fakeCluster) AwaitClusterState(expected disco.ClusterState, timeout time.Duration) (err error) {
+	// You might reasonably ask why we don't use context.WithTimeout for this.
+	// The answer is that "context deadline exceeded" would be ambiguous as to
+	// whether it was an internal error produced by the etcd stuff, or an error
+	// introduced by this function.
+	now := time.Now()
+	state, err := f.nodes[0].ClusterState(context.Background())
+	if err != nil {
+		return err
+	}
 	for state != expected {
+		// wait a bit to see if it recovered
+		time.Sleep(50 * time.Millisecond)
 		state, err = f.nodes[0].ClusterState(context.Background())
 		if err != nil {
 			return err
 		}
+		if state != expected {
+			nodes := f.nodes[0].Nodes()
+			for _, n := range nodes {
+				f.tb.Logf("  %s: %s", n.ID, n.State)
+			}
+		}
+		elapsed := time.Since(now)
+		if elapsed >= timeout {
+			return fmt.Errorf("cluster did not reach state %q after %v", expected, timeout)
+		}
+		f.tb.Logf("cluster state after %v: %q", elapsed, state)
 	}
 	return nil
 }
 
 // MustAwaitClusterState verifies that node 0 thinks the cluster is in the
 // requested state, or fails a test.
-func (f *fakeCluster) MustAwaitClusterState(expected disco.ClusterState) {
-	err := f.AwaitClusterState(expected)
+func (f *fakeCluster) MustAwaitClusterState(expected disco.ClusterState, timeout time.Duration) {
+	err := f.AwaitClusterState(expected, timeout)
 	if err != nil {
 		f.tb.Fatalf("awaiting cluster state %s: %v", expected, err)
 	}

etcd/leasedkv_test.go

@@ -31,35 +31,26 @@ func TestClusterKv(t *testing.T) {
 	if err != nil {
 		t.Fatalf("starting cluster: %v", err)
 	}
-	c.MustAwaitClusterState(disco.ClusterStateDown)
+	c.MustAwaitClusterState(disco.ClusterStateDown, 10*time.Second)
 	err = c.BringUp()
 	if err != nil {
 		t.Fatalf("bringing up cluster: %v", err)
 	}
-	c.MustAwaitClusterState(disco.ClusterStateNormal)
+	c.MustAwaitClusterState(disco.ClusterStateNormal, 10*time.Second)
 	_, err = c.Elect()
 	if err != nil {
 		t.Fatalf("trying to cause election: %v", err)
 	}
 	ctx := context.TODO()
-	err = c.nodes[0].SetState(ctx, disco.NodeStateStarting)
-	assert.NoError(t, err)
-
-	err = c.nodes[1].SetState(ctx, disco.NodeStateStarting)
-	assert.NoError(t, err)
-
-	c.MustAwaitClusterState(disco.ClusterStateStarting)
-
-	err = c.nodes[0].SetState(ctx, disco.NodeStateStarted)
-	assert.NoError(t, err)
-
-	err = c.nodes[1].SetState(ctx, disco.NodeStateStarted)
-	assert.NoError(t, err)
-
+	c.nodes[0].SetState(ctx, disco.NodeStateStarting)
+	c.nodes[1].SetState(ctx, disco.NodeStateStarting)
+	c.MustAwaitClusterState(disco.ClusterStateStarting, 10*time.Second)
+	c.nodes[0].SetState(ctx, disco.NodeStateStarted)
+	c.nodes[1].SetState(ctx, disco.NodeStateStarted)
 	// Two of three nodes are up, one is down, we have 2 replicas, so
 	// we should be able to handle reads but not writes, so we're in
 	// a Degraded state.
-	c.MustAwaitClusterState(disco.ClusterStateDegraded)
+	c.MustAwaitClusterState(disco.ClusterStateDegraded, 10*time.Second)
 	err = c.Stop()
 	if err != nil {
 		t.Fatalf("stopping cluster: %v", err)