feat(archive): make cold storage production ready

.github/workflows/common.yml

@@ -120,6 +120,120 @@ jobs:
         shell: bash
         run: |
           ./protolock.sh status --uptodate
+
+  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
+        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
+        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 "$NUGET_GITHUB_ACTOR" \
+            --password "$NUGET_GITHUB_TOKEN" \
+            --store-password-in-clear-text
+      - name: Run archive storage contract tests
+        run: |
+          dotnet test \
+            --configuration Release \
+            -p:Platform=x64 \
+            -p:ContinuousIntegrationBuild=true \
+            -p:RunS3Tests=true \
+            --filter "FullyQualifiedName~S3ReaderTests|FullyQualifiedName~S3WriterTests|FullyQualifiedName~S3MetricsTests|FullyQualifiedName~S3FixtureLifecycleTests" \
+            --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_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
+        env:
+          RUSTFS_CONTAINER_ID: ${{ job.services.rustfs.id }}
+        run: |
+          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_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_RECOVERY_PHASE: verify-cleanup
+          RUSTFS_CONTAINER_ID: ${{ job.services.rustfs.id }}
+        run: |
+          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 \
+            --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
+        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

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`.

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

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

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<TLogFormat, TStreamId>
 
 	public readonly ClusterVNode Node;
 	public TFChunkDb Db => Node.Db;
+	public string DbPath => _dbPath;
 	private readonly string _dbPath;
 	private readonly bool _isReadOnlyReplica;
 	private readonly TaskCompletionSource<bool> _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<string, string>[] {
+		var configuration = new List<KeyValuePair<string, string>> {
 				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());