feat(operations): add capacity and resilience gates (#18)

This commit is contained in:
KyuubiYoru
2026-07-16 15:57:01 +02:00
parent 08729ae25c
commit 609dad7cf1
21 changed files with 1759 additions and 107 deletions
@@ -1,3 +1,4 @@
using System.Runtime.CompilerServices;
[assembly: InternalsVisibleTo("FinalFactory.Rendezvous.Tests")]
[assembly: InternalsVisibleTo("FinalFactory.Rendezvous.Capacity")]
@@ -11,16 +11,29 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
private readonly DateTimeOffset _wallOrigin;
private readonly TimeSpan _monotonicOrigin;
private readonly Dictionary<SessionListingId, ListingEntry> _listings = [];
private readonly Dictionary<string, int> _listingCountsByOwner = new(StringComparer.Ordinal);
private readonly PriorityQueue<DeadlineEntry<SessionListingId>, long> _listingExpiries = new();
private readonly HashSet<SessionListingId> _scheduledListingExpiries = [];
private readonly Dictionary<LeaseId, SessionListingId> _leases = [];
private readonly Dictionary<MediationHandle, SessionListingId> _presenceHandles = [];
private readonly Dictionary<MediationHandle, PresenceEntry> _presence = [];
private readonly PriorityQueue<DeadlineEntry<MediationHandle>, long> _presenceExpiries = new();
private readonly HashSet<MediationHandle> _scheduledPresenceExpiries = [];
private readonly Dictionary<JoinAttemptId, AttemptEntry> _attempts = [];
private readonly PriorityQueue<AttemptExpiry, long> _attemptExpiries = new();
private readonly Dictionary<TenantScope, int> _attemptCountsByScope = [];
private readonly Dictionary<SessionListingId, HashSet<JoinAttemptId>> _attemptsByListing = [];
private readonly PriorityQueue<DeadlineEntry<JoinAttemptId>, long> _attemptExpiries = new();
private readonly Dictionary<JoinAttemptId, OutcomeReportEntry> _outcomeReports = [];
private readonly Dictionary<SessionListingId, HashSet<JoinAttemptId>> _outcomesByListing = [];
private readonly PriorityQueue<DeadlineEntry<JoinAttemptId>, long> _outcomeExpiries = new();
private readonly Dictionary<MediationHandle, JoinAttemptId> _attemptHandles = [];
private readonly Dictionary<string, IdempotencyEntry> _idempotency = new(StringComparer.Ordinal);
private readonly PriorityQueue<DeadlineEntry<string>, long> _idempotencyExpiries = new();
private readonly Dictionary<string, TimeSpan> _replay = new(StringComparer.Ordinal);
private readonly PriorityQueue<DeadlineEntry<string>, long> _replayExpiries = new();
private readonly Dictionary<string, TimeSpan> _revocations = new(StringComparer.Ordinal);
private readonly PriorityQueue<DeadlineEntry<string>, long> _revocationExpiries = new();
private readonly HashSet<string> _scheduledRevocationExpiries = new(StringComparer.Ordinal);
private TimeSpan? _drainDeadline;
private TimeSpan _nextUdpMaintenance;
private long _maintenanceSweepCount;
@@ -47,6 +60,22 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
public Guid InstanceId { get; }
internal long MaintenanceSweepCount => Interlocked.Read(ref _maintenanceSweepCount);
internal int ScheduledExpiryEntryCount
{
get
{
lock (_gate)
{
return _listingExpiries.Count
+ _presenceExpiries.Count
+ _attemptExpiries.Count
+ _outcomeExpiries.Count
+ _idempotencyExpiries.Count
+ _replayExpiries.Count
+ _revocationExpiries.Count;
}
}
}
public bool IsAvailable
{
@@ -161,10 +190,8 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
if (_listings.Count >= _options.MaxListings
|| _idempotency.Count >= _options.MaxIdempotencyEntries
|| _listings.Values.Count(entry => string.Equals(
entry.Definition.OwnerSubject,
command.Listing.OwnerSubject,
StringComparison.Ordinal)) >= command.OwnerListingLimit)
|| _listingCountsByOwner.GetValueOrDefault(command.Listing.OwnerSubject)
>= command.OwnerListingLimit)
{
return new(StoreResultCode.CapacityExceeded);
}
@@ -183,12 +210,21 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
WallDeadline(now, _options.LeaseLifetime),
version: 1);
_listings.Add(frozen.ListingId, entry);
ScheduleMutableDeadline(
_listingExpiries,
_scheduledListingExpiries,
frozen.ListingId,
entry.LeaseDeadline);
_listingCountsByOwner[frozen.OwnerSubject] =
_listingCountsByOwner.GetValueOrDefault(frozen.OwnerSubject) + 1;
_leases.Add(frozen.LeaseId, frozen.ListingId);
_presenceHandles.Add(frozen.HostPresenceHandle, frozen.ListingId);
_idempotency.Add(idempotencyKey, new(
IdempotencyEntry idempotency = new(
command.RequestFingerprint,
frozen.ListingId,
now + _options.IdempotencyLifetime));
now + _options.IdempotencyLifetime);
_idempotency.Add(idempotencyKey, idempotency);
EnqueueDeadline(_idempotencyExpiries, idempotencyKey, idempotency.Deadline);
return new(StoreResultCode.Success, Snapshot(entry));
}, cancellationToken);
@@ -346,10 +382,20 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
return new(StoreResultCode.CapacityExceeded);
}
_presence[command.Handle] = new(
bool isNewPresence = !_presence.ContainsKey(command.Handle);
PresenceEntry presence = new(
command.PublicEndpoint,
command.LocalEndpoint,
now + _options.PresenceLifetime);
_presence[command.Handle] = presence;
if (isNewPresence)
{
ScheduleMutableDeadline(
_presenceExpiries,
_scheduledPresenceExpiries,
command.Handle,
presence.Deadline);
}
return new(StoreResultCode.Success, Snapshot(entry));
}, cancellationToken, eagerCleanup: false);
@@ -444,8 +490,7 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
if (_attempts.Count >= _options.MaxJoinAttempts
|| _outcomeReports.Count >= _options.MaxOutcomeReports
|| _idempotency.Count >= _options.MaxIdempotencyEntries
|| _attempts.Values.Count(entry => entry.Command.Scope == command.Scope)
>= command.ScopeAttemptLimit)
|| _attemptCountsByScope.GetValueOrDefault(command.Scope) >= command.ScopeAttemptLimit)
{
return new(StoreResultCode.CapacityExceeded);
}
@@ -461,19 +506,25 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
now + _options.JoinAttemptLifetime,
WallDeadline(now, _options.JoinAttemptLifetime));
_attempts.Add(command.AttemptId, attempt);
_attemptExpiries.Enqueue(
new AttemptExpiry(command.AttemptId, attempt.Deadline),
attempt.Deadline.Ticks);
_outcomeReports.Add(command.AttemptId, new(
AddToIndex(_attemptsByListing, command.ListingId, command.AttemptId);
_attemptCountsByScope[command.Scope] =
_attemptCountsByScope.GetValueOrDefault(command.Scope) + 1;
EnqueueDeadline(_attemptExpiries, command.AttemptId, attempt.Deadline);
OutcomeReportEntry outcome = new(
command.ListingId,
command.ClientSubject,
command.ClientCapabilityFingerprint,
now + _options.JoinAttemptLifetime + _options.IdempotencyLifetime));
now + _options.JoinAttemptLifetime + _options.IdempotencyLifetime);
_outcomeReports.Add(command.AttemptId, outcome);
AddToIndex(_outcomesByListing, command.ListingId, command.AttemptId);
EnqueueDeadline(_outcomeExpiries, command.AttemptId, outcome.Deadline);
_attemptHandles.Add(command.MediationHandle, command.AttemptId);
_idempotency.Add(idempotencyKey, new(
IdempotencyEntry idempotency = new(
command.RequestFingerprint,
command.AttemptId,
now + _options.IdempotencyLifetime));
now + _options.IdempotencyLifetime);
_idempotency.Add(idempotencyKey, idempotency);
EnqueueDeadline(_idempotencyExpiries, idempotencyKey, idempotency.Deadline);
return new(StoreResultCode.Success, Snapshot(attempt));
}, cancellationToken);
@@ -757,7 +808,9 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
throw new ArgumentOutOfRangeException(nameof(consumption), "Replay lifetime exceeds the configured ceiling.");
}
_replay.Add(key, now + lifetime);
TimeSpan deadline = now + lifetime;
_replay.Add(key, deadline);
EnqueueDeadline(_replayExpiries, key, deadline);
return new(StoreResultCode.Success, true);
}, cancellationToken);
@@ -794,7 +847,24 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
+ _presence.Count
+ _attempts.Count
+ _outcomeReports.Count;
_revocations[subject] = now + lifetime;
TimeSpan deadline = now + lifetime;
bool isNewRevocation = !_revocations.TryGetValue(subject, out TimeSpan existingDeadline);
if (!isNewRevocation && existingDeadline > deadline)
{
// A repeated operator action may extend protection but cannot silently
// shorten an already-authoritative security revocation.
deadline = existingDeadline;
}
_revocations[subject] = deadline;
if (isNewRevocation)
{
ScheduleMutableDeadline(
_revocationExpiries,
_scheduledRevocationExpiries,
subject,
deadline);
}
SessionListingId[] listings = _listings
.Where(item => string.Equals(item.Value.Definition.OwnerSubject, subject, StringComparison.Ordinal))
.Select(static item => item.Key)
@@ -818,7 +888,7 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
}
foreach (JoinAttemptId attemptId in outcomeReports)
{
_outcomeReports.Remove(attemptId);
RemoveOutcome(attemptId);
}
int activeResourcesAfter = _listings.Count
@@ -911,63 +981,42 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
ClearActiveState();
}
_expiryChurn += RemoveExpired(_revocations, now);
_expiryChurn += RemoveExpired(_replay, now);
string[] expiredIdempotency = _idempotency
.Where(item => item.Value.Deadline <= now)
.Select(static item => item.Key)
.ToArray();
_expiryChurn += expiredIdempotency.Length;
foreach (string key in expiredIdempotency)
{
_idempotency.Remove(key);
}
MediationHandle[] expiredPresence = _presence
.Where(item => item.Value.Deadline <= now)
.Select(static item => item.Key)
.ToArray();
_expiryChurn += expiredPresence.Length;
foreach (MediationHandle handle in expiredPresence)
{
_presence.Remove(handle);
}
_expiryChurn += RemoveExpiredMutableDeadlines(
_revocations,
_revocationExpiries,
_scheduledRevocationExpiries,
now);
_expiryChurn += RemoveExpiredDeadlines(_replay, _replayExpiries, now);
_expiryChurn += RemoveExpiredIdempotency(now);
_expiryChurn += RemoveExpiredPresence(now);
_expiryChurn += RemoveExpiredAttempts(now);
JoinAttemptId[] expiredOutcomes = _outcomeReports
.Where(item => item.Value.Deadline <= now)
.Select(static item => item.Key)
.ToArray();
_expiryChurn += expiredOutcomes.Length;
foreach (JoinAttemptId attemptId in expiredOutcomes)
{
_outcomeReports.Remove(attemptId);
}
SessionListingId[] expiredListings = _listings
.Where(item => item.Value.LeaseDeadline <= now)
.Select(static item => item.Key)
.ToArray();
_expiryChurn += expiredListings.Length;
foreach (SessionListingId listingId in expiredListings)
{
RemoveListing(listingId);
}
_expiryChurn += RemoveExpiredOutcomes(now);
_expiryChurn += RemoveExpiredListings(now);
}
private void ClearActiveState()
{
_listings.Clear();
_listingCountsByOwner.Clear();
_listingExpiries.Clear();
_scheduledListingExpiries.Clear();
_leases.Clear();
_presenceHandles.Clear();
_presence.Clear();
_presenceExpiries.Clear();
_scheduledPresenceExpiries.Clear();
_attempts.Clear();
_attemptCountsByScope.Clear();
_attemptsByListing.Clear();
_attemptExpiries.Clear();
_outcomeReports.Clear();
_outcomesByListing.Clear();
_outcomeExpiries.Clear();
_attemptHandles.Clear();
_idempotency.Clear();
_idempotencyExpiries.Clear();
_replay.Clear();
_replayExpiries.Clear();
}
private void RemoveListing(SessionListingId listingId)
@@ -978,23 +1027,23 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
}
_leases.Remove(listing.Definition.LeaseId);
DecrementCount(_listingCountsByOwner, listing.Definition.OwnerSubject);
_presenceHandles.Remove(listing.Definition.HostPresenceHandle);
_presence.Remove(listing.Definition.HostPresenceHandle);
foreach (JoinAttemptId attemptId in _attempts
.Where(item => item.Value.Command.ListingId == listingId)
.Select(static item => item.Key)
.ToArray())
if (_attemptsByListing.TryGetValue(listingId, out HashSet<JoinAttemptId>? attempts))
{
RemoveAttempt(attemptId);
foreach (JoinAttemptId attemptId in attempts.ToArray())
{
RemoveAttempt(attemptId);
}
}
foreach (JoinAttemptId attemptId in _outcomeReports
.Where(item => item.Value.ListingId == listingId)
.Select(static item => item.Key)
.ToArray())
if (_outcomesByListing.TryGetValue(listingId, out HashSet<JoinAttemptId>? outcomes))
{
_outcomeReports.Remove(attemptId);
foreach (JoinAttemptId attemptId in outcomes.ToArray())
{
RemoveOutcome(attemptId);
}
}
}
@@ -1003,20 +1052,75 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
if (_attempts.Remove(attemptId, out AttemptEntry? attempt))
{
_attemptHandles.Remove(attempt.Command.MediationHandle);
DecrementCount(_attemptCountsByScope, attempt.Command.Scope);
RemoveFromIndex(_attemptsByListing, attempt.Command.ListingId, attemptId);
}
}
private void RemoveOutcome(JoinAttemptId attemptId)
{
if (_outcomeReports.Remove(attemptId, out OutcomeReportEntry? outcome))
{
RemoveFromIndex(_outcomesByListing, outcome.ListingId, attemptId);
}
}
private static void AddToIndex<TKey>(
Dictionary<TKey, HashSet<JoinAttemptId>> index,
TKey key,
JoinAttemptId attemptId)
where TKey : notnull
{
if (!index.TryGetValue(key, out HashSet<JoinAttemptId>? values))
{
values = [];
index.Add(key, values);
}
values.Add(attemptId);
}
private static void RemoveFromIndex<TKey>(
Dictionary<TKey, HashSet<JoinAttemptId>> index,
TKey key,
JoinAttemptId attemptId)
where TKey : notnull
{
if (index.TryGetValue(key, out HashSet<JoinAttemptId>? values)
&& values.Remove(attemptId)
&& values.Count == 0)
{
index.Remove(key);
}
}
private static void DecrementCount<TKey>(Dictionary<TKey, int> counts, TKey key)
where TKey : notnull
{
int remaining = counts[key] - 1;
if (remaining == 0)
{
counts.Remove(key);
}
else
{
counts[key] = remaining;
}
}
private int RemoveExpiredAttempts(TimeSpan now)
{
int removed = 0;
while (_attemptExpiries.TryPeek(out AttemptExpiry candidate, out long deadlineTicks)
while (_attemptExpiries.TryPeek(
out DeadlineEntry<JoinAttemptId> candidate,
out long deadlineTicks)
&& deadlineTicks <= now.Ticks)
{
_attemptExpiries.Dequeue();
if (_attempts.TryGetValue(candidate.AttemptId, out AttemptEntry? current)
if (_attempts.TryGetValue(candidate.Key, out AttemptEntry? current)
&& current.Deadline == candidate.Deadline)
{
RemoveAttempt(candidate.AttemptId);
RemoveAttempt(candidate.Key);
removed++;
}
}
@@ -1024,6 +1128,184 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
return removed;
}
private int RemoveExpiredListings(TimeSpan now)
{
int removed = 0;
while (_listingExpiries.TryPeek(
out DeadlineEntry<SessionListingId> candidate,
out long deadlineTicks)
&& deadlineTicks <= now.Ticks)
{
_listingExpiries.Dequeue();
_scheduledListingExpiries.Remove(candidate.Key);
if (!_listings.TryGetValue(candidate.Key, out ListingEntry? current))
{
continue;
}
if (current.LeaseDeadline > now)
{
ScheduleMutableDeadline(
_listingExpiries,
_scheduledListingExpiries,
candidate.Key,
current.LeaseDeadline);
}
else
{
RemoveListing(candidate.Key);
removed++;
}
}
return removed;
}
private int RemoveExpiredPresence(TimeSpan now)
{
int removed = 0;
while (_presenceExpiries.TryPeek(
out DeadlineEntry<MediationHandle> candidate,
out long deadlineTicks)
&& deadlineTicks <= now.Ticks)
{
_presenceExpiries.Dequeue();
_scheduledPresenceExpiries.Remove(candidate.Key);
if (!_presence.TryGetValue(candidate.Key, out PresenceEntry? current))
{
continue;
}
if (current.Deadline > now)
{
ScheduleMutableDeadline(
_presenceExpiries,
_scheduledPresenceExpiries,
candidate.Key,
current.Deadline);
}
else if (_presence.Remove(candidate.Key))
{
removed++;
}
}
return removed;
}
private int RemoveExpiredOutcomes(TimeSpan now)
{
int removed = 0;
while (_outcomeExpiries.TryPeek(
out DeadlineEntry<JoinAttemptId> candidate,
out long deadlineTicks)
&& deadlineTicks <= now.Ticks)
{
_outcomeExpiries.Dequeue();
if (_outcomeReports.TryGetValue(candidate.Key, out OutcomeReportEntry? current)
&& current.Deadline == candidate.Deadline)
{
RemoveOutcome(candidate.Key);
removed++;
}
}
return removed;
}
private int RemoveExpiredIdempotency(TimeSpan now)
{
int removed = 0;
while (_idempotencyExpiries.TryPeek(
out DeadlineEntry<string> candidate,
out long deadlineTicks)
&& deadlineTicks <= now.Ticks)
{
_idempotencyExpiries.Dequeue();
if (_idempotency.TryGetValue(candidate.Key, out IdempotencyEntry? current)
&& current.Deadline == candidate.Deadline
&& _idempotency.Remove(candidate.Key))
{
removed++;
}
}
return removed;
}
private static int RemoveExpiredDeadlines<TKey>(
Dictionary<TKey, TimeSpan> entries,
PriorityQueue<DeadlineEntry<TKey>, long> expiries,
TimeSpan now)
where TKey : notnull
{
int removed = 0;
while (expiries.TryPeek(out DeadlineEntry<TKey> candidate, out long deadlineTicks)
&& deadlineTicks <= now.Ticks)
{
expiries.Dequeue();
if (entries.TryGetValue(candidate.Key, out TimeSpan current)
&& current == candidate.Deadline
&& entries.Remove(candidate.Key))
{
removed++;
}
}
return removed;
}
private static int RemoveExpiredMutableDeadlines<TKey>(
Dictionary<TKey, TimeSpan> entries,
PriorityQueue<DeadlineEntry<TKey>, long> expiries,
HashSet<TKey> scheduled,
TimeSpan now)
where TKey : notnull
{
int removed = 0;
while (expiries.TryPeek(out DeadlineEntry<TKey> candidate, out long deadlineTicks)
&& deadlineTicks <= now.Ticks)
{
expiries.Dequeue();
scheduled.Remove(candidate.Key);
if (!entries.TryGetValue(candidate.Key, out TimeSpan current))
{
continue;
}
if (current > now)
{
ScheduleMutableDeadline(expiries, scheduled, candidate.Key, current);
}
else if (entries.Remove(candidate.Key))
{
removed++;
}
}
return removed;
}
private static void ScheduleMutableDeadline<TKey>(
PriorityQueue<DeadlineEntry<TKey>, long> expiries,
HashSet<TKey> scheduled,
TKey key,
TimeSpan deadline)
where TKey : notnull
{
if (scheduled.Add(key))
{
EnqueueDeadline(expiries, key, deadline);
}
}
private static void EnqueueDeadline<TKey>(
PriorityQueue<DeadlineEntry<TKey>, long> expiries,
TKey key,
TimeSpan deadline)
where TKey : notnull =>
expiries.Enqueue(new(key, deadline), deadline.Ticks);
private bool HandleExists(MediationHandle handle) =>
_presenceHandles.ContainsKey(handle) || _attemptHandles.ContainsKey(handle);
@@ -1063,20 +1345,6 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
IsCancelled = entry.IsCancelled,
};
private static int RemoveExpired(Dictionary<string, TimeSpan> entries, TimeSpan now)
{
string[] expired = entries
.Where(item => item.Value <= now)
.Select(static item => item.Key)
.ToArray();
foreach (string key in expired)
{
entries.Remove(key);
}
return expired.Length;
}
private static void ValidateListing(ListingDefinition listing)
{
ArgumentNullException.ThrowIfNull(listing);
@@ -1219,7 +1487,8 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
public bool IsCancelled { get; set; }
}
private readonly record struct AttemptExpiry(JoinAttemptId AttemptId, TimeSpan Deadline);
private readonly record struct DeadlineEntry<TKey>(TKey Key, TimeSpan Deadline)
where TKey : notnull;
private sealed class OutcomeReportEntry(
SessionListingId listingId,