From 97238d80ad77c1e5fd4f19a4dc36777c64f39663 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sun, 6 Sep 2026 10:54:51 +0900 Subject: [PATCH] rdma: reserve session publication before completion calls A native completion call fires the teardown callback before it returns, so an outcome recorded after the call is too late: every successful transfer published as an expiry, and PUT failures and panics lost their real outcome to the callback's placeholder. The READY handler now reserves the session record before invoking any completion call. A reserved record is invisible to the callback, and the handler publishes the real outcome exactly once after the result is known. A deferred safety net publishes on panic unwinds before the unwind finalizer retires the session. Successful PUTs forward the backend-assigned ETag and version into the object-created event, and the audit record carries the transferred byte count as the object size. A READY whose re-authorization fails now publishes the denial itself before canceling, instead of letting the cancel publish an expiry. The native side keeps the terminal reason on every failure exit: a verify failure no longer overwrites the wire-failure outcome, and QP transition or re-arm failures record the wire failure they return. The unclaimed-teardown classification follows the wire mapping, so every transfer-level failure the client would see as 502 publishes as the same 502 instead of a diverging per-cause code. The metrics bucket tag is documented as absent on synthesized publications: route parameters come from route matching, which a synthesized request never runs; the audit log derives the bucket from the path and stays accurate. --- cuwrapper/rc/rc_server_abi.cpp | 10 ++- rdma/rcroutes/ops_linux.go | 123 ++++++++++++++++++++++---------- rdma/rcroutes/ops_linux_test.go | 91 ++++++++++++++--------- rdma/rcroutes/routes_linux.go | 60 +++++++++------- 4 files changed, 185 insertions(+), 99 deletions(-) diff --git a/cuwrapper/rc/rc_server_abi.cpp b/cuwrapper/rc/rc_server_abi.cpp index 948cce72..e9787db7 100644 --- a/cuwrapper/rc/rc_server_abi.cpp +++ b/cuwrapper/rc/rc_server_abi.cpp @@ -879,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 */ @@ -887,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; @@ -968,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; @@ -981,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; diff --git a/rdma/rcroutes/ops_linux.go b/rdma/rcroutes/ops_linux.go index 096b34a7..0adf6630 100644 --- a/rdma/rcroutes/ops_linux.go +++ b/rdma/rcroutes/ops_linux.go @@ -60,6 +60,26 @@ type opsEmitter struct { 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 +} + +// setCommitMeta records the backend-assigned object metadata for +// the success publication's event payload. +func (e *opsEmitter) setCommitMeta(etag, version string) { + if e == nil { + return + } + e.etag = etag + e.hasEtag = true + e.version = version + e.hasVer = true } // synthesize builds a fiber context whose path and request locals @@ -69,6 +89,11 @@ type opsEmitter struct { // 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 @@ -121,15 +146,28 @@ func (e *opsEmitter) publish(err error, bytes int64) { 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 fires at commit time only; error // publications never carry it. if e.ops.Events != nil && err == nil && e.isPut { - e.ops.Events.SendEvent(ctx, s3event.EventMeta{ + 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.ops.Events.SendEvent(ctx, meta) } } @@ -160,10 +198,7 @@ func httpStatusFromError(err error) int { return routeError(err).HTTPStatusCode } -// sessionOutcome is the terminal outcome of a tracked session, -// recorded by whichever path observes the final result first. A -// recorded outcome is final: the teardown callback consumes it and -// publishes, so exactly one publication happens per session. +// sessionOutcome is the terminal outcome of a tracked session. type sessionOutcome struct { err error // nil on success byt int64 // bytes transferred on success @@ -174,6 +209,11 @@ type sessionOutcome struct { 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 } // opsTracker owns terminal publication: each session publishes @@ -273,23 +313,36 @@ func (t *opsTracker) failOutcome(sessionID string, err error) { rec.emit.publish(err, 0) } -// recordOutcome stores the terminal outcome of a completing -// transfer. It never publishes: publication belongs to the teardown -// callback, which fires exactly once per session after the -// completion call returns. Recording is idempotent - the first -// outcome wins - so a panic-unwind path that records after the -// normal path changed nothing. -func (t *opsTracker) recordOutcome(sessionID string, err error, bytes int64) { +// 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 emitter when +// the record exists and was not reserved yet. +func (t *opsTracker) reserve(sessionID string) *opsEmitter { if t == nil { - return + return nil } t.mu.Lock() defer t.mu.Unlock() rec, ok := t.sessions[sessionID] - if !ok || rec.out.done { + if !ok || rec.reserved { + return nil + } + rec.reserved = true + return rec.emit +} + +// publishReserved publishes through a reserved record and drops it: +// the single publication of a request-owned session outcome. +func (t *opsTracker) publishReserved(sessionID string, emit *opsEmitter, err error, bytes int64) { + if t == nil { return } - rec.out = sessionOutcome{err: err, byt: bytes, done: true} + t.mu.Lock() + delete(t.sessions, sessionID) + t.mu.Unlock() + if emit != nil { + emit.publish(err, bytes) + } } func (t *opsTracker) loadOps() OpsServices { @@ -313,6 +366,14 @@ func (t *opsTracker) onTerminal(ev rcserver.TerminalEvent) { 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. + if rec.reserved { + t.mu.Unlock() + return + } delete(t.sessions, ev.SessionID) t.mu.Unlock() @@ -328,15 +389,17 @@ func (t *opsTracker) onTerminal(ev rcserver.TerminalEvent) { // expiredError renders an unclaimed teardown as the error the // publication carries, derived from the native outcome so the -// record names the real terminal reason instead of a generic one. +// 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): + case int(rcserver.ReadyWireFail), int(rcserver.ReadyVerifyFail), + int(rcserver.ReadyTimeout): return rcserver.ErrWire - case int(rcserver.ReadyTimeout): - return errTimeoutExpired - case int(rcserver.ReadyVerifyFail): - return errVerifyFailed default: return errSessionExpired } @@ -385,19 +448,3 @@ var errSessionExpired = sessionExpiredError{APIError: s3err.APIError{ Description: "The RDMA transfer session expired before completion", HTTPStatusCode: 500, }} - -// errTimeoutExpired and errVerifyFailed classify unclaimed teardowns -// whose native outcome names a specific READY failure: the record -// then matches what the wire response for that failure carries. -var ( - errTimeoutExpired = sessionExpiredError{APIError: s3err.APIError{ - Code: "RdmaTransferTimeout", - Description: "The RDMA transfer timed out before completion", - HTTPStatusCode: 504, - }} - errVerifyFailed = sessionExpiredError{APIError: s3err.APIError{ - Code: "RdmaTransferVerifyFailed", - Description: "The RDMA transfer failed verification before completion", - HTTPStatusCode: 502, - }} -) diff --git a/rdma/rcroutes/ops_linux_test.go b/rdma/rcroutes/ops_linux_test.go index 58e99f1c..a1ee154f 100644 --- a/rdma/rcroutes/ops_linux_test.go +++ b/rdma/rcroutes/ops_linux_test.go @@ -8,17 +8,10 @@ // // Unless required by applicable law or agreed to in writing, // software distributed under the License is distributed on an -<<<<<<< Updated upstream -// "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. -======= // "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. ->>>>>>> Stashed changes //go:build linux && amd64 && cgo @@ -34,9 +27,11 @@ import ( "github.com/versity/versitygw/s3err" ) -// The publication model: request paths only record outcomes; the -// teardown callback is the single publisher. The tests exercise -// the table semantics that guarantee exactly-once publication. +// 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() @@ -47,7 +42,7 @@ func TestOpsTrackerCallbackPublishesExpiry(t *testing.T) { } // The reaper path publishes for a session no READY ever - // recorded and removes the entry. + // 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) @@ -60,50 +55,73 @@ func TestOpsTrackerCallbackPublishesExpiry(t *testing.T) { } } -func TestOpsTrackerRecordedOutcomeWins(t *testing.T) { +func TestOpsTrackerReserveBlocksCallback(t *testing.T) { tr := newOpsTracker() tr.register("sess-2", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", true, time.Now()) - // The READY path records the real outcome; the callback then - // consumes it instead of publishing an expiry. - tr.recordOutcome("sess-2", nil, 4096) + // The READY path reserves before its completion call; the + // callback the call fires synchronously must skip the record. + emit := tr.reserve("sess-2") + if emit == nil { + t.Fatal("reserve returned nil for a live session") + } tr.onTerminal(rcserver.TerminalEvent{SessionID: "sess-2"}) - if got := len(tr.sessions); got != 0 { - t.Fatalf("session survived terminal: %d", got) + if got := len(tr.sessions); got != 1 { + t.Fatalf("callback consumed a reserved record: %d", got) } - // A late second record is a no-op (entry gone). - tr.recordOutcome("sess-2", nil, 1) + // The request path then publishes and drops the entry. + tr.publishReserved("sess-2", emit, 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 TestOpsTrackerFirstRecordWins(t *testing.T) { +func TestOpsTrackerReserveIsExclusive(t *testing.T) { tr := newOpsTracker() tr.register("sess-3", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", false, time.Now()) - tr.recordOutcome("sess-3", nil, 100) - tr.recordOutcome("sess-3", errors.New("late"), 0) - - tr.mu.Lock() - rec := tr.sessions["sess-3"] - tr.mu.Unlock() - if rec == nil || rec.out.err != nil || rec.out.byt != 100 { - t.Fatalf("second record overwrote the first: %+v", rec.out) + 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() + 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() - tr.register("sess-4", auth.Account{Access: "ak"}, "us-east-1", + tr.register("sess-5", auth.Account{Access: "ak"}, "us-east-1", "bkt", "obj", false, time.Now()) - tr.unregister("sess-4") + 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-4"}) + tr.onTerminal(rcserver.TerminalEvent{SessionID: "sess-5"}) if got := len(tr.sessions); got != 0 { t.Fatalf("terminal resurrected entry: %d", got) } @@ -113,9 +131,9 @@ func TestOpsTrackerUnknownSession(t *testing.T) { tr := newOpsTracker() // Unknown sessions and the nil tracker are silent no-ops. var nilTracker *opsTracker - nilTracker.recordOutcome("ghost", nil, 1) + nilTracker.reserve("ghost") nilTracker.onTerminal(rcserver.TerminalEvent{SessionID: "ghost"}) - tr.recordOutcome("ghost", nil, 1) + tr.reserve("ghost") tr.onTerminal(rcserver.TerminalEvent{SessionID: "ghost"}) if got := len(tr.sessions); got != 0 { t.Fatalf("ghost session materialized: %d", got) @@ -164,14 +182,17 @@ func TestHttpStatusFromError(t *testing.T) { } 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), "RdmaTransferVerifyFailed", 502}, - {int(rcserver.ReadyTimeout), "RdmaTransferTimeout", 504}, + {int(rcserver.ReadyVerifyFail), "RdmaTransferFailed", 502}, + {int(rcserver.ReadyTimeout), "RdmaTransferFailed", 502}, {int(rcserver.ReadyOK), "SessionExpired", 500}, } for _, tc := range cases { diff --git a/rdma/rcroutes/routes_linux.go b/rdma/rcroutes/routes_linux.go index 580f8d32..a120f0cc 100644 --- a/rdma/rcroutes/routes_linux.go +++ b/rdma/rcroutes/routes_linux.go @@ -384,7 +384,13 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error { return errors.New("invalid session target") } 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: cancel the session, + // and publish the real denial - not an expiry - as its + // outcome. The cancel tears the session down now; a + // plain publish (consume-or-noop) would race it, so + // the record is consumed here. + err = mapRcError(err) + h.ops.failOutcome(sessionID, err) _ = h.svc.Cancel(sessionID, principal) return err } @@ -420,26 +426,31 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error { // this handler owns the completion ref. A panic or early // unwind must still release it so the session can be reaped. // - // This handler only RECORDS the outcome; publication belongs - // to the teardown callback, which every native completion - // call below fires exactly once. The deferred recorder covers - // panic unwinds too, so every path through this handler - // leaves a final outcome behind. - recorded := false - record := func(err error, bytes int64) { - if !recorded { - recorded = true - h.ops.recordOutcome(sessionID, err, bytes) + // Every native completion call below (FinishFinal, and + // FinishPut inside commitPut) fires the teardown callback + // synchronously, BEFORE the call returns - so the outcome + // cannot be recorded after the call. The handler reserves the + // publication instead: a reserved record is invisible to the + // callback, and this handler publishes exactly once after the + // result is known. The deferred safety net publishes on any + // unwind that bypassed the normal paths. + emit := h.ops.reserve(sessionID) + published := false + publish := func(err error, bytes int64) { + if !published { + published = true + h.ops.publishReserved(sessionID, emit, err, bytes) } } finalized := false defer func() { if !finalized { - // The unwind finalizer retires the session, which - // fires the callback; make sure it carries an - // outcome even on a panic path. - if !recorded { - record(errPanicked(), 0) + // 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 { + publish(errPanicked(), 0) } _ = h.svc.FinishFinal(sessionID) } @@ -458,27 +469,26 @@ func (h *Handler) readyCore(ctx fiber.Ctx) error { finalized = true } if err != nil { - record(mapRcError(err), 0) + publish(mapRcError(err), 0) return err } // The FINAL wire reply carries the stored object's // metadata, which the backend assigned at commit time. resp.Etag = put.ETag resp.VersionID = put.VersionID + // The success publication's object-created event carries + // the same commit metadata. + emit.setCommitMeta(put.ETag, put.VersionID) } else if err := h.svc.FinishFinal(sessionID); err != nil { - record(mapRcError(err), 0) + publish(mapRcError(err), 0) return mapRcError(err) } else { finalized = true } - // The transfer completed: record the terminal outcome with - // the byte count the data plane reported. FinishPut (inside - // commitPut) or the FinishFinal above already fired the - // teardown callback, which publishes this outcome - or, on a - // timing edge where the callback ran first, the deferred - // finalizer's FinishFinal does. - record(nil, int64(resp.BytesTransferred)) + // The transfer completed: publish the terminal record with + // the byte count the data plane reported. + publish(nil, int64(resp.BytesTransferred)) // Wire reply per the hipobj-rc-v2 contract: protocol echo, // cookie echo, transferred bytes, and object metadata.