Merge pull request #2366 from potatogim/rc-parity-pr4

rdma: publish RC transfer outcomes into the gateway operational services

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,
@@ -1287,13 +1286,45 @@ func runGateway(ctx context.Context, be backend.Backend) error {
 		// nil on success without doing so. Wrap it so a verified
 		// request reaches the route handler, while errors end the
 		// chain as usual.
+		rcH := rcroutes.New(rcSvc, be, iamSvc, readonly, disableACLs, int(rcMaxSessions))
+		// 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
+			// logger and event schema read the region from
+			// there, so set it for every verified request.
+			utils.ContextKeyRegion.Set(ctx, region)
+			// 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)
 			}
 			return ctx.Next()
 		}
-		rcH := rcroutes.New(rcSvc, be, iamSvc, readonly, disableACLs)
+		// The gateway builds the access logger, metrics manager,
+		// and event sender inside RunVersityGW; hand them to the
+		// RC routes as soon as they exist so finished transfers
+		// publish into them.
+		cfg.OnServicesReady = func(s embedgw.OpsServices) {
+			rcH.SetOpsServices(rcroutes.OpsServices{
+				Logger:  s.Logger,
+				Metrics: s.Metrics,
+				Events:  s.Events,
+			})
+		}
 		cfg.S3Options = append(s3Opts,
 			s3api.WithRoute("POST", "/.hipobj-rc/prepare", rcAuth, rcH.Prepare),
 			s3api.WithRoute("POST", "/.hipobj-rc/ready", rcAuth, rcH.Ready),

cuwrapper/rc/rc_server_abi.cpp

@@ -101,6 +101,11 @@ struct rc_server {
    * races an in-flight sink pointer swap. */
   rc_log_fn log_fn = nullptr;
   void *log_ctx = nullptr;
+  /* Terminal notification sink: same lifetime contract as log_fn
+   * (installed once at init, cleared by destroy after the reaper
+   * joined). Fired by reapSession with no lock held. */
+  rc_terminal_fn term_fn = nullptr;
+  void *term_ctx = nullptr;
   std::atomic<uint64_t> epoch_counter{1};
   /* resource accounting (global buckets; per-principal map). */
   std::mutex acct_mtx;
@@ -253,6 +258,13 @@ void reapSession(rc_server *srv, RcSession *s) {
    * is actually gone: a surviving QP still holds the device. */
   if (destroyed) hipObj::v2::releaseDevice(srv->device);
   limitsRelease(srv, s->principal, s->staging_len);
+  /* Terminal notification: the sink runs after every server-side
+   * bookkeeping above so it observes the session as fully gone,
+   * and no lock is held here per the callback contract. */
+  rc_terminal_fn tfn = srv->term_fn;
+  if (tfn)
+    tfn(srv->term_ctx, s->core.id.c_str(), s->last_outcome,
+        (uint64_t)s->staging_len);
 }
 
 /* Runs the reap pass: sessions marked reap_pending (or in the
@@ -301,6 +313,13 @@ void rc_server_set_log_sink(rc_server *srv, rc_log_fn fn, void *ctx) {
   srv->log_ctx = ctx;
 }
 
+void rc_server_set_terminal_notify(rc_server *srv, rc_terminal_fn fn,
+                                   void *ctx) {
+  if (!srv) return;
+  srv->term_fn = fn;
+  srv->term_ctx = ctx;
+}
+
 int rc_server_init(const rc_device_opts *opts, rc_server **out) {
   if (!opts || !out) return RC_E_ARG;
   if (!hipObj::ibv.ensureLoaded()) {
@@ -860,6 +879,7 @@ int rc_ready_transfer(rc_server *srv, const rc_ready_req *req,
                                       /*destLid*/ 0, destGid,
                                       s->core.clientPsn) != 0) {
     g.lock();
+    s->last_outcome = RC_READY_WIRE_FAIL;
     s->reap_pending = true;
     s->active_ref = 0; /* roll the completion ref back: no data
                         * phase will run for this session */
@@ -868,6 +888,7 @@ int rc_ready_transfer(rc_server *srv, const rc_ready_req *req,
   if (hipObj::v2::transitionQpToRtsV2(conn, srv->device,
                                       s->core.serverPsn) != 0) {
     g.lock();
+    s->last_outcome = RC_READY_WIRE_FAIL;
     s->reap_pending = true;
     s->active_ref = 0;
     return RC_E_WIRE;
@@ -949,7 +970,9 @@ int rc_ready_transfer(rc_server *srv, const rc_ready_req *req,
           if (after->active_ref > 0) after->active_ref--;
         } else {
           /* Cannot re-arm the QP (or the session died mid-reset):
-           * not retryable. */
+           * not retryable. Record the wire failure so the
+           * teardown publication classifies it as one. */
+          after->last_outcome = RC_READY_WIRE_FAIL;
           after->reap_pending = true;
           if (after->active_ref > 0) after->active_ref--;
           return RC_E_WIRE;
@@ -962,6 +985,10 @@ int rc_ready_transfer(rc_server *srv, const rc_ready_req *req,
       after->active_ref = 0;
       return RC_E_WIRE;
     case hipObj::v2::DataPhaseResult::VerifyFail:
+      after->last_outcome = RC_READY_VERIFY_FAIL;
+      after->reap_pending = true;
+      after->active_ref = 0;
+      return RC_E_WIRE;
     case hipObj::v2::DataPhaseResult::WireFail:
       after->last_outcome = RC_READY_WIRE_FAIL;
       after->reap_pending = true;

cuwrapper/rc/rc_server_abi.h

@@ -52,6 +52,21 @@ typedef struct rc_server rc_server;
 
 void rc_server_set_log_sink(rc_server *srv, rc_log_fn fn, void *ctx);
 
+/* Terminal session notification. Fired exactly once per session
+ * when the reaper destroys it (expiry, CANCEL, or destroy),
+ * carrying the last outcome observed for the session. `id` is
+ * only valid for the duration of the call. Called with no server
+ * lock held; the sink must return promptly and must not call back
+ * into the server. Sessions that end inside a READY/FinishPut
+ * handler still fire this notification after the handler's own
+ * terminal bookkeeping, so the sink can treat it as the single
+ * authoritative "the session is gone" signal. */
+typedef void (*rc_terminal_fn)(void *ctx, const char *id, int outcome,
+                               uint64_t bytes);
+
+void rc_server_set_terminal_notify(rc_server *srv, rc_terminal_fn fn,
+                                   void *ctx);
+
 /* Device selection: matching GID prefix when gid_hint is set,
  * otherwise the first verbs device. */
 typedef struct {

embedgw/embedgw.go

@@ -520,6 +520,13 @@ type Config struct {
 	// as request middleware.
 	S3Options []s3api.Option
 
+	// OnServicesReady runs after the operational services (access
+	// logger, metrics manager, event sender) are created and before
+	// the S3 server is built, so embedders can wire them into
+	// components constructed earlier (such as the RDMA control
+	// routes). A nil callback is skipped.
+	OnServicesReady func(OpsServices)
+
 	// Version, Build, and BuildTime are displayed in the startup banner.
 	// All three are optional; omit or leave empty to suppress the field.
 	Version   string
@@ -527,6 +534,14 @@ type Config struct {
 	BuildTime string
 }
 
+// OpsServices bundles the operational service instances handed to
+// the Config.OnServicesReady callback.
+type OpsServices struct {
+	Logger  s3log.AuditLogger
+	Metrics metrics.Manager
+	Events  s3event.S3EventSender
+}
+
 // TODO: remove gatewayRunning once package-level globals (bucket-name
 // validation, debug logging) are eliminated and concurrent calls are safe.
 var gatewayRunning atomic.Bool
@@ -784,7 +799,16 @@ func RunVersityGW(ctx context.Context, be backend.Backend, cfg *Config) error {
 		return fmt.Errorf("setup logger: %w", err)
 	}
 
-	metricsManager, err := metrics.NewManager(ctx, metrics.Config{
+	// The metrics manager must outlive the gateway context: RC
+	// teardown publications drain during backend shutdown, after
+	// this context is cancelled. A manager bound to ctx would
+	// silently discard those final datapoints, so it runs on its
+	// own context and closes with the other sinks below.
+	metricsCtx, metricsStop := context.WithCancel(context.Background())
+	// The cancel runs when this function returns - after the
+	// shutdown sequence below finishes draining every sink.
+	defer metricsStop()
+	metricsManager, err := metrics.NewManager(metricsCtx, metrics.Config{
 		ServiceName:      cfg.MetricsService,
 		StatsdServers:    cfg.StatsdServers,
 		DogStatsdServers: cfg.DogstatsServers,
@@ -853,6 +877,14 @@ func RunVersityGW(ctx context.Context, be backend.Backend, cfg *Config) error {
 		}))
 	}
 
+	if cfg.OnServicesReady != nil {
+		cfg.OnServicesReady(OpsServices{
+			Logger:  loggers.S3Logger,
+			Metrics: metricsManager,
+			Events:  evSender,
+		})
+	}
+
 	srv, err := s3api.New(be, middlewares.RootUserConfig{
 		Access: cfg.RootUserAccess,
 		Secret: cfg.RootUserSecret,
@@ -1211,6 +1243,13 @@ Loop:
 	}
 
 	if metricsManager != nil {
+		// Cancel the metrics context first: the forwarder exits
+		// through it, draining the buffered datapoints, and
+		// Close then only waits for the forwarder and closes the
+		// publishers. The channel itself never closes, so late
+		// producers (a handler outliving the HTTP shutdown
+		// timeout) drop their datapoint instead of panicking.
+		metricsStop()
 		metricsManager.Close()
 	}
 

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,13 +143,31 @@ 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 closes the RC service, drains operational publications,
+// then shuts the wrapped backend down, once. The drain runs AFTER
+// the RC close: closing RC quiesces the native reaper (every
+// teardown callback has returned by the time Close returns), so no
+// producer can enqueue behind the drain - enqueue-then-worker-exit
+// stranding is impossible by ordering rather than by locking.
 func (b *BackendShutdownAfterRC) Shutdown() {
 	b.closed.Do(func() {
 		b.rc.Close()
+		if d := b.ops.Load(); d != nil {
+			(*d).Shutdown()
+		}
 		b.Backend.Shutdown()
 	})
 }

metrics/metrics.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/gofiber/fiber/v3"
 	"github.com/versity/versitygw/s3err"
@@ -44,18 +45,31 @@ type Tag struct {
 // Manager is the interface definition for metrics manager
 type Manager interface {
 	Send(ctx fiber.Ctx, err error, action string, count int64, status int)
+	// SendWithBucket is Send with the bucket dimension stated
+	// by the caller. The S3 middleware derives the bucket from
+	// the matched route, which synthesized contexts (RDMA
+	// operational records) cannot reproduce: they pass the
+	// captured bucket explicitly instead.
+	SendWithBucket(ctx fiber.Ctx, err error, action string, count int64, status int, bucket string)
 	Close()
 }
 
 // manager is a manager of metrics plugins
 type manager struct {
-	wg  sync.WaitGroup
-	ctx context.Context
+	wg     sync.WaitGroup
+	ctx    context.Context
+	cancel context.CancelFunc
 
 	config Config
 
 	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 {
@@ -80,9 +94,15 @@ func NewManager(ctx context.Context, conf Config) (Manager, error) {
 
 	addDataChan := make(chan datapoint, dataItemCount)
 
+	// Derive a cancellable child of the caller context: closing
+	// the manager cancels it itself (a standalone user of the
+	// API has no external cancellation to rely on), while the
+	// gateway shutdown path keeps its own context propagation.
+	mctx, mcancel := context.WithCancel(ctx)
 	mgr := &manager{
 		addDataChan: addDataChan,
-		ctx:         ctx,
+		ctx:         mctx,
+		cancel:      mcancel,
 		config:      conf,
 	}
 
@@ -93,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)
@@ -106,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)
@@ -137,6 +162,28 @@ func (m *manager) Send(ctx fiber.Ctx, err error, action string, count int64, sta
 		reqTags = append(reqTags, Tag{Key: "bucket", Value: bucket})
 	}
 
+	m.send(ctx, err, action, count, status, reqTags)
+}
+
+// SendWithBucket reports with the bucket dimension supplied by the
+// caller; see the Manager interface.
+func (m *manager) SendWithBucket(ctx fiber.Ctx, err error, action string, count int64, status int, bucket string) {
+	if action == "" {
+		action = ActionUndetected
+	}
+	a := ActionMap[action]
+	reqTags := []Tag{
+		{Key: "method", Value: ctx.Method()},
+		{Key: "api", Value: a.Service},
+		{Key: "action", Value: a.Name},
+	}
+	if bucket != "" {
+		reqTags = append(reqTags, Tag{Key: "bucket", Value: bucket})
+	}
+	m.send(ctx, err, action, count, status, reqTags)
+}
+
+func (m *manager) send(ctx fiber.Ctx, err error, action string, count int64, status int, reqTags []Tag) {
 	reqStatus := status
 
 	if err != nil {
@@ -186,7 +233,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
 	}
 
@@ -196,6 +243,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:
@@ -203,10 +257,17 @@ func (m *manager) add(key string, value int64, tags ...Tag) {
 	}
 }
 
-// Close closes metrics channels, waits for data to complete, closes all plugins
+// Close stops the manager: producers drop new datapoints, the
+// forwarder drains the buffered ones and exits through the
+// canceled context, and the publishers flush and close. The
+// datapoint channel itself is never closed - a producer racing
+// the closure would panic - so the closed flag and the context
+// cancellation carry the shutdown instead.
 func (m *manager) Close() {
-	// drain the datapoint channels
-	close(m.addDataChan)
+	m.closed.Store(true)
+	// Self-owned cancellation terminates the forwarder wherever
+	// it is waiting; the external context is only a second path.
+	m.cancel()
 	m.wg.Wait()
 
 	// close all publishers
@@ -222,12 +283,36 @@ type publisher interface {
 }
 
 func (m *manager) addForwarder(addChan <-chan datapoint) {
-	for data := range addChan {
-		for _, s := range m.publishers {
-			s.Add(data.key, data.value, data.tags...)
+	defer m.wg.Done()
+	for {
+		select {
+		case data, ok := <-addChan:
+			if !ok {
+				return
+			}
+			for _, s := range m.publishers {
+				s.Add(data.key, data.value, data.tags...)
+			}
+		case <-m.ctx.Done():
+			// The channel is never closed (producers race its
+			// closure otherwise); termination is the context.
+			// Drain whatever the buffer still holds so late
+			// datapoints are not lost, then exit.
+			for {
+				select {
+				case data, ok := <-addChan:
+					if !ok {
+						return
+					}
+					for _, s := range m.publishers {
+						s.Add(data.key, data.value, data.tags...)
+					}
+				default:
+					return
+				}
+			}
 		}
 	}
-	m.wg.Done()
 }
 
 type datapoint struct {

rdma/rcroutes/ops_linux.go

@@ -0,0 +1,831 @@
+// Copyright 2026 Versity Software
+// Copyright 2026 Gluesys Inc. and Jihyeon Gim
+// This file is licensed under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux && amd64 && cgo
+
+package rcroutes
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/gofiber/fiber/v3"
+	"github.com/valyala/fasthttp"
+
+	"github.com/versity/versitygw/auth"
+	"github.com/versity/versitygw/metrics"
+	"github.com/versity/versitygw/rdma/rcserver"
+	"github.com/versity/versitygw/s3api/utils"
+	"github.com/versity/versitygw/s3err"
+	"github.com/versity/versitygw/s3event"
+	"github.com/versity/versitygw/s3log"
+)
+
+// OpsServices carries the operational service instances the RC routes
+// report into. All three may be nil; publication then becomes a no-op
+// so the control plane works without any configured backend.
+type OpsServices struct {
+	Logger  s3log.AuditLogger
+	Metrics metrics.Manager
+	Events  s3event.S3EventSender
+}
+
+// opsEmitter is the operational context captured at PREPARE and held
+// until the final outcome is known: enough to synthesize an access
+// record carrying the session's object rather than the wire path.
+// Every string field is owned storage: nothing may reference the
+// request's pooled buffers once PREPARE returns, because fasthttp
+// reuses them for the next request.
+type opsEmitter struct {
+	ops    OpsServices
+	app    *fiber.App
+	acct   auth.Account
+	region string
+	bucket string
+	key    string
+	isPut  bool
+	start  time.Time
+	// Commit metadata the PUT path fills in before publishing the
+	// success record, so the object-created event carries the
+	// backend-assigned ETag and version like the regular put
+	// pipeline's event does.
+	etag    string
+	version string
+	hasEtag bool
+	hasVer  bool
+	// committed records that the backend created the object.
+	// The creation event keys off this fact, not off the final
+	// publication's error status: a committed PUT whose native
+	// finalizer later failed still created the object, and its
+	// creation event must not be lost.
+	committed bool
+	// eventSent guards the creation event against a second
+	// publication path (stashed terminal consumed by a release).
+	eventSent bool
+}
+
+// markCommitted records the backend commit fact together with the
+// backend-assigned object metadata. The creation event fires for
+// any publication after this, including an error publication from
+// a failed native finalizer: the object exists regardless of the
+// finalizer's fate.
+func (e *opsEmitter) markCommitted(etag, version string) {
+	if e == nil {
+		return
+	}
+	e.etag = etag
+	e.hasEtag = true
+	e.version = version
+	e.hasVer = true
+	e.committed = true
+}
+
+// synthesize builds a fiber context whose path and request locals
+// describe the session's logical object operation, so the standard
+// access-log and event pipelines observe GET/PUT of bucket/key
+// instead of the fixed RDMA control path. The app runs with
+// Immutable: string accessors copy instead of exposing the pooled
+// context buffer, which matters because event senders serialize
+// asynchronously and would otherwise read a reused buffer.
+//
+// Route parameters cannot be populated this way (they come from
+// route matching, which a synthesized request never runs), so the
+// metrics bucket tag is absent on RC publications; the audit log
+// derives the bucket from the path instead and stays accurate.
+func (e *opsEmitter) synthesize() (fiber.Ctx, func()) {
+	ctx := e.app.AcquireCtx(&fasthttp.RequestCtx{})
+	method := fiber.MethodGet
+	if e.isPut {
+		method = fiber.MethodPut
+	}
+	ctx.Method(method)
+	// The access logger and the event schema both split this path
+	// into bucket/key, so the synthesized path must be the object
+	// path in canonical form.
+	ctx.Path("/" + e.bucket + "/" + e.key)
+	utils.ContextKeyAccount.Set(ctx, e.acct)
+	utils.ContextKeyRegion.Set(ctx, e.region)
+	utils.ContextKeyStartTime.Set(ctx, e.start)
+	utils.ContextKeyIsRoot.Set(ctx, false)
+	requestID, hostID := utils.EnsureRequestIDs(ctx)
+	ctx.Request().Header.Add("X-Amz-Request-Id", requestID)
+	ctx.Request().Header.Add("X-Amz-Id-2", hostID)
+	return ctx, func() { e.app.ReleaseCtx(ctx) }
+}
+
+// publish emits the final audit record, request metric, and (for a
+// committed PUT) the object-created event. Exactly-once delivery is
+// the tracker's job; this method just performs one emission.
+//
+// The operational sinks classify plain errors as 500 on their own
+// and unwrap nothing, so the publication always hands them the
+// error in its normalized S3 form: the audit log, the metric, and
+// the wire response then carry the same classification.
+func (e *opsEmitter) publish(err error, bytes int64) {
+	if e == nil || (e.ops.Logger == nil && e.ops.Metrics == nil && e.ops.Events == nil) {
+		return
+	}
+	sinkErr := normalizeSinkError(err)
+	ctx, release := e.synthesize()
+	defer release()
+
+	action := metrics.ActionGetObject
+	if e.isPut {
+		action = metrics.ActionPutObject
+	}
+	status := http.StatusOK
+	if sinkErr != nil {
+		status = sinkErr.(s3err.APIError).HTTPStatusCode
+	}
+
+	if e.ops.Metrics != nil {
+		// The bucket dimension comes from the captured session,
+		// not the route: the synthesized context has no matched
+		// route, so Params("bucket") would be empty here.
+		e.ops.Metrics.SendWithBucket(ctx, sinkErr, action, bytes, status, e.bucket)
+	}
+	if e.ops.Logger != nil {
+		e.ops.Logger.Log(ctx, sinkErr, nil, s3log.LogMeta{
+			Action: action,
+			// The object size field reports the transferred
+			// byte count the record carries, so a successful
+			// GET/PUT shows real bytes instead of zero.
+			ObjectSize: bytes,
+		})
+	}
+	// The object-created event keys off the backend commit fact,
+	// not the publication's error status: a committed PUT whose
+	// native finalizer failed still created the object, so its
+	// creation event must survive. Uncommitted PUTs (backend
+	// failure) never carry it.
+	if e.ops.Events != nil && e.committed && e.isPut && !e.eventSent {
+		meta := s3event.EventMeta{
+			EventName:  s3event.EventObjectCreatedPut,
+			ObjectSize: bytes,
+		}
+		if e.hasEtag {
+			etag := e.etag
+			meta.ObjectETag = &etag
+		}
+		if e.hasVer {
+			ver := e.version
+			meta.VersionId = &ver
+		}
+		e.eventSent = true
+		e.ops.Events.SendEvent(ctx, meta)
+	}
+}
+
+// normalizeSinkError renders any operation error as the plain
+// s3err.APIError the sinks expect: wrapped S3 errors keep their
+// payload (the audit loggers assert the S3Error interface directly
+// and would misclassify a wrapper), and non-S3 errors map through
+// the same route error mapping the wire response uses.
+func normalizeSinkError(err error) error {
+	if err == nil {
+		return nil
+	}
+	var s3Err s3err.S3Error
+	if errors.As(err, &s3Err) {
+		return s3Err.BaseError()
+	}
+	return routeError(err)
+}
+
+// httpStatusFromError maps an operation error to the HTTP status
+// the S3 surface would have answered with, using the same route
+// error mapping as the wire response so operational records never
+// disagree with what the client saw.
+func httpStatusFromError(err error) int {
+	if err == nil {
+		return 200
+	}
+	return routeError(err).HTTPStatusCode
+}
+
+// sessionOutcome is the terminal outcome of a tracked session.
+type sessionOutcome struct {
+	err  error // nil on success
+	byt  int64 // bytes transferred on success
+	done bool  // outcome recorded
+}
+
+// sessionRecord is one tracked session with its captured context.
+type sessionRecord struct {
+	emit *opsEmitter
+	out  sessionOutcome
+	// reserved marks a record the request path owns: the native
+	// teardown callback skips it (the request path will publish
+	// exactly once itself), so a completion call that fires the
+	// callback before returning cannot publish a placeholder.
+	reserved bool
+	// terminal marks a teardown that arrived while the record was
+	// reserved: the native session is gone and no second callback
+	// will come, so a later release of the reservation resolves
+	// the stashed outcome instead of leaving an orphan.
+	terminal bool
+	// claimGen identifies the current reservation. Each reserve
+	// bumps it, so a release or publication from an earlier
+	// reservation is rejected even though the record's emitter
+	// pointer is reused across claims.
+	claimGen uint64
+}
+
+// opsTracker owns terminal publication: each session publishes
+// exactly once. The request paths only ever RECORD an outcome; the
+// native teardown callback - which the ABI guarantees fires exactly
+// once per destroyed session, after every completion call - is the
+// 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
+	// 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 guards the accept-vs-drain boundary: overflow, and
+	// the stopped transition, change only under it. Sinks never
+	// 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
+	reqBacklog atomic.Int64
+	reqDropped atomic.Int64
+	// pubPending counts queued-but-unpublished session records.
+	// The native side releases its session quota when it fires
+	// the teardown notification, not when the audit record lands,
+	// so successive sessions can queue more records than the
+	// live-session limit allows. Admission control closes that
+	// gap: a new session is refused while too many of its
+	// predecessors' records are still unpublished, so a stalled
+	// sink delays new sessions instead of accumulating memory.
+	pubPending atomic.Int64
+	// sessionLimit is the native concurrent-session quota; the
+	// admission budget scales with it.
+	sessionLimit int
+	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
+	isReq bool
+}
+
+// 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
+
+// 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
+// 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(sessionLimit int) *opsTracker {
+	t := &opsTracker{
+		sessions: map[string]*sessionRecord{},
+		app: fiber.New(fiber.Config{
+			Immutable: true,
+		}),
+		pubq:         make(chan pubJob, pubQueueSoftCap),
+		done:         make(chan struct{}),
+		drain:        make(chan struct{}),
+		sessionLimit: sessionLimit,
+	}
+	go func() {
+		defer close(t.done)
+		for {
+			select {
+			case job, ok := <-t.pubq:
+				if !ok {
+					return
+				}
+				t.run(job)
+				// Service the overflow list after every
+				// channel job: bursts that exceed the
+				// buffer publish as soon as the sink
+				// recovers instead of waiting for
+				// shutdown. The list is detached under
+				// the lock and published outside it, so a
+				// slow sink never blocks a dispatcher.
+				for _, job := range t.takeOverflow() {
+					t.run(job)
+				}
+			case <-t.drain:
+				// Drain mode. The accept-vs-drain boundary:
+				// under pubmu the worker marks itself
+				// stopped, empties the channel and detaches
+				// the overflow list. A dispatch that
+				// acquires the mutex before the stopped
+				// transition is drained here; one that
+				// acquires it after sees stopped (or done,
+				// closed only after the unlock) and takes
+				// its post-drain path. Sinks run after the
+				// unlock, never under the lock.
+				t.pubmu.Lock()
+				t.stopped = true
+				var pending []pubJob
+				for {
+					select {
+					case job, ok := <-t.pubq:
+						if !ok {
+							t.pubmu.Unlock()
+							for _, job := range pending {
+								t.run(job)
+							}
+							return
+						}
+						pending = append(pending, job)
+					default:
+						pending = append(pending, t.overflow...)
+						t.overflow = nil
+						t.pubmu.Unlock()
+						for _, job := range pending {
+							t.run(job)
+						}
+						return
+					}
+				}
+			}
+		}
+	}()
+	return t
+}
+
+// takeOverflow detaches the overflow list under pubmu. Called by
+// the worker only; the caller publishes the returned jobs outside
+// the lock.
+func (t *opsTracker) takeOverflow() []pubJob {
+	t.pubmu.Lock()
+	defer t.pubmu.Unlock()
+	pending := t.overflow
+	t.overflow = nil
+	return pending
+}
+
+// run publishes one job and releases its reservations: the
+// request-backlog slot and the session admission credit the
+// record was holding.
+func (t *opsTracker) run(job pubJob) {
+	job.emit.publish(job.err, job.byt)
+	if job.isReq {
+		t.reqBacklog.Add(-1)
+		return
+	}
+	t.pubPending.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.
+// 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
+		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)
+		}
+	})
+}
+
+// dispatch hands a publication to the worker without ever
+// 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) {
+	// The pubmu critical section is the accept-vs-drain boundary:
+	// the drain sweep marks stopped under the same lock, so an
+	// append either lands before the sweep (and is drained) or
+	// observes stopped and runs inline.
+	t.pubmu.Lock()
+	if t.stopped {
+		t.pubmu.Unlock()
+		t.run(job)
+		return
+	}
+	select {
+	case t.pubq <- job:
+		t.pubmu.Unlock()
+	default:
+		t.overflow = append(t.overflow, job)
+		t.pubmu.Unlock()
+	}
+}
+
+// SetOpsServices installs the operational service instances. The
+// gateway creates the logger, metrics manager, and event sender
+// after the RC routes exist, so the tracker starts empty and the
+// services arrive here. Sessions registered before the injection
+// publish nothing (there are none: the gateway wires this before
+// it starts serving).
+func (t *opsTracker) SetOpsServices(ops OpsServices) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.ops = ops
+}
+
+// register captures the operational context of a successfully
+// created session so the terminal outcome can be published later.
+// The strings are cloned: they originate from the request's pooled
+// header buffer, which does not survive the response.
+//
+// The account is captured by value but its string fields still
+// reference request storage on some IAM paths, so the sink-relevant
+// identity is cloned as well.
+// errPubBacklog reports admission refusal: too many earlier
+// sessions still have unpublished audit records, so accepting
+// another would grow the publication backlog without bound while
+// a sink is stalled.
+var errPubBacklog = errors.New("publication backlog at capacity")
+
+func (t *opsTracker) register(sessionID string, acct auth.Account,
+	region, bucket, key string, isPut bool, start time.Time) error {
+	// Admission control: the native quota counts live sessions,
+	// but teardown notifications fire before the audit records
+	// land, so session turnover can queue more records than the
+	// quota bounds. Refusing new sessions while the unpublished
+	// backlog reaches the quota turns a stalled sink into
+	// latency (the client retries) instead of unbounded memory.
+	// The check and the credit acquisition share the session
+	// mutex so concurrent registrations cannot each observe the
+	// same headroom and overshoot together. Unbounded when
+	// sessionLimit is unset (tests).
+	if t.sessionLimit > 0 {
+		t.mu.Lock()
+		full := t.pubPending.Load() >= int64(t.sessionLimit)
+		if !full {
+			t.pubPending.Add(1)
+		}
+		t.mu.Unlock()
+		if full {
+			return errPubBacklog
+		}
+	} else {
+		t.mu.Lock()
+		t.pubPending.Add(1)
+		t.mu.Unlock()
+	}
+	acct.Access = strings.Clone(acct.Access)
+	emit := &opsEmitter{
+		ops:    t.loadOps(),
+		app:    t.app,
+		acct:   acct,
+		region: strings.Clone(region),
+		bucket: strings.Clone(bucket),
+		key:    strings.Clone(key),
+		isPut:  isPut,
+		start:  start,
+	}
+
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.sessions[sessionID] = &sessionRecord{emit: emit}
+	return nil
+}
+
+// unregister drops a session entry whose PREPARE finalization
+// failed before the session was committed: the native side either
+// rejected it (no callback will come) or already reaped it (the
+// callback found no record and published nothing). The failure
+// itself is published as a request record by the caller.
+func (t *opsTracker) unregister(sessionID string) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if _, ok := t.sessions[sessionID]; ok {
+		delete(t.sessions, sessionID)
+		// Release the admission credit the registration took:
+		// no callback will ever publish for this entry, so
+		// leaving the credit held would permanently shrink the
+		// admission budget.
+		t.pubPending.Add(-1)
+	}
+}
+
+// failOutcome publishes a failed finalization exactly once: when
+// 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. 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 && !rec.reserved {
+		delete(t.sessions, sessionID)
+	} else {
+		ok = false
+	}
+	t.mu.Unlock()
+	if !ok {
+		return
+	}
+	t.dispatch(pubJob{emit: rec.emit, err: err})
+}
+
+// reservation couples the emitter with the generation of the
+// claim that owns it: release and publication validate the
+// generation, so a stale claim cannot act on a newer one.
+type reservation struct {
+	emit *opsEmitter
+	gen  uint64
+}
+
+// reserve marks a session record as owned by its request path: the
+// teardown callback skips a reserved record because the request
+// path publishes the real outcome itself. Returns the reservation
+// when the record exists and was not reserved yet.
+func (t *opsTracker) reserve(sessionID string) *reservation {
+	if t == nil {
+		return nil
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	rec, ok := t.sessions[sessionID]
+	if !ok || rec.reserved {
+		return nil
+	}
+	rec.reserved = true
+	rec.claimGen++
+	return &reservation{emit: rec.emit, gen: rec.claimGen}
+}
+
+// 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.
+// If the native session already tore down while the record was
+// reserved (terminal stashed), the session is gone: consume the
+// record and publish the stashed outcome, since no second
+// callback will arrive.
+func (t *opsTracker) releaseReservation(sessionID string, rsv *reservation) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	rec, ok := t.sessions[sessionID]
+	if !ok || !rec.reserved || rsv == nil || rec.claimGen != rsv.gen {
+		t.mu.Unlock()
+		return
+	}
+	if !rec.terminal {
+		rec.reserved = false
+		t.mu.Unlock()
+		return
+	}
+	delete(t.sessions, sessionID)
+	t.mu.Unlock()
+	t.dispatch(pubJob{emit: rec.emit, err: rec.out.err, byt: rec.out.byt})
+}
+
+// publishReserved publishes through a reserved record and drops it:
+// the single publication of a request-owned session outcome.
+func (t *opsTracker) publishReserved(sessionID string, rsv *reservation, err error, bytes int64) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	rec, ok := t.sessions[sessionID]
+	// Ownership check: only the current reservation generation
+	// publishes. A stale claim (its reservation was released or
+	// superseded) must not delete or publish the current
+	// owner's record.
+	if !ok || rsv == nil || rec.claimGen != rsv.gen {
+		t.mu.Unlock()
+		return
+	}
+	delete(t.sessions, sessionID)
+	t.mu.Unlock()
+	t.dispatch(pubJob{emit: rsv.emit, err: err, byt: bytes})
+}
+
+func (t *opsTracker) loadOps() OpsServices {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.ops
+}
+
+// onTerminal is the native teardown callback: the single publisher
+// of session records. It consumes the recorded outcome (success,
+// failure, or expiry when no outcome was ever recorded) and removes
+// the entry, so exactly one publication happens per session no
+// matter which path confirmed the result.
+func (t *opsTracker) onTerminal(ev rcserver.TerminalEvent) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	rec, ok := t.sessions[ev.SessionID]
+	if !ok {
+		t.mu.Unlock()
+		return
+	}
+	// A reserved record belongs to its request path, which
+	// publishes the real outcome itself: the callback (fired
+	// synchronously by a completion call, before the request
+	// path could confirm the result) must not touch it. But the
+	// terminal is still a fact: if the reservation is released
+	// later (claim rollback) and no second callback will ever
+	// come - the native session is gone - the stashed event
+	// resolves then, instead of being lost.
+	if rec.reserved {
+		if !rec.out.done {
+			rec.out = sessionOutcome{err: expiredError(ev), done: true}
+		}
+		rec.terminal = true
+		t.mu.Unlock()
+		return
+	}
+	delete(t.sessions, ev.SessionID)
+	t.mu.Unlock()
+
+	if !rec.out.done {
+		// No request path ever confirmed a result: the session
+		// expired, was abandoned, or was canceled. The event's
+		// outcome carries the native reason.
+		rec.out = sessionOutcome{err: expiredError(ev), done: true}
+	}
+	t.dispatch(pubJob{emit: rec.emit, err: rec.out.err, byt: rec.out.byt})
+}
+
+// expiredError renders an unclaimed teardown as the error the
+// publication carries, derived from the native outcome so the
+// record names the real terminal reason. The classification stays
+// aligned with the wire mapping: every transfer-level failure the
+// READY call reports as RC_E_WIRE (wire, verify, or execution
+// timeout) publishes as the same 502 the client would have seen,
+// and only a session that expired without any transfer attempt
+// keeps the expiry code.
+func expiredError(ev rcserver.TerminalEvent) error {
+	switch ev.Outcome {
+	case int(rcserver.ReadyWireFail), int(rcserver.ReadyVerifyFail),
+		int(rcserver.ReadyTimeout):
+		return rcserver.ErrWire
+	default:
+		return errSessionExpired
+	}
+}
+
+// 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
+	}
+	acct.Access = strings.Clone(acct.Access)
+	emit := &opsEmitter{
+		ops:    t.loadOps(),
+		app:    t.app,
+		acct:   acct,
+		region: strings.Clone(regionFromCtx(ctx)),
+		bucket: strings.Clone(bucket),
+		key:    strings.Clone(key),
+		isPut:  isPut,
+		start:  time.Now(),
+	}
+	// The accept-vs-drain boundary decides: a record accepted
+	// before the drain sweep is published by the worker; one
+	// that arrives after is dropped here (not published inline),
+	// because a request publication has no owner left to
+	// guarantee its sinks are still open.
+	t.dispatchOrDrop(pubJob{emit: emit, err: err})
+}
+
+// dispatchOrDrop is dispatch with request-publication semantics:
+// after the worker stopped through the drain the job is dropped
+// 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 {
+		// Overload policy: drop and count. The record carries no
+		// session and no owner can reissue it. Incremented under
+		// pubmu so Shutdown's report (also under pubmu via the
+		// drain's stopped transition) cannot miss it.
+		t.reqDropped.Add(1)
+		t.pubmu.Unlock()
+		return
+	}
+	job.isReq = true
+	t.reqBacklog.Add(1)
+	select {
+	case t.pubq <- job:
+		t.pubmu.Unlock()
+	default:
+		t.overflow = append(t.overflow, job)
+		t.pubmu.Unlock()
+	}
+}
+
+// regionFromCtx reads the region the gateway middleware stored on
+// the live request; the synthesized publication reuses it.
+func regionFromCtx(ctx fiber.Ctx) string {
+	if v, ok := utils.ContextKeyRegion.Get(ctx).(string); ok {
+		return v
+	}
+	return ""
+}
+
+// sessionExpiredError is the S3 error an expired or abandoned
+// session publishes: an internal error whose code names the
+// expiry, so the audit log keeps a descriptive code.
+type sessionExpiredError struct {
+	s3err.APIError
+}
+
+var errSessionExpired = sessionExpiredError{APIError: s3err.APIError{
+	Code:           "SessionExpired",
+	Description:    "The RDMA transfer session expired before completion",
+	HTTPStatusCode: 500,
+}}

rdma/rcroutes/ops_linux_test.go

@@ -0,0 +1,409 @@
+// Copyright 2026 Versity Software
+// Copyright 2026 Gluesys Inc. and Jihyeon Gim
+// This file is licensed under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux && amd64 && cgo
+
+package rcroutes
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"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
+// record before any native completion call (which fires the
+// teardown callback synchronously, before the call returns), the
+// callback skips reserved records, and the request path publishes
+// exactly once. Unreserved records are published by the callback.
+
+func TestOpsTrackerCallbackPublishesExpiry(t *testing.T) {
+	tr := newOpsTracker(0)
+	tr.register("sess-1", auth.Account{Access: "ak"}, "us-east-1",
+		"bkt", "obj", false, time.Now())
+	if got := len(tr.sessions); got != 1 {
+		t.Fatalf("registered sessions = %d, want 1", got)
+	}
+
+	// The reaper path publishes for a session no READY ever
+	// reserved and removes the entry.
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "sess-1"})
+	if got := len(tr.sessions); got != 0 {
+		t.Fatalf("session survived terminal: %d", got)
+	}
+
+	// A second terminal (double reap) finds nothing.
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "sess-1"})
+	if got := len(tr.sessions); got != 0 {
+		t.Fatalf("double terminal left residue: %d", got)
+	}
+}
+
+func TestOpsTrackerReserveBlocksCallback(t *testing.T) {
+	tr := newOpsTracker(0)
+	tr.register("sess-2", auth.Account{Access: "ak"}, "us-east-1",
+		"bkt", "obj", true, time.Now())
+
+	// The READY path reserves before its completion call; the
+	// callback the call fires synchronously must skip the record.
+	rsv := tr.reserve("sess-2")
+	if rsv == nil {
+		t.Fatal("reserve returned nil for a live session")
+	}
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "sess-2"})
+	if got := len(tr.sessions); got != 1 {
+		t.Fatalf("callback consumed a reserved record: %d", got)
+	}
+
+	// The request path then publishes and drops the entry.
+	tr.publishReserved("sess-2", rsv, nil, 4096)
+	if got := len(tr.sessions); got != 0 {
+		t.Fatalf("publishReserved left residue: %d", got)
+	}
+
+	// A second reserve of the consumed entry is nil.
+	if again := tr.reserve("sess-2"); again != nil {
+		t.Fatal("reserve succeeded for a consumed entry")
+	}
+}
+
+func TestOpsTrackerReserveIsExclusive(t *testing.T) {
+	tr := newOpsTracker(0)
+	tr.register("sess-3", auth.Account{Access: "ak"}, "us-east-1",
+		"bkt", "obj", false, time.Now())
+
+	if first := tr.reserve("sess-3"); first == nil {
+		t.Fatal("first reserve failed")
+	}
+	if second := tr.reserve("sess-3"); second != nil {
+		t.Fatal("double reserve succeeded")
+	}
+}
+
+func TestOpsTrackerFailOutcome(t *testing.T) {
+	tr := newOpsTracker(0)
+	tr.register("sess-4", auth.Account{Access: "ak"}, "us-east-1",
+		"bkt", "obj", false, time.Now())
+
+	// Consume-or-noop: present entry is consumed.
+	tr.failOutcome("sess-4", errors.New("x"))
+	if got := len(tr.sessions); got != 0 {
+		t.Fatalf("failOutcome left residue: %d", got)
+	}
+	// A second call after the callback already consumed is a
+	// silent no-op, not a double publication.
+	tr.failOutcome("sess-4", errors.New("y"))
+}
+
+func TestOpsTrackerUnregister(t *testing.T) {
+	tr := newOpsTracker(0)
+	tr.register("sess-5", auth.Account{Access: "ak"}, "us-east-1",
+		"bkt", "obj", false, time.Now())
+	tr.unregister("sess-5")
+	if got := len(tr.sessions); got != 0 {
+		t.Fatalf("unregister left entries: %d", got)
+	}
+	// The teardown callback for the unregistered session is a
+	// silent no-op (native side already rejected or reaped it).
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "sess-5"})
+	if got := len(tr.sessions); got != 0 {
+		t.Fatalf("terminal resurrected entry: %d", got)
+	}
+}
+
+func TestOpsTrackerUnknownSession(t *testing.T) {
+	tr := newOpsTracker(0)
+	// Unknown sessions and the nil tracker are silent no-ops.
+	var nilTracker *opsTracker
+	nilTracker.reserve("ghost")
+	nilTracker.onTerminal(rcserver.TerminalEvent{SessionID: "ghost"})
+	tr.reserve("ghost")
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "ghost"})
+	if got := len(tr.sessions); got != 0 {
+		t.Fatalf("ghost session materialized: %d", got)
+	}
+}
+
+func TestNormalizeSinkError(t *testing.T) {
+	if normalizeSinkError(nil) != nil {
+		t.Fatal("nil error should stay nil")
+	}
+	// Plain errors map through the route error mapping.
+	got := normalizeSinkError(errors.New("x"))
+	apiErr, ok := got.(s3err.APIError)
+	if !ok || apiErr.HTTPStatusCode != 500 {
+		t.Fatalf("plain error => %#v", got)
+	}
+	// Wrapped S3 errors are extracted to their base form so the
+	// audit loggers' direct assertion classifies them correctly.
+	wrapped := errWrapped{s3err.GetAPIError(s3err.ErrNoSuchBucket)}
+	got = normalizeSinkError(wrapped)
+	apiErr, ok = got.(s3err.APIError)
+	if !ok || apiErr.Code != "NoSuchBucket" || apiErr.HTTPStatusCode != 404 {
+		t.Fatalf("wrapped error => %#v", got)
+	}
+}
+
+type errWrapped struct{ s3err.S3Error }
+
+func (errWrapped) Error() string { return "wrapped" }
+
+func TestHttpStatusFromError(t *testing.T) {
+	if got := httpStatusFromError(nil); got != 200 {
+		t.Fatalf("nil error => %d, want 200", got)
+	}
+	if got := httpStatusFromError(errors.New("x")); got != 500 {
+		t.Fatalf("plain error => %d, want 500", got)
+	}
+	if got := httpStatusFromError(s3err.GetAPIError(s3err.ErrAccessDenied)); got != 403 {
+		t.Fatalf("access denied => %d, want 403", got)
+	}
+	// A resource-limit rejection maps to the wire status, not a
+	// generic 500.
+	if got := httpStatusFromError(rcserver.ErrLimit); got != 429 {
+		t.Fatalf("limit error => %d, want 429", got)
+	}
+}
+
+func TestExpiredErrorClassification(t *testing.T) {
+	// Every transfer-level failure the READY call reports as
+	// RC_E_WIRE publishes as the same 502 the wire response
+	// carries; only an unattempted session keeps the expiry code.
+	cases := []struct {
+		outcome int
+		code    string
+		status  int
+	}{
+		{int(rcserver.ReadyWireFail), "RdmaTransferFailed", 502},
+		{int(rcserver.ReadyVerifyFail), "RdmaTransferFailed", 502},
+		{int(rcserver.ReadyTimeout), "RdmaTransferFailed", 502},
+		{int(rcserver.ReadyOK), "SessionExpired", 500},
+	}
+	for _, tc := range cases {
+		err := expiredError(rcserver.TerminalEvent{Outcome: tc.outcome})
+		apiErr := normalizeSinkError(err).(s3err.APIError)
+		if apiErr.Code != tc.code || apiErr.HTTPStatusCode != tc.status {
+			t.Fatalf("outcome %d => %s/%d, want %s/%d",
+				tc.outcome, apiErr.Code, apiErr.HTTPStatusCode,
+				tc.code, tc.status)
+		}
+	}
+}
+
+// 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 }
+
+// TestOpsTrackerPublishesExactlyOncePerSession drives the tracker
+// 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(0)
+	tr.SetOpsServices(OpsServices{Logger: rl})
+
+	// 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 success with bytes.
+	tr.register("s-res", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
+	rsv := tr.reserve("s-res")
+	if rsv == nil {
+		t.Fatal("reserve failed")
+	}
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-res"})
+	tr.publishReserved("s-res", rsv, nil, 128)
+
+	// Denial path while reserved: failOutcome must not steal the
+	// publication; the owner's success record is the only one.
+	tr.register("s-den", auth.Account{Access: "ak"}, "r", "b", "o", true, time.Now())
+	rsv2 := tr.reserve("s-den")
+	if rsv2 == nil {
+		t.Fatal("reserve failed")
+	}
+	tr.failOutcome("s-den", errors.New("denied"))
+	tr.publishReserved("s-den", rsv2, 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())
+	rsv3 := tr.reserve("s-rel")
+	if rsv3 == nil {
+		t.Fatal("reserve failed")
+	}
+	tr.releaseReservation("s-rel", rsv3)
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-rel"})
+
+	// M1 regression: a terminal arriving while reserved is stashed,
+	// and a later claim-rollback release consumes it and publishes the
+	// expiry - the record is not orphaned.
+	tr.register("s-stash", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
+	rsvS := tr.reserve("s-stash")
+	if rsvS == nil {
+		t.Fatal("reserve failed")
+	}
+	tr.onTerminal(rcserver.TerminalEvent{SessionID: "s-stash"})
+	tr.releaseReservation("s-stash", rsvS)
+
+	// M2 regression: a second reservation of a live record is
+	// refused, so a duplicate READY cannot claim the transfer
+	// while another request owns the publication. The owner then
+	// completes normally.
+	tr.register("s-dbl", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
+	rsvD := tr.reserve("s-dbl")
+	if rsvD == nil {
+		t.Fatal("first reserve failed")
+	}
+	if tr.reserve("s-dbl") != nil {
+		t.Fatal("double reserve succeeded")
+	}
+	tr.publishReserved("s-dbl", rsvD, nil, 64)
+
+	// Ownership: a stale emitter must not publish or consume the
+	// current record; the real owner still can, even after the
+	// callback fired (stashed) underneath it.
+	tr.register("s-own", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
+	rsvO := tr.reserve("s-own")
+	if rsvO == nil {
+		t.Fatal("reserve failed")
+	}
+	// 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") // 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", 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.
+	tr.register("s-fail", auth.Account{Access: "ak"}, "r", "b", "o", false, time.Now())
+	tr.failOutcome("s-fail", errors.New("x"))
+
+	// 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) != 8 {
+		t.Fatalf("published %d records, want 8: %+v", len(rl.logs), rl.logs)
+	}
+	// 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-stash stashed terminal consumed by release
+		{false, 64},  // s-dbl owner publishes after refused double reserve
+		{false, 32},  // s-own current owner publishes; stale token no-op
+		{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)
+		}
+	}
+}
+
+func TestOpsTrackerAdmissionAtomicUnderConcurrency(t *testing.T) {
+	tr := newOpsTracker(8)
+	// One predecessor record is already pending.
+	if err := tr.register("s-seed", auth.Account{Access: "a"},
+		"r", "b", "k", false, time.Now()); err != nil {
+		t.Fatalf("seed registration: %v", err)
+	}
+
+	const rounds = 16
+	var wg sync.WaitGroup
+	var admitted atomic.Int64
+	var refused atomic.Int64
+	for i := 0; i < rounds; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			err := tr.register(fmt.Sprintf("s-%d", i), auth.Account{Access: "a"},
+				"r", "b", "k", false, time.Now())
+			if err == nil {
+				admitted.Add(1)
+			} else if errors.Is(err, errPubBacklog) {
+				refused.Add(1)
+			} else {
+				t.Errorf("registration %d: unexpected error %v", i, err)
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	if got := admitted.Load(); got != 7 {
+		t.Fatalf("admitted %d registrations, want exactly 7 (limit 8, 1 pending)", got)
+	}
+	if got := refused.Load(); got != rounds-7 {
+		t.Fatalf("refused %d registrations, want %d", got, rounds-7)
+	}
+	if got := tr.pubPending.Load(); got != 8 {
+		t.Fatalf("pending credits = %d, want 8", got)
+	}
+}

rdma/rcroutes/routes_linux.go

@@ -32,6 +32,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
@@ -41,6 +42,7 @@ import (
 	"github.com/versity/versitygw/backend"
 	"github.com/versity/versitygw/rdma/rcserver"
 	"github.com/versity/versitygw/s3api/utils"
+	"github.com/versity/versitygw/s3err"
 	"github.com/versity/versitygw/s3response"
 )
 
@@ -76,13 +78,50 @@ type Handler struct {
 	iam        auth.IAMService
 	readonly   bool
 	disableACL bool
+	// ops owns terminal publication into the operational services
+	// (access log, request metrics, object events); nil keeps the
+	// routes uninstrumented.
+	ops *opsTracker
 }
 
-// New builds the route handler around a started RC service.
+// PublishAuthFailure emits an operation record for a request whose
+// authentication failed before any route logic ran. The gateway
+// auth adapter calls it so signature failures appear in the access
+// log like they do on the S3 surface.
+func (h *Handler) PublishAuthFailure(ctx fiber.Ctx, err error) {
+	h.ops.publishRequest(ctx, auth.Account{}, err, "", "", false)
+}
+
+// SetOpsServices injects the operational service instances once the
+// gateway has created them, and wires the native teardown callback
+// that publishes sessions no request path ever completed.
+func (h *Handler) SetOpsServices(ops OpsServices) {
+	if h.ops == nil {
+		return
+	}
+	h.ops.SetOpsServices(ops)
+	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.
 func New(svc *rcserver.RCSvc, be backend.Backend, iam auth.IAMService,
-	readonly, disableACL bool) *Handler {
+	readonly, disableACL bool, sessionLimit int) *Handler {
 	return &Handler{svc: svc, be: be, iam: iam,
-		readonly: readonly, disableACL: disableACL}
+		readonly: readonly, disableACL: disableACL,
+		ops: newOpsTracker(sessionLimit)}
 }
 
 // principalID derives the session identity digest from the
@@ -100,6 +139,14 @@ func errNotAdmitted() error {
 	return errRouteUnavailable{}
 }
 
+// ErrNotAdmitted is the route error for requests that lost the
+// race with shutdown: the RC service stopped admitting, so the
+// request cannot be served. Exposed for the admission barrier in
+// the route middleware, which runs before the handlers.
+func ErrNotAdmitted() error {
+	return errNotAdmitted()
+}
+
 func invalidHeader(name, value string) error {
 	return fmt.Errorf("invalid %s header: %q: %w",
 		name, value, errRouteBadRequest{})
@@ -123,42 +170,53 @@ func (h *Handler) prepareCore(ctx fiber.Ctx) error {
 	}
 	defer h.svc.Leave()
 
-	if proto := ctx.Get(hdrProtocol); proto != protocolValue {
-		return invalidHeader(hdrProtocol, proto)
-	}
-
 	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
 	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
 
+	// Header parse failures end the request before authorization;
+	// publish them as request records too, with whatever object
+	// identity and operation the malformed headers still carried.
+	publishHeaderErr := func(err error, isPut bool) error {
+		target := ctx.Get(hdrTarget)
+		bucket, key, _ := splitTarget(target)
+		h.ops.publishRequest(ctx, acct, err, bucket, key, isPut)
+		return err
+	}
+
+	if proto := ctx.Get(hdrProtocol); proto != protocolValue {
+		return publishHeaderErr(invalidHeader(hdrProtocol, proto), false)
+	}
+
 	op := strings.ToUpper(ctx.Get(hdrOp))
 	if op != "GET" && op != "PUT" {
-		return invalidHeader(hdrOp, ctx.Get(hdrOp))
+		return publishHeaderErr(invalidHeader(hdrOp, ctx.Get(hdrOp)), false)
 	}
+	isPut := op == "PUT"
 	target := ctx.Get(hdrTarget)
 	bucket, key, ok := splitTarget(target)
 	if !ok {
-		return invalidHeader(hdrTarget, target)
+		return publishHeaderErr(invalidHeader(hdrTarget, target), isPut)
 	}
 	size, err := parseUint(ctx.Get(hdrSize), 10, 64)
 	if err != nil || size == 0 {
-		return invalidHeader(hdrSize, ctx.Get(hdrSize))
+		return publishHeaderErr(invalidHeader(hdrSize, ctx.Get(hdrSize)), isPut)
 	}
 	offset, err := parseUint(ctx.Get(hdrOffset), 10, 64)
 	if err != nil {
-		return invalidHeader(hdrOffset, ctx.Get(hdrOffset))
+		return publishHeaderErr(invalidHeader(hdrOffset, ctx.Get(hdrOffset)), isPut)
 	}
 	psn, err := parseUint(ctx.Get(hdrPsn), 16, 32)
 	if err != nil || psn == 0 || psn > 0xffffff {
-		return invalidHeader(hdrPsn, ctx.Get(hdrPsn))
+		return publishHeaderErr(invalidHeader(hdrPsn, ctx.Get(hdrPsn)), isPut)
 	}
 	cookie, err := parseUint(ctx.Get(hdrCookie), 16, 32)
 	if err != nil || cookie == 0 {
-		return invalidHeader(hdrCookie, ctx.Get(hdrCookie))
+		return publishHeaderErr(invalidHeader(hdrCookie, ctx.Get(hdrCookie)), isPut)
 	}
-	isPut := op == "PUT"
 
 	// Authorize through the regular object-access chain.
 	if err := h.authorize(ctx, acct, isRoot, bucket, key, isPut); err != nil {
+		h.ops.publishRequest(ctx, acct, err, bucket, key, isPut)
 		return err
 	}
 
@@ -173,6 +231,7 @@ func (h *Handler) prepareCore(ctx fiber.Ctx) error {
 		ClientToken: ctx.Get(hdrToken),
 	})
 	if err != nil {
+		h.ops.publishRequest(ctx, acct, mapRcError(err), bucket, key, isPut)
 		return mapRcError(err)
 	}
 
@@ -181,13 +240,43 @@ func (h *Handler) prepareCore(ctx fiber.Ctx) error {
 	if !isPut {
 		if err := h.stageGet(ctx, resp.SessionID, bucket, key, offset, size); err != nil {
 			_ = h.svc.FinishPrepare(resp.SessionID, false)
+			h.ops.publishRequest(ctx, acct, err, bucket, key, isPut)
 			return err
 		}
 	}
+
+	// Register before the finalizing call: FinishPrepare can reap
+	// an already-expired session and fire the teardown callback
+	// synchronously, and a registered record (or a parked early
+	// notification) keeps that publication from being lost.
+	// Refusal here is admission control: earlier sessions still
+	// hold unpublished audit records, so the new session is
+	// rejected before the native side commits it.
+	if err := h.ops.register(resp.SessionID, acct,
+		regionFromCtx(ctx), bucket, key, isPut, time.Now()); err != nil {
+		_ = h.svc.FinishPrepare(resp.SessionID, false)
+		// The audit record carries the same SlowDown the wire
+		// shows, so operator-side accounting matches what the
+		// client saw.
+		apiErr := s3err.GetAPIError(s3err.ErrSlowDown)
+		h.ops.publishRequest(ctx, acct, apiErr, bucket, key, isPut)
+		return apiErr
+	}
 	if err := h.svc.FinishPrepare(resp.SessionID, true); err != nil {
+		// The finalization failed. Exactly one publication
+		// covers it: the finalizing call already reaped the
+		// session and its callback published the recorded
+		// outcome, or the native side rejected the call and no
+		// callback is coming, in which case the entry is
+		// published here.
+		h.ops.failOutcome(resp.SessionID, mapRcError(err))
 		return mapRcError(err)
 	}
 
+	// The session now owns the operation record: the terminal
+	// path (READY/FinishPut completion, CANCEL, or the expiry
+	// reaper) publishes the final outcome exactly once.
+
 	// Wire reply per the hipobj-rc-v2 contract: protocol echo,
 	// the server endpoint as "200:<token>", session id, and PSN.
 	ctx.Set(hdrProtocol, protocolValue)
@@ -324,8 +413,38 @@ 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.
+	rsv := h.ops.reserve(sessionID)
+	if rsv == nil {
+		// Another READY holds the publication reservation for
+		// this session. Publication ownership must track native
+		// transfer ownership: proceeding without the reservation
+		// would let this request win the native claim while a
+		// different request owns the publication, losing the
+		// record on completion. Answer as a duplicate claim.
+		return fmt.Errorf("transfer in progress: %w", rcserver.ErrDouble)
+	}
+	// Panic safety starts at the reservation: an unwind anywhere
+	// below (authorization, backend I/O) must still retire the
+	// reservation so the record is not orphaned - the terminal
+	// callback can only stash under a reservation, and nobody
+	// else would ever release it. releaseReservation is a no-op
+	// once a later publish consumed the record.
+	defer h.ops.releaseReservation(sessionID, rsv)
+	publish := func(err error, bytes int64) {
+		h.ops.publishReserved(sessionID, rsv, err, bytes)
+	}
 	if err := h.authorize(ctx, acct, isRoot, bucket, key, info.Op == 1); err != nil {
-		// Permission revoked mid-session: cancel the session.
+		// Permission revoked mid-session: publish the real
+		// denial - not an expiry - as the outcome, release the
+		// reservation, and cancel the session.
+		err = mapRcError(err)
+		publish(err, 0)
 		_ = h.svc.Cancel(sessionID, principal)
 		return err
 	}
@@ -344,7 +463,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, rsv)
 		return mapRcError(err)
 	}
 
@@ -352,17 +475,43 @@ 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, rsv)
 		return fmt.Errorf("peer busy: %w", rcserver.ErrDouble)
 	}
 
 	// The claim succeeded: from here until the response commits,
 	// this handler owns the completion ref. A panic or early
 	// unwind must still release it so the session can be reaped.
+	//
+	// 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 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
+	doPublish := func(err error, bytes int64) {
+		if !published {
+			published = true
+			publish(err, bytes)
+		}
+	}
 	finalized := false
 	defer func() {
 		if !finalized {
+			// The unwind finalizer retires the session. The
+			// reservation keeps its callback a no-op, so the
+			// publication must come from here on a panic or
+			// early-unwind path.
+			if !published {
+				doPublish(errPanicked(), 0)
+			}
 			_ = h.svc.FinishFinal(sessionID)
 		}
 	}()
@@ -375,11 +524,20 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error {
 		// finalizer retires exactly at that point; a failure
 		// *before* the borrow still falls back to the
 		// finalizer path below.
-		put, gd, err := h.commitPut(ctx, sessionID, bucket, key, sizeOf(resp))
-		if gd {
+		put, viewDone, committed, err := h.commitPut(ctx, sessionID, bucket, key, sizeOf(resp))
+		if viewDone {
 			finalized = true
 		}
+		if put != nil {
+			// The backend committed the object. Record the
+			// fact before anything else can fail: the audit
+			// record and creation event must reflect the
+			// commit even when the native finalizer below
+			// errors out.
+			rsv.emit.markCommitted(put.ETag, put.VersionID)
+		}
 		if err != nil {
+			doPublish(mapRcError(err), committed)
 			return err
 		}
 		// The FINAL wire reply carries the stored object's
@@ -387,11 +545,16 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error {
 		resp.Etag = put.ETag
 		resp.VersionID = put.VersionID
 	} else if err := h.svc.FinishFinal(sessionID); err != nil {
+		doPublish(mapRcError(err), 0)
 		return mapRcError(err)
 	} else {
 		finalized = true
 	}
 
+	// The transfer completed: publish the terminal record with
+	// the byte count the data plane reported.
+	doPublish(nil, int64(resp.BytesTransferred))
+
 	// Wire reply per the hipobj-rc-v2 contract: protocol echo,
 	// cookie echo, transferred bytes, and object metadata.
 	ctx.Set(hdrProtocol, protocolValue)
@@ -421,18 +584,18 @@ func sizeOf(resp *rcserver.ReadyResponse) uint64 {
 // panic-safe defer releases the view if the handler unwinds before
 // FinishPut runs.
 func (h *Handler) commitPut(ctx fiber.Ctx, sessionID, bucket, key string,
-	size uint64) (*s3response.PutObjectOutput, bool, error) {
+	size uint64) (put *s3response.PutObjectOutput, viewDone bool, committed int64, err error) {
 	view, err := h.svc.GetPutData(sessionID)
 	if err != nil {
-		return nil, false, mapRcError(err)
+		return nil, false, 0, mapRcError(err)
 	}
 	// Panic-safe ownership: if anything below unwinds, the view is
 	// still returned exactly once (the ABI consumes the handle a
 	// single time; a redundant FinishPut after a commit is a
 	// no-op STALE).
-	committed := false
+	putDone := false
 	defer func() {
-		if !committed {
+		if !putDone {
 			_ = h.svc.FinishPut(*view, false, "", "")
 		}
 	}()
@@ -447,13 +610,19 @@ func (h *Handler) commitPut(ctx fiber.Ctx, sessionID, bucket, key string,
 		Body:          bytes.NewReader(view.Buf),
 	})
 	if err != nil {
-		return nil, true, err
+		// The object was not created; the audit records zero
+		// transferred bytes. The view was consumed above.
+		return nil, true, 0, err
 	}
+	// The object exists from here on. The value below is the
+	// committed byte count, reported even when the native
+	// finalizer fails, so the audit record reflects the commit.
+	committed = contentLength
 	if err := h.svc.FinishPut(*view, true, res.ETag, res.VersionID); err != nil {
-		return nil, true, mapRcError(err)
+		return &res, true, committed, mapRcError(err)
 	}
-	committed = true
-	return &res, true, nil
+	putDone = true
+	return &res, true, committed, nil
 }
 
 // Cancel handles CANCEL: authenticated owner tears the session down.
@@ -624,3 +793,14 @@ func mapRcError(err error) error {
 	}
 	return err
 }
+
+// errPanicked is the outcome recorded when a panic unwinds the
+// READY handler after the completion ref was claimed: the record
+// keeps the failure even though the panic itself propagates.
+func errPanicked() error {
+	return s3err.APIError{
+		Code:           "InternalRDMAError",
+		Description:    "The RDMA transfer ended without a confirmed result",
+		HTTPStatusCode: 500,
+	}
+}

rdma/rcroutes/routes_stub.go

@@ -19,6 +19,8 @@
 package rcroutes
 
 import (
+	"errors"
+
 	"github.com/gofiber/fiber/v3"
 
 	"github.com/versity/versitygw/auth"
@@ -30,7 +32,7 @@ type Handler struct{}
 
 // New builds a stub route handler; the routes answer 501.
 func New(svc any, be backend.Backend, iam auth.IAMService,
-	readonly, disableACL bool) *Handler {
+	readonly, disableACL bool, sessionLimit int) *Handler {
 	return &Handler{}
 }
 
@@ -48,3 +50,27 @@ func (h *Handler) Ready(ctx fiber.Ctx) error { return notImplemented(ctx) }
 
 // Cancel is a stub handler that answers 501 Not Implemented.
 func (h *Handler) Cancel(ctx fiber.Ctx) error { return notImplemented(ctx) }
+
+// OpsServices carries the operational service instances (stub
+// mirror; the fields exist only to keep the embedding surface
+// platform-independent).
+type OpsServices struct {
+	Logger  any
+	Metrics any
+	Events  any
+}
+
+// SetOpsServices is a stub: without RDMA support there is nothing
+// to publish into.
+func (h *Handler) SetOpsServices(ops OpsServices) {}
+
+// PublishAuthFailure is a stub mirror of the linux handler.
+func (h *Handler) PublishAuthFailure(ctx fiber.Ctx, err error) {}
+
+// ErrNotAdmitted is a stub mirror of the linux helper.
+func ErrNotAdmitted() error {
+	return errors.New("rc routes unavailable")
+}
+
+// Shutdown is a stub mirror of the linux handler.
+func (h *Handler) Shutdown() {}

rdma/rcserver/rcserver_linux.go

@@ -34,6 +34,8 @@ package rcserver
 extern void rcgo_log_sink(void *ctx, int level, char *msg,
                           char *file, int line);
 extern void rcgo_snapshot_cb(rc_session_snapshot *rec, void *ctx);
+extern void rcgo_terminal_cb(void *ctx, char *id, int outcome,
+                             uint64_t bytes);
 */
 import "C"
 
@@ -547,6 +549,53 @@ var (
 	snapshotSink *[]SessionSnapshot
 )
 
+// TerminalEvent describes one reaped session, delivered through the
+// terminal notification sink.
+type TerminalEvent struct {
+	SessionID string
+	Outcome   int // RC_READY_* value observed at teardown
+	Bytes     uint64
+}
+
+// terminalNotify serializes the trampoline callback and stores the
+// subscriber. The RC service has a single instance per gateway, so
+// one process-wide sink matches the log-sink pattern.
+var (
+	terminalNotifyMu sync.Mutex
+	terminalNotify   func(TerminalEvent)
+)
+
+//export rcgo_terminal_cb
+func rcgo_terminal_cb(_ unsafe.Pointer, id *C.char, outcome C.int, bytes C.uint64_t) {
+	if id == nil {
+		return
+	}
+	terminalNotifyMu.Lock()
+	fn := terminalNotify
+	terminalNotifyMu.Unlock()
+	if fn == nil {
+		return
+	}
+	fn(TerminalEvent{
+		SessionID: C.GoString(id),
+		Outcome:   int(outcome),
+		Bytes:     uint64(bytes),
+	})
+}
+
+// SetTerminalNotify installs the session-teardown callback. The C
+// side invokes it with no lock held; the callback must return
+// promptly and must not call back into the service.
+func (s *RCSvc) SetTerminalNotify(fn func(TerminalEvent)) {
+	terminalNotifyMu.Lock()
+	terminalNotify = fn
+	terminalNotifyMu.Unlock()
+	if s.srv != nil {
+		C.rc_server_set_terminal_notify(s.srv,
+			(*[0]byte)(C.rcgo_terminal_cb), nil)
+	}
+}
+
 //export rcgo_snapshot_cb
 func rcgo_snapshot_cb(rec *C.rc_session_snapshot, _ unsafe.Pointer) {
 	sink := snapshotSink

rdma/rcserver/rcserver_stub.go

@@ -178,6 +178,16 @@ func (s *RCSvc) SessionsSnapshot() ([]SessionSnapshot, error) {
 	return nil, errNotSupported
 }
 
+// TerminalEvent is a stub mirror of the linux record.
+type TerminalEvent struct {
+	SessionID string
+	Outcome   int
+	Bytes     uint64
+}
+
+// SetTerminalNotify is a stub.
+func (s *RCSvc) SetTerminalNotify(fn func(TerminalEvent)) {}
+
 // ReadyTransfer is a stub.
 func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) {
 	return nil, errNotSupported

s3api/controllers/base_test.go

@@ -282,7 +282,9 @@ func (m *mockEvSender) Close() error                               { return nil
 type mockMetricsManager struct{}
 
 func (m *mockMetricsManager) Send(_ fiber.Ctx, _ error, _ string, _ int64, _ int) {}
-func (m *mockMetricsManager) Close()                                              {}
+func (m *mockMetricsManager) SendWithBucket(_ fiber.Ctx, _ error, _ string, _ int64, _ int, _ string) {
+}
+func (m *mockMetricsManager) Close() {}
 
 func TestProcessController(t *testing.T) {
 	payload, err := xml.Marshal(s3response.Bucket{