From d30ace72ad90f3f2d25e9cdcb54f9d5afc08343c Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sun, 6 Sep 2026 12:12:53 +0900 Subject: [PATCH] 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 | 74 +++++++++++++++++++++++++++---- rdma/rcroutes/ops_linux_test.go | 78 +++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 8 deletions(-) diff --git a/rdma/rcroutes/ops_linux.go b/rdma/rcroutes/ops_linux.go index 0adf6630..9879df7c 100644 --- a/rdma/rcroutes/ops_linux.go +++ b/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 diff --git a/rdma/rcroutes/ops_linux_test.go b/rdma/rcroutes/ops_linux_test.go index a1ee154f..be33847e 100644 --- a/rdma/rcroutes/ops_linux_test.go +++ b/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) + } +}