admin: AdminForward Phase 3 — register gRPC server + wire follower-side bridge (#648)

Phase 3 of Task #26 / AdminForward design 3.3. Closes the wiring loop
opened by #635 (leader-side dispatcher) and #644 (follower-side client +
handler integration): the admin HTTP DynamoHandler now hands
`ErrTablesNotLeader` writes off to the in-process `LeaderForwarder`, and
the leader-side gRPC `AdminForward` service is registered on every node
so the forwarded RPC actually lands somewhere.

## Summary

- **`admin.ServerDeps.Forwarder`** (3315bf4) — new optional field on
the high-level admin server composition surface. `admin.NewServer` calls
`WithLeaderForwarder` when set; nil leaves the existing 503 +
Retry-After:1 fallback untouched. Single-node / leader-only deployments
are unaffected.
- **`main_admin_forward.go`** (f3e529c) — production bridge between
`kv.GRPCConnCache` and the admin layer's typed `PBAdminForwardClient`
(via `pb.NewAdminForwardClient`). Includes `buildLeaderForwarder`,
`roleStoreFromFlags`, `adminForwardServerDeps`, and a
`readyForRegistration` gate.
- **Runtime wiring** (19f6b70) — `runtimeServerRunner.start()` now
creates `dynamoServer` BEFORE `startRaftServers`, so the leader-side
`ForwardServer` can use it as its `TablesSource`. `startRaftServers`
accepts the new `adminForwardServerDeps` bundle and delegates to
`registerAdminForwardServer`. `startServers` constructs a single
`*kv.GRPCConnCache` and forwards it (with the `Coordinator`) to
`startAdminFromFlags`, which in turn builds the production
`LeaderForwarder` via `buildAdminLeaderForwarder` and passes it through
`ServerDeps.Forwarder`.

## What is NOT in this PR

- Rolling-upgrade compatibility flag (criterion 5) — still deferred
behind a cluster-version bump.
- Election-period retry loop on the follower's bridge — the handler
returns 503 + Retry-After:1, the SPA / curl re-issues, and the bridge
dials again on the next attempt. Inline retry inside the handler would
just hide the latency from operators.

## Test plan

- [x] `go build ./...`
- [x] `go vet ./...`
- [x] `golangci-lint run` (main + admin packages: 0 issues)
- [x] `go test ./internal/admin/ -count=1 -race` — full admin suite
passes (existing 21 forward tests + 1 new
`TestServer_ServerDepsForwarderIsWired`)
- [x] `go test . -count=1 -race` — main package passes (4 new bridge /
role-store / readyForRegistration tests)
- [ ] `adapter` package times out at 240s on this branch AND on `main` —
pre-existing flake unrelated to this PR (verified by checking out main
and running the same suite)
- [ ] End-to-end smoke against a 3-node cluster — needs the local Jepsen
runner; blocked on the next PR or a manual run

## Acceptance criteria coverage

| # | Criterion | This PR |
|---|---|---|
| 1 | Leader direct write | ✓ (in main since #634) |
| 2 | Follower forwards transparently | ✓ — wiring complete; both
leader-side gRPC service AND follower-side bridge are now plumbed |
| 3 | Election-period 503 + retry | ✓ — handler still returns 503 +
Retry-After:1 when no leader is known; the SPA / client retries and the
next attempt re-dials |
| 4 | Leader demotes stale full role | ✓ (in main since #635) |
| 5 | Rolling-upgrade compat flag | ⏳ deferred (cluster-version bump) |
| 6 | `forwarded_from` in audit log | ✓ (in main since #635 — bridge now
stamps `--raftId` onto every forwarded request) |

## Self-review (5 lenses)

1. **Data loss**: No FSM / Raft / Pebble path changes; admin writes
still go through the same `AdminCreateTable` / `AdminDeleteTable` Raft
proposal once they reach the leader.
2. **Concurrency**: `runner.start()` reorder confirmed safe —
`startDynamoDBServer` only depends on `coordinate` + `shardStore`
(already constructed); raft TCP listeners are independent of the dynamo
HTTP listener. Worst case during the 100 ms-or-so reorder window is the
same "no leader yet" 503 the old order would have produced if a request
landed before raft converged.
3. **Performance**: One additional `&kv.GRPCConnCache{}` allocation per
process; one closure for the resolver. No hot-path changes.
4. **Data consistency**: Forwarded requests re-validate the principal at
the leader (criterion 4, already shipped); commit-ts ordering is
unchanged because the leader's `AdminCreateTable` path does its own
`HLC.Next`.
5. **Test coverage**: 4 new main-package tests + 1 new admin-package
server-level test. `TestServer_ServerDepsForwarderIsWired` is the
regression test for "future refactor drops the forwarder before it
reaches the dynamo handler" — exactly the failure mode the wiring change
in this PR introduces the risk of.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **New Features**
* Added leader-forwarding capability for admin requests sent to follower
nodes
  * Implemented role-based access control for admin API operations
* Enhanced DynamoDB request routing to automatically direct operations
to leader nodes

* **Tests**
* Added comprehensive integration and unit tests validating
leader-forwarding paths

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

internal/admin/server.go

@@ -39,6 +39,14 @@ type ServerDeps struct {
 	// cluster page deploy without standing up the dynamo bridge.
 	Tables TablesSource
 
+	// Forwarder is the LeaderForwarder that the Dynamo handler hands
+	// off ErrTablesNotLeader writes to (design 3.3, AdminForward).
+	// Optional: a nil value disables follower→leader forwarding, in
+	// which case the handler surfaces 503 + Retry-After: 1 directly.
+	// Single-node and leader-only deployments leave this nil; multi-
+	// node clusters wire the production gRPC client.
+	Forwarder LeaderForwarder
+
 	// StaticFS is the embed.FS (or any fs.FS) backing the SPA. May be
 	// nil during early development; the router renders 404 for
 	// /admin/assets/* and the SPA fallback in that case.
@@ -110,9 +118,13 @@ func NewServer(deps ServerDeps) (*Server, error) {
 		// operators must restart the listener for revocation to
 		// take effect, but the JWT no longer extends a revoked
 		// key past the next request.
-		dynamo = NewDynamoHandler(deps.Tables).
+		dynamoHandler := NewDynamoHandler(deps.Tables).
 			WithLogger(logger).
 			WithRoleStore(MapRoleStore(deps.Roles))
+		if deps.Forwarder != nil {
+			dynamoHandler = dynamoHandler.WithLeaderForwarder(deps.Forwarder)
+		}
+		dynamo = dynamoHandler
 	}
 	mux := buildAPIMux(auth, deps.Verifier, cluster, dynamo, logger)
 	router := NewRouter(mux, deps.StaticFS)

internal/admin/server_test.go

@@ -453,6 +453,57 @@ func TestServer_DynamoDeleteTable_FullRoleHappyPath(t *testing.T) {
 	_ = resp.Body.Close()
 }
 
+// TestServer_ServerDepsForwarderIsWired confirms that a non-nil
+// ServerDeps.Forwarder reaches the DynamoHandler so a follower-side
+// CreateTable that the source returns ErrTablesNotLeader for is
+// transparently forwarded. Without the wire, the request would 503 +
+// Retry-After:1 — the no-forwarder fallback path. With it, the
+// leader's response status (here, 201) is replayed verbatim.
+func TestServer_ServerDepsForwarderIsWired(t *testing.T) {
+	src := &notLeaderSource{}
+	fwd := &stubLeaderForwarder{createRes: &ForwardResult{
+		StatusCode:  http.StatusCreated,
+		Payload:     []byte(`{"name":"users"}`),
+		ContentType: "application/json; charset=utf-8",
+	}}
+	clk := fixedClock(time.Unix(1_700_000_000, 0).UTC())
+	signer := newSignerForTest(t, 1, clk)
+	verifier := newVerifierForTest(t, []byte{1}, clk)
+	cluster := ClusterInfoFunc(func(_ context.Context) (ClusterInfo, error) {
+		return ClusterInfo{NodeID: "node-1", Version: "0.1.0"}, nil
+	})
+	srv, err := NewServer(ServerDeps{
+		Signer:      signer,
+		Verifier:    verifier,
+		Credentials: MapCredentialStore{"AKIA_ADMIN": "ADMIN_SECRET"},
+		Roles:       map[string]Role{"AKIA_ADMIN": RoleFull},
+		ClusterInfo: cluster,
+		Tables:      src,
+		Forwarder:   fwd,
+		AuthOpts:    AuthServiceOpts{Clock: clk},
+	})
+	require.NoError(t, err)
+
+	ts := httptest.NewServer(srv.Handler())
+	defer ts.Close()
+	cookies := loginAsFullAdminAndCookies(t, ts)
+	body := strings.NewReader(`{"table_name":"users","partition_key":{"name":"id","type":"S"}}`)
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodPost,
+		ts.URL+"/admin/api/v1/dynamo/tables", body)
+	require.NoError(t, err)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set(csrfHeaderName, csrfHeaderFromCookies(cookies))
+	for _, c := range cookies {
+		req.AddCookie(c)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusCreated, resp.StatusCode)
+	require.Equal(t, "users", fwd.lastCreateInput.TableName,
+		"forwarder must be invoked when source returns ErrTablesNotLeader")
+	_ = resp.Body.Close()
+}
+
 func TestServer_WriteRejectsMissingCSRF(t *testing.T) {
 	// Login to obtain a session, then hit cluster with POST to trigger
 	// CSRF on what the router normally rejects as method_not_allowed.

main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/bootjp/elastickv/adapter"
 	"github.com/bootjp/elastickv/distribution"
 	internalutil "github.com/bootjp/elastickv/internal"
+	"github.com/bootjp/elastickv/internal/admin"
 	"github.com/bootjp/elastickv/internal/memwatch"
 	internalraftadmin "github.com/bootjp/elastickv/internal/raftadmin"
 	"github.com/bootjp/elastickv/internal/raftengine"
@@ -680,6 +681,40 @@ func startServers(in serversInput) error {
 	if err != nil {
 		return err
 	}
+	// roleStore + connCache are gated on *adminEnabled. With admin
+	// disabled, building either is wasted work AND a security
+	// regression risk: a non-empty -adminFullAccessKeys flag would
+	// otherwise still flip forwardDeps.readyForRegistration() to
+	// true, registering the leader-side gRPC AdminForward service
+	// and re-exposing the table-write surface a follower-direct
+	// admin call could reach (Codex P1, CodeRabbit Major on #648).
+	// The HTTP admin listener already short-circuits in
+	// startAdminFromFlags when *adminEnabled is false; the gRPC path
+	// must do the same.
+	var (
+		roleStore admin.RoleStore
+		connCache *kv.GRPCConnCache
+	)
+	if *adminEnabled {
+		roleStore = roleStoreFromFlags(parseCSV(*adminFullAccessKeys), parseCSV(*adminReadOnlyAccessKeys))
+		// connCache is shared between the follower-side LeaderForwarder
+		// (built inside startAdminFromFlags) and any future bridge that
+		// dials the leader's gRPC ports. Keeping a single instance per
+		// process means the two paths re-use TLS / HTTP/2 connections
+		// rather than each maintaining a parallel pool. The shutdown
+		// goroutine drains the cache on context cancellation so the
+		// accumulated HTTP/2 connections are not leaked when the
+		// process exits gracefully (Claude review on #648).
+		connCache = &kv.GRPCConnCache{}
+		cache := connCache
+		in.eg.Go(func() error {
+			<-in.ctx.Done()
+			if err := cache.Close(); err != nil {
+				return errors.Wrap(err, "close admin gRPC connection cache")
+			}
+			return nil
+		})
+	}
 	runner := runtimeServerRunner{
 		ctx:             in.ctx,
 		lc:              in.lc,
@@ -711,6 +746,7 @@ func startServers(in serversInput) error {
 		pprofAddress:    *pprofAddr,
 		pprofToken:      *pprofToken,
 		metricsRegistry: in.metricsRegistry,
+		roleStore:       roleStore,
 	}
 	if err := runner.start(); err != nil {
 		return err
@@ -720,7 +756,14 @@ func startServers(in serversInput) error {
 	// Passing nil here would leave the admin dashboard with no
 	// access to table metadata; the admin handler answers
 	// /admin/api/v1/dynamo/* with 404 in that case.
-	if err := startAdminFromFlags(in.ctx, in.lc, in.eg, in.runtimes, runner.dynamoServer); err != nil {
+	//
+	// in.coordinate + connCache are forwarded so the admin HTTP
+	// dynamo handler can construct its production LeaderForwarder
+	// (Phase 3 of design 3.3): when the local node is a follower,
+	// the handler hands ErrTablesNotLeader writes to the forwarder
+	// which dials the leader over the cached gRPC pool. Without these
+	// the handler falls back to 503 + Retry-After:1.
+	if err := startAdminFromFlags(in.ctx, in.lc, in.eg, in.runtimes, runner.dynamoServer, in.coordinate, connCache); err != nil {
 		return waitErrgroupAfterStartupFailure(in.cancel, in.eg, err)
 	}
 	return nil
@@ -975,7 +1018,9 @@ func startRaftServers(
 	proposalObserverForGroup func(uint64) kv.ProposalObserver,
 	adminServer *adapter.AdminServer,
 	adminGRPCOpts adminGRPCInterceptors,
+	forwardDeps adminForwardServerDeps,
 ) error {
+	forwardLogger := slog.Default().With(slog.String("component", "admin"))
 	// extraOptsCap reserves slots for the unary + stream admin interceptor
 	// options appended below. Sized as a constant so the magic-number
 	// linter does not complain.
@@ -1005,6 +1050,7 @@ func startRaftServers(
 		if adminServer != nil {
 			pb.RegisterAdminServer(gs, adminServer)
 		}
+		registerAdminForwardServer(gs, forwardDeps, forwardLogger)
 		rt.registerGRPC(gs)
 		internalraftadmin.RegisterOperationalServices(ctx, gs, rt.engine, []string{"RawKV"})
 		reflection.Register(gs)
@@ -1255,12 +1301,37 @@ type runtimeServerRunner struct {
 	// field is unexported on purpose — it is package-private state,
 	// not a public API. Nil until start() reaches the dynamo step.
 	dynamoServer *adapter.DynamoDBServer
+
+	// roleStore is the access-key → role index the leader-side
+	// gRPC AdminForward service uses to re-validate the principal
+	// on every forwarded write. Mirrors what admin.Config.RoleIndex
+	// produces inside startAdminFromFlags; built up-front in
+	// startServers so registerAdminForwardServer in startRaftServers
+	// does not need to wait for the (later) admin-config parse.
+	// Nil when no admin access keys are configured.
+	roleStore admin.RoleStore
 }
 
 func (r *runtimeServerRunner) start() error {
 	if err := startRedisServer(r.ctx, r.lc, r.eg, r.redisAddress, r.shardStore, r.coordinate, r.leaderRedis, r.pubsubRelay, r.metricsRegistry, r.readTracker); err != nil {
 		return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err)
 	}
+	// startDynamoDBServer must run BEFORE startRaftServers so the
+	// resulting DynamoDBServer is available to the leader-side gRPC
+	// AdminForward registration in startRaftServers (design 3.3).
+	// Both servers listen on different addresses; the dynamo HTTP
+	// listener accepting traffic before raft TCP listeners are up
+	// is no different from the existing startup-race semantics — a
+	// hit in that window already returned 503 before this reorder.
+	dynamoServer, err := startDynamoDBServer(r.ctx, r.lc, r.eg, r.dynamoAddress, r.shardStore, r.coordinate, r.leaderDynamo, r.metricsRegistry, r.readTracker)
+	if err != nil {
+		return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err)
+	}
+	r.dynamoServer = dynamoServer
+	forwardDeps := adminForwardServerDeps{
+		tables: newDynamoTablesSource(r.dynamoServer),
+		roles:  r.roleStore,
+	}
 	if err := startRaftServers(
 		r.ctx,
 		r.lc,
@@ -1275,14 +1346,10 @@ func (r *runtimeServerRunner) start() error {
 		},
 		r.adminServer,
 		r.adminGRPCOpts,
+		forwardDeps,
 	); err != nil {
 		return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err)
 	}
-	dynamoServer, err := startDynamoDBServer(r.ctx, r.lc, r.eg, r.dynamoAddress, r.shardStore, r.coordinate, r.leaderDynamo, r.metricsRegistry, r.readTracker)
-	if err != nil {
-		return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err)
-	}
-	r.dynamoServer = dynamoServer
 	if err := startS3Server(r.ctx, r.lc, r.eg, r.s3Address, r.shardStore, r.coordinate, r.leaderS3, r.s3Region, r.s3CredsFile, r.s3PathStyleOnly, r.readTracker); err != nil {
 		return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err)
 	}

main_admin.go

@@ -68,7 +68,15 @@ type adminListenerConfig struct {
 // without touching --s3CredentialsFile: pulling the admin feature into
 // a hard dependency on that file would break deployments that never
 // intended to use it.
-func startAdminFromFlags(ctx context.Context, lc *net.ListenConfig, eg *errgroup.Group, runtimes []*raftGroupRuntime, dynamoServer *adapter.DynamoDBServer) error {
+func startAdminFromFlags(
+	ctx context.Context,
+	lc *net.ListenConfig,
+	eg *errgroup.Group,
+	runtimes []*raftGroupRuntime,
+	dynamoServer *adapter.DynamoDBServer,
+	coordinate kv.Coordinator,
+	connCache *kv.GRPCConnCache,
+) error {
 	if !*adminEnabled {
 		return nil
 	}
@@ -109,10 +117,40 @@ func startAdminFromFlags(ctx context.Context, lc *net.ListenConfig, eg *errgroup
 	}
 	clusterSrc := newClusterInfoSource(*raftId, buildVersion(), runtimes)
 	tablesSrc := newDynamoTablesSource(dynamoServer)
-	_, err = startAdminServer(ctx, lc, eg, cfg, staticCreds, clusterSrc, tablesSrc, buildVersion())
+	forwarder, err := buildAdminLeaderForwarder(coordinate, connCache, *raftId)
+	if err != nil {
+		return errors.Wrap(err, "build admin leader forwarder")
+	}
+	_, err = startAdminServer(ctx, lc, eg, cfg, staticCreds, clusterSrc, tablesSrc, forwarder, buildVersion())
 	return err
 }
 
+// buildAdminLeaderForwarder constructs the production LeaderForwarder
+// for the dynamo HTTP handler when the wiring is complete enough to
+// reach a remote leader. The bridge tolerates a nil connCache (and a
+// nil coordinate) so single-node / leader-only builds — where the
+// dashboard always hits a leader — can ship without paying the
+// forwarder's wiring cost. tablesSrc itself can be nil for cluster-
+// only builds; that's handled higher up by ServerDeps.Tables == nil.
+func buildAdminLeaderForwarder(coordinate kv.Coordinator, connCache *kv.GRPCConnCache, nodeID string) (admin.LeaderForwarder, error) {
+	if coordinate == nil || connCache == nil {
+		// Returning (nil, nil) is the explicit "no forwarder" signal
+		// — the handler falls back to 503 + Retry-After:1 on
+		// ErrTablesNotLeader. The function-level doc comment above
+		// describes this contract; the nilnil linter is not enabled
+		// in .golangci.yaml so no suppression directive is needed
+		// (Claude review on #648).
+		return nil, nil
+	}
+	if nodeID == "" {
+		// admin.NewGRPCForwardClient enforces this too; surfacing
+		// it here keeps the misconfiguration message in the wiring
+		// layer rather than buried under a Wrap chain.
+		return nil, errors.New("admin forward bridge: --raftId is required")
+	}
+	return buildLeaderForwarder(coordinate, connCache, nodeID)
+}
+
 // newDynamoTablesSource adapts *adapter.DynamoDBServer to the
 // admin.TablesSource interface. The bridge stays in this file (rather
 // than internal/admin) so the admin package stays free of the heavy
@@ -346,14 +384,15 @@ func startAdminServer(
 	creds map[string]string,
 	cluster admin.ClusterInfoSource,
 	tables admin.TablesSource,
+	forwarder admin.LeaderForwarder,
 	version string,
 ) (string, error) {
 	adminCfg := buildAdminConfig(cfg)
 	enabled, err := checkAdminConfig(&adminCfg, cluster)
 	if err != nil || !enabled {
 		return "", err
 	}
-	server, err := buildAdminHTTPServer(&adminCfg, creds, cluster, tables)
+	server, err := buildAdminHTTPServer(&adminCfg, creds, cluster, tables, forwarder)
 	if err != nil {
 		return "", err
 	}
@@ -393,7 +432,7 @@ func checkAdminConfig(adminCfg *admin.Config, cluster admin.ClusterInfoSource) (
 	return true, nil
 }
 
-func buildAdminHTTPServer(adminCfg *admin.Config, creds map[string]string, cluster admin.ClusterInfoSource, tables admin.TablesSource) (*admin.Server, error) {
+func buildAdminHTTPServer(adminCfg *admin.Config, creds map[string]string, cluster admin.ClusterInfoSource, tables admin.TablesSource, forwarder admin.LeaderForwarder) (*admin.Server, error) {
 	primaryKeys, err := adminCfg.DecodedSigningKeys()
 	if err != nil {
 		return nil, errors.Wrap(err, "decode admin signing keys")
@@ -413,6 +452,7 @@ func buildAdminHTTPServer(adminCfg *admin.Config, creds map[string]string, clust
 		Roles:       adminCfg.RoleIndex(),
 		ClusterInfo: cluster,
 		Tables:      tables,
+		Forwarder:   forwarder,
 		StaticFS:    nil,
 		AuthOpts: admin.AuthServiceOpts{
 			InsecureCookie: adminCfg.AllowInsecureDevCookie,