Skip to content

Fix run service job renewal retry #4671

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
ishu599 wants to merge 1 commit into
base: main
Choose a base branch
from
+103 −4
Open

Fix run service job renewal retry #4671

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
11 changes: 7 additions & 4 deletions src/Runner.Listener/JobDispatcher.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -747,7 +747,7 @@ internal async Task RenewJobRequestAsync(Pipelines.AgentJobRequestMessage messag

private async Task RenewJobRequestAsync(IRunServer runServer, Guid planId, Guid jobId, TaskCompletionSource<int> firstJobRequestRenewed, CancellationToken token)
{
TaskAgentJobRequest request = null;
DateTime? lastSuccessfulLockedUntil = null;
int firstRenewRetryLimit = 5;
int encounteringError = 0;

Expand All @@ -759,6 +759,7 @@ private async Task RenewJobRequestAsync(IRunServer runServer, Guid planId, Guid
{
var renewResponse = await runServer.RenewJobAsync(planId, jobId, token);
Trace.Info($"Successfully renew job {jobId}, job is valid till {renewResponse.LockedUntil}");
lastSuccessfulLockedUntil = renewResponse.LockedUntil;

if (!firstJobRequestRenewed.Task.IsCompleted)
{
Expand Down Expand Up @@ -807,7 +808,7 @@ private async Task RenewJobRequestAsync(IRunServer runServer, Guid planId, Guid
else
{
// retry till reach lockeduntil + 5 mins extra buffer.
remainingTime = request.LockedUntil.Value + TimeSpan.FromMinutes(5) - DateTime.UtcNow;
remainingTime = lastSuccessfulLockedUntil.Value + TimeSpan.FromMinutes(5) - DateTime.UtcNow;
}

if (remainingTime > TimeSpan.Zero)
Expand All @@ -820,7 +821,7 @@ private async Task RenewJobRequestAsync(IRunServer runServer, Guid planId, Guid
}
else
{
Trace.Info($"Retrying lock renewal for job {jobId}. Job is valid until {request.LockedUntil.Value}.");
Trace.Info($"Retrying lock renewal for job {jobId}. Job is valid until {lastSuccessfulLockedUntil.Value}.");
if (encounteringError > 5)
{
delayTime = BackoffTimerHelper.GetRandomBackoff(TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30));
Expand Down Expand Up @@ -854,6 +855,7 @@ private async Task RenewJobRequestAsync(IRunServer runServer, Guid planId, Guid
private async Task RenewJobRequestAsync(IRunnerServer runnerServer, int poolId, long requestId, Guid lockToken, string orchestrationId, TaskCompletionSource<int> firstJobRequestRenewed, CancellationToken token)
{
TaskAgentJobRequest request = null;
DateTime? lastSuccessfulLockedUntil = null;
int firstRenewRetryLimit = 5;
int encounteringError = 0;

Expand All @@ -865,6 +867,7 @@ private async Task RenewJobRequestAsync(IRunnerServer runnerServer, int poolId,
{
request = await runnerServer.RenewAgentRequestAsync(poolId, requestId, lockToken, orchestrationId, token);
Trace.Info($"Successfully renew job request {requestId}, job is valid till {request.LockedUntil.Value}");
lastSuccessfulLockedUntil = request.LockedUntil;

if (!firstJobRequestRenewed.Task.IsCompleted)
{
Expand Down Expand Up @@ -936,7 +939,7 @@ private async Task RenewJobRequestAsync(IRunnerServer runnerServer, int poolId,
}
else
{
Trace.Info($"Retrying lock renewal for jobrequest {requestId}. Job is valid until {request.LockedUntil.Value}.");
Trace.Info($"Retrying lock renewal for jobrequest {requestId}. Job is valid until {lastSuccessfulLockedUntil.Value}.");
if (encounteringError > 5)
{
delayTime = BackoffTimerHelper.GetRandomBackoff(TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30));
Expand Down
96 changes: 96 additions & 0 deletions src/Test/L0/Listener/JobDispatcherL0.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.Reflection;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using GitHub.DistributedTask.ObjectTemplating.Tokens;
Expand Down Expand Up @@ -344,6 +345,101 @@ public async void DispatcherRenewJobRequestStopOnJobTokenExpiredExceptions()
}
}

[Fact]
[Trait("Level", "L0")]
[Trait("Category", "Runner")]
public async Task DispatcherRenewJobOnRunServiceRetryAfterGenericException()
{
// Arrange
using (var hc = new TestHostContext(this))
{
int poolId = 1;
Int64 requestId = 1000;
int count = 0;

TaskCompletionSource<int> firstJobRequestRenewed = new();
CancellationTokenSource cancellationTokenSource = new();

hc.SetSingleton<IRunServer>(_runServer.Object);
hc.SetSingleton<IConfigurationStore>(_configurationStore.Object);
_configurationStore.Setup(x => x.GetSettings())
.Returns(new RunnerSettings() { PoolId = 1 });

_ = _runServer.Setup(x => x.RenewJobAsync(
It.IsAny<Guid>(),
It.IsAny<Guid>(),
It.IsAny<CancellationToken>()))
.Returns(() =>
{
count++;

if (count == 1)
{
// First renewal succeeds.
return Task.FromResult(new RenewJobResponse
{
LockedUntil = DateTime.UtcNow.AddMinutes(5)
});
}
else if (count == 2)
{
// Second renewal fails with a generic exception.
throw new HttpRequestException("Simulated renewal failure.");
}
else if (count == 3)
{
// Retry succeeds.
cancellationTokenSource.Cancel();
return Task.FromResult(new RenewJobResponse
{
LockedUntil = DateTime.UtcNow.AddMinutes(5)
});
}

throw new InvalidOperationException("Should not reach here.");
});

var jobDispatcher = new JobDispatcher();
jobDispatcher.Initialize(hc);
EnableRunServiceJobForJobDispatcher(jobDispatcher);

// Set the value of the _isRunServiceJob field to true.
var isRunServiceJobField = typeof(JobDispatcher).GetField(
"_isRunServiceJob",
BindingFlags.NonPublic | BindingFlags.Instance);

isRunServiceJobField.SetValue(jobDispatcher, true);

// Act
await jobDispatcher.RenewJobRequestAsync(
GetAgentJobRequestMessage(),
GetServiceEndpoint(),
poolId,
requestId,
Guid.Empty,
Guid.NewGuid().ToString(),
firstJobRequestRenewed,
cancellationTokenSource.Token);

// Assert
Assert.True(
firstJobRequestRenewed.Task.IsCompletedSuccessfully,
"First renew should succeed.");

Assert.Equal(3, count);
Assert.True(
cancellationTokenSource.IsCancellationRequested,
"Cancellation should only happen after the retry succeeds.");

_runServer.Verify(
x => x.RenewJobAsync(
It.IsAny<Guid>(),
It.IsAny<Guid>(),
It.IsAny<CancellationToken>()),
Times.Exactly(3));
}
}

[Fact]
[Trait("Level", "L0")]
[Trait("Category", "Runner")]
Expand Down