admin: AdminForward integration for S3 admin ops (P2 slice 2b)

Slice 2b of P2 (docs/design/2026_04_24_proposed_admin_dashboard.md
Section 3.3.2 + 4.1): a follower-side S3 admin write
(POST /buckets, PUT /buckets/{name}/acl, DELETE /buckets/{name})
now hands off to the leader transparently, completing the same
end-to-end forwarding contract Dynamo writes received in #644 +

Stacked on #669 (P2 slice 2a). Once #669 merges, this rebases
cleanly onto main.

Proto (proto/admin_forward.proto):
- Three new ADMIN_OP enum values (CREATE_BUCKET / DELETE_BUCKET /
  PUT_BUCKET_ACL) appended after the Dynamo block so existing
  wire-format integers stay stable. Regenerated with the pinned
  protoc 29.3 / protoc-gen-go 1.36.11 / protoc-gen-go-grpc 1.6.1.

Leader-side ForwardServer (internal/admin/forward_server.go):
- WithBucketsSource lets deployments wire the S3 dispatcher
  optionally — Dynamo-only builds keep the receiver nil and the
  three new operations return 501 NotImplemented.
- Three new dispatch arms: handleCreateBucket / handleDeleteBucket /
  handlePutBucketAcl. Each one mirrors the leader-direct HTTP
  path's payload contract (NUL-byte rejection, 64 KiB limit,
  DisallowUnknownFields, trailing-token rejection, slash-in-name
  rejection) so a hostile follower cannot smuggle a payload past
  validations the leader-direct path enforces.
- forwardBucketsErrorResponse mirrors forwardErrorResponse on the
  Dynamo side: ErrBucketsForbidden / NotLeader / NotFound /
  AlreadyExists / NotEmpty + ValidationError each map to the same
  HTTP status the leader-direct writeBucketsError produces, so
  forwarded and leader-direct responses are byte-for-byte
  indistinguishable from the SPA's view.
- isStructuredSourceError extended to recognise the bucket
  sentinels so they are NOT logged at LevelError on the leader.
- Forward's switch was extracted into dispatchForward to keep the
  parent function under cyclop's 10-branch ceiling as the operation
  enum grew.

Follower-side LeaderForwarder (internal/admin/forward_client.go):
- Interface gains ForwardCreateBucket / ForwardDeleteBucket /
  ForwardPutBucketAcl. PutBucketAcl carries both the bucket name
  (from the URL path) and the new ACL (from the request body) in
  one JSON payload — same approach handleDeleteBucket takes for
  the bucket name.
- gRPCForwardClient methods reuse the existing forward() helper
  for transport, so connection-cache reuse and ErrLeaderUnavailable
  signalling behave identically across resource types.

Handler integration (internal/admin/s3_handler.go):
- New `forwarder LeaderForwarder` field + WithLeaderForwarder method.
- handleCreate / handlePutACL / handleDelete now consult tryForward*
  helpers when the source returned ErrBucketsNotLeader; the helpers
  are gated on `errors.Is(err, ErrBucketsNotLeader) && forwarder != nil`
  so a leader-direct rejection (already-exists, not-found, etc.) is
  never re-applied at the leader.
- writeForwardResult / writeForwardFailure mirror the Dynamo handler's
  pattern: nosniff + Cache-Control:no-store + Retry-After:1 on 503.
  ErrLeaderUnavailable does NOT log at LevelError (elections are
  routine); transport errors do log so operators can investigate.

Wiring (main.go + main_admin_forward.go):
- adminForwardServerDeps gains a `buckets` field; readyForRegistration
  still requires only TablesSource + RoleStore (so cluster-only or
  Dynamo-only builds keep registering Dynamo forwarding without S3).
- runtimeServerRunner.start() now creates *adapter.S3Server BEFORE
  startRaftServers (in addition to dynamoServer) so the leader-side
  ForwardServer registration sees both adapters. The reorder is safe:
  each adapter listens on its own address and the raft TCP listeners
  are independent.
- ServerDeps.Forwarder now plumbs through buildS3HandlerForDeps too,
  so the follower's S3Handler picks up the same LeaderForwarder
  instance the Dynamo handler does.

Tests:
- 9 forward-server tests covering the three new bucket operations:
  happy path / no-BucketsSource→501 / bad-JSON 400 / already-exists
  409 / not-empty 409 / slash-in-name 400 / missing-acl 400 /
  payload-too-large 413 (sweep over all three ops).
- 4 forward-client tests covering ForwardCreateBucket /
  ForwardDeleteBucket / ForwardPutBucketAcl happy-path payload
  shapes + ErrLeaderUnavailable on no-leader.
- stubLeaderForwarder gains 3 bucket-side forward methods so
  existing dynamo tests still satisfy the LeaderForwarder
  interface, and the new stub fields let bucket-handler tests
  verify the forward arguments.
- 6 handler integration tests on S3Handler.tryForward*: forwarded
  create / delete / put-acl happy paths (replay leader's status
  + payload + content-type), forwarder ErrLeaderUnavailable → 503
  + Retry-After, transport-error → 503 + no leakage, and a 3-axis
  gate sweep proving the forwarder is NOT invoked on
  AlreadyExists / Forbidden / generic source errors.

Closes design 3.3.2 acceptance criteria 2 (transparent forwarding)
+ 6 (forwarded_from in audit log) for S3 admin writes; criterion 3
(election-period 503 + retry) is also live for S3 because the
existing tryForward helpers reuse the same fallback paths.

Author: bootjp

internal/admin/forward_client.go

@@ -32,6 +32,17 @@ type LeaderForwarder interface {
 	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)
+	// ForwardCreateBucket issues a forwarded POST /s3/buckets on
+	// the leader's behalf. The leader echoes back the same JSON
+	// envelope a leader-direct call would produce.
+	ForwardCreateBucket(ctx context.Context, principal AuthPrincipal, in CreateBucketRequest) (*ForwardResult, error)
+	// ForwardDeleteBucket is the delete-side counterpart.
+	ForwardDeleteBucket(ctx context.Context, principal AuthPrincipal, name string) (*ForwardResult, error)
+	// ForwardPutBucketAcl issues a forwarded PUT
+	// /s3/buckets/{name}/acl. Both the bucket name and the
+	// new ACL travel inside the proto payload — see the leader-side
+	// handlePutBucketAcl for the JSON shape.
+	ForwardPutBucketAcl(ctx context.Context, principal AuthPrincipal, name, acl string) (*ForwardResult, error)
 }
 
 // ForwardResult is the leader's response replayed for the SPA. The
@@ -142,6 +153,43 @@ func (c *gRPCForwardClient) ForwardDeleteTable(ctx context.Context, principal Au
 	return c.forward(ctx, pb.AdminOperation_ADMIN_OP_DELETE_TABLE, principal, payload)
 }
 
+// ForwardCreateBucket serialises `in` as JSON and dispatches the
+// CreateBucket operation. Same contract as ForwardCreateTable —
+// gRPC errors are wrapped, ErrLeaderUnavailable on no-leader.
+func (c *gRPCForwardClient) ForwardCreateBucket(ctx context.Context, principal AuthPrincipal, in CreateBucketRequest) (*ForwardResult, error) {
+	payload, err := json.Marshal(in)
+	if err != nil {
+		return nil, pkgerrors.Wrap(err, "admin forward: marshal create-bucket request")
+	}
+	return c.forward(ctx, pb.AdminOperation_ADMIN_OP_CREATE_BUCKET, principal, payload)
+}
+
+// ForwardDeleteBucket serialises the bucket name as `{"name":"..."}`
+// to match the leader-side handleDeleteBucket contract.
+func (c *gRPCForwardClient) ForwardDeleteBucket(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-bucket request")
+	}
+	return c.forward(ctx, pb.AdminOperation_ADMIN_OP_DELETE_BUCKET, principal, payload)
+}
+
+// ForwardPutBucketAcl carries both the bucket name (from the URL
+// path) and the new ACL (from the request body) in the proto
+// payload — see handlePutBucketAcl for the leader-side decode.
+func (c *gRPCForwardClient) ForwardPutBucketAcl(ctx context.Context, principal AuthPrincipal, name, acl string) (*ForwardResult, error) {
+	payload, err := json.Marshal(struct {
+		Name string `json:"name"`
+		ACL  string `json:"acl"`
+	}{Name: name, ACL: acl})
+	if err != nil {
+		return nil, pkgerrors.Wrap(err, "admin forward: marshal put-bucket-acl request")
+	}
+	return c.forward(ctx, pb.AdminOperation_ADMIN_OP_PUT_BUCKET_ACL, principal, payload)
+}
+
 func (c *gRPCForwardClient) forward(ctx context.Context, op pb.AdminOperation, principal AuthPrincipal, payload []byte) (*ForwardResult, error) {
 	addr := c.resolver()
 	if addr == "" {

internal/admin/forward_client_test.go

@@ -182,3 +182,72 @@ func TestGRPCForwardClient_FillsMissingContentType(t *testing.T) {
 		CreateTableRequest{TableName: "t", PartitionKey: CreateTableAttribute{Name: "id", Type: "S"}})
 	require.Equal(t, "application/json; charset=utf-8", res.ContentType)
 }
+
+func TestGRPCForwardClient_ForwardCreateBucket_HappyPath(t *testing.T) {
+	conn := &stubForwardConn{resp: &pb.AdminForwardResponse{
+		StatusCode:  http.StatusCreated,
+		Payload:     []byte(`{"bucket_name":"public-assets","acl":"public-read"}`),
+		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 := CreateBucketRequest{BucketName: "public-assets", ACL: "public-read"}
+	res, err := fwd.ForwardCreateBucket(context.Background(),
+		AuthPrincipal{AccessKey: "AKIA_F", Role: RoleFull}, in)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusCreated, res.StatusCode)
+	require.Equal(t, pb.AdminOperation_ADMIN_OP_CREATE_BUCKET, conn.lastReq.GetOperation())
+	require.Equal(t, "follower-2", conn.lastReq.GetForwardedFrom())
+	var roundtripped CreateBucketRequest
+	require.NoError(t, json.Unmarshal(conn.lastReq.GetPayload(), &roundtripped))
+	require.Equal(t, in, roundtripped)
+}
+
+func TestGRPCForwardClient_ForwardDeleteBucket_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.ForwardDeleteBucket(context.Background(),
+		AuthPrincipal{AccessKey: "AKIA_F", Role: RoleFull}, "orders")
+	require.NoError(t, err)
+	require.Equal(t, http.StatusNoContent, res.StatusCode)
+	require.Equal(t, pb.AdminOperation_ADMIN_OP_DELETE_BUCKET, conn.lastReq.GetOperation())
+	require.JSONEq(t, `{"name":"orders"}`, string(conn.lastReq.GetPayload()))
+}
+
+func TestGRPCForwardClient_ForwardPutBucketAcl_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-4")
+
+	res, err := fwd.ForwardPutBucketAcl(context.Background(),
+		AuthPrincipal{AccessKey: "AKIA_F", Role: RoleFull}, "orders", "public-read")
+	require.NoError(t, err)
+	require.Equal(t, http.StatusNoContent, res.StatusCode)
+	require.Equal(t, pb.AdminOperation_ADMIN_OP_PUT_BUCKET_ACL, conn.lastReq.GetOperation())
+	// Both name and acl travel in the payload — see handlePutBucketAcl
+	// for the leader-side decode contract.
+	require.JSONEq(t, `{"name":"orders","acl":"public-read"}`, string(conn.lastReq.GetPayload()))
+}
+
+func TestGRPCForwardClient_BucketOps_NoLeaderReturnsErrLeaderUnavailable(t *testing.T) {
+	dial := &stubConnFactory{conn: &stubForwardConn{}}
+	fwd, _ := NewGRPCForwardClient(fixedResolver(""), dial, "f")
+
+	_, err := fwd.ForwardCreateBucket(context.Background(), AuthPrincipal{Role: RoleFull},
+		CreateBucketRequest{BucketName: "x"})
+	require.ErrorIs(t, err, ErrLeaderUnavailable)
+	_, err = fwd.ForwardDeleteBucket(context.Background(), AuthPrincipal{Role: RoleFull}, "x")
+	require.ErrorIs(t, err, ErrLeaderUnavailable)
+	_, err = fwd.ForwardPutBucketAcl(context.Background(), AuthPrincipal{Role: RoleFull}, "x", "private")
+	require.ErrorIs(t, err, ErrLeaderUnavailable)
+}

internal/admin/forward_integration_test.go

@@ -20,11 +20,28 @@ type stubLeaderForwarder struct {
 	createErr error
 	deleteRes *ForwardResult
 	deleteErr error
-
-	lastCreatePrincipal AuthPrincipal
-	lastCreateInput     CreateTableRequest
-	lastDeletePrincipal AuthPrincipal
-	lastDeleteName      string
+	// Bucket-side counterparts ship with P2 slice 2b. Tests that
+	// only exercise Dynamo behaviour leave these zero values; the
+	// stub honours the same nil-result-and-no-error contract the
+	// table side uses on a "did not call us" path.
+	createBucketRes *ForwardResult
+	createBucketErr error
+	deleteBucketRes *ForwardResult
+	deleteBucketErr error
+	putACLRes       *ForwardResult
+	putACLErr       error
+
+	lastCreatePrincipal       AuthPrincipal
+	lastCreateInput           CreateTableRequest
+	lastDeletePrincipal       AuthPrincipal
+	lastDeleteName            string
+	lastCreateBucketPrincipal AuthPrincipal
+	lastCreateBucketInput     CreateBucketRequest
+	lastDeleteBucketPrincipal AuthPrincipal
+	lastDeleteBucketName      string
+	lastPutACLPrincipal       AuthPrincipal
+	lastPutACLBucket          string
+	lastPutACLValue           string
 }
 
 func (s *stubLeaderForwarder) ForwardCreateTable(_ context.Context, principal AuthPrincipal, in CreateTableRequest) (*ForwardResult, error) {
@@ -45,6 +62,34 @@ func (s *stubLeaderForwarder) ForwardDeleteTable(_ context.Context, principal Au
 	return s.deleteRes, nil
 }
 
+func (s *stubLeaderForwarder) ForwardCreateBucket(_ context.Context, principal AuthPrincipal, in CreateBucketRequest) (*ForwardResult, error) {
+	s.lastCreateBucketPrincipal = principal
+	s.lastCreateBucketInput = in
+	if s.createBucketErr != nil {
+		return nil, s.createBucketErr
+	}
+	return s.createBucketRes, nil
+}
+
+func (s *stubLeaderForwarder) ForwardDeleteBucket(_ context.Context, principal AuthPrincipal, name string) (*ForwardResult, error) {
+	s.lastDeleteBucketPrincipal = principal
+	s.lastDeleteBucketName = name
+	if s.deleteBucketErr != nil {
+		return nil, s.deleteBucketErr
+	}
+	return s.deleteBucketRes, nil
+}
+
+func (s *stubLeaderForwarder) ForwardPutBucketAcl(_ context.Context, principal AuthPrincipal, name, acl string) (*ForwardResult, error) {
+	s.lastPutACLPrincipal = principal
+	s.lastPutACLBucket = name
+	s.lastPutACLValue = acl
+	if s.putACLErr != nil {
+		return nil, s.putACLErr
+	}
+	return s.putACLRes, nil
+}
+
 // notLeaderSource is a TablesSource that always returns
 // ErrTablesNotLeader on writes — i.e., it simulates the local
 // node being a follower. Used to exercise the forwarder path