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

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 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
File filter

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;
}
}
19 changes: 8 additions & 11 deletions src/MongoDB.Driver/Core/Operations/RetryabilityHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
/* Copyright 2018-present MongoDB Inc.
/* Copyright 2010-present MongoDB Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down 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
100 changes: 51 additions & 49 deletions src/MongoDB.Driver/Core/Operations/RetryableReadContext.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
/* Copyright 2019-present MongoDB Inc.
/* Copyright 2010-present MongoDB Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand All @@ -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.AcquireOrReplaceChannel(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.AcquireOrReplaceChannelAsync(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,58 @@ public void Dispose()
}
}

public void ReplaceChannel(IChannelHandle channel)
internal void AcquireOrReplaceChannel(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers)
{
var attempt = 1;
while (true)
{
operationContext.ThrowIfTimedOutOrCanceled();
ReplaceChannelSource(Binding.GetReadChannelSource(operationContext, deprioritizedServers));
try
{
ReplaceChannel(ChannelSource.GetChannel(operationContext));
return;
}
catch (Exception ex) when (RetryableReadOperationExecutor.ShouldConnectionAcquireBeRetried(operationContext, this, ex, attempt))
{
attempt++;
}
}
}

internal async Task AcquireOrReplaceChannelAsync(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers)
{
var attempt = 1;
while (true)
{
operationContext.ThrowIfTimedOutOrCanceled();
ReplaceChannelSource(await Binding.GetReadChannelSourceAsync(operationContext, deprioritizedServers).ConfigureAwait(false));
try
{
ReplaceChannel(await ChannelSource.GetChannelAsync(operationContext).ConfigureAwait(false));
return;
}
catch (Exception ex) when (RetryableReadOperationExecutor.ShouldConnectionAcquireBeRetried(operationContext, this, ex, attempt))
{
attempt++;
}
}
}

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