fix(ha): redeem join tokens atomically

internal/api/server/api_pki.go

@@ -7,13 +7,14 @@
 package server
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"time"
 
 	"github.com/ellanetworks/core/internal/cluster/pkiissuer"
-	"github.com/ellanetworks/core/internal/db"
 	"github.com/ellanetworks/core/internal/logger"
 	"github.com/ellanetworks/core/internal/pki"
 )
@@ -22,6 +23,15 @@ const (
 	PKIMintJoinTokenAction = "pki_mint_join_token" // #nosec G101 -- audit action name
 )
 
+// A leader that has just been promoted registers its own cluster
+// certificate as part of leader init; minting before that commit
+// lands would embed a stale pin, so the handler waits it out rather
+// than failing a request that is about to become serviceable.
+const (
+	mintReadyWait = 15 * time.Second
+	mintReadyPoll = 100 * time.Millisecond
+)
+
 // pkiAdminEndpoint resolves the pkiissuer.Service at request time and
 // dispatches to build. Returns 503 until the issuer service has been
 // installed by runtime and its join-HMAC key is committed.
@@ -57,10 +67,9 @@ type MintJoinTokenResponse struct {
 }
 
 // PKIMintJoinToken handles POST /api/v1/cluster/pki/join-tokens.
-// The handler runs on the leader (the public API forwards to it),
-// so dbInstance.NodeID() identifies the leader whose pin gets
-// embedded in the token's claims.
-func PKIMintJoinToken(dbInstance *db.Database, svc *pkiissuer.Service) http.Handler {
+// Minting is leader-only and is not forwarded: a request that lands
+// on a follower fails.
+func PKIMintJoinToken(svc *pkiissuer.Service) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		var req MintJoinTokenRequest
 		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
@@ -80,9 +89,17 @@ func PKIMintJoinToken(dbInstance *db.Database, svc *pkiissuer.Service) http.Hand
 			ttl = 30 * time.Minute
 		}
 
-		token, err := svc.MintJoinToken(r.Context(), req.NodeID, ttl, dbInstance.NodeID())
+		token, err := mintWhenReady(r.Context(), svc, req.NodeID, ttl)
 		if err != nil {
+			if errors.Is(err, pkiissuer.ErrNotReady) {
+				w.Header().Set("Retry-After", "1")
+				writeError(r.Context(), w, http.StatusServiceUnavailable, "mint token", err, logger.APILog)
+
+				return
+			}
+
 			writeError(r.Context(), w, http.StatusInternalServerError, "mint token", err, logger.APILog)
+
 			return
 		}
 
@@ -102,3 +119,22 @@ func PKIMintJoinToken(dbInstance *db.Database, svc *pkiissuer.Service) http.Hand
 		}, http.StatusCreated, logger.APILog)
 	})
 }
+
+// mintWhenReady retries MintJoinToken while the issuer reports
+// ErrNotReady, up to mintReadyWait.
+func mintWhenReady(ctx context.Context, svc *pkiissuer.Service, nodeID int, ttl time.Duration) (string, error) {
+	deadline := time.Now().Add(mintReadyWait)
+
+	for {
+		token, err := svc.MintJoinToken(ctx, nodeID, ttl)
+		if !errors.Is(err, pkiissuer.ErrNotReady) || time.Now().After(deadline) {
+			return token, err
+		}
+
+		select {
+		case <-ctx.Done():
+			return "", ctx.Err()
+		case <-time.After(mintReadyPoll):
+		}
+	}
+}

internal/api/server/api_pki_test.go

@@ -71,7 +71,9 @@ func TestPKIAdminEndpoints_MintToken(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	issuer := pkiissuer.New(env.DB)
+	var leaderFP string
+
+	issuer := pkiissuer.New(env.DB, func() string { return leaderFP })
 	if err := issuer.Bootstrap(context.Background()); err != nil {
 		t.Fatalf("Bootstrap: %v", err)
 	}
@@ -83,9 +85,11 @@ func TestPKIAdminEndpoints_MintToken(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	leaderFP = pki.Fingerprint(leaderCert)
+
 	if err := env.DB.UpsertClusterNodeCert(context.Background(), &db.ClusterNodeCert{
 		NodeID:      1,
-		Fingerprint: pki.Fingerprint(leaderCert),
+		Fingerprint: leaderFP,
 		CertPEM:     string(pki.EncodeCertPEM(leaderCert)),
 		AddedAt:     time.Now().Unix(),
 	}); err != nil {
@@ -149,7 +153,7 @@ func TestPKIAdminEndpoints_InstalledButNotBootstrapped503(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	server.SetPKIIssuer(pkiissuer.New(env.DB))
+	server.SetPKIIssuer(pkiissuer.New(env.DB, nil))
 	t.Cleanup(func() { server.SetPKIIssuer(nil) })
 
 	admin, err := initializeAndRefresh(env.Server.URL, env.Server.Client())

internal/api/server/cluster_http_forward.go

@@ -117,12 +117,40 @@ func mapApplyErrorToHTTP(ctx context.Context, w http.ResponseWriter, err error)
 		writeProposeForwardError(ctx, w, http.StatusServiceUnavailable,
 			"raft busy or shutting down", err)
 
+	case errors.Is(err, db.ErrMigrationPending):
+		writeProposeForwardCodedError(ctx, w, http.StatusServiceUnavailable,
+			"schema migration pending", ellaraft.ForwardCodeMigrationPend, err)
+
 	default:
+		if code := forwardCodeForDomainErr(err); code != "" {
+			writeProposeForwardCodedError(ctx, w, http.StatusConflict,
+				err.Error(), code, err)
+
+			return
+		}
+
 		writeProposeForwardError(ctx, w, http.StatusInternalServerError,
 			"apply failed", err)
 	}
 }
 
+func forwardCodeForDomainErr(err error) string {
+	switch {
+	case errors.Is(err, db.ErrJoinTokenAlreadyConsumed):
+		return ellaraft.ForwardCodeTokenConsumed
+	case errors.Is(err, db.ErrJoinTokenExpired):
+		return ellaraft.ForwardCodeTokenExpired
+	case errors.Is(err, db.ErrJoinTokenNodeMismatch):
+		return ellaraft.ForwardCodeTokenNodeMism
+	case errors.Is(err, db.ErrAlreadyExists):
+		return ellaraft.ForwardCodeAlreadyExists
+	case errors.Is(err, db.ErrNotFound):
+		return ellaraft.ForwardCodeNotFound
+	default:
+		return ""
+	}
+}
+
 func writeProposeForwardError(ctx context.Context, w http.ResponseWriter, status int, message string, cause error) {
 	writeProposeForwardCodedError(ctx, w, status, message, "", cause)
 }

internal/api/server/cluster_pki.go

@@ -14,6 +14,7 @@ import (
 	"bufio"
 	"bytes"
 	"encoding/json"
+	"errors"
 	"io"
 	"net"
 	"net/http"
@@ -22,8 +23,10 @@ import (
 	"github.com/ellanetworks/core/internal/cluster/listener"
 	"github.com/ellanetworks/core/internal/cluster/pkiagent"
 	"github.com/ellanetworks/core/internal/cluster/pkiissuer"
+	"github.com/ellanetworks/core/internal/db"
 	"github.com/ellanetworks/core/internal/logger"
 	"github.com/ellanetworks/core/internal/pki"
+	hraft "github.com/hashicorp/raft"
 	"go.uber.org/zap"
 )
 
@@ -54,6 +57,11 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler {
 
 		peerNodeID, hasPeer := peerNodeIDFromContext(r.Context())
 
+		var (
+			fp   string
+			pins []db.ClusterNodeCert
+		)
+
 		switch {
 		case hasPeer:
 			// mTLS path: the cert's owner is the only caller who
@@ -65,6 +73,12 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler {
 				return
 			}
 
+			fp, pins, err = svc.RegisterCert(r.Context(), req.NodeID, []byte(req.CertPEM))
+			if err != nil {
+				writeError(r.Context(), w, clusterPKIStatus(err, http.StatusBadRequest), "register cert", err, logger.APILog)
+				return
+			}
+
 		default:
 			if req.Token == "" {
 				writeError(r.Context(), w, http.StatusUnauthorized,
@@ -73,26 +87,13 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler {
 				return
 			}
 
-			claims, err := svc.VerifyAndConsumeJoinToken(r.Context(), req.Token)
+			fp, pins, err = svc.RedeemJoinToken(r.Context(), req.Token, req.NodeID, []byte(req.CertPEM))
 			if err != nil {
-				writeError(r.Context(), w, http.StatusUnauthorized, "verify join token", err, logger.APILog)
-				return
-			}
-
-			if claims.NodeID != req.NodeID {
-				writeError(r.Context(), w, http.StatusForbidden,
-					"node-id in body does not match token claims", nil, logger.APILog)
-
+				writeError(r.Context(), w, clusterPKIStatus(err, http.StatusUnauthorized), "redeem join token", err, logger.APILog)
 				return
 			}
 		}
 
-		fp, pins, err := svc.RegisterCert(r.Context(), req.NodeID, []byte(req.CertPEM))
-		if err != nil {
-			writeError(r.Context(), w, http.StatusBadRequest, "register cert", err, logger.APILog)
-			return
-		}
-
 		records := make([]pkiagent.PinRecord, 0, len(pins))
 		for _, p := range pins {
 			records = append(records, pkiagent.PinRecord{NodeID: p.NodeID, Fingerprint: p.Fingerprint})
@@ -107,6 +108,22 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler {
 	})
 }
 
+func clusterPKIStatus(err error, deny int) int {
+	switch {
+	case errors.Is(err, db.ErrMigrationPending),
+		errors.Is(err, db.ErrProposeTimeout),
+		errors.Is(err, db.ErrNotFound),
+		errors.Is(err, hraft.ErrNotLeader):
+		return http.StatusServiceUnavailable
+
+	case errors.Is(err, db.ErrOutcomeUnknown):
+		return http.StatusConflict
+
+	default:
+		return deny
+	}
+}
+
 // RegisterBootstrapALPN dispatches POST /cluster/pki/register on
 // the bootstrap ALPN (no client cert) and closes the connection
 // after one request.

internal/api/server/cluster_pki_e2e_test.go

@@ -62,7 +62,7 @@ func TestClusterPKI_JoinFlowEndToEnd(t *testing.T) {
 		t.Fatalf("preregister leader pin: %v", err)
 	}
 
-	issuer := pkiissuer.New(leaderDB)
+	issuer := pkiissuer.New(leaderDB, func() string { return pki.Fingerprint(leaderAgent.Leaf().Leaf) })
 	if err := issuer.Bootstrap(ctx); err != nil {
 		t.Fatalf("issuer bootstrap: %v", err)
 	}
@@ -95,7 +95,7 @@ func TestClusterPKI_JoinFlowEndToEnd(t *testing.T) {
 	t.Cleanup(leaderLn.Stop)
 
 	// Mint a token for nodeID 2 (the joiner). Leader is nodeID 1.
-	token, err := issuer.MintJoinToken(ctx, 2, 5*time.Minute, 1)
+	token, err := issuer.MintJoinToken(ctx, 2, 5*time.Minute)
 	if err != nil {
 		t.Fatalf("mint join token: %v", err)
 	}

internal/api/server/server.go

@@ -261,7 +261,7 @@ func NewHandler(cfg HandlerConfig) http.Handler {
 	// request time (set by runtime after first-leader bootstrap), so
 	// these routes can be registered before the issuer is ready.
 	mux.HandleFunc("POST /api/v1/cluster/pki/join-tokens", Authenticate(jwtSecret, dbInstance, Authorize(PermManageCluster, pkiAdminEndpoint(func(svc *pkiissuer.Service) http.Handler {
-		return PKIMintJoinToken(dbInstance, svc)
+		return PKIMintJoinToken(svc)
 	}))).ServeHTTP)
 
 	// Fallback to UI