admin: scaffold AdminForward follower-side client (Task #26 phase 1)

Adds the LeaderForwarder interface and a gRPC-backed implementation
that the dynamo HTTP handler will invoke when the local node is a
follower (the source returned ErrTablesNotLeader). This commit only
introduces the client + its tests; wiring it into the handler and
the main_admin.go bridge is the next phase.

Pieces:
- LeaderForwarder interface with ForwardCreateTable /
  ForwardDeleteTable, decoupled from proto so the handler stays
  proto-free.
- ForwardResult is the transport-neutral envelope the handler
  re-emits verbatim, so a forwarded request looks identical to a
  leader-direct call from the SPA's perspective.
- ErrLeaderUnavailable for the "no leader known" path
  (acceptance criterion 3 — the handler will turn this into 503
  + Retry-After: 1 once wired up).
- gRPCForwardClient builds the AdminForwardRequest from the admin
  CreateTableRequest / table name + AuthPrincipal, dials via a
  GRPCConnFactory abstraction, and re-shapes the response.
- forwarded_from is populated with the local node id so the
  leader's audit log line carries the trace (criterion 6, leader
  side already in #635).
- Defensive: status_code == 0 (proto zero value) is upgraded to
  502 Bad Gateway; missing ContentType is filled with the JSON
  default. Both surface transport bugs rather than producing
  silently-malformed SPA responses.

Tests cover:
- constructor input validation (3 cases)
- happy paths for both Forward operations including principal,
  payload, op enum, and forwarded_from round-trip
- ErrLeaderUnavailable when the resolver returns ""
- dial / rpc errors propagated with cockroachdb/errors wrapping
- zero status code upgrade
- missing content type fallback

Not in this commit: handler integration, gRPC server registration
in main.go, election-period retry. Those land in follow-ups before
the PR opens.

Author: bootjp

internal/admin/forward_client.go

@@ -0,0 +1,181 @@
+package admin
+
+import (
+	"context"
+	"errors"
+	"net/http"
+
+	pb "github.com/bootjp/elastickv/proto"
+	pkgerrors "github.com/cockroachdb/errors"
+	"github.com/goccy/go-json"
+)
+
+// LeaderForwarder is the contract the admin HTTP handler invokes
+// when the local node is a follower (the source returned
+// ErrTablesNotLeader). Implementations dial the current leader
+// over the AdminForward gRPC service and return the leader's
+// response in a transport-neutral shape so the handler can re-emit
+// it verbatim.
+//
+// Defining this interface in the admin package — rather than wiring
+// pb.AdminForwardClient directly into the handler — keeps the
+// admin HTTP layer free of any proto-level coupling and lets tests
+// substitute a deterministic stub. The bridge in main_admin.go
+// provides the production implementation that uses
+// kv.GRPCConnCache + the raft engine's leader address.
+type LeaderForwarder interface {
+	// ForwardCreateTable issues a forwarded CreateTable on the
+	// leader's behalf. The response is the leader's structured
+	// AdminForwardResponse re-shaped into ForwardResult so the
+	// handler does not need to import proto.
+	ForwardCreateTable(ctx context.Context, principal AuthPrincipal, in CreateTableRequest) (*ForwardResult, error)
+	// ForwardDeleteTable is the delete-side counterpart.
+	ForwardDeleteTable(ctx context.Context, principal AuthPrincipal, name string) (*ForwardResult, error)
+}
+
+// ForwardResult is the leader's response replayed for the SPA. The
+// handler writes Payload verbatim with the given status code and
+// content type, so a forwarded request is indistinguishable from a
+// leader-direct call.
+type ForwardResult struct {
+	StatusCode  int
+	Payload     []byte
+	ContentType string
+}
+
+// ErrLeaderUnavailable is returned when the forwarder cannot reach
+// any leader — typically during a Raft election or a cluster split.
+// The handler maps it to 503 + Retry-After: 1 so the SPA / client
+// re-issues the request after a short delay (acceptance criterion 3).
+var ErrLeaderUnavailable = errors.New("admin: raft leader currently unavailable")
+
+// LeaderAddressResolver returns the current Raft leader's address
+// for the local node's group, or "" if no leader is known. The
+// production wiring uses raftengine.Engine.LeaderAddr / the
+// cluster's address map; tests inject a fixed string.
+type LeaderAddressResolver func() string
+
+// GRPCConnFactory is the small surface AdminForwardClient needs
+// from kv.GRPCConnCache. Pulling out an interface lets tests
+// substitute an in-memory dialer without spinning up a TCP
+// listener and lets the bridge use the existing connection cache
+// without copy-paste.
+type GRPCConnFactory interface {
+	// ConnFor returns a gRPC client connection to addr, reusing
+	// the cached entry if one exists. addr "" is a programming
+	// error and may panic; callers must check leader-empty before
+	// dialling.
+	ConnFor(addr string) (PBAdminForwardClient, error)
+}
+
+// PBAdminForwardClient narrows pb.AdminForwardClient to just the
+// methods this package uses. The narrower interface keeps the test
+// stub implementation small.
+type PBAdminForwardClient interface {
+	Forward(ctx context.Context, in *pb.AdminForwardRequest, opts ...grpcCallOption) (*pb.AdminForwardResponse, error)
+}
+
+// grpcCallOption is a re-export of google.golang.org/grpc.CallOption
+// kept private so callers do not import proto's grpc dep directly.
+type grpcCallOption = interface{}
+
+// gRPCForwardClient is the production LeaderForwarder. Construct
+// one with NewGRPCForwardClient. Two collaborators are required:
+//   - resolver: returns the current leader address, or "" if absent
+//   - dial:     turns an address into a PBAdminForwardClient (the
+//     bridge wraps kv.GRPCConnCache to satisfy this)
+//
+// nodeID is echoed into the leader's audit log via
+// AdminForwardRequest.forwarded_from (acceptance criterion 6).
+type gRPCForwardClient struct {
+	resolver LeaderAddressResolver
+	dial     GRPCConnFactory
+	nodeID   string
+}
+
+// NewGRPCForwardClient constructs the production LeaderForwarder.
+// All three parameters must be non-nil / non-empty; otherwise the
+// constructor returns nil and a wiring-error so a misconfigured
+// build refuses to start rather than producing 500s at runtime.
+func NewGRPCForwardClient(resolver LeaderAddressResolver, dial GRPCConnFactory, nodeID string) (LeaderForwarder, error) {
+	if resolver == nil {
+		return nil, errors.New("admin forwarder: leader address resolver is required")
+	}
+	if dial == nil {
+		return nil, errors.New("admin forwarder: gRPC connection factory is required")
+	}
+	if nodeID == "" {
+		return nil, errors.New("admin forwarder: node id is required for audit log enrichment")
+	}
+	return &gRPCForwardClient{resolver: resolver, dial: dial, nodeID: nodeID}, nil
+}
+
+// ForwardCreateTable serialises `in` as JSON and dispatches the
+// CreateTable operation to the leader. Returns ErrLeaderUnavailable
+// when no leader address is known; gRPC-level errors are wrapped
+// and surfaced unchanged so the handler can decide whether to
+// retry, log, or 500.
+func (c *gRPCForwardClient) ForwardCreateTable(ctx context.Context, principal AuthPrincipal, in CreateTableRequest) (*ForwardResult, error) {
+	payload, err := json.Marshal(in)
+	if err != nil {
+		// CreateTableRequest is plain string fields; Marshal
+		// cannot fail in practice. Surface the error rather than
+		// silently dropping the request.
+		return nil, pkgerrors.Wrap(err, "admin forward: marshal create-table request")
+	}
+	return c.forward(ctx, pb.AdminOperation_ADMIN_OP_CREATE_TABLE, principal, payload)
+}
+
+// ForwardDeleteTable serialises the table name as `{"name":"..."}`
+// to match the leader-side handleDelete contract.
+func (c *gRPCForwardClient) ForwardDeleteTable(ctx context.Context, principal AuthPrincipal, name string) (*ForwardResult, error) {
+	payload, err := json.Marshal(struct {
+		Name string `json:"name"`
+	}{Name: name})
+	if err != nil {
+		return nil, pkgerrors.Wrap(err, "admin forward: marshal delete-table request")
+	}
+	return c.forward(ctx, pb.AdminOperation_ADMIN_OP_DELETE_TABLE, principal, payload)
+}
+
+func (c *gRPCForwardClient) forward(ctx context.Context, op pb.AdminOperation, principal AuthPrincipal, payload []byte) (*ForwardResult, error) {
+	addr := c.resolver()
+	if addr == "" {
+		return nil, ErrLeaderUnavailable
+	}
+	cli, err := c.dial.ConnFor(addr)
+	if err != nil {
+		return nil, pkgerrors.Wrap(err, "admin forward: dial leader")
+	}
+	resp, err := cli.Forward(ctx, &pb.AdminForwardRequest{
+		Principal: &pb.AdminPrincipal{
+			AccessKey: principal.AccessKey,
+			Role:      string(principal.Role),
+		},
+		Operation:     op,
+		Payload:       payload,
+		ForwardedFrom: c.nodeID,
+	})
+	if err != nil {
+		return nil, pkgerrors.Wrap(err, "admin forward: rpc")
+	}
+	out := &ForwardResult{
+		StatusCode:  int(resp.GetStatusCode()),
+		Payload:     resp.GetPayload(),
+		ContentType: resp.GetContentType(),
+	}
+	if out.ContentType == "" {
+		// The leader server always sets a content type, but be
+		// defensive: a future change that drops it must not
+		// produce a SPA response with no Content-Type header.
+		out.ContentType = "application/json; charset=utf-8"
+	}
+	if out.StatusCode == 0 {
+		// status_code 0 is the proto's zero value, which the
+		// leader server never emits intentionally. Treat it as a
+		// transport bug rather than a 200, since 0 is not a valid
+		// HTTP status.
+		out.StatusCode = http.StatusBadGateway
+	}
+	return out, nil
+}

internal/admin/forward_client_test.go

@@ -0,0 +1,183 @@
+package admin
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"testing"
+
+	pb "github.com/bootjp/elastickv/proto"
+	"github.com/goccy/go-json"
+	"github.com/stretchr/testify/require"
+)
+
+// stubForwardConn is the in-memory PBAdminForwardClient the
+// follower-side tests use. It captures the last request and
+// returns whatever response/error the test prepared.
+type stubForwardConn struct {
+	resp    *pb.AdminForwardResponse
+	err     error
+	lastReq *pb.AdminForwardRequest
+}
+
+func (s *stubForwardConn) Forward(_ context.Context, in *pb.AdminForwardRequest, _ ...grpcCallOption) (*pb.AdminForwardResponse, error) {
+	s.lastReq = in
+	if s.err != nil {
+		return nil, s.err
+	}
+	return s.resp, nil
+}
+
+// stubConnFactory wraps stubForwardConn so the gRPCForwardClient
+// can resolve an address to a client. The captured `addr` lets
+// tests prove the resolver round-tripped correctly.
+type stubConnFactory struct {
+	conn     *stubForwardConn
+	dialErr  error
+	lastAddr string
+}
+
+func (s *stubConnFactory) ConnFor(addr string) (PBAdminForwardClient, error) {
+	s.lastAddr = addr
+	if s.dialErr != nil {
+		return nil, s.dialErr
+	}
+	return s.conn, nil
+}
+
+// fixedResolver returns a closure that always resolves to addr.
+// Pulling this into a helper keeps the per-test setup compact.
+func fixedResolver(addr string) LeaderAddressResolver {
+	return func() string { return addr }
+}
+
+func TestNewGRPCForwardClient_RejectsMissingDeps(t *testing.T) {
+	cases := []struct {
+		name     string
+		resolver LeaderAddressResolver
+		dial     GRPCConnFactory
+		nodeID   string
+		expect   string
+	}{
+		{"nil resolver", nil, &stubConnFactory{}, "n1", "leader address resolver"},
+		{"nil dial", fixedResolver(""), nil, "n1", "gRPC connection factory"},
+		{"empty node id", fixedResolver(""), &stubConnFactory{}, "", "node id is required"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			fwd, err := NewGRPCForwardClient(tc.resolver, tc.dial, tc.nodeID)
+			require.Error(t, err)
+			require.Nil(t, fwd)
+			require.Contains(t, err.Error(), tc.expect)
+		})
+	}
+}
+
+func TestGRPCForwardClient_ForwardCreateTable_HappyPath(t *testing.T) {
+	conn := &stubForwardConn{resp: &pb.AdminForwardResponse{
+		StatusCode:  http.StatusCreated,
+		Payload:     []byte(`{"name":"users"}`),
+		ContentType: "application/json; charset=utf-8",
+	}}
+	dial := &stubConnFactory{conn: conn}
+	fwd, err := NewGRPCForwardClient(fixedResolver("leader.local:7000"), dial, "follower-2")
+	require.NoError(t, err)
+
+	in := CreateTableRequest{TableName: "users", PartitionKey: CreateTableAttribute{Name: "id", Type: "S"}}
+	res, err := fwd.ForwardCreateTable(context.Background(), AuthPrincipal{AccessKey: "AKIA_F", Role: RoleFull}, in)
+	require.NoError(t, err)
+	require.NotNil(t, res)
+	require.Equal(t, http.StatusCreated, res.StatusCode)
+	require.JSONEq(t, `{"name":"users"}`, string(res.Payload))
+	require.Equal(t, "application/json; charset=utf-8", res.ContentType)
+
+	// The captured request must carry the principal, the canonical
+	// op enum, the JSON-encoded body, and the follower node id.
+	require.Equal(t, "leader.local:7000", dial.lastAddr)
+	require.NotNil(t, conn.lastReq)
+	require.Equal(t, pb.AdminOperation_ADMIN_OP_CREATE_TABLE, conn.lastReq.GetOperation())
+	require.Equal(t, "AKIA_F", conn.lastReq.GetPrincipal().GetAccessKey())
+	require.Equal(t, "full", conn.lastReq.GetPrincipal().GetRole())
+	require.Equal(t, "follower-2", conn.lastReq.GetForwardedFrom())
+	var roundtripped CreateTableRequest
+	require.NoError(t, json.Unmarshal(conn.lastReq.GetPayload(), &roundtripped))
+	require.Equal(t, in, roundtripped)
+}
+
+func TestGRPCForwardClient_ForwardDeleteTable_HappyPath(t *testing.T) {
+	conn := &stubForwardConn{resp: &pb.AdminForwardResponse{
+		StatusCode:  http.StatusNoContent,
+		ContentType: "application/json; charset=utf-8",
+	}}
+	dial := &stubConnFactory{conn: conn}
+	fwd, _ := NewGRPCForwardClient(fixedResolver("leader.local:7000"), dial, "follower-3")
+
+	res, err := fwd.ForwardDeleteTable(context.Background(), AuthPrincipal{AccessKey: "AKIA_F", Role: RoleFull}, "users")
+	require.NoError(t, err)
+	require.Equal(t, http.StatusNoContent, res.StatusCode)
+	require.Empty(t, res.Payload)
+	require.Equal(t, pb.AdminOperation_ADMIN_OP_DELETE_TABLE, conn.lastReq.GetOperation())
+	require.JSONEq(t, `{"name":"users"}`, string(conn.lastReq.GetPayload()))
+	require.Equal(t, "follower-3", conn.lastReq.GetForwardedFrom())
+}
+
+func TestGRPCForwardClient_NoLeaderReturnsErrLeaderUnavailable(t *testing.T) {
+	dial := &stubConnFactory{conn: &stubForwardConn{}}
+	fwd, _ := NewGRPCForwardClient(fixedResolver(""), dial, "f")
+
+	_, err := fwd.ForwardCreateTable(context.Background(), AuthPrincipal{Role: RoleFull},
+		CreateTableRequest{TableName: "t", PartitionKey: CreateTableAttribute{Name: "id", Type: "S"}})
+	require.ErrorIs(t, err, ErrLeaderUnavailable)
+	// Connection must not have been dialled when no leader is known.
+	require.Empty(t, dial.lastAddr)
+}
+
+func TestGRPCForwardClient_DialErrorPropagated(t *testing.T) {
+	dial := &stubConnFactory{dialErr: errors.New("network unreachable")}
+	fwd, _ := NewGRPCForwardClient(fixedResolver("leader:1"), dial, "f")
+
+	_, err := fwd.ForwardCreateTable(context.Background(), AuthPrincipal{Role: RoleFull},
+		CreateTableRequest{TableName: "t", PartitionKey: CreateTableAttribute{Name: "id", Type: "S"}})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "network unreachable")
+}
+
+func TestGRPCForwardClient_GRPCErrorPropagated(t *testing.T) {
+	conn := &stubForwardConn{err: errors.New("rpc deadline exceeded")}
+	dial := &stubConnFactory{conn: conn}
+	fwd, _ := NewGRPCForwardClient(fixedResolver("leader:1"), dial, "f")
+
+	_, err := fwd.ForwardDeleteTable(context.Background(), AuthPrincipal{Role: RoleFull}, "x")
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "rpc deadline exceeded")
+}
+
+func TestGRPCForwardClient_ZeroStatusUpgradesTo502(t *testing.T) {
+	// status_code 0 is the proto zero value — never emitted
+	// intentionally by the leader. The forwarder must not treat
+	// that as 200; mapping it to 502 surfaces the transport bug.
+	conn := &stubForwardConn{resp: &pb.AdminForwardResponse{StatusCode: 0}}
+	dial := &stubConnFactory{conn: conn}
+	fwd, _ := NewGRPCForwardClient(fixedResolver("leader:1"), dial, "f")
+
+	res, err := fwd.ForwardCreateTable(context.Background(), AuthPrincipal{Role: RoleFull},
+		CreateTableRequest{TableName: "t", PartitionKey: CreateTableAttribute{Name: "id", Type: "S"}})
+	require.NoError(t, err)
+	require.Equal(t, http.StatusBadGateway, res.StatusCode)
+}
+
+func TestGRPCForwardClient_FillsMissingContentType(t *testing.T) {
+	// A future leader-server change that omits ContentType must
+	// still produce a SPA-readable response. The forwarder fills
+	// in the default JSON content type defensively.
+	conn := &stubForwardConn{resp: &pb.AdminForwardResponse{
+		StatusCode: http.StatusCreated,
+		Payload:    []byte(`{"name":"u"}`),
+	}}
+	dial := &stubConnFactory{conn: conn}
+	fwd, _ := NewGRPCForwardClient(fixedResolver("leader:1"), dial, "f")
+
+	res, _ := fwd.ForwardCreateTable(context.Background(), AuthPrincipal{Role: RoleFull},
+		CreateTableRequest{TableName: "t", PartitionKey: CreateTableAttribute{Name: "id", Type: "S"}})
+	require.Equal(t, "application/json; charset=utf-8", res.ContentType)
+}