Add the download pipeline: sniffing, redirects, throttling, verdicts
Second half of the collector foundation. Still nothing in the app references it; the pipeline is tested end to end against a deliberately badly behaved loopback server before anything depends on it. Types come from the bytes, never from the URL, the extension or Content-Type - two of those three are chosen by whoever serves the file, and a host must not get to pick the extension of a file written to the user's disk. Animation is a separate question from kind: GIF89a proves nothing without a second image descriptor, and a PNG is an APNG only if acTL precedes the first IDAT, so both are walked properly rather than guessed. Timeouts are split three ways because HttpClient.Timeout covers the whole response: any value large enough for a 30 MB file is also large enough for a dead connection to hang on. Connect, headers and a per-read idle deadline let both be strict. Redirects are followed by hand since the shared proxy handler disables them, which is what allows a hop cap, loop detection and refusing a jump to a data: URL. The lease verdict is a pure function, because ProxyLease's constructor is internal to the domain and no test can fabricate one. Its rule is that the verdict describes the transport, not the resource: a 404 is a working proxy, and so is a 429 - blaming the proxy for an origin's rate limit would make the pool rotate away from a good address in response to being asked to slow down. Cancellation reports nothing at all. Throttling exists to be obeyed. It is raised only by the host's own 429 and 503, and never by rotating to another proxy. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Leonid Pershin
co-authored by
Claude Opus 5
Claude Opus 5
parent
181f974a37
commit
1742c094e9
@@ -0,0 +1,344 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Globalization;
|
||||
using System.IO.Compression;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Text;
|
||||
|
||||
namespace AvParser.Infrastructure.Tests.Collecting;
|
||||
|
||||
/// <summary>One request as the server saw it.</summary>
|
||||
/// <param name="Path">Requested path.</param>
|
||||
/// <param name="Headers">Request headers, lower-cased keys.</param>
|
||||
/// <param name="ArrivedUtc">When the request line was read.</param>
|
||||
/// <param name="CompletedUtc">When the response finished being written.</param>
|
||||
internal sealed record RecordedRequest(
|
||||
string Path,
|
||||
IReadOnlyDictionary<string, string> Headers,
|
||||
DateTimeOffset ArrivedUtc,
|
||||
DateTimeOffset CompletedUtc
|
||||
);
|
||||
|
||||
/// <summary>How the server should answer one path.</summary>
|
||||
internal sealed record Reply
|
||||
{
|
||||
/// <summary>Status code to send.</summary>
|
||||
public int Status { get; init; } = 200;
|
||||
|
||||
/// <summary>Body bytes. Ignored for statuses that carry none.</summary>
|
||||
public byte[] Body { get; init; } = [];
|
||||
|
||||
/// <summary>Value for the <c>Content-Type</c> header, if any.</summary>
|
||||
public string? ContentType { get; init; }
|
||||
|
||||
/// <summary>Extra headers, sent verbatim.</summary>
|
||||
public IReadOnlyDictionary<string, string> Headers { get; init; } =
|
||||
new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
/// <summary>Send the body with chunked transfer encoding instead of a length.</summary>
|
||||
public bool Chunked { get; init; }
|
||||
|
||||
/// <summary>Compress the body and declare <c>Content-Encoding: gzip</c>.</summary>
|
||||
public bool Gzip { get; init; }
|
||||
|
||||
/// <summary>Declare this length instead of the real one — for testing truncation.</summary>
|
||||
public long? DeclaredLength { get; init; }
|
||||
|
||||
/// <summary>Write only this many body bytes, then close the connection mid-transfer.</summary>
|
||||
public int? TruncateAfter { get; init; }
|
||||
|
||||
/// <summary>Wait this long before sending the status line.</summary>
|
||||
public TimeSpan HeaderDelay { get; init; }
|
||||
|
||||
/// <summary>Wait this long between body writes.</summary>
|
||||
public TimeSpan BodyDelay { get; init; }
|
||||
|
||||
/// <summary>Bytes per body write, when dripping.</summary>
|
||||
public int BodyChunkSize { get; init; } = int.MaxValue;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A deliberately badly behaved HTTP server on loopback.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Raw sockets rather than <c>HttpListener</c> or Kestrel, because most of what needs testing is
|
||||
/// below the level either of those will let you reach: declaring four kilobytes and sending three
|
||||
/// hundred bytes, dripping one byte at a time, or closing mid-chunk. A well-behaved server cannot
|
||||
/// produce the failures a collector has to survive.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Port 0 lets the OS assign a free port, so test classes never collide and nothing has to be
|
||||
/// reserved or cleaned up between runs.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class LoopbackServer : IAsyncDisposable
|
||||
{
|
||||
private readonly TcpListener _listener;
|
||||
private readonly CancellationTokenSource _shutdown = new();
|
||||
private readonly ConcurrentDictionary<string, Func<Reply>> _routes = new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentBag<RecordedRequest> _requests = [];
|
||||
private readonly Task _acceptLoop;
|
||||
|
||||
public LoopbackServer()
|
||||
{
|
||||
_listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
_listener.Start();
|
||||
|
||||
var port = ((IPEndPoint)_listener.LocalEndpoint).Port;
|
||||
BaseAddress = new Uri($"http://127.0.0.1:{port.ToString(CultureInfo.InvariantCulture)}/");
|
||||
|
||||
_acceptLoop = Task.Run(AcceptAsync);
|
||||
}
|
||||
|
||||
/// <summary>Root of this server.</summary>
|
||||
public Uri BaseAddress { get; }
|
||||
|
||||
/// <summary>Every request served, in no particular order.</summary>
|
||||
public IReadOnlyCollection<RecordedRequest> Requests => [.. _requests];
|
||||
|
||||
/// <summary>How many requests were served.</summary>
|
||||
public int RequestCount => _requests.Count;
|
||||
|
||||
/// <summary>Registers a fixed reply for a path.</summary>
|
||||
public Uri Map(string path, Reply reply) => Map(path, () => reply);
|
||||
|
||||
/// <summary>Registers a reply computed per request, for sequences.</summary>
|
||||
public Uri Map(string path, Func<Reply> reply)
|
||||
{
|
||||
_routes[path] = reply;
|
||||
|
||||
return new Uri(BaseAddress, path);
|
||||
}
|
||||
|
||||
/// <summary>Registers a body served as-is with a 200.</summary>
|
||||
public Uri MapBody(string path, byte[] body, string? contentType = null) =>
|
||||
Map(path, new Reply { Body = body, ContentType = contentType });
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _shutdown.CancelAsync();
|
||||
_listener.Stop();
|
||||
|
||||
try
|
||||
{
|
||||
await _acceptLoop;
|
||||
}
|
||||
catch (Exception ex) when (ex is OperationCanceledException or SocketException or ObjectDisposedException)
|
||||
{
|
||||
// Shutting down a listener out from under its accept loop is the normal way to stop it.
|
||||
}
|
||||
|
||||
_shutdown.Dispose();
|
||||
}
|
||||
|
||||
private async Task AcceptAsync()
|
||||
{
|
||||
while (!_shutdown.IsCancellationRequested)
|
||||
{
|
||||
TcpClient client;
|
||||
|
||||
try
|
||||
{
|
||||
client = await _listener.AcceptTcpClientAsync(_shutdown.Token);
|
||||
}
|
||||
catch (Exception ex) when (ex is OperationCanceledException or SocketException or ObjectDisposedException)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_ = Task.Run(() => ServeAsync(client));
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ServeAsync(TcpClient client)
|
||||
{
|
||||
using (client)
|
||||
{
|
||||
try
|
||||
{
|
||||
await using var stream = client.GetStream();
|
||||
|
||||
var (path, headers) = await ReadRequestAsync(stream);
|
||||
var arrived = DateTimeOffset.UtcNow;
|
||||
|
||||
try
|
||||
{
|
||||
var reply = _routes.TryGetValue(path, out var factory)
|
||||
? factory()
|
||||
: new Reply { Status = 404, Body = "not found"u8.ToArray() };
|
||||
|
||||
if (reply.HeaderDelay > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(reply.HeaderDelay, _shutdown.Token);
|
||||
}
|
||||
|
||||
await WriteReplyAsync(stream, reply);
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Recorded even when the client hung up mid-response: several tests arrange
|
||||
// exactly that, and they still need to know the request arrived.
|
||||
_requests.Add(new RecordedRequest(path, headers, arrived, DateTimeOffset.UtcNow));
|
||||
}
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or OperationCanceledException or SocketException)
|
||||
{
|
||||
// A client that hangs up mid-response is exactly what several tests arrange.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task<(string Path, Dictionary<string, string> Headers)> ReadRequestAsync(NetworkStream stream)
|
||||
{
|
||||
var buffer = new List<byte>(1024);
|
||||
var single = new byte[1];
|
||||
|
||||
while (buffer.Count < 16 * 1024)
|
||||
{
|
||||
var read = await stream.ReadAsync(single);
|
||||
|
||||
if (read == 0)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
buffer.Add(single[0]);
|
||||
|
||||
if (
|
||||
buffer.Count >= 4
|
||||
&& buffer[^4] == (byte)'\r'
|
||||
&& buffer[^3] == (byte)'\n'
|
||||
&& buffer[^2] == (byte)'\r'
|
||||
&& buffer[^1] == (byte)'\n'
|
||||
)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
var text = Encoding.ASCII.GetString([.. buffer]);
|
||||
var lines = text.Split("\r\n", StringSplitOptions.RemoveEmptyEntries);
|
||||
var path =
|
||||
lines.Length > 0
|
||||
? lines[0].Split(' ') is { Length: >= 2 } parts
|
||||
? parts[1]
|
||||
: "/"
|
||||
: "/";
|
||||
|
||||
var headers = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
|
||||
foreach (var line in lines.Skip(1))
|
||||
{
|
||||
var colon = line.IndexOf(':', StringComparison.Ordinal);
|
||||
|
||||
if (colon > 0)
|
||||
{
|
||||
headers[line[..colon].Trim()] = line[(colon + 1)..].Trim();
|
||||
}
|
||||
}
|
||||
|
||||
return (path, headers);
|
||||
}
|
||||
|
||||
private async Task WriteReplyAsync(NetworkStream stream, Reply reply)
|
||||
{
|
||||
var body = reply.Gzip ? Compress(reply.Body) : reply.Body;
|
||||
|
||||
var head = new StringBuilder();
|
||||
head.Append(CultureInfo.InvariantCulture, $"HTTP/1.1 {reply.Status} {Describe(reply.Status)}\r\n");
|
||||
head.Append("Connection: close\r\n");
|
||||
|
||||
if (reply.ContentType is not null)
|
||||
{
|
||||
head.Append(CultureInfo.InvariantCulture, $"Content-Type: {reply.ContentType}\r\n");
|
||||
}
|
||||
|
||||
if (reply.Gzip)
|
||||
{
|
||||
head.Append("Content-Encoding: gzip\r\n");
|
||||
}
|
||||
|
||||
foreach (var (name, value) in reply.Headers)
|
||||
{
|
||||
head.Append(CultureInfo.InvariantCulture, $"{name}: {value}\r\n");
|
||||
}
|
||||
|
||||
if (reply.Chunked)
|
||||
{
|
||||
head.Append("Transfer-Encoding: chunked\r\n");
|
||||
}
|
||||
else
|
||||
{
|
||||
var declared = reply.DeclaredLength ?? body.Length;
|
||||
head.Append(CultureInfo.InvariantCulture, $"Content-Length: {declared}\r\n");
|
||||
}
|
||||
|
||||
head.Append("\r\n");
|
||||
|
||||
await stream.WriteAsync(Encoding.ASCII.GetBytes(head.ToString()), _shutdown.Token);
|
||||
await stream.FlushAsync(_shutdown.Token);
|
||||
|
||||
var limit = Math.Min(reply.TruncateAfter ?? body.Length, body.Length);
|
||||
var chunk = Math.Max(1, Math.Min(reply.BodyChunkSize, limit == 0 ? 1 : limit));
|
||||
|
||||
for (var offset = 0; offset < limit; offset += chunk)
|
||||
{
|
||||
var count = Math.Min(chunk, limit - offset);
|
||||
|
||||
if (reply.Chunked)
|
||||
{
|
||||
var header = count.ToString("x", CultureInfo.InvariantCulture) + "\r\n";
|
||||
await stream.WriteAsync(Encoding.ASCII.GetBytes(header), _shutdown.Token);
|
||||
}
|
||||
|
||||
await stream.WriteAsync(body.AsMemory(offset, count), _shutdown.Token);
|
||||
|
||||
if (reply.Chunked)
|
||||
{
|
||||
await stream.WriteAsync("\r\n"u8.ToArray(), _shutdown.Token);
|
||||
}
|
||||
|
||||
await stream.FlushAsync(_shutdown.Token);
|
||||
|
||||
if (reply.BodyDelay > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(reply.BodyDelay, _shutdown.Token);
|
||||
}
|
||||
}
|
||||
|
||||
// A truncated reply deliberately omits the terminal chunk: the client must notice.
|
||||
if (reply.Chunked && reply.TruncateAfter is null)
|
||||
{
|
||||
await stream.WriteAsync("0\r\n\r\n"u8.ToArray(), _shutdown.Token);
|
||||
await stream.FlushAsync(_shutdown.Token);
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] Compress(byte[] body)
|
||||
{
|
||||
using var output = new MemoryStream();
|
||||
|
||||
using (var gzip = new GZipStream(output, CompressionLevel.Fastest, leaveOpen: true))
|
||||
{
|
||||
gzip.Write(body);
|
||||
}
|
||||
|
||||
return output.ToArray();
|
||||
}
|
||||
|
||||
private static string Describe(int status) =>
|
||||
status switch
|
||||
{
|
||||
200 => "OK",
|
||||
301 => "Moved Permanently",
|
||||
302 => "Found",
|
||||
403 => "Forbidden",
|
||||
404 => "Not Found",
|
||||
410 => "Gone",
|
||||
429 => "Too Many Requests",
|
||||
500 => "Internal Server Error",
|
||||
503 => "Service Unavailable",
|
||||
_ => "Status",
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user