From 60bec2c00de78a862b7ed7b6f368ee319c3d8e2b Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sun, 6 Sep 2026 23:40:51 +0900 Subject: [PATCH] rdma: gate session admission on audit publication backlog The native side releases its session quota when it fires the teardown notification, before the audit record lands in a sink, so session turnover can queue more unpublished records than the live-session limit allows. Hold an admission credit per session from registration until its record is published, and refuse new sessions while the backlog of unpublished session records reaches the native session quota: the refusal rolls the prepare back, still publishes the request-level audit record, and answers SlowDown so the client retries. A stalled sink now turns into latency instead of unbounded memory. Count dropped request records under the publication mutex so the shutdown drop-count report cannot miss an increment racing it. --- cmd/vgwrdma/main.go | 2 +- rdma/rcroutes/ops_linux.go | 62 ++++++++++++++++++++++++++------- rdma/rcroutes/ops_linux_test.go | 14 ++++---- rdma/rcroutes/routes_linux.go | 15 +++++--- rdma/rcroutes/routes_stub.go | 2 +- 5 files changed, 69 insertions(+), 26 deletions(-) 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{} }