fix(cluster): improve error handling on failed machine subscription in cluster store

internal/machine/caddyconfig/controller.go

@@ -107,7 +107,7 @@ func (c *Controller) Run(ctx context.Context) error {
 		select {
 		case _, ok := <-changes:
 			if !ok {
-				return fmt.Errorf("containers subscription failed")
+				return fmt.Errorf("subscription to container changes in cluster store failed")
 			}
 			c.log.Debug("Cluster containers changed, regenerating Caddy configuration.")
 

internal/machine/cluster.go

@@ -630,74 +630,52 @@ func (cc *clusterController) syncDockerContainers(ctx context.Context) error {
 }
 
 // handleMachineChanges subscribes to machine changes in the cluster and reconfigures the network peers accordingly
-// when changes occur.
+// when changes occur. It returns an error when the subscription fails.
 func (cc *clusterController) handleMachineChanges(ctx context.Context) error {
-	for {
+	machines, changes, err := cc.store.SubscribeMachines(ctx)
-		// Retry to subscribe to machine changes indefinitely until the context is done.
+	if err != nil {
-		boff := backoff.WithContext(backoff.NewExponentialBackOff(
+		return fmt.Errorf("subscribe to machine changes: %w", err)
-			backoff.WithInitialInterval(1*time.Second),
+	}
-			backoff.WithMaxInterval(60*time.Second),
+	slog.Info("Subscribed to machine changes in the cluster to reconfigure network peers.")
-			backoff.WithMaxElapsedTime(0),
-		), ctx)
 
-		var (
+	// Assume the initial store synchronization when this machine first joined the cluster has already been completed.
-			machines []*pb.MachineInfo
+	// So the machines should not be empty. But we still have a safety check to not reconfigure with an empty list,
-			changes  <-chan struct{}
+	// which would remove all peers and lock this machine out of the cluster.
-			err      error
+	// A list containing only this machine is a valid state (e.g. all other machines were removed) and should still
-		)
+	// trigger reconfiguration to drop any stale peers.
-		subscribe := func() error {
+	if len(machines) > 0 {
-			if machines, changes, err = cc.store.SubscribeMachines(ctx); err != nil {
+		slog.Info("Reconfiguring network peers with the current machines.", "machines", len(machines))
-				slog.Info("Failed to subscribe to machine changes, retrying.", "err", err)
+		if err = cc.configurePeers(machines); err != nil {
-			}
+			slog.Error("Failed to configure peers.", "err", err)
-			return err
-		}
-		if err = backoff.Retry(subscribe, boff); err != nil {
-			if errors.Is(err, context.Canceled) {
-				return nil
-			}
-			slog.Error("Unexpected error while retrying to subscribe to machine changes.", "err", err)
-			continue
-		}
-		slog.Info("Subscribed to machine changes in the cluster to reconfigure network peers.")
-
-		// The machine store may be empty when a machine first joins the cluster, before store synchronization
-		// completes. Skip configuration now and apply it when the store changes are received.
-		// TODO: remove this check after ensuring the store is actually synced to the latest known state at this point.
-		//  See TODO in waitStoreSync.
-		if len(machines) > 0 {
-			slog.Info("Reconfiguring network peers with the current machines.", "machines", len(machines))
-			if err = cc.configurePeers(machines); err != nil {
-				slog.Error("Failed to configure peers.", "err", err)
-			}
-		}
-		// For simplicity, reconfigure all peers on any change.
-		for {
-			select {
-			// TODO: test when Corrosion fails and the subscription fails to resubscribe (after 1 minute). It seems
-			//  the changes channel will be closed and this will become a busy loop. Perhaps, the outer for loop should
-			//  be reworked as well.
-			case <-changes:
-				slog.Info("Cluster machines changed, reconfiguring network peers.")
-				if machines, err = cc.store.ListMachines(ctx); err != nil {
-					slog.Error("Failed to list machines.", "err", err)
-					continue
-				}
-				// Skip reconfiguration if the machines list is empty. This can happen when joining the cluster.
-				// Corrosion can notifies about table changes before the data is fully replicated.
-				// Reconfiguring with an empty list would remove all peers and lock this machine out of the cluster.
-				// See https://github.com/psviderski/uncloud/issues/155.
-				if len(machines) == 0 {
-					slog.Debug("Skipping peer reconfiguration: machines list in store is empty.")
-					continue
-				}
-				if err = cc.configurePeers(machines); err != nil {
-					slog.Error("Failed to configure peers.", "err", err)
-				}
-			case <-ctx.Done():
-				return nil
-			}
 		}
 	}
+
+	// For simplicity, reconfigure all peers on any change. The subscription closes the changes channel both on context
+	// cancellation and when the subscription fails, so this loop exits in both cases.
+	for range changes {
+		slog.Info("Cluster machines changed, reconfiguring network peers.")
+		if machines, err = cc.store.ListMachines(ctx); err != nil {
+			slog.Error("Failed to list machines.", "err", err)
+			continue
+		}
+		// A safety check for the exceptional case when something bad happened with the store. Reconfiguring with an
+		// empty list would remove all peers and lock this machine out of the cluster.
+		// See https://github.com/psviderski/uncloud/issues/155.
+		if len(machines) == 0 {
+			slog.Debug("Skipping peer reconfiguration: machines list in store is empty.")
+			continue
+		}
+		if err = cc.configurePeers(machines); err != nil {
+			slog.Error("Failed to configure peers.", "err", err)
+		}
+	}
+
+	// The changes channel was closed. It's a clean shutdown if the context was cancelled, otherwise the subscription
+	// failed and we return an error to fail the controller.
+	if ctx.Err() != nil {
+		return nil
+	}
+	return fmt.Errorf("subscription to machine changes in cluster store failed")
 }
 
 func (cc *clusterController) configurePeers(machines []*pb.MachineInfo) error {

internal/machine/dns/resolver.go

@@ -50,7 +50,7 @@ func (r *ClusterResolver) Run(ctx context.Context) error {
 		select {
 		case _, ok := <-changes:
 			if !ok {
-				return fmt.Errorf("containers subscription failed")
+				return fmt.Errorf("subscription to container changes in cluster store failed")
 			}
 			r.log.Debug("Cluster containers changed, updating DNS records.")
 

internal/machine/store/container.go

@@ -208,6 +208,8 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error
 // SubscribeContainers returns a list of containers and a channel that signals changes to the list. The channel doesn't
 // receive any values, it just signals when a container(s) has been added, updated, or deleted in the database.
 // The result excludes orphan containers whose machine is no longer in the cluster.
+// The channel is closed when the containers are no longer subscribable: either the provided context is cancelled or
+// the underlying subscription fails.
 func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-chan struct{}, error) {
 	// TODO: figure out whether we need sync_status at all (not used at the moment).
 	q := sq.Select("c.id", "c.container", "c.machine_id", "c.sync_status", "c.updated_at").

internal/machine/store/store.go

@@ -265,6 +265,8 @@ func (s *Store) DeleteMachine(ctx context.Context, id string) error {
 
 // SubscribeMachines returns a list of machines and a channel that signals changes to the list. The channel doesn't
 // receive any values, it just signals when a machine has been added, updated, or deleted in the database.
+// The channel is closed when the machines are no longer subscribable: either the provided context is cancelled or
+// the underlying subscription fails.
 func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-chan struct{}, error) {
 	sub, err := s.corro.SubscribeContext(ctx, "SELECT id, info FROM machines ORDER BY name", nil, false)
 	if err != nil {