rdma: gate the RC data plane behind a dedicated flag

Add --rdma-rc-enable (VGW_RDMA_RC_ENABLE, default false) so the RC control routes and data plane start without implying the cuObject v1 backend. The global CLI hook resolves the mode first: gateway commands require either --rdma-ip or --rdma-rc-enable, and neither path implies the other, so a v2-only deployment boots without a v1 address. The v1 port, retry, pool, and DCI validations also ran for every mode, so stale v1 environment values blocked v2-only startup with unrelated errors. Those validations moved behind the v1 check as a cgo-free helper in internal/rdmamode, exercised alongside the mode matrix, and the CQ-depth limit keeps its 32-bit boundary check there. The RC data plane builds its IAM service, starts the session server, and mounts the three control routes behind SigV4. Startup and shutdown own the backend chain through idempotent guards: the gateway wraps the input backend in a once guard and defers a rollback closure that follows the chain as it grows; the completed v1 chain gets its own once owner, and the RC service is closed first through a backend wrapper installed right after a successful session-server init. Startup failures close exactly what was built, the RunVersityGW lifecycle consumes the same guards instead of closing again, and the RC sessions drain before the backend chain shuts down.
2026-09-23 08:24:17 +00:00 · 2026-09-04 19:44:45 +09:00
parent 1596531003
commit 726d65dbc9
3 changed files with 439 additions and 29 deletions
@@ -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{
@@ -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}
+}
@@ -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)
+	}
+}