349 lines
13 KiB
C#
349 lines
13 KiB
C#
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using Nexus.Api.Data;
|
|
using Nexus.Api.Models;
|
|
using Nexus.Api.Repositories;
|
|
using Nexus.Api.Services;
|
|
using Xunit;
|
|
|
|
namespace Nexus.Api.Tests;
|
|
|
|
public sealed class StaleTaskRecoveryTests
|
|
{
|
|
[Fact]
|
|
public async Task ResetStaleInProgressTasksAsync_OnlyResetsStaleInProgressTasks_AndWritesActivity()
|
|
{
|
|
await using var fixture = await TaskWorkflowFixture.CreateAsync();
|
|
var staleTimestamp = DateTimeOffset.UtcNow.AddHours(-3);
|
|
|
|
var staleInProgress = await fixture.TaskRepository.AddAsync(new WorkTask
|
|
{
|
|
Title = "Stale in progress",
|
|
State = "In progress",
|
|
Source = "iris",
|
|
UpdatedAt = staleTimestamp,
|
|
CreatedAt = staleTimestamp
|
|
}, CancellationToken.None);
|
|
|
|
await fixture.ActivityRepository.AddAsync(new ActivityEvent
|
|
{
|
|
Type = "comment",
|
|
Message = "Previous agent note",
|
|
TaskId = staleInProgress.Id,
|
|
CreatedAt = staleTimestamp.AddMinutes(15)
|
|
}, CancellationToken.None);
|
|
|
|
var staleBlocked = await fixture.TaskRepository.AddAsync(new WorkTask
|
|
{
|
|
Title = "Blocked task",
|
|
State = "Blocked",
|
|
Source = "iris",
|
|
UpdatedAt = staleTimestamp,
|
|
CreatedAt = staleTimestamp
|
|
}, CancellationToken.None);
|
|
|
|
var staleReview = await fixture.TaskRepository.AddAsync(new WorkTask
|
|
{
|
|
Title = "Review task",
|
|
State = "Review",
|
|
Source = "iris",
|
|
UpdatedAt = staleTimestamp,
|
|
CreatedAt = staleTimestamp
|
|
}, CancellationToken.None);
|
|
|
|
var staleDone = await fixture.TaskRepository.AddAsync(new WorkTask
|
|
{
|
|
Title = "Done task",
|
|
State = "Done",
|
|
Source = "iris",
|
|
UpdatedAt = staleTimestamp,
|
|
CreatedAt = staleTimestamp
|
|
}, CancellationToken.None);
|
|
|
|
var staleBacklog = await fixture.TaskRepository.AddAsync(new WorkTask
|
|
{
|
|
Title = "Backlog task",
|
|
State = "Backlog",
|
|
Source = "iris",
|
|
UpdatedAt = staleTimestamp,
|
|
CreatedAt = staleTimestamp
|
|
}, CancellationToken.None);
|
|
|
|
var freshInProgress = await fixture.TaskRepository.AddAsync(new WorkTask
|
|
{
|
|
Title = "Fresh in progress",
|
|
State = "In progress",
|
|
Source = "iris",
|
|
UpdatedAt = DateTimeOffset.UtcNow.AddMinutes(-30),
|
|
CreatedAt = staleTimestamp
|
|
}, CancellationToken.None);
|
|
|
|
var resetCount = await fixture.StaleTaskRecoveryService.ResetStaleInProgressTasksAsync(TimeSpan.FromHours(2), CancellationToken.None);
|
|
|
|
Assert.Equal(1, resetCount);
|
|
Assert.Equal("Backlog", (await fixture.TaskService.GetByIdAsync(staleInProgress.Id, CancellationToken.None))!.State);
|
|
Assert.Equal("Blocked", (await fixture.TaskService.GetByIdAsync(staleBlocked.Id, CancellationToken.None))!.State);
|
|
Assert.Equal("Review", (await fixture.TaskService.GetByIdAsync(staleReview.Id, CancellationToken.None))!.State);
|
|
Assert.Equal("Done", (await fixture.TaskService.GetByIdAsync(staleDone.Id, CancellationToken.None))!.State);
|
|
Assert.Equal("Backlog", (await fixture.TaskService.GetByIdAsync(staleBacklog.Id, CancellationToken.None))!.State);
|
|
Assert.Equal("In progress", (await fixture.TaskService.GetByIdAsync(freshInProgress.Id, CancellationToken.None))!.State);
|
|
|
|
var activity = await fixture.TaskService.GetTaskActivityAsync(staleInProgress.Id, CancellationToken.None);
|
|
var resetActivity = activity.FirstOrDefault(entry => entry.Message.Contains("stale recovery", StringComparison.Ordinal));
|
|
|
|
Assert.NotNull(resetActivity);
|
|
Assert.Contains("reason=stale-recovery", resetActivity!.Message, StringComparison.Ordinal);
|
|
Assert.Contains("previous status In progress", resetActivity.Message, StringComparison.Ordinal);
|
|
Assert.Contains("stale reference", resetActivity.Message, StringComparison.Ordinal);
|
|
Assert.Contains("last activity", resetActivity.Message, StringComparison.Ordinal);
|
|
Assert.Contains("new status Backlog", resetActivity.Message, StringComparison.Ordinal);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ResetStaleInProgressTasksAsync_RevalidatesCurrentTaskBeforeReset()
|
|
{
|
|
var staleTimestamp = DateTimeOffset.UtcNow.AddHours(-3);
|
|
var taskId = Guid.NewGuid();
|
|
var staleCandidate = new WorkTask
|
|
{
|
|
Id = taskId,
|
|
Title = "Changed during recovery scan",
|
|
State = "In progress",
|
|
Source = "iris",
|
|
UpdatedAt = staleTimestamp,
|
|
CreatedAt = staleTimestamp
|
|
};
|
|
var currentTask = new WorkTask
|
|
{
|
|
Id = taskId,
|
|
Title = "Changed during recovery scan",
|
|
State = "Review",
|
|
Source = "iris",
|
|
UpdatedAt = staleTimestamp,
|
|
CreatedAt = staleTimestamp
|
|
};
|
|
var taskRepository = new FakeTaskRepository(staleCandidate, currentTask);
|
|
var activityRepository = new FakeActivityRepository();
|
|
var liveUpdateService = new FakeLiveUpdateService();
|
|
var recoveryService = new StaleTaskRecoveryService(
|
|
taskRepository,
|
|
activityRepository,
|
|
liveUpdateService);
|
|
|
|
var resetCount = await recoveryService.ResetStaleInProgressTasksAsync(TimeSpan.FromHours(2), CancellationToken.None);
|
|
|
|
Assert.Equal(0, resetCount);
|
|
Assert.Equal("Review", currentTask.State);
|
|
Assert.Equal(0, taskRepository.ResetCount);
|
|
Assert.Empty(activityRepository.Added);
|
|
Assert.Equal(0, liveUpdateService.PublishCount);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task BackgroundService_RunRecoveryOnceAsync_UsesConfiguredThreshold_AndCallsRecoveryService()
|
|
{
|
|
var fakeRecoveryService = new FakeStaleTaskRecoveryService();
|
|
var services = new ServiceCollection();
|
|
services.AddScoped<IStaleTaskRecoveryService>(_ => fakeRecoveryService);
|
|
|
|
await using var provider = services.BuildServiceProvider();
|
|
var backgroundService = new StaleTaskRecoveryBackgroundService(
|
|
provider.GetRequiredService<IServiceScopeFactory>(),
|
|
new TestOptionsMonitor<StaleTaskRecoveryOptions>(new StaleTaskRecoveryOptions
|
|
{
|
|
StaleHours = 4,
|
|
IntervalMinutes = 30
|
|
}),
|
|
NullLogger<StaleTaskRecoveryBackgroundService>.Instance);
|
|
|
|
var resetCount = await backgroundService.RunRecoveryOnceAsync(CancellationToken.None);
|
|
|
|
Assert.Equal(1, fakeRecoveryService.CallCount);
|
|
Assert.Equal(TimeSpan.FromHours(4), fakeRecoveryService.LastThreshold);
|
|
Assert.Equal(7, resetCount);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task BackgroundService_StartAsync_RunsRecoveryWithoutWaitingForFullInterval()
|
|
{
|
|
var fakeRecoveryService = new FakeStaleTaskRecoveryService();
|
|
var firstCall = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
fakeRecoveryService.OnCall = () => firstCall.TrySetResult(true);
|
|
|
|
var services = new ServiceCollection();
|
|
services.AddScoped<IStaleTaskRecoveryService>(_ => fakeRecoveryService);
|
|
|
|
await using var provider = services.BuildServiceProvider();
|
|
var backgroundService = new StaleTaskRecoveryBackgroundService(
|
|
provider.GetRequiredService<IServiceScopeFactory>(),
|
|
new TestOptionsMonitor<StaleTaskRecoveryOptions>(new StaleTaskRecoveryOptions
|
|
{
|
|
StaleHours = 2,
|
|
IntervalMinutes = 30
|
|
}),
|
|
NullLogger<StaleTaskRecoveryBackgroundService>.Instance);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
await backgroundService.StartAsync(cts.Token);
|
|
await firstCall.Task.WaitAsync(cts.Token);
|
|
await backgroundService.StopAsync(CancellationToken.None);
|
|
|
|
Assert.True(fakeRecoveryService.CallCount >= 1);
|
|
Assert.Equal(TimeSpan.FromHours(2), fakeRecoveryService.LastThreshold);
|
|
}
|
|
|
|
[Fact]
|
|
public void TaskRecoveryOptions_BindsStaleHoursFromEnvironmentOverride()
|
|
{
|
|
const string key = "TaskRecovery__StaleHours";
|
|
var originalValue = Environment.GetEnvironmentVariable(key);
|
|
|
|
try
|
|
{
|
|
Environment.SetEnvironmentVariable(key, "5");
|
|
|
|
var configuration = new ConfigurationBuilder()
|
|
.AddInMemoryCollection(new Dictionary<string, string?>
|
|
{
|
|
[$"{StaleTaskRecoveryOptions.SectionName}:StaleHours"] = "2",
|
|
[$"{StaleTaskRecoveryOptions.SectionName}:IntervalMinutes"] = "30"
|
|
})
|
|
.AddEnvironmentVariables()
|
|
.Build();
|
|
|
|
var options = configuration.GetSection(StaleTaskRecoveryOptions.SectionName).Get<StaleTaskRecoveryOptions>();
|
|
|
|
Assert.NotNull(options);
|
|
Assert.Equal(5, options!.StaleHours);
|
|
Assert.Equal(30, options.IntervalMinutes);
|
|
}
|
|
finally
|
|
{
|
|
Environment.SetEnvironmentVariable(key, originalValue);
|
|
}
|
|
}
|
|
}
|
|
|
|
file sealed class FakeTaskRepository(WorkTask staleCandidate, WorkTask currentTask) : ITaskRepository
|
|
{
|
|
public int ResetCount { get; private set; }
|
|
|
|
public Task<List<WorkTask>> GetAllAsync(CancellationToken ct = default)
|
|
=> Task.FromResult(new List<WorkTask> { staleCandidate });
|
|
|
|
public ValueTask<WorkTask?> GetByIdAsync(Guid id, CancellationToken ct = default)
|
|
=> ValueTask.FromResult<WorkTask?>(id == currentTask.Id ? currentTask : null);
|
|
|
|
public Task<bool> TryResetStaleInProgressToBacklogAsync(
|
|
Guid id,
|
|
DateTimeOffset staleBefore,
|
|
DateTimeOffset updatedAt,
|
|
CancellationToken ct = default)
|
|
{
|
|
if (id != currentTask.Id
|
|
|| !string.Equals(currentTask.State, "In progress", StringComparison.OrdinalIgnoreCase)
|
|
|| currentTask.UpdatedAt >= staleBefore)
|
|
{
|
|
return Task.FromResult(false);
|
|
}
|
|
|
|
ResetCount++;
|
|
currentTask.State = "Backlog";
|
|
currentTask.UpdatedAt = updatedAt;
|
|
return Task.FromResult(true);
|
|
}
|
|
|
|
public Task UpdateAsync(WorkTask task, CancellationToken ct = default)
|
|
{
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public Task<List<WorkTask>> GetPendingApprovalAsync(CancellationToken ct = default)
|
|
=> Task.FromResult(new List<WorkTask>());
|
|
|
|
public Task<WorkTask> AddAsync(WorkTask task, CancellationToken ct = default)
|
|
=> Task.FromResult(task);
|
|
|
|
public Task DeleteAsync(WorkTask task, CancellationToken ct = default)
|
|
=> Task.CompletedTask;
|
|
|
|
public Task<int> CountAsync(CancellationToken ct = default)
|
|
=> Task.FromResult(0);
|
|
|
|
public Task<int> CountByStateAsync(string state, CancellationToken ct = default)
|
|
=> Task.FromResult(0);
|
|
|
|
public Task<WorkTask?> GetLastBlockedAsync(CancellationToken ct = default)
|
|
=> Task.FromResult<WorkTask?>(null);
|
|
}
|
|
|
|
file sealed class FakeActivityRepository : IActivityRepository
|
|
{
|
|
public List<ActivityEvent> Added { get; } = [];
|
|
|
|
public Task<List<ActivityEvent>> GetRecentAsync(int take, CancellationToken ct = default)
|
|
=> Task.FromResult(new List<ActivityEvent>());
|
|
|
|
public Task<List<ActivityEvent>> GetRecentForTasksAsync(IEnumerable<Guid> taskIds, CancellationToken ct = default)
|
|
=> Task.FromResult(new List<ActivityEvent>());
|
|
|
|
public Task<(List<ActivityEvent> Items, int TotalCount)> GetPagedAsync(
|
|
string? type,
|
|
string? sort,
|
|
int page,
|
|
int pageSize,
|
|
CancellationToken ct = default)
|
|
=> Task.FromResult((new List<ActivityEvent>(), 0));
|
|
|
|
public Task<List<ActivityEvent>> GetByAgentAsync(string agentId, int take, CancellationToken ct = default)
|
|
=> Task.FromResult(new List<ActivityEvent>());
|
|
|
|
public Task<ActivityEvent> AddAsync(ActivityEvent activity, CancellationToken ct = default)
|
|
{
|
|
Added.Add(activity);
|
|
return Task.FromResult(activity);
|
|
}
|
|
}
|
|
|
|
file sealed class FakeLiveUpdateService : ILiveUpdateService
|
|
{
|
|
public int PublishCount { get; private set; }
|
|
public long CurrentSequence => PublishCount;
|
|
|
|
public Task<LiveUpdateSubscription> SubscribeAsync(long? afterSequence = null, CancellationToken ct = default)
|
|
=> throw new NotSupportedException();
|
|
|
|
public LiveUpdateEnvelope Publish(string type, object payload, string channel = "dashboard")
|
|
{
|
|
PublishCount++;
|
|
return new LiveUpdateEnvelope(type, DateTimeOffset.UtcNow, payload, PublishCount, channel);
|
|
}
|
|
}
|
|
|
|
file sealed class FakeStaleTaskRecoveryService : IStaleTaskRecoveryService
|
|
{
|
|
public int CallCount { get; private set; }
|
|
public TimeSpan LastThreshold { get; private set; }
|
|
public Action? OnCall { get; set; }
|
|
|
|
public Task<int> ResetStaleInProgressTasksAsync(TimeSpan staleThreshold, CancellationToken ct = default)
|
|
{
|
|
CallCount++;
|
|
LastThreshold = staleThreshold;
|
|
OnCall?.Invoke();
|
|
return Task.FromResult(7);
|
|
}
|
|
}
|
|
|
|
file sealed class TestOptionsMonitor<T>(T currentValue) : IOptionsMonitor<T>
|
|
{
|
|
public T CurrentValue { get; private set; } = currentValue;
|
|
|
|
public T Get(string? name) => CurrentValue;
|
|
|
|
public IDisposable? OnChange(Action<T, string?> listener) => null;
|
|
}
|