Controller: Connect stream, worker sessions and the worker client

README.md

@@ -3,8 +3,9 @@
 Core of [Opifica](https://github.com/opifica), an open-source CI/CD system built in the EU
 where every job runs in a fresh virtual machine.
 
-> **Status:** early development. The pipeline language, the API definitions and the worker kit's warm pool
-> work and are tested. The controller, agent and signer are skeletons.
+> **Status:** early development. Working and tested: the pipeline language, the worker kit's warm pool, and
+> workers joining and staying connected to the controller. Scheduling and the agent come next; see
+> [docs/ROADMAP.md](docs/ROADMAP.md).
 > See the [organisation page](https://github.com/opifica) for the architecture.
 
 ## What lives here
@@ -15,9 +16,10 @@ where every job runs in a fresh virtual machine.
 | `schema/` | JSON Schema for pipeline files, for editor autocompletion | Working |
 | `cmd/opifica` | CLI: `validate`, `plan`, `schema` | Working |
 | `proto/` | API definitions: worker ↔ controller, worker ↔ guest agent (Protobuf, ConnectRPC) | Draft |
-| `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, image verification | Working |
+| `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, image verification, controller client | Working |
 | `pkg/worker/workertest` | Fake hypervisor for testing worker code without VMs | Working |
-| `cmd/controller` | API, Git providers, pipeline engine, scheduler, secrets broker | Skeleton |
+| `cmd/controller` | Join tokens, worker joining over TLS, the Connect stream, worker revocation | Working |
+| | Scheduler, runs, logs, secrets, Git providers | Not started |
 | `cmd/agent` | Runs inside each VM, executes steps, streams logs and results | Skeleton |
 | `cmd/signer` | Signing service for release artifacts | Not started |
 
@@ -111,9 +113,22 @@ broken.yaml:11:5: unknown field "secret"
 Requires Go 1.27 or later.
 
 ```sh
-make tools      # install buf and the protoc plugins (once)
+make tools      # install buf, the protoc plugins and sqlc (once)
 make            # lint, test and build into bin/
-make generate   # after changing proto/; generated code in gen/ is committed
+make generate   # after changing proto/ or SQL queries; generated code is committed
+```
+
+Database tests need PostgreSQL. Either set `OPIFICA_TEST_DATABASE_URL` to a server the tests may create databases
+on, or install the PostgreSQL server binaries (e.g. `apt install postgresql-17`): the tests then start their own
+throwaway server. Without either, those tests are skipped.
+
+### Trying the controller
+
+```sh
+export OPIFICA_DATABASE_URL=postgres://localhost/opifica
+go run ./cmd/controller token create -pool trusted   # prints a join token and the CA fingerprint
+go run ./cmd/controller serve                        # TLS on :8443
+go run ./cmd/controller worker list
 ```
 
 ## Stack

cmd/controller/main.go

@@ -2,28 +2,27 @@
 // integration, pipeline engine, scheduler and secrets broker.
 //
 // Implemented so far: the database, the built-in certificate authority, join
-// tokens and Join. Every other RPC returns "unimplemented".
+// tokens, Join and the Connect stream (sessions, heartbeats, revocation).
+// Scheduling comes next.
 package main
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"flag"
 	"fmt"
 	"io"
 	"log/slog"
-	"net/http"
 	"os"
 	"os/signal"
 	"strings"
 	"syscall"
+	"text/tabwriter"
 	"time"
 
-	"connectrpc.com/connect"
+	"github.com/google/uuid"
 
-	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
-	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/controller/api"
 	"github.com/opifica/opifica/internal/controller/ca"
 	"github.com/opifica/opifica/internal/controller/store"
 	"github.com/opifica/opifica/internal/controller/workers"
@@ -36,6 +35,8 @@ const usage = `Usage: controller <command> [options]
 Commands:
   serve          Run the controller
   token create   Create a one-time join token for a new build machine
+  worker list    List workers
+  worker revoke  Revoke a worker; it is disconnected and can't reconnect
   version        Print the version
 
 Common options (or environment variables):
@@ -57,6 +58,10 @@ func run(args []string, stdout, stderr io.Writer) int {
 		return serve(args[1:], stderr)
 	case args[0] == "token" && len(args) > 1 && args[1] == "create":
 		return tokenCreate(args[2:], stdout, stderr)
+	case args[0] == "worker" && len(args) > 1 && args[1] == "list":
+		return workerList(args[2:], stdout, stderr)
+	case args[0] == "worker" && len(args) > 1 && args[1] == "revoke":
+		return workerRevoke(args[2:], stdout, stderr)
 	case args[0] == "version":
 		fmt.Fprintln(stdout, version.Version)
 		return 0
@@ -121,42 +126,19 @@ func serve(args []string, stderr io.Writer) int {
 	defer closeFn()
 	svc.Log = log
 
-	tlsCfg, err := tlsConfig(svc.CA, splitList(*hosts))
+	tlsCfg, err := api.TLSConfig(svc.CA, splitList(*hosts))
 	if err != nil {
 		log.Error("issuing server certificate", "err", err)
 		return 1
 	}
 	log.Info("certificate authority ready", "fingerprint", svc.CA.Fingerprint())
-	if err := server.RunTLS(ctx, *addr, newHandler(svc), tlsCfg, log); err != nil {
+	if err := server.RunTLS(ctx, *addr, api.NewHandler(svc), tlsCfg, log); err != nil {
 		log.Error("server stopped", "err", err)
 		return 1
 	}
 	return 0
 }
 
-// tlsConfig issues a fresh server certificate and accepts, but doesn't yet
-// require, worker client certificates: Join is called before a worker has
-// one. RPCs that need a worker identity check it themselves.
-func tlsConfig(authority *ca.CA, hosts []string) (*tls.Config, error) {
-	certPEM, keyPEM, err := authority.IssueServerCert(hosts)
-	if err != nil {
-		return nil, err
-	}
-	// Send the CA too, so a joining worker can check it against the pinned
-	// fingerprint before it has the CA bundle.
-	certPEM = append(certPEM, authority.CertPEM()...)
-	pair, err := tls.X509KeyPair(certPEM, keyPEM)
-	if err != nil {
-		return nil, err
-	}
-	return &tls.Config{
-		Certificates: []tls.Certificate{pair},
-		ClientCAs:    authority.Pool(),
-		ClientAuth:   tls.VerifyClientCertIfGiven,
-		MinVersion:   tls.VersionTLS13,
-	}, nil
-}
-
 func defaultHosts() string {
 	hosts := []string{"localhost", "127.0.0.1", "::1"}
 	if h, err := os.Hostname(); err == nil && h != "" {
@@ -175,27 +157,6 @@ func splitList(s string) []string {
 	return out
 }
 
-func newHandler(svc *workers.Service) http.Handler {
-	mux := http.NewServeMux()
-	mux.HandleFunc("GET /healthz", server.Health)
-	mux.Handle(opificav1connect.NewWorkerServiceHandler(&workerHandler{svc: svc}))
-	return mux
-}
-
-// workerHandler implements the API workers connect to.
-type workerHandler struct {
-	opificav1connect.UnimplementedWorkerServiceHandler
-	svc *workers.Service
-}
-
-func (h *workerHandler) Join(ctx context.Context, req *connect.Request[opificav1.JoinRequest]) (*connect.Response[opificav1.JoinResponse], error) {
-	resp, err := h.svc.Join(ctx, req.Msg)
-	if err != nil {
-		return nil, err
-	}
-	return connect.NewResponse(resp), nil
-}
-
 type labelFlag map[string]string
 
 func (l labelFlag) String() string { return "" }
@@ -252,3 +213,74 @@ The token is shown only once.
 `, *pool, tok.ExpiresAt.Format(time.RFC3339), tok.Secret, tok.CAFingerprint)
 	return 0
 }
+
+func workerList(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("worker list", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	svc, closeFn, err := c.open(ctx)
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	defer closeFn()
+	ws, err := svc.Store.ListWorkers(ctx)
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	tw := tabwriter.NewWriter(stdout, 0, 4, 2, ' ', 0)
+	fmt.Fprintln(tw, "ID\tHOSTNAME\tOS\tARCH\tPOOL\tJOBS\tLAST SEEN\tSTATUS")
+	for _, w := range ws {
+		seen, status := "never", "active"
+		if w.LastSeenAt != nil {
+			seen = time.Since(*w.LastSeenAt).Round(time.Second).String() + " ago"
+		}
+		if w.RevokedAt != nil {
+			status = "revoked"
+		}
+		fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%d\t%s\t%s\n", w.ID, w.Hostname, w.Os, w.Arch, w.Pool, w.MaxConcurrentJobs, seen, status)
+	}
+	tw.Flush()
+	return 0
+}
+
+func workerRevoke(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("worker revoke", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	fs.Usage = func() { fmt.Fprintln(stderr, "Usage: controller worker revoke [options] <worker-id>") }
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	if fs.NArg() != 1 {
+		fs.Usage()
+		return 2
+	}
+	id, err := uuid.Parse(fs.Arg(0))
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: invalid worker id %q\n", fs.Arg(0))
+		return 2
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	svc, closeFn, err := c.open(ctx)
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	defer closeFn()
+	if err := svc.Revoke(ctx, id); err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	fmt.Fprintf(stdout, "Worker %s revoked. A connected controller disconnects it within %v.\n", id, workers.DefaultTouchInterval)
+	return 0
+}

cmd/controller/main_test.go

@@ -22,6 +22,7 @@ import (
 
 	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
 	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/controller/api"
 	"github.com/opifica/opifica/internal/server"
 	"github.com/opifica/opifica/internal/testutil/pgtest"
 	"github.com/opifica/opifica/pkg/worker"
@@ -56,15 +57,15 @@ func TestTokenCreateAndJoinOverTLS(t *testing.T) {
 	}
 	defer closeFn()
 	svc.Log = slog.New(slog.NewTextHandler(io.Discard, nil))
-	tlsCfg, err := tlsConfig(svc.CA, []string{"127.0.0.1"})
+	tlsCfg, err := api.TLSConfig(svc.CA, []string{"127.0.0.1"})
 	if err != nil {
 		t.Fatal(err)
 	}
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatal(err)
 	}
-	go server.ServeTLS(ctx, ln, newHandler(svc), tlsCfg, svc.Log)
+	go server.ServeTLS(ctx, ln, api.NewHandler(svc), tlsCfg, svc.Log)
 
 	// The worker has no CA yet: it accepts the controller only if the
 	// certificate chains to a CA with the pinned fingerprint.
@@ -94,11 +95,11 @@ func TestTokenCreateAndJoinOverTLS(t *testing.T) {
 		t.Errorf("Join response = %+v", resp.Msg)
 	}
 
-	// Still unimplemented: the stream comes next.
+	// Connect without a client certificate is refused.
 	stream := wc.Connect(ctx)
 	stream.Send(&opificav1.ConnectRequest{})
-	if _, err := stream.Receive(); connect.CodeOf(err) != connect.CodeUnimplemented {
-		t.Errorf("Connect: err = %v, want unimplemented", err)
+	if _, err := stream.Receive(); connect.CodeOf(err) != connect.CodeUnauthenticated {
+		t.Errorf("Connect without a certificate: err = %v, want unauthenticated", err)
 	}
 	stream.CloseRequest()
 	stream.CloseResponse()

docs/ROADMAP.md

@@ -29,20 +29,23 @@ result are stored. Tested in ordinary CI, using the fake hypervisor with a real
 - [x] **Certificate authority:** built-in CA that issues worker client certificates and the server certificate
 - [x] **Join:** admins create one-time join tokens (`controller token create`); `Join` checks the token and signs
   the worker's CSR; first contact is verified against the CA fingerprint printed with the token
-- [ ] **Connect stream:** Hello, heartbeats, worker registry, disconnect and reconnect handling
+- [x] **Connect stream:** Hello, heartbeats, worker registry, disconnect and reconnect handling, revocation
+  (`controller worker list`, `controller worker revoke`)
 - [ ] **Leases:** job assignment with a lease renewed by heartbeats; expired leases put the job back in the queue
 - [ ] **Scheduler:** match planned jobs to workers by OS, architecture, image, labels and pool; respect `needs`
 - [ ] **Runs:** create a run from a pipeline file and an event; track job and run state
 - [ ] **Logs:** receive log chunks, drop duplicates after reconnects, store them on disk
 - [ ] **Secrets:** encrypted store; requests checked against the job token and policy
 - [ ] **Admin config:** static file for pools, routing (event → pool) and secret policy
 - [x] **TLS** on the worker API, with worker client certificates accepted
-- [ ] **mTLS required** for `Connect`, and certificate renewal
+- [x] **mTLS required** for `Connect`; the certificate must be the worker's current one
+- [ ] Certificate renewal before the 30-day expiry
 - [ ] **Admin API + CLI:** `opifica worker token create`, `opifica runs list`, `opifica logs <job>`
 
 ### Worker kit and agent
 
-- [ ] Controller client: join, connect, reconnect with backoff, heartbeats from pool stats
+- [x] Controller client: join (saving the identity to disk), connect, reconnect with backoff, heartbeats
+- [ ] Heartbeats filled from the warm pool and job runner
 - [ ] Job runner: lease a VM → checkout → run steps through the agent → stream masked logs → collect reports
   and artifacts → destroy the VM
 - [ ] Local Git mirror per repository

internal/controller/api/api.go

@@ -0,0 +1,66 @@
+// Package api assembles the controller's HTTP handler and TLS settings.
+package api
+
+import (
+	"context"
+	"crypto/tls"
+	"net/http"
+
+	"connectrpc.com/connect"
+
+	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
+	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/controller/ca"
+	"github.com/opifica/opifica/internal/controller/workers"
+	"github.com/opifica/opifica/internal/server"
+)
+
+// NewHandler returns the controller's HTTP handler.
+func NewHandler(svc *workers.Service) http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /healthz", server.Health)
+	mux.Handle(opificav1connect.NewWorkerServiceHandler(&workerHandler{svc: svc}))
+	return workers.WithPeerCert(mux)
+}
+
+// TLSConfig issues a fresh server certificate for hosts. Worker client
+// certificates are verified when presented but not required at the TLS
+// layer, because Join is called before a worker has one; Connect requires
+// one itself.
+func TLSConfig(authority *ca.CA, hosts []string) (*tls.Config, error) {
+	certPEM, keyPEM, err := authority.IssueServerCert(hosts)
+	if err != nil {
+		return nil, err
+	}
+	// Send the CA too, so a joining worker can check it against the pinned
+	// fingerprint before it has the CA bundle.
+	certPEM = append(certPEM, authority.CertPEM()...)
+	pair, err := tls.X509KeyPair(certPEM, keyPEM)
+	if err != nil {
+		return nil, err
+	}
+	return &tls.Config{
+		Certificates: []tls.Certificate{pair},
+		ClientCAs:    authority.Pool(),
+		ClientAuth:   tls.VerifyClientCertIfGiven,
+		MinVersion:   tls.VersionTLS13,
+	}, nil
+}
+
+// workerHandler implements the API workers connect to.
+type workerHandler struct {
+	opificav1connect.UnimplementedWorkerServiceHandler
+	svc *workers.Service
+}
+
+func (h *workerHandler) Join(ctx context.Context, req *connect.Request[opificav1.JoinRequest]) (*connect.Response[opificav1.JoinResponse], error) {
+	resp, err := h.svc.Join(ctx, req.Msg)
+	if err != nil {
+		return nil, err
+	}
+	return connect.NewResponse(resp), nil
+}
+
+func (h *workerHandler) Connect(ctx context.Context, stream *connect.BidiStream[opificav1.ConnectRequest, opificav1.ConnectResponse]) error {
+	return h.svc.Connect(ctx, workers.PeerCert(ctx), stream)
+}