fix: propagate feed job cancellation

Jobs/RssFeedJob.cs

@@ -17,9 +17,10 @@ public class RssFeedJob(DB db, RssFeedService rssFeed, DiscordWebhookService dis
 {
     public async Task Execute(IJobExecutionContext context)
     {
+        CancellationToken cancellationToken = context.CancellationToken;
         List<RssSubscription> subscriptions = await db.RssSubscriptions
             .Include(s => s.Webhook)
-            .ToListAsync();
+            .ToListAsync(cancellationToken);
 
         if (subscriptions.Count == 0)
             return;
@@ -28,29 +29,30 @@ public async Task Execute(IJobExecutionContext context)
 
         foreach (IGrouping<string, RssSubscription> group in subscriptions.GroupBy(s => s.FeedUrl))
         {
+            cancellationToken.ThrowIfCancellationRequested();
             string feedUrl = group.Key;
             List<RssSubscription> subs = group.ToList();
 
-            (string? _, string? _, IReadOnlyList<RssFeedService.FeedEntry> entries) = await rssFeed.FetchAsync(feedUrl);
+            (string? _, string? _, IReadOnlyList<RssFeedService.FeedEntry> entries) = await rssFeed.FetchAsync(feedUrl, cancellationToken);
             if (entries.Count == 0)
                 continue;
 
             List<string> entryIds = entries.Select(e => e.EntryId).ToList();
             HashSet<string> seen = (await db.RssSeenEntries
                     .Where(v => v.FeedUrl == feedUrl && entryIds.Contains(v.EntryId))
                     .Select(v => v.EntryId)
-                    .ToListAsync())
+                    .ToListAsync(cancellationToken))
                 .ToHashSet();
 
             // If nothing from this feed has ever been seen, this is an initial run: mark
             // everything seen and only post the latest entry to avoid backfilling history.
             // Check all history for the feed because older seen entries may have rolled out of
             // the feed's current response.
-            bool initialSeed = !await HasFeedHistoryAsync(db, feedUrl);
+            bool initialSeed = !await HasFeedHistoryAsync(db, feedUrl, cancellationToken);
             if (initialSeed)
             {
                 RssFeedService.FeedEntry latest = entries.OrderByDescending(e => e.Published).First();
-                if (!await DispatchAsync(latest, subs, SendAsync))
+                if (!await DispatchAsync(latest, subs, (sub, content) => SendAsync(sub, content, cancellationToken), cancellationToken))
                     continue;
 
                 foreach (RssFeedService.FeedEntry entry in entries)
@@ -64,10 +66,11 @@ public async Task Execute(IJobExecutionContext context)
 
             foreach (RssFeedService.FeedEntry entry in entries.OrderBy(e => e.Published))
             {
+                cancellationToken.ThrowIfCancellationRequested();
                 if (seen.Contains(entry.EntryId))
                     continue;
 
-                if (!await DispatchAsync(entry, subs, SendAsync))
+                if (!await DispatchAsync(entry, subs, (sub, content) => SendAsync(sub, content, cancellationToken), cancellationToken))
                     continue;
 
                 db.RssSeenEntries.Add(new RssSeenEntry { FeedUrl = feedUrl, EntryId = entry.EntryId, SeenAt = DateTime.UtcNow });
@@ -77,13 +80,14 @@ public async Task Execute(IJobExecutionContext context)
         }
 
         if (changed)
-            await db.SaveChangesAsync();
+            await db.SaveChangesAsync(cancellationToken);
     }
 
     internal static async Task<bool> DispatchAsync(
         RssFeedService.FeedEntry entry,
         IReadOnlyList<RssSubscription> subs,
-        Func<RssSubscription, string, Task<bool>> sendAsync)
+        Func<RssSubscription, string, Task<bool>> sendAsync,
+        CancellationToken cancellationToken = default)
     {
         string content = !string.IsNullOrWhiteSpace(entry.Link) ? entry.Link : entry.Title;
         if (string.IsNullOrWhiteSpace(content))
@@ -92,25 +96,26 @@ internal static async Task<bool> DispatchAsync(
         bool allSucceeded = true;
         foreach (RssSubscription sub in subs)
         {
+            cancellationToken.ThrowIfCancellationRequested();
             if (!await sendAsync(sub, content))
                 allSucceeded = false;
         }
 
         return allSucceeded;
     }
 
-    internal static Task<bool> HasFeedHistoryAsync(DB db, string feedUrl) =>
-        db.RssSeenEntries.AnyAsync(entry => entry.FeedUrl == feedUrl);
+    internal static Task<bool> HasFeedHistoryAsync(DB db, string feedUrl, CancellationToken cancellationToken = default) =>
+        db.RssSeenEntries.AnyAsync(entry => entry.FeedUrl == feedUrl, cancellationToken);
 
-    private async Task<bool> SendAsync(RssSubscription sub, string content)
+    private async Task<bool> SendAsync(RssSubscription sub, string content, CancellationToken cancellationToken)
     {
         if (sub.Webhook == null)
         {
             logsService.Log($"RssFeedJob: no webhook available for {sub.FeedUrl} in channel {sub.ChannelDiscordId}", LogSeverity.Warning);
             return false;
         }
 
-        bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, content, sub.DisplayName, sub.AvatarUrl);
+        bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, content, sub.DisplayName, sub.AvatarUrl, cancellationToken);
         if (!ok)
             logsService.Log($"RssFeedJob: failed to post entry from {sub.FeedUrl} to channel {sub.ChannelDiscordId}", LogSeverity.Warning);
 

Jobs/YoutubeRssJob.cs

@@ -20,9 +20,10 @@ public class YoutubeRssJob(DB db, YoutubeFeedService youtubeFeed, DiscordWebhook
 
     public async Task Execute(IJobExecutionContext context)
     {
+        CancellationToken cancellationToken = context.CancellationToken;
         List<YoutubeSubscription> subscriptions = await db.YoutubeSubscriptions
             .Include(s => s.Webhook)
-            .ToListAsync();
+            .ToListAsync(cancellationToken);
 
         if (subscriptions.Count == 0)
             return;
@@ -31,18 +32,19 @@ public async Task Execute(IJobExecutionContext context)
 
         foreach (IGrouping<string, YoutubeSubscription> group in subscriptions.GroupBy(s => s.YoutubeChannelId))
         {
+            cancellationToken.ThrowIfCancellationRequested();
             string youtubeChannelId = group.Key;
             List<YoutubeSubscription> subs = group.ToList();
 
-            (string? channelTitle, IReadOnlyList<YoutubeFeedService.VideoEntry> entries) = await youtubeFeed.FetchFeedAsync(youtubeChannelId);
+            (string? channelTitle, IReadOnlyList<YoutubeFeedService.VideoEntry> entries) = await youtubeFeed.FetchFeedAsync(youtubeChannelId, cancellationToken);
             if (entries.Count == 0)
                 continue;
 
             // Refresh cached identity (title / avatar) used as the webhook username + avatar.
             string username = !string.IsNullOrWhiteSpace(channelTitle) ? channelTitle! : subs[0].YoutubeChannelTitle;
             string? avatar = subs.Select(s => s.YoutubeAvatarUrl).FirstOrDefault(a => !string.IsNullOrWhiteSpace(a));
             if (string.IsNullOrWhiteSpace(avatar))
-                avatar = await YoutubeUtils.GetChannelAvatarAsync(HttpClient, youtubeChannelId);
+                avatar = await YoutubeUtils.GetChannelAvatarAsync(HttpClient, youtubeChannelId, cancellationToken);
 
             foreach (YoutubeSubscription sub in subs)
             {
@@ -62,18 +64,18 @@ public async Task Execute(IJobExecutionContext context)
             HashSet<string> seen = (await db.YoutubeSeenVideos
                     .Where(v => videoIds.Contains(v.VideoId))
                     .Select(v => v.VideoId)
-                    .ToListAsync())
+                    .ToListAsync(cancellationToken))
                 .ToHashSet();
 
             // If nothing from this channel has ever been seen, this is an initial run for it:
             // mark everything seen and only post the latest video to avoid backfilling history.
             // Check all history for the channel because older seen videos may have rolled out of
             // the feed's current response.
-            bool initialSeed = !await HasFeedHistoryAsync(db, youtubeChannelId);
+            bool initialSeed = !await HasFeedHistoryAsync(db, youtubeChannelId, cancellationToken);
             if (initialSeed)
             {
                 YoutubeFeedService.VideoEntry latest = entries.OrderByDescending(e => e.Published).First();
-                if (!await DispatchAsync(subs, sub => SendAsync(sub, latest, username, avatar)))
+                if (!await DispatchAsync(subs, sub => SendAsync(sub, latest, username, avatar, cancellationToken), cancellationToken))
                     continue;
 
                 foreach (YoutubeFeedService.VideoEntry entry in entries)
@@ -87,10 +89,11 @@ public async Task Execute(IJobExecutionContext context)
 
             foreach (YoutubeFeedService.VideoEntry entry in entries.OrderBy(e => e.Published))
             {
+                cancellationToken.ThrowIfCancellationRequested();
                 if (seen.Contains(entry.VideoId))
                     continue;
 
-                if (!await DispatchAsync(subs, sub => SendAsync(sub, entry, username, avatar)))
+                if (!await DispatchAsync(subs, sub => SendAsync(sub, entry, username, avatar, cancellationToken), cancellationToken))
                     continue;
 
                 db.YoutubeSeenVideos.Add(new YoutubeSeenVideo { YoutubeChannelId = youtubeChannelId, VideoId = entry.VideoId, SeenAt = DateTime.UtcNow });
@@ -100,39 +103,42 @@ public async Task Execute(IJobExecutionContext context)
         }
 
         if (changed)
-            await db.SaveChangesAsync();
+            await db.SaveChangesAsync(cancellationToken);
     }
 
     internal static async Task<bool> DispatchAsync(
         IReadOnlyList<YoutubeSubscription> subs,
-        Func<YoutubeSubscription, Task<bool>> sendAsync)
+        Func<YoutubeSubscription, Task<bool>> sendAsync,
+        CancellationToken cancellationToken = default)
     {
         bool allSucceeded = true;
         foreach (YoutubeSubscription sub in subs)
         {
+            cancellationToken.ThrowIfCancellationRequested();
             if (!await sendAsync(sub))
                 allSucceeded = false;
         }
 
         return allSucceeded;
     }
 
-    internal static Task<bool> HasFeedHistoryAsync(DB db, string youtubeChannelId) =>
-        db.YoutubeSeenVideos.AnyAsync(video => video.YoutubeChannelId == youtubeChannelId);
+    internal static Task<bool> HasFeedHistoryAsync(DB db, string youtubeChannelId, CancellationToken cancellationToken = default) =>
+        db.YoutubeSeenVideos.AnyAsync(video => video.YoutubeChannelId == youtubeChannelId, cancellationToken);
 
     private async Task<bool> SendAsync(
         YoutubeSubscription sub,
         YoutubeFeedService.VideoEntry entry,
         string username,
-        string? avatar)
+        string? avatar,
+        CancellationToken cancellationToken)
     {
         if (sub.Webhook == null)
         {
             logsService.Log($"YoutubeRssJob: no webhook available for {sub.YoutubeChannelId} in channel {sub.ChannelDiscordId}", LogSeverity.Warning);
             return false;
         }
 
-        bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, entry.Link, username, avatar);
+        bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, entry.Link, username, avatar, cancellationToken);
         if (!ok)
             logsService.Log($"YoutubeRssJob: failed to post {entry.VideoId} to channel {sub.ChannelDiscordId}", LogSeverity.Warning);
 

Morpheus.Tests/RssFeedJobTests.cs

@@ -98,4 +98,26 @@ [new RssSubscription()],
         Assert.True(succeeded);
         Assert.False(sent);
     }
+
+    [Fact]
+    public async Task DispatchAsync_WhenCallerCancels_PropagatesCancellationBeforeDelivery()
+    {
+        RssFeedService.FeedEntry entry = new("entry-1", "Entry", "https://example.com/entry-1", DateTime.UtcNow);
+        using CancellationTokenSource cts = new();
+        await cts.CancelAsync();
+        bool sent = false;
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
+            RssFeedJob.DispatchAsync(
+                entry,
+                [new RssSubscription()],
+                (_, _) =>
+                {
+                    sent = true;
+                    return Task.FromResult(true);
+                },
+                cts.Token));
+
+        Assert.False(sent);
+    }
 }

Morpheus.Tests/YoutubeRssJobTests.cs

@@ -79,4 +79,24 @@ Task<bool> SendAsync(YoutubeSubscription _)
         Assert.True(secondSucceeded);
         Assert.Equal(4, attempts);
     }
+
+    [Fact]
+    public async Task DispatchAsync_WhenCallerCancels_PropagatesCancellationBeforeDelivery()
+    {
+        using CancellationTokenSource cts = new();
+        await cts.CancelAsync();
+        bool sent = false;
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
+            YoutubeRssJob.DispatchAsync(
+                [new YoutubeSubscription()],
+                _ =>
+                {
+                    sent = true;
+                    return Task.FromResult(true);
+                },
+                cts.Token));
+
+        Assert.False(sent);
+    }
 }

Morpheus.Tests/YoutubeUtilsTests.cs

@@ -27,6 +27,18 @@ public async Task ResolveChannelIdAsync_DoesNotRequestNonYoutubeUrls(string inpu
         Assert.Empty(handler.RequestedUris);
     }
 
+    [Fact]
+    public async Task GetChannelAvatarAsync_WhenCallerCancels_PropagatesCancellation()
+    {
+        RecordingHandler handler = new(_ => throw new InvalidOperationException("Unexpected HTTP request."));
+        using HttpClient httpClient = new(handler);
+        using CancellationTokenSource cts = new();
+        await cts.CancelAsync();
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
+            YoutubeUtils.GetChannelAvatarAsync(httpClient, "UC123", cts.Token));
+    }
+
     [Theory]
     [InlineData("https://www.youtube.com/@channel", "/@channel", "")]
     [InlineData("youtube.com/user/channel", "/user/channel", "")]
@@ -74,6 +86,9 @@ private sealed class RecordingHandler(Func<HttpRequestMessage, HttpResponseMessa
 
         protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
         {
+            if (cancellationToken.IsCancellationRequested)
+                return Task.FromCanceled<HttpResponseMessage>(cancellationToken);
+
             if (request.RequestUri != null)
                 RequestedUris.Add(request.RequestUri);
 

Utilities/YoutubeUtils.cs

@@ -154,7 +154,7 @@ private static bool IsSupportedYoutubePath(string path) =>
     /// Uses the Innertube (youtubei) browse API to fetch a channel's avatar URL.
     /// Returns null on error or if thumbnails are not found.
     /// </summary>
-    public static async Task<string?> GetChannelAvatarAsync(HttpClient httpClient, string channelId)
+    public static async Task<string?> GetChannelAvatarAsync(HttpClient httpClient, string channelId, CancellationToken cancellationToken = default)
     {
         ArgumentNullException.ThrowIfNull(httpClient);
         if (string.IsNullOrWhiteSpace(channelId))
@@ -165,10 +165,10 @@ private static bool IsSupportedYoutubePath(string path) =>
             // 1) fetch youtube homepage to extract INNERTUBE_API_KEY and client version
             using HttpRequestMessage homeReq = new(HttpMethod.Get, "https://www.youtube.com");
             homeReq.Headers.UserAgent.ParseAdd(BrowserUserAgent);
-            using HttpResponseMessage homeResp = await httpClient.SendAsync(homeReq).ConfigureAwait(false);
+            using HttpResponseMessage homeResp = await httpClient.SendAsync(homeReq, cancellationToken).ConfigureAwait(false);
             if (!homeResp.IsSuccessStatusCode)
                 return null;
-            string homeHtml = await homeResp.Content.ReadAsStringAsync().ConfigureAwait(false);
+            string homeHtml = await homeResp.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
 
             static string? ExtractValue(string hay, string marker)
             {
@@ -201,10 +201,10 @@ private static bool IsSupportedYoutubePath(string path) =>
             };
             req.Headers.UserAgent.ParseAdd(BrowserUserAgent);
 
-            using HttpResponseMessage resp = await httpClient.SendAsync(req).ConfigureAwait(false);
+            using HttpResponseMessage resp = await httpClient.SendAsync(req, cancellationToken).ConfigureAwait(false);
             if (!resp.IsSuccessStatusCode)
                 return null;
-            string body = await resp.Content.ReadAsStringAsync().ConfigureAwait(false);
+            string body = await resp.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
             using JsonDocument doc = JsonDocument.Parse(body);
             JsonElement root = doc.RootElement;
 
@@ -266,9 +266,12 @@ static bool TryTraverse(JsonElement el, string[] path, out JsonElement result)
                 string? best = BestFromSources(thumbs2);
                 if (!string.IsNullOrWhiteSpace(best)) return best;
             }
-
             return null;
         }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
         catch (Exception)
         {
             return null;