diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 55128c6c..bd7579d9 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -1286,7 +1286,7 @@ func runGateway(ctx context.Context, be backend.Backend) error { // nil on success without doing so. Wrap it so a verified // request reaches the route handler, while errors end the // chain as usual. - rcH := rcroutes.New(rcSvc, be, iamSvc, readonly, disableACLs) + rcH := rcroutes.New(rcSvc, be, iamSvc, readonly, disableACLs, int(rcMaxSessions)) // The RC shutdown wrapper drains queued operational // publications before the sinks close; the drain hook is // the route handler, which is built only now (it needs diff --git a/rdma/rcroutes/ops_linux.go b/rdma/rcroutes/ops_linux.go index b7a33ea0..11aa5695 100644 --- a/rdma/rcroutes/ops_linux.go +++ b/rdma/rcroutes/ops_linux.go @@ -286,10 +286,22 @@ type opsTracker struct { overflow []pubJob reqBacklog atomic.Int64 reqDropped atomic.Int64 - stopped bool - done chan struct{} - drain chan struct{} - drainOnce sync.Once + // pubPending counts queued-but-unpublished session records. + // The native side releases its session quota when it fires + // the teardown notification, not when the audit record lands, + // so successive sessions can queue more records than the + // live-session limit allows. Admission control closes that + // gap: a new session is refused while too many of its + // predecessors' records are still unpublished, so a stalled + // sink delays new sessions instead of accumulating memory. + pubPending atomic.Int64 + // sessionLimit is the native concurrent-session quota; the + // admission budget scales with it. + sessionLimit int + stopped bool + done chan struct{} + drain chan struct{} + drainOnce sync.Once } // pubJob is one deferred publication handed to the worker. @@ -327,15 +339,16 @@ const pubRequestBacklogCap = 4096 // whatever the sinks cannot keep up with. Each job is a few // pointers; a stalled sink delays records, it does not lose // them. -func newOpsTracker() *opsTracker { +func newOpsTracker(sessionLimit int) *opsTracker { t := &opsTracker{ sessions: map[string]*sessionRecord{}, app: fiber.New(fiber.Config{ Immutable: true, }), - pubq: make(chan pubJob, pubQueueSoftCap), - done: make(chan struct{}), - drain: make(chan struct{}), + pubq: make(chan pubJob, pubQueueSoftCap), + done: make(chan struct{}), + drain: make(chan struct{}), + sessionLimit: sessionLimit, } go func() { defer close(t.done) @@ -408,13 +421,16 @@ func (t *opsTracker) takeOverflow() []pubJob { return pending } -// run publishes one job and releases its request-backlog -// reservation, if any. +// run publishes one job and releases its reservations: the +// request-backlog slot and the session admission credit the +// record was holding. func (t *opsTracker) run(job pubJob) { job.emit.publish(job.err, job.byt) if job.isReq { t.reqBacklog.Add(-1) + return } + t.pubPending.Add(-1) } // Shutdown drains pending publications and stops the worker. The @@ -490,8 +506,24 @@ func (t *opsTracker) SetOpsServices(ops OpsServices) { // The account is captured by value but its string fields still // reference request storage on some IAM paths, so the sink-relevant // identity is cloned as well. +// errPubBacklog reports admission refusal: too many earlier +// sessions still have unpublished audit records, so accepting +// another would grow the publication backlog without bound while +// a sink is stalled. +var errPubBacklog = errors.New("publication backlog at capacity") + func (t *opsTracker) register(sessionID string, acct auth.Account, - region, bucket, key string, isPut bool, start time.Time) { + region, bucket, key string, isPut bool, start time.Time) error { + // Admission control: the native quota counts live sessions, + // but teardown notifications fire before the audit records + // land, so session turnover can queue more records than the + // quota bounds. Refusing new sessions while the unpublished + // backlog exceeds the quota turns a stalled sink into + // latency (the client retries) instead of unbounded memory. + // Unbounded when sessionLimit is unset (tests). + if t.sessionLimit > 0 && t.pubPending.Load() >= int64(t.sessionLimit) { + return errPubBacklog + } acct.Access = strings.Clone(acct.Access) emit := &opsEmitter{ ops: t.loadOps(), @@ -506,7 +538,9 @@ func (t *opsTracker) register(sessionID string, acct auth.Account, t.mu.Lock() defer t.mu.Unlock() + t.pubPending.Add(1) t.sessions[sessionID] = &sessionRecord{emit: emit} + return nil } // unregister drops a session entry whose PREPARE finalization @@ -733,10 +767,12 @@ func (t *opsTracker) dispatchOrDrop(job pubJob) { return } if t.reqBacklog.Load() >= pubRequestBacklogCap { - t.pubmu.Unlock() // Overload policy: drop and count. The record carries no - // session and no owner can reissue it. + // session and no owner can reissue it. Incremented under + // pubmu so Shutdown's report (also under pubmu via the + // drain's stopped transition) cannot miss it. t.reqDropped.Add(1) + t.pubmu.Unlock() return } job.isReq = true diff --git a/rdma/rcroutes/ops_linux_test.go b/rdma/rcroutes/ops_linux_test.go index 09a2c3d2..312622eb 100644 --- a/rdma/rcroutes/ops_linux_test.go +++ b/rdma/rcroutes/ops_linux_test.go @@ -38,7 +38,7 @@ import ( // exactly once. Unreserved records are published by the callback. func TestOpsTrackerCallbackPublishesExpiry(t *testing.T) { - tr := newOpsTracker() + tr := newOpsTracker(0) tr.register("sess-1", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", false, time.Now()) if got := len(tr.sessions); got != 1 { @@ -60,7 +60,7 @@ func TestOpsTrackerCallbackPublishesExpiry(t *testing.T) { } func TestOpsTrackerReserveBlocksCallback(t *testing.T) { - tr := newOpsTracker() + tr := newOpsTracker(0) tr.register("sess-2", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", true, time.Now()) @@ -88,7 +88,7 @@ func TestOpsTrackerReserveBlocksCallback(t *testing.T) { } func TestOpsTrackerReserveIsExclusive(t *testing.T) { - tr := newOpsTracker() + tr := newOpsTracker(0) tr.register("sess-3", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", false, time.Now()) @@ -101,7 +101,7 @@ func TestOpsTrackerReserveIsExclusive(t *testing.T) { } func TestOpsTrackerFailOutcome(t *testing.T) { - tr := newOpsTracker() + tr := newOpsTracker(0) tr.register("sess-4", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", false, time.Now()) @@ -116,7 +116,7 @@ func TestOpsTrackerFailOutcome(t *testing.T) { } func TestOpsTrackerUnregister(t *testing.T) { - tr := newOpsTracker() + tr := newOpsTracker(0) tr.register("sess-5", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", false, time.Now()) tr.unregister("sess-5") @@ -132,7 +132,7 @@ func TestOpsTrackerUnregister(t *testing.T) { } func TestOpsTrackerUnknownSession(t *testing.T) { - tr := newOpsTracker() + tr := newOpsTracker(0) // Unknown sessions and the nil tracker are silent no-ops. var nilTracker *opsTracker nilTracker.reserve("ghost") @@ -237,7 +237,7 @@ func (r *recordingLogger) Shutdown() error { return nil } // publishes exactly one record with its own outcome and bytes. func TestOpsTrackerPublishesExactlyOncePerSession(t *testing.T) { rl := &recordingLogger{} - tr := newOpsTracker() + tr := newOpsTracker(0) tr.SetOpsServices(OpsServices{Logger: rl}) // Expiry path: callback publishes a zero-byte error record. diff --git a/rdma/rcroutes/routes_linux.go b/rdma/rcroutes/routes_linux.go index d1e185dd..382beefe 100644 --- a/rdma/rcroutes/routes_linux.go +++ b/rdma/rcroutes/routes_linux.go @@ -118,10 +118,10 @@ func (h *Handler) Shutdown() { // operational services arrive later through SetOpsServices, once // the gateway has created them. func New(svc *rcserver.RCSvc, be backend.Backend, iam auth.IAMService, - readonly, disableACL bool) *Handler { + readonly, disableACL bool, sessionLimit int) *Handler { return &Handler{svc: svc, be: be, iam: iam, readonly: readonly, disableACL: disableACL, - ops: newOpsTracker()} + ops: newOpsTracker(sessionLimit)} } // principalID derives the session identity digest from the @@ -249,8 +249,15 @@ func (h *Handler) prepareCore(ctx fiber.Ctx) error { // an already-expired session and fire the teardown callback // synchronously, and a registered record (or a parked early // notification) keeps that publication from being lost. - h.ops.register(resp.SessionID, acct, - regionFromCtx(ctx), bucket, key, isPut, time.Now()) + // Refusal here is admission control: earlier sessions still + // hold unpublished audit records, so the new session is + // rejected before the native side commits it. + if err := h.ops.register(resp.SessionID, acct, + regionFromCtx(ctx), bucket, key, isPut, time.Now()); err != nil { + _ = h.svc.FinishPrepare(resp.SessionID, false) + h.ops.publishRequest(ctx, acct, err, bucket, key, isPut) + return s3err.GetAPIError(s3err.ErrSlowDown) + } if err := h.svc.FinishPrepare(resp.SessionID, true); err != nil { // The finalization failed. Exactly one publication // covers it: the finalizing call already reaped the diff --git a/rdma/rcroutes/routes_stub.go b/rdma/rcroutes/routes_stub.go index b735c47c..f3b5e03c 100644 --- a/rdma/rcroutes/routes_stub.go +++ b/rdma/rcroutes/routes_stub.go @@ -32,7 +32,7 @@ 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 { + readonly, disableACL bool, sessionLimit int) *Handler { return &Handler{} }