feat(audit): Parquet cold archive of expired AuditLog partitions before the purge (ScadaBridge#42)
Audit roadmap T2 (alog.md §15.2): with AuditLog:Archive:Enabled (default off)
the purge actor writes each expired monthly partition to Parquet under an
absolute AuditLog:Archive:Directory before switching it out, so audit rows
outlive RetentionDays.
- Parquet.Net 6.1.0 (fully managed; NuGetAudit clean) in central package
management.
- AuditArchiveParquet: data file = the ten stored columns, OccurredAtUtc as a
UTC microsecond timestamp plus exact OccurredAtUtcTicks, and ChainSeq/RowHash;
tombstone file = the month's per-channel purge tombstones. Buffered row
groups; the files are plain Parquet (DuckDB/pyarrow read them).
- AuditLogPartitionArchive (IAuditLogArchive): reads the partition's rows,
chain heads and tombstones in one transaction under the chain lock SHARED,
writes {Directory}/{YYYY}/auditlog-{YYYY-MM}-{stamp}Z.parquet,
.tombstones.parquet and a .manifest.json (counts, SHA-256 per file, each
month's head and prior anchor) via tmp + rename, manifest last. Archives are
never overwritten.
- Archive before purge: SwitchOutPartitionAsync gains an internal guarded
overload. Under the EXCLUSIVE chain lock it re-counts the partition and
re-reads each month's LastSeq and THROWs 51002/51003 (nothing purged) unless
both still equal the archive's, so the switch's checkpoint anchor is exactly
the archive's chain end. The manifest is then marked Purged.
- Fail closed: a partition that cannot be archived is not purged; the failure
takes the existing AuditLogPurgeFailedEvent / PurgeFailures path. Success
publishes AuditLogArchivedEvent.
- Verify: file hashes and counts against the manifest, the chain recomputed
from the archived rows and tombstones (AuditChainVerifier), and the chain end
against the database anchor (Matches / Superseded / NotPurged / Mismatch).
- AuditLogArchiveOptionsValidator: Directory required when enabled and must be
absolute; RowGroupSize, timeouts bounded.
No migration: the purger role already reads AuditLog, the head and tombstones.
Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
@@ -78,6 +78,9 @@
|
||||
<PackageVersion Include="NSubstitute" Version="5.3.0" />
|
||||
<PackageVersion Include="OPCFoundation.NetStandard.Opc.Ua.Client" Version="1.5.378.106" />
|
||||
<PackageVersion Include="OpenTelemetry.Api" Version="1.15.3" />
|
||||
<!-- Parquet audit archival + the parquet format of `audit export` (ScadaBridge#42). 6.1.0 is fully
|
||||
managed (Snappier / ZstdSharp / K4os LZ4, no native IronCompress) and NuGetAudit-clean. -->
|
||||
<PackageVersion Include="Parquet.Net" Version="6.1.0" />
|
||||
<PackageVersion Include="Serilog" Version="4.3.1" />
|
||||
<PackageVersion Include="Serilog.AspNetCore" Version="10.0.0" />
|
||||
<PackageVersion Include="Serilog.Sinks.Console" Version="6.1.1" />
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
using ZB.MOM.WW.Configuration;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
|
||||
/// <summary>
|
||||
/// Validates <see cref="AuditLogArchiveOptions"/> (section <c>AuditLog:Archive</c>; Parquet
|
||||
/// cold archive, ScadaBridge#42) at host startup. An enabled archive must name an absolute
|
||||
/// directory: a relative one would resolve against the service's working directory, which
|
||||
/// differs between the two central nodes and moves on every redeploy.
|
||||
/// </summary>
|
||||
public sealed class AuditLogArchiveOptionsValidator : OptionsValidatorBase<AuditLogArchiveOptions>
|
||||
{
|
||||
/// <inheritdoc />
|
||||
protected override void Validate(ValidationBuilder builder, AuditLogArchiveOptions options)
|
||||
{
|
||||
if (options.Enabled)
|
||||
{
|
||||
builder.RequireThat(!string.IsNullOrWhiteSpace(options.Directory),
|
||||
"AuditLog:Archive:Directory is required when AuditLog:Archive:Enabled is true.");
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(options.Directory))
|
||||
{
|
||||
builder.RequireThat(Path.IsPathFullyQualified(options.Directory),
|
||||
$"AuditLog:Archive:Directory ('{options.Directory}') must be an absolute path reachable by both central nodes.");
|
||||
}
|
||||
|
||||
builder.RequireThat(options.RowGroupSize is >= 1_000 and <= 1_000_000,
|
||||
$"AuditLog:Archive:{nameof(AuditLogArchiveOptions.RowGroupSize)} ({options.RowGroupSize}) must be between 1000 and 1000000.");
|
||||
builder.RequireThat(options.ReadCommandTimeoutMinutes > 0,
|
||||
$"AuditLog:Archive:{nameof(AuditLogArchiveOptions.ReadCommandTimeoutMinutes)} ({options.ReadCommandTimeoutMinutes}) must be > 0.");
|
||||
builder.RequireThat(options.ChainLockTimeoutSeconds > 0,
|
||||
$"AuditLog:Archive:{nameof(AuditLogArchiveOptions.ChainLockTimeoutSeconds)} ({options.ChainLockTimeoutSeconds}) must be > 0.");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
|
||||
/// <summary>
|
||||
/// Published on the actor-system EventStream by <see cref="AuditLogPurgeActor"/> when an
|
||||
/// expired partition was written to a Parquet archive and then switched out
|
||||
/// (ScadaBridge#42). Followed by the usual <see cref="AuditLogPurgedEvent"/>.
|
||||
/// </summary>
|
||||
/// <param name="MonthBoundary">The partition's lower boundary (UTC).</param>
|
||||
/// <param name="ArchiveName">The archive name (<c>audit archive verify --name</c> takes it).</param>
|
||||
/// <param name="RowCount">Rows written to the archive.</param>
|
||||
/// <param name="DataBytes">Size of the archive's data file in bytes.</param>
|
||||
public sealed record AuditLogArchivedEvent(
|
||||
DateTime MonthBoundary,
|
||||
string ArchiveName,
|
||||
long RowCount,
|
||||
long DataBytes);
|
||||
@@ -6,6 +6,7 @@ using Microsoft.Extensions.Options;
|
||||
using ZB.MOM.WW.ScadaBridge.AuditLog.Configuration;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
|
||||
@@ -20,6 +21,9 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
/// <see cref="IAuditLogRepository.SwitchOutPartitionAsync"/>, a
|
||||
/// metadata-only staging-table switch (no index
|
||||
/// drop/rebuild brackets it).</item>
|
||||
/// <item>When <c>AuditLog:Archive:Enabled</c> (ScadaBridge#42, default off), each boundary
|
||||
/// is first written to a Parquet archive via <see cref="IAuditLogArchive"/> and then
|
||||
/// switched out by the archive's guarded purge instead of the plain switch.</item>
|
||||
/// <item>Publishes <see cref="AuditLogPurgedEvent"/> on the actor-system
|
||||
/// EventStream so the central health collector + ops surfaces
|
||||
/// can subscribe without coupling to this actor.</item>
|
||||
@@ -188,6 +192,27 @@ public class AuditLogPurgeActor : ReceiveActor
|
||||
return;
|
||||
}
|
||||
|
||||
// Parquet cold archive (ScadaBridge#42, default off). When enabled, every expired
|
||||
// partition is archived first and switched out only by the archive's guarded purge,
|
||||
// so a partition that cannot be archived is never dropped (fail closed: the table
|
||||
// grows and the failure surfaces on the health counter instead of rows being lost).
|
||||
var archiveOptions = scope.ServiceProvider.GetService<IOptions<AuditLogArchiveOptions>>()?.Value;
|
||||
IAuditLogArchive? archive = null;
|
||||
if (archiveOptions?.Enabled == true)
|
||||
{
|
||||
try
|
||||
{
|
||||
archive = scope.ServiceProvider.GetRequiredService<IAuditLogArchive>();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "AuditLog archival is enabled but IAuditLogArchive could not be resolved; skipping partition purge.");
|
||||
PublishPurgeFailure(eventStream, purgeFailureCounter, DateTime.MinValue, ex, 0);
|
||||
await RunPerChannelOverridesAsync(repository, eventStream, purgeFailureCounter).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
foreach (var boundary in boundaries)
|
||||
{
|
||||
// Per-boundary try/catch: one bad partition (transient SQL
|
||||
@@ -196,9 +221,22 @@ public class AuditLogPurgeActor : ReceiveActor
|
||||
var sw = Stopwatch.StartNew();
|
||||
try
|
||||
{
|
||||
var rowsDeleted = await repository
|
||||
.SwitchOutPartitionAsync(boundary, _purgeOptions.ResolvedMaintenanceCommandTimeout)
|
||||
.ConfigureAwait(false);
|
||||
long rowsDeleted;
|
||||
if (archive is null)
|
||||
{
|
||||
rowsDeleted = await repository
|
||||
.SwitchOutPartitionAsync(boundary, _purgeOptions.ResolvedMaintenanceCommandTimeout)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
var manifest = await archive.ArchivePartitionAsync(boundary).ConfigureAwait(false);
|
||||
rowsDeleted = await archive
|
||||
.PurgeArchivedPartitionAsync(manifest, _purgeOptions.ResolvedMaintenanceCommandTimeout)
|
||||
.ConfigureAwait(false);
|
||||
eventStream.Publish(new AuditLogArchivedEvent(boundary, manifest.Name, manifest.RowCount, manifest.DataBytes));
|
||||
}
|
||||
|
||||
sw.Stop();
|
||||
|
||||
eventStream.Publish(
|
||||
|
||||
@@ -55,6 +55,12 @@ public static class ServiceCollectionExtensions
|
||||
/// <summary>Configuration section bound to <see cref="AuditLogHashChainOptions"/> (ScadaBridge#41).</summary>
|
||||
public const string HashChainSectionName = "AuditLog:HashChain";
|
||||
|
||||
/// <summary>
|
||||
/// Configuration section bound to
|
||||
/// <see cref="ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance.AuditLogArchiveOptions"/> (ScadaBridge#42).
|
||||
/// </summary>
|
||||
public const string ArchiveSectionName = "AuditLog:Archive";
|
||||
|
||||
/// <summary>Configuration section bound to <see cref="ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeOptions"/>.</summary>
|
||||
public const string PurgeSectionName = "AuditLog:Purge";
|
||||
|
||||
@@ -405,6 +411,12 @@ public static class ServiceCollectionExtensions
|
||||
config, HashChainSectionName);
|
||||
services.AddHostedService<AuditLogHashChainService>();
|
||||
|
||||
// Parquet cold archive (ScadaBridge#42). Default OFF; when enabled the purge actor
|
||||
// archives each expired partition before switching it out. IAuditLogArchive is
|
||||
// registered by AddConfigurationDatabase (central only) and reads these options.
|
||||
services.AddValidatedOptions<ConfigurationDatabase.Maintenance.AuditLogArchiveOptions, AuditLogArchiveOptionsValidator>(
|
||||
config, ArchiveSectionName);
|
||||
|
||||
// I1 (review): bind the two central-singleton options HERE rather than in
|
||||
// AddAuditLogCentralReconciliationClient. AkkaHostedService.RegisterCentralActors
|
||||
// resolves IOptions<AuditLogPurgeOptions> / <SiteAuditReconciliationOptions>
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
|
||||
/// <summary>
|
||||
/// Parquet cold archive of the central <c>AuditLog</c> (audit roadmap T2, alog.md §15.2,
|
||||
/// ScadaBridge#42): a monthly partition is written to Parquet before the retention purge
|
||||
/// switches it out, so audit rows outlive the SQL retention window.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Archive before purge.</b> <see cref="ArchivePartitionAsync"/> writes the partition's
|
||||
/// rows (with their hash-chain columns), the month's chain head and its per-channel purge
|
||||
/// tombstones; <see cref="PurgeArchivedPartitionAsync"/> then switches the partition out
|
||||
/// inside a guard that refuses to drop it if the partition gained or lost a row, or its
|
||||
/// chain advanced, since the archive was written. A partition is therefore never purged
|
||||
/// without an archive that holds exactly the rows the purge drops.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Hash chain.</b> The switch records each month's chain end as the checkpoint anchor
|
||||
/// (ScadaBridge#41); the archive records the same chain end, so <see cref="VerifyAsync"/>
|
||||
/// can recompute the archived chain and tie its end to the anchor still in the database.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public interface IAuditLogArchive
|
||||
{
|
||||
/// <summary>True when an archive directory is configured (listing and verifying need one; archiving needs <c>Enabled</c> too).</summary>
|
||||
bool IsConfigured { get; }
|
||||
|
||||
/// <summary>True when the retention purge archives partitions before dropping them (<c>AuditLog:Archive:Enabled</c>).</summary>
|
||||
bool IsEnabled { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Writes the partition whose lower boundary is <paramref name="partitionBoundaryUtc"/> to
|
||||
/// a new archive (data file, tombstone file, then manifest with status
|
||||
/// <see cref="AuditArchiveStatus.Archived"/>).
|
||||
/// </summary>
|
||||
/// <param name="partitionBoundaryUtc">The <c>pf_AuditLog_Month</c> lower boundary (day 1, midnight UTC).</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The written manifest.</returns>
|
||||
Task<AuditArchiveManifest> ArchivePartitionAsync(DateTime partitionBoundaryUtc, CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Switches the archived partition out of <c>dbo.AuditLog</c>, guarded against the
|
||||
/// partition having changed since <paramref name="manifest"/> was written, then marks the
|
||||
/// manifest <see cref="AuditArchiveStatus.Purged"/>.
|
||||
/// </summary>
|
||||
/// <param name="manifest">The manifest <see cref="ArchivePartitionAsync"/> returned.</param>
|
||||
/// <param name="commandTimeout">Maintenance command timeout for the switch.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The approximate number of rows the switch discarded.</returns>
|
||||
/// <exception cref="AuditArchiveStaleException">The partition changed after it was archived; nothing was purged.</exception>
|
||||
Task<long> PurgeArchivedPartitionAsync(AuditArchiveManifest manifest, TimeSpan? commandTimeout = null, CancellationToken ct = default);
|
||||
|
||||
/// <summary>Lists every archive manifest under the archive directory, oldest partition first.</summary>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The manifests.</returns>
|
||||
Task<IReadOnlyList<AuditArchiveManifest>> ListAsync(CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Verifies one archive: file integrity against the manifest, the hash chain recomputed
|
||||
/// over the archived rows, and the chain end against the database's checkpoint anchor.
|
||||
/// </summary>
|
||||
/// <param name="name">The archive name (as <see cref="ListAsync"/> reports it).</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The verification result.</returns>
|
||||
/// <exception cref="FileNotFoundException">No archive has that name.</exception>
|
||||
Task<AuditArchiveVerificationResult> VerifyAsync(string name, CancellationToken ct = default);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Thrown by <see cref="IAuditLogArchive.PurgeArchivedPartitionAsync"/> when the partition
|
||||
/// gained or lost rows, or its hash chain advanced, after the archive was written. Nothing
|
||||
/// is purged; the next purge tick writes a fresh archive.
|
||||
/// </summary>
|
||||
public sealed class AuditArchiveStaleException : Exception
|
||||
{
|
||||
/// <summary>Initializes a new instance with a message and the SQL error that carried it.</summary>
|
||||
/// <param name="message">The message.</param>
|
||||
/// <param name="inner">The underlying exception.</param>
|
||||
public AuditArchiveStaleException(string message, Exception? inner = null) : base(message, inner) { }
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
using System.Text.Json.Serialization;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
|
||||
/// <summary>
|
||||
/// Lifecycle of one AuditLog Parquet archive (audit roadmap T2, ScadaBridge#42).
|
||||
/// </summary>
|
||||
[JsonConverter(typeof(JsonStringEnumConverter<AuditArchiveStatus>))]
|
||||
public enum AuditArchiveStatus
|
||||
{
|
||||
/// <summary>
|
||||
/// The archive files are durable but the partition switch that should follow has not
|
||||
/// been confirmed: either it failed (the archive went stale and a later tick wrote a
|
||||
/// fresh one) or the node stopped between the switch and the manifest update. Verify
|
||||
/// tells the two apart: only the second one's chain end matches the database anchor.
|
||||
/// </summary>
|
||||
Archived,
|
||||
|
||||
/// <summary>The archived partition was switched out of <c>dbo.AuditLog</c>; the archive is the only copy.</summary>
|
||||
Purged,
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One chain partition (UTC calendar month) as it stood when its rows were archived: the
|
||||
/// chain head and the checkpoint anchor an earlier partition purge had left, so the archived
|
||||
/// rows can be re-verified with <c>AuditChainVerifier</c> long after the database dropped them.
|
||||
/// </summary>
|
||||
/// <param name="Partition">The chain partition, <c>YYYY-MM</c>.</param>
|
||||
/// <param name="LastSeq">The head's last sequence number at archive time.</param>
|
||||
/// <param name="LastHash">The head's last hash at archive time (lowercase hex).</param>
|
||||
/// <param name="PriorPurgedThroughSeq">The checkpoint anchor an earlier purge of this month left, or null.</param>
|
||||
/// <param name="PriorPurgedThroughHash">The hash at <paramref name="PriorPurgedThroughSeq"/> (lowercase hex), or null.</param>
|
||||
/// <param name="TombstoneCount">Per-channel purge tombstones of this month carried in the archive.</param>
|
||||
public sealed record AuditArchiveChainState(
|
||||
string Partition,
|
||||
long LastSeq,
|
||||
string LastHash,
|
||||
long? PriorPurgedThroughSeq,
|
||||
string? PriorPurgedThroughHash,
|
||||
long TombstoneCount);
|
||||
|
||||
/// <summary>
|
||||
/// The JSON manifest written beside every AuditLog Parquet archive (ScadaBridge#42). It is
|
||||
/// written LAST, so its presence is the commit marker for the archive's two Parquet files.
|
||||
/// </summary>
|
||||
/// <param name="FormatVersion">Manifest/file layout version (currently 1).</param>
|
||||
/// <param name="Name">Archive name, <c>auditlog-YYYY-MM-yyyyMMddTHHmmssfffZ</c>; the file stem of all three files.</param>
|
||||
/// <param name="PartitionBoundaryUtc">The <c>pf_AuditLog_Month</c> lower boundary of the archived partition.</param>
|
||||
/// <param name="Partition">The archived partition, <c>YYYY-MM</c>.</param>
|
||||
/// <param name="ArchivedAtUtc">When the archive was written.</param>
|
||||
/// <param name="Status">Whether the partition switch that follows the archive has been confirmed.</param>
|
||||
/// <param name="PurgedAtUtc">When the partition switch completed, or null.</param>
|
||||
/// <param name="RowCount">Rows in the data file.</param>
|
||||
/// <param name="ChainedRowCount">Rows that carried a ChainSeq/RowHash.</param>
|
||||
/// <param name="UnchainedRowCount">Rows the chaining job had not reached (not covered by the chain).</param>
|
||||
/// <param name="MinOccurredAtUtc">Earliest OccurredAtUtc in the data file, or null when empty.</param>
|
||||
/// <param name="MaxOccurredAtUtc">Latest OccurredAtUtc in the data file, or null when empty.</param>
|
||||
/// <param name="DataFile">Data file name, relative to the manifest's directory.</param>
|
||||
/// <param name="DataSha256">SHA-256 of the data file (lowercase hex).</param>
|
||||
/// <param name="DataBytes">Size of the data file in bytes.</param>
|
||||
/// <param name="TombstoneFile">Tombstone file name, relative to the manifest's directory.</param>
|
||||
/// <param name="TombstoneSha256">SHA-256 of the tombstone file (lowercase hex).</param>
|
||||
/// <param name="TombstoneCount">Tombstones in the tombstone file.</param>
|
||||
/// <param name="Chains">One entry per chain partition that had a head when the rows were archived.</param>
|
||||
public sealed record AuditArchiveManifest(
|
||||
int FormatVersion,
|
||||
string Name,
|
||||
DateTime PartitionBoundaryUtc,
|
||||
string Partition,
|
||||
DateTime ArchivedAtUtc,
|
||||
AuditArchiveStatus Status,
|
||||
DateTime? PurgedAtUtc,
|
||||
long RowCount,
|
||||
long ChainedRowCount,
|
||||
long UnchainedRowCount,
|
||||
DateTime? MinOccurredAtUtc,
|
||||
DateTime? MaxOccurredAtUtc,
|
||||
string DataFile,
|
||||
string DataSha256,
|
||||
long DataBytes,
|
||||
string TombstoneFile,
|
||||
string TombstoneSha256,
|
||||
long TombstoneCount,
|
||||
IReadOnlyList<AuditArchiveChainState> Chains);
|
||||
|
||||
/// <summary>
|
||||
/// How an archive's recorded chain end relates to the checkpoint anchor the partition purge
|
||||
/// left on the database's chain head.
|
||||
/// </summary>
|
||||
[JsonConverter(typeof(JsonStringEnumConverter<AuditArchiveAnchorStatus>))]
|
||||
public enum AuditArchiveAnchorStatus
|
||||
{
|
||||
/// <summary>The database anchor is exactly the archive's chain end (sequence and hash).</summary>
|
||||
Matches,
|
||||
|
||||
/// <summary>
|
||||
/// The database anchor is past the archive's chain end: rows that arrived late for this
|
||||
/// month were chained and purged (and archived) again afterwards.
|
||||
/// </summary>
|
||||
Superseded,
|
||||
|
||||
/// <summary>The partition has not been purged: the archive's rows are still in <c>dbo.AuditLog</c>.</summary>
|
||||
NotPurged,
|
||||
|
||||
/// <summary>
|
||||
/// The anchor disagrees with the archive (an earlier sequence number, a different hash, or
|
||||
/// no chain head at all): the archive or the head was altered.
|
||||
/// </summary>
|
||||
Mismatch,
|
||||
}
|
||||
|
||||
/// <summary>Verification of one chain partition inside an archive.</summary>
|
||||
/// <param name="Chain">The chain recomputed over the archived rows, the archived head and tombstones.</param>
|
||||
/// <param name="Anchor">How the archived chain end relates to the database's checkpoint anchor.</param>
|
||||
/// <param name="AnchorDetail">Human-readable explanation of <paramref name="Anchor"/>.</param>
|
||||
public sealed record AuditArchiveChainVerification(
|
||||
AuditChainVerificationResult Chain,
|
||||
AuditArchiveAnchorStatus Anchor,
|
||||
string AnchorDetail);
|
||||
|
||||
/// <summary>
|
||||
/// Result of <c>audit archive verify</c> (ScadaBridge#42): the archive files match their
|
||||
/// manifest, every archived chain recomputes to the archived head, and each chain end is the
|
||||
/// anchor the partition purge left in the database.
|
||||
/// </summary>
|
||||
/// <param name="Name">The archive verified.</param>
|
||||
/// <param name="Partition">The archived partition, <c>YYYY-MM</c>.</param>
|
||||
/// <param name="ManifestStatus">The manifest's lifecycle status.</param>
|
||||
/// <param name="Verified">True when the files are intact, no chain is broken and no anchor mismatches.</param>
|
||||
/// <param name="FilesIntact">True when both files' size and SHA-256 and the row/tombstone counts match the manifest.</param>
|
||||
/// <param name="FileProblem">Why <paramref name="FilesIntact"/> is false, or null.</param>
|
||||
/// <param name="RowsRead">Rows read from the data file.</param>
|
||||
/// <param name="Chains">Per chain partition verification.</param>
|
||||
public sealed record AuditArchiveVerificationResult(
|
||||
string Name,
|
||||
string Partition,
|
||||
AuditArchiveStatus ManifestStatus,
|
||||
bool Verified,
|
||||
bool FilesIntact,
|
||||
string? FileProblem,
|
||||
long RowsRead,
|
||||
IReadOnlyList<AuditArchiveChainVerification> Chains);
|
||||
|
||||
/// <summary>Result of <c>audit archive list</c> (ScadaBridge#42).</summary>
|
||||
/// <param name="Enabled">Whether the retention purge is archiving partitions (<c>AuditLog:Archive:Enabled</c>).</param>
|
||||
/// <param name="Archives">Every archive manifest, oldest partition first.</param>
|
||||
public sealed record AuditArchiveListResult(bool Enabled, IReadOnlyList<AuditArchiveManifest> Archives);
|
||||
@@ -0,0 +1,428 @@
|
||||
using System.Globalization;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Security.Cryptography;
|
||||
using Parquet;
|
||||
using Parquet.Schema;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
/// <summary>
|
||||
/// One per-channel purge tombstone carried in an archive (ScadaBridge#41/#42): the link a
|
||||
/// sanctioned per-channel retention DELETE removed from the middle of a month's chain.
|
||||
/// </summary>
|
||||
/// <param name="ChainMonth">The chain partition's first instant (UTC).</param>
|
||||
/// <param name="ChainSeq">The purged row's sequence number.</param>
|
||||
/// <param name="RowHash">The purged row's 32-byte hash.</param>
|
||||
/// <param name="EventId">The purged row's EventId.</param>
|
||||
/// <param name="PurgedAtUtc">When the per-channel purge removed it.</param>
|
||||
public sealed record AuditArchiveTombstone(
|
||||
DateTime ChainMonth,
|
||||
long ChainSeq,
|
||||
byte[] RowHash,
|
||||
Guid EventId,
|
||||
DateTime PurgedAtUtc);
|
||||
|
||||
/// <summary>
|
||||
/// The Parquet layout of an AuditLog archive (audit roadmap T2, ScadaBridge#42), format v1.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Data file</b> — one row per archived <c>dbo.AuditLog</c> row, the ten stored (writable)
|
||||
/// columns plus the two hash-chain columns, so a row can be re-hashed with
|
||||
/// <see cref="AuditChainHasher"/> exactly as the database chained it:
|
||||
/// <c>EventId</c> (string, <c>D</c> format), <c>OccurredAtUtc</c> (timestamp, UTC, microseconds —
|
||||
/// for analytics), <c>OccurredAtUtcTicks</c> (int64 — the exact <c>datetime2(7)</c> value the hash
|
||||
/// covers), <c>Actor</c>, <c>Action</c>, <c>Outcome</c>, <c>Category</c>, <c>Target</c>,
|
||||
/// <c>SourceNode</c>, <c>CorrelationId</c>, <c>DetailsJson</c> (strings), <c>ChainSeq</c> (int64,
|
||||
/// null when unchained) and <c>RowHash</c> (binary, null when unchained). The persisted computed
|
||||
/// columns (Kind, Status, SourceSiteId, …) are not stored: they are derived from
|
||||
/// <c>DetailsJson</c>, which the hash covers.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Tombstone file</b> — <c>ChainMonth</c> (<c>YYYY-MM</c>), <c>ChainSeq</c>, <c>RowHash</c>,
|
||||
/// <c>EventId</c>, <c>PurgedAtUtc</c>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Both are plain Parquet: DuckDB (<c>SELECT * FROM 'auditlog-….parquet'</c>), pandas/pyarrow
|
||||
/// and Spark read them without any ScadaBridge code.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class AuditArchiveParquet
|
||||
{
|
||||
/// <summary>The archive layout version written into the file metadata and the manifest.</summary>
|
||||
public const int FormatVersion = 1;
|
||||
|
||||
/// <summary>Key-value metadata key carrying <see cref="FormatVersion"/>.</summary>
|
||||
public const string FormatVersionKey = "scadabridge.auditlog.archive.formatVersion";
|
||||
|
||||
/// <summary>Key-value metadata key carrying the archived partition (<c>YYYY-MM</c>).</summary>
|
||||
public const string PartitionKey = "scadabridge.auditlog.archive.partition";
|
||||
|
||||
private static readonly DataField<string> EventIdField = new("EventId", false);
|
||||
private static readonly DateTimeDataField OccurredAtField =
|
||||
new("OccurredAtUtc", DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: false);
|
||||
private static readonly DataField<long> OccurredAtTicksField = new("OccurredAtUtcTicks", false);
|
||||
private static readonly DataField<string> ActorField = new("Actor", true);
|
||||
private static readonly DataField<string> ActionField = new("Action", false);
|
||||
private static readonly DataField<string> OutcomeField = new("Outcome", false);
|
||||
private static readonly DataField<string> CategoryField = new("Category", false);
|
||||
private static readonly DataField<string> TargetField = new("Target", true);
|
||||
private static readonly DataField<string> SourceNodeField = new("SourceNode", true);
|
||||
private static readonly DataField<string> CorrelationIdField = new("CorrelationId", true);
|
||||
private static readonly DataField<string> DetailsJsonField = new("DetailsJson", true);
|
||||
private static readonly DataField<long> ChainSeqField = new("ChainSeq", true);
|
||||
private static readonly DataField<byte[]> RowHashField = new("RowHash", true);
|
||||
|
||||
private static readonly DataField<string> TombstoneMonthField = new("ChainMonth", false);
|
||||
private static readonly DataField<long> TombstoneSeqField = new("ChainSeq", false);
|
||||
private static readonly DataField<byte[]> TombstoneHashField = new("RowHash", false);
|
||||
private static readonly DataField<string> TombstoneEventIdField = new("EventId", false);
|
||||
private static readonly DateTimeDataField TombstonePurgedAtField =
|
||||
new("PurgedAtUtc", DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: false);
|
||||
|
||||
/// <summary>The data-file schema.</summary>
|
||||
public static ParquetSchema DataSchema { get; } = new(
|
||||
EventIdField, OccurredAtField, OccurredAtTicksField, ActorField, ActionField, OutcomeField, CategoryField,
|
||||
TargetField, SourceNodeField, CorrelationIdField, DetailsJsonField, ChainSeqField, RowHashField);
|
||||
|
||||
/// <summary>The tombstone-file schema.</summary>
|
||||
public static ParquetSchema TombstoneSchema { get; } = new(
|
||||
TombstoneMonthField, TombstoneSeqField, TombstoneHashField, TombstoneEventIdField, TombstonePurgedAtField);
|
||||
|
||||
/// <summary>Opens a buffered writer of archive data rows over <paramref name="output"/>.</summary>
|
||||
/// <param name="output">Destination stream (need not be seekable).</param>
|
||||
/// <param name="partition">The archived partition (<c>YYYY-MM</c>), stamped into the file metadata.</param>
|
||||
/// <param name="rowGroupSize">Rows per Parquet row group.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The writer; dispose it to flush the last row group and the footer.</returns>
|
||||
public static async Task<DataWriter> CreateDataWriterAsync(Stream output, string partition, int rowGroupSize, CancellationToken ct = default)
|
||||
{
|
||||
var writer = await ParquetWriter.CreateAsync(DataSchema, output, cancellationToken: ct).ConfigureAwait(false);
|
||||
writer.CustomMetadata = Metadata(partition);
|
||||
return new DataWriter(writer, rowGroupSize);
|
||||
}
|
||||
|
||||
/// <summary>Opens a buffered writer of tombstones over <paramref name="output"/>.</summary>
|
||||
/// <param name="output">Destination stream (need not be seekable).</param>
|
||||
/// <param name="partition">The archived partition (<c>YYYY-MM</c>), stamped into the file metadata.</param>
|
||||
/// <param name="rowGroupSize">Rows per Parquet row group.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The writer; dispose it to flush the last row group and the footer.</returns>
|
||||
public static async Task<TombstoneWriter> CreateTombstoneWriterAsync(Stream output, string partition, int rowGroupSize, CancellationToken ct = default)
|
||||
{
|
||||
var writer = await ParquetWriter.CreateAsync(TombstoneSchema, output, cancellationToken: ct).ConfigureAwait(false);
|
||||
writer.CustomMetadata = Metadata(partition);
|
||||
return new TombstoneWriter(writer, rowGroupSize);
|
||||
}
|
||||
|
||||
/// <summary>Streams the rows of an archive data file in file order.</summary>
|
||||
/// <param name="input">A readable, seekable stream over the data file.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The rows, as the chain hasher sees them.</returns>
|
||||
public static async IAsyncEnumerable<AuditChainRow> ReadRowsAsync(Stream input, [EnumeratorCancellation] CancellationToken ct = default)
|
||||
{
|
||||
await using var reader = await ParquetReader.CreateAsync(input, leaveStreamOpen: true, cancellationToken: ct).ConfigureAwait(false);
|
||||
for (var g = 0; g < reader.RowGroupCount; g++)
|
||||
{
|
||||
using var group = reader.OpenRowGroupReader(g);
|
||||
var n = checked((int)group.RowCount);
|
||||
var eventIds = await Strings(group, EventIdField, n, ct).ConfigureAwait(false);
|
||||
var ticks = new long[n];
|
||||
await group.ReadAsync<long>(OccurredAtTicksField, ticks, cancellationToken: ct).ConfigureAwait(false);
|
||||
var actors = await Strings(group, ActorField, n, ct).ConfigureAwait(false);
|
||||
var actions = await Strings(group, ActionField, n, ct).ConfigureAwait(false);
|
||||
var outcomes = await Strings(group, OutcomeField, n, ct).ConfigureAwait(false);
|
||||
var categories = await Strings(group, CategoryField, n, ct).ConfigureAwait(false);
|
||||
var targets = await Strings(group, TargetField, n, ct).ConfigureAwait(false);
|
||||
var nodes = await Strings(group, SourceNodeField, n, ct).ConfigureAwait(false);
|
||||
var correlations = await Strings(group, CorrelationIdField, n, ct).ConfigureAwait(false);
|
||||
var details = await Strings(group, DetailsJsonField, n, ct).ConfigureAwait(false);
|
||||
var seqs = new long?[n];
|
||||
await group.ReadAsync<long>(ChainSeqField, seqs, cancellationToken: ct).ConfigureAwait(false);
|
||||
var hashes = new byte[]?[n];
|
||||
await group.ReadAsync(RowHashField, hashes!, cancellationToken: ct).ConfigureAwait(false);
|
||||
|
||||
for (var i = 0; i < n; i++)
|
||||
{
|
||||
yield return new AuditChainRow(
|
||||
Guid.Parse(eventIds[i]!),
|
||||
new DateTime(ticks[i], DateTimeKind.Utc),
|
||||
actors[i],
|
||||
actions[i]!,
|
||||
outcomes[i]!,
|
||||
categories[i]!,
|
||||
targets[i],
|
||||
nodes[i],
|
||||
correlations[i] is { } c ? Guid.Parse(c) : null,
|
||||
details[i],
|
||||
seqs[i],
|
||||
hashes[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Streams the tombstones of an archive tombstone file.</summary>
|
||||
/// <param name="input">A readable, seekable stream over the tombstone file.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The tombstones.</returns>
|
||||
public static async IAsyncEnumerable<AuditArchiveTombstone> ReadTombstonesAsync(Stream input, [EnumeratorCancellation] CancellationToken ct = default)
|
||||
{
|
||||
await using var reader = await ParquetReader.CreateAsync(input, leaveStreamOpen: true, cancellationToken: ct).ConfigureAwait(false);
|
||||
for (var g = 0; g < reader.RowGroupCount; g++)
|
||||
{
|
||||
using var group = reader.OpenRowGroupReader(g);
|
||||
var n = checked((int)group.RowCount);
|
||||
var months = await Strings(group, TombstoneMonthField, n, ct).ConfigureAwait(false);
|
||||
var seqs = new long[n];
|
||||
await group.ReadAsync<long>(TombstoneSeqField, seqs, cancellationToken: ct).ConfigureAwait(false);
|
||||
var hashes = new byte[]?[n];
|
||||
await group.ReadAsync(TombstoneHashField, hashes!, cancellationToken: ct).ConfigureAwait(false);
|
||||
var eventIds = await Strings(group, TombstoneEventIdField, n, ct).ConfigureAwait(false);
|
||||
var purgedAt = new DateTime[n];
|
||||
await group.ReadAsync<DateTime>(TombstonePurgedAtField, purgedAt, cancellationToken: ct).ConfigureAwait(false);
|
||||
|
||||
for (var i = 0; i < n; i++)
|
||||
{
|
||||
yield return new AuditArchiveTombstone(
|
||||
ParseMonth(months[i]!),
|
||||
seqs[i],
|
||||
hashes[i]!,
|
||||
Guid.Parse(eventIds[i]!),
|
||||
DateTime.SpecifyKind(purgedAt[i], DateTimeKind.Utc));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Parses a <c>YYYY-MM</c> chain partition label to its first instant (UTC).</summary>
|
||||
/// <param name="partition">The label.</param>
|
||||
/// <returns>Day 1, midnight UTC.</returns>
|
||||
public static DateTime ParseMonth(string partition)
|
||||
{
|
||||
var month = DateTime.ParseExact(partition, "yyyy-MM", CultureInfo.InvariantCulture, DateTimeStyles.None);
|
||||
return new DateTime(month.Year, month.Month, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
}
|
||||
|
||||
private static Dictionary<string, string> Metadata(string partition) => new()
|
||||
{
|
||||
[FormatVersionKey] = FormatVersion.ToString(CultureInfo.InvariantCulture),
|
||||
[PartitionKey] = partition,
|
||||
};
|
||||
|
||||
private static async Task<string?[]> Strings(ParquetRowGroupReader group, DataField<string> field, int n, CancellationToken ct)
|
||||
{
|
||||
var values = new string?[n];
|
||||
await group.ReadAsync(field, values!, cancellationToken: ct).ConfigureAwait(false);
|
||||
return values;
|
||||
}
|
||||
|
||||
/// <summary>Buffers archive data rows and writes them a row group at a time.</summary>
|
||||
public sealed class DataWriter : IAsyncDisposable
|
||||
{
|
||||
private readonly ParquetWriter _writer;
|
||||
private readonly int _rowGroupSize;
|
||||
private readonly List<AuditChainRow> _buffer;
|
||||
private bool _disposed;
|
||||
|
||||
internal DataWriter(ParquetWriter writer, int rowGroupSize)
|
||||
{
|
||||
_writer = writer;
|
||||
_rowGroupSize = Math.Max(1, rowGroupSize);
|
||||
_buffer = new List<AuditChainRow>(Math.Min(_rowGroupSize, 65_536));
|
||||
}
|
||||
|
||||
/// <summary>Rows added so far.</summary>
|
||||
public long RowCount { get; private set; }
|
||||
|
||||
/// <summary>Adds one row, flushing a row group when the buffer is full.</summary>
|
||||
/// <param name="row">The row.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>A task that completes when the row is buffered (or its row group written).</returns>
|
||||
public async Task AddAsync(AuditChainRow row, CancellationToken ct = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(row);
|
||||
_buffer.Add(row);
|
||||
RowCount++;
|
||||
if (_buffer.Count >= _rowGroupSize)
|
||||
{
|
||||
await FlushAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Writes any buffered rows as a row group, then the file footer.</summary>
|
||||
/// <returns>A task that completes when the file is finished.</returns>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
await FlushAsync(CancellationToken.None).ConfigureAwait(false);
|
||||
await _writer.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task FlushAsync(CancellationToken ct)
|
||||
{
|
||||
if (_buffer.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var rows = _buffer;
|
||||
using (var group = _writer.CreateRowGroup())
|
||||
{
|
||||
await group.WriteAsync(EventIdField, rows.Select(r => r.EventId.ToString("D")).ToArray()).ConfigureAwait(false);
|
||||
await group.WriteAsync<DateTime>(OccurredAtField, rows.Select(r => r.OccurredAtUtc).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync<long>(OccurredAtTicksField, rows.Select(r => r.OccurredAtUtc.Ticks).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync(ActorField, rows.Select(r => r.Actor).ToArray()!).ConfigureAwait(false);
|
||||
await group.WriteAsync(ActionField, rows.Select(r => r.Action).ToArray()).ConfigureAwait(false);
|
||||
await group.WriteAsync(OutcomeField, rows.Select(r => r.Outcome).ToArray()).ConfigureAwait(false);
|
||||
await group.WriteAsync(CategoryField, rows.Select(r => r.Category).ToArray()).ConfigureAwait(false);
|
||||
await group.WriteAsync(TargetField, rows.Select(r => r.Target).ToArray()!).ConfigureAwait(false);
|
||||
await group.WriteAsync(SourceNodeField, rows.Select(r => r.SourceNode).ToArray()!).ConfigureAwait(false);
|
||||
await group.WriteAsync(CorrelationIdField, rows.Select(r => r.CorrelationId?.ToString("D")).ToArray()!).ConfigureAwait(false);
|
||||
await group.WriteAsync(DetailsJsonField, rows.Select(r => r.DetailsJson).ToArray()!).ConfigureAwait(false);
|
||||
await group.WriteAsync<long>(ChainSeqField, rows.Select(r => r.ChainSeq).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync(RowHashField, rows.Select(r => r.RowHash).ToArray()!).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_buffer.Clear();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Buffers tombstones and writes them a row group at a time.</summary>
|
||||
public sealed class TombstoneWriter : IAsyncDisposable
|
||||
{
|
||||
private readonly ParquetWriter _writer;
|
||||
private readonly int _rowGroupSize;
|
||||
private readonly List<AuditArchiveTombstone> _buffer = [];
|
||||
private bool _disposed;
|
||||
|
||||
internal TombstoneWriter(ParquetWriter writer, int rowGroupSize)
|
||||
{
|
||||
_writer = writer;
|
||||
_rowGroupSize = Math.Max(1, rowGroupSize);
|
||||
}
|
||||
|
||||
/// <summary>Tombstones added so far.</summary>
|
||||
public long Count { get; private set; }
|
||||
|
||||
/// <summary>Adds one tombstone, flushing a row group when the buffer is full.</summary>
|
||||
/// <param name="tombstone">The tombstone.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>A task that completes when the tombstone is buffered (or its row group written).</returns>
|
||||
public async Task AddAsync(AuditArchiveTombstone tombstone, CancellationToken ct = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(tombstone);
|
||||
_buffer.Add(tombstone);
|
||||
Count++;
|
||||
if (_buffer.Count >= _rowGroupSize)
|
||||
{
|
||||
await FlushAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Writes any buffered tombstones as a row group, then the file footer.</summary>
|
||||
/// <returns>A task that completes when the file is finished.</returns>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
await FlushAsync(CancellationToken.None).ConfigureAwait(false);
|
||||
await _writer.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task FlushAsync(CancellationToken ct)
|
||||
{
|
||||
if (_buffer.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var rows = _buffer;
|
||||
using (var group = _writer.CreateRowGroup())
|
||||
{
|
||||
await group.WriteAsync(TombstoneMonthField, rows.Select(t => AuditChainHasher.PartitionLabel(t.ChainMonth)).ToArray()).ConfigureAwait(false);
|
||||
await group.WriteAsync<long>(TombstoneSeqField, rows.Select(t => t.ChainSeq).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync(TombstoneHashField, rows.Select(t => t.RowHash).ToArray()).ConfigureAwait(false);
|
||||
await group.WriteAsync(TombstoneEventIdField, rows.Select(t => t.EventId.ToString("D")).ToArray()).ConfigureAwait(false);
|
||||
await group.WriteAsync<DateTime>(TombstonePurgedAtField, rows.Select(t => t.PurgedAtUtc).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_buffer.Clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A forward-only write-through stream that computes the SHA-256 and length of everything
|
||||
/// written to the inner stream, so an archive file is hashed in the same pass that writes it.
|
||||
/// </summary>
|
||||
internal sealed class HashingWriteStream : Stream
|
||||
{
|
||||
private readonly Stream _inner;
|
||||
private readonly IncrementalHash _hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256);
|
||||
|
||||
public HashingWriteStream(Stream inner) => _inner = inner;
|
||||
|
||||
public long BytesWritten { get; private set; }
|
||||
|
||||
public override bool CanRead => false;
|
||||
|
||||
public override bool CanSeek => false;
|
||||
|
||||
public override bool CanWrite => true;
|
||||
|
||||
public override long Length => throw new NotSupportedException();
|
||||
|
||||
public override long Position
|
||||
{
|
||||
get => BytesWritten;
|
||||
set => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
/// <summary>Returns the lowercase-hex SHA-256 of everything written so far (call once, at the end).</summary>
|
||||
public string HashHex()
|
||||
{
|
||||
var hex = Convert.ToHexStringLower(_hash.GetHashAndReset());
|
||||
_hash.Dispose();
|
||||
return hex;
|
||||
}
|
||||
|
||||
public override void Write(byte[] buffer, int offset, int count) => Write(buffer.AsSpan(offset, count));
|
||||
|
||||
public override void Write(ReadOnlySpan<byte> buffer)
|
||||
{
|
||||
_hash.AppendData(buffer);
|
||||
_inner.Write(buffer);
|
||||
BytesWritten += buffer.Length;
|
||||
}
|
||||
|
||||
public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_hash.AppendData(buffer.Span);
|
||||
await _inner.WriteAsync(buffer, cancellationToken).ConfigureAwait(false);
|
||||
BytesWritten += buffer.Length;
|
||||
}
|
||||
|
||||
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
||||
=> WriteAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
|
||||
|
||||
public override void Flush() => _inner.Flush();
|
||||
|
||||
public override Task FlushAsync(CancellationToken cancellationToken) => _inner.FlushAsync(cancellationToken);
|
||||
|
||||
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
|
||||
|
||||
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
|
||||
|
||||
public override void SetLength(long value) => throw new NotSupportedException();
|
||||
|
||||
// The hash is NOT released on Dispose: the Parquet writer may dispose this stream when it
|
||||
// finishes, and HashHex() is read after that. HashHex() releases it.
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
/// <summary>
|
||||
/// Options for the AuditLog Parquet cold archive (section <c>AuditLog:Archive</c>; audit
|
||||
/// roadmap T2, alog.md §15.2, ScadaBridge#42). Validated at startup by
|
||||
/// <c>AuditLogArchiveOptionsValidator</c> in the AuditLog component.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Default OFF.</b> With <see cref="Enabled"/> false the retention purge drops expired
|
||||
/// partitions exactly as before. With it true the purge writes each expired partition to
|
||||
/// Parquet under <see cref="Directory"/> first and refuses to drop a partition it could not
|
||||
/// archive — so a full or unreachable archive volume stops the purge (and the table grows)
|
||||
/// rather than losing audit rows.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Directory.</b> An absolute path both central nodes can reach (a shared volume or UNC
|
||||
/// share): the purge runs on whichever node is active, and <c>audit archive list/verify</c>
|
||||
/// run on whichever node serves the management call. Files land in a <c>YYYY</c>
|
||||
/// sub-directory per partition year. Object storage (S3/Azure Blob) is not wired — mount it
|
||||
/// as a file system or copy the files on from there.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class AuditLogArchiveOptions
|
||||
{
|
||||
/// <summary>Whether the retention purge archives each partition before dropping it (default false).</summary>
|
||||
public bool Enabled { get; set; }
|
||||
|
||||
/// <summary>Absolute archive root directory; required when <see cref="Enabled"/>.</summary>
|
||||
public string? Directory { get; set; }
|
||||
|
||||
/// <summary>Rows per Parquet row group (default 50,000; bounds archive-writer memory).</summary>
|
||||
public int RowGroupSize { get; set; } = 50_000;
|
||||
|
||||
/// <summary>
|
||||
/// Command timeout, in minutes, for reading one partition out of SQL Server (default 60).
|
||||
/// A month on a busy central is millions of rows.
|
||||
/// </summary>
|
||||
public int ReadCommandTimeoutMinutes { get; set; } = 60;
|
||||
|
||||
/// <summary>
|
||||
/// How long, in seconds, archiving waits for a chaining pass or another purge to release
|
||||
/// the hash-chain lock before giving up for this tick (default 300).
|
||||
/// </summary>
|
||||
public int ChainLockTimeoutSeconds { get; set; } = 300;
|
||||
}
|
||||
@@ -50,7 +50,8 @@ public sealed class AuditLogHashChain : IAuditLogHashChain
|
||||
// 4 parameters per row; 200 rows = 800 parameters, far under SQL Server's 2,100 cap.
|
||||
private const int UpdateChunkRows = 200;
|
||||
|
||||
private const string RowColumns =
|
||||
/// <summary>The ten writable AuditLog columns, in table order — the hashed row content.</summary>
|
||||
internal const string RowColumns =
|
||||
"EventId, OccurredAtUtc, Actor, Action, Outcome, Category, Target, SourceNode, CorrelationId, DetailsJson";
|
||||
|
||||
private readonly ScadaBridgeDbContext _context;
|
||||
@@ -344,7 +345,11 @@ public sealed class AuditLogHashChain : IAuditLogHashChain
|
||||
return await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false) == 1;
|
||||
}
|
||||
|
||||
private static AuditChainRow ReadRow(DbDataReader reader, bool withLink)
|
||||
/// <summary>
|
||||
/// Reads one row selected as <see cref="RowColumns"/> (plus <c>ChainSeq, RowHash</c> when
|
||||
/// <paramref name="withLink"/>), strictly in ordinal order so it is safe under SequentialAccess.
|
||||
/// </summary>
|
||||
internal static AuditChainRow ReadRow(DbDataReader reader, bool withLink)
|
||||
{
|
||||
// Ordinals follow RowColumns (+ ChainSeq, RowHash). Read strictly in order — the
|
||||
// verification reader is opened with SequentialAccess.
|
||||
@@ -390,7 +395,7 @@ public sealed class AuditLogHashChain : IAuditLogHashChain
|
||||
cmd.Parameters.Add(new SqlParameter("@monthEnd", SqlDbType.DateTime2) { Scale = 7, Value = month.AddMonths(1) });
|
||||
}
|
||||
|
||||
private static async Task<bool> OpenAsync(DbConnection conn, CancellationToken ct)
|
||||
internal static async Task<bool> OpenAsync(DbConnection conn, CancellationToken ct)
|
||||
{
|
||||
if (conn.State == ConnectionState.Open)
|
||||
{
|
||||
|
||||
+635
@@ -0,0 +1,635 @@
|
||||
using System.Data;
|
||||
using System.Data.Common;
|
||||
using System.Globalization;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text.Json;
|
||||
using System.Text.RegularExpressions;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
/// <summary>
|
||||
/// SQL Server + file-system implementation of <see cref="IAuditLogArchive"/> (audit roadmap
|
||||
/// T2, alog.md §15.2, ScadaBridge#42). See the interface for the archive-before-purge
|
||||
/// contract and <see cref="AuditArchiveParquet"/> for the file layout.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Layout.</b> <c>{Directory}/{YYYY}/{name}.parquet</c> (rows),
|
||||
/// <c>{name}.tombstones.parquet</c> (per-channel purge tombstones) and
|
||||
/// <c>{name}.manifest.json</c>, where <c>name</c> is
|
||||
/// <c>auditlog-{YYYY-MM}-{archivedAt:yyyyMMddTHHmmssfff}Z</c>. Each file is written to a
|
||||
/// <c>.tmp</c> sibling, flushed to disk and renamed; the manifest goes last, so a manifest
|
||||
/// only ever names complete files. Archives are never overwritten: a retried purge writes a
|
||||
/// new archive, and the superseded one keeps status <see cref="AuditArchiveStatus.Archived"/>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Consistency with the hash chain.</b> The partition is read inside one transaction that
|
||||
/// holds the chain lock SHARED, so no chaining pass or purge moves a head or writes a
|
||||
/// tombstone while the rows, heads and tombstones are read. The purge then re-checks the
|
||||
/// row count and heads under the EXCLUSIVE lock (<see cref="AuditPartitionSwitchGuard"/>).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed partial class AuditLogPartitionArchive : IAuditLogArchive
|
||||
{
|
||||
private static readonly JsonSerializerOptions ManifestJson = new()
|
||||
{
|
||||
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
|
||||
WriteIndented = true,
|
||||
};
|
||||
|
||||
private readonly ScadaBridgeDbContext _context;
|
||||
private readonly AuditLogArchiveOptions _options;
|
||||
private readonly ILogger<AuditLogPartitionArchive> _logger;
|
||||
private readonly TimeProvider _time;
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="AuditLogPartitionArchive"/> class.</summary>
|
||||
/// <param name="context">The database context whose connection is used.</param>
|
||||
/// <param name="options">Archive options.</param>
|
||||
/// <param name="logger">Optional logger.</param>
|
||||
/// <param name="time">Optional clock (tests).</param>
|
||||
public AuditLogPartitionArchive(
|
||||
ScadaBridgeDbContext context,
|
||||
AuditLogArchiveOptions options,
|
||||
ILogger<AuditLogPartitionArchive>? logger = null,
|
||||
TimeProvider? time = null)
|
||||
{
|
||||
_context = context ?? throw new ArgumentNullException(nameof(context));
|
||||
_options = options ?? throw new ArgumentNullException(nameof(options));
|
||||
_logger = logger ?? NullLogger<AuditLogPartitionArchive>.Instance;
|
||||
_time = time ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool IsConfigured => !string.IsNullOrWhiteSpace(_options.Directory);
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool IsEnabled => _options.Enabled && IsConfigured;
|
||||
|
||||
/// <summary>Returns true when <paramref name="name"/> is a well-formed archive name.</summary>
|
||||
/// <param name="name">The candidate name.</param>
|
||||
/// <returns>True for <c>auditlog-YYYY-MM-yyyyMMddTHHmmssfffZ</c>.</returns>
|
||||
public static bool IsValidName(string? name) => name is not null && NamePattern().IsMatch(name);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<AuditArchiveManifest> ArchivePartitionAsync(DateTime partitionBoundaryUtc, CancellationToken ct = default)
|
||||
{
|
||||
if (!_options.Enabled || !IsConfigured)
|
||||
{
|
||||
throw new InvalidOperationException("AuditLog archival is not enabled (AuditLog:Archive:Enabled / Directory).");
|
||||
}
|
||||
|
||||
var boundary = DateTime.SpecifyKind(partitionBoundaryUtc, DateTimeKind.Utc);
|
||||
var partition = AuditChainHasher.PartitionLabel(AuditChainHasher.MonthOf(boundary));
|
||||
var archivedAt = _time.GetUtcNow().UtcDateTime;
|
||||
var name = $"auditlog-{partition}-{archivedAt.ToString("yyyyMMdd'T'HHmmssfff", CultureInfo.InvariantCulture)}Z";
|
||||
var directory = Path.Combine(_options.Directory!, partition[..4]);
|
||||
Directory.CreateDirectory(directory);
|
||||
|
||||
var dataFile = name + ".parquet";
|
||||
var tombstoneFile = name + ".tombstones.parquet";
|
||||
var dataTmp = Path.Combine(directory, dataFile + ".tmp");
|
||||
var tombstoneTmp = Path.Combine(directory, tombstoneFile + ".tmp");
|
||||
|
||||
var chains = new List<AuditArchiveChainState>();
|
||||
long rowCount = 0, chained = 0;
|
||||
DateTime? minOccurred = null, maxOccurred = null;
|
||||
string dataSha, tombstoneSha;
|
||||
long dataBytes, tombstoneCount;
|
||||
|
||||
try
|
||||
{
|
||||
var conn = _context.Database.GetDbConnection();
|
||||
var openedHere = await AuditLogHashChain.OpenAsync(conn, ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
await using var tx = await conn.BeginTransactionAsync(IsolationLevel.ReadCommitted, ct).ConfigureAwait(false);
|
||||
if (!await AuditLogHashChain.TryAcquireLockAsync(
|
||||
conn, tx, "Shared", TimeSpan.FromSeconds(Math.Max(1, _options.ChainLockTimeoutSeconds)), ct)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
await tx.RollbackAsync(ct).ConfigureAwait(false);
|
||||
throw new TimeoutException(
|
||||
$"The AuditLog hash-chain lock stayed busy for {_options.ChainLockTimeoutSeconds}s; partition {partition} not archived this tick.");
|
||||
}
|
||||
|
||||
// 1. Chain heads of every month the partition holds (normally exactly one).
|
||||
var tombstoneCounts = new Dictionary<DateTime, long>();
|
||||
var heads = new List<(DateTime Month, long LastSeq, byte[] LastHash, long? PurgedSeq, byte[]? PurgedHash)>();
|
||||
await using (var cmd = Command(conn, tx,
|
||||
"SELECT ChainMonth, LastSeq, LastHash, PurgedThroughSeq, PurgedThroughHash FROM dbo.AuditLogChainHead " +
|
||||
"WHERE $partition.pf_AuditLog_Month(CAST(ChainMonth AS datetime2(7))) = $partition.pf_AuditLog_Month(@boundary) " +
|
||||
"ORDER BY ChainMonth;", boundary))
|
||||
{
|
||||
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
heads.Add((
|
||||
DateTime.SpecifyKind(reader.GetDateTime(0), DateTimeKind.Utc),
|
||||
reader.GetInt64(1),
|
||||
(byte[])reader.GetValue(2),
|
||||
reader.IsDBNull(3) ? null : reader.GetInt64(3),
|
||||
reader.IsDBNull(4) ? null : (byte[])reader.GetValue(4)));
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Per-channel purge tombstones (verification bridges gaps with them).
|
||||
await using (var file = NewFile(tombstoneTmp))
|
||||
{
|
||||
var hashing = new HashingWriteStream(file);
|
||||
await using (var writer = await AuditArchiveParquet
|
||||
.CreateTombstoneWriterAsync(hashing, partition, _options.RowGroupSize, ct).ConfigureAwait(false))
|
||||
{
|
||||
await using var cmd = Command(conn, tx,
|
||||
"SELECT ChainMonth, ChainSeq, RowHash, EventId, PurgedAtUtc FROM dbo.AuditLogChainTombstone " +
|
||||
"WHERE $partition.pf_AuditLog_Month(CAST(ChainMonth AS datetime2(7))) = $partition.pf_AuditLog_Month(@boundary) " +
|
||||
"ORDER BY ChainMonth, ChainSeq;", boundary);
|
||||
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
var tombstone = new AuditArchiveTombstone(
|
||||
DateTime.SpecifyKind(reader.GetDateTime(0), DateTimeKind.Utc),
|
||||
reader.GetInt64(1),
|
||||
(byte[])reader.GetValue(2),
|
||||
reader.GetGuid(3),
|
||||
DateTime.SpecifyKind(reader.GetDateTime(4), DateTimeKind.Utc));
|
||||
tombstoneCounts[tombstone.ChainMonth] = tombstoneCounts.GetValueOrDefault(tombstone.ChainMonth) + 1;
|
||||
await writer.AddAsync(tombstone, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
tombstoneCount = writer.Count;
|
||||
}
|
||||
|
||||
await hashing.FlushAsync(ct).ConfigureAwait(false);
|
||||
file.Flush(flushToDisk: true);
|
||||
tombstoneSha = hashing.HashHex();
|
||||
}
|
||||
|
||||
// 3. The rows, grouped by chain month and in chain order within it (unchained
|
||||
// rows last), which is the order verification needs.
|
||||
await using (var file = NewFile(dataTmp))
|
||||
{
|
||||
var hashing = new HashingWriteStream(file);
|
||||
await using (var writer = await AuditArchiveParquet
|
||||
.CreateDataWriterAsync(hashing, partition, _options.RowGroupSize, ct).ConfigureAwait(false))
|
||||
{
|
||||
await using var cmd = Command(conn, tx,
|
||||
$"SELECT {AuditLogHashChain.RowColumns}, ChainSeq, RowHash FROM dbo.AuditLog " +
|
||||
"WHERE $PARTITION.pf_AuditLog_Month(OccurredAtUtc) = $partition.pf_AuditLog_Month(@boundary) " +
|
||||
"ORDER BY DATEFROMPARTS(YEAR(OccurredAtUtc), MONTH(OccurredAtUtc), 1), " +
|
||||
"CASE WHEN ChainSeq IS NULL THEN 1 ELSE 0 END, ChainSeq, OccurredAtUtc, EventId;", boundary);
|
||||
cmd.CommandTimeout = Math.Max(1, _options.ReadCommandTimeoutMinutes) * 60;
|
||||
await using var reader = await cmd.ExecuteReaderAsync(CommandBehavior.SequentialAccess, ct).ConfigureAwait(false);
|
||||
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
var row = AuditLogHashChain.ReadRow(reader, withLink: true);
|
||||
if (row.ChainSeq is not null)
|
||||
{
|
||||
chained++;
|
||||
}
|
||||
|
||||
minOccurred = minOccurred is { } lo && lo <= row.OccurredAtUtc ? lo : row.OccurredAtUtc;
|
||||
maxOccurred = maxOccurred is { } hi && hi >= row.OccurredAtUtc ? hi : row.OccurredAtUtc;
|
||||
await writer.AddAsync(row, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
rowCount = writer.RowCount;
|
||||
}
|
||||
|
||||
await hashing.FlushAsync(ct).ConfigureAwait(false);
|
||||
file.Flush(flushToDisk: true);
|
||||
dataSha = hashing.HashHex();
|
||||
dataBytes = hashing.BytesWritten;
|
||||
}
|
||||
|
||||
await tx.CommitAsync(ct).ConfigureAwait(false);
|
||||
|
||||
foreach (var head in heads)
|
||||
{
|
||||
chains.Add(new AuditArchiveChainState(
|
||||
AuditChainHasher.PartitionLabel(head.Month),
|
||||
head.LastSeq,
|
||||
Convert.ToHexStringLower(head.LastHash),
|
||||
head.PurgedSeq,
|
||||
head.PurgedHash is null ? null : Convert.ToHexStringLower(head.PurgedHash),
|
||||
tombstoneCounts.GetValueOrDefault(head.Month)));
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (openedHere)
|
||||
{
|
||||
await conn.CloseAsync().ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
File.Move(dataTmp, Path.Combine(directory, dataFile));
|
||||
File.Move(tombstoneTmp, Path.Combine(directory, tombstoneFile));
|
||||
}
|
||||
catch
|
||||
{
|
||||
TryDelete(dataTmp);
|
||||
TryDelete(tombstoneTmp);
|
||||
throw;
|
||||
}
|
||||
|
||||
var manifest = new AuditArchiveManifest(
|
||||
AuditArchiveParquet.FormatVersion,
|
||||
name,
|
||||
boundary,
|
||||
partition,
|
||||
archivedAt,
|
||||
AuditArchiveStatus.Archived,
|
||||
PurgedAtUtc: null,
|
||||
rowCount,
|
||||
chained,
|
||||
rowCount - chained,
|
||||
minOccurred,
|
||||
maxOccurred,
|
||||
dataFile,
|
||||
dataSha,
|
||||
dataBytes,
|
||||
tombstoneFile,
|
||||
tombstoneSha,
|
||||
tombstoneCount,
|
||||
chains);
|
||||
await WriteManifestAsync(directory, manifest, ct).ConfigureAwait(false);
|
||||
|
||||
_logger.LogInformation(
|
||||
"Archived AuditLog partition {Partition} to {Archive}: {Rows} rows ({Chained} chained), {Tombstones} tombstones, {Bytes} bytes.",
|
||||
partition, name, rowCount, chained, tombstoneCount, dataBytes);
|
||||
return manifest;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<long> PurgeArchivedPartitionAsync(
|
||||
AuditArchiveManifest manifest, TimeSpan? commandTimeout = null, CancellationToken ct = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(manifest);
|
||||
var guard = new AuditPartitionSwitchGuard(
|
||||
manifest.RowCount,
|
||||
manifest.Chains.Select(c => (AuditArchiveParquet.ParseMonth(c.Partition), c.LastSeq)).ToList());
|
||||
|
||||
long purged;
|
||||
try
|
||||
{
|
||||
purged = await new AuditLogRepository(_context)
|
||||
.SwitchOutPartitionAsync(manifest.PartitionBoundaryUtc, guard, commandTimeout, ct)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex) when (FindSqlError(ex) is { } number
|
||||
&& number is AuditPartitionSwitchGuard.RowCountChangedError or AuditPartitionSwitchGuard.ChainAdvancedError)
|
||||
{
|
||||
throw new AuditArchiveStaleException(
|
||||
$"Partition {manifest.Partition} changed after archive {manifest.Name} was written; nothing was purged, "
|
||||
+ "the next purge tick archives it again.", ex);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var directory = Path.Combine(_options.Directory!, manifest.Partition[..4]);
|
||||
await WriteManifestAsync(directory, manifest with
|
||||
{
|
||||
Status = AuditArchiveStatus.Purged,
|
||||
PurgedAtUtc = _time.GetUtcNow().UtcDateTime,
|
||||
}, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
|
||||
{
|
||||
// The rows are gone and the archive files are complete; only the status flag is
|
||||
// stale. Verify still ties the archive to the purge anchor (anchor Matches).
|
||||
_logger.LogError(ex,
|
||||
"AuditLog partition {Partition} was purged but archive {Archive}'s manifest could not be marked Purged.",
|
||||
manifest.Partition, manifest.Name);
|
||||
}
|
||||
|
||||
return purged;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<IReadOnlyList<AuditArchiveManifest>> ListAsync(CancellationToken ct = default)
|
||||
{
|
||||
RequireConfigured();
|
||||
if (!Directory.Exists(_options.Directory))
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
var manifests = new List<AuditArchiveManifest>();
|
||||
foreach (var path in Directory.EnumerateFiles(_options.Directory!, "*.manifest.json", SearchOption.AllDirectories))
|
||||
{
|
||||
if (!IsValidName(Path.GetFileName(path)[..^".manifest.json".Length]))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
manifests.Add(await ReadManifestAsync(path, ct).ConfigureAwait(false));
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Skipping unreadable AuditLog archive manifest {Path}.", path);
|
||||
}
|
||||
}
|
||||
|
||||
return manifests
|
||||
.OrderBy(m => m.PartitionBoundaryUtc)
|
||||
.ThenBy(m => m.ArchivedAtUtc)
|
||||
.ToList();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<AuditArchiveVerificationResult> VerifyAsync(string name, CancellationToken ct = default)
|
||||
{
|
||||
RequireConfigured();
|
||||
if (!IsValidName(name))
|
||||
{
|
||||
throw new ArgumentException(
|
||||
$"'{name}' is not an archive name (expected auditlog-YYYY-MM-yyyyMMddTHHmmssfffZ).", nameof(name));
|
||||
}
|
||||
|
||||
var directory = Path.Combine(_options.Directory!, name.Substring("auditlog-".Length, 4));
|
||||
var manifestPath = Path.Combine(directory, name + ".manifest.json");
|
||||
if (!File.Exists(manifestPath))
|
||||
{
|
||||
throw new FileNotFoundException($"No AuditLog archive named '{name}'.", manifestPath);
|
||||
}
|
||||
|
||||
var manifest = await ReadManifestAsync(manifestPath, ct).ConfigureAwait(false);
|
||||
var problems = new List<string>();
|
||||
var dataPath = Path.Combine(directory, Path.GetFileName(manifest.DataFile));
|
||||
var tombstonePath = Path.Combine(directory, Path.GetFileName(manifest.TombstoneFile));
|
||||
await CheckFileAsync(dataPath, manifest.DataSha256, manifest.DataBytes, problems, ct).ConfigureAwait(false);
|
||||
await CheckFileAsync(tombstonePath, manifest.TombstoneSha256, null, problems, ct).ConfigureAwait(false);
|
||||
if (problems.Count > 0)
|
||||
{
|
||||
return new AuditArchiveVerificationResult(
|
||||
manifest.Name, manifest.Partition, manifest.Status, Verified: false, FilesIntact: false,
|
||||
string.Join(" ", problems), RowsRead: 0, []);
|
||||
}
|
||||
|
||||
// Tombstones per month, then one streaming verifier per chain month.
|
||||
var tombstones = new Dictionary<DateTime, Dictionary<long, byte[]>>();
|
||||
long tombstonesRead = 0;
|
||||
await using (var input = File.OpenRead(tombstonePath))
|
||||
{
|
||||
await foreach (var t in AuditArchiveParquet.ReadTombstonesAsync(input, ct).ConfigureAwait(false))
|
||||
{
|
||||
if (!tombstones.TryGetValue(t.ChainMonth, out var byMonth))
|
||||
{
|
||||
tombstones[t.ChainMonth] = byMonth = [];
|
||||
}
|
||||
|
||||
byMonth[t.ChainSeq] = t.RowHash;
|
||||
tombstonesRead++;
|
||||
}
|
||||
}
|
||||
|
||||
var states = manifest.Chains.ToDictionary(c => AuditArchiveParquet.ParseMonth(c.Partition));
|
||||
var verifiers = new SortedDictionary<DateTime, AuditChainVerifier>();
|
||||
var unchained = new Dictionary<DateTime, long>();
|
||||
AuditChainVerifier VerifierFor(DateTime month)
|
||||
{
|
||||
if (!verifiers.TryGetValue(month, out var verifier))
|
||||
{
|
||||
AuditChainHeadState? head = states.TryGetValue(month, out var s)
|
||||
? new AuditChainHeadState(
|
||||
s.LastSeq,
|
||||
Convert.FromHexString(s.LastHash),
|
||||
s.PriorPurgedThroughSeq,
|
||||
s.PriorPurgedThroughHash is null ? null : Convert.FromHexString(s.PriorPurgedThroughHash))
|
||||
: null;
|
||||
verifier = new AuditChainVerifier(month, head, tombstones.GetValueOrDefault(month) ?? []);
|
||||
verifiers[month] = verifier;
|
||||
}
|
||||
|
||||
return verifier;
|
||||
}
|
||||
|
||||
foreach (var month in states.Keys)
|
||||
{
|
||||
VerifierFor(month);
|
||||
}
|
||||
|
||||
long rowsRead = 0;
|
||||
await using (var input = File.OpenRead(dataPath))
|
||||
{
|
||||
await foreach (var row in AuditArchiveParquet.ReadRowsAsync(input, ct).ConfigureAwait(false))
|
||||
{
|
||||
rowsRead++;
|
||||
var month = AuditChainHasher.MonthOf(row.OccurredAtUtc);
|
||||
if (row.ChainSeq is null)
|
||||
{
|
||||
unchained[month] = unchained.GetValueOrDefault(month) + 1;
|
||||
VerifierFor(month);
|
||||
continue;
|
||||
}
|
||||
|
||||
VerifierFor(month).Accept(row);
|
||||
}
|
||||
}
|
||||
|
||||
if (rowsRead != manifest.RowCount)
|
||||
{
|
||||
problems.Add($"The data file holds {rowsRead} rows; the manifest records {manifest.RowCount}.");
|
||||
}
|
||||
|
||||
if (tombstonesRead != manifest.TombstoneCount)
|
||||
{
|
||||
problems.Add($"The tombstone file holds {tombstonesRead} tombstones; the manifest records {manifest.TombstoneCount}.");
|
||||
}
|
||||
|
||||
var anchors = await ReadAnchorsAsync(verifiers.Keys, ct).ConfigureAwait(false);
|
||||
var results = new List<AuditArchiveChainVerification>();
|
||||
foreach (var (month, verifier) in verifiers)
|
||||
{
|
||||
var chain = verifier.Complete(unchained.GetValueOrDefault(month));
|
||||
var (anchor, detail) = CompareAnchor(states.GetValueOrDefault(month), anchors.GetValueOrDefault(month));
|
||||
results.Add(new AuditArchiveChainVerification(chain, anchor, detail));
|
||||
}
|
||||
|
||||
var filesIntact = problems.Count == 0;
|
||||
var verified = filesIntact
|
||||
&& results.All(r => r.Chain.Status != AuditChainStatus.Broken && r.Anchor != AuditArchiveAnchorStatus.Mismatch);
|
||||
return new AuditArchiveVerificationResult(
|
||||
manifest.Name, manifest.Partition, manifest.Status, verified, filesIntact,
|
||||
filesIntact ? null : string.Join(" ", problems), rowsRead, results);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Compares an archived chain end with the database's current checkpoint anchor for the month.
|
||||
/// </summary>
|
||||
internal static (AuditArchiveAnchorStatus Status, string Detail) CompareAnchor(
|
||||
AuditArchiveChainState? archived, (bool HeadExists, long? PurgedSeq, byte[]? PurgedHash) database)
|
||||
{
|
||||
if (archived is null)
|
||||
{
|
||||
// No chain head existed when the rows were archived (chaining off, or the month was
|
||||
// never reached): nothing to tie. A later anchor can only come from a later purge.
|
||||
return database.PurgedSeq is null
|
||||
? (AuditArchiveAnchorStatus.Matches, "No hash chain existed for this month when it was archived.")
|
||||
: (AuditArchiveAnchorStatus.Superseded,
|
||||
$"No chain at archive time; a later purge anchored the month at ChainSeq {database.PurgedSeq}.");
|
||||
}
|
||||
|
||||
if (!database.HeadExists)
|
||||
{
|
||||
return (AuditArchiveAnchorStatus.Mismatch,
|
||||
$"The archive records a chain through ChainSeq {archived.LastSeq} but the database has no chain head for {archived.Partition}.");
|
||||
}
|
||||
|
||||
if (database.PurgedSeq is null)
|
||||
{
|
||||
return (AuditArchiveAnchorStatus.NotPurged,
|
||||
$"{archived.Partition} has not been purged: the archived rows are still in dbo.AuditLog.");
|
||||
}
|
||||
|
||||
if (database.PurgedSeq == archived.LastSeq)
|
||||
{
|
||||
return database.PurgedHash is not null
|
||||
&& Convert.ToHexStringLower(database.PurgedHash) == archived.LastHash.ToLowerInvariant()
|
||||
? (AuditArchiveAnchorStatus.Matches,
|
||||
$"The archived chain ends at the purge anchor (ChainSeq {archived.LastSeq}).")
|
||||
: (AuditArchiveAnchorStatus.Mismatch,
|
||||
$"The purge anchor is at ChainSeq {archived.LastSeq} but its hash differs from the archive's chain end.");
|
||||
}
|
||||
|
||||
return database.PurgedSeq > archived.LastSeq
|
||||
? (AuditArchiveAnchorStatus.Superseded,
|
||||
$"The purge anchor (ChainSeq {database.PurgedSeq}) is past this archive's chain end ({archived.LastSeq}): "
|
||||
+ "late rows were purged and archived again afterwards.")
|
||||
: (AuditArchiveAnchorStatus.Mismatch,
|
||||
$"The purge anchor (ChainSeq {database.PurgedSeq}) is before this archive's chain end ({archived.LastSeq}).");
|
||||
}
|
||||
|
||||
private async Task<Dictionary<DateTime, (bool HeadExists, long? PurgedSeq, byte[]? PurgedHash)>> ReadAnchorsAsync(
|
||||
IEnumerable<DateTime> months, CancellationToken ct)
|
||||
{
|
||||
var result = new Dictionary<DateTime, (bool, long?, byte[]?)>();
|
||||
var conn = _context.Database.GetDbConnection();
|
||||
var openedHere = await AuditLogHashChain.OpenAsync(conn, ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
foreach (var month in months)
|
||||
{
|
||||
await using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = "SELECT PurgedThroughSeq, PurgedThroughHash FROM dbo.AuditLogChainHead WHERE ChainMonth = @month;";
|
||||
cmd.Parameters.Add(new SqlParameter("@month", SqlDbType.Date) { Value = month.Date });
|
||||
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||||
result[month] = await reader.ReadAsync(ct).ConfigureAwait(false)
|
||||
? (true,
|
||||
reader.IsDBNull(0) ? null : reader.GetInt64(0),
|
||||
reader.IsDBNull(1) ? null : (byte[])reader.GetValue(1))
|
||||
: (false, null, null);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (openedHere)
|
||||
{
|
||||
await conn.CloseAsync().ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private static async Task CheckFileAsync(string path, string expectedSha, long? expectedBytes, List<string> problems, CancellationToken ct)
|
||||
{
|
||||
if (!File.Exists(path))
|
||||
{
|
||||
problems.Add($"{Path.GetFileName(path)} is missing.");
|
||||
return;
|
||||
}
|
||||
|
||||
await using var input = File.OpenRead(path);
|
||||
if (expectedBytes is { } bytes && input.Length != bytes)
|
||||
{
|
||||
problems.Add($"{Path.GetFileName(path)} is {input.Length} bytes; the manifest records {bytes}.");
|
||||
return;
|
||||
}
|
||||
|
||||
var actual = Convert.ToHexStringLower(await SHA256.HashDataAsync(input, ct).ConfigureAwait(false));
|
||||
if (!string.Equals(actual, expectedSha, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
problems.Add($"{Path.GetFileName(path)}'s SHA-256 does not match the manifest: the file was altered.");
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task WriteManifestAsync(string directory, AuditArchiveManifest manifest, CancellationToken ct)
|
||||
{
|
||||
var path = Path.Combine(directory, manifest.Name + ".manifest.json");
|
||||
var tmp = path + ".tmp";
|
||||
await using (var file = NewFile(tmp))
|
||||
{
|
||||
await JsonSerializer.SerializeAsync(file, manifest, ManifestJson, ct).ConfigureAwait(false);
|
||||
file.Flush(flushToDisk: true);
|
||||
}
|
||||
|
||||
File.Move(tmp, path, overwrite: true);
|
||||
}
|
||||
|
||||
private static async Task<AuditArchiveManifest> ReadManifestAsync(string path, CancellationToken ct)
|
||||
{
|
||||
await using var input = File.OpenRead(path);
|
||||
return await JsonSerializer.DeserializeAsync<AuditArchiveManifest>(input, ManifestJson, ct).ConfigureAwait(false)
|
||||
?? throw new JsonException($"Empty manifest {path}.");
|
||||
}
|
||||
|
||||
private static FileStream NewFile(string path)
|
||||
=> new(path, FileMode.Create, FileAccess.Write, FileShare.None, bufferSize: 1 << 16, FileOptions.Asynchronous);
|
||||
|
||||
private static DbCommand Command(DbConnection conn, DbTransaction tx, string sql, DateTime boundary)
|
||||
{
|
||||
var cmd = conn.CreateCommand();
|
||||
cmd.Transaction = tx;
|
||||
cmd.CommandText = sql;
|
||||
cmd.Parameters.Add(new SqlParameter("@boundary", SqlDbType.DateTime2) { Scale = 7, Value = boundary });
|
||||
return cmd;
|
||||
}
|
||||
|
||||
private static int? FindSqlError(Exception? ex)
|
||||
{
|
||||
for (; ex is not null; ex = ex.InnerException)
|
||||
{
|
||||
if (ex is SqlException sql)
|
||||
{
|
||||
return sql.Number;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private static void TryDelete(string path)
|
||||
{
|
||||
try
|
||||
{
|
||||
File.Delete(path);
|
||||
}
|
||||
catch (IOException)
|
||||
{
|
||||
}
|
||||
catch (UnauthorizedAccessException)
|
||||
{
|
||||
}
|
||||
}
|
||||
|
||||
private void RequireConfigured()
|
||||
{
|
||||
if (!IsConfigured)
|
||||
{
|
||||
throw new InvalidOperationException("No AuditLog archive directory is configured (AuditLog:Archive:Directory).");
|
||||
}
|
||||
}
|
||||
|
||||
[GeneratedRegex(@"^auditlog-\d{4}-(0[1-9]|1[0-2])-\d{8}T\d{9}Z$", RegexOptions.CultureInvariant)]
|
||||
private static partial Regex NamePattern();
|
||||
}
|
||||
+56
@@ -0,0 +1,56 @@
|
||||
using System.Globalization;
|
||||
using System.Text;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
/// <summary>
|
||||
/// What an archive captured about a partition (ScadaBridge#42), checked inside the
|
||||
/// partition-switch transaction so the purge drops exactly the rows the archive holds.
|
||||
/// </summary>
|
||||
/// <param name="ExpectedRows">Rows the archive holds for the partition.</param>
|
||||
/// <param name="ExpectedHeads">Each chain month's <c>LastSeq</c> at archive time.</param>
|
||||
public sealed record AuditPartitionSwitchGuard(long ExpectedRows, IReadOnlyList<(DateTime ChainMonth, long LastSeq)> ExpectedHeads)
|
||||
{
|
||||
/// <summary>SQL error number THROWn when the partition row count changed.</summary>
|
||||
public const int RowCountChangedError = 51002;
|
||||
|
||||
/// <summary>SQL error number THROWn when a chain head in the partition changed.</summary>
|
||||
public const int ChainAdvancedError = 51003;
|
||||
|
||||
/// <summary>
|
||||
/// Renders the guard as T-SQL for the switch batch. Every value is a number or a
|
||||
/// <c>yyyy-MM-dd</c> date produced here, so inlining them as literals is injection-safe and
|
||||
/// keeps the switch a parameterless <c>ExecuteSqlRaw</c> batch.
|
||||
/// </summary>
|
||||
/// <param name="monthBoundaryLiteral">The boundary literal the switch batch already uses.</param>
|
||||
/// <returns>The guard statements (run after the chain lock, inside the TRY).</returns>
|
||||
internal string ToSql(string monthBoundaryLiteral)
|
||||
{
|
||||
var sql = new StringBuilder();
|
||||
sql.AppendLine();
|
||||
sql.AppendLine(" -- 0b. Archive guard (ScadaBridge#42): the archive must hold exactly the rows this");
|
||||
sql.AppendLine(" -- switch drops. Under the exclusive chain lock, so nothing chains or per-channel");
|
||||
sql.AppendLine(" -- purges between this check and the SWITCH.");
|
||||
sql.AppendLine(CultureInfo.InvariantCulture, $" DECLARE @guardPartition int = $partition.pf_AuditLog_Month('{monthBoundaryLiteral}');");
|
||||
sql.AppendLine(" IF (SELECT COUNT_BIG(*) FROM dbo.AuditLog WHERE $PARTITION.pf_AuditLog_Month(OccurredAtUtc) = @guardPartition)");
|
||||
sql.AppendLine(CultureInfo.InvariantCulture, $" <> {ExpectedRows}");
|
||||
sql.AppendLine(CultureInfo.InvariantCulture, $" THROW {RowCountChangedError}, 'AuditLog archive is stale: the partition row count changed after it was archived.', 1;");
|
||||
sql.AppendLine(" DECLARE @guardHeads TABLE (ChainMonth date NOT NULL PRIMARY KEY, LastSeq bigint NOT NULL);");
|
||||
foreach (var (month, lastSeq) in ExpectedHeads)
|
||||
{
|
||||
sql.AppendLine(CultureInfo.InvariantCulture,
|
||||
$" INSERT INTO @guardHeads (ChainMonth, LastSeq) VALUES ('{month:yyyy-MM-dd}', {lastSeq});");
|
||||
}
|
||||
|
||||
sql.AppendLine(" IF EXISTS (");
|
||||
sql.AppendLine(" SELECT h.ChainMonth, h.LastSeq FROM dbo.AuditLogChainHead h");
|
||||
sql.AppendLine(" WHERE $partition.pf_AuditLog_Month(CAST(h.ChainMonth AS datetime2(7))) = @guardPartition");
|
||||
sql.AppendLine(" EXCEPT SELECT ChainMonth, LastSeq FROM @guardHeads)");
|
||||
sql.AppendLine(" OR EXISTS (");
|
||||
sql.AppendLine(" SELECT ChainMonth, LastSeq FROM @guardHeads");
|
||||
sql.AppendLine(" EXCEPT SELECT h.ChainMonth, h.LastSeq FROM dbo.AuditLogChainHead h");
|
||||
sql.AppendLine(" WHERE $partition.pf_AuditLog_Month(CAST(h.ChainMonth AS datetime2(7))) = @guardPartition)");
|
||||
sql.AppendLine(CultureInfo.InvariantCulture, $" THROW {ChainAdvancedError}, 'AuditLog archive is stale: the hash chain advanced after it was archived.', 1;");
|
||||
return sql.ToString();
|
||||
}
|
||||
}
|
||||
@@ -477,7 +477,24 @@ VALUES
|
||||
};
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<long> SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default)
|
||||
public Task<long> SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default)
|
||||
=> SwitchOutPartitionAsync(monthBoundary, guard: null, commandTimeout, ct);
|
||||
|
||||
/// <summary>
|
||||
/// The partition switch-out, optionally guarded by an archive's expectations
|
||||
/// (ScadaBridge#42). With a <paramref name="guard"/>, the switch transaction — after it
|
||||
/// holds the exclusive chain lock — re-counts the partition and re-reads its months'
|
||||
/// chain heads, and THROWs (error 51002 / 51003, nothing purged) when either differs
|
||||
/// from what the archive captured. That is what makes "archive, then purge" safe
|
||||
/// against a row arriving, a per-channel purge or a chaining pass in between.
|
||||
/// </summary>
|
||||
/// <param name="monthBoundary">The partition's lower boundary.</param>
|
||||
/// <param name="guard">Archive expectations, or null for the unguarded purge.</param>
|
||||
/// <param name="commandTimeout">Maintenance command timeout.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The approximate number of rows discarded.</returns>
|
||||
internal async Task<long> SwitchOutPartitionAsync(
|
||||
DateTime monthBoundary, Maintenance.AuditPartitionSwitchGuard? guard, TimeSpan? commandTimeout, CancellationToken ct)
|
||||
{
|
||||
// The switch batch below runs via
|
||||
// ExecuteSqlRaw with NO EF user-transaction — it carries its own server-side
|
||||
@@ -526,6 +543,11 @@ VALUES
|
||||
WHERE $PARTITION.pf_AuditLog_Month(OccurredAtUtc) =
|
||||
$partition.pf_AuditLog_Month('{monthBoundaryStr}');";
|
||||
|
||||
// ScadaBridge#42 archive guard (empty for the unguarded purge). Runs under the chain
|
||||
// lock, so no chaining pass or per-channel purge can move the partition between the
|
||||
// check and the SWITCH; an INSERT racing it is caught by the next tick's archive.
|
||||
var guardSql = guard is null ? string.Empty : guard.ToSql(monthBoundaryStr);
|
||||
|
||||
var sql = $@"
|
||||
BEGIN TRY
|
||||
BEGIN TRANSACTION;
|
||||
@@ -536,7 +558,7 @@ VALUES
|
||||
DECLARE @chainLock int;
|
||||
EXEC @chainLock = sp_getapplock @Resource = N'{Maintenance.AuditChainLock.Resource}', @LockMode = 'Exclusive', @LockOwner = 'Transaction', @LockTimeout = -1;
|
||||
IF @chainLock < 0 THROW 51000, 'AuditLog hash-chain lock was not granted.', 1;
|
||||
|
||||
{guardSql}
|
||||
-- 1. Defensive cleanup for databases created before
|
||||
-- AlignAuditLogEventIdUniqueness: the migration drops the
|
||||
-- non-aligned UX_AuditLog_EventId, but a database restored from an
|
||||
|
||||
@@ -114,6 +114,15 @@ public static class ServiceCollectionExtensions
|
||||
// AuditLogHashChainService, verified by the VerifyAuditChain management command.
|
||||
services.AddScoped<IAuditLogHashChain, AuditLogHashChain>();
|
||||
|
||||
// AuditLog Parquet cold archive (ScadaBridge#42): written by the purge actor before a
|
||||
// partition switch, listed/verified by the ListAuditArchives / VerifyAuditArchive
|
||||
// management commands. AuditLog:Archive is bound by AddAuditLogCentral; without it the
|
||||
// defaults apply (disabled, no directory).
|
||||
services.AddScoped<IAuditLogArchive>(sp => new AuditLogPartitionArchive(
|
||||
sp.GetRequiredService<ScadaBridgeDbContext>(),
|
||||
sp.GetService<Microsoft.Extensions.Options.IOptions<AuditLogArchiveOptions>>()?.Value ?? new AuditLogArchiveOptions(),
|
||||
sp.GetService<Microsoft.Extensions.Logging.ILogger<AuditLogPartitionArchive>>()));
|
||||
|
||||
// Opt-in config-audit retention (ScadaBridge:ConfigAudit:RetentionDays, default 0 = off).
|
||||
// Options binding to configuration is done by the host; without it the defaults apply
|
||||
// and the purge service is a no-op.
|
||||
|
||||
+1
@@ -12,6 +12,7 @@
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Parquet.Net" />
|
||||
<PackageReference Include="Microsoft.EntityFrameworkCore" />
|
||||
<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" />
|
||||
<PackageReference Include="Microsoft.EntityFrameworkCore.Design">
|
||||
|
||||
@@ -4,6 +4,7 @@ using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using NSubstitute;
|
||||
using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
using ZB.MOM.WW.ScadaBridge.AuditLog.Configuration;
|
||||
using ZB.MOM.WW.Audit;
|
||||
@@ -688,4 +689,107 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
|
||||
|
||||
StopPurgeActor(actor);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Parquet cold archive (ScadaBridge#42)
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
private IActorRef CreateArchivingActor(RecordingRepo repo, IAuditLogArchive archive, bool enabled)
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
services.AddScoped<IAuditLogRepository>(_ => repo);
|
||||
services.AddScoped(_ => archive);
|
||||
services.AddSingleton(Options.Create(new ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance.AuditLogArchiveOptions
|
||||
{
|
||||
Enabled = enabled,
|
||||
Directory = "/archive",
|
||||
}));
|
||||
var sp = services.BuildServiceProvider();
|
||||
return Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
|
||||
sp,
|
||||
Options.Create(FastTickOptions(TimeSpan.FromHours(1))),
|
||||
Options.Create(new AuditLogOptions()),
|
||||
NullLogger<AuditLogPurgeActor>.Instance)));
|
||||
}
|
||||
|
||||
private static AuditArchiveManifest ManifestFor(DateTime boundary) => new(
|
||||
1, $"auditlog-{boundary:yyyy-MM}-20261109T120000000Z", boundary, boundary.ToString("yyyy-MM"),
|
||||
DateTime.UtcNow, AuditArchiveStatus.Archived, null, 7, 7, 0, null, null,
|
||||
"d.parquet", "00", 99, "t.parquet", "00", 0, []);
|
||||
|
||||
/// <summary>
|
||||
/// With archiving enabled every expired partition is archived and then purged by the
|
||||
/// archive's guarded purge — never by the plain, unguarded switch.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ArchiveEnabled_ArchivesEachPartition_ThenPurgesThroughTheArchive()
|
||||
{
|
||||
var boundary = new DateTime(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
var repo = new RecordingRepo { Boundaries = [boundary] };
|
||||
var archive = Substitute.For<IAuditLogArchive>();
|
||||
var manifest = ManifestFor(boundary);
|
||||
archive.ArchivePartitionAsync(boundary, Arg.Any<CancellationToken>()).Returns(manifest);
|
||||
archive.PurgeArchivedPartitionAsync(manifest, Arg.Any<TimeSpan?>(), Arg.Any<CancellationToken>())
|
||||
.Returns(7L);
|
||||
var archived = CreateTestProbe();
|
||||
Sys.EventStream.Subscribe(archived.Ref, typeof(AuditLogArchivedEvent));
|
||||
var purged = SubscribePurged();
|
||||
|
||||
var actor = CreateArchivingActor(repo, archive, enabled: true);
|
||||
actor.Tell(AuditLogPurgeActor.PurgeTick.Instance);
|
||||
|
||||
var a = archived.ExpectMsg<AuditLogArchivedEvent>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal(manifest.Name, a.ArchiveName);
|
||||
Assert.Equal(7, purged.ExpectMsg<AuditLogPurgedEvent>(TimeSpan.FromSeconds(5)).RowsDeleted);
|
||||
Assert.Empty(repo.SwitchedBoundaries);
|
||||
archive.Received(1).PurgeArchivedPartitionAsync(manifest, Arg.Any<TimeSpan?>(), Arg.Any<CancellationToken>());
|
||||
|
||||
StopPurgeActor(actor);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fail closed: a partition that could not be archived is not purged at all, and the
|
||||
/// failure surfaces as a purge failure.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ArchiveFailure_LeavesThePartitionInPlace_AndPublishesAFailure()
|
||||
{
|
||||
var boundary = new DateTime(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
var repo = new RecordingRepo { Boundaries = [boundary] };
|
||||
var archive = Substitute.For<IAuditLogArchive>();
|
||||
archive.ArchivePartitionAsync(boundary, Arg.Any<CancellationToken>())
|
||||
.Returns<Task<AuditArchiveManifest>>(_ => throw new IOException("archive volume full"));
|
||||
var failed = CreateTestProbe();
|
||||
Sys.EventStream.Subscribe(failed.Ref, typeof(AuditLogPurgeFailedEvent));
|
||||
|
||||
var actor = CreateArchivingActor(repo, archive, enabled: true);
|
||||
actor.Tell(AuditLogPurgeActor.PurgeTick.Instance);
|
||||
|
||||
var evt = failed.ExpectMsg<AuditLogPurgeFailedEvent>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal(boundary, evt.MonthBoundary);
|
||||
Assert.Contains("archive volume full", evt.Error);
|
||||
Assert.Empty(repo.SwitchedBoundaries);
|
||||
archive.DidNotReceiveWithAnyArgs().PurgeArchivedPartitionAsync(default!, default, default);
|
||||
|
||||
StopPurgeActor(actor);
|
||||
}
|
||||
|
||||
/// <summary>With archiving configured but disabled, the purge is the plain switch and the archive is untouched.</summary>
|
||||
[Fact]
|
||||
public void ArchiveDisabled_UsesThePlainSwitch()
|
||||
{
|
||||
var boundary = new DateTime(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
var repo = new RecordingRepo { Boundaries = [boundary] };
|
||||
var archive = Substitute.For<IAuditLogArchive>();
|
||||
var purged = SubscribePurged();
|
||||
|
||||
var actor = CreateArchivingActor(repo, archive, enabled: false);
|
||||
actor.Tell(AuditLogPurgeActor.PurgeTick.Instance);
|
||||
|
||||
purged.ExpectMsg<AuditLogPurgedEvent>(TimeSpan.FromSeconds(5));
|
||||
Assert.Equal([boundary], repo.SwitchedBoundaries);
|
||||
archive.DidNotReceiveWithAnyArgs().ArchivePartitionAsync(default, default);
|
||||
|
||||
StopPurgeActor(actor);
|
||||
}
|
||||
}
|
||||
|
||||
+164
@@ -0,0 +1,164 @@
|
||||
using Parquet;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Maintenance;
|
||||
|
||||
/// <summary>
|
||||
/// Storage-free tests for the AuditLog archive Parquet layout (ScadaBridge#42): rows and
|
||||
/// tombstones round-trip exactly (tick-precise timestamps, NULL vs empty strings, unicode,
|
||||
/// the chain columns), the archived rows re-verify against the archived chain head, and an
|
||||
/// edit made inside the Parquet file breaks that verification.
|
||||
/// </summary>
|
||||
public class AuditArchiveParquetTests
|
||||
{
|
||||
private static readonly DateTime Month = new(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
/// <summary>Rows written to the archive layout read back equal, column for column, across row groups.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Rows_RoundTrip_Exactly_AcrossRowGroups()
|
||||
{
|
||||
var (rows, _) = BuildChain(2_503);
|
||||
rows.Add(Row(9_999) with { ChainSeq = null, RowHash = null }); // an unchained row
|
||||
|
||||
var bytes = await WriteAsync(rows, rowGroupSize: 1_000);
|
||||
var back = await ReadAsync(bytes);
|
||||
|
||||
Assert.Equal(rows.Select(Key), back.Select(Key));
|
||||
await using var reader = await ParquetReader.CreateAsync(new MemoryStream(bytes));
|
||||
Assert.Equal(3, reader.RowGroupCount);
|
||||
Assert.Equal("1", reader.CustomMetadata[AuditArchiveParquet.FormatVersionKey]);
|
||||
Assert.Equal("2025-10", reader.CustomMetadata[AuditArchiveParquet.PartitionKey]);
|
||||
}
|
||||
|
||||
/// <summary>Tombstones round-trip exactly.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Tombstones_RoundTrip_Exactly()
|
||||
{
|
||||
var tombstones = Enumerable.Range(1, 5).Select(i => new AuditArchiveTombstone(
|
||||
Month, i * 3, Enumerable.Repeat((byte)i, 32).ToArray(), Guid.NewGuid(),
|
||||
new DateTime(2026, 1, 2, 3, 4, 5, DateTimeKind.Utc).AddMilliseconds(i))).ToList();
|
||||
|
||||
var stream = new MemoryStream();
|
||||
await using (var writer = await AuditArchiveParquet.CreateTombstoneWriterAsync(stream, "2025-10", 2))
|
||||
{
|
||||
foreach (var t in tombstones)
|
||||
{
|
||||
await writer.AddAsync(t);
|
||||
}
|
||||
}
|
||||
|
||||
var back = new List<AuditArchiveTombstone>();
|
||||
await foreach (var t in AuditArchiveParquet.ReadTombstonesAsync(new MemoryStream(stream.ToArray())))
|
||||
{
|
||||
back.Add(t);
|
||||
}
|
||||
|
||||
Assert.Equal(
|
||||
tombstones.Select(t => (t.ChainMonth, t.ChainSeq, Convert.ToHexString(t.RowHash), t.EventId, t.PurgedAtUtc)),
|
||||
back.Select(t => (t.ChainMonth, t.ChainSeq, Convert.ToHexString(t.RowHash), t.EventId, t.PurgedAtUtc)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The rows read back from an archive re-verify against the archived head — the same
|
||||
/// check <c>audit archive verify</c> makes long after the database dropped them.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ArchivedRows_ReverifyAgainstTheArchivedHead()
|
||||
{
|
||||
var (rows, head) = BuildChain(50);
|
||||
var verifier = new AuditChainVerifier(Month, head, new Dictionary<long, byte[]>());
|
||||
foreach (var row in await ReadAsync(await WriteAsync(rows, 16)))
|
||||
{
|
||||
verifier.Accept(row);
|
||||
}
|
||||
|
||||
var result = verifier.Complete(0);
|
||||
Assert.Equal(AuditChainStatus.Intact, result.Status);
|
||||
Assert.Equal(50, result.RowsVerified);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Positive control for the test above: a row edited inside the archive (re-written
|
||||
/// file, as a forger would) fails re-verification at that row.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task RowEditedInsideTheArchive_FailsReverification()
|
||||
{
|
||||
var (rows, head) = BuildChain(50);
|
||||
rows[17] = rows[17] with { DetailsJson = "{\"kind\":\"ApiCall\",\"forged\":true}" };
|
||||
|
||||
var verifier = new AuditChainVerifier(Month, head, new Dictionary<long, byte[]>());
|
||||
foreach (var row in await ReadAsync(await WriteAsync(rows, 16)))
|
||||
{
|
||||
verifier.Accept(row);
|
||||
}
|
||||
|
||||
var result = verifier.Complete(0);
|
||||
Assert.Equal(AuditChainStatus.Broken, result.Status);
|
||||
Assert.Equal(AuditChainBreakKind.HashMismatch, result.FirstBreak!.Kind);
|
||||
Assert.Equal(18, result.FirstBreak.ChainSeq);
|
||||
}
|
||||
|
||||
internal static (List<AuditChainRow> Rows, AuditChainHeadState Head) BuildChain(int count)
|
||||
{
|
||||
var rows = new List<AuditChainRow>();
|
||||
var previous = AuditChainHasher.GenesisHash.ToArray();
|
||||
for (var i = 1; i <= count; i++)
|
||||
{
|
||||
var row = Row(i);
|
||||
var hash = AuditChainHasher.ComputeRowHash(previous, Month, i, row);
|
||||
rows.Add(row with { ChainSeq = i, RowHash = hash });
|
||||
previous = hash;
|
||||
}
|
||||
|
||||
return (rows, new AuditChainHeadState(count, previous, null, null));
|
||||
}
|
||||
|
||||
private static AuditChainRow Row(int i) => new(
|
||||
Guid.NewGuid(),
|
||||
// Odd ticks on purpose: the archive must keep datetime2(7) precision the hash covers.
|
||||
Month.AddMinutes(i).AddTicks(1_234_567 + i),
|
||||
i % 5 == 0 ? null : i % 7 == 0 ? string.Empty : "api-key-" + i,
|
||||
"ApiOutbound.ApiCall",
|
||||
i % 3 == 0 ? "Failure" : "Success",
|
||||
"ApiOutbound",
|
||||
i % 4 == 0 ? null : "erp-ü-日本",
|
||||
i % 6 == 0 ? null : "central-a",
|
||||
i % 2 == 0 ? Guid.NewGuid() : null,
|
||||
i % 8 == 0 ? null : $"{{\"kind\":\"ApiCall\",\"status\":\"Delivered\",\"n\":{i}}}");
|
||||
|
||||
private static string Key(AuditChainRow r) => string.Join('|',
|
||||
r.EventId, r.OccurredAtUtc.Ticks, r.OccurredAtUtc.Kind, r.Actor ?? "<null>", r.Action, r.Outcome, r.Category,
|
||||
r.Target ?? "<null>", r.SourceNode ?? "<null>", r.CorrelationId?.ToString() ?? "<null>", r.DetailsJson ?? "<null>",
|
||||
r.ChainSeq?.ToString() ?? "<null>", r.RowHash is null ? "<null>" : Convert.ToHexString(r.RowHash));
|
||||
|
||||
private static async Task<byte[]> WriteAsync(IEnumerable<AuditChainRow> rows, int rowGroupSize)
|
||||
{
|
||||
var stream = new MemoryStream();
|
||||
await using (var writer = await AuditArchiveParquet.CreateDataWriterAsync(stream, "2025-10", rowGroupSize))
|
||||
{
|
||||
foreach (var row in rows)
|
||||
{
|
||||
await writer.AddAsync(row);
|
||||
}
|
||||
}
|
||||
|
||||
return stream.ToArray();
|
||||
}
|
||||
|
||||
private static async Task<List<AuditChainRow>> ReadAsync(byte[] bytes)
|
||||
{
|
||||
var rows = new List<AuditChainRow>();
|
||||
await foreach (var row in AuditArchiveParquet.ReadRowsAsync(new MemoryStream(bytes)))
|
||||
{
|
||||
rows.Add(row);
|
||||
}
|
||||
|
||||
return rows;
|
||||
}
|
||||
}
|
||||
+420
@@ -0,0 +1,420 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using ZB.MOM.WW.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Migrations;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Maintenance;
|
||||
|
||||
/// <summary>
|
||||
/// SQL Server integration tests for the AuditLog Parquet archive (ScadaBridge#42) against the
|
||||
/// real migrated, partitioned schema: archive-then-purge leaves the month's hash chain
|
||||
/// verifiable and the archive re-verifiable against the purge anchor; the purge refuses to
|
||||
/// drop a partition that changed after it was archived; tampering with an archive is caught.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Every test owns a distinct month (partition) of the shared fixture database; the chain
|
||||
/// and the per-channel purge are table-wide, so each test drains the chain itself.
|
||||
/// </remarks>
|
||||
public sealed class AuditLogPartitionArchiveTests : IClassFixture<MsSqlMigrationFixture>, IDisposable
|
||||
{
|
||||
private readonly MsSqlMigrationFixture _fixture;
|
||||
private readonly DirectoryInfo _root = Directory.CreateTempSubdirectory("auditlog-archive-");
|
||||
|
||||
/// <summary>Creates the test class with the shared migrated SQL Server fixture.</summary>
|
||||
/// <param name="fixture">The shared migration fixture.</param>
|
||||
public AuditLogPartitionArchiveTests(MsSqlMigrationFixture fixture) => _fixture = fixture;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose() => _root.Delete(recursive: true);
|
||||
|
||||
/// <summary>
|
||||
/// Archive, then purge: the archive holds exactly the partition's rows (chained,
|
||||
/// unchained, with their chain columns) and its tombstones; after the switch the month's
|
||||
/// chain still verifies from the anchor, and the archive verifies against that anchor.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task ArchiveThenPurge_KeepsTheChainVerifiable_AndTheArchiveMatchesTheAnchor()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
var month = Month(2026, 2);
|
||||
await InsertAsync(Enumerable.Range(1, 10).Select(i => NewEvent(month.AddDays(i).AddTicks(i * 7_777))));
|
||||
// Two rows of a short-retention channel, purged per channel mid-chain → tombstones.
|
||||
await InsertAsync([NewEvent(month.AddHours(1), AuditChannel.DbOutbound), NewEvent(month.AddHours(2), AuditChannel.DbOutbound)]);
|
||||
await DrainAsync();
|
||||
await using (var ctx = CreateContext())
|
||||
{
|
||||
Assert.Equal(2, await new AuditLogRepository(ctx).PurgeChannelOlderThanAsync("DbOutbound", month.AddDays(1), 100));
|
||||
}
|
||||
|
||||
await InsertAsync([NewEvent(month.AddDays(20))]); // written after the drain: unchained
|
||||
var before = await PartitionRowsAsync(month);
|
||||
Assert.Equal(11, before.Count);
|
||||
|
||||
var manifest = await ArchiveAsync(month);
|
||||
Assert.Equal(AuditArchiveStatus.Archived, manifest.Status);
|
||||
Assert.Equal(11, manifest.RowCount);
|
||||
Assert.Equal(10, manifest.ChainedRowCount);
|
||||
Assert.Equal(1, manifest.UnchainedRowCount);
|
||||
Assert.Equal(2, manifest.TombstoneCount);
|
||||
var chain = Assert.Single(manifest.Chains);
|
||||
Assert.Equal("2026-02", chain.Partition);
|
||||
Assert.Equal(12, chain.LastSeq);
|
||||
|
||||
// Row equality: the archive's data file holds exactly the partition's stored rows.
|
||||
var archived = await ReadArchiveRowsAsync(manifest);
|
||||
Assert.Equal(before.Select(Key).Order(), archived.Select(Key).Order());
|
||||
|
||||
await using (var ctx = CreateContext())
|
||||
{
|
||||
Assert.Equal(11, await Archive(ctx).PurgeArchivedPartitionAsync(manifest));
|
||||
}
|
||||
|
||||
Assert.Empty(await PartitionRowsAsync(month));
|
||||
var monthChain = await VerifyMonthAsync(month);
|
||||
Assert.Equal(AuditChainStatus.Intact, monthChain.Status);
|
||||
Assert.Equal(12, monthChain.PurgedThroughSeq);
|
||||
|
||||
var verification = await VerifyArchiveAsync(manifest.Name);
|
||||
Assert.True(verification.Verified, verification.FileProblem);
|
||||
Assert.Equal(AuditArchiveStatus.Purged, verification.ManifestStatus);
|
||||
Assert.Equal(11, verification.RowsRead);
|
||||
var v = Assert.Single(verification.Chains);
|
||||
Assert.Equal(AuditChainStatus.Intact, v.Chain.Status);
|
||||
Assert.Equal(10, v.Chain.RowsVerified);
|
||||
Assert.Equal(2, v.Chain.TombstonesApplied);
|
||||
Assert.Equal(1, v.Chain.UnchainedRows);
|
||||
Assert.Equal(AuditArchiveAnchorStatus.Matches, v.Anchor);
|
||||
|
||||
await using (var ctx = CreateContext())
|
||||
{
|
||||
var listed = Assert.Single(await Archive(ctx).ListAsync(), m => m.Partition == "2026-02");
|
||||
Assert.Equal(AuditArchiveStatus.Purged, listed.Status);
|
||||
Assert.NotNull(listed.PurgedAtUtc);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A row that lands in the partition after it was archived makes the purge refuse
|
||||
/// (nothing dropped); a fresh archive then purges. The stale archive still verifies as a
|
||||
/// complete, un-purged copy.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task RowArrivingAfterTheArchive_StopsThePurge()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
var month = Month(2026, 3);
|
||||
await InsertAsync(Enumerable.Range(1, 5).Select(i => NewEvent(month.AddDays(i))));
|
||||
await DrainAsync();
|
||||
var stale = await ArchiveAsync(month);
|
||||
|
||||
await InsertAsync([NewEvent(month.AddDays(9))]);
|
||||
await using (var ctx = CreateContext())
|
||||
{
|
||||
var ex = await Assert.ThrowsAsync<AuditArchiveStaleException>(() => Archive(ctx).PurgeArchivedPartitionAsync(stale));
|
||||
Assert.Equal(AuditPartitionSwitchGuard.RowCountChangedError, ((SqlException)ex.InnerException!).Number);
|
||||
}
|
||||
|
||||
Assert.Equal(6, (await PartitionRowsAsync(month)).Count);
|
||||
Assert.Null((await VerifyMonthAsync(month)).PurgedThroughSeq);
|
||||
|
||||
var staleVerification = await VerifyArchiveAsync(stale.Name);
|
||||
Assert.True(staleVerification.Verified);
|
||||
Assert.Equal(AuditArchiveStatus.Archived, staleVerification.ManifestStatus);
|
||||
Assert.Equal(AuditArchiveAnchorStatus.NotPurged, Assert.Single(staleVerification.Chains).Anchor);
|
||||
|
||||
// Positive control: a fresh archive holds the new row and is allowed to purge.
|
||||
var fresh = await ArchiveAsync(month);
|
||||
Assert.Equal(6, fresh.RowCount);
|
||||
await using (var ctx = CreateContext())
|
||||
{
|
||||
Assert.Equal(6, await Archive(ctx).PurgeArchivedPartitionAsync(fresh));
|
||||
}
|
||||
|
||||
Assert.Empty(await PartitionRowsAsync(month));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A chaining pass between the archive and the purge (the row count is unchanged, the
|
||||
/// chain head moved) also stops the purge: otherwise the anchor would cover links the
|
||||
/// archive does not hold.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task ChainAdvancingAfterTheArchive_StopsThePurge()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
var month = Month(2026, 4);
|
||||
await InsertAsync(Enumerable.Range(1, 4).Select(i => NewEvent(month.AddDays(i))));
|
||||
await DrainAsync();
|
||||
await InsertAsync([NewEvent(month.AddDays(8))]); // unchained at archive time
|
||||
var manifest = await ArchiveAsync(month);
|
||||
Assert.Equal(1, manifest.UnchainedRowCount);
|
||||
|
||||
await DrainAsync(); // links the row: head 4 → 5, row count unchanged
|
||||
await using (var ctx = CreateContext())
|
||||
{
|
||||
var ex = await Assert.ThrowsAsync<AuditArchiveStaleException>(() => Archive(ctx).PurgeArchivedPartitionAsync(manifest));
|
||||
Assert.Equal(AuditPartitionSwitchGuard.ChainAdvancedError, ((SqlException)ex.InnerException!).Number);
|
||||
}
|
||||
|
||||
Assert.Equal(5, (await PartitionRowsAsync(month)).Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tampering with a purged archive is caught three ways: a flipped byte (file hash), a
|
||||
/// re-written file with an edited row and a re-hashed manifest (chain), and a forger who
|
||||
/// also recomputes the whole chain (the purge anchor in the database no longer matches).
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task TamperedArchive_FailsVerification()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
var month = Month(2026, 5);
|
||||
await InsertAsync(Enumerable.Range(1, 8).Select(i => NewEvent(month.AddDays(i))));
|
||||
await DrainAsync();
|
||||
var manifest = await ArchiveAndPurgeAsync(month);
|
||||
Assert.True((await VerifyArchiveAsync(manifest.Name)).Verified);
|
||||
|
||||
var dataPath = Path.Combine(_root.FullName, "2026", manifest.DataFile);
|
||||
var manifestPath = Path.Combine(_root.FullName, "2026", manifest.Name + ".manifest.json");
|
||||
var original = await File.ReadAllBytesAsync(dataPath);
|
||||
var originalManifest = await File.ReadAllTextAsync(manifestPath);
|
||||
|
||||
// 1. A flipped byte.
|
||||
var flipped = (byte[])original.Clone();
|
||||
flipped[original.Length / 2] ^= 0xFF;
|
||||
await File.WriteAllBytesAsync(dataPath, flipped);
|
||||
var result = await VerifyArchiveAsync(manifest.Name);
|
||||
Assert.False(result.Verified);
|
||||
Assert.False(result.FilesIntact);
|
||||
Assert.Contains("SHA-256", result.FileProblem);
|
||||
|
||||
// 2. An edited row in a re-written file, manifest hash/size updated to match.
|
||||
var rows = await ReadArchiveRowsAsync(manifest, original);
|
||||
var victim = rows[3];
|
||||
rows[3] = victim with { Target = "forged" };
|
||||
await RewriteAsync(rows, manifest, dataPath, manifestPath, lastHash: null);
|
||||
result = await VerifyArchiveAsync(manifest.Name);
|
||||
Assert.False(result.Verified);
|
||||
Assert.True(result.FilesIntact);
|
||||
var brk = Assert.Single(result.Chains).Chain.FirstBreak!;
|
||||
Assert.Equal(AuditChainBreakKind.HashMismatch, brk.Kind);
|
||||
Assert.Equal(victim.EventId, brk.EventId);
|
||||
|
||||
// 3. The forger recomputes every link and the manifest's chain end: the archive is
|
||||
// self-consistent, but its end no longer matches the anchor the purge left.
|
||||
var previous = AuditChainHasher.GenesisHash.ToArray();
|
||||
for (var i = 0; i < rows.Count; i++)
|
||||
{
|
||||
var hash = AuditChainHasher.ComputeRowHash(previous, month, rows[i].ChainSeq!.Value, rows[i]);
|
||||
rows[i] = rows[i] with { RowHash = hash };
|
||||
previous = hash;
|
||||
}
|
||||
|
||||
await RewriteAsync(rows, manifest, dataPath, manifestPath, lastHash: Convert.ToHexStringLower(previous));
|
||||
result = await VerifyArchiveAsync(manifest.Name);
|
||||
Assert.False(result.Verified);
|
||||
var c = Assert.Single(result.Chains);
|
||||
Assert.Equal(AuditChainStatus.Intact, c.Chain.Status);
|
||||
Assert.Equal(AuditArchiveAnchorStatus.Mismatch, c.Anchor);
|
||||
|
||||
// Positive control: the untouched archive still verifies.
|
||||
await File.WriteAllBytesAsync(dataPath, original);
|
||||
await File.WriteAllTextAsync(manifestPath, originalManifest);
|
||||
Assert.True((await VerifyArchiveAsync(manifest.Name)).Verified);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Rows that arrive for a month after it was purged chain on from the anchor; archiving
|
||||
/// and purging them again produces a second archive that resumes from the first one's end.
|
||||
/// The first archive then reports its anchor as superseded, and both verify.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task LateRows_AreArchivedAgain_FromTheAnchor()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
var month = Month(2026, 6);
|
||||
await InsertAsync(Enumerable.Range(1, 3).Select(i => NewEvent(month.AddDays(i))));
|
||||
await DrainAsync();
|
||||
var first = await ArchiveAndPurgeAsync(month);
|
||||
|
||||
await InsertAsync([NewEvent(month.AddDays(15)), NewEvent(month.AddDays(16))]);
|
||||
await DrainAsync();
|
||||
await Task.Delay(5); // distinct archive name (millisecond stamp)
|
||||
var second = await ArchiveAndPurgeAsync(month);
|
||||
var chain = Assert.Single(second.Chains);
|
||||
Assert.Equal(3, chain.PriorPurgedThroughSeq);
|
||||
Assert.Equal(5, chain.LastSeq);
|
||||
|
||||
var secondResult = await VerifyArchiveAsync(second.Name);
|
||||
Assert.True(secondResult.Verified);
|
||||
Assert.Equal(AuditArchiveAnchorStatus.Matches, Assert.Single(secondResult.Chains).Anchor);
|
||||
Assert.Equal(2, Assert.Single(secondResult.Chains).Chain.RowsVerified);
|
||||
|
||||
var firstResult = await VerifyArchiveAsync(first.Name);
|
||||
Assert.True(firstResult.Verified);
|
||||
Assert.Equal(AuditArchiveAnchorStatus.Superseded, Assert.Single(firstResult.Chains).Anchor);
|
||||
}
|
||||
|
||||
/// <summary>A name that is not an archive name (e.g. a path) is rejected before any file access.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task VerifyRejectsPathLikeNames()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
await using var ctx = CreateContext();
|
||||
await Assert.ThrowsAsync<ArgumentException>(() => Archive(ctx).VerifyAsync("../../etc/passwd"));
|
||||
await Assert.ThrowsAsync<FileNotFoundException>(() => Archive(ctx).VerifyAsync("auditlog-2026-01-20260101T000000000Z"));
|
||||
}
|
||||
|
||||
// ── helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
private static DateTime Month(int year, int month) => new(year, month, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private AuditLogPartitionArchive Archive(ScadaBridgeDbContext ctx) => new(ctx, new AuditLogArchiveOptions
|
||||
{
|
||||
Enabled = true,
|
||||
Directory = _root.FullName,
|
||||
RowGroupSize = 1_000,
|
||||
});
|
||||
|
||||
private async Task<AuditArchiveManifest> ArchiveAsync(DateTime month)
|
||||
{
|
||||
await using var ctx = CreateContext();
|
||||
return await Archive(ctx).ArchivePartitionAsync(month);
|
||||
}
|
||||
|
||||
private async Task<AuditArchiveManifest> ArchiveAndPurgeAsync(DateTime month)
|
||||
{
|
||||
var manifest = await ArchiveAsync(month);
|
||||
await using var ctx = CreateContext();
|
||||
await Archive(ctx).PurgeArchivedPartitionAsync(manifest);
|
||||
return manifest;
|
||||
}
|
||||
|
||||
private async Task<AuditArchiveVerificationResult> VerifyArchiveAsync(string name)
|
||||
{
|
||||
await using var ctx = CreateContext();
|
||||
return await Archive(ctx).VerifyAsync(name);
|
||||
}
|
||||
|
||||
private async Task<AuditChainVerificationResult> VerifyMonthAsync(DateTime month)
|
||||
{
|
||||
await using var ctx = CreateContext();
|
||||
return await new AuditLogHashChain(ctx).VerifyMonthAsync(month);
|
||||
}
|
||||
|
||||
private async Task<List<AuditChainRow>> ReadArchiveRowsAsync(AuditArchiveManifest manifest, byte[]? bytes = null)
|
||||
{
|
||||
bytes ??= await File.ReadAllBytesAsync(Path.Combine(_root.FullName, manifest.Partition[..4], manifest.DataFile));
|
||||
var rows = new List<AuditChainRow>();
|
||||
await foreach (var row in AuditArchiveParquet.ReadRowsAsync(new MemoryStream(bytes)))
|
||||
{
|
||||
rows.Add(row);
|
||||
}
|
||||
|
||||
return rows;
|
||||
}
|
||||
|
||||
private static async Task RewriteAsync(
|
||||
List<AuditChainRow> rows, AuditArchiveManifest manifest, string dataPath, string manifestPath, string? lastHash)
|
||||
{
|
||||
var stream = new MemoryStream();
|
||||
await using (var writer = await AuditArchiveParquet.CreateDataWriterAsync(stream, manifest.Partition, 1_000))
|
||||
{
|
||||
foreach (var row in rows)
|
||||
{
|
||||
await writer.AddAsync(row);
|
||||
}
|
||||
}
|
||||
|
||||
var bytes = stream.ToArray();
|
||||
await File.WriteAllBytesAsync(dataPath, bytes);
|
||||
var forged = manifest with
|
||||
{
|
||||
Status = AuditArchiveStatus.Purged,
|
||||
DataSha256 = Convert.ToHexStringLower(System.Security.Cryptography.SHA256.HashData(bytes)),
|
||||
DataBytes = bytes.Length,
|
||||
Chains = lastHash is null ? manifest.Chains : [manifest.Chains[0] with { LastHash = lastHash }],
|
||||
};
|
||||
await File.WriteAllTextAsync(manifestPath, JsonSerializer.Serialize(forged, new JsonSerializerOptions
|
||||
{
|
||||
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
|
||||
}));
|
||||
}
|
||||
|
||||
private async Task<List<AuditChainRow>> PartitionRowsAsync(DateTime month)
|
||||
{
|
||||
await using var conn = new SqlConnection(_fixture.ConnectionString);
|
||||
await conn.OpenAsync();
|
||||
await using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = $"SELECT {AuditLogHashChain.RowColumns}, ChainSeq, RowHash FROM dbo.AuditLog "
|
||||
+ "WHERE OccurredAtUtc >= @s AND OccurredAtUtc < @e;";
|
||||
cmd.Parameters.Add(new SqlParameter("@s", System.Data.SqlDbType.DateTime2) { Value = month });
|
||||
cmd.Parameters.Add(new SqlParameter("@e", System.Data.SqlDbType.DateTime2) { Value = month.AddMonths(1) });
|
||||
var rows = new List<AuditChainRow>();
|
||||
await using var reader = await cmd.ExecuteReaderAsync();
|
||||
while (await reader.ReadAsync())
|
||||
{
|
||||
rows.Add(AuditLogHashChain.ReadRow(reader, withLink: true));
|
||||
}
|
||||
|
||||
return rows;
|
||||
}
|
||||
|
||||
private static string Key(AuditChainRow r) => string.Join('|',
|
||||
r.EventId, r.OccurredAtUtc.Ticks, r.Actor ?? "<null>", r.Action, r.Outcome, r.Category, r.Target ?? "<null>",
|
||||
r.SourceNode ?? "<null>", r.CorrelationId?.ToString() ?? "<null>", r.DetailsJson ?? "<null>",
|
||||
r.ChainSeq?.ToString() ?? "<null>", r.RowHash is null ? "<null>" : Convert.ToHexString(r.RowHash));
|
||||
|
||||
private async Task InsertAsync(IEnumerable<AuditEvent> events)
|
||||
{
|
||||
await using var ctx = CreateContext();
|
||||
var repo = new AuditLogRepository(ctx);
|
||||
foreach (var e in events)
|
||||
{
|
||||
await repo.InsertIfNotExistsAsync(e);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task DrainAsync()
|
||||
{
|
||||
await using var ctx = CreateContext();
|
||||
var chain = new AuditLogHashChain(ctx);
|
||||
for (var i = 0; i < 1000; i++)
|
||||
{
|
||||
var r = await chain.ChainPendingAsync(500);
|
||||
if (r.RowsChained == 0 && !r.LockBusy)
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
throw new InvalidOperationException("Chain did not drain.");
|
||||
}
|
||||
|
||||
private ScadaBridgeDbContext CreateContext()
|
||||
=> new(new DbContextOptionsBuilder<ScadaBridgeDbContext>().UseSqlServer(_fixture.ConnectionString).Options);
|
||||
|
||||
private static AuditEvent NewEvent(DateTime occurredAtUtc, AuditChannel channel = AuditChannel.ApiOutbound) =>
|
||||
ScadaBridgeAuditEventFactory.Create(
|
||||
channel: channel,
|
||||
kind: channel == AuditChannel.DbOutbound ? AuditKind.DbWrite : AuditKind.ApiCall,
|
||||
status: AuditStatus.Delivered,
|
||||
occurredAtUtc: occurredAtUtc,
|
||||
actor: "archive-test",
|
||||
target: "erp",
|
||||
sourceNode: "central-a",
|
||||
sourceSiteId: "archive-test",
|
||||
correlationId: Guid.NewGuid());
|
||||
}
|
||||
Reference in New Issue
Block a user