diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index bbb9d963..a2db8939 100644 --- a/cmd/vgwrdma/main.go +++ b/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 diff --git a/internal/rdmamode/rdmamode.go b/internal/rdmamode/rdmamode.go index 92cad9fa..a4e861ea 100644 --- a/internal/rdmamode/rdmamode.go +++ b/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() }) diff --git a/rdma/rcroutes/ops_linux.go b/rdma/rcroutes/ops_linux.go index 9879df7c..f33b9b3f 100644 --- a/rdma/rcroutes/ops_linux.go +++ b/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) { diff --git a/rdma/rcroutes/ops_linux_test.go b/rdma/rcroutes/ops_linux_test.go index be33847e..5853773f 100644 --- a/rdma/rcroutes/ops_linux_test.go +++ b/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) + } } } diff --git a/rdma/rcroutes/routes_linux.go b/rdma/rcroutes/routes_linux.go index a120f0cc..252874d9 100644 --- a/rdma/rcroutes/routes_linux.go +++ b/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) } diff --git a/rdma/rcroutes/routes_stub.go b/rdma/rcroutes/routes_stub.go index 6a48adab..d9c4b1f0 100644 --- a/rdma/rcroutes/routes_stub.go +++ b/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() {}