From c19169a5809fa34a3386b28a695374c58714aca6 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 15 Aug 2026 05:32:33 -0400 Subject: [PATCH 1/4] feat(archive): make cold storage production ready Signed-off-by: Yordis Prieto --- .github/workflows/common.yml | 125 +++++++++ docs/diagnostics/metrics.md | 10 + docs/operations.md | 7 +- .../registry/trogon/eventstore/metrics.yaml | 46 ++++ .../Helpers/MiniClusterNode.cs | 35 ++- .../when_archiving_and_restoring_a_cluster.cs | 250 ++++++++++++++++++ .../Integration/specification_with_cluster.cs | 61 ++--- .../EventStore.Core.XUnit.Tests.csproj | 3 + .../ArchiveCatchup/ArchiveCatchupTests.cs | 181 ++++++++++++- .../ArchiveCatchup/FakeArchiveStorage.cs | 41 ++- .../Archive/ArchiveLifecycleSoakTests.cs | 240 +++++++++++++++++ .../Services/Archive/ArchiveMetricsTests.cs | 102 +++++++ .../Services/Archive/ArchiverServiceTests.cs | 65 ++++- .../ArchiveStorageReaderMetricsTests.cs | 227 ++++++++++++++++ .../Storage/ArchiveStorageTestsBase.cs | 74 +++++- .../Archive/Storage/S3RestartRecoveryTests.cs | 135 ++++++++++ .../Services/Archive/Storage/S3Tests.cs | 69 +++++ src/EventStore.Core/ClusterVNode.cs | 17 +- .../Configuration/ClusterVNodeOptions.cs | 2 +- src/EventStore.Core/MetricsBootstrapper.cs | 21 +- .../Archive/ArchiveCatchup/ArchiveCatchup.cs | 57 +++- .../Services/Archive/ArchiveMetrics.cs | 149 +++++++++++ .../Archive/ArchivePlugableComponent.cs | 6 +- .../Archive/Archiver/ArchiverService.cs | 32 ++- .../Archive/Storage/ArchiveReadStream.cs | 222 ++++++++++++++++ .../Archive/Storage/ArchiveStorageFactory.cs | 8 +- .../Storage/ArchiveStorageReaderMetrics.cs | 97 +++++++ .../Services/Archive/Storage/S3Storage.cs | 6 +- .../Generated/MetricDefinitions.g.cs | 30 +++ 29 files changed, 2234 insertions(+), 84 deletions(-) create mode 100644 src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs create mode 100644 src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveLifecycleSoakTests.cs create mode 100644 src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveMetricsTests.cs create mode 100644 src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs create mode 100644 src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs create mode 100644 src/EventStore.Core/Services/Archive/ArchiveMetrics.cs create mode 100644 src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs create mode 100644 src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs diff --git a/.github/workflows/common.yml b/.github/workflows/common.yml index 1b340abd32..e6fa087284 100644 --- a/.github/workflows/common.yml +++ b/.github/workflows/common.yml @@ -120,6 +120,131 @@ jobs: shell: bash run: | ./protolock.sh status --uptodate + + archive-storage-contract: + runs-on: ubuntu-latest + name: Archive Storage Contract + services: + rustfs: + image: rustfs/rustfs:1.0.0-beta.11@sha256:84ce557a0245a06a9aae5516f55ee0f007fca78d41df356f419306fdc0cb168c + ports: + - 9000:9000 + env: + RUSTFS_ACCESS_KEY: archive-contract + RUSTFS_SECRET_KEY: archive-contract-secret-key + options: >- + --health-cmd "curl --fail http://localhost:9000/health" + --health-interval 2s + --health-timeout 2s + --health-retries 30 + steps: + - name: Checkout + uses: actions/checkout@v7 + - name: Install net10.0 + uses: actions/setup-dotnet@v6 + with: + dotnet-version: 10.0.x + - name: Set up .NET NuGet authentication + run: | + dotnet nuget add source "https://nuget.pkg.github.com/TrogonStack/index.json" \ + --name "github" \ + --username "${{ github.actor }}" \ + --password "${{ secrets.GITHUB_TOKEN }}" \ + --store-password-in-clear-text + - name: Run archive storage contract tests + env: + EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 + EVENTSTORE_S3_TEST_REGION: us-east-1 + EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract + EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key + run: | + dotnet test \ + --configuration Release \ + -p:Platform=x64 \ + -p:ContinuousIntegrationBuild=true \ + -p:RunS3Tests=true \ + --filter "FullyQualifiedName~S3ReaderTests|FullyQualifiedName~S3WriterTests|FullyQualifiedName~S3MetricsTests" \ + --logger:GitHubActions \ + src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj + + - name: Seed archive restart recovery data + id: seed_archive_recovery + env: + EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 + EVENTSTORE_S3_TEST_REGION: us-east-1 + EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract + EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key + EVENTSTORE_S3_RECOVERY_BUCKET: archive-recovery-${{ github.run_id }}-${{ github.run_attempt }} + EVENTSTORE_S3_RECOVERY_PHASE: seed + run: | + dotnet test \ + --configuration Release \ + --no-build \ + -p:Platform=x64 \ + -p:RunS3Tests=true \ + --filter "FullyQualifiedName~S3RestartRecoveryTests" \ + --logger:GitHubActions \ + src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj + + - name: Stop RustFS without removing its data + run: | + timeout 20 docker stop --timeout 10 "${{ job.services.rustfs.id }}" + test "$(docker inspect --format '{{.State.Status}}' "${{ job.services.rustfs.id }}")" = "exited" + + - name: Assert archive storage is unavailable + env: + EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 + EVENTSTORE_S3_TEST_REGION: us-east-1 + EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract + EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key + EVENTSTORE_S3_RECOVERY_BUCKET: archive-recovery-${{ github.run_id }}-${{ github.run_attempt }} + EVENTSTORE_S3_RECOVERY_PHASE: unavailable + run: | + timeout 20 dotnet test \ + --configuration Release \ + --no-build \ + -p:Platform=x64 \ + -p:RunS3Tests=true \ + --filter "FullyQualifiedName~S3RestartRecoveryTests" \ + --logger:GitHubActions \ + src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj + + - name: Restart RustFS and verify archive recovery + if: ${{ always() && steps.seed_archive_recovery.outcome == 'success' }} + env: + EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 + EVENTSTORE_S3_TEST_REGION: us-east-1 + EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract + EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key + EVENTSTORE_S3_RECOVERY_BUCKET: archive-recovery-${{ github.run_id }}-${{ github.run_attempt }} + EVENTSTORE_S3_RECOVERY_PHASE: verify-cleanup + run: | + docker start "${{ job.services.rustfs.id }}" + timeout 30 bash -c -- 'until curl --output /dev/null --silent --fail http://localhost:9000/health; do sleep 1; done' + dotnet test \ + --configuration Release \ + --no-build \ + -p:Platform=x64 \ + -p:RunS3Tests=true \ + --filter "FullyQualifiedName~S3RestartRecoveryTests" \ + --logger:GitHubActions \ + src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj + + - name: Run archive cluster restore gate + env: + EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 + EVENTSTORE_S3_TEST_REGION: us-east-1 + EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract + EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key + run: | + dotnet test \ + --configuration Release \ + -p:Platform=x64 \ + -p:ContinuousIntegrationBuild=true \ + --filter "TestCategory=ArchiveIntegration" \ + --logger:GitHubActions \ + src/EventStore.Core.Tests/EventStore.Core.Tests.csproj + docker-compose: runs-on: ubuntu-latest name: Docker Compose Smoke Test diff --git a/docs/diagnostics/metrics.md b/docs/diagnostics/metrics.md index ae011b7174..def9fb6d36 100644 --- a/docs/diagnostics/metrics.md +++ b/docs/diagnostics/metrics.md @@ -87,6 +87,16 @@ Queue and message labels come from the `QueueLabels` and `MessageTypes` regular- The `trogon.eventstore.storage.activity` attribute is `read` or `write`. The checkpoint read kind currently emitted by the node is `non_flushed`. +### Archive + +| Metric | Instrument | Unit | Attributes | Description | +| --- | --- | --- | --- | --- | +| `trogon.eventstore.archive.checkpoint.lag` | Gauge | `By` | None | Replicated transaction log bytes not yet covered by the archive checkpoint | +| `trogon.eventstore.archive.chunk.pending.count` | UpDownCounter | `{chunk}` | None | Chunks waiting for commit, queued for persistence, or currently being persisted | +| `trogon.eventstore.archive.retry.count` | Counter | `{retry}` | `trogon.eventstore.activity.name` | Archive retries by operation | +| `trogon.eventstore.archive.failure.count` | Counter | `{failure}` | `trogon.eventstore.activity.name` | Archive failures by operation | +| `trogon.eventstore.archive.read.duration` | Histogram | `s` | `trogon.eventstore.activity.name`, `trogon.eventstore.activity.outcome` | Remote archive read duration through stream completion or disposal | + ### Persistent subscriptions Every persistent subscription instrument includes `trogon.eventstore.persistent_subscription.stream` and `trogon.eventstore.persistent_subscription.group`. diff --git a/docs/operations.md b/docs/operations.md index 2e0829d57b..573203ee18 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -138,9 +138,10 @@ Reads for chunks no longer stored locally issue object-storage requests. Their l on the S3 service and network. When read concurrency is limited, slower archive reads may cause other reads to wait. -There are currently no dedicated archive queue-depth or archive-checkpoint metrics. The -`eventstore-logical-chunk-read-distribution` metric measures how far reads are from the log tail when event-read -metrics are enabled. +Monitor `trogon.eventstore.archive.checkpoint.lag` and +`trogon.eventstore.archive.chunk.pending.count` to detect an archiver falling behind. Retry and failure counters +identify storage or recovery problems by operation, while `trogon.eventstore.archive.read.duration` reports the +latency and outcome of remote reads. See [Metrics](diagnostics/metrics.md#archive) for the complete definitions. Archive storage supplements backups; it does not replace them. Only completed, committed chunks through the archive checkpoint are uploaded. Keep normal database and index backups. During startup, an archive-enabled diff --git a/otel/semconv/registry/trogon/eventstore/metrics.yaml b/otel/semconv/registry/trogon/eventstore/metrics.yaml index 0540fa0f04..13ff2554c5 100644 --- a/otel/semconv/registry/trogon/eventstore/metrics.yaml +++ b/otel/semconv/registry/trogon/eventstore/metrics.yaml @@ -312,6 +312,52 @@ groups: metric_name: trogon.eventstore.storage.chunk.read.distance instrument: histogram unit: "{chunk}" + - id: metric.trogon.eventstore.archive.checkpoint.lag + type: metric + stability: development + brief: Replicated transaction log bytes not yet covered by the archive checkpoint. + metric_name: trogon.eventstore.archive.checkpoint.lag + instrument: gauge + unit: By + - id: metric.trogon.eventstore.archive.chunk.pending.count + type: metric + stability: development + brief: Number of archive chunks waiting for or undergoing persistence. + metric_name: trogon.eventstore.archive.chunk.pending.count + instrument: updowncounter + unit: "{chunk}" + - id: metric.trogon.eventstore.archive.retry.count + type: metric + stability: development + brief: Number of retries initiated by archive operations. + metric_name: trogon.eventstore.archive.retry.count + instrument: counter + unit: "{retry}" + attributes: + - ref: trogon.eventstore.activity.name + requirement_level: required + - id: metric.trogon.eventstore.archive.failure.count + type: metric + stability: development + brief: Number of failed archive operations. + metric_name: trogon.eventstore.archive.failure.count + instrument: counter + unit: "{failure}" + attributes: + - ref: trogon.eventstore.activity.name + requirement_level: required + - id: metric.trogon.eventstore.archive.read.duration + type: metric + stability: development + brief: Duration of remote archive read requests. + metric_name: trogon.eventstore.archive.read.duration + instrument: histogram + unit: s + attributes: + - ref: trogon.eventstore.activity.name + requirement_level: required + - ref: trogon.eventstore.activity.outcome + requirement_level: required - id: metric.trogon.eventstore.persistent_subscription.connection.count type: metric stability: development diff --git a/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs b/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs index ceafee2d86..444ab592d2 100644 --- a/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs +++ b/src/EventStore.Core.Tests/Helpers/MiniClusterNode.cs @@ -16,6 +16,7 @@ using EventStore.Core.Certificates; using EventStore.Core.Data; using EventStore.Core.Messages; +using EventStore.Core.Services.Archive; using EventStore.Core.Services.Monitoring; using EventStore.Core.Services.PersistentSubscription.ConsumerStrategy; using EventStore.Core.Services.Storage.ReaderIndex; @@ -49,6 +50,7 @@ public class MiniClusterNode public readonly ClusterVNode Node; public TFChunkDb Db => Node.Db; + public string DbPath => _dbPath; private readonly string _dbPath; private readonly bool _isReadOnlyReplica; private readonly TaskCompletionSource _started = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -64,7 +66,9 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp, IPEndPoint httpEndPoint, EndPoint[] gossipSeeds, ISubsystem[] subsystems = null, bool enableTrustedAuth = false, int memTableSize = 1000, bool disableFlushToDisk = false, bool readOnlyReplica = false, int nodePriority = 0, - string intHostAdvertiseAs = null, IExpiryStrategy expiryStrategy = null) + string intHostAdvertiseAs = null, IExpiryStrategy expiryStrategy = null, + ArchiveOptions archiveOptions = null, bool archiver = false, + int clusterSize = 3, bool unsafeAllowSurplusNodes = false) { RunningTime.Start(); @@ -100,14 +104,15 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp, DiscoverViaDns = false, ClusterDns = string.Empty, GossipSeed = gossipSeeds, - ClusterSize = 3, + ClusterSize = clusterSize, NodePriority = nodePriority, GossipIntervalMs = 2_000, GossipAllowedDifferenceMs = 1_000, GossipTimeoutMs = 2_000, DeadMemberRemovalPeriodSec = 1_800_000, ReadOnlyReplica = readOnlyReplica, - Archiver = false, + Archiver = archiver, + UnsafeAllowSurplusNodes = unsafeAllowSurplusNodes, StreamInfoCacheCapacity = 10_000 }, Interface = new() @@ -144,15 +149,33 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp, PlugableComponents = subsystems }; - var inMemConf = new ConfigurationBuilder() - .AddInMemoryCollection(new KeyValuePair[] { + var configuration = new List> { new("EventStore:TcpPlugin:NodeTcpPort", externalTcp.Port.ToString()), new("EventStore:TcpPlugin:EnableExternalTcp", "true"), new("EventStore:TcpUnitTestPlugin:NodeTcpPort", externalTcp.Port.ToString()), new("EventStore:TcpUnitTestPlugin:NodeHeartbeatInterval", "10000"), new("EventStore:TcpUnitTestPlugin:NodeHeartbeatTimeout", "10000"), new("EventStore:TcpUnitTestPlugin:Insecure", options.Application.Insecure.ToString()), - }).Build(); + }; + + if (archiveOptions is not null) + { + configuration.AddRange([ + new("EventStore:Archive:Enabled", archiveOptions.Enabled.ToString()), + new("EventStore:Archive:StorageType", archiveOptions.StorageType.ToString()), + new("EventStore:Archive:S3:Bucket", archiveOptions.S3.Bucket), + new("EventStore:Archive:S3:Region", archiveOptions.S3.Region), + new("EventStore:Archive:S3:AccessKeyId", archiveOptions.S3.AccessKeyId), + new("EventStore:Archive:S3:SecretAccessKey", archiveOptions.S3.SecretAccessKey), + new("EventStore:Archive:S3:ServiceUrl", archiveOptions.S3.ServiceUrl), + new("EventStore:Archive:RetainAtLeast:Days", archiveOptions.RetainAtLeast.Days.ToString()), + new("EventStore:Archive:RetainAtLeast:LogicalBytes", archiveOptions.RetainAtLeast.LogicalBytes.ToString()), + ]); + } + + var inMemConf = new ConfigurationBuilder() + .AddInMemoryCollection(configuration) + .Build(); var serverCertificate = ssl_connections.GetServerCertificate(); var trustedRootCertificates = new X509Certificate2Collection(ssl_connections.GetRootCertificate()); diff --git a/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs new file mode 100644 index 0000000000..c49c6174ce --- /dev/null +++ b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs @@ -0,0 +1,250 @@ +using System; +using System.IO; +using System.Linq; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using Amazon.S3; +using Amazon.S3.Model; +using EventStore.ClientAPI; +using EventStore.Core.Data; +using EventStore.Core.Messages; +using EventStore.Core.Messaging; +using EventStore.Core.Services; +using EventStore.Core.Services.Archive; +using EventStore.Core.Services.Archive.Naming; +using EventStore.Core.Services.Archive.Storage; +using EventStore.Core.Services.UserManagement; +using EventStore.Core.Tests.Helpers; +using EventStore.Core.TransactionLog.Chunks.TFChunk; +using EventStore.Core.TransactionLog.FileNamingStrategy; +using NUnit.Framework; + +namespace EventStore.Core.Tests.Integration.Archive; + +[TestFixture(typeof(LogFormat.V2), typeof(string))] +[Category("ArchiveIntegration")] +[NonParallelizable] +public class when_archiving_and_restoring_a_cluster + : specification_with_cluster +{ + private const string Stream = "archive-soak"; + private const string ArchiveCheckpointFile = "archive.chk"; + private const int ArchiverNodeIndex = 3; + private const int SoakIterations = 3; + private const int EventsPerIteration = 12; + private static readonly TimeSpan GateTimeout = TimeSpan.FromMinutes(3); + private static readonly TimeSpan SoakTimeout = TimeSpan.FromMinutes(15); + + private readonly string _bucket = $"archive-soak-{Guid.NewGuid():N}"; + private readonly ArchiveOptions _archiveOptions; + private AmazonS3Client _s3Client; + private S3Reader _archiveReader; + private long _archivedCheckpoint; + private int _restoredNodeIndex; + private int _completedIterations; + protected override int NodeCount => 4; + protected override TimeSpan GivenTimeout => SoakTimeout; + + public when_archiving_and_restoring_a_cluster() + { + var endpoint = Environment.GetEnvironmentVariable("EVENTSTORE_S3_TEST_ENDPOINT"); + var region = Environment.GetEnvironmentVariable("EVENTSTORE_S3_TEST_REGION") ?? "us-east-1"; + var accessKey = Environment.GetEnvironmentVariable("EVENTSTORE_S3_TEST_ACCESS_KEY"); + var secretKey = Environment.GetEnvironmentVariable("EVENTSTORE_S3_TEST_SECRET_KEY"); + + _archiveOptions = new() + { + Enabled = !string.IsNullOrWhiteSpace(endpoint), + StorageType = StorageType.S3, + S3 = new() + { + Bucket = _bucket, + Region = region, + AccessKeyId = accessKey ?? string.Empty, + SecretAccessKey = secretKey ?? string.Empty, + ServiceUrl = endpoint ?? string.Empty, + }, + RetainAtLeast = new() { Days = 0, LogicalBytes = 0 }, + }; + } + + protected override void BeforeNodesStart() + { + if (!_archiveOptions.Enabled) + { + Assert.Ignore("The archive integration endpoint is not configured."); + } + + _s3Client = new AmazonS3Client( + _archiveOptions.S3.AccessKeyId, + _archiveOptions.S3.SecretAccessKey, + new AmazonS3Config + { + ServiceURL = _archiveOptions.S3.ServiceUrl, + AuthenticationRegion = _archiveOptions.S3.Region, + ForcePathStyle = true, + }); + _s3Client.PutBucketAsync(new PutBucketRequest { BucketName = _bucket }).GetAwaiter().GetResult(); + var checkpointInitialized = new S3Writer(_archiveOptions.S3, ArchiveCheckpointFile) + .SetCheckpoint(0L, CancellationToken.None) + .AsTask() + .GetAwaiter() + .GetResult(); + if (!checkpointInitialized) + { + throw new InvalidOperationException("Failed to initialize the archive checkpoint."); + } + + var archiveNamer = new ArchiveChunkNamer( + new VersionedPatternFileNamingStrategy(PathName, "chunk-")); + _archiveReader = new S3Reader(_archiveOptions.S3, archiveNamer, ArchiveCheckpointFile); + } + + protected override MiniClusterNode CreateNode( + int index, + Endpoints endpoints, + EndPoint[] gossipSeeds, + bool wait = true) => + new( + PathName, + index, + endpoints.InternalTcp, + endpoints.ExternalTcp, + endpoints.HttpEndPoint, + gossipSeeds, + readOnlyReplica: index == ArchiverNodeIndex, + archiveOptions: _archiveOptions.Enabled ? _archiveOptions : null, + archiver: index == ArchiverNodeIndex); + + protected override IEventStoreConnection CreateConnection() => + EventStoreConnection.Create( + ConnectionSettings.Create().DisableServerCertificateValidation(), + GetLeader().ExternalTcpEndPoint); + + protected override async Task Given() + { + var payload = new byte[256 * 1024]; + new Random(1729).NextBytes(payload); + var leader = GetLeader(); + var archiver = _nodes[ArchiverNodeIndex]; + AssertEx.IsOrBecomesTrue( + () => _nodes.Any(node => + node.DebugIndex != ArchiverNodeIndex && node.NodeState == VNodeState.Follower), + GateTimeout, + $"A voting follower did not become ready. States={string.Join(", ", _nodes.Select(node => node.NodeState))}"); + _restoredNodeIndex = Array.FindIndex(_nodes, + node => node.NodeState == VNodeState.Follower && node.DebugIndex != ArchiverNodeIndex); + Assert.That(_restoredNodeIndex, Is.GreaterThanOrEqualTo(0)); + + for (var iteration = 0; iteration < SoakIterations; iteration++) + { + for (var eventNumber = 0; eventNumber < EventsPerIteration; eventNumber++) + { + await _conn.AppendToStreamAsync( + Stream, + EventStore.ClientAPI.ExpectedVersion.Any, + new EventData(Guid.NewGuid(), "archive-event", isJson: false, payload, Array.Empty())); + } + + AssertEx.IsOrBecomesTrue( + () => archiver.Db.Config.WriterCheckpoint.Read() >= leader.Db.Config.WriterCheckpoint.Read(), + GateTimeout, + $"The archiver did not replicate iteration {iteration + 1}."); + + var previousCheckpoint = _archivedCheckpoint; + await WaitForArchiveCheckpoint(previousCheckpoint + 2L * MiniNode.ChunkSize); + _archivedCheckpoint = await _archiveReader.GetCheckpoint(CancellationToken.None); + var coldChunkNumber = (int)(previousCheckpoint / MiniNode.ChunkSize); + + await StartScavenge(leader); + AssertEx.IsOrBecomesTrue( + () => leader.Db.Manager.GetChunk(coldChunkNumber).IsRemote, + GateTimeout, + $"Iteration {iteration + 1} did not replace chunk {coldChunkNumber} with an archive locator."); + Assert.That( + (await leader.Db.Manager.GetChunk(coldChunkNumber).TryReadFirst(CancellationToken.None)).Success, + Is.True); + + await RestoreNode(_restoredNodeIndex, coldChunkNumber); + _completedIterations++; + } + } + + private static async Task StartScavenge(MiniClusterNode leader) + { + var scavengeStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + leader.Node.MainQueue.Publish(new ClientMessage.ScavengeDatabase( + new CallbackEnvelope(scavengeStarted.SetResult), + Guid.NewGuid(), + SystemAccounts.System, + startFromChunk: 0, + threads: 1, + threshold: null, + throttlePercent: null, + syncOnly: false)); + Assert.That(await scavengeStarted.Task.WaitAsync(GateTimeout), + Is.TypeOf()); + } + + private async Task RestoreNode(int nodeIndex, int coldChunkNumber) + { + await _nodes[nodeIndex].Shutdown(keepDb: true); + Directory.Delete(_nodes[nodeIndex].DbPath, recursive: true); + + var restored = CreateNode(nodeIndex, _nodeEndpoints[nodeIndex], GossipSeedsFor(nodeIndex)); + restored.Start(); + _nodes[nodeIndex] = restored; + await restored.Started.WaitAsync(GateTimeout); + + AssertEx.IsOrBecomesTrue( + () => restored.Db.Config.WriterCheckpoint.Read() >= _archivedCheckpoint, + GateTimeout, + "The restored node did not catch up from the archive before rejoining replication."); + Assert.That( + (await restored.Db.Manager.GetChunk(coldChunkNumber).TryReadFirst(CancellationToken.None)).Success, + Is.True); + } + + private EndPoint[] GossipSeedsFor(int nodeIndex) => + _nodeEndpoints + .Where((_, index) => index != nodeIndex) + .Select(x => (EndPoint)x.HttpEndPoint) + .ToArray(); + + private async Task WaitForArchiveCheckpoint(long minimum) + { + using var timeout = new CancellationTokenSource(GateTimeout); + while (await _archiveReader.GetCheckpoint(timeout.Token) < minimum) + { + await Task.Delay(100, timeout.Token); + } + } + + [OneTimeTearDown] + public override async Task TestFixtureTearDown() + { + await base.TestFixtureTearDown(); + if (_s3Client is null) + { + return; + } + + var objects = await _s3Client.ListObjectsV2Async(new ListObjectsV2Request { BucketName = _bucket }); + foreach (var item in objects.S3Objects) + { + await _s3Client.DeleteObjectAsync(_bucket, item.Key); + } + await _s3Client.DeleteBucketAsync(_bucket); + _s3Client.Dispose(); + } + + [Test] + public void archive_restore_gate_completed() + { + Assert.That(_completedIterations, Is.EqualTo(SoakIterations)); + Assert.That(_archivedCheckpoint, Is.GreaterThanOrEqualTo(SoakIterations * 2L * MiniNode.ChunkSize)); + Assert.That(_nodes[ArchiverNodeIndex].NodeState, Is.EqualTo(VNodeState.ReadOnlyReplica)); + Assert.That(_nodes[_restoredNodeIndex].NodeState, Is.AnyOf(VNodeState.Follower, VNodeState.Leader)); + } +} diff --git a/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs b/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs index c8b1646fc5..fb555b69dc 100644 --- a/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs +++ b/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs @@ -15,10 +15,11 @@ namespace EventStore.Core.Tests.Integration; public abstract class specification_with_cluster : SpecificationWithDirectoryPerTestFixture { - protected readonly MiniClusterNode[] _nodes = new MiniClusterNode[3]; - protected readonly Endpoints[] _nodeEndpoints = new Endpoints[3]; + protected MiniClusterNode[] _nodes; + protected Endpoints[] _nodeEndpoints; protected IEventStoreConnection _conn; protected virtual TimeSpan GivenTimeout { get; } = TimeSpan.FromMinutes(2); + protected virtual int NodeCount => 3; private readonly Dictionary>> _nodeCreationFactory = new(); @@ -79,16 +80,14 @@ public override async Task TestFixtureSetUp() MiniNodeLogging.Setup(); - _nodeEndpoints[0] = new Endpoints(); - _nodeEndpoints[1] = new Endpoints(); - _nodeEndpoints[2] = new Endpoints(); - - _nodeEndpoints[0].DisposeSockets(); - _nodeEndpoints[1].DisposeSockets(); - _nodeEndpoints[2].DisposeSockets(); + _nodes = new MiniClusterNode[NodeCount]; + _nodeEndpoints = Enumerable.Range(0, NodeCount).Select(_ => new Endpoints()).ToArray(); + foreach (var endpoints in _nodeEndpoints) + { + endpoints.DisposeSockets(); + } - var duplicates = _nodeEndpoints[0].Ports().Concat(_nodeEndpoints[1].Ports()) - .Concat(_nodeEndpoints[2].Ports()) + var duplicates = _nodeEndpoints.SelectMany(x => x.Ports()) .GroupBy(x => x) .Where(g => g.Count() > 1) .Select(x => x.Key) @@ -96,25 +95,25 @@ public override async Task TestFixtureSetUp() Assert.IsEmpty(duplicates); - _nodeCreationFactory.Add(0, wait => CreateNode(0, - _nodeEndpoints[0], new[] { _nodeEndpoints[1].HttpEndPoint, _nodeEndpoints[2].HttpEndPoint }, - wait)); - _nodeCreationFactory.Add(1, wait => CreateNode(1, - _nodeEndpoints[1], new[] { _nodeEndpoints[0].HttpEndPoint, _nodeEndpoints[2].HttpEndPoint }, - wait)); - _nodeCreationFactory.Add(2, wait => CreateNode(2, - _nodeEndpoints[2], new[] { _nodeEndpoints[0].HttpEndPoint, _nodeEndpoints[1].HttpEndPoint }, - wait)); - - _nodes[0] = _nodeCreationFactory[0](true); - _nodes[1] = _nodeCreationFactory[1](true); - _nodes[2] = _nodeCreationFactory[2](true); + for (var index = 0; index < NodeCount; index++) + { + var nodeIndex = index; + _nodeCreationFactory.Add(nodeIndex, wait => CreateNode( + nodeIndex, + _nodeEndpoints[nodeIndex], + _nodeEndpoints.Where((_, otherIndex) => otherIndex != nodeIndex) + .Select(x => (EndPoint)x.HttpEndPoint) + .ToArray(), + wait)); + _nodes[nodeIndex] = _nodeCreationFactory[nodeIndex](true); + } BeforeNodesStart(); - _nodes[0].Start(); - _nodes[1].Start(); - _nodes[2].Start(); + foreach (var node in _nodes) + { + node.Start(); + } try { @@ -125,7 +124,8 @@ public override async Task TestFixtureSetUp() if (_nodes.Count(x => x.Started.IsCompletedSuccessfully) < 2) { MiniNodeLogging.WriteLogs(); - throw new TimeoutException($"Cluster nodes did not start. Statuses: {_nodes[0].NodeState}/{_nodes[1].NodeState}/{_nodes[2].NodeState}", ex); + throw new TimeoutException( + $"Cluster nodes did not start. Statuses: {string.Join('/', _nodes.Select(x => x.NodeState))}", ex); } } @@ -188,10 +188,7 @@ public void AfterEachTest() public override async Task TestFixtureTearDown() { _conn?.Close(); - await Task.WhenAll( - _nodes[0].Shutdown(), - _nodes[1].Shutdown(), - _nodes[2].Shutdown()); + await Task.WhenAll(_nodes.Select(x => x.Shutdown())); MiniNodeLogging.Clear(); diff --git a/src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj b/src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj index 10756c46e3..193960d579 100644 --- a/src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj +++ b/src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj @@ -3,6 +3,9 @@ true true + + $(DefineConstants);RUN_S3_TESTS + diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs index 709618cc8f..a7c541e132 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs @@ -4,8 +4,10 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; +using EventStore.Core.Services.Archive; using EventStore.Core.Services.Archive.Storage.Exceptions; using EventStore.Core.TransactionLog.Checkpoint; +using EventStore.Core.TransactionLog.Chunks; using Xunit; namespace EventStore.Core.XUnit.Tests.Services.Archive.ArchiveCatchup; @@ -31,6 +33,7 @@ private struct Sut public ICheckpoint WriterCheckpoint { get; init; } public ICheckpoint ChaserCheckpoint { get; init; } public ICheckpoint EpochCheckpoint { get; init; } + public RecordingArchiveMetrics Metrics { get; init; } } private Sut CreateSut( @@ -38,12 +41,16 @@ private Sut CreateSut( long? archiveCheckpoint = null, string[] dbChunks = null, string[] archiveChunks = null, - Action onGetChunk = null) + Action onGetChunk = null, + Action onGetCheckpoint = null, + Func getChunk = null, + TimeSpan? retryInterval = null, + int chunkSize = ChunkSize) { dbCheckpoint ??= 0L; archiveCheckpoint ??= 0L; - dbChunks ??= CreateChunkList(dbCheckpoint.Value); - archiveChunks ??= CreateChunkList(archiveCheckpoint.Value); + dbChunks ??= CreateChunkList(dbCheckpoint.Value, chunkSize); + archiveChunks ??= CreateChunkList(archiveCheckpoint.Value, chunkSize); foreach (var dbChunk in dbChunks) { @@ -51,10 +58,13 @@ private Sut CreateSut( } var archive = new FakeArchiveStorage( - chunkSize: ChunkSize, + chunkSize, archiveChunks, archiveCheckpoint.Value, - onGetChunk); + onGetChunk, + onGetCheckpoint, + getChunk); + var metrics = new RecordingArchiveMetrics(); var writerCheckpoint = new InMemoryCheckpoint(dbCheckpoint.Value); var chaserCheckpoint = new InMemoryCheckpoint(dbCheckpoint.Value); @@ -64,9 +74,11 @@ private Sut CreateSut( writerCheckpoint: writerCheckpoint, chaserCheckpoint: chaserCheckpoint, epochCheckpoint: epochCheckpoint, - chunkSize: ChunkSize, + chunkSize, fileNamingStrategy: new CustomNamingStrategy(), - archiveStorageFactory: archive + archiveStorageFactory: archive, + metrics: metrics, + retryInterval: retryInterval ?? TimeSpan.Zero ); return new Sut @@ -77,17 +89,18 @@ private Sut CreateSut( ArchiveChunks = archiveChunks, WriterCheckpoint = writerCheckpoint, ChaserCheckpoint = chaserCheckpoint, - EpochCheckpoint = epochCheckpoint + EpochCheckpoint = epochCheckpoint, + Metrics = metrics }; } - private string[] CreateChunkList(long checkpoint) + private string[] CreateChunkList(long checkpoint, int chunkSize = ChunkSize) { var chunks = new List(); var namingStrategy = new CustomNamingStrategy(); - var numChunks = checkpoint / ChunkSize; - if (checkpoint % ChunkSize != 0) + var numChunks = checkpoint / chunkSize; + if (checkpoint % chunkSize != 0) { numChunks++; } @@ -100,6 +113,23 @@ private string[] CreateChunkList(long checkpoint) return chunks.ToArray(); } + [Fact] + public async Task catches_up_when_a_completed_chunk_is_smaller_than_the_preallocation_size() + { + const int preallocationSize = 8192; + const string chunkFile = "chunk-0.0"; + var sut = CreateSut( + archiveCheckpoint: preallocationSize, + chunkSize: preallocationSize); + var receivedLength = sut.Archive.CreateChunkBytes(chunkFile).Length; + Assert.True(receivedLength < preallocationSize); + + await sut.Catchup.Run(); + + Assert.Equal(preallocationSize, sut.WriterCheckpoint.Read()); + Assert.Equal(receivedLength, new FileInfo(Path.Combine(DbPath, chunkFile)).Length); + } + [Theory] [InlineData(0L, 0L)] // db is in sync with archive, both have no data [InlineData(1000L, 1000L)] // db is in sync with archive, both have one chunk @@ -211,6 +241,8 @@ public async Task retries_if_a_chunk_is_deleted_from_the_archive_when_catching_u Assert.Equal(2000L, sut.WriterCheckpoint.Read()); Assert.Equal(2000L, sut.ChaserCheckpoint.Read()); Assert.Equal(-1, sut.EpochCheckpoint.Read()); + Assert.Equal([ArchiveOperation.CatchUpChunk], sut.Metrics.Failures); + Assert.Equal([ArchiveOperation.CatchUpChunk], sut.Metrics.Retries); return; @@ -227,6 +259,118 @@ void OnGetChunk(string chunkFile) } } + [Fact] + public async Task records_checkpoint_failure_before_retrying() + { + using var cts = new CancellationTokenSource(); + var sut = CreateSut(onGetCheckpoint: () => + { + cts.Cancel(); + throw new InvalidOperationException("checkpoint unavailable"); + }); + + await Assert.ThrowsAsync(() => sut.Catchup.Run(cts.Token)); + + Assert.Equal([ArchiveOperation.CatchUpCheckpoint], sut.Metrics.Failures); + Assert.Equal([ArchiveOperation.CatchUpCheckpoint], sut.Metrics.Retries); + } + + [Fact] + public async Task resumes_after_checkpoint_outage_on_restart() + { + using var outage = new CancellationTokenSource(); + var unavailable = true; + var sut = CreateSut( + archiveCheckpoint: ChunkSize, + onGetCheckpoint: () => + { + if (!unavailable) + { + return; + } + + outage.Cancel(); + throw new IOException("archive unavailable"); + }); + + await Assert.ThrowsAnyAsync(() => sut.Catchup.Run(outage.Token)); + Assert.Equal(0, sut.WriterCheckpoint.Read()); + Assert.Equal(0, sut.ChaserCheckpoint.Read()); + + unavailable = false; + await sut.Catchup.Run(); + + Assert.Equal(ChunkSize, sut.WriterCheckpoint.Read()); + Assert.Equal(ChunkSize, sut.ChaserCheckpoint.Read()); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task retries_corrupt_or_truncated_chunk_on_restart(bool truncated) + { + FakeArchiveStorage archive = null; + var firstRead = true; + using var cancellation = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); + var sut = CreateSut( + archiveCheckpoint: ChunkSize, + retryInterval: TimeSpan.FromMinutes(1), + getChunk: chunkFile => + { + if (!firstRead) + { + return archive.CreateChunk(chunkFile); + } + + firstRead = false; + var bytes = archive.CreateChunkBytes(chunkFile); + if (truncated) + { + Array.Resize(ref bytes, bytes.Length - 1); + } + else + { + bytes[ChunkHeader.Size] ^= byte.MaxValue; + } + return new MemoryStream(bytes); + }); + archive = sut.Archive; + + await Assert.ThrowsAnyAsync(() => sut.Catchup.Run(cancellation.Token)); + Assert.Equal(0, sut.WriterCheckpoint.Read()); + Assert.Equal(0, sut.ChaserCheckpoint.Read()); + + await sut.Catchup.Run(); + + Assert.Equal(ChunkSize, sut.WriterCheckpoint.Read()); + Assert.Equal(ChunkSize, sut.ChaserCheckpoint.Read()); + } + + [Fact] + public async Task backs_off_when_a_listed_chunk_remains_missing() + { + var attempts = 0; + var secondAttempt = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var cancellation = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); + var sut = CreateSut( + archiveCheckpoint: ChunkSize, + retryInterval: TimeSpan.FromMinutes(1), + onGetChunk: _ => + { + if (Interlocked.Increment(ref attempts) == 2) + { + secondAttempt.TrySetResult(); + } + + throw new ChunkDeletedException(); + }); + + await Assert.ThrowsAnyAsync(() => sut.Catchup.Run(cancellation.Token)); + + Assert.False(secondAttempt.Task.IsCompleted); + Assert.Equal(1, attempts); + } + [Fact] public async Task propagates_cancellation_while_fetching_chunk() { @@ -275,3 +419,18 @@ private IEnumerable ListBackedUpChunks() .Order(); } } + +internal sealed class RecordingArchiveMetrics : IArchiveMetrics +{ + public List Failures { get; } = []; + public List Retries { get; } = []; + + public void SetReplicationPosition(long position) { } + public void SetCheckpoint(long position) { } + public void SetUncommittedChunks(int count) { } + public void SetQueuedChunks(int count) { } + public void SetActiveChunks(int count) { } + public void RecordRetry(ArchiveOperation operation) => Retries.Add(operation); + public void RecordFailure(ArchiveOperation operation) => Failures.Add(operation); + public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) { } +} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs index 2ede6e7000..da8525f1a2 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.IO; using System.Linq; +using System.Security.Cryptography; using System.Threading; using System.Threading.Tasks; using EventStore.Core.Services.Archive.Naming; @@ -24,6 +25,8 @@ internal class FakeArchiveStorage : IArchiveStorageWriter, IArchiveStorageReader private int _listings; private readonly Action _onGetChunk; + private readonly Action _onGetCheckpoint; + private readonly Func _getChunk; public string[] ChunkGets { @@ -39,12 +42,20 @@ public string[] ChunkGets private readonly List _chunkGets; - public FakeArchiveStorage(int chunkSize, string[] chunks, long checkpoint, Action onGetChunk) + public FakeArchiveStorage( + int chunkSize, + string[] chunks, + long checkpoint, + Action onGetChunk, + Action onGetCheckpoint = null, + Func getChunk = null) { _chunkSize = chunkSize; _chunks = chunks; _checkpoint = checkpoint; _onGetChunk = onGetChunk; + _onGetCheckpoint = onGetCheckpoint; + _getChunk = getChunk; _chunkGets = new(); } @@ -62,6 +73,7 @@ public ValueTask GetChunk(string chunkFile, long start, long end, Cancel public ValueTask GetCheckpoint(CancellationToken ct) { + _onGetCheckpoint?.Invoke(); return ValueTask.FromResult(_checkpoint); } @@ -91,14 +103,35 @@ public ValueTask GetChunk(string chunkFile, CancellationToken ct) } _onGetChunk?.Invoke(chunkFile); - var chunk = new byte[ChunkHeader.Size + _chunkSize]; + if (_getChunk is not null) + { + return ValueTask.FromResult(_getChunk(chunkFile)); + } + + return ValueTask.FromResult(CreateChunk(chunkFile)); + } + + public Stream CreateChunk(string chunkFile) + { + return new MemoryStream(CreateChunkBytes(chunkFile)); + } + + public byte[] CreateChunkBytes(string chunkFile) + { + var chunk = new byte[TFChunk.GetAlignedSize(ChunkHeader.Size + ChunkFooter.Size)]; var chunkStartNumber = _customNamingStrategy.GetIndexFor(chunkFile); var chunkEndNumber = _customNamingStrategy.GetVersionFor(chunkFile); var header = CreateChunkHeader(chunkStartNumber, chunkEndNumber); header.Format(chunk.AsSpan()[..ChunkHeader.Size]); - var stream = new MemoryStream(chunk); - return ValueTask.FromResult((Stream)stream); + var footerOffset = chunk.Length - ChunkFooter.Size; + new ChunkFooter(true, physicalDataSize: 0, logicalDataSize: 0, mapSize: 0) + .Format(chunk.AsSpan(footerOffset, ChunkFooter.Size)); + using var hash = IncrementalHash.CreateHash(HashAlgorithmName.MD5); + hash.AppendData(chunk.AsSpan(0, chunk.Length - ChunkFooter.ChecksumSize)); + new ChunkFooter(true, physicalDataSize: 0, logicalDataSize: 0, mapSize: 0, hash) + .Format(chunk.AsSpan(footerOffset, ChunkFooter.Size)); + return chunk; } public IAsyncEnumerable ListChunks(CancellationToken ct) diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveLifecycleSoakTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveLifecycleSoakTests.cs new file mode 100644 index 0000000000..6f154127df --- /dev/null +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveLifecycleSoakTests.cs @@ -0,0 +1,240 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Core.Bus; +using EventStore.Core.Data; +using EventStore.Core.Messages; +using EventStore.Core.Messaging; +using EventStore.Core.Services.Archive.Archiver; +using EventStore.Core.Services.Archive.Archiver.Unmerger; +using EventStore.Core.Services.Archive.Naming; +using EventStore.Core.Services.Archive.Storage; +using EventStore.Core.Tests.TransactionLog; +using EventStore.Core.TransactionLog.Checkpoint; +using EventStore.Core.TransactionLog.Chunks; +using EventStore.Core.TransactionLog.Chunks.TFChunk; +using EventStore.Core.TransactionLog.FileNamingStrategy; +using EventStore.Core.Transforms.Identity; +using EventStore.Plugins.Transforms; +using Xunit; + +namespace EventStore.Core.XUnit.Tests.Services.Archive; + +using ArchiveCatchupService = Core.Services.Archive.ArchiveCatchup.ArchiveCatchup; + +public class ArchiveLifecycleSoakTests : DirectoryPerTest +{ + private const int ChunkCount = 8; + private const int ChunkSize = 4096; + private const string ArchiveCheckpointFile = "archive.chk"; + + [Fact] + public async Task archives_removes_reads_and_recovers_multiple_nodes_across_restart() + { + var leaderPath = CreateDirectory("leader"); + var archivePath = CreateDirectory("archive"); + var followerPaths = new[] { CreateDirectory("follower-1"), CreateDirectory("follower-2") }; + var leaderNaming = new VersionedPatternFileNamingStrategy(leaderPath, "chunk-"); + var archiveNamer = new ArchiveChunkNamer( + new VersionedPatternFileNamingStrategy(archivePath, "chunk-")); + var archive = new LocalArchiveStorage(archivePath, archiveNamer, ArchiveCheckpointFile); + var archiveFactory = new ArchiveStorageFactoryAdapter(archive); + var chunks = await CreateCompletedChunks(leaderNaming); + var archiver = new ArchiverService(new NoOpSubscriber(), archiveFactory, new UnexpectedUnmerger(), archiveNamer); + + foreach (var chunk in chunks) + { + archiver.Handle(new SystemMessage.ChunkLoaded(chunk)); + } + + archiver.Handle(new ReplicationTrackingMessage.ReplicatedTo(chunks[^1].ChunkEndPosition)); + await WaitForCheckpoint(archive, chunks[^1].ChunkEndPosition); + + Assert.Equal( + Enumerable.Range(0, ChunkCount).Select(archiveNamer.GetFileNameFor), + await archive.ListChunks(CancellationToken.None).ToArrayAsync()); + + foreach (var chunk in chunks) + { + File.Delete(chunk.ChunkLocator); + } + + Assert.Empty(Directory.EnumerateFiles(leaderPath, "chunk-*")); + await VerifyColdReads(leaderPath, archive, chunks); + + await Task.WhenAll(followerPaths.Select(path => CatchUpNode(path, archiveFactory))); + foreach (var followerPath in followerPaths) + { + await VerifyRestart(followerPath, archiveFactory, chunks[^1].ChunkEndPosition); + await VerifyLocalChunks(followerPath); + } + + archiver.Handle(new SystemMessage.BecomeShuttingDown(Guid.NewGuid(), exitProcess: true, shutdownHttp: true)); + } + + private string CreateDirectory(string name) + { + var path = Path.Combine(Fixture.Directory, name); + Directory.CreateDirectory(path); + return path; + } + + private static async Task CreateCompletedChunks(IVersionedFileNamingStrategy namingStrategy) + { + var fileSystem = new ChunkLocalFileSystem(namingStrategy); + var chunks = new List(ChunkCount); + for (var chunkNumber = 0; chunkNumber < ChunkCount; chunkNumber++) + { + var chunk = await TFChunk.CreateNew( + fileSystem, + namingStrategy.GetFilenameFor(chunkNumber, 0), + ChunkSize, + chunkNumber, + chunkNumber, + isScavenged: true, + unbuffered: false, + writethrough: false, + reduceFileCachePressure: false, + asyncIO: false, + new TFChunkTracker.NoOp(), + new IdentityChunkTransformFactory(), + CancellationToken.None); + await chunk.CompleteScavenge([], CancellationToken.None); + chunks.Add(chunk.ChunkInfo); + chunk.Dispose(); + } + + return chunks.ToArray(); + } + + private static async Task WaitForCheckpoint(IArchiveStorageReader archive, long expected) + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + while (await archive.GetCheckpoint(timeout.Token) != expected) + { + await Task.Delay(25, timeout.Token); + } + } + + private static async Task VerifyColdReads( + string leaderPath, + IArchiveStorageReader archive, + IReadOnlyList chunks) + { + var fileSystem = new FileSystemWithArchive( + ChunkSize, + new PrefixingLocatorCodec(), + new ChunkLocalFileSystem(new VersionedPatternFileNamingStrategy(leaderPath, "chunk-")), + archive); + + for (var chunkNumber = 0; chunkNumber < chunks.Count; chunkNumber++) + { + using var chunk = await TFChunk.FromCompletedFile( + fileSystem, + $"archived-chunk-{chunkNumber}", + verifyHash: false, + unbufferedRead: false, + new TFChunkTracker.NoOp(), + static _ => new IdentityChunkTransformFactory(), + token: CancellationToken.None); + + Assert.True(chunk.IsRemote); + Assert.True(chunk.ChunkFooter.IsCompleted); + Assert.Equal(chunkNumber, chunk.ChunkHeader.ChunkStartNumber); + Assert.Equal(chunkNumber, chunk.ChunkHeader.ChunkEndNumber); + } + } + + private static async Task CatchUpNode(string dbPath, IArchiveStorageFactory archiveFactory) + { + using var checkpoints = new NodeCheckpoints(dbPath); + var catchup = CreateCatchup(dbPath, checkpoints, archiveFactory); + await catchup.Run(); + + Assert.Equal((long)ChunkCount * ChunkSize, checkpoints.Writer.Read()); + Assert.Equal((long)ChunkCount * ChunkSize, checkpoints.Chaser.Read()); + Assert.Equal(-1, checkpoints.Epoch.Read()); + } + + private static async Task VerifyRestart(string dbPath, IArchiveStorageFactory archiveFactory, long expectedCheckpoint) + { + using var checkpoints = new NodeCheckpoints(dbPath, mustExist: true); + Assert.Equal(expectedCheckpoint, checkpoints.Writer.Read()); + await CreateCatchup(dbPath, checkpoints, archiveFactory).Run(); + Assert.Equal(expectedCheckpoint, checkpoints.Writer.Read()); + Assert.Equal(expectedCheckpoint, checkpoints.Chaser.Read()); + } + + private static ArchiveCatchupService CreateCatchup( + string dbPath, + NodeCheckpoints checkpoints, + IArchiveStorageFactory archiveFactory) => + new( + dbPath, + checkpoints.Writer, + checkpoints.Chaser, + checkpoints.Epoch, + ChunkSize, + new VersionedPatternFileNamingStrategy(dbPath, "chunk-"), + archiveFactory); + + private static async Task VerifyLocalChunks(string dbPath) + { + var namingStrategy = new VersionedPatternFileNamingStrategy(dbPath, "chunk-"); + var fileSystem = new ChunkLocalFileSystem(namingStrategy); + for (var chunkNumber = 0; chunkNumber < ChunkCount; chunkNumber++) + { + using var chunk = await TFChunk.FromCompletedFile( + fileSystem, + namingStrategy.GetFilenameFor(chunkNumber, 1), + verifyHash: true, + unbufferedRead: false, + new TFChunkTracker.NoOp(), + static _ => new IdentityChunkTransformFactory(), + token: CancellationToken.None); + Assert.Equal(chunkNumber, chunk.ChunkHeader.ChunkStartNumber); + } + } + + private sealed class NodeCheckpoints : IDisposable + { + public FileCheckpoint Writer { get; } + public FileCheckpoint Chaser { get; } + public FileCheckpoint Epoch { get; } + + public NodeCheckpoints(string dbPath, bool mustExist = false) + { + Writer = new(Path.Combine(dbPath, "writer.chk"), "writer", mustExist); + Chaser = new(Path.Combine(dbPath, "chaser.chk"), "chaser", mustExist); + Epoch = new(Path.Combine(dbPath, "epoch.chk"), "epoch", mustExist, initValue: -1); + } + + public void Dispose() + { + Writer.Close(flush: true); + Chaser.Close(flush: true); + Epoch.Close(flush: true); + } + } + + private sealed class ArchiveStorageFactoryAdapter(LocalArchiveStorage storage) : IArchiveStorageFactory + { + public IArchiveStorageReader CreateReader() => storage; + public IArchiveStorageWriter CreateWriter() => storage; + } + + private sealed class NoOpSubscriber : ISubscriber + { + public void Subscribe(IAsyncHandle handler) where T : Message { } + public void Unsubscribe(IAsyncHandle handler) where T : Message { } + } + + private sealed class UnexpectedUnmerger : IChunkUnmerger + { + public IAsyncEnumerable Unmerge(string chunkPath, int chunkStartNumber, int chunkEndNumber) => + throw new InvalidOperationException("The lifecycle test archives only individual chunks."); + } +} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveMetricsTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveMetricsTests.cs new file mode 100644 index 0000000000..4c7afaf769 --- /dev/null +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveMetricsTests.cs @@ -0,0 +1,102 @@ +using System; +using System.Diagnostics.Metrics; +using System.Linq; +using EventStore.Core.Services.Archive; +using EventStore.Core.XUnit.Tests.Metrics; +using TrogonEventStore.SemanticConventions; +using Xunit; + +namespace EventStore.Core.XUnit.Tests.Services.Archive; + +public class ArchiveMetricsTests +{ + [Fact] + public void observes_checkpoint_lag_and_pending_chunks() + { + using var meter = new Meter($"{typeof(ArchiveMetricsTests)}"); + using var listener = new TestMeterListener(meter); + var sut = new ArchiveMetrics(meter); + + sut.SetReplicationPosition(500); + sut.SetCheckpoint(125); + sut.SetUncommittedChunks(2); + sut.SetQueuedChunks(3); + sut.SetActiveChunks(1); + listener.Observe(); + + Assert.Equal(375, Assert.Single(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveCheckpointLag.Name)).Value); + Assert.Equal(6, Assert.Single(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveChunkPendingCount.Name)).Value); + } + + [Fact] + public void checkpoint_lag_never_reports_a_negative_value() + { + using var meter = new Meter($"{typeof(ArchiveMetricsTests)}-negative-lag"); + using var listener = new TestMeterListener(meter); + var sut = new ArchiveMetrics(meter); + + sut.SetReplicationPosition(100); + sut.SetCheckpoint(200); + listener.Observe(); + + Assert.Equal(0, Assert.Single(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveCheckpointLag.Name)).Value); + } + + [Fact] + public void does_not_publish_archiver_state_for_archive_readers() + { + using var meter = new Meter($"{typeof(ArchiveMetricsTests)}-reader"); + using var listener = new TestMeterListener(meter); + _ = new ArchiveMetrics(meter, observeArchiverState: false); + + listener.Observe(); + + Assert.Empty(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveCheckpointLag.Name)); + Assert.Empty(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveChunkPendingCount.Name)); + } + + [Fact] + public void records_retries_and_failures_by_operation() + { + using var meter = new Meter($"{typeof(ArchiveMetricsTests)}-counts"); + using var listener = new TestMeterListener(meter); + var sut = new ArchiveMetrics(meter); + + sut.RecordRetry(ArchiveOperation.StoreChunk); + sut.RecordFailure(ArchiveOperation.ReadRange); + + var retry = Assert.Single(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveRetryCount.Name)); + Assert.Equal(1, retry.Value); + Assert.Contains(retry.Tags, tag => + tag.Key == TrogonAttributeNames.ActivityName && (string)tag.Value == "store-chunk"); + + var failure = Assert.Single(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveFailureCount.Name)); + Assert.Equal(1, failure.Value); + Assert.Contains(failure.Tags, tag => + tag.Key == TrogonAttributeNames.ActivityName && (string)tag.Value == "read-range"); + } + + [Fact] + public void records_remote_read_duration_and_outcome() + { + using var meter = new Meter($"{typeof(ArchiveMetricsTests)}-duration"); + using var listener = new TestMeterListener(meter); + var sut = new ArchiveMetrics(meter); + + sut.RecordRead(ArchiveOperation.ReadFull, TimeSpan.FromMilliseconds(1250), succeeded: true); + + var measurement = Assert.Single(listener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveReadDuration.Name)); + Assert.Equal(1.25, measurement.Value); + Assert.Equal( + [(TrogonAttributeNames.ActivityName, "read-full"), (TrogonAttributeNames.ActivityOutcome, "success")], + measurement.Tags.Select(tag => (tag.Key, (string)tag.Value))); + } +} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs index df2992f656..f5d356e09d 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs @@ -8,6 +8,7 @@ using EventStore.Core.Data; using EventStore.Core.Messages; using EventStore.Core.Messaging; +using EventStore.Core.Services.Archive; using EventStore.Core.Services.Archive.Archiver; using EventStore.Core.Services.Archive.Archiver.Unmerger; using EventStore.Core.Services.Archive.Naming; @@ -25,17 +26,38 @@ private static (ArchiverService, FakeArchiveStorage) CreateSut( TimeSpan? chunkStorageDelay = null, TimeSpan? checkpointStorageDelay = null, string[] existingChunks = null, - long? existingCheckpoint = null) + long? existingCheckpoint = null, + IArchiveMetrics metrics = null) { var archive = new FakeArchiveStorage( chunkStorageDelay ?? TimeSpan.Zero, checkpointStorageDelay ?? TimeSpan.Zero, existingChunks ?? Array.Empty(), existingCheckpoint ?? 0L); - var service = new ArchiverService(new FakeSubscriber(), archive, new FakeUnmerger(), new FakeArchiveChunkNamer()); + var service = new ArchiverService( + new FakeSubscriber(), archive, new FakeUnmerger(), new FakeArchiveChunkNamer(), metrics); return (service, archive); } + [Fact] + public async Task reports_archive_backlog_and_checkpoint_lag_inputs() + { + var metrics = new ArchiverRecordingMetrics(); + var (sut, archive) = CreateSut(metrics: metrics); + var chunkInfo = GetChunkInfo(0, 0); + + sut.Handle(new SystemMessage.ChunkCompleted(chunkInfo)); + sut.Handle(new ReplicationTrackingMessage.ReplicatedTo(chunkInfo.ChunkEndPosition)); + + await WaitFor(archive, numStores: 1, numCheckpoints: 1); + + Assert.Equal(chunkInfo.ChunkEndPosition, metrics.ReplicationPosition); + Assert.Equal(chunkInfo.ChunkEndPosition, metrics.Checkpoint); + Assert.Equal(1, metrics.MaxUncommittedChunks); + Assert.Equal(1, metrics.MaxQueuedChunks); + Assert.Equal(1, metrics.MaxActiveChunks); + } + private static ChunkInfo GetChunkInfo(int chunkStartNumber, int chunkEndNumber, bool complete = true, bool remote = false) { return new ChunkInfo @@ -492,3 +514,42 @@ public IAsyncEnumerable ListChunks(CancellationToken ct) return _existingChunks.ToAsyncEnumerable(); } } + +internal sealed class ArchiverRecordingMetrics : IArchiveMetrics +{ + private long _replicationPosition; + private long _checkpoint; + private int _maxUncommittedChunks; + private int _maxQueuedChunks; + private int _maxActiveChunks; + + public long ReplicationPosition => Interlocked.Read(ref _replicationPosition); + public long Checkpoint => Interlocked.Read(ref _checkpoint); + public int MaxUncommittedChunks => Volatile.Read(ref _maxUncommittedChunks); + public int MaxQueuedChunks => Volatile.Read(ref _maxQueuedChunks); + public int MaxActiveChunks => Volatile.Read(ref _maxActiveChunks); + + public void SetReplicationPosition(long position) => Interlocked.Exchange(ref _replicationPosition, position); + public void SetCheckpoint(long position) => Interlocked.Exchange(ref _checkpoint, position); + public void SetUncommittedChunks(int count) => UpdateMax(ref _maxUncommittedChunks, count); + public void SetQueuedChunks(int count) => UpdateMax(ref _maxQueuedChunks, count); + public void SetActiveChunks(int count) => UpdateMax(ref _maxActiveChunks, count); + public void RecordRetry(ArchiveOperation operation) { } + public void RecordFailure(ArchiveOperation operation) { } + public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) { } + + private static void UpdateMax(ref int target, int value) + { + var current = Volatile.Read(ref target); + while (value > current) + { + var observed = Interlocked.CompareExchange(ref target, value, current); + if (observed == current) + { + return; + } + + current = observed; + } + } +} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs new file mode 100644 index 0000000000..eb05a4682e --- /dev/null +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs @@ -0,0 +1,227 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Core.Services.Archive; +using EventStore.Core.Services.Archive.Naming; +using EventStore.Core.Services.Archive.Storage; +using Xunit; + +namespace EventStore.Core.XUnit.Tests.Services.Archive.Storage; + +public class ArchiveStorageReaderMetricsTests +{ + [Theory] + [InlineData(ArchiveOperation.ReadMetadata)] + [InlineData(ArchiveOperation.ReadFull)] + [InlineData(ArchiveOperation.ReadRange)] + public async Task records_successful_remote_requests(ArchiveOperation operation) + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(), metrics); + + await InvokeAndConsume(sut, metrics, operation); + + var read = Assert.Single(metrics.Reads); + Assert.Equal(operation, read.Operation); + Assert.True(read.Succeeded); + Assert.True(read.Duration >= TimeSpan.Zero); + Assert.Empty(metrics.Failures); + } + + [Fact] + public async Task records_stream_failures_during_remote_content_reads() + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(streamFails: true), metrics); + await using var stream = await sut.GetChunk("chunk", CancellationToken.None); + + Assert.Empty(metrics.Reads); + await Assert.ThrowsAsync(async () => + await stream.ReadExactlyAsync(new byte[1], CancellationToken.None)); + + Assert.Equal(ArchiveOperation.ReadFull, Assert.Single(metrics.Failures)); + Assert.False(Assert.Single(metrics.Reads).Succeeded); + } + + [Theory] + [InlineData(ArchiveOperation.ReadMetadata)] + [InlineData(ArchiveOperation.ReadFull)] + [InlineData(ArchiveOperation.ReadRange)] + public async Task records_failed_remote_requests(ArchiveOperation operation) + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(fail: true), metrics); + + await Assert.ThrowsAsync(() => Invoke(sut, operation)); + + Assert.Equal(operation, Assert.Single(metrics.Failures)); + var read = Assert.Single(metrics.Reads); + Assert.Equal(operation, read.Operation); + Assert.False(read.Succeeded); + } + + [Theory] + [InlineData(ArchiveOperation.ReadMetadata)] + [InlineData(ArchiveOperation.ReadFull)] + [InlineData(ArchiveOperation.ReadRange)] + public async Task does_not_record_cancelled_remote_requests_as_failures(ArchiveOperation operation) + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(cancel: true), metrics); + + await Assert.ThrowsAnyAsync(() => Invoke(sut, operation)); + + Assert.Empty(metrics.Failures); + Assert.Empty(metrics.Reads); + } + + [Fact] + public async Task does_not_record_cancelled_content_reads_as_failures() + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(streamCancels: true), metrics); + await using var stream = await sut.GetChunk("chunk", CancellationToken.None); + + await Assert.ThrowsAnyAsync(async () => + await stream.ReadExactlyAsync(new byte[1], CancellationToken.None)); + + Assert.Empty(metrics.Failures); + Assert.Empty(metrics.Reads); + } + + [Fact] + public async Task does_not_record_cancelled_synchronous_disposal_as_a_failure() + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(disposeCancels: true), metrics); + var stream = await sut.GetChunk("chunk", CancellationToken.None); + + Assert.Throws(stream.Dispose); + + Assert.Empty(metrics.Failures); + Assert.Empty(metrics.Reads); + } + + private static async Task Invoke(IArchiveStorageReader reader, ArchiveOperation operation) + { + switch (operation) + { + case ArchiveOperation.ReadMetadata: + await reader.GetChunkLength("chunk", CancellationToken.None); + break; + case ArchiveOperation.ReadFull: + await reader.GetChunk("chunk", CancellationToken.None); + break; + case ArchiveOperation.ReadRange: + await reader.GetChunk("chunk", 0, 1, CancellationToken.None); + break; + default: + throw new ArgumentOutOfRangeException(nameof(operation), operation, null); + } + } + + private static async Task InvokeAndConsume( + IArchiveStorageReader reader, + RecordingArchiveMetrics metrics, + ArchiveOperation operation) + { + if (operation == ArchiveOperation.ReadMetadata) + { + await reader.GetChunkLength("chunk", CancellationToken.None); + return; + } + + await using var stream = operation switch + { + ArchiveOperation.ReadFull => await reader.GetChunk("chunk", CancellationToken.None), + ArchiveOperation.ReadRange => await reader.GetChunk("chunk", 0, 1, CancellationToken.None), + _ => throw new ArgumentOutOfRangeException(nameof(operation), operation, null) + }; + Assert.Empty(metrics.Reads); + await stream.CopyToAsync(Stream.Null); + } + + private sealed class StubReader( + bool fail = false, + bool streamFails = false, + bool cancel = false, + bool streamCancels = false, + bool disposeCancels = false) : IArchiveStorageReader + { + public IArchiveChunkNamer ChunkNamer { get; } = new StubChunkNamer(); + + public ValueTask GetCheckpoint(CancellationToken ct) => ValueTask.FromResult(0L); + + public ValueTask GetChunkLength(string chunkFile, CancellationToken ct) => + cancel + ? ValueTask.FromException(new OperationCanceledException()) + : fail ? ValueTask.FromException(new InvalidOperationException()) : ValueTask.FromResult(1L); + + public ValueTask GetChunk(string chunkFile, long start, long end, CancellationToken ct) => + GetChunk(chunkFile, ct); + + public ValueTask GetChunk(string chunkFile, CancellationToken ct) => + cancel + ? ValueTask.FromException(new OperationCanceledException()) + : fail + ? ValueTask.FromException(new InvalidOperationException()) + : ValueTask.FromResult( + streamFails + ? new FailingReadStream() + : streamCancels + ? new CanceledReadStream() + : disposeCancels ? new CanceledDisposeStream() : new MemoryStream([1])); + + public async IAsyncEnumerable ListChunks([EnumeratorCancellation] CancellationToken ct) + { + await Task.CompletedTask; + yield break; + } + } + + private sealed class StubChunkNamer : IArchiveChunkNamer + { + public string Prefix => "chunk-"; + public string GetFileNameFor(int logicalChunkNumber) => $"chunk-{logicalChunkNumber}"; + } + + private sealed class FailingReadStream : MemoryStream + { + public override ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) => + ValueTask.FromException(new IOException("remote read failed")); + } + + private sealed class CanceledReadStream : MemoryStream + { + public override ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) => + ValueTask.FromException(new OperationCanceledException()); + } + + private sealed class CanceledDisposeStream : MemoryStream + { + protected override void Dispose(bool disposing) => throw new OperationCanceledException(); + } + + private sealed class RecordingArchiveMetrics : IArchiveMetrics + { + public List Failures { get; } = []; + public List<(ArchiveOperation Operation, TimeSpan Duration, bool Succeeded)> Reads { get; } = []; + + public void SetReplicationPosition(long position) { } + public void SetCheckpoint(long position) { } + public void SetUncommittedChunks(int count) { } + public void SetQueuedChunks(int count) { } + public void SetActiveChunks(int count) { } + public void RecordRetry(ArchiveOperation operation) { } + public void RecordFailure(ArchiveOperation operation) => Failures.Add(operation); + public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) => + Reads.Add((operation, duration, succeeded)); + } +} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs index ea64afc527..f700ed9b14 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs @@ -1,5 +1,10 @@ +using System; using System.IO; using System.Security.Cryptography; +using System.Threading.Tasks; +using Amazon.Runtime; +using Amazon.S3; +using Amazon.S3.Model; using EventStore.Core.Services.Archive; using EventStore.Core.Services.Archive.Naming; using EventStore.Core.Services.Archive.Storage; @@ -10,6 +15,8 @@ namespace EventStore.Core.XUnit.Tests.Services.Archive.Storage; public abstract class ArchiveStorageTestsBase : DirectoryPerTest { protected const string ChunkPrefix = "chunk-"; + private readonly string _bucket = $"archive-contract-{Guid.NewGuid():N}"; + private AmazonS3Client _s3Client; protected string ArchivePath => Path.Combine(Fixture.Directory, "archive"); protected string DbPath => Path.Combine(Fixture.Directory, "db"); protected abstract StorageType StorageType { get; } @@ -20,7 +27,53 @@ public ArchiveStorageTestsBase() Directory.CreateDirectory(DbPath); } - protected IArchiveStorageFactory CreateSutFactory(StorageType storageType) + public override async Task InitializeAsync() + { + await base.InitializeAsync(); + + if (StorageType != StorageType.S3) + { + return; + } + + var options = CreateS3Options(); + _s3Client = new AmazonS3Client( + new BasicAWSCredentials(options.AccessKeyId, options.SecretAccessKey), + new AmazonS3Config + { + ServiceURL = options.ServiceUrl, + AuthenticationRegion = options.Region, + ForcePathStyle = true, + }); + + await _s3Client.PutBucketAsync(new PutBucketRequest { BucketName = options.Bucket }); + } + + public override async Task DisposeAsync() + { + try + { + if (_s3Client is not null) + { + var objects = await _s3Client.ListObjectsV2Async(new ListObjectsV2Request { BucketName = _bucket }); + foreach (var item in objects.S3Objects ?? []) + { + await _s3Client.DeleteObjectAsync(_bucket, item.Key); + } + + await _s3Client.DeleteBucketAsync(_bucket); + } + } + finally + { + _s3Client?.Dispose(); + await base.DisposeAsync(); + } + } + + protected IArchiveStorageFactory CreateSutFactory( + StorageType storageType, + IArchiveMetrics archiveMetrics = null) { var namingStrategy = new VersionedPatternFileNamingStrategy(ArchivePath, ChunkPrefix); var chunkNamer = new ArchiveChunkNamer(namingStrategy); @@ -28,12 +81,27 @@ protected IArchiveStorageFactory CreateSutFactory(StorageType storageType) new() { StorageType = storageType, - S3 = new() { Bucket = "archiver-unit-tests", Region = "eu-west-1", } + S3 = CreateS3Options(), }, - chunkNamer); + chunkNamer, + archiveMetrics); return factory; } + private S3Options CreateS3Options() => new() + { + Bucket = _bucket, + Region = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_REGION"), + AccessKeyId = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_ACCESS_KEY"), + SecretAccessKey = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_SECRET_KEY"), + ServiceUrl = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_ENDPOINT"), + }; + + private static string GetRequiredEnvironmentVariable(string name) => + Environment.GetEnvironmentVariable(name) is { Length: > 0 } value + ? value + : throw new InvalidOperationException($"{name} must be configured to run S3 contract tests"); + protected IArchiveStorageWriter CreateWriterSut(StorageType storageType) => CreateSutFactory(storageType).CreateWriter(); diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs new file mode 100644 index 0000000000..80f77aaa26 --- /dev/null +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs @@ -0,0 +1,135 @@ +using System; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using Amazon.Runtime; +using Amazon.S3; +using Amazon.S3.Model; +using EventStore.Core.Services.Archive; +using EventStore.Core.Services.Archive.Naming; +using EventStore.Core.Services.Archive.Storage; +using EventStore.Core.TransactionLog.FileNamingStrategy; +using Xunit; + +namespace EventStore.Core.XUnit.Tests.Services.Archive.Storage; + +#if RUN_S3_TESTS +public class S3RestartRecoveryTests +{ + private const string ArchiveCheckpointFile = "archive.chk"; + private const string ChunkFile = "chunk-000000.000000"; + private const long Checkpoint = 4_194_304; + private static readonly byte[] Payload = Enumerable.Range(0, 4096).Select(x => (byte)(x % 251)).ToArray(); + + [Fact] + public async Task executes_the_requested_restart_recovery_phase() + { + var phase = GetRequiredEnvironmentVariable("EVENTSTORE_S3_RECOVERY_PHASE"); + var options = CreateOptions(); + + switch (phase) + { + case "seed": + await Seed(options); + break; + case "unavailable": + await AssertUnavailable(options); + break; + case "verify-cleanup": + await VerifyAndCleanup(options); + break; + default: + throw new InvalidOperationException($"Unsupported restart recovery phase: {phase}"); + } + } + + private static async Task Seed(S3Options options) + { + using var client = CreateClient(options); + await client.PutBucketAsync(new PutBucketRequest { BucketName = options.Bucket }); + + var localChunk = Path.GetTempFileName(); + try + { + await File.WriteAllBytesAsync(localChunk, Payload); + var writer = new S3Writer(options, ArchiveCheckpointFile); + Assert.True(await writer.StoreChunk(localChunk, ChunkFile, CancellationToken.None)); + Assert.True(await writer.SetCheckpoint(Checkpoint, CancellationToken.None)); + } + finally + { + File.Delete(localChunk); + } + } + + private static async Task AssertUnavailable(S3Options options) + { + var reader = CreateReader(options); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + var stopwatch = Stopwatch.StartNew(); + + await Assert.ThrowsAnyAsync(async () => + await reader.GetCheckpoint(timeout.Token)); + + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(10), + $"Storage unavailability was not detected within the bounded interval: {stopwatch.Elapsed}"); + } + + private static async Task VerifyAndCleanup(S3Options options) + { + using var client = CreateClient(options); + try + { + var reader = CreateReader(options); + Assert.Equal(Checkpoint, await reader.GetCheckpoint(CancellationToken.None)); + + await using var chunk = await reader.GetChunk(ChunkFile, CancellationToken.None); + using var copy = new MemoryStream(); + await chunk.CopyToAsync(copy); + Assert.Equal(Payload, copy.ToArray()); + } + finally + { + var objects = await client.ListObjectsV2Async(new ListObjectsV2Request { BucketName = options.Bucket }); + foreach (var item in objects.S3Objects ?? []) + { + await client.DeleteObjectAsync(options.Bucket, item.Key); + } + await client.DeleteBucketAsync(options.Bucket); + } + } + + private static S3Reader CreateReader(S3Options options) + { + var namingStrategy = new VersionedPatternFileNamingStrategy(Path.GetTempPath(), "chunk-"); + return new S3Reader(options, new ArchiveChunkNamer(namingStrategy), ArchiveCheckpointFile); + } + + private static AmazonS3Client CreateClient(S3Options options) => + new( + new BasicAWSCredentials(options.AccessKeyId, options.SecretAccessKey), + new AmazonS3Config + { + ServiceURL = options.ServiceUrl, + AuthenticationRegion = options.Region, + ForcePathStyle = true, + }); + + private static S3Options CreateOptions() => new() + { + Bucket = GetRequiredEnvironmentVariable("EVENTSTORE_S3_RECOVERY_BUCKET"), + Region = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_REGION"), + AccessKeyId = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_ACCESS_KEY"), + SecretAccessKey = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_SECRET_KEY"), + ServiceUrl = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_ENDPOINT"), + }; + + private static string GetRequiredEnvironmentVariable(string name) => + Environment.GetEnvironmentVariable(name) is { Length: > 0 } value + ? value + : throw new InvalidOperationException($"{name} must be configured to run the S3 restart recovery gate"); +} +#endif diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs index 7e7078e308..317e856790 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs @@ -1,4 +1,14 @@ +using System; +using System.Diagnostics.Metrics; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; using EventStore.Core.Services.Archive; +using EventStore.Core.Services.Archive.Storage.Exceptions; +using EventStore.Core.XUnit.Tests.Metrics; +using TrogonEventStore.SemanticConventions; +using Xunit; namespace EventStore.Core.XUnit.Tests.Services.Archive.Storage; @@ -12,4 +22,63 @@ public class S3WriterTests : ArchiveStorageWriterTests { protected override StorageType StorageType => StorageType.S3; } + +public class S3MetricsTests : ArchiveStorageTestsBase +{ + protected override StorageType StorageType => StorageType.S3; + + [Fact] + public async Task records_remote_reads_and_failures_through_the_s3_factory() + { + using var meter = new Meter($"{typeof(S3MetricsTests)}"); + using var durationListener = new TestMeterListener(meter); + using var failureListener = new TestMeterListener(meter); + var metrics = new ArchiveMetrics(meter); + var factory = CreateSutFactory(StorageType, metrics); + var writer = factory.CreateWriter(); + var reader = factory.CreateReader(); + var chunkFile = reader.ChunkNamer.GetFileNameFor(0); + + Assert.True(await writer.StoreChunk( + CreateLocalChunk(0, 0), + chunkFile, + CancellationToken.None)); + + await using (var stream = await reader.GetChunk(chunkFile, CancellationToken.None)) + { + await stream.CopyToAsync(Stream.Null); + } + + await using (var stream = await reader.GetChunk(chunkFile, 100, 200, CancellationToken.None)) + { + await stream.CopyToAsync(Stream.Null); + } + + await Assert.ThrowsAsync(async () => + { + await reader.GetChunk("missing-chunk", CancellationToken.None); + }); + + var durations = durationListener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveReadDuration.Name); + Assert.Contains(durations, measurement => HasTags(measurement, "read-full", "success")); + Assert.Contains(durations, measurement => HasTags(measurement, "read-range", "success")); + Assert.Contains(durations, measurement => HasTags(measurement, "read-full", "error")); + + var failure = Assert.Single(failureListener.RetrieveMeasurements( + MetricDefinitions.TrogonEventstoreArchiveFailureCount.Name)); + Assert.Equal(1, failure.Value); + Assert.Contains(failure.Tags, tag => + tag.Key == TrogonAttributeNames.ActivityName && (string)tag.Value == "read-full"); + } + + private static bool HasTags( + TestMeterListener.TestMeasurement measurement, + string activity, + string outcome) => + measurement.Tags.Any(tag => + tag.Key == TrogonAttributeNames.ActivityName && (string)tag.Value == activity) && + measurement.Tags.Any(tag => + tag.Key == TrogonAttributeNames.ActivityOutcome && (string)tag.Value == outcome); +} #endif diff --git a/src/EventStore.Core/ClusterVNode.cs b/src/EventStore.Core/ClusterVNode.cs index 39e9b9d2b8..0398e37dc1 100644 --- a/src/EventStore.Core/ClusterVNode.cs +++ b/src/EventStore.Core/ClusterVNode.cs @@ -329,13 +329,18 @@ public ClusterVNode(ClusterVNodeOptions options, NodeInfo = new VNodeInfo(instanceId.Value, debugIndex, intTcp, intSecIp, extTcp, extSecIp, httpEndPoint, options.Cluster.ReadOnlyReplica); + var metricsConfiguration = MetricsConfiguration.Get(configuration); + var trackers = new Trackers(); + MetricsBootstrapper.BootstrapArchive( + metricsConfiguration, + trackers, + archiveOptions.Enabled, + options.Cluster.Archiver); var dbConfig = CreateDbConfig( out var statsHelper, out var readerThreadsCount, out var workerThreadsCount); - var trackers = new Trackers(); - var metricsConfiguration = MetricsConfiguration.Get(configuration); MetricsBootstrapper.Bootstrap(metricsConfiguration, dbConfig, trackers); static bool WatchSlowMessages(TimeSpan threshold) => threshold > TimeSpan.Zero; @@ -448,7 +453,8 @@ TFChunkDbConfig CreateDbConfig( { var archiveReader = new ArchiveStorageFactory( archiveOptions, - new ArchiveChunkNamer(_fileNamingStrategy)) + new ArchiveChunkNamer(_fileNamingStrategy), + trackers.ArchiveMetrics) .CreateReader(); chunkFileSystem = new FileSystemWithArchive( @@ -1379,7 +1385,8 @@ GossipAdvertiseInfo GetGossipAdvertiseInfo() // todo: consider if we can/should reuse the same reader elsewhere var archiveReader = new ArchiveStorageFactory( archiveOptions, - new ArchiveChunkNamer(_fileNamingStrategy)).CreateReader(); + new ArchiveChunkNamer(_fileNamingStrategy), + trackers.ArchiveMetrics).CreateReader(); chunkDeleter = new ChunkDeleter( logger: logger, archiveCheckpoint: new AdvancingCheckpoint(archiveReader.GetCheckpoint), @@ -1657,7 +1664,9 @@ IServiceCollection ConfigureNodeServices(IServiceCollection services) services .AddSingleton(telemetryService) // for correct disposal .AddSingleton(_readIndex) + .AddSingleton(Db.TransformManager) .AddSingleton(standardComponents) + .AddSingleton(trackers.ArchiveMetrics) .AddSingleton(authorizationGateway) .AddSingleton(certificateProvider) .AddSingleton>(new List { new IdentityDbTransform() }) diff --git a/src/EventStore.Core/Configuration/ClusterVNodeOptions.cs b/src/EventStore.Core/Configuration/ClusterVNodeOptions.cs index dcb6e5115d..28e4595ddd 100644 --- a/src/EventStore.Core/Configuration/ClusterVNodeOptions.cs +++ b/src/EventStore.Core/Configuration/ClusterVNodeOptions.cs @@ -415,7 +415,7 @@ public record ClusterOptions "or accept writes from clients.")] public bool ReadOnlyReplica { get; init; } = false; - [Description("Sets this node as an Archiver node. Requires ReadOnlyReplica to be true. Experimental.")] + [Description("Sets this node as an Archiver node. Requires ReadOnlyReplica to be true.")] public bool Archiver { get; init; } = false; [Description("Allow more nodes than the cluster size to join the cluster as clones. " + diff --git a/src/EventStore.Core/MetricsBootstrapper.cs b/src/EventStore.Core/MetricsBootstrapper.cs index 3eba419d01..00f2c0e4e4 100644 --- a/src/EventStore.Core/MetricsBootstrapper.cs +++ b/src/EventStore.Core/MetricsBootstrapper.cs @@ -6,6 +6,7 @@ using EventStore.Core.Diagnostics; using EventStore.Core.Index; using EventStore.Core.Metrics; +using EventStore.Core.Services.Archive; using EventStore.Core.Services.VNode; using EventStore.Core.TransactionLog; using EventStore.Core.TransactionLog.Checkpoint; @@ -18,6 +19,7 @@ namespace EventStore.Core; public class Trackers { + internal Meter CoreMeter { get; set; } public IInaugurationStatusTracker InaugurationStatusTracker { get; set; } = new NodeStatusTracker.NoOp(); public IIndexStatusTracker IndexStatusTracker { get; set; } = new IndexStatusTracker.NoOp(); public INodeStatusTracker NodeStatusTracker { get; set; } = new NodeStatusTracker.NoOp(); @@ -34,6 +36,7 @@ public class Trackers public IElectionCounterTracker ElectionCounterTracker { get; set; } = new ElectionsCounterTracker.NoOp(); public IPersistentSubscriptionTracker PersistentSubscriptionTracker { get; set; } = IPersistentSubscriptionTracker.NoOp; + public IArchiveMetrics ArchiveMetrics { get; set; } = IArchiveMetrics.NoOp; } public class GrpcTrackers @@ -68,6 +71,21 @@ public class GossipTrackers public static class MetricsBootstrapper { + public static void BootstrapArchive( + Conf conf, + Trackers trackers, + bool archiveEnabled, + bool isArchiver) + { + if (!archiveEnabled || conf.ExpectedScrapeIntervalSeconds <= 0) + { + return; + } + + trackers.CoreMeter ??= TelemetryMeterFactory.Create(TelemetryMeterInstrumentation.CoreName); + trackers.ArchiveMetrics = new ArchiveMetrics(trackers.CoreMeter, observeArchiverState: isArchiver); + } + public static void Bootstrap( Conf conf, TFChunkDbConfig dbConfig, @@ -84,7 +102,8 @@ public static void Bootstrap( return; } - var coreMeter = TelemetryMeterFactory.Create(TelemetryMeterInstrumentation.CoreName); + var coreMeter = trackers.CoreMeter ??= + TelemetryMeterFactory.Create(TelemetryMeterInstrumentation.CoreName); var statusMetric = new StatusMetric(coreMeter, MetricDefinitions.TrogonEventstoreComponentStatus); var grpcMethodMetric = new DurationMetric(coreMeter, MetricDefinitions.TrogonEventstoreGrpcServerCallDuration); var gossipLatencyMetric = new DurationMetric(coreMeter, MetricDefinitions.TrogonEventstoreGossipExchangeDuration); diff --git a/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs b/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs index 044ee60129..0fc3bfba7b 100644 --- a/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs +++ b/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs @@ -7,7 +7,10 @@ using EventStore.Core.Services.Archive.Storage.Exceptions; using EventStore.Core.TransactionLog.Checkpoint; using EventStore.Core.TransactionLog.Chunks; +using EventStore.Core.TransactionLog.Chunks.TFChunk; using EventStore.Core.TransactionLog.FileNamingStrategy; +using EventStore.Core.Transforms; +using EventStore.Plugins.Transforms; using Serilog; namespace EventStore.Core.Services.Archive.ArchiveCatchup; @@ -30,9 +33,12 @@ public class ArchiveCatchup : IClusterVNodeStartupTask private readonly int _chunkSize; private readonly IVersionedFileNamingStrategy _fileNamingStrategy; private readonly IArchiveStorageReader _archiveReader; + private readonly IArchiveMetrics _metrics; + private readonly Func _getTransformFactory; + private readonly TimeSpan _retryInterval; private static readonly ILogger Log = Serilog.Log.ForContext(); - private static readonly TimeSpan RetryInterval = TimeSpan.FromMinutes(1); + private static readonly TimeSpan DefaultRetryInterval = TimeSpan.FromMinutes(1); public ArchiveCatchup( string dbPath, @@ -41,7 +47,10 @@ public ArchiveCatchup( ICheckpoint epochCheckpoint, int chunkSize, IVersionedFileNamingStrategy fileNamingStrategy, - IArchiveStorageFactory archiveStorageFactory) + IArchiveStorageFactory archiveStorageFactory, + IArchiveMetrics metrics = null, + Func getTransformFactory = null, + TimeSpan? retryInterval = null) { _dbPath = dbPath; _writerCheckpoint = writerCheckpoint; @@ -50,6 +59,9 @@ public ArchiveCatchup( _chunkSize = chunkSize; _fileNamingStrategy = fileNamingStrategy; _archiveReader = archiveStorageFactory.CreateReader(); + _metrics = metrics ?? IArchiveMetrics.NoOp; + _getTransformFactory = getTransformFactory ?? DbTransformManager.Default.GetFactoryForExistingChunk; + _retryInterval = retryInterval ?? DefaultRetryInterval; } public async Task Run(CancellationToken ct = default) @@ -156,8 +168,10 @@ private async Task GetArchiveCheckpoint(CancellationToken ct) } catch (Exception ex) { - Log.Error(ex, "Failed to get archive checkpoint. Retrying in: {interval}", RetryInterval); - await Task.Delay(RetryInterval, ct); + _metrics.RecordFailure(ArchiveOperation.CatchUpCheckpoint); + _metrics.RecordRetry(ArchiveOperation.CatchUpCheckpoint); + Log.Error(ex, "Failed to get archive checkpoint. Retrying in: {interval}", _retryInterval); + await Task.Delay(_retryInterval, ct); } } while (true); } @@ -183,11 +197,12 @@ private async Task FetchAndCommitChunk(string chunkFile, CancellationToken private async Task FetchChunk(string chunkFile, string destinationPath, CancellationToken ct) { + string tempPath = null; try { Log.Information("Fetching {chunk} from the archive", chunkFile); - var tempPath = Path.Combine(_dbPath, Guid.NewGuid() + ".archive.tmp"); + tempPath = Path.Combine(_dbPath, Guid.NewGuid() + ".archive.tmp"); await using (var inputStream = await _archiveReader.GetChunk(chunkFile, ct)) { @@ -203,8 +218,19 @@ private async Task FetchChunk(string chunkFile, string destinationPath, Ca }); await inputStream.CopyToAsync(outputStream, ct); + outputStream.SetLength(outputStream.Position); } + using (await TFChunk.FromCompletedFile( + new ChunkLocalFileSystem(_fileNamingStrategy), + tempPath, + verifyHash: true, + unbufferedRead: false, + tracker: new TFChunkTracker.NoOp(), + getTransformFactory: _getTransformFactory, + token: ct)) + { } + if (File.Exists(destinationPath)) { var backupPath = $"{destinationPath}.archive.bkup"; @@ -214,14 +240,18 @@ private async Task FetchChunk(string chunkFile, string destinationPath, Ca } File.Move(tempPath, destinationPath); + tempPath = null; return true; } catch (ChunkDeletedException) { + _metrics.RecordFailure(ArchiveOperation.CatchUpChunk); + _metrics.RecordRetry(ArchiveOperation.CatchUpChunk); Log.Warning( - "Failed to fetch {chunk} from the archive as it was deleted. This can happen if the archive is being scavenged.", - chunkFile); + "Failed to fetch {chunk} from the archive as it was deleted. This can happen if the archive is being scavenged. Retrying in {interval}.", + chunkFile, _retryInterval); + await Task.Delay(_retryInterval, ct); return false; } catch (OperationCanceledException) @@ -230,10 +260,19 @@ private async Task FetchChunk(string chunkFile, string destinationPath, Ca } catch (Exception ex) { - Log.Error(ex, "Failed to fetch {chunk} from the archive. Retrying in {interval}", chunkFile, RetryInterval); - await Task.Delay(RetryInterval, ct); + _metrics.RecordFailure(ArchiveOperation.CatchUpChunk); + _metrics.RecordRetry(ArchiveOperation.CatchUpChunk); + Log.Error(ex, "Failed to fetch {chunk} from the archive. Retrying in {interval}", chunkFile, _retryInterval); + await Task.Delay(_retryInterval, ct); return false; } + finally + { + if (tempPath is not null) + { + File.Delete(tempPath); + } + } } private async Task CommitChunk(string chunkPath, CancellationToken ct) diff --git a/src/EventStore.Core/Services/Archive/ArchiveMetrics.cs b/src/EventStore.Core/Services/Archive/ArchiveMetrics.cs new file mode 100644 index 0000000000..1d3a9c9f81 --- /dev/null +++ b/src/EventStore.Core/Services/Archive/ArchiveMetrics.cs @@ -0,0 +1,149 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Threading; +using EventStore.Core.Metrics; +using TrogonEventStore.SemanticConventions; + +namespace EventStore.Core.Services.Archive; + +public enum ArchiveOperation +{ + StoreChunk, + SetCheckpoint, + LoadCheckpoint, + ReadMetadata, + ReadFull, + ReadRange, + CatchUpCheckpoint, + CatchUpChunk, + Service +} + +public interface IArchiveMetrics +{ + void SetReplicationPosition(long position); + void SetCheckpoint(long position); + void SetUncommittedChunks(int count); + void SetQueuedChunks(int count); + void SetActiveChunks(int count); + void RecordRetry(ArchiveOperation operation); + void RecordFailure(ArchiveOperation operation); + void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded); + + public static IArchiveMetrics NoOp { get; } = new NoOpArchiveMetrics(); + + private sealed class NoOpArchiveMetrics : IArchiveMetrics + { + public void SetReplicationPosition(long position) { } + public void SetCheckpoint(long position) { } + public void SetUncommittedChunks(int count) { } + public void SetQueuedChunks(int count) { } + public void SetActiveChunks(int count) { } + public void RecordRetry(ArchiveOperation operation) { } + public void RecordFailure(ArchiveOperation operation) { } + public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) { } + } +} + +public sealed class ArchiveMetrics : IArchiveMetrics +{ + private readonly Counter _retries; + private readonly Counter _failures; + private readonly Histogram _readDuration; + private long _replicationPosition; + private long _checkpoint; + private int _uncommittedChunks; + private int _queuedChunks; + private int _activeChunks; + + public ArchiveMetrics(Meter meter, bool observeArchiverState = true) + { + ArgumentNullException.ThrowIfNull(meter); + + if (observeArchiverState) + { + var checkpointLag = MetricDefinitions.TrogonEventstoreArchiveCheckpointLag; + checkpointLag.EnsureInstrumentKind(MetricInstrumentKind.Gauge); + meter.CreateObservableGauge( + checkpointLag.Name, + ObserveCheckpointLag, + checkpointLag.Unit, + checkpointLag.Description); + + var pendingChunks = MetricDefinitions.TrogonEventstoreArchiveChunkPendingCount; + pendingChunks.EnsureInstrumentKind(MetricInstrumentKind.UpDownCounter); + meter.CreateObservableUpDownCounter( + pendingChunks.Name, + ObservePendingChunks, + pendingChunks.Unit, + pendingChunks.Description); + } + + _retries = CreateCounter(meter, MetricDefinitions.TrogonEventstoreArchiveRetryCount); + _failures = CreateCounter(meter, MetricDefinitions.TrogonEventstoreArchiveFailureCount); + + var readDuration = MetricDefinitions.TrogonEventstoreArchiveReadDuration; + readDuration.EnsureInstrumentKind(MetricInstrumentKind.Histogram); + _readDuration = meter.CreateHistogram( + readDuration.Name, + readDuration.Unit, + readDuration.Description); + } + + public void SetReplicationPosition(long position) => + Interlocked.Exchange(ref _replicationPosition, position); + + public void SetCheckpoint(long position) => + Interlocked.Exchange(ref _checkpoint, position); + + public void SetUncommittedChunks(int count) => + Interlocked.Exchange(ref _uncommittedChunks, count); + + public void SetQueuedChunks(int count) => + Interlocked.Exchange(ref _queuedChunks, count); + + public void SetActiveChunks(int count) => + Interlocked.Exchange(ref _activeChunks, count); + + public void RecordRetry(ArchiveOperation operation) => + _retries.Add(1, ActivityName(operation)); + + public void RecordFailure(ArchiveOperation operation) => + _failures.Add(1, ActivityName(operation)); + + public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) => + _readDuration.Record( + duration.TotalSeconds, + ActivityName(operation), + new KeyValuePair(TrogonAttributeNames.ActivityOutcome, succeeded ? "success" : "error")); + + private long ObserveCheckpointLag() => + Math.Max(0, Interlocked.Read(ref _replicationPosition) - Interlocked.Read(ref _checkpoint)); + + private long ObservePendingChunks() => + Volatile.Read(ref _uncommittedChunks) + + Volatile.Read(ref _queuedChunks) + + Volatile.Read(ref _activeChunks); + + private static Counter CreateCounter(Meter meter, MetricDefinition definition) + { + definition.EnsureInstrumentKind(MetricInstrumentKind.Counter); + return meter.CreateCounter(definition.Name, definition.Unit, definition.Description); + } + + private static KeyValuePair ActivityName(ArchiveOperation operation) => + new(TrogonAttributeNames.ActivityName, operation switch + { + ArchiveOperation.StoreChunk => "store-chunk", + ArchiveOperation.SetCheckpoint => "set-checkpoint", + ArchiveOperation.LoadCheckpoint => "load-checkpoint", + ArchiveOperation.ReadMetadata => "read-metadata", + ArchiveOperation.ReadFull => "read-full", + ArchiveOperation.ReadRange => "read-range", + ArchiveOperation.CatchUpCheckpoint => "catch-up-checkpoint", + ArchiveOperation.CatchUpChunk => "catch-up-chunk", + ArchiveOperation.Service => "service", + _ => throw new ArgumentOutOfRangeException(nameof(operation), operation, null) + }); +} diff --git a/src/EventStore.Core/Services/Archive/ArchivePlugableComponent.cs b/src/EventStore.Core/Services/Archive/ArchivePlugableComponent.cs index 6444326167..f01867c8dd 100644 --- a/src/EventStore.Core/Services/Archive/ArchivePlugableComponent.cs +++ b/src/EventStore.Core/Services/Archive/ArchivePlugableComponent.cs @@ -5,6 +5,7 @@ using EventStore.Core.Services.Archive.Naming; using EventStore.Core.Services.Archive.Storage; using EventStore.Core.TransactionLog.FileNamingStrategy; +using EventStore.Core.Transforms; using EventStore.Plugins; using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.Configuration; @@ -69,6 +70,7 @@ private static IReadOnlyList AddArchiveCatchupTask( } var standardComponents = serviceProvider.GetRequiredService(); + var transformManager = serviceProvider.GetRequiredService(); newStartupTasks.Add(new ArchiveCatchup.ArchiveCatchup( dbPath: standardComponents.DbConfig.Path, writerCheckpoint: standardComponents.DbConfig.WriterCheckpoint, @@ -76,7 +78,9 @@ private static IReadOnlyList AddArchiveCatchupTask( epochCheckpoint: standardComponents.DbConfig.EpochCheckpoint, chunkSize: standardComponents.DbConfig.ChunkSize, serviceProvider.GetRequiredService(), - serviceProvider.GetRequiredService())); + serviceProvider.GetRequiredService(), + serviceProvider.GetRequiredService(), + transformManager.GetFactoryForExistingChunk)); return newStartupTasks; } diff --git a/src/EventStore.Core/Services/Archive/Archiver/ArchiverService.cs b/src/EventStore.Core/Services/Archive/Archiver/ArchiverService.cs index 691fd658c8..323ca68014 100644 --- a/src/EventStore.Core/Services/Archive/Archiver/ArchiverService.cs +++ b/src/EventStore.Core/Services/Archive/Archiver/ArchiverService.cs @@ -35,6 +35,7 @@ public class ArchiverService : private readonly Channel _archiveSignal; private readonly IChunkUnmerger _chunkUnmerger; private readonly IArchiveChunkNamer _chunkNamer; + private readonly IArchiveMetrics _metrics; private readonly TimeSpan RetryInterval = TimeSpan.FromMinutes(1); private long _replicationPosition; @@ -45,13 +46,15 @@ public ArchiverService( ISubscriber mainBus, IArchiveStorageFactory archiveStorageFactory, IChunkUnmerger chunkUnmerger, - IArchiveChunkNamer chunkNamer) + IArchiveChunkNamer chunkNamer, + IArchiveMetrics metrics = null) { _mainBus = mainBus; _archiveWriter = archiveStorageFactory.CreateWriter(); _archiveReader = archiveStorageFactory.CreateReader(); _chunkUnmerger = chunkUnmerger; _chunkNamer = chunkNamer; + _metrics = metrics ?? IArchiveMetrics.NoOp; _uncommittedChunks = new(); _chunksToArchive = new(); @@ -95,6 +98,7 @@ public void Handle(SystemMessage.ChunkCompleted message) if (chunkInfo.ChunkEndPosition > _replicationPosition) { _uncommittedChunks.Enqueue(chunkInfo); + _metrics.SetUncommittedChunks(_uncommittedChunks.Count); return; } @@ -114,6 +118,7 @@ public void Handle(SystemMessage.ChunkSwitched message) public void Handle(ReplicationTrackingMessage.ReplicatedTo message) { _replicationPosition = Math.Max(_replicationPosition, message.LogPosition); + _metrics.SetReplicationPosition(_replicationPosition); ProcessUncommittedChunks(); if (_archivingStarted) @@ -169,6 +174,7 @@ private void ProcessUncommittedChunks() } _uncommittedChunks.Dequeue(); + _metrics.SetUncommittedChunks(_uncommittedChunks.Count); ScheduleChunkForArchiving(chunkInfo, "new"); } } @@ -183,6 +189,7 @@ private void ScheduleChunkForArchiving(ChunkInfo chunkInfo, string chunkType) } _chunksToArchive[(chunkInfo.ChunkStartNumber, chunkInfo.ChunkEndNumber, chunkInfo.ChunkLocator)] = chunkInfo; + _metrics.SetQueuedChunks(_chunksToArchive.Count); } _archiveSignal.Writer.TryWrite(true); @@ -201,7 +208,15 @@ private async Task ArchiveChunks(CancellationToken ct) while (TryGetNextChunkToArchive(out var chunkInfo)) { - await ArchiveChunk(chunkInfo, ct); + _metrics.SetActiveChunks(1); + try + { + await ArchiveChunk(chunkInfo, ct); + } + finally + { + _metrics.SetActiveChunks(0); + } } } } @@ -216,6 +231,7 @@ private bool TryGetNextChunkToArchive(out ChunkInfo chunkInfo) if (candidate.ChunkEndPosition <= _checkpoint) { _chunksToArchive.Remove(key); + _metrics.SetQueuedChunks(_chunksToArchive.Count); continue; } @@ -225,6 +241,7 @@ private bool TryGetNextChunkToArchive(out ChunkInfo chunkInfo) } _chunksToArchive.Remove(key); + _metrics.SetQueuedChunks(_chunksToArchive.Count); chunkInfo = candidate; return true; } @@ -268,6 +285,8 @@ private async Task ArchiveChunk(ChunkInfo chunkInfo, CancellationToken ct) while (!await _archiveWriter.StoreChunk(chunkToStore, destinationFile, ct)) { + _metrics.RecordFailure(ArchiveOperation.StoreChunk); + _metrics.RecordRetry(ArchiveOperation.StoreChunk); Log.Warning("Archiving of {chunkFile}{chunkDetails} failed. Retrying in: {retryInterval}.", Path.GetFileName(chunkPath), chunksUnmerged ? $" (logical chunk no.: {logicalChunkNumber})" : string.Empty, @@ -287,6 +306,8 @@ private async Task ArchiveChunk(ChunkInfo chunkInfo, CancellationToken ct) { while (!await _archiveWriter.SetCheckpoint(chunkInfo.ChunkEndPosition, ct)) { + _metrics.RecordFailure(ArchiveOperation.SetCheckpoint); + _metrics.RecordRetry(ArchiveOperation.SetCheckpoint); Log.Warning( "Failed to set the archive checkpoint to: 0x{checkpoint:X}. Retrying in: {retryInterval}.", chunkInfo.ChunkEndPosition, RetryInterval); @@ -296,6 +317,7 @@ private async Task ArchiveChunk(ChunkInfo chunkInfo, CancellationToken ct) lock (_chunksToArchiveLock) { _checkpoint = chunkInfo.ChunkEndPosition; + _metrics.SetCheckpoint(_checkpoint); } Log.Debug("Archive checkpoint set to: 0x{checkpoint:X}", chunkInfo.ChunkEndPosition); @@ -314,6 +336,7 @@ private async Task ArchiveChunk(ChunkInfo chunkInfo, CancellationToken ct) } catch (Exception ex) { + _metrics.RecordFailure(ArchiveOperation.Service); Log.Error(ex, "Archiving of {chunkFile} failed.", chunkFile); throw; } @@ -326,6 +349,7 @@ private async Task LoadArchiveCheckpoint(CancellationToken ct) try { _checkpoint = await _archiveReader.GetCheckpoint(ct); + _metrics.SetCheckpoint(_checkpoint); Log.Debug("Archive checkpoint is: 0x{checkpoint:X}", _checkpoint); return; } @@ -335,6 +359,8 @@ private async Task LoadArchiveCheckpoint(CancellationToken ct) } catch (Exception ex) { + _metrics.RecordFailure(ArchiveOperation.LoadCheckpoint); + _metrics.RecordRetry(ArchiveOperation.LoadCheckpoint); Log.Warning(ex, "Failed to load the archive checkpoint. Retrying in: {retryInterval}.", RetryInterval); await Task.Delay(RetryInterval, ct); } @@ -356,6 +382,8 @@ private void ScheduleExistingChunksForArchiving() _chunksToArchive.Remove(key); } + + _metrics.SetQueuedChunks(_chunksToArchive.Count); } _archiveSignal.Writer.TryWrite(true); diff --git a/src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs b/src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs new file mode 100644 index 0000000000..43f33b4107 --- /dev/null +++ b/src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs @@ -0,0 +1,222 @@ +using System; +using System.Diagnostics; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace EventStore.Core.Services.Archive.Storage; + +internal sealed class ArchiveReadStream( + Stream inner, + IArchiveMetrics metrics, + ArchiveOperation operation, + long started) : Stream +{ + private int _recorded; + + public override bool CanRead => inner.CanRead; + public override bool CanSeek => inner.CanSeek; + public override bool CanWrite => inner.CanWrite; + public override long Length => inner.Length; + public override long Position + { + get => inner.Position; + set => inner.Position = value; + } + + public override void Flush() => inner.Flush(); + + public override int Read(byte[] buffer, int offset, int count) + { + try + { + var read = inner.Read(buffer, offset, count); + RecordEndOfStream(read); + return read; + } + catch (OperationCanceledException) + { + Ignore(); + throw; + } + catch + { + RecordFailure(); + throw; + } + } + + public override int Read(Span buffer) + { + try + { + var read = inner.Read(buffer); + RecordEndOfStream(read); + return read; + } + catch (OperationCanceledException) + { + Ignore(); + throw; + } + catch + { + RecordFailure(); + throw; + } + } + + public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + try + { + var read = await inner.ReadAsync(buffer, cancellationToken); + RecordEndOfStream(read); + return read; + } + catch (OperationCanceledException) + { + Ignore(); + throw; + } + catch + { + RecordFailure(); + throw; + } + } + + public override async Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken) + { + try + { + var read = await inner.ReadAsync(buffer, offset, count, cancellationToken); + RecordEndOfStream(read); + return read; + } + catch (OperationCanceledException) + { + Ignore(); + throw; + } + catch + { + RecordFailure(); + throw; + } + } + + public override int ReadByte() + { + try + { + var value = inner.ReadByte(); + if (value < 0) + { + RecordSuccess(); + } + + return value; + } + catch (OperationCanceledException) + { + Ignore(); + throw; + } + catch + { + RecordFailure(); + throw; + } + } + + public override long Seek(long offset, SeekOrigin origin) => inner.Seek(offset, origin); + public override void SetLength(long value) => inner.SetLength(value); + public override void Write(byte[] buffer, int offset, int count) => inner.Write(buffer, offset, count); + + protected override void Dispose(bool disposing) + { + if (!disposing) + { + base.Dispose(disposing); + return; + } + + try + { + inner.Dispose(); + RecordSuccess(); + } + catch (OperationCanceledException) + { + Ignore(); + throw; + } + catch + { + RecordFailure(); + throw; + } + finally + { + base.Dispose(disposing); + } + } + + public override async ValueTask DisposeAsync() + { + try + { + await inner.DisposeAsync(); + RecordSuccess(); + } + catch (OperationCanceledException) + { + Ignore(); + throw; + } + catch + { + RecordFailure(); + throw; + } + finally + { + GC.SuppressFinalize(this); + } + } + + private void RecordEndOfStream(int bytesRead) + { + if (bytesRead == 0) + { + RecordSuccess(); + } + } + + private void RecordSuccess() => Record(succeeded: true); + + private void RecordFailure() + => Record(succeeded: false); + + private void Ignore() => Interlocked.Exchange(ref _recorded, 1); + + private void Record(bool succeeded) + { + if (Interlocked.Exchange(ref _recorded, 1) != 0) + { + return; + } + + if (!succeeded) + { + metrics.RecordFailure(operation); + } + + metrics.RecordRead(operation, Stopwatch.GetElapsedTime(started), succeeded); + } +} diff --git a/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageFactory.cs b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageFactory.cs index 01a2ff8988..5456ae7b67 100644 --- a/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageFactory.cs +++ b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageFactory.cs @@ -5,18 +5,22 @@ namespace EventStore.Core.Services.Archive.Storage; public class ArchiveStorageFactory( ArchiveOptions options, - IArchiveChunkNamer chunkNamer) : IArchiveStorageFactory + IArchiveChunkNamer chunkNamer, + IArchiveMetrics archiveMetrics = null) : IArchiveStorageFactory { private const string ArchiveCheckpointFile = "archive.chk"; + private readonly IArchiveMetrics _archiveMetrics = archiveMetrics ?? IArchiveMetrics.NoOp; public IArchiveStorageReader CreateReader() { - return options.StorageType switch + var reader = options.StorageType switch { StorageType.Unspecified => throw new InvalidOperationException("Please specify an Archive StorageType"), StorageType.S3 => new S3Reader(options.S3, chunkNamer, ArchiveCheckpointFile), _ => throw new ArgumentOutOfRangeException(nameof(options.StorageType)) }; + + return new ArchiveStorageReaderMetrics(reader, _archiveMetrics); } public IArchiveStorageWriter CreateWriter() diff --git a/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs new file mode 100644 index 0000000000..ee7e1cd36e --- /dev/null +++ b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs @@ -0,0 +1,97 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Core.Services.Archive.Naming; + +namespace EventStore.Core.Services.Archive.Storage; + +public sealed class ArchiveStorageReaderMetrics( + IArchiveStorageReader inner, + IArchiveMetrics metrics) : IArchiveStorageReader +{ + public IArchiveChunkNamer ChunkNamer => inner.ChunkNamer; + + public ValueTask GetCheckpoint(CancellationToken ct) => inner.GetCheckpoint(ct); + + public async ValueTask GetChunkLength(string chunkFile, CancellationToken ct) => + await MeasureMetadata( + ArchiveOperation.ReadMetadata, + () => inner.GetChunkLength(chunkFile, ct)); + + public async ValueTask GetChunk(string chunkFile, long start, long end, CancellationToken ct) + { + var started = Stopwatch.GetTimestamp(); + try + { + var stream = await inner.GetChunk(chunkFile, start, end, ct); + return new ArchiveReadStream(stream, metrics, ArchiveOperation.ReadRange, started); + } + catch (OperationCanceledException) + { + throw; + } + catch + { + RecordFailedRead(ArchiveOperation.ReadRange, started); + throw; + } + } + + public async ValueTask GetChunk(string chunkFile, CancellationToken ct) + { + var started = Stopwatch.GetTimestamp(); + try + { + var stream = await inner.GetChunk(chunkFile, ct); + return new ArchiveReadStream(stream, metrics, ArchiveOperation.ReadFull, started); + } + catch (OperationCanceledException) + { + throw; + } + catch + { + RecordFailedRead(ArchiveOperation.ReadFull, started); + throw; + } + } + + public async IAsyncEnumerable ListChunks([EnumeratorCancellation] CancellationToken ct) + { + await foreach (var chunk in inner.ListChunks(ct).WithCancellation(ct)) + { + yield return chunk; + } + } + + private async ValueTask MeasureMetadata(ArchiveOperation operation, Func> action) + { + var started = Stopwatch.GetTimestamp(); + try + { + var result = await action(); + metrics.RecordRead(operation, Stopwatch.GetElapsedTime(started), succeeded: true); + return result; + } + catch (OperationCanceledException) + { + throw; + } + catch + { + metrics.RecordFailure(operation); + metrics.RecordRead(operation, Stopwatch.GetElapsedTime(started), succeeded: false); + throw; + } + } + + private void RecordFailedRead(ArchiveOperation operation, long started) + { + metrics.RecordFailure(operation); + metrics.RecordRead(operation, Stopwatch.GetElapsedTime(started), succeeded: false); + } +} diff --git a/src/EventStore.Core/Services/Archive/Storage/S3Storage.cs b/src/EventStore.Core/Services/Archive/Storage/S3Storage.cs index f941fdd13d..d253b85651 100644 --- a/src/EventStore.Core/Services/Archive/Storage/S3Storage.cs +++ b/src/EventStore.Core/Services/Archive/Storage/S3Storage.cs @@ -14,13 +14,13 @@ public static IAwsS3BlobStorage Create(S3Options options) { var sessionToken = string.IsNullOrWhiteSpace(options.SessionToken) ? null : options.SessionToken; - return (IAwsS3BlobStorage)StorageFactory.Blobs.AwsS3( + return (IAwsS3BlobStorage)StorageFactory.Blobs.MinIO( options.AccessKeyId, options.SecretAccessKey, - sessionToken, options.Bucket, options.Region, - options.ServiceUrl); + options.ServiceUrl, + sessionToken); } if (HasExplicitCredentials(options)) diff --git a/src/TrogonEventStore.SemanticConventions/Generated/MetricDefinitions.g.cs b/src/TrogonEventStore.SemanticConventions/Generated/MetricDefinitions.g.cs index 9f7c93097b..8d9fbcb11f 100644 --- a/src/TrogonEventStore.SemanticConventions/Generated/MetricDefinitions.g.cs +++ b/src/TrogonEventStore.SemanticConventions/Generated/MetricDefinitions.g.cs @@ -7,6 +7,31 @@ namespace TrogonEventStore.SemanticConventions { public static class MetricDefinitions { + public static MetricDefinition TrogonEventstoreArchiveCheckpointLag { get; } = new MetricDefinition( + "trogon.eventstore.archive.checkpoint.lag", + "By", + "Replicated transaction log bytes not yet covered by the archive checkpoint.", + MetricInstrumentKind.Gauge); + public static MetricDefinition TrogonEventstoreArchiveChunkPendingCount { get; } = new MetricDefinition( + "trogon.eventstore.archive.chunk.pending.count", + "{chunk}", + "Number of archive chunks waiting for or undergoing persistence.", + MetricInstrumentKind.UpDownCounter); + public static MetricDefinition TrogonEventstoreArchiveFailureCount { get; } = new MetricDefinition( + "trogon.eventstore.archive.failure.count", + "{failure}", + "Number of failed archive operations.", + MetricInstrumentKind.Counter); + public static MetricDefinition TrogonEventstoreArchiveReadDuration { get; } = new MetricDefinition( + "trogon.eventstore.archive.read.duration", + "s", + "Duration of remote archive read requests.", + MetricInstrumentKind.Histogram); + public static MetricDefinition TrogonEventstoreArchiveRetryCount { get; } = new MetricDefinition( + "trogon.eventstore.archive.retry.count", + "{retry}", + "Number of retries initiated by archive operations.", + MetricInstrumentKind.Counter); public static MetricDefinition TrogonEventstoreCacheOperationCount { get; } = new MetricDefinition( "trogon.eventstore.cache.operation.count", "{operation}", @@ -220,6 +245,11 @@ public static class MetricDefinitions public static IReadOnlyList All { get; } = Array.AsReadOnly(new[] { + TrogonEventstoreArchiveCheckpointLag, + TrogonEventstoreArchiveChunkPendingCount, + TrogonEventstoreArchiveFailureCount, + TrogonEventstoreArchiveReadDuration, + TrogonEventstoreArchiveRetryCount, TrogonEventstoreCacheOperationCount, TrogonEventstoreCacheResourceCount, TrogonEventstoreCacheResourceSize, From f2ba3347ba3bb624e00f3538f9e98bfb6b05dc8d Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 15 Aug 2026 06:18:45 -0400 Subject: [PATCH 2/4] fix(archive): preserve recovery guarantees under failure Signed-off-by: Yordis Prieto --- .github/workflows/common.yml | 49 +++++------- .../when_archiving_and_restoring_a_cluster.cs | 60 +++++++++++---- .../Integration/specification_with_cluster.cs | 5 +- .../ArchiveCatchup/ArchiveCatchupTests.cs | 74 +++++++++++++------ .../ArchiveCatchup/FakeArchiveStorage.cs | 2 +- .../Services/Archive/ArchiverServiceTests.cs | 44 +---------- .../Archive/RecordingArchiveMetrics.cs | 49 ++++++++++++ .../ArchiveStorageReaderMetricsTests.cs | 64 ++++++++++++---- .../Storage/ArchiveStorageTestsBase.cs | 2 +- .../Archive/Storage/S3RestartRecoveryTests.cs | 6 +- .../Services/Archive/Storage/S3Tests.cs | 5 +- .../Archive/ArchiveCatchup/ArchiveCatchup.cs | 31 ++++++-- .../Archive/Storage/ArchiveReadStream.cs | 12 +-- .../Storage/ArchiveStorageReaderMetrics.cs | 16 ++++ 14 files changed, 277 insertions(+), 142 deletions(-) create mode 100644 src/EventStore.Core.XUnit.Tests/Services/Archive/RecordingArchiveMetrics.cs diff --git a/.github/workflows/common.yml b/.github/workflows/common.yml index e6fa087284..9de2916ba0 100644 --- a/.github/workflows/common.yml +++ b/.github/workflows/common.yml @@ -124,6 +124,12 @@ jobs: archive-storage-contract: runs-on: ubuntu-latest name: Archive Storage Contract + env: + EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 + EVENTSTORE_S3_TEST_REGION: us-east-1 + EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract + EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key + EVENTSTORE_S3_RECOVERY_BUCKET: archive-recovery-${{ github.run_id }}-${{ github.run_attempt }} services: rustfs: image: rustfs/rustfs:1.0.0-beta.11@sha256:84ce557a0245a06a9aae5516f55ee0f007fca78d41df356f419306fdc0cb168c @@ -140,23 +146,23 @@ jobs: steps: - name: Checkout uses: actions/checkout@v7 + with: + persist-credentials: false - name: Install net10.0 uses: actions/setup-dotnet@v6 with: dotnet-version: 10.0.x - name: Set up .NET NuGet authentication + env: + NUGET_GITHUB_ACTOR: ${{ github.actor }} + NUGET_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} run: | dotnet nuget add source "https://nuget.pkg.github.com/TrogonStack/index.json" \ --name "github" \ - --username "${{ github.actor }}" \ - --password "${{ secrets.GITHUB_TOKEN }}" \ + --username "$NUGET_GITHUB_ACTOR" \ + --password "$NUGET_GITHUB_TOKEN" \ --store-password-in-clear-text - name: Run archive storage contract tests - env: - EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 - EVENTSTORE_S3_TEST_REGION: us-east-1 - EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract - EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key run: | dotnet test \ --configuration Release \ @@ -170,11 +176,6 @@ jobs: - name: Seed archive restart recovery data id: seed_archive_recovery env: - EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 - EVENTSTORE_S3_TEST_REGION: us-east-1 - EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract - EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key - EVENTSTORE_S3_RECOVERY_BUCKET: archive-recovery-${{ github.run_id }}-${{ github.run_attempt }} EVENTSTORE_S3_RECOVERY_PHASE: seed run: | dotnet test \ @@ -187,17 +188,14 @@ jobs: src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj - name: Stop RustFS without removing its data + env: + RUSTFS_CONTAINER_ID: ${{ job.services.rustfs.id }} run: | - timeout 20 docker stop --timeout 10 "${{ job.services.rustfs.id }}" - test "$(docker inspect --format '{{.State.Status}}' "${{ job.services.rustfs.id }}")" = "exited" + timeout 20 docker stop --timeout 10 "$RUSTFS_CONTAINER_ID" + test "$(docker inspect --format '{{.State.Status}}' "$RUSTFS_CONTAINER_ID")" = "exited" - name: Assert archive storage is unavailable env: - EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 - EVENTSTORE_S3_TEST_REGION: us-east-1 - EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract - EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key - EVENTSTORE_S3_RECOVERY_BUCKET: archive-recovery-${{ github.run_id }}-${{ github.run_attempt }} EVENTSTORE_S3_RECOVERY_PHASE: unavailable run: | timeout 20 dotnet test \ @@ -212,14 +210,10 @@ jobs: - name: Restart RustFS and verify archive recovery if: ${{ always() && steps.seed_archive_recovery.outcome == 'success' }} env: - EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 - EVENTSTORE_S3_TEST_REGION: us-east-1 - EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract - EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key - EVENTSTORE_S3_RECOVERY_BUCKET: archive-recovery-${{ github.run_id }}-${{ github.run_attempt }} EVENTSTORE_S3_RECOVERY_PHASE: verify-cleanup + RUSTFS_CONTAINER_ID: ${{ job.services.rustfs.id }} run: | - docker start "${{ job.services.rustfs.id }}" + docker start "$RUSTFS_CONTAINER_ID" timeout 30 bash -c -- 'until curl --output /dev/null --silent --fail http://localhost:9000/health; do sleep 1; done' dotnet test \ --configuration Release \ @@ -231,11 +225,6 @@ jobs: src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj - name: Run archive cluster restore gate - env: - EVENTSTORE_S3_TEST_ENDPOINT: http://localhost:9000 - EVENTSTORE_S3_TEST_REGION: us-east-1 - EVENTSTORE_S3_TEST_ACCESS_KEY: archive-contract - EVENTSTORE_S3_TEST_SECRET_KEY: archive-contract-secret-key run: | dotnet test \ --configuration Release \ diff --git a/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs index c49c6174ce..8464b5ff3c 100644 --- a/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs +++ b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs @@ -69,13 +69,19 @@ public when_archiving_and_restoring_a_cluster() }; } - protected override void BeforeNodesStart() + [OneTimeSetUp] + public override async Task TestFixtureSetUp() { if (!_archiveOptions.Enabled) { Assert.Ignore("The archive integration endpoint is not configured."); } + await base.TestFixtureSetUp(); + } + + protected override void BeforeNodesStart() + { _s3Client = new AmazonS3Client( _archiveOptions.S3.AccessKeyId, _archiveOptions.S3.SecretAccessKey, @@ -126,19 +132,11 @@ protected override async Task Given() { var payload = new byte[256 * 1024]; new Random(1729).NextBytes(payload); - var leader = GetLeader(); var archiver = _nodes[ArchiverNodeIndex]; - AssertEx.IsOrBecomesTrue( - () => _nodes.Any(node => - node.DebugIndex != ArchiverNodeIndex && node.NodeState == VNodeState.Follower), - GateTimeout, - $"A voting follower did not become ready. States={string.Join(", ", _nodes.Select(node => node.NodeState))}"); - _restoredNodeIndex = Array.FindIndex(_nodes, - node => node.NodeState == VNodeState.Follower && node.DebugIndex != ArchiverNodeIndex); - Assert.That(_restoredNodeIndex, Is.GreaterThanOrEqualTo(0)); for (var iteration = 0; iteration < SoakIterations; iteration++) { + var leader = await ReconnectToLeader(); for (var eventNumber = 0; eventNumber < EventsPerIteration; eventNumber++) { await _conn.AppendToStreamAsync( @@ -166,16 +164,42 @@ await _conn.AppendToStreamAsync( (await leader.Db.Manager.GetChunk(coldChunkNumber).TryReadFirst(CancellationToken.None)).Success, Is.True); + _restoredNodeIndex = GetVotingFollowerIndex(); await RestoreNode(_restoredNodeIndex, coldChunkNumber); _completedIterations++; } } + private async Task> ReconnectToLeader() + { + var leader = GetLeader(); + _conn?.Close(); + _conn = EventStoreConnection.Create( + ConnectionSettings.Create().DisableServerCertificateValidation(), + leader.ExternalTcpEndPoint); + await _conn.ConnectAsync(); + return leader; + } + + private int GetVotingFollowerIndex() + { + AssertEx.IsOrBecomesTrue( + () => _nodes.Any(node => + node.DebugIndex != ArchiverNodeIndex && node.NodeState == VNodeState.Follower), + GateTimeout, + $"A voting follower did not become ready. States={string.Join(", ", _nodes.Select(node => node.NodeState))}"); + + var followerIndex = Array.FindIndex(_nodes, + node => node.DebugIndex != ArchiverNodeIndex && node.NodeState == VNodeState.Follower); + Assert.That(followerIndex, Is.GreaterThanOrEqualTo(0)); + return followerIndex; + } + private static async Task StartScavenge(MiniClusterNode leader) { var scavengeStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); leader.Node.MainQueue.Publish(new ClientMessage.ScavengeDatabase( - new CallbackEnvelope(scavengeStarted.SetResult), + new CallbackEnvelope(message => scavengeStarted.TrySetResult(message)), Guid.NewGuid(), SystemAccounts.System, startFromChunk: 0, @@ -215,9 +239,17 @@ private EndPoint[] GossipSeedsFor(int nodeIndex) => private async Task WaitForArchiveCheckpoint(long minimum) { using var timeout = new CancellationTokenSource(GateTimeout); - while (await _archiveReader.GetCheckpoint(timeout.Token) < minimum) + var checkpoint = 0L; + try + { + while ((checkpoint = await _archiveReader.GetCheckpoint(timeout.Token)) < minimum) + { + await Task.Delay(100, timeout.Token); + } + } + catch (OperationCanceledException) when (timeout.IsCancellationRequested) { - await Task.Delay(100, timeout.Token); + Assert.Fail($"The archive checkpoint stalled at {checkpoint}; expected at least {minimum}."); } } @@ -231,7 +263,7 @@ public override async Task TestFixtureTearDown() } var objects = await _s3Client.ListObjectsV2Async(new ListObjectsV2Request { BucketName = _bucket }); - foreach (var item in objects.S3Objects) + foreach (var item in objects.S3Objects ?? []) { await _s3Client.DeleteObjectAsync(_bucket, item.Key); } diff --git a/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs b/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs index fb555b69dc..6b8aa33884 100644 --- a/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs +++ b/src/EventStore.Core.Tests/Integration/specification_with_cluster.cs @@ -188,7 +188,10 @@ public void AfterEachTest() public override async Task TestFixtureTearDown() { _conn?.Close(); - await Task.WhenAll(_nodes.Select(x => x.Shutdown())); + if (_nodes is not null) + { + await Task.WhenAll(_nodes.Where(node => node is not null).Select(node => node.Shutdown())); + } MiniNodeLogging.Clear(); diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs index a7c541e132..07532c308a 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/ArchiveCatchupTests.cs @@ -45,7 +45,8 @@ private Sut CreateSut( Action onGetCheckpoint = null, Func getChunk = null, TimeSpan? retryInterval = null, - int chunkSize = ChunkSize) + int chunkSize = ChunkSize, + Action deleteTempFile = null) { dbCheckpoint ??= 0L; archiveCheckpoint ??= 0L; @@ -78,7 +79,8 @@ private Sut CreateSut( fileNamingStrategy: new CustomNamingStrategy(), archiveStorageFactory: archive, metrics: metrics, - retryInterval: retryInterval ?? TimeSpan.Zero + retryInterval: retryInterval ?? TimeSpan.Zero, + deleteTempFile: deleteTempFile ); return new Sut @@ -123,6 +125,7 @@ public async Task catches_up_when_a_completed_chunk_is_smaller_than_the_prealloc chunkSize: preallocationSize); var receivedLength = sut.Archive.CreateChunkBytes(chunkFile).Length; Assert.True(receivedLength < preallocationSize); + Assert.Equal(receivedLength, await sut.Archive.GetChunkLength(chunkFile, CancellationToken.None)); await sut.Catchup.Run(); @@ -263,13 +266,15 @@ void OnGetChunk(string chunkFile) public async Task records_checkpoint_failure_before_retrying() { using var cts = new CancellationTokenSource(); - var sut = CreateSut(onGetCheckpoint: () => - { - cts.Cancel(); - throw new InvalidOperationException("checkpoint unavailable"); - }); + var sut = CreateSut( + retryInterval: TimeSpan.FromMinutes(1), + onGetCheckpoint: () => + { + cts.Cancel(); + throw new InvalidOperationException("checkpoint unavailable"); + }); - await Assert.ThrowsAsync(() => sut.Catchup.Run(cts.Token)); + await Assert.ThrowsAnyAsync(() => sut.Catchup.Run(cts.Token)); Assert.Equal([ArchiveOperation.CatchUpCheckpoint], sut.Metrics.Failures); Assert.Equal([ArchiveOperation.CatchUpCheckpoint], sut.Metrics.Retries); @@ -346,6 +351,44 @@ public async Task retries_corrupt_or_truncated_chunk_on_restart(bool truncated) Assert.Equal(ChunkSize, sut.ChaserCheckpoint.Read()); } + [Fact] + public async Task cleans_up_before_backoff_and_ignores_cleanup_failures() + { + FakeArchiveStorage archive = null; + var returnCorruptChunk = true; + var cleanupAttempted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var cancellation = new CancellationTokenSource(); + var sut = CreateSut( + archiveCheckpoint: ChunkSize, + retryInterval: TimeSpan.FromMinutes(1), + getChunk: chunkFile => + { + var bytes = archive.CreateChunkBytes(chunkFile); + if (returnCorruptChunk) + { + bytes[ChunkHeader.Size] ^= byte.MaxValue; + } + return new MemoryStream(bytes); + }, + deleteTempFile: _ => + { + cleanupAttempted.TrySetResult(); + throw new IOException("cleanup failed"); + }); + archive = sut.Archive; + + var firstRun = sut.Catchup.Run(cancellation.Token); + await cleanupAttempted.Task.WaitAsync(TimeSpan.FromSeconds(1)); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(() => firstRun); + Assert.Equal(0, sut.WriterCheckpoint.Read()); + + returnCorruptChunk = false; + await sut.Catchup.Run(); + + Assert.Equal(ChunkSize, sut.WriterCheckpoint.Read()); + } + [Fact] public async Task backs_off_when_a_listed_chunk_remains_missing() { @@ -419,18 +462,3 @@ private IEnumerable ListBackedUpChunks() .Order(); } } - -internal sealed class RecordingArchiveMetrics : IArchiveMetrics -{ - public List Failures { get; } = []; - public List Retries { get; } = []; - - public void SetReplicationPosition(long position) { } - public void SetCheckpoint(long position) { } - public void SetUncommittedChunks(int count) { } - public void SetQueuedChunks(int count) { } - public void SetActiveChunks(int count) { } - public void RecordRetry(ArchiveOperation operation) => Retries.Add(operation); - public void RecordFailure(ArchiveOperation operation) => Failures.Add(operation); - public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) { } -} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs index da8525f1a2..100acbfcc2 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiveCatchup/FakeArchiveStorage.cs @@ -79,7 +79,7 @@ public ValueTask GetCheckpoint(CancellationToken ct) public ValueTask GetChunkLength(string chunkFile, CancellationToken ct) { - return ValueTask.FromResult((long)(ChunkHeader.Size + _chunkSize)); + return ValueTask.FromResult((long)TFChunk.GetAlignedSize(ChunkHeader.Size + ChunkFooter.Size)); } private ChunkHeader CreateChunkHeader(int chunkStartNumber, int chunkEndNumber) diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs index f5d356e09d..08cb5ea8a0 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/ArchiverServiceTests.cs @@ -42,7 +42,7 @@ private static (ArchiverService, FakeArchiveStorage) CreateSut( [Fact] public async Task reports_archive_backlog_and_checkpoint_lag_inputs() { - var metrics = new ArchiverRecordingMetrics(); + var metrics = new RecordingArchiveMetrics(); var (sut, archive) = CreateSut(metrics: metrics); var chunkInfo = GetChunkInfo(0, 0); @@ -50,6 +50,9 @@ public async Task reports_archive_backlog_and_checkpoint_lag_inputs() sut.Handle(new ReplicationTrackingMessage.ReplicatedTo(chunkInfo.ChunkEndPosition)); await WaitFor(archive, numStores: 1, numCheckpoints: 1); + AssertEx.IsOrBecomesTrue( + () => metrics.Checkpoint == chunkInfo.ChunkEndPosition, + timeout: TimeSpan.FromSeconds(10)); Assert.Equal(chunkInfo.ChunkEndPosition, metrics.ReplicationPosition); Assert.Equal(chunkInfo.ChunkEndPosition, metrics.Checkpoint); @@ -514,42 +517,3 @@ public IAsyncEnumerable ListChunks(CancellationToken ct) return _existingChunks.ToAsyncEnumerable(); } } - -internal sealed class ArchiverRecordingMetrics : IArchiveMetrics -{ - private long _replicationPosition; - private long _checkpoint; - private int _maxUncommittedChunks; - private int _maxQueuedChunks; - private int _maxActiveChunks; - - public long ReplicationPosition => Interlocked.Read(ref _replicationPosition); - public long Checkpoint => Interlocked.Read(ref _checkpoint); - public int MaxUncommittedChunks => Volatile.Read(ref _maxUncommittedChunks); - public int MaxQueuedChunks => Volatile.Read(ref _maxQueuedChunks); - public int MaxActiveChunks => Volatile.Read(ref _maxActiveChunks); - - public void SetReplicationPosition(long position) => Interlocked.Exchange(ref _replicationPosition, position); - public void SetCheckpoint(long position) => Interlocked.Exchange(ref _checkpoint, position); - public void SetUncommittedChunks(int count) => UpdateMax(ref _maxUncommittedChunks, count); - public void SetQueuedChunks(int count) => UpdateMax(ref _maxQueuedChunks, count); - public void SetActiveChunks(int count) => UpdateMax(ref _maxActiveChunks, count); - public void RecordRetry(ArchiveOperation operation) { } - public void RecordFailure(ArchiveOperation operation) { } - public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) { } - - private static void UpdateMax(ref int target, int value) - { - var current = Volatile.Read(ref target); - while (value > current) - { - var observed = Interlocked.CompareExchange(ref target, value, current); - if (observed == current) - { - return; - } - - current = observed; - } - } -} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/RecordingArchiveMetrics.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/RecordingArchiveMetrics.cs new file mode 100644 index 0000000000..9afd2e5943 --- /dev/null +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/RecordingArchiveMetrics.cs @@ -0,0 +1,49 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using EventStore.Core.Services.Archive; + +namespace EventStore.Core.XUnit.Tests.Services.Archive; + +internal sealed class RecordingArchiveMetrics : IArchiveMetrics +{ + private long _replicationPosition; + private long _checkpoint; + private int _maxUncommittedChunks; + private int _maxQueuedChunks; + private int _maxActiveChunks; + + public List Failures { get; } = []; + public List Retries { get; } = []; + public List<(ArchiveOperation Operation, TimeSpan Duration, bool Succeeded)> Reads { get; } = []; + public long ReplicationPosition => Interlocked.Read(ref _replicationPosition); + public long Checkpoint => Interlocked.Read(ref _checkpoint); + public int MaxUncommittedChunks => Volatile.Read(ref _maxUncommittedChunks); + public int MaxQueuedChunks => Volatile.Read(ref _maxQueuedChunks); + public int MaxActiveChunks => Volatile.Read(ref _maxActiveChunks); + + public void SetReplicationPosition(long position) => Interlocked.Exchange(ref _replicationPosition, position); + public void SetCheckpoint(long position) => Interlocked.Exchange(ref _checkpoint, position); + public void SetUncommittedChunks(int count) => UpdateMax(ref _maxUncommittedChunks, count); + public void SetQueuedChunks(int count) => UpdateMax(ref _maxQueuedChunks, count); + public void SetActiveChunks(int count) => UpdateMax(ref _maxActiveChunks, count); + public void RecordRetry(ArchiveOperation operation) => Retries.Add(operation); + public void RecordFailure(ArchiveOperation operation) => Failures.Add(operation); + public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) => + Reads.Add((operation, duration, succeeded)); + + private static void UpdateMax(ref int target, int value) + { + var current = Volatile.Read(ref target); + while (value > current) + { + var observed = Interlocked.CompareExchange(ref target, value, current); + if (observed == current) + { + return; + } + + current = observed; + } + } +} diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs index eb05a4682e..a704fa05b0 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs @@ -7,6 +7,7 @@ using EventStore.Core.Services.Archive; using EventStore.Core.Services.Archive.Naming; using EventStore.Core.Services.Archive.Storage; +using EventStore.Core.Services.Archive.Storage.Exceptions; using Xunit; namespace EventStore.Core.XUnit.Tests.Services.Archive.Storage; @@ -46,6 +47,38 @@ await Assert.ThrowsAsync(async () => Assert.False(Assert.Single(metrics.Reads).Succeeded); } + [Fact] + public async Task an_empty_read_does_not_hide_a_later_stream_failure() + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(emptyThenFails: true), metrics); + await using var stream = await sut.GetChunk("chunk", CancellationToken.None); + + Assert.Equal(0, await stream.ReadAsync(Memory.Empty)); + Assert.Empty(metrics.Reads); + + await Assert.ThrowsAsync(async () => + await stream.ReadExactlyAsync(new byte[1], CancellationToken.None)); + Assert.Equal(ArchiveOperation.ReadFull, Assert.Single(metrics.Failures)); + Assert.False(Assert.Single(metrics.Reads).Succeeded); + } + + [Theory] + [InlineData(ArchiveOperation.ReadFull)] + [InlineData(ArchiveOperation.ReadRange)] + public async Task missing_chunks_record_an_unsuccessful_read_without_a_storage_failure(ArchiveOperation operation) + { + var metrics = new RecordingArchiveMetrics(); + var sut = new ArchiveStorageReaderMetrics(new StubReader(deleted: true), metrics); + + await Assert.ThrowsAsync(() => Invoke(sut, operation)); + + Assert.Empty(metrics.Failures); + var read = Assert.Single(metrics.Reads); + Assert.Equal(operation, read.Operation); + Assert.False(read.Succeeded); + } + [Theory] [InlineData(ArchiveOperation.ReadMetadata)] [InlineData(ArchiveOperation.ReadFull)] @@ -149,7 +182,9 @@ private sealed class StubReader( bool streamFails = false, bool cancel = false, bool streamCancels = false, - bool disposeCancels = false) : IArchiveStorageReader + bool disposeCancels = false, + bool emptyThenFails = false, + bool deleted = false) : IArchiveStorageReader { public IArchiveChunkNamer ChunkNamer { get; } = new StubChunkNamer(); @@ -166,6 +201,8 @@ public ValueTask GetChunk(string chunkFile, long start, long end, Cancel public ValueTask GetChunk(string chunkFile, CancellationToken ct) => cancel ? ValueTask.FromException(new OperationCanceledException()) + : deleted + ? ValueTask.FromException(new ChunkDeletedException()) : fail ? ValueTask.FromException(new InvalidOperationException()) : ValueTask.FromResult( @@ -173,7 +210,9 @@ public ValueTask GetChunk(string chunkFile, CancellationToken ct) => ? new FailingReadStream() : streamCancels ? new CanceledReadStream() - : disposeCancels ? new CanceledDisposeStream() : new MemoryStream([1])); + : disposeCancels + ? new CanceledDisposeStream() + : emptyThenFails ? new EmptyThenFailingReadStream() : new MemoryStream([1])); public async IAsyncEnumerable ListChunks([EnumeratorCancellation] CancellationToken ct) { @@ -209,19 +248,14 @@ private sealed class CanceledDisposeStream : MemoryStream protected override void Dispose(bool disposing) => throw new OperationCanceledException(); } - private sealed class RecordingArchiveMetrics : IArchiveMetrics + private sealed class EmptyThenFailingReadStream : MemoryStream { - public List Failures { get; } = []; - public List<(ArchiveOperation Operation, TimeSpan Duration, bool Succeeded)> Reads { get; } = []; - - public void SetReplicationPosition(long position) { } - public void SetCheckpoint(long position) { } - public void SetUncommittedChunks(int count) { } - public void SetQueuedChunks(int count) { } - public void SetActiveChunks(int count) { } - public void RecordRetry(ArchiveOperation operation) { } - public void RecordFailure(ArchiveOperation operation) => Failures.Add(operation); - public void RecordRead(ArchiveOperation operation, TimeSpan duration, bool succeeded) => - Reads.Add((operation, duration, succeeded)); + public override ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) => + buffer.IsEmpty + ? ValueTask.FromResult(0) + : ValueTask.FromException(new IOException("remote read failed")); } + } diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs index f700ed9b14..254ad8c44c 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs @@ -81,7 +81,7 @@ protected IArchiveStorageFactory CreateSutFactory( new() { StorageType = storageType, - S3 = CreateS3Options(), + S3 = storageType == StorageType.S3 ? CreateS3Options() : new(), }, chunkNamer, archiveMetrics); diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs index 80f77aaa26..a749f233de 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3RestartRecoveryTests.cs @@ -3,6 +3,7 @@ using System.IO; using System.Linq; using System.Net; +using System.Net.Http; using System.Threading; using System.Threading.Tasks; using Amazon.Runtime; @@ -71,10 +72,11 @@ private static async Task AssertUnavailable(S3Options options) using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var stopwatch = Stopwatch.StartNew(); - await Assert.ThrowsAnyAsync(async () => + await Assert.ThrowsAsync(async () => await reader.GetCheckpoint(timeout.Token)); - Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(10), + Assert.False(timeout.IsCancellationRequested); + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(5), $"Storage unavailability was not detected within the bounded interval: {stopwatch.Elapsed}"); } diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs index 317e856790..f7622f7536 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs @@ -65,11 +65,8 @@ await Assert.ThrowsAsync(async () => Assert.Contains(durations, measurement => HasTags(measurement, "read-range", "success")); Assert.Contains(durations, measurement => HasTags(measurement, "read-full", "error")); - var failure = Assert.Single(failureListener.RetrieveMeasurements( + Assert.Empty(failureListener.RetrieveMeasurements( MetricDefinitions.TrogonEventstoreArchiveFailureCount.Name)); - Assert.Equal(1, failure.Value); - Assert.Contains(failure.Tags, tag => - tag.Key == TrogonAttributeNames.ActivityName && (string)tag.Value == "read-full"); } private static bool HasTags( diff --git a/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs b/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs index 0fc3bfba7b..7ddcca7c64 100644 --- a/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs +++ b/src/EventStore.Core/Services/Archive/ArchiveCatchup/ArchiveCatchup.cs @@ -36,6 +36,7 @@ public class ArchiveCatchup : IClusterVNodeStartupTask private readonly IArchiveMetrics _metrics; private readonly Func _getTransformFactory; private readonly TimeSpan _retryInterval; + private readonly Action _deleteTempFile; private static readonly ILogger Log = Serilog.Log.ForContext(); private static readonly TimeSpan DefaultRetryInterval = TimeSpan.FromMinutes(1); @@ -50,7 +51,8 @@ public ArchiveCatchup( IArchiveStorageFactory archiveStorageFactory, IArchiveMetrics metrics = null, Func getTransformFactory = null, - TimeSpan? retryInterval = null) + TimeSpan? retryInterval = null, + Action deleteTempFile = null) { _dbPath = dbPath; _writerCheckpoint = writerCheckpoint; @@ -62,6 +64,7 @@ public ArchiveCatchup( _metrics = metrics ?? IArchiveMetrics.NoOp; _getTransformFactory = getTransformFactory ?? DbTransformManager.Default.GetFactoryForExistingChunk; _retryInterval = retryInterval ?? DefaultRetryInterval; + _deleteTempFile = deleteTempFile ?? File.Delete; } public async Task Run(CancellationToken ct = default) @@ -246,6 +249,7 @@ private async Task FetchChunk(string chunkFile, string destinationPath, Ca } catch (ChunkDeletedException) { + DeleteTempFile(ref tempPath); _metrics.RecordFailure(ArchiveOperation.CatchUpChunk); _metrics.RecordRetry(ArchiveOperation.CatchUpChunk); Log.Warning( @@ -260,6 +264,7 @@ private async Task FetchChunk(string chunkFile, string destinationPath, Ca } catch (Exception ex) { + DeleteTempFile(ref tempPath); _metrics.RecordFailure(ArchiveOperation.CatchUpChunk); _metrics.RecordRetry(ArchiveOperation.CatchUpChunk); Log.Error(ex, "Failed to fetch {chunk} from the archive. Retrying in {interval}", chunkFile, _retryInterval); @@ -268,10 +273,26 @@ private async Task FetchChunk(string chunkFile, string destinationPath, Ca } finally { - if (tempPath is not null) - { - File.Delete(tempPath); - } + DeleteTempFile(ref tempPath); + } + } + + private void DeleteTempFile(ref string tempPath) + { + var path = tempPath; + tempPath = null; + if (path is null) + { + return; + } + + try + { + _deleteTempFile(path); + } + catch (Exception ex) + { + Log.Warning(ex, "Failed to delete temporary archive file {tempFile}", Path.GetFileName(path)); } } diff --git a/src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs b/src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs index 43f33b4107..7eb93def5a 100644 --- a/src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs +++ b/src/EventStore.Core/Services/Archive/Storage/ArchiveReadStream.cs @@ -31,7 +31,7 @@ public override int Read(byte[] buffer, int offset, int count) try { var read = inner.Read(buffer, offset, count); - RecordEndOfStream(read); + RecordEndOfStream(read, count); return read; } catch (OperationCanceledException) @@ -51,7 +51,7 @@ public override int Read(Span buffer) try { var read = inner.Read(buffer); - RecordEndOfStream(read); + RecordEndOfStream(read, buffer.Length); return read; } catch (OperationCanceledException) @@ -71,7 +71,7 @@ public override async ValueTask ReadAsync(Memory buffer, Cancellation try { var read = await inner.ReadAsync(buffer, cancellationToken); - RecordEndOfStream(read); + RecordEndOfStream(read, buffer.Length); return read; } catch (OperationCanceledException) @@ -95,7 +95,7 @@ public override async Task ReadAsync( try { var read = await inner.ReadAsync(buffer, offset, count, cancellationToken); - RecordEndOfStream(read); + RecordEndOfStream(read, count); return read; } catch (OperationCanceledException) @@ -190,9 +190,9 @@ public override async ValueTask DisposeAsync() } } - private void RecordEndOfStream(int bytesRead) + private void RecordEndOfStream(int bytesRead, int requestedBytes) { - if (bytesRead == 0) + if (bytesRead == 0 && requestedBytes > 0) { RecordSuccess(); } diff --git a/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs index ee7e1cd36e..c1d17dc49a 100644 --- a/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs +++ b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs @@ -6,6 +6,7 @@ using System.Threading; using System.Threading.Tasks; using EventStore.Core.Services.Archive.Naming; +using EventStore.Core.Services.Archive.Storage.Exceptions; namespace EventStore.Core.Services.Archive.Storage; @@ -34,6 +35,11 @@ public async ValueTask GetChunk(string chunkFile, long start, long end, { throw; } + catch (ChunkDeletedException) + { + RecordUnsuccessfulRead(ArchiveOperation.ReadRange, started); + throw; + } catch { RecordFailedRead(ArchiveOperation.ReadRange, started); @@ -53,6 +59,11 @@ public async ValueTask GetChunk(string chunkFile, CancellationToken ct) { throw; } + catch (ChunkDeletedException) + { + RecordUnsuccessfulRead(ArchiveOperation.ReadFull, started); + throw; + } catch { RecordFailedRead(ArchiveOperation.ReadFull, started); @@ -92,6 +103,11 @@ private async ValueTask MeasureMetadata(ArchiveOperation operation, Func Date: Sat, 15 Aug 2026 06:44:00 -0400 Subject: [PATCH 3/4] fix(archive): keep missing chunks out of failure alerts Signed-off-by: Yordis Prieto --- .../Archive/Storage/ArchiveStorageReaderMetricsTests.cs | 3 +++ .../Services/Archive/Storage/ArchiveStorageReaderMetrics.cs | 5 +++++ 2 files changed, 8 insertions(+) diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs index a704fa05b0..387c6e9419 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageReaderMetricsTests.cs @@ -64,6 +64,7 @@ await Assert.ThrowsAsync(async () => } [Theory] + [InlineData(ArchiveOperation.ReadMetadata)] [InlineData(ArchiveOperation.ReadFull)] [InlineData(ArchiveOperation.ReadRange)] public async Task missing_chunks_record_an_unsuccessful_read_without_a_storage_failure(ArchiveOperation operation) @@ -193,6 +194,8 @@ private sealed class StubReader( public ValueTask GetChunkLength(string chunkFile, CancellationToken ct) => cancel ? ValueTask.FromException(new OperationCanceledException()) + : deleted + ? ValueTask.FromException(new ChunkDeletedException()) : fail ? ValueTask.FromException(new InvalidOperationException()) : ValueTask.FromResult(1L); public ValueTask GetChunk(string chunkFile, long start, long end, CancellationToken ct) => diff --git a/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs index c1d17dc49a..778c722c1f 100644 --- a/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs +++ b/src/EventStore.Core/Services/Archive/Storage/ArchiveStorageReaderMetrics.cs @@ -92,6 +92,11 @@ private async ValueTask MeasureMetadata(ArchiveOperation operation, Func Date: Sat, 15 Aug 2026 07:03:28 -0400 Subject: [PATCH 4/4] fix(archive): preserve setup failures during cleanup Signed-off-by: Yordis Prieto --- .github/workflows/common.yml | 2 +- .../when_archiving_and_restoring_a_cluster.cs | 21 ++++++++++--- .../Storage/ArchiveStorageTestsBase.cs | 6 ++-- .../Services/Archive/Storage/S3Tests.cs | 31 +++++++++++++++++++ 4 files changed, 52 insertions(+), 8 deletions(-) diff --git a/.github/workflows/common.yml b/.github/workflows/common.yml index 9de2916ba0..311d2c06fb 100644 --- a/.github/workflows/common.yml +++ b/.github/workflows/common.yml @@ -169,7 +169,7 @@ jobs: -p:Platform=x64 \ -p:ContinuousIntegrationBuild=true \ -p:RunS3Tests=true \ - --filter "FullyQualifiedName~S3ReaderTests|FullyQualifiedName~S3WriterTests|FullyQualifiedName~S3MetricsTests" \ + --filter "FullyQualifiedName~S3ReaderTests|FullyQualifiedName~S3WriterTests|FullyQualifiedName~S3MetricsTests|FullyQualifiedName~S3FixtureLifecycleTests" \ --logger:GitHubActions \ src/EventStore.Core.XUnit.Tests/EventStore.Core.XUnit.Tests.csproj diff --git a/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs index 8464b5ff3c..781da7f1ce 100644 --- a/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs +++ b/src/EventStore.Core.Tests/Integration/Archive/when_archiving_and_restoring_a_cluster.cs @@ -38,6 +38,7 @@ public class when_archiving_and_restoring_a_cluster private readonly string _bucket = $"archive-soak-{Guid.NewGuid():N}"; private readonly ArchiveOptions _archiveOptions; + private bool _bucketCreated; private AmazonS3Client _s3Client; private S3Reader _archiveReader; private long _archivedCheckpoint; @@ -92,6 +93,7 @@ protected override void BeforeNodesStart() ForcePathStyle = true, }); _s3Client.PutBucketAsync(new PutBucketRequest { BucketName = _bucket }).GetAwaiter().GetResult(); + _bucketCreated = true; var checkpointInitialized = new S3Writer(_archiveOptions.S3, ArchiveCheckpointFile) .SetCheckpoint(0L, CancellationToken.None) .AsTask() @@ -262,13 +264,22 @@ public override async Task TestFixtureTearDown() return; } - var objects = await _s3Client.ListObjectsV2Async(new ListObjectsV2Request { BucketName = _bucket }); - foreach (var item in objects.S3Objects ?? []) + try + { + if (_bucketCreated) + { + var objects = await _s3Client.ListObjectsV2Async(new ListObjectsV2Request { BucketName = _bucket }); + foreach (var item in objects.S3Objects ?? []) + { + await _s3Client.DeleteObjectAsync(_bucket, item.Key); + } + await _s3Client.DeleteBucketAsync(_bucket); + } + } + finally { - await _s3Client.DeleteObjectAsync(_bucket, item.Key); + _s3Client.Dispose(); } - await _s3Client.DeleteBucketAsync(_bucket); - _s3Client.Dispose(); } [Test] diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs index 254ad8c44c..48ce8c23f7 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/ArchiveStorageTestsBase.cs @@ -16,6 +16,7 @@ public abstract class ArchiveStorageTestsBase : DirectoryPerTest { protected const string ChunkPrefix = "chunk-"; private readonly string _bucket = $"archive-contract-{Guid.NewGuid():N}"; + private bool _bucketCreated; private AmazonS3Client _s3Client; protected string ArchivePath => Path.Combine(Fixture.Directory, "archive"); protected string DbPath => Path.Combine(Fixture.Directory, "db"); @@ -47,13 +48,14 @@ public override async Task InitializeAsync() }); await _s3Client.PutBucketAsync(new PutBucketRequest { BucketName = options.Bucket }); + _bucketCreated = true; } public override async Task DisposeAsync() { try { - if (_s3Client is not null) + if (_bucketCreated) { var objects = await _s3Client.ListObjectsV2Async(new ListObjectsV2Request { BucketName = _bucket }); foreach (var item in objects.S3Objects ?? []) @@ -88,7 +90,7 @@ protected IArchiveStorageFactory CreateSutFactory( return factory; } - private S3Options CreateS3Options() => new() + protected virtual S3Options CreateS3Options() => new() { Bucket = _bucket, Region = GetRequiredEnvironmentVariable("EVENTSTORE_S3_TEST_REGION"), diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs index f7622f7536..cdeb2fd64e 100644 --- a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/S3Tests.cs @@ -78,4 +78,35 @@ private static bool HasTags( measurement.Tags.Any(tag => tag.Key == TrogonAttributeNames.ActivityOutcome && (string)tag.Value == outcome); } + +public class S3FixtureLifecycleTests : ArchiveStorageTestsBase +{ + protected override StorageType StorageType => StorageType.S3; + + [Fact] + public async Task teardown_does_not_access_a_bucket_that_setup_failed_to_create() + { + var failedFixture = new FailedBucketFixture(); + await Assert.ThrowsAsync(failedFixture.InitializeAsync); + await failedFixture.DisposeAsync(); + } + + private sealed class FailedBucketFixture : ArchiveStorageTestsBase + { + protected override StorageType StorageType => StorageType.S3; + + protected override S3Options CreateS3Options() + { + var options = base.CreateS3Options(); + return new() + { + Bucket = options.Bucket, + Region = options.Region, + AccessKeyId = options.AccessKeyId, + SecretAccessKey = $"{options.SecretAccessKey}-invalid", + ServiceUrl = options.ServiceUrl, + }; + } + } +} #endif