fix: fix zombie runners if control-plane was down when runners restar…

…ted (#33)

Upeon restarting, the control-plane previously had no way of knowing if
a registration was from an existing runner or a new one. Added a new
runner-generated UUID key to the runners table to distinguish between
the two cases.

.gitattributes

@@ -0,0 +1 @@
+/console/src/protos/**/* linguist-generated=true

controlplane/controlplane.go

@@ -76,19 +76,14 @@ type Service struct {
 }
 
 func New(ctx context.Context, dal *dal.DAL, heartbeatTimeout time.Duration, artefactChunkSize int) (*Service, error) {
-	logger := log.FromContext(ctx)
-	count, err := dal.DeleteStaleRunners(context.Background(), heartbeatTimeout)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to delete stale runners")
-	}
-	logger.Warnf("Deleted %d stale runners", count)
-	return &Service{
+	svc := &Service{
 		dal:               dal,
 		heartbeatTimeout:  heartbeatTimeout,
 		artefactChunkSize: artefactChunkSize,
-	}, nil
+	}
+	go svc.reapStaleRunners(ctx)
+	return svc, nil
 }
-
 func (s *Service) Deploy(ctx context.Context, req *connect.Request[ftlv1.DeployRequest]) (*connect.Response[ftlv1.DeployResponse], error) {
 	panic("unimplemented")
 }
@@ -110,6 +105,10 @@ func (s *Service) RegisterRunner(ctx context.Context, req *connect.ClientStream[
 	if endpoint.Scheme != "http" && endpoint.Scheme != "https" {
 		return nil, connect.NewError(connect.CodeUnavailable, errors.Errorf("invalid endpoint scheme %q", endpoint.Scheme))
 	}
+	key, err := uuid.Parse(msg.Key)
+	if err != nil {
+		return nil, connect.NewError(connect.CodeFailedPrecondition, errors.Wrap(err, "invalid key"))
+	}
 
 	// Check if we can contact the runner.
 	client := rpc.Dial(ftlv1connect.NewRunnerServiceClient, endpoint.String())
@@ -119,7 +118,7 @@ func (s *Service) RegisterRunner(ctx context.Context, req *connect.ClientStream[
 		return nil, connect.NewError(connect.CodeUnavailable, errors.Wrap(err, "failed to connect to runner"))
 	}
 
-	runnerID, err := s.dal.RegisterRunner(ctx, msg.Language, endpoint)
+	runnerID, err := s.dal.RegisterRunner(ctx, key, msg.Language, endpoint)
 	if errors.Is(err, dal.ErrConflict) {
 		return nil, connect.NewError(connect.CodeAlreadyExists, err)
 	} else if err != nil {
@@ -302,3 +301,21 @@ func (s *Service) getDeployment(ctx context.Context, key string) (*dal.Deploymen
 	}
 	return deployment, nil
 }
+
+func (s *Service) reapStaleRunners(ctx context.Context) {
+	logger := log.FromContext(ctx)
+	for {
+		count, err := s.dal.DeleteStaleRunners(context.Background(), s.heartbeatTimeout)
+		if err != nil {
+			logger.Errorf(err, "Failed to delete stale runners")
+		} else if count > 0 {
+			logger.Warnf("Deleted %d stale runners", count)
+		}
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-time.After(s.heartbeatTimeout):
+		}
+	}
+}

controlplane/controlplane_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/alecthomas/concurrency"
 	connect "github.com/bufbuild/connect-go"
 	grpcreflect "github.com/bufbuild/connect-grpcreflect-go"
+	"github.com/google/uuid"
 	"github.com/jpillora/backoff"
 
 	"github.com/TBD54566975/ftl/common/log"
@@ -26,14 +27,17 @@ func TestControlPlaneRegisterRunnerHeartbeatClose(t *testing.T) {
 
 	stream := client.RegisterRunner(ctx)
 	t.Cleanup(func() { _, _ = stream.CloseAndReceive() })
+	key := uuid.NewString()
 	err := stream.Send(&ftlv1.RegisterRunnerRequest{
+		Key:      key,
 		Language: "go",
 		Endpoint: bind.String(),
 	})
 	assert.NoError(t, err)
 	time.Sleep(time.Millisecond * 100)
 
 	err = stream.Send(&ftlv1.RegisterRunnerRequest{
+		Key:      key,
 		Language: "go",
 		Endpoint: bind.String(),
 	})
@@ -54,8 +58,10 @@ func TestControlPlaneRegisterRunnerHeartbeatClose(t *testing.T) {
 func TestControlPlaneRegisterRunnerHeartbeatTimeout(t *testing.T) {
 	db, client, bind, ctx := startForTesting(t)
 
+	key := uuid.NewString()
 	stream := client.RegisterRunner(ctx)
 	err := stream.Send(&ftlv1.RegisterRunnerRequest{
+		Key:      key,
 		Language: "go",
 		Endpoint: bind.String(),
 	})

controlplane/internal/dal/dal.go

@@ -205,8 +205,8 @@ type RunnerID int64
 // RegisterRunner registers a new runner.
 //
 // It will return ErrConflict if a runner with the same endpoint already exists.
-func (d *DAL) RegisterRunner(ctx context.Context, language string, endpoint *url.URL) (RunnerID, error) {
-	id, err := d.db.RegisterRunner(ctx, language, endpoint.String())
+func (d *DAL) RegisterRunner(ctx context.Context, key uuid.UUID, language string, endpoint *url.URL) (RunnerID, error) {
+	id, err := d.db.RegisterRunner(ctx, key, language, endpoint.String())
 	if isPGConflict(err) {
 		return 0, errors.Wrap(ErrConflict, "runner already registered")
 	}

controlplane/internal/sql/queries.sql

@@ -70,7 +70,8 @@ FROM artefacts a
 WHERE a.id = @id;
 
 -- name: RegisterRunner :one
-INSERT INTO runners (language, endpoint) VALUES ($1, $2)
+INSERT INTO runners (key, language, endpoint) VALUES ($1, $2, $3)
+ON CONFLICT (key) DO UPDATE SET language = $2, endpoint = $3
 RETURNING id;
 
 -- name: DeleteStaleRunners :one

controlplane/internal/sql/schema/001_init.sql

@@ -41,6 +41,8 @@ CREATE INDEX deployment_artefacts_deployment_id_idx ON deployment_artefacts (dep
 -- Runners are processes that are available to run modules.
 CREATE TABLE runners (
   id BIGSERIAL PRIMARY KEY NOT NULL,
+  -- Unique identifier for this runner, generated at startup.
+  key UUID UNIQUE NOT NULL,
   last_seen TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT (NOW() AT TIME ZONE 'utc'),
   language VARCHAR(64) NOT NULL,
   endpoint VARCHAR(255) UNIQUE NOT NULL,