Retry analytics on concurrent-modification conflicts instead of erroring
The reconciliation worker resets position checkpoints via bulk UPDATE while
the analytics worker saves recalculated positions for the same trader —
MySQL then rejects the stale save ("Record has changed since last read").
The engine run is now retried once with a fresh context (re-reading the
concurrent change); if it still collides, it logs a warning and lets the
next cycle pick the trader up again (LastAnalyzedAt stays unset).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
850c47a404
commit
940a99fec2
@@ -86,9 +86,33 @@ public class TraderAnalyticsWorker : BackgroundService
|
|||||||
_logger.LogInformation("Found {Count} active or unanalyzed traders to update", traderIds.Count);
|
_logger.LogInformation("Found {Count} active or unanalyzed traders to update", traderIds.Count);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async Task MarkJobFailedAsync(int traderId, Exception ex)
|
||||||
|
{
|
||||||
|
if (activeJob != null && activeJob.TraderId == traderId)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
using var jobScope = _services.CreateScope();
|
||||||
|
var updateJobRepo = jobScope.ServiceProvider.GetRequiredService<IJobRepository>();
|
||||||
|
activeJob.Status = Predictalytics.Domain.Enums.JobStatus.Failed;
|
||||||
|
activeJob.CompletedAt = DateTime.UtcNow;
|
||||||
|
activeJob.ErrorMessage = ex.Message;
|
||||||
|
await updateJobRepo.UpdateAsync(activeJob, CancellationToken.None);
|
||||||
|
}
|
||||||
|
catch { /* Ignore secondary errors */ }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
foreach (var id in traderIds)
|
foreach (var id in traderIds)
|
||||||
{
|
{
|
||||||
if (ct.IsCancellationRequested) break;
|
if (ct.IsCancellationRequested) break;
|
||||||
|
|
||||||
|
// Concurrent writers (reconciliation checkpoint resets, pollers) can touch the
|
||||||
|
// same TraderPositions/Traders rows mid-save ("Record has changed since last
|
||||||
|
// read"). A retry with a fresh context re-reads their changes and succeeds.
|
||||||
|
const int maxAttempts = 2;
|
||||||
|
for (int attempt = 1; attempt <= maxAttempts; attempt++)
|
||||||
|
{
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
using var traderScope = _services.CreateScope();
|
using var traderScope = _services.CreateScope();
|
||||||
@@ -149,27 +173,38 @@ public class TraderAnalyticsWorker : BackgroundService
|
|||||||
activeJob.CompletedAt = DateTime.UtcNow;
|
activeJob.CompletedAt = DateTime.UtcNow;
|
||||||
await updateJobRepo.UpdateAsync(activeJob, ct);
|
await updateJobRepo.UpdateAsync(activeJob, ct);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
break; // trader done
|
||||||
}
|
}
|
||||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||||
{
|
{
|
||||||
_logger.LogInformation("Cancellation requested during analysis, stopping batch.");
|
_logger.LogInformation("Cancellation requested during analysis, stopping batch.");
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
catch (DbUpdateException ex) when (attempt < maxAttempts)
|
||||||
|
{
|
||||||
|
_logger.LogWarning("Concurrent modification while analyzing trader {TraderId} — retrying with a fresh context. ({Message})",
|
||||||
|
id, ex.InnerException?.Message ?? ex.Message);
|
||||||
|
await Task.Delay(250, CancellationToken.None);
|
||||||
|
}
|
||||||
|
catch (DbUpdateException ex)
|
||||||
|
{
|
||||||
|
// Still colliding after the retry. LastAnalyzedAt stays unset, so this
|
||||||
|
// trader is picked up again next cycle — a warning is enough.
|
||||||
|
_logger.LogWarning("Trader {TraderId} skipped after repeated concurrent modifications — will retry next cycle. ({Message})",
|
||||||
|
id, ex.InnerException?.Message ?? ex.Message);
|
||||||
|
await MarkJobFailedAsync(id, ex);
|
||||||
|
break;
|
||||||
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
_logger.LogError(ex, "Error recalculating positions/PnL for trader {TraderId}", id);
|
_logger.LogError(ex, "Error recalculating positions/PnL for trader {TraderId}", id);
|
||||||
if (activeJob != null && activeJob.TraderId == id)
|
await MarkJobFailedAsync(id, ex);
|
||||||
{
|
break;
|
||||||
try {
|
|
||||||
using var jobScope = _services.CreateScope();
|
|
||||||
var updateJobRepo = jobScope.ServiceProvider.GetRequiredService<IJobRepository>();
|
|
||||||
activeJob.Status = Predictalytics.Domain.Enums.JobStatus.Failed;
|
|
||||||
activeJob.CompletedAt = DateTime.UtcNow;
|
|
||||||
activeJob.ErrorMessage = ex.Message;
|
|
||||||
await updateJobRepo.UpdateAsync(activeJob, CancellationToken.None);
|
|
||||||
} catch { /* Ignore secondary errors */ }
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (ct.IsCancellationRequested) break;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (traderIds.Count > 0)
|
if (traderIds.Count > 0)
|
||||||
|
|||||||
Reference in New Issue
Block a user