rdma: bound publication backpressure and close the reservation window

Overflow publications ran one goroutine per job: a stalled sink
with a full queue grew them without bound (a probe reached a
thousand blocked calls). Overflow now runs inline under a bounded
semaphore, so at most a fixed number of callers wait and every
record still publishes.

The publication worker outlived the operational sinks: gateway
shutdown closed the RC service and then the sinks while
publications were still queued, losing terminal records (reached
"file already closed" with the real file logger). The shutdown
wrapper now drains the publication queue before closing the RC
service, and the route handler exposes the drain; late terminals
after the drain publish inline instead of queueing behind a
stopped worker.

The publication reservation happened after the transfer claim
returned: in the window between the claim and the reservation, a
concurrent READY's re-authorization denial consumed the unreserved
record, so the claimant's successful transfer lost its publication
to the denial. The READY handler now reserves before the claim and
before re-authorization; a rolled-back claim (wire failure, peer
busy) releases the reservation instead of publishing, so the
session keeps its record for the next claimant or the reaper.

The tracker test now joins the worker through the shutdown drain
and asserts per-session outcomes and byte counts instead of an
aggregate count that a pending fifth record could satisfy.

cmd/vgwrdma/main.go

@@ -1276,7 +1276,6 @@ func runGateway(ctx context.Context, be backend.Backend) error {
 		// rollback closure and the RunVersityGW lifecycle a
 		// single, ordered owner of both steps.
 		be = rdmamode.WrapBackendShutdownAfterRC(be, rcSvc)
-
 		rcVerify := middlewares.VerifyV4Signature(
 			middlewares.RootUserConfig{
 				Access: gwcli.RootUserAccess,
@@ -1288,6 +1287,13 @@ func runGateway(ctx context.Context, be backend.Backend) error {
 		// request reaches the route handler, while errors end the
 		// chain as usual.
 		rcH := rcroutes.New(rcSvc, be, iamSvc, readonly, disableACLs)
+		// 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
+		// the wrapped backend).
+		if w, ok := be.(*rdmamode.BackendShutdownAfterRC); ok {
+			w.SetOpsDrainer(rcH)
+		}
 		rcAuth := func(ctx fiber.Ctx) error {
 			// The RC routes run before the default-values
 			// middleware sets the request locals; the access

internal/rdmamode/rdmamode.go

@@ -21,6 +21,7 @@ import (
 	"math"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/versity/versitygw/backend"
@@ -124,6 +125,12 @@ func V2ValidationError(s V2Settings) string {
 // idempotent.
 type Closer interface{ Close() }
 
+// OpsDrainer drains operational publications (audit records,
+// events) that the RC teardown path queued before the RC service
+// closes, so nothing is left waiting on sinks that are about to
+// close. It is idempotent.
+type OpsDrainer interface{ Shutdown() }
+
 // 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
@@ -136,12 +143,28 @@ type Closer interface{ Close() }
 type BackendShutdownAfterRC struct {
 	backend.Backend
 	rc     Closer
+	ops    atomic.Pointer[OpsDrainer]
 	closed sync.Once
 }
 
-// Shutdown closes the RC service, then the wrapped backend, once.
+// SetOpsDrainer installs the operational publication drainer. The
+// route handler that owns the publications is built after this
+// wrapper (it needs the wrapped backend), so the drainer arrives
+// via this setter; installs after Shutdown ran are dropped, since
+// the drain window has passed.
+func (b *BackendShutdownAfterRC) SetOpsDrainer(d OpsDrainer) {
+	b.ops.Store(&d)
+}
+
+// Shutdown drains operational publications, closes the RC service,
+// then shuts the wrapped backend down, once. The publication drain
+// runs before the RC close: RC teardown itself queues publications,
+// so the queue must still be moving while the sessions drain.
 func (b *BackendShutdownAfterRC) Shutdown() {
 	b.closed.Do(func() {
+		if d := b.ops.Load(); d != nil {
+			(*d).Shutdown()
+		}
 		b.rc.Close()
 		b.Backend.Shutdown()
 	})

rdma/rcroutes/ops_linux.go

@@ -233,12 +233,15 @@ type sessionRecord struct {
 // guarantee that no record is silently dropped while still capping
 // how long a callback may wait.
 type opsTracker struct {
-	mu       sync.Mutex
-	ops      OpsServices
-	sessions map[string]*sessionRecord
-	app      *fiber.App
-	pubq     chan pubJob
-	done     chan struct{}
+	mu        sync.Mutex
+	ops       OpsServices
+	sessions  map[string]*sessionRecord
+	app       *fiber.App
+	pubq      chan pubJob
+	overflow  chan struct{}
+	done      chan struct{}
+	drain     chan struct{}
+	drainOnce sync.Once
 }
 
 // pubJob is one deferred publication handed to the worker.
@@ -249,49 +252,104 @@ type pubJob struct {
 }
 
 // pubQueueDepth bounds how many publications may wait in the
-// handoff queue before the callback falls back to inline
-// execution.
+// handoff queue before the caller falls back to inline execution.
 const pubQueueDepth = 256
 
+// pubOverflowSlots bounds how many callers may run overflow
+// publications inline at once; further callers block on the
+// semaphore until a slot frees.
+const pubOverflowSlots = 8
+
 func newOpsTracker() *opsTracker {
 	t := &opsTracker{
 		sessions: map[string]*sessionRecord{},
 		app: fiber.New(fiber.Config{
 			Immutable: true,
 		}),
-		pubq: make(chan pubJob, pubQueueDepth),
-		done: make(chan struct{}),
+		pubq:     make(chan pubJob, pubQueueDepth),
+		overflow: make(chan struct{}, pubOverflowSlots),
+		done:     make(chan struct{}),
+		drain:    make(chan struct{}),
 	}
 	go func() {
 		defer close(t.done)
-		for job := range t.pubq {
-			job.emit.publish(job.err, job.byt)
+		for {
+			select {
+			case job, ok := <-t.pubq:
+				if !ok {
+					return
+				}
+				job.emit.publish(job.err, job.byt)
+			case <-t.drain:
+				// Drain mode: empty whatever is already
+				// queued, then exit. Producers past this
+				// point publish inline.
+				for {
+					select {
+					case job, ok := <-t.pubq:
+						if !ok {
+							return
+						}
+						job.emit.publish(job.err, job.byt)
+					default:
+						return
+					}
+				}
+			}
 		}
 	}()
 	return t
 }
 
+// Shutdown drains pending publications and stops the worker. The
+// gateway must call this BEFORE closing the operational sinks: a
+// queued publication that runs after its logger closed is lost.
+// After Shutdown, dispatch publishes inline (the queue no longer
+// moves), so late terminals still record instead of vanishing.
+func (t *opsTracker) Shutdown() {
+	if t == nil {
+		return
+	}
+	t.drainOnce.Do(func() {
+		close(t.drain)
+		<-t.done
+	})
+}
+
 // dispatch hands a publication to the worker. It must never block
 // indefinitely: when the queue is full (the worker itself stuck in
 // a sink), the publication runs inline so the record is still
 // delivered and the caller - possibly the native reaper - returns.
+// dispatch hands a publication to the worker. It must never block
+// indefinitely: when the queue is full the caller runs the
+// publication itself under an overflow semaphore, which bounds how
+// many overflow publications may wait at once. Overflow callers
+// beyond the semaphore block - the alternative (one goroutine per
+// job) grows without bound under a stalled sink, and dropping the
+// record loses the publication entirely. The callers that reach
+// overflow are the native reaper or a request handler; waiting
+// there is bounded by the semaphore and by the worker draining,
+// and is the price of never losing a record.
 func (t *opsTracker) dispatch(job pubJob) {
+	// After the worker exited (shutdown drain), the queue no
+	// longer moves: publish inline so the record is not
+	// stranded behind a send nobody will receive.
+	select {
+	case <-t.done:
+		job.emit.publish(job.err, job.byt)
+		return
+	default:
+	}
 	select {
 	case t.pubq <- job:
 		return
 	default:
 	}
-	select {
-	case <-t.done:
-		// Worker exited (shutdown path): publish inline.
-		job.emit.publish(job.err, job.byt)
-		return
-	default:
-	}
-	// Queue full and worker alive but stalled. Rather than block
-	// the caller, drop the job onto a goroutine: publication order
-	// is not part of the contract, and dropping is worse.
-	go job.emit.publish(job.err, job.byt)
+	// Queue full and worker alive. Run inline under the overflow
+	// semaphore.
+	t.overflow <- struct{}{}
+	defer func() { <-t.overflow }()
+	job.emit.publish(job.err, job.byt)
 }
 
 // SetOpsServices installs the operational service instances. The
@@ -390,6 +448,22 @@ func (t *opsTracker) reserve(sessionID string) *opsEmitter {
 	return rec.emit
 }
 
+// releaseReservation returns a reserved record to the pool
+// without publishing: the transfer claim it was held for rolled
+// back, so the session lives on and the next claimant (another
+// READY, or the reaper) must still find an unreserved record.
+func (t *opsTracker) releaseReservation(sessionID string, emit *opsEmitter) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	rec, ok := t.sessions[sessionID]
+	if ok && rec.reserved && rec.emit == emit {
+		rec.reserved = false
+	}
+	t.mu.Unlock()
+}
+
 // publishReserved publishes through a reserved record and drops it:
 // the single publication of a request-owned session outcome.
 func (t *opsTracker) publishReserved(sessionID string, emit *opsEmitter, err error, bytes int64) {

rdma/rcroutes/ops_linux_test.go

@@ -231,27 +231,21 @@ func (r *recordingLogger) Log(ctx fiber.Ctx, err error, body []byte, meta s3log.
 func (r *recordingLogger) HangUp() error   { return nil }
 func (r *recordingLogger) Shutdown() error { return nil }
 
-func (r *recordingLogger) count() int {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	return len(r.logs)
-}
-
 // TestOpsTrackerPublishesExactlyOncePerSession drives the tracker
-// with a recording sink and asserts the published record count:
-// one per registered session no matter how the ownership played
-// out (callback expiry, reserved request path, denial).
+// with a recording sink, joins the publication worker through
+// Shutdown, and asserts the per-session record: each session
+// publishes exactly one record with its own outcome and bytes.
 func TestOpsTrackerPublishesExactlyOncePerSession(t *testing.T) {
 	rl := &recordingLogger{}
 	tr := newOpsTracker()
 	tr.SetOpsServices(OpsServices{Logger: rl})
 
-	// Expiry path: callback publishes.
+	// Expiry path: callback publishes a zero-byte error record.
 	tr.register("s-exp", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
 	tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-exp"})
 
 	// Reserved path: reserve, callback fires (skipped), the
-	// request path publishes.
+	// request path publishes success with bytes.
 	tr.register("s-res", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
 	emit := tr.reserve("s-res")
 	if emit == nil {
@@ -261,7 +255,7 @@ func TestOpsTrackerPublishesExactlyOncePerSession(t *testing.T) {
 	tr.publishReserved("s-res", emit, nil, 128)
 
 	// Denial path while reserved: failOutcome must not steal the
-	// publication; the request path still owns it.
+	// publication; the owner's success record is the only one.
 	tr.register("s-den", auth.Account{Access: "ak"}, "r", "b", "o", true, time.Now())
 	emit2 := tr.reserve("s-den")
 	if emit2 == nil {
@@ -270,16 +264,49 @@ func TestOpsTrackerPublishesExactlyOncePerSession(t *testing.T) {
 	tr.failOutcome("s-den", errors.New("denied"))
 	tr.publishReserved("s-den", emit2, nil, 256)
 
+	// Released reservation: the record returns to the pool and
+	// the reaper (or the next claimant) can still publish it.
+	tr.register("s-rel", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
+	emit3 := tr.reserve("s-rel")
+	if emit3 == nil {
+		t.Fatal("reserve failed")
+	}
+	tr.releaseReservation("s-rel", emit3)
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-rel"})
+
 	// Consume-or-noop denial of an unreserved session.
 	tr.register("s-fail", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
 	tr.failOutcome("s-fail", errors.New("x"))
 
-	// Let the publication worker drain.
-	deadline := time.Now().Add(2 * time.Second)
-	for rl.count() < 4 && time.Now().Before(deadline) {
-		time.Sleep(time.Millisecond)
+	// Join the worker: Shutdown drains everything queued and
+	// stops it, so counting after Shutdown sees the final state.
+	tr.Shutdown()
+
+	rl.mu.Lock()
+	defer rl.mu.Unlock()
+	if len(rl.logs) != 5 {
+		t.Fatalf("published %d records, want 5: %+v", len(rl.logs), rl.logs)
 	}
-	if got := rl.count(); got != 4 {
-		t.Fatalf("published %d records, want 4", got)
+	// The recording sink cannot see session IDs directly (they
+	// live in the synthesized context), so assert the observable
+	// contract: error/bytes pairings, one per session, in the
+	// dispatch order above.
+	type outcome struct {
+		isErr bool
+		bytes int64
+	}
+	want := []outcome{
+		{true, 0},    // s-exp expiry
+		{false, 128}, // s-res success
+		{false, 256}, // s-den success (denial was skipped)
+		{true, 0},    // s-rel expiry after release
+		{true, 0},    // s-fail denial
+	}
+	for i, w := range want {
+		got := rl.logs[i]
+		if (got.err != nil) != w.isErr || got.bytes != w.bytes {
+			t.Fatalf("record %d = (err=%v, bytes=%d), want (err=%v, bytes=%d)",
+				i, got.err, got.bytes, w.isErr, w.bytes)
+		}
 	}
 }

rdma/rcroutes/routes_linux.go

@@ -103,6 +103,17 @@ func (h *Handler) SetOpsServices(ops OpsServices) {
 	h.svc.SetTerminalNotify(h.ops.onTerminal)
 }
 
+// Shutdown drains pending operational publications and stops the
+// publication worker. Call before the operational sinks (audit
+// logger, metrics, events) close: a queued publication that runs
+// after its sink closed is lost.
+func (h *Handler) Shutdown() {
+	if h.ops == nil {
+		return
+	}
+	h.ops.Shutdown()
+}
+
 // New builds the route handler around a started RC service. The
 // operational services arrive later through SetOpsServices, once
 // the gateway has created them.
@@ -383,14 +394,22 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error {
 	if !ok {
 		return errors.New("invalid session target")
 	}
+	// Reserve the publication BEFORE the transfer claim and before
+	// re-authorization: once ReadyTransfer returns this handler
+	// holds the native completion reference, and a concurrent
+	// READY's denial (or the reaper) must not be able to consume
+	// the record in the window between the claim and the
+	// reservation. A reserved record is invisible to both.
+	emit := h.ops.reserve(sessionID)
+	publish := func(err error, bytes int64) {
+		h.ops.publishReserved(sessionID, emit, err, bytes)
+	}
 	if err := h.authorize(ctx, acct, isRoot, bucket, key, info.Op == 1); err != nil {
-		// Permission revoked mid-session: cancel the session,
-		// and publish the real denial - not an expiry - as its
-		// outcome. The cancel tears the session down now; a
-		// plain publish (consume-or-noop) would race it, so
-		// the record is consumed here.
+		// Permission revoked mid-session: publish the real
+		// denial - not an expiry - as the outcome, release the
+		// reservation, and cancel the session.
 		err = mapRcError(err)
-		h.ops.failOutcome(sessionID, err)
+		publish(err, 0)
 		_ = h.svc.Cancel(sessionID, principal)
 		return err
 	}
@@ -409,7 +428,11 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error {
 		// 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.
+		// transferring on. The publication reservation held
+		// the record for this claim; release it without
+		// publishing so the surviving path (the other READY,
+		// or the eventual reaper) still owns it.
+		h.ops.releaseReservation(sessionID, emit)
 		return mapRcError(err)
 	}
 
@@ -417,8 +440,11 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error {
 	// 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.
+	// so answer 409 without any finalizer. The reservation is
+	// released the same way: the session stays with the record
+	// the next READY (or the reaper) will claim.
 	if resp.Outcome == rcserver.ReadyBusy {
+		h.ops.releaseReservation(sessionID, emit)
 		return fmt.Errorf("peer busy: %w", rcserver.ErrDouble)
 	}
 
@@ -429,17 +455,16 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error {
 	// Every native completion call below (FinishFinal, and
 	// FinishPut inside commitPut) fires the teardown callback
 	// synchronously, BEFORE the call returns - so the outcome
-	// cannot be recorded after the call. The handler reserves the
-	// publication instead: a reserved record is invisible to the
-	// callback, and this handler publishes exactly once after the
-	// result is known. The deferred safety net publishes on any
-	// unwind that bypassed the normal paths.
-	emit := h.ops.reserve(sessionID)
+	// cannot be recorded after the call. The reservation was made
+	// before the transfer claim (above), so the callback is a
+	// no-op for this session and this handler publishes exactly
+	// once after the result is known. The deferred safety net
+	// publishes on any unwind that bypassed the normal paths.
 	published := false
-	publish := func(err error, bytes int64) {
+	doPublish := func(err error, bytes int64) {
 		if !published {
 			published = true
-			h.ops.publishReserved(sessionID, emit, err, bytes)
+			publish(err, bytes)
 		}
 	}
 	finalized := false
@@ -450,7 +475,7 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error {
 			// publication must come from here on a panic or
 			// early-unwind path.
 			if !published {
-				publish(errPanicked(), 0)
+				doPublish(errPanicked(), 0)
 			}
 			_ = h.svc.FinishFinal(sessionID)
 		}

rdma/rcroutes/routes_stub.go

@@ -64,3 +64,6 @@ func (h *Handler) SetOpsServices(ops OpsServices) {}
 
 // PublishAuthFailure is a stub mirror of the linux handler.
 func (h *Handler) PublishAuthFailure(ctx fiber.Ctx, err error) {}
+
+// Shutdown is a stub mirror of the linux handler.
+func (h *Handler) Shutdown() {}