perf(site-runtime): instance reconcile sweep reads metadata only and fetches config_json just for the instances it repairs (review 2026-10-21 01/SR-P1)

Adds SiteStorageService.GetAllDeployedConfigMetadataAsync. The sweep decides
from the projection, then fetches single rows for its starts/restarts off-
thread and re-validates them (generation guard, revision match, pipeline/
running guards), releasing the per-revision claim of any repair that does
not act. Boot load unchanged.
This commit is contained in:
Joseph Doherty
2026-10-22 06:42:06 -04:00
parent c782744b07
commit 960c35060c
6 changed files with 415 additions and 42 deletions
@@ -104,6 +104,8 @@ flowchart TD
>
> Both are closed by one **periodic sweep owned by the Deployment Manager singleton** (`ScadaBridge:SiteRuntime:InstanceSweepIntervalSeconds`, default 60, validated `>= 0`; `0` disables it). Each tick re-reads the durable rows off-thread and, on the actor thread, for every **enabled** stored config: starts the Instance Actor if none is running (row 78), or — if one is running at a revision the store has since replaced — **re-applies the stored config through the exact per-instance apply/restart path a central redeploy uses** (row 72). A sweep is chosen over a LocalDb change hook deliberately: a hook would cover only the replication arrival path, only on replicated sites, and `ILocalDb` exposes no change-notification API at all, whereas the sweep is arrival-path-agnostic (replication seeding, post-boot snapshot resync, a hand-restored file, any future path). See `docs/plans/2026-08-15-instance-sweep-self-heal-design.md`.
>
> **The sweep reads metadata, not configs (review 2026-10-21 01/SR-P1).** Each tick compares only name / enabled / revision, but it used to read every row's `config_json` (~70 KB worst case) — on a 500-instance site ~35 MB of large-object-heap strings a minute, forever, on a perfectly healthy site. The tick now reads the `GetAllDeployedConfigMetadataAsync` projection (no `config_json`), decides its repairs on the actor thread exactly as before, and only then fetches the full row — one single-row read per instance it is about to start or restart — off-thread. When those rows land every decision is **re-validated**: a fetch this node has since mutated past is dropped whole by the same store-write generation guard, a row that vanished, was disabled or now carries a different revision than the one claimed is skipped, and the deploy-pipeline / running-actor guards are re-checked; a repair that does not act gives back its once-per-revision claim so the next tick re-evaluates it. Ticks stay suppressed until the fetch lands. The boot load keeps the full read — it runs once and needs every blob.
> The **row-72 decision is to converge, not merely warn**: the stored row *is* what central successfully deployed, so an actor serving the config central believes was replaced is the actual fault. Reusing `HandleDeploy` means the compile gate, warm-then-gate ordering, redeploy-supersede, the static-override / native-alarm reset and all-or-nothing-per-instance are the production ones — and a stop still does **not** cancel an in-flight script run, exactly as for a central redeploy.
>
> Guards, all shared with the row-67 handler: a **disabled** instance is never started; the sweep **defers to any deploy pipeline that owns the instance** (compile warm in flight, predecessor still terminating, deployment awaiting its persistence/init join) so it can never interleave with a real redeploy; it is **inert until boot-time Instance Actor creation has finished**, so a tick between staggered batches does not "heal" instances the boot load is about to create; it makes **one repair attempt per (instance, revision) pair**, which is what stops a periodic self-heal becoming a restart loop for an actor that dies during init or a stored config the compile gate rejects (a rejection leaves the running instance untouched — the gate returns before anything is stopped); and reads never overlap. One guard is **not** in that list because it needed adding later: deferring to an in-flight deploy pipeline says nothing about a snapshot that landed *after* the pipeline finished, which is the stale-snapshot race — a tick whose read raced this node's own deploy/enable/disable/delete is discarded whole by the store-write generation check (see **Stale-snapshot guard** below). A sweep that finds nothing is **completely silent**. A repair emits the same Warning `deployment` site-event pattern as row 67, naming the trigger (`deployment-manager sweep self-heal`) so it is distinguishable from a startup-reconcile heal, and a stale-revision restart names **both** revisions; a rejected restart adds an Error event. The synthetic redeploy replies to the singleton itself rather than to `NoSender`, because dead letters are a monitored site health metric.
@@ -107,6 +107,13 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// </remarks>
private readonly Func<string, Task<DeployedInstance?>> _deployedConfigLoader;
/// <summary>
/// Reads every deployed row's identity metadata WITHOUT <c>config_json</c> for the instance
/// reconcile sweep (review 2026-10-21 01/SR-P1). Defaults to
/// <see cref="SiteStorageService.GetAllDeployedConfigMetadataAsync"/>; overridable via the ctor.
/// Deliberately separate from <see cref="_configLoader"/>, which stays the boot-only full read.
/// </summary>
private readonly Func<Task<List<DeployedConfigMetadata>>> _configMetadataLoader;
/// <summary>
/// Reads the durable <c>data_connection_definitions</c> rows (deferred register rows 89 + 90).
/// Defaults to <see cref="SiteStorageService.GetAllDataConnectionDefinitionsAsync"/>;
/// overridable via the ctor so a test can drive a read failure.
@@ -315,6 +322,10 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// ensure-started heal uses; defaults to reading from <paramref name="storage"/>. Primarily
/// for tests that need to hold the read open mid-flight to drive the freshness race.
/// </param>
/// <param name="configMetadataLoader">
/// Optional override for the instance reconcile sweep's metadata-only read (01/SR-P1);
/// defaults to reading from <paramref name="storage"/>.
/// </param>
/// <param name="secretsMirror">
/// This node's <see cref="DeployedSecretsMirror"/>,
/// reconciled synchronously by <see cref="HandleDeployArtifacts"/> right after a shipped
@@ -339,7 +350,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
ScriptExecutionScheduler? scriptScheduler = null,
Func<Task<List<StoredDataConnectionDefinition>>>? connectionDefinitionLoader = null,
Func<string, Task<DeployedInstance?>>? deployedConfigLoader = null,
DeployedSecretsMirror? secretsMirror = null)
DeployedSecretsMirror? secretsMirror = null,
Func<Task<List<DeployedConfigMetadata>>>? configMetadataLoader = null)
{
_scriptScheduler = scriptScheduler;
_secretsMirror = secretsMirror;
@@ -360,6 +372,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
: TimeSpan.Zero;
_configLoader = configLoader ?? (() => _storage.GetAllDeployedConfigsAsync());
_deployedConfigLoader = deployedConfigLoader ?? (name => _storage.GetDeployedConfigAsync(name));
_configMetadataLoader = configMetadataLoader ?? (() => _storage.GetAllDeployedConfigMetadataAsync());
_connectionDefinitionLoader = connectionDefinitionLoader
?? (() => _storage.GetAllDataConnectionDefinitionsAsync());
_logger = logger;
@@ -405,6 +418,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// no longer matches the stored one (row 72).
Receive<InstanceSweepTick>(_ => HandleInstanceSweepTick());
Receive<InstanceSweepLoaded>(HandleInstanceSweepLoaded);
// 01/SR-P1: the full rows of the instances a sweep decided to start/restart.
Receive<InstanceSweepRepairsLoaded>(HandleInstanceSweepRepairsLoaded);
// Terminal reply of a sweep-driven stale-revision restart. Every OTHER deploy path
// replies to central (or to a test probe); only the sweep makes this actor its own
// deployer, so a response landing here is always its own and is surfaced, never
@@ -624,20 +639,39 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
ReadStoreSnapshotAsync()
{
var configs = await _configLoader().ConfigureAwait(false);
return (configs, await ReadConnectionDefinitionsAsync().ConfigureAwait(false));
}
IReadOnlyList<StoredDataConnectionDefinition>? definitions = null;
/// <summary>
/// The sweep's variant of <see cref="ReadStoreSnapshotAsync"/> (review 2026-10-21 01/SR-P1):
/// identical failure semantics, but the configs half is the metadata-only projection — the
/// sweep compares name / enabled / revision only, and fetches a full row solely for the
/// instances it goes on to start or restart (<see cref="HandleInstanceSweepRepairsLoaded"/>).
/// </summary>
private async Task<(List<DeployedConfigMetadata> Configs, IReadOnlyList<StoredDataConnectionDefinition>? Definitions)>
ReadSweepSnapshotAsync()
{
var configs = await _configMetadataLoader().ConfigureAwait(false);
return (configs, await ReadConnectionDefinitionsAsync().ConfigureAwait(false));
}
/// <summary>
/// The definitions half of a store snapshot; a failure is swallowed into
/// <see langword="null"/> (keep the overlay) — see <see cref="ReadStoreSnapshotAsync"/>.
/// </summary>
private async Task<IReadOnlyList<StoredDataConnectionDefinition>?> ReadConnectionDefinitionsAsync()
{
try
{
definitions = await _connectionDefinitionLoader().ConfigureAwait(false);
return await _connectionDefinitionLoader().ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(
ex, "Could not read the data connection definitions — the DCL arbitration "
+ "overlay keeps its previous contents until the next read succeeds");
return null;
}
return (configs, definitions);
}
/// <inheritdoc />
@@ -1949,7 +1983,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// Row 101: capture on the actor thread BEFORE the off-thread read starts.
var generationAtRead = _storeWriteGeneration;
ReadStoreSnapshotAsync().ContinueWith(t => t.IsCompletedSuccessfully
ReadSweepSnapshotAsync().ContinueWith(t => t.IsCompletedSuccessfully
? new InstanceSweepLoaded(t.Result.Configs, null, t.Result.Definitions, generationAtRead)
: new InstanceSweepLoaded([], t.Exception?.GetBaseException().Message))
.PipeTo(Self);
@@ -1993,6 +2027,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// row out of its own file on the next tick.
ApplyConnectionDefinitionOverlay(msg.ConnectionDefinitions);
var repairs = new List<SweepRepair>();
foreach (var config in msg.Configs)
{
var instanceName = config.InstanceUniqueName;
@@ -2010,10 +2045,94 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
if (!_instanceActors.ContainsKey(instanceName))
{
// Row 78 — an enabled config with no actor.
if (!TryClaimSweepAttempt(instanceName, config.RevisionHash))
continue;
if (TryClaimSweepAttempt(instanceName, config.RevisionHash))
repairs.Add(new SweepRepair(instanceName, config.RevisionHash, Restart: false));
continue;
}
StartStoredInstance(config, SweepReason,
// Row 72 — a running actor serving a revision the store has replaced.
if (IsRunningStaleRevision(config))
repairs.Add(new SweepRepair(instanceName, config.RevisionHash, Restart: true));
}
if (repairs.Count == 0)
return;
// 01/SR-P1: the snapshot carried no config_json, so fetch the full rows of exactly the
// instances being repaired — off-thread, one single-row read each — and act when they
// land. Ticks stay suppressed meanwhile so a slow fetch cannot overlap the next read.
_instanceSweepInFlight = true;
var generationAtFetch = _storeWriteGeneration;
var loader = _deployedConfigLoader;
Task.Run(async () =>
{
var rows = new Dictionary<string, DeployedInstance?>(StringComparer.Ordinal);
foreach (var repair in repairs)
rows[repair.InstanceName] = await loader(repair.InstanceName).ConfigureAwait(false);
return rows;
}).ContinueWith(t => t.IsCompletedSuccessfully
? new InstanceSweepRepairsLoaded(repairs, t.Result, null, generationAtFetch)
: new InstanceSweepRepairsLoaded(
repairs, null, t.Exception?.GetBaseException().Message, generationAtFetch))
.PipeTo(Self);
}
/// <summary>
/// Actor-thread half of a sweep's repairs (review 2026-10-21 01/SR-P1): the full rows for the
/// instances <see cref="HandleInstanceSweepLoaded"/> decided to start or restart have landed.
///
/// <para>
/// Every decision is RE-VALIDATED against the fetched row and the current runtime state,
/// because both may have moved while the fetch was in flight: a snapshot this node has since
/// mutated past is dropped whole (row 101's generation guard, exactly as for the metadata
/// read); a row that disappeared, was disabled, or now carries a DIFFERENT revision than the
/// one claimed is skipped; and the deploy-pipeline / running-actor guards are re-checked. A
/// repair that does not act RELEASES its per-revision claim, so the next tick re-evaluates it
/// rather than the instance being silently written off.
/// </para>
/// </summary>
private void HandleInstanceSweepRepairsLoaded(InstanceSweepRepairsLoaded msg)
{
_instanceSweepInFlight = false;
if (msg.Error != null || msg.Rows == null)
{
_logger.LogWarning(
"Instance reconcile sweep could not read the configs it meant to repair: {Error}", msg.Error);
foreach (var repair in msg.Repairs)
ReleaseSweepAttempt(repair.InstanceName, repair.RevisionHash);
return;
}
if (IsStoreSnapshotStale(msg.StoreGenerationAtRead, "instance sweep repair"))
{
foreach (var repair in msg.Repairs)
ReleaseSweepAttempt(repair.InstanceName, repair.RevisionHash);
return;
}
foreach (var repair in msg.Repairs)
{
var name = repair.InstanceName;
msg.Rows.TryGetValue(name, out var row);
if (row == null || !row.IsEnabled ||
!string.Equals(row.RevisionHash, repair.RevisionHash, StringComparison.Ordinal) ||
InstanceIsOwnedByADeployPipeline(name))
{
ReleaseSweepAttempt(name, repair.RevisionHash);
continue;
}
if (!repair.Restart)
{
if (_instanceActors.ContainsKey(name))
{
ReleaseSweepAttempt(name, repair.RevisionHash);
continue;
}
StartStoredInstance(row, SweepReason,
"An enabled instance had a deployed config in the site store but no running "
+ "Instance Actor. Configs that arrive by LocalDb replication (a standby "
+ "seeded from its peer, or a snapshot resync) are written after the "
@@ -2022,8 +2141,29 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
continue;
}
// Row 72 — a running actor serving a revision the store has replaced.
TryRestartStaleInstance(config);
if (!_instanceActors.ContainsKey(name) ||
!_runningRevisions.TryGetValue(name, out var runningRevision) ||
string.Equals(runningRevision, row.RevisionHash, StringComparison.Ordinal))
{
ReleaseSweepAttempt(name, repair.RevisionHash);
continue;
}
RestartStaleInstance(row, runningRevision);
}
}
/// <summary>
/// Gives back a sweep claim that was taken but not acted on, so the next tick may try the
/// same (instance, revision) again. Only releases the exact claim — a newer revision claimed
/// in the meantime is left alone.
/// </summary>
private void ReleaseSweepAttempt(string instanceName, string revisionHash)
{
if (_sweepAttemptedRevisions.TryGetValue(instanceName, out var claimed) &&
string.Equals(claimed, revisionHash, StringComparison.Ordinal))
{
_sweepAttemptedRevisions.Remove(instanceName);
}
}
@@ -2075,15 +2215,22 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// entirely while a deploy pipeline owns the instance, so it can never interleave with a
/// real redeploy.
/// </para>
///
/// <para>
/// This half only DECIDES and claims the attempt from the metadata snapshot; the config body
/// is fetched afterwards and the restart performed by <see cref="RestartStaleInstance"/>
/// (review 2026-10-21 01/SR-P1).
/// </para>
/// </summary>
private void TryRestartStaleInstance(DeployedInstance config)
/// <returns><see langword="true"/> when a restart should be attempted (claim taken).</returns>
private bool IsRunningStaleRevision(DeployedConfigMetadata config)
{
var instanceName = config.InstanceUniqueName;
var storedRevision = config.RevisionHash;
// No stored revision to compare against — nothing can be concluded, so do nothing.
if (string.IsNullOrEmpty(storedRevision))
return;
return false;
// No recorded running revision. Unreachable while CreateInstanceActor is the single
// creation funnel; treated as "unknown", never as "stale", so a future path that
@@ -2091,14 +2238,23 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
if (!_runningRevisions.TryGetValue(instanceName, out var runningRevision) ||
string.IsNullOrEmpty(runningRevision))
{
return;
return false;
}
if (string.Equals(runningRevision, storedRevision, StringComparison.Ordinal))
return;
return false;
if (!TryClaimSweepAttempt(instanceName, storedRevision))
return;
return TryClaimSweepAttempt(instanceName, storedRevision);
}
/// <summary>
/// Second half of the row-72 repair: re-applies the fetched stored row through the redeploy
/// path (see <see cref="IsRunningStaleRevision"/> for the decision and its safety rules).
/// </summary>
private void RestartStaleInstance(DeployedInstance config, string runningRevision)
{
var instanceName = config.InstanceUniqueName;
var storedRevision = config.RevisionHash;
_logger.LogWarning(
"Instance {Instance} is running stale revision {RunningRevision} while the deployed-config "
@@ -2556,7 +2712,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// STOPPING live connections that nothing re-creates until an instance redeploy; rows push
/// superseded configs onto live connections and flip-flop with the next fresh read.</item>
/// <item>CONFIGS half, deploy-revision leg: a snapshot carrying the revision a deploy has just
/// replaced makes <see cref="TryRestartStaleInstance"/> "repair" the instance BACKWARD —
/// replaced makes <see cref="RestartStaleInstance"/> "repair" the instance BACKWARD —
/// it redeploys the snapshot's own (old) config JSON through the redeploy path.
/// <see cref="InstanceIsOwnedByADeployPipeline"/> does not cover this: it only holds while
/// the deploy is in flight, and the stale snapshot can land after the pipeline released
@@ -3901,11 +4057,32 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// complete, and the next tick is already a fresh read.
/// </param>
internal record InstanceSweepLoaded(
List<DeployedInstance> Configs,
List<DeployedConfigMetadata> Configs,
string? Error,
IReadOnlyList<StoredDataConnectionDefinition>? ConnectionDefinitions = null,
long StoreGenerationAtRead = 0);
/// <summary>
/// One repair a sweep decided on from the metadata snapshot (01/SR-P1): start a missing
/// instance, or restart one running a stale revision, at <paramref name="RevisionHash"/>.
/// </summary>
internal sealed record SweepRepair(string InstanceName, string RevisionHash, bool Restart);
/// <summary>
/// Carries the full rows of a sweep's repairs back onto the actor thread (01/SR-P1).
/// <paramref name="Rows"/> is null when the fetch failed (<paramref name="Error"/> set); a
/// <see langword="null"/> VALUE means the row is gone.
/// </summary>
/// <param name="StoreGenerationAtRead">
/// <see cref="_storeWriteGeneration"/> when the fetch was dispatched — same row-101 guard as
/// <see cref="InstanceSweepLoaded"/>.
/// </param>
internal sealed record InstanceSweepRepairsLoaded(
IReadOnlyList<SweepRepair> Repairs,
IReadOnlyDictionary<string, DeployedInstance?>? Rows,
string? Error,
long StoreGenerationAtRead);
/// <summary>
/// Row 101: piped back to self once an off-thread mutation of one of the two snapshot tables
/// has COMMITTED, purely so <see cref="_storeWriteGeneration"/> can be bumped on the actor
@@ -194,6 +194,45 @@ public class SiteStorageService
IsEnabled: reader.GetInt64(3) != 0);
}
/// <summary>
/// Returns the deployment-identity metadata of EVERY deployed instance —
/// <c>(name, deployment id, revision hash, is_enabled)</c> — without reading the
/// <c>config_json</c> column.
/// </summary>
/// <remarks>
/// Review 2026-10-21 01/SR-P1: the Deployment Manager's instance reconcile sweep runs every
/// <c>InstanceSweepIntervalSeconds</c> (default 60 s) and compares only name / enabled /
/// revision, but it used to read through <see cref="GetAllDeployedConfigsAsync"/> — on a
/// 500-instance site with ~70 KB configs that is ~35 MB of large-object-heap strings per tick
/// on a perfectly healthy site, forever. The sweep now reads this projection and fetches the
/// full row via <see cref="GetDeployedConfigAsync"/> only for the instances it actually starts
/// or restarts. The boot load still uses <see cref="GetAllDeployedConfigsAsync"/> — it runs
/// once and needs every blob.
/// </remarks>
/// <returns>A task that resolves to one metadata row per deployed instance.</returns>
public async Task<List<DeployedConfigMetadata>> GetAllDeployedConfigMetadataAsync()
{
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
SELECT instance_unique_name, deployment_id, revision_hash, is_enabled
FROM deployed_configurations";
var results = new List<DeployedConfigMetadata>();
await using var reader = await command.ExecuteReaderAsync();
while (await reader.ReadAsync())
{
results.Add(new DeployedConfigMetadata(
InstanceUniqueName: reader.GetString(0),
DeploymentId: reader.GetString(1),
RevisionHash: reader.GetString(2),
IsEnabled: reader.GetInt64(3) != 0));
}
return results;
}
/// <summary>
/// Stores or updates a deployed instance configuration. Uses UPSERT semantics.
/// </summary>
@@ -0,0 +1,136 @@
using System.Text.Json;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.TestSupport;
using ZB.MOM.WW.ScadaBridge.TestSupport;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;
/// <summary>
/// Review 2026-10-21 01/SR-P1: the instance reconcile sweep compared only name / enabled /
/// revision, yet materialised every row's <c>config_json</c> on every tick — tens of MB of LOH
/// strings per minute on a healthy large site. It now reads a metadata-only projection and
/// fetches a full row only for an instance it actually starts or restarts.
/// </summary>
public class InstanceSweepMetadataReadTests : TestKit, IDisposable
{
private readonly SiteStorageService _storage;
private readonly ScriptCompilationService _compilationService;
private readonly SharedScriptLibrary _sharedScriptLibrary;
private readonly TestLocalDb _localDb;
private int _fullTableReads;
private int _singleRowReads;
/// <summary>Creates a temporary LocalDb-backed site store per test.</summary>
public InstanceSweepMetadataReadTests()
{
_localDb = TestLocalDb.CreateTemp("sweep-metadata-read");
_storage = new SiteStorageService(_localDb.Db, NullLogger<SiteStorageService>.Instance);
_storage.InitializeAsync().GetAwaiter().GetResult();
_compilationService = new ScriptCompilationService(NullLogger<ScriptCompilationService>.Instance);
_sharedScriptLibrary = new SharedScriptLibrary(
_compilationService, NullLogger<SharedScriptLibrary>.Instance);
}
/// <inheritdoc />
void IDisposable.Dispose()
{
Shutdown();
var path = _localDb.Path;
_localDb.Dispose();
TestLocalDb.DeleteFiles(path);
}
private static string ConfigJson(string name) => JsonSerializer.Serialize(new FlattenedConfiguration
{
InstanceUniqueName = name,
Attributes = [new ResolvedAttribute { CanonicalName = "A", Value = "1", DataType = "Int32" }]
});
private async Task<List<DeployedInstance>> CountingFullLoad()
{
Interlocked.Increment(ref _fullTableReads);
return await _storage.GetAllDeployedConfigsAsync();
}
private async Task<DeployedInstance?> CountingSingleRowLoad(string name)
{
Interlocked.Increment(ref _singleRowReads);
return await _storage.GetDeployedConfigAsync(name);
}
private IActorRef CreateBootedDeploymentManager()
{
var health = new DeploymentManagerStartupProbe();
Func<Task<List<DeployedInstance>>> fullLoad = CountingFullLoad;
Func<string, Task<DeployedInstance?>> rowLoad = CountingSingleRowLoad;
var dm = ActorOf(Props.Create(() => new DeploymentManagerActor(
_storage,
_compilationService,
_sharedScriptLibrary,
null,
new SiteRuntimeOptions { InstanceSweepIntervalSeconds = 1, StartupBatchDelayMs = 10 },
NullLogger<DeploymentManagerActor>.Instance,
null,
health,
null,
null,
null,
null,
fullLoad,
null,
null,
null,
rowLoad,
null)));
AwaitCondition(() => health.ReportCount > 0, TimeSpan.FromSeconds(20), TimeSpan.FromMilliseconds(25));
return dm;
}
private async Task<bool> IsRunningAsync(IActorRef dm, string name)
{
var snapshot = await dm.Ask<DebugViewSnapshot>(
new DebugSnapshotRequest(name, Guid.NewGuid().ToString("N")), TimeSpan.FromSeconds(5));
return !snapshot.InstanceNotFound;
}
/// <summary>A healthy site's sweep never reads a config blob: one full read (boot), zero row reads.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task HealthySweep_ReadsNoConfigJson()
{
await _storage.StoreDeployedConfigAsync("P1", ConfigJson("P1"), "d1", "r1", true);
await _storage.StoreDeployedConfigAsync("P2", ConfigJson("P2"), "d2", "r2", true);
var dm = CreateBootedDeploymentManager();
await AwaitAssertAsync(async () => Assert.True(await IsRunningAsync(dm, "P2")), TimeSpan.FromSeconds(10));
await Task.Delay(TimeSpan.FromSeconds(4)); // several sweep ticks
Assert.Equal(1, Volatile.Read(ref _fullTableReads));
Assert.Equal(0, Volatile.Read(ref _singleRowReads));
}
/// <summary>
/// A repair still works: a row that appears after boot (as CDC replication would deliver it)
/// is started, its config fetched by a single-row read.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task SweepStart_FetchesOnlyTheRowItStarts()
{
var dm = CreateBootedDeploymentManager();
await _storage.StoreDeployedConfigAsync("Late1", ConfigJson("Late1"), "d1", "r1", true);
await AwaitAssertAsync(async () => Assert.True(await IsRunningAsync(dm, "Late1")), TimeSpan.FromSeconds(15));
Assert.Equal(1, Volatile.Read(ref _fullTableReads));
Assert.Equal(1, Volatile.Read(ref _singleRowReads));
}
}
@@ -532,22 +532,16 @@ public class InstanceSweepSelfHealTests : TestKit, IDisposable
/// done since makes it stale. Connection definitions are passed as <see langword="null"/>
/// (= "the definitions read failed, keep the overlay") so these tests exercise the CONFIG
/// half of the snapshot in isolation.
/// <para>
/// Metadata only (review 2026-10-21 01/SR-P1): the sweep fetches the config body of an
/// instance it repairs from the store, so a test expecting a repair must ALSO have the row
/// (at the same revision) in storage.
/// </para>
/// </summary>
private static DeploymentManagerActor.InstanceSweepLoaded SweepSnapshot(
string instanceName, string revision, string configJson, long generationAtRead,
bool isEnabled = true) =>
string instanceName, string revision, long generationAtRead, bool isEnabled = true) =>
new(
[
new DeployedInstance
{
InstanceUniqueName = instanceName,
ConfigJson = configJson,
DeploymentId = $"dep-{revision}",
RevisionHash = revision,
IsEnabled = isEnabled,
DeployedAt = DateTimeOffset.UtcNow.ToString("O")
}
],
[new DeployedConfigMetadata(instanceName, $"dep-{revision}", revision, isEnabled)],
null,
ConnectionDefinitions: null,
StoreGenerationAtRead: generationAtRead);
@@ -585,9 +579,7 @@ public class InstanceSweepSelfHealTests : TestKit, IDisposable
var afterDeploys = await ResolveInstanceChildAsync(dm, instanceName);
// Generation 0 = captured before either deploy mutated the store.
dm.Tell(SweepSnapshot(
instanceName, "rev-1", MakeConfigJson(instanceName, attributeValue: "111"),
generationAtRead: 0));
dm.Tell(SweepSnapshot(instanceName, "rev-1", generationAtRead: 0));
await Task.Delay(SeveralSweeps);
@@ -625,9 +617,13 @@ public class InstanceSweepSelfHealTests : TestKit, IDisposable
await AwaitAssertAsync(async () =>
Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));
dm.Tell(SweepSnapshot(
instanceName, "rev-2", MakeConfigJson(instanceName, attributeValue: "222"),
generationAtRead: 0));
// The store is refreshed out of band (as replication would), so the sweep's single-row
// fetch finds the rev-2 body the metadata snapshot names. A direct storage write does not
// pass through the actor, so the generation stays 0.
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName, attributeValue: "222"),
"dep-rev-2", "rev-2", true);
dm.Tell(SweepSnapshot(instanceName, "rev-2", generationAtRead: 0));
await AwaitAssertAsync(async () =>
{
@@ -671,8 +667,7 @@ public class InstanceSweepSelfHealTests : TestKit, IDisposable
Assert.False(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(15));
// Generation 0 = captured before the deploy and the delete mutated the store.
dm.Tell(SweepSnapshot(
instanceName, "rev-1", MakeConfigJson(instanceName), generationAtRead: 0));
dm.Tell(SweepSnapshot(instanceName, "rev-1", generationAtRead: 0));
await Task.Delay(SeveralSweeps);
@@ -693,8 +688,9 @@ public class InstanceSweepSelfHealTests : TestKit, IDisposable
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);
dm.Tell(SweepSnapshot(
instanceName, "rev-1", MakeConfigJson(instanceName), generationAtRead: 0));
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName), "dep-rev-1", "rev-1", true);
dm.Tell(SweepSnapshot(instanceName, "rev-1", generationAtRead: 0));
await AwaitAssertAsync(async () =>
Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));
@@ -147,6 +147,29 @@ public class SiteStorageServiceTests : IAsyncLifetime, IDisposable
Assert.Null(result);
}
/// <summary>
/// Review 2026-10-21 01/SR-P1: the all-rows metadata projection the instance sweep reads
/// carries the same identity as the full read, for every row, without the config blob.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task GetAllDeployedConfigMetadataAsync_MatchesTheFullReadForEveryRow()
{
await _storage.StoreDeployedConfigAsync(
"Pump1", "{\"test\":true}", "dep-001", "sha256:abc", isEnabled: true);
await _storage.StoreDeployedConfigAsync(
"Pump2", "{\"other\":1}", "dep-002", "sha256:def", isEnabled: false);
var metadata = (await _storage.GetAllDeployedConfigMetadataAsync())
.OrderBy(m => m.InstanceUniqueName).ToList();
var full = (await _storage.GetAllDeployedConfigsAsync())
.OrderBy(c => c.InstanceUniqueName).ToList();
Assert.Equal(
full.Select(c => (c.InstanceUniqueName, c.DeploymentId, c.RevisionHash, c.IsEnabled)),
metadata.Select(m => (m.InstanceUniqueName, m.DeploymentId, m.RevisionHash, m.IsEnabled)));
}
/// <summary>Verifies that a metadata read returns the same identity as reading all configs and picking the match.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]