diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index e5d3bf4d..bd7579d9 100644 --- a/cmd/vgwrdma/main.go +++ b/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), diff --git a/cuwrapper/rc/rc_server_abi.cpp b/cuwrapper/rc/rc_server_abi.cpp index 7d7c5e6a..e9787db7 100644 --- a/cuwrapper/rc/rc_server_abi.cpp +++ b/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 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; diff --git a/cuwrapper/rc/rc_server_abi.h b/cuwrapper/rc/rc_server_abi.h index f852c19b..9193524a 100644 --- a/cuwrapper/rc/rc_server_abi.h +++ b/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 { diff --git a/embedgw/embedgw.go b/embedgw/embedgw.go index f7c6daa9..b5b9d9f8 100644 --- a/embedgw/embedgw.go +++ b/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() } diff --git a/internal/rdmamode/rdmamode.go b/internal/rdmamode/rdmamode.go index 92cad9fa..a3fd4720 100644 --- a/internal/rdmamode/rdmamode.go +++ b/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() }) } diff --git a/metrics/metrics.go b/metrics/metrics.go index d96f2ded..ded4c676 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -22,6 +22,7 @@ import ( "os" "strings" "sync" + "sync/atomic" "github.com/gofiber/fiber/v3" "github.com/versity/versitygw/s3err" @@ -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 { diff --git a/rdma/rcroutes/ops_linux.go b/rdma/rcroutes/ops_linux.go new file mode 100644 index 00000000..eb8370f0 --- /dev/null +++ b/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, +}} diff --git a/rdma/rcroutes/ops_linux_test.go b/rdma/rcroutes/ops_linux_test.go new file mode 100644 index 00000000..ae1d6e65 --- /dev/null +++ b/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) + } +} diff --git a/rdma/rcroutes/routes_linux.go b/rdma/rcroutes/routes_linux.go index e0a6c818..394d9893 100644 --- a/rdma/rcroutes/routes_linux.go +++ b/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:", 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, + } +} diff --git a/rdma/rcroutes/routes_stub.go b/rdma/rcroutes/routes_stub.go index 387f4a5e..f3b5e03c 100644 --- a/rdma/rcroutes/routes_stub.go +++ b/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() {} diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index 2713c1c7..aad73e0c 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/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 diff --git a/rdma/rcserver/rcserver_stub.go b/rdma/rcserver/rcserver_stub.go index 99e6b220..493e3b1a 100644 --- a/rdma/rcserver/rcserver_stub.go +++ b/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 diff --git a/s3api/controllers/base_test.go b/s3api/controllers/base_test.go index 46e01d9b..b2169e92 100644 --- a/s3api/controllers/base_test.go +++ b/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{