Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ public IO_Ably_ConnectionDetailsMessageSerializer(MsgPack.Serialization.Serializ
}

protected override void PackToCore(MsgPack.Packer packer, IO.Ably.ConnectionDetails objectTree) {
packer.PackMapHeader(7);
packer.PackMapHeader(8);
this._serializer0.PackTo(packer, "clientId");
this._serializer0.PackTo(packer, objectTree.ClientId);
this._serializer0.PackTo(packer, "connectionKey");
Expand All @@ -44,6 +44,8 @@ protected override void PackToCore(MsgPack.Packer packer, IO.Ably.ConnectionDeta
this._serializer1.PackTo(packer, objectTree.ConnectionStateTtl);
this._serializer0.PackTo(packer, "maxFrameSize");
this._serializer2.PackTo(packer, objectTree.MaxFrameSize);
this._serializer0.PackTo(packer, "maxIdleInterval");
this._serializer1.PackTo(packer, objectTree.MaxIdleInterval);
this._serializer0.PackTo(packer, "maxInboundRate");
this._serializer2.PackTo(packer, objectTree.MaxInboundRate);
this._serializer0.PackTo(packer, "maxMessageSize");
Expand Down Expand Up @@ -112,6 +114,33 @@ protected override IO.Ably.ConnectionDetails UnpackFromCore(MsgPack.Unpacker unp
result.MaxFrameSize = nullable2.Value;
}
unpacked = (unpacked + 1);
System.Nullable<System.TimeSpan> nullableMaxIdleInterval = default(System.Nullable<System.TimeSpan>);
if ((unpacked < itemsCount)) {
if ((unpacker.Read() == false)) {
throw MsgPack.Serialization.SerializationExceptions.NewMissingItem(4);
}
if (((unpacker.IsArrayHeader == false)
&& (unpacker.IsMapHeader == false))) {
nullableMaxIdleInterval = this._serializer1.UnpackFrom(unpacker);
}
else {
MsgPack.Unpacker disposableMaxIdleInterval = default(MsgPack.Unpacker);
disposableMaxIdleInterval = unpacker.ReadSubtree();
try {
nullableMaxIdleInterval = this._serializer1.UnpackFrom(disposableMaxIdleInterval);
}
finally {
if (((disposableMaxIdleInterval == null)
== false)) {
disposableMaxIdleInterval.Dispose();
}
}
}
}
if (nullableMaxIdleInterval.HasValue) {
result.MaxIdleInterval = nullableMaxIdleInterval;
}
unpacked = (unpacked + 1);
System.Nullable<long> nullable3 = default(System.Nullable<long>);
if ((unpacked < itemsCount)) {
nullable3 = MsgPack.Serialization.UnpackHelpers.UnpackNullableInt64Value(unpacker, typeof(IO.Ably.ConnectionDetails), "Int64 maxInboundRate");
Expand Down Expand Up @@ -230,7 +259,35 @@ protected override IO.Ably.ConnectionDetails UnpackFromCore(MsgPack.Unpacker unp
}
}
else {
unpacker.Skip();
if ((key == "maxIdleInterval")) {
System.Nullable<System.TimeSpan> nullableMaxIdle = default(System.Nullable<System.TimeSpan>);
if ((unpacker.Read() == false)) {
throw MsgPack.Serialization.SerializationExceptions.NewMissingItem(i);
}
if (((unpacker.IsArrayHeader == false)
&& (unpacker.IsMapHeader == false))) {
nullableMaxIdle = this._serializer1.UnpackFrom(unpacker);
}
else {
MsgPack.Unpacker disposableMaxIdle = default(MsgPack.Unpacker);
disposableMaxIdle = unpacker.ReadSubtree();
try {
nullableMaxIdle = this._serializer1.UnpackFrom(disposableMaxIdle);
}
finally {
if (((disposableMaxIdle == null)
== false)) {
disposableMaxIdle.Dispose();
}
}
}
if (nullableMaxIdle.HasValue) {
result.MaxIdleInterval = nullableMaxIdle;
}
}
else {
unpacker.Skip();
}
}
}
}
Expand Down
27 changes: 24 additions & 3 deletions src/IO.Ably.Shared/AblyAuth.cs
Original file line number Diff line number Diff line change
Expand Up @@ -305,11 +305,27 @@ public virtual async Task<TokenDetails> RequestTokenAsync(TokenParams tokenParam
bool shouldCatch = true;
try
{
var callbackResult = await authOptions.AuthCallback(tokenParams);
// RSA4c bounds an auth attempt by realtimeRequestTimeout. This one is awaited on
// the realtime workflow's single reader thread, so an authCallback that never
// returns would stall channel state, inbound messages and the idle monitor.
//
// Task.Run rather than awaiting the delegate directly: TimeoutAfter extends an
// already-created Task, so a callback whose body runs synchronously - the most
// ordinary shape in C# - would block before there is anything to bound. It
// cannot cancel the callback; an abandoned one runs to completion.
var callbackResult = await Task.Run(() => authOptions.AuthCallback(tokenParams))
.TimeoutAfter(Options.RealtimeRequestTimeout, null);
Comment thread
AndyTWF marked this conversation as resolved.

if (callbackResult == null)
{
throw new AblyException("AuthCallback returned null", ErrorCodes.ClientAuthProviderRequestFailed);
// Covers a callback that returned null and one that timed out - TimeoutAfter
// yields null for both, and the wrapper below replaces the message anyway.
// ClientCallbackError, not ClientAuthProviderRequestFailed: the wrapper
// rethrows as 80019 per RSA4c1 with this as the cause, so 80019 here would
// duplicate its own parent. ably-js uses 40170 for the same case.
throw new AblyException(
$"AuthCallback returned null or did not return within {Options.RealtimeRequestTimeout.TotalSeconds}s",
ErrorCodes.ClientCallbackError);
}

if (callbackResult is TokenDetails)
Expand Down Expand Up @@ -341,12 +357,17 @@ public virtual async Task<TokenDetails> RequestTokenAsync(TokenParams tokenParam
: HttpStatusCode.Unauthorized;
}

// RSA4c1 wants the cause "set to the underlying cause". The four argument overload
// assigns InnerException instead, which is not the spec's field and is not
// serialised as one.
throw new AblyException(
new ErrorInfo(
"Error calling AuthCallback, token request failed. See inner exception for details.",
ErrorCodes.ClientAuthProviderRequestFailed,
statusCode,
ex),
href: null,
cause: (ex as AblyException)?.ErrorInfo ?? new ErrorInfo(ex.Message),
innerException: ex),
ex);
}
}
Expand Down
7 changes: 7 additions & 0 deletions src/IO.Ably.Shared/AuthOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,13 @@ public class AuthOptions
/// <summary>
/// The callback used to get a new <see cref="IO.Ably.TokenDetails"/> or <see cref="IO.Ably.TokenRequest"/>.
/// AuthCallback is used by internally by <see cref="IO.Ably.AblyAuth"/>.RequestTokenAsync.
/// <para>
/// The callback is bounded by <see cref="ClientOptions.RealtimeRequestTimeout"/> per RSA4c,
/// but cannot be cancelled: one that overruns is abandoned and keeps running, so a later auth
/// attempt may invoke it again concurrently and implementations must tolerate that. A result
/// returned after the bound is discarded, and each abandoned invocation holds a thread pool
/// worker until it returns.
/// </para>
/// </summary>
public Func<TokenParams, Task<object>> AuthCallback { get; set; }

Expand Down
77 changes: 74 additions & 3 deletions src/IO.Ably.Shared/ClientOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,54 @@ public bool UseBinaryProtocol
/// </summary>
public TimeSpan ChannelRetryTimeout { get; set; } = Defaults.ChannelRetryTimeout;

private TimeSpan _realtimeRequestTimeout = Defaults.RealtimeRequestTimeout;
private int _heartbeatMonitorDelay = 1000;

/// <summary>
/// How long the library waits for Ably to answer before treating a realtime request as
/// having failed. Applies while establishing a connection, while awaiting a response to a
/// Heartbeat, Connect, Attach, Detach or Close, and as part of the RTN23a idle timeout.
/// Default: 10s. Must be at least one millisecond, and values beyond a minute or so are
/// rarely useful. Disabling the timeout is not supported - the timeouts this value drives are
/// all required to fire - so both Timeout.InfiniteTimeSpan and TimeSpan.MaxValue are rejected.
/// TO3l11 - https://sdk.ably.com/builds/ably/specification/main/features/#TO3l11.
/// </summary>
/// <exception cref="ArgumentOutOfRangeException">when set below one millisecond, or to an
/// interval too large for the underlying timers to fire.</exception>
public TimeSpan RealtimeRequestTimeout
{
get => _realtimeRequestTimeout;

set
{
// The lower bound is one millisecond, not zero. CountdownTimer hands the delay to
// System.Threading.Timer as (int)TotalMilliseconds, so anything under a millisecond
// truncates to a zero delay timer - the same hot loop a literal zero produces, and
// just as quiet. A negative value stops the timer firing at all, and
// Timeout.InfiniteTimeSpan is -1ms, which matters because both Task.Delay and
// System.Threading.Timer accept it as genuine infinity and it would disable RTN14c,
// RTN12b, RTL4f, RTL5f and RSA4c outright.
//
// The upper bound is the tightest limit across the sinks this value reaches on every
// framework shipped: Task.Delay allows uint.MaxValue - 1 ms on .NET 6+ but only
// Int32.MaxValue on .NET Framework, Mono and Xamarin, and CountdownTimer casts to
// int for System.Threading.Timer. So Int32.MaxValue ms - a bound on the arithmetic,
// not a supported configuration, which is why the message names the useful range.
if (value < TimeSpan.FromMilliseconds(1) || value.TotalMilliseconds > int.MaxValue)
{
throw new ArgumentOutOfRangeException(
nameof(RealtimeRequestTimeout),
value,
"RealtimeRequestTimeout must be at least one millisecond. The default is 10s; " +
"values beyond a minute or so are rarely useful. Disabling the timeout is " +
"not supported - the connect, close, attach, detach and auth timeouts it " +
"drives are all required to fire.");
}

_realtimeRequestTimeout = value;
}
}

/// <summary>
/// Timeout for opening an http request.
/// Default: 4s.
Expand Down Expand Up @@ -390,7 +438,32 @@ public bool UseBinaryProtocol
/// connection has been lost.
/// Defaults: 1000.
/// </summary>
public int HeartbeatMonitorDelay { get; set; } = 1000;
public int HeartbeatMonitorDelay
{
get => _heartbeatMonitorDelay;

set
{
// This is the granularity of RTN23a idle detection, and the monitor driving it is a
// fire-and-forget loop, so a value it cannot wait on takes detection out for the life
// of the client. Zero is a hot loop, queueing a command per scheduler tick. Minus one
// is Timeout.Infinite, which Task.Delay accepts as genuine infinity, so the monitor
// ticks once and is then silent for good - nothing thrown, nothing logged. Below
// minus one it throws inside the loop instead. None of the three can be what a caller
// meant, so this rejects rather than clamping and quietly overriding them.
if (value < 1)
{
throw new ArgumentOutOfRangeException(
nameof(HeartbeatMonitorDelay),
value,
"HeartbeatMonitorDelay must be at least one millisecond. The default is 1000. " +
"It is how often RTN23a idle detection is evaluated, so a large value delays " +
"noticing a dead connection, and turning it off is not supported.");
}

_heartbeatMonitorDelay = value;
}
}

/// <summary>
/// If enabled, every REST request to Ably includes a `request_id` query string parameter.
Expand Down Expand Up @@ -436,8 +509,6 @@ internal Func<DateTimeOffset> NowFunc

internal bool SkipInternetCheck { get; set; }

internal TimeSpan RealtimeRequestTimeout { get; set; } = Defaults.RealtimeRequestTimeout;

/// <summary>
/// Default constructor for ClientOptions.
/// </summary>
Expand Down
15 changes: 12 additions & 3 deletions src/IO.Ably.Shared/Realtime/Connection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -178,13 +178,22 @@ private void HandleNetworkStateChange(NetworkState state)
/// <summary>
/// Connection#CreateRecoveryKey is an attribute composed of the connectionKey, messageSerial and channelSerials (RTN16g, RTN16g1, RTN16h).
/// </summary>
/// <returns>recoveryKey.</returns>
/// <returns>
/// The recovery key, or <see cref="string.Empty"/> where RTN16g3 calls for null. This SDK
/// returns empty strings rather than nulls for absent string values throughout, and callers
/// pass the result straight back as ClientOptions.Recover, which treats the two alike - so
/// returning null instead would break every consumer testing the result with IsNotEmpty for
/// no behavioural gain. ably-js returns null here.
/// </returns>
public string CreateRecoveryKey()
Comment thread
AndyTWF marked this conversation as resolved.
{
// RTN16g3, which replaces RTN16g2 as of specification 6.1.0 - null in CLOSED, CLOSING
// and FAILED, and SUSPENDED is deliberately not among them. RTN8d and RTN9d keep the
// key through SUSPENDED because RTN14h always attempts a resume, so the connection is
// still recoverable there and the key has to be available to hand over.
if (Key.IsEmpty() || InnerState.State == Realtime.ConnectionState.Closing
|| InnerState.State == Realtime.ConnectionState.Closed
|| InnerState.State == Realtime.ConnectionState.Failed
|| InnerState.State == Realtime.ConnectionState.Suspended)
|| InnerState.State == Realtime.ConnectionState.Failed)
{
return string.Empty;
}
Expand Down
13 changes: 12 additions & 1 deletion src/IO.Ably.Shared/Realtime/Presence.cs
Original file line number Diff line number Diff line change
Expand Up @@ -792,11 +792,22 @@ private void SendQueuedMessages()

private void FailQueuedMessages(ErrorInfo reason)
{
// RTL11 wants "an ErrorInfo indicating the failure". TaskWrapper has no branch for
// (false, null) and faults the task with a bare Exception, so the reason is coalesced
// here rather than at each call site.
var error = reason ?? ErrorInfo.ReasonUnknown;

while (!PendingPresenceQueue.IsEmpty)
{
if (PendingPresenceQueue.TryDequeue(out var queuedPresenceMessage))
{
queuedPresenceMessage.Callback?.Invoke(false, reason);
// Guarded like every other callback site: a throwing application callback would
// otherwise abort the rest of the queue and skip the RTP5a map clearing in
// ChannelDetachedOrFailed, which RealtimeChannels then swallows per channel.
ActionUtils.SafeExecute(
() => queuedPresenceMessage.Callback?.Invoke(false, error),
Logger,
nameof(FailQueuedMessages));
}
}
}
Expand Down
Loading
Loading