-
Notifications
You must be signed in to change notification settings - Fork 721
Stream snapshot download without storing the archive #12862
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
svlachakis
wants to merge
19
commits into
master
Choose a base branch
from
feature/snapshot-streaming-download
base: master
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from 2 commits
Commits
Show all changes
19 commits
Select commit
Hold shift + click to select a range
26279e0
fix: check snapshot disk space before download
svlachakis 5da50e5
feat: stream snapshot download without storing the archive
svlachakis 91f6604
fix: harden snapshot streaming per review
svlachakis 7dcf349
ci: add Nethermind.Init.Snapshot.Test to the test matrix
svlachakis f4aa509
fix: snapshot streaming review round two
svlachakis 06f8ac4
style: drop usings made redundant by implicit usings
svlachakis a5000a9
fix: snapshot streaming review round three
svlachakis 2d339d3
style: drop stale using, single source for the stall timeout
svlachakis 5f5c887
fix: close three silent empty-database paths
svlachakis d6c6497
fix: release stream buffers eagerly on dispose
svlachakis 8ac81bf
Merge branch 'master' into feature/snapshot-streaming-download
svlachakis ecf86f4
fix: address review round five
svlachakis 0d67d7e
refactor: unify snapshot failure policy, harden cleanup and hot path
svlachakis c0ab337
fix: restart the download when the checkpoint outlives the archive
svlachakis 048341d
fix: keep cleanup failures out of the streaming failure handler
svlachakis c64ae86
fix: stop promising an outcome the cleanup can still deny
svlachakis 328feb4
style: drop the using left behind by the tar helper move
svlachakis f13275b
fix: require a checksum unless the server proves entity identity
svlachakis 834b61e
fix: close the gaps the ETag gate opened
svlachakis File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
179 changes: 179 additions & 0 deletions
179
src/Nethermind/Nethermind.Init.Snapshot.Test/FlakySnapshotServer.cs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,179 @@ | ||
| // SPDX-FileCopyrightText: 2026 Demerzel Solutions Limited | ||
| // SPDX-License-Identifier: LGPL-3.0-only | ||
|
|
||
| using System.Collections.Concurrent; | ||
| using System.Net; | ||
|
|
||
| namespace Nethermind.Init.Snapshot.Test; | ||
|
|
||
| internal sealed class FlakySnapshotServer : IDisposable | ||
| { | ||
| private readonly HttpListener _listener; | ||
| private readonly ConcurrentDictionary<string, int> _attemptsPerRange = new(); | ||
| private int _requestCount; | ||
| private int _switchAfterRequests = int.MaxValue; | ||
| private byte[] _newContent = []; | ||
| private string? _newETag; | ||
|
|
||
| public FlakySnapshotServer() | ||
| { | ||
| (_listener, int port) = StartListener(); | ||
| Url = $"http://127.0.0.1:{port}/snapshot.tar.zst"; | ||
| Task.Run(AcceptLoopAsync); | ||
| } | ||
|
|
||
| public string Url { get; } | ||
|
|
||
| public byte[] Content { get; set; } = []; | ||
|
|
||
| public string? ETag { get; set; } = "\"v1\""; | ||
|
|
||
| public bool SupportsRanges { get; set; } = true; | ||
|
|
||
| public int? DropFirstAttemptPerRangeAfterBytes { get; set; } | ||
|
|
||
| public int? FailWithNotFoundAfterRequests { get; set; } | ||
|
|
||
| public int RequestCount => _requestCount; | ||
|
|
||
| public void SwitchSourceAfterRequests(int requestCount, byte[] newContent, string? newETag) | ||
| { | ||
| _newContent = newContent; | ||
| _newETag = newETag; | ||
| _switchAfterRequests = requestCount; | ||
| } | ||
|
|
||
| private static (HttpListener Listener, int Port) StartListener() | ||
| { | ||
| for (int attempt = 0; ; attempt++) | ||
| { | ||
| HttpListener listener = new(); | ||
| int port = Random.Shared.Next(20000, 60000); | ||
| listener.Prefixes.Add($"http://127.0.0.1:{port}/"); | ||
| try | ||
| { | ||
| listener.Start(); | ||
| return (listener, port); | ||
| } | ||
| catch (HttpListenerException) when (attempt < 5) | ||
| { | ||
| listener.Close(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public void Dispose() | ||
| { | ||
| _listener.Stop(); | ||
| _listener.Close(); | ||
| } | ||
|
|
||
| private async Task AcceptLoopAsync() | ||
| { | ||
| while (_listener.IsListening) | ||
| { | ||
| HttpListenerContext context; | ||
| try | ||
| { | ||
| context = await _listener.GetContextAsync(); | ||
| } | ||
| catch | ||
| { | ||
| return; | ||
| } | ||
|
|
||
| _ = Task.Run(() => HandleAsync(context)); | ||
| } | ||
| } | ||
|
|
||
| private async Task HandleAsync(HttpListenerContext context) | ||
| { | ||
| int requestNumber = Interlocked.Increment(ref _requestCount); | ||
| byte[] content = requestNumber > _switchAfterRequests ? _newContent : Content; | ||
| string? etag = requestNumber > _switchAfterRequests ? _newETag : ETag; | ||
| HttpListenerResponse response = context.Response; | ||
|
|
||
| try | ||
| { | ||
| if (FailWithNotFoundAfterRequests is int failAfter && requestNumber > failAfter) | ||
| { | ||
| response.StatusCode = 404; | ||
| response.Close(); | ||
| return; | ||
| } | ||
|
|
||
| if (etag is not null) | ||
| response.Headers["ETag"] = etag; | ||
| string? rangeHeader = context.Request.Headers["Range"]; | ||
| string? ifRange = context.Request.Headers["If-Range"]; | ||
| long from = 0; | ||
| long to = content.Length - 1; | ||
| bool ranged = SupportsRanges | ||
| && rangeHeader is not null | ||
| && (ifRange is null || ifRange == etag) | ||
| && TryParseRange(rangeHeader, content.Length, ref from, ref to); | ||
|
|
||
| if (ranged && from >= content.Length) | ||
| { | ||
| response.StatusCode = 416; | ||
| response.Headers["Content-Range"] = $"bytes */{content.Length}"; | ||
| response.Close(); | ||
| return; | ||
| } | ||
|
|
||
| if (ranged) | ||
| { | ||
| response.StatusCode = 206; | ||
| response.Headers["Content-Range"] = $"bytes {from}-{to}/{content.Length}"; | ||
| } | ||
| else | ||
| { | ||
| response.StatusCode = 200; | ||
| from = 0; | ||
| to = content.Length - 1; | ||
| } | ||
|
|
||
| long length = to - from + 1; | ||
| response.ContentLength64 = length; | ||
|
|
||
| string rangeKey = rangeHeader ?? "full"; | ||
| int attempt = _attemptsPerRange.AddOrUpdate(rangeKey, 1, static (_, previous) => previous + 1); | ||
| if (DropFirstAttemptPerRangeAfterBytes is int dropAfter && attempt == 1 && length > dropAfter) | ||
| { | ||
| await response.OutputStream.WriteAsync(content.AsMemory((int)from, dropAfter)); | ||
| response.Abort(); | ||
| return; | ||
| } | ||
|
|
||
| await response.OutputStream.WriteAsync(content.AsMemory((int)from, (int)length)); | ||
| response.Close(); | ||
| } | ||
| catch | ||
| { | ||
| try | ||
| { | ||
| response.Abort(); | ||
| } | ||
| catch | ||
| { | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static bool TryParseRange(string rangeHeader, long contentLength, ref long from, ref long to) | ||
| { | ||
| if (!rangeHeader.StartsWith("bytes=", StringComparison.Ordinal)) | ||
| return false; | ||
|
|
||
| string[] parts = rangeHeader["bytes=".Length..].Split('-'); | ||
| if (parts.Length != 2 || !long.TryParse(parts[0], out long start)) | ||
| return false; | ||
|
|
||
| from = start; | ||
| to = parts[1].Length > 0 && long.TryParse(parts[1], out long end) | ||
| ? Math.Min(end, contentLength - 1) | ||
| : contentLength - 1; | ||
| return true; | ||
| } | ||
|
|
||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Low — the test server discards its accept-loop task and later suppresses background failures in empty catches. An unexpected listener or handler failure therefore cannot reach NUnit, which can turn server bugs into false greens or opaque timeouts. Retain and observe the background tasks during teardown, narrow expected shutdown handling, and surface unexpected exceptions; the empty catches around the hang cancellation and response abort need the same treatment.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed in f13275b. The accept loop and every handler task are retained;
Disposestops the listener, waits for them, and rethrows anything unexpected as anAggregateException, so a server bug fails the test instead of turning into a false green or a timeout.The catches are narrowed:
HttpListenerException,IOException,ObjectDisposedExceptionandOperationCanceledExceptionare the expected shapes when a client disconnects (which the fault-injection tests do deliberately) and are ignored; anything else is recorded. The hang path awaits its cancellation instead of swallowing it.