fix: do not include outdated containers from the removed machine in DNS and Caddy configs

cmd/uncloud/machine/rm.go

@@ -151,20 +151,22 @@ func remove(ctx context.Context, uncli *cli.CLI, nameOrID string, opts removeOpt
 		fmt.Println()
 	}
 
-	if _, err = client.RemoveMachine(ctx, &pb.RemoveMachineRequest{Id: m.Id}); err != nil {
+	// Initiate reset before removing the machine from the cluster to stop it from updating the cluster store.
-		return fmt.Errorf("remove machine from cluster: %w", err)
+	// This is still optimistic as Reset only triggers the reset process that runs asynchronously.
-	}
-	fmt.Printf("Machine '%s' removed from the cluster.\n", m.Name)
-
 	if reset && reachable {
 		_, err = client.MachineClient.Reset(rmCtx, &pb.ResetRequest{})
 		if err != nil {
-			fmt.Printf("WARNING: Failed to reset machine: %v\n", err)
+			tui.PrintWarning(fmt.Sprintf("Failed to reset machine: %v\n", err))
 		} else {
 			fmt.Println("Machine reset initiated and will complete in the background.")
 		}
 	}
 
+	if _, err = client.RemoveMachine(ctx, &pb.RemoveMachineRequest{Id: m.Id}); err != nil {
+		return fmt.Errorf("remove machine from cluster: %w", err)
+	}
+	fmt.Printf("Machine '%s' removed from the cluster.\n", m.Name)
+
 	// Remove the connection to the machine from the uncloud config if it exists.
 	if uncli.Config != nil {
 		contextName := uncli.ContextOverrideOrCurrent()

internal/machine/cluster/cluster.go

@@ -334,6 +334,13 @@ func (c *Cluster) RemoveMachine(ctx context.Context, req *pb.RemoveMachineReques
 		return nil, status.Error(codes.InvalidArgument, "machine ID not set")
 	}
 
+	// Cleanup machine containers from the store that could be left if the machine is unavailable
+	// removed with --no-reset, or didn't have time to finish propagating changes before resetting.
+	if err := c.store.DeleteContainers(ctx, store.DeleteOptions{MachineIDs: []string{req.Id}}); err != nil {
+		slog.Error("Failed to delete container records from the cluster store for the machine being removed.",
+			"id", req.Id, "err", err)
+	}
+
 	if err := c.store.DeleteMachine(ctx, req.Id); err != nil {
 		if errors.Is(err, store.ErrMachineNotFound) {
 			return nil, status.Errorf(codes.NotFound, "machine not found: %s", req.Id)

internal/machine/store/container.go

@@ -46,7 +46,10 @@ type ServiceIDOrNameOptions struct {
 }
 
 type DeleteOptions struct {
+	// IDs filters containers by their container IDs.
 	IDs []string
+	// MachineIDs filters containers by the machine IDs they are running on.
+	MachineIDs []string
 }
 
 // CreateOrUpdateContainer creates a new container record or updates an existing one in the store database.
@@ -100,21 +103,24 @@ func normaliseContainerForStore(ctr *api.ServiceContainer) {
 }
 
 // ListContainers returns a list of container records from the store database that match the given options.
+// The result excludes orphan containers whose machine is no longer in the cluster.
 func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]ContainerRecord, error) {
-	q := sq.Select("id", "container", "machine_id", "sync_status", "updated_at").From("containers").
+	q := sq.Select("c.id", "c.container", "c.machine_id", "c.sync_status", "c.updated_at").
-		Where(sq.Eq{"sync_status": SyncStatusSynced})
+		From("containers c").
+		Join("machines m ON m.id = c.machine_id").
+		Where(sq.Eq{"c.sync_status": SyncStatusSynced})
 
 	if len(opts.MachineIDs) > 0 {
-		q = q.Where(sq.Eq{"machine_id": opts.MachineIDs})
+		q = q.Where(sq.Eq{"c.machine_id": opts.MachineIDs})
 	}
 
 	if opts.ServiceIDOrName.ID != "" || opts.ServiceIDOrName.Name != "" {
 		var conditions []sq.Sqlizer
 		if opts.ServiceIDOrName.ID != "" {
-			conditions = append(conditions, sq.Eq{"service_id": opts.ServiceIDOrName.ID})
+			conditions = append(conditions, sq.Eq{"c.service_id": opts.ServiceIDOrName.ID})
 		}
 		if opts.ServiceIDOrName.Name != "" {
-			conditions = append(conditions, sq.Eq{"service_name": opts.ServiceIDOrName.Name})
+			conditions = append(conditions, sq.Eq{"c.service_name": opts.ServiceIDOrName.Name})
 		}
 		q = q.Where(sq.Or(conditions))
 	}
@@ -172,16 +178,19 @@ func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]Contain
 }
 
 // DeleteContainers deletes container records from the store database that match the given options.
+// If no filter is set, all container records are deleted. Filters are combined with AND.
 func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error {
-	query := "DELETE FROM containers"
+	q := sq.Delete("containers")
-	var args []any
-
 	if len(opts.IDs) > 0 {
-		query += " WHERE id IN (?" + strings.Repeat(", ?", len(opts.IDs)-1) + ")"
+		q = q.Where(sq.Eq{"id": opts.IDs})
-		args = make([]any, len(opts.IDs))
+	}
-		for i, id := range opts.IDs {
+	if len(opts.MachineIDs) > 0 {
-			args[i] = id
+		q = q.Where(sq.Eq{"machine_id": opts.MachineIDs})
-		}
+	}
+
+	query, args, err := q.ToSql()
+	if err != nil {
+		return fmt.Errorf("build query: %w", err)
 	}
 
 	res, err := s.corro.ExecContext(ctx, query, args...)
@@ -189,7 +198,8 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error
 		return fmt.Errorf("delete query: %w", err)
 	}
 	if res.RowsAffected > 0 {
-		slog.Debug("Container records deleted from store DB.", "ids", opts.IDs, "count", res.RowsAffected)
+		slog.Debug("Container records deleted from store DB.",
+			"ids", opts.IDs, "machine_ids", opts.MachineIDs, "count", res.RowsAffected)
 	}
 
 	return nil
@@ -197,10 +207,13 @@ 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.
 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("id", "container", "machine_id", "sync_status", "updated_at").From("containers").
+	q := sq.Select("c.id", "c.container", "c.machine_id", "c.sync_status", "c.updated_at").
-		Where(sq.Eq{"sync_status": SyncStatusSynced})
+		From("containers c").
+		Join("machines m ON m.id = c.machine_id").
+		Where(sq.Eq{"c.sync_status": SyncStatusSynced})
 	query, args, err := q.ToSql()
 	if err != nil {
 		return nil, nil, fmt.Errorf("build query: %w", err)

test/e2e/machine_test.go

@@ -3,7 +3,9 @@ package e2e
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	"github.com/psviderski/uncloud/internal/ucind"
@@ -434,4 +436,68 @@ func TestMachineOperations(t *testing.T) {
 		}
 		assert.Equal(t, len(targetMachine.Machine.Network.Endpoints), len(updatedMachine.Network.Endpoints))
 	})
+
+	t.Run("remove machine clears container records from cluster store", func(t *testing.T) {
+		// Deploy a global service with an HTTP ingress port so the auto-generated Caddyfile lists
+		// each container's IP as an upstream.
+		serviceName := "test-machine-rm-cleanup"
+		spec := api.ServiceSpec{
+			Name: serviceName,
+			Mode: api.ServiceModeGlobal,
+			Container: api.ContainerSpec{
+				Image: "portainer/pause:latest",
+			},
+			Ports: []api.PortSpec{
+				{
+					Hostname:      "test-machine-rm-cleanup.example.com",
+					ContainerPort: 8000,
+					Protocol:      api.ProtocolHTTP,
+					Mode:          api.PortModeIngress,
+				},
+			},
+		}
+
+		t.Cleanup(func() {
+			err := cli.RemoveService(ctx, serviceName)
+			if err != nil && !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+		})
+		_, err = cli.NewDeployment(spec, nil).Run(ctx)
+		require.NoError(t, err)
+
+		svc, err := cli.InspectService(ctx, serviceName)
+		require.NoError(t, err)
+
+		containers := serviceContainersByMachine(svc)
+
+		removedMachine := c.Machines[2]
+		removedIP := containers[removedMachine.ID][0].Container.UncloudNetworkIP().String()
+		keptIP := containers[c.Machines[0].ID][0].Container.UncloudNetworkIP().String()
+
+		// The Caddyfile contains both upstream IPs before the machine removal.
+		require.Eventually(t, func() bool {
+			cfg, err := cli.Caddy.GetConfig(ctx, nil)
+			if err != nil {
+				return false
+			}
+			return strings.Contains(cfg.Caddyfile, removedIP) && strings.Contains(cfg.Caddyfile, keptIP)
+		}, 15*time.Second, 100*time.Millisecond,
+			"expected Caddyfile to include both the to-be-removed and kept container IPs")
+
+		_, err = cli.RemoveMachine(ctx, &pb.RemoveMachineRequest{Id: removedMachine.ID})
+		require.NoError(t, err)
+
+		// After removal, the removed machine's container record should be gone from the cluster store,
+		// so the Caddy controller regenerates a Caddyfile without that upstream while keeping the
+		// upstreams for the still-running containers.
+		require.Eventually(t, func() bool {
+			cfg, err := cli.Caddy.GetConfig(ctx, nil)
+			if err != nil {
+				return false
+			}
+			return !strings.Contains(cfg.Caddyfile, removedIP) && strings.Contains(cfg.Caddyfile, keptIP)
+		}, 15*time.Second, 100*time.Millisecond,
+			"expected Caddyfile to drop the removed machine's container IP and keep the rest")
+	})
 }