diff --git a/Directory.Packages.props b/Directory.Packages.props index 3b92baa4..4863f214 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -78,6 +78,9 @@ + + diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogArchiveOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogArchiveOptionsValidator.cs new file mode 100644 index 00000000..fa069df6 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogArchiveOptionsValidator.cs @@ -0,0 +1,36 @@ +using ZB.MOM.WW.Configuration; +using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Validates (section AuditLog:Archive; 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. +/// +public sealed class AuditLogArchiveOptionsValidator : OptionsValidatorBase +{ + /// + 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."); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogArchivedEvent.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogArchivedEvent.cs new file mode 100644 index 00000000..94cc8c65 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogArchivedEvent.cs @@ -0,0 +1,16 @@ +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Published on the actor-system EventStream by when an +/// expired partition was written to a Parquet archive and then switched out +/// (ScadaBridge#42). Followed by the usual . +/// +/// The partition's lower boundary (UTC). +/// The archive name (audit archive verify --name takes it). +/// Rows written to the archive. +/// Size of the archive's data file in bytes. +public sealed record AuditLogArchivedEvent( + DateTime MonthBoundary, + string ArchiveName, + long RowCount, + long DataBytes); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs index 8ef23a1d..ef1fc6f2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs @@ -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; /// , a /// metadata-only staging-table switch (no index /// drop/rebuild brackets it). +/// When AuditLog:Archive:Enabled (ScadaBridge#42, default off), each boundary +/// is first written to a Parquet archive via and then +/// switched out by the archive's guarded purge instead of the plain switch. /// Publishes on the actor-system /// EventStream so the central health collector + ops surfaces /// can subscribe without coupling to this actor. @@ -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>()?.Value; + IAuditLogArchive? archive = null; + if (archiveOptions?.Enabled == true) + { + try + { + archive = scope.ServiceProvider.GetRequiredService(); + } + 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( diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs index 2219173c..4b5db4d9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs @@ -55,6 +55,12 @@ public static class ServiceCollectionExtensions /// Configuration section bound to (ScadaBridge#41). public const string HashChainSectionName = "AuditLog:HashChain"; + /// + /// Configuration section bound to + /// (ScadaBridge#42). + /// + public const string ArchiveSectionName = "AuditLog:Archive"; + /// Configuration section bound to . public const string PurgeSectionName = "AuditLog:Purge"; @@ -405,6 +411,12 @@ public static class ServiceCollectionExtensions config, HashChainSectionName); services.AddHostedService(); + // 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( + config, ArchiveSectionName); + // I1 (review): bind the two central-singleton options HERE rather than in // AddAuditLogCentralReconciliationClient. AkkaHostedService.RegisterCentralActors // resolves IOptions / diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IAuditLogArchive.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IAuditLogArchive.cs new file mode 100644 index 00000000..3fff5186 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IAuditLogArchive.cs @@ -0,0 +1,82 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; + +/// +/// Parquet cold archive of the central AuditLog (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. +/// +/// +/// +/// Archive before purge. writes the partition's +/// rows (with their hash-chain columns), the month's chain head and its per-channel purge +/// tombstones; 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. +/// +/// +/// Hash chain. The switch records each month's chain end as the checkpoint anchor +/// (ScadaBridge#41); the archive records the same chain end, so +/// can recompute the archived chain and tie its end to the anchor still in the database. +/// +/// +public interface IAuditLogArchive +{ + /// True when an archive directory is configured (listing and verifying need one; archiving needs Enabled too). + bool IsConfigured { get; } + + /// True when the retention purge archives partitions before dropping them (AuditLog:Archive:Enabled). + bool IsEnabled { get; } + + /// + /// Writes the partition whose lower boundary is to + /// a new archive (data file, tombstone file, then manifest with status + /// ). + /// + /// The pf_AuditLog_Month lower boundary (day 1, midnight UTC). + /// Cancellation token. + /// The written manifest. + Task ArchivePartitionAsync(DateTime partitionBoundaryUtc, CancellationToken ct = default); + + /// + /// Switches the archived partition out of dbo.AuditLog, guarded against the + /// partition having changed since was written, then marks the + /// manifest . + /// + /// The manifest returned. + /// Maintenance command timeout for the switch. + /// Cancellation token. + /// The approximate number of rows the switch discarded. + /// The partition changed after it was archived; nothing was purged. + Task PurgeArchivedPartitionAsync(AuditArchiveManifest manifest, TimeSpan? commandTimeout = null, CancellationToken ct = default); + + /// Lists every archive manifest under the archive directory, oldest partition first. + /// Cancellation token. + /// The manifests. + Task> ListAsync(CancellationToken ct = default); + + /// + /// 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. + /// + /// The archive name (as reports it). + /// Cancellation token. + /// The verification result. + /// No archive has that name. + Task VerifyAsync(string name, CancellationToken ct = default); +} + +/// +/// Thrown by 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. +/// +public sealed class AuditArchiveStaleException : Exception +{ + /// Initializes a new instance with a message and the SQL error that carried it. + /// The message. + /// The underlying exception. + public AuditArchiveStaleException(string message, Exception? inner = null) : base(message, inner) { } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Audit/AuditArchive.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Audit/AuditArchive.cs new file mode 100644 index 00000000..e3a762d8 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Audit/AuditArchive.cs @@ -0,0 +1,147 @@ +using System.Text.Json.Serialization; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; + +/// +/// Lifecycle of one AuditLog Parquet archive (audit roadmap T2, ScadaBridge#42). +/// +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum AuditArchiveStatus +{ + /// + /// 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. + /// + Archived, + + /// The archived partition was switched out of dbo.AuditLog; the archive is the only copy. + Purged, +} + +/// +/// 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 AuditChainVerifier long after the database dropped them. +/// +/// The chain partition, YYYY-MM. +/// The head's last sequence number at archive time. +/// The head's last hash at archive time (lowercase hex). +/// The checkpoint anchor an earlier purge of this month left, or null. +/// The hash at (lowercase hex), or null. +/// Per-channel purge tombstones of this month carried in the archive. +public sealed record AuditArchiveChainState( + string Partition, + long LastSeq, + string LastHash, + long? PriorPurgedThroughSeq, + string? PriorPurgedThroughHash, + long TombstoneCount); + +/// +/// 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. +/// +/// Manifest/file layout version (currently 1). +/// Archive name, auditlog-YYYY-MM-yyyyMMddTHHmmssfffZ; the file stem of all three files. +/// The pf_AuditLog_Month lower boundary of the archived partition. +/// The archived partition, YYYY-MM. +/// When the archive was written. +/// Whether the partition switch that follows the archive has been confirmed. +/// When the partition switch completed, or null. +/// Rows in the data file. +/// Rows that carried a ChainSeq/RowHash. +/// Rows the chaining job had not reached (not covered by the chain). +/// Earliest OccurredAtUtc in the data file, or null when empty. +/// Latest OccurredAtUtc in the data file, or null when empty. +/// Data file name, relative to the manifest's directory. +/// SHA-256 of the data file (lowercase hex). +/// Size of the data file in bytes. +/// Tombstone file name, relative to the manifest's directory. +/// SHA-256 of the tombstone file (lowercase hex). +/// Tombstones in the tombstone file. +/// One entry per chain partition that had a head when the rows were archived. +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 Chains); + +/// +/// How an archive's recorded chain end relates to the checkpoint anchor the partition purge +/// left on the database's chain head. +/// +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum AuditArchiveAnchorStatus +{ + /// The database anchor is exactly the archive's chain end (sequence and hash). + Matches, + + /// + /// 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. + /// + Superseded, + + /// The partition has not been purged: the archive's rows are still in dbo.AuditLog. + NotPurged, + + /// + /// 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. + /// + Mismatch, +} + +/// Verification of one chain partition inside an archive. +/// The chain recomputed over the archived rows, the archived head and tombstones. +/// How the archived chain end relates to the database's checkpoint anchor. +/// Human-readable explanation of . +public sealed record AuditArchiveChainVerification( + AuditChainVerificationResult Chain, + AuditArchiveAnchorStatus Anchor, + string AnchorDetail); + +/// +/// Result of audit archive verify (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. +/// +/// The archive verified. +/// The archived partition, YYYY-MM. +/// The manifest's lifecycle status. +/// True when the files are intact, no chain is broken and no anchor mismatches. +/// True when both files' size and SHA-256 and the row/tombstone counts match the manifest. +/// Why is false, or null. +/// Rows read from the data file. +/// Per chain partition verification. +public sealed record AuditArchiveVerificationResult( + string Name, + string Partition, + AuditArchiveStatus ManifestStatus, + bool Verified, + bool FilesIntact, + string? FileProblem, + long RowsRead, + IReadOnlyList Chains); + +/// Result of audit archive list (ScadaBridge#42). +/// Whether the retention purge is archiving partitions (AuditLog:Archive:Enabled). +/// Every archive manifest, oldest partition first. +public sealed record AuditArchiveListResult(bool Enabled, IReadOnlyList Archives); diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditArchiveParquet.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditArchiveParquet.cs new file mode 100644 index 00000000..8d4ac259 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditArchiveParquet.cs @@ -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; + +/// +/// 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. +/// +/// The chain partition's first instant (UTC). +/// The purged row's sequence number. +/// The purged row's 32-byte hash. +/// The purged row's EventId. +/// When the per-channel purge removed it. +public sealed record AuditArchiveTombstone( + DateTime ChainMonth, + long ChainSeq, + byte[] RowHash, + Guid EventId, + DateTime PurgedAtUtc); + +/// +/// The Parquet layout of an AuditLog archive (audit roadmap T2, ScadaBridge#42), format v1. +/// +/// +/// +/// Data file — one row per archived dbo.AuditLog row, the ten stored (writable) +/// columns plus the two hash-chain columns, so a row can be re-hashed with +/// exactly as the database chained it: +/// EventId (string, D format), OccurredAtUtc (timestamp, UTC, microseconds — +/// for analytics), OccurredAtUtcTicks (int64 — the exact datetime2(7) value the hash +/// covers), Actor, Action, Outcome, Category, Target, +/// SourceNode, CorrelationId, DetailsJson (strings), ChainSeq (int64, +/// null when unchained) and RowHash (binary, null when unchained). The persisted computed +/// columns (Kind, Status, SourceSiteId, …) are not stored: they are derived from +/// DetailsJson, which the hash covers. +/// +/// +/// Tombstone file — ChainMonth (YYYY-MM), ChainSeq, RowHash, +/// EventId, PurgedAtUtc. +/// +/// +/// Both are plain Parquet: DuckDB (SELECT * FROM 'auditlog-….parquet'), pandas/pyarrow +/// and Spark read them without any ScadaBridge code. +/// +/// +public static class AuditArchiveParquet +{ + /// The archive layout version written into the file metadata and the manifest. + public const int FormatVersion = 1; + + /// Key-value metadata key carrying . + public const string FormatVersionKey = "scadabridge.auditlog.archive.formatVersion"; + + /// Key-value metadata key carrying the archived partition (YYYY-MM). + public const string PartitionKey = "scadabridge.auditlog.archive.partition"; + + private static readonly DataField EventIdField = new("EventId", false); + private static readonly DateTimeDataField OccurredAtField = + new("OccurredAtUtc", DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: false); + private static readonly DataField OccurredAtTicksField = new("OccurredAtUtcTicks", false); + private static readonly DataField ActorField = new("Actor", true); + private static readonly DataField ActionField = new("Action", false); + private static readonly DataField OutcomeField = new("Outcome", false); + private static readonly DataField CategoryField = new("Category", false); + private static readonly DataField TargetField = new("Target", true); + private static readonly DataField SourceNodeField = new("SourceNode", true); + private static readonly DataField CorrelationIdField = new("CorrelationId", true); + private static readonly DataField DetailsJsonField = new("DetailsJson", true); + private static readonly DataField ChainSeqField = new("ChainSeq", true); + private static readonly DataField RowHashField = new("RowHash", true); + + private static readonly DataField TombstoneMonthField = new("ChainMonth", false); + private static readonly DataField TombstoneSeqField = new("ChainSeq", false); + private static readonly DataField TombstoneHashField = new("RowHash", false); + private static readonly DataField TombstoneEventIdField = new("EventId", false); + private static readonly DateTimeDataField TombstonePurgedAtField = + new("PurgedAtUtc", DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: false); + + /// The data-file schema. + public static ParquetSchema DataSchema { get; } = new( + EventIdField, OccurredAtField, OccurredAtTicksField, ActorField, ActionField, OutcomeField, CategoryField, + TargetField, SourceNodeField, CorrelationIdField, DetailsJsonField, ChainSeqField, RowHashField); + + /// The tombstone-file schema. + public static ParquetSchema TombstoneSchema { get; } = new( + TombstoneMonthField, TombstoneSeqField, TombstoneHashField, TombstoneEventIdField, TombstonePurgedAtField); + + /// Opens a buffered writer of archive data rows over . + /// Destination stream (need not be seekable). + /// The archived partition (YYYY-MM), stamped into the file metadata. + /// Rows per Parquet row group. + /// Cancellation token. + /// The writer; dispose it to flush the last row group and the footer. + public static async Task 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); + } + + /// Opens a buffered writer of tombstones over . + /// Destination stream (need not be seekable). + /// The archived partition (YYYY-MM), stamped into the file metadata. + /// Rows per Parquet row group. + /// Cancellation token. + /// The writer; dispose it to flush the last row group and the footer. + public static async Task 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); + } + + /// Streams the rows of an archive data file in file order. + /// A readable, seekable stream over the data file. + /// Cancellation token. + /// The rows, as the chain hasher sees them. + public static async IAsyncEnumerable 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(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(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]); + } + } + } + + /// Streams the tombstones of an archive tombstone file. + /// A readable, seekable stream over the tombstone file. + /// Cancellation token. + /// The tombstones. + public static async IAsyncEnumerable 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(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(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)); + } + } + } + + /// Parses a YYYY-MM chain partition label to its first instant (UTC). + /// The label. + /// Day 1, midnight UTC. + 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 Metadata(string partition) => new() + { + [FormatVersionKey] = FormatVersion.ToString(CultureInfo.InvariantCulture), + [PartitionKey] = partition, + }; + + private static async Task Strings(ParquetRowGroupReader group, DataField field, int n, CancellationToken ct) + { + var values = new string?[n]; + await group.ReadAsync(field, values!, cancellationToken: ct).ConfigureAwait(false); + return values; + } + + /// Buffers archive data rows and writes them a row group at a time. + public sealed class DataWriter : IAsyncDisposable + { + private readonly ParquetWriter _writer; + private readonly int _rowGroupSize; + private readonly List _buffer; + private bool _disposed; + + internal DataWriter(ParquetWriter writer, int rowGroupSize) + { + _writer = writer; + _rowGroupSize = Math.Max(1, rowGroupSize); + _buffer = new List(Math.Min(_rowGroupSize, 65_536)); + } + + /// Rows added so far. + public long RowCount { get; private set; } + + /// Adds one row, flushing a row group when the buffer is full. + /// The row. + /// Cancellation token. + /// A task that completes when the row is buffered (or its row group written). + 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); + } + } + + /// Writes any buffered rows as a row group, then the file footer. + /// A task that completes when the file is finished. + 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(OccurredAtField, rows.Select(r => r.OccurredAtUtc).ToArray(), cancellationToken: ct).ConfigureAwait(false); + await group.WriteAsync(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(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(); + } + } + + /// Buffers tombstones and writes them a row group at a time. + public sealed class TombstoneWriter : IAsyncDisposable + { + private readonly ParquetWriter _writer; + private readonly int _rowGroupSize; + private readonly List _buffer = []; + private bool _disposed; + + internal TombstoneWriter(ParquetWriter writer, int rowGroupSize) + { + _writer = writer; + _rowGroupSize = Math.Max(1, rowGroupSize); + } + + /// Tombstones added so far. + public long Count { get; private set; } + + /// Adds one tombstone, flushing a row group when the buffer is full. + /// The tombstone. + /// Cancellation token. + /// A task that completes when the tombstone is buffered (or its row group written). + 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); + } + } + + /// Writes any buffered tombstones as a row group, then the file footer. + /// A task that completes when the file is finished. + 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(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(TombstonePurgedAtField, rows.Select(t => t.PurgedAtUtc).ToArray(), cancellationToken: ct).ConfigureAwait(false); + } + + _buffer.Clear(); + } + } +} + +/// +/// 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. +/// +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(); + } + + /// Returns the lowercase-hex SHA-256 of everything written so far (call once, at the end). + 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 buffer) + { + _hash.AppendData(buffer); + _inner.Write(buffer); + BytesWritten += buffer.Length; + } + + public override async ValueTask WriteAsync(ReadOnlyMemory 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. +} diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogArchiveOptions.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogArchiveOptions.cs new file mode 100644 index 00000000..8042c72d --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogArchiveOptions.cs @@ -0,0 +1,46 @@ +namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance; + +/// +/// Options for the AuditLog Parquet cold archive (section AuditLog:Archive; audit +/// roadmap T2, alog.md §15.2, ScadaBridge#42). Validated at startup by +/// AuditLogArchiveOptionsValidator in the AuditLog component. +/// +/// +/// +/// Default OFF. With false the retention purge drops expired +/// partitions exactly as before. With it true the purge writes each expired partition to +/// Parquet under 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. +/// +/// +/// Directory. An absolute path both central nodes can reach (a shared volume or UNC +/// share): the purge runs on whichever node is active, and audit archive list/verify +/// run on whichever node serves the management call. Files land in a YYYY +/// 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. +/// +/// +public sealed class AuditLogArchiveOptions +{ + /// Whether the retention purge archives each partition before dropping it (default false). + public bool Enabled { get; set; } + + /// Absolute archive root directory; required when . + public string? Directory { get; set; } + + /// Rows per Parquet row group (default 50,000; bounds archive-writer memory). + public int RowGroupSize { get; set; } = 50_000; + + /// + /// Command timeout, in minutes, for reading one partition out of SQL Server (default 60). + /// A month on a busy central is millions of rows. + /// + public int ReadCommandTimeoutMinutes { get; set; } = 60; + + /// + /// 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). + /// + public int ChainLockTimeoutSeconds { get; set; } = 300; +} diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogHashChain.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogHashChain.cs index 701e3425..51c882ed 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogHashChain.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogHashChain.cs @@ -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 = + /// The ten writable AuditLog columns, in table order — the hashed row content. + 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) + /// + /// Reads one row selected as (plus ChainSeq, RowHash when + /// ), strictly in ordinal order so it is safe under SequentialAccess. + /// + 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 OpenAsync(DbConnection conn, CancellationToken ct) + internal static async Task OpenAsync(DbConnection conn, CancellationToken ct) { if (conn.State == ConnectionState.Open) { diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogPartitionArchive.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogPartitionArchive.cs new file mode 100644 index 00000000..54af5355 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditLogPartitionArchive.cs @@ -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; + +/// +/// SQL Server + file-system implementation of (audit roadmap +/// T2, alog.md §15.2, ScadaBridge#42). See the interface for the archive-before-purge +/// contract and for the file layout. +/// +/// +/// +/// Layout. {Directory}/{YYYY}/{name}.parquet (rows), +/// {name}.tombstones.parquet (per-channel purge tombstones) and +/// {name}.manifest.json, where name is +/// auditlog-{YYYY-MM}-{archivedAt:yyyyMMddTHHmmssfff}Z. Each file is written to a +/// .tmp 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 . +/// +/// +/// Consistency with the hash chain. 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 (). +/// +/// +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 _logger; + private readonly TimeProvider _time; + + /// Initializes a new instance of the class. + /// The database context whose connection is used. + /// Archive options. + /// Optional logger. + /// Optional clock (tests). + public AuditLogPartitionArchive( + ScadaBridgeDbContext context, + AuditLogArchiveOptions options, + ILogger? logger = null, + TimeProvider? time = null) + { + _context = context ?? throw new ArgumentNullException(nameof(context)); + _options = options ?? throw new ArgumentNullException(nameof(options)); + _logger = logger ?? NullLogger.Instance; + _time = time ?? TimeProvider.System; + } + + /// + public bool IsConfigured => !string.IsNullOrWhiteSpace(_options.Directory); + + /// + public bool IsEnabled => _options.Enabled && IsConfigured; + + /// Returns true when is a well-formed archive name. + /// The candidate name. + /// True for auditlog-YYYY-MM-yyyyMMddTHHmmssfffZ. + public static bool IsValidName(string? name) => name is not null && NamePattern().IsMatch(name); + + /// + public async Task 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(); + 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(); + 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; + } + + /// + public async Task 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; + } + + /// + public async Task> ListAsync(CancellationToken ct = default) + { + RequireConfigured(); + if (!Directory.Exists(_options.Directory)) + { + return []; + } + + var manifests = new List(); + 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(); + } + + /// + public async Task 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(); + 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>(); + 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(); + var unchained = new Dictionary(); + 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(); + 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); + } + + /// + /// Compares an archived chain end with the database's current checkpoint anchor for the month. + /// + 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> ReadAnchorsAsync( + IEnumerable months, CancellationToken ct) + { + var result = new Dictionary(); + 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 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 ReadManifestAsync(string path, CancellationToken ct) + { + await using var input = File.OpenRead(path); + return await JsonSerializer.DeserializeAsync(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(); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditPartitionSwitchGuard.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditPartitionSwitchGuard.cs new file mode 100644 index 00000000..2a3dd42b --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Maintenance/AuditPartitionSwitchGuard.cs @@ -0,0 +1,56 @@ +using System.Globalization; +using System.Text; + +namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance; + +/// +/// 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. +/// +/// Rows the archive holds for the partition. +/// Each chain month's LastSeq at archive time. +public sealed record AuditPartitionSwitchGuard(long ExpectedRows, IReadOnlyList<(DateTime ChainMonth, long LastSeq)> ExpectedHeads) +{ + /// SQL error number THROWn when the partition row count changed. + public const int RowCountChangedError = 51002; + + /// SQL error number THROWn when a chain head in the partition changed. + public const int ChainAdvancedError = 51003; + + /// + /// Renders the guard as T-SQL for the switch batch. Every value is a number or a + /// yyyy-MM-dd date produced here, so inlining them as literals is injection-safe and + /// keeps the switch a parameterless ExecuteSqlRaw batch. + /// + /// The boundary literal the switch batch already uses. + /// The guard statements (run after the chain lock, inside the TRY). + 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(); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs index 3049ce8d..11e4c45b 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs @@ -477,7 +477,24 @@ VALUES }; /// - public async Task SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) + public Task SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) + => SwitchOutPartitionAsync(monthBoundary, guard: null, commandTimeout, ct); + + /// + /// The partition switch-out, optionally guarded by an archive's expectations + /// (ScadaBridge#42). With a , 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. + /// + /// The partition's lower boundary. + /// Archive expectations, or null for the unguarded purge. + /// Maintenance command timeout. + /// Cancellation token. + /// The approximate number of rows discarded. + internal async Task 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 diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs index 4fcc5191..9c2dd1cd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs @@ -114,6 +114,15 @@ public static class ServiceCollectionExtensions // AuditLogHashChainService, verified by the VerifyAuditChain management command. services.AddScoped(); + // 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(sp => new AuditLogPartitionArchive( + sp.GetRequiredService(), + sp.GetService>()?.Value ?? new AuditLogArchiveOptions(), + sp.GetService>())); + // 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. diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.csproj b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.csproj index acf9f91c..9dcd1810 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.csproj +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.csproj @@ -12,6 +12,7 @@ + diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs index f1e0a0fe..8f9e58bb 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs @@ -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(_ => 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.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, []); + + /// + /// With archiving enabled every expired partition is archived and then purged by the + /// archive's guarded purge — never by the plain, unguarded switch. + /// + [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(); + var manifest = ManifestFor(boundary); + archive.ArchivePartitionAsync(boundary, Arg.Any()).Returns(manifest); + archive.PurgeArchivedPartitionAsync(manifest, Arg.Any(), Arg.Any()) + .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(TimeSpan.FromSeconds(5)); + Assert.Equal(manifest.Name, a.ArchiveName); + Assert.Equal(7, purged.ExpectMsg(TimeSpan.FromSeconds(5)).RowsDeleted); + Assert.Empty(repo.SwitchedBoundaries); + archive.Received(1).PurgeArchivedPartitionAsync(manifest, Arg.Any(), Arg.Any()); + + StopPurgeActor(actor); + } + + /// + /// Fail closed: a partition that could not be archived is not purged at all, and the + /// failure surfaces as a purge failure. + /// + [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(); + archive.ArchivePartitionAsync(boundary, Arg.Any()) + .Returns>(_ => 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(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); + } + + /// With archiving configured but disabled, the purge is the plain switch and the archive is untouched. + [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(); + var purged = SubscribePurged(); + + var actor = CreateArchivingActor(repo, archive, enabled: false); + actor.Tell(AuditLogPurgeActor.PurgeTick.Instance); + + purged.ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal([boundary], repo.SwitchedBoundaries); + archive.DidNotReceiveWithAnyArgs().ArchivePartitionAsync(default, default); + + StopPurgeActor(actor); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Maintenance/AuditArchiveParquetTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Maintenance/AuditArchiveParquetTests.cs new file mode 100644 index 00000000..d95880b2 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Maintenance/AuditArchiveParquetTests.cs @@ -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; + +/// +/// 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. +/// +public class AuditArchiveParquetTests +{ + private static readonly DateTime Month = new(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc); + + /// Rows written to the archive layout read back equal, column for column, across row groups. + /// A task that represents the asynchronous operation. + [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]); + } + + /// Tombstones round-trip exactly. + /// A task that represents the asynchronous operation. + [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(); + 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))); + } + + /// + /// The rows read back from an archive re-verify against the archived head — the same + /// check audit archive verify makes long after the database dropped them. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ArchivedRows_ReverifyAgainstTheArchivedHead() + { + var (rows, head) = BuildChain(50); + var verifier = new AuditChainVerifier(Month, head, new Dictionary()); + 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); + } + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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()); + 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 Rows, AuditChainHeadState Head) BuildChain(int count) + { + var rows = new List(); + 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 ?? "", r.Action, r.Outcome, r.Category, + r.Target ?? "", r.SourceNode ?? "", r.CorrelationId?.ToString() ?? "", r.DetailsJson ?? "", + r.ChainSeq?.ToString() ?? "", r.RowHash is null ? "" : Convert.ToHexString(r.RowHash)); + + private static async Task WriteAsync(IEnumerable 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> ReadAsync(byte[] bytes) + { + var rows = new List(); + await foreach (var row in AuditArchiveParquet.ReadRowsAsync(new MemoryStream(bytes))) + { + rows.Add(row); + } + + return rows; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Maintenance/AuditLogPartitionArchiveTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Maintenance/AuditLogPartitionArchiveTests.cs new file mode 100644 index 00000000..3a61776a --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Maintenance/AuditLogPartitionArchiveTests.cs @@ -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; + +/// +/// 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. +/// +/// +/// 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. +/// +public sealed class AuditLogPartitionArchiveTests : IClassFixture, IDisposable +{ + private readonly MsSqlMigrationFixture _fixture; + private readonly DirectoryInfo _root = Directory.CreateTempSubdirectory("auditlog-archive-"); + + /// Creates the test class with the shared migrated SQL Server fixture. + /// The shared migration fixture. + public AuditLogPartitionArchiveTests(MsSqlMigrationFixture fixture) => _fixture = fixture; + + /// + public void Dispose() => _root.Delete(recursive: true); + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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); + } + } + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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(() => 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)); + } + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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(() => Archive(ctx).PurgeArchivedPartitionAsync(manifest)); + Assert.Equal(AuditPartitionSwitchGuard.ChainAdvancedError, ((SqlException)ex.InnerException!).Number); + } + + Assert.Equal(5, (await PartitionRowsAsync(month)).Count); + } + + /// + /// 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). + /// + /// A task that represents the asynchronous operation. + [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); + } + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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); + } + + /// A name that is not an archive name (e.g. a path) is rejected before any file access. + /// A task that represents the asynchronous operation. + [SkippableFact] + public async Task VerifyRejectsPathLikeNames() + { + Skip.IfNot(_fixture.Available, _fixture.SkipReason); + await using var ctx = CreateContext(); + await Assert.ThrowsAsync(() => Archive(ctx).VerifyAsync("../../etc/passwd")); + await Assert.ThrowsAsync(() => 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 ArchiveAsync(DateTime month) + { + await using var ctx = CreateContext(); + return await Archive(ctx).ArchivePartitionAsync(month); + } + + private async Task ArchiveAndPurgeAsync(DateTime month) + { + var manifest = await ArchiveAsync(month); + await using var ctx = CreateContext(); + await Archive(ctx).PurgeArchivedPartitionAsync(manifest); + return manifest; + } + + private async Task VerifyArchiveAsync(string name) + { + await using var ctx = CreateContext(); + return await Archive(ctx).VerifyAsync(name); + } + + private async Task VerifyMonthAsync(DateTime month) + { + await using var ctx = CreateContext(); + return await new AuditLogHashChain(ctx).VerifyMonthAsync(month); + } + + private async Task> ReadArchiveRowsAsync(AuditArchiveManifest manifest, byte[]? bytes = null) + { + bytes ??= await File.ReadAllBytesAsync(Path.Combine(_root.FullName, manifest.Partition[..4], manifest.DataFile)); + var rows = new List(); + await foreach (var row in AuditArchiveParquet.ReadRowsAsync(new MemoryStream(bytes))) + { + rows.Add(row); + } + + return rows; + } + + private static async Task RewriteAsync( + List 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> 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(); + 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 ?? "", r.Action, r.Outcome, r.Category, r.Target ?? "", + r.SourceNode ?? "", r.CorrelationId?.ToString() ?? "", r.DetailsJson ?? "", + r.ChainSeq?.ToString() ?? "", r.RowHash is null ? "" : Convert.ToHexString(r.RowHash)); + + private async Task InsertAsync(IEnumerable 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().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()); +}