admin: register AdminForward for S3-only deployments (Codex P1)

Codex on PR #673 caught that `readyForRegistration()` predicate
gated registration on `tables != nil`, which meant a cluster
started with `--dynamoAddr` empty but S3 enabled never registered
the AdminForward gRPC service at all. Followers attempting to
forward S3 admin writes (CREATE_BUCKET / DELETE_BUCKET /
PUT_BUCKET_ACL) hit gRPC `Unimplemented` and surfaced as 503 to
the SPA, instead of transparently reaching the leader. The S3
write path was advertised as "follower-forwarded" by P2 slice 2b
but actually only worked when Dynamo was also configured.

Fix is symmetric: relax the predicate to require RoleStore plus
**at least one** source (TablesSource or BucketsSource) and have
the dispatcher reject ops for the missing surface with the same
501 it already returns when the inverse source is nil.

Per CLAUDE.md "test the bug first" convention, add the
regression test before the fix:

  TestForwardServer_DynamoOps_NoTablesSourceReturns501
    — symmetric with the existing
    TestForwardServer_BucketOps_NoBucketsSourceReturns501.
    With nil source, CREATE_TABLE used to slip past the source
    nil-deref because validation rejected the payload first;
    DELETE_TABLE panicked on `s.source.AdminDeleteTable`. Both
    now return 501.

Plus three readyForRegistration assertions:
  - buckets-only deployment registers (the regression case)
  - missing-roles + buckets-only does NOT register
  - tables-only / both-sources continue to register

While in there, the per-op nil-source guards I added tipped the
linter's dupl detector. Refactored the dispatch path to keep
forward_server tidy:

1. **`checkOpAvailability`**: pulls the per-op nil-source switch
   out of `dispatchForward` so the latter stays under the
   cyclomatic ceiling. One place to update when a future op
   joins the enum, instead of five.

2. **`decodeNamedPayload`**: the {"name": "..."} JSON shape is
   shared between Dynamo handleDelete and S3 handleDeleteBucket
   (NUL guard, DisallowUnknownFields, dec.More(), empty-name
   reject, slash reject — every guard had a paired comment
   citing the original review that introduced it). Centralising
   the helper keeps that audit trail in one place; the opLabel
   parameter ("delete" / "delete-bucket") preserves the
   path-specific wording in error responses.

3. **`auditDeleteSuccess`**: handleDelete and handleDeleteBucket
   each emitted a 7-line `admin_audit` block that differed only
   in the operation label and the `table` vs `bucket` field
   name. Folded into one helper so a future delete-shaped op
   automatically picks up the same field set.

4. **`notImplementedForwardResponse`**: deduplicates the 501
   construction across the five Dynamo + S3 op cases.

No behaviour change to any of the existing successful paths;
all forward_server tests pass unchanged. Tests added: the
S3-only registration assertion + the new symmetric
DynamoOps_NoTablesSourceReturns501 sweep.

go test -race ./internal/admin/ . — passes.
golangci-lint run ./internal/admin/... . — 0 issues.

Author: bootjp

internal/admin/forward_server.go

@@ -107,8 +107,20 @@ func (s *ForwardServer) Forward(ctx context.Context, req *pb.AdminForwardRequest
 // handler. Pulled out so Forward stays under the cyclomatic ceiling
 // as the operation enum grows; the principal-validation +
 // forwarded_from sanitisation logic stays in Forward where it belongs.
+//
+// Source-availability checks live in checkOpAvailability rather than
+// in each handler: a Dynamo-only build has s.source != nil but
+// s.buckets == nil, and an S3-only build (Codex P1 on PR #673) has
+// the inverse. Centralising the check means every operation gets a
+// consistent 501 error shape and a future op cannot ship without the
+// operator-visible "not configured" message that the existing ops
+// promise.
 func (s *ForwardServer) dispatchForward(ctx context.Context, principal AuthPrincipal, forwardedFrom string, req *pb.AdminForwardRequest) (*pb.AdminForwardResponse, error) {
-	switch req.GetOperation() {
+	op := req.GetOperation()
+	if resp, err, ok := s.checkOpAvailability(op); !ok {
+		return resp, err
+	}
+	switch op {
 	case pb.AdminOperation_ADMIN_OP_CREATE_TABLE:
 		return s.handleCreate(ctx, principal, forwardedFrom, req)
 	case pb.AdminOperation_ADMIN_OP_DELETE_TABLE:
@@ -126,6 +138,43 @@ func (s *ForwardServer) dispatchForward(ctx context.Context, principal AuthPrinc
 	}
 }
 
+// checkOpAvailability returns (resp, err, true) when dispatchForward
+// should continue to the per-op handler, or (resp, err, false) when
+// the leader's build does not include the source the requested
+// operation needs (S3-only deployment served a Dynamo op, or vice
+// versa). Pulling the per-op switch out keeps dispatchForward's
+// cyclomatic count under the linter ceiling as the enum grows.
+func (s *ForwardServer) checkOpAvailability(op pb.AdminOperation) (*pb.AdminForwardResponse, error, bool) {
+	switch op {
+	case pb.AdminOperation_ADMIN_OP_CREATE_TABLE, pb.AdminOperation_ADMIN_OP_DELETE_TABLE:
+		if s.source == nil {
+			resp, err := notImplementedForwardResponse("DynamoDB")
+			return resp, err, false
+		}
+	case pb.AdminOperation_ADMIN_OP_CREATE_BUCKET,
+		pb.AdminOperation_ADMIN_OP_DELETE_BUCKET,
+		pb.AdminOperation_ADMIN_OP_PUT_BUCKET_ACL:
+		if s.buckets == nil {
+			resp, err := notImplementedForwardResponse("S3")
+			return resp, err, false
+		}
+	case pb.AdminOperation_ADMIN_OP_UNSPECIFIED:
+		// Unknown-op rejection is dispatchForward's responsibility,
+		// not this gate's. Falling through to ok=true lets the main
+		// switch's default branch produce the canonical 400 message.
+	}
+	return nil, nil, true
+}
+
+// notImplementedForwardResponse produces the 501 response a follower
+// receives when the leader is built without the source for this
+// operation's surface. surface is the human-facing label that ends
+// up in the error message ("DynamoDB" or "S3").
+func notImplementedForwardResponse(surface string) (*pb.AdminForwardResponse, error) {
+	return rejectForward(http.StatusNotImplemented, "not_implemented",
+		surface+" admin forwarding is not configured on this leader")
+}
+
 func (s *ForwardServer) validatePrincipal(p *pb.AdminPrincipal) (AuthPrincipal, bool) {
 	accessKey := p.GetAccessKey()
 	if accessKey == "" {
@@ -182,65 +231,24 @@ func (s *ForwardServer) handleDelete(ctx context.Context, principal AuthPrincipa
 	// proto stays operation-agnostic — there is no operation-specific
 	// field in AdminForwardRequest, by design (adding one per op
 	// would couple every new admin endpoint to the proto schema).
-	payload := req.GetPayload()
-	if len(payload) > adminForwardPayloadLimit {
-		return rejectForward(http.StatusRequestEntityTooLarge, "payload_too_large",
-			"forwarded payload exceeds the 64 KiB admin limit")
-	}
-	// Mirror decodeCreateTableRequest's NUL-byte guard: goccy/go-json
-	// treats raw NUL as end-of-input so dec.More() would otherwise
-	// miss `{"name":"users"}\x00{"extra":1}` payloads. Codex P2 on
-	// PR #635 flagged this as the same smuggling vector that the
-	// HTTP create path already covers.
-	if bytes.IndexByte(payload, 0) >= 0 {
-		return rejectForward(http.StatusBadRequest, "invalid_body", "delete payload contains a NUL byte")
-	}
-	dec := json.NewDecoder(bytes.NewReader(payload))
-	dec.DisallowUnknownFields()
-	var body struct {
-		Name string `json:"name"`
-	}
-	if err := dec.Decode(&body); err != nil {
-		return rejectForward(http.StatusBadRequest, "invalid_body", "delete payload is not valid JSON")
-	}
-	if dec.More() {
-		return rejectForward(http.StatusBadRequest, "invalid_body", "delete payload has trailing data")
+	name, rejection, err := decodeNamedPayload(req.GetPayload(), "delete")
+	if rejection != nil || err != nil {
+		return rejection, err
 	}
-	if body.Name == "" {
-		return rejectForward(http.StatusBadRequest, "invalid_body", "delete payload missing name")
-	}
-	// Reject slash-bearing names symmetrically with the HTTP
-	// handleDelete and handleDescribe paths. Without this, a
-	// forwarded call could act on `foo/bar` while a leader-direct
-	// call would 404 — divergent behaviour Codex P2 flagged on
-	// PR #635.
-	if strings.ContainsRune(body.Name, '/') {
-		return rejectForward(http.StatusBadRequest, "invalid_body", "delete payload name must not contain '/'")
-	}
-	if err := s.source.AdminDeleteTable(ctx, principal, body.Name); err != nil {
-		s.logUnexpectedSourceError(ctx, "delete_table", body.Name, forwardedFrom, err)
+	if err := s.source.AdminDeleteTable(ctx, principal, name); err != nil {
+		s.logUnexpectedSourceError(ctx, "delete_table", name, forwardedFrom, err)
 		return forwardErrorResponse("delete", err), nil
 	}
-	s.logger.LogAttrs(ctx, slog.LevelInfo, "admin_audit",
-		slog.String("actor", principal.AccessKey),
-		slog.String("role", string(principal.Role)),
-		slog.String("forwarded_from", forwardedFrom),
-		slog.String("operation", "delete_table"),
-		slog.String("table", body.Name),
-	)
+	s.auditDeleteSuccess(ctx, principal, forwardedFrom, "delete_table", "table", name)
 	return &pb.AdminForwardResponse{StatusCode: http.StatusNoContent}, nil
 }
 
 // handleCreateBucket dispatches a forwarded POST /s3/buckets call.
 // Mirrors handleCreate (Dynamo) but decodes a CreateBucketRequest
-// and routes through BucketsSource. Returns 501 NotImplemented when
-// no BucketsSource is wired so a follower whose S3 surface was
-// disabled gets a deterministic error instead of a generic 500.
+// and routes through BucketsSource. dispatchForward gates this on
+// s.buckets != nil so callers reach here only when S3 forwarding is
+// configured.
 func (s *ForwardServer) handleCreateBucket(ctx context.Context, principal AuthPrincipal, forwardedFrom string, req *pb.AdminForwardRequest) (*pb.AdminForwardResponse, error) {
-	if s.buckets == nil {
-		return rejectForward(http.StatusNotImplemented, "not_implemented",
-			"S3 admin forwarding is not configured on this leader")
-	}
 	payload := req.GetPayload()
 	if len(payload) > adminForwardPayloadLimit {
 		return rejectForward(http.StatusRequestEntityTooLarge, "payload_too_large",
@@ -291,52 +299,33 @@ func (s *ForwardServer) handleCreateBucket(ctx context.Context, principal AuthPr
 // a JSON object with a single "name" field, which the bridge
 // generates from the URL path.
 func (s *ForwardServer) handleDeleteBucket(ctx context.Context, principal AuthPrincipal, forwardedFrom string, req *pb.AdminForwardRequest) (*pb.AdminForwardResponse, error) {
-	if s.buckets == nil {
-		return rejectForward(http.StatusNotImplemented, "not_implemented",
-			"S3 admin forwarding is not configured on this leader")
+	name, rejection, err := decodeNamedPayload(req.GetPayload(), "delete-bucket")
+	if rejection != nil || err != nil {
+		return rejection, err
 	}
-	payload := req.GetPayload()
-	if len(payload) > adminForwardPayloadLimit {
-		return rejectForward(http.StatusRequestEntityTooLarge, "payload_too_large",
-			"forwarded payload exceeds the 64 KiB admin limit")
-	}
-	if bytes.IndexByte(payload, 0) >= 0 {
-		return rejectForward(http.StatusBadRequest, "invalid_body",
-			"delete-bucket payload contains a NUL byte")
-	}
-	dec := json.NewDecoder(bytes.NewReader(payload))
-	dec.DisallowUnknownFields()
-	var body struct {
-		Name string `json:"name"`
-	}
-	if err := dec.Decode(&body); err != nil {
-		return rejectForward(http.StatusBadRequest, "invalid_body",
-			"delete-bucket payload is not valid JSON")
-	}
-	if dec.More() {
-		return rejectForward(http.StatusBadRequest, "invalid_body",
-			"delete-bucket payload has trailing data")
-	}
-	if body.Name == "" {
-		return rejectForward(http.StatusBadRequest, "invalid_body",
-			"delete-bucket payload missing name")
-	}
-	if strings.ContainsRune(body.Name, '/') {
-		return rejectForward(http.StatusBadRequest, "invalid_body",
-			"delete-bucket payload name must not contain '/'")
-	}
-	if err := s.buckets.AdminDeleteBucket(ctx, principal, body.Name); err != nil {
-		s.logUnexpectedSourceError(ctx, "delete_bucket", body.Name, forwardedFrom, err)
+	if err := s.buckets.AdminDeleteBucket(ctx, principal, name); err != nil {
+		s.logUnexpectedSourceError(ctx, "delete_bucket", name, forwardedFrom, err)
 		return forwardBucketsErrorResponse("delete", err), nil
 	}
+	s.auditDeleteSuccess(ctx, principal, forwardedFrom, "delete_bucket", "bucket", name)
+	return &pb.AdminForwardResponse{StatusCode: http.StatusNoContent}, nil
+}
+
+// auditDeleteSuccess emits the admin_audit slog line both Dynamo and
+// S3 forwarded delete handlers need. Centralised so handleDelete and
+// handleDeleteBucket do not diverge on the field set, and so a future
+// handler that mirrors the same delete shape (e.g. delete-namespace)
+// keeps the audit-log contract by reusing this helper rather than
+// re-emitting a hand-rolled subset. resourceField is "table" or
+// "bucket"; opLabel is the audit "operation" value.
+func (s *ForwardServer) auditDeleteSuccess(ctx context.Context, principal AuthPrincipal, forwardedFrom, opLabel, resourceField, name string) {
 	s.logger.LogAttrs(ctx, slog.LevelInfo, "admin_audit",
 		slog.String("actor", principal.AccessKey),
 		slog.String("role", string(principal.Role)),
 		slog.String("forwarded_from", forwardedFrom),
-		slog.String("operation", "delete_bucket"),
-		slog.String("bucket", body.Name),
+		slog.String("operation", opLabel),
+		slog.String(resourceField, name),
 	)
-	return &pb.AdminForwardResponse{StatusCode: http.StatusNoContent}, nil
 }
 
 // handlePutBucketAcl dispatches a forwarded PUT
@@ -345,10 +334,6 @@ func (s *ForwardServer) handleDeleteBucket(ctx context.Context, principal AuthPr
 // operation-agnostic — same approach handleDeleteBucket takes for
 // the bucket name.
 func (s *ForwardServer) handlePutBucketAcl(ctx context.Context, principal AuthPrincipal, forwardedFrom string, req *pb.AdminForwardRequest) (*pb.AdminForwardResponse, error) {
-	if s.buckets == nil {
-		return rejectForward(http.StatusNotImplemented, "not_implemented",
-			"S3 admin forwarding is not configured on this leader")
-	}
 	payload := req.GetPayload()
 	if len(payload) > adminForwardPayloadLimit {
 		return rejectForward(http.StatusRequestEntityTooLarge, "payload_too_large",
@@ -458,6 +443,64 @@ func sanitiseForwardedFrom(s string) string {
 	}, s)
 }
 
+// decodeNamedPayload validates and decodes the {"name": "..."} JSON
+// shape both the Dynamo and S3 delete forwarders accept. Returns the
+// decoded name on success, or a populated rejection response (and
+// nil name) on a 400 / 413. opLabel is the human-facing prefix that
+// goes into the rejection messages ("delete" for Dynamo,
+// "delete-bucket" for S3) so the response identifies which path
+// rejected — and so a future op (e.g. "describe-bucket") that
+// reuses this helper still produces an actionable error.
+//
+// All four guards mirror the leader-direct HTTP path:
+//   - 64 KiB payload cap (matches adminForwardPayloadLimit elsewhere)
+//   - NUL-byte rejection (goccy/go-json treats raw NUL as end-of-
+//     input; without this guard `{"name":"x"}\x00{"extra":1}`
+//     payloads slip past dec.More(); Codex P2 on PR #635)
+//   - DisallowUnknownFields + dec.More() trailing-token rejection
+//   - empty + slash-bearing name rejection (the HTTP handlers
+//     already 404 slash-bearing names; the forwarded path has to
+//     reject symmetrically or a hostile follower could act on
+//     `foo/bar` while a leader-direct call would not).
+func decodeNamedPayload(payload []byte, opLabel string) (string, *pb.AdminForwardResponse, error) {
+	if len(payload) > adminForwardPayloadLimit {
+		resp, err := rejectForward(http.StatusRequestEntityTooLarge, "payload_too_large",
+			"forwarded payload exceeds the 64 KiB admin limit")
+		return "", resp, err
+	}
+	if bytes.IndexByte(payload, 0) >= 0 {
+		resp, err := rejectForward(http.StatusBadRequest, "invalid_body",
+			opLabel+" payload contains a NUL byte")
+		return "", resp, err
+	}
+	dec := json.NewDecoder(bytes.NewReader(payload))
+	dec.DisallowUnknownFields()
+	var body struct {
+		Name string `json:"name"`
+	}
+	if err := dec.Decode(&body); err != nil {
+		resp, rerr := rejectForward(http.StatusBadRequest, "invalid_body",
+			opLabel+" payload is not valid JSON")
+		return "", resp, rerr
+	}
+	if dec.More() {
+		resp, err := rejectForward(http.StatusBadRequest, "invalid_body",
+			opLabel+" payload has trailing data")
+		return "", resp, err
+	}
+	if body.Name == "" {
+		resp, err := rejectForward(http.StatusBadRequest, "invalid_body",
+			opLabel+" payload missing name")
+		return "", resp, err
+	}
+	if strings.ContainsRune(body.Name, '/') {
+		resp, err := rejectForward(http.StatusBadRequest, "invalid_body",
+			opLabel+" payload name must not contain '/'")
+		return "", resp, err
+	}
+	return body.Name, nil, nil
+}
+
 // forwardErrorResponse re-encodes a TablesSource error in the
 // structured shape the follower's handler can re-emit verbatim. This
 // is the leader-side counterpart of writeTablesError: every status /

internal/admin/forward_server_test.go

@@ -517,6 +517,46 @@ func TestForwardServer_BucketOps_NoBucketsSourceReturns501(t *testing.T) {
 	}
 }
 
+func TestForwardServer_DynamoOps_NoTablesSourceReturns501(t *testing.T) {
+	// S3-only deployments construct ForwardServer with a nil
+	// TablesSource so leaders can still register the gRPC service
+	// for follower-forwarded bucket writes. Dynamo operations must
+	// then reject cleanly with 501 instead of dereferencing the nil
+	// source and panicking. Symmetric with
+	// TestForwardServer_BucketOps_NoBucketsSourceReturns501 for the
+	// inverse Dynamo-only build (Codex P1 on PR #673).
+	cases := []struct {
+		name    string
+		op      pb.AdminOperation
+		payload []byte
+	}{
+		{
+			name:    "create_table",
+			op:      pb.AdminOperation_ADMIN_OP_CREATE_TABLE,
+			payload: mustJSON(t, CreateTableRequest{TableName: "orders"}),
+		},
+		{
+			name:    "delete_table",
+			op:      pb.AdminOperation_ADMIN_OP_DELETE_TABLE,
+			payload: []byte(`{"name":"orders"}`),
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := NewForwardServer(nil, fullPrincipalRoleStore(), nil).
+				WithBucketsSource(&stubBucketsSource{})
+			resp, err := srv.Forward(context.Background(), &pb.AdminForwardRequest{
+				Principal: &pb.AdminPrincipal{AccessKey: "AKIA_FULL", Role: "full"},
+				Operation: tc.op,
+				Payload:   tc.payload,
+			})
+			require.NoError(t, err)
+			require.Equal(t, int32(http.StatusNotImplemented), resp.GetStatusCode())
+			require.Contains(t, string(resp.GetPayload()), "not_implemented")
+		})
+	}
+}
+
 func TestForwardServer_CreateBucket_RejectsWhitespacePaddedName(t *testing.T) {
 	// Validation parity with the HTTP path's
 	// validateCreateBucketRequest: a name like " bucket " must

main_admin_forward.go

@@ -73,12 +73,17 @@ type adminForwardServerDeps struct {
 
 // readyForRegistration reports whether the bundle has enough
 // collaborators to construct + register a ForwardServer.
-// TablesSource and RoleStore are both required: the registered
-// service would 500 every Dynamo forwarded call without them.
-// BucketsSource is optional — when nil, the leader-side dispatcher
-// rejects S3 operations with 501, but Dynamo forwarding still works.
+// RoleStore is always required (the principal re-evaluation step
+// has no fallback). At least one of TablesSource or BucketsSource
+// must be present — registering with neither would respond 501 to
+// every operation, which is functionally identical to not
+// registering at all. The dispatcher emits 501 for whichever
+// source is nil so an S3-only or Dynamo-only build still serves
+// its half of the admin surface (Codex P1 on PR #673: an S3-only
+// cluster previously skipped registration entirely and surfaced
+// follower forwards as gRPC Unimplemented / 503).
 func (d adminForwardServerDeps) readyForRegistration() bool {
-	return d.tables != nil && d.roles != nil
+	return d.roles != nil && (d.tables != nil || d.buckets != nil)
 }
 
 // registerAdminForwardServer attaches the leader-side gRPC