rdma: keep reaper callbacks off sink threads

The teardown callback ran audit, metrics, and event sinks inline on
whatever thread fired it, which is the native reaper thread: one
blocking sink (a synchronous file write on a stalled filesystem)
would stall reaping for every other session. Publications now hand
off to a dedicated worker through a bounded queue; a full queue
falls back to a detached goroutine, so the callback never waits on
a sink and no record is dropped.

A denial issued while a completion owner holds the reservation
(concurrent READY re-authorization failure) no longer steals the
publication: reserved records are invisible to the denial path,
which previously produced a denial record plus the owner's success
record for one transfer.

The tracker tests now drive a recording audit sink and assert the
published record count across expiry, reserved, denied-while-
reserved, and consume-or-noop paths: one record per session.

rdma/rcroutes/ops_linux.go

@@ -223,20 +223,75 @@ type sessionRecord struct {
 // single publisher. This removes every ownership race: a recorded
 // outcome cannot be double-published, and a record the callback
 // already consumed cannot be resurrected.
+//
+// Sink execution never runs on the caller's thread: the native
+// reaper invokes the callback, and an operational sink can block
+// (a synchronous file write on a stalled filesystem), which would
+// stall reaping for every other session. Publications hand off to
+// a dedicated worker through a bounded queue; when the queue is
+// full the publication runs inline as a last resort, keeping the
+// 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{}
 }
 
+// pubJob is one deferred publication handed to the worker.
+type pubJob struct {
+	emit *opsEmitter
+	err  error
+	byt  int64
+}
+
+// pubQueueDepth bounds how many publications may wait in the
+// handoff queue before the callback falls back to inline
+// execution.
+const pubQueueDepth = 256
+
 func newOpsTracker() *opsTracker {
-	return &opsTracker{
+	t := &opsTracker{
 		sessions: map[string]*sessionRecord{},
 		app: fiber.New(fiber.Config{
 			Immutable: true,
 		}),
+		pubq: make(chan pubJob, pubQueueDepth),
+		done: make(chan struct{}),
 	}
+	go func() {
+		defer close(t.done)
+		for job := range t.pubq {
+			job.emit.publish(job.err, job.byt)
+		}
+	}()
+	return t
+}
+
+// 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.
+func (t *opsTracker) dispatch(job pubJob) {
+	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)
 }
 
 // SetOpsServices installs the operational service instances. The
@@ -296,21 +351,25 @@ func (t *opsTracker) unregister(sessionID string) {
 // the finalizing call already reaped the session its callback
 // published (the entry is gone, this is a no-op); when no callback
 // will ever come (the native side rejected the call) the entry is
-// consumed and published here.
+// consumed and published here. A reserved record belongs to an
+// in-flight completion owner (a concurrent READY's denial must not
+// steal its publication), so it is left untouched.
 func (t *opsTracker) failOutcome(sessionID string, err error) {
 	if t == nil {
 		return
 	}
 	t.mu.Lock()
 	rec, ok := t.sessions[sessionID]
-	if ok {
+	if ok && !rec.reserved {
 		delete(t.sessions, sessionID)
+	} else {
+		ok = false
 	}
 	t.mu.Unlock()
 	if !ok {
 		return
 	}
-	rec.emit.publish(err, 0)
+	t.dispatch(pubJob{emit: rec.emit, err: err})
 }
 
 // reserve marks a session record as owned by its request path: the
@@ -341,7 +400,7 @@ func (t *opsTracker) publishReserved(sessionID string, emit *opsEmitter, err err
 	delete(t.sessions, sessionID)
 	t.mu.Unlock()
 	if emit != nil {
-		emit.publish(err, bytes)
+		t.dispatch(pubJob{emit: emit, err: err, byt: bytes})
 	}
 }
 
@@ -383,8 +442,7 @@ func (t *opsTracker) onTerminal(ev rcserver.TerminalEvent) {
 		// outcome carries the native reason.
 		rec.out = sessionOutcome{err: expiredError(ev), done: true}
 	}
-	emit := rec.emit
-	emit.publish(rec.out.err, rec.out.byt)
+	t.dispatch(pubJob{emit: rec.emit, err: rec.out.err, byt: rec.out.byt})
 }
 
 // expiredError renders an unclaimed teardown as the error the
@@ -424,7 +482,7 @@ func (t *opsTracker) publishRequest(ctx fiber.Ctx, acct auth.Account,
 		isPut:  isPut,
 		start:  time.Now(),
 	}
-	emit.publish(err, 0)
+	t.dispatch(pubJob{emit: emit, err: err})
 }
 
 // regionFromCtx reads the region the gateway middleware stored on

rdma/rcroutes/ops_linux_test.go

@@ -19,12 +19,16 @@ package rcroutes
 
 import (
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/gofiber/fiber/v3"
+
 	"github.com/versity/versitygw/auth"
 	"github.com/versity/versitygw/rdma/rcserver"
 	"github.com/versity/versitygw/s3err"
+	"github.com/versity/versitygw/s3log"
 )
 
 // The publication model: the request path reserves a session
@@ -205,3 +209,77 @@ func TestExpiredErrorClassification(t *testing.T) {
 		}
 	}
 }
+
+// recordingLogger captures audit publications so tests can assert
+// what the sinks actually received.
+type recordingLogger struct {
+	mu   sync.Mutex
+	logs []recLog
+}
+
+type recLog struct {
+	err   error
+	bytes int64
+}
+
+func (r *recordingLogger) Log(ctx fiber.Ctx, err error, body []byte, meta s3log.LogMeta) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.logs = append(r.logs, recLog{err: err, bytes: meta.ObjectSize})
+}
+
+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).
+func TestOpsTrackerPublishesExactlyOncePerSession(t *testing.T) {
+	rl := &recordingLogger{}
+	tr := newOpsTracker()
+	tr.SetOpsServices(OpsServices{Logger: rl})
+
+	// Expiry path: callback publishes.
+	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.
+	tr.register("s-res", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
+	emit := tr.reserve("s-res")
+	if emit == nil {
+		t.Fatal("reserve failed")
+	}
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-res"})
+	tr.publishReserved("s-res", emit, nil, 128)
+
+	// Denial path while reserved: failOutcome must not steal the
+	// publication; the request path still owns it.
+	tr.register("s-den", auth.Account{Access: "ak"}, "r", "b", "o", true, time.Now())
+	emit2 := tr.reserve("s-den")
+	if emit2 == nil {
+		t.Fatal("reserve failed")
+	}
+	tr.failOutcome("s-den", errors.New("denied"))
+	tr.publishReserved("s-den", emit2, nil, 256)
+
+	// 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)
+	}
+	if got := rl.count(); got != 4 {
+		t.Fatalf("published %d records, want 4", got)
+	}
+}