mirror of
https://github.com/jakejarvis/sofa.git
synced 2026-10-16 01:05:42 -04:00
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 <noreply@anthropic.com>
This commit is contained in:
4 files changed
+151
-26
No files matched your search
+37
-20
@@ -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 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 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() {
|
||||
|
||||
@@ -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<T>(
|
||||
items: readonly T[],
|
||||
fn: (item: T) => Promise<void>,
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -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<void>>(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<void>>();
|
||||
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" }]);
|
||||
});
|
||||
});
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user