diff --git a/bun.lockb b/bun.lockb index bad57d5..83b76e0 100755 Binary files a/bun.lockb and b/bun.lockb differ diff --git a/package.json b/package.json index 1f7172e..e958f0a 100644 --- a/package.json +++ b/package.json @@ -24,6 +24,7 @@ "@opentui/core": "^0.1.77", "@opentui/solid": "^0.1.77", "date-fns": "^4.1.0", + "effect": "^3", "solid-js": "^1.9.9" } } diff --git a/src/effects/feed-refresh.ts b/src/effects/feed-refresh.ts new file mode 100644 index 0000000..562c564 --- /dev/null +++ b/src/effects/feed-refresh.ts @@ -0,0 +1,85 @@ +/** + * Feed-refresh batch as an Effect program. + * + * Replaces the hand-rolled worker pool (mapWithConcurrency) + per-feed + * fetch/apply plumbing in stores/feed.ts with Effect's structured + * concurrency: + * - `Effect.forEach(..., { concurrency })` bounds in-flight fetches to + * `concurrency` (starts exactly that many fibers; each completion pulls + * the next feed — identical semantics to the old shared-counter pool). + * - `Effect.timeout` bounds each feed's fetch to `timeoutMs`. It runs + * through the `Clock` service, so under `TestContext` the TestClock + * drives it deterministically (no real 20s wait in tests). + * - Failures are folded to a null result: a failed or timed-out feed is + * left untouched instead of failing the batch. + * - The apply callback runs inside each feed's own fiber, so a feed's + * refreshed episodes land AS ITS OWN FETCH COMPLETES — the + * per-feed-apply-as-it-lands contract, no Promise.all barrier. + * + * The store boundary (stores/feed.ts) supplies the real fetch and apply + * closures and runs the program with Effect.runPromise. + */ + +import { Duration, Effect } from "effect" +import type { Episode } from "../types/episode" +import type { Feed } from "../types/feed" + +/** Result of fetching one feed's RSS. `episodes: null` means the fetch + * failed or timed out — callers must leave that feed untouched. */ +export interface RefreshFetchResult { + episodes: Episode[] | null + coverUrl: string | undefined +} + +/** Result guaranteed to have parsed episodes (the apply path only). */ +export interface RefreshSuccess { + episodes: Episode[] + coverUrl: string | undefined +} + +export interface RefreshBatchOptions { + /** Max simultaneous in-flight fetches. */ + concurrency: number + /** Per-feed fetch timeout in milliseconds. */ + timeoutMs: number +} + +/** Fold any failure (network error, timeout, rejection) to a null result so + * one bad feed can never fail the batch. */ +const failedResult: RefreshFetchResult = { episodes: null, coverUrl: undefined } + +/** Fetch one feed with a timeout, applying its result as its own fetch + * lands. A failed or timed-out fetch yields null — the feed is untouched. */ +const refreshOne = ( + feed: Feed, + fetchOne: (feed: Feed) => Promise, + applyOne: (feed: Feed, result: RefreshSuccess) => void, + timeoutMs: number, +): Effect.Effect => + Effect.tryPromise(() => fetchOne(feed)).pipe( + Effect.timeout(Duration.millis(timeoutMs)), + Effect.catchAll(() => Effect.succeed(failedResult)), + Effect.flatMap((result) => { + if (result.episodes === null) return Effect.void + // Capture the narrowed array before the closure — TS drops the + // `episodes !== null` narrowing inside Effect.sync's callback. + const episodes = result.episodes + return Effect.sync(() => applyOne(feed, { episodes, coverUrl: result.coverUrl })) + }), + ) + +/** Refresh every feed with bounded concurrency. Each feed's refreshed + * episodes are applied as its own fetch lands (no barrier); a failed or + * timed-out feed is left untouched. The program never fails — failures + * are folded to per-feed no-ops. */ +export const refreshFeedsBatch = ( + feeds: readonly Feed[], + fetchOne: (feed: Feed) => Promise, + applyOne: (feed: Feed, result: RefreshSuccess) => void, + options: RefreshBatchOptions, +): Effect.Effect => + Effect.forEach( + feeds, + (feed) => refreshOne(feed, fetchOne, applyOne, options.timeoutMs), + { concurrency: options.concurrency, discard: true }, + ) diff --git a/src/stores/feed.ts b/src/stores/feed.ts index 25f1898..57ce46e 100644 --- a/src/stores/feed.ts +++ b/src/stores/feed.ts @@ -4,6 +4,8 @@ */ import { createSignal } from "solid-js"; +import { Effect } from "effect"; +import { refreshFeedsBatch } from "../effects/feed-refresh"; import { FeedVisibility } from "../types/feed"; import type { Feed, FeedFilter, FeedSortField } from "../types/feed"; import type { Podcast } from "../types/podcast"; @@ -249,31 +251,6 @@ export function sameRefreshWindow( return fetched.every((e) => signatures.has(episodeSignature(e))); } -/** Run `fn` over every item with at most `limit` executions in flight — a - * classic worker pool. Workers pull indexes from a shared counter, so the - * first `limit` calls start immediately and each completion frees its slot - * for the next item; results are assembled in INPUT order regardless of - * completion order. A hung `fn` holds at most one slot. */ -async function mapWithConcurrency( - items: T[], - limit: number, - fn: (item: T) => Promise, -): Promise { - const results = new Array(items.length); - let nextIndex = 0; - const workers = Array.from( - { length: Math.min(limit, items.length) }, - async () => { - let i: number; - while ((i = nextIndex++) < items.length) { - results[i] = await fn(items[i]); - } - }, - ); - await Promise.all(workers); - return results; -} - function createFeedStore() { const [feeds, setFeeds] = createSignal([]); const [sources, setSources] = createSignal([ @@ -644,40 +621,40 @@ function createFeedStore() { })(), "Refreshing"); }; - /** Refresh all feeds — bounded concurrency (at most FETCH_CONCURRENCY - * in-flight requests), and each feed's refreshed episodes are applied - * AS ITS OWN FETCH LANDS (no Promise.all barrier). Per-feed apply is - * safe because applyRefreshedEpisodes keeps unchanged feeds' object - * identity and lastUpdated (union merge), so each feed's refreshed - * episodes render as its own fetch resolves — the order flapping the - * old atomic barrier existed to hide can no longer happen. */ + /** Refresh all feeds via the Effect batch program (effects/feed-refresh): + * bounded concurrency (at most FETCH_CONCURRENCY in-flight requests) + * and each feed's refreshed episodes applied AS ITS OWN FETCH LANDS + * (no barrier — the apply runs inside the feed's own fiber). Per-feed + * apply is safe because applyRefreshedEpisodes keeps unchanged feeds' + * object identity and lastUpdated (union merge), so each feed's + * refreshed episodes render as its own fetch resolves — the order + * flapping the old atomic barrier existed to hide can no longer + * happen. A failed or timed-out fetch (null episodes) leaves that + * feed untouched. */ const refreshAllFeeds = async () => { setIsLoadingFeeds(true); try { - await mapWithConcurrency( - feeds(), - FETCH_CONCURRENCY, - async (feed) => { - const { episodes, coverUrl } = await fetchEpisodes( - feed.podcast.feedUrl, - MAX_EPISODES_REFRESH, - feed.id, - ); - // A failed fetch (null) leaves that feed untouched. - if (!episodes) return; - setFeeds((prev) => { - let updated = applyRefreshedEpisodes(prev, feed.id, episodes); - if (coverUrl) { - updated = updated.map((f) => - f.id === feed.id && !f.podcast.coverUrl && coverUrl - ? { ...f, podcast: { ...f.podcast, coverUrl } } - : f, - ); - } - if (updated !== prev) scheduleSaveFeeds(); - return updated; - }); - }, + await Effect.runPromise( + refreshFeedsBatch( + feeds(), + (feed) => + fetchEpisodes(feed.podcast.feedUrl, MAX_EPISODES_REFRESH, feed.id), + (feed, { episodes, coverUrl }) => { + setFeeds((prev) => { + let updated = applyRefreshedEpisodes(prev, feed.id, episodes); + if (coverUrl) { + updated = updated.map((f) => + f.id === feed.id && !f.podcast.coverUrl && coverUrl + ? { ...f, podcast: { ...f.podcast, coverUrl } } + : f, + ); + } + if (updated !== prev) scheduleSaveFeeds(); + return updated; + }); + }, + { concurrency: FETCH_CONCURRENCY, timeoutMs: FETCH_TIMEOUT_MS }, + ), ); // Global auto-download: one idempotent pass after the batch. runAutoDownload(); diff --git a/tests/feed-refresh-effect.test.ts b/tests/feed-refresh-effect.test.ts new file mode 100644 index 0000000..2e60624 --- /dev/null +++ b/tests/feed-refresh-effect.test.ts @@ -0,0 +1,230 @@ +/** + * Feed-refresh Effect program tests (src/effects/feed-refresh.ts). + * + * These test the Effect program in isolation — no store singleton, no + * network, no fake timers. The fetch/apply closures are injected, and the + * `Clock` service comes from TestContext's TestClock, so timeouts are driven + * deterministically with TestClock.adjust instead of real 20s waits. + * + * Contracts pinned here (mirrored at the store level by + * feed-nonblocking.test.ts / feed-refresh.test.ts against a real Bun.serve): + * 1. Bounded concurrency — never more than `concurrency` fetches in + * flight, and the pool pulls the next feed as one completes. + * 2. Per-feed apply as its own fetch lands (no barrier). + * 3. A timed-out fetch leaves that feed untouched and does not stall the + * batch (TestClock.adjust fires the timeout deterministically). + * 4. A rejecting fetch leaves that feed untouched and does not fail the + * batch. + */ + +import { test, expect } from "bun:test" +import { Duration, Effect, Fiber, TestClock, TestContext } from "effect" +import { + refreshFeedsBatch, + type RefreshFetchResult, +} from "../src/effects/feed-refresh" +import type { Feed } from "../src/types/feed" +import type { Podcast } from "../src/types/podcast" +import type { Episode } from "../src/types/episode" + +const makePodcast = (id: string): Podcast => ({ + id, + title: `Show ${id}`, + description: `Show ${id} description`, + feedUrl: `http://example.com/${id}.xml`, + lastUpdated: new Date(0), + isSubscribed: true, +}) + +const makeFeed = (id: string): Feed => ({ + id, + podcast: makePodcast(id), + episodes: [], + visibility: "public" as Feed["visibility"], + sourceId: "test", + lastUpdated: new Date(0), + isPinned: false, +}) + +const makeEpisode = (id: string): Episode => ({ + id, + podcastId: "pod", + title: `Ep ${id}`, + description: "", + audioUrl: `https://example.com/${id}.mp3`, + duration: 60, + pubDate: new Date(0), +}) + +/** Resolve an episode result without dragging in the full RSS shape. */ +const ok = (episodeIds: string[]): RefreshFetchResult => ({ + episodes: episodeIds.map(makeEpisode), + coverUrl: undefined, +}) + +/** One macrotask turn — lets microtask-scheduled Effect fibers run. */ +const tick = (): Promise => { + const { promise, resolve } = Promise.withResolvers() + setImmediate(resolve) + return promise +} + +/** A resolvable fetch gate: the pool parks on `promise` until the test + * resolves it. (Promise.withResolvers's return type is not in tsconfig's + * ES2015.Promise lib, hence the explicit shape.) */ +interface Gate { + promise: Promise + resolve: (value: RefreshFetchResult) => void +} + +/** Poll `cond` across up to `iterations` event-loop turns. */ +async function pollUntil( + cond: () => boolean, + iterations = 500, +): Promise { + for (let i = 0; i < iterations; i++) { + if (cond()) return true + await tick() + } + return cond() +} + +test("bounds in-flight fetches to the configured concurrency", async () => { + const feeds = Array.from({ length: 10 }, (_, i) => makeFeed(`feed-${i}`)) + let inFlight = 0 + let maxInFlight = 0 + const gates: Gate[] = [] + const applied: string[] = [] + + const program = refreshFeedsBatch( + feeds, + (feed) => { + inFlight++ + if (inFlight > maxInFlight) maxInFlight = inFlight + const gate = Promise.withResolvers() + gates.push(gate) + return gate.promise.finally(() => { + inFlight-- + }) + }, + (feed) => { + applied.push(feed.id) + }, + { concurrency: 4, timeoutMs: 60_000 }, + ) + + // Run the batch in flight (NOT awaited) and observe the pool from + // outside via the gate side effects. + const done = Effect.runPromise(program) + // The pool starts exactly `concurrency` fetches up front. + const sawStart = await pollUntil(() => gates.length >= 4) + expect(sawStart).toBe(true) + expect(maxInFlight).toBe(4) + expect(gates.length).toBe(4) + + // Resolve one gate: the pool pulls the next feed, still bounded at 4. + gates[0].resolve(ok(["a"])) + const sawPull = await pollUntil(() => gates.length >= 5) + expect(sawPull).toBe(true) + expect(maxInFlight).toBeLessThanOrEqual(4) + + // Release everything, re-draining as the pool pulls new gates, until + // every feed has been fetched and applied. + while (applied.length < 10) { + for (const gate of gates.splice(0)) gate.resolve(ok(["x"])) + await tick() + } + await done + expect(maxInFlight).toBeLessThanOrEqual(4) + expect(applied).toHaveLength(10) +}) + +test("applies each feed as its own fetch lands (no barrier)", async () => { + const a = makeFeed("a") + const b = makeFeed("b") + const applied: string[] = [] + let bCalled = false + const gateB = Promise.withResolvers() + + const program = refreshFeedsBatch( + [a, b], + (feed) => { + if (feed.id === "a") return Promise.resolve(ok(["a-1"])) + bCalled = true + return gateB.promise + }, + (feed) => { + applied.push(feed.id) + }, + { concurrency: 4, timeoutMs: 60_000 }, + ) + + // Run the batch in flight; A's fetch resolves and applies while B's is + // still parked at the gate. + const done = Effect.runPromise(program) + const aApplied = await pollUntil(() => applied.includes("a")) + expect(aApplied).toBe(true) + expect(bCalled).toBe(true) + expect(applied).toEqual(["a"]) + expect(applied).not.toContain("b") + + gateB.resolve(ok(["b-1"])) + await done + expect(applied).toEqual(["a", "b"]) +}) + +test("a timed-out fetch leaves that feed untouched, without stalling the batch", async () => { + const fast = makeFeed("fast") + const hung = makeFeed("hung") + const applied: string[] = [] + // A promise that never settles — the fetch hangs past the timeout. + const never = new Promise(() => {}) + + const program = refreshFeedsBatch( + [fast, hung], + (feed) => + feed.id === "fast" + ? Promise.resolve(ok(["f-1"])) + : never, + (feed) => { + applied.push(feed.id) + }, + { concurrency: 4, timeoutMs: 5_000 }, + ) + + const timed = Effect.gen(function* () { + const fiber = yield* Effect.fork(program) + // Advance the TestClock past the timeout: the hung fetch's + // Effect.timeout fires deterministically — no real 5s wait. + yield* TestClock.adjust(Duration.millis(5_000)) + yield* Fiber.join(fiber) + }) + await Effect.runPromise( + timed.pipe(Effect.provide(TestContext.TestContext)), + ) + + // The fast feed applied; the hung one was dropped, and the batch + // completed anyway. + expect(applied).toEqual(["fast"]) +}) + +test("a rejecting fetch leaves that feed untouched and does not fail the batch", async () => { + const bad = makeFeed("bad") + const good = makeFeed("good") + const applied: string[] = [] + + const program = refreshFeedsBatch( + [bad, good], + (feed) => + feed.id === "bad" + ? Promise.reject(new Error("feed exploded")) + : Promise.resolve(ok(["g-1"])), + (feed) => { + applied.push(feed.id) + }, + { concurrency: 4, timeoutMs: 60_000 }, + ) + + await Effect.runPromise(program) + expect(applied).toEqual(["good"]) +})