Skip to content

CSHARP-3537: CSOT: retryable reads and writes #1717

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Jun 30, 2025
Merged

CSHARP-3537: CSOT: retryable reads and writes #1717

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a76715e
CSHARP-3537: CSOT: retryable reads and writes
Jun 25, 2025
3e5a4cd
pr
Jun 25, 2025
6cf58a2
PR
Jun 25, 2025
dc2d1ea
pr
Jun 26, 2025
fefc460
Fix NullReferenceException
Jun 26, 2025
e6b11d9
Pr
Jun 27, 2025
d936f03
Pr
Jun 27, 2025
cbb79b8
Remove unintended change
Jun 27, 2025
25ced32
Pr
Jun 28, 2025
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
9 changes: 5 additions & 4 deletions src/MongoDB.Driver/Core/Operations/BulkMixedWriteOperation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,10 +139,9 @@ public WriteConcern WriteConcern
public BulkWriteOperationResult Execute(OperationContext operationContext, IWriteBinding binding)
{
using (BeginOperation())
using (var context = RetryableWriteContext.Create(operationContext, binding, _retryRequested))
using (var context = RetryableWriteContext.Create(operationContext, binding, IsOperationRetryable()))
{
EnsureHintIsSupportedIfAnyRequestHasHint();
context.DisableRetriesIfAnyWriteRequestIsNotRetryable(_requests);
var helper = new BatchHelper(_requests, _isOrdered, _writeConcern);
foreach (var batch in helper.GetBatches())
{
Expand All @@ -155,10 +154,9 @@ public BulkWriteOperationResult Execute(OperationContext operationContext, IWrit
public async Task<BulkWriteOperationResult> ExecuteAsync(OperationContext operationContext, IWriteBinding binding)
{
using (BeginOperation())
using (var context = await RetryableWriteContext.CreateAsync(operationContext, binding, _retryRequested).ConfigureAwait(false))
using (var context = await RetryableWriteContext.CreateAsync(operationContext, binding, IsOperationRetryable()).ConfigureAwait(false))
{
EnsureHintIsSupportedIfAnyRequestHasHint();
context.DisableRetriesIfAnyWriteRequestIsNotRetryable(_requests);
var helper = new BatchHelper(_requests, _isOrdered, _writeConcern);
foreach (var batch in helper.GetBatches())
{
Expand All @@ -168,6 +166,9 @@ public async Task<BulkWriteOperationResult> ExecuteAsync(OperationContext operat
}
}

private bool IsOperationRetryable()
=> _retryRequested && _requests.All(r => r.IsRetryable());

private IDisposable BeginOperation() =>
// Execution starts with the first request
EventContext.BeginOperation(null, _requests.FirstOrDefault()?.RequestType.ToString().ToLower());
Expand Down
9 changes: 5 additions & 4 deletions src/MongoDB.Driver/Core/Operations/BulkUnmixedWriteOperationBase.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,9 +129,8 @@ public BulkWriteOperationResult Execute(OperationContext operationContext, Retry
public BulkWriteOperationResult Execute(OperationContext operationContext, IWriteBinding binding)
{
using (BeginOperation())
using (var context = RetryableWriteContext.Create(operationContext, binding, _retryRequested))
using (var context = RetryableWriteContext.Create(operationContext, binding, IsOperationRetryable()))
{
context.DisableRetriesIfAnyWriteRequestIsNotRetryable(_requests);
return Execute(operationContext, context);
}
}
Expand All @@ -146,9 +145,8 @@ public Task<BulkWriteOperationResult> ExecuteAsync(OperationContext operationCon
public async Task<BulkWriteOperationResult> ExecuteAsync(OperationContext operationContext, IWriteBinding binding)
{
using (BeginOperation())
using (var context = await RetryableWriteContext.CreateAsync(operationContext, binding, _retryRequested).ConfigureAwait(false))
using (var context = await RetryableWriteContext.CreateAsync(operationContext, binding, IsOperationRetryable()).ConfigureAwait(false))
{
context.DisableRetriesIfAnyWriteRequestIsNotRetryable(_requests);
return await ExecuteAsync(operationContext, context).ConfigureAwait(false);
}
}
Expand All @@ -159,6 +157,9 @@ public async Task<BulkWriteOperationResult> ExecuteAsync(OperationContext operat
protected abstract bool RequestHasHint(TWriteRequest request);

// private methods
private bool IsOperationRetryable()
=> _retryRequested && _requests.All(r => r.IsRetryable());

private IDisposable BeginOperation() =>
EventContext.BeginOperation(null, _requests.FirstOrDefault()?.RequestType.ToString().ToLower());

Expand Down
3 changes: 1 addition & 2 deletions src/MongoDB.Driver/Core/Operations/DeleteRequest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
*/

using MongoDB.Bson;
using MongoDB.Driver.Core.Connections;
using MongoDB.Driver.Core.Misc;

namespace MongoDB.Driver.Core.Operations
Expand All @@ -36,6 +35,6 @@ public DeleteRequest(BsonDocument filter)
public int Limit { get; init; }

// public methods
public override bool IsRetryable(ConnectionDescription connectionDescription) => Limit != 0;
public override bool IsRetryable() => Limit != 0;
}
}
3 changes: 1 addition & 2 deletions src/MongoDB.Driver/Core/Operations/InsertRequest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
*/

using MongoDB.Bson;
using MongoDB.Driver.Core.Connections;
using MongoDB.Driver.Core.Misc;

namespace MongoDB.Driver.Core.Operations
Expand All @@ -32,6 +31,6 @@ public InsertRequest(BsonDocument document)
public BsonDocument Document { get; }

// public methods
public override bool IsRetryable(ConnectionDescription connectionDescription) => true;
public override bool IsRetryable() => true;
}
}
17 changes: 7 additions & 10 deletions src/MongoDB.Driver/Core/Operations/RetryabilityHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,20 +135,17 @@ public static bool IsResumableChangeStreamException(Exception exception, int max
{
return exception is MongoException mongoException ? mongoException.HasErrorLabel(ResumableChangeStreamErrorLabel) : false;
}
else

if (exception is MongoCommandException commandException)
{
var commandException = exception as MongoCommandException;
if (commandException != null)
var code = (ServerErrorCode)commandException.Code;
if (__resumableChangeStreamErrorCodes.Contains(code))
{
var code = (ServerErrorCode)commandException.Code;
if (__resumableChangeStreamErrorCodes.Contains(code))
{
return true;
}
return true;
}

return __resumableChangeStreamExceptions.Contains(exception.GetType());
}

return __resumableChangeStreamExceptions.Contains(exception.GetType());
}

/// <summary>
Expand Down
112 changes: 64 additions & 48 deletions src/MongoDB.Driver/Core/Operations/RetryableReadContext.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,11 @@
*/

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MongoDB.Driver.Core.Bindings;
using MongoDB.Driver.Core.Misc;
using MongoDB.Driver.Core.Servers;

namespace MongoDB.Driver.Core.Operations
{
Expand All @@ -29,41 +31,33 @@ public static RetryableReadContext Create(OperationContext operationContext, IRe
var context = new RetryableReadContext(binding, retryRequested);
try
{
context.Initialize(operationContext);

ChannelPinningHelper.PinChannellIfRequired(
context.ChannelSource,
context.Channel,
context.Binding.Session);

return context;
context.Initialize(operationContext, null);
}
catch
{
context.Dispose();
throw;
}

ChannelPinningHelper.PinChannellIfRequired(context.ChannelSource, context.Channel, context.Binding.Session);
return context;
}

public static async Task<RetryableReadContext> CreateAsync(OperationContext operationContext, IReadBinding binding, bool retryRequested)
{
var context = new RetryableReadContext(binding, retryRequested);
try
{
await context.InitializeAsync(operationContext).ConfigureAwait(false);

ChannelPinningHelper.PinChannellIfRequired(
context.ChannelSource,
context.Channel,
context.Binding.Session);

return context;
await context.InitializeAsync(operationContext, null).ConfigureAwait(false);
}
catch
{
context.Dispose();
throw;
}

ChannelPinningHelper.PinChannellIfRequired(context.ChannelSource, context.Channel, context.Binding.Session);
return context;
}
#endregion

Expand Down Expand Up @@ -96,50 +90,72 @@ public void Dispose()
}
}

public void ReplaceChannel(IChannelHandle channel)
internal void Initialize(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers)
{
var attempt = 1;
while (true)
{
try
{
ReplaceChannelSource(Binding.GetReadChannelSource(operationContext, deprioritizedServers));
ReplaceChannel(ChannelSource.GetChannel(operationContext));
return;
}
catch (Exception ex)
{
var innerException = ex is MongoAuthenticationException mongoAuthenticationException ? mongoAuthenticationException.InnerException : ex;
if (RetryableReadOperationExecutor.ShouldRetryOperation(operationContext, this, innerException, attempt))
{
attempt++;
}
else
{
throw;
}
}
}
}

internal async Task InitializeAsync(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers)
{
var attempt = 1;
while (true)
{
try
{
ReplaceChannelSource(await Binding.GetReadChannelSourceAsync(operationContext, deprioritizedServers).ConfigureAwait(false));
ReplaceChannel(await ChannelSource.GetChannelAsync(operationContext).ConfigureAwait(false));
return;
}
catch (Exception ex)
{
var innerException = ex is MongoAuthenticationException mongoAuthenticationException ? mongoAuthenticationException.InnerException : ex;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not to continue extracting the exception in ShouldRetryOperation? Just to reuse the code.
Then you also can keep using the When pattern, so more compact code.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

if (RetryableReadOperationExecutor.ShouldRetryOperation(operationContext, this, innerException, attempt))
{
attempt++;
}
else
{
throw;
}
}
}
}

private void ReplaceChannel(IChannelHandle channel)
{
Ensure.IsNotNull(channel, nameof(channel));
_channel?.Dispose();
_channel = channel;
}

public void ReplaceChannelSource(IChannelSourceHandle channelSource)
private void ReplaceChannelSource(IChannelSourceHandle channelSource)
{
Ensure.IsNotNull(channelSource, nameof(channelSource));
_channelSource?.Dispose();
_channel?.Dispose();
_channelSource = channelSource;
_channel = null;
}

private void Initialize(OperationContext operationContext)
{
_channelSource = _binding.GetReadChannelSource(operationContext);

try
{
_channel = _channelSource.GetChannel(operationContext);
}
catch (Exception ex) when (RetryableReadOperationExecutor.ShouldConnectionAcquireBeRetried(this, ex))
{
ReplaceChannelSource(_binding.GetReadChannelSource(operationContext));
ReplaceChannel(_channelSource.GetChannel(operationContext));
}
}

private async Task InitializeAsync(OperationContext operationContext)
{
_channelSource = await _binding.GetReadChannelSourceAsync(operationContext).ConfigureAwait(false);

try
{
_channel = await _channelSource.GetChannelAsync(operationContext).ConfigureAwait(false);
}
catch (Exception ex) when (RetryableReadOperationExecutor.ShouldConnectionAcquireBeRetried(this, ex))
{
ReplaceChannelSource(await _binding.GetReadChannelSourceAsync(operationContext).ConfigureAwait(false));
ReplaceChannel(await _channelSource.GetChannelAsync(operationContext).ConfigureAwait(false));
}
}
}
}
Loading