Only disable invalid feeds after 1 day of failed attempts

Author: synzen

services/user-feeds/migrations/Migration20250209001416.ts

@@ -0,0 +1,13 @@
+import { Migration } from '@mikro-orm/migrations';
+
+export class Migration20250209001416 extends Migration {
+
+  async up(): Promise<void> {
+    this.addSql('alter table "feed_retry_record" add column "created_at" timestamptz(0) null;');
+  }
+
+  async down(): Promise<void> {
+    this.addSql('alter table "feed_retry_record" drop column "created_at";');
+  }
+
+}

services/user-feeds/src/feed-event-handler/entities/feed-retry-record.entity.ts

@@ -20,6 +20,12 @@ export class FeedRetryRecord {
   @Property()
   attempts_so_far: number;
 
+  @Property({
+    nullable: true,
+    type: "timestamp with time zone",
+  })
+  created_at: Date = new Date();
+
   constructor(data: Omit<FeedRetryRecord, "id">) {
     this.feed_id = data.feed_id;
     this.attempts_so_far = data.attempts_so_far;

services/user-feeds/src/feed-event-handler/feed-event-handler.service.ts

@@ -41,6 +41,7 @@ import { EntityRepository } from "@mikro-orm/postgresql";
 import { z } from "zod";
 import { CacheStorageService } from "../cache-storage/cache-storage.service";
 import { PartitionedFeedArticleFieldStoreService } from "../articles/partitioned-feed-article-field-store.service";
+import dayjs from "dayjs";
 
 @Injectable()
 export class FeedEventHandlerService {
@@ -366,8 +367,6 @@ export class FeedEventHandlerService {
 
       await this.updateFeedArticlesInCache({ event, articles: allArticles });
 
-      // START TEMPORARY - Should revisit this for a more robust retry strategy
-
       const foundRetryRecord = await this.feedRetryRecordRepo.findOne(
         {
           feed_id: event.data.feed.id,
@@ -389,8 +388,6 @@ export class FeedEventHandlerService {
         });
       }
 
-      // END TEMPORARY
-
       if (!articles.length) {
         this.debugLog(
           `Debug ${event.data.feed.id}: Ignoring feed event due to no` +
@@ -484,17 +481,31 @@ export class FeedEventHandlerService {
           stack: (err as Error).stack,
         });
 
-        const retryRecord = await this.feedRetryRecordRepo.findOne(
-          {
-            feed_id: event.data.feed.id,
-          },
-          {
-            fields: ["id", "attempts_so_far"],
-          }
-        );
+        const retryRecord = await this.feedRetryRecordRepo.findOne({
+          feed_id: event.data.feed.id,
+        });
 
-        // Rudimentary retry to alleviate some pressure
-        if (retryRecord?.attempts_so_far && retryRecord.attempts_so_far >= 8) {
+        const createdAt = retryRecord?.created_at;
+        let disableFeed = false;
+
+        if (createdAt) {
+          const cutoffTime = dayjs(createdAt).add(1, "day");
+
+          if (dayjs().isAfter(cutoffTime)) {
+            logger.info(
+              `Feed ${event.data.feed.id} exceeded retry cutoff date for invalid feed` +
+                `, sending disable event`,
+              {
+                xml: err.feedText,
+              }
+            );
+
+            disableFeed = true;
+          }
+        } else if (
+          retryRecord?.attempts_so_far &&
+          retryRecord.attempts_so_far >= 8
+        ) {
           logger.info(
             `Feed ${event.data.feed.id} exceeded retry limit for invalid feed` +
               `, sending disable event`,
@@ -503,6 +514,11 @@ export class FeedEventHandlerService {
             }
           );
 
+          disableFeed = true;
+        }
+
+        // Rudimentary retry to alleviate some pressure
+        if (disableFeed && retryRecord) {
           this.amqpConnection.publish(
             "",
             MessageBrokerQueue.FeedRejectedDisableFeed,
@@ -532,6 +548,7 @@ export class FeedEventHandlerService {
           await this.feedRetryRecordRepo.upsert({
             feed_id: event.data.feed.id,
             attempts_so_far: (retryRecord?.attempts_so_far || 0) + 1,
+            created_at: retryRecord?.created_at || new Date(),
           });
         }