rdma: cap the session-less publication backlog

Bound the records a stalled sink can accumulate from requests
that never opened a session (failed authentications): beyond
4096 queued, dispatchOrDrop drops the record and counts it, and
shutdown reports the drop count once. Session publications stay
uncapped - each session publishes exactly once and the session
table has a hard limit, so their backlog is structurally bounded.

Cancel the metrics child context on constructor failure so a
malformed publisher endpoint does not leak the derived context
onto the parent.

metrics/metrics.go

@@ -113,6 +113,7 @@ func NewManager(ctx context.Context, conf Config) (Manager, error) {
 		for server := range statsdServers {
 			statsd, err := newStatsd(server, conf.ServiceName)
 			if err != nil {
+				mcancel()
 				return nil, err
 			}
 			mgr.publishers = append(mgr.publishers, statsd)
@@ -126,6 +127,10 @@ func NewManager(ctx context.Context, conf Config) (Manager, error) {
 		for server := range dogStatsdServers {
 			dogStatsd, err := newDogStatsd(server, conf.ServiceName)
 			if err != nil {
+				// The derived child context would otherwise stay
+				// attached to the parent until the parent is
+				// canceled.
+				mcancel()
 				return nil, err
 			}
 			mgr.publishers = append(mgr.publishers, dogStatsd)

rdma/rcroutes/ops_linux.go

@@ -19,9 +19,12 @@ package rcroutes
 
 import (
 	"errors"
+	"fmt"
 	"net/http"
+	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gofiber/fiber/v3"
@@ -279,19 +282,22 @@ type opsTracker struct {
 	// execute under pubmu - the worker detaches queued work and
 	// publishes outside the lock - so a slow sink delays records
 	// but never blocks a dispatcher.
-	pubmu     sync.Mutex
-	overflow  []pubJob
-	stopped   bool
-	done      chan struct{}
-	drain     chan struct{}
-	drainOnce sync.Once
+	pubmu      sync.Mutex
+	overflow   []pubJob
+	reqBacklog atomic.Int64
+	reqDropped atomic.Int64
+	stopped    bool
+	done       chan struct{}
+	drain      chan struct{}
+	drainOnce  sync.Once
 }
 
 // pubJob is one deferred publication handed to the worker.
 type pubJob struct {
-	emit *opsEmitter
-	err  error
-	byt  int64
+	emit  *opsEmitter
+	err   error
+	byt   int64
+	isReq bool
 }
 
 // pubQueueSoftCap is the buffered pre-allocation of the
@@ -300,6 +306,16 @@ type pubJob struct {
 // native callback.
 const pubQueueSoftCap = 256
 
+// pubRequestBacklogCap bounds the queued records that carry no
+// session. Session publications are structurally bounded (each
+// session publishes exactly once and the session table has a
+// hard limit), but request publications - failed authentications
+// - arrive with no session at all, and a stalled sink would let
+// them accumulate without limit. Beyond this depth the record is
+// dropped and counted, trading a bounded window of lost
+// request-audit records for memory safety under overload.
+const pubRequestBacklogCap = 4096
+
 // 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
@@ -329,7 +345,7 @@ func newOpsTracker() *opsTracker {
 				if !ok {
 					return
 				}
-				job.emit.publish(job.err, job.byt)
+				t.run(job)
 				// Service the overflow list after every
 				// channel job: bursts that exceed the
 				// buffer publish as soon as the sink
@@ -338,7 +354,7 @@ func newOpsTracker() *opsTracker {
 				// the lock and published outside it, so a
 				// slow sink never blocks a dispatcher.
 				for _, job := range t.takeOverflow() {
-					job.emit.publish(job.err, job.byt)
+					t.run(job)
 				}
 			case <-t.drain:
 				// Drain mode. The accept-vs-drain boundary:
@@ -360,7 +376,7 @@ func newOpsTracker() *opsTracker {
 						if !ok {
 							t.pubmu.Unlock()
 							for _, job := range pending {
-								job.emit.publish(job.err, job.byt)
+								t.run(job)
 							}
 							return
 						}
@@ -370,7 +386,7 @@ func newOpsTracker() *opsTracker {
 						t.overflow = nil
 						t.pubmu.Unlock()
 						for _, job := range pending {
-							job.emit.publish(job.err, job.byt)
+							t.run(job)
 						}
 						return
 					}
@@ -392,6 +408,15 @@ func (t *opsTracker) takeOverflow() []pubJob {
 	return pending
 }
 
+// run publishes one job and releases its request-backlog
+// reservation, if any.
+func (t *opsTracker) run(job pubJob) {
+	job.emit.publish(job.err, job.byt)
+	if job.isReq {
+		t.reqBacklog.Add(-1)
+	}
+}
+
 // 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.
@@ -404,6 +429,12 @@ func (t *opsTracker) Shutdown() {
 	t.drainOnce.Do(func() {
 		close(t.drain)
 		<-t.done
+		if n := t.reqDropped.Load(); n > 0 {
+			// Overload during shutdown: records without a session
+			// were dropped once the request backlog hit its cap.
+			// Surfaced once here rather than per record.
+			fmt.Fprintf(os.Stderr, "rdma-rc: dropped %d request audit records at the publication backlog cap\n", n)
+		}
 	})
 }
 
@@ -427,7 +458,7 @@ func (t *opsTracker) dispatch(job pubJob) {
 	t.pubmu.Lock()
 	if t.stopped {
 		t.pubmu.Unlock()
-		job.emit.publish(job.err, job.byt)
+		t.run(job)
 		return
 	}
 	select {
@@ -692,13 +723,24 @@ func (t *opsTracker) publishRequest(ctx fiber.Ctx, acct auth.Account,
 
 // dispatchOrDrop is dispatch with request-publication semantics:
 // after the worker stopped through the drain the job is dropped
-// instead of published inline.
+// instead of published inline, and the queued backlog of session-
+// less records is capped so a stalled sink cannot accumulate them
+// without bound.
 func (t *opsTracker) dispatchOrDrop(job pubJob) {
 	t.pubmu.Lock()
 	if t.stopped {
 		t.pubmu.Unlock()
 		return
 	}
+	if t.reqBacklog.Load() >= pubRequestBacklogCap {
+		t.pubmu.Unlock()
+		// Overload policy: drop and count. The record carries no
+		// session and no owner can reissue it.
+		t.reqDropped.Add(1)
+		return
+	}
+	job.isReq = true
+	t.reqBacklog.Add(1)
 	select {
 	case t.pubq <- job:
 		t.pubmu.Unlock()