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.