fix: race on cluster init by waiting for corrosion service to become ready with schema applied

internal/machine/cluster.go

@@ -108,11 +108,13 @@ func (cc *clusterController) Run(ctx context.Context) error {
 		if err := cc.corroService.Restart(ctx); err != nil {
 			return fmt.Errorf("restart corrosion service: %w", err)
 		}
+		slog.Info("Corrosion service restarted.")
 	} else {
 		slog.Info("Starting corrosion service.")
 		if err := cc.corroService.Start(ctx); err != nil {
 			return fmt.Errorf("start corrosion service: %w", err)
 		}
+		slog.Info("Corrosion service started.")
 	}
 
 	errGroup, ctx := errgroup.WithContext(ctx)

internal/machine/corroservice/docker.go

@@ -27,29 +27,33 @@ type DockerService struct {
 	User    string
 }
 
-func NewDockerService(cli *client.Client, image, name, dataDir string) *DockerService {
-	return &DockerService{
-		Client:  cli,
-		Image:   image,
-		Name:    name,
-		DataDir: dataDir,
-	}
-}
-
 func (s *DockerService) Start(ctx context.Context) error {
 	_, err := s.Client.ContainerInspect(ctx, s.Name)
 	if err != nil {
-		if client.IsErrNotFound(err) {
-			return s.startNewContainer(ctx)
+		if !client.IsErrNotFound(err) {
+			return fmt.Errorf("inspect container %q: %w", s.Name, err)
+		}
+		if err = s.startNewContainer(ctx); err != nil {
+			return err
+		}
+	} else {
+		// Container already exists.
+		// TODO: recreate only if the container configuration has to be changed.
+		if err = s.Client.ContainerRemove(ctx, s.Name, container.RemoveOptions{Force: true}); err != nil {
+			return fmt.Errorf("remove container %q: %w", s.Name, err)
+		}
+		if err = s.startNewContainer(ctx); err != nil {
+			return err
 		}
-		return fmt.Errorf("inspect container %q: %w", s.Name, err)
-	}
-	// TODO: recreate only if the container configuration has to be changed.
-	if err = s.Client.ContainerRemove(ctx, s.Name, container.RemoveOptions{Force: true}); err != nil {
-		return fmt.Errorf("remove container %q: %w", s.Name, err)
 	}
 
-	return s.startNewContainer(ctx)
+	slog.Debug("Waiting for corrosion service to be ready.")
+	if err = WaitReady(ctx, s.DataDir); err != nil {
+		return err
+	}
+	slog.Debug("Corrosion service is ready.")
+
+	return nil
 }
 
 func (s *DockerService) Stop(ctx context.Context) error {

internal/machine/corroservice/service.go

@@ -1,6 +1,16 @@
 package corroservice
 
-import "context"
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/BurntSushi/toml"
+	"github.com/cenkalti/backoff/v4"
+	"github.com/psviderski/uncloud/internal/corrosion"
+)
 
 type Service interface {
 	Start(ctx context.Context) error
@@ -8,3 +18,45 @@ type Service interface {
 	Restart(ctx context.Context) error
 	Running() bool
 }
+
+// WaitReady waits for the Corrosion service to be ready with the uncloud schema applied.
+func WaitReady(ctx context.Context, dataDir string) error {
+	// Read the config file to get the API address.
+	configPath := filepath.Join(dataDir, "config.toml")
+	configData, err := os.ReadFile(configPath)
+	if err != nil {
+		return fmt.Errorf("read config file: %w", err)
+	}
+
+	var config Config
+	if err = toml.Unmarshal(configData, &config); err != nil {
+		return fmt.Errorf("unmarshal config: %w", err)
+	}
+
+	corro, err := corrosion.NewAPIClient(config.API.Addr)
+	if err != nil {
+		return fmt.Errorf("create corrosion API client: %w", err)
+	}
+
+	// Corrosion starts serving the API before applying the schema. Query the cluster table with exponential backoff
+	// to check if the uncloud schema has been applied.
+	checkReady := func() error {
+		rows, err := corro.QueryContext(ctx, "SELECT 1 FROM cluster LIMIT 1")
+		if err != nil {
+			return fmt.Errorf("query cluster table: %w", err)
+		}
+		defer rows.Close()
+
+		return nil
+	}
+
+	b := backoff.NewExponentialBackOff(
+		backoff.WithInitialInterval(50*time.Millisecond),
+		backoff.WithMaxInterval(1*time.Second),
+		backoff.WithMaxElapsedTime(15*time.Second),
+	)
+	if err = backoff.Retry(checkReady, backoff.WithContext(b, ctx)); err != nil {
+		return fmt.Errorf("corrosion service did not become ready: %w", err)
+	}
+	return nil
+}

internal/machine/corroservice/systemd.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"log/slog"
 	"os/exec"
-	"time"
 )
 
 const DefaultSystemdUnit = "uncloud-corrosion.service"
@@ -44,20 +43,15 @@ func (s *SystemdService) startOrRestart(ctx context.Context, cmd string) error {
 	if _, err := exec.Command("systemctl", cmd, s.Unit).Output(); err != nil {
 		return fmt.Errorf("systemctl %s %s: %w", cmd, s.Unit, err)
 	}
-	slog.Info(fmt.Sprintf("Corrosion systemd service %sed.", cmd), "unit", s.Unit)
+	slog.Debug(fmt.Sprintf("Corrosion systemd service %sed.", cmd), "unit", s.Unit)
 
-	// Optimistically wait for the corrosion service to start and initialise the database schema before proceeding.
-	timer := time.NewTimer(2 * time.Second)
-	defer timer.Stop()
-
-	select {
-	case <-timer.C:
-	case <-ctx.Done():
-		return nil
+	slog.Debug("Waiting for corrosion service to be ready.")
+	if err := WaitReady(ctx, s.DataDir); err != nil {
+		return err
 	}
-
-	// TODO: run a goroutine to check the status of the service and log any errors in the uncloud log.
+	slog.Debug("Corrosion service is ready.")
 	s.running = true
+
 	return nil
 }
 

internal/machine/machine.go

@@ -321,6 +321,7 @@ func (m *Machine) Run(ctx context.Context) error {
 		if err := m.config.CorrosionService.Start(ctx); err != nil {
 			return fmt.Errorf("start corrosion service: %w", err)
 		}
+		slog.Info("Corrosion service started.")
 	}
 
 	// Use an errgroup to coordinate error handling and graceful shutdown of multiple machine components.