diff --git a/src/statichost/StaticHost/Live/LiveEndpoints.cs b/src/statichost/StaticHost/Live/LiveEndpoints.cs index 827a65b7f5..96d88de128 100644 --- a/src/statichost/StaticHost/Live/LiveEndpoints.cs +++ b/src/statichost/StaticHost/Live/LiveEndpoints.cs @@ -334,6 +334,8 @@ private static async Task YouTubeVerify( return Results.NotFound(); } + logger.LogInformation("YouTube WebSub subscription verified for {Topic}; granted lease {LeaseSeconds}s.", + topic, leaseSeconds); return Results.Text(challenge, "text/plain"); } diff --git a/src/statichost/StaticHost/Live/LiveStatusServiceCollectionExtensions.cs b/src/statichost/StaticHost/Live/LiveStatusServiceCollectionExtensions.cs index 9287968e0c..cbfb9fc95e 100644 --- a/src/statichost/StaticHost/Live/LiveStatusServiceCollectionExtensions.cs +++ b/src/statichost/StaticHost/Live/LiveStatusServiceCollectionExtensions.cs @@ -12,7 +12,7 @@ public static class LiveStatusServiceCollectionExtensions /// Registers Redis-backed canonical state, a per-process /// , named /// instances ("twitch", "twitch-id", - /// "youtube", "youtube-pubsub") with standard resilience, the + /// "youtube", "youtube-pubsub"), the /// Twitch token provider + Helix client, the YouTube Data + WebSub client, /// and the two workers /// ( and ). @@ -49,8 +49,9 @@ public static TBuilder AddLiveStatus(this TBuilder builder) .AddStandardResilienceHandler(); builder.Services.AddHttpClient(YouTubeClient.HttpClientName) .AddStandardResilienceHandler(); - builder.Services.AddHttpClient(YouTubeClient.PubSubHttpClientName) - .AddStandardResilienceHandler(); + // Subscription POST retries are scheduled in Redis, not inside the HTTP request. + builder.Services.AddHttpClient(YouTubeClient.PubSubHttpClientName, + client => client.Timeout = TimeSpan.FromSeconds(30)); builder.Services.AddSingleton(); builder.Services.AddSingleton(); diff --git a/src/statichost/StaticHost/Live/README.md b/src/statichost/StaticHost/Live/README.md index 08f4272242..d4ed37d3f0 100644 --- a/src/statichost/StaticHost/Live/README.md +++ b/src/statichost/StaticHost/Live/README.md @@ -33,9 +33,35 @@ Two `BackgroundService` workers keep the state current: | `YouTubeWebSubService` | PubSubHubbub `videos.xml` push | `videos.list` every 2 min while live; `search.list` every 30 min while idle | Webhook signature and parsing logic remains separate and unit-testable. -Outgoing HTTP is performed by named, resilient -`HttpClient` instances (`twitch`, `twitch-id`, `youtube`, -`youtube-pubsub`) registered with `AddStandardResilienceHandler`. +The `twitch`, `twitch-id`, and `youtube` named HTTP clients use +`AddStandardResilienceHandler`. The `youtube-pubsub` client instead uses a +30-second timeout without in-request retries. Redis schedules subscription +retries separately from live-status polling. + +### YouTube subscription failures + +Both subscription requests and verification use Google's documented topic: +`https://www.youtube.com/feeds/videos.xml?channel_id=CHANNEL_ID`. + +Failed subscription requests back off on a 2, 5, 15, 30, then 60-minute schedule, +with each delay jittered between 90% and 100% of that duration. The worker +retries on its first normal tick at or after the persisted deadline. Failure +counts and deadlines are shared in Redis so restarts or leader changes do not +restart the retry sequence. Successful verification resets the backoff; +acceptance of the subscription POST alone does not. A channel change starts a +fresh sequence. + +Each failed attempt logs one concise warning for expected HTTP errors or +timeouts, including the failure count and next retry deadline. Exception +details and hub error response bodies are available at Debug; unexpected +exceptions remain Error-level with their stack traces. Successful verification +is logged at Information. Shutdown or leadership cancellation does not count +as an upstream failure. + +The live and idle polling intervals remain unchanged during subscription +backoff. Google documents feed notifications for uploads and video +title/description updates, not guaranteed broadcast start/stop events, so +fallback polling is still necessary. ## Configuration @@ -345,8 +371,11 @@ project or `Aspire.Dev.slnx` trigger that job. - Existing SSE connections retain the last local snapshot during a temporary Redis interruption. New snapshot and SSE requests require canonical Redis state. -- Provider loops log failures and retry on their configured intervals. -- Standard resilience pipeline on every named `HttpClient`. +- Provider loops log failures and retry on their configured intervals, with + separate shared backoff for failed YouTube subscription requests. +- Standard resilience pipelines protect Data API and Twitch requests; + YouTube subscription POSTs use the bounded timeout and shared retry schedule + described above. - Reconciliation timers are the safety net for missed individual webhooks. - SSE heartbeats every 15 s defeat proxy idle-timeouts; the client uses exponential backoff with a `visibilitychange`-aware reconnect. diff --git a/src/statichost/StaticHost/Live/YouTube/RedisYouTubeWebSubSubscriptionState.cs b/src/statichost/StaticHost/Live/YouTube/RedisYouTubeWebSubSubscriptionState.cs index 8c7dec73f0..8d1d4ee756 100644 --- a/src/statichost/StaticHost/Live/YouTube/RedisYouTubeWebSubSubscriptionState.cs +++ b/src/statichost/StaticHost/Live/YouTube/RedisYouTubeWebSubSubscriptionState.cs @@ -25,14 +25,19 @@ internal sealed class RedisYouTubeWebSubSubscriptionState( }, cancellationToken); - public async ValueTask MarkRequestFailedAsync( + public async ValueTask MarkRequestFailedAsync( YouTubeWebSubSubscriptionRequest request, CancellationToken cancellationToken = default) { - _ = await UpdateAsync( - current => new StateTransition( - YouTubeWebSubSubscriptionTransitions.MarkRequestFailed(current, request), - true), + var failedAt = timeProvider.GetUtcNow(); + return await UpdateAsync( + current => + { + var next = YouTubeWebSubSubscriptionTransitions.MarkRequestFailed(current, request, failedAt); + return new StateTransition( + next, + next == current ? null : next.Retry); + }, cancellationToken).ConfigureAwait(false); } diff --git a/src/statichost/StaticHost/Live/YouTube/YouTubeClient.cs b/src/statichost/StaticHost/Live/YouTube/YouTubeClient.cs index 6a9b9c72a1..106715fea7 100644 --- a/src/statichost/StaticHost/Live/YouTube/YouTubeClient.cs +++ b/src/statichost/StaticHost/Live/YouTube/YouTubeClient.cs @@ -114,8 +114,8 @@ public async Task SubscribeAsync( var c = httpFactory.CreateClient(PubSubHttpClientName); c.BaseAddress ??= new Uri("https://pubsubhubbub.appspot.com/"); - var topic = $"https://www.youtube.com/xml/feeds/videos.xml?channel_id={channelId}"; - var form = new FormUrlEncodedContent( + var topic = YouTubeWebSubSubscriptionTransitions.TopicFor(channelId); + using var form = new FormUrlEncodedContent( [ new KeyValuePair("hub.callback", callbackUrl), new KeyValuePair("hub.topic", topic), @@ -130,7 +130,7 @@ public async Task SubscribeAsync( if (!response.IsSuccessStatusCode) { var body = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); - logger.LogWarning("YouTube WebSub subscribe failed: {Status} {Body}", response.StatusCode, body); + logger.LogDebug("YouTube WebSub subscribe failed: {Status} {Body}", response.StatusCode, body); response.EnsureSuccessStatusCode(); } } diff --git a/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubService.cs b/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubService.cs index 7a3fd0478c..f60955e26b 100644 --- a/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubService.cs +++ b/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubService.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.Options; +using Polly.Timeout; namespace StaticHost.Live.YouTube; @@ -128,17 +129,32 @@ await client.SubscribeAsync( } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - await _subscriptions.MarkRequestFailedAsync( - request, - CancellationToken.None).ConfigureAwait(false); + // Let the short send reservation expire on shutdown or leadership loss. throw; } catch (Exception ex) { - await _subscriptions.MarkRequestFailedAsync( + var retry = await _subscriptions.MarkRequestFailedAsync( request, CancellationToken.None).ConfigureAwait(false); - logger.LogWarning(ex, "YouTube WebSub subscribe failed; will retry next tick."); + if (ex is HttpRequestException or OperationCanceledException or TimeoutRejectedException) + { + if (retry is not null) + { + logger.LogWarning( + "YouTube WebSub subscribe failed ({FailureType}, HTTP {StatusCode}); attempt {FailureCount}. " + + "Next subscription attempt no earlier than {RetryAt}. Live-status polling continues.", + ex.GetType().Name, + (ex as HttpRequestException)?.StatusCode, + retry.FailureCount, + retry.RetryAt); + } + logger.LogDebug(ex, "YouTube WebSub subscription request failure details."); + } + else + { + logger.LogError(ex, "Unexpected YouTube WebSub subscription failure."); + } } if (requestSent) diff --git a/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubSubscriptionState.cs b/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubSubscriptionState.cs index 67991cc1ec..c6f756e38b 100644 --- a/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubSubscriptionState.cs +++ b/src/statichost/StaticHost/Live/YouTube/YouTubeWebSubSubscriptionState.cs @@ -15,7 +15,7 @@ public interface IYouTubeWebSubSubscriptionState DateTimeOffset now, CancellationToken cancellationToken = default); - ValueTask MarkRequestFailedAsync( + ValueTask MarkRequestFailedAsync( YouTubeWebSubSubscriptionRequest request, CancellationToken cancellationToken = default); @@ -41,12 +41,19 @@ public sealed record YouTubeWebSubSubscriptionRequest( DateTimeOffset RequestedAt, DateTimeOffset? SentAt = null); +public sealed record YouTubeWebSubRetryState( + string Topic, + int FailureCount, + DateTimeOffset RetryAt); + internal sealed record YouTubeWebSubSubscriptionData( YouTubeWebSubSubscriptionRequest? Pending, string? ActiveTopic, DateTimeOffset RenewAt, YouTubeWebSubConfirmation? RecentConfirmation = null) { + public YouTubeWebSubRetryState? Retry { get; init; } + public static YouTubeWebSubSubscriptionData Empty { get; } = new( Pending: null, ActiveTopic: null, @@ -79,6 +86,14 @@ internal static class YouTubeWebSubSubscriptionTransitions out YouTubeWebSubSubscriptionData next) { var topic = TopicFor(channelId); + if (current.Retry is { } retry && + string.Equals(retry.Topic, topic, StringComparison.Ordinal) && + now < retry.RetryAt) + { + next = current; + return null; + } + var pending = current.Pending; if (pending is not null) @@ -110,14 +125,41 @@ internal static class YouTubeWebSubSubscriptionTransitions Convert.ToHexStringLower(RandomNumberGenerator.GetBytes(32)), now); - next = current with { Pending = request, RecentConfirmation = null }; + next = current with + { + Pending = request, + RecentConfirmation = null, + Retry = string.Equals(current.Retry?.Topic, topic, StringComparison.Ordinal) ? current.Retry : null, + }; return request; } public static YouTubeWebSubSubscriptionData MarkRequestFailed( YouTubeWebSubSubscriptionData current, - YouTubeWebSubSubscriptionRequest request) => - RequestsMatch(current.Pending, request) ? current with { Pending = null } : current; + YouTubeWebSubSubscriptionRequest request, + DateTimeOffset failedAt) + { + if (!RequestsMatch(current.Pending, request)) + { + return current; + } + + var failures = (int)Math.Min((long)(current.Retry?.FailureCount ?? 0) + 1, int.MaxValue); + var minutes = failures switch + { + 1 => 2, + 2 => 5, + 3 => 15, + 4 => 30, + _ => 60, + }; + var delay = TimeSpan.FromMinutes(minutes * (0.9 + Random.Shared.NextDouble() * 0.1)); + return current with + { + Pending = null, + Retry = new YouTubeWebSubRetryState(request.Topic, failures, failedAt.Add(delay)), + }; + } public static YouTubeWebSubSubscriptionData MarkRequestSent( YouTubeWebSubSubscriptionData current, @@ -185,8 +227,8 @@ public static bool HasValidConfirmationShape( !string.IsNullOrEmpty(verifyToken) && leaseSeconds > 0; - private static string TopicFor(string channelId) => - $"https://www.youtube.com/xml/feeds/videos.xml?channel_id={channelId}"; + internal static string TopicFor(string channelId) => + $"https://www.youtube.com/feeds/videos.xml?channel_id={channelId}"; private static bool RequestsMatch( YouTubeWebSubSubscriptionRequest? current, diff --git a/tests/StaticHost.Tests/Live/InMemoryLiveStatusInfrastructure.cs b/tests/StaticHost.Tests/Live/InMemoryLiveStatusInfrastructure.cs index a46656fa6f..6a4959fc6f 100644 --- a/tests/StaticHost.Tests/Live/InMemoryLiveStatusInfrastructure.cs +++ b/tests/StaticHost.Tests/Live/InMemoryLiveStatusInfrastructure.cs @@ -112,7 +112,7 @@ public ValueTask DisposeAsync() } } -internal sealed class YouTubeWebSubSubscriptionState : IYouTubeWebSubSubscriptionState +internal sealed class YouTubeWebSubSubscriptionState(TimeProvider? timeProvider = null) : IYouTubeWebSubSubscriptionState { private readonly Lock _gate = new(); private YouTubeWebSubSubscriptionData _state = YouTubeWebSubSubscriptionData.Empty; @@ -135,7 +135,7 @@ internal sealed class YouTubeWebSubSubscriptionState : IYouTubeWebSubSubscriptio } } - public ValueTask MarkRequestFailedAsync( + public ValueTask MarkRequestFailedAsync( YouTubeWebSubSubscriptionRequest request, CancellationToken cancellationToken = default) { @@ -143,10 +143,11 @@ public ValueTask MarkRequestFailedAsync( lock (_gate) { - _state = YouTubeWebSubSubscriptionTransitions.MarkRequestFailed(_state, request); + var previous = _state; + _state = YouTubeWebSubSubscriptionTransitions.MarkRequestFailed( + _state, request, (timeProvider ?? TimeProvider.System).GetUtcNow()); + return ValueTask.FromResult(_state == previous ? null : _state.Retry); } - - return ValueTask.CompletedTask; } public ValueTask MarkRequestSentAsync( diff --git a/tests/StaticHost.Tests/Live/RedisYouTubeIntegrationTests.cs b/tests/StaticHost.Tests/Live/RedisYouTubeIntegrationTests.cs index 9233eb8924..34d48a3fad 100644 --- a/tests/StaticHost.Tests/Live/RedisYouTubeIntegrationTests.cs +++ b/tests/StaticHost.Tests/Live/RedisYouTubeIntegrationTests.cs @@ -222,23 +222,24 @@ public async Task WebSubReplacement_RejectsAbandonedOwnersLateWritesAndVerificat } [Fact] - public async Task WebSubFailedSend_AllowsAnotherWorkerToRetryWithoutClearingItsReservation() + public async Task WebSubFailedSend_SharesBackoffWithoutClearingAnotherWorkersReservation() { var now = DateTimeOffset.UtcNow; var first = SubscriptionState(Redis.First); var second = SubscriptionState(Redis.Second); var failed = Assert.IsType( await first.TryBeginSubscriptionAsync("channel-123", now)); - await first.MarkRequestFailedAsync(failed); + var failure = Assert.IsType(await first.MarkRequestFailedAsync(failed)); + Assert.Null(await second.TryBeginSubscriptionAsync("channel-123", failure.RetryAt.AddTicks(-1))); var retry = Assert.IsType( - await second.TryBeginSubscriptionAsync("channel-123", now.AddSeconds(1))); - await second.MarkRequestSentAsync(retry, now.AddSeconds(2)); + await second.TryBeginSubscriptionAsync("channel-123", failure.RetryAt)); + await second.MarkRequestSentAsync(retry, failure.RetryAt.AddSeconds(1)); await first.MarkRequestFailedAsync(failed); await first.MarkRequestSentAsync(failed, now.AddMinutes(1)); - Assert.Null(await first.TryBeginSubscriptionAsync("channel-123", now.AddMinutes(9))); + Assert.Null(await first.TryBeginSubscriptionAsync("channel-123", failure.RetryAt.AddMinutes(9))); Assert.True(await first.TryConfirmSubscriptionAsync( - "subscribe", retry.Topic, retry.VerifyToken, 3_600, now.AddMinutes(9))); + "subscribe", retry.Topic, retry.VerifyToken, 3_600, failure.RetryAt.AddMinutes(9))); } private static RedisYouTubeWebSubSubscriptionState SubscriptionState(IConnectionMultiplexer connection) => diff --git a/tests/StaticHost.Tests/Live/YouTubeClientTests.cs b/tests/StaticHost.Tests/Live/YouTubeClientTests.cs index 055e44ea43..7ecc32120c 100644 --- a/tests/StaticHost.Tests/Live/YouTubeClientTests.cs +++ b/tests/StaticHost.Tests/Live/YouTubeClientTests.cs @@ -1,3 +1,5 @@ +using Microsoft.Extensions.Hosting; + namespace StaticHost.Tests.Live; public sealed class YouTubeClientTests @@ -117,7 +119,27 @@ await client.SubscribeAsync( Assert.Contains("hub.secret=webhook-secret", request.Content, StringComparison.Ordinal); Assert.Contains("hub.lease_seconds=432000", request.Content, StringComparison.Ordinal); Assert.Contains("hub.callback=https%3A%2F%2Fexample.com%2Fapi%2Flive%2Fyoutube%2Fwebhook", request.Content, StringComparison.Ordinal); - Assert.Contains("hub.topic=https%3A%2F%2Fwww.youtube.com%2Fxml%2Ffeeds%2Fvideos.xml%3Fchannel_id%3Dchannel-123", request.Content, StringComparison.Ordinal); + Assert.Contains("hub.topic=https%3A%2F%2Fwww.youtube.com%2Ffeeds%2Fvideos.xml%3Fchannel_id%3Dchannel-123", request.Content, StringComparison.Ordinal); + } + + [Fact] + public async Task PubSubHttpClient_HasBoundedTimeoutAndDoesNotRetryPosts() + { + var builder = Host.CreateApplicationBuilder(); + builder.AddLiveStatus(); + var handler = new RecordingHttpMessageHandler(_ => new HttpResponseMessage(HttpStatusCode.ServiceUnavailable)); + builder.Services.AddHttpClient(YouTubeClient.PubSubHttpClientName) + .ConfigurePrimaryHttpMessageHandler(() => handler); + using var host = builder.Build(); + var client = host.Services.GetRequiredService() + .CreateClient(YouTubeClient.PubSubHttpClientName); + using var form = new FormUrlEncodedContent([]); + + using var response = await client.PostAsync("https://pubsubhubbub.appspot.com/subscribe", form); + + Assert.Equal(TimeSpan.FromSeconds(30), client.Timeout); + Assert.Equal(HttpStatusCode.ServiceUnavailable, response.StatusCode); + Assert.Single(handler.Requests); } private static YouTubeClient CreateClient( diff --git a/tests/StaticHost.Tests/Live/YouTubeWebSubRetryTests.cs b/tests/StaticHost.Tests/Live/YouTubeWebSubRetryTests.cs new file mode 100644 index 0000000000..db81f1af6e --- /dev/null +++ b/tests/StaticHost.Tests/Live/YouTubeWebSubRetryTests.cs @@ -0,0 +1,103 @@ +namespace StaticHost.Tests.Live; + +public sealed class YouTubeWebSubRetryTests +{ + [Fact] + public async Task FailedRequests_BackOffAcrossWorkersWithBoundedJitter() + { + var redis = new TestLiveStatusRedis(); + var time = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + + for (var attempt = 1; attempt <= 7; attempt++) + { + var state = CreateState(redis, time); + var request = Assert.IsType( + await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + Assert.Equal("https://www.youtube.com/feeds/videos.xml?channel_id=channel", request.Topic); + time.Advance(TimeSpan.FromSeconds(30)); + var failedAt = time.GetUtcNow(); + var retry = Assert.IsType(await state.MarkRequestFailedAsync(request)); + var maximumMinutes = attempt switch { 1 => 2, 2 => 5, 3 => 15, 4 => 30, _ => 60 }; + + Assert.Equal(attempt, retry.FailureCount); + Assert.InRange(retry.RetryAt - failedAt, + TimeSpan.FromMinutes(maximumMinutes * 0.9), + TimeSpan.FromMinutes(maximumMinutes)); + var otherWorker = CreateState(redis, time); + Assert.Null(await otherWorker.TryBeginSubscriptionAsync("channel", retry.RetryAt.AddTicks(-1))); + Assert.Null(await otherWorker.MarkRequestFailedAsync(request)); + time.SetUtcNow(retry.RetryAt); + } + } + + [Fact] + public async Task OnlyVerificationResetsFailures_AndLateFailureCannotUndoRecovery() + { + var redis = new TestLiveStatusRedis(); + var time = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + var state = CreateState(redis, time); + var first = Assert.IsType( + await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + var failure = Assert.IsType(await state.MarkRequestFailedAsync(first)); + time.SetUtcNow(failure.RetryAt); + var second = Assert.IsType( + await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + await state.MarkRequestSentAsync(second, time.GetUtcNow()); + var secondFailure = Assert.IsType(await state.MarkRequestFailedAsync(second)); + Assert.Equal(2, secondFailure.FailureCount); + time.SetUtcNow(secondFailure.RetryAt); + var third = Assert.IsType( + await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + + Assert.True(await CreateState(redis, time).TryConfirmSubscriptionAsync( + "subscribe", third.Topic, third.VerifyToken, 3600, time.GetUtcNow())); + Assert.Null(await state.MarkRequestFailedAsync(third)); + Assert.Null(await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + time.SetUtcNow(await state.GetRenewAtAsync()); + var renewal = Assert.IsType( + await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + var renewalFailure = Assert.IsType(await state.MarkRequestFailedAsync(renewal)); + Assert.Equal(1, renewalFailure.FailureCount); + } + + [Fact] + public async Task ChannelChangeIsNotBlockedByPreviousChannelsBackoff() + { + var redis = new TestLiveStatusRedis(); + var time = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + var state = CreateState(redis, time); + var old = Assert.IsType( + await state.TryBeginSubscriptionAsync("old-channel", time.GetUtcNow())); + await state.MarkRequestFailedAsync(old); + + var replacement = Assert.IsType( + await state.TryBeginSubscriptionAsync("new-channel", time.GetUtcNow())); + Assert.Null(await state.MarkRequestFailedAsync(old)); + var retry = Assert.IsType(await state.MarkRequestFailedAsync(replacement)); + Assert.Equal(1, retry.FailureCount); + Assert.Equal(replacement.Topic, retry.Topic); + } + + [Fact] + public async Task ExistingRedisPayloadWithoutRetryState_ResubscribesToCorrectedTopic() + { + var redis = new TestLiveStatusRedis(); + redis.Set(LiveStatusRedisKeys.YouTubeSubscription, """ + {"version":1,"data":{"pending":null, + "activeTopic":"https://www.youtube.com/xml/feeds/videos.xml?channel_id=channel", + "renewAt":"2030-01-01T00:00:00+00:00","recentConfirmation":null}} + """); + var time = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + var state = CreateState(redis, time); + + var request = Assert.IsType( + await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + + Assert.Equal("https://www.youtube.com/feeds/videos.xml?channel_id=channel", request.Topic); + Assert.True(await state.TryConfirmSubscriptionAsync( + "subscribe", request.Topic, request.VerifyToken, 3600, time.GetUtcNow())); + } + + private static RedisYouTubeWebSubSubscriptionState CreateState(TestLiveStatusRedis redis, TimeProvider time) => + new(redis, time, NullLogger.Instance); +} diff --git a/tests/StaticHost.Tests/Live/YouTubeWebSubServiceTests.cs b/tests/StaticHost.Tests/Live/YouTubeWebSubServiceTests.cs index f40461e8f0..502740134b 100644 --- a/tests/StaticHost.Tests/Live/YouTubeWebSubServiceTests.cs +++ b/tests/StaticHost.Tests/Live/YouTubeWebSubServiceTests.cs @@ -1,3 +1,6 @@ +using Microsoft.Extensions.Logging; +using Polly.Timeout; + namespace StaticHost.Tests.Live; public sealed class YouTubeWebSubServiceTests @@ -330,6 +333,108 @@ await LiveStatusEndpointRouteBuilderExtensions.ConfirmYouTubeLiveStatusAsync( Assert.Equal(["channel-123"], client.LiveLookups); } + [Theory] + [InlineData("polly-timeout")] + [InlineData("http-timeout")] + [InlineData("http-error")] + [InlineData("unexpected")] + public async Task TickAsync_BacksOffSubscriptionsWithoutStoppingLivePolling(string failureKind) + { + Exception failure = failureKind switch + { + "polly-timeout" => new TimeoutRejectedException("timed out"), + "http-timeout" => new TaskCanceledException("timed out"), + "http-error" => new HttpRequestException("unavailable", null, HttpStatusCode.ServiceUnavailable), + _ => new InvalidOperationException("unexpected"), + }; + var time = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + var client = new TestYouTubeClient { SubscriptionException = failure }; + client.LiveResults.Enqueue(new YouTubeLiveResult(true, "video")); + client.VideoResults.Enqueue(new YouTubeLiveResult(true, "video")); + client.VideoResults.Enqueue(new YouTubeLiveResult(true, "video")); + using var broadcaster = LiveTestHelpers.CreateBroadcaster(timeProvider: time); + var logger = new RecordingLogger(); + using var service = new YouTubeWebSubService( + client, broadcaster, + new TestOptionsMonitor(new LiveStatusOptions + { + YouTube = new YouTubeOptions + { + ApiKey = "api-key", ChannelId = "channel", WebhookSecret = "secret", + }, + }), + logger, time, new YouTubeWebSubSubscriptionState(time), new SingleInstanceLiveStatusCoordination()); + + await service.TickAsync(CancellationToken.None); + var report = Assert.Single(logger.Entries, entry => entry.Level >= LogLevel.Warning); + if (failureKind == "unexpected") + { + Assert.Equal(LogLevel.Error, report.Level); + Assert.Same(failure, report.Exception); + } + else + { + Assert.Equal(LogLevel.Warning, report.Level); + Assert.Null(report.Exception); + Assert.Contains("Next subscription attempt", report.Message, StringComparison.Ordinal); + Assert.Contains(logger.Entries, entry => entry.Level == LogLevel.Debug && entry.Exception == failure); + } + + time.Advance(TimeSpan.FromMinutes(2)); + await service.TickAsync(CancellationToken.None); + Assert.Equal(2, client.Subscriptions.Count); + time.Advance(TimeSpan.FromMinutes(2)); + await service.TickAsync(CancellationToken.None); + + Assert.Equal(2, client.Subscriptions.Count); + Assert.Equal(2, logger.Entries.Count(entry => entry.Level >= LogLevel.Warning)); + Assert.Single(client.LiveLookups); + Assert.Equal(2, client.VideoLookups.Count); + Assert.True(broadcaster.Current.YouTube.Live); + } + + [Fact] + public async Task TickAsync_LeadershipCancellationDoesNotLogOrCountAsSubscriptionFailure() + { + using var cancellation = new CancellationTokenSource(); + var time = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + var state = new YouTubeWebSubSubscriptionState(time); + var client = new TestYouTubeClient + { + OnSubscribe = cancellation.Cancel, + SubscriptionException = new OperationCanceledException(cancellation.Token), + }; + var logger = new RecordingLogger(); + using var broadcaster = LiveTestHelpers.CreateBroadcaster(); + using var service = new YouTubeWebSubService( + client, broadcaster, + new TestOptionsMonitor(new LiveStatusOptions + { + YouTube = new YouTubeOptions + { + ApiKey = "api-key", ChannelId = "channel", WebhookSecret = "secret", + }, + }), + logger, time, state, new SingleInstanceLiveStatusCoordination()); + + await Assert.ThrowsAsync(() => service.TickAsync(cancellation.Token)); + + Assert.Empty(logger.Entries); + Assert.Empty(client.LiveLookups); + time.Advance(TimeSpan.FromSeconds(30)); + Assert.NotNull(await state.TryBeginSubscriptionAsync("channel", time.GetUtcNow())); + } + + private sealed class RecordingLogger : ILogger + { + public List<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = []; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, + Exception? exception, Func formatter) => + Entries.Add((logLevel, exception, formatter(state, exception))); + } + private sealed class TestYouTubeClient : IYouTubeClient { public string? ResolvedChannelId { get; set; } = "channel-123"; @@ -340,6 +445,10 @@ private sealed class TestYouTubeClient : IYouTubeClient public Exception? LiveLookupException { get; set; } + public Exception? SubscriptionException { get; set; } + + public Action? OnSubscribe { get; set; } + public List Subscriptions { get; } = []; public int ResolveChannelIdCalls { get; private set; } @@ -380,7 +489,8 @@ public Task SubscribeAsync( CancellationToken cancellationToken) { Subscriptions.Add(new Subscription(channelId, callbackUrl, secret, verifyToken, lease)); - return Task.CompletedTask; + OnSubscribe?.Invoke(); + return SubscriptionException is null ? Task.CompletedTask : Task.FromException(SubscriptionException); } }