Skip to content

fix(transactions): improve recovery liveness and diagnostics #10374

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
ReubenBond merged 4 commits into from
Aug 10, 2026
Merged

fix(transactions): improve recovery liveness and diagnostics #10374

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dac63d5
fix(transactions): improve recovery liveness and diagnostics
ReubenBond Aug 8, 2026
76c7216
fix(transactions): preserve lock expiration transaction id
ReubenBond Aug 9, 2026
9784c63
fix(transactions): stabilize recovery tests
ReubenBond Aug 10, 2026
693e7a4
fix(transactions): update storage queue test setup
ReubenBond Aug 10, 2026
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
4 changes: 4 additions & 0 deletions src/Orleans.Transactions.TestKit.Base/Orleans.Transactions.TestKit.Base.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,4 +26,8 @@
<ProjectReference Include="$(SourceRoot)src\Orleans.Transactions\Orleans.Transactions.csproj" />
</ItemGroup>

<ItemGroup>
<InternalsVisibleTo Include="Orleans.Transactions.Tests" />
</ItemGroup>

</Project>
533 changes: 533 additions & 0 deletions src/Orleans.Transactions.TestKit.Base/TestRunners/TransactionRecoveryEventObserver.cs
View file
Open in desktop

Large diffs are not rendered by default.

204 changes: 204 additions & 0 deletions src/Orleans.Transactions.TestKit.Base/TestRunners/TransactionRecoveryFailureObservation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,204 @@
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

namespace Orleans.Transactions.TestKit;

internal static class TransactionRecoveryFailureObservation
{
internal enum OutcomeKind
{
FailureObserved,
StoppedWithoutFailure,
AttemptTimedOut,
}

internal sealed record Outcome<TFailure, TProducerResult>(
OutcomeKind Kind,
TFailure? Failure,
TProducerResult ProducerResult,
bool ProducerSettled,
TimeSpan Elapsed,
TimeSpan DrainElapsed)
where TFailure : class;

public static async Task ObserveAsync(Task task, Action<Exception, long> onFailure)
{
try
{
await task.ConfigureAwait(false);
}
catch (Exception exception)
{
onFailure(exception, Stopwatch.GetTimestamp());
}
}

public static bool IsPremature(long observedAt, long shutdownRequestedAt) => observedAt < shutdownRequestedAt;

internal static async Task<Outcome<TFailure, TProducerResult>> DetectAsync<TFailure, TProducerResult>(
Task<TProducerResult> producer,
Task<TFailure> firstFailure,
CancellationTokenSource stopProducing,
TimeSpan responseWindow,
TimeSpan schedulingMargin)
where TFailure : class
{
ArgumentOutOfRangeException.ThrowIfLessThan(responseWindow, TimeSpan.Zero);
ArgumentOutOfRangeException.ThrowIfLessThan(schedulingMargin, TimeSpan.Zero);

var startedAt = Stopwatch.GetTimestamp();
var responseDeadline = GetDeadline(startedAt, responseWindow);
var drainDeadline = GetDeadline(responseDeadline, schedulingMargin);

await WaitUntilAsync(firstFailure, producer, responseDeadline);
if (firstFailure.IsCompleted)
{
return await CompleteFailureAsync(
producer,
firstFailure,
stopProducing,
startedAt,
drainDeadline);
}

if (producer.IsCompleted)
{
var producerResult = await producer.ConfigureAwait(false);
if (firstFailure.IsCompleted)
{
return await CompleteFailureAsync(
producer,
firstFailure,
stopProducing,
startedAt,
drainDeadline);
}

return new(
OutcomeKind.StoppedWithoutFailure,
Failure: null,
producerResult,
ProducerSettled: true,
Stopwatch.GetElapsedTime(startedAt),
DrainElapsed: TimeSpan.Zero);
}

stopProducing.Cancel();
var drainStartedAt = Stopwatch.GetTimestamp();
await WaitUntilAsync(firstFailure, producer, drainDeadline);

if (firstFailure.IsCompleted)
{
return await CompleteFailureAsync(
producer,
firstFailure,
stopProducing,
startedAt,
drainDeadline,
drainStartedAt);
}

if (producer.IsCompleted)
{
var producerResult = await producer.ConfigureAwait(false);
if (firstFailure.IsCompleted)
{
return await CompleteFailureAsync(
producer,
firstFailure,
stopProducing,
startedAt,
drainDeadline,
drainStartedAt);
}

return new(
OutcomeKind.StoppedWithoutFailure,
Failure: null,
producerResult,
ProducerSettled: true,
Stopwatch.GetElapsedTime(startedAt),
Stopwatch.GetElapsedTime(drainStartedAt));
}

producer.Ignore();
return new(
OutcomeKind.AttemptTimedOut,
Failure: null,
ProducerResult: default!,
ProducerSettled: false,
Stopwatch.GetElapsedTime(startedAt),
Stopwatch.GetElapsedTime(drainStartedAt));
}

private static async Task<Outcome<TFailure, TProducerResult>> CompleteFailureAsync<TFailure, TProducerResult>(
Task<TProducerResult> producer,
Task<TFailure> firstFailure,
CancellationTokenSource stopProducing,
long startedAt,
long drainDeadline,
long? drainStartedAt = null)
where TFailure : class
{
stopProducing.Cancel();
var failure = await firstFailure.ConfigureAwait(false);
var drainStarted = drainStartedAt ?? Stopwatch.GetTimestamp();
if (!producer.IsCompleted)
{
await WaitUntilAsync(producer, drainDeadline);
}

if (producer.IsCompleted)
{
return new(
OutcomeKind.FailureObserved,
failure,
await producer.ConfigureAwait(false),
ProducerSettled: true,
Stopwatch.GetElapsedTime(startedAt),
Stopwatch.GetElapsedTime(drainStarted));
}

producer.Ignore();
return new(
OutcomeKind.FailureObserved,
failure,
ProducerResult: default!,
ProducerSettled: false,
Stopwatch.GetElapsedTime(startedAt),
Stopwatch.GetElapsedTime(drainStarted));
}

private static async Task WaitUntilAsync(Task first, long deadline)
{
while (!first.IsCompleted)
{
var now = Stopwatch.GetTimestamp();
if (now >= deadline)
{
return;
}

await Task.WhenAny(first, Task.Delay(Stopwatch.GetElapsedTime(now, deadline))).ConfigureAwait(false);
}
}

private static async Task WaitUntilAsync(Task first, Task second, long deadline)
{
while (!first.IsCompleted && !second.IsCompleted)
{
var now = Stopwatch.GetTimestamp();
if (now >= deadline)
{
return;
}

await Task.WhenAny(first, second, Task.Delay(Stopwatch.GetElapsedTime(now, deadline))).ConfigureAwait(false);
}
}

private static long GetDeadline(long startedAt, TimeSpan duration)
=> checked(startedAt + (long)(duration.TotalSeconds * Stopwatch.Frequency));
}
Loading
Loading