feat(health): oldest-parked-message age on the site health report (parked-retention aging signal)
This commit is contained in:
@@ -47,7 +47,14 @@ public record SiteHealthReport(
|
|||||||
// hosted service. Point-in-time (not reset on collect) — mirrors the
|
// hosted service. Point-in-time (not reset on collect) — mirrors the
|
||||||
// SiteAuditBacklog pattern. Defaults to 0 so existing producers / tests that
|
// SiteAuditBacklog pattern. Defaults to 0 so existing producers / tests that
|
||||||
// don't wire the poller stay valid.
|
// don't wire the poller stay valid.
|
||||||
long SiteEventLogWriteFailures = 0);
|
long SiteEventLogWriteFailures = 0,
|
||||||
|
// Age (seconds) of the oldest parked
|
||||||
|
// store-and-forward message, computed at report time on the site; null when no
|
||||||
|
// rows are parked. Parked rows persist until operator action, so this aging
|
||||||
|
// signal surfaces a forgotten site's stale backlog (count alone hides age).
|
||||||
|
// Additive-only trailing member — defaults to null for producers / tests that
|
||||||
|
// don't populate it.
|
||||||
|
double? OldestParkedMessageAgeSeconds = null);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Broadcast wrapper used between central nodes to keep per-node
|
/// Broadcast wrapper used between central nodes to keep per-node
|
||||||
|
|||||||
@@ -107,6 +107,15 @@ public class HealthReportSender : BackgroundService
|
|||||||
{
|
{
|
||||||
var parkedCount = await _sfStorage.GetParkedMessageCountAsync();
|
var parkedCount = await _sfStorage.GetParkedMessageCountAsync();
|
||||||
_collector.SetParkedMessageCount(parkedCount);
|
_collector.SetParkedMessageCount(parkedCount);
|
||||||
|
|
||||||
|
// Parked-retention aging signal: age of the oldest parked row
|
||||||
|
// (null when none are parked), so a forgotten site's stale
|
||||||
|
// backlog is visible by age, not just count.
|
||||||
|
var oldestParked = await _sfStorage.GetOldestParkedCreatedAtAsync();
|
||||||
|
_collector.SetOldestParkedMessageAgeSeconds(
|
||||||
|
oldestParked.HasValue
|
||||||
|
? (DateTimeOffset.UtcNow - oldestParked.Value).TotalSeconds
|
||||||
|
: null);
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -111,6 +111,19 @@ public interface ISiteHealthCollector
|
|||||||
/// <param name="count">The number of parked messages.</param>
|
/// <param name="count">The number of parked messages.</param>
|
||||||
void SetParkedMessageCount(int count);
|
void SetParkedMessageCount(int count);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Sets the age (seconds) of the oldest parked store-and-forward message for the
|
||||||
|
/// next <see cref="CollectReport"/> call; <c>null</c> when no rows are parked.
|
||||||
|
/// Point-in-time (not reset on collect), refreshed by the parked-count poller in
|
||||||
|
/// <c>HealthReportSender</c>. Default interface implementation is a no-op so
|
||||||
|
/// existing test fakes continue to compile without per-fake updates.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="ageSeconds">Age in seconds of the oldest parked row, or <c>null</c> if none are parked.</param>
|
||||||
|
void SetOldestParkedMessageAgeSeconds(double? ageSeconds)
|
||||||
|
{
|
||||||
|
// Default no-op so test fakes do not need to be updated.
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Replace the latest cumulative
|
/// Replace the latest cumulative
|
||||||
/// site-event-log write-failure count (SQLite error, disk full,
|
/// site-event-log write-failure count (SQLite error, disk full,
|
||||||
|
|||||||
@@ -25,6 +25,10 @@ public class SiteHealthCollector : ISiteHealthCollector
|
|||||||
private IReadOnlyDictionary<string, int> _sfBufferDepths = new Dictionary<string, int>();
|
private IReadOnlyDictionary<string, int> _sfBufferDepths = new Dictionary<string, int>();
|
||||||
private int _deployedInstanceCount, _enabledInstanceCount, _disabledInstanceCount;
|
private int _deployedInstanceCount, _enabledInstanceCount, _disabledInstanceCount;
|
||||||
private int _parkedMessageCount;
|
private int _parkedMessageCount;
|
||||||
|
// Age (seconds) of the oldest parked message, stored as raw double bits so it can
|
||||||
|
// be read/written atomically via Interlocked (double? is neither atomic nor
|
||||||
|
// volatile-able). NaN is the "no parked rows" sentinel (ages are always >= 0).
|
||||||
|
private long _oldestParkedAgeBits = BitConverter.DoubleToInt64Bits(double.NaN);
|
||||||
private volatile string _nodeHostname = "";
|
private volatile string _nodeHostname = "";
|
||||||
private volatile IReadOnlyList<Commons.Messages.Health.NodeStatus>? _clusterNodes;
|
private volatile IReadOnlyList<Commons.Messages.Health.NodeStatus>? _clusterNodes;
|
||||||
private volatile bool _isActiveNode;
|
private volatile bool _isActiveNode;
|
||||||
@@ -123,6 +127,20 @@ public class SiteHealthCollector : ISiteHealthCollector
|
|||||||
Interlocked.Exchange(ref _parkedMessageCount, count);
|
Interlocked.Exchange(ref _parkedMessageCount, count);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void SetOldestParkedMessageAgeSeconds(double? ageSeconds)
|
||||||
|
{
|
||||||
|
Interlocked.Exchange(ref _oldestParkedAgeBits,
|
||||||
|
BitConverter.DoubleToInt64Bits(ageSeconds ?? double.NaN));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Reads the atomically-stored oldest-parked age, mapping the NaN sentinel back to null.</summary>
|
||||||
|
private double? ReadOldestParkedAgeSeconds()
|
||||||
|
{
|
||||||
|
var value = BitConverter.Int64BitsToDouble(Interlocked.Read(ref _oldestParkedAgeBits));
|
||||||
|
return double.IsNaN(value) ? null : value;
|
||||||
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public void SetNodeHostname(string hostname) => _nodeHostname = hostname;
|
public void SetNodeHostname(string hostname) => _nodeHostname = hostname;
|
||||||
|
|
||||||
@@ -214,6 +232,7 @@ public class SiteHealthCollector : ISiteHealthCollector
|
|||||||
SiteAuditWriteFailures: siteAuditWriteFailures,
|
SiteAuditWriteFailures: siteAuditWriteFailures,
|
||||||
AuditRedactionFailure: auditRedactionFailures,
|
AuditRedactionFailure: auditRedactionFailures,
|
||||||
SiteAuditBacklog: _siteAuditBacklog,
|
SiteAuditBacklog: _siteAuditBacklog,
|
||||||
SiteEventLogWriteFailures: Interlocked.Read(ref _siteEventLogWriteFailures));
|
SiteEventLogWriteFailures: Interlocked.Read(ref _siteEventLogWriteFailures),
|
||||||
|
OldestParkedMessageAgeSeconds: ReadOldestParkedAgeSeconds());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -663,6 +663,26 @@ public class StoreAndForwardStorage
|
|||||||
return Convert.ToInt32(result);
|
return Convert.ToInt32(result);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets the <c>created_at</c> of the oldest parked message, or <c>null</c> when
|
||||||
|
/// no rows are parked. Backs the parked-retention aging signal on the site health
|
||||||
|
/// report: parked rows persist until an operator acts, so their <i>age</i> — not
|
||||||
|
/// just their count — is what tells an operator a forgotten site's backlog is
|
||||||
|
/// growing stale.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that resolves to the oldest parked row's creation time, or <c>null</c> if none are parked.</returns>
|
||||||
|
public async Task<DateTimeOffset?> GetOldestParkedCreatedAtAsync()
|
||||||
|
{
|
||||||
|
await using var conn = await OpenConnectionAsync();
|
||||||
|
await using var cmd = conn.CreateCommand();
|
||||||
|
cmd.CommandText = "SELECT MIN(created_at) FROM sf_messages WHERE status = @parked";
|
||||||
|
cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
|
||||||
|
var result = await cmd.ExecuteScalarAsync();
|
||||||
|
return result is null or DBNull
|
||||||
|
? null
|
||||||
|
: DateTimeOffset.Parse((string)result);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Gets total message count by status.
|
/// Gets total message count by status.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|||||||
@@ -790,4 +790,22 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable
|
|||||||
Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-1"));
|
Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-1"));
|
||||||
Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-2"));
|
Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-2"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── Task 23: oldest-parked-age health signal ──
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task GetOldestParkedCreatedAt_ReturnsOldestParkedRow_OrNull()
|
||||||
|
{
|
||||||
|
Assert.Null(await _storage.GetOldestParkedCreatedAtAsync());
|
||||||
|
|
||||||
|
var old = NewMsg("old-parked", StoreAndForwardMessageStatus.Parked, DateTimeOffset.UtcNow.AddDays(-3));
|
||||||
|
var recent = NewMsg("new-parked", StoreAndForwardMessageStatus.Parked, DateTimeOffset.UtcNow.AddHours(-1));
|
||||||
|
var pending = NewMsg("pending", StoreAndForwardMessageStatus.Pending, DateTimeOffset.UtcNow.AddDays(-9)); // must not count
|
||||||
|
await _storage.EnqueueAsync(old);
|
||||||
|
await _storage.EnqueueAsync(recent);
|
||||||
|
await _storage.EnqueueAsync(pending);
|
||||||
|
|
||||||
|
var oldest = await _storage.GetOldestParkedCreatedAtAsync();
|
||||||
|
Assert.Equal(old.CreatedAt, oldest!.Value, TimeSpan.FromSeconds(1));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user