chris/musicshare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

Make Plex polling faster and its errors specific

Browse Source 
Servers advertise many addresses (public, relay, LAN, Docker bridges).
Use the last one that worked and race the rest only if it fails. Friends'
servers that keep failing back off exponentially instead of slowing
every poll. Errors now name each server and the underlying cause, e.g.
"blackcoffee: ECONNREFUSED", instead of undici's generic "fetch failed".

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MzFywujuH9TiuXbr9vjoX8
This commit is contained in:
chris
2026-10-03 16:02:38 -04:00
co-authored by Claude Opus 5.5
parent 60e6a89297
commit cb3e3d11a9
2 changed files with 112 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/providers/plex.ts
+77 -33
View File
@@ -61,8 +61,11 @@ interface PlexServer {
name: string;
owned: boolean;
tokenEnc: string;
/** candidate base URLs, best first */
/** candidate base URLs, last known good first */
uris: string[];
/** Backoff for shared (friends') servers that keep failing, so they don't slow every poll. */
failures?: number;
retryAt?: number;
}
interface PlexMeta {
@@ -116,26 +119,48 @@ interface PlexSessionTrack {
Player?: { state: 'playing' | 'paused' | 'buffering' };
}
/** Try each candidate URI until one answers; moves the working one to the front. */
const REQUEST_TIMEOUT_MS = 5000;
const MAX_BACKOFF_MS = 30 * 60_000;
/** Human-readable reason, e.g. "ECONNREFUSED" rather than undici's generic "fetch failed". */
function describe(err: unknown): string {
if (err instanceof UnsafeUrlError) return 'only reachable on a private network';
if (err instanceof HttpError) return `HTTP ${err.status}`;
const e = err as { name?: string; message?: string; cause?: { code?: string; message?: string } };
if (e?.name === 'TimeoutError') return 'timed out';
return e?.cause?.code ?? e?.cause?.message ?? e?.message ?? String(err);
}
async function sessionsAt(uri: string, token: string): Promise<PlexSessionTrack[]> {
const base = await assertSafeServerUrl(uri);
const res = await fetchJson<{ MediaContainer: { Metadata?: PlexSessionTrack[] } }>(
`${base}/status/sessions`,
{ headers: plexHeaders(token), ...serverFetchOptions() },
REQUEST_TIMEOUT_MS,
);
return res?.MediaContainer.Metadata ?? [];
}
/**
* Servers advertise many addresses (public, relay, LAN, Docker bridges). Try the
* last one that worked; if it fails, race the rest and remember the winner.
*/
async function fetchSessions(server: PlexServer): Promise<PlexSessionTrack[]> {
const token = decrypt(server.tokenEnc);
let lastError: unknown;
for (const [i, uri] of server.uris.entries()) {
const [first, ...rest] = server.uris;
if (!first) throw new Error('no addresses');
try {
return await sessionsAt(first, token);
} catch (err) {
if (rest.length === 0 || (err instanceof HttpError && err.status === 401)) throw err;
try {
const base = await assertSafeServerUrl(uri);
const res = await fetchJson<{ MediaContainer: { Metadata?: PlexSessionTrack[] } }>(
`${base}/status/sessions`,
{ headers: plexHeaders(token), ...serverFetchOptions() },
5000,
);
if (i > 0) server.uris = [uri, ...server.uris.filter((u) => u !== uri)];
return res?.MediaContainer.Metadata ?? [];
} catch (e) {
lastError = e;
if (e instanceof HttpError && e.status === 401) break; // token no longer valid for this server
const winner = await Promise.any(rest.map((uri) => sessionsAt(uri, token).then((sessions) => ({ uri, sessions }))));
server.uris = [winner.uri, ...server.uris.filter((u) => u !== winner.uri)];
return winner.sessions;
} catch {
throw err; // report why the preferred address failed
}
}
throw lastError instanceof Error ? lastError : new Error(`${server.name}: unreachable`);
}
function isUsersSession(s: PlexSessionTrack, accountId: string, server: PlexServer): boolean {
@@ -154,27 +179,46 @@ export const plexProvider: PollingProvider = {
meta.discoveredAt = Date.now();
}
const results = await Promise.allSettled(servers.map((s) => fetchSessions(s)));
await db
.update(schema.connections)
.set({ meta: { ...meta, servers } })
.where(eq(schema.connections.id, conn.id));
if (servers.length > 0 && results.every((r) => r.status === 'rejected')) {
// force rediscovery next time; server addresses may have changed
await db
.update(schema.connections)
.set({ meta: { ...meta, servers, discoveredAt: 0 } })
.where(eq(schema.connections.id, conn.id));
const reason = (results[0] as PromiseRejectedResult).reason;
if (reason instanceof UnsafeUrlError) throw new Error('Your Plex server is only reachable on a private network');
throw new Error(`Couldn't reach your Plex server${servers.length > 1 ? 's' : ''}: ${reason?.message ?? reason}`);
}
const now = Date.now();
const due = servers.filter((s) => s.owned || !s.retryAt || s.retryAt <= now);
const attempted = due.length > 0 ? due : servers;
const results = await Promise.all(
attempted.map(async (server) => {
try {
const sessions = await fetchSessions(server);
delete server.failures;
delete server.retryAt;
return { server, sessions };
} catch (err) {
if (!server.owned) {
server.failures = (server.failures ?? 0) + 1;
server.retryAt = now + Math.min(MAX_BACKOFF_MS, 60_000 * 2 ** (server.failures - 1));
}
return { server, error: describe(err) };
}
}),
);
const mine = results
.flatMap((r, i) => (r.status === 'fulfilled' ? r.value.map((s) => ({ s, server: servers[i]! })) : []))
.flatMap(({ server, sessions }) => (sessions ?? []).map((s) => ({ s, server })))
.filter(({ s, server }) => s.type === 'track' && isUsersSession(s, conn.externalUserId, server))
.map(({ s }) => s);
const failed = results.filter((r) => r.error);
const ownedAttempted = results.filter((r) => r.server.owned);
const allFailed = failed.length === results.length;
const ownServersDown = ownedAttempted.length > 0 && ownedAttempted.every((r) => r.error);
await db
.update(schema.connections)
// force rediscovery next time if nothing answered; server addresses may have changed
.set({ meta: { ...meta, servers, ...(allFailed ? { discoveredAt: 0 } : {}) } })
.where(eq(schema.connections.id, conn.id));
if (mine.length === 0 && (allFailed || ownServersDown)) {
const details = (ownServersDown && !allFailed ? ownedAttempted : failed).map((r) => `${r.server.name}: ${r.error}`);
throw new Error(`Couldn't reach your Plex server${details.length > 1 ? 's' : ''} (${details.join('; ')})`);
}
const session = mine.find((s) => s.Player?.state === 'playing') ?? mine[0];
if (!session) return null;
apps/api/test/connections.test.ts
+35 -2
View File
@@ -160,7 +160,7 @@ describe('OAuth return URLs', () => {
});
describe('Plex polling', () => {
async function plexConnection(uris: string[]) {
async function plexConnection(uris: string[], extraServers: object[] = []) {
const { user } = await createUser(app(), 'alice');
return upsertConnection({
userId: user.id,
@@ -168,9 +168,14 @@ describe('Plex polling', () => {
externalUserId: '12345',
displayName: null,
accessTokenEnc: encrypt('account-token'),
meta: { discoveredAt: Date.now(), servers: [{ name: 'Home', owned: true, tokenEnc: encrypt('server-token'), uris }] },
meta: {
discoveredAt: Date.now(),
servers: [{ name: 'Home', owned: true, tokenEnc: encrypt('server-token'), uris }, ...extraServers],
},
});
}
const friendServer = (name: string, uri: string) => ({ name, owned: false, tokenEnc: encrypt('t'), uris: [uri] });
const refused = () => Promise.reject(Object.assign(new TypeError('fetch failed'), { cause: { code: 'ECONNREFUSED' } }));
const sessions = {
MediaContainer: {
@@ -208,4 +213,32 @@ describe('Plex polling', () => {
await expect(plexProvider.poll(conn)).rejects.toThrow(/private network/);
expect(fetchMock).not.toHaveBeenCalled();
});
it("backs off a friend's server that keeps failing, so it stops slowing every poll", async () => {
let friendCalls = 0;
vi.stubGlobal(
'fetch',
routeFetch([
[/^https:\/\/8\.8\.8\.8/, () => json(sessions)],
[/^https:\/\/9\.9\.9\.9/, () => (friendCalls++, refused())],
]),
);
const conn = await plexConnection(['https://8.8.8.8:32400'], [friendServer('Friend', 'https://9.9.9.9:32400')]);
expect((await plexProvider.poll(conn))?.track.title).toBe('Digital Love');
const [row] = await db.select().from(schema.connections).where(eq(schema.connections.id, conn.id));
await plexProvider.poll(row!);
expect(friendCalls).toBe(1); // skipped on the second poll
});
it("reports the owner's server being down even when friends' servers answer", async () => {
vi.stubGlobal(
'fetch',
routeFetch([
[/^https:\/\/8\.8\.8\.8/, () => refused()],
[/^https:\/\/9\.9\.9\.9/, () => json({ MediaContainer: {} })],
]),
);
const conn = await plexConnection(['https://8.8.8.8:32400'], [friendServer('Friend', 'https://9.9.9.9:32400')]);
await expect(plexProvider.poll(conn)).rejects.toThrow("Couldn't reach your Plex server (Home: ECONNREFUSED)");
});
});