Revert "feat: invert control plane ↔ agent to an agent-initiated gRPC stream "

.gitignore

@@ -3,10 +3,6 @@
 *.exe
 *.out
 
-# Local Firecracker assets: kernel image, converted rootfs cache, and per-VM
-# work dirs the agent creates at runtime (some root-owned). Not source.
-/fc-assets/
-
 # Environment
 .env
 

Makefile

@@ -1,11 +1,10 @@
-.PHONY: run run-agent build build-agent test test-e2e test-kvm test-bpf tidy fmt bpf-generate proto-generate
+.PHONY: run run-agent build build-agent test test-e2e test-kvm test-bpf tidy fmt bpf-generate
 
 run:
 	go run ./cmd/server
 
-# Run a host agent (P3). The agent dials the control plane's gRPC AgentLink and
-# holds a stream open; override the target via env, e.g.
-# MODE=agent CONTROL_PLANE_GRPC_ADDR=localhost:8090 AGENT_RUNTIME=fake.
+# Run a host agent (P3). Defaults target a local control plane; override via env,
+# e.g. MODE=agent CONTROL_PLANE_URL=... ADVERTISE_ADDR=... PORT=9000.
 run-agent:
 	MODE=agent go run ./cmd/agent
 
@@ -50,19 +49,6 @@ bpf-generate:
 	bpftool btf dump file /sys/kernel/btf/vmlinux format c > $(BPF_DIR)/vmlinux.h
 	go generate ./$(BPF_DIR)/...
 
-# Regenerate the gRPC AgentLink stubs from proto/agentlink/agentlink.proto. This
-# is a MAINTAINER step; the outputs (internal/agentlink/pb/*.pb.go) are committed
-# so `go build`/CI need only the Go toolchain. Run after editing the .proto, then
-# `git add` the regenerated files. Requires protoc plus the Go plugins:
-#   go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
-#   go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
-proto-generate:
-	@command -v protoc >/dev/null || { echo "need protoc (protobuf-compiler)"; exit 1; }
-	protoc \
-		--go_out=. --go_opt=module=github.com/aarani/craftling-go \
-		--go-grpc_out=. --go-grpc_opt=module=github.com/aarani/craftling-go \
-		proto/agentlink/agentlink.proto
-
 tidy:
 	go mod tidy
 

cmd/agent/main.go

@@ -1,17 +1,17 @@
-// Command agent is the host-side worker (P3). It dials the control plane and
-// holds a persistent gRPC stream open over which the control plane pushes VM
-// lifecycle commands (provision/start/stop/deprovision); the agent runs them
-// against its local Runtime and answers on the same stream. It has no inbound
-// API — the open stream both delivers commands and proves the host's liveness.
+// Command agent is the host-side worker (P3). It exposes a VM API the control
+// plane calls to provision/start/stop/deprovision local VMs, and it registers +
+// heartbeats with the control plane so the scheduler can place servers on it.
 //
 // It ships with the in-memory FakeRuntime; a real Firecracker driver (P4) slots
 // in behind the same Runtime interface without changing this wiring.
 package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
+	"net/http"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -27,6 +27,15 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	// heartbeatInterval is how often the agent proves liveness to the control
+	// plane. It must be comfortably below the control plane's host TTL (30s).
+	heartbeatInterval = 10 * time.Second
+	// registerRetryInterval is how long to wait between registration attempts
+	// while the control plane is unreachable.
+	registerRetryInterval = 5 * time.Second
+)
+
 func main() {
 	cfg := config.Load()
 
@@ -36,28 +45,57 @@ func main() {
 	}
 	defer func() { _ = zlog.Sync() }()
 
-	// The runtime that actually runs VMs, driven by commands off the link.
+	advertiseAddr := cfg.Agent.AdvertiseAddr
+	if advertiseAddr == "" {
+		// Best-effort default so a local single-host run works out of the box.
+		advertiseAddr = "localhost:" + cfg.Port
+	}
+
+	// The runtime that actually runs VMs, fronted by the agent HTTP API.
 	rt, err := newRuntime(cfg, zlog)
 	if err != nil {
 		zlog.Fatal("init runtime", zap.Error(err))
 	}
+	srv := &http.Server{
+		Addr:         ":" + cfg.Port,
+		Handler:      agent.NewRouter(rt, zlog),
+		ReadTimeout:  10 * time.Second,
+		WriteTimeout: 10 * time.Second,
+		IdleTimeout:  60 * time.Second,
+	}
 
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 
-	// Hold a persistent connection to the control plane for the agent's lifetime.
-	// RunLink blocks until ctx is cancelled, reconnecting on its own if the stream
-	// drops, so this is the agent's main loop.
-	zlog.Info("connecting to control plane", zap.String("addr", cfg.Agent.ControlPlaneGRPCAddr))
-	agent.RunLink(ctx, cfg.Agent.ControlPlaneGRPCAddr, rt, agent.LinkInfo{
+	go func() {
+		zlog.Info("agent listening",
+			zap.String("port", cfg.Port), zap.String("advertise_addr", advertiseAddr))
+		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			zlog.Fatal("agent listen failed", zap.Error(err))
+		}
+	}()
+
+	// Register with the control plane and keep the host alive via heartbeats.
+	cp := agent.NewCPClient(cfg.Agent.ControlPlaneURL, &http.Client{Timeout: 10 * time.Second})
+	go runRegistration(ctx, zlog, cp, agent.RegisterRequest{
 		ID:            cfg.Agent.ID,
 		Hostname:      cfg.Agent.Hostname,
+		Address:       advertiseAddr,
 		Zone:          cfg.Agent.Zone,
-		AgentVersion:  cfg.Agent.Version,
 		CPUsTotal:     cfg.Agent.CPUsTotal,
 		MemoryMBTotal: cfg.Agent.MemoryMBTotal,
-	}, zlog)
+		AgentVersion:  cfg.Agent.Version,
+	})
 
+	<-ctx.Done()
+	stop()
+	zlog.Info("shutting down agent...")
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := srv.Shutdown(shutdownCtx); err != nil {
+		zlog.Fatal("forced shutdown", zap.Error(err))
+	}
 	zlog.Info("agent exited")
 }
 
@@ -123,3 +161,57 @@ func imageCacheDir(fc config.FirecrackerConfig) string {
 	}
 	return filepath.Join(workDir, "images")
 }
+
+// runRegistration registers the host then heartbeats on an interval until ctx is
+// cancelled. A heartbeat that the control plane rejects with "not found" (it was
+// restarted and forgot us) triggers a re-register, restoring the same identity.
+func runRegistration(ctx context.Context, log *zap.Logger, cp *agent.CPClient, req agent.RegisterRequest) {
+	id := register(ctx, log, cp, req)
+	if id == "" {
+		return // ctx cancelled before we registered
+	}
+	// Re-register under the assigned id so identity stays stable across restarts.
+	req.ID = id
+
+	ticker := time.NewTicker(heartbeatInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			found, err := cp.Heartbeat(ctx, id)
+			if err != nil {
+				log.Warn("heartbeat failed", zap.Error(err))
+				continue
+			}
+			if !found {
+				log.Warn("control plane forgot host; re-registering", zap.String("id", id))
+				if newID := register(ctx, log, cp, req); newID != "" {
+					id = newID
+					req.ID = newID
+				}
+			}
+		}
+	}
+}
+
+// register retries registration until it succeeds or ctx is cancelled,
+// returning the assigned host id (empty on cancellation).
+func register(ctx context.Context, log *zap.Logger, cp *agent.CPClient, req agent.RegisterRequest) string {
+	for {
+		id, err := cp.Register(ctx, req)
+		if err == nil {
+			log.Info("registered with control plane", zap.String("id", id))
+			return id
+		}
+		log.Warn("register failed; retrying", zap.Error(err))
+
+		select {
+		case <-ctx.Done():
+			return ""
+		case <-time.After(registerRetryInterval):
+		}
+	}
+}

cmd/server/main.go

@@ -4,14 +4,12 @@ import (
 	"context"
 	"errors"
 	"log"
-	"net"
 	"net/http"
 	"os/signal"
 	"syscall"
 	"time"
 
-	"github.com/aarani/craftling-go/internal/agentlink"
-	pb "github.com/aarani/craftling-go/internal/agentlink/pb"
+	"github.com/aarani/craftling-go/internal/agent"
 	"github.com/aarani/craftling-go/internal/config"
 	"github.com/aarani/craftling-go/internal/db"
 	"github.com/aarani/craftling-go/internal/handler"
@@ -24,7 +22,6 @@ import (
 	"github.com/aarani/craftling-go/internal/seed"
 	"github.com/aarani/craftling-go/internal/worldstore"
 	"go.uber.org/zap"
-	"google.golang.org/grpc"
 )
 
 const (
@@ -37,6 +34,8 @@ const (
 	// hostHeartbeatTTL is how long a host may go without heartbeating before it
 	// is marked down.
 	hostHeartbeatTTL = 30 * time.Second
+	// agentCallTimeout bounds each control-plane→agent VM API call.
+	agentCallTimeout = 10 * time.Second
 	// worldGCInterval is how often the durable world store is swept for
 	// snapshots belonging to no live server (P5b).
 	worldGCInterval = time.Hour
@@ -74,10 +73,8 @@ func main() {
 	dbCancel()
 
 	// The fleet inventory lives in process memory (P1). It is shared between the
-	// agent link hub (which registers/heartbeats hosts as their streams come and
-	// go) and the host reaper.
+	// HTTP handlers (register/heartbeat) and the host reaper.
 	hostRepo := repository.NewHostRepository()
-	gameServerRepo := repository.NewGameServerRepository(pool)
 
 	router := handler.NewRouter(cfg, zlog, pool, hostRepo)
 
@@ -89,18 +86,6 @@ func main() {
 		IdleTimeout:  60 * time.Second,
 	}
 
-	// The hub is the control plane's end of the persistent agent connection:
-	// agents dial the gRPC listener and hold a stream open, and the hub pushes VM
-	// commands down it. It registers hosts (reconstructing committed capacity
-	// from the durable server records) and tracks liveness off the stream.
-	hub := agentlink.NewHub(hostRepo, gameServerRepo, zlog)
-	grpcSrv := grpc.NewServer()
-	pb.RegisterAgentLinkServer(grpcSrv, hub)
-	grpcLis, err := net.Listen("tcp", ":"+cfg.GRPCPort)
-	if err != nil {
-		zlog.Fatal("listen for agent gRPC", zap.Error(err))
-	}
-
 	// ctx is cancelled on the first interrupt/terminate signal, which both
 	// stops the reaper and triggers graceful shutdown.
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -118,8 +103,8 @@ func main() {
 	// then drives the VM by calling the assigned host's agent (the control plane
 	// never touches KVM itself).
 	sched := scheduler.New(hostRepo)
-	prov := provisioner.NewRemote(hub)
-	rec := reconciler.New(gameServerRepo, prov, sched, zlog)
+	prov := provisioner.NewRemote(hostRepo, agent.NewClient(&http.Client{Timeout: agentCallTimeout}))
+	rec := reconciler.New(repository.NewGameServerRepository(pool), prov, sched, zlog)
 	go rec.Run(ctx, reconcileInterval)
 
 	// If a durable world store is configured, periodically GC snapshots that no
@@ -131,17 +116,9 @@ func main() {
 	if err != nil {
 		zlog.Warn("world store unavailable; world GC disabled", zap.Error(err))
 	} else if worldStore != nil {
-		go reaper.Worlds(ctx, zlog, worldStore, gameServerRepo, worldGCInterval)
+		go reaper.Worlds(ctx, zlog, worldStore, repository.NewGameServerRepository(pool), worldGCInterval)
 	}
 
-	// Serve the agent gRPC link alongside the HTTP API.
-	go func() {
-		zlog.Info("agent gRPC listening", zap.String("port", cfg.GRPCPort))
-		if err := grpcSrv.Serve(grpcLis); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
-			zlog.Fatal("agent gRPC serve failed", zap.Error(err))
-		}
-	}()
-
 	// Start the server in a goroutine so it doesn't block graceful shutdown handling.
 	go func() {
 		zlog.Info("server listening", zap.String("port", cfg.Port), zap.String("env", cfg.Env))
@@ -154,8 +131,6 @@ func main() {
 	stop() // restore default signal handling so a second signal force-quits
 	zlog.Info("shutting down server...")
 
-	grpcSrv.GracefulStop()
-
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	if err := srv.Shutdown(shutdownCtx); err != nil {

docker-compose.yml

@@ -38,10 +38,6 @@ services:
       ADMIN_PASSWORD: adminpassword
     ports:
       - "8082:8080"
-    # The gRPC AgentLink listener (default :8090) is reachable to agents over the
-    # compose network; no host publish is needed.
-    expose:
-      - "8090"
     depends_on:
       db:
         condition: service_healthy
@@ -70,11 +66,11 @@ services:
       dockerfile: Dockerfile.agent
     environment:
       MODE: agent
+      PORT: "9000"
       APP_ENV: production
-      # The agent dials this and holds a stream open; the control plane pushes VM
-      # commands down it. No inbound address is advertised — the control plane
-      # never dials the agent.
-      CONTROL_PLANE_GRPC_ADDR: server:8090
+      CONTROL_PLANE_URL: http://server:8080
+      # Address the control plane calls back (reachable on the compose network).
+      ADVERTISE_ADDR: agent:9000
       # Player-facing connect host VMs report (override for real connectivity).
       ADVERTISE_HOST: 127.0.0.1
       AGENT_RUNTIME: firecracker
@@ -94,8 +90,9 @@ services:
     privileged: true
     volumes:
       - ./fc-assets:/var/lib/craftling
-    # No published ports: the agent serves no inbound API. It reaches the control
-    # plane's gRPC link at server:8090 over the compose network.
+    ports:
+      # Exposed for debugging the agent API directly.
+      - "9000:9000"
     depends_on:
       - server
     restart: unless-stopped

go.mod

@@ -22,8 +22,6 @@ require (
 	go.uber.org/zap v1.28.0
 	golang.org/x/crypto v0.52.0
 	golang.org/x/sys v0.46.0
-	google.golang.org/grpc v1.81.1
-	google.golang.org/protobuf v1.36.11
 )
 
 require (
@@ -136,7 +134,7 @@ require (
 	golang.org/x/net v0.55.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/text v0.37.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/ini.v1 v1.67.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -133,8 +133,6 @@ github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM=
 github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
 github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY=
 github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
-github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
-github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
@@ -331,12 +329,6 @@ golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk=
 golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
 golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
-gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 h1:XF8+t6QQiS0o9ArVan/HW8Q7cycNPGsJf6GA2nXxYAg=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
-google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ=
-google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=