diff --git a/.github/workflows/nethermind-tests.yml b/.github/workflows/nethermind-tests.yml index 3648164d032..872cd770175 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 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 00000000000..9219ed7fa92 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs @@ -0,0 +1,215 @@ +// 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 readonly CancellationTokenSource _hangCts = new(); + private int _requestCount; + private int _hangConsumed; + private int _rangeIgnored; + private int _redirected; + private int _switchAfterRequests = int.MaxValue; + private byte[] _newContent = []; + private string? _newETag; + + public FlakySnapshotServer() + { + (_listener, int port) = TestHttpListener.Start(); + 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 bool OmitContentLength { get; set; } + + public bool RotateETagEveryRequest { get; set; } + + public int? HangOnceAfterBytes { get; set; } + + public int? ServerErrorFirstRequests { get; set; } + + public bool IgnoreRangeOnce { get; set; } + + public bool RedirectFirstRequest { get; set; } + + public int RequestCount => _requestCount; + + public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, string? newETag) + { + _newContent = newContent; + _newETag = newETag; + _switchAfterRequests = requestCount; + } + + public void Dispose() + { + _hangCts.Cancel(); + _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; + if (RotateETagEveryRequest) + etag = $"\"v{requestNumber}\""; + HttpListenerResponse response = context.Response; + + try + { + if (RedirectFirstRequest && Interlocked.Exchange(ref _redirected, 1) == 0) + { + response.StatusCode = 302; + response.Headers["Location"] = Url; + response.Close(); + return; + } + + if (FailWithNotFoundAfterRequests is int failAfter && requestNumber > failAfter) + { + response.StatusCode = 404; + response.Close(); + 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"]; + 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 && IgnoreRangeOnce && rangeHeader != "bytes=0-0" && Interlocked.Exchange(ref _rangeIgnored, 1) == 0) + ranged = false; + + 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; + if (OmitContentLength) + response.SendChunked = true; + 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(Timeout.InfiniteTimeSpan, _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) + { + 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 new file mode 100644 index 00000000000..bfd841cd708 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/InitDatabaseSnapshotTests.cs @@ -0,0 +1,306 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.IO.Abstractions; +using System.Net; +using System.Security.Cryptography; +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(SnapshotDiskSpace.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"); + } + + [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(archive.Length * 2L)); + + 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"); + } + + [TestCase(0, TestName = "Zero")] + [TestCase(17, TestName = "AboveMaximum")] + public void Execute_StreamingConnectionsOutOfRange_Throws(int connections) + { + _snapshotConfig.Streaming = true; + _snapshotConfig.StreamingConnections = connections; + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(60)); + + Assert.ThrowsAsync(() => step.Execute(timeout.Token), + "a connection count outside 1-16 must be rejected before any network activity"); + } + + [Test] + public async Task Execute_VerifiedCheckpointWithoutArchive_StreamsFromScratch() + { + using FlakySnapshotServer server = new(); + Dictionary files = TestArchive.BuildFiles(); + byte[] archive = TestArchive.BuildTarZst(files); + server.Content = archive; + _snapshotConfig.SnapshotFileName = "snapshot.tar.zst"; + AdvanceCheckpoint(SnapshotStage.Verified); + _snapshotConfig.DownloadUrl = server.Url; + _snapshotConfig.Streaming = true; + _snapshotConfig.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(archive.Length * 2L)); + + await step.Execute(CancellationToken.None); + + Assert.That(File.Exists(Path.Combine(_dbPath, "state/a42.sst")), Is.True, + "a checkpoint past Downloaded with no archive on disk must stream instead of skipping the download stage"); + } + + [TestCase(nameof(SnapshotStage.Verified), TestName = "Verified")] + [TestCase(nameof(SnapshotStage.Extracted), TestName = "Extracted")] + [TestCase(nameof(SnapshotStage.Completed), TestName = "Completed")] + public async Task Execute_CheckpointPastDownloadWithoutArchiveOrDatabase_RedownloadsAndExtracts(string stage) + { + using FlakySnapshotServer server = new(); + server.Content = TestArchive.BuildTar(); + AdvanceCheckpoint(Enum.Parse(stage)); + _snapshotConfig.DownloadUrl = server.Url; + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + await step.Execute(CancellationToken.None); + + Assert.That(File.Exists(Path.Combine(_dbPath, "state.bin")), Is.True, + "a checkpoint past the download with neither archive nor database must restart the download instead of failing startup forever"); + } + + [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"); + } + + [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")); + _snapshotConfig.SnapshotFileName = "snapshot.tar.zst"; + AdvanceCheckpoint(SnapshotStage.Completed); + _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"); + } + + [Test] + public void Execute_ArchiveNotMatchingStripComponents_Throws() + { + File.WriteAllBytes(_snapshotPath, TestArchive.BuildTarWithoutTopLevelDirectory()); + AdvanceCheckpoint(SnapshotStage.Verified); + InitDatabaseSnapshot step = new(_api, DrivesWithFreeSpace(long.MaxValue)); + + InvalidOperationException 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( + long totalSize, long existingSize, long expected) => + Assert.That(SnapshotDiskSpace.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() + { + File.WriteAllBytes(_snapshotPath, TestArchive.BuildTar(SnapshotPayloadSize)); + 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) = TestHttpListener.Start(); + + _ = 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"); + } + + 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 00000000000..f74e0e631da --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/Nethermind.Init.Snapshot.Test.csproj @@ -0,0 +1,19 @@ + + + + + + enable + enable + + + + + + + + + + + + diff --git a/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotDownloaderTests.cs b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotDownloaderTests.cs new file mode 100644 index 00000000000..ac34eebc44e --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotDownloaderTests.cs @@ -0,0 +1,94 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using Nethermind.Core.Test.IO; +using Nethermind.Logging; +using NUnit.Framework; + +namespace Nethermind.Init.Snapshot.Test; + +[TestFixture] +public class SnapshotDownloaderTests +{ + private FlakySnapshotServer _server = null!; + private TempPath _tempDir = null!; + private string _destinationPath = null!; + + [SetUp] + public void SetUp() + { + _server = new FlakySnapshotServer(); + _tempDir = TempPath.GetTempDirectory(); + Directory.CreateDirectory(_tempDir.Path); + _destinationPath = Path.Combine(_tempDir.Path, "snapshot.tar.zst"); + } + + [TearDown] + public void TearDown() + { + _server.Dispose(); + _tempDir.Dispose(); + } + + [Test] + public async Task DownloadAsync_NoFileOnDisk_DownloadsExactContent() + { + byte[] content = BuildContent(); + _server.Content = content; + using SnapshotDownloader downloader = new(LimboLogs.Instance); + + await downloader.DownloadAsync(_server.Url, _destinationPath, CancellationToken.None); + + Assert.That(File.ReadAllBytes(_destinationPath), Is.EqualTo(content), + "a fresh download must write the exact remote bytes"); + } + + [Test] + public async Task DownloadAsync_PartialFileOnDisk_ResumesWithoutCorruption() + { + byte[] content = BuildContent(); + _server.Content = content; + File.WriteAllBytes(_destinationPath, content[..30_000]); + using SnapshotDownloader downloader = new(LimboLogs.Instance); + + await downloader.DownloadAsync(_server.Url, _destinationPath, CancellationToken.None); + + Assert.That(File.ReadAllBytes(_destinationPath), Is.EqualTo(content), + "a resumed download must append exactly the missing suffix, not replay the full body"); + } + + [Test] + public async Task DownloadAsync_FileAlreadyComplete_LeavesItUntouched() + { + byte[] content = BuildContent(); + _server.Content = content; + File.WriteAllBytes(_destinationPath, content); + using SnapshotDownloader downloader = new(LimboLogs.Instance); + + await downloader.DownloadAsync(_server.Url, _destinationPath, CancellationToken.None); + + Assert.That(File.ReadAllBytes(_destinationPath), Is.EqualTo(content), + "a complete file must be recognized via 416 and left as is"); + } + + [Test] + public async Task DownloadAsync_ServerRedirects_FollowsAndDownloads() + { + byte[] content = BuildContent(); + _server.Content = content; + _server.RedirectFirstRequest = true; + using SnapshotDownloader downloader = new(LimboLogs.Instance); + + await downloader.DownloadAsync(_server.Url, _destinationPath, CancellationToken.None); + + Assert.That(File.ReadAllBytes(_destinationPath), Is.EqualTo(content), + "a redirect must be followed manually so the range header survives"); + } + + private static byte[] BuildContent() + { + byte[] content = new byte[100_000]; + new Random(42).NextBytes(content); + return content; + } +} 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 00000000000..d2fa63eada9 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/SnapshotHttpStreamTests.cs @@ -0,0 +1,176 @@ +// 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; + using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(60)); + + Assert.ThrowsAsync( + () => DownloadAsync(connections: 2, cancellationToken: timeout.Token), + "a permanent HTTP error must abort the stream instead of retrying forever"); + } + + [Test] + public async Task Read_ServerIgnoresOneRangeRequest_RetriesAndDeliversExactContent() + { + byte[] content = BuildContent(100_000); + _server.Content = content; + _server.IgnoreRangeOnce = true; + + (byte[] delivered, byte[] hash) = await DownloadAsync(connections: 2); + + Assert.That(delivered, Is.EqualTo(content), "a single 200 answer to a range request must be retried instead of corrupting the stream"); + Assert.That(hash, Is.EqualTo(SHA256.HashData(content)), "the retried chunk must be hashed exactly once"); + } + + [Test] + public async Task Dispose_AfterFullDownload_ReleasesPooledBuffers() + { + byte[] content = BuildContent(100_000); + _server.Content = content; + SnapshotStreamSettings settings = new(3, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50), TimeSpan.FromSeconds(30), ComputeChecksum: true); + using SnapshotHttpClient client = new(); + SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(_server.Url, CancellationToken.None); + SnapshotHttpStream stream = new(client, _server.Url, remoteInfo, settings, LimboLogs.Instance, CancellationToken.None); + using MemoryStream delivered = new(); + await Task.Run(() => stream.CopyTo(delivered)); + + await stream.DisposeAsync(); + + Assert.That(stream.PooledBufferCount, Is.EqualTo(0), + "disposing the stream must release the pooled chunk buffers instead of keeping them reachable"); + } + + [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, CancellationToken cancellationToken = default) + { + SnapshotStreamSettings settings = new(connections, TestChunkSize, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50), stallTimeout ?? TimeSpan.FromSeconds(30), ComputeChecksum: true); + using SnapshotHttpClient client = new(); + SnapshotRemoteInfo remoteInfo = await client.ProbeAsync(_server.Url, cancellationToken); + await using SnapshotHttpStream stream = new(client, _server.Url, remoteInfo, settings, LimboLogs.Instance, cancellationToken); + using MemoryStream delivered = new(); + await Task.Run(() => stream.CopyTo(delivered)); + byte[] hash = (await stream.FinishAsync(cancellationToken))!; + 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 00000000000..4b74a7e25b9 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/StreamingSnapshotInitializerTests.cs @@ -0,0 +1,252 @@ +// 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 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_ArchiveNotMatchingStripComponents_DeletesDatabaseAndThrows() + { + byte[] archive = TestArchive.BuildTarZstWithoutTopLevelDirectory(); + _server.Content = archive; + _config.Checksum = Convert.ToHexString(SHA256.HashData(archive)); + SnapshotCheckpoint checkpoint = CreateCheckpoint(); + + InvalidOperationException exception = Assert.ThrowsAsync( + () => CreateInitializer().InitializeAsync(checkpoint, CancellationToken.None), + "an extraction that produced no files is a configuration error and must fail startup in both modes")!; + + Assert.That(exception.Message, Does.Contain("StripComponents"), + "the failure must point the operator at the strip configuration"); + Assert.That(SnapshotDatabase.Exists(_dbPath), Is.False, + "the partially created database directory must be cleaned up before failing"); + Assert.That(checkpoint.Read(), Is.EqualTo(SnapshotStage.Started), + "the checkpoint must not advance when nothing was extracted"); + } + + [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 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() + { + _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() + { + _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), TimeSpan.FromSeconds(30)), + 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 00000000000..4c68bf7e22d --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot.Test/TestArchive.cs @@ -0,0 +1,111 @@ +// 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 PaxGlobalExtendedAttributesTarEntry(new Dictionary { ["comment"] = "test" })); + 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); + } + } + + return Compress(tarBuffer); + } + + public static byte[] BuildTar(int payloadSize = 1000) + { + using MemoryStream tarBuffer = new(); + using (TarWriter writer = new(tarBuffer, leaveOpen: true)) + { + writer.WriteEntry(new PaxTarEntry(TarEntryType.Directory, "data")); + writer.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, "data/state.bin") + { + DataStream = new MemoryStream(new byte[payloadSize]) + }); + } + + return tarBuffer.ToArray(); + } + + public static byte[] BuildTarWithoutTopLevelDirectory() + { + using MemoryStream tarBuffer = new(); + using (TarWriter writer = new(tarBuffer, leaveOpen: true)) + { + writer.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, "state.bin") + { + DataStream = new MemoryStream(new byte[1000]) + }); + } + + return tarBuffer.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); + } + + return Compress(tarBuffer); + } + + 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" }); + } + + return Compress(tarBuffer); + } + + 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, + }; + } + + private static byte[] Compress(MemoryStream tarBuffer) + { + tarBuffer.Position = 0; + using MemoryStream compressed = new(); + using (CompressionStream zstd = new(compressed, leaveOpen: true)) + tarBuffer.CopyTo(zstd); + return compressed.ToArray(); + } +} 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 00000000000..e4f3d212128 --- /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 dd7c4985021..6ab2a451d5e 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 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. Peak buffer memory is (connections + 1) x 64 MiB. 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 9c9962f9b18..ab795061361 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/InitDatabaseSnapshot.cs @@ -3,12 +3,10 @@ using System.Buffers; using System.IO.Abstractions; -using System.Net; using System.Security.Cryptography; using Autofac.Features.AttributeFilters; using Nethermind.Api; using Nethermind.Api.Steps; -using Nethermind.Core.Extensions; using Nethermind.Init.Steps; using Nethermind.Logging; @@ -26,6 +24,7 @@ public class InitDatabaseSnapshot( INethermindApi api, [KeyFilter(nameof(IInitConfig.BaseDbPath))] IDriveInfo[] drives) : IStep { + private const int MaxStreamingConnections = 16; private const int ExtractionRestartDelaySeconds = 5; private const int InitialRetryDelaySeconds = 5; private const int MaxRetryDelaySeconds = 300; @@ -57,16 +56,20 @@ 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 is < 1 or > MaxStreamingConnections) + 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 (Path.Exists(dbPath)) + if (SnapshotDatabase.Exists(dbPath)) { if (checkpoint.Read() < SnapshotStage.Extracted) { if (_logger.IsInfo) _logger.Info("Extraction did not complete last time. Restarting. To interrupt press Ctrl^C"); await Task.Delay(TimeSpan.FromSeconds(ExtractionRestartDelaySeconds), cancellationToken).ConfigureAwait(false); - Directory.Delete(dbPath, true); + SnapshotDatabase.Delete(dbPath); } else { @@ -75,8 +78,34 @@ 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(File.Exists(snapshotPath) ? SnapshotStage.Downloaded : SnapshotStage.Started); + } - Directory.CreateDirectory(snapshotConfig.SnapshotDirectory); + if (checkpoint.Read() >= SnapshotStage.Downloaded && !File.Exists(snapshotPath)) + { + if (_logger.IsWarn) + _logger.Warn($"The snapshot checkpoint indicates a completed download, but no archive exists at {snapshotPath}. Restarting the download."); + checkpoint.Advance(SnapshotStage.Started); + } + + if (snapshotConfig.Streaming) + { + if (checkpoint.Read() < SnapshotStage.Downloaded) + { + 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); await DownloadWithRetryAsync(downloader, snapshotUrl, snapshotPath, checkpoint, cancellationToken).ConfigureAwait(false); @@ -109,6 +138,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); @@ -119,10 +150,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."); @@ -157,28 +185,18 @@ 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 (!SnapshotChecksum.Verify(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); @@ -191,28 +209,37 @@ private async Task ExtractAsync( if (checkpoint.Read() >= SnapshotStage.Extracted) return; - CheckDiskSpace(snapshotPath); + SnapshotDiskSpace.Check(drives, SnapshotDiskSpace.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.ProbeAsync(url, cancellationToken).ConfigureAwait(false)).Length; + } + 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; + } - long snapshotSize = api.FileSystem.FileInfo.New(snapshotPath).Length; - long required = (long)(snapshotSize * 2.5); - - foreach (IDriveInfo drive in drives) + if (totalSize is null) { - if (drive.AvailableFreeSpace < required) - throw new IOException( - $"Insufficient disk space on '{drive.RootDirectory.FullName}' to extract snapshot: " + - $"need at least {required} bytes, {drive.AvailableFreeSpace} available."); + if (_logger.IsWarn) + _logger.Warn("The server did not report the snapshot size. Skipping the pre-download disk space check."); + return; } + + SnapshotDiskSpace.Check(drives, SnapshotDiskSpace.GetRequiredSpaceForDownload(totalSize.Value, existingSize), "download and extract"); } private async Task ComputeChecksumAsync(string filePath, CancellationToken cancellationToken) @@ -220,10 +247,10 @@ private async Task ComputeChecksumAsync(string filePath, CancellationTok long fileSize = new FileInfo(filePath).Length; using IncrementalHash hasher = IncrementalHash.CreateHash(HashAlgorithmName.SHA256); byte[] buffer = ArrayPool.Shared.Rent(ChecksumBufferSize); - byte[] checksum; - await using (FileStream fileStream = new(filePath, FileMode.Open, FileAccess.Read, - FileShare.None, bufferSize: 1, FileOptions.Asynchronous | FileOptions.SequentialScan)) + try { + await using FileStream fileStream = new(filePath, FileMode.Open, FileAccess.Read, + FileShare.None, bufferSize: 1, FileOptions.Asynchronous | FileOptions.SequentialScan); long bytesHashed = 0; DateTime nextLog = DateTime.UtcNow.AddSeconds(ChecksumProgressIntervalSeconds); @@ -239,9 +266,12 @@ private async Task ComputeChecksumAsync(string filePath, CancellationTok nextLog = DateTime.UtcNow.AddSeconds(ChecksumProgressIntervalSeconds); } } - checksum = hasher.GetHashAndReset(); + + return hasher.GetHashAndReset(); + } + finally + { + ArrayPool.Shared.Return(buffer); } - ArrayPool.Shared.Return(buffer); - return checksum; } } diff --git a/src/Nethermind/Nethermind.Init.Snapshot/Nethermind.Init.Snapshot.csproj b/src/Nethermind/Nethermind.Init.Snapshot/Nethermind.Init.Snapshot.csproj index e9f2b426edf..7e06814b574 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/SnapshotArchiveFormat.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotArchiveFormat.cs new file mode 100644 index 00000000000..04747b2f153 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotArchiveFormat.cs @@ -0,0 +1,16 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +namespace Nethermind.Init.Snapshot; + +internal static class SnapshotArchiveFormat +{ + public static bool IsZip(string extension) => + extension is ".zip"; + + public static bool IsTarBased(string extension, string innerExtension) => + extension is ".tar" or ".zst" or ".zstd" or ".gz" or ".bz2" or ".xz" || innerExtension == ".tar"; + + public static bool IsStreamable(string extension) => + extension is ".tar" or ".zst" or ".zstd" or ".gz"; +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotChecksum.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotChecksum.cs new file mode 100644 index 00000000000..f113773bc61 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotChecksum.cs @@ -0,0 +1,26 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using Nethermind.Core.Extensions; +using Nethermind.Logging; + +namespace Nethermind.Init.Snapshot; + +internal static class SnapshotChecksum +{ + public static bool Verify(byte[] actual, string expectedHex, string onMismatch, ILogger logger) + { + 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; + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotConfig.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotConfig.cs index 01d1ca6688c..7ce8a55c3be 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/SnapshotDatabase.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDatabase.cs new file mode 100644 index 00000000000..fae86d8ae91 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDatabase.cs @@ -0,0 +1,64 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +namespace Nethermind.Init.Snapshot; + +internal static class SnapshotDatabase +{ + private const string MountArtifact = "lost+found"; + + public static bool Exists(string dbPath) + { + if (File.Exists(dbPath)) + return true; + if (!Directory.Exists(dbPath)) + return false; + + try + { + foreach (string entry in Directory.EnumerateFileSystemEntries(dbPath)) + { + if (Path.GetFileName(entry) != MountArtifact) + return true; + } + } + catch (Exception e) when (e is UnauthorizedAccessException or IOException) + { + return true; + } + + return false; + } + + public static void Delete(string dbPath) + { + if (!Directory.Exists(dbPath)) + return; + + foreach (string entry in Directory.GetFileSystemEntries(dbPath)) + { + if (Path.GetFileName(entry) == MountArtifact) + continue; + + if (Directory.Exists(entry)) + Directory.Delete(entry, true); + else + File.Delete(entry); + } + + TryRemoveDirectoryItself(dbPath); + } + + private static bool TryRemoveDirectoryItself(string dbPath) + { + try + { + Directory.Delete(dbPath); + return true; + } + catch (Exception e) when (e is IOException or UnauthorizedAccessException) + { + return false; + } + } +} diff --git a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDiskSpace.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDiskSpace.cs new file mode 100644 index 00000000000..063d09139f1 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDiskSpace.cs @@ -0,0 +1,28 @@ +// SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited +// SPDX-License-Identifier: LGPL-3.0-only + +using System.IO.Abstractions; + +namespace Nethermind.Init.Snapshot; + +internal static class SnapshotDiskSpace +{ + private const double ExtractionSpaceMultiplier = 1.5; + + public static long GetRequiredSpaceForDownload(long totalSize, long existingSize) => + totalSize - existingSize + GetRequiredSpaceForExtraction(totalSize); + + public static long GetRequiredSpaceForExtraction(long snapshotSize) => + (long)(snapshotSize * ExtractionSpaceMultiplier); + + public static void Check(IDriveInfo[] drives, long required, string operation) + { + foreach (IDriveInfo drive in drives) + { + if (drive.AvailableFreeSpace < required) + throw new IOException( + $"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/SnapshotDownloader.cs b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotDownloader.cs index 535c4f728b6..564478b62de 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); + private static readonly TimeSpan StallTimeout = SnapshotHttpClient.DefaultStallTimeout; - // A single HttpClient is shared for all retries to preserve the connection pool. - private readonly HttpClient _httpClient = new(new HttpClientHandler { AllowAutoRedirect = false }); + 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, StallTimeout, cancellationToken).ConfigureAwait(false); await CopyWithProgressAsync(contentStream, fileStream, progress, cancellationToken).ConfigureAwait(false); @@ -83,7 +81,10 @@ public async Task DownloadAsync(string url, string destinationPath, Cancellation _logger.Info($"Snapshot downloaded to {destinationPath}."); } - public void Dispose() => _httpClient.Dispose(); + public Task ProbeAsync(string url, CancellationToken cancellationToken) => + _client.ProbeAsync(url, cancellationToken); + + public void Dispose() => _client.Dispose(); private static (FileMode fileMode, long bytesToSkip, long? totalSize) ResolveCopyStrategy( HttpStatusCode statusCode, long existingSize, long? contentLength) => @@ -99,74 +100,16 @@ 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) { + 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; @@ -179,17 +122,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 b1e68619fce..0b5b1a17e5c 100644 --- a/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotExtractor.cs @@ -21,7 +21,29 @@ internal sealed class SnapshotExtractor(ILogManager logManager) /// are stripped (equivalent to tar --strip-components). /// public Task ExtractAsync(string archivePath, string destinationPath, int stripComponents, CancellationToken cancellationToken) => - Task.Run(() => Extract(archivePath, destinationPath, stripComponents, cancellationToken), cancellationToken); + Task.Factory.StartNew( + () => Extract(archivePath, destinationPath, stripComponents, cancellationToken), + cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + public Task ExtractTarStreamAsync(Stream archiveStream, string destinationPath, string extension, int stripComponents, CancellationToken cancellationToken) => + Task.Factory.StartNew(() => + { + 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(); + } + + EnsureNotEmpty(destinationPath); + }, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default); private void Extract(string archivePath, string destinationPath, int stripComponents, CancellationToken cancellationToken) { @@ -31,19 +53,22 @@ private void Extract(string archivePath, string destinationPath, int stripCompon string extension = Path.GetExtension(archivePath).ToLowerInvariant(); string innerExtension = Path.GetExtension(Path.GetFileNameWithoutExtension(archivePath)).ToLowerInvariant(); - if (IsZip(extension)) + if (SnapshotArchiveFormat.IsZip(extension)) ExtractZip(archivePath, destinationPath, cancellationToken); - else if (IsTarArchive(extension, innerExtension)) + else if (SnapshotArchiveFormat.IsTarBased(extension, innerExtension)) ExtractTar(archivePath, destinationPath, extension, stripComponents, cancellationToken); else throw new NotSupportedException($"Unsupported snapshot archive format: {archivePath}"); - } - private static bool IsZip(string extension) => - extension is ".zip"; + EnsureNotEmpty(destinationPath); + } - private static bool IsTarArchive(string extension, string innerExtension) => - extension is ".tar" or ".zst" or ".zstd" or ".gz" or ".bz2" or ".xz" || innerExtension == ".tar"; + private static void EnsureNotEmpty(string destinationPath) + { + if (!SnapshotDatabase.Exists(destinationPath)) + throw new InvalidOperationException( + $"The archive produced no files under '{destinationPath}'. Check Snapshot.StripComponents against the archive layout."); + } private static void ExtractZip(string archivePath, string destinationPath, CancellationToken cancellationToken) { @@ -54,12 +79,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; @@ -68,6 +98,9 @@ private static void ExtractTar(string archivePath, string destinationPath, strin { cancellationToken.ThrowIfCancellationRequested(); + if (entry.EntryType is TarEntryType.GlobalExtendedAttributes) + continue; + string? strippedPath = StripLeadingComponents(entry.Name, stripComponents); if (strippedPath is null) continue; @@ -79,19 +112,21 @@ private static void ExtractTar(string archivePath, string destinationPath, strin 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}."); } } - 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 00000000000..9fa01925c1c --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpClient.cs @@ -0,0 +1,117 @@ +// 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 +{ + 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); + + private readonly HttpClient _httpClient = new(new HttpClientHandler { AllowAutoRedirect = false }) + { + Timeout = Timeout.InfiniteTimeSpan + }; + + 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); + + 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 e) when (headerCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) + { + throw new HttpRequestException($"No response headers received within {HeaderTimeout.TotalSeconds}s.", e); + } + + 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, TimeSpan stallTimeout, CancellationToken cancellationToken) + { + using StallGuardedReader reader = new(stallTimeout, cancellationToken); + byte[] buffer = ArrayPool.Shared.Rent(SkipBufferSize); + try + { + long remaining = bytesToSkip; + while (remaining > 0) + { + 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 + { + 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 00000000000..9894bb2cb9f --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/SnapshotHttpStream.cs @@ -0,0 +1,472 @@ +// 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, TimeSpan StallTimeout, bool ComputeChecksum = false) +{ + public static SnapshotStreamSettings Default(int connections) => + new(connections, 64 * 1024 * 1024, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(300), SnapshotHttpClient.DefaultStallTimeout); +} + +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; + 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(); + _hasher = settings.ComputeChecksum ? IncrementalHash.CreateHash(HashAlgorithmName.SHA256) : null; + _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + bool ranged = remoteInfo is { SupportsRanges: true, Length: not null }; + int producerCount = ranged ? settings.Connections : 1; + _window = new SemaphoreSlim(producerCount + 1, producerCount + 1); + _progress = new ProgressReporter(ProgressLabel, logManager, (ulong)(remoteInfo.Length ?? 0), ProgressInterval); + _progress.Logger.SetFormat(SnapshotProgress.FormatBytes(ProgressLabel, remoteInfo.Length)); + + if (ranged) + { + _chunkCount = (remoteInfo.Length!.Value + settings.ChunkSize - 1) / settings.ChunkSize; + _producers = new Task[producerCount]; + for (int i = 0; i < producerCount; 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) + { + ReturnBuffer(_current.Buffer!); + _current = default; + _window.Release(); + } + + return toCopy; + } + + public async Task FinishAsync(CancellationToken cancellationToken) + { + await Task.Factory.StartNew(Drain, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).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(); + _buffers.Clear(); + _current = default; + _pending.Clear(); + } + + 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) + { + using StallGuardedReader reader = new(_settings.StallTimeout, _cts.Token); + 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 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; + } + + return; + } + catch (HttpRequestException e) when (SnapshotHttpClient.IsPermanentHttpError(e)) + { + throw; + } + catch (Exception e) when (IsRetryable(e)) + { + 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; + using StallGuardedReader reader = new(_settings.StallTimeout, _cts.Token); + 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 reader.ReadAsync(content, buffer.AsMemory(filled, _settings.ChunkSize - filled)).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 (IsRetryable(e)) + { + 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, _settings.StallTimeout, _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 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 void ReturnBuffer(byte[] buffer) => _buffers.Add(buffer); + + internal int PooledBufferCount => _buffers.Count; + + 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 00000000000..0ce63da9c27 --- /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 00000000000..186411f3eba --- /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/StallGuardedReader.cs b/src/Nethermind/Nethermind.Init.Snapshot/StallGuardedReader.cs new file mode 100644 index 00000000000..06034a6672a --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/StallGuardedReader.cs @@ -0,0 +1,35 @@ +// 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 + { + return await content.ReadAsync(buffer, _stallCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException e) when (!cancellationToken.IsCancellationRequested) + { + throw new IOException($"No data received for {stallTimeout.TotalSeconds}s.", e); + } + finally + { + if (!_stallCts.TryReset()) + Recreate(); + } + } + + 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 new file mode 100644 index 00000000000..0cb8ec90422 --- /dev/null +++ b/src/Nethermind/Nethermind.Init.Snapshot/StreamingSnapshotInitializer.cs @@ -0,0 +1,195 @@ +// 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(); + using SnapshotHttpClient client = new(); + + for (int attempt = 1; attempt <= MaxSourceChangedRestarts; attempt++) + { + 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."); + + LogMode(remoteInfo); + CheckDiskSpace(remoteInfo.Length); + + bool verified; + try + { + verified = await StreamAndExtractAsync(client, remoteInfo, checkpoint, cancellationToken).ConfigureAwait(false); + } + catch (SnapshotSourceChangedException e) + { + if (_logger.IsWarn) + _logger.Warn($"{e.Message} Restarting the snapshot download."); + DeleteDatabase(); + continue; + } + catch (Exception e) when (e is IOException or InvalidDataException or EndOfStreamException or ZstdException) + { + if (_logger.IsError) + _logger.Error($"Snapshot streaming failed: {e.Message} Deleting the partially extracted database."); + DeleteDatabase(); + LogContinuingWithoutSnapshot(); + return; + } + catch (Exception e) when (e is not OperationCanceledException) + { + if (_logger.IsError) + _logger.Error("Snapshot streaming failed. Deleting the partially extracted database.", e); + DeleteDatabase(); + throw; + } + + if (!verified) + { + DeleteDatabase(); + LogContinuingWithoutSnapshot(); + } + + return; + } + + 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( + SnapshotHttpClient client, SnapshotRemoteInfo remoteInfo, SnapshotCheckpoint checkpoint, CancellationToken cancellationToken) + { + string? expectedChecksum = config.Checksum; + byte[]? checksum; + await using (SnapshotHttpStream stream = new( + client, url, remoteInfo, settings with { ComputeChecksum = expectedChecksum is not null }, logManager, cancellationToken)) + { + SnapshotExtractor extractor = new(logManager); + string extension = Path.GetExtension(config.SnapshotFileName).ToLowerInvariant(); + await extractor.ExtractTarStreamAsync(stream, dbPath, extension, config.StripComponents, cancellationToken).ConfigureAwait(false); + checksum = await stream.FinishAsync(cancellationToken).ConfigureAwait(false); + } + + if (expectedChecksum is null) + { + if (_logger.IsWarn) + _logger.Warn("Snapshot checksum is not configured."); + } + else if (checksum is null + || !SnapshotChecksum.Verify(checksum, expectedChecksum, "Deleting the extracted database.", _logger)) + { + return false; + } + + checkpoint.Advance(SnapshotStage.Completed); + if (_logger.IsInfo) + _logger.Info("Database successfully initialized from streamed snapshot."); + return true; + } + + 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(); + if (!SnapshotArchiveFormat.IsStreamable(extension)) + throw new NotSupportedException( + $"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() + { + 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; + } + + SnapshotDiskSpace.Check(drives, SnapshotDiskSpace.GetRequiredSpaceForExtraction(snapshotLength.Value), "extract"); + } + + private void LogContinuingWithoutSnapshot() + { + if (_logger.IsInfo) + _logger.Info("The node will continue running without a snapshot."); + } + + private void DeleteDatabase() + { + try + { + SnapshotDatabase.Delete(dbPath); + } + catch (Exception e) when (e is IOException or UnauthorizedAccessException) + { + throw new IOException( + $"Could not clean up the database at {dbPath}, so the node must not start on top of it. Delete it manually before restarting.", e); + } + } +} diff --git a/src/Nethermind/Nethermind.slnx b/src/Nethermind/Nethermind.slnx index c7884b9e584..fe8f6044f5c 100644 --- a/src/Nethermind/Nethermind.slnx +++ b/src/Nethermind/Nethermind.slnx @@ -76,6 +76,7 @@ +