HeiKyu/Rendezvous
2
0
Fork 0
Code Issues 27 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement scoped join attempts and tickets (#10)
quality-gate / quality (push) Successful in 1m1s
Details

Browse Source 
Closes #10
This commit is contained in:
KyuubiYoru
2026-07-16 06:56:30 +02:00
parent 06c3973ce7
commit 1baa1055dc
30 changed files with 2057 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/FinalFactory.Rendezvous.Server/State/EphemeralStateContracts.cs
+45 -2
View File
@@ -35,6 +35,7 @@ internal sealed record EphemeralStoreOptions
public TimeSpan LeaseLifetime { get; init; } = TimeSpan.FromSeconds(60);
public TimeSpan PresenceLifetime { get; init; } = TimeSpan.FromSeconds(20);
public TimeSpan JoinAttemptLifetime { get; init; } = TimeSpan.FromSeconds(30);
public TimeSpan ConnectionTicketLifetime { get; init; } = TimeSpan.FromSeconds(20);
public TimeSpan ReplayLifetime { get; init; } = TimeSpan.FromSeconds(30);
public TimeSpan IdempotencyLifetime { get; init; } = TimeSpan.FromMinutes(2);
public TimeSpan GracefulDrainLifetime { get; init; } = TimeSpan.FromSeconds(30);
@@ -50,9 +51,17 @@ internal sealed record EphemeralStoreOptions
RequireDuration(LeaseLifetime, TimeSpan.FromSeconds(60), nameof(LeaseLifetime));
RequireDuration(PresenceLifetime, TimeSpan.FromSeconds(20), nameof(PresenceLifetime));
RequireDuration(JoinAttemptLifetime, TimeSpan.FromSeconds(30), nameof(JoinAttemptLifetime));
RequireDuration(ConnectionTicketLifetime, TimeSpan.FromSeconds(20), nameof(ConnectionTicketLifetime));
RequireDuration(ReplayLifetime, TimeSpan.FromSeconds(30), nameof(ReplayLifetime));
RequireDuration(IdempotencyLifetime, TimeSpan.FromMinutes(10), nameof(IdempotencyLifetime));
RequireDuration(GracefulDrainLifetime, TimeSpan.FromSeconds(30), nameof(GracefulDrainLifetime));
if (ConnectionTicketLifetime > JoinAttemptLifetime)
{
throw new ArgumentOutOfRangeException(
nameof(ConnectionTicketLifetime),
"Connection tickets cannot outlive their join attempt.");
}
if (IdempotencyLifetime < LeaseLifetime || IdempotencyLifetime < JoinAttemptLifetime)
{
throw new ArgumentOutOfRangeException(
@@ -246,7 +255,11 @@ internal sealed record CreateJoinAttemptCommand
public required uint ProtocolVersion { get; init; }
public required SecretFingerprint HostCapabilityFingerprint { get; init; }
public required SecretFingerprint ClientCapabilityFingerprint { get; init; }
public required SecretFingerprint ConnectionTicketFingerprint { get; init; }
public required string CapabilityDerivationSalt { get; init; }
public int ScopeAttemptLimit { get; init; } = int.MaxValue;
public override string ToString() => "[CreateJoinAttemptCommand: credentials redacted]";
}
internal sealed record AttemptEndpointBinding(
@@ -261,12 +274,28 @@ internal sealed record StoredJoinAttempt
public required SessionListingId ListingId { get; init; }
public required string ClientSubject { get; init; }
public required uint ProtocolVersion { get; init; }
public required string IdempotencyKey { get; init; }
public required string RequestFingerprint { get; init; }
public required string CapabilityDerivationSalt { get; init; }
public required SecretFingerprint HostCapabilityFingerprint { get; init; }
public required SecretFingerprint ClientCapabilityFingerprint { get; init; }
public required SecretFingerprint ConnectionTicketFingerprint { get; init; }
public required DateTimeOffset ExpiresAt { get; init; }
public required DateTimeOffset ConnectionTicketExpiresAt { get; init; }
public AttemptEndpointBinding? HostEndpoint { get; init; }
public AttemptEndpointBinding? ClientEndpoint { get; init; }
public required bool IntroductionConsumed { get; init; }
public required bool ConnectionTicketConsumed { get; init; }
public override string ToString() => $"[StoredJoinAttempt {AttemptId}; credentials redacted]";
}
internal sealed record HostJoinAttemptQuery(
SessionListingId ListingId,
SecretFingerprint LeaseFingerprint,
int MaximumResults,
JoinAttemptId? AfterAttemptId = null);
internal sealed record BindAttemptEndpointCommand(
MediationHandle Handle,
AttemptPeerRole Role,
@@ -275,9 +304,20 @@ internal sealed record BindAttemptEndpointCommand(
ObservedEndpoint? LocalEndpoint);
internal sealed record IntroductionEndpoints(
JoinAttemptId AttemptId,
StoredJoinAttempt Attempt,
AttemptEndpointBinding Host,
AttemptEndpointBinding Client);
AttemptEndpointBinding Client)
{
public JoinAttemptId AttemptId => Attempt.AttemptId;
}
internal sealed record CancelJoinAttemptCommand(
JoinAttemptId AttemptId,
SecretFingerprint ClientCapabilityFingerprint);
internal sealed record ConsumeConnectionTicketCommand(
JoinAttemptId AttemptId,
SecretFingerprint ConnectionTicketFingerprint);
internal sealed record ReplayConsumption(
string Namespace,
@@ -316,8 +356,11 @@ internal interface IEphemeralRendezvousStore
StoreResult<IReadOnlyList<StoredListing>> BrowseVisibleListings(VisibleListingQuery query, CancellationToken cancellationToken = default);
StoreResult<StoredListing> BindHostPresence(BindHostPresenceCommand command, CancellationToken cancellationToken = default);
StoreResult<StoredJoinAttempt> CreateJoinAttempt(CreateJoinAttemptCommand command, CancellationToken cancellationToken = default);
StoreResult<IReadOnlyList<StoredJoinAttempt>> BrowseHostJoinAttempts(HostJoinAttemptQuery query, CancellationToken cancellationToken = default);
StoreResult<bool> CancelJoinAttempt(CancelJoinAttemptCommand command, CancellationToken cancellationToken = default);
StoreResult<StoredJoinAttempt> BindAttemptEndpoint(BindAttemptEndpointCommand command, CancellationToken cancellationToken = default);
StoreResult<IntroductionEndpoints> ConsumeIntroduction(MediationHandle handle, CancellationToken cancellationToken = default);
StoreResult<bool> ConsumeConnectionTicket(ConsumeConnectionTicketCommand command, CancellationToken cancellationToken = default);
StoreResult<bool> ConsumeReplay(ReplayConsumption consumption, CancellationToken cancellationToken = default);
StoreResult<bool> RevokeListing(SessionListingId listingId, CancellationToken cancellationToken = default);
StoreResult<int> RevokePrincipal(string subject, TimeSpan lifetime, CancellationToken cancellationToken = default);
src/FinalFactory.Rendezvous.Server/State/InMemoryEphemeralRendezvousStore.cs
+121 -2
View File
@@ -390,6 +390,66 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
return new(StoreResultCode.Success, Snapshot(attempt));
}, cancellationToken);
public StoreResult<IReadOnlyList<StoredJoinAttempt>> BrowseHostJoinAttempts(
HostJoinAttemptQuery query,
CancellationToken cancellationToken = default) => Atomic<IReadOnlyList<StoredJoinAttempt>>(_ =>
{
ArgumentNullException.ThrowIfNull(query);
if (query.ListingId.Value == Guid.Empty
|| !query.LeaseFingerprint.IsValid
|| query.MaximumResults is < 1 or > ContractLimits.BrowserPageMaxItems + 1)
{
throw new ArgumentException("Host attempt query invariants are invalid.", nameof(query));
}
if (!_available)
{
return new(StoreResultCode.ServiceUnavailable);
}
if (!_listings.TryGetValue(query.ListingId, out ListingEntry? listing)
|| listing.Definition.LeaseFingerprint != query.LeaseFingerprint)
{
return new(StoreResultCode.NotFound);
}
IReadOnlyList<StoredJoinAttempt> attempts = _attempts.Values
.Where(entry => entry.Command.ListingId == query.ListingId
&& !entry.IntroductionConsumed
&& (!query.AfterAttemptId.HasValue
|| entry.Command.AttemptId.Value.CompareTo(query.AfterAttemptId.Value.Value) > 0))
.OrderBy(static entry => entry.Command.AttemptId.Value)
.Take(query.MaximumResults)
.Select(Snapshot)
.ToArray();
return new(StoreResultCode.Success, attempts);
}, cancellationToken);
public StoreResult<bool> CancelJoinAttempt(
CancelJoinAttemptCommand command,
CancellationToken cancellationToken = default) => Atomic<bool>(_ =>
{
ArgumentNullException.ThrowIfNull(command);
if (command.AttemptId.Value == Guid.Empty || !command.ClientCapabilityFingerprint.IsValid)
{
throw new ArgumentException("Join cancellation invariants are invalid.", nameof(command));
}
if (!_available)
{
return new(StoreResultCode.ServiceUnavailable);
}
if (!_attempts.TryGetValue(command.AttemptId, out AttemptEntry? attempt)
|| attempt.ClientCapabilityFingerprint != command.ClientCapabilityFingerprint)
{
return new(StoreResultCode.NotFound);
}
RemoveAttempt(command.AttemptId);
return new(StoreResultCode.Success, true);
}, cancellationToken);
public StoreResult<StoredJoinAttempt> BindAttemptEndpoint(
BindAttemptEndpointCommand command,
CancellationToken cancellationToken = default) => Atomic<StoredJoinAttempt>(_ =>
@@ -447,7 +507,7 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
public StoreResult<IntroductionEndpoints> ConsumeIntroduction(
MediationHandle handle,
CancellationToken cancellationToken = default) => Atomic<IntroductionEndpoints>(_ =>
CancellationToken cancellationToken = default) => Atomic<IntroductionEndpoints>(now =>
{
if (!_available)
{
@@ -471,12 +531,57 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
}
attempt.IntroductionConsumed = true;
TimeSpan ticketLifetime = TimeSpan.FromTicks(Math.Min(
_options.ConnectionTicketLifetime.Ticks,
(attempt.Deadline - now).Ticks));
attempt.TicketDeadline = now + ticketLifetime;
attempt.TicketWallExpiresAt = WallDeadline(now, ticketLifetime);
return new(StoreResultCode.Success, new(
attempt.Command.AttemptId,
Snapshot(attempt),
attempt.HostEndpoint,
attempt.ClientEndpoint));
}, cancellationToken);
public StoreResult<bool> ConsumeConnectionTicket(
ConsumeConnectionTicketCommand command,
CancellationToken cancellationToken = default) => Atomic<bool>(now =>
{
ArgumentNullException.ThrowIfNull(command);
if (command.AttemptId.Value == Guid.Empty || !command.ConnectionTicketFingerprint.IsValid)
{
throw new ArgumentException("Connection ticket invariants are invalid.", nameof(command));
}
if (!_available)
{
return new(StoreResultCode.ServiceUnavailable);
}
if (!_attempts.TryGetValue(command.AttemptId, out AttemptEntry? attempt)
|| attempt.ConnectionTicketFingerprint != command.ConnectionTicketFingerprint)
{
return new(StoreResultCode.NotFound);
}
if (!attempt.IntroductionConsumed)
{
return new(StoreResultCode.Conflict);
}
if (!attempt.TicketDeadline.HasValue || attempt.TicketDeadline.Value <= now)
{
return new(StoreResultCode.Expired);
}
if (attempt.ConnectionTicketConsumed)
{
return new(StoreResultCode.ReplayRejected);
}
attempt.ConnectionTicketConsumed = true;
return new(StoreResultCode.Success, true);
}, cancellationToken);
public StoreResult<bool> ConsumeReplay(
ReplayConsumption consumption,
CancellationToken cancellationToken = default) => Atomic<bool>(now =>
@@ -714,10 +819,18 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
ListingId = entry.Command.ListingId,
ClientSubject = entry.Command.ClientSubject,
ProtocolVersion = entry.Command.ProtocolVersion,
IdempotencyKey = entry.Command.IdempotencyKey,
RequestFingerprint = entry.Command.RequestFingerprint,
CapabilityDerivationSalt = entry.Command.CapabilityDerivationSalt,
HostCapabilityFingerprint = entry.Command.HostCapabilityFingerprint,
ClientCapabilityFingerprint = entry.Command.ClientCapabilityFingerprint,
ConnectionTicketFingerprint = entry.Command.ConnectionTicketFingerprint,
ExpiresAt = entry.WallExpiresAt,
ConnectionTicketExpiresAt = entry.TicketWallExpiresAt ?? default,
HostEndpoint = entry.HostEndpoint,
ClientEndpoint = entry.ClientEndpoint,
IntroductionConsumed = entry.IntroductionConsumed,
ConnectionTicketConsumed = entry.ConnectionTicketConsumed,
};
private static void RemoveExpired(Dictionary<string, TimeSpan> entries, TimeSpan now)
@@ -789,6 +902,8 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
|| command.ProtocolVersion == 0
|| !command.HostCapabilityFingerprint.IsValid
|| !command.ClientCapabilityFingerprint.IsValid
|| !command.ConnectionTicketFingerprint.IsValid
|| !IsDerivationSaltValid(command.CapabilityDerivationSalt)
|| command.ScopeAttemptLimit <= 0)
{
throw new ArgumentException("Join attempt invariants are invalid.", nameof(command));
@@ -853,11 +968,15 @@ internal sealed class InMemoryEphemeralRendezvousStore : IEphemeralRendezvousSto
public CreateJoinAttemptCommand Command { get; } = command;
public SecretFingerprint HostCapabilityFingerprint { get; } = command.HostCapabilityFingerprint;
public SecretFingerprint ClientCapabilityFingerprint { get; } = command.ClientCapabilityFingerprint;
public SecretFingerprint ConnectionTicketFingerprint { get; } = command.ConnectionTicketFingerprint;
public TimeSpan Deadline { get; } = deadline;
public DateTimeOffset WallExpiresAt { get; } = wallExpiresAt;
public TimeSpan? TicketDeadline { get; set; }
public DateTimeOffset? TicketWallExpiresAt { get; set; }
public AttemptEndpointBinding? HostEndpoint { get; set; }
public AttemptEndpointBinding? ClientEndpoint { get; set; }
public bool IntroductionConsumed { get; set; }
public bool ConnectionTicketConsumed { get; set; }
}
private sealed record IdempotencyEntry(
src/FinalFactory.Rendezvous.Server/State/StoreResultMapping.cs
+20
View File
@@ -0,0 +1,20 @@
using FinalFactory.Rendezvous.Contracts;
namespace FinalFactory.Rendezvous.Server.State;
internal static class StoreResultMapping
{
public static RendezvousErrorCode ToContractError(this StoreResultCode code) => code switch
{
StoreResultCode.Success => RendezvousErrorCode.None,
StoreResultCode.NotFound => RendezvousErrorCode.NotFound,
StoreResultCode.Expired => RendezvousErrorCode.Expired,
StoreResultCode.Revoked => RendezvousErrorCode.Forbidden,
StoreResultCode.Conflict => RendezvousErrorCode.Conflict,
StoreResultCode.CapacityExceeded => RendezvousErrorCode.CapacityExceeded,
StoreResultCode.ReplayRejected => RendezvousErrorCode.ReplayRejected,
StoreResultCode.Draining or StoreResultCode.ServiceUnavailable =>
RendezvousErrorCode.ServiceUnavailable,
_ => RendezvousErrorCode.InternalError,
};
}