diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 354be6e3..55128c6c 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -1300,16 +1300,14 @@ func runGateway(ctx context.Context, be backend.Backend) error { // logger and event schema read the region from // there, so set it for every verified request. utils.ContextKeyRegion.Set(ctx, region) - // Admission barrier: verification can block on IAM - // lookups, and its failure publications must not - // outlive the RC shutdown sequence (sinks close - // right after the RC service drains). Requests that - // lose the race to shutdown are dropped - their - // audit record would land after the sinks closed. - if !rcSvc.TryEnter() { - return rcroutes.WriteRouteError(ctx, rcroutes.ErrNotAdmitted()) - } - defer rcSvc.Leave() + // Verification runs outside the admission barrier: + // signature checks can block on IAM lookups that + // carry no cancellation, and holding the barrier + // across them would let one stalled lookup defer + // RC shutdown indefinitely. The handlers enforce + // admission themselves; a failure publication + // produced here checks the drain state before + // dispatching, so it cannot outlive the sinks. if err := rcVerify(ctx); err != nil { rcH.PublishAuthFailure(ctx, err) return rcroutes.WriteRouteError(ctx, err) diff --git a/metrics/metrics.go b/metrics/metrics.go index 0d725dba..2b538b1f 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -22,6 +22,7 @@ import ( "os" "strings" "sync" + "sync/atomic" "github.com/gofiber/fiber/v3" "github.com/versity/versitygw/s3err" @@ -62,6 +63,12 @@ type manager struct { publishers []publisher addDataChan chan datapoint + // closed gates senders against Close: the datapoint channel + // is closed to drain the forwarder, and a send on a closed + // channel panics. Producers that lose this race (an S3 + // handler still finishing after the shutdown timeout) drop + // their update instead of taking the process down. + closed atomic.Bool } type Config struct { @@ -214,7 +221,7 @@ func (m *manager) increment(key string, tags ...Tag) { // add adds value to key func (m *manager) add(key string, value int64, tags ...Tag) { - if m.ctx.Err() != nil { + if m.ctx.Err() != nil || m.closed.Load() { return } @@ -224,6 +231,13 @@ func (m *manager) add(key string, value int64, tags ...Tag) { tags: tags, } + // The send races Close for last-producer position: the + // closed check above and the channel close in Close are not + // atomic, so the send below can still observe a closed + // channel. Recovering here turns that race into a dropped + // datapoint, which is the documented contract for late + // producers. + defer func() { _ = recover() }() select { case m.addDataChan <- d: default: @@ -233,6 +247,11 @@ func (m *manager) add(key string, value int64, tags ...Tag) { // Close closes metrics channels, waits for data to complete, closes all plugins func (m *manager) Close() { + // Stop accepting new datapoints before closing the channel: + // producers check the flag and drop their update, so only a + // producer already between the check and the send can race, + // and that producer recovers instead of panicking. + m.closed.Store(true) // drain the datapoint channels close(m.addDataChan) m.wg.Wait() diff --git a/rdma/rcroutes/ops_linux.go b/rdma/rcroutes/ops_linux.go index 8f23b527..5c5a1e15 100644 --- a/rdma/rcroutes/ops_linux.go +++ b/rdma/rcroutes/ops_linux.go @@ -263,11 +263,18 @@ 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 + mu sync.Mutex + ops OpsServices + sessions map[string]*sessionRecord + app *fiber.App + pubq chan pubJob + // overflow holds publications that arrived while the queue + // buffer was full. A native callback must never wait on a + // slow sink, so dispatch appends here (under pubmu) instead + // of blocking or running the sink itself, and the worker + // drains this list after the channel empties. + pubmu sync.Mutex + overflow []pubJob done chan struct{} drain chan struct{} drainOnce sync.Once @@ -280,28 +287,30 @@ type pubJob struct { byt int64 } -// pubQueueCapacity returns the publication queue depth for a -// given session limit. Each session publishes exactly one -// terminal record (enforced by the reservation protocol), and the -// session count never exceeds the configured limit, so a queue -// this deep can never fill - native callbacks always hand off -// without waiting. A floor keeps tests and degenerate zero -// configs sane. -func pubQueueCapacity(maxSessions uint32) int { - n := int(maxSessions) + 1 // +1: shutdown drain may race one last dispatch - if n < 64 { - n = 64 - } - return n -} +// pubQueueSoftCap is the buffered pre-allocation of the +// publication queue, not a bound: the overflow list in dispatch +// holds whatever exceeds it, so a slow sink never blocks a +// native callback. +const pubQueueSoftCap = 256 -func newOpsTracker(queueDepth int) *opsTracker { +// newOpsTracker builds the tracker. The publication queue is +// conceptually unbounded: a callback thread must never run a +// sink (a blocked sink would stall the native reaper and defer +// RC shutdown), so dispatch always hands off without waiting, +// whatever the backlog. Capacity accounting cannot bound the +// backlog - queued records accumulate across successive sessions +// and authentication failures consume no session at all - so the +// worker is the only sink executor and the queue absorbs +// 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 { t := &opsTracker{ sessions: map[string]*sessionRecord{}, app: fiber.New(fiber.Config{ Immutable: true, }), - pubq: make(chan pubJob, queueDepth), + pubq: make(chan pubJob, pubQueueSoftCap), done: make(chan struct{}), drain: make(chan struct{}), } @@ -315,9 +324,13 @@ func newOpsTracker(queueDepth int) *opsTracker { } 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. + // Drain mode: empty the channel, then the + // overflow list, then exit. Producers past + // this point append to overflow; the final + // sweep below runs only what arrived before + // the drain signal, and a producer racing + // the sweep re-enqueues through dispatch's + // post-drain path. for { select { case job, ok := <-t.pubq: @@ -326,6 +339,13 @@ func newOpsTracker(queueDepth int) *opsTracker { } job.emit.publish(job.err, job.byt) default: + t.pubmu.Lock() + pending := t.overflow + t.overflow = nil + t.pubmu.Unlock() + for _, job := range pending { + job.emit.publish(job.err, job.byt) + } return } } @@ -351,12 +371,17 @@ func (t *opsTracker) Shutdown() { } // dispatch hands a publication to the worker without ever -// blocking the caller. The queue is sized to the session limit -// and each session publishes exactly once, so the send below -// always has room while the worker runs; after the worker exits -// (shutdown drain) the publication runs inline, because the queue -// no longer moves. Either way the native callback - which may be -// the reaper - returns promptly and the record is never lost. +// blocking the caller or running a sink on the calling thread: +// the native reaper invokes terminal callbacks, and an +// operational sink can block indefinitely, which must never +// stall reaping or RC shutdown. The channel buffer absorbs the +// common case; when it is full the job goes to the overflow +// list, which the worker drains after the channel. After the +// worker exits (shutdown drain), a session-terminal job is +// published inline - its producer (Close, after quiescing +// native producers) is not a native callback - while request +// publications are dropped by publishRequest before reaching +// here. func (t *opsTracker) dispatch(job pubJob) { select { case <-t.done: @@ -367,10 +392,9 @@ func (t *opsTracker) dispatch(job pubJob) { select { case t.pubq <- job: default: - // Unreachable while the capacity contract holds; kept - // as a safety net so a mis-sized queue degrades to - // inline publication instead of blocking the reaper. - job.emit.publish(job.err, job.byt) + t.pubmu.Lock() + t.overflow = append(t.overflow, job) + t.pubmu.Unlock() } } @@ -597,11 +621,22 @@ func expiredError(ev rcserver.TerminalEvent) error { // publishRequest emits an operation record for a request that ended // before any session existed (authentication, authorization, or // header failures): no tracking table entry, single emission. +// These requests run outside the admission barrier (verification +// may block on uncancellable IAM lookups), so a record produced +// after the shutdown drain began is dropped rather than published +// into closed sinks. func (t *opsTracker) publishRequest(ctx fiber.Ctx, acct auth.Account, err error, bucket, key string, isPut bool) { if t == nil { return } + select { + case <-t.done: + // The worker already exited through the drain; sinks + // are closing. Drop the record. + return + default: + } acct.Access = strings.Clone(acct.Access) emit := &opsEmitter{ ops: t.loadOps(), diff --git a/rdma/rcroutes/ops_linux_test.go b/rdma/rcroutes/ops_linux_test.go index 12f9dc60..09a2c3d2 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(pubQueueCapacity(0)) + tr := newOpsTracker() 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(pubQueueCapacity(0)) + tr := newOpsTracker() 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(pubQueueCapacity(0)) + tr := newOpsTracker() 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(pubQueueCapacity(0)) + tr := newOpsTracker() 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(pubQueueCapacity(0)) + tr := newOpsTracker() 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(pubQueueCapacity(0)) + tr := newOpsTracker() // 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(pubQueueCapacity(0)) + tr := newOpsTracker() tr.SetOpsServices(OpsServices{Logger: rl}) // Expiry path: callback publishes a zero-byte error record. @@ -307,18 +307,23 @@ func TestOpsTrackerPublishesExactlyOncePerSession(t *testing.T) { if rsvO == nil { t.Fatal("reserve failed") } - // Stale reservation: released, re-reserved by another claim, - // then the stale token tries to publish. The generation check - // must reject the stale token while the current owner still + // Stale generation: the original owner releases, another + // claim re-reserves, and then the ORIGINAL token tries both + // release and publish. The generation check must reject the + // stale token on both paths while the current owner still // publishes. - rsvO2 := tr.reserve("s-own") + rsvO2 := tr.reserve("s-own") // refused: still reserved by rsvO + if rsvO2 != nil { + t.Fatal("double reserve succeeded") + } tr.releaseReservation("s-own", rsvO) rsvB := tr.reserve("s-own") if rsvB == nil { t.Fatal("re-reserve after release failed") } - tr.publishReserved("s-own", rsvO2, nil, 999) // stale: no-op - tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-own"}) + tr.publishReserved("s-own", rsvO, nil, 999) // stale: no-op + tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-own"}) // stashes under rsvB + tr.releaseReservation("s-own", rsvO) // stale: no-op, keeps rsvB tr.publishReserved("s-own", rsvB, nil, 32) // Consume-or-noop denial of an unreserved session. diff --git a/rdma/rcroutes/routes_linux.go b/rdma/rcroutes/routes_linux.go index d25b5164..d1e185dd 100644 --- a/rdma/rcroutes/routes_linux.go +++ b/rdma/rcroutes/routes_linux.go @@ -121,7 +121,7 @@ func New(svc *rcserver.RCSvc, be backend.Backend, iam auth.IAMService, readonly, disableACL bool) *Handler { return &Handler{svc: svc, be: be, iam: iam, readonly: readonly, disableACL: disableACL, - ops: newOpsTracker(pubQueueCapacity(svc.MaxSessions()))} + ops: newOpsTracker()} } // principalID derives the session identity digest from the diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index 6a2fef27..aad73e0c 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/rdma/rcserver/rcserver_linux.go @@ -174,13 +174,12 @@ type SessionInfo struct { // unblock), waits for every entered call to leave, then tears the // server down; it is idempotent and safe from any goroutine. type RCSvc struct { - srv *C.rc_server - closing atomic.Bool - ops atomic.Int64 - once sync.Once - ctx context.Context - cancel context.CancelFunc - maxSessions uint32 + srv *C.rc_server + closing atomic.Bool + ops atomic.Int64 + once sync.Once + ctx context.Context + cancel context.CancelFunc } // Context returns the service-lifetime context. Handlers bind @@ -189,14 +188,6 @@ func (s *RCSvc) Context() context.Context { return s.ctx } -// MaxSessions reports the configured global session limit. The -// operational publication pipeline sizes its queue against it: -// each session publishes exactly one terminal record, so a queue -// this deep can never fill. -func (s *RCSvc) MaxSessions() uint32 { - return s.maxSessions -} - // rcLogSink receives every diagnostic line the C server emits. // It is stateless and process-global on purpose: the sink must be // valid from init through destroy, independent of any single @@ -272,7 +263,7 @@ func Init(opts DeviceOpts) (*RCSvc, error) { } installLogSink(srv, opts.Debug) ctx, cancel := context.WithCancel(context.Background()) - return &RCSvc{srv: srv, ctx: ctx, cancel: cancel, maxSessions: opts.MaxSessions}, nil + return &RCSvc{srv: srv, ctx: ctx, cancel: cancel}, nil } // TryEnter admits a request into the service. It returns false once