From 26279e0cc6c08a974ab9d82a0f83643b016815ad Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:50:19 +0300 Subject: [PATCH 01/11] fix: check snapshot disk space before download The 2.5x guard ran at the extraction stage, after the tarball was already on disk, so passing it effectively required ~3.5x the snapshot in volume space. Probe the snapshot size upfront and demand the remaining download bytes plus a 1.5x extraction estimate before downloading; at extraction demand only the 1.5x estimate. The pre-check runs outside the retry loop so low disk aborts instead of retrying forever. --- .../InitDatabaseSnapshotTests.cs | 204 ++++++++++++++++++ .../Nethermind.Init.Snapshot.Test.csproj | 14 ++ .../InitDatabaseSnapshot.cs | 41 +++- .../Nethermind.Init.Snapshot.csproj | 4 + .../SnapshotDownloader.cs | 10 + src/Nethermind/Nethermind.slnx | 1 + 6 files changed, 268 insertions(+), 6 deletions(-) create mode 100644 src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs new file mode 100644 index 000000000000..2a353c6efae4 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs @@ -0,0 +1,204 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System; +using System.Formats.Tar; +using System.IO; +using System.IO.Abstractions; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using Nethermind.Api; +using Nethermind.Core.Test.IO; +using Nethermind.Logging; +using NSubstitute; +using NUnit.Framework; +using Testably.Abstractions; + +namespace Nethermind.Init.Snapshot.Test; + +public class InitDatabaseSnapshotTests +{ + private const int SnapshotPayloadSize = 100_000; + + private TempPath _tempDir = null!; + private string _dbPath = null!; + private string _snapshotPath = null!; + private SnapshotConfig _snapshotConfig = null!; + private INethermindApi _api = null!; + + [SetUp] + public void SetUp() + { + _tempDir = TempPath.GetTempDirectory(); + string snapshotDirectory = Path.Combine(_tempDir.Path, "snapshot"); + Directory.CreateDirectory(snapshotDirectory); + _dbPath = Path.Combine(_tempDir.Path, "db"); + _snapshotConfig = new SnapshotConfig + { + Enabled = true, + DownloadUrl = "http://127.0.0.1:1/snapshot.tar", + SnapshotDirectory = snapshotDirectory, + SnapshotFileName = "snapshot.tar", + StripComponents = 1 + }; + _snapshotPath = Path.Combine(snapshotDirectory, _snapshotConfig.SnapshotFileName); + + _api = Substitute.For(); + _api.Config().Returns(_snapshotConfig); + _api.Config().Returns(new InitConfig { BaseDbPath = _dbPath }); + _api.LogManager.Returns(LimboLogs.Instance); + _api.FileSystem.Returns(new RealFileSystem()); + } + + [TearDown] + public void TearDown() => _tempDir.Dispose(); + + [Test] + public async Task Execute_FreeSpaceCoversExtractionButNotLegacyMultiplier_ExtractsSnapshot() + { + long snapshotSize = WriteSnapshotTar(); + AdvanceCheckpoint(SnapshotStage.Verified); + long freeSpace = snapshotSize * 2; + Assert.That(freeSpace, Is.GreaterThanOrEqualTo(InitDatabaseSnapshot.GetRequiredSpaceForExtraction(snapshotSize)), + "precondition: free space must cover the extraction estimate"); + Assert.That(freeSpace, Is.LessThan((long)(snapshotSize * 2.5)), + "precondition: free space must be below the legacy 2.5x requirement to prove the regression is fixed"); + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(freeSpace)); + + await step.Execute(CancellationToken.None); + + Assert.That(File.Exists(Path.Combine(_dbPath, "state.bin")), Is.True, + "the snapshot content should be extracted into the database directory"); + Assert.That(File.Exists(_snapshotPath), Is.False, + "the snapshot archive should be deleted after a successful extraction"); + } + + [Test] + public void Execute_FreeSpaceBelowExtractionEstimate_ThrowsIOException() + { + long snapshotSize = WriteSnapshotTar(); + AdvanceCheckpoint(SnapshotStage.Verified); + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(snapshotSize)); + + IOException exception = Assert.ThrowsAsync(() => step.Execute(CancellationToken.None))!; + + Assert.That(exception.Message, Does.Contain("Insufficient disk space"), + "the extraction should be rejected when free space is below the extraction estimate"); + Assert.That(Directory.Exists(_dbPath), Is.False, + "nothing should be extracted when free space is insufficient"); + } + + [Test] + public void Execute_FreeSpaceBelowDownloadRequirement_ThrowsIOExceptionBeforeDownloading() + { + using SnapshotServer server = SnapshotServer.Start(contentLength: 1_000_000); + _snapshotConfig.DownloadUrl = server.Url; + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(1_000_000)); + + IOException exception = Assert.ThrowsAsync(() => step.Execute(CancellationToken.None))!; + + Assert.That(exception.Message, Does.Contain("Insufficient disk space"), + "the download should be rejected when free space cannot fit the snapshot and its extraction"); + Assert.That(File.Exists(_snapshotPath), Is.False, + "no bytes should be downloaded when free space is insufficient"); + } + + [TestCase(1_000, 0, 2_500, TestName = "FreshDownload")] + [TestCase(1_000, 400, 2_100, TestName = "ResumedDownload")] + public void GetRequiredSpaceForDownload_ForGivenSizes_AddsRemainingBytesToExtractionEstimate( + long totalSize, long existingSize, long expected) => + Assert.That(InitDatabaseSnapshot.GetRequiredSpaceForDownload(totalSize, existingSize), Is.EqualTo(expected), + "the pre-download requirement should be the remaining bytes plus the extraction estimate of the full snapshot"); + + private long WriteSnapshotTar() + { + using (FileStream fileStream = File.Create(_snapshotPath)) + using (TarWriter tarWriter = new(fileStream)) + { + tarWriter.WriteEntry(new PaxTarEntry(TarEntryType.Directory, "data")); + PaxTarEntry fileEntry = new(TarEntryType.RegularFile, "data/state.bin") + { + DataStream = new MemoryStream(new byte[SnapshotPayloadSize]) + }; + tarWriter.WriteEntry(fileEntry); + } + + return new FileInfo(_snapshotPath).Length; + } + + private void AdvanceCheckpoint(SnapshotStage stage) => + new SnapshotCheckpoint(_snapshotConfig, LimboLogs.Instance).Advance(stage); + + private static IDriveInfo[] DrivesWithFreeSpace(long freeSpace) + { + IDriveInfo drive = Substitute.For(); + drive.AvailableFreeSpace.Returns(freeSpace); + drive.RootDirectory.FullName.Returns("/"); + return [drive]; + } + + private sealed class SnapshotServer : IDisposable + { + private readonly HttpListener _listener; + + public string Url { get; } + + private SnapshotServer(HttpListener listener, string url) + { + _listener = listener; + Url = url; + } + + public static SnapshotServer Start(long contentLength) + { + (HttpListener listener, int port) = StartListener(); + + _ = Task.Run(async () => + { + while (listener.IsListening) + { + try + { + HttpListenerContext context = await listener.GetContextAsync(); + context.Response.ContentLength64 = contentLength; + byte[] chunk = new byte[1024]; + await context.Response.OutputStream.WriteAsync(chunk); + context.Response.OutputStream.Flush(); + } + catch (Exception) + { + return; + } + } + }); + + return new SnapshotServer(listener, $"http://127.0.0.1:{port}/snapshot.tar"); + } + + private static (HttpListener Listener, int Port) StartListener() + { + for (int attempt = 0; ; attempt++) + { + HttpListener listener = new(); + int port = Random.Shared.Next(20000, 60000); + listener.Prefixes.Add($"http://127.0.0.1:{port}/"); + try + { + listener.Start(); + return (listener, port); + } + catch (HttpListenerException) when (attempt < 5) + { + listener.Close(); + } + } + } + + public void Dispose() + { + _listener.Stop(); + _listener.Close(); + } + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj b/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj new file mode 100644 index 000000000000..41845560d254 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj @@ -0,0 +1,14 @@ + + + + + + + + + + + + + + diff --git a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs index 316862eb0699..72bb45a33320 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs @@ -26,6 +26,7 @@ public class InitDatabaseSnapshot( INethermindApi api, [KeyFilter(nameof(IInitConfig.BaseDbPath))] IDriveInfo[] drives) : IStep { + private const double ExtractionSpaceMultiplier = 1.5; private const int ExtractionRestartDelaySeconds = 5; private const int InitialRetryDelaySeconds = 5; private const int MaxRetryDelaySeconds = 300; @@ -109,6 +110,8 @@ private async Task DownloadWithRetryAsync( if (checkpoint.Read() >= SnapshotStage.Downloaded) return; + await CheckDiskSpaceBeforeDownloadAsync(downloader, url, destinationPath, cancellationToken).ConfigureAwait(false); + TimeSpan retryDelay = TimeSpan.FromSeconds(InitialRetryDelaySeconds); long lastSize = GetFileSize(destinationPath); @@ -191,26 +194,52 @@ private async Task ExtractAsync( if (checkpoint.Read() >= SnapshotStage.Extracted) return; - CheckDiskSpace(snapshotPath); + CheckDiskSpace(GetRequiredSpaceForExtraction(GetFileSize(snapshotPath)), "extract"); SnapshotExtractor extractor = new(api.LogManager); await extractor.ExtractAsync(snapshotPath, dbPath, stripComponents, cancellationToken).ConfigureAwait(false); checkpoint.Advance(SnapshotStage.Extracted); } - private void CheckDiskSpace(string snapshotPath) + private async Task CheckDiskSpaceBeforeDownloadAsync( + SnapshotDownloader downloader, string url, string destinationPath, CancellationToken cancellationToken) { - if (drives.Length == 0) + long existingSize = GetFileSize(destinationPath); + long? totalSize; + try + { + totalSize = await downloader.GetTotalSizeAsync(url, existingSize, cancellationToken).ConfigureAwait(false); + } + catch (Exception e) when (e is IOException or HttpRequestException) + { + if (_logger.IsWarn) + _logger.Warn($"Could not determine the snapshot size upfront. Skipping the pre-download disk space check. Error: {e.Message}"); + return; + } + + if (totalSize is null) + { + if (_logger.IsWarn) + _logger.Warn("The server did not report the snapshot size. Skipping the pre-download disk space check."); return; + } - long snapshotSize = api.FileSystem.FileInfo.New(snapshotPath).Length; - long required = (long)(snapshotSize * 2.5); + CheckDiskSpace(GetRequiredSpaceForDownload(totalSize.Value, existingSize), "download and extract"); + } + + internal static long GetRequiredSpaceForDownload(long totalSize, long existingSize) => + totalSize - existingSize + GetRequiredSpaceForExtraction(totalSize); + internal static long GetRequiredSpaceForExtraction(long snapshotSize) => + (long)(snapshotSize * ExtractionSpaceMultiplier); + + private void CheckDiskSpace(long required, string operation) + { foreach (IDriveInfo drive in drives) { if (drive.AvailableFreeSpace < required) throw new IOException( - $"Insufficient disk space on '{drive.RootDirectory.FullName}' to extract snapshot: " + + $"Insufficient disk space on '{drive.RootDirectory.FullName}' to {operation} the snapshot: " + $"need at least {required} bytes, {drive.AvailableFreeSpace} available."); } } diff --git a/src/Nethermind/Nethermind.Init.Snapshot/Nethermind.Init.Snapshot.csproj b/src/Nethermind/Nethermind.Init.Snapshot/Nethermind.Init.Snapshot.csproj index e9f2b426edf8..7e06814b5747 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/Nethermind.Init.Snapshot.csproj +++ b/src/Nethermind/Nethermind.Init.Snapshot/Nethermind.Init.Snapshot.csproj @@ -12,6 +12,10 @@ + + + <_Parameter1>Nethermind.Init.Snapshot.Test + diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs index 535c4f728b6d..36b100e9801b 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs @@ -83,6 +83,16 @@ public async Task DownloadAsync(string url, string destinationPath, Cancellation _logger.Info($"Snapshot downloaded to {destinationPath}."); } + public async Task GetTotalSizeAsync(string url, long existingSize, CancellationToken cancellationToken) + { + using HttpResponseMessage response = await SendWithRangeAsync(_httpClient, url, existingSize, cancellationToken).ConfigureAwait(false); + + if (response.StatusCode == HttpStatusCode.RequestedRangeNotSatisfiable) + return existingSize; + + return ResolveCopyStrategy(response.StatusCode, existingSize, response.Content.Headers.ContentLength).totalSize; + } + public void Dispose() => _httpClient.Dispose(); private static (FileMode fileMode, long bytesToSkip, long? totalSize) ResolveCopyStrategy( diff --git a/src/Nethermind/Nethermind.slnx b/src/Nethermind/Nethermind.slnx index c7884b9e584d..fe8f6044f5cb 100644 --- a/src/Nethermind/Nethermind.slnx +++ b/src/Nethermind/Nethermind.slnx @@ -76,6 +76,7 @@ + From 5da50e5434ca44e22ff0364f2b01cabf8bb5d84c Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 22:38:22 +0300 Subject: [PATCH 02/11] feat: stream snapshot download without storing the archive --- .../FlakySnapshotServer.cs | 179 +++++++ .../InitDatabaseSnapshotTests.cs | 31 ++ .../Nethermind.Init.Snapshot.Test.csproj | 5 + .../SnapshotHttpStreamTests.cs | 131 +++++ .../StreamingSnapshotInitializerTests.cs | 174 +++++++ .../TestArchive.cs | 52 ++ .../ISnapshotConfig.cs | 6 + .../InitDatabaseSnapshot.cs | 68 ++- .../SnapshotConfig.cs | 4 + .../SnapshotDownloader.cs | 91 +--- .../SnapshotExtractor.cs | 37 +- .../SnapshotHttpClient.cs | 99 ++++ .../SnapshotHttpStream.cs | 454 ++++++++++++++++++ .../SnapshotProgress.cs | 23 + .../SnapshotSourceChangedException.cs | 6 + .../StreamingSnapshotInitializer.cs | 133 +++++ 16 files changed, 1381 insertions(+), 112 deletions(-) create mode 100644 src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot/SnapshotProgress.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot/SnapshotSourceChangedException.cs create mode 100644 src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs new file mode 100644 index 000000000000..44ceb5f35acf --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs @@ -0,0 +1,179 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.Collections.Concurrent; +using System.Net; + +namespace Nethermind.Init.Snapshot.Test; + +internal sealed class FlakySnapshotServer : IDisposable +{ + private readonly HttpListener _listener; + private readonly ConcurrentDictionary _attemptsPerRange = new(); + private int _requestCount; + private int _switchAfterRequests = int.MaxValue; + private byte[] _newContent = []; + private string? _newETag; + + public FlakySnapshotServer() + { + (_listener, int port) = StartListener(); + Url = $"http://127.0.0.1:{port}/snapshot.tar.zst"; + Task.Run(AcceptLoopAsync); + } + + public string Url { get; } + + public byte[] Content { get; set; } = []; + + public string? ETag { get; set; } = "\"v1\""; + + public bool SupportsRanges { get; set; } = true; + + public int? DropFirstAttemptPerRangeAfterBytes { get; set; } + + public int? FailWithNotFoundAfterRequests { get; set; } + + public int RequestCount => _requestCount; + + public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, string? newETag) + { + _newContent = newContent; + _newETag = newETag; + _switchAfterRequests = requestCount; + } + + private static (HttpListener Listener, int Port) StartListener() + { + for (int attempt = 0; ; attempt++) + { + HttpListener listener = new(); + int port = Random.Shared.Next(20000, 60000); + listener.Prefixes.Add($"http://127.0.0.1:{port}/"); + try + { + listener.Start(); + return (listener, port); + } + catch (HttpListenerException) when (attempt < 5) + { + listener.Close(); + } + } + } + + public void Dispose() + { + _listener.Stop(); + _listener.Close(); + } + + private async Task AcceptLoopAsync() + { + while (_listener.IsListening) + { + HttpListenerContext context; + try + { + context = await _listener.GetContextAsync(); + } + catch + { + return; + } + + _ = Task.Run(() => HandleAsync(context)); + } + } + + private async Task HandleAsync(HttpListenerContext context) + { + int requestNumber = Interlocked.Increment(ref _requestCount); + byte[] content = requestNumber > _switchAfterRequests ? _newContent : Content; + string? etag = requestNumber > _switchAfterRequests ? _newETag : ETag; + HttpListenerResponse response = context.Response; + + try + { + if (FailWithNotFoundAfterRequests is int failAfter && requestNumber > failAfter) + { + response.StatusCode = 404; + response.Close(); + return; + } + + if (etag is not null) + response.Headers["ETag"] = etag; + string? rangeHeader = context.Request.Headers["Range"]; + string? ifRange = context.Request.Headers["If-Range"]; + long from = 0; + long to = content.Length - 1; + bool ranged = SupportsRanges + && rangeHeader is not null + && (ifRange is null || ifRange == etag) + && TryParseRange(rangeHeader, content.Length, ref from, ref to); + + if (ranged && from >= content.Length) + { + response.StatusCode = 416; + response.Headers["Content-Range"] = $"bytes */{content.Length}"; + response.Close(); + return; + } + + if (ranged) + { + response.StatusCode = 206; + response.Headers["Content-Range"] = $"bytes {from}-{to}/{content.Length}"; + } + else + { + response.StatusCode = 200; + from = 0; + to = content.Length - 1; + } + + long length = to - from + 1; + response.ContentLength64 = length; + + string rangeKey = rangeHeader ?? "full"; + int attempt = _attemptsPerRange.AddOrUpdate(rangeKey, 1, static (_, previous) => previous + 1); + if (DropFirstAttemptPerRangeAfterBytes is int dropAfter && attempt == 1 && length > dropAfter) + { + await response.OutputStream.WriteAsync(content.AsMemory((int)from, dropAfter)); + response.Abort(); + return; + } + + await response.OutputStream.WriteAsync(content.AsMemory((int)from, (int)length)); + response.Close(); + } + catch + { + try + { + response.Abort(); + } + catch + { + } + } + } + + private static bool TryParseRange(string rangeHeader, long contentLength, ref long from, ref long to) + { + if (!rangeHeader.StartsWith("bytes=", StringComparison.Ordinal)) + return false; + + string[] parts = rangeHeader["bytes=".Length..].Split('-'); + if (parts.Length != 2 || !long.TryParse(parts[0], out long start)) + return false; + + from = start; + to = parts[1].Length > 0 && long.TryParse(parts[1], out long end) + ? Math.Min(end, contentLength - 1) + : contentLength - 1; + return true; + } + +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs index 2a353c6efae4..dc28f2e6a51a 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs @@ -4,6 +4,7 @@ using System; using System.Formats.Tar; using System.IO; +using System.Security.Cryptography; using System.IO.Abstractions; using System.Net; using System.Threading; @@ -104,6 +105,36 @@ public void Execute_FreeSpaceBelowDownloadRequirement_ThrowsIOExceptionBeforeDow "no bytes should be downloaded when free space is insufficient"); } + [Test] + public async Task Execute_StreamingEnabled_ExtractsSnapshotFromStream() + { + using FlakySnapshotServer server = new(); + Dictionary files = TestArchive.BuildFiles(); + byte[] archive = TestArchive.BuildTarZst(files); + server.Content = archive; + _snapshotConfig.SnapshotFileName = "snapshot.tar.zst"; + _snapshotConfig.DownloadUrl = server.Url; + _snapshotConfig.Streaming = true; + _snapshotConfig.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + await step.Execute(CancellationToken.None); + + Assert.That(File.Exists(Path.Combine(_dbPath, "state/a42.sst")), Is.True, + "the streaming path must extract the snapshot into the database directory"); + } + + [Test] + public void Execute_StreamingConnectionsNotPositive_Throws() + { + _snapshotConfig.Streaming = true; + _snapshotConfig.StreamingConnections = 0; + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + Assert.ThrowsAsync(() => step.Execute(CancellationToken.None), + "a non-positive connection count must be rejected before any network activity"); + } + [TestCase(1_000, 0, 2_500, TestName = "FreshDownload")] [TestCase(1_000, 400, 2_100, TestName = "ResumedDownload")] public void GetRequiredSpaceForDownload_ForGivenSizes_AddsRemainingBytesToExtractionEstimate( diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj b/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj index 41845560d254..f74e0e631da5 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj @@ -2,6 +2,11 @@ + + enable + enable + + diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs new file mode 100644 index 000000000000..4a6f14cfd5e9 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs @@ -0,0 +1,131 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.Security.Cryptography; +using Nethermind.Logging; +using NUnit.Framework; + +namespace Nethermind.Init.Snapshot.Test; + +[TestFixture] +public class SnapshotHttpStreamTests +{ + private const int TestChunkSize = 16384; + + private FlakySnapshotServer _server = null!; + + [SetUp] + public void SetUp() => _server = new FlakySnapshotServer(); + + [TearDown] + public void TearDown() => _server.Dispose(); + + [TestCase(1000, TestName = "SmallerThanOneChunk")] + [TestCase(2 * TestChunkSize, TestName = "ExactChunkMultiple")] + [TestCase(100_000, TestName = "PartialLastChunk")] + public async Task Read_ChunkedDownload_DeliversExactContentAndHash(int contentLength) + { + byte[] content = BuildContent(contentLength); + _server.Content = content; + + (byte[] delivered, byte[] hash) = await DownloadAsync(connections: 3); + + Assert.That(delivered, Is.EqualTo(content), "a chunked download must deliver the exact source bytes in order"); + Assert.That(hash, Is.EqualTo(SHA256.HashData(content)), "the incremental hash must cover every byte exactly once"); + } + + [Test] + public async Task Read_EveryRangeDroppedOnFirstAttempt_DeliversExactContent() + { + byte[] content = BuildContent(100_000); + _server.Content = content; + _server.DropFirstAttemptPerRangeAfterBytes = 5000; + + (byte[] delivered, byte[] hash) = await DownloadAsync(connections: 3); + + Assert.That(delivered, Is.EqualTo(content), "every chunk must survive a dropped connection and be re-fetched"); + Assert.That(hash, Is.EqualTo(SHA256.HashData(content)), "retried chunks must not be hashed twice"); + } + + [Test] + public async Task Read_ServerWithoutRangeSupport_DeliversExactContent() + { + byte[] content = BuildContent(100_000); + _server.Content = content; + _server.SupportsRanges = false; + + (byte[] delivered, byte[] hash) = await DownloadAsync(connections: 3); + + Assert.That(delivered, Is.EqualTo(content), "the sequential fallback must deliver the exact source bytes"); + Assert.That(hash, Is.EqualTo(SHA256.HashData(content)), "the sequential fallback must hash every byte exactly once"); + } + + [Test] + public async Task Read_ServerWithoutRangeSupportDropsConnections_ResumesBySkipping() + { + byte[] content = BuildContent(100_000); + _server.Content = content; + _server.SupportsRanges = false; + _server.DropFirstAttemptPerRangeAfterBytes = 30_000; + + (byte[] delivered, byte[] hash) = await DownloadAsync(connections: 3); + + Assert.That(delivered, Is.EqualTo(content), "a resumed connection on a rangeless server must skip already delivered bytes, not replay them"); + Assert.That(hash, Is.EqualTo(SHA256.HashData(content)), "skipped bytes must not be hashed twice"); + } + + [Test] + public void Read_SourceChangesMidDownload_ThrowsSnapshotSourceChanged() + { + _server.Content = BuildContent(100_000); + _server.SwitchSourceAfterRequests(2, BuildContent(500), "\"v2\""); + + Assert.ThrowsAsync( + () => DownloadAsync(connections: 1), + "a snapshot replaced on the server mid-download must abort the stream instead of mixing two objects"); + } + + [Test] + public void Read_SourceLengthChangesOnRangelessServerWithoutETag_ThrowsSnapshotSourceChanged() + { + _server.Content = BuildContent(100_000); + _server.SupportsRanges = false; + _server.ETag = null; + _server.DropFirstAttemptPerRangeAfterBytes = 30_000; + _server.SwitchSourceAfterRequests(2, BuildContent(50_000), null); + + Assert.ThrowsAsync( + () => DownloadAsync(connections: 1), + "a rotated source without an ETag must be detected by its length instead of splicing two objects together"); + } + + [Test] + public void Read_ServerReturnsNotFoundMidDownload_ThrowsWithoutRetrying() + { + _server.Content = BuildContent(100_000); + _server.FailWithNotFoundAfterRequests = 1; + + Assert.ThrowsAsync( + () => DownloadAsync(connections: 2), + "a permanent HTTP error must abort the stream instead of retrying forever"); + } + + private async Task<(byte[] Delivered, byte[] Hash)> DownloadAsync(int connections) + { + SnapshotStreamSettings settings = new(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50)); + using SnapshotHttpClient client = new(); + SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(_server.Url, CancellationToken.None); + await using SnapshotHttpStream stream = new(client, _server.Url, remoteInfo, settings, LimboLogs.Instance, CancellationToken.None); + using MemoryStream delivered = new(); + await Task.Run(() => stream.CopyTo(delivered)); + byte[] hash = await stream.FinishAsync(CancellationToken.None); + return (delivered.ToArray(), hash); + } + + private static byte[] BuildContent(int length) + { + byte[] content = new byte[length]; + new Random(42).NextBytes(content); + return content; + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs new file mode 100644 index 000000000000..ef71956e177b --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs @@ -0,0 +1,174 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.IO.Abstractions; +using System.Security.Cryptography; +using Nethermind.Core.Test.IO; +using Nethermind.Logging; +using NSubstitute; +using NUnit.Framework; + +namespace Nethermind.Init.Snapshot.Test; + +[TestFixture] +public class StreamingSnapshotInitializerTests +{ + private const int TestChunkSize = 16384; + + private FlakySnapshotServer _server = null!; + private TempPath _tempDir = null!; + private string _dbPath = null!; + private SnapshotConfig _config = null!; + + [SetUp] + public void SetUp() + { + _server = new FlakySnapshotServer(); + _tempDir = TempPath.GetTempDirectory(); + _dbPath = Path.Combine(_tempDir.Path, "db"); + string snapshotDirectory = Path.Combine(_tempDir.Path, "snapshot"); + Directory.CreateDirectory(snapshotDirectory); + _config = new SnapshotConfig + { + Enabled = true, + DownloadUrl = _server.Url, + SnapshotDirectory = snapshotDirectory, + SnapshotFileName = "snapshot.tar.zst", + Streaming = true, + }; + } + + [TearDown] + public void TearDown() + { + _server.Dispose(); + _tempDir.Dispose(); + } + + [Test] + public async Task InitializeAsync_ValidSnapshot_ExtractsDatabaseAndCompletesCheckpoint() + { + Dictionary files = TestArchive.BuildFiles(); + byte[] archive = TestArchive.BuildTarZst(files); + _server.Content = archive; + _config.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + string staleArchivePath = Path.Combine(_config.SnapshotDirectory, _config.SnapshotFileName); + File.WriteAllBytes(staleArchivePath, [1, 2, 3]); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None); + + AssertExtracted(files); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Completed), "a verified extraction must complete the checkpoint"); + Assert.That(File.Exists(staleArchivePath), Is.False, "a stale archive file must be deleted since streaming never uses it"); + } + + [Test] + public async Task InitializeAsync_EveryRangeDroppedOnFirstAttempt_ExtractsDatabase() + { + Dictionary files = TestArchive.BuildFiles(); + byte[] archive = TestArchive.BuildTarZst(files); + _server.Content = archive; + _server.DropFirstAttemptPerRangeAfterBytes = 2000; + _config.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None); + + AssertExtracted(files); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Completed), "dropped connections must be resumed without corrupting the extraction"); + } + + [Test] + public async Task InitializeAsync_ChecksumMismatch_DeletesDatabase() + { + byte[] archive = TestArchive.BuildTarZst(TestArchive.BuildFiles()); + _server.Content = archive; + _config.Checksum = Convert.ToHexString(SHA256.HashData([9, 9, 9])); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None); + + Assert.That(Directory.Exists(_dbPath), Is.False, "a database extracted from an unverified snapshot must be deleted"); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Started), "the checkpoint must not advance when the checksum fails"); + } + + [Test] + public async Task InitializeAsync_CorruptArchive_DeletesDatabaseWithoutThrowing() + { + byte[] garbage = new byte[50_000]; + new Random(42).NextBytes(garbage); + _server.Content = garbage; + _config.Checksum = Convert.ToHexString(SHA256.HashData(garbage)); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None); + + Assert.That(Directory.Exists(_dbPath), Is.False, "a corrupt archive must not leave a partially extracted database behind"); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Started), "the checkpoint must not advance when extraction fails"); + } + + [Test] + public void InitializeAsync_ZipArchiveConfigured_Throws() + { + _config.SnapshotFileName = "snapshot.zip"; + + Assert.ThrowsAsync( + () => CreateInitializer().InitializeAsync(CreateCheckpoint(), CancellationToken.None), + "zip archives cannot be extracted from a stream and must be rejected upfront"); + } + + [Test] + public async Task InitializeAsync_SourceChangesOnce_RestartsAndCompletesWithNewSource() + { + Dictionary oldFiles = TestArchive.BuildFiles(seed: 42); + byte[] oldArchive = TestArchive.BuildTarZst(oldFiles); + Dictionary newFiles = TestArchive.BuildFiles(seed: 43); + byte[] newArchive = TestArchive.BuildTarZst(newFiles); + Assert.That(oldArchive.Length, Is.GreaterThan(TestChunkSize), "precondition: the old archive must span multiple chunks so the switch happens mid-download"); + _server.Content = oldArchive; + _server.SwitchSourceAfterRequests(2, newArchive, "\"v2\""); + _config.Checksum = Convert.ToHexString(SHA256.HashData(newArchive)); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer(connections: 1).InitializeAsync(checkpoint, CancellationToken.None); + + AssertExtracted(newFiles); + Assert.That(File.Exists(Path.Combine(_dbPath, "state/a42.sst")), Is.False, + "files extracted from the abandoned first attempt must not survive the restart"); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Completed), "a source change must restart the download and complete with the new object"); + } + + [Test] + public void InitializeAsync_InsufficientDiskSpace_Throws() + { + _server.Content = TestArchive.BuildTarZst(TestArchive.BuildFiles()); + IDriveInfo drive = Substitute.For(); + drive.AvailableFreeSpace.Returns(10); + drive.RootDirectory.FullName.Returns("/db-drive"); + + IOException exception = Assert.ThrowsAsync( + () => CreateInitializer(drives: [drive]).InitializeAsync(CreateCheckpoint(), CancellationToken.None), + "the disk space check must fail before any byte is extracted")!; + + Assert.That(exception.Message, Does.Contain("Insufficient disk space"), "the failure must be the disk space guard, not a download error"); + Assert.That(Directory.Exists(_dbPath), Is.False, "nothing must be extracted when free space is insufficient"); + } + + private StreamingSnapshotInitializer CreateInitializer(int connections = 2, IDriveInfo[]? drives = null) => + new(_config, _server.Url, _dbPath, drives ?? [], + new SnapshotStreamSettings(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50)), + LimboLogs.Instance); + + private SnapshotCheckpoint CreateCheckpoint() => new(_config, LimboLogs.Instance); + + private void AssertExtracted(IReadOnlyDictionary files) + { + foreach ((string name, byte[] data) in files) + { + string path = Path.Combine(_dbPath, name); + Assert.That(File.Exists(path), Is.True, $"extracted file '{name}' must exist because it is part of the archive"); + Assert.That(File.ReadAllBytes(path), Is.EqualTo(data), $"extracted file '{name}' must match the archived content byte for byte"); + } + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs new file mode 100644 index 000000000000..5c7e9a5c5c0c --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs @@ -0,0 +1,52 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.Formats.Tar; +using ZstdSharp; + +namespace Nethermind.Init.Snapshot.Test; + +internal static class TestArchive +{ + public static byte[] BuildTarZst(IReadOnlyDictionary files) + { + using MemoryStream tarBuffer = new(); + using (TarWriter writer = new(tarBuffer, leaveOpen: true)) + { + writer.WriteEntry(new PaxTarEntry(TarEntryType.Directory, "db")); + HashSet directories = []; + foreach (string name in files.Keys) + { + string? parent = Path.GetDirectoryName(name); + if (!string.IsNullOrEmpty(parent) && directories.Add(parent)) + writer.WriteEntry(new PaxTarEntry(TarEntryType.Directory, $"db/{parent}")); + } + + foreach ((string name, byte[] data) in files) + { + PaxTarEntry entry = new(TarEntryType.RegularFile, $"db/{name}") { DataStream = new MemoryStream(data) }; + writer.WriteEntry(entry); + } + } + + tarBuffer.Position = 0; + using MemoryStream compressed = new(); + using (CompressionStream zstd = new(compressed, leaveOpen: true)) + tarBuffer.CopyTo(zstd); + return compressed.ToArray(); + } + + public static Dictionary BuildFiles(int seed = 42) + { + Random random = new(seed); + byte[] state = new byte[50_000]; + byte[] headers = new byte[30_000]; + random.NextBytes(state); + random.NextBytes(headers); + return new Dictionary + { + [$"state/a{seed}.sst"] = state, + [$"headers/b{seed}.sst"] = headers, + }; + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs b/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs index dd7c4985021a..85d179356030 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs @@ -24,4 +24,10 @@ public interface ISnapshotConfig : IConfig [ConfigItem(Description = "Number of leading path components to strip when extracting a tar archive (passed as --strip-components to tar). Must be non-negative. Set this to match the depth of the snapshot path embedded in the archive.", DefaultValue = "1")] int StripComponents { get; set; } + + [ConfigItem(Description = "Whether to stream the snapshot directly into the database directory without storing the archive file, reducing peak disk usage to the extracted size. Interrupted connections are resumed automatically. Supported for tar-based archives only.", DefaultValue = "false")] + bool Streaming { get; set; } + + [ConfigItem(Description = "The number of parallel connections the streaming snapshot download uses when the server supports range requests.", DefaultValue = "4")] + int StreamingConnections { get; set; } } diff --git a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs index 72bb45a33320..15140a527014 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs @@ -58,6 +58,9 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) if (snapshotConfig.StripComponents < 0) throw new InvalidOperationException($"Snapshot.StripComponents must be non-negative, got {snapshotConfig.StripComponents}."); + if (snapshotConfig.Streaming && snapshotConfig.StreamingConnections < 1) + throw new InvalidOperationException($"Snapshot.StreamingConnections must be positive, got {snapshotConfig.StreamingConnections}."); + SnapshotCheckpoint checkpoint = new(snapshotConfig, api.LogManager); if (Path.Exists(dbPath)) @@ -79,6 +82,15 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) Directory.CreateDirectory(snapshotConfig.SnapshotDirectory); + if (snapshotConfig.Streaming) + { + StreamingSnapshotInitializer initializer = new( + snapshotConfig, snapshotUrl, dbPath, drives, + SnapshotStreamSettings.Default(snapshotConfig.StreamingConnections), api.LogManager); + await initializer.InitializeAsync(checkpoint, cancellationToken).ConfigureAwait(false); + return; + } + using SnapshotDownloader downloader = new(api.LogManager); await DownloadWithRetryAsync(downloader, snapshotUrl, snapshotPath, checkpoint, cancellationToken).ConfigureAwait(false); @@ -122,10 +134,7 @@ private async Task DownloadWithRetryAsync( await downloader.DownloadAsync(url, destinationPath, cancellationToken).ConfigureAwait(false); break; } - catch (HttpRequestException e) when ( - e.StatusCode is >= HttpStatusCode.BadRequest and < HttpStatusCode.InternalServerError - and not HttpStatusCode.TooManyRequests - and not HttpStatusCode.RequestedRangeNotSatisfiable) + catch (HttpRequestException e) when (SnapshotHttpClient.IsPermanentHttpError(e)) { if (_logger.IsError) _logger.Error($"Snapshot download failed with permanent HTTP error {(int?)e.StatusCode}. Aborting."); @@ -160,41 +169,54 @@ private async Task VerifyChecksumAsync( if (checkpoint.Read() >= SnapshotStage.Verified) return true; - if (config.Checksum is null) - { - if (_logger.IsWarn) - _logger.Warn("Snapshot checksum is not configured."); - } - else + if (config.Checksum is not null) { if (_logger.IsInfo) _logger.Info($"Verifying snapshot checksum {config.Checksum}."); - byte[] expected = Bytes.FromHexString(config.Checksum); byte[] actual = await ComputeChecksumAsync(snapshotPath, cancellationToken).ConfigureAwait(false); - - if (!Bytes.AreEqual(actual, expected)) - { - if (_logger.IsError) - _logger.Error($"Snapshot checksum verification failed. Expected: {config.Checksum}, actual: {Convert.ToHexString(actual).ToLowerInvariant()}. Aborting snapshot initialization, but the node will continue running."); + if (!VerifyChecksum(actual, config.Checksum, "Aborting snapshot initialization, but the node will continue running.", _logger)) return false; - } - - if (_logger.IsInfo) - _logger.Info("Snapshot checksum verified."); + } + else if (_logger.IsWarn) + { + _logger.Warn("Snapshot checksum is not configured."); } checkpoint.Advance(SnapshotStage.Verified); return true; } + internal static bool VerifyChecksum(byte[] actual, string? expectedHex, string onMismatch, ILogger logger) + { + if (expectedHex is null) + { + if (logger.IsWarn) + logger.Warn("Snapshot checksum is not configured."); + return true; + } + + byte[] expected = Bytes.FromHexString(expectedHex); + if (Bytes.AreEqual(actual, expected)) + { + if (logger.IsInfo) + logger.Info("Snapshot checksum verified."); + return true; + } + + if (logger.IsError) + logger.Error( + $"Snapshot checksum verification failed. Expected: {expectedHex}, actual: {Convert.ToHexString(actual).ToLowerInvariant()}. {onMismatch}"); + return false; + } + private async Task ExtractAsync( string snapshotPath, string dbPath, int stripComponents, SnapshotCheckpoint checkpoint, CancellationToken cancellationToken) { if (checkpoint.Read() >= SnapshotStage.Extracted) return; - CheckDiskSpace(GetRequiredSpaceForExtraction(GetFileSize(snapshotPath)), "extract"); + CheckDiskSpace(drives, GetRequiredSpaceForExtraction(GetFileSize(snapshotPath)), "extract"); SnapshotExtractor extractor = new(api.LogManager); await extractor.ExtractAsync(snapshotPath, dbPath, stripComponents, cancellationToken).ConfigureAwait(false); @@ -224,7 +246,7 @@ private async Task CheckDiskSpaceBeforeDownloadAsync( return; } - CheckDiskSpace(GetRequiredSpaceForDownload(totalSize.Value, existingSize), "download and extract"); + CheckDiskSpace(drives, GetRequiredSpaceForDownload(totalSize.Value, existingSize), "download and extract"); } internal static long GetRequiredSpaceForDownload(long totalSize, long existingSize) => @@ -233,7 +255,7 @@ internal static long GetRequiredSpaceForDownload(long totalSize, long existingSi internal static long GetRequiredSpaceForExtraction(long snapshotSize) => (long)(snapshotSize * ExtractionSpaceMultiplier); - private void CheckDiskSpace(long required, string operation) + internal static void CheckDiskSpace(IDriveInfo[] drives, long required, string operation) { foreach (IDriveInfo drive in drives) { diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotConfig.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotConfig.cs index 01d1ca6688c1..7ce8a55c3bea 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotConfig.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotConfig.cs @@ -16,4 +16,8 @@ public class SnapshotConfig : ISnapshotConfig public string SnapshotFileName { get; set; } = "snapshot.zip"; public int StripComponents { get; set; } = 1; + + public bool Streaming { get; set; } + + public int StreamingConnections { get; set; } = 4; } diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs index 36b100e9801b..799af5fdba94 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs @@ -11,18 +11,15 @@ namespace Nethermind.Init.Snapshot; /// /// Downloads a snapshot file from a URL with resumable download support. -/// Manually follows HTTP redirects to preserve the Range header, which standard -/// HttpClient strips on auto-redirect. /// internal sealed class SnapshotDownloader(ILogManager logManager) : IDisposable { private const int BufferSize = 65536; - private const int MaxRedirects = 10; private const int ResumeWarningDelaySeconds = 5; private static readonly TimeSpan ProgressInterval = TimeSpan.FromSeconds(5); - // A single HttpClient is shared for all retries to preserve the connection pool. - private readonly HttpClient _httpClient = new(new HttpClientHandler { AllowAutoRedirect = false }); + // A single client is shared for all retries to preserve the connection pool. + private readonly SnapshotHttpClient _client = new(); private readonly ILogger _logger = logManager.GetClassLogger(); /// @@ -49,7 +46,8 @@ public async Task DownloadAsync(string url, string destinationPath, Cancellation await Task.Delay(TimeSpan.FromSeconds(ResumeWarningDelaySeconds), cancellationToken).ConfigureAwait(false); } - using HttpResponseMessage response = await SendWithRangeAsync(_httpClient, url, existingSize, cancellationToken).ConfigureAwait(false); + using HttpResponseMessage response = await _client.GetAsync( + url, existingSize > 0 ? new RangeHeaderValue(existingSize, null) : null, ifRange: null, cancellationToken).ConfigureAwait(false); if (_logger.IsInfo) _logger.Info($"Server response: {response.StatusCode}, ETag: {response.Headers.ETag}, Last-Modified: {response.Content.Headers.LastModified}"); @@ -71,11 +69,11 @@ public async Task DownloadAsync(string url, string destinationPath, Cancellation ulong initialProgress = fileMode == FileMode.Append ? (ulong)existingSize : 0UL; using ProgressReporter progress = new("Snapshot download", logManager, (ulong)(totalSize ?? 0), ProgressInterval); - progress.Logger.SetFormat(FormatBytes(totalSize)); + progress.Logger.SetFormat(SnapshotProgress.FormatBytes("Snapshot download", totalSize)); progress.Update(initialProgress); if (bytesToSkip > 0) - await SkipBytesAsync(contentStream, bytesToSkip, cancellationToken).ConfigureAwait(false); + await SnapshotHttpClient.SkipAsync(contentStream, bytesToSkip, cancellationToken).ConfigureAwait(false); await CopyWithProgressAsync(contentStream, fileStream, progress, cancellationToken).ConfigureAwait(false); @@ -85,7 +83,8 @@ public async Task DownloadAsync(string url, string destinationPath, Cancellation public async Task GetTotalSizeAsync(string url, long existingSize, CancellationToken cancellationToken) { - using HttpResponseMessage response = await SendWithRangeAsync(_httpClient, url, existingSize, cancellationToken).ConfigureAwait(false); + using HttpResponseMessage response = await _client.GetAsync( + url, existingSize > 0 ? new RangeHeaderValue(existingSize, null) : null, ifRange: null, cancellationToken).ConfigureAwait(false); if (response.StatusCode == HttpStatusCode.RequestedRangeNotSatisfiable) return existingSize; @@ -93,7 +92,7 @@ public async Task DownloadAsync(string url, string destinationPath, Cancellation return ResolveCopyStrategy(response.StatusCode, existingSize, response.Content.Headers.ContentLength).totalSize; } - public void Dispose() => _httpClient.Dispose(); + public void Dispose() => _client.Dispose(); private static (FileMode fileMode, long bytesToSkip, long? totalSize) ResolveCopyStrategy( HttpStatusCode statusCode, long existingSize, long? contentLength) => @@ -109,65 +108,6 @@ private static (FileMode fileMode, long bytesToSkip, long? totalSize) ResolveCop _ => throw new IOException($"Unexpected HTTP status: {statusCode}") }; - private static async Task SkipBytesAsync(Stream stream, long bytesToSkip, CancellationToken cancellationToken) - { - byte[] buffer = ArrayPool.Shared.Rent(BufferSize); - try - { - long remaining = bytesToSkip; - while (remaining > 0) - { - int chunk = (int)Math.Min(buffer.Length, remaining); - await stream.ReadAtLeastAsync(buffer.AsMemory(0, chunk), chunk, throwOnEndOfStream: true, cancellationToken).ConfigureAwait(false); - remaining -= chunk; - } - } - finally - { - ArrayPool.Shared.Return(buffer); - } - } - - private static async Task SendWithRangeAsync( - HttpClient httpClient, string url, long existingSize, CancellationToken cancellationToken) - { - Uri currentUri = new(url); - - for (int redirects = 0; redirects < MaxRedirects; redirects++) - { - using HttpRequestMessage request = new(HttpMethod.Get, currentUri); - if (existingSize > 0) - request.Headers.Range = new RangeHeaderValue(existingSize, null); - - HttpResponseMessage response = await httpClient.SendAsync( - request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false); - - switch (response.StatusCode) - { - case HttpStatusCode.MovedPermanently - or HttpStatusCode.Found - or HttpStatusCode.SeeOther - or HttpStatusCode.TemporaryRedirect - or HttpStatusCode.PermanentRedirect: - { - Uri? location = response.Headers.Location; - response.Dispose(); - if (location is null) - throw new IOException("Redirect response missing Location header."); - currentUri = new Uri(currentUri, location); // resolve relative redirects - continue; - } - // Let the caller handle 416 — it means the file is already complete. - case HttpStatusCode.RequestedRangeNotSatisfiable: - return response; - default: - return response.EnsureSuccessStatusCode(); - } - } - - throw new IOException("Too many redirects while downloading snapshot."); - } - private static async Task CopyWithProgressAsync( Stream source, FileStream destination, ProgressReporter progress, CancellationToken cancellationToken) { @@ -189,17 +129,4 @@ private static async Task CopyWithProgressAsync( } } - private static Func FormatBytes(long? totalBytes) => - totalBytes is null - ? static logger => $"Snapshot download {HumanReadableSize(logger.CurrentValue)}" - : logger => $"Snapshot download {HumanReadableSize(logger.CurrentValue)} out of {HumanReadableSize((ulong)totalBytes.Value)}"; - - private static string HumanReadableSize(ulong byteCount) => - byteCount switch - { - < MemorySizes.KiB => $"{byteCount:0.##}B", - < MemorySizes.MiB => $"{(float)byteCount / MemorySizes.KiB:0.##}KB", - < MemorySizes.GiB => $"{(float)byteCount / MemorySizes.MiB:0.##}MB", - _ => $"{(float)byteCount / MemorySizes.GiB:0.##}GB", - }; } diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs index b1e68619fced..b40f9853a5c6 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs @@ -23,6 +23,24 @@ internal sealed class SnapshotExtractor(ILogManager logManager) public Task ExtractAsync(string archivePath, string destinationPath, int stripComponents, CancellationToken cancellationToken) => Task.Run(() => Extract(archivePath, destinationPath, stripComponents, cancellationToken), cancellationToken); + public Task ExtractTarStreamAsync(Stream archiveStream, string destinationPath, string extension, int stripComponents, CancellationToken cancellationToken) => + Task.Run(() => + { + if (_logger.IsInfo) + _logger.Info($"Extracting streamed snapshot to {destinationPath}. Do not interrupt!"); + + Stream decompressedStream = OpenDecompressedStream(archiveStream, extension, leaveOpen: true); + try + { + ExtractTarEntries(decompressedStream, destinationPath, stripComponents, cancellationToken); + } + finally + { + if (!ReferenceEquals(decompressedStream, archiveStream)) + decompressedStream.Dispose(); + } + }, cancellationToken); + private void Extract(string archivePath, string destinationPath, int stripComponents, CancellationToken cancellationToken) { if (_logger.IsInfo) @@ -54,12 +72,17 @@ private static void ExtractZip(string archivePath, string destinationPath, Cance } private static void ExtractTar(string archivePath, string destinationPath, string extension, int stripComponents, CancellationToken cancellationToken) + { + using FileStream fileStream = File.OpenRead(archivePath); + using Stream decompressedStream = OpenDecompressedStream(fileStream, extension, leaveOpen: false); + ExtractTarEntries(decompressedStream, destinationPath, stripComponents, cancellationToken); + } + + private static void ExtractTarEntries(Stream decompressedStream, string destinationPath, int stripComponents, CancellationToken cancellationToken) { Directory.CreateDirectory(destinationPath); - using FileStream fileStream = File.OpenRead(archivePath); - using Stream decompressedStream = OpenDecompressedStream(fileStream, extension); - using TarReader tarReader = new(decompressedStream); + using TarReader tarReader = new(decompressedStream, leaveOpen: true); string destinationRoot = destinationPath.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; @@ -84,14 +107,14 @@ private static void ExtractTar(string archivePath, string destinationPath, strin } } - private static Stream OpenDecompressedStream(Stream fileStream, string extension) => + private static Stream OpenDecompressedStream(Stream archiveStream, string extension, bool leaveOpen) => extension switch { - ".zst" or ".zstd" => new DecompressionStream(fileStream), - ".gz" => new GZipStream(fileStream, CompressionMode.Decompress), + ".zst" or ".zstd" => new DecompressionStream(archiveStream, leaveOpen: leaveOpen), + ".gz" => new GZipStream(archiveStream, CompressionMode.Decompress, leaveOpen), // .bz2 and .xz are matched by IsTarArchive but have no decompression support in .NET BCL. ".bz2" or ".xz" => throw new NotSupportedException($"Tar compression format '{extension}' is not supported. Use .gz or .zst instead."), - _ => fileStream + _ => archiveStream }; /// diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs new file mode 100644 index 000000000000..32ee16edb2ed --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs @@ -0,0 +1,99 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.Buffers; +using System.Net; +using System.Net.Http.Headers; + +namespace Nethermind.Init.Snapshot; + +internal sealed record SnapshotRemoteInfo(long? Length, EntityTagHeaderValue? ETag, bool SupportsRanges); + +internal sealed class SnapshotHttpClient : IDisposable +{ + private const int MaxRedirects = 10; + private const int SkipBufferSize = 65536; + + private readonly HttpClient _httpClient = new(new HttpClientHandler { AllowAutoRedirect = false }); + + public async Task ProbeAsync(string url, CancellationToken cancellationToken) + { + using HttpResponseMessage response = await GetAsync(url, new RangeHeaderValue(0, 0), ifRange: null, cancellationToken).ConfigureAwait(false); + EntityTagHeaderValue? etag = response.Headers.ETag; + + if (response.StatusCode == HttpStatusCode.PartialContent) + { + long? length = response.Content.Headers.ContentRange?.Length; + return new SnapshotRemoteInfo(length, etag, length is not null); + } + + return new SnapshotRemoteInfo(response.Content.Headers.ContentLength, etag, false); + } + + public async Task GetAsync( + string url, RangeHeaderValue? range, EntityTagHeaderValue? ifRange, CancellationToken cancellationToken) + { + Uri currentUri = new(url); + + for (int redirects = 0; redirects < MaxRedirects; redirects++) + { + using HttpRequestMessage request = new(HttpMethod.Get, currentUri); + if (range is not null) + request.Headers.Range = range; + if (ifRange is not null && !ifRange.IsWeak) + request.Headers.IfRange = new RangeConditionHeaderValue(ifRange); + + HttpResponseMessage response = await _httpClient.SendAsync( + request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false); + + switch (response.StatusCode) + { + case HttpStatusCode.MovedPermanently + or HttpStatusCode.Found + or HttpStatusCode.SeeOther + or HttpStatusCode.TemporaryRedirect + or HttpStatusCode.PermanentRedirect: + { + Uri? location = response.Headers.Location; + response.Dispose(); + if (location is null) + throw new IOException("Redirect response missing Location header."); + currentUri = new Uri(currentUri, location); + continue; + } + case HttpStatusCode.RequestedRangeNotSatisfiable: + return response; + default: + return response.EnsureSuccessStatusCode(); + } + } + + throw new IOException("Too many redirects while downloading snapshot."); + } + + public void Dispose() => _httpClient.Dispose(); + + public static bool IsPermanentHttpError(HttpRequestException e) => + e.StatusCode is >= HttpStatusCode.BadRequest and < HttpStatusCode.InternalServerError + and not HttpStatusCode.TooManyRequests + and not HttpStatusCode.RequestedRangeNotSatisfiable; + + public static async Task SkipAsync(Stream content, long bytesToSkip, CancellationToken cancellationToken) + { + byte[] buffer = ArrayPool.Shared.Rent(SkipBufferSize); + try + { + long remaining = bytesToSkip; + while (remaining > 0) + { + int chunk = (int)Math.Min(SkipBufferSize, remaining); + await content.ReadAtLeastAsync(buffer.AsMemory(0, chunk), chunk, throwOnEndOfStream: true, cancellationToken).ConfigureAwait(false); + remaining -= chunk; + } + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs new file mode 100644 index 000000000000..ad2b8dd83403 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs @@ -0,0 +1,454 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.Buffers; +using System.Collections.Concurrent; +using System.Net; +using System.Net.Http.Headers; +using System.Runtime.ExceptionServices; +using System.Security.Cryptography; +using Nethermind.Core; +using Nethermind.Logging; + +namespace Nethermind.Init.Snapshot; + +internal sealed record SnapshotStreamSettings(int Connections, int ChunkSize, TimeSpan InitialRetryDelay, TimeSpan MaxRetryDelay) +{ + public static SnapshotStreamSettings Default(int connections) => + new(connections, 64 * 1024 * 1024, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(300)); +} + +internal sealed class SnapshotHttpStream : Stream +{ + private const string ProgressLabel = "Snapshot streaming"; + private const int DrainBufferSize = 65536; + private const int MaxRangeRejections = 5; + private static readonly TimeSpan ProgressInterval = TimeSpan.FromSeconds(5); + + private readonly SnapshotHttpClient _client; + private readonly string _url; + private readonly SnapshotRemoteInfo _remoteInfo; + private readonly SnapshotStreamSettings _settings; + private readonly ILogger _logger; + private readonly IncrementalHash _hasher = IncrementalHash.CreateHash(HashAlgorithmName.SHA256); + private readonly CancellationTokenSource _cts; + private readonly SemaphoreSlim _window; + private readonly ConcurrentDictionary> _pending = new(); + private readonly ConcurrentBag _buffers = []; + private readonly ProgressReporter _progress; + private readonly Task[] _producers; + private readonly long _chunkCount; + + private Exception? _fault; + private long _nextChunkToFetch = -1; + private long _consumeIndex; + private Chunk _current; + private int _currentConsumed; + private long _position; + private bool _finished; + private int _disposed; + + public SnapshotHttpStream( + SnapshotHttpClient client, + string url, + SnapshotRemoteInfo remoteInfo, + SnapshotStreamSettings settings, + ILogManager logManager, + CancellationToken cancellationToken) + { + _client = client; + _url = url; + _remoteInfo = remoteInfo; + _settings = settings; + _logger = logManager.GetClassLogger(); + _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + _window = new SemaphoreSlim(settings.Connections + 1, settings.Connections + 1); + _progress = new ProgressReporter(ProgressLabel, logManager, (ulong)(remoteInfo.Length ?? 0), ProgressInterval); + _progress.Logger.SetFormat(SnapshotProgress.FormatBytes(ProgressLabel, remoteInfo.Length)); + + if (remoteInfo is { SupportsRanges: true, Length: long length }) + { + _chunkCount = (length + settings.ChunkSize - 1) / settings.ChunkSize; + _producers = new Task[settings.Connections]; + for (int i = 0; i < settings.Connections; i++) + _producers[i] = Task.Run(FetchChunksAsync); + } + else + { + _producers = [Task.Run(ReadSequentiallyAsync)]; + } + } + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => _remoteInfo.Length ?? throw new NotSupportedException(); + + public override long Position + { + get => _position; + set => throw new NotSupportedException(); + } + + public override int Read(byte[] buffer, int offset, int count) => Read(buffer.AsSpan(offset, count)); + + public override int Read(Span buffer) + { + if (buffer.IsEmpty || _finished) + return 0; + + if (_current.Buffer is null && !TryTakeNextChunk()) + return 0; + + int toCopy = Math.Min(buffer.Length, _current.Length - _currentConsumed); + _current.Buffer!.AsSpan(_currentConsumed, toCopy).CopyTo(buffer); + _hasher.AppendData(_current.Buffer!, _currentConsumed, toCopy); + _currentConsumed += toCopy; + _position += toCopy; + _progress.Update((ulong)_position); + + if (_currentConsumed == _current.Length) + { + _buffers.Add(_current.Buffer!); + _current = default; + _window.Release(); + } + + return toCopy; + } + + public async Task FinishAsync(CancellationToken cancellationToken) + { + await Task.Run(Drain, cancellationToken).ConfigureAwait(false); + if (_remoteInfo.Length is long expected && _position != expected) + throw new IOException($"Snapshot stream ended after {_position} bytes, expected {expected}."); + return _hasher.GetHashAndReset(); + } + + public override void Flush() { } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + public override async ValueTask DisposeAsync() + { + await DisposeCoreAsync().ConfigureAwait(false); + await base.DisposeAsync().ConfigureAwait(false); + } + + protected override void Dispose(bool disposing) + { + if (disposing) + DisposeCoreAsync().GetAwaiter().GetResult(); + base.Dispose(disposing); + } + + private async Task DisposeCoreAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + _cts.Cancel(); + await Task.WhenAll(_producers).ConfigureAwait(false); + _progress.Dispose(); + _hasher.Dispose(); + _cts.Dispose(); + _window.Dispose(); + } + + private bool TryTakeNextChunk() + { + Chunk chunk = AwaitChunk(_consumeIndex); + _pending.TryRemove(_consumeIndex, out _); + _consumeIndex++; + + if (chunk.Buffer is null) + { + _finished = true; + return false; + } + + _current = chunk; + _currentConsumed = 0; + return true; + } + + private Chunk AwaitChunk(long index) + { + try + { + return GetOrAddPending(index).Task.GetAwaiter().GetResult(); + } + catch (OperationCanceledException) + { + Exception? fault = Volatile.Read(ref _fault); + if (fault is not null and not OperationCanceledException) + ExceptionDispatchInfo.Capture(fault).Throw(); + throw; + } + } + + private void Drain() + { + byte[] scratch = ArrayPool.Shared.Rent(DrainBufferSize); + try + { + while (Read(scratch, 0, DrainBufferSize) > 0) + { + } + } + finally + { + ArrayPool.Shared.Return(scratch); + } + } + + private async Task FetchChunksAsync() + { + try + { + while (true) + { + await _window.WaitAsync(_cts.Token).ConfigureAwait(false); + long index = Interlocked.Increment(ref _nextChunkToFetch); + if (index >= _chunkCount) + { + _window.Release(); + if (index == _chunkCount) + Deliver(index, default); + return; + } + + long offset = index * (long)_settings.ChunkSize; + int length = (int)Math.Min(_settings.ChunkSize, _remoteInfo.Length!.Value - offset); + byte[] buffer = RentBuffer(); + await FetchChunkAsync(offset, buffer, length).ConfigureAwait(false); + Deliver(index, new Chunk(buffer, length)); + } + } + catch (Exception e) + { + Fail(e); + } + } + + private async Task FetchChunkAsync(long offset, byte[] buffer, int length) + { + TimeSpan retryDelay = _settings.InitialRetryDelay; + int rangeRejections = 0; + int received = 0; + int receivedAtLastFailure = 0; + + while (true) + { + try + { + using HttpResponseMessage response = await _client.GetAsync( + _url, new RangeHeaderValue(offset + received, offset + length - 1), _remoteInfo.ETag, _cts.Token).ConfigureAwait(false); + EnsureSourceUnchanged(response); + + if (response.StatusCode != HttpStatusCode.PartialContent) + { + if (++rangeRejections >= MaxRangeRejections) + throw new InvalidDataException( + $"Server repeatedly answered a range request with {(int)response.StatusCode}. Disable Snapshot.Streaming or use a server that supports range requests."); + throw new IOException($"Server answered a range request at {offset + received} with {(int)response.StatusCode}."); + } + + long? from = response.Content.Headers.ContentRange?.From; + if (from != offset + received) + throw new InvalidDataException($"Server returned a range starting at {from}, expected {offset + received}."); + + await using Stream content = await response.Content.ReadAsStreamAsync(_cts.Token).ConfigureAwait(false); + while (received < length) + { + int read = await content.ReadAsync(buffer.AsMemory(received, length - received), _cts.Token).ConfigureAwait(false); + if (read == 0) + throw new IOException($"Connection ended {received} bytes into a {length} byte chunk at {offset}."); + received += read; + } + + return; + } + catch (HttpRequestException e) when (SnapshotHttpClient.IsPermanentHttpError(e)) + { + throw; + } + catch (Exception e) when (e is IOException or HttpRequestException && e is not SnapshotSourceChangedException) + { + if (received > receivedAtLastFailure) + { + receivedAtLastFailure = received; + retryDelay = _settings.InitialRetryDelay; + } + + if (_logger.IsWarn) + _logger.Warn($"Snapshot chunk at {offset} failed after {received} bytes. Retrying in {retryDelay.TotalSeconds}s. Error: {e.Message}"); + await Task.Delay(retryDelay, _cts.Token).ConfigureAwait(false); + retryDelay = Min(retryDelay * 2, _settings.MaxRetryDelay); + } + } + } + + private async Task ReadSequentiallyAsync() + { + HttpResponseMessage? response = null; + try + { + long produced = 0; + long index = 0; + int filled = 0; + byte[] buffer = RentBuffer(); + TimeSpan retryDelay = _settings.InitialRetryDelay; + Stream? content = null; + + while (true) + { + try + { + if (content is null) + (response, content) = await ConnectSequentialAsync(produced + filled).ConfigureAwait(false); + + int read = await content.ReadAsync(buffer.AsMemory(filled, _settings.ChunkSize - filled), _cts.Token).ConfigureAwait(false); + if (read == 0) + { + if (_remoteInfo.Length is long expected && produced + filled < expected) + throw new IOException($"Connection ended after {produced + filled} bytes, expected {expected}."); + break; + } + + retryDelay = _settings.InitialRetryDelay; + filled += read; + if (filled == _settings.ChunkSize) + { + await _window.WaitAsync(_cts.Token).ConfigureAwait(false); + Deliver(index++, new Chunk(buffer, filled)); + produced += filled; + buffer = RentBuffer(); + filled = 0; + } + } + catch (HttpRequestException e) when (SnapshotHttpClient.IsPermanentHttpError(e)) + { + throw; + } + catch (Exception e) when (e is IOException or HttpRequestException && e is not SnapshotSourceChangedException) + { + response?.Dispose(); + response = null; + content = null; + if (_logger.IsWarn) + _logger.Warn($"Snapshot stream interrupted after {produced + filled} bytes. Retrying in {retryDelay.TotalSeconds}s. Error: {e.Message}"); + await Task.Delay(retryDelay, _cts.Token).ConfigureAwait(false); + retryDelay = Min(retryDelay * 2, _settings.MaxRetryDelay); + } + } + + if (filled > 0) + { + await _window.WaitAsync(_cts.Token).ConfigureAwait(false); + Deliver(index++, new Chunk(buffer, filled)); + } + + Deliver(index, default); + } + catch (Exception e) + { + Fail(e); + } + finally + { + response?.Dispose(); + } + } + + private async Task<(HttpResponseMessage Response, Stream Content)> ConnectSequentialAsync(long skip) + { + RangeHeaderValue? range = skip > 0 ? new RangeHeaderValue(skip, null) : null; + HttpResponseMessage response = await _client.GetAsync(_url, range, _remoteInfo.ETag, _cts.Token).ConfigureAwait(false); + try + { + EnsureSourceUnchanged(response); + + if (response.StatusCode == HttpStatusCode.RequestedRangeNotSatisfiable) + return (response, Stream.Null); + + Stream content = await response.Content.ReadAsStreamAsync(_cts.Token).ConfigureAwait(false); + + if (response.StatusCode == HttpStatusCode.PartialContent) + { + long? from = response.Content.Headers.ContentRange?.From; + if (from != skip) + throw new InvalidDataException($"Server returned a range starting at {from}, expected {skip}."); + } + else if (skip > 0) + { + if (_logger.IsWarn) + _logger.Warn($"Server does not support range requests. Re-reading {skip} already received bytes to resume."); + await SnapshotHttpClient.SkipAsync(content, skip, _cts.Token).ConfigureAwait(false); + } + + return (response, content); + } + catch + { + response.Dispose(); + throw; + } + } + + private void EnsureSourceUnchanged(HttpResponseMessage response) + { + EntityTagHeaderValue? expected = _remoteInfo.ETag; + EntityTagHeaderValue? actual = response.Headers.ETag; + if (expected is not null && actual is not null && expected.Tag != actual.Tag) + throw new SnapshotSourceChangedException($"Snapshot changed on the server during the download (ETag {expected.Tag} -> {actual.Tag})."); + + long? total = response.StatusCode switch + { + HttpStatusCode.PartialContent or HttpStatusCode.RequestedRangeNotSatisfiable => response.Content.Headers.ContentRange?.Length, + HttpStatusCode.OK => response.Content.Headers.ContentLength, + _ => null + }; + if (total is not null && _remoteInfo.Length is not null && total != _remoteInfo.Length) + throw new SnapshotSourceChangedException($"Snapshot size changed on the server during the download ({_remoteInfo.Length} -> {total} bytes)."); + } + + private void Deliver(long index, Chunk chunk) => GetOrAddPending(index).TrySetResult(chunk); + + private TaskCompletionSource GetOrAddPending(long index) + { + TaskCompletionSource tcs = _pending.GetOrAdd( + index, static _ => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)); + if (Volatile.Read(ref _fault) is not null || _cts.IsCancellationRequested) + tcs.TrySetCanceled(); + return tcs; + } + + private void Fail(Exception e) + { + Interlocked.CompareExchange(ref _fault, e, null); + try + { + _cts.Cancel(); + } + catch (ObjectDisposedException) + { + } + + foreach (KeyValuePair> pending in _pending) + pending.Value.TrySetCanceled(); + } + + private byte[] RentBuffer() => _buffers.TryTake(out byte[]? buffer) ? buffer : new byte[_settings.ChunkSize]; + + private static TimeSpan Min(TimeSpan left, TimeSpan right) => left < right ? left : right; + + private readonly record struct Chunk(byte[]? Buffer, int Length); +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotProgress.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotProgress.cs new file mode 100644 index 000000000000..0ce63da9c27d --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotProgress.cs @@ -0,0 +1,23 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using Nethermind.Core; + +namespace Nethermind.Init.Snapshot; + +internal static class SnapshotProgress +{ + public static Func FormatBytes(string prefix, long? totalBytes) => + totalBytes is null + ? logger => $"{prefix} {HumanReadableSize(logger.CurrentValue)}" + : logger => $"{prefix} {HumanReadableSize(logger.CurrentValue)} out of {HumanReadableSize((ulong)totalBytes.Value)}"; + + private static string HumanReadableSize(ulong byteCount) => + byteCount switch + { + < MemorySizes.KiB => $"{byteCount:0.##}B", + < MemorySizes.MiB => $"{(float)byteCount / MemorySizes.KiB:0.##}KB", + < MemorySizes.GiB => $"{(float)byteCount / MemorySizes.MiB:0.##}MB", + _ => $"{(float)byteCount / MemorySizes.GiB:0.##}GB", + }; +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotSourceChangedException.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotSourceChangedException.cs new file mode 100644 index 000000000000..186411f3ebad --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotSourceChangedException.cs @@ -0,0 +1,6 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +namespace Nethermind.Init.Snapshot; + +internal sealed class SnapshotSourceChangedException(string message) : IOException(message); diff --git a/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs b/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs new file mode 100644 index 000000000000..77decb1f7d19 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs @@ -0,0 +1,133 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.IO.Abstractions; +using Nethermind.Logging; +using ZstdSharp; + +namespace Nethermind.Init.Snapshot; + +internal sealed class StreamingSnapshotInitializer( + ISnapshotConfig config, + string url, + string dbPath, + IDriveInfo[] drives, + SnapshotStreamSettings settings, + ILogManager logManager) +{ + private const int MaxSourceChangedRestarts = 3; + + private readonly ILogger _logger = logManager.GetClassLogger(); + + public async Task InitializeAsync(SnapshotCheckpoint checkpoint, CancellationToken cancellationToken) + { + EnsureStreamableArchive(config.SnapshotFileName); + DeleteStaleArchive(); + + for (int attempt = 1; ; attempt++) + { + try + { + await StreamAndExtractAsync(checkpoint, cancellationToken).ConfigureAwait(false); + return; + } + catch (SnapshotSourceChangedException e) when (attempt < MaxSourceChangedRestarts) + { + if (_logger.IsWarn) + _logger.Warn($"{e.Message} Restarting the snapshot download."); + DeleteDatabase(); + } + } + } + + private async Task StreamAndExtractAsync(SnapshotCheckpoint checkpoint, CancellationToken cancellationToken) + { + using SnapshotHttpClient client = new(); + SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(url, cancellationToken).ConfigureAwait(false); + LogMode(remoteInfo); + CheckDiskSpace(remoteInfo.Length); + + await using SnapshotHttpStream stream = new(client, url, remoteInfo, settings, logManager, cancellationToken); + SnapshotExtractor extractor = new(logManager); + string extension = Path.GetExtension(config.SnapshotFileName).ToLowerInvariant(); + byte[] checksum; + try + { + await extractor.ExtractTarStreamAsync(stream, dbPath, extension, config.StripComponents, cancellationToken).ConfigureAwait(false); + checksum = await stream.FinishAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception e) when (e is IOException or InvalidDataException or EndOfStreamException or ZstdException + && e is not SnapshotSourceChangedException) + { + if (_logger.IsError) + _logger.Error($"Snapshot streaming failed: {e.Message} Deleting the partially extracted database; the node will continue running."); + DeleteDatabase(); + return; + } + + if (!VerifyChecksum(checksum)) + { + DeleteDatabase(); + return; + } + + checkpoint.Advance(SnapshotStage.Completed); + if (_logger.IsInfo) + _logger.Info("Database successfully initialized from streamed snapshot."); + } + + private static void EnsureStreamableArchive(string fileName) + { + string extension = Path.GetExtension(fileName).ToLowerInvariant(); + if (extension is not (".tar" or ".zst" or ".zstd" or ".gz")) + throw new NotSupportedException( + $"Snapshot streaming supports only tar-based archives (.tar, .tar.zst, .tar.gz); got '{fileName}'. Disable Snapshot.Streaming to use other formats."); + } + + private void DeleteStaleArchive() + { + string archivePath = Path.Combine(config.SnapshotDirectory, config.SnapshotFileName); + if (!File.Exists(archivePath)) + return; + + if (_logger.IsWarn) + _logger.Warn($"Deleting snapshot file {archivePath}; the streaming download does not use it."); + File.Delete(archivePath); + } + + private void LogMode(SnapshotRemoteInfo remoteInfo) + { + if (!remoteInfo.SupportsRanges) + { + if (_logger.IsWarn) + _logger.Warn("Snapshot server does not support range requests. Streaming with a single connection; every resumed connection re-reads the file from the beginning."); + } + else if (_logger.IsInfo) + { + _logger.Info($"Streaming snapshot from {url} with {settings.Connections} connections."); + } + } + + private void CheckDiskSpace(long? snapshotLength) + { + if (snapshotLength is null) + { + if (_logger.IsWarn) + _logger.Warn("Snapshot size is unknown. Skipping the disk space check."); + return; + } + + long required = InitDatabaseSnapshot.GetRequiredSpaceForExtraction(snapshotLength.Value); + InitDatabaseSnapshot.CheckDiskSpace(drives, required, "extract"); + } + + private bool VerifyChecksum(byte[] actual) => + InitDatabaseSnapshot.VerifyChecksum( + actual, config.Checksum, "Deleting the extracted database; the node will continue running.", _logger); + + private void DeleteDatabase() + { + if (Directory.Exists(dbPath)) + Directory.Delete(dbPath, true); + } +} From 91f6604d8bab1b68c4b6407fc21118647d52a953 Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:02:12 +0300 Subject: [PATCH 03/11] fix: harden snapshot streaming per review Stall timeout on body reads with retry, reject tar link entries, keep a fully downloaded archive and extract it two-phase, bound StreamingConnections to 16, require a checksum when the size is unknown, graceful stop after repeated source changes --- .../FlakySnapshotServer.cs | 28 +++------- .../InitDatabaseSnapshotTests.cs | 46 ++++++++-------- .../SnapshotHttpStreamTests.cs | 2 +- .../StreamingSnapshotInitializerTests.cs | 29 +++++++++- .../TestArchive.cs | 16 ++++++ .../TestHttpListener.cs | 28 ++++++++++ .../ISnapshotConfig.cs | 2 +- .../InitDatabaseSnapshot.cs | 21 +++++--- .../SnapshotExtractor.cs | 7 ++- .../SnapshotHttpStream.cs | 53 ++++++++++++++++--- .../StreamingSnapshotInitializer.cs | 13 +++-- 11 files changed, 179 insertions(+), 66 deletions(-) create mode 100644 src/Nethermind/Nethermind.Init.Snapshot.Test/TestHttpListener.cs diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs index 44ceb5f35acf..27fd82b4e6b9 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs @@ -17,7 +17,7 @@ internal sealed class FlakySnapshotServer : IDisposable public FlakySnapshotServer() { - (_listener, int port) = StartListener(); + (_listener, int port) = TestHttpListener.Start(); Url = $"http://127.0.0.1:{port}/snapshot.tar.zst"; Task.Run(AcceptLoopAsync); } @@ -34,6 +34,8 @@ public FlakySnapshotServer() public int? FailWithNotFoundAfterRequests { get; set; } + public bool OmitContentLength { get; set; } + public int RequestCount => _requestCount; public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, string? newETag) @@ -43,25 +45,6 @@ public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, strin _switchAfterRequests = requestCount; } - private static (HttpListener Listener, int Port) StartListener() - { - for (int attempt = 0; ; attempt++) - { - HttpListener listener = new(); - int port = Random.Shared.Next(20000, 60000); - listener.Prefixes.Add($"http://127.0.0.1:{port}/"); - try - { - listener.Start(); - return (listener, port); - } - catch (HttpListenerException) when (attempt < 5) - { - listener.Close(); - } - } - } - public void Dispose() { _listener.Stop(); @@ -134,7 +117,10 @@ private async Task HandleAsync(HttpListenerContext context) } long length = to - from + 1; - response.ContentLength64 = length; + if (OmitContentLength) + response.SendChunked = true; + else + response.ContentLength64 = length; string rangeKey = rangeHeader ?? "full"; int attempt = _attemptsPerRange.AddOrUpdate(rangeKey, 1, static (_, previous) => previous + 1); diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs index dc28f2e6a51a..9bb31974fb01 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs @@ -124,15 +124,32 @@ public async Task Execute_StreamingEnabled_ExtractsSnapshotFromStream() "the streaming path must extract the snapshot into the database directory"); } - [Test] - public void Execute_StreamingConnectionsNotPositive_Throws() + [TestCase(0, TestName = "Zero")] + [TestCase(17, TestName = "AboveMaximum")] + public void Execute_StreamingConnectionsOutOfRange_Throws(int connections) { _snapshotConfig.Streaming = true; - _snapshotConfig.StreamingConnections = 0; + _snapshotConfig.StreamingConnections = connections; InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); Assert.ThrowsAsync(() => step.Execute(CancellationToken.None), - "a non-positive connection count must be rejected before any network activity"); + "a connection count outside 1-16 must be rejected before any network activity"); + } + + [Test] + public async Task Execute_StreamingEnabledWithDownloadedArchive_ExtractsWithoutStreaming() + { + WriteSnapshotTar(); + AdvanceCheckpoint(SnapshotStage.Verified); + _snapshotConfig.Streaming = true; + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + await step.Execute(CancellationToken.None); + + Assert.That(File.Exists(Path.Combine(_dbPath, "state.bin")), Is.True, + "an already downloaded archive must be extracted through the two-phase path instead of being deleted and re-streamed"); + Assert.That(File.Exists(_snapshotPath), Is.False, + "the archive must be deleted after a successful two-phase extraction"); } [TestCase(1_000, 0, 2_500, TestName = "FreshDownload")] @@ -183,7 +200,7 @@ private SnapshotServer(HttpListener listener, string url) public static SnapshotServer Start(long contentLength) { - (HttpListener listener, int port) = StartListener(); + (HttpListener listener, int port) = TestHttpListener.Start(); _ = Task.Run(async () => { @@ -207,25 +224,6 @@ public static SnapshotServer Start(long contentLength) return new SnapshotServer(listener, $"http://127.0.0.1:{port}/snapshot.tar"); } - private static (HttpListener Listener, int Port) StartListener() - { - for (int attempt = 0; ; attempt++) - { - HttpListener listener = new(); - int port = Random.Shared.Next(20000, 60000); - listener.Prefixes.Add($"http://127.0.0.1:{port}/"); - try - { - listener.Start(); - return (listener, port); - } - catch (HttpListenerException) when (attempt < 5) - { - listener.Close(); - } - } - } - public void Dispose() { _listener.Stop(); diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs index 4a6f14cfd5e9..9f1b42fc9282 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs @@ -112,7 +112,7 @@ public void Read_ServerReturnsNotFoundMidDownload_ThrowsWithoutRetrying() private async Task<(byte[] Delivered, byte[] Hash)> DownloadAsync(int connections) { - SnapshotStreamSettings settings = new(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50)); + SnapshotStreamSettings settings = new(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50), TimeSpan.FromSeconds(30)); using SnapshotHttpClient client = new(); SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(_server.Url, CancellationToken.None); await using SnapshotHttpStream stream = new(client, _server.Url, remoteInfo, settings, LimboLogs.Instance, CancellationToken.None); diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs index ef71956e177b..1aefeb121399 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs @@ -108,6 +108,33 @@ public async Task InitializeAsync_CorruptArchive_DeletesDatabaseWithoutThrowing( Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Started), "the checkpoint must not advance when extraction fails"); } + [Test] + public async Task InitializeAsync_ArchiveWithSymlinkEntry_DeletesDatabaseWithoutThrowing() + { + byte[] archive = TestArchive.BuildTarZstWithSymlink(); + _server.Content = archive; + _config.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None); + + Assert.That(Directory.Exists(_dbPath), Is.False, + "an archive with link entries must be rejected and the partial extraction deleted, because links can escape the database directory"); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Started), "the checkpoint must not advance when extraction is rejected"); + } + + [Test] + public void InitializeAsync_UnknownLengthWithoutChecksum_Throws() + { + _server.Content = TestArchive.BuildTarZst(TestArchive.BuildFiles()); + _server.SupportsRanges = false; + _server.OmitContentLength = true; + + Assert.ThrowsAsync( + () => CreateInitializer().InitializeAsync(CreateCheckpoint(), CancellationToken.None), + "without a length and without a checksum a truncated download would be undetectable, so streaming must refuse to start"); + } + [Test] public void InitializeAsync_ZipArchiveConfigured_Throws() { @@ -157,7 +184,7 @@ public void InitializeAsync_InsufficientDiskSpace_Throws() private StreamingSnapshotInitializer CreateInitializer(int connections = 2, IDriveInfo[]? drives = null) => new(_config, _server.Url, _dbPath, drives ?? [], - new SnapshotStreamSettings(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50)), + new SnapshotStreamSettings(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50), TimeSpan.FromSeconds(30)), LimboLogs.Instance); private SnapshotCheckpoint CreateCheckpoint() => new(_config, LimboLogs.Instance); diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs index 5c7e9a5c5c0c..e30bc6d31c12 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs @@ -36,6 +36,22 @@ public static byte[] BuildTarZst(IReadOnlyDictionary files) return compressed.ToArray(); } + public static byte[] BuildTarZstWithSymlink() + { + using MemoryStream tarBuffer = new(); + using (TarWriter writer = new(tarBuffer, leaveOpen: true)) + { + writer.WriteEntry(new PaxTarEntry(TarEntryType.Directory, "db")); + writer.WriteEntry(new PaxTarEntry(TarEntryType.SymbolicLink, "db/escape") { LinkName = "/tmp" }); + } + + tarBuffer.Position = 0; + using MemoryStream compressed = new(); + using (CompressionStream zstd = new(compressed, leaveOpen: true)) + tarBuffer.CopyTo(zstd); + return compressed.ToArray(); + } + public static Dictionary BuildFiles(int seed = 42) { Random random = new(seed); diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/TestHttpListener.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestHttpListener.cs new file mode 100644 index 000000000000..e4f3d2121283 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestHttpListener.cs @@ -0,0 +1,28 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.Net; + +namespace Nethermind.Init.Snapshot.Test; + +internal static class TestHttpListener +{ + public static (HttpListener Listener, int Port) Start() + { + for (int attempt = 0; ; attempt++) + { + HttpListener listener = new(); + int port = Random.Shared.Next(20000, 60000); + listener.Prefixes.Add($"http://127.0.0.1:{port}/"); + try + { + listener.Start(); + return (listener, port); + } + catch (HttpListenerException) when (attempt < 5) + { + listener.Close(); + } + } + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs b/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs index 85d179356030..ac5a72e0cb99 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs @@ -28,6 +28,6 @@ public interface ISnapshotConfig : IConfig [ConfigItem(Description = "Whether to stream the snapshot directly into the database directory without storing the archive file, reducing peak disk usage to the extracted size. Interrupted connections are resumed automatically. Supported for tar-based archives only.", DefaultValue = "false")] bool Streaming { get; set; } - [ConfigItem(Description = "The number of parallel connections the streaming snapshot download uses when the server supports range requests.", DefaultValue = "4")] + [ConfigItem(Description = "The number of parallel connections the streaming snapshot download uses when the server supports range requests. Each connection buffers up to 64 MiB in memory. Allowed range: 1-16.", DefaultValue = "4")] int StreamingConnections { get; set; } } diff --git a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs index 15140a527014..ccf301dac964 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs @@ -27,6 +27,7 @@ public class InitDatabaseSnapshot( [KeyFilter(nameof(IInitConfig.BaseDbPath))] IDriveInfo[] drives) : IStep { private const double ExtractionSpaceMultiplier = 1.5; + private const int MaxStreamingConnections = 16; private const int ExtractionRestartDelaySeconds = 5; private const int InitialRetryDelaySeconds = 5; private const int MaxRetryDelaySeconds = 300; @@ -58,8 +59,8 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) if (snapshotConfig.StripComponents < 0) throw new InvalidOperationException($"Snapshot.StripComponents must be non-negative, got {snapshotConfig.StripComponents}."); - if (snapshotConfig.Streaming && snapshotConfig.StreamingConnections < 1) - throw new InvalidOperationException($"Snapshot.StreamingConnections must be positive, got {snapshotConfig.StreamingConnections}."); + if (snapshotConfig.Streaming && snapshotConfig.StreamingConnections is < 1 or > MaxStreamingConnections) + throw new InvalidOperationException($"Snapshot.StreamingConnections must be between 1 and {MaxStreamingConnections}, got {snapshotConfig.StreamingConnections}."); SnapshotCheckpoint checkpoint = new(snapshotConfig, api.LogManager); @@ -84,11 +85,17 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) if (snapshotConfig.Streaming) { - StreamingSnapshotInitializer initializer = new( - snapshotConfig, snapshotUrl, dbPath, drives, - SnapshotStreamSettings.Default(snapshotConfig.StreamingConnections), api.LogManager); - await initializer.InitializeAsync(checkpoint, cancellationToken).ConfigureAwait(false); - return; + if (checkpoint.Read() < SnapshotStage.Downloaded || !File.Exists(snapshotPath)) + { + StreamingSnapshotInitializer initializer = new( + snapshotConfig, snapshotUrl, dbPath, drives, + SnapshotStreamSettings.Default(snapshotConfig.StreamingConnections), api.LogManager); + await initializer.InitializeAsync(checkpoint, cancellationToken).ConfigureAwait(false); + return; + } + + if (_logger.IsWarn) + _logger.Warn($"A fully downloaded snapshot archive exists at {snapshotPath}. Extracting it instead of streaming."); } using SnapshotDownloader downloader = new(api.LogManager); diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs index b40f9853a5c6..30306295904b 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs @@ -91,6 +91,9 @@ private static void ExtractTarEntries(Stream decompressedStream, string destinat { cancellationToken.ThrowIfCancellationRequested(); + if (entry.EntryType is TarEntryType.GlobalExtendedAttributes) + continue; + string? strippedPath = StripLeadingComponents(entry.Name, stripComponents); if (strippedPath is null) continue; @@ -102,8 +105,10 @@ private static void ExtractTarEntries(Stream decompressedStream, string destinat if (entry.EntryType is TarEntryType.Directory) Directory.CreateDirectory(destinationEntryPath); - else + else if (entry.EntryType is TarEntryType.RegularFile or TarEntryType.V7RegularFile) entry.ExtractToFile(destinationEntryPath, overwrite: true); + else + throw new IOException($"Tar entry '{entry.Name}' has unsupported type {entry.EntryType}."); } } diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs index ad2b8dd83403..980e5a506ead 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs @@ -12,10 +12,10 @@ namespace Nethermind.Init.Snapshot; -internal sealed record SnapshotStreamSettings(int Connections, int ChunkSize, TimeSpan InitialRetryDelay, TimeSpan MaxRetryDelay) +internal sealed record SnapshotStreamSettings(int Connections, int ChunkSize, TimeSpan InitialRetryDelay, TimeSpan MaxRetryDelay, TimeSpan StallTimeout) { public static SnapshotStreamSettings Default(int connections) => - new(connections, 64 * 1024 * 1024, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(300)); + new(connections, 64 * 1024 * 1024, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(300), TimeSpan.FromMinutes(2)); } internal sealed class SnapshotHttpStream : Stream @@ -268,7 +268,7 @@ private async Task FetchChunkAsync(long offset, byte[] buffer, int length) await using Stream content = await response.Content.ReadAsStreamAsync(_cts.Token).ConfigureAwait(false); while (received < length) { - int read = await content.ReadAsync(buffer.AsMemory(received, length - received), _cts.Token).ConfigureAwait(false); + int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(received, length - received)).ConfigureAwait(false); if (read == 0) throw new IOException($"Connection ended {received} bytes into a {length} byte chunk at {offset}."); received += read; @@ -280,7 +280,7 @@ private async Task FetchChunkAsync(long offset, byte[] buffer, int length) { throw; } - catch (Exception e) when (e is IOException or HttpRequestException && e is not SnapshotSourceChangedException) + catch (Exception e) when (IsRetryable(e)) { if (received > receivedAtLastFailure) { @@ -315,7 +315,7 @@ private async Task ReadSequentiallyAsync() if (content is null) (response, content) = await ConnectSequentialAsync(produced + filled).ConfigureAwait(false); - int read = await content.ReadAsync(buffer.AsMemory(filled, _settings.ChunkSize - filled), _cts.Token).ConfigureAwait(false); + int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(filled, _settings.ChunkSize - filled)).ConfigureAwait(false); if (read == 0) { if (_remoteInfo.Length is long expected && produced + filled < expected) @@ -338,7 +338,7 @@ private async Task ReadSequentiallyAsync() { throw; } - catch (Exception e) when (e is IOException or HttpRequestException && e is not SnapshotSourceChangedException) + catch (Exception e) when (IsRetryable(e)) { response?.Dispose(); response = null; @@ -391,7 +391,7 @@ private async Task ReadSequentiallyAsync() { if (_logger.IsWarn) _logger.Warn($"Server does not support range requests. Re-reading {skip} already received bytes to resume."); - await SnapshotHttpClient.SkipAsync(content, skip, _cts.Token).ConfigureAwait(false); + await SkipWithStallTimeoutAsync(content, skip).ConfigureAwait(false); } return (response, content); @@ -446,6 +446,45 @@ private void Fail(Exception e) pending.Value.TrySetCanceled(); } + private async Task ReadWithStallTimeoutAsync(Stream content, Memory buffer) + { + using CancellationTokenSource stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); + stallCts.CancelAfter(_settings.StallTimeout); + try + { + return await content.ReadAsync(buffer, stallCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!_cts.IsCancellationRequested) + { + throw new IOException($"No data received for {_settings.StallTimeout.TotalSeconds}s."); + } + } + + private async Task SkipWithStallTimeoutAsync(Stream content, long bytesToSkip) + { + byte[] scratch = ArrayPool.Shared.Rent(DrainBufferSize); + try + { + long remaining = bytesToSkip; + while (remaining > 0) + { + int read = await ReadWithStallTimeoutAsync(content, scratch.AsMemory(0, (int)Math.Min(DrainBufferSize, remaining))).ConfigureAwait(false); + if (read == 0) + throw new EndOfStreamException($"Connection ended while skipping {bytesToSkip} already received bytes."); + remaining -= read; + } + } + finally + { + ArrayPool.Shared.Return(scratch); + } + } + + private bool IsRetryable(Exception e) => + e is not SnapshotSourceChangedException + && !_cts.IsCancellationRequested + && e is IOException or HttpRequestException or OperationCanceledException; + private byte[] RentBuffer() => _buffers.TryTake(out byte[]? buffer) ? buffer : new byte[_settings.ChunkSize]; private static TimeSpan Min(TimeSpan left, TimeSpan right) => left < right ? left : right; diff --git a/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs b/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs index 77decb1f7d19..80f5f9660c7a 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs @@ -24,26 +24,33 @@ public async Task InitializeAsync(SnapshotCheckpoint checkpoint, CancellationTok EnsureStreamableArchive(config.SnapshotFileName); DeleteStaleArchive(); - for (int attempt = 1; ; attempt++) + for (int attempt = 1; attempt <= MaxSourceChangedRestarts; attempt++) { try { await StreamAndExtractAsync(checkpoint, cancellationToken).ConfigureAwait(false); return; } - catch (SnapshotSourceChangedException e) when (attempt < MaxSourceChangedRestarts) + catch (SnapshotSourceChangedException e) { if (_logger.IsWarn) _logger.Warn($"{e.Message} Restarting the snapshot download."); DeleteDatabase(); } } + + if (_logger.IsError) + _logger.Error($"The snapshot kept changing on the server across {MaxSourceChangedRestarts} attempts. Giving up; the node will continue running."); } private async Task StreamAndExtractAsync(SnapshotCheckpoint checkpoint, CancellationToken cancellationToken) { using SnapshotHttpClient client = new(); SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(url, cancellationToken).ConfigureAwait(false); + if (remoteInfo.Length is null && config.Checksum is null) + throw new InvalidOperationException( + "The server does not report the snapshot size and Snapshot.Checksum is not set, so a truncated download could not be detected. Set Snapshot.Checksum or disable Snapshot.Streaming."); + LogMode(remoteInfo); CheckDiskSpace(remoteInfo.Length); @@ -81,7 +88,7 @@ private static void EnsureStreamableArchive(string fileName) string extension = Path.GetExtension(fileName).ToLowerInvariant(); if (extension is not (".tar" or ".zst" or ".zstd" or ".gz")) throw new NotSupportedException( - $"Snapshot streaming supports only tar-based archives (.tar, .tar.zst, .tar.gz); got '{fileName}'. Disable Snapshot.Streaming to use other formats."); + $"Snapshot streaming supports only tar-based archives (.tar, .tar.zst, .tar.gz), but Snapshot.SnapshotFileName is '{fileName}'. Set Snapshot.SnapshotFileName to match the archive or disable Snapshot.Streaming."); } private void DeleteStaleArchive() From 7dcf34911950b724a52a5dc192c5c28c7532a096 Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:08:51 +0300 Subject: [PATCH 04/11] ci: add Nethermind.Init.Snapshot.Test to the test matrix --- .github/workflows/nethermind-tests.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/nethermind-tests.yml b/.github/workflows/nethermind-tests.yml index 3648164d0326..872cd7701754 100644 --- a/.github/workflows/nethermind-tests.yml +++ b/.github/workflows/nethermind-tests.yml @@ -74,6 +74,7 @@ jobs: - Nethermind.HealthChecks.Test - Nethermind.History.Test - Nethermind.Hive.Test + - Nethermind.Init.Snapshot.Test - Nethermind.JsonRpc.Test - Nethermind.JsonRpc.TraceStore.Test - Nethermind.KeyStore.Test From f4aa5098127488bc1aa7242f2b43a865bfe296b9 Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:18:11 +0300 Subject: [PATCH 05/11] fix: snapshot streaming review round two Explicit infinite HttpClient timeout with a 100s header guard so body reads are governed only by the stall timeout, one stall CTS per response instead of per read, document that a node restart restarts the download, cover exhausted restarts and stalls --- .../FlakySnapshotServer.cs | 24 +++++++++++++++++++ .../SnapshotHttpStreamTests.cs | 17 +++++++++++-- .../StreamingSnapshotInitializerTests.cs | 14 +++++++++++ .../ISnapshotConfig.cs | 2 +- .../SnapshotHttpClient.cs | 20 +++++++++++++--- .../SnapshotHttpStream.cs | 24 +++++++++++++------ 6 files changed, 88 insertions(+), 13 deletions(-) diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs index 27fd82b4e6b9..ab3ead5565ce 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs @@ -10,7 +10,9 @@ internal sealed class FlakySnapshotServer : IDisposable { private readonly HttpListener _listener; private readonly ConcurrentDictionary _attemptsPerRange = new(); + private readonly CancellationTokenSource _hangCts = new(); private int _requestCount; + private int _hangConsumed; private int _switchAfterRequests = int.MaxValue; private byte[] _newContent = []; private string? _newETag; @@ -36,6 +38,10 @@ public FlakySnapshotServer() public bool OmitContentLength { get; set; } + public bool RotateETagEveryRequest { get; set; } + + public int? HangOnceAfterBytes { get; set; } + public int RequestCount => _requestCount; public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, string? newETag) @@ -47,6 +53,7 @@ public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, strin public void Dispose() { + _hangCts.Cancel(); _listener.Stop(); _listener.Close(); } @@ -74,6 +81,8 @@ private async Task HandleAsync(HttpListenerContext context) int requestNumber = Interlocked.Increment(ref _requestCount); byte[] content = requestNumber > _switchAfterRequests ? _newContent : Content; string? etag = requestNumber > _switchAfterRequests ? _newETag : ETag; + if (RotateETagEveryRequest) + etag = $"\"v{requestNumber}\""; HttpListenerResponse response = context.Response; try @@ -122,6 +131,21 @@ private async Task HandleAsync(HttpListenerContext context) else response.ContentLength64 = length; + if (HangOnceAfterBytes is int hangAfter && length > hangAfter && Interlocked.Exchange(ref _hangConsumed, 1) == 0) + { + await response.OutputStream.WriteAsync(content.AsMemory((int)from, hangAfter)); + await response.OutputStream.FlushAsync(); + try + { + await Task.Delay(TimeSpan.FromSeconds(30), _hangCts.Token); + } + catch (OperationCanceledException) + { + } + response.Abort(); + return; + } + string rangeKey = rangeHeader ?? "full"; int attempt = _attemptsPerRange.AddOrUpdate(rangeKey, 1, static (_, previous) => previous + 1); if (DropFirstAttemptPerRangeAfterBytes is int dropAfter && attempt == 1 && length > dropAfter) diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs index 9f1b42fc9282..653d758400a3 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs @@ -110,9 +110,22 @@ public void Read_ServerReturnsNotFoundMidDownload_ThrowsWithoutRetrying() "a permanent HTTP error must abort the stream instead of retrying forever"); } - private async Task<(byte[] Delivered, byte[] Hash)> DownloadAsync(int connections) + [Test] + public async Task Read_ServerStallsOnce_DetectsStallAndResumes() + { + byte[] content = BuildContent(100_000); + _server.Content = content; + _server.HangOnceAfterBytes = 2000; + + (byte[] delivered, byte[] hash) = await DownloadAsync(connections: 2, stallTimeout: TimeSpan.FromMilliseconds(250)); + + Assert.That(delivered, Is.EqualTo(content), "a connection that stops delivering bytes must be detected as stalled and the chunk re-fetched"); + Assert.That(hash, Is.EqualTo(SHA256.HashData(content)), "bytes received before the stall must be hashed exactly once"); + } + + private async Task<(byte[] Delivered, byte[] Hash)> DownloadAsync(int connections, TimeSpan? stallTimeout = null) { - SnapshotStreamSettings settings = new(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50), TimeSpan.FromSeconds(30)); + SnapshotStreamSettings settings = new(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50), stallTimeout ?? TimeSpan.FromSeconds(30)); using SnapshotHttpClient client = new(); SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(_server.Url, CancellationToken.None); await using SnapshotHttpStream stream = new(client, _server.Url, remoteInfo, settings, LimboLogs.Instance, CancellationToken.None); diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs index 1aefeb121399..909a64a5bcca 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs @@ -166,6 +166,20 @@ public async Task InitializeAsync_SourceChangesOnce_RestartsAndCompletesWithNewS Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Completed), "a source change must restart the download and complete with the new object"); } + [Test] + public async Task InitializeAsync_SourceKeepsChanging_GivesUpWithoutThrowing() + { + _server.Content = TestArchive.BuildTarZst(TestArchive.BuildFiles()); + _server.RotateETagEveryRequest = true; + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer(connections: 1).InitializeAsync(checkpoint, CancellationToken.None); + + Assert.That(Directory.Exists(_dbPath), Is.False, + "after exhausting the restart budget the partial database must be deleted so the node can continue without a snapshot"); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Started), "the checkpoint must not advance when the download never completes"); + } + [Test] public void InitializeAsync_InsufficientDiskSpace_Throws() { diff --git a/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs b/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs index ac5a72e0cb99..ee05d600c346 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/ISnapshotConfig.cs @@ -25,7 +25,7 @@ public interface ISnapshotConfig : IConfig [ConfigItem(Description = "Number of leading path components to strip when extracting a tar archive (passed as --strip-components to tar). Must be non-negative. Set this to match the depth of the snapshot path embedded in the archive.", DefaultValue = "1")] int StripComponents { get; set; } - [ConfigItem(Description = "Whether to stream the snapshot directly into the database directory without storing the archive file, reducing peak disk usage to the extracted size. Interrupted connections are resumed automatically. Supported for tar-based archives only.", DefaultValue = "false")] + [ConfigItem(Description = "Whether to stream the snapshot directly into the database directory without storing the archive file, reducing peak disk usage to the extracted size. Interrupted connections are resumed automatically within a run, but a node restart discards all progress and starts the download over, since no archive is kept on disk. Supported for tar-based archives only.", DefaultValue = "false")] bool Streaming { get; set; } [ConfigItem(Description = "The number of parallel connections the streaming snapshot download uses when the server supports range requests. Each connection buffers up to 64 MiB in memory. Allowed range: 1-16.", DefaultValue = "4")] diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs index 32ee16edb2ed..ccaddadf9437 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs @@ -13,8 +13,12 @@ internal sealed class SnapshotHttpClient : IDisposable { private const int MaxRedirects = 10; private const int SkipBufferSize = 65536; + private static readonly TimeSpan HeaderTimeout = TimeSpan.FromSeconds(100); - private readonly HttpClient _httpClient = new(new HttpClientHandler { AllowAutoRedirect = false }); + private readonly HttpClient _httpClient = new(new HttpClientHandler { AllowAutoRedirect = false }) + { + Timeout = Timeout.InfiniteTimeSpan + }; public async Task ProbeAsync(string url, CancellationToken cancellationToken) { @@ -43,8 +47,18 @@ public async Task GetAsync( if (ifRange is not null && !ifRange.IsWeak) request.Headers.IfRange = new RangeConditionHeaderValue(ifRange); - HttpResponseMessage response = await _httpClient.SendAsync( - request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false); + using CancellationTokenSource headerCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + headerCts.CancelAfter(HeaderTimeout); + HttpResponseMessage response; + try + { + response = await _httpClient.SendAsync( + request, HttpCompletionOption.ResponseHeadersRead, headerCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (headerCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) + { + throw new HttpRequestException($"No response headers received within {HeaderTimeout.TotalSeconds}s."); + } switch (response.StatusCode) { diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs index 980e5a506ead..8124a0fbf236 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs @@ -266,9 +266,10 @@ private async Task FetchChunkAsync(long offset, byte[] buffer, int length) throw new InvalidDataException($"Server returned a range starting at {from}, expected {offset + received}."); await using Stream content = await response.Content.ReadAsStreamAsync(_cts.Token).ConfigureAwait(false); + using CancellationTokenSource stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); while (received < length) { - int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(received, length - received)).ConfigureAwait(false); + int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(received, length - received), stallCts).ConfigureAwait(false); if (read == 0) throw new IOException($"Connection ended {received} bytes into a {length} byte chunk at {offset}."); received += read; @@ -299,6 +300,7 @@ private async Task FetchChunkAsync(long offset, byte[] buffer, int length) private async Task ReadSequentiallyAsync() { HttpResponseMessage? response = null; + CancellationTokenSource? stallCts = null; try { long produced = 0; @@ -313,9 +315,12 @@ private async Task ReadSequentiallyAsync() try { if (content is null) + { (response, content) = await ConnectSequentialAsync(produced + filled).ConfigureAwait(false); + stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); + } - int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(filled, _settings.ChunkSize - filled)).ConfigureAwait(false); + int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(filled, _settings.ChunkSize - filled), stallCts!).ConfigureAwait(false); if (read == 0) { if (_remoteInfo.Length is long expected && produced + filled < expected) @@ -340,6 +345,8 @@ private async Task ReadSequentiallyAsync() } catch (Exception e) when (IsRetryable(e)) { + stallCts?.Dispose(); + stallCts = null; response?.Dispose(); response = null; content = null; @@ -364,6 +371,7 @@ private async Task ReadSequentiallyAsync() } finally { + stallCts?.Dispose(); response?.Dispose(); } } @@ -446,15 +454,16 @@ private void Fail(Exception e) pending.Value.TrySetCanceled(); } - private async Task ReadWithStallTimeoutAsync(Stream content, Memory buffer) + private async Task ReadWithStallTimeoutAsync(Stream content, Memory buffer, CancellationTokenSource stallCts) { - using CancellationTokenSource stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); stallCts.CancelAfter(_settings.StallTimeout); try { - return await content.ReadAsync(buffer, stallCts.Token).ConfigureAwait(false); + int read = await content.ReadAsync(buffer, stallCts.Token).ConfigureAwait(false); + stallCts.CancelAfter(Timeout.InfiniteTimeSpan); + return read; } - catch (OperationCanceledException) when (!_cts.IsCancellationRequested) + catch (OperationCanceledException) when (stallCts.IsCancellationRequested && !_cts.IsCancellationRequested) { throw new IOException($"No data received for {_settings.StallTimeout.TotalSeconds}s."); } @@ -462,13 +471,14 @@ private async Task ReadWithStallTimeoutAsync(Stream content, Memory b private async Task SkipWithStallTimeoutAsync(Stream content, long bytesToSkip) { + using CancellationTokenSource stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); byte[] scratch = ArrayPool.Shared.Rent(DrainBufferSize); try { long remaining = bytesToSkip; while (remaining > 0) { - int read = await ReadWithStallTimeoutAsync(content, scratch.AsMemory(0, (int)Math.Min(DrainBufferSize, remaining))).ConfigureAwait(false); + int read = await ReadWithStallTimeoutAsync(content, scratch.AsMemory(0, (int)Math.Min(DrainBufferSize, remaining)), stallCts).ConfigureAwait(false); if (read == 0) throw new EndOfStreamException($"Connection ended while skipping {bytesToSkip} already received bytes."); remaining -= read; From 06f8ac454ab98ded5385d62758b87cd4a9acda18 Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:27:12 +0300 Subject: [PATCH 06/11] style: drop usings made redundant by implicit usings --- .../InitDatabaseSnapshotTests.cs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs index 9bb31974fb01..f487ebfb3e6a 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs @@ -1,14 +1,10 @@ // SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited // SPDX-License-Identifier: LGPL-3.0-only -using System; using System.Formats.Tar; -using System.IO; -using System.Security.Cryptography; using System.IO.Abstractions; using System.Net; -using System.Threading; -using System.Threading.Tasks; +using System.Security.Cryptography; using Nethermind.Api; using Nethermind.Core.Test.IO; using Nethermind.Logging; From a5000a9b61611a229b1bb08dc0d58fb91cd6ffc4 Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:34:29 +0300 Subject: [PATCH 07/11] fix: snapshot streaming review round three Shared StallGuardedReader (TryReset, no per-read CTS, no disarm race) used by streaming, skip, and the two-phase downloader; probe retried with backoff instead of failing startup on one blip --- .../FlakySnapshotServer.cs | 9 ++++ .../StreamingSnapshotInitializerTests.cs | 17 ++++++ .../SnapshotDownloader.cs | 6 ++- .../SnapshotHttpClient.cs | 14 ++--- .../SnapshotHttpStream.cs | 52 ++----------------- .../StallGuardedReader.cs | 34 ++++++++++++ .../StreamingSnapshotInitializer.cs | 25 ++++++++- 7 files changed, 101 insertions(+), 56 deletions(-) create mode 100644 src/Nethermind/Nethermind.Init.Snapshot/StallGuardedReader.cs diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs index ab3ead5565ce..1fb257cf57a5 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs @@ -42,6 +42,8 @@ public FlakySnapshotServer() public int? HangOnceAfterBytes { get; set; } + public int? ServerErrorFirstRequests { get; set; } + public int RequestCount => _requestCount; public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, string? newETag) @@ -94,6 +96,13 @@ private async Task HandleAsync(HttpListenerContext context) return; } + if (ServerErrorFirstRequests is int errorCount && requestNumber <= errorCount) + { + response.StatusCode = 500; + response.Close(); + return; + } + if (etag is not null) response.Headers["ETag"] = etag; string? rangeHeader = context.Request.Headers["Range"]; diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs index 909a64a5bcca..99f13aaf6d68 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs @@ -166,6 +166,23 @@ public async Task InitializeAsync_SourceChangesOnce_RestartsAndCompletesWithNewS Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Completed), "a source change must restart the download and complete with the new object"); } + [Test] + public async Task InitializeAsync_ProbeFailsTransientlyOnce_RetriesAndCompletes() + { + Dictionary files = TestArchive.BuildFiles(); + byte[] archive = TestArchive.BuildTarZst(files); + _server.Content = archive; + _server.ServerErrorFirstRequests = 1; + _config.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None); + + AssertExtracted(files); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Completed), + "a transient probe failure must be retried instead of failing node startup"); + } + [Test] public async Task InitializeAsync_SourceKeepsChanging_GivesUpWithoutThrowing() { diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs index 799af5fdba94..65a74119dffe 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs @@ -17,6 +17,7 @@ internal sealed class SnapshotDownloader(ILogManager logManager) : IDisposable private const int BufferSize = 65536; private const int ResumeWarningDelaySeconds = 5; private static readonly TimeSpan ProgressInterval = TimeSpan.FromSeconds(5); + private static readonly TimeSpan StallTimeout = TimeSpan.FromMinutes(2); // A single client is shared for all retries to preserve the connection pool. private readonly SnapshotHttpClient _client = new(); @@ -73,7 +74,7 @@ public async Task DownloadAsync(string url, string destinationPath, Cancellation progress.Update(initialProgress); if (bytesToSkip > 0) - await SnapshotHttpClient.SkipAsync(contentStream, bytesToSkip, cancellationToken).ConfigureAwait(false); + await SnapshotHttpClient.SkipAsync(contentStream, bytesToSkip, StallTimeout, cancellationToken).ConfigureAwait(false); await CopyWithProgressAsync(contentStream, fileStream, progress, cancellationToken).ConfigureAwait(false); @@ -111,12 +112,13 @@ private static (FileMode fileMode, long bytesToSkip, long? totalSize) ResolveCop private static async Task CopyWithProgressAsync( Stream source, FileStream destination, ProgressReporter progress, CancellationToken cancellationToken) { + using StallGuardedReader reader = new(StallTimeout, cancellationToken); byte[] buffer = ArrayPool.Shared.Rent(BufferSize); try { ulong downloaded = progress.Logger.CurrentValue; int bytesRead; - while ((bytesRead = await source.ReadAsync(buffer, cancellationToken).ConfigureAwait(false)) > 0) + while ((bytesRead = await reader.ReadAsync(source, buffer.AsMemory(0, BufferSize)).ConfigureAwait(false)) > 0) { await destination.WriteAsync(buffer.AsMemory(0, bytesRead), cancellationToken).ConfigureAwait(false); downloaded += (ulong)bytesRead; diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs index ccaddadf9437..18a7f9870be6 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs @@ -55,9 +55,9 @@ public async Task GetAsync( response = await _httpClient.SendAsync( request, HttpCompletionOption.ResponseHeadersRead, headerCts.Token).ConfigureAwait(false); } - catch (OperationCanceledException) when (headerCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) + catch (OperationCanceledException e) when (headerCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) { - throw new HttpRequestException($"No response headers received within {HeaderTimeout.TotalSeconds}s."); + throw new HttpRequestException($"No response headers received within {HeaderTimeout.TotalSeconds}s.", e); } switch (response.StatusCode) @@ -92,17 +92,19 @@ public static bool IsPermanentHttpError(HttpRequestException e) => and not HttpStatusCode.TooManyRequests and not HttpStatusCode.RequestedRangeNotSatisfiable; - public static async Task SkipAsync(Stream content, long bytesToSkip, CancellationToken cancellationToken) + public static async Task SkipAsync(Stream content, long bytesToSkip, TimeSpan stallTimeout, CancellationToken cancellationToken) { + using StallGuardedReader reader = new(stallTimeout, cancellationToken); byte[] buffer = ArrayPool.Shared.Rent(SkipBufferSize); try { long remaining = bytesToSkip; while (remaining > 0) { - int chunk = (int)Math.Min(SkipBufferSize, remaining); - await content.ReadAtLeastAsync(buffer.AsMemory(0, chunk), chunk, throwOnEndOfStream: true, cancellationToken).ConfigureAwait(false); - remaining -= chunk; + int read = await reader.ReadAsync(content, buffer.AsMemory(0, (int)Math.Min(SkipBufferSize, remaining))).ConfigureAwait(false); + if (read == 0) + throw new EndOfStreamException($"Connection ended while skipping {bytesToSkip} already received bytes."); + remaining -= read; } } finally diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs index 8124a0fbf236..ed68df475258 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs @@ -240,6 +240,7 @@ private async Task FetchChunksAsync() private async Task FetchChunkAsync(long offset, byte[] buffer, int length) { + using StallGuardedReader reader = new(_settings.StallTimeout, _cts.Token); TimeSpan retryDelay = _settings.InitialRetryDelay; int rangeRejections = 0; int received = 0; @@ -266,10 +267,9 @@ private async Task FetchChunkAsync(long offset, byte[] buffer, int length) throw new InvalidDataException($"Server returned a range starting at {from}, expected {offset + received}."); await using Stream content = await response.Content.ReadAsStreamAsync(_cts.Token).ConfigureAwait(false); - using CancellationTokenSource stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); while (received < length) { - int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(received, length - received), stallCts).ConfigureAwait(false); + int read = await reader.ReadAsync(content, buffer.AsMemory(received, length - received)).ConfigureAwait(false); if (read == 0) throw new IOException($"Connection ended {received} bytes into a {length} byte chunk at {offset}."); received += read; @@ -300,7 +300,7 @@ private async Task FetchChunkAsync(long offset, byte[] buffer, int length) private async Task ReadSequentiallyAsync() { HttpResponseMessage? response = null; - CancellationTokenSource? stallCts = null; + using StallGuardedReader reader = new(_settings.StallTimeout, _cts.Token); try { long produced = 0; @@ -315,12 +315,9 @@ private async Task ReadSequentiallyAsync() try { if (content is null) - { (response, content) = await ConnectSequentialAsync(produced + filled).ConfigureAwait(false); - stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); - } - int read = await ReadWithStallTimeoutAsync(content, buffer.AsMemory(filled, _settings.ChunkSize - filled), stallCts!).ConfigureAwait(false); + int read = await reader.ReadAsync(content, buffer.AsMemory(filled, _settings.ChunkSize - filled)).ConfigureAwait(false); if (read == 0) { if (_remoteInfo.Length is long expected && produced + filled < expected) @@ -345,8 +342,6 @@ private async Task ReadSequentiallyAsync() } catch (Exception e) when (IsRetryable(e)) { - stallCts?.Dispose(); - stallCts = null; response?.Dispose(); response = null; content = null; @@ -371,7 +366,6 @@ private async Task ReadSequentiallyAsync() } finally { - stallCts?.Dispose(); response?.Dispose(); } } @@ -399,7 +393,7 @@ private async Task ReadSequentiallyAsync() { if (_logger.IsWarn) _logger.Warn($"Server does not support range requests. Re-reading {skip} already received bytes to resume."); - await SkipWithStallTimeoutAsync(content, skip).ConfigureAwait(false); + await SnapshotHttpClient.SkipAsync(content, skip, _settings.StallTimeout, _cts.Token).ConfigureAwait(false); } return (response, content); @@ -454,42 +448,6 @@ private void Fail(Exception e) pending.Value.TrySetCanceled(); } - private async Task ReadWithStallTimeoutAsync(Stream content, Memory buffer, CancellationTokenSource stallCts) - { - stallCts.CancelAfter(_settings.StallTimeout); - try - { - int read = await content.ReadAsync(buffer, stallCts.Token).ConfigureAwait(false); - stallCts.CancelAfter(Timeout.InfiniteTimeSpan); - return read; - } - catch (OperationCanceledException) when (stallCts.IsCancellationRequested && !_cts.IsCancellationRequested) - { - throw new IOException($"No data received for {_settings.StallTimeout.TotalSeconds}s."); - } - } - - private async Task SkipWithStallTimeoutAsync(Stream content, long bytesToSkip) - { - using CancellationTokenSource stallCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); - byte[] scratch = ArrayPool.Shared.Rent(DrainBufferSize); - try - { - long remaining = bytesToSkip; - while (remaining > 0) - { - int read = await ReadWithStallTimeoutAsync(content, scratch.AsMemory(0, (int)Math.Min(DrainBufferSize, remaining)), stallCts).ConfigureAwait(false); - if (read == 0) - throw new EndOfStreamException($"Connection ended while skipping {bytesToSkip} already received bytes."); - remaining -= read; - } - } - finally - { - ArrayPool.Shared.Return(scratch); - } - } - private bool IsRetryable(Exception e) => e is not SnapshotSourceChangedException && !_cts.IsCancellationRequested diff --git a/src/Nethermind/Nethermind.Init.Snapshot/StallGuardedReader.cs b/src/Nethermind/Nethermind.Init.Snapshot/StallGuardedReader.cs new file mode 100644 index 000000000000..ec69349cc0a1 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/StallGuardedReader.cs @@ -0,0 +1,34 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +namespace Nethermind.Init.Snapshot; + +internal sealed class StallGuardedReader(TimeSpan stallTimeout, CancellationToken cancellationToken) : IDisposable +{ + private CancellationTokenSource _stallCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + + public async Task ReadAsync(Stream content, Memory buffer) + { + _stallCts.CancelAfter(stallTimeout); + try + { + int read = await content.ReadAsync(buffer, _stallCts.Token).ConfigureAwait(false); + if (!_stallCts.TryReset()) + Recreate(); + return read; + } + catch (OperationCanceledException e) when (!cancellationToken.IsCancellationRequested) + { + Recreate(); + throw new IOException($"No data received for {stallTimeout.TotalSeconds}s.", e); + } + } + + public void Dispose() => _stallCts.Dispose(); + + private void Recreate() + { + _stallCts.Dispose(); + _stallCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs b/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs index 80f5f9660c7a..c0220e4fe4a5 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs @@ -46,7 +46,7 @@ public async Task InitializeAsync(SnapshotCheckpoint checkpoint, CancellationTok private async Task StreamAndExtractAsync(SnapshotCheckpoint checkpoint, CancellationToken cancellationToken) { using SnapshotHttpClient client = new(); - SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(url, cancellationToken).ConfigureAwait(false); + SnapshotRemoteInfo remoteInfo = await ProbeWithRetryAsync(client, cancellationToken).ConfigureAwait(false); if (remoteInfo.Length is null && config.Checksum is null) throw new InvalidOperationException( "The server does not report the snapshot size and Snapshot.Checksum is not set, so a truncated download could not be detected. Set Snapshot.Checksum or disable Snapshot.Streaming."); @@ -83,6 +83,29 @@ private async Task StreamAndExtractAsync(SnapshotCheckpoint checkpoint, Cancella _logger.Info("Database successfully initialized from streamed snapshot."); } + private async Task ProbeWithRetryAsync(SnapshotHttpClient client, CancellationToken cancellationToken) + { + TimeSpan retryDelay = settings.InitialRetryDelay; + while (true) + { + try + { + return await client.ProbeAsync(url, cancellationToken).ConfigureAwait(false); + } + catch (HttpRequestException e) when (SnapshotHttpClient.IsPermanentHttpError(e)) + { + throw; + } + catch (Exception e) when (e is IOException or HttpRequestException) + { + if (_logger.IsWarn) + _logger.Warn($"Snapshot probe failed. Retrying in {retryDelay.TotalSeconds}s. Error: {e.Message}"); + await Task.Delay(retryDelay, cancellationToken).ConfigureAwait(false); + retryDelay = retryDelay * 2 > settings.MaxRetryDelay ? settings.MaxRetryDelay : retryDelay * 2; + } + } + } + private static void EnsureStreamableArchive(string fileName) { string extension = Path.GetExtension(fileName).ToLowerInvariant(); From 2d339d360349a254ade6bf150e4cacdd5828e95f Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:44:18 +0300 Subject: [PATCH 08/11] style: drop stale using, single source for the stall timeout --- src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs | 1 - src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs | 2 +- src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs | 2 ++ src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs | 2 +- 4 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs index ccf301dac964..da53ff50dd10 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs @@ -3,7 +3,6 @@ using System.Buffers; using System.IO.Abstractions; -using System.Net; using System.Security.Cryptography; using Autofac.Features.AttributeFilters; using Nethermind.Api; diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs index 65a74119dffe..5e0a092584fe 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs @@ -17,7 +17,7 @@ internal sealed class SnapshotDownloader(ILogManager logManager) : IDisposable private const int BufferSize = 65536; private const int ResumeWarningDelaySeconds = 5; private static readonly TimeSpan ProgressInterval = TimeSpan.FromSeconds(5); - private static readonly TimeSpan StallTimeout = TimeSpan.FromMinutes(2); + private static readonly TimeSpan StallTimeout = SnapshotHttpClient.DefaultStallTimeout; // A single client is shared for all retries to preserve the connection pool. private readonly SnapshotHttpClient _client = new(); diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs index 18a7f9870be6..9fa01925c1c3 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs @@ -11,6 +11,8 @@ internal sealed record SnapshotRemoteInfo(long? Length, EntityTagHeaderValue? ET internal sealed class SnapshotHttpClient : IDisposable { + internal static readonly TimeSpan DefaultStallTimeout = TimeSpan.FromMinutes(2); + private const int MaxRedirects = 10; private const int SkipBufferSize = 65536; private static readonly TimeSpan HeaderTimeout = TimeSpan.FromSeconds(100); diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs index ed68df475258..c8a84774db82 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs @@ -15,7 +15,7 @@ namespace Nethermind.Init.Snapshot; internal sealed record SnapshotStreamSettings(int Connections, int ChunkSize, TimeSpan InitialRetryDelay, TimeSpan MaxRetryDelay, TimeSpan StallTimeout) { public static SnapshotStreamSettings Default(int connections) => - new(connections, 64 * 1024 * 1024, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(300), TimeSpan.FromMinutes(2)); + new(connections, 64 * 1024 * 1024, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(300), SnapshotHttpClient.DefaultStallTimeout); } internal sealed class SnapshotHttpStream : Stream From 5f5c887db01786e8f6bf94808745e4973611233f Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:01:32 +0300 Subject: [PATCH 09/11] fix: close three silent empty-database paths An extraction that writes zero files (StripComponents mismatch) now fails instead of completing, a completed checkpoint with a missing or empty database directory reinitializes instead of skipping, and a mounted-but-empty db directory (lost+found only) no longer counts as an existing database --- .../InitDatabaseSnapshotTests.cs | 21 ++++++++++++++++ .../StreamingSnapshotInitializerTests.cs | 16 +++++++++++++ .../TestArchive.cs | 16 +++++++++++++ .../InitDatabaseSnapshot.cs | 24 ++++++++++++++++++- .../SnapshotExtractor.cs | 12 ++++++++++ 5 files changed, 88 insertions(+), 1 deletion(-) diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs index f487ebfb3e6a..0ed264eea22e 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs @@ -148,6 +148,27 @@ public async Task Execute_StreamingEnabledWithDownloadedArchive_ExtractsWithoutS "the archive must be deleted after a successful two-phase extraction"); } + [Test] + public async Task Execute_CompletedCheckpointButEmptyDatabaseDirectory_ReinitializesFromSnapshot() + { + using FlakySnapshotServer server = new(); + Dictionary files = TestArchive.BuildFiles(); + byte[] archive = TestArchive.BuildTarZst(files); + server.Content = archive; + Directory.CreateDirectory(Path.Combine(_dbPath, "lost+found")); + AdvanceCheckpoint(SnapshotStage.Completed); + _snapshotConfig.SnapshotFileName = "snapshot.tar.zst"; + _snapshotConfig.DownloadUrl = server.Url; + _snapshotConfig.Streaming = true; + _snapshotConfig.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + await step.Execute(CancellationToken.None); + + Assert.That(File.Exists(Path.Combine(_dbPath, "state/a42.sst")), Is.True, + "a stale completed checkpoint with a missing or empty database must reinitialize instead of skipping and leaving the node to sync from genesis"); + } + [TestCase(1_000, 0, 2_500, TestName = "FreshDownload")] [TestCase(1_000, 400, 2_100, TestName = "ResumedDownload")] public void GetRequiredSpaceForDownload_ForGivenSizes_AddsRemainingBytesToExtractionEstimate( diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs index 99f13aaf6d68..ad8d48cb6951 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs @@ -135,6 +135,22 @@ public void InitializeAsync_UnknownLengthWithoutChecksum_Throws() "without a length and without a checksum a truncated download would be undetectable, so streaming must refuse to start"); } + [Test] + public async Task InitializeAsync_ArchiveNotMatchingStripComponents_DeletesDatabaseWithoutThrowing() + { + byte[] archive = TestArchive.BuildTarZstWithoutTopLevelDirectory(); + _server.Content = archive; + _config.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + await CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None); + + Assert.That(Directory.Exists(_dbPath), Is.False, + "an extraction that produced no files must be treated as a failure, not silently completed"); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Started), + "the checkpoint must not advance when nothing was extracted"); + } + [Test] public void InitializeAsync_ZipArchiveConfigured_Throws() { diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs index e30bc6d31c12..46a1e71f1b21 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs @@ -36,6 +36,22 @@ public static byte[] BuildTarZst(IReadOnlyDictionary files) return compressed.ToArray(); } + public static byte[] BuildTarZstWithoutTopLevelDirectory() + { + using MemoryStream tarBuffer = new(); + using (TarWriter writer = new(tarBuffer, leaveOpen: true)) + { + PaxTarEntry entry = new(TarEntryType.RegularFile, "state.bin") { DataStream = new MemoryStream(new byte[10_000]) }; + writer.WriteEntry(entry); + } + + tarBuffer.Position = 0; + using MemoryStream compressed = new(); + using (CompressionStream zstd = new(compressed, leaveOpen: true)) + tarBuffer.CopyTo(zstd); + return compressed.ToArray(); + } + public static byte[] BuildTarZstWithSymlink() { using MemoryStream tarBuffer = new(); diff --git a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs index da53ff50dd10..f7affa1bd597 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs @@ -63,7 +63,7 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) SnapshotCheckpoint checkpoint = new(snapshotConfig, api.LogManager); - if (Path.Exists(dbPath)) + if (DatabaseExists(dbPath)) { if (checkpoint.Read() < SnapshotStage.Extracted) { @@ -79,6 +79,12 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) return; } } + else if (checkpoint.Read() >= SnapshotStage.Extracted) + { + if (_logger.IsWarn) + _logger.Warn($"The snapshot checkpoint indicates a completed extraction, but the database at {dbPath} is missing or empty. Reinitializing from the snapshot."); + checkpoint.Advance(SnapshotStage.Started); + } Directory.CreateDirectory(snapshotConfig.SnapshotDirectory); @@ -163,6 +169,22 @@ private async Task DownloadWithRetryAsync( checkpoint.Advance(SnapshotStage.Downloaded); } + private static bool DatabaseExists(string dbPath) + { + if (File.Exists(dbPath)) + return true; + if (!Directory.Exists(dbPath)) + return false; + + foreach (string entry in Directory.EnumerateFileSystemEntries(dbPath)) + { + if (Path.GetFileName(entry) != "lost+found") + return true; + } + + return false; + } + private long GetFileSize(string path) { IFileInfo file = api.FileSystem.FileInfo.New(path); diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs index 30306295904b..c92a1e87a0c3 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs @@ -85,6 +85,7 @@ private static void ExtractTarEntries(Stream decompressedStream, string destinat using TarReader tarReader = new(decompressedStream, leaveOpen: true); string destinationRoot = destinationPath.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; + int extractedFiles = 0; TarEntry? entry; while ((entry = tarReader.GetNextEntry()) is not null) @@ -104,12 +105,23 @@ private static void ExtractTarEntries(Stream decompressedStream, string destinat throw new IOException($"Tar entry '{entry.Name}' would extract outside the destination directory."); if (entry.EntryType is TarEntryType.Directory) + { Directory.CreateDirectory(destinationEntryPath); + } else if (entry.EntryType is TarEntryType.RegularFile or TarEntryType.V7RegularFile) + { entry.ExtractToFile(destinationEntryPath, overwrite: true); + extractedFiles++; + } else + { throw new IOException($"Tar entry '{entry.Name}' has unsupported type {entry.EntryType}."); + } } + + if (extractedFiles == 0) + throw new IOException( + $"The archive produced no files under '{destinationPath}'. Check Snapshot.StripComponents against the archive layout."); } private static Stream OpenDecompressedStream(Stream archiveStream, string extension, bool leaveOpen) => From d6c6497e49ef121d2c3258bc102f3cfa1739e2d5 Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:18:39 +0300 Subject: [PATCH 10/11] fix: release stream buffers eagerly on dispose --- src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs index c8a84774db82..4b438e82826c 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs @@ -160,6 +160,9 @@ private async Task DisposeCoreAsync() _hasher.Dispose(); _cts.Dispose(); _window.Dispose(); + _buffers.Clear(); + _current = default; + _pending.Clear(); } private bool TryTakeNextChunk() From ecf86f4828c9908010ad7f753db525bcb5cb6c25 Mon Sep 17 00:00:00 2001 From: Stavros Vlachakis <89769224+svlachakis@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:36:08 +0300 Subject: [PATCH 11/11] fix: address review round five Checkpoint rewind keeps a complete archive (Downloaded, not Started), the empty-extraction guard covers zip and tar in one place using the database-exists predicate, and DatabaseExists tolerates unreadable directories --- .../InitDatabaseSnapshotTests.cs | 37 +++++++++++++++++++ .../InitDatabaseSnapshot.cs | 20 ++++++---- .../SnapshotExtractor.cs | 23 ++++++------ 3 files changed, 61 insertions(+), 19 deletions(-) diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs index 0ed264eea22e..8e79a0ff1f5b 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs @@ -169,6 +169,43 @@ public async Task Execute_CompletedCheckpointButEmptyDatabaseDirectory_Reinitial "a stale completed checkpoint with a missing or empty database must reinitialize instead of skipping and leaving the node to sync from genesis"); } + [Test] + public void Execute_ArchiveNotMatchingStripComponents_Throws() + { + using (FileStream fileStream = File.Create(_snapshotPath)) + using (TarWriter tarWriter = new(fileStream)) + { + PaxTarEntry fileEntry = new(TarEntryType.RegularFile, "state.bin") + { + DataStream = new MemoryStream(new byte[1000]) + }; + tarWriter.WriteEntry(fileEntry); + } + AdvanceCheckpoint(SnapshotStage.Verified); + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + IOException exception = Assert.ThrowsAsync(() => step.Execute(CancellationToken.None))!; + + Assert.That(exception.Message, Does.Contain("StripComponents"), + "an extraction that produced no files must point the operator at the strip configuration"); + } + + [Test] + public async Task Execute_ExtractedCheckpointWithArchivePresentButNoDatabase_ReusesArchive() + { + WriteSnapshotTar(); + AdvanceCheckpoint(SnapshotStage.Extracted); + _snapshotConfig.Streaming = true; + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + await step.Execute(CancellationToken.None); + + Assert.That(File.Exists(Path.Combine(_dbPath, "state.bin")), Is.True, + "a complete archive left by a crash between extraction and completion must be reused instead of re-downloaded"); + Assert.That(File.Exists(_snapshotPath), Is.False, + "the archive must be deleted after the successful extraction"); + } + [TestCase(1_000, 0, 2_500, TestName = "FreshDownload")] [TestCase(1_000, 400, 2_100, TestName = "ResumedDownload")] public void GetRequiredSpaceForDownload_ForGivenSizes_AddsRemainingBytesToExtractionEstimate( diff --git a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs index d3270573523c..ce987903f0c1 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs @@ -62,6 +62,7 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) throw new InvalidOperationException($"Snapshot.StreamingConnections must be between 1 and {MaxStreamingConnections}, got {snapshotConfig.StreamingConnections}."); SnapshotCheckpoint checkpoint = new(snapshotConfig, api.LogManager); + Directory.CreateDirectory(snapshotConfig.SnapshotDirectory); if (DatabaseExists(dbPath)) { @@ -83,11 +84,9 @@ private async Task InitDbFromSnapshotAsync(CancellationToken cancellationToken) { if (_logger.IsWarn) _logger.Warn($"The snapshot checkpoint indicates a completed extraction, but the database at {dbPath} is missing or empty. Reinitializing from the snapshot."); - checkpoint.Advance(SnapshotStage.Started); + checkpoint.Advance(File.Exists(snapshotPath) ? SnapshotStage.Downloaded : SnapshotStage.Started); } - Directory.CreateDirectory(snapshotConfig.SnapshotDirectory); - if (snapshotConfig.Streaming) { if (checkpoint.Read() < SnapshotStage.Downloaded || !File.Exists(snapshotPath)) @@ -169,17 +168,24 @@ private async Task DownloadWithRetryAsync( checkpoint.Advance(SnapshotStage.Downloaded); } - private static bool DatabaseExists(string dbPath) + internal static bool DatabaseExists(string dbPath) { if (File.Exists(dbPath)) return true; if (!Directory.Exists(dbPath)) return false; - foreach (string entry in Directory.EnumerateFileSystemEntries(dbPath)) + try { - if (Path.GetFileName(entry) != "lost+found") - return true; + foreach (string entry in Directory.EnumerateFileSystemEntries(dbPath)) + { + if (Path.GetFileName(entry) != "lost+found") + return true; + } + } + catch (Exception e) when (e is UnauthorizedAccessException or IOException) + { + return true; } return false; diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs index c92a1e87a0c3..237cbb6cb4ff 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs @@ -39,6 +39,8 @@ public Task ExtractTarStreamAsync(Stream archiveStream, string destinationPath, if (!ReferenceEquals(decompressedStream, archiveStream)) decompressedStream.Dispose(); } + + EnsureNotEmpty(destinationPath); }, cancellationToken); private void Extract(string archivePath, string destinationPath, int stripComponents, CancellationToken cancellationToken) @@ -55,6 +57,15 @@ private void Extract(string archivePath, string destinationPath, int stripCompon ExtractTar(archivePath, destinationPath, extension, stripComponents, cancellationToken); else throw new NotSupportedException($"Unsupported snapshot archive format: {archivePath}"); + + EnsureNotEmpty(destinationPath); + } + + private static void EnsureNotEmpty(string destinationPath) + { + if (!InitDatabaseSnapshot.DatabaseExists(destinationPath)) + throw new IOException( + $"The archive produced no files under '{destinationPath}'. Check Snapshot.StripComponents against the archive layout."); } private static bool IsZip(string extension) => @@ -85,7 +96,6 @@ private static void ExtractTarEntries(Stream decompressedStream, string destinat using TarReader tarReader = new(decompressedStream, leaveOpen: true); string destinationRoot = destinationPath.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; - int extractedFiles = 0; TarEntry? entry; while ((entry = tarReader.GetNextEntry()) is not null) @@ -105,23 +115,12 @@ private static void ExtractTarEntries(Stream decompressedStream, string destinat throw new IOException($"Tar entry '{entry.Name}' would extract outside the destination directory."); if (entry.EntryType is TarEntryType.Directory) - { Directory.CreateDirectory(destinationEntryPath); - } else if (entry.EntryType is TarEntryType.RegularFile or TarEntryType.V7RegularFile) - { entry.ExtractToFile(destinationEntryPath, overwrite: true); - extractedFiles++; - } else - { throw new IOException($"Tar entry '{entry.Name}' has unsupported type {entry.EntryType}."); - } } - - if (extractedFiles == 0) - throw new IOException( - $"The archive produced no files under '{destinationPath}'. Check Snapshot.StripComponents against the archive layout."); } private static Stream OpenDecompressedStream(Stream archiveStream, string extension, bool leaveOpen) =>