Skip to content
Open
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);

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
47 changes: 45 additions & 2 deletions src/IO.Ably.Shared/ClientOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,51 @@ public bool UseBinaryProtocol
/// </summary>
public TimeSpan ChannelRetryTimeout { get; set; } = Defaults.ChannelRetryTimeout;

private TimeSpan _realtimeRequestTimeout = Defaults.RealtimeRequestTimeout;

/// <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 a positive interval, 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 to zero, a negative value, or an
/// interval too large for the underlying timers to fire.</exception>
public TimeSpan RealtimeRequestTimeout
{
get => _realtimeRequestTimeout;

set
{
// A non-positive value fails quietly rather than loudly: zero turns the RTN14c
// connect timeout into a hot reconnect loop, and a negative one stops the timer
// firing at all. Timeout.InfiniteTimeSpan is -1ms, so the lower bound catches it -
// 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.
//
// 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.Zero || value.TotalMilliseconds > int.MaxValue)
{
throw new ArgumentOutOfRangeException(
nameof(RealtimeRequestTimeout),
value,
"RealtimeRequestTimeout must be a positive interval. 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 @@ -436,8 +481,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
7 changes: 5 additions & 2 deletions src/IO.Ably.Shared/Realtime/Connection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -181,10 +181,13 @@ private void HandleNetworkStateChange(NetworkState state)
/// <returns>recoveryKey.</returns>
public string CreateRecoveryKey()
{
// 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