From 2e7f2667dc704364428366205b07c2a91b3e982b Mon Sep 17 00:00:00 2001 From: Jake Jarvis Date: Sun, 4 Oct 2026 14:58:46 -0400 Subject: [PATCH] fix(cron): isolate per-title failures in refresh jobs Wrap the four title-looping cron jobs in a runIsolated helper so one failing title (e.g. removed from TMDB) no longer aborts the whole run; the job only fails when every item fails. Also include library shell titles (NULL lastFetchedAt) in the nightly stale query so they get hydrated. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cron.ts | 63 ++++++++++++++++--------- packages/core/src/cron.ts | 29 ++++++++++++ packages/core/test/cron.test.ts | 82 +++++++++++++++++++++++++++++++-- packages/db/src/queries/cron.ts | 9 +++- 4 files changed, 154 insertions(+), 29 deletions(-) diff --git a/apps/server/src/cron.ts b/apps/server/src/cron.ts index f9c9106..3e7f913 100644 --- a/apps/server/src/cron.ts +++ b/apps/server/src/cron.ts @@ -16,6 +16,7 @@ import { getTitleByIdForCron, getTitleIdsWithStaleSeasons, getTitlesWithFreshRecommendations, + runIsolated, startCronRun, } from "@sofa/core/cron"; import { @@ -107,20 +108,28 @@ async function nightlyRefreshLibrary() { // Library titles: 7 days const staleLibrary = getStaleLibraryTitles(libraryIds, libraryStale); - for (const { id } of staleLibrary) { - await refreshTitle(id); - await Bun.sleep(RATE_LIMIT_MS); - } + await runIsolated( + staleLibrary.map((t) => t.id), + async (id) => { + await refreshTitle(id); + await Bun.sleep(RATE_LIMIT_MS); + }, + (id, err) => log.warn(`Failed to refresh title ${id}:`, err), + ); // Non-library titles: 30 days - const nonLibrary = getStaleNonLibraryTitlesForRefresh(nonLibraryStale, 50); + const nonLibraryIds = getStaleNonLibraryTitlesForRefresh(nonLibraryStale, 50) + .map((t) => t.id) + .filter((id) => !libraryIds.includes(id)); - for (const t of nonLibrary) { - if (!libraryIds.includes(t.id)) { - await refreshTitle(t.id); + await runIsolated( + nonLibraryIds, + async (id) => { + await refreshTitle(id); await Bun.sleep(RATE_LIMIT_MS); - } - } + }, + (id, err) => log.warn(`Failed to refresh title ${id}:`, err), + ); } // Refresh availability for library titles where stale @@ -131,12 +140,14 @@ async function refreshAvailabilityJob() { const { withOffers, withStaleOffers } = getStaleAvailabilityTitles(libraryIds, stale); - for (const titleId of libraryIds) { - if (withStaleOffers.has(titleId) || !withOffers.has(titleId)) { - await refreshAvailability(titleId); + await runIsolated( + libraryIds.filter((id) => withStaleOffers.has(id) || !withOffers.has(id)), + async (id) => { + await refreshAvailability(id); await Bun.sleep(RATE_LIMIT_MS); - } - } + }, + (id, err) => log.warn(`Failed to refresh availability for title ${id}:`, err), + ); } async function refreshRecommendationsJob() { @@ -148,10 +159,14 @@ async function refreshRecommendationsJob() { `Refreshing recommendations for ${staleIds.length} of ${libraryIds.length} library titles`, ); - for (const titleId of staleIds) { - await refreshRecommendations(titleId); - await Bun.sleep(RATE_LIMIT_MS); - } + await runIsolated( + staleIds, + async (id) => { + await refreshRecommendations(id); + await Bun.sleep(RATE_LIMIT_MS); + }, + (id, err) => log.warn(`Failed to refresh recommendations for title ${id}:`, err), + ); } async function refreshTvChildrenJob() { @@ -164,14 +179,16 @@ async function refreshTvChildrenJob() { const tvIds = tvShows.map((s) => s.id); const titlesWithStaleSeasons = getTitleIdsWithStaleSeasons(tvIds, stale); - for (const show of tvShows) { - if (titlesWithStaleSeasons.has(show.id)) { + await runIsolated( + tvShows.filter((s) => titlesWithStaleSeasons.has(s.id)), + async (show) => { const details = await getTvDetails(show.tmdbId); await refreshTvChildren(show.id, show.tmdbId, details.number_of_seasons); await syncTvChildArt(show.id, { warmCache: true }); await Bun.sleep(RATE_LIMIT_MS); - } - } + }, + (show, err) => log.warn(`Failed to refresh episodes for TV show ${show.id}:`, err), + ); } async function cacheImagesJob() { diff --git a/packages/core/src/cron.ts b/packages/core/src/cron.ts index 0b4a89b..db6ebd5 100644 --- a/packages/core/src/cron.ts +++ b/packages/core/src/cron.ts @@ -14,6 +14,35 @@ import { updateCronRunSuccess, } from "@sofa/db/queries/cron"; +/** + * Run `fn` for each item in order, isolating failures so one bad item (e.g. a + * title TMDB has removed) doesn't abort the rest of a cron job. Each failure is + * reported via `onItemError`. Throws only when at least one item was attempted + * and every attempt failed, so a total outage still marks the run as failed. + */ +export async function runIsolated( + items: readonly T[], + fn: (item: T) => Promise, + onItemError: (item: T, err: unknown) => void, +): Promise<{ attempted: number; failed: number }> { + let failed = 0; + let firstError: unknown; + for (const item of items) { + try { + await fn(item); + } catch (err) { + failed++; + firstError ??= err; + onItemError(item, err); + } + } + if (items.length > 0 && failed === items.length) { + const reason = firstError instanceof Error ? firstError.message : String(firstError); + throw new Error(`All ${items.length} items failed (first error: ${reason})`); + } + return { attempted: items.length, failed }; +} + export function startCronRun(jobName: string) { return insertCronRunReturning(jobName); } diff --git a/packages/core/test/cron.test.ts b/packages/core/test/cron.test.ts index ea72659..3802fab 100644 --- a/packages/core/test/cron.test.ts +++ b/packages/core/test/cron.test.ts @@ -1,9 +1,15 @@ -import { beforeEach, describe, expect, test } from "vitest"; +import { beforeEach, describe, expect, test, vi } from "vitest"; -import { cronRuns } from "@sofa/db/schema"; -import { clearAllTables, eq, testDb } from "@sofa/test/db"; +import { cronRuns, titles } from "@sofa/db/schema"; +import { clearAllTables, eq, insertTitle, testDb } from "@sofa/test/db"; -import { completeCronRun, failCronRun, startCronRun } from "../src/cron"; +import { + completeCronRun, + failCronRun, + getStaleLibraryTitles, + runIsolated, + startCronRun, +} from "../src/cron"; beforeEach(() => { clearAllTables(); @@ -51,3 +57,71 @@ describe("failCronRun", () => { expect(row?.errorMessage).toBe("string error"); }); }); + +describe("runIsolated", () => { + test("continues after a failing item", async () => { + const fn = vi.fn<(item: number) => Promise>(async (item) => { + if (item === 2) throw new Error("boom"); + }); + const onItemError = vi.fn<(item: number, err: unknown) => void>(); + + const result = await runIsolated([1, 2, 3], fn, onItemError); + + expect(fn).toHaveBeenCalledTimes(3); + expect(onItemError).toHaveBeenCalledTimes(1); + expect(onItemError).toHaveBeenCalledWith(2, expect.any(Error)); + expect(result).toEqual({ attempted: 3, failed: 1 }); + }); + + test("throws when every item fails", async () => { + const onItemError = vi.fn<(item: number, err: unknown) => void>(); + await expect( + runIsolated( + [1, 2], + async () => { + throw new Error("down"); + }, + onItemError, + ), + ).rejects.toThrow("All 2 items failed"); + expect(onItemError).toHaveBeenCalledTimes(2); + }); + + test("resolves for an empty list", async () => { + const fn = vi.fn<(item: number) => Promise>(); + const result = await runIsolated([], fn, vi.fn<(item: number, err: unknown) => void>()); + expect(result).toEqual({ attempted: 0, failed: 0 }); + expect(fn).not.toHaveBeenCalled(); + }); +}); + +describe("getStaleLibraryTitles", () => { + const staleDate = new Date("2026-01-08T00:00:00Z"); + + function setFetched(id: string, date: Date) { + testDb.update(titles).set({ lastFetchedAt: date }).where(eq(titles.id, id)).run(); + } + + test("returns a shell title with NULL lastFetchedAt", () => { + insertTitle({ id: "t-shell", tmdbId: 1 }); + expect(getStaleLibraryTitles(["t-shell"], staleDate)).toEqual([{ id: "t-shell" }]); + }); + + test("returns a title fetched before the stale date", () => { + insertTitle({ id: "t-old", tmdbId: 2 }); + setFetched("t-old", new Date("2026-01-01T00:00:00Z")); + expect(getStaleLibraryTitles(["t-old"], staleDate)).toEqual([{ id: "t-old" }]); + }); + + test("does not return a title fetched after the stale date", () => { + insertTitle({ id: "t-fresh", tmdbId: 3 }); + setFetched("t-fresh", new Date("2026-01-09T00:00:00Z")); + expect(getStaleLibraryTitles(["t-fresh"], staleDate)).toEqual([]); + }); + + test("does not return titles outside the passed id list", () => { + insertTitle({ id: "t-shell", tmdbId: 1 }); + insertTitle({ id: "t-other", tmdbId: 4 }); + expect(getStaleLibraryTitles(["t-shell"], staleDate)).toEqual([{ id: "t-shell" }]); + }); +}); diff --git a/packages/db/src/queries/cron.ts b/packages/db/src/queries/cron.ts index 06f6574..c1f2fcc 100644 --- a/packages/db/src/queries/cron.ts +++ b/packages/db/src/queries/cron.ts @@ -1,4 +1,4 @@ -import { and, eq, gte, inArray, isNotNull, lt, or } from "drizzle-orm"; +import { and, eq, gte, inArray, isNotNull, isNull, lt, or } from "drizzle-orm"; import { db } from "../client"; import { @@ -47,7 +47,12 @@ export function getStaleTitles(titleIds: string[], staleDate: Date) { return db .select({ id: titles.id }) .from(titles) - .where(and(inArray(titles.id, titleIds), lt(titles.lastFetchedAt, staleDate))) + .where( + and( + inArray(titles.id, titleIds), + or(isNull(titles.lastFetchedAt), lt(titles.lastFetchedAt, staleDate)), + ), + ) .all(); }