diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index e7ed3251..2da868f2 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -24,6 +24,9 @@ jobs: run: | go get -v -t -d ./... + - name: Build RDMA session server archive + run: make rdma/librcserver.a + - name: Test run: go test -coverprofile profile.txt -race -v -timeout 30s -tags=github ./... diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 45975089..22e02103 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -16,13 +16,12 @@ package main import ( "context" + "errors" "fmt" "log" - "math" "net/http" _ "net/http/pprof" "os" - "strings" "sync" "github.com/gofiber/fiber/v3" @@ -35,6 +34,7 @@ import ( "github.com/versity/versitygw/debuglogger" "github.com/versity/versitygw/embedgw" "github.com/versity/versitygw/internal/netutil" + "github.com/versity/versitygw/internal/rdmamode" "github.com/versity/versitygw/rdma" "github.com/versity/versitygw/rdma/rcroutes" "github.com/versity/versitygw/rdma/rcserver" @@ -111,6 +111,7 @@ var ( socketPerm string rdmaIP string rcGidHint string + rdmaRCEnable bool rdmaPort uint poolBufSize int poolBufCount int @@ -242,11 +243,16 @@ documentation can be found in the GitHub wiki.`, ctx.IsSet("rdma-cq-depth") || ctx.IsSet("rdma-retry-count") - // Only commands that actually start a gateway need --rdma-ip; admin, - // utils, help, and version subcommands print output and exit without - // ever calling runGateway. - if gatewayCommands[ctx.Args().First()] && strings.TrimSpace(rdmaIP) == "" { - return fmt.Errorf("rdma-ip is required") + // Only commands that actually start a gateway need an + // RDMA path; admin, utils, help, and version + // subcommands print output and exit without ever + // calling runGateway. Either the cuObject v1 address + // or the RC v2 flag enables one. + if gatewayCommands[ctx.Args().First()] { + v1On, v2On := rdmamode.Mode(rdmaIP, rdmaRCEnable) + if !v1On && !v2On { + return fmt.Errorf("either --rdma-ip or --rdma-rc-enable is required") + } } return nil }, @@ -886,6 +892,12 @@ func initFlags() []cli.Flag { EnvVars: []string{"VGW_RC_GID_HINT"}, Destination: &rcGidHint, }, + &cli.BoolFlag{ + Name: "rdma-rc-enable", + Usage: "enable the hipobj-rc-v2 RC control routes and data plane (independent of --rdma-ip)", + EnvVars: []string{"VGW_RDMA_RC_ENABLE"}, + Destination: &rdmaRCEnable, + }, &cli.UintFlag{ Name: "rdma-port", Usage: "port for RDMA listener", @@ -967,27 +979,36 @@ func runGateway(ctx context.Context, be backend.Backend) error { if gwcli.CopyObjectThreshold < 1 { return fmt.Errorf("copy-object-threshold must be positive") } - if rdmaPort > 65535 { - return fmt.Errorf("rdma-port %d is out of range (0-65535)", rdmaPort) - } - if rdmaRetryCount > 7 { - return fmt.Errorf("rdma-retry-count %d is out of range (0-7)", rdmaRetryCount) - } - if poolBufSize <= 0 { - return fmt.Errorf("pool-buf-size %d must be positive", poolBufSize) - } - if poolBufCount <= 0 { - return fmt.Errorf("pool-buf-count %d must be positive", poolBufCount) - } - if rdmaTunablesSet && rdmaNumDCIs <= 0 { - return fmt.Errorf("rdma-num-dcis %d must be positive", rdmaNumDCIs) - } - if rdmaTunablesSet && rdmaCQDepth > math.MaxUint32 { - return fmt.Errorf("rdma-cq-depth %d exceeds maximum %d", rdmaCQDepth, uint32(math.MaxUint32)) + v1On, v2On := rdmamode.Mode(rdmaIP, rdmaRCEnable) + if v1On { + // v1-only settings are irrelevant when the cuObject + // backend is not running; stale environment values must + // not block v2-only startup. + v1s := rdmamode.V1Settings{ + Port: rdmaPort, + RetryCount: rdmaRetryCount, + PoolBufSize: poolBufSize, + PoolBufCnt: poolBufCount, + TunablesSet: rdmaTunablesSet, + NumDCIs: rdmaNumDCIs, + CQDepth: rdmaCQDepth, + } + if msg := rdmamode.V1ValidationError(v1s); msg != "" { + return errors.New(msg) + } } + // The gateway command owns the input backend from here on: + // every later failure path must close the whole chain it has + // built so far, exactly once. The closure reads be at run + // time, so the rollback covers the v1 chain and the RC + // wrapper once those layers are added; the once guard keeps + // the shared close with the RunVersityGW lifecycle single. + be = rdmamode.WrapShutdownOnce(be) + defer func() { be.Shutdown() }() + var s3Opts []s3api.Option - if rdmaIP != "" { + if v1On { rdma.ConfigureTelemetry(debug) var tunables *rdma.RDMATunables @@ -1010,7 +1031,11 @@ func runGateway(ctx context.Context, be backend.Backend) error { if err != nil { return err } - be = cuserverBackend + // The v1 layer has no idempotent Shutdown of its own; put + // the once owner around the completed chain so the + // rollback closure and the RunVersityGW lifecycle both + // stop at the same single close. + be = rdmamode.WrapShutdownOnce(cuserverBackend) s3Opts = append(s3Opts, s3api.WithMiddleware("/", cumiddleware.CuObjMiddleware)) } @@ -1116,11 +1141,13 @@ func runGateway(ctx context.Context, be backend.Backend) error { BuildTime: BuildTime, } - if rdmaIP != "" { + if v2On { // RC data plane: build the IAM service the gateway will // use so the control routes authenticate against the // same account store, then start the session server and - // mount the hipobj-rc-v2 routes on the S3 port. + // mount the hipobj-rc-v2 routes on the S3 port. Enabled + // by --rdma-rc-enable independently of the cuObject + // --rdma-ip flag. iamSvc, err := auth.New(cfg.IamOpts()) if err != nil { return fmt.Errorf("setup iam for rdma routes: %w", err) @@ -1148,7 +1175,11 @@ func runGateway(ctx context.Context, be backend.Backend) error { if err != nil { return err } - defer rcSvc.Close() + // Close the RC service before the backend chain so the + // control routes drain first; the wrapper also gives the + // rollback closure and the RunVersityGW lifecycle a + // single, ordered owner of both steps. + be = rdmamode.WrapBackendShutdownAfterRC(be, rcSvc) rcVerify := middlewares.VerifyV4Signature( middlewares.RootUserConfig{ @@ -1162,7 +1193,7 @@ func runGateway(ctx context.Context, be backend.Backend) error { // chain as usual. rcAuth := func(ctx fiber.Ctx) error { if err := rcVerify(ctx); err != nil { - return err + return rcroutes.WriteRouteError(ctx, err) } return ctx.Next() } diff --git a/internal/rdmamode/rdmamode.go b/internal/rdmamode/rdmamode.go new file mode 100644 index 00000000..a5b40aca --- /dev/null +++ b/internal/rdmamode/rdmamode.go @@ -0,0 +1,121 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an "AS +// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +// express or implied. See the License for the specific language +// governing permissions and limitations under the License. + +// Package rdmamode resolves which RDMA paths a gateway runs and +// orders their shutdown. It is deliberately free of cgo so the +// behavior is testable on any build platform. +package rdmamode + +import ( + "fmt" + "math" + "strings" + "sync" + + "github.com/versity/versitygw/backend" +) + +// Mode resolves which RDMA paths run: the cuObject v1 backend +// follows the legacy --rdma-ip contract, and the hipobj-rc-v2 +// control routes follow --rdma-rc-enable. Neither flag implies +// the other. +func Mode(rdmaIP string, rcEnable bool) (v1, v2 bool) { + return strings.TrimSpace(rdmaIP) != "", rcEnable +} + +// V1Settings carries the cuObject v1 tunings that are validated +// only while the v1 backend runs. +type V1Settings struct { + Port uint + RetryCount uint + PoolBufSize int + PoolBufCnt int + TunablesSet bool + NumDCIs int + CQDepth uint +} + +// V1ValidationError describes the first invalid v1 setting, or +// the empty string when every setting is valid. Stale v1 values +// in the environment must not block v2-only or plain-S3 +// startup, so the gateway consults this only when v1 is on. +func V1ValidationError(s V1Settings) string { + switch { + case s.Port > 65535: + return fmt.Sprintf("rdma-port %d is out of range (0-65535)", s.Port) + case s.RetryCount > 7: + return fmt.Sprintf("rdma-retry-count %d is out of range (0-7)", s.RetryCount) + case s.PoolBufSize <= 0: + return fmt.Sprintf("pool-buf-size %d must be positive", s.PoolBufSize) + case s.PoolBufCnt <= 0: + return fmt.Sprintf("pool-buf-count %d must be positive", s.PoolBufCnt) + case s.TunablesSet && s.NumDCIs <= 0: + return fmt.Sprintf("rdma-num-dcis %d must be positive", s.NumDCIs) + case s.TunablesSet && s.CQDepth > math.MaxUint32: + return fmt.Sprintf("rdma-cq-depth %d exceeds the 32-bit limit", s.CQDepth) + default: + return "" + } +} + +// Closer is the close operation of the RC session service. It is +// idempotent. +type Closer interface{ Close() } + +// BackendShutdownAfterRC forwards a backend and closes the RC +// service before the wrapped backend shuts down. The RC handlers +// reference the backend and IAM service, so the RC service must +// stop accepting and drain before either dependency is closed by +// the gateway lifecycle. The whole shutdown, including the +// delegated backend call, runs exactly once: several closers +// (the gateway lifecycle and a deferred rollback in the startup +// path) may call Shutdown on the same instance, and the wrapped +// backends are not guaranteed to be idempotent. +type BackendShutdownAfterRC struct { + backend.Backend + rc Closer + closed sync.Once +} + +// Shutdown closes the RC service, then the wrapped backend, once. +func (b *BackendShutdownAfterRC) Shutdown() { + b.closed.Do(func() { + b.rc.Close() + b.Backend.Shutdown() + }) +} + +// ShutdownOnceBackend makes a backend Shutdown idempotent. The +// gateway startup path defers a rollback close while the gateway +// lifecycle also shuts its backend down; a bare backend has no +// idempotency guarantee of its own. +type ShutdownOnceBackend struct { + backend.Backend + once sync.Once +} + +// Shutdown closes the wrapped backend at most once. +func (b *ShutdownOnceBackend) Shutdown() { + b.once.Do(func() { b.Backend.Shutdown() }) +} + +// WrapShutdownOnce returns be with an idempotent Shutdown. +func WrapShutdownOnce(be backend.Backend) backend.Backend { + return &ShutdownOnceBackend{Backend: be} +} + +// WrapBackendShutdownAfterRC returns be with a Shutdown that +// closes rc first. +func WrapBackendShutdownAfterRC(be backend.Backend, rc Closer) backend.Backend { + return &BackendShutdownAfterRC{Backend: be, rc: rc} +} diff --git a/internal/rdmamode/rdmamode_test.go b/internal/rdmamode/rdmamode_test.go new file mode 100644 index 00000000..abeaa515 --- /dev/null +++ b/internal/rdmamode/rdmamode_test.go @@ -0,0 +1,258 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an "AS +// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +// express or implied. See the License for the specific language +// governing permissions and limitations under the License. + +package rdmamode + +import ( + "math" + "runtime" + "sync" + "testing" + "time" + + "github.com/versity/versitygw/backend" +) + +func TestModeMatrix(t *testing.T) { + tests := []struct { + name string + rdmaIP string + rcEnable bool + wantV1, wantV2 bool + }{ + {"neither", "", false, false, false}, + {"v1 only", "192.0.2.1", false, true, false}, + {"v2 only", "", true, false, true}, + {"both", "192.0.2.1", true, true, true}, + {"v1 ip with whitespace", " ", false, false, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + v1, v2 := Mode(tt.rdmaIP, tt.rcEnable) + if v1 != tt.wantV1 || v2 != tt.wantV2 { + t.Fatalf("Mode(%q, %v) = (%v, %v), want (%v, %v)", + tt.rdmaIP, tt.rcEnable, + v1, v2, tt.wantV1, tt.wantV2) + } + }) + } +} + +func TestV1ValidationCQDepthBoundary(t *testing.T) { + // The v1 tunable narrows to uint32; the boundary itself must + // pass and the first value beyond it must fail. + base := V1Settings{ + Port: 19100, RetryCount: 2, PoolBufSize: 1024, PoolBufCnt: 16, + NumDCIs: 8, TunablesSet: true, + } + base.CQDepth = math.MaxUint32 + if msg := V1ValidationError(base); msg != "" { + t.Fatalf("MaxUint32 rejected: %q", msg) + } + base.CQDepth = math.MaxUint32 + 1 + if msg := V1ValidationError(base); msg == "" { + t.Fatal("MaxUint32+1 accepted") + } +} + +func TestV1ValidationError(t *testing.T) { + if msg := V1ValidationError(V1Settings{ + Port: 99999, RetryCount: 99, PoolBufSize: -1, + }); msg == "" { + t.Fatal("expected an error for invalid settings") + } + if msg := V1ValidationError(V1Settings{ + Port: 19100, RetryCount: 2, PoolBufSize: 1024, PoolBufCnt: 16, + }); msg != "" { + t.Fatalf("valid settings reported %q", msg) + } +} + +// sequenceRC records when Close starts and finishes, so tests +// can prove the backend waits for it. +type sequenceRC struct { + mu sync.Mutex + started int + finished int + release chan struct{} +} + +func (r *sequenceRC) Close() { + r.mu.Lock() + r.started++ + r.mu.Unlock() + <-r.release + r.mu.Lock() + r.finished++ + r.mu.Unlock() +} + +func (r *sequenceRC) counts() (started, finished int) { + r.mu.Lock() + defer r.mu.Unlock() + return r.started, r.finished +} + +// sequenceBackend records Shutdown calls and whether the RC close +// had finished when each ran. +type sequenceBackend struct { + backend.BackendUnsupported + rc *sequenceRC + mu sync.Mutex + shutdowns int + rcDone bool +} + +func (b *sequenceBackend) Shutdown() { + started, finished := b.rc.counts() + b.mu.Lock() + defer b.mu.Unlock() + b.shutdowns++ + b.rcDone = started >= 1 && finished >= 1 +} + +func TestBackendShutdownClosesRCFirst(t *testing.T) { + rc := &sequenceRC{release: make(chan struct{})} + be := &sequenceBackend{rc: rc} + wrapped := WrapBackendShutdownAfterRC(be, rc) + + done := make(chan struct{}) + go func() { + wrapped.Shutdown() + close(done) + }() + + // While RC close is blocked, the backend must not shut down. + // Wait for the close to start; goroutine scheduling needs a + // moment even though the channel blocks it from finishing. + started := make(chan struct{}) + go func() { + for { + s, _ := rc.counts() + if s >= 1 { + close(started) + return + } + runtime.Gosched() + } + }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("RC close never started") + } + be.mu.Lock() + early := be.shutdowns + be.mu.Unlock() + if early != 0 { + t.Fatalf("backend shut down %d times before RC close finished", early) + } + + close(rc.release) + <-done + + be.mu.Lock() + shutdowns, rcDone := be.shutdowns, be.rcDone + be.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("backend shutdown %d times, want 1", shutdowns) + } + if !rcDone { + t.Fatal("backend shut down before RC close finished") + } +} + +func TestBackendShutdownExactlyOnce(t *testing.T) { + // The gateway lifecycle and a startup rollback may both call + // Shutdown on the wrapped backend; every step must run once. + rc := &sequenceRC{release: make(chan struct{})} + close(rc.release) + be := &sequenceBackend{rc: rc} + wrapped := WrapBackendShutdownAfterRC(be, rc) + + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + wrapped.Shutdown() + }() + } + wg.Wait() + + s, f := rc.counts() + if s != 1 || f != 1 { + t.Fatalf("RC close started %d, finished %d; want 1, 1", s, f) + } + be.mu.Lock() + shutdowns := be.shutdowns + be.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("backend shutdown %d times, want exactly 1", shutdowns) + } +} + +func TestShutdownOnceBackendClosesOnce(t *testing.T) { + // A rollback defer and the gateway lifecycle can both call + // Shutdown on a bare backend; the once wrapper must collapse + // them into a single close. + rc := &sequenceRC{release: make(chan struct{})} + close(rc.release) + be := &sequenceBackend{rc: rc} + wrapped := WrapShutdownOnce(be) + + var wg sync.WaitGroup + for i := 0; i < 3; i++ { + wg.Add(1) + go func() { + defer wg.Done() + wrapped.Shutdown() + }() + } + wg.Wait() + + be.mu.Lock() + shutdowns := be.shutdowns + be.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("backend shutdown %d times, want 1", shutdowns) + } +} + +func TestRCWrapperChainsOntoOnceBackend(t *testing.T) { + // The gateway wraps the base backend once, then the RC + // wrapper on top. The RC close runs before the delegated + // shutdown, and repeated calls through either layer reach + // the base backend exactly once. + rc := &sequenceRC{release: make(chan struct{})} + close(rc.release) + base := &sequenceBackend{rc: rc} + onceWrapped := WrapShutdownOnce(base) + rcWrapped := WrapBackendShutdownAfterRC(onceWrapped, rc) + + rcWrapped.Shutdown() + rcWrapped.Shutdown() + onceWrapped.Shutdown() + + s, f := rc.counts() + if s != 1 || f != 1 { + t.Fatalf("RC close started %d, finished %d; want 1, 1", s, f) + } + base.mu.Lock() + shutdowns := base.shutdowns + base.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("base backend shutdown %d times, want 1", shutdowns) + } +} diff --git a/rdma/rcroutes/errors.go b/rdma/rcroutes/errors.go new file mode 100644 index 00000000..007c048e --- /dev/null +++ b/rdma/rcroutes/errors.go @@ -0,0 +1,132 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an "AS +// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +// express or implied. See the License for the specific language +// governing permissions and limitations under the License. + +package rcroutes + +import ( + "errors" + + "github.com/gofiber/fiber/v3" + + "github.com/versity/versitygw/debuglogger" + "github.com/versity/versitygw/s3api/utils" + "github.com/versity/versitygw/s3err" +) + +// ErrRouteNotImplemented is the platform-stub answer of the +// control routes on builds without the RC data plane. +type ErrRouteNotImplemented struct{} + +func (ErrRouteNotImplemented) Error() string { + return "RDMA not supported on this platform" +} + +// isRouteNotImplemented reports whether err is the platform-stub +// answer of the control routes. The marker type is shared, so +// every build classifies it the same way. +func isRouteNotImplemented(err error) bool { + var ni ErrRouteNotImplemented + return errors.As(err, &ni) +} + +// WriteRouteError renders err as the terminal S3-style XML response +// of an RC control route. The production S3 error handler converts +// ordinary Fiber errors into a generic 500 response, so the route +// must send its final status and body itself. Errors that already +// carry S3 semantics (authentication, authorization, object +// backend) keep their status and code; anything else is mapped to +// a protocol error without exposing internal detail. The gateway +// auth adapter uses it for the same reason. +func WriteRouteError(ctx fiber.Ctx, err error) error { + requestID, hostID := utils.EnsureRequestIDs(ctx) + + apiErr := routeError(err) + if isRouteNotImplemented(err) { + apiErr = s3err.APIError{ + Code: "NotImplemented", + Description: "RDMA is not supported on this platform", + HTTPStatusCode: fiber.StatusNotImplemented, + } + } + if apiErr.HTTPStatusCode == fiber.StatusInternalServerError { + logInternalRouteError(ctx, err) + } + + // A full S3 error keeps its own richer XML body - per-type + // diagnostics such as signature details would be lost if + // only the base error were serialized. + var body []byte + var s3Err s3err.S3Error + if errors.As(err, &s3Err) { + body = s3Err.XMLBody(requestID, hostID) + } else { + body = apiErr.XMLBody(requestID, hostID) + } + + ctx.Response().Header.SetContentType(fiber.MIMEApplicationXML) + return ctx.Status(apiErr.HTTPStatusCode).Send(body) +} + +// routeError resolves err to its S3-style response. Errors that +// already carry S3 semantics (authentication, authorization, +// object backend) keep their status and code; RC transport errors +// map to the closest protocol error. +func routeError(err error) s3err.APIError { + var s3Err s3err.S3Error + if errors.As(err, &s3Err) { + return s3Err.BaseError() + } + + code, status, description := routeErrorDetails(err) + return s3err.APIError{ + Code: code, + Description: description, + HTTPStatusCode: status, + } +} + +// logInternalRouteError keeps a server-side trace of unexpected +// failures. The wire response stays generic; without this the +// terminal serializer would hide the production diagnostics the +// global error handler used to log. +func logInternalRouteError(ctx fiber.Ctx, err error) { + debuglogger.InternalError(err) +} + +// routeErrorDetails maps a non-S3 route error to its protocol +// error code, HTTP status, and description. +func routeErrorDetails(err error) (code string, status int, description string) { + switch { + case isRouteBadRequest(err): + return "InvalidRdmaRequest", fiber.StatusBadRequest, + "Invalid RDMA control request" + case isRouteNotFound(err): + return "NoSuchRdmaSession", fiber.StatusNotFound, + "No such RDMA session" + case isRouteConflict(err): + return "RdmaSessionConflict", fiber.StatusConflict, + "RDMA session state conflict" + case isRouteLimit(err): + return "RdmaResourceLimit", fiber.StatusTooManyRequests, + "RDMA resource limit reached" + case isRouteBadGateway(err): + return "RdmaTransferFailed", fiber.StatusBadGateway, + "RDMA transfer failed" + case isRouteUnavailable(err): + return "RdmaServiceUnavailable", fiber.StatusServiceUnavailable, + "RDMA service is shutting down" + default: + return s3err.GetAPIError(s3err.ErrInternalError).Code, + fiber.StatusInternalServerError, "Internal Error" + } +} diff --git a/rdma/rcroutes/errors_linux.go b/rdma/rcroutes/errors_linux.go new file mode 100644 index 00000000..25b9f4a1 --- /dev/null +++ b/rdma/rcroutes/errors_linux.go @@ -0,0 +1,85 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an "AS +// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +// express or implied. See the License for the specific language +// governing permissions and limitations under the License. + +//go:build linux && amd64 && cgo + +package rcroutes + +import ( + "errors" + + "github.com/versity/versitygw/rdma/rcserver" +) + +// This file keeps the Linux-only error classifiers next to the +// shared terminal serializer in errors.go. The marker types below +// carry only the intended protocol class; the HTTP status and XML +// body are decided in one place. + +// errRouteBadRequest marks a malformed control request (invalid +// header value, bad argument). +type errRouteBadRequest struct{} + +func (errRouteBadRequest) Error() string { return "invalid RDMA control request" } + +// errRouteUnavailable marks admission refusal during shutdown. +type errRouteUnavailable struct{} + +func (errRouteUnavailable) Error() string { + return "RDMA service is shutting down" +} + +// isRouteBadRequest reports whether err is a malformed-request +// class error: an invalid header value from the route handlers or +// a rejected argument, short transfer, or oversized value from +// the session server. +func isRouteBadRequest(err error) bool { + var bad errRouteBadRequest + return errors.As(err, &bad) || + errors.Is(err, rcserver.ErrShort) || + errors.Is(err, rcserver.ErrTrunc) || + errors.Is(err, rcserver.ErrArg) +} + +// isRouteNotFound reports whether err identifies an unknown or +// cross-principal session; both map to the same response so the +// session id is not disclosed across principals. +func isRouteNotFound(err error) bool { + return errors.Is(err, rcserver.ErrNoSession) +} + +// isRouteConflict reports whether err is a stale, duplicate, or +// wrong-state session error. +func isRouteConflict(err error) bool { + return errors.Is(err, rcserver.ErrStale) || + errors.Is(err, rcserver.ErrState) || + errors.Is(err, rcserver.ErrDouble) +} + +// isRouteLimit reports whether err was caused by a configured +// resource limit. +func isRouteLimit(err error) bool { + return errors.Is(err, rcserver.ErrLimit) +} + +// isRouteBadGateway reports whether the RC wire transfer failed. +func isRouteBadGateway(err error) bool { + return errors.Is(err, rcserver.ErrWire) +} + +// isRouteUnavailable reports whether the RC service refused +// admission or is shutting down. +func isRouteUnavailable(err error) bool { + var unavail errRouteUnavailable + return errors.As(err, &unavail) +} diff --git a/rdma/rcroutes/errors_mapping_linux_test.go b/rdma/rcroutes/errors_mapping_linux_test.go new file mode 100644 index 00000000..74adc866 --- /dev/null +++ b/rdma/rcroutes/errors_mapping_linux_test.go @@ -0,0 +1,130 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an "AS +// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +// express or implied. See the License for the specific language +// governing permissions and limitations under the License. + +//go:build linux && amd64 && cgo + +package rcroutes + +import ( + "errors" + "fmt" + "net/http" + "testing" + + "github.com/versity/versitygw/rdma/rcserver" +) + +// This file verifies the mapping from session-server failures to +// protocol error responses. It runs where the Linux session +// server links; the s3api package exercises the shared +// serializer through the production S3 server on every platform. + +func TestRouteErrorProtocolMapping(t *testing.T) { + tests := []struct { + name string + err error + code string + want int + }{ + { + name: "invalid header is 400", + err: invalidHeader(hdrProtocol, "bogus"), + code: "InvalidRdmaRequest", + want: http.StatusBadRequest, + }, + { + name: "unknown session is 404", + err: rcserver.ErrNoSession, + code: "NoSuchRdmaSession", + want: http.StatusNotFound, + }, + { + name: "owner mismatch answers as unknown session", + err: fmt.Errorf("session owner mismatch: %w", rcserver.ErrNoSession), + code: "NoSuchRdmaSession", + want: http.StatusNotFound, + }, + { + name: "stale session is 409", + err: rcserver.ErrStale, + code: "RdmaSessionConflict", + want: http.StatusConflict, + }, + { + name: "duplicate borrow is 409", + err: fmt.Errorf("peer busy: %w", rcserver.ErrDouble), + code: "RdmaSessionConflict", + want: http.StatusConflict, + }, + { + name: "wrong session state is 409", + err: rcserver.ErrState, + code: "RdmaSessionConflict", + want: http.StatusConflict, + }, + { + name: "resource limit is 429", + err: rcserver.ErrLimit, + code: "RdmaResourceLimit", + want: http.StatusTooManyRequests, + }, + { + name: "wire failure is 502", + err: rcserver.ErrWire, + code: "RdmaTransferFailed", + want: http.StatusBadGateway, + }, + { + name: "short transfer is 400", + err: rcserver.ErrShort, + code: "InvalidRdmaRequest", + want: http.StatusBadRequest, + }, + { + name: "value too long is 400", + err: rcserver.ErrTrunc, + code: "InvalidRdmaRequest", + want: http.StatusBadRequest, + }, + { + name: "invalid argument is 400", + err: rcserver.ErrArg, + code: "InvalidRdmaRequest", + want: http.StatusBadRequest, + }, + { + name: "admission refusal is 503", + err: errNotAdmitted(), + code: "RdmaServiceUnavailable", + want: http.StatusServiceUnavailable, + }, + { + name: "unexpected error is 500", + err: errors.New("boom"), + code: "InternalError", + want: http.StatusInternalServerError, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := routeError(tt.err) + if got.HTTPStatusCode != tt.want { + t.Fatalf("status = %d, want %d", got.HTTPStatusCode, tt.want) + } + if got.Code != tt.code { + t.Fatalf("code = %q, want %q", got.Code, tt.code) + } + }) + } +} diff --git a/rdma/rcroutes/errors_stub.go b/rdma/rcroutes/errors_stub.go new file mode 100644 index 00000000..97e1c178 --- /dev/null +++ b/rdma/rcroutes/errors_stub.go @@ -0,0 +1,41 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an "AS +// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +// express or implied. See the License for the specific language +// governing permissions and limitations under the License. + +//go:build !(linux && amd64 && cgo) + +package rcroutes + +// isRouteBadRequest reports whether err is a malformed-request +// class error. The stub never classifies request errors because it +// answers before validation. +func isRouteBadRequest(err error) bool { return false } + +// isRouteNotFound reports whether err identifies an unknown +// session. Never true on the stub. +func isRouteNotFound(err error) bool { return false } + +// isRouteConflict reports whether err is a session-state conflict. +// Never true on the stub. +func isRouteConflict(err error) bool { return false } + +// isRouteLimit reports whether err was caused by a resource +// limit. Never true on the stub. +func isRouteLimit(err error) bool { return false } + +// isRouteBadGateway reports whether the RC wire transfer failed. +// Never true on the stub. +func isRouteBadGateway(err error) bool { return false } + +// isRouteUnavailable reports whether the RC service refused +// admission. Never true on the stub. +func isRouteUnavailable(err error) bool { return false } diff --git a/rdma/rcroutes/routes_linux.go b/rdma/rcroutes/routes_linux.go index 11298cc2..e0a6c818 100644 --- a/rdma/rcroutes/routes_linux.go +++ b/rdma/rcroutes/routes_linux.go @@ -97,18 +97,27 @@ func principalID(acct auth.Account) rcserver.PrincipalID { } func errNotAdmitted() error { - return fiber.NewError(fiber.StatusServiceUnavailable, - "RDMA service is shutting down") + return errRouteUnavailable{} } func invalidHeader(name, value string) error { - return fiber.NewError(fiber.StatusBadRequest, - fmt.Sprintf("invalid %s header: %q", name, value)) + return fmt.Errorf("invalid %s header: %q: %w", + name, value, errRouteBadRequest{}) } // Prepare handles PREPARE: authorize the object access, create the // session, and (GET) stage the object into the session buffer. func (h *Handler) Prepare(ctx fiber.Ctx) error { + // Serialize any error at the route boundary: the + // production S3 error handler turns ordinary Fiber + // errors into a generic 500 response. + if err := h.prepareCore(ctx); err != nil { + return WriteRouteError(ctx, err) + } + return nil +} + +func (h *Handler) prepareCore(ctx fiber.Ctx) error { if !h.svc.TryEnter() { return errNotAdmitted() } @@ -256,6 +265,16 @@ func (h *Handler) stageGet(ctx fiber.Ctx, sessionID, bucket, key string, // the session's target, then run the transfer. PUT picks up the // received data and hands it to the object backend. func (h *Handler) Ready(ctx fiber.Ctx) error { + // Serialize any error at the route boundary: the + // production S3 error handler turns ordinary Fiber + // errors into a generic 500 response. + if err := h.readyCore(ctx); err != nil { + return WriteRouteError(ctx, err) + } + return nil +} + +func (h *Handler) readyCore(ctx fiber.Ctx) error { if !h.svc.TryEnter() { return errNotAdmitted() } @@ -295,7 +314,7 @@ func (h *Handler) Ready(ctx fiber.Ctx) error { if err != nil { // Owner mismatch and unknown session both answer 404 so // the session id is not disclosed cross-principal. - return fiber.NewError(fiber.StatusNotFound, "no such RDMA session") + return mapRcError(err) } // Re-run authorization for the session's stored target and @@ -303,7 +322,7 @@ func (h *Handler) Ready(ctx fiber.Ctx) error { // request. bucket, key, ok := splitTarget(info.Target) if !ok { - return fiber.NewError(fiber.StatusInternalServerError, "session target") + return errors.New("invalid session target") } if err := h.authorize(ctx, acct, isRoot, bucket, key, info.Op == 1); err != nil { // Permission revoked mid-session: cancel the session. @@ -335,7 +354,7 @@ func (h *Handler) Ready(ctx fiber.Ctx) error { // rolled the claim back (state Prepared, no completion ref), // so answer 409 without any finalizer. if resp.Outcome == rcserver.ReadyBusy { - return fiber.NewError(fiber.StatusConflict, "peer busy") + return fmt.Errorf("peer busy: %w", rcserver.ErrDouble) } // The claim succeeded: from here until the response commits, @@ -439,6 +458,16 @@ func (h *Handler) commitPut(ctx fiber.Ctx, sessionID, bucket, key string, // Cancel handles CANCEL: authenticated owner tears the session down. func (h *Handler) Cancel(ctx fiber.Ctx) error { + // Serialize any error at the route boundary: the + // production S3 error handler turns ordinary Fiber + // errors into a generic 500 response. + if err := h.cancelCore(ctx); err != nil { + return WriteRouteError(ctx, err) + } + return nil +} + +func (h *Handler) cancelCore(ctx fiber.Ctx) error { if !h.svc.TryEnter() { return errNotAdmitted() } @@ -583,30 +612,15 @@ func formatHex(v uint64) string { } // mapRcError translates ABI statuses into HTTP-shaped failures. +// mapRcError wraps a session-server error so the shared terminal +// serializer in errors.go can classify it. Owner mismatch answers +// the same 404 as an unknown session so a session id is never +// disclosed across principals. func mapRcError(err error) error { - var status int - var msg string switch { - case errors.Is(err, rcserver.ErrNoSession): - status, msg = fiber.StatusNotFound, "no such RDMA session" - case errors.Is(err, rcserver.ErrStale): - status, msg = fiber.StatusConflict, "stale RDMA session handle" case errors.Is(err, rcserver.ErrSession): - status, msg = fiber.StatusForbidden, "RDMA session owner mismatch" - case errors.Is(err, rcserver.ErrState), errors.Is(err, rcserver.ErrDouble): - status, msg = fiber.StatusConflict, "wrong RDMA session state" - case errors.Is(err, rcserver.ErrLimit): - status, msg = fiber.StatusTooManyRequests, "RDMA resource limit" - case errors.Is(err, rcserver.ErrWire): - status, msg = fiber.StatusBadGateway, "RDMA transfer failed" - case errors.Is(err, rcserver.ErrShort): - status, msg = fiber.StatusBadRequest, "short RDMA transfer" - case errors.Is(err, rcserver.ErrTrunc): - status, msg = fiber.StatusBadRequest, "RDMA value too long" - case errors.Is(err, rcserver.ErrArg): - status, msg = fiber.StatusBadRequest, "invalid RDMA argument" - default: - status, msg = fiber.StatusInternalServerError, "RDMA internal error" + return fmt.Errorf("session owner mismatch: %w", + rcserver.ErrNoSession) } - return fiber.NewError(status, msg) + return err } diff --git a/rdma/rcroutes/routes_stub.go b/rdma/rcroutes/routes_stub.go index c692750b..387f4a5e 100644 --- a/rdma/rcroutes/routes_stub.go +++ b/rdma/rcroutes/routes_stub.go @@ -34,16 +34,17 @@ func New(svc any, be backend.Backend, iam auth.IAMService, return &Handler{} } -func notImplemented() error { - return fiber.NewError(fiber.StatusNotImplemented, - "RDMA not supported on this platform") +// notImplemented answers 501 through the shared terminal error +// serializer so the response shape matches the Linux routes. +func notImplemented(ctx fiber.Ctx) error { + return WriteRouteError(ctx, ErrRouteNotImplemented{}) } // Prepare is a stub handler that answers 501 Not Implemented. -func (h *Handler) Prepare(ctx fiber.Ctx) error { return notImplemented() } +func (h *Handler) Prepare(ctx fiber.Ctx) error { return notImplemented(ctx) } // Ready is a stub handler that answers 501 Not Implemented. -func (h *Handler) Ready(ctx fiber.Ctx) error { return notImplemented() } +func (h *Handler) Ready(ctx fiber.Ctx) error { return notImplemented(ctx) } // Cancel is a stub handler that answers 501 Not Implemented. -func (h *Handler) Cancel(ctx fiber.Ctx) error { return notImplemented() } +func (h *Handler) Cancel(ctx fiber.Ctx) error { return notImplemented(ctx) } diff --git a/s3api/rc_route_error_test.go b/s3api/rc_route_error_test.go new file mode 100644 index 00000000..3cfe7c46 --- /dev/null +++ b/s3api/rc_route_error_test.go @@ -0,0 +1,212 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an "AS +// IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +// express or implied. See the License for the specific language +// governing permissions and limitations under the License. + +package s3api + +import ( + "bytes" + "encoding/xml" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gofiber/fiber/v3" + + "github.com/versity/versitygw/rdma/rcroutes" + "github.com/versity/versitygw/s3err" +) + +// This file pins the wire behavior of RDMA control-route error +// responses at the production server boundary: the real error +// handler, the real request-ID middleware, and the terminal +// serializer the RDMA routes use. The RC route handlers live in +// rdma/rcroutes; these tests exercise the same response path a +// client observes. + +type rcErrorResponse struct { + XMLName xml.Name `xml:"Error"` + Code string `xml:"Code"` + Message string `xml:"Message"` + RequestID string `xml:"RequestId"` + HostID string `xml:"HostId"` +} + +func TestRCRouteErrorPreservesS3Error(t *testing.T) { + server, err := newTestS3ApiServer( + WithRoute(http.MethodPost, "/.hipobj-rc/op", func(ctx fiber.Ctx) error { + return rcroutes.WriteRouteError(ctx, + s3err.GetAPIError(s3err.ErrAccessDenied)) + }), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + resp, err := server.app.Test(httptest.NewRequest(http.MethodPost, "/.hipobj-rc/op", nil)) + if err != nil { + t.Fatalf("app.Test() error = %v", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusForbidden) + } + var er rcErrorResponse + if err := xml.NewDecoder(resp.Body).Decode(&er); err != nil { + t.Fatalf("decode XML body: %v", err) + } + if er.Code != "AccessDenied" { + t.Fatalf("code = %q, want AccessDenied", er.Code) + } + if er.RequestID == "" || er.HostID == "" { + t.Fatal("response missing RequestId or HostId") + } + if got := resp.Header.Get("x-amz-request-id"); got != er.RequestID { + t.Fatalf("body RequestId %q != header %q", er.RequestID, got) + } + if got := resp.Header.Get("x-amz-id-2"); got != er.HostID { + t.Fatalf("body HostId %q != header %q", er.HostID, got) + } + if ct := resp.Header.Get("Content-Type"); ct != fiber.MIMEApplicationXML { + t.Fatalf("content-type = %q, want %q", ct, fiber.MIMEApplicationXML) + } +} + +func TestRCRouteErrorKeepsSubtypeDiagnostics(t *testing.T) { + // A wrapped per-type S3 error must keep both its status and + // its subtype-only XML fields; serializing only the base + // error would drop the diagnostics. + server, err := newTestS3ApiServer( + WithRoute(http.MethodPost, "/.hipobj-rc/op", func(ctx fiber.Ctx) error { + inner := s3err.GetAPIError(s3err.ErrSignatureDoesNotMatch) + wrapped := s3err.SignatureDoesNotMatchError{ + AWSAccessKeyId: "AKIAEXAMPLE", + // The remaining diagnostic fields flow from the + // base through the subtype constructor in + // production; the wire contract under test is + // that the subtype body is used verbatim. + APIError: inner, + StringToSign: "EXAMPLE-STRING-TO-SIGN", + } + return rcroutes.WriteRouteError(ctx, + fmt.Errorf("auth: %w", wrapped)) + }), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + resp, err := server.app.Test(httptest.NewRequest(http.MethodPost, "/.hipobj-rc/op", nil)) + if err != nil { + t.Fatalf("app.Test() error = %v", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusForbidden) + } + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + for _, want := range []string{"SignatureDoesNotMatch", "AKIAEXAMPLE", + "EXAMPLE-STRING-TO-SIGN"} { + if !bytes.Contains(body, []byte(want)) { + t.Fatalf("body missing %q: %s", want, body) + } + } +} + +func TestRCRouteErrorRawFiberIs500(t *testing.T) { + // Control case: an ordinary fiber error collapses into the + // generic 500 of the production error handler. This is the + // behavior WriteRouteError exists to avoid. + server, err := newTestS3ApiServer( + WithRoute(http.MethodPost, "/.hipobj-rc/op", func(ctx fiber.Ctx) error { + return fiber.NewError(fiber.StatusTeapot, "raw") + }), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + resp, err := server.app.Test(httptest.NewRequest(http.MethodPost, "/.hipobj-rc/op", nil)) + if err != nil { + t.Fatalf("app.Test() error = %v", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusInternalServerError { + t.Fatalf("status = %d, want %d (production handler collapses fiber errors)", + resp.StatusCode, http.StatusInternalServerError) + } +} + +func TestRCRouteErrorUnexpectedIsGeneric500(t *testing.T) { + server, err := newTestS3ApiServer( + WithRoute(http.MethodPost, "/.hipobj-rc/op", func(ctx fiber.Ctx) error { + return rcroutes.WriteRouteError(ctx, errors.New("boom")) + }), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + resp, err := server.app.Test(httptest.NewRequest(http.MethodPost, "/.hipobj-rc/op", nil)) + if err != nil { + t.Fatalf("app.Test() error = %v", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusInternalServerError { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusInternalServerError) + } + var er rcErrorResponse + if err := xml.NewDecoder(resp.Body).Decode(&er); err != nil { + t.Fatalf("decode XML body: %v", err) + } + if er.Code != "InternalError" { + t.Fatalf("code = %q, want InternalError", er.Code) + } +} + +func TestRCRouteErrorStubNotImplemented(t *testing.T) { + server, err := newTestS3ApiServer( + WithRoute(http.MethodPost, "/.hipobj-rc/op", func(ctx fiber.Ctx) error { + return rcroutes.WriteRouteError(ctx, rcroutes.ErrRouteNotImplemented{}) + }), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + resp, err := server.app.Test(httptest.NewRequest(http.MethodPost, "/.hipobj-rc/op", nil)) + if err != nil { + t.Fatalf("app.Test() error = %v", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusNotImplemented { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusNotImplemented) + } + var er rcErrorResponse + if err := xml.NewDecoder(resp.Body).Decode(&er); err != nil { + t.Fatalf("decode XML body: %v", err) + } + if er.Code != "NotImplemented" { + t.Fatalf("code = %q, want NotImplemented", er.Code) + } +}