chore: update ucind cluster to wait for initial store sync and cluster readiness

internal/ucind/cluster.go

@@ -119,17 +119,16 @@ func (p *Provisioner) CreateCluster(ctx context.Context, name string, opts Creat
 func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error {
 	// Init a new cluster on the first machine.
 	initMachine := machines[0]
-
-	if err := WaitMachineReady(ctx, initMachine, 30*time.Second); err != nil {
-		return fmt.Errorf("wait for machine %q to be ready: %w", initMachine.Name, err)
-	}
-
 	initClient, err := initMachine.Connect(ctx)
 	if err != nil {
 		return fmt.Errorf("create machine client over TCP '%s': %w", initMachine.APIAddress, err)
 	}
 	defer initClient.Close()
 
+	if err := initClient.WaitMachineReady(ctx, 30*time.Second); err != nil {
+		return fmt.Errorf("wait for machine %q to be ready: %w", initMachine.Name, err)
+	}
+
 	req := &pb.InitClusterRequest{
 		MachineName: initMachine.Name,
 		Network:     pb.NewIPPrefix(cluster.DefaultNetwork),
@@ -138,14 +137,22 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
 	if err != nil {
 		return fmt.Errorf("init cluster: %w", err)
 	}
+
 	fmt.Printf("Cluster %q initialised with machine %q\n", initMachine.ClusterName, initResp.Machine.Name)
+	fmt.Printf("Waiting for cluster to be ready...")
+	if err = initClient.WaitClusterReady(ctx, 30*time.Second); err != nil {
+		return fmt.Errorf("wait for cluster to be ready: %w", err)
+	}
+	fmt.Println(" done.")
+
+	// Get the current store DB version from the init machine to pass to the join requests.
+	inspectResp, err := initClient.MachineClient.InspectMachine(ctx, &emptypb.Empty{})
+	if err != nil {
+		return fmt.Errorf("inspect init machine: %w", err)
+	}
 
 	// Join the rest of the machines to the cluster.
 	for _, m := range machines[1:] {
-		if err = WaitMachineReady(ctx, m, 30*time.Second); err != nil {
-			return fmt.Errorf("wait for machine %q to be ready: %w", m.Name, err)
-		}
-
 		cli, err := m.Connect(ctx)
 		if err != nil {
 			return fmt.Errorf("create machine client over TCP '%s': %w", m.APIAddress, err)
@@ -153,6 +160,10 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
 		//goland:noinspection GoDeferInLoop
 		defer cli.Close()
 
+		if err := cli.WaitMachineReady(ctx, 30*time.Second); err != nil {
+			return fmt.Errorf("wait for machine %q to be ready: %w", m.Name, err)
+		}
+
 		tokenResp, err := cli.Token(ctx, &emptypb.Empty{})
 		if err != nil {
 			return fmt.Errorf("get machine token: %w", err)
@@ -183,6 +194,7 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
 		joinReq := &pb.JoinClusterRequest{
 			Machine:           addResp.Machine,
 			OtherMachines:     []*pb.MachineInfo{initResp.Machine},
+			MinStoreDbVersion: inspectResp.Machines[0].StoreDbVersion,
 		}
 		if _, err = cli.JoinCluster(ctx, joinReq); err != nil {
 			return fmt.Errorf("join cluster: %w", err)
@@ -256,16 +268,16 @@ func (p *Provisioner) InspectCluster(ctx context.Context, name string) (Cluster,
 // WaitClusterReady waits for all machines in the cluster to be ready and UP.
 func (p *Provisioner) WaitClusterReady(ctx context.Context, c Cluster, timeout time.Duration) error {
 	firstMachine := c.Machines[0]
-	if err := WaitMachineReady(ctx, firstMachine, timeout); err != nil {
-		return fmt.Errorf("wait for machine '%s' to be ready: %w", firstMachine.Name, err)
-	}
-
 	cli, err := firstMachine.Connect(ctx)
 	if err != nil {
 		return fmt.Errorf("connect to machine over TCP '%s': %w", firstMachine.APIAddress, err)
 	}
 	defer cli.Close()
 
+	if err = cli.WaitClusterReady(ctx, timeout); err != nil {
+		return fmt.Errorf("wait for cluster to be ready: %w", err)
+	}
+
 	boff := backoff.WithContext(backoff.NewExponentialBackOff(
 		backoff.WithInitialInterval(100*time.Millisecond),
 		backoff.WithMaxInterval(1*time.Second),
@@ -289,7 +301,25 @@ func (p *Provisioner) WaitClusterReady(ctx context.Context, c Cluster, timeout t
 		}
 		return nil
 	}
-	return backoff.Retry(checkMachinesUp, boff)
+	if err = backoff.Retry(checkMachinesUp, boff); err != nil {
+		return err
+	}
+
+	// Wait for each machine to sync the cluster store and be ready to serve cluster requests.
+	for _, m := range c.Machines[1:] {
+		mcli, err := m.Connect(ctx)
+		if err != nil {
+			return fmt.Errorf("connect to machine over TCP '%s': %w", m.APIAddress, err)
+		}
+		//goland:noinspection GoDeferInLoop
+		defer mcli.Close()
+
+		if err = mcli.WaitClusterReady(ctx, timeout); err != nil {
+			return fmt.Errorf("wait for cluster to be ready on machine '%s': %w", m.Name, err)
+		}
+	}
+
+	return nil
 }
 
 func (p *Provisioner) RemoveCluster(ctx context.Context, name string) error {

internal/ucind/machine.go

@@ -9,7 +9,6 @@ import (
 	"net/netip"
 	"time"
 
-	"github.com/cenkalti/backoff/v4"
 	"github.com/containerd/errdefs"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/image"
@@ -17,7 +16,6 @@ import (
 	"github.com/psviderski/uncloud/internal/secret"
 	"github.com/psviderski/uncloud/pkg/client"
 	"github.com/psviderski/uncloud/pkg/client/connector"
-	"google.golang.org/protobuf/types/known/emptypb"
 )
 
 const (
@@ -194,26 +192,3 @@ func randomMachineName() (string, error) {
 	}
 	return "machine-" + suffix, nil
 }
-
-// WaitMachineReady waits for the machine API to respond.
-func WaitMachineReady(ctx context.Context, m Machine, timeout time.Duration) error {
-	cli, err := m.Connect(ctx)
-	if err != nil {
-		return fmt.Errorf("connect to machine over TCP '%s': %w", m.APIAddress, err)
-	}
-	defer cli.Close()
-
-	boff := backoff.WithContext(backoff.NewExponentialBackOff(
-		backoff.WithInitialInterval(100*time.Millisecond),
-		backoff.WithMaxInterval(10*time.Second),
-		backoff.WithMaxElapsedTime(timeout),
-	), ctx)
-
-	inspect := func() error {
-		if _, err := cli.Inspect(ctx, &emptypb.Empty{}); err != nil {
-			return fmt.Errorf("inspect machine: %w", err)
-		}
-		return nil
-	}
-	return backoff.Retry(inspect, boff)
-}

pkg/client/machine.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	"github.com/psviderski/uncloud/pkg/api"
 	"google.golang.org/grpc/codes"
@@ -100,3 +102,46 @@ func (cli *Client) RenameMachine(ctx context.Context, nameOrID, newName string)
 
 	return cli.UpdateMachine(ctx, req)
 }
+
+// WaitMachineReady waits for the machine API on the connected machine to respond.
+func (cli *Client) WaitMachineReady(ctx context.Context, timeout time.Duration) error {
+	boff := backoff.WithContext(backoff.NewExponentialBackOff(
+		backoff.WithInitialInterval(100*time.Millisecond),
+		backoff.WithMaxInterval(1*time.Second),
+		backoff.WithMaxElapsedTime(timeout),
+	), ctx)
+
+	inspect := func() error {
+		if _, err := cli.Inspect(ctx, &emptypb.Empty{}); err != nil {
+			return fmt.Errorf("inspect machine: %w", err)
+		}
+		return nil
+	}
+	return backoff.Retry(inspect, boff)
+}
+
+// WaitClusterReady waits for the connected machine to be ready to server cluster requests.
+func (cli *Client) WaitClusterReady(ctx context.Context, timeout time.Duration) error {
+	// Backoff is not really needed here as the default service config for the gRPC client is already
+	// doing retries with backoff for Unavailable errors. However, it's still convenient to use backoff
+	// to control the overall timeout for the operation.
+	boff := backoff.WithContext(backoff.NewExponentialBackOff(
+		backoff.WithInitialInterval(100*time.Millisecond),
+		backoff.WithMaxInterval(1*time.Second),
+		backoff.WithMaxElapsedTime(timeout),
+	), ctx)
+
+	listMachines := func() error {
+		_, err := cli.ListMachines(ctx, nil)
+		if err != nil {
+			if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {
+				// Machine is not ready yet, retry.
+				return err
+			}
+			// Other non-Unavailable errors should not be retried.
+			return backoff.Permanent(err)
+		}
+		return nil
+	}
+	return backoff.Retry(listMachines, boff)
+}

test/e2e/cluster_test.go

@@ -55,7 +55,7 @@ func createTestCluster(
 	})
 
 	if waitReady {
-		require.NoError(t, p.WaitClusterReady(ctx, c, 15*time.Second))
+		require.NoError(t, p.WaitClusterReady(ctx, c, 30*time.Second))
 	}
 
 	return c, p
@@ -73,7 +73,6 @@ func TestClusterLifecycle(t *testing.T) {
 		// Create a client for each machine and wait for it to be ready.
 		clients := make([]*client.Client, len(c.Machines))
 		for i, m := range c.Machines {
-			require.NoError(t, ucind.WaitMachineReady(ctx, m, 5*time.Second))
 			clients[i], err = m.Connect(ctx)
 			require.NoError(t, err)
 			//goland:noinspection GoDeferInLoop
@@ -86,12 +85,11 @@ func TestClusterLifecycle(t *testing.T) {
 			require.Eventually(t, func() bool {
 				machines, err := cli.ListMachines(ctx, nil)
 				if err != nil {
-					// FailedPrecondition "cluster is not initialised" is expected until the store is reconciled.
+					// Unavailable "machine is not ready to serve cluster requests" is expected until
-					if s, ok := status.FromError(err); ok {
+					// the store is reconciled.
-						if s.Code() == codes.FailedPrecondition {
+					if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {
 						return false
 					}
-					}
 					require.NoError(t, err)
 				}