diff --git a/docs/requirements/Component-DeploymentManager.md b/docs/requirements/Component-DeploymentManager.md
index 8f559bc5..2f66ee3e 100644
--- a/docs/requirements/Component-DeploymentManager.md
+++ b/docs/requirements/Component-DeploymentManager.md
@@ -207,6 +207,22 @@ a cancelled bulk deploy can be left holding its lock.
one, so the two entry points cannot drift on deployment identity, idempotency, lock
coverage, or optimistic concurrency.
+### Startup-reconcile staging
+
+A site node that missed a deploy reports its inventory on startup and central
+(`Communication/ReconcileService`) stages fresh `PendingDeployment` fetch tokens for
+the gap. The pass costs a **fixed number of round trips whatever the gap size**
+(review 2026-10-21 03/P-1): the expected-set projection (no config JSON), ONE
+untracked read of the gap instances' snapshot configs
+(`GetDeployedSnapshotsByInstanceIdsAsync`), and ONE batched insert-if-absent
+(`StagePendingBatchIfAbsentAsync` — expired rows deleted set-based, live rows from an
+in-flight deploy or the peer node's concurrent reconcile handed back instead of
+superseded, one `SaveChanges`; a failed batch save falls back to per-row staging so
+one bad row costs only itself). Central's work is bounded by `QueryTimeout`, the
+site's own deadline for the reply. On the site, `SiteReconciliationActor` retries a
+failed or partially-failed pass up to **3** times with exponential back-off (30 s,
+60 s) instead of waiting for the next node restart.
+
### Terminal deployment-record retention
Deployment records are insert-only — one row per deploy attempt — so without a
diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs
index 51f67081..c8d71116 100644
--- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs
@@ -341,6 +341,37 @@ public interface IDeploymentManagerRepository
string configurationJson, string token, DateTimeOffset createdAtUtc, DateTimeOffset expiresAtUtc,
CancellationToken cancellationToken = default);
+ ///
+ /// Reads the deployed config snapshots of in ONE untracked
+ /// query — the startup-reconcile gap read. Only the gap's ids should be passed: this is the
+ /// read that materialises each snapshot's ConfigurationJson.
+ ///
+ /// Instances whose snapshots to read; duplicates are harmless.
+ /// A cancellation token that can be used to cancel the operation.
+ /// Instance id → snapshot, omitting instances with no snapshot (deleted race).
+ Task> GetDeployedSnapshotsByInstanceIdsAsync(
+ IReadOnlyCollection instanceIds, CancellationToken cancellationToken = default);
+
+ ///
+ /// Batch form of for startup reconcile: expired
+ /// pending rows for the candidates' instances are deleted, then every candidate whose
+ /// instance has no LIVE pending row is inserted, all in one SaveChanges. A live row
+ /// (an in-flight deploy, or the peer node's concurrent reconcile) is never superseded.
+ /// Self-contained: commits its own writes.
+ ///
+ /// One new pending row per instance (later duplicates for the same instance are ignored).
+ /// "Now": rows with ExpiresAtUtc <= nowUtc are expired.
+ /// A cancellation token that can be used to cancel the operation.
+ ///
+ /// Instance id → the instance's live pending row after the call: the candidate itself
+ /// (reference-equal) when it was staged, otherwise the pre-existing live row, returned
+ /// WITHOUT its ConfigurationJson (empty string). An instance whose candidate could not
+ /// be staged and has no live row (e.g. the instance was deleted mid-reconcile) is omitted.
+ ///
+ Task> StagePendingBatchIfAbsentAsync(
+ IReadOnlyList candidates, DateTimeOffset nowUtc,
+ CancellationToken cancellationToken = default);
+
// Instance lookups for deployment pipeline
///
/// Gets an instance by ID.
diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ReconcileService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ReconcileService.cs
index 28e440a4..21ef8c7f 100644
--- a/src/ZB.MOM.WW.ScadaBridge.Communication/ReconcileService.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ReconcileService.cs
@@ -1,5 +1,6 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Deployment;
@@ -18,6 +19,11 @@ namespace ZB.MOM.WW.ScadaBridge.Communication;
/// replicate. The node fetches each gap item's config over the existing token-gated HTTP
/// endpoint; this service stages those tokens as PendingDeployment rows.
///
+/// The whole pass is a fixed number of round trips regardless of gap size: the expected-set
+/// projection, ONE untracked read of the gap's snapshot configs, and ONE batched
+/// insert-if-absent of the pending rows. Work is bounded by
+/// — the site's own deadline for the reply.
+///
/// Scoped service (holds scoped repositories); resolved per-request by the actor inside a
/// DI scope. The actor stays thin: it captures Sender, resolves this service in a
/// scope, awaits , and pipes the response back.
@@ -73,92 +79,96 @@ public class ReconcileService
Array.Empty(), Array.Empty(), baseUrl);
}
- // 2. Central's expected deployed set for this site (instances with a snapshot).
+ // Central does no more work than the site will wait for: the node's own Ask/gRPC
+ // deadline is QueryTimeout, so past it every staged token would go to a caller that
+ // has already given up. (The actor-lifetime token passed in is not request-scoped.)
+ using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+ if (_options.QueryTimeout > TimeSpan.Zero)
+ deadline.CancelAfter(_options.QueryTimeout);
+ var ct = deadline.Token;
+
+ // 2. Central's expected deployed set for this site (instances with a snapshot) — a
+ // projection carrying each snapshot's DeploymentId/RevisionHash, no config JSON.
var expected = await _deploymentRepository
- .GetExpectedDeploymentsForSiteAsync(site.Id, cancellationToken)
+ .GetExpectedDeploymentsForSiteAsync(site.Id, ct)
.ConfigureAwait(false);
var localMap = request.LocalNameToRevisionHash;
- var gap = new List();
-
- var now = DateTimeOffset.UtcNow;
- var expiresAt = now + _options.PendingDeploymentTtl;
// 3. GAP = expected items the node is MISSING (name absent locally) or STALE
// (local revision != expected revision). Current items are omitted.
- foreach (var exp in expected)
- {
- var present = localMap.TryGetValue(exp.InstanceUniqueName, out var localHash);
- var stale = present && !string.Equals(localHash, exp.RevisionHash, StringComparison.Ordinal);
- if (present && !stale)
- continue; // node already has the current revision
+ var gapExpected = expected
+ .Where(exp => !localMap.TryGetValue(exp.InstanceUniqueName, out var localHash)
+ || !string.Equals(localHash, exp.RevisionHash, StringComparison.Ordinal))
+ .ToList();
- // 4. Read the frozen snapshot config to stage. Null = the snapshot was deleted
- // between the expected-set query and now (instance removed mid-reconcile);
- // skip it — best-effort reconcile re-runs on the next node startup.
- var snapshot = await _deploymentRepository
- .GetDeployedSnapshotByInstanceIdAsync(exp.InstanceId, cancellationToken)
+ var gap = new List(gapExpected.Count);
+ if (gapExpected.Count > 0)
+ {
+ // 4. ONE read of the frozen snapshot configs, for the gap only — this is the only
+ // query that materialises ConfigurationJson. A missing entry means the snapshot
+ // was deleted between the expected-set query and now (instance removed
+ // mid-reconcile); it is skipped.
+ var snapshots = await _deploymentRepository
+ .GetDeployedSnapshotsByInstanceIdsAsync(
+ gapExpected.Select(e => e.InstanceId).ToList(), ct)
.ConfigureAwait(false);
- if (snapshot == null)
+
+ var now = DateTimeOffset.UtcNow;
+ var expiresAt = now + _options.PendingDeploymentTtl;
+
+ // Stage with the snapshot's DeploymentId so the gap item's DeploymentId + token point
+ // the node at the right pending row to fetch.
+ var staging = new List<(ExpectedDeployment Expected, PendingDeployment Candidate)>(gapExpected.Count);
+ foreach (var exp in gapExpected)
{
- _logger.LogDebug(
- "Reconcile: snapshot for instance {Instance} disappeared (deleted race); skipping",
- exp.InstanceUniqueName);
- continue;
+ if (!snapshots.TryGetValue(exp.InstanceId, out var snapshot))
+ {
+ _logger.LogDebug(
+ "Reconcile: snapshot for instance {Instance} disappeared (deleted race); skipping",
+ exp.InstanceUniqueName);
+ continue;
+ }
+
+ staging.Add((exp, new PendingDeployment(
+ snapshot.DeploymentId, exp.InstanceId, snapshot.RevisionHash,
+ snapshot.ConfigurationJson, DeploymentFetchToken.Generate(), now, expiresAt)));
}
- var token = DeploymentFetchToken.Generate();
-
- // Stage with the snapshot's DeploymentId as the deploymentId so the gap item's
- // DeploymentId + token point the node at the right pending row to fetch.
- //
- // StagePendingIfAbsent is insert-if-absent: if BOTH site nodes are concurrently
- // missing the same instance (e.g. fresh container start / cleared SQLite after a
- // successful deploy), both attempt to stage here. The first succeeds (true); the
- // second gets false and is handled in the !staged branch below — it returns the
- // existing pending row's token so it heals in the same round, rather than being
- // omitted. Deploy-time supersession serializes via the per-instance operation lock.
- var staged = await _deploymentRepository.StagePendingIfAbsentAsync(
- exp.InstanceId, snapshot.DeploymentId, exp.RevisionHash,
- snapshot.ConfigurationJson, token, now, expiresAt, cancellationToken)
+ // Insert-if-absent, in one save. If BOTH site nodes are concurrently missing the same
+ // instance (fresh container start / cleared SQLite after a successful deploy), both
+ // reconcile here; the loser finds the winner's LIVE row and is handed ITS
+ // DeploymentId/RevisionHash/Token (the fetch token is multi-use within its TTL), so
+ // both nodes heal in the same round instead of the second waiting for a restart. The
+ // same applies to an in-flight deploy's row: its config is newer than the snapshot,
+ // and fetching it is still correct (the site's guarded write handles ordering).
+ // Deploy-time supersession serializes via the per-instance operation lock.
+ var liveRows = await _deploymentRepository
+ .StagePendingBatchIfAbsentAsync(staging.Select(x => x.Candidate).ToList(), now, ct)
.ConfigureAwait(false);
- if (!staged)
+ foreach (var (exp, candidate) in staging)
{
- // A pending row already exists for this instance — either a CONCURRENT reconcile
- // from the other site node (both nodes' SQLite empty after a fresh/cleared deploy,
- // both reconciling at startup) or an in-flight deploy. Do NOT omit the item: if we
- // did, the second concurrently-missing node would get 0 fetched and stay unhealed
- // until a later restart. Instead, read the EXISTING pending row and emit a gap item
- // carrying ITS DeploymentId/RevisionHash/Token. The fetch token is multi-use within
- // its TTL, so both nodes fetch the same pending config and heal in the same round.
- // (If the existing row is from an in-flight deploy its config is newer than the
- // snapshot — fetching it is still correct; the site's guarded write handles ordering.)
- var existing = await _deploymentRepository
- .GetPendingDeploymentByInstanceIdAsync(exp.InstanceId, now, cancellationToken)
- .ConfigureAwait(false);
- if (existing != null)
+ if (!liveRows.TryGetValue(candidate.InstanceId, out var row))
+ {
+ // Raced away: no live row could be staged or found (e.g. the instance was
+ // deleted mid-reconcile). Omit it — the node retries on its next pass.
+ _logger.LogDebug(
+ "Reconcile: no pending row could be staged for instance {Instance}; omitting from gap",
+ exp.InstanceUniqueName);
+ continue;
+ }
+
+ if (!ReferenceEquals(row, candidate))
{
_logger.LogDebug(
"Reconcile: pending row already exists for instance {Instance} (concurrent reconcile or in-flight deploy); returning existing token so this node heals too",
exp.InstanceUniqueName);
- gap.Add(new ReconcileGapItem(
- exp.InstanceUniqueName, existing.DeploymentId, existing.RevisionHash,
- exp.IsEnabled, existing.Token));
}
- else
- {
- // Raced away: the pending row was purged between the stage attempt and this
- // read. Omit it — reconcile is best-effort and the node retries next round.
- _logger.LogDebug(
- "Reconcile: pending row for instance {Instance} disappeared between stage and read (purged race); omitting from gap",
- exp.InstanceUniqueName);
- }
- continue;
- }
- gap.Add(new ReconcileGapItem(
- exp.InstanceUniqueName, snapshot.DeploymentId, exp.RevisionHash, exp.IsEnabled, token));
+ gap.Add(new ReconcileGapItem(
+ exp.InstanceUniqueName, row.DeploymentId, row.RevisionHash, exp.IsEnabled, row.Token));
+ }
}
// 5. ORPHANS = names the node has locally that central no longer considers deployed.
diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs
index ef99c45a..ef0d07be 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs
@@ -573,6 +573,145 @@ public class DeploymentManagerRepository : IDeploymentManagerRepository
return !liveExists;
}
+ ///
+ public async Task> GetDeployedSnapshotsByInstanceIdsAsync(
+ IReadOnlyCollection instanceIds, CancellationToken cancellationToken = default)
+ {
+ ArgumentNullException.ThrowIfNull(instanceIds);
+ if (instanceIds.Count == 0)
+ return new Dictionary();
+
+ var ids = instanceIds.Distinct().ToArray();
+ return await _dbContext.Set()
+ .AsNoTracking()
+ .Where(s => ids.Contains(s.InstanceId))
+ .ToDictionaryAsync(s => s.InstanceId, cancellationToken);
+ }
+
+ ///
+ public async Task> StagePendingBatchIfAbsentAsync(
+ IReadOnlyList candidates, DateTimeOffset nowUtc,
+ CancellationToken cancellationToken = default)
+ {
+ ArgumentNullException.ThrowIfNull(candidates);
+ var byInstance = new Dictionary();
+ foreach (var candidate in candidates)
+ byInstance.TryAdd(candidate.InstanceId, candidate);
+ if (byInstance.Count == 0)
+ return byInstance;
+
+ var ids = byInstance.Keys.ToArray();
+
+ // Same two-step contract as StagePendingIfAbsentAsync, set-based: drop EXPIRED rows
+ // first (they would otherwise read as in-flight and collide on the DeploymentId unique
+ // index when a snapshot's own DeploymentId is re-staged), then insert only where no
+ // LIVE row exists. A live row is never superseded.
+ await DeletePendingWhereAsync(
+ p => ids.Contains(p.InstanceId) && p.ExpiresAtUtc <= nowUtc, cancellationToken);
+
+ var live = await ReadLivePendingAsync(ids, nowUtc, cancellationToken);
+ var added = new List();
+ foreach (var (instanceId, candidate) in byInstance)
+ {
+ if (live.ContainsKey(instanceId))
+ continue;
+ added.Add(candidate);
+ }
+
+ if (added.Count == 0)
+ return live;
+
+ await _dbContext.Set().AddRangeAsync(added, cancellationToken);
+ try
+ {
+ await _dbContext.SaveChangesAsync(cancellationToken);
+ }
+ catch (DbUpdateException)
+ {
+ // The batch lost a race — most commonly the peer site node's concurrent reconcile
+ // inserted the same snapshot DeploymentId first (unique index), or an instance was
+ // deleted mid-reconcile (FK). Un-track the failed inserts and fall back to staging
+ // one row at a time, so one bad row costs only itself and a row the peer staged is
+ // picked up as the live row it now is.
+ foreach (var row in added)
+ _dbContext.Entry(row).State = EntityState.Detached;
+
+ return await StagePendingOneByOneAsync(byInstance, nowUtc, cancellationToken);
+ }
+
+ foreach (var row in added)
+ live[row.InstanceId] = row;
+ return live;
+ }
+
+ ///
+ /// Fallback for after its single save failed:
+ /// stages each candidate through , skipping (and
+ /// omitting from the result) any single row that still cannot be written.
+ ///
+ private async Task> StagePendingOneByOneAsync(
+ IReadOnlyDictionary byInstance, DateTimeOffset nowUtc,
+ CancellationToken cancellationToken)
+ {
+ var result = new Dictionary();
+ foreach (var (instanceId, candidate) in byInstance)
+ {
+ bool staged;
+ try
+ {
+ staged = await StagePendingIfAbsentAsync(
+ instanceId, candidate.DeploymentId, candidate.RevisionHash,
+ candidate.ConfigurationJson, candidate.Token,
+ nowUtc, candidate.ExpiresAtUtc, cancellationToken);
+ }
+ catch (DbUpdateException)
+ {
+ DetachPendingChanges();
+ continue;
+ }
+
+ if (staged)
+ {
+ result[instanceId] = candidate;
+ continue;
+ }
+
+ var existing = await ReadLivePendingAsync([instanceId], nowUtc, cancellationToken);
+ if (existing.TryGetValue(instanceId, out var row))
+ result[instanceId] = row;
+ }
+
+ return result;
+ }
+
+ ///
+ /// Reads the most recent LIVE pending row per instance, projected WITHOUT
+ /// ConfigurationJson (the caller only needs the fetch coordinates), untracked.
+ ///
+ private async Task> ReadLivePendingAsync(
+ int[] instanceIds, DateTimeOffset nowUtc, CancellationToken cancellationToken)
+ {
+ var rows = await _dbContext.Set()
+ .AsNoTracking()
+ .Where(p => instanceIds.Contains(p.InstanceId) && p.ExpiresAtUtc > nowUtc)
+ .Select(p => new
+ {
+ p.Id, p.DeploymentId, p.InstanceId, p.RevisionHash, p.Token, p.CreatedAtUtc, p.ExpiresAtUtc
+ })
+ .ToListAsync(cancellationToken);
+
+ // At most one live row per instance by design; on a hypothetical duplicate take the
+ // most recent, mirroring GetPendingDeploymentByInstanceIdAsync's ordering.
+ return rows
+ .GroupBy(r => r.InstanceId)
+ .Select(g => g.OrderByDescending(r => r.CreatedAtUtc).ThenByDescending(r => r.Id).First())
+ .ToDictionary(
+ r => r.InstanceId,
+ r => new PendingDeployment(
+ r.DeploymentId, r.InstanceId, r.RevisionHash, string.Empty, r.Token,
+ r.CreatedAtUtc, r.ExpiresAtUtc) { Id = r.Id });
+ }
+
// --- Instance lookups for deployment pipeline ---
///
diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReconciliationActor.cs
index be5827fe..fa742543 100644
--- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReconciliationActor.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReconciliationActor.cs
@@ -8,8 +8,8 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
///
/// Runs on EVERY site node (NOT a singleton) so a standby that was DOWN during a deploy
-/// self-heals on its next restart. On startup the actor performs one best-effort
-/// reconciliation pass:
+/// self-heals on its next restart. On startup the actor performs a best-effort
+/// reconciliation pass (retried a bounded number of times, see remarks):
///
/// - read the node's local deployed inventory from SQLite,
/// - report it to central via the SiteCommunicationActor Ask
@@ -23,10 +23,18 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
///
///
/// Best-effort throughout. A central-unreachable / timed-out Ask is caught, logged
-/// at Warning, and the pass simply ends — reconcile re-runs on the next node startup; it is a
-/// self-heal, not a critical path. A per-item fetch/write failure is caught and logged, then
-/// the remaining gap items continue (one bad item must not abort the rest). The actor never
-/// crashes on these failures.
+/// at Warning, and the pass ends; it is a self-heal, not a critical path. A per-item
+/// fetch/write failure is caught and logged, then the remaining gap items continue (one bad
+/// item must not abort the rest). The actor never crashes on these failures.
+///
+///
+/// Bounded retry. A pass that did not complete (central unreachable, Ask timeout,
+/// local read error) or completed with failed items is re-run, up to maxAttempts
+/// passes in total, after an exponential back-off (retryBaseDelay, then 2x, 4x...).
+/// Each retry re-reads the local inventory, so items an earlier pass healed are no longer in
+/// the gap. Before this, a failed pass waited for the NEXT node restart — leaving exactly the
+/// node this actor exists to heal unhealed. After the last attempt the actor logs and stops
+/// trying until the next startup.
///
///
/// The pass runs after a small startup delay (so the site's central transport has time to
@@ -63,6 +71,12 @@ public sealed class SiteReconciliationActor : ReceiveActor, IWithTimers
private readonly ILogger _logger;
private readonly TimeSpan _initialDelay;
private readonly TimeSpan _askTimeout;
+ private readonly int _maxAttempts;
+ private readonly TimeSpan _retryBaseDelay;
+
+ // Number of passes started so far (1-based once the first pass runs). Only touched on
+ // the actor thread.
+ private int _attempt;
/// Akka timer scheduler injected by the framework via .
public ITimerScheduler Timers { get; set; } = null!;
@@ -93,6 +107,13 @@ public sealed class SiteReconciliationActor : ReceiveActor, IWithTimers
/// node's reconcile pass found the gap. Optional: when null the pass stores configs and
/// starts nothing, exactly as it did before.
///
+ ///
+ /// Total passes allowed per actor lifetime when a pass fails or leaves failed items.
+ /// Defaults to 3; values below 1 are treated as 1 (no retry).
+ ///
+ ///
+ /// Back-off before the first retry, doubled for each later one. Defaults to 30 seconds.
+ ///
public SiteReconciliationActor(
SiteStorageService storage,
IDeploymentConfigFetcher configFetcher,
@@ -102,7 +123,9 @@ public sealed class SiteReconciliationActor : ReceiveActor, IWithTimers
ILogger logger,
TimeSpan? initialDelay = null,
TimeSpan? askTimeout = null,
- IActorRef? deploymentManager = null)
+ IActorRef? deploymentManager = null,
+ int maxAttempts = 3,
+ TimeSpan? retryBaseDelay = null)
{
_storage = storage;
_configFetcher = configFetcher;
@@ -113,23 +136,33 @@ public sealed class SiteReconciliationActor : ReceiveActor, IWithTimers
_logger = logger;
_initialDelay = initialDelay ?? TimeSpan.FromSeconds(5);
_askTimeout = askTimeout ?? TimeSpan.FromSeconds(30);
+ _maxAttempts = Math.Max(1, maxAttempts);
+ _retryBaseDelay = retryBaseDelay ?? TimeSpan.FromSeconds(30);
- Receive(_ => RunReconcilePassAsync().PipeTo(Self));
+ Receive(_ =>
+ {
+ _attempt++;
+ RunReconcilePassAsync().PipeTo(Self);
+ });
Receive(HandleReconcilePassResult);
// Defensive: RunReconcilePassAsync is designed never to throw (it returns a faulted
// ReconcilePassResult instead), but if anything unexpected faults the piped Task the
// Status.Failure would otherwise go to dead letters silently. Log it instead.
- Receive(f => _logger.LogWarning(f.Cause,
- "Reconcile pass faulted unexpectedly for site {Site} node {Node}",
- _siteIdentifier, _nodeId));
+ Receive(f =>
+ {
+ _logger.LogWarning(f.Cause,
+ "Reconcile pass faulted unexpectedly for site {Site} node {Node}",
+ _siteIdentifier, _nodeId);
+ ScheduleRetryIfAllowed();
+ });
}
///
protected override void PreStart()
{
base.PreStart();
- // One-shot pass after a small delay so the site's central transport can connect first.
+ // First pass after a small delay so the site's central transport can connect first.
// Non-blocking: the timer fires RunReconcile back onto this actor's mailbox.
Timers.StartSingleTimer(StartupTimerKey, RunReconcile.Instance, _initialDelay);
_logger.LogInformation(
@@ -239,16 +272,24 @@ public sealed class SiteReconciliationActor : ReceiveActor, IWithTimers
if (result.Error != null)
{
// Best-effort: a failed pass (central unreachable, Ask timeout, local read error) is
- // logged at Warning and the actor stays alive. Reconcile re-runs on the next startup.
+ // logged at Warning and the actor stays alive; it is retried with back-off up to the
+ // attempt cap, then left to the next startup.
_logger.LogWarning(result.Error,
- "Reconcile pass for site {Site} node {Node} did not complete (central unreachable or read error) — will retry on next startup",
- _siteIdentifier, _nodeId);
+ "Reconcile pass {Attempt}/{MaxAttempts} for site {Site} node {Node} did not complete (central unreachable or read error)",
+ _attempt, _maxAttempts, _siteIdentifier, _nodeId);
+ ScheduleRetryIfAllowed();
return;
}
_logger.LogInformation(
- "Reconcile pass for site {Site} node {Node} complete: {Fetched} fetched, {Failed} failed, {Orphans} orphan(s)",
- _siteIdentifier, _nodeId, result.Fetched, result.Failed, result.Orphans);
+ "Reconcile pass {Attempt}/{MaxAttempts} for site {Site} node {Node} complete: {Fetched} fetched, {Failed} failed, {Orphans} orphan(s)",
+ _attempt, _maxAttempts, _siteIdentifier, _nodeId, result.Fetched, result.Failed, result.Orphans);
+
+ // Items that failed to fetch/store are still missing: retry the pass (the re-read
+ // inventory makes the gap only what is still missing). The healed instances below are
+ // started now regardless — a partial pass still healed them.
+ if (result.Failed > 0)
+ ScheduleRetryIfAllowed();
// Heal the RUNTIME, not just the file (row 67). Storing a config starts nothing: the
// Deployment Manager's boot-time SQLite load has already run by now, so without this
@@ -268,9 +309,30 @@ public sealed class SiteReconciliationActor : ReceiveActor, IWithTimers
}
}
+ ///
+ /// Schedules the next reconcile pass after an exponential back-off, unless the attempt cap
+ /// is reached — in which case the actor logs and leaves the heal to the next node startup.
+ ///
+ private void ScheduleRetryIfAllowed()
+ {
+ if (_attempt >= _maxAttempts)
+ {
+ _logger.LogWarning(
+ "Reconcile for site {Site} node {Node} gave up after {Attempts} attempt(s) — will retry on next startup",
+ _siteIdentifier, _nodeId, _attempt);
+ return;
+ }
+
+ var delay = TimeSpan.FromTicks(_retryBaseDelay.Ticks * (1L << Math.Min(_attempt - 1, 16)));
+ _logger.LogInformation(
+ "Reconcile for site {Site} node {Node}: retrying in {Delay} (attempt {Next}/{MaxAttempts})",
+ _siteIdentifier, _nodeId, delay, _attempt + 1, _maxAttempts);
+ Timers.StartSingleTimer(StartupTimerKey, RunReconcile.Instance, delay);
+ }
+
// ── Internal messages ──
- /// Self-tick that drives the one-shot startup reconcile pass.
+ /// Self-tick that drives a startup reconcile pass (the first, or a retry).
private sealed class RunReconcile
{
public static readonly RunReconcile Instance = new();
diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs
index ebb2660a..b9be76fe 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs
@@ -40,13 +40,17 @@ public class CentralCommunicationActorReconcileTests : TestKit
{
new(2, "inst-B", "rev2", "dep-B", true),
});
- deploymentRepo.GetDeployedSnapshotByInstanceIdAsync(2, Arg.Any())
- .Returns(new DeployedConfigSnapshot("dep-B", "rev2", "{\"cfg\":\"B\"}"));
- deploymentRepo.StagePendingIfAbsentAsync(
- Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(),
- Arg.Any(), Arg.Any(), Arg.Any(),
- Arg.Any())
- .Returns(true);
+ deploymentRepo.GetDeployedSnapshotsByInstanceIdsAsync(
+ Arg.Any>(), Arg.Any())
+ .Returns(new Dictionary
+ {
+ [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>(), Arg.Any(), Arg.Any())
+ .Returns(ci => (IReadOnlyDictionary)ci.ArgAt>(0)
+ .ToDictionary(p => p.InstanceId));
var options = Options.Create(new CommunicationOptions
{
diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ReconcileServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ReconcileServiceTests.cs
index 13227a73..744ddad0 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ReconcileServiceTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ReconcileServiceTests.cs
@@ -25,12 +25,43 @@ public class ReconcileServiceTests
Substitute.For();
private readonly ISiteRepository _siteRepo = Substitute.For();
- private ReconcileService CreateService(TimeSpan? ttl = null)
+ // Fake repository state behind the two batched reconcile calls.
+ private readonly Dictionary _snapshots = new();
+ private readonly Dictionary _existingLive = new();
+ private readonly HashSet _racedAway = new();
+
+ /// Wires the batched snapshot read and batched stage to the fake state above.
+ public ReconcileServiceTests()
+ {
+ _deploymentRepo.GetDeployedSnapshotsByInstanceIdsAsync(
+ Arg.Any>(), Arg.Any())
+ .Returns(ci => (IReadOnlyDictionary)ci.ArgAt>(0)
+ .Where(_snapshots.ContainsKey)
+ .ToDictionary(id => id, id => _snapshots[id]));
+
+ _deploymentRepo.StagePendingBatchIfAbsentAsync(
+ Arg.Any>(), Arg.Any(), Arg.Any())
+ .Returns(ci =>
+ {
+ var live = new Dictionary();
+ foreach (var candidate in ci.ArgAt>(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)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.Instance);
@@ -48,15 +79,11 @@ public class ReconcileServiceTests
.Returns(expected.ToList());
private void SnapshotFor(ExpectedDeployment exp) =>
- _deploymentRepo.GetDeployedSnapshotByInstanceIdAsync(exp.InstanceId, Arg.Any())
- .Returns(new DeployedConfigSnapshot(exp.DeploymentId, exp.RevisionHash, $"{{\"cfg\":\"{exp.InstanceUniqueName}\"}}"));
-
- private void StageReturns(bool result) =>
- _deploymentRepo.StagePendingIfAbsentAsync(
- Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(),
- Arg.Any(), Arg.Any(), Arg.Any(),
- Arg.Any())
- .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);
}
- /// Verifies that when staging a pending deployment returns false and a row already exists, the gap carries the existing token.
+ /// Verifies that when a live pending row already exists for a gap instance, the gap carries the existing token.
/// A task that represents the asynchronous operation.
[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(), Arg.Any(), Arg.Any(),
- Arg.Any(), Arg.Any(), Arg.Any(),
- Arg.Any())
- .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(), Arg.Any())
- .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);
}
- /// Verifies that when staging a pending deployment returns false and no row exists, that gap item is omitted.
+ /// Verifies that when no pending row can be staged or found, that gap item is omitted.
/// A task that represents the asynchronous operation.
[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(), Arg.Any(), Arg.Any(),
- Arg.Any(), Arg.Any(), Arg.Any(),
- Arg.Any())
- .Returns(false);
-
- // The pending row was purged between the stage attempt and the read.
- _deploymentRepo.GetPendingDeploymentByInstanceIdAsync(3, Arg.Any(), Arg.Any())
- .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(), Arg.Any());
}
+
+ // ── review 2026-10-21 03/P-1: batched reconcile + bounded work ──
+
+ ///
+ /// 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.
+ ///
+ /// A task that represents the asynchronous operation.
+ [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>(ids => ids.Order().SequenceEqual(new[] { 2, 3 })),
+ Arg.Any());
+ await _deploymentRepo.Received(1).StagePendingBatchIfAbsentAsync(
+ Arg.Is>(rows => rows.Count == 2),
+ Arg.Any(), Arg.Any());
+ await _deploymentRepo.DidNotReceive().GetDeployedSnapshotByInstanceIdAsync(
+ Arg.Any(), Arg.Any());
+ await _deploymentRepo.DidNotReceive().StagePendingIfAbsentAsync(
+ Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(),
+ Arg.Any(), Arg.Any(), Arg.Any(),
+ Arg.Any());
+ }
+
+ /// A node with nothing missing costs no snapshot read and no staging at all.
+ /// A task that represents the asynchronous operation.
+ [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>(), Arg.Any());
+ await _deploymentRepo.DidNotReceive().StagePendingBatchIfAbsentAsync(
+ Arg.Any>(), Arg.Any(), Arg.Any());
+ }
+
+ ///
+ /// Central stops working once the site's own deadline (QueryTimeout) has passed instead of
+ /// staging tokens for a caller that has already given up.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task Reconcile_WorkIsBoundedByQueryTimeout()
+ {
+ SiteResolves();
+ _deploymentRepo.GetExpectedDeploymentsForSiteAsync(SiteId, Arg.Any())
+ .Returns(async ci =>
+ {
+ await Task.Delay(TimeSpan.FromSeconds(30), ci.ArgAt(1));
+ return (IReadOnlyList)new List();
+ });
+
+ var service = CreateService(queryTimeout: TimeSpan.FromMilliseconds(100));
+
+ await Assert.ThrowsAnyAsync(
+ () => service.ReconcileAsync(Request()));
+ }
}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/ReconcileRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/ReconcileRepositoryTests.cs
index 556a71e1..adb6fbc2 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/ReconcileRepositoryTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/ReconcileRepositoryTests.cs
@@ -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));
+
+ /// The gap snapshot read returns only the requested instances, untracked.
+ /// A task that represents the asynchronous operation.
+ [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());
+ }
+
+ ///
+ /// 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.
+ ///
+ /// A task that represents the asynchronous operation.
+ [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().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"));
+ }
+
+ ///
+ /// 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.
+ ///
+ /// A task that represents the asynchronous operation.
+ [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"));
+ }
}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReconciliationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReconciliationActorTests.cs
index f256d0ab..f610eff3 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReconciliationActorTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReconciliationActorTests.cs
@@ -53,7 +53,9 @@ public class SiteReconciliationActorTests : TestKit, IDisposable
IActorRef siteCommunicationActor,
IDeploymentConfigFetcher fetcher,
ILogger? 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.Instance,
TimeSpan.FromMilliseconds(50), // initial delay — fast for tests
- askTimeout ?? TimeSpan.FromSeconds(5)))); // ask timeout
+ askTimeout ?? TimeSpan.FromSeconds(5), // ask timeout
+ null,
+ maxAttempts,
+ retryBaseDelay)));
/// Verifies that an instance missing at the site is fetched from central and stored.
/// A task that represents the asynchronous operation.
@@ -296,6 +301,68 @@ public class SiteReconciliationActorTests : TestKit, IDisposable
Assert.Empty(fetcher.Calls);
}
+ ///
+ /// 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.
+ ///
+ [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(TimeSpan.FromSeconds(5));
+ commProbe.ExpectMsg(TimeSpan.FromSeconds(5));
+ commProbe.ExpectMsg(TimeSpan.FromSeconds(5));
+ commProbe.ExpectNoMsg(TimeSpan.FromSeconds(1.5));
+ Assert.Empty(fetcher.Calls);
+ }
+
+ ///
+ /// review 2026-10-21 03/P-1: a pass that completes with a failed item is retried, and the
+ /// retry heals the item.
+ ///
+ /// A task that represents the asynchronous operation.
+ [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(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(TimeSpan.FromSeconds(5));
+ commProbe.Reply(gap);
+
+ var retry = commProbe.ExpectMsg(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));
+ }
+
///
/// In-test fake : runs a per-deploymentId behavior
/// (return config JSON or throw, as a Task) and records every call's coords thread-safely