From 535cc9d5216e597ee9f9191c5be90fa074b2636e Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Thu, 3 Sep 2026 04:35:53 +0900 Subject: [PATCH] feat: add the hipobj-rc-v2 control routes to the vgwrdma gateway * rdma: add the hipobj-rc-v2 control routes to the vgwrdma gateway Mount the three control routes (prepare, ready, cancel) on the S3 port behind the standard SigV4 middleware. The routes own authentication-adjacent policy the C server cannot see: the middleware wrapper yields to the handler on success, READY and CANCEL re-read the account through the IAM cache bypass so mid-flow deletions and credential rotations take effect immediately, and every object access re-authorizes against the decoded bucket and key. The READY handler implements the session ownership contract: the completion-reference finalizer installs only after the transfer claim succeeds, the PUT path hands the reference to the put view exactly at the borrow point, and the FINAL reply carries the stored object's metadata. Backend I/O runs under a context merged with the RC service context so shutdown unblocks in-flight handlers, with a bounded pool for the fresh IAM lookups. vgwrdma starts the session server alongside the gateway when an RDMA interface is configured, tears it down on exit, and shuts the IAM service down on any startup failure. embedgw learns the readonly flag for the object access checks the routes share. Signed-off-by: Jihyeon Gim * rdma: add the missing stub handlers for non-Linux builds The non-Linux rcroutes stub exposed only Register while the vgwrdma gateway registers the prepare/ready/cancel handlers directly, so cross-compiling cmd/vgwrdma failed with undefined methods. Add the three stub handlers answering 501 Not Implemented and let Register reuse them, matching the Linux Handler API surface. * auth: drop the duplicated GetUserAccountFresh definition The rebase onto main (which already carries GetUserAccountFresh from the iam-cache-fresh change) kept both copies of the method, breaking the build with a redeclaration error. Remove the second copy so the method is defined once. * rdma: address the review findings on the control route wiring Drop the unused Handler.Register from both build variants: the gateway mounts the three control routes through s3api.WithRoute so the SigV4 verifier wrapper (rcAuth) runs in front of each handler, and nothing else calls Register. Clear iamOwned only when RunVersityGW returns nil. It shuts the IAM service down itself at the end of its shutdown sequence, but its early failure paths return before reaching that point, so the deferred shutdown must keep covering those errors. Remove the unused rcserver.SessionInfo parameter from sizeOf; the transferred byte count comes from the READY response alone. * rdma: keep transient IAM failures retryable in the fresh revalidation The fresh account revalidation turned every GetUserAccountFresh error into 403, which reports transient backend failures (LDAP timeouts, network errors) as a revoked account and leaves the client no room to retry. Only a confirmed missing account (auth.ErrNoSuchUser) means that; answer anything else with 503 so clients can retry the request. * rdma: make the IAM shutdown exactly-once and keep gateway errors visible The gateway and RunVersityGW share the IAM service, and which side shut it down could not be told from the return value: runtime failures return after RunVersityGW already shut the service down, while early setup failures return before any shutdown happens. The iamOwned flag therefore either shut the service down twice or leaked it depending on the error, and the error itself was dropped. Wrap the service so Shutdown runs exactly once no matter which side calls it, keep the deferred shutdown for every early failure path, and return the gateway error again. The wrapper re-exposes the optional interfaces (fresh account reads, signing keys, policy evaluation, fixed bucket ownership) so feature detection through the IAM service keeps working. * rdma: reuse the SigV4 account for RC control requests READY and CANCEL are independently authenticated SigV4 requests. Use the account resolved by the normal SigV4 path instead of bypassing the IAM cache a second time. This aligns RC revocation latency with other signed S3 requests and removes the extra backend IAM lookup, its concurrency cap, and the RC-specific IAM error mapping. The session owner check and the READY target and operation authorization are unchanged. * rdma: reword the READY reauthorization comment The comment implied a revocation inside the session window always takes effect at READY, but the account used here is the one SigV4 resolved, which may be a cached entry. State what the check does without claiming account-cache freshness. * rdma: preserve IAM cache behavior and standalone region --------- Signed-off-by: Jihyeon Gim Co-authored-by: Ben McClelland --- cmd/vgwrdma/main.go | 113 +++++- embedgw/embedgw.go | 122 ++++--- rdma/rcroutes/routes_linux.go | 612 ++++++++++++++++++++++++++++++++ rdma/rcroutes/routes_stub.go | 49 +++ rdma/rcserver/rcserver_linux.go | 15 +- 5 files changed, 842 insertions(+), 69 deletions(-) create mode 100644 rdma/rcroutes/routes_linux.go create mode 100644 rdma/rcroutes/routes_stub.go diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 0596ec9c..2185934f 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -23,8 +23,11 @@ import ( _ "net/http/pprof" "os" "strings" + "sync" + "github.com/gofiber/fiber/v3" "github.com/urfave/cli/v2" + "github.com/versity/versitygw/auth" "github.com/versity/versitygw/backend" "github.com/versity/versitygw/cmd/internal/gwcli" "github.com/versity/versitygw/cubackend" @@ -33,7 +36,10 @@ import ( "github.com/versity/versitygw/embedgw" "github.com/versity/versitygw/internal/netutil" "github.com/versity/versitygw/rdma" + "github.com/versity/versitygw/rdma/rcroutes" + "github.com/versity/versitygw/rdma/rcserver" "github.com/versity/versitygw/s3api" + "github.com/versity/versitygw/s3api/middlewares" ) var ( @@ -104,6 +110,7 @@ var ( mpMaxParts int socketPerm string rdmaIP string + rcGidHint string rdmaPort uint poolBufSize int poolBufCount int @@ -123,6 +130,38 @@ var ( BuildTime = "none" ) +type standaloneIAMExtensions interface { + auth.SigningKeyProvider + auth.PolicyEvaluator + auth.FixedBucketOwner +} + +type shutdownOnceService struct { + auth.IAMService + once sync.Once + err error +} + +func (s *shutdownOnceService) Shutdown() error { + s.once.Do(func() { + s.err = s.IAMService.Shutdown() + }) + return s.err +} + +type shutdownOnceStandaloneService struct { + *shutdownOnceService + standaloneIAMExtensions +} + +func wrapIAMShutdownOnce(iam auth.IAMService) auth.IAMService { + wrapped := &shutdownOnceService{IAMService: iam} + if standalone, ok := iam.(standaloneIAMExtensions); ok { + return &shutdownOnceStandaloneService{wrapped, standalone} + } + return wrapped +} + // gatewayCommands are the subcommands that call gwcli.RunGateway (and // therefore need --rdma-ip); admin, utils, help, and version do not. var gatewayCommands = map[string]bool{ @@ -840,6 +879,12 @@ func initFlags() []cli.Flag { EnvVars: []string{"VGW_RDMA_IP"}, Destination: &rdmaIP, }, + &cli.StringFlag{ + Name: "rc-gid-hint", + Usage: "dotted GID prefix selecting the verbs device for the hipobj-rc-v2 RC data plane (default: first device)", + EnvVars: []string{"VGW_RC_GID_HINT"}, + Destination: &rcGidHint, + }, &cli.UintFlag{ Name: "rdma-port", Usage: "port for RDMA listener", @@ -969,7 +1014,7 @@ func runGateway(ctx context.Context, be backend.Backend) error { s3Opts = append(s3Opts, s3api.WithMiddleware("/", cumiddleware.CuObjMiddleware)) } - return embedgw.RunVersityGW(ctx, be, &embedgw.Config{ + cfg := &embedgw.Config{ RootUserAccess: gwcli.RootUserAccess, RootUserSecret: gwcli.RootUserSecret, Region: region, @@ -1065,9 +1110,71 @@ func runGateway(ctx context.Context, be backend.Backend) error { WebsiteKeyFile: websiteKeyFile, WebsiteNoTLS: websiteNoTLS, SigHup: gwcli.SigHup, - S3Options: s3Opts, Version: Version, Build: Build, BuildTime: BuildTime, - }) + } + + if rdmaIP != "" { + // 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. + iamSvc, err := auth.New(cfg.IamOpts()) + if err != nil { + return fmt.Errorf("setup iam for rdma routes: %w", err) + } + iamSvc = wrapIAMShutdownOnce(iamSvc) + cfg.IAMService = iamSvc + // RunVersityGW may shut IAM down before returning. This defer + // also covers errors before and during gateway startup. + defer func() { + _ = iamSvc.Shutdown() + }() + + rcSvc, err := rcserver.Init(rcserver.DeviceOpts{ + GidHint: rcGidHint, + Port: 1, + MaxSessions: 1024, + MaxUserSessions: 64, + MaxStagingBytes: 4 << 30, + MaxUserStagingBytes: 1 << 30, + MaxQPs: 1024, + MaxUserQPs: 16, + TPrepMs: 100000, + TExecMs: 30000, + }) + if err != nil { + return err + } + defer rcSvc.Close() + + rcVerify := middlewares.VerifyV4Signature( + middlewares.RootUserConfig{ + Access: gwcli.RootUserAccess, + Secret: gwcli.RootUserSecret, + }, iamSvc, region, false, true, false) + // Fiber only runs the next handler in the chain when the + // previous one calls ctx.Next; VerifyV4Signature returns + // nil on success without doing so. Wrap it so a verified + // request reaches the route handler, while errors end the + // chain as usual. + rcAuth := func(ctx fiber.Ctx) error { + if err := rcVerify(ctx); err != nil { + return err + } + return ctx.Next() + } + rcH := rcroutes.New(rcSvc, be, iamSvc, readonly, disableACLs) + cfg.S3Options = append(s3Opts, + s3api.WithRoute("POST", "/.hipobj-rc/prepare", rcAuth, rcH.Prepare), + s3api.WithRoute("POST", "/.hipobj-rc/ready", rcAuth, rcH.Ready), + s3api.WithRoute("POST", "/.hipobj-rc/cancel", rcAuth, rcH.Cancel), + ) + } else { + cfg.S3Options = s3Opts + } + + runErr := embedgw.RunVersityGW(ctx, be, cfg) + return runErr } diff --git a/embedgw/embedgw.go b/embedgw/embedgw.go index 1a0c72a6..6065581b 100644 --- a/embedgw/embedgw.go +++ b/embedgw/embedgw.go @@ -524,6 +524,70 @@ type Config struct { // validation, debug logging) are eliminated and concurrent calls are safe. var gatewayRunning atomic.Bool +// IamOpts translates the Config's IAM backend trigger fields into +// auth.Opts. Split out so embedders can build the same IAM service +// the gateway itself would construct (e.g. to mount routes that +// authenticate against it before the gateway starts). +func (c *Config) IamOpts() *auth.Opts { + return &auth.Opts{ + RootAccount: auth.Account{ + Access: c.RootUserAccess, + Secret: c.RootUserSecret, + Role: auth.RoleAdmin, + }, + Dir: c.IAMDir, + LDAPServerURL: c.LDAPServerURL, + LDAPBindDN: c.LDAPBindDN, + LDAPPassword: c.LDAPPassword, + LDAPQueryBase: c.LDAPQueryBase, + LDAPObjClasses: c.LDAPObjClasses, + LDAPAccessAtr: c.LDAPAccessAttr, + LDAPSecretAtr: c.LDAPSecretAttr, + LDAPRoleAtr: c.LDAPRoleAttr, + LDAPUserIdAtr: c.LDAPUserIDAttr, + LDAPGroupIdAtr: c.LDAPGroupIDAttr, + LDAPProjectIdAtr: c.LDAPProjectIDAttr, + LDAPTLSSkipVerify: c.LDAPTLSSkipVerify, + VaultEndpointURL: c.VaultEndpointURL, + VaultNamespace: c.VaultNamespace, + VaultSecretStoragePath: c.VaultSecretStoragePath, + VaultSecretStorageNamespace: c.VaultSecretStorageNamespace, + VaultAuthMethod: c.VaultAuthMethod, + VaultAuthNamespace: c.VaultAuthNamespace, + VaultMountPath: c.VaultMountPath, + VaultRootToken: c.VaultRootToken, + VaultRoleId: c.VaultRoleID, + VaultRoleSecret: c.VaultRoleSecret, + VaultServerCert: c.VaultServerCert, + VaultClientCert: c.VaultClientCert, + VaultClientCertKey: c.VaultClientCertKey, + S3Access: c.S3IAMAccess, + S3Secret: c.S3IAMSecret, + S3Region: c.S3IAMRegion, + S3Bucket: c.S3IAMBucket, + S3Endpoint: c.S3IAMEndpoint, + S3DisableSSlVerfiy: c.S3IAMDisableSSLVerify, + CacheDisable: c.IAMCacheDisable, + CacheTTL: c.IAMCacheTTL, + CachePrune: c.IAMCachePrune, + IpaHost: c.IpaHost, + IpaVaultName: c.IpaVaultName, + IpaUser: c.IpaUser, + IpaPassword: c.IpaPassword, + IpaInsecure: c.IpaInsecure, + StandaloneIAMEndpoint: c.StandaloneIAMEndpoint, + StandaloneIAMAccess: c.StandaloneIAMAccess, + StandaloneIAMSecret: c.StandaloneIAMSecret, + StandaloneClientCert: c.StandaloneClientCert, + StandaloneClientCertKey: c.StandaloneClientCertKey, + StandaloneServerCA: c.StandaloneServerCA, + StandaloneDefaultUserID: c.StandaloneDefaultUserID, + StandaloneDefaultGroupID: c.StandaloneDefaultGroupID, + StandaloneDefaultProjectID: c.StandaloneDefaultProjectID, + StandaloneRegion: c.Region, + } +} + // RunVersityGW starts the VersityGW gateway with the supplied backend and // configuration. It blocks until ctx is cancelled, or an error occurs. All // subsystems are gracefully shut down before the function returns. @@ -698,63 +762,7 @@ func RunVersityGW(ctx context.Context, be backend.Backend, cfg *Config) error { iam := cfg.IAMService if iam == nil { - iam, err = auth.New(&auth.Opts{ - RootAccount: auth.Account{ - Access: cfg.RootUserAccess, - Secret: cfg.RootUserSecret, - Role: auth.RoleAdmin, - }, - Dir: cfg.IAMDir, - LDAPServerURL: cfg.LDAPServerURL, - LDAPBindDN: cfg.LDAPBindDN, - LDAPPassword: cfg.LDAPPassword, - LDAPQueryBase: cfg.LDAPQueryBase, - LDAPObjClasses: cfg.LDAPObjClasses, - LDAPAccessAtr: cfg.LDAPAccessAttr, - LDAPSecretAtr: cfg.LDAPSecretAttr, - LDAPRoleAtr: cfg.LDAPRoleAttr, - LDAPUserIdAtr: cfg.LDAPUserIDAttr, - LDAPGroupIdAtr: cfg.LDAPGroupIDAttr, - LDAPProjectIdAtr: cfg.LDAPProjectIDAttr, - LDAPTLSSkipVerify: cfg.LDAPTLSSkipVerify, - VaultEndpointURL: cfg.VaultEndpointURL, - VaultNamespace: cfg.VaultNamespace, - VaultSecretStoragePath: cfg.VaultSecretStoragePath, - VaultSecretStorageNamespace: cfg.VaultSecretStorageNamespace, - VaultAuthMethod: cfg.VaultAuthMethod, - VaultAuthNamespace: cfg.VaultAuthNamespace, - VaultMountPath: cfg.VaultMountPath, - VaultRootToken: cfg.VaultRootToken, - VaultRoleId: cfg.VaultRoleID, - VaultRoleSecret: cfg.VaultRoleSecret, - VaultServerCert: cfg.VaultServerCert, - VaultClientCert: cfg.VaultClientCert, - VaultClientCertKey: cfg.VaultClientCertKey, - S3Access: cfg.S3IAMAccess, - S3Secret: cfg.S3IAMSecret, - S3Region: cfg.S3IAMRegion, - S3Bucket: cfg.S3IAMBucket, - S3Endpoint: cfg.S3IAMEndpoint, - S3DisableSSlVerfiy: cfg.S3IAMDisableSSLVerify, - CacheDisable: cfg.IAMCacheDisable, - CacheTTL: cfg.IAMCacheTTL, - CachePrune: cfg.IAMCachePrune, - IpaHost: cfg.IpaHost, - IpaVaultName: cfg.IpaVaultName, - IpaUser: cfg.IpaUser, - IpaPassword: cfg.IpaPassword, - IpaInsecure: cfg.IpaInsecure, - StandaloneIAMEndpoint: cfg.StandaloneIAMEndpoint, - StandaloneIAMAccess: cfg.StandaloneIAMAccess, - StandaloneIAMSecret: cfg.StandaloneIAMSecret, - StandaloneClientCert: cfg.StandaloneClientCert, - StandaloneClientCertKey: cfg.StandaloneClientCertKey, - StandaloneServerCA: cfg.StandaloneServerCA, - StandaloneDefaultUserID: cfg.StandaloneDefaultUserID, - StandaloneDefaultGroupID: cfg.StandaloneDefaultGroupID, - StandaloneDefaultProjectID: cfg.StandaloneDefaultProjectID, - StandaloneRegion: cfg.Region, - }) + iam, err = auth.New(cfg.IamOpts()) if err != nil { return fmt.Errorf("setup iam: %w", err) } diff --git a/rdma/rcroutes/routes_linux.go b/rdma/rcroutes/routes_linux.go new file mode 100644 index 00000000..11298cc2 --- /dev/null +++ b/rdma/rcroutes/routes_linux.go @@ -0,0 +1,612 @@ +// 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 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 serves the /.hipobj-rc/{prepare,ready,cancel} +// terminal routes of the hipobj-rc-v2 two-phase transfer protocol. +// +// The routes are the control plane only: routing, SigV4 +// authentication, authorization, and the object backend stay on the +// Go side; the RC session, QP/CQ/MR lifetimes, and the data phase +// live in the C session server bound by rdma/rcserver. +package rcroutes + +import ( + "bytes" + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "net/url" + "strconv" + "strings" + + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/gofiber/fiber/v3" + + "github.com/versity/versitygw/auth" + "github.com/versity/versitygw/backend" + "github.com/versity/versitygw/rdma/rcserver" + "github.com/versity/versitygw/s3api/utils" + "github.com/versity/versitygw/s3response" +) + +// Wire headers of the two-phase protocol (lowercase on the wire; +// Fiber's Get is case-insensitive). Names mirror the hipobj-rc-v2 +// wire contract shared with the client library. +const ( + hdrProtocol = "x-amz-rdma-protocol" + hdrToken = "x-amz-rdma-token" + hdrPsn = "x-amz-rdma-psn" + hdrCookie = "x-amz-rdma-cookie" + hdrOp = "x-amz-rdma-op" + hdrTarget = "x-amz-rdma-target" + hdrSize = "x-amz-rdma-size" + hdrOffset = "x-amz-rdma-offset" + hdrSession = "x-amz-rdma-session" + hdrQpn = "x-amz-rdma-qpn" + hdrMrAddr = "x-amz-rdma-mr-addr" + hdrMrRkey = "x-amz-rdma-mr-rkey" + + protocolValue = "hipobj-rc-v2" + + hdrReply = "x-amz-rdma-reply" + hdrBytes = "x-amz-rdma-bytes-transferred" + hdrEtag = "x-amz-rdma-etag" + hdrVersionID = "x-amz-rdma-version-id" +) + +// Handler serves the three control routes. +type Handler struct { + svc *rcserver.RCSvc + be backend.Backend + iam auth.IAMService + readonly bool + disableACL bool +} + +// New builds the route handler around a started RC service. +func New(svc *rcserver.RCSvc, be backend.Backend, iam auth.IAMService, + readonly, disableACL bool) *Handler { + return &Handler{svc: svc, be: be, iam: iam, + readonly: readonly, disableACL: disableACL} +} + +// principalID derives the session identity digest from the +// authenticated account: SHA-256 over the access key and secret, so +// READY/CANCEL owner checks compare the credential, not a display +// name. +func principalID(acct auth.Account) rcserver.PrincipalID { + h := sha256.Sum256([]byte(acct.Access + ":" + acct.Secret)) + var p rcserver.PrincipalID + copy(p[:], h[:]) + return p +} + +func errNotAdmitted() error { + return fiber.NewError(fiber.StatusServiceUnavailable, + "RDMA service is shutting down") +} + +func invalidHeader(name, value string) error { + return fiber.NewError(fiber.StatusBadRequest, + fmt.Sprintf("invalid %s header: %q", name, value)) +} + +// 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 { + if !h.svc.TryEnter() { + return errNotAdmitted() + } + defer h.svc.Leave() + + if proto := ctx.Get(hdrProtocol); proto != protocolValue { + return invalidHeader(hdrProtocol, proto) + } + + acct := utils.ContextKeyAccount.Get(ctx).(auth.Account) + isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool) + + op := strings.ToUpper(ctx.Get(hdrOp)) + if op != "GET" && op != "PUT" { + return invalidHeader(hdrOp, ctx.Get(hdrOp)) + } + target := ctx.Get(hdrTarget) + bucket, key, ok := splitTarget(target) + if !ok { + return invalidHeader(hdrTarget, target) + } + size, err := parseUint(ctx.Get(hdrSize), 10, 64) + if err != nil || size == 0 { + return invalidHeader(hdrSize, ctx.Get(hdrSize)) + } + offset, err := parseUint(ctx.Get(hdrOffset), 10, 64) + if err != nil { + return invalidHeader(hdrOffset, ctx.Get(hdrOffset)) + } + psn, err := parseUint(ctx.Get(hdrPsn), 16, 32) + if err != nil || psn == 0 || psn > 0xffffff { + return invalidHeader(hdrPsn, ctx.Get(hdrPsn)) + } + cookie, err := parseUint(ctx.Get(hdrCookie), 16, 32) + if err != nil || cookie == 0 { + return invalidHeader(hdrCookie, ctx.Get(hdrCookie)) + } + isPut := op == "PUT" + + // Authorize through the regular object-access chain. + if err := h.authorize(ctx, acct, isRoot, bucket, key, isPut); err != nil { + return err + } + + resp, err := h.svc.Prepare(rcserver.PrepareRequest{ + Principal: principalID(acct), + Op: map[bool]uint8{false: 0, true: 1}[isPut], + Target: target, + Offset: offset, + Size: size, + ClientPsn: uint32(psn), + Cookie: uint32(cookie), + ClientToken: ctx.Get(hdrToken), + }) + if err != nil { + return mapRcError(err) + } + + // GET: stage the object into the session buffer before the + // PREPARE response commits the session. + if !isPut { + if err := h.stageGet(ctx, resp.SessionID, bucket, key, offset, size); err != nil { + _ = h.svc.FinishPrepare(resp.SessionID, false) + return err + } + } + if err := h.svc.FinishPrepare(resp.SessionID, true); err != nil { + return mapRcError(err) + } + + // Wire reply per the hipobj-rc-v2 contract: protocol echo, + // the server endpoint as "200:", session id, and PSN. + ctx.Set(hdrProtocol, protocolValue) + if resp.ReplyToken != "" { + ctx.Set(hdrReply, "200:"+resp.ReplyToken) + } else { + ctx.Set(hdrReply, "200:"+strings.Repeat("0", 88)) + } + ctx.Set(hdrSession, resp.SessionID) + ctx.Set(hdrPsn, formatPsn(resp.ServerPsn)) + if resp.ServerQpn != 0 { + ctx.Set(hdrQpn, formatHex(uint64(resp.ServerQpn))) + } + if resp.StagingAddr != 0 { + ctx.Set(hdrMrAddr, formatHex(resp.StagingAddr)) + ctx.Set(hdrMrRkey, formatHex(uint64(resp.StagingRkey))) + } + return ctx.SendStatus(fiber.StatusOK) +} + +// stageGet reads the object range into the session staging buffer +// and records the staged length on the session. +func (h *Handler) stageGet(ctx fiber.Ctx, sessionID, bucket, key string, + offset, size uint64) error { + lease, err := h.svc.BorrowStaging(sessionID) + if err != nil { + return mapRcError(err) + } + committed := false + defer func() { + if !committed { + _ = h.svc.FinishStaging(*lease, false, 0, "", "") + } + }() + + acceptRange := fmt.Sprintf("bytes=%d-%d", offset, offset+size-1) + // Bind the object read to the service context so gateway + // shutdown cancels it through RCSvc.Close instead of letting + // the ops wait spin on a stalled backend. + objCtx, stopSvc := svcCtx(ctx.RequestCtx(), h.svc.Context()) + defer stopSvc() + res, err := h.be.GetObject(objCtx, &s3.GetObjectInput{ + Bucket: &bucket, + Key: &key, + Range: &acceptRange, + }) + if err != nil { + return err + } + if res.Body != nil { + defer res.Body.Close() + } + written, rerr := io.ReadFull(res.Body, lease.Buf) + if errors.Is(rerr, io.ErrUnexpectedEOF) || errors.Is(rerr, io.EOF) { + rerr = nil + } + if rerr != nil { + return rerr + } + etag, version := "", "" + if res.ETag != nil { + etag = *res.ETag + } + if res.VersionId != nil { + version = *res.VersionId + } + if err := h.svc.FinishStaging(*lease, true, int(written), etag, version); err != nil { + return mapRcError(err) + } + committed = true + return nil +} + +// Ready handles READY: re-authenticate, re-check authorization for +// 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 { + if !h.svc.TryEnter() { + return errNotAdmitted() + } + defer h.svc.Leave() + + if proto := ctx.Get(hdrProtocol); proto != protocolValue { + return invalidHeader(hdrProtocol, proto) + } + + acct := utils.ContextKeyAccount.Get(ctx).(auth.Account) + isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool) + + principal := principalID(acct) + + sessionID := ctx.Get(hdrSession) + if sessionID == "" { + return invalidHeader(hdrSession, "") + } + cookie, err := parseUint(ctx.Get(hdrCookie), 16, 32) + if err != nil || cookie == 0 { + return invalidHeader(hdrCookie, ctx.Get(hdrCookie)) + } + qpn, err := parseUint(ctx.Get(hdrQpn), 16, 32) + if err != nil || qpn == 0 { + return invalidHeader(hdrQpn, ctx.Get(hdrQpn)) + } + mrAddr, err := parseUint(ctx.Get(hdrMrAddr), 16, 64) + if err != nil { + return invalidHeader(hdrMrAddr, ctx.Get(hdrMrAddr)) + } + mrRkey, err := parseUint(ctx.Get(hdrMrRkey), 16, 32) + if err != nil { + return invalidHeader(hdrMrRkey, ctx.Get(hdrMrRkey)) + } + + info, err := h.svc.SessionInfo(sessionID, principal) + 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") + } + + // Re-run authorization for the session's stored target and + // operation using the account authenticated for this READY + // request. + bucket, key, ok := splitTarget(info.Target) + if !ok { + return fiber.NewError(fiber.StatusInternalServerError, "session target") + } + if err := h.authorize(ctx, acct, isRoot, bucket, key, info.Op == 1); err != nil { + // Permission revoked mid-session: cancel the session. + _ = h.svc.Cancel(sessionID, principal) + return err + } + + resp, err := h.svc.ReadyTransfer(rcserver.ReadyRequest{ + Principal: principal, + SessionID: sessionID, + Cookie: uint32(cookie), + ClientQpn: uint32(qpn), + ClientMrAddr: mrAddr, + ClientMrRkey: uint32(mrRkey), + }) + if err != nil { + // The transfer claim rolled back server-side (wire + // failure or duplicate READY): this request holds no + // completion ref, so no local finalizer may run + // either. A second concurrent READY must not be able + // to reap a session the first one is still + // transferring on. + return mapRcError(err) + } + + // Busy reports as RC_OK with a retryable outcome carried in + // the same response (atomic with the transfer result, so a + // concurrent READY cannot rewrite it); the server already + // 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") + } + + // The claim succeeded: from here until the response commits, + // this handler owns the completion ref. A panic or early + // unwind must still release it so the session can be reaped. + finalized := false + defer func() { + if !finalized { + _ = h.svc.FinishFinal(sessionID) + } + }() + + if info.Op == 1 { + // PUT: pick up the received bytes and store them. + // Once GetPutData succeeds the completion ref belongs + // to the put view and FPU is its only owner (v0.13: no + // FF after GD, success or failure), so the outer + // finalizer retires exactly at that point; a failure + // *before* the borrow still falls back to the + // finalizer path below. + put, gd, err := h.commitPut(ctx, sessionID, bucket, key, sizeOf(resp)) + if gd { + finalized = true + } + if err != nil { + return err + } + // The FINAL wire reply carries the stored object's + // metadata, which the backend assigned at commit time. + resp.Etag = put.ETag + resp.VersionID = put.VersionID + } else if err := h.svc.FinishFinal(sessionID); err != nil { + return mapRcError(err) + } else { + finalized = true + } + + // Wire reply per the hipobj-rc-v2 contract: protocol echo, + // cookie echo, transferred bytes, and object metadata. + ctx.Set(hdrProtocol, protocolValue) + ctx.Set(hdrBytes, strconv.FormatUint(resp.BytesTransferred, 10)) + ctx.Set(hdrCookie, formatCookie(resp.CookieEcho)) + if resp.Etag != "" { + ctx.Set(hdrEtag, resp.Etag) + } + if resp.VersionID != "" { + ctx.Set(hdrVersionID, resp.VersionID) + } + return ctx.SendStatus(fiber.StatusOK) +} + +func sizeOf(resp *rcserver.ReadyResponse) uint64 { + if resp == nil { + return 0 + } + return resp.BytesTransferred +} + +// commitPut borrows the PUT view and stores it through the regular +// object-put backend path. The second return value reports whether +// the borrow (GetPutData) succeeded: from that point the put view +// owns the completion ref and FinishPut is its only release, so +// the caller must not run the session finalizer anymore. A +// panic-safe defer releases the view if the handler unwinds before +// FinishPut runs. +func (h *Handler) commitPut(ctx fiber.Ctx, sessionID, bucket, key string, + size uint64) (*s3response.PutObjectOutput, bool, error) { + view, err := h.svc.GetPutData(sessionID) + if err != nil { + return nil, false, mapRcError(err) + } + // Panic-safe ownership: if anything below unwinds, the view is + // still returned exactly once (the ABI consumes the handle a + // single time; a redundant FinishPut after a commit is a + // no-op STALE). + committed := false + defer func() { + if !committed { + _ = h.svc.FinishPut(*view, false, "", "") + } + }() + + contentLength := int64(len(view.Buf)) + putCtx, stopSvc := svcCtx(ctx.RequestCtx(), h.svc.Context()) + defer stopSvc() + res, err := h.be.PutObject(putCtx, s3response.PutObjectInput{ + Bucket: &bucket, + Key: &key, + ContentLength: &contentLength, + Body: bytes.NewReader(view.Buf), + }) + if err != nil { + return nil, true, err + } + if err := h.svc.FinishPut(*view, true, res.ETag, res.VersionID); err != nil { + return nil, true, mapRcError(err) + } + committed = true + return &res, true, nil +} + +// Cancel handles CANCEL: authenticated owner tears the session down. +func (h *Handler) Cancel(ctx fiber.Ctx) error { + if !h.svc.TryEnter() { + return errNotAdmitted() + } + defer h.svc.Leave() + + if proto := ctx.Get(hdrProtocol); proto != protocolValue { + return invalidHeader(hdrProtocol, proto) + } + + acct := utils.ContextKeyAccount.Get(ctx).(auth.Account) + sessionID := ctx.Get(hdrSession) + if sessionID == "" { + return invalidHeader(hdrSession, "") + } + if err := h.svc.Cancel(sessionID, principalID(acct)); err != nil { + if errors.Is(err, rcserver.ErrStale) { + // Already gone: idempotent success for the owner. + return ctx.SendStatus(fiber.StatusOK) + } + return mapRcError(err) + } + ctx.Set(hdrProtocol, protocolValue) + return ctx.SendStatus(fiber.StatusOK) +} + +// authorize runs the object access checks (ACL + policy), plus the +// retention/object-lock re-check for PUT, mirroring the regular +// object controllers. The backend lookups run under a context +// merged with the RC service context so gateway shutdown cancels +// them too. +func (h *Handler) authorize(ctx fiber.Ctx, acct auth.Account, isRoot bool, + bucket, key string, isPut bool) error { + authCtx, stopSvc := svcCtx(ctx.RequestCtx(), h.svc.Context()) + defer stopSvc() + acl, err := h.be.GetBucketAcl(authCtx, + &s3.GetBucketAclInput{Bucket: &bucket}) + if err != nil { + return err + } + parsedAcl, err := auth.ParseACL(acl) + if err != nil { + return err + } + action := auth.GetObjectAction + perm := auth.PermissionRead + if isPut { + action = auth.PutObjectAction + perm = auth.PermissionWrite + } + if err := auth.VerifyAccess(ctx, h.be, auth.AccessOptions{ + Acl: parsedAcl, + AclPermission: perm, + IsRoot: isRoot, + Acc: acct, + Bucket: bucket, + Object: key, + Actions: []auth.Action{action}, + Readonly: h.readonly, + DisableACL: h.disableACL, + Iam: h.iam, + }); err != nil { + return err + } + if isPut { + if err := auth.CheckObjectAccess(ctx, bucket, acct, + []types.ObjectIdentifier{{Key: &key}}, auth.BypassOverwrite, + false, h.be, h.iam, true); err != nil { + return err + } + } + return nil +} + +// svcCtx returns a context bound to both the request and the RC +// service lifetime. Callers must defer the returned stop function +// so the watcher registered on the service context detaches when +// the backend call completes normally, instead of accumulating +// one per request until shutdown. +func svcCtx(request context.Context, svc context.Context) (context.Context, func()) { + merged, cancel := context.WithCancel(request) + stop := context.AfterFunc(svc, cancel) + return merged, func() { + stop() + cancel() + } +} + +// splitTarget splits "/bucket/key[?query]" (the canonical wire +// form; a leading slash separates the bucket from the key) and +// percent-decodes each segment: the wire form is the canonical +// percent-encoded path, so the decoded bucket/key must feed the +// authorization and object I/O, not the raw encoding. +func splitTarget(target string) (bucket, key string, ok bool) { + if target == "" || !strings.HasPrefix(target, "/") { + return "", "", false + } + target = target[1:] + if i := strings.IndexByte(target, '?'); i >= 0 { + target = target[:i] + } + bucket, key, found := strings.Cut(target, "/") + if !found || bucket == "" || key == "" { + return "", "", false + } + bucket, err := url.PathUnescape(bucket) + if err != nil { + return "", "", false + } + key, err = url.PathUnescape(key) + if err != nil || key == "" { + return "", "", false + } + return bucket, key, true +} + +// parseUint parses a decimal or bare-hex (wire) unsigned value. +// base selects the interpretation: hex fields (PSN, cookie, QPN, +// MR addr/rkey) arrive as bare hex without a 0x prefix; size and +// offset are decimal. +func parseUint(s string, base, bits int) (uint64, error) { + if s == "" { + return 0, errors.New("empty") + } + return strconv.ParseUint(s, base, bits) +} + +// formatPsn renders a PSN as 6 uppercase zero-padded hex chars. +func formatPsn(psn uint32) string { + return fmt.Sprintf("%06X", psn) +} + +// formatCookie renders a cookie as 8 uppercase zero-padded hex +// chars (the wire echo form). +func formatCookie(c uint32) string { + return fmt.Sprintf("%08X", c) +} + +// formatHex renders a value as bare lowercase hex (no 0x prefix), +// the wire form for QPN/MR fields. +func formatHex(v uint64) string { + return strconv.FormatUint(v, 16) +} + +// mapRcError translates ABI statuses into HTTP-shaped failures. +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 fiber.NewError(status, msg) +} diff --git a/rdma/rcroutes/routes_stub.go b/rdma/rcroutes/routes_stub.go new file mode 100644 index 00000000..c692750b --- /dev/null +++ b/rdma/rcroutes/routes_stub.go @@ -0,0 +1,49 @@ +// 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 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 serves the /.hipobj-rc control routes of the +// hipobj-rc-v2 two-phase transfer protocol. This file is a stub +// for platforms without RDMA support. +package rcroutes + +import ( + "github.com/gofiber/fiber/v3" + + "github.com/versity/versitygw/auth" + "github.com/versity/versitygw/backend" +) + +// Handler serves the three control routes (stub). +type Handler struct{} + +// New builds a stub route handler; the routes answer 501. +func New(svc any, be backend.Backend, iam auth.IAMService, + readonly, disableACL bool) *Handler { + return &Handler{} +} + +func notImplemented() error { + return fiber.NewError(fiber.StatusNotImplemented, + "RDMA not supported on this platform") +} + +// Prepare is a stub handler that answers 501 Not Implemented. +func (h *Handler) Prepare(ctx fiber.Ctx) error { return notImplemented() } + +// Ready is a stub handler that answers 501 Not Implemented. +func (h *Handler) Ready(ctx fiber.Ctx) error { return notImplemented() } + +// Cancel is a stub handler that answers 501 Not Implemented. +func (h *Handler) Cancel(ctx fiber.Ctx) error { return notImplemented() } diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index 17a8064f..b2f3b9a5 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/rdma/rcserver/rcserver_linux.go @@ -231,15 +231,12 @@ func (s *RCSvc) Leave() { // Convergence: every blocking call an RC handler makes after // TryEnter is bounded - GET staging, PUT commit, and the // bucket-ACL lookup run under the service context and unblock on -// the cancel below; the fresh IAM revalidation selects on the -// same context and its lookup goroutines are capped (a stalled -// context-less IAM backend can strand at most a fixed number of -// them; further requests fail fast); and the remaining -// authorization helpers (VerifyAccess, CheckObjectAccess) -// consume the fiber request context, which the gateway shuts -// down before RunVersityGW returns - and Close runs after -// RunVersityGW returns in every current caller - so those calls -// are cancelled by the fiber shutdown that precedes Close. +// the cancel below; and the authorization helpers (VerifyAccess, +// CheckObjectAccess) consume the fiber request context, which the +// gateway shuts down before RunVersityGW returns - and Close +// runs after RunVersityGW returns in every current caller - so +// those calls are cancelled by the fiber shutdown that precedes +// Close. func (s *RCSvc) Close() { s.once.Do(func() { s.closing.Store(true)