diff --git a/src/AvParser.Infrastructure/Collecting/FetchOptions.cs b/src/AvParser.Infrastructure/Collecting/FetchOptions.cs new file mode 100644 index 0000000..8d97b16 --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/FetchOptions.cs @@ -0,0 +1,91 @@ +using AvParser.Core.Collecting; +using AvParser.Infrastructure.Proxies; + +namespace AvParser.Infrastructure.Collecting; + +/// Limits and deadlines for downloading one item. +/// +/// Every ceiling here exists because the other side chooses the bytes. A missing cap is not a +/// generous default, it is a remote party deciding how much of the user's disk to fill. +/// +public sealed record FetchOptions +{ + /// Largest item to accept. Anything bigger is refused, ideally before the body starts. + public long MaxItemBytes { get; init; } = 32L * 1024 * 1024; + + /// Smallest item to accept; below this it is a tracking pixel or a spacer, not media. + public long MinItemBytes { get; init; } = 1024; + + /// How many redirects to follow before giving up. + public int MaxRedirects { get; init; } = 5; + + /// Connection, header and idle deadlines. + public HttpClientTimeouts Timeouts { get; init; } = HttpClientTimeouts.Default; + + /// Backstop on one item, however well behaved the transfer looks. + public TimeSpan MaxItemDuration { get; init; } = TimeSpan.FromMinutes(5); + + /// Whether a missing proxy is a hard failure rather than a direct connection. + public bool RequireProxy { get; init; } + + /// Identifies the collector to origins that care. + public string UserAgent { get; init; } = "AvParser/0.1"; + + /// Kinds to keep. Anything recognised but absent is refused after sniffing. + public IReadOnlySet AllowedKinds { get; init; } = + new HashSet + { + MediaKind.Jpeg, + MediaKind.Png, + MediaKind.Gif, + MediaKind.WebP, + MediaKind.Avif, + MediaKind.Mp4, + MediaKind.WebM, + }; + + /// Hashes already known to be dead-link placeholders. + public IReadOnlySet Tombstones { get; init; } = new HashSet(StringComparer.Ordinal); +} + +/// What one download attempt produced. +/// How it ended, in the vocabulary the journal records. +/// Address after redirects. +public sealed record FetchResult(SeenOutcome Outcome, Uri FinalUrl) +{ + /// The content, when there is any. + public MediaBlob? Blob { get; init; } + + /// + /// The staged file, when one survived. + /// + /// + /// Ownership passes to the caller: it must be handed to the store or deleted. The fetcher only + /// leaves one here when the download completed and verified. + /// + public string? TempPath { get; init; } + + /// Status of the final response, or 0 when none arrived. + public int HttpStatus { get; init; } + + /// What the origin claimed the content was. + public string? ContentType { get; init; } + + /// Proxy the content came through, or null when direct. + public string? ProxyKey { get; init; } + + /// Localisation code for the failure, matching a Parse.Error.{Code} key. + public string? ErrorCode { get; init; } + + /// Detail for the log and the error row. + public string? ErrorDetail { get; init; } + + /// How long the attempt took. + public TimeSpan Elapsed { get; init; } + + /// How long the origin asked us to wait, when it did. + public TimeSpan? RetryAfter { get; init; } + + /// Whether the item was stored-worthy. + public bool IsSuccess => Outcome is SeenOutcome.Stored or SeenOutcome.Duplicate; +} diff --git a/src/AvParser.Infrastructure/Collecting/HostThrottle.cs b/src/AvParser.Infrastructure/Collecting/HostThrottle.cs new file mode 100644 index 0000000..03812bd --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/HostThrottle.cs @@ -0,0 +1,177 @@ +using System.Collections.Concurrent; +using Microsoft.Extensions.Logging; + +namespace AvParser.Infrastructure.Collecting; + +/// +/// Keeps the collector polite to each origin, independently of how many workers are running. +/// +/// +/// +/// This exists to be obeyed, not worked around. It is raised only by the host's own signals — a +/// 429 or a 503 with Retry-After — and the proxy pool plays no part in it: rotating away +/// from a rate limit would be evading the limit rather than respecting it, so a refusal slows this +/// collector down and never redirects it through a different address. +/// +/// +/// Concurrency and pacing are separate knobs because they answer different questions: how many +/// connections an origin will tolerate at once, and how fast it will tolerate them arriving. +/// +/// +public sealed class HostThrottle( + int maxConcurrentPerHost, + TimeSpan minimumInterval, + ILogger logger, + TimeProvider? timeProvider = null +) : IDisposable +{ + private readonly ConcurrentDictionary _hosts = new(StringComparer.OrdinalIgnoreCase); + private readonly int _maxConcurrent = Math.Clamp(maxConcurrentPerHost, 1, 64); + private readonly TimeSpan _minimumInterval = minimumInterval < TimeSpan.Zero ? TimeSpan.Zero : minimumInterval; + private readonly ILogger _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + private readonly TimeProvider _time = timeProvider ?? TimeProvider.System; + + /// How long this host is still refusing requests, or zero when it is not. + public TimeSpan CooldownRemaining(Uri url) + { + ArgumentNullException.ThrowIfNull(url); + + if (!_hosts.TryGetValue(url.IdnHost, out var state)) + { + return TimeSpan.Zero; + } + + var remaining = state.CoolingUntil - _time.GetUtcNow(); + + return remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero; + } + + /// + /// Waits for a turn against this host. + /// + /// A token to dispose when the request is finished. + public async ValueTask AcquireAsync(Uri url, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(url); + + var state = _hosts.GetOrAdd(url.IdnHost, _ => new HostState(_maxConcurrent)); + + await state.Slots.WaitAsync(cancellationToken).ConfigureAwait(false); + + try + { + var wait = state.ReserveNextSlot(_time.GetUtcNow(), _minimumInterval); + + if (wait > TimeSpan.Zero) + { + await Task.Delay(wait, _time, cancellationToken).ConfigureAwait(false); + } + } + catch + { + state.Slots.Release(); + throw; + } + + return new Turn(state); + } + + /// Records that an origin asked us to slow down. + /// Address that was refused. + /// What the origin asked for, when it said. + public void ReportRateLimited(Uri url, TimeSpan? retryAfter) + { + ArgumentNullException.ThrowIfNull(url); + + var state = _hosts.GetOrAdd(url.IdnHost, _ => new HostState(_maxConcurrent)); + var consecutive = state.RecordRefusal(); + var pause = retryAfter ?? RetryAfter.Backoff(consecutive); + + state.CoolingUntil = _time.GetUtcNow() + pause; + + _logger.LogInformation( + "{Host} asked to slow down; pausing it for {Seconds:F0}s", + url.IdnHost, + pause.TotalSeconds + ); + } + + /// Records that an origin answered normally, ending any backoff escalation. + public void ReportSuccess(Uri url) + { + ArgumentNullException.ThrowIfNull(url); + + if (_hosts.TryGetValue(url.IdnHost, out var state)) + { + state.ResetRefusals(); + } + } + + /// + public void Dispose() + { + foreach (var state in _hosts.Values) + { + state.Slots.Dispose(); + } + + _hosts.Clear(); + } + + private sealed class HostState(int maxConcurrent) + { + private readonly Lock _gate = new(); + private long _nextAllowedTicks; + private int _consecutiveRefusals; + + public SemaphoreSlim Slots { get; } = new(maxConcurrent, maxConcurrent); + + public DateTimeOffset CoolingUntil { get; set; } + + /// Claims the next pacing slot and reports how long to wait for it. + public TimeSpan ReserveNextSlot(DateTimeOffset now, TimeSpan interval) + { + if (interval <= TimeSpan.Zero) + { + return TimeSpan.Zero; + } + + lock (_gate) + { + var earliest = new DateTimeOffset(_nextAllowedTicks, TimeSpan.Zero); + var start = earliest > now ? earliest : now; + + _nextAllowedTicks = (start + interval).UtcTicks; + + return start - now; + } + } + + public int RecordRefusal() + { + lock (_gate) + { + return ++_consecutiveRefusals; + } + } + + public void ResetRefusals() + { + lock (_gate) + { + _consecutiveRefusals = 0; + } + } + } + + private sealed class Turn(HostState state) : IDisposable + { + private HostState? _state = state; + + public void Dispose() + { + // Guarded so a double dispose cannot inflate the semaphore's count. + Interlocked.Exchange(ref _state, null)?.Slots.Release(); + } + } +} diff --git a/src/AvParser.Infrastructure/Collecting/LeaseVerdict.cs b/src/AvParser.Infrastructure/Collecting/LeaseVerdict.cs new file mode 100644 index 0000000..73269cc --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/LeaseVerdict.cs @@ -0,0 +1,114 @@ +using System.Net; +using System.Net.Sockets; +using System.Security.Authentication; + +namespace AvParser.Infrastructure.Collecting; + +/// What a download should tell the proxy pool about the proxy it used. +public enum LeaseVerdict +{ + /// The proxy carried the request. Nothing about the resource changes this. + Success = 0, + + /// The proxy itself is the problem. + Failure = 1, + + /// No opinion; the pool must learn nothing from this attempt. + Neutral = 2, +} + +/// A verdict and why it was reached. +/// What to report. +/// Short machine-ish reason, recorded against the proxy on failure. +public readonly record struct LeaseOutcome(LeaseVerdict Verdict, string? Reason); + +/// +/// Decides what a download tells the proxy pool. +/// +/// +/// +/// The governing rule is that the verdict describes the transport, not the resource. A 404 +/// is a perfectly successful use of a proxy: the request went out, an answer came back, and the +/// only thing wrong is that the picture is gone. Recording that as a proxy failure would quarantine +/// working proxies at exactly the rate that dead links appear, which on a free list is most of the +/// time. +/// +/// +/// A 429 is deliberately a success for the same reason, and for one more: it is the origin's rate +/// limit, and blaming the proxy for it would make the pool rotate away from a working address in +/// response to being asked to slow down. +/// +/// +/// A pure function, because the alternative is untestable: ProxyLease's constructor is +/// internal to the domain assembly and no test can fabricate one. +/// +/// +public static class LeaseVerdicts +{ + /// Decides the verdict for one completed or failed attempt. + /// Status of the final response, when one arrived. + /// Exception that ended the attempt, when one did. + /// Whether any body bytes arrived before it ended. + /// Whether the user stopped it. + public static LeaseOutcome Decide(HttpStatusCode? status, Exception? failure, bool bytesReceived, bool cancelled) + { + // Cancellation is not evidence about anything. Treating it as failure would quarantine + // healthy proxies every time somebody presses Stop. + if (cancelled || failure is OperationCanceledException) + { + return new LeaseOutcome(LeaseVerdict.Neutral, null); + } + + if (failure is not null) + { + return new LeaseOutcome(LeaseVerdict.Failure, DescribeFailure(failure, bytesReceived)); + } + + if (status is null) + { + return new LeaseOutcome(LeaseVerdict.Failure, "no response"); + } + + return (int)status switch + { + // The proxy is answering about itself, or the gateway between us and the origin broke. + 407 => new LeaseOutcome(LeaseVerdict.Failure, "proxy auth"), + 502 => new LeaseOutcome(LeaseVerdict.Failure, "bad gateway"), + 504 => new LeaseOutcome(LeaseVerdict.Failure, "gateway timeout"), + _ => new LeaseOutcome(LeaseVerdict.Success, null), + }; + } + + private static string DescribeFailure(Exception failure, bool bytesReceived) + { + if (failure is TimeoutException) + { + // A stall part-way through is a different complaint from never answering at all. + return bytesReceived ? "stalled" : "timeout"; + } + + if (failure is AuthenticationException) + { + return "tls"; + } + + if (failure is HttpRequestException http) + { + if (http.InnerException is SocketException socket) + { + return socket.SocketErrorCode switch + { + SocketError.ConnectionRefused => "refused", + SocketError.ConnectionReset => "reset", + SocketError.HostNotFound or SocketError.NoData => "dns", + SocketError.TimedOut => "timeout", + _ => "connection failed", + }; + } + + return http.InnerException is AuthenticationException ? "tls" : "request failed"; + } + + return failure is IOException && bytesReceived ? "truncated" : "failed"; + } +} diff --git a/src/AvParser.Infrastructure/Collecting/MediaFetcher.cs b/src/AvParser.Infrastructure/Collecting/MediaFetcher.cs new file mode 100644 index 0000000..2556cf9 --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/MediaFetcher.cs @@ -0,0 +1,636 @@ +using System.Buffers; +using System.Diagnostics; +using System.Net; +using System.Net.Http.Headers; +using System.Security.Cryptography; +using AvParser.Core.Collecting; +using AvParser.Infrastructure.Media; +using AvParser.Infrastructure.Proxies; +using Microsoft.Extensions.Logging; + +namespace AvParser.Infrastructure.Collecting; + +/// Downloads one candidate, verifies it, and stages it for the store. +public interface IMediaFetcher +{ + /// Fetches one candidate. + /// + /// Never throws for an ordinary network or content failure — those come back as a + /// so the run continues. Cancellation still propagates. + /// + Task FetchAsync( + MediaCandidate candidate, + FetchOptions options, + CancellationToken cancellationToken = default + ); +} + +/// +/// The one place that knows how to get bytes off the network safely. +/// +/// +/// +/// Redirects are followed here because the shared proxy handler disables automatic redirects, and +/// following them by hand is what makes it possible to cap the hops, spot a loop, refuse a jump to +/// a non-HTTP scheme and re-enter the throttle when the host changes. +/// +/// +/// The invariant everything else depends on: a staged file is returned only when the transfer +/// reached the end and, where the length was knowable, matched it. Every other path deletes it in a +/// finally. A partial image that reached the blob store would be indistinguishable from a +/// real one for ever after. +/// +/// +public sealed class MediaFetcher( + IProxiedHttpClientFactory clients, + BlobStore blobs, + HostThrottle throttle, + ILogger logger +) : IMediaFetcher +{ + /// Big enough that a 32 MB file is a few hundred reads, small enough to pool cheaply. + private const int BufferSize = 64 * 1024; + + private readonly IProxiedHttpClientFactory _clients = clients ?? throw new ArgumentNullException(nameof(clients)); + private readonly BlobStore _blobs = blobs ?? throw new ArgumentNullException(nameof(blobs)); + private readonly HostThrottle _throttle = throttle ?? throw new ArgumentNullException(nameof(throttle)); + private readonly ILogger _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + + /// + public async Task FetchAsync( + MediaCandidate candidate, + FetchOptions options, + CancellationToken cancellationToken = default + ) + { + ArgumentNullException.ThrowIfNull(candidate); + ArgumentNullException.ThrowIfNull(options); + + var stopwatch = Stopwatch.StartNew(); + + using var budget = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + budget.CancelAfter(options.MaxItemDuration); + + try + { + var result = await FetchCoreAsync(candidate, options, stopwatch, budget.Token, cancellationToken) + .ConfigureAwait(false); + + // Cancelling a socket read surfaces as an IOException as often as an + // OperationCanceledException, and the handlers below would otherwise turn a stop into + // an ordinary per-item failure. One check here covers every return path. + cancellationToken.ThrowIfCancellationRequested(); + + return result; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (OperationCanceledException) + { + // The backstop fired rather than the user: an item that takes minutes is a failure of + // this attempt, not of the run. + return Failure(candidate.Url, SeenOutcome.Timeout, "Stalled", "item budget exceeded", stopwatch); + } + catch (ProxyUnavailableException) + { + return Failure(candidate.Url, SeenOutcome.Failed, "NoProxy", "no live proxy", stopwatch); + } + } + + private async Task FetchCoreAsync( + MediaCandidate candidate, + FetchOptions options, + Stopwatch stopwatch, + CancellationToken token, + CancellationToken userToken + ) + { + using var leased = await _clients + .LeaseAsync(options.Timeouts, options.RequireProxy, token) + .ConfigureAwait(false); + + HttpStatusCode? status = null; + Exception? failure = null; + var bytesReceived = false; + + try + { + var visited = new HashSet(StringComparer.OrdinalIgnoreCase); + var current = candidate.Url; + var referer = candidate.Referer; + + for (var hop = 0; ; hop++) + { + if (hop > options.MaxRedirects) + { + return Failure(current, SeenOutcome.Failed, "TooManyRedirects", null, stopwatch, leased.ProxyKey); + } + + if (!visited.Add(current.AbsoluteUri)) + { + return Failure(current, SeenOutcome.Failed, "RedirectLoop", null, stopwatch, leased.ProxyKey); + } + + using var turn = await _throttle.AcquireAsync(current, token).ConfigureAwait(false); + + using var request = BuildRequest(current, referer, options); + using var headerBudget = CancellationTokenSource.CreateLinkedTokenSource(token); + headerBudget.CancelAfter(options.Timeouts.Headers); + + HttpResponseMessage response; + try + { + response = await leased + .Client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, headerBudget.Token) + .ConfigureAwait(false); + } + catch (OperationCanceledException) when (!userToken.IsCancellationRequested) + { + failure = new TimeoutException("headers"); + return Failure(current, SeenOutcome.Timeout, "Stalled", "headers", stopwatch, leased.ProxyKey); + } + + using (response) + { + status = response.StatusCode; + + if (IsRedirect(response.StatusCode)) + { + var next = ResolveRedirect(current, response.Headers.Location); + + if (next is null) + { + return Failure( + current, + SeenOutcome.Failed, + "BadRedirect", + null, + stopwatch, + leased.ProxyKey + ); + } + + referer = current; + current = next; + continue; + } + + if (response.StatusCode is HttpStatusCode.TooManyRequests or HttpStatusCode.ServiceUnavailable) + { + var wait = ReadRetryAfter(response.Headers.RetryAfter); + _throttle.ReportRateLimited(current, wait); + + return Failure( + current, + SeenOutcome.RateLimited, + "RateLimited", + null, + stopwatch, + leased.ProxyKey, + (int)response.StatusCode + ) with + { + RetryAfter = wait, + }; + } + + if (!response.IsSuccessStatusCode) + { + var outcome = response.StatusCode is HttpStatusCode.Gone or HttpStatusCode.NotFound + ? SeenOutcome.Gone + : SeenOutcome.Failed; + + return Failure( + current, + outcome, + "HttpStatus", + ((int)response.StatusCode).ToString(System.Globalization.CultureInfo.InvariantCulture), + stopwatch, + leased.ProxyKey, + (int)response.StatusCode + ); + } + + _throttle.ReportSuccess(current); + + var declared = response.Content.Headers.ContentLength; + if (declared > options.MaxItemBytes) + { + // Refused without reading a byte of it: the point of a size cap is not to + // download the thing and then decide it was too big. + return Failure( + current, + SeenOutcome.TooLarge, + "TooLarge", + null, + stopwatch, + leased.ProxyKey, + (int)response.StatusCode + ); + } + + var download = await DownloadAsync(response, current, options, stopwatch, leased, token, userToken) + .ConfigureAwait(false); + + bytesReceived = download.BytesReceived; + failure = download.Failure; + + return download.Result; + } + } + } + catch (HttpRequestException ex) + { + failure = ex; + return Failure(candidate.Url, SeenOutcome.Failed, "RequestFailed", ex.Message, stopwatch, leased.ProxyKey); + } + catch (IOException ex) + { + failure = ex; + return Failure(candidate.Url, SeenOutcome.Failed, "RequestFailed", ex.Message, stopwatch, leased.ProxyKey); + } + finally + { + var verdict = LeaseVerdicts.Decide(status, failure, bytesReceived, userToken.IsCancellationRequested); + + switch (verdict.Verdict) + { + case LeaseVerdict.Success: + leased.Lease?.ReportSuccess(stopwatch.Elapsed); + break; + case LeaseVerdict.Failure: + leased.Lease?.ReportFailure(verdict.Reason); + break; + default: + // Neutral: say nothing, so the pool learns nothing from a cancelled attempt. + break; + } + } + } + + private async Task<(FetchResult Result, bool BytesReceived, Exception? Failure)> DownloadAsync( + HttpResponseMessage response, + Uri url, + FetchOptions options, + Stopwatch stopwatch, + LeasedHttpClient leased, + CancellationToken token, + CancellationToken userToken + ) + { + var contentType = response.Content.Headers.ContentType?.MediaType; + var status = (int)response.StatusCode; + var declared = response.Content.Headers.ContentLength; + + // Automatic decompression strips the encoding, so a declared length that came with one + // described the compressed body and cannot be compared against what we read. + var lengthIsComparable = declared.HasValue && response.Content.Headers.ContentEncoding.Count == 0; + + var temp = _blobs.CreateTempPath(); + var buffer = ArrayPool.Shared.Rent(BufferSize); + var prefix = new byte[MediaSignatures.PrefixLength]; + var prefixLength = 0; + var total = 0L; + var kind = MediaKind.Unknown; + var sniffed = false; + string? hash = null; + Exception? failure = null; + + try + { + // Scoped so the file is closed before it is read back below. Writing with FileShare.None + // is deliberate — nothing else may see a partial file — which makes closing it first a + // requirement rather than tidiness. + { + using var hasher = IncrementalHash.CreateHash(HashAlgorithmName.SHA256); + await using var source = await response.Content.ReadAsStreamAsync(token).ConfigureAwait(false); + await using var destination = new FileStream( + temp, + FileMode.Create, + FileAccess.Write, + FileShare.None, + BufferSize, + useAsync: true + ); + + using var idle = CancellationTokenSource.CreateLinkedTokenSource(token); + + while (true) + { + // Rescheduled before every read: the rule is "no bytes for N seconds", not + // "the whole body within N seconds". + idle.CancelAfter(options.Timeouts.Idle); + + int read; + try + { + read = await source.ReadAsync(buffer.AsMemory(0, BufferSize), idle.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!userToken.IsCancellationRequested) + { + failure = new TimeoutException("idle"); + return ( + Failure(url, SeenOutcome.Timeout, "Stalled", null, stopwatch, leased.ProxyKey, status), + total > 0, + failure + ); + } + catch (Exception ex) when (ex is HttpRequestException or IOException) + { + // The origin hung up mid-body. When it had declared a length, that is + // precisely a truncated transfer; the distinction matters because a + // truncated file must never reach the store. + failure = ex; + var truncated = lengthIsComparable && total > 0; + + return ( + Failure( + url, + SeenOutcome.Failed, + truncated ? "Truncated" : "RequestFailed", + truncated ? null : ex.Message, + stopwatch, + leased.ProxyKey, + status + ), + total > 0, + failure + ); + } + + if (read == 0) + { + break; + } + + total += read; + + if (total > options.MaxItemBytes) + { + return ( + Failure(url, SeenOutcome.TooLarge, "TooLarge", null, stopwatch, leased.ProxyKey, status), + true, + null + ); + } + + if (prefixLength < prefix.Length) + { + var take = Math.Min(prefix.Length - prefixLength, read); + buffer.AsSpan(0, take).CopyTo(prefix.AsSpan(prefixLength)); + prefixLength += take; + } + + if (!sniffed && prefixLength >= prefix.Length) + { + sniffed = true; + kind = MediaSignatures.Detect(prefix); + + if (Reject(kind, prefix, options) is { } early) + { + return ( + Failure( + url, + early.Outcome, + early.Code, + contentType, + stopwatch, + leased.ProxyKey, + status + ), + true, + null + ); + } + } + + hasher.AppendData(buffer, 0, read); + await destination.WriteAsync(buffer.AsMemory(0, read), token).ConfigureAwait(false); + } + + await destination.FlushAsync(token).ConfigureAwait(false); + hash = Convert.ToHexStringLower(hasher.GetCurrentHash()); + } + + // A body shorter than the length it announced is a truncated transfer, not a small file. + if (lengthIsComparable && total != declared!.Value) + { + failure = new IOException("truncated"); + return ( + Failure(url, SeenOutcome.Failed, "Truncated", null, stopwatch, leased.ProxyKey, status), + true, + failure + ); + } + + // Files shorter than the sniff window never reached the check inside the loop. + if (!sniffed) + { + kind = MediaSignatures.Detect(prefix.AsSpan(0, prefixLength)); + + if (Reject(kind, prefix.AsSpan(0, prefixLength), options) is { } rejection) + { + return ( + Failure( + url, + rejection.Outcome, + rejection.Code, + contentType, + stopwatch, + leased.ProxyKey, + status + ), + true, + null + ); + } + } + + if (total < options.MinItemBytes) + { + return ( + Failure(url, SeenOutcome.TooSmall, "TooSmall", null, stopwatch, leased.ProxyKey, status), + true, + null + ); + } + + if (options.Tombstones.Contains(hash!)) + { + return ( + Failure(url, SeenOutcome.Placeholder, "Placeholder", null, stopwatch, leased.ProxyKey, status), + true, + null + ); + } + + var blob = await DescribeAsync(temp, hash!, kind, total, prefix, token).ConfigureAwait(false); + + var result = new FetchResult(SeenOutcome.Stored, url) + { + Blob = blob, + TempPath = temp, + HttpStatus = status, + ContentType = contentType, + ProxyKey = leased.ProxyKey, + Elapsed = stopwatch.Elapsed, + }; + + temp = null; // ownership passes to the caller + return (result, true, null); + } + finally + { + ArrayPool.Shared.Return(buffer); + + // Anything that did not run to a verified end leaves nothing behind. This is what keeps + // the blob store free of half-written files. + if (temp is not null) + { + TryDelete(temp); + } + } + } + + /// Re-reads the head of the staged file to settle animation and dimensions. + /// + /// A second pass rather than a streaming state machine: whether a PNG is animated depends on a + /// chunk that can sit anywhere before the first IDAT, and tracking that across arbitrary + /// read boundaries is a great deal of fiddly code to avoid re-reading a megabyte that is still + /// in the page cache. + /// + private static async Task DescribeAsync( + string path, + string hash, + MediaKind kind, + long length, + byte[] prefix, + CancellationToken token + ) + { + var blob = MediaBlob.Create(hash, kind, length); + var (width, height) = MediaSignatures.ReadDimensions(kind, prefix); + + var headLength = (int)Math.Min(length, MediaSignatures.AnimationScanLimit); + var head = new byte[headLength]; + + await using (var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read)) + { + await stream.ReadExactlyAsync(head.AsMemory(0, headLength), token).ConfigureAwait(false); + } + + return blob with + { + Width = width, + Height = height, + IsAnimated = MediaSignatures.DetectAnimation(kind, head), + }; + } + + private static (SeenOutcome Outcome, string Code)? Reject( + MediaKind kind, + ReadOnlySpan prefix, + FetchOptions options + ) + { + if (MediaSignatures.LooksLikeHtml(prefix)) + { + // A dead link answered with an error page and a 200 is the normal behaviour of several + // large image hosts; without this the store fills with copies of their apology page. + return (SeenOutcome.NotMedia, "NotMedia"); + } + + if (kind == MediaKind.Unknown) + { + return (SeenOutcome.NotMedia, "NotMedia"); + } + + return options.AllowedKinds.Contains(kind) ? null : (SeenOutcome.UnsupportedType, "UnsupportedType"); + } + + private static HttpRequestMessage BuildRequest(Uri url, Uri? referer, FetchOptions options) + { + var request = new HttpRequestMessage(HttpMethod.Get, url); + + request.Headers.UserAgent.ParseAdd(options.UserAgent); + request.Headers.Accept.ParseAdd("image/avif,image/webp,image/apng,image/*,video/*;q=0.8,*/*;q=0.5"); + + if (referer is not null) + { + request.Headers.Referrer = referer; + } + + return request; + } + + private static bool IsRedirect(HttpStatusCode status) => + status + is HttpStatusCode.MovedPermanently + or HttpStatusCode.Found + or HttpStatusCode.SeeOther + or HttpStatusCode.TemporaryRedirect + or HttpStatusCode.PermanentRedirect; + + /// Resolves a Location header, refusing anything that is not HTTP. + private static Uri? ResolveRedirect(Uri current, Uri? location) + { + if (location is null) + { + return null; + } + + var resolved = location.IsAbsoluteUri ? location : new Uri(current, location); + + // data:, file: and friends have no business being the target of a media fetch. + return resolved.Scheme is "http" or "https" ? resolved : null; + } + + private static TimeSpan? ReadRetryAfter(RetryConditionHeaderValue? header) + { + if (header is null) + { + return null; + } + + var raw = header.Delta is { } delta + ? ((int)delta.TotalSeconds).ToString(System.Globalization.CultureInfo.InvariantCulture) + : header.Date?.ToString("r", System.Globalization.CultureInfo.InvariantCulture); + + return RetryAfter.Parse(raw, DateTimeOffset.UtcNow, out _); + } + + private static FetchResult Failure( + Uri url, + SeenOutcome outcome, + string code, + string? detail, + Stopwatch stopwatch, + string? proxyKey = null, + int status = 0 + ) => + new(outcome, url) + { + ErrorCode = code, + ErrorDetail = detail, + HttpStatus = status, + ProxyKey = proxyKey, + Elapsed = stopwatch.Elapsed, + }; + + private void TryDelete(string path) + { + try + { + if (File.Exists(path)) + { + File.Delete(path); + } + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) + { + _logger.LogWarning(ex, "Could not remove the staged file {Path}", path); + } + } +} diff --git a/src/AvParser.Infrastructure/Collecting/MediaSignatures.cs b/src/AvParser.Infrastructure/Collecting/MediaSignatures.cs new file mode 100644 index 0000000..d2dd922 --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/MediaSignatures.cs @@ -0,0 +1,371 @@ +using System.Buffers.Binary; +using AvParser.Core.Collecting; + +namespace AvParser.Infrastructure.Collecting; + +/// +/// Decides what a byte sequence is by looking at the bytes. +/// +/// +/// +/// The URL, the extension and the Content-Type header are all wrong often enough to be +/// useless, and two of the three are chosen by whoever is serving the file. Everything the store +/// records about a type — including the extension it is written under — comes from here. +/// +/// +/// Kind detection needs only a short prefix. Animation does not: whether a PNG is an APNG depends +/// on a chunk that may sit anywhere before the first IDAT, and whether a GIF moves depends +/// on there being a second image descriptor. Those take , which is +/// given as much of the head of the file as the caller cares to read. +/// +/// +internal static class MediaSignatures +{ + /// Bytes needed before can decide. + public const int PrefixLength = 32; + + /// + /// How much of a file is worth re-reading to settle animation. + /// + /// + /// A bound rather than a correct answer: an APNG could in principle bury acTL past this, + /// but a megabyte of leading metadata does not occur in practice and an unbounded walk over a + /// hostile file does. + /// + public const int AnimationScanLimit = 1024 * 1024; + + /// PNG's eight-byte signature, chosen by the format to survive text-mode transfers. + private static ReadOnlySpan PngMagic => [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]; + + /// EBML magic, shared by WebM and Matroska. + private static ReadOnlySpan EbmlMagic => [0x1A, 0x45, 0xDF, 0xA3]; + + /// Identifies the content from its leading bytes. + public static MediaKind Detect(ReadOnlySpan prefix) + { + if (prefix.Length >= 3 && prefix[0] == 0xFF && prefix[1] == 0xD8 && prefix[2] == 0xFF) + { + return MediaKind.Jpeg; + } + + if (prefix.Length >= 8 && prefix[..8].SequenceEqual(PngMagic)) + { + return MediaKind.Png; + } + + if (prefix.Length >= 6 && (prefix[..6].SequenceEqual("GIF87a"u8) || prefix[..6].SequenceEqual("GIF89a"u8))) + { + return MediaKind.Gif; + } + + if (prefix.Length >= 12 && prefix[..4].SequenceEqual("RIFF"u8) && prefix.Slice(8, 4).SequenceEqual("WEBP"u8)) + { + return MediaKind.WebP; + } + + if (prefix.Length >= 12 && prefix.Slice(4, 4).SequenceEqual("ftyp"u8)) + { + return FromIsoBrand(prefix.Slice(8, 4)); + } + + if (prefix.Length >= 4 && prefix[..4].SequenceEqual(EbmlMagic)) + { + // EBML covers both WebM and Matroska; only the DocType tells them apart. + return IsWebmDocType(prefix) ? MediaKind.WebM : MediaKind.Unknown; + } + + return MediaKind.Unknown; + } + + /// + /// Whether the prefix looks like a web page rather than media. + /// + /// + /// The single cheapest defence against a dead link answered with 200 and an error page, which + /// is the normal behaviour of several large image hosts. + /// + public static bool LooksLikeHtml(ReadOnlySpan prefix) + { + var start = 0; + + if (prefix.Length >= 3 && prefix[0] == 0xEF && prefix[1] == 0xBB && prefix[2] == 0xBF) + { + start = 3; + } + + while (start < prefix.Length && prefix[start] is (byte)' ' or (byte)'\t' or (byte)'\r' or (byte)'\n') + { + start++; + } + + var body = prefix[start..]; + + return StartsWithIgnoringCase(body, " + /// Decides whether the content has more than one frame. + /// + /// Type already established by . + /// Leading bytes, ideally up to . + /// when the content is known to animate. + public static bool DetectAnimation(MediaKind kind, ReadOnlySpan head) => + kind switch + { + MediaKind.Gif => GifHasSecondFrame(head), + MediaKind.Png => PngHasAnimationControl(head), + MediaKind.WebP => WebpHasAnimationFlag(head), + _ => false, + }; + + /// Reads pixel dimensions when the header makes it cheap. + /// JPEG is absent on purpose: it needs a walk over segments, and nothing needs it yet. + public static (int? Width, int? Height) ReadDimensions(MediaKind kind, ReadOnlySpan prefix) => + kind switch + { + MediaKind.Png when prefix.Length >= 24 && prefix.Slice(12, 4).SequenceEqual("IHDR"u8) => ( + (int)BinaryPrimitives.ReadUInt32BigEndian(prefix.Slice(16, 4)), + (int)BinaryPrimitives.ReadUInt32BigEndian(prefix.Slice(20, 4)) + ), + MediaKind.Gif when prefix.Length >= 10 => ( + BinaryPrimitives.ReadUInt16LittleEndian(prefix.Slice(6, 2)), + BinaryPrimitives.ReadUInt16LittleEndian(prefix.Slice(8, 2)) + ), + // Only the extended WebP form carries a canvas size in a fixed place. + MediaKind.WebP when prefix.Length >= 30 && prefix.Slice(12, 4).SequenceEqual("VP8X"u8) => ( + ReadUInt24LittleEndian(prefix.Slice(24, 3)) + 1, + ReadUInt24LittleEndian(prefix.Slice(27, 3)) + 1 + ), + _ => (null, null), + }; + + private static MediaKind FromIsoBrand(ReadOnlySpan brand) + { + if (brand.SequenceEqual("avif"u8) || brand.SequenceEqual("avis"u8)) + { + return MediaKind.Avif; + } + + return + brand.SequenceEqual("isom"u8) + || brand.SequenceEqual("iso2"u8) + || brand.SequenceEqual("iso4"u8) + || brand.SequenceEqual("mp41"u8) + || brand.SequenceEqual("mp42"u8) + || brand.SequenceEqual("avc1"u8) + || brand.SequenceEqual("dash"u8) + || brand.SequenceEqual("M4V "u8) + ? MediaKind.Mp4 + : MediaKind.Unknown; + } + + /// Looks for the EBML DocType element and checks it says "webm". + private static bool IsWebmDocType(ReadOnlySpan prefix) + { + // Element id 0x4282 within the EBML header. Scanning the first few dozen bytes for it is + // cheaper and far shorter than a general EBML parser, and the header is tiny by spec. + for (var offset = 4; offset + 3 < prefix.Length && offset < 64; offset++) + { + if (prefix[offset] != 0x42 || prefix[offset + 1] != 0x82) + { + continue; + } + + var length = prefix[offset + 2]; + var start = offset + 3; + + if (length is 0 or > 16 || start + length > prefix.Length) + { + return false; + } + + return prefix.Slice(start, length).StartsWith("webm"u8); + } + + return false; + } + + /// A GIF animates when it carries more than one image descriptor. + /// + /// The GIF89a version marker proves nothing on its own — plenty of still images use it — + /// so the block structure has to actually be walked. + /// + private static bool GifHasSecondFrame(ReadOnlySpan data) + { + if (data.Length < 13) + { + return false; + } + + var packed = data[10]; + var offset = 13; + + if ((packed & 0x80) != 0) + { + offset += 3 * (1 << ((packed & 0x07) + 1)); + } + + var frames = 0; + + while (offset < data.Length) + { + var block = data[offset++]; + + if (block == 0x3B) + { + break; + } + + if (block == 0x21) + { + if (offset >= data.Length) + { + break; + } + + offset++; + offset = SkipSubBlocks(data, offset); + + if (offset < 0) + { + break; + } + + continue; + } + + if (block != 0x2C) + { + // Something unexpected: stop rather than guess our way through a malformed file. + break; + } + + if (++frames >= 2) + { + return true; + } + + if (offset + 9 > data.Length) + { + break; + } + + var localPacked = data[offset + 8]; + offset += 9; + + if ((localPacked & 0x80) != 0) + { + offset += 3 * (1 << ((localPacked & 0x07) + 1)); + } + + if (offset >= data.Length) + { + break; + } + + offset++; + offset = SkipSubBlocks(data, offset); + + if (offset < 0) + { + break; + } + } + + return frames >= 2; + } + + private static int SkipSubBlocks(ReadOnlySpan data, int offset) + { + while (offset < data.Length) + { + var size = data[offset++]; + + if (size == 0) + { + return offset; + } + + offset += size; + } + + return -1; + } + + /// A PNG is an APNG when acTL appears before the first IDAT. + private static bool PngHasAnimationControl(ReadOnlySpan data) + { + var offset = 8; + + while (offset + 8 <= data.Length) + { + var length = BinaryPrimitives.ReadUInt32BigEndian(data.Slice(offset, 4)); + var type = data.Slice(offset + 4, 4); + + if (type.SequenceEqual("acTL"u8)) + { + return true; + } + + if (type.SequenceEqual("IDAT"u8)) + { + // Past this point an acTL would be ignored by decoders, so it is not an APNG. + return false; + } + + if (length > int.MaxValue - 12) + { + return false; + } + + offset += 12 + (int)length; + } + + return false; + } + + /// An extended WebP declares animation in its VP8X flag byte. + private static bool WebpHasAnimationFlag(ReadOnlySpan data) + { + if (data.Length < 21 || !data.Slice(12, 4).SequenceEqual("VP8X"u8)) + { + // Plain VP8 and VP8L are single frames by definition. + return false; + } + + return (data[20] & 0x02) != 0; + } + + private static int ReadUInt24LittleEndian(ReadOnlySpan value) => + value[0] | (value[1] << 8) | (value[2] << 16); + + private static bool StartsWithIgnoringCase(ReadOnlySpan value, ReadOnlySpan prefix) + { + if (value.Length < prefix.Length) + { + return false; + } + + for (var index = 0; index < prefix.Length; index++) + { + var left = value[index]; + var right = prefix[index]; + + if (left >= 'A' && left <= 'Z') + { + left += 32; + } + + if (left != right) + { + return false; + } + } + + return true; + } +} diff --git a/src/AvParser.Infrastructure/Collecting/RetryAfter.cs b/src/AvParser.Infrastructure/Collecting/RetryAfter.cs new file mode 100644 index 0000000..bcfa55a --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/RetryAfter.cs @@ -0,0 +1,81 @@ +using System.Globalization; + +namespace AvParser.Infrastructure.Collecting; + +/// +/// Reads the Retry-After header an origin sends with a 429 or a 503. +/// +/// +/// The header comes in two forms, and the date form is measured against the server's clock +/// while we compare it to ours. A few minutes of skew turns a two-second pause into a wait past any +/// reasonable patience, so the result is clamped and the clamp is reported — silently waiting an +/// hour looks exactly like a hang. +/// +internal static class RetryAfter +{ + /// Never pause for less than this; a zero-second header is not a licence to hammer. + public static readonly TimeSpan Minimum = TimeSpan.FromSeconds(1); + + /// Never pause for more than this, whatever the header says. + public static readonly TimeSpan Maximum = TimeSpan.FromSeconds(120); + + /// Parses the header value. + /// Raw header text: delta-seconds or an HTTP date. + /// Current time, for the date form. + /// Whether the value had to be pulled into range. + /// How long to wait, or when the header made no sense. + public static TimeSpan? Parse(string? value, DateTimeOffset now, out bool clamped) + { + clamped = false; + + if (string.IsNullOrWhiteSpace(value)) + { + return null; + } + + var text = value.Trim(); + + if (long.TryParse(text, NumberStyles.Integer, CultureInfo.InvariantCulture, out var seconds)) + { + // Negative and absurd values both occur; both mean "the origin is not being helpful". + return Clamp(TimeSpan.FromSeconds(Math.Clamp(seconds, 0, 86_400)), out clamped); + } + + if (DateTimeOffset.TryParse(text, CultureInfo.InvariantCulture, DateTimeStyles.AdjustToUniversal, out var when)) + { + var delta = when - now; + + return delta <= TimeSpan.Zero ? Clamp(TimeSpan.Zero, out clamped) : Clamp(delta, out clamped); + } + + return null; + } + + /// Backoff to use when the origin rate-limits without saying for how long. + /// How many refusals in a row this host has produced. + public static TimeSpan Backoff(int consecutive) + { + var exponent = Math.Clamp(consecutive - 1, 0, 8); + var seconds = 2d * Math.Pow(2, exponent); + + return TimeSpan.FromSeconds(Math.Min(seconds, Maximum.TotalSeconds)); + } + + private static TimeSpan Clamp(TimeSpan value, out bool clamped) + { + if (value < Minimum) + { + clamped = value != Minimum; + return Minimum; + } + + if (value > Maximum) + { + clamped = true; + return Maximum; + } + + clamped = false; + return value; + } +} diff --git a/src/AvParser.Infrastructure/Proxies/ProxiedHttpClientFactory.cs b/src/AvParser.Infrastructure/Proxies/ProxiedHttpClientFactory.cs index c6e34d4..ea56ed3 100644 --- a/src/AvParser.Infrastructure/Proxies/ProxiedHttpClientFactory.cs +++ b/src/AvParser.Infrastructure/Proxies/ProxiedHttpClientFactory.cs @@ -2,11 +2,76 @@ using AvParser.Core.Proxies; namespace AvParser.Infrastructure.Proxies; +/// +/// The three deadlines a download actually needs. +/// +/// +/// One timeout is not enough, and using one for everything is actively wrong: HttpClient.Timeout +/// is a deadline on the whole response, so any single value large enough for a 30 MB file is +/// also large enough for a dead connection to hang on, and any value small enough to catch the dead +/// connection kills the download. Splitting it lets both be strict. +/// +/// Establishing the TCP or CONNECT tunnel. +/// Waiting for the response line and headers. +/// Longest gap between two body reads before the transfer is called stalled. +public readonly record struct HttpClientTimeouts(TimeSpan Connect, TimeSpan Headers, TimeSpan Idle) +{ + /// Reasonable defaults for fetching media. + public static HttpClientTimeouts Default { get; } = + new(TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(20)); +} + +/// Thrown when proxy-only operation was demanded and the pool had nothing to give. +/// +/// An exception rather than a null lease because the failure mode it prevents is silent: the +/// factory would otherwise hand back a perfectly working direct client, and the request +/// would go out from the user's own address precisely when they asked for it not to. +/// +public sealed class ProxyUnavailableException : Exception +{ + /// Creates the exception with the default message. + public ProxyUnavailableException() + : base("No live proxy is available and direct connections are not allowed.") { } + + /// Creates the exception with a specific message. + public ProxyUnavailableException(string message) + : base(message) { } + + /// Creates the exception with a message and a cause. + public ProxyUnavailableException(string message, Exception innerException) + : base(message, innerException) { } +} + +/// An and the proxy lease it is bound to, owned together. +/// +/// Paired so a caller cannot dispose one and leak the other. Disposing without having reported an +/// outcome leaves the pool none the wiser, which is the correct neutral behaviour for cancellation. +/// +public sealed class LeasedHttpClient(HttpClient client, ProxyLease? lease) : IDisposable +{ + /// The client. Routed through when there is one. + public HttpClient Client { get; } = client ?? throw new ArgumentNullException(nameof(client)); + + /// The proxy in use, or when the connection is direct. + public ProxyLease? Lease { get; } = lease; + + /// Address of the proxy in use, for recording provenance. + public string? ProxyKey => Lease?.Endpoint.Key; + + /// + public void Dispose() + { + Client.Dispose(); + + // Safe after an explicit verdict: the lease records only the first report it is given. + Lease?.Dispose(); + } +} + /// Creates instances bound to a specific proxy. /// /// Not IHttpClientFactory: that exists to share handlers across requests, and a per-proxy -/// handler is the opposite of shareable. This is the seam the parser layer will use once it -/// starts making real requests. +/// handler is the opposite of shareable. /// public interface IProxiedHttpClientFactory { @@ -26,6 +91,30 @@ public interface IProxiedHttpClientFactory TimeSpan? timeout = null, CancellationToken cancellationToken = default ); + + /// Creates a client with the three deadlines set separately. + /// + /// is left infinite on purpose; the caller enforces the header + /// and idle deadlines with linked tokens so a long download is not mistaken for a hang. + /// + HttpClient Create(ProxyEndpoint? endpoint, HttpClientTimeouts timeouts); + + /// + /// Takes a proxy from the pool and returns it paired with a client. + /// + /// Deadlines for the client. + /// + /// When set, having no live proxy throws instead of quietly connecting directly. + /// + /// Cancels acquisition. + /// + /// was set and the pool had nothing live. + /// + Task LeaseAsync( + HttpClientTimeouts timeouts, + bool requireProxy, + CancellationToken cancellationToken = default + ); } /// @@ -53,4 +142,32 @@ public sealed class ProxiedHttpClientFactory(IProxyPool pool) : IProxiedHttpClie var lease = await _pool.AcquireAsync(cancellationToken).ConfigureAwait(false); return (Create(lease?.Endpoint, timeout), lease); } + + /// + public HttpClient Create(ProxyEndpoint? endpoint, HttpClientTimeouts timeouts) + { + var handler = ProxyHandlerFactory.CreateHandler(endpoint, timeouts.Connect); + + // Infinite on purpose: this deadline covers the entire response including the body, so any + // value large enough for real media is useless as a liveness check. The caller enforces the + // header and idle deadlines separately, which catches hangs without killing big downloads. + return new HttpClient(handler, disposeHandler: true) { Timeout = Timeout.InfiniteTimeSpan }; + } + + /// + public async Task LeaseAsync( + HttpClientTimeouts timeouts, + bool requireProxy, + CancellationToken cancellationToken = default + ) + { + var lease = await _pool.AcquireAsync(cancellationToken).ConfigureAwait(false); + + if (lease is null && requireProxy) + { + throw new ProxyUnavailableException(); + } + + return new LeasedHttpClient(Create(lease?.Endpoint, timeouts), lease); + } } diff --git a/tests/AvParser.Infrastructure.Tests/Collecting/LeaseVerdictTests.cs b/tests/AvParser.Infrastructure.Tests/Collecting/LeaseVerdictTests.cs new file mode 100644 index 0000000..dab4cb6 --- /dev/null +++ b/tests/AvParser.Infrastructure.Tests/Collecting/LeaseVerdictTests.cs @@ -0,0 +1,159 @@ +using System.Net; +using System.Net.Sockets; +using System.Security.Authentication; +using AvParser.Infrastructure.Collecting; + +namespace AvParser.Infrastructure.Tests.Collecting; + +/// +/// The verdict table, tested as a pure function. +/// +/// +/// It has to be a pure function to be testable at all: ProxyLease's constructor is internal +/// to AvParser.Core, which declares no InternalsVisibleTo, so nothing outside the +/// domain assembly can fabricate one to assert against. +/// +public class LeaseVerdictTests +{ + [Theory] + [InlineData(200)] + [InlineData(204)] + public void A_delivered_response_is_a_working_proxy(int status) => + Decide(status).Verdict.ShouldBe(LeaseVerdict.Success); + + [Theory] + [InlineData(400)] + [InlineData(403)] + [InlineData(404)] + [InlineData(410)] + [InlineData(451)] + [InlineData(500)] + [InlineData(503)] + public void The_resource_being_broken_is_not_the_proxy_being_broken(int status) + { + // A dead link is the normal case on a free list. Recording it against the proxy would + // quarantine working proxies at exactly the rate that dead links appear. + Decide(status).Verdict.ShouldBe(LeaseVerdict.Success); + } + + [Fact] + public void A_rate_limit_is_never_the_proxy_s_fault() + { + // And treating it as one would make the pool rotate away from a working address in + // response to being asked politely to slow down. + Decide(429).Verdict.ShouldBe(LeaseVerdict.Success); + } + + [Theory] + [InlineData(407, "proxy auth")] + [InlineData(502, "bad gateway")] + [InlineData(504, "gateway timeout")] + public void Statuses_the_proxy_itself_produces_count_against_it(int status, string reason) + { + var outcome = Decide(status); + + outcome.Verdict.ShouldBe(LeaseVerdict.Failure); + outcome.Reason.ShouldBe(reason); + } + + [Fact] + public void Cancellation_teaches_the_pool_nothing() + { + // The deliberate neutral case: pressing Stop must not quarantine healthy proxies. + LeaseVerdicts.Decide(null, null, bytesReceived: true, cancelled: true).Verdict.ShouldBe(LeaseVerdict.Neutral); + LeaseVerdicts + .Decide(HttpStatusCode.OK, new OperationCanceledException(), true, false) + .Verdict.ShouldBe(LeaseVerdict.Neutral); + } + + [Fact] + public void No_response_at_all_is_a_failure() => + LeaseVerdicts.Decide(null, null, false, false).Verdict.ShouldBe(LeaseVerdict.Failure); + + [Fact] + public void A_refused_connection_is_a_failure() + { + var failure = new HttpRequestException("boom", new SocketException((int)SocketError.ConnectionRefused)); + var outcome = LeaseVerdicts.Decide(null, failure, false, false); + + outcome.Verdict.ShouldBe(LeaseVerdict.Failure); + outcome.Reason.ShouldBe("refused"); + } + + [Fact] + public void A_tls_handshake_failure_names_itself() + { + var outcome = LeaseVerdicts.Decide(null, new AuthenticationException("no"), false, false); + + outcome.Verdict.ShouldBe(LeaseVerdict.Failure); + outcome.Reason.ShouldBe("tls"); + } + + [Fact] + public void A_timeout_before_any_bytes_reads_differently_from_one_after() + { + LeaseVerdicts.Decide(null, new TimeoutException(), false, false).Reason.ShouldBe("timeout"); + LeaseVerdicts.Decide(null, new TimeoutException(), true, false).Reason.ShouldBe("stalled"); + } + + [Fact] + public void A_body_cut_short_counts_against_the_proxy() + { + var outcome = LeaseVerdicts.Decide(HttpStatusCode.OK, new IOException("short"), true, false); + + outcome.Verdict.ShouldBe(LeaseVerdict.Failure); + outcome.Reason.ShouldBe("truncated"); + } + + private static LeaseOutcome Decide(int status) => + LeaseVerdicts.Decide((HttpStatusCode)status, null, bytesReceived: true, cancelled: false); +} + +public class RetryAfterTests +{ + private static readonly DateTimeOffset Now = new(2026, 8, 13, 12, 0, 0, TimeSpan.Zero); + + [Fact] + public void Delta_seconds_are_read() + { + RetryAfter.Parse("30", Now, out var clamped).ShouldBe(TimeSpan.FromSeconds(30)); + clamped.ShouldBeFalse(); + } + + [Fact] + public void An_http_date_is_read_relative_to_now() => + RetryAfter.Parse(Now.AddSeconds(45).ToString("r"), Now, out _).ShouldBe(TimeSpan.FromSeconds(45)); + + [Fact] + public void An_absurd_wait_is_pulled_back_into_range() + { + // A date measured against a skewed server clock otherwise turns a short pause into + // something indistinguishable from a hang. + RetryAfter.Parse("86400", Now, out var clamped).ShouldBe(RetryAfter.Maximum); + clamped.ShouldBeTrue(); + } + + [Fact] + public void A_date_already_in_the_past_still_pauses_briefly() + { + RetryAfter.Parse(Now.AddMinutes(-5).ToString("r"), Now, out _).ShouldBe(RetryAfter.Minimum); + RetryAfter.Parse("0", Now, out _).ShouldBe(RetryAfter.Minimum); + RetryAfter.Parse("-30", Now, out _).ShouldBe(RetryAfter.Minimum); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData(" ")] + [InlineData("soon")] + public void Nonsense_reads_as_absent(string? value) => RetryAfter.Parse(value, Now, out _).ShouldBeNull(); + + [Fact] + public void Backoff_grows_and_then_stops_growing() + { + RetryAfter.Backoff(1).ShouldBe(TimeSpan.FromSeconds(2)); + RetryAfter.Backoff(2).ShouldBe(TimeSpan.FromSeconds(4)); + RetryAfter.Backoff(3).ShouldBe(TimeSpan.FromSeconds(8)); + RetryAfter.Backoff(20).ShouldBe(RetryAfter.Maximum); + } +} diff --git a/tests/AvParser.Infrastructure.Tests/Collecting/LoopbackServer.cs b/tests/AvParser.Infrastructure.Tests/Collecting/LoopbackServer.cs new file mode 100644 index 0000000..d2c7679 --- /dev/null +++ b/tests/AvParser.Infrastructure.Tests/Collecting/LoopbackServer.cs @@ -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; + +/// One request as the server saw it. +/// Requested path. +/// Request headers, lower-cased keys. +/// When the request line was read. +/// When the response finished being written. +internal sealed record RecordedRequest( + string Path, + IReadOnlyDictionary Headers, + DateTimeOffset ArrivedUtc, + DateTimeOffset CompletedUtc +); + +/// How the server should answer one path. +internal sealed record Reply +{ + /// Status code to send. + public int Status { get; init; } = 200; + + /// Body bytes. Ignored for statuses that carry none. + public byte[] Body { get; init; } = []; + + /// Value for the Content-Type header, if any. + public string? ContentType { get; init; } + + /// Extra headers, sent verbatim. + public IReadOnlyDictionary Headers { get; init; } = + new Dictionary(StringComparer.OrdinalIgnoreCase); + + /// Send the body with chunked transfer encoding instead of a length. + public bool Chunked { get; init; } + + /// Compress the body and declare Content-Encoding: gzip. + public bool Gzip { get; init; } + + /// Declare this length instead of the real one — for testing truncation. + public long? DeclaredLength { get; init; } + + /// Write only this many body bytes, then close the connection mid-transfer. + public int? TruncateAfter { get; init; } + + /// Wait this long before sending the status line. + public TimeSpan HeaderDelay { get; init; } + + /// Wait this long between body writes. + public TimeSpan BodyDelay { get; init; } + + /// Bytes per body write, when dripping. + public int BodyChunkSize { get; init; } = int.MaxValue; +} + +/// +/// A deliberately badly behaved HTTP server on loopback. +/// +/// +/// +/// Raw sockets rather than HttpListener 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. +/// +/// +/// 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. +/// +/// +internal sealed class LoopbackServer : IAsyncDisposable +{ + private readonly TcpListener _listener; + private readonly CancellationTokenSource _shutdown = new(); + private readonly ConcurrentDictionary> _routes = new(StringComparer.Ordinal); + private readonly ConcurrentBag _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); + } + + /// Root of this server. + public Uri BaseAddress { get; } + + /// Every request served, in no particular order. + public IReadOnlyCollection Requests => [.. _requests]; + + /// How many requests were served. + public int RequestCount => _requests.Count; + + /// Registers a fixed reply for a path. + public Uri Map(string path, Reply reply) => Map(path, () => reply); + + /// Registers a reply computed per request, for sequences. + public Uri Map(string path, Func reply) + { + _routes[path] = reply; + + return new Uri(BaseAddress, path); + } + + /// Registers a body served as-is with a 200. + public Uri MapBody(string path, byte[] body, string? contentType = null) => + Map(path, new Reply { Body = body, ContentType = contentType }); + + /// + 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 Headers)> ReadRequestAsync(NetworkStream stream) + { + var buffer = new List(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(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", + }; +} diff --git a/tests/AvParser.Infrastructure.Tests/Collecting/MediaFetcherTests.cs b/tests/AvParser.Infrastructure.Tests/Collecting/MediaFetcherTests.cs new file mode 100644 index 0000000..94da8e4 --- /dev/null +++ b/tests/AvParser.Infrastructure.Tests/Collecting/MediaFetcherTests.cs @@ -0,0 +1,571 @@ +using System.Net; +using System.Text; +using AvParser.Core.Collecting; +using AvParser.Core.Proxies; +using AvParser.Infrastructure.Collecting; +using AvParser.Infrastructure.Media; +using AvParser.Infrastructure.Proxies; +using AvParser.Infrastructure.Storage; +using Microsoft.Extensions.Logging.Abstractions; + +namespace AvParser.Infrastructure.Tests.Collecting; + +/// +/// A factory that never uses a proxy but mirrors the real handler's configuration. +/// +/// +/// Matching AllowAutoRedirect = false and AutomaticDecompression = All matters: with +/// redirects handled by the BCL these tests would exercise its logic rather than the fetcher's. +/// +internal sealed class DirectHttpClientFactory : IProxiedHttpClientFactory +{ + public HttpClient Create(ProxyEndpoint? endpoint, TimeSpan? timeout = null) => + Create(endpoint, HttpClientTimeouts.Default); + + public HttpClient Create(ProxyEndpoint? endpoint, HttpClientTimeouts timeouts) => + new( + new SocketsHttpHandler + { + AllowAutoRedirect = false, + ConnectTimeout = timeouts.Connect, + AutomaticDecompression = DecompressionMethods.All, + }, + disposeHandler: true + ) + { + Timeout = Timeout.InfiniteTimeSpan, + }; + + public Task<(HttpClient Client, ProxyLease? Lease)> CreateFromPoolAsync( + TimeSpan? timeout = null, + CancellationToken cancellationToken = default + ) => Task.FromResult<(HttpClient, ProxyLease?)>((Create(null, timeout), null)); + + public Task LeaseAsync( + HttpClientTimeouts timeouts, + bool requireProxy, + CancellationToken cancellationToken = default + ) => + requireProxy + ? throw new ProxyUnavailableException() + : Task.FromResult(new LeasedHttpClient(Create(null, timeouts), null)); +} + +public sealed class MediaFetcherTests : IAsyncLifetime +{ + private readonly string _root = Path.Combine(Path.GetTempPath(), "AvParserTests", Guid.NewGuid().ToString("N")); + + private LoopbackServer _server = null!; + private AppPaths _paths = null!; + private BlobStore _blobs = null!; + private HostThrottle _throttle = null!; + private MediaFetcher _fetcher = null!; + + public ValueTask InitializeAsync() + { + _paths = new AppPaths(_root); + _paths.EnsureCreated(); + + _server = new LoopbackServer(); + _blobs = new BlobStore(_paths, NullLogger.Instance); + _throttle = Throttle(); + _fetcher = new MediaFetcher( + new DirectHttpClientFactory(), + _blobs, + _throttle, + NullLogger.Instance + ); + + return ValueTask.CompletedTask; + } + + public async ValueTask DisposeAsync() + { + await _server.DisposeAsync(); + _throttle.Dispose(); + + if (Directory.Exists(_root)) + { + try + { + Directory.Delete(_root, recursive: true); + } + catch (IOException) + { + // Not worth failing a green test over. + } + } + } + + private HostThrottle Throttle(int concurrent = 4, int intervalMs = 0) => + new(concurrent, TimeSpan.FromMilliseconds(intervalMs), NullLogger.Instance); + + /// Options with the floor lowered, since the sample files are deliberately tiny. + private static FetchOptions Options(params (string Key, object Value)[] _) => + new() { MinItemBytes = 1, Timeouts = HttpClientTimeouts.Default }; + + private Task FetchAsync(Uri url, FetchOptions? options = null) => + _fetcher.FetchAsync( + new MediaCandidate(url) { SourceId = "test" }, + options ?? Options(), + TestContext.Current.CancellationToken + ); + + /// Waits for the server's own bookkeeping to catch up with the client. + private async Task> WaitForRequestsAsync(int count, Func match) + { + for (var attempt = 0; attempt < 100; attempt++) + { + var matched = _server.Requests.Where(match).OrderBy(r => r.ArrivedUtc).ToList(); + + if (matched.Count >= count) + { + return matched; + } + + await Task.Delay(20, TestContext.Current.CancellationToken); + } + + var found = _server.Requests.Where(match).OrderBy(r => r.ArrivedUtc).ToList(); + found.Count.ShouldBe(count); + + return found; + } + + private int StagedFileCount() => + Directory.Exists(_paths.MediaTempDirectory) ? Directory.EnumerateFiles(_paths.MediaTempDirectory).Count() : 0; + + [Fact] + public async Task A_valid_image_is_downloaded_and_hashed() + { + var url = _server.MapBody("/a.png", Samples.Png(), "image/png"); + + var result = await FetchAsync(url); + + result.Outcome.ShouldBe(SeenOutcome.Stored); + result.Blob!.Kind.ShouldBe(MediaKind.Png); + result.Blob.Sha256.Length.ShouldBe(64); + result.TempPath.ShouldNotBeNull(); + File.Exists(result.TempPath).ShouldBeTrue(); + } + + [Fact] + public async Task The_extension_comes_from_the_signature_not_from_the_content_type() + { + // The origin says PNG and serves a GIF. Believing the header would write a .png that no + // viewer opens, and would let a host dictate the extension of a file on the user's disk. + var url = _server.MapBody("/liar.png", Samples.Gif(), "image/png"); + + var result = await FetchAsync(url); + + result.Blob!.Kind.ShouldBe(MediaKind.Gif); + result.Blob.Extension.ShouldBe(".gif"); + result.ContentType.ShouldBe("image/png"); + } + + [Fact] + public async Task An_error_page_behind_a_200_is_refused() + { + var url = _server.MapBody("/gone.jpg", Samples.Html(), "image/jpeg"); + + var result = await FetchAsync(url); + + result.Outcome.ShouldBe(SeenOutcome.NotMedia); + result.TempPath.ShouldBeNull(); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_kind_the_user_excluded_is_refused() + { + var url = _server.MapBody("/clip.mp4", Samples.Mp4()); + var options = Options() with { AllowedKinds = new HashSet { MediaKind.Png } }; + + (await FetchAsync(url, options)).Outcome.ShouldBe(SeenOutcome.UnsupportedType); + } + + [Theory] + [InlineData(403, SeenOutcome.Failed)] + [InlineData(404, SeenOutcome.Gone)] + [InlineData(410, SeenOutcome.Gone)] + [InlineData(500, SeenOutcome.Failed)] + public async Task Error_statuses_are_reported_without_staging_anything(int status, SeenOutcome expected) + { + var url = _server.Map($"/e{status}", new Reply { Status = status, Body = "no"u8.ToArray() }); + + var result = await FetchAsync(url); + + result.Outcome.ShouldBe(expected); + result.HttpStatus.ShouldBe(status); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_rate_limit_is_honoured_rather_than_worked_around() + { + var url = _server.Map( + "/busy", + new Reply + { + Status = 429, + Headers = new Dictionary(StringComparer.OrdinalIgnoreCase) { ["Retry-After"] = "3" }, + } + ); + + var result = await FetchAsync(url); + + result.Outcome.ShouldBe(SeenOutcome.RateLimited); + result.RetryAfter.ShouldBe(TimeSpan.FromSeconds(3)); + + // The host is now cooling: the collector slows down instead of switching proxy. + _throttle.CooldownRemaining(url).ShouldBeGreaterThan(TimeSpan.Zero); + } + + [Fact] + public async Task A_service_unavailable_without_a_hint_still_backs_off() + { + var url = _server.Map("/down", new Reply { Status = 503 }); + + (await FetchAsync(url)).Outcome.ShouldBe(SeenOutcome.RateLimited); + _throttle.CooldownRemaining(url).ShouldBeGreaterThan(TimeSpan.Zero); + } + + [Fact] + public async Task Headers_that_never_arrive_time_out() + { + var url = _server.Map("/slowhead", new Reply { Body = Samples.Png(), HeaderDelay = TimeSpan.FromSeconds(3) }); + + var options = Options() with + { + Timeouts = HttpClientTimeouts.Default with { Headers = TimeSpan.FromMilliseconds(300) }, + }; + + (await FetchAsync(url, options)).Outcome.ShouldBe(SeenOutcome.Timeout); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_body_that_stops_arriving_is_abandoned_and_leaves_nothing_behind() + { + var url = _server.Map( + "/drip", + new Reply + { + Body = Encoding.ASCII.GetBytes(new string('x', 4096)), + BodyChunkSize = 8, + BodyDelay = TimeSpan.FromSeconds(2), + } + ); + + var options = Options() with + { + Timeouts = HttpClientTimeouts.Default with { Idle = TimeSpan.FromMilliseconds(300) }, + }; + + var result = await FetchAsync(url, options); + + result.Outcome.ShouldBe(SeenOutcome.Timeout); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_body_shorter_than_declared_is_a_truncated_transfer_not_a_small_file() + { + // The single most important rejection: a partial image promoted into the blob store would + // be indistinguishable from a real one for ever afterwards. + var body = Samples.Png(); + var url = _server.Map( + "/short", + new Reply + { + Body = body, + DeclaredLength = body.Length + 5000, + TruncateAfter = body.Length, + } + ); + + var result = await FetchAsync(url); + + result.Outcome.ShouldBe(SeenOutcome.Failed); + result.ErrorCode.ShouldBe("Truncated"); + result.TempPath.ShouldBeNull(); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_declared_length_over_the_cap_is_refused_before_the_body_is_read() + { + var body = Encoding.ASCII.GetBytes(new string('x', 8192)); + var url = _server.Map("/huge", new Reply { Body = body }); + var options = Options() with { MaxItemBytes = 1024 }; + + var result = await FetchAsync(url, options); + + result.Outcome.ShouldBe(SeenOutcome.TooLarge); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_chunked_body_that_grows_past_the_cap_is_cut_off() + { + // No Content-Length to check against, so the cap has to be enforced as bytes arrive. + var body = Encoding.ASCII.GetBytes(new string('x', 8192)); + var url = _server.Map( + "/huge-chunked", + new Reply + { + Body = body, + Chunked = true, + BodyChunkSize = 256, + } + ); + var options = Options() with { MaxItemBytes = 1024 }; + + (await FetchAsync(url, options)).Outcome.ShouldBe(SeenOutcome.TooLarge); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_well_formed_chunked_body_is_accepted() + { + var url = _server.Map( + "/chunked.png", + new Reply + { + Body = Samples.Png(), + Chunked = true, + BodyChunkSize = 7, + } + ); + + var result = await FetchAsync(url); + + result.Outcome.ShouldBe(SeenOutcome.Stored); + result.Blob!.Length.ShouldBe(Samples.Png().Length); + } + + [Fact] + public async Task The_same_image_served_gzipped_and_plain_is_one_blob() + { + // The handler decompresses transparently, so the hash must be over the decoded bytes. + var plain = _server.MapBody("/plain.png", Samples.Png()); + var zipped = _server.Map("/zipped.png", new Reply { Body = Samples.Png(), Gzip = true }); + + var first = await FetchAsync(plain); + var second = await FetchAsync(zipped); + + first.Blob!.Sha256.ShouldBe(second.Blob!.Sha256); + first.Blob.Length.ShouldBe(second.Blob.Length); + } + + [Fact] + public async Task A_tracking_pixel_is_below_the_floor() + { + var url = _server.MapBody("/pixel.gif", Samples.Gif()); + + // The default floor, rather than the lowered one the other tests use. + var result = await FetchAsync(url, new FetchOptions()); + + result.Outcome.ShouldBe(SeenOutcome.TooSmall); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task A_redirect_chain_is_followed_and_the_final_address_recorded() + { + var final = _server.MapBody("/final.png", Samples.Png()); + _server.Map("/hop2", Redirect(302, "/final.png")); + var start = _server.Map("/hop1", Redirect(301, "/hop2")); + + var result = await FetchAsync(start); + + result.Outcome.ShouldBe(SeenOutcome.Stored); + result.FinalUrl.ShouldBe(final); + } + + [Fact] + public async Task A_relative_location_is_resolved_against_the_current_address() + { + _server.MapBody("/other.png", Samples.Png()); + var start = _server.Map("/rel", Redirect(302, "/other.png")); + + (await FetchAsync(start)).Outcome.ShouldBe(SeenOutcome.Stored); + } + + [Fact] + public async Task A_redirect_loop_is_broken() + { + _server.Map("/loopB", Redirect(302, "/loopA")); + var start = _server.Map("/loopA", Redirect(302, "/loopB")); + + var result = await FetchAsync(start); + + result.ErrorCode.ShouldBe("RedirectLoop"); + } + + [Fact] + public async Task A_chain_longer_than_the_cap_gives_up() + { + // Distinct hops, so this is the hop budget rather than the loop detector doing the work. + for (var hop = 0; hop < 10; hop++) + { + _server.Map($"/chain{hop}", Redirect(302, $"/chain{hop + 1}")); + } + + var result = await FetchAsync(new Uri(_server.BaseAddress, "/chain0"), Options() with { MaxRedirects = 3 }); + + result.ErrorCode.ShouldBe("TooManyRedirects"); + } + + [Fact] + public async Task A_redirect_to_a_non_http_scheme_is_refused() + { + var start = _server.Map("/evil", Redirect(302, "data:image/png;base64,AAAA")); + + (await FetchAsync(start)).ErrorCode.ShouldBe("BadRedirect"); + } + + [Fact] + public async Task A_known_placeholder_is_recognised_by_its_hash() + { + var url = _server.MapBody("/dead.png", Samples.Png()); + var probe = await FetchAsync(url); + File.Delete(probe.TempPath!); + + var options = Options() with + { + Tombstones = new HashSet(StringComparer.Ordinal) { probe.Blob!.Sha256 }, + }; + + var result = await FetchAsync(url, options); + + result.Outcome.ShouldBe(SeenOutcome.Placeholder); + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task Animation_is_detected_end_to_end() + { + var still = _server.MapBody("/still.gif", Samples.Gif(frames: 1)); + var moving = _server.MapBody("/moving.gif", Samples.Gif(frames: 3)); + + (await FetchAsync(still)).Blob!.IsAnimated.ShouldBeFalse(); + (await FetchAsync(moving)).Blob!.IsAnimated.ShouldBeTrue(); + } + + [Fact] + public async Task Dimensions_are_recorded_when_the_header_carries_them() + { + var url = _server.MapBody("/big.png", Samples.Png(1024, 768)); + + var blob = (await FetchAsync(url)).Blob!; + + blob.Width.ShouldBe(1024); + blob.Height.ShouldBe(768); + } + + [Fact] + public async Task A_referer_is_sent_when_the_candidate_carries_one() + { + var url = _server.MapBody("/ref.png", Samples.Png()); + var referer = new Uri("https://example.test/page"); + + await _fetcher.FetchAsync( + new MediaCandidate(url) { SourceId = "test", Referer = referer }, + Options(), + TestContext.Current.CancellationToken + ); + + var request = _server.Requests.Single(r => r.Path == "/ref.png"); + request.Headers["Referer"].ShouldBe(referer.AbsoluteUri); + request.Headers["User-Agent"].ShouldContain("AvParser"); + } + + [Fact] + public async Task One_request_at_a_time_per_host_when_the_cap_says_so() + { + _throttle.Dispose(); + _throttle = Throttle(concurrent: 1); + _fetcher = new MediaFetcher( + new DirectHttpClientFactory(), + _blobs, + _throttle, + NullLogger.Instance + ); + + // Real media padded out, so neither request is refused after 32 bytes and cut short — + // the point of the test is two full responses that could have overlapped. + var body = (byte[])[.. Samples.Png(), .. new byte[2048]]; + var slow = new Reply + { + Body = body, + BodyChunkSize = 256, + BodyDelay = TimeSpan.FromMilliseconds(40), + }; + + _server.Map("/one", slow); + _server.Map("/two", slow); + + await Task.WhenAll( + FetchAsync(new Uri(_server.BaseAddress, "/one")), + FetchAsync(new Uri(_server.BaseAddress, "/two")) + ); + + // The server records a request from its own task, which can lag the client by a moment. + var requests = await WaitForRequestsAsync(2, r => r.Path is "/one" or "/two"); + + // Serialised, the second request cannot arrive until the first body has been written: + // nine 256-byte writes at 40 ms apiece. Run concurrently the gap would be near zero. + // Compared against arrival rather than completion because the server records completion + // from its own task, which can lag the client by a few milliseconds. + var gap = requests[1].ArrivedUtc - requests[0].ArrivedUtc; + gap.ShouldBeGreaterThan(TimeSpan.FromMilliseconds(200)); + } + + [Fact] + public async Task Cancelling_mid_download_leaves_no_staged_file() + { + // Real media, or the fetcher rightly rejects it after 32 bytes and there is nothing left + // to cancel. + var url = _server.Map( + "/cancel", + new Reply + { + Body = [.. Samples.Png(), .. new byte[65536]], + BodyChunkSize = 128, + BodyDelay = TimeSpan.FromMilliseconds(30), + } + ); + + using var cancellation = new CancellationTokenSource(); + var fetch = _fetcher.FetchAsync(new MediaCandidate(url) { SourceId = "test" }, Options(), cancellation.Token); + + await Task.Delay(200, TestContext.Current.CancellationToken); + await cancellation.CancelAsync(); + + await Should.ThrowAsync(async () => await fetch); + + StagedFileCount().ShouldBe(0); + } + + [Fact] + public async Task Demanding_a_proxy_that_is_not_there_fails_the_item_rather_than_going_direct() + { + // Otherwise the request leaves from the user's own address at precisely the moment they + // asked for it not to. + var url = _server.MapBody("/direct.png", Samples.Png()); + var options = Options() with { RequireProxy = true }; + + var result = await FetchAsync(url, options); + + result.ErrorCode.ShouldBe("NoProxy"); + _server.Requests.ShouldNotContain(r => r.Path == "/direct.png"); + } + + private static Reply Redirect(int status, string location) => + new() + { + Status = status, + Headers = new Dictionary(StringComparer.OrdinalIgnoreCase) { ["Location"] = location }, + }; +} diff --git a/tests/AvParser.Infrastructure.Tests/Collecting/MediaSignatureTests.cs b/tests/AvParser.Infrastructure.Tests/Collecting/MediaSignatureTests.cs new file mode 100644 index 0000000..4050b00 --- /dev/null +++ b/tests/AvParser.Infrastructure.Tests/Collecting/MediaSignatureTests.cs @@ -0,0 +1,383 @@ +using System.Buffers.Binary; +using System.Text; +using AvParser.Core.Collecting; +using AvParser.Infrastructure.Collecting; + +namespace AvParser.Infrastructure.Tests.Collecting; + +/// Builds the smallest byte sequences that are still legitimately each format. +internal static class Samples +{ + private static readonly byte[] PngSignature = [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]; + + public static byte[] Jpeg() => [0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10, .. "JFIF\0"u8, .. new byte[24]]; + + public static byte[] Png(int width = 640, int height = 480) + { + var bytes = new List(PngSignature); + bytes.AddRange(Chunk("IHDR", Ihdr(width, height))); + bytes.AddRange(Chunk("IDAT", new byte[8])); + + return [.. bytes]; + } + + public static byte[] Apng(int width = 64, int height = 64) + { + var bytes = new List(PngSignature); + bytes.AddRange(Chunk("IHDR", Ihdr(width, height))); + // acTL must precede IDAT to count; a decoder ignores it afterwards and so do we. + bytes.AddRange(Chunk("acTL", [0, 0, 0, 2, 0, 0, 0, 0])); + bytes.AddRange(Chunk("IDAT", new byte[8])); + + return [.. bytes]; + } + + /// A PNG carrying acTL only after IDAT — a still image with junk appended. + public static byte[] PngWithLateAnimationChunk() + { + var bytes = new List(PngSignature); + bytes.AddRange(Chunk("IHDR", Ihdr(8, 8))); + bytes.AddRange(Chunk("IDAT", new byte[8])); + bytes.AddRange(Chunk("acTL", [0, 0, 0, 2, 0, 0, 0, 0])); + + return [.. bytes]; + } + + public static byte[] Gif(int frames = 1, int width = 12, int height = 34, bool modern = true) + { + var bytes = new List(Encoding.ASCII.GetBytes(modern ? "GIF89a" : "GIF87a")); + + bytes.AddRange(LittleEndian16(width)); + bytes.AddRange(LittleEndian16(height)); + bytes.Add(0x00); // no global colour table, which keeps the offsets simple + bytes.Add(0x00); + bytes.Add(0x00); + + for (var frame = 0; frame < frames; frame++) + { + bytes.Add(0x2C); + bytes.AddRange(LittleEndian16(0)); + bytes.AddRange(LittleEndian16(0)); + bytes.AddRange(LittleEndian16(width)); + bytes.AddRange(LittleEndian16(height)); + bytes.Add(0x00); // no local colour table + bytes.Add(0x02); // LZW minimum code size + bytes.Add(0x03); // one sub-block of three bytes + bytes.AddRange([0x01, 0x02, 0x03]); + bytes.Add(0x00); // sub-block terminator + } + + bytes.Add(0x3B); + + return [.. bytes]; + } + + /// An animated GIF whose frames are separated by graphic control extensions. + public static byte[] GifWithExtensions() + { + var bytes = new List(Encoding.ASCII.GetBytes("GIF89a")); + bytes.AddRange(LittleEndian16(4)); + bytes.AddRange(LittleEndian16(4)); + bytes.Add(0x80 | 0x01); // global colour table, 2^(1+1) = 4 entries + bytes.Add(0x00); + bytes.Add(0x00); + bytes.AddRange(new byte[3 * 4]); + + for (var frame = 0; frame < 2; frame++) + { + bytes.Add(0x21); + bytes.Add(0xF9); + bytes.Add(0x04); + bytes.AddRange([0x00, 0x0A, 0x00, 0x00]); + bytes.Add(0x00); + + bytes.Add(0x2C); + bytes.AddRange(LittleEndian16(0)); + bytes.AddRange(LittleEndian16(0)); + bytes.AddRange(LittleEndian16(4)); + bytes.AddRange(LittleEndian16(4)); + bytes.Add(0x00); + bytes.Add(0x02); + bytes.Add(0x02); + bytes.AddRange([0x01, 0x02]); + bytes.Add(0x00); + } + + bytes.Add(0x3B); + + return [.. bytes]; + } + + public static byte[] WebpStill() + { + var body = new List("WEBP"u8.ToArray()); + body.AddRange("VP8 "u8.ToArray()); + body.AddRange(LittleEndian32(16)); + body.AddRange(new byte[16]); + + return Riff(body); + } + + public static byte[] WebpLossless() + { + var body = new List("WEBP"u8.ToArray()); + body.AddRange("VP8L"u8.ToArray()); + body.AddRange(LittleEndian32(16)); + body.AddRange(new byte[16]); + + return Riff(body); + } + + public static byte[] WebpExtended(bool animated, int width = 100, int height = 50) + { + var body = new List("WEBP"u8.ToArray()); + body.AddRange("VP8X"u8.ToArray()); + body.AddRange(LittleEndian32(10)); + body.Add((byte)(animated ? 0x02 : 0x10)); // ANIM, versus a plain alpha flag + body.AddRange([0x00, 0x00, 0x00]); + body.AddRange(LittleEndian24(width - 1)); + body.AddRange(LittleEndian24(height - 1)); + body.AddRange(new byte[8]); + + return Riff(body); + } + + public static byte[] Mp4(string brand = "isom") => + [0x00, 0x00, 0x00, 0x20, .. "ftyp"u8, .. Encoding.ASCII.GetBytes(brand), .. new byte[20]]; + + public static byte[] Avif() => Mp4("avif"); + + public static byte[] WebM() => Ebml("webm"); + + public static byte[] Matroska() => Ebml("matroska"); + + public static byte[] Html(string body = "Image not found") => + Encoding.UTF8.GetBytes(body); + + private static byte[] Ebml(string docType) + { + var bytes = new List { 0x1A, 0x45, 0xDF, 0xA3 }; + bytes.AddRange([0x9F, 0x42, 0x86, 0x81, 0x01]); // a plausible-looking EBML header preamble + bytes.Add(0x42); + bytes.Add(0x82); + bytes.Add((byte)docType.Length); + bytes.AddRange(Encoding.ASCII.GetBytes(docType)); + bytes.AddRange(new byte[16]); + + return [.. bytes]; + } + + private static byte[] Riff(List body) + { + var bytes = new List("RIFF"u8.ToArray()); + bytes.AddRange(LittleEndian32(body.Count)); + bytes.AddRange(body); + + return [.. bytes]; + } + + private static byte[] Ihdr(int width, int height) + { + var data = new byte[13]; + BinaryPrimitives.WriteUInt32BigEndian(data.AsSpan(0, 4), (uint)width); + BinaryPrimitives.WriteUInt32BigEndian(data.AsSpan(4, 4), (uint)height); + data[8] = 8; + + return data; + } + + private static byte[] Chunk(string type, byte[] data) + { + var bytes = new List(); + var length = new byte[4]; + BinaryPrimitives.WriteUInt32BigEndian(length, (uint)data.Length); + + bytes.AddRange(length); + bytes.AddRange(Encoding.ASCII.GetBytes(type)); + bytes.AddRange(data); + bytes.AddRange(new byte[4]); // CRC, never checked here + + return [.. bytes]; + } + + private static byte[] LittleEndian16(int value) => [(byte)(value & 0xFF), (byte)((value >> 8) & 0xFF)]; + + private static byte[] LittleEndian24(int value) => + [(byte)(value & 0xFF), (byte)((value >> 8) & 0xFF), (byte)((value >> 16) & 0xFF)]; + + private static byte[] LittleEndian32(int value) => + [(byte)(value & 0xFF), (byte)((value >> 8) & 0xFF), (byte)((value >> 16) & 0xFF), (byte)((value >> 24) & 0xFF)]; +} + +public class MediaSignatureTests +{ + [Fact] + public void Jpeg_is_recognised() => MediaSignatures.Detect(Samples.Jpeg()).ShouldBe(MediaKind.Jpeg); + + [Fact] + public void Png_is_recognised() => MediaSignatures.Detect(Samples.Png()).ShouldBe(MediaKind.Png); + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void Both_gif_versions_are_recognised(bool modern) => + MediaSignatures.Detect(Samples.Gif(modern: modern)).ShouldBe(MediaKind.Gif); + + [Fact] + public void Webp_is_recognised() => MediaSignatures.Detect(Samples.WebpStill()).ShouldBe(MediaKind.WebP); + + [Theory] + [InlineData("isom")] + [InlineData("mp42")] + [InlineData("avc1")] + public void Mp4_brands_are_recognised(string brand) => + MediaSignatures.Detect(Samples.Mp4(brand)).ShouldBe(MediaKind.Mp4); + + [Fact] + public void Avif_is_told_apart_from_mp4() => MediaSignatures.Detect(Samples.Avif()).ShouldBe(MediaKind.Avif); + + [Fact] + public void WebM_is_recognised() => MediaSignatures.Detect(Samples.WebM()).ShouldBe(MediaKind.WebM); + + [Fact] + public void Matroska_is_not_mistaken_for_webm() + { + // Both open with the same EBML magic; only the DocType separates them. + MediaSignatures.Detect(Samples.Matroska()).ShouldBe(MediaKind.Unknown); + } + + [Fact] + public void Anything_unrecognised_is_unknown() + { + MediaSignatures.Detect(Samples.Html()).ShouldBe(MediaKind.Unknown); + MediaSignatures.Detect(new byte[32]).ShouldBe(MediaKind.Unknown); + MediaSignatures.Detect([]).ShouldBe(MediaKind.Unknown); + MediaSignatures.Detect([0xFF]).ShouldBe(MediaKind.Unknown); + } + + [Fact] + public void A_truncated_prefix_never_throws() + { + var png = Samples.Png(); + + for (var length = 0; length < png.Length; length++) + { + _ = MediaSignatures.Detect(png.AsSpan(0, length)); + _ = MediaSignatures.DetectAnimation(MediaKind.Png, png.AsSpan(0, length)); + _ = MediaSignatures.ReadDimensions(MediaKind.Png, png.AsSpan(0, length)); + } + } + + [Theory] + [InlineData("")] + [InlineData("gone")] + [InlineData(" \n")] + [InlineData("")] + public void An_error_page_behind_a_200_is_spotted(string body) => + MediaSignatures.LooksLikeHtml(Samples.Html(body)).ShouldBeTrue(); + + [Fact] + public void Real_media_is_not_mistaken_for_html() + { + MediaSignatures.LooksLikeHtml(Samples.Png()).ShouldBeFalse(); + MediaSignatures.LooksLikeHtml(Samples.Gif()).ShouldBeFalse(); + MediaSignatures.LooksLikeHtml(Samples.Jpeg()).ShouldBeFalse(); + } + + [Fact] + public void A_single_frame_gif_does_not_animate() => + MediaSignatures.DetectAnimation(MediaKind.Gif, Samples.Gif(frames: 1)).ShouldBeFalse(); + + [Fact] + public void A_two_frame_gif_animates() => + MediaSignatures.DetectAnimation(MediaKind.Gif, Samples.Gif(frames: 2)).ShouldBeTrue(); + + [Fact] + public void A_gif_with_control_extensions_between_frames_animates() + { + // Real animated GIFs interleave graphic control extensions and use a global colour table; + // walking past both is where a naive scanner goes wrong. + MediaSignatures.DetectAnimation(MediaKind.Gif, Samples.GifWithExtensions()).ShouldBeTrue(); + } + + [Fact] + public void A_plain_png_does_not_animate() => + MediaSignatures.DetectAnimation(MediaKind.Png, Samples.Png()).ShouldBeFalse(); + + [Fact] + public void An_apng_animates() => MediaSignatures.DetectAnimation(MediaKind.Png, Samples.Apng()).ShouldBeTrue(); + + [Fact] + public void An_animation_chunk_after_the_first_idat_does_not_count() + { + // Decoders ignore acTL once IDAT has started, so claiming animation here would be a lie. + MediaSignatures.DetectAnimation(MediaKind.Png, Samples.PngWithLateAnimationChunk()).ShouldBeFalse(); + } + + [Fact] + public void A_plain_webp_does_not_animate() + { + MediaSignatures.DetectAnimation(MediaKind.WebP, Samples.WebpStill()).ShouldBeFalse(); + MediaSignatures.DetectAnimation(MediaKind.WebP, Samples.WebpLossless()).ShouldBeFalse(); + MediaSignatures.DetectAnimation(MediaKind.WebP, Samples.WebpExtended(animated: false)).ShouldBeFalse(); + } + + [Fact] + public void An_extended_webp_with_the_anim_flag_animates() => + MediaSignatures.DetectAnimation(MediaKind.WebP, Samples.WebpExtended(animated: true)).ShouldBeTrue(); + + [Fact] + public void Png_dimensions_are_read() + { + var (width, height) = MediaSignatures.ReadDimensions(MediaKind.Png, Samples.Png(1920, 1080)); + + width.ShouldBe(1920); + height.ShouldBe(1080); + } + + [Fact] + public void Gif_dimensions_are_read() + { + var (width, height) = MediaSignatures.ReadDimensions(MediaKind.Gif, Samples.Gif(width: 320, height: 240)); + + width.ShouldBe(320); + height.ShouldBe(240); + } + + [Fact] + public void Extended_webp_dimensions_are_read() + { + // Stored as canvas-minus-one, which is the kind of off-by-one that ships unnoticed. + var (width, height) = MediaSignatures.ReadDimensions( + MediaKind.WebP, + Samples.WebpExtended(animated: true, width: 800, height: 600) + ); + + width.ShouldBe(800); + height.ShouldBe(600); + } + + [Fact] + public void Dimensions_are_absent_rather_than_guessed() + { + MediaSignatures.ReadDimensions(MediaKind.Jpeg, Samples.Jpeg()).ShouldBe((null, null)); + MediaSignatures.ReadDimensions(MediaKind.WebP, Samples.WebpStill()).ShouldBe((null, null)); + } + + [Fact] + public void Garbage_never_throws() + { + var random = new Random(20260813); + + for (var attempt = 0; attempt < 500; attempt++) + { + var bytes = new byte[random.Next(0, 200)]; + random.NextBytes(bytes); + + var kind = MediaSignatures.Detect(bytes); + _ = MediaSignatures.DetectAnimation(kind, bytes); + _ = MediaSignatures.ReadDimensions(kind, bytes); + _ = MediaSignatures.LooksLikeHtml(bytes); + } + } +}