fix(reconcile): batch startup reconcile into fixed round trips, bound central work by QueryTimeout, retry failed site passes with back-off (review 2026-10-21 03/P-1)

This commit is contained in:
Joseph Doherty
2026-10-22 06:25:57 -04:00
parent d6a9a79c27
commit 7c83a4aa94
9 changed files with 638 additions and 130 deletions
@@ -40,13 +40,17 @@ public class CentralCommunicationActorReconcileTests : TestKit
{
new(2, "inst-B", "rev2", "dep-B", true),
});
deploymentRepo.GetDeployedSnapshotByInstanceIdAsync(2, Arg.Any<CancellationToken>())
.Returns(new DeployedConfigSnapshot("dep-B", "rev2", "{\"cfg\":\"B\"}"));
deploymentRepo.StagePendingIfAbsentAsync(
Arg.Any<int>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(),
Arg.Any<string>(), Arg.Any<DateTimeOffset>(), Arg.Any<DateTimeOffset>(),
Arg.Any<CancellationToken>())
.Returns(true);
deploymentRepo.GetDeployedSnapshotsByInstanceIdsAsync(
Arg.Any<IReadOnlyCollection<int>>(), Arg.Any<CancellationToken>())
.Returns(new Dictionary<int, DeployedConfigSnapshot>
{
[2] = new("dep-B", "rev2", "{\"cfg\":\"B\"}") { InstanceId = 2 },
});
// Every candidate is staged fresh: the batch hands each one back as its live row.
deploymentRepo.StagePendingBatchIfAbsentAsync(
Arg.Any<IReadOnlyList<PendingDeployment>>(), Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>())
.Returns(ci => (IReadOnlyDictionary<int, PendingDeployment>)ci.ArgAt<IReadOnlyList<PendingDeployment>>(0)
.ToDictionary(p => p.InstanceId));
var options = Options.Create(new CommunicationOptions
{
@@ -25,12 +25,43 @@ public class ReconcileServiceTests
Substitute.For<IDeploymentManagerRepository>();
private readonly ISiteRepository _siteRepo = Substitute.For<ISiteRepository>();
private ReconcileService CreateService(TimeSpan? ttl = null)
// Fake repository state behind the two batched reconcile calls.
private readonly Dictionary<int, DeployedConfigSnapshot> _snapshots = new();
private readonly Dictionary<int, PendingDeployment> _existingLive = new();
private readonly HashSet<int> _racedAway = new();
/// <summary>Wires the batched snapshot read and batched stage to the fake state above.</summary>
public ReconcileServiceTests()
{
_deploymentRepo.GetDeployedSnapshotsByInstanceIdsAsync(
Arg.Any<IReadOnlyCollection<int>>(), Arg.Any<CancellationToken>())
.Returns(ci => (IReadOnlyDictionary<int, DeployedConfigSnapshot>)ci.ArgAt<IReadOnlyCollection<int>>(0)
.Where(_snapshots.ContainsKey)
.ToDictionary(id => id, id => _snapshots[id]));
_deploymentRepo.StagePendingBatchIfAbsentAsync(
Arg.Any<IReadOnlyList<PendingDeployment>>(), Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>())
.Returns(ci =>
{
var live = new Dictionary<int, PendingDeployment>();
foreach (var candidate in ci.ArgAt<IReadOnlyList<PendingDeployment>>(0))
{
if (_existingLive.TryGetValue(candidate.InstanceId, out var existing))
live[candidate.InstanceId] = existing;
else if (!_racedAway.Contains(candidate.InstanceId))
live[candidate.InstanceId] = candidate;
}
return (IReadOnlyDictionary<int, PendingDeployment>)live;
});
}
private ReconcileService CreateService(TimeSpan? ttl = null, TimeSpan? queryTimeout = null)
{
var options = Options.Create(new CommunicationOptions
{
CentralFetchBaseUrl = BaseUrl,
PendingDeploymentTtl = ttl ?? TimeSpan.FromMinutes(5),
QueryTimeout = queryTimeout ?? TimeSpan.FromSeconds(30),
});
return new ReconcileService(
_deploymentRepo, _siteRepo, options, NullLogger<ReconcileService>.Instance);
@@ -48,15 +79,11 @@ public class ReconcileServiceTests
.Returns(expected.ToList());
private void SnapshotFor(ExpectedDeployment exp) =>
_deploymentRepo.GetDeployedSnapshotByInstanceIdAsync(exp.InstanceId, Arg.Any<CancellationToken>())
.Returns(new DeployedConfigSnapshot(exp.DeploymentId, exp.RevisionHash, $"{{\"cfg\":\"{exp.InstanceUniqueName}\"}}"));
private void StageReturns(bool result) =>
_deploymentRepo.StagePendingIfAbsentAsync(
Arg.Any<int>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(),
Arg.Any<string>(), Arg.Any<DateTimeOffset>(), Arg.Any<DateTimeOffset>(),
Arg.Any<CancellationToken>())
.Returns(result);
_snapshots[exp.InstanceId] = new DeployedConfigSnapshot(
exp.DeploymentId, exp.RevisionHash, $"{{\"cfg\":\"{exp.InstanceUniqueName}\"}}")
{
InstanceId = exp.InstanceId
};
private static ReconcileSiteRequest Request(params (string Name, string Rev)[] local) =>
new(SiteIdentifier, "node-a",
@@ -74,7 +101,6 @@ public class ReconcileServiceTests
ExpectedSet(a, b, c);
SnapshotFor(b);
SnapshotFor(c);
StageReturns(true);
// Node has A current, B stale (revOLD), C missing entirely.
var response = await CreateService().ReconcileAsync(
@@ -109,7 +135,6 @@ public class ReconcileServiceTests
SiteResolves();
var a = Expected(1, "inst-A", "rev1", "dep-A");
ExpectedSet(a);
StageReturns(true);
// inst-A is current; inst-Z is not deployed centrally → orphan.
var response = await CreateService().ReconcileAsync(
@@ -120,10 +145,10 @@ public class ReconcileServiceTests
Assert.Equal("inst-Z", orphan);
}
/// <summary>Verifies that when staging a pending deployment returns false and a row already exists, the gap carries the existing token.</summary>
/// <summary>Verifies that when a live pending row already exists for a gap instance, the gap carries the existing token.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Reconcile_StagePendingReturnsFalse_ExistingRowPresent_ReturnsExistingTokenInGap()
public async Task Reconcile_LiveRowAlreadyExists_ReturnsExistingTokenInGap()
{
// Live-found bug: two site nodes concurrently missing the SAME instance both reconcile at
// startup. The first stages a pending row; the second's StagePendingIfAbsent returns false.
@@ -138,19 +163,11 @@ public class ReconcileServiceTests
// B stages fresh (true). C already has a pending row (a concurrent reconcile from the
// other node, or an in-flight deploy, staged it first) → stage returns false.
StageReturns(true);
_deploymentRepo.StagePendingIfAbsentAsync(
3, Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(),
Arg.Any<string>(), Arg.Any<DateTimeOffset>(), Arg.Any<DateTimeOffset>(),
Arg.Any<CancellationToken>())
.Returns(false);
// The existing pending row for inst-C carries the FIRST node's deploymentId + token.
var nowUtc = DateTimeOffset.UtcNow;
_deploymentRepo.GetPendingDeploymentByInstanceIdAsync(3, Arg.Any<DateTimeOffset?>(), Arg.Any<CancellationToken>())
.Returns(new PendingDeployment(
"dep-C-existing", 3, "rev3-existing", "{\"cfg\":\"inst-C\"}",
"tok-C-existing", nowUtc, nowUtc.AddMinutes(5)));
_existingLive[3] = new PendingDeployment(
"dep-C-existing", 3, "rev3-existing", string.Empty,
"tok-C-existing", nowUtc, nowUtc.AddMinutes(5));
var response = await CreateService().ReconcileAsync(Request(/* empty local inventory */));
@@ -168,10 +185,10 @@ public class ReconcileServiceTests
Assert.True(gapC.IsEnabled);
}
/// <summary>Verifies that when staging a pending deployment returns false and no row exists, that gap item is omitted.</summary>
/// <summary>Verifies that when no pending row can be staged or found, that gap item is omitted.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Reconcile_StagePendingReturnsFalse_NoExistingRow_OmitsThatGapItem()
public async Task Reconcile_NoLiveRowStagedOrFound_OmitsThatGapItem()
{
// Fallback path: stage returns false, but the pending row raced away (was purged between
// the stage attempt and the read). Omit the item — the node retries on the next reconcile.
@@ -182,16 +199,8 @@ public class ReconcileServiceTests
SnapshotFor(b);
SnapshotFor(c);
StageReturns(true);
_deploymentRepo.StagePendingIfAbsentAsync(
3, Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(),
Arg.Any<string>(), Arg.Any<DateTimeOffset>(), Arg.Any<DateTimeOffset>(),
Arg.Any<CancellationToken>())
.Returns(false);
// The pending row was purged between the stage attempt and the read.
_deploymentRepo.GetPendingDeploymentByInstanceIdAsync(3, Arg.Any<DateTimeOffset?>(), Arg.Any<CancellationToken>())
.Returns((PendingDeployment?)null);
// No row could be staged or found for inst-C (raced away).
_racedAway.Add(3);
var response = await CreateService().ReconcileAsync(Request(/* empty local inventory */));
@@ -209,7 +218,6 @@ public class ReconcileServiceTests
var b = Expected(2, "inst-B", "rev2", "dep-B");
ExpectedSet(b);
// No snapshot configured for inst-B → repo returns null (deleted race).
StageReturns(true);
var response = await CreateService().ReconcileAsync(Request());
@@ -234,4 +242,79 @@ public class ReconcileServiceTests
await _deploymentRepo.DidNotReceive()
.GetExpectedDeploymentsForSiteAsync(Arg.Any<int>(), Arg.Any<CancellationToken>());
}
// ── review 2026-10-21 03/P-1: batched reconcile + bounded work ──
/// <summary>
/// The reconcile pass is a fixed number of round trips: snapshot JSON is read ONCE, only
/// for the gap, and the pending rows are staged in ONE batch — never per gap instance.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Reconcile_ReadsGapSnapshotsOnce_AndStagesInOneBatch()
{
SiteResolves();
var a = Expected(1, "inst-A", "rev1", "dep-A");
var b = Expected(2, "inst-B", "rev2", "dep-B");
var c = Expected(3, "inst-C", "rev3", "dep-C");
ExpectedSet(a, b, c);
SnapshotFor(a);
SnapshotFor(b);
SnapshotFor(c);
var response = await CreateService().ReconcileAsync(Request(("inst-A", "rev1")));
Assert.Equal(2, response.Gap.Count);
await _deploymentRepo.Received(1).GetDeployedSnapshotsByInstanceIdsAsync(
Arg.Is<IReadOnlyCollection<int>>(ids => ids.Order().SequenceEqual(new[] { 2, 3 })),
Arg.Any<CancellationToken>());
await _deploymentRepo.Received(1).StagePendingBatchIfAbsentAsync(
Arg.Is<IReadOnlyList<PendingDeployment>>(rows => rows.Count == 2),
Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>());
await _deploymentRepo.DidNotReceive().GetDeployedSnapshotByInstanceIdAsync(
Arg.Any<int>(), Arg.Any<CancellationToken>());
await _deploymentRepo.DidNotReceive().StagePendingIfAbsentAsync(
Arg.Any<int>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(),
Arg.Any<string>(), Arg.Any<DateTimeOffset>(), Arg.Any<DateTimeOffset>(),
Arg.Any<CancellationToken>());
}
/// <summary>A node with nothing missing costs no snapshot read and no staging at all.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Reconcile_NoGap_DoesNoSnapshotReadOrStaging()
{
SiteResolves();
ExpectedSet(Expected(1, "inst-A", "rev1", "dep-A"));
var response = await CreateService().ReconcileAsync(Request(("inst-A", "rev1")));
Assert.Empty(response.Gap);
await _deploymentRepo.DidNotReceive().GetDeployedSnapshotsByInstanceIdsAsync(
Arg.Any<IReadOnlyCollection<int>>(), Arg.Any<CancellationToken>());
await _deploymentRepo.DidNotReceive().StagePendingBatchIfAbsentAsync(
Arg.Any<IReadOnlyList<PendingDeployment>>(), Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>());
}
/// <summary>
/// Central stops working once the site's own deadline (QueryTimeout) has passed instead of
/// staging tokens for a caller that has already given up.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Reconcile_WorkIsBoundedByQueryTimeout()
{
SiteResolves();
_deploymentRepo.GetExpectedDeploymentsForSiteAsync(SiteId, Arg.Any<CancellationToken>())
.Returns(async ci =>
{
await Task.Delay(TimeSpan.FromSeconds(30), ci.ArgAt<CancellationToken>(1));
return (IReadOnlyList<ExpectedDeployment>)new List<ExpectedDeployment>();
});
var service = CreateService(queryTimeout: TimeSpan.FromMilliseconds(100));
await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => service.ReconcileAsync(Request()));
}
}
@@ -203,4 +203,100 @@ public class ReconcileRepositoryTests : IDisposable
// The reconcile row must not exist
Assert.Null(await _repository.GetPendingDeploymentByIdAsync("dep-reconcile"));
}
// --- review 2026-10-21 03/P-1: batched reconcile reads/stages ---
private static PendingDeployment Candidate(int instanceId, string deploymentId, DateTimeOffset now) =>
new(deploymentId, instanceId, "rev-" + deploymentId, $"{{\"cfg\":\"{deploymentId}\"}}",
"tok-" + deploymentId, now, now.AddMinutes(5));
/// <summary>The gap snapshot read returns only the requested instances, untracked.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task GetDeployedSnapshotsByInstanceIds_ReturnsRequestedOnly_Untracked()
{
var siteId = await SeedSiteAsync("SiteBatch");
var a = await SeedInstanceAsync("BatchA", siteId);
var b = await SeedInstanceAsync("BatchB", siteId);
var c = await SeedInstanceAsync("BatchC", siteId);
await SeedSnapshotAsync(a, "dep-a", "rev-a");
await SeedSnapshotAsync(b, "dep-b", "rev-b");
_context.ChangeTracker.Clear();
var snapshots = await _repository.GetDeployedSnapshotsByInstanceIdsAsync(new[] { a, c });
var only = Assert.Single(snapshots);
Assert.Equal(a, only.Key);
Assert.Equal("dep-a", only.Value.DeploymentId);
Assert.Empty(_context.ChangeTracker.Entries<DeployedConfigSnapshot>());
}
/// <summary>
/// The batch stage inserts every candidate without a live row in one call, hands back an
/// existing live row (without its JSON) instead of superseding it, and drops expired rows.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task StagePendingBatchIfAbsent_StagesAbsent_KeepsLive_ReplacesExpired()
{
var siteId = await SeedSiteAsync("SiteStage");
var fresh = await SeedInstanceAsync("StageFresh", siteId);
var inFlight = await SeedInstanceAsync("StageLive", siteId);
var expired = await SeedInstanceAsync("StageExpired", siteId);
var now = DateTimeOffset.UtcNow;
_context.Set<PendingDeployment>().AddRange(
new PendingDeployment("dep-live", inFlight, "rev-live", "{\"live\":true}", "tok-live",
now.AddMinutes(-1), now.AddMinutes(4)),
// Same DeploymentId the reconcile re-stages — the expired row must go first.
new PendingDeployment("dep-exp", expired, "rev-old", "{\"old\":true}", "tok-old",
now.AddMinutes(-20), now.AddMinutes(-10)));
await _context.SaveChangesAsync();
_context.ChangeTracker.Clear();
var freshCandidate = Candidate(fresh, "dep-fresh", now);
var liveCandidate = Candidate(inFlight, "dep-other", now);
var expiredCandidate = Candidate(expired, "dep-exp", now);
var result = await _repository.StagePendingBatchIfAbsentAsync(
new[] { freshCandidate, liveCandidate, expiredCandidate }, now);
Assert.Equal(3, result.Count);
Assert.Same(freshCandidate, result[fresh]);
Assert.Same(expiredCandidate, result[expired]);
var keptLive = result[inFlight];
Assert.NotSame(liveCandidate, keptLive);
Assert.Equal("dep-live", keptLive.DeploymentId);
Assert.Equal("tok-live", keptLive.Token);
Assert.Equal(string.Empty, keptLive.ConfigurationJson);
Assert.Equal("tok-dep-exp", (await _repository.GetPendingDeploymentByIdAsync("dep-exp"))!.Token);
Assert.NotNull(await _repository.GetPendingDeploymentByIdAsync("dep-fresh"));
Assert.Null(await _repository.GetPendingDeploymentByIdAsync("dep-other"));
}
/// <summary>
/// A batch whose single save fails (here: one candidate's instance was deleted, FK) falls
/// back to per-row staging, so the good rows are still staged and only the bad one is
/// omitted.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task StagePendingBatchIfAbsent_OneBadRow_OthersStillStaged()
{
var siteId = await SeedSiteAsync("SiteFk");
var good = await SeedInstanceAsync("FkGood", siteId);
var now = DateTimeOffset.UtcNow;
const int deletedInstanceId = 987_654;
var goodCandidate = Candidate(good, "dep-good", now);
var result = await _repository.StagePendingBatchIfAbsentAsync(
new[] { goodCandidate, Candidate(deletedInstanceId, "dep-gone", now) }, now);
var only = Assert.Single(result);
Assert.Equal(good, only.Key);
Assert.NotNull(await _repository.GetPendingDeploymentByIdAsync("dep-good"));
Assert.Null(await _repository.GetPendingDeploymentByIdAsync("dep-gone"));
}
}
@@ -53,7 +53,9 @@ public class SiteReconciliationActorTests : TestKit, IDisposable
IActorRef siteCommunicationActor,
IDeploymentConfigFetcher fetcher,
ILogger<SiteReconciliationActor>? logger = null,
TimeSpan? askTimeout = null) =>
TimeSpan? askTimeout = null,
int maxAttempts = 3,
TimeSpan? retryBaseDelay = null) =>
ActorOf(Props.Create(() => new SiteReconciliationActor(
_storage,
fetcher,
@@ -62,7 +64,10 @@ public class SiteReconciliationActorTests : TestKit, IDisposable
NodeId,
logger ?? Microsoft.Extensions.Logging.Abstractions.NullLogger<SiteReconciliationActor>.Instance,
TimeSpan.FromMilliseconds(50), // initial delay — fast for tests
askTimeout ?? TimeSpan.FromSeconds(5)))); // ask timeout
askTimeout ?? TimeSpan.FromSeconds(5), // ask timeout
null,
maxAttempts,
retryBaseDelay)));
/// <summary>Verifies that an instance missing at the site is fetched from central and stored.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
@@ -296,6 +301,68 @@ public class SiteReconciliationActorTests : TestKit, IDisposable
Assert.Empty(fetcher.Calls);
}
/// <summary>
/// review 2026-10-21 03/P-1: a pass whose Ask to central fails is retried with back-off up
/// to the attempt cap — not left until the next node restart — and then stops.
/// </summary>
[Fact]
public void CentralAskFails_IsRetried_UpToTheAttemptCap()
{
var fetcher = new FakeConfigFetcher(_ => Task.FromResult("never"));
var commProbe = CreateTestProbe();
CreateReconciliationActor(
commProbe, fetcher,
askTimeout: TimeSpan.FromMilliseconds(200),
maxAttempts: 3,
retryBaseDelay: TimeSpan.FromMilliseconds(100));
// Never replied to → three passes (1 + 2 retries), then no more.
commProbe.ExpectMsg<ReconcileSiteRequest>(TimeSpan.FromSeconds(5));
commProbe.ExpectMsg<ReconcileSiteRequest>(TimeSpan.FromSeconds(5));
commProbe.ExpectMsg<ReconcileSiteRequest>(TimeSpan.FromSeconds(5));
commProbe.ExpectNoMsg(TimeSpan.FromSeconds(1.5));
Assert.Empty(fetcher.Calls);
}
/// <summary>
/// review 2026-10-21 03/P-1: a pass that completes with a failed item is retried, and the
/// retry heals the item.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task FailedItem_IsRetried_AndHealsOnTheNextPass()
{
const string config = "{\"instanceUniqueName\":\"Flaky\"}";
var calls = 0;
var fetcher = new FakeConfigFetcher(_ => Interlocked.Increment(ref calls) == 1
? Task.FromException<string>(new DeploymentConfigFetchException("transient", isSuperseded: false))
: Task.FromResult(config));
var commProbe = CreateTestProbe();
CreateReconciliationActor(
commProbe, fetcher, maxAttempts: 3, retryBaseDelay: TimeSpan.FromMilliseconds(100));
var gap = new ReconcileSiteResponse(
[new ReconcileGapItem("Flaky", "depF", "revF", true, "tok-f")], [], "http://central:9000");
commProbe.ExpectMsg<ReconcileSiteRequest>(TimeSpan.FromSeconds(5));
commProbe.Reply(gap);
var retry = commProbe.ExpectMsg<ReconcileSiteRequest>(TimeSpan.FromSeconds(5));
Assert.False(retry.LocalNameToRevisionHash.ContainsKey("Flaky"));
commProbe.Reply(gap);
await AwaitAssertAsync(async () =>
{
var configs = await _storage.GetAllDeployedConfigsAsync();
Assert.Single(configs, c => c.InstanceUniqueName == "Flaky");
}, TimeSpan.FromSeconds(5));
// Healed on pass 2 → no third pass.
commProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(600));
}
/// <summary>
/// In-test fake <see cref="IDeploymentConfigFetcher"/>: runs a per-deploymentId behavior
/// (return config JSON or throw, as a Task) and records every call's coords thread-safely