feat(sessions): add UnattachedSessionGraceSeconds to reap never-subscribed idle sessions

A client that opens a session and vanishes before it ever streams (e.g.
OtOpcUa's GalaxyDriver on a failed start) pinned a session slot and a live
x86 worker for the full DefaultLeaseSeconds (1800 s), because detach-grace
only applies to sessions that once had a subscriber.

MxGateway:Sessions:UnattachedSessionGraceSeconds (int, default 0 = off,
validated >= 0) makes the existing lease sweep reap a Ready session that
has never attached an event subscriber once it has been idle (no unary
activity) for the window, with the close reason unattached-grace-expired.
Ready-only, idle-from-last-activity and first-subscriber-cancels keep it
from catching slow starts or unary-only clients.

GatewaySession.ExtendLease had no production callers and did not fit the
rule, so it is removed; tests move leases through TouchClientActivity.
This commit is contained in:
Joseph Doherty
2026-10-22 06:31:39 -04:00
parent 2b165cd3a4
commit 285ef64160
9 changed files with 258 additions and 35 deletions
@@ -325,6 +325,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
options.FaultedGraceSeconds,
"MxGateway:Sessions:FaultedGraceSeconds must be zero or greater (0 reaps a faulted session on the next sweep).",
builder);
AddIfNegative(
options.UnattachedSessionGraceSeconds,
"MxGateway:Sessions:UnattachedSessionGraceSeconds must be zero or greater (0 disables reaping of never-subscribed sessions).",
builder);
AddIfNegative(
options.WorkerReadyWaitTimeoutMs,
"MxGateway:Sessions:WorkerReadyWaitTimeoutMs must be greater than or equal to zero.",
@@ -57,6 +57,26 @@ public sealed class SessionOptions
/// </summary>
public int FaultedGraceSeconds { get; init; }
/// <summary>
/// Gets the idle window, in seconds, after which the lease monitor reaps a session that
/// has <em>never</em> attached an event-stream subscriber. A client that opens a session
/// and vanishes before it ever streams (for example OtOpcUa's <c>GalaxyDriver</c> failing
/// part-way through its start-up) otherwise pins a session slot and a live x86 worker for
/// the full <see cref="DefaultLeaseSeconds"/> (30 minutes by default). The window is
/// measured from the session's last client activity — open or the most recent unary
/// command — so a client that only ever reads and writes is not reaped while it keeps
/// calling, and it stops applying the moment the session's first subscriber attaches
/// (from then on <see cref="DetachGraceSeconds"/> and the normal lease govern it). A value
/// of <c>0</c> (the default) disables the sweep. Must be greater than or equal to zero.
/// </summary>
/// <remarks>
/// Set it comfortably above the time a well-behaved client needs between
/// <c>OpenSession</c> and its first <c>StreamEvents</c>, and remember the close lands on
/// the next sweep after the window elapses (up to <see cref="LeaseSweepIntervalSeconds"/>
/// later).
/// </remarks>
public int UnattachedSessionGraceSeconds { get; init; }
/// <summary>
/// Gets a value indicating whether multiple event subscribers are allowed per session.
/// </summary>
@@ -30,6 +30,7 @@ public sealed class GatewaySession
private int _activeEventSubscriberCount;
private readonly TimeSpan _detachGrace;
private readonly TimeSpan _faultedGrace;
private readonly TimeSpan _unattachedGrace;
private readonly TimeSpan _workerReadyWaitTimeout;
private DateTimeOffset? _detachedAtUtc;
private DateTimeOffset? _faultedAtUtc;
@@ -169,6 +170,12 @@ public sealed class GatewaySession
/// The tags come from the owner key, never from the client's wire request, so a client
/// cannot label its own session with another tenant's tag.
/// </param>
/// <param name="unattachedGrace">
/// Idle window after which the lease monitor reaps a Ready session that has never attached
/// an event subscriber (see <see cref="IsUnattachedGraceExpired"/>), so a client that
/// opens and vanishes does not pin a worker for the full lease. <see cref="TimeSpan.Zero"/>
/// (the default) disables it.
/// </param>
public GatewaySession(
string sessionId,
string backendName,
@@ -188,7 +195,8 @@ public sealed class GatewaySession
TimeSpan workerReadyWaitTimeout = default,
ArrayAddressNormalizer? addressNormalizer = null,
TimeSpan faultedGrace = default,
IReadOnlyList<string>? ownerDashboardTags = null)
IReadOnlyList<string>? ownerDashboardTags = null,
TimeSpan unattachedGrace = default)
{
if (string.IsNullOrWhiteSpace(sessionId))
{
@@ -231,6 +239,7 @@ public sealed class GatewaySession
_eventStreaming = eventStreaming ?? SessionEventStreaming.Default;
_detachGrace = detachGrace > TimeSpan.Zero ? detachGrace : TimeSpan.Zero;
_faultedGrace = faultedGrace > TimeSpan.Zero ? faultedGrace : TimeSpan.Zero;
_unattachedGrace = unattachedGrace > TimeSpan.Zero ? unattachedGrace : TimeSpan.Zero;
_workerReadyWaitTimeout = workerReadyWaitTimeout > TimeSpan.Zero ? workerReadyWaitTimeout : TimeSpan.Zero;
_addressNormalizer = addressNormalizer;
}
@@ -1005,18 +1014,6 @@ public sealed class GatewaySession
}
}
/// <summary>
/// Extends the session lease to the specified expiration time.
/// </summary>
/// <param name="leaseExpiresAt">Timestamp when the lease expires.</param>
public void ExtendLease(DateTimeOffset leaseExpiresAt)
{
lock (_syncRoot)
{
_leaseExpiresAt = leaseExpiresAt;
}
}
/// <summary>
/// Determines whether the session lease has expired.
/// </summary>
@@ -1073,6 +1070,25 @@ public sealed class GatewaySession
}
}
/// <summary>
/// Determines whether a <see cref="SessionState.Ready"/> session that has <em>never</em>
/// attached an event-stream subscriber has been idle — no unary command since open or
/// since its last one — for at least the configured unattached grace. The lease monitor
/// reaps such a session exactly as it reaps an expired lease, so a client that opened and
/// vanished does not pin a worker for the full lease. Always <see langword="false"/> when
/// the grace is disabled, when the session ever had a subscriber (detach-grace and the
/// normal lease govern it from then on) or when one is attached now.
/// </summary>
/// <param name="now">Current timestamp for comparison.</param>
/// <returns><see langword="true"/> if the session is reapable as never-subscribed and idle; otherwise <see langword="false"/>.</returns>
public bool IsUnattachedGraceExpired(DateTimeOffset now)
{
lock (_syncRoot)
{
return IsUnattachedGraceExpiredCore(now);
}
}
/// <summary>
/// Attaches an event subscriber and returns a lease whose
/// <see cref="IEventSubscriberLease.Reader"/> reads the fanned public
@@ -1851,7 +1867,10 @@ public sealed class GatewaySession
// Re-verify eligibility atomically. If a subscriber reattached between the sweep's
// eligibility check and this point, neither condition holds and we decline.
bool eligible = IsLeaseExpiredCore(now) || IsFaultedReapableCore(now) || IsDetachGraceExpiredCore(now);
bool eligible = IsLeaseExpiredCore(now)
|| IsFaultedReapableCore(now)
|| IsDetachGraceExpiredCore(now)
|| IsUnattachedGraceExpiredCore(now);
if (!eligible)
{
alreadyClosing = false;
@@ -1882,6 +1901,17 @@ public sealed class GatewaySession
|| _faultedAtUtc is null
|| now - _faultedAtUtc.Value >= _faultedGrace);
// Ready only: a session still starting its worker is not idle, it is waiting, and reaping it
// mid-startup would turn a slow worker start into an OpenFailed for a client that did nothing
// wrong. Idle is measured from the last client activity (open or last unary command) rather
// than from open so a unary-only client that keeps calling is never reaped by this rule.
private bool IsUnattachedGraceExpiredCore(DateTimeOffset now)
=> _unattachedGrace > TimeSpan.Zero
&& _state is SessionState.Ready
&& !_everHadEventSubscriber
&& _activeEventSubscriberCount == 0
&& now - _lastClientActivityAt >= _unattachedGrace;
// Final terminal transition; under _syncRoot to keep _state writes single-lock.
// Closed is unconditionally terminal — TransitionTo refuses to overwrite it —
// so we don't need to re-check the precondition here.
@@ -21,6 +21,7 @@ public sealed class SessionManager : ISessionManager
public const string LeaseExpiredReason = "lease-expired";
public const string DetachGraceExpiredReason = "detach-grace-expired";
public const string FaultedReason = "faulted-reaped";
public const string UnattachedGraceExpiredReason = "unattached-grace-expired";
// Bounded so a mass expiry (or a host stop with a full registry) cannot stampede
// worker-process teardown: every concurrent close is one x86 worker being shut down or
@@ -363,12 +364,14 @@ public sealed class SessionManager : ISessionManager
{
// A session is swept when its normal lease has expired, it has FAULTED (a faulted
// session is permanently unusable yet otherwise pins a session slot and a live x86
// worker until its DefaultLeaseSeconds lease expires), OR its detach-grace retention
// worker until its DefaultLeaseSeconds lease expires), its detach-grace retention
// window has elapsed (last external subscriber dropped and no client reconnected
// within DetachGraceSeconds). All three are the same teardown; only the reason differs
// so operators can tell a fault reap from a short reconnect-window expiry from a long
// idle-lease expiry in logs/metrics. Precedence when several fire simultaneously:
// lease-expired, then faulted, then detach-grace.
// within DetachGraceSeconds), OR it never attached a subscriber and has been idle for
// UnattachedSessionGraceSeconds (a client that opened and vanished). All four are the
// same teardown; only the reason differs so operators can tell a fault reap from a
// short reconnect-window expiry from a never-subscribed reap from a long idle-lease
// expiry in logs/metrics. Precedence when several fire simultaneously: lease-expired,
// then faulted, then detach-grace, then unattached-grace.
//
// TOCTOU note: eligibility is re-verified atomically inside TryBeginCloseIfExpired
// under _syncRoot, so a client that reattaches a subscriber between the check above
@@ -379,7 +382,9 @@ public sealed class SessionManager : ISessionManager
? FaultedReason
: session.IsDetachGraceExpired(now)
? DetachGraceExpiredReason
: null;
: session.IsUnattachedGraceExpired(now)
? UnattachedGraceExpiredReason
: null;
if (reason is null)
{
continue;
@@ -708,7 +713,8 @@ public sealed class SessionManager : ISessionManager
TimeSpan.FromMilliseconds(Math.Max(0, _options.Sessions.WorkerReadyWaitTimeoutMs)),
_addressNormalizer,
TimeSpan.FromSeconds(Math.Max(0, _options.Sessions.FaultedGraceSeconds)),
ownerDashboardTags);
ownerDashboardTags,
TimeSpan.FromSeconds(Math.Max(0, _options.Sessions.UnattachedSessionGraceSeconds)));
}
private static string CreateClientCorrelationId(
@@ -51,6 +51,7 @@
"DefaultLeaseSeconds": 1800,
"LeaseSweepIntervalSeconds": 30,
"DetachGraceSeconds": 30,
"UnattachedSessionGraceSeconds": 0,
"AllowMultipleEventSubscribers": false,
"MaxEventSubscribersPerSession": 8
},
@@ -619,6 +619,34 @@ public sealed class GatewayOptionsValidatorTests
Assert.True(result.Succeeded);
}
/// <summary>Verifies a negative <see cref="SessionOptions.UnattachedSessionGraceSeconds"/> fails validation.</summary>
[Fact]
public void Validate_Fails_WhenUnattachedSessionGraceSecondsIsNegative()
{
GatewayOptions options = CloneWithSessions(
ValidOptions(),
new SessionOptions { UnattachedSessionGraceSeconds = -1 });
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
Assert.True(result.Failed);
Assert.Contains(
result.Failures!,
f => f.Contains("MxGateway:Sessions:UnattachedSessionGraceSeconds"));
}
/// <summary>Verifies a zero (off) and a positive <see cref="SessionOptions.UnattachedSessionGraceSeconds"/> pass validation.</summary>
/// <param name="seconds">Grace window under test.</param>
[Theory]
[InlineData(0)]
[InlineData(120)]
public void Validate_Succeeds_WhenUnattachedSessionGraceSecondsIsZeroOrPositive(int seconds)
{
GatewayOptions options = CloneWithSessions(
ValidOptions(),
new SessionOptions { UnattachedSessionGraceSeconds = seconds });
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
Assert.True(result.Succeeded);
}
// -------------------------------------------------------------------------
// ReplayBufferCapacity / ReplayRetentionSeconds validation
// -------------------------------------------------------------------------
@@ -1129,8 +1129,8 @@ public sealed class SessionManagerTests
GatewaySession expiredSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
GatewaySession activeSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None);
DateTimeOffset now = DateTimeOffset.UtcNow;
expiredSession.ExtendLease(now.AddSeconds(-1));
activeSession.ExtendLease(now.AddMinutes(5));
SetLeaseExpiry(expiredSession, now.AddSeconds(-1));
SetLeaseExpiry(activeSession, now.AddMinutes(5));
int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None);
@@ -1194,6 +1194,130 @@ public sealed class SessionManagerTests
Assert.Equal(0, snapshot.OpenSessions);
}
/// <summary>
/// With <c>UnattachedSessionGraceSeconds</c> set, a session that never attached an event
/// subscriber and has been idle for the window is reaped by the ordinary lease sweep even
/// though its 30-minute lease is nowhere near expiry. This is the knob for clients that
/// open a session and vanish (OtOpcUa's GalaxyDriver on a failed start): without it each
/// such open pins an x86 worker for <c>DefaultLeaseSeconds</c>.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task CloseExpiredLeasesAsync_ReapsNeverSubscribedSessionAfterUnattachedGrace()
{
FakeWorkerClient workerClient = new();
FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
using GatewayMetrics metrics = new();
SessionManager manager = CreateManager(
new FakeSessionWorkerClientFactory(workerClient),
metrics: metrics,
options: CreateOptions(unattachedSessionGraceSeconds: 60),
timeProvider: clock);
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
clock.Advance(TimeSpan.FromSeconds(59));
int closedBefore = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(0, closedBefore);
Assert.Equal(SessionState.Ready, session.State);
clock.Advance(TimeSpan.FromSeconds(1));
int closedAfter = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(1, closedAfter);
Assert.Equal(SessionState.Closed, session.State);
Assert.Equal(1, workerClient.ShutdownCount);
Assert.False(manager.TryGetSession(session.SessionId, out _));
Assert.Equal(1, metrics.GetSnapshot().SessionsClosed);
Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
}
/// <summary>
/// The window is an idle timeout, not a hard deadline from open: unary activity restarts
/// it, so a client that only ever reads and writes (never streams) is not reaped while it
/// keeps calling.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task CloseExpiredLeasesAsync_UnattachedGrace_IsRestartedByUnaryActivity()
{
FakeWorkerClient workerClient = new();
FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
SessionManager manager = CreateManager(
new FakeSessionWorkerClientFactory(workerClient),
options: CreateOptions(unattachedSessionGraceSeconds: 60),
timeProvider: clock);
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
clock.Advance(TimeSpan.FromSeconds(59));
await manager.InvokeAsync(session.SessionId, CreateCommand(MxCommandKind.Ping), CancellationToken.None);
clock.Advance(TimeSpan.FromSeconds(59));
int closedWhileActive = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(0, closedWhileActive);
Assert.Equal(SessionState.Ready, session.State);
clock.Advance(TimeSpan.FromSeconds(1));
int closedAfterIdle = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(1, closedAfterIdle);
Assert.Equal(SessionState.Closed, session.State);
}
/// <summary>
/// A session that ever had an event subscriber is outside this knob's remit — once it has
/// streamed, detach-grace and the normal lease govern it — and a session with a subscriber
/// attached right now is never idle.
/// </summary>
/// <param name="keepSubscriberAttached">Whether the subscriber is still attached at sweep time.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
[Theory]
[InlineData(true)]
[InlineData(false)]
public async Task CloseExpiredLeasesAsync_UnattachedGrace_DoesNotReapSessionThatSubscribed(bool keepSubscriberAttached)
{
FakeWorkerClient workerClient = new();
FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
SessionManager manager = CreateManager(
new FakeSessionWorkerClientFactory(workerClient),
options: CreateOptions(detachGraceSeconds: 0, unattachedSessionGraceSeconds: 60),
timeProvider: clock);
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
IDisposable subscriber = session.AttachEventSubscriber(maxSubscribers: 1);
if (!keepSubscriberAttached)
{
subscriber.Dispose();
}
clock.Advance(TimeSpan.FromSeconds(61));
int closedCount = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(0, closedCount);
Assert.Equal(SessionState.Ready, session.State);
Assert.Equal(0, workerClient.ShutdownCount);
subscriber.Dispose();
}
/// <summary>
/// The knob is off by default: a never-subscribed idle session lives for its full lease.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task CloseExpiredLeasesAsync_UnattachedGrace_IsDisabledByDefault()
{
FakeWorkerClient workerClient = new();
FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
SessionManager manager = CreateManager(
new FakeSessionWorkerClientFactory(workerClient),
options: CreateOptions(),
timeProvider: clock);
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
clock.Advance(TimeSpan.FromMinutes(20));
int closedCount = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(0, closedCount);
Assert.Equal(SessionState.Ready, session.State);
}
/// <summary>Verifies that an expired-lease sweep leaves a session with an active event subscriber open.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
@@ -1203,7 +1327,7 @@ public sealed class SessionManagerTests
SessionManager manager = CreateManager(new FakeSessionWorkerClientFactory(workerClient));
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
DateTimeOffset now = DateTimeOffset.UtcNow;
session.ExtendLease(now.AddSeconds(-1));
SetLeaseExpiry(session, now.AddSeconds(-1));
using IDisposable eventSubscriber = session.AttachEventSubscriber(maxSubscribers: 1);
int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None);
@@ -1310,8 +1434,8 @@ public sealed class SessionManagerTests
DateTimeOffset now = DateTimeOffset.UtcNow;
// Both leases are far in the future, so only the fault can reap the faulted session.
faultedSession.ExtendLease(now.AddMinutes(30));
healthySession.ExtendLease(now.AddMinutes(30));
SetLeaseExpiry(faultedSession, now.AddMinutes(30));
SetLeaseExpiry(healthySession, now.AddMinutes(30));
faultedSession.MarkFaulted("test fault");
int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None);
@@ -1372,8 +1496,8 @@ public sealed class SessionManagerTests
GatewaySession firstSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
GatewaySession secondSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None);
DateTimeOffset now = DateTimeOffset.UtcNow;
firstSession.ExtendLease(now.AddSeconds(-1));
secondSession.ExtendLease(now.AddSeconds(-1));
SetLeaseExpiry(firstSession, now.AddSeconds(-1));
SetLeaseExpiry(secondSession, now.AddSeconds(-1));
int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None);
@@ -1425,8 +1549,8 @@ public sealed class SessionManagerTests
GatewaySession failingSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
GatewaySession healthySession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None);
DateTimeOffset now = DateTimeOffset.UtcNow;
failingSession.ExtendLease(now.AddSeconds(-1));
healthySession.ExtendLease(now.AddSeconds(-1));
SetLeaseExpiry(failingSession, now.AddSeconds(-1));
SetLeaseExpiry(healthySession, now.AddSeconds(-1));
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
async () => await manager.CloseExpiredLeasesAsync(now, CancellationToken.None));
@@ -1592,7 +1716,8 @@ public sealed class SessionManagerTests
int defaultLeaseSeconds = 1800,
int detachGraceSeconds = 0,
int workerReadyWaitTimeoutMs = 0,
int faultedGraceSeconds = 0)
int faultedGraceSeconds = 0,
int unattachedSessionGraceSeconds = 0)
{
return new GatewayOptions
{
@@ -1604,6 +1729,7 @@ public sealed class SessionManagerTests
DetachGraceSeconds = detachGraceSeconds,
WorkerReadyWaitTimeoutMs = workerReadyWaitTimeoutMs,
FaultedGraceSeconds = faultedGraceSeconds,
UnattachedSessionGraceSeconds = unattachedSessionGraceSeconds,
},
Worker = new WorkerOptions
{