96a5740 · atproto-bridge

Changed files (6) +175 -9

pds/lib/deadline.js +54 -0

…	…	@@ -0,0 +1,54 @@
	1	// withDeadline — a settle guarantee for upstream calls.
	2	//
	3	// Both `doFetch` editions (server-bare.js over bare-http1, server.js over global fetch) reject only
	4	// when the request emits an error. That covers a REFUSED connection (~5ms) but NOT an unreachable
	5	// HOST: with packets blackholed, bare-http1's global agent (`agent.js:217`, timeout 5000) destroys
	6	// the socket, and `client-connection.js:73-75` turns that into a bare `req.destroy()` — no error
	7	// event, so nothing rejects and the proxied XRPC hangs until the caller gives up. Measured: still
	8	// pending after 30s. Passing `timeout` per-request does not help either, because the agent merges
	9	// its own opts LAST (`agent.js:128`), overwriting yours.
	10	//
	11	// So the deadline lives here, above the transport, and hands back an `onTimeout` hook so the caller
	12	// can destroy the in-flight request rather than leak the socket.
	13
	14	export const UPSTREAM_TIMEOUT = 'UPSTREAM_TIMEOUT'
	15
	16	/**
	17	* Race `promise` against a deadline.
	18	*
	19	* @param {Promise} promise the in-flight upstream call
	20	* @param {number} timeoutMs deadline in ms; <= 0 (or non-finite) disables the guard
	21	* @param {Function} [onTimeout] called ONCE if the deadline blows — destroy the request here
	22	* @returns {Promise} the promise's value, or rejects with a UPSTREAM_TIMEOUT error
	23	*/
	24	export function withDeadline (promise, timeoutMs, onTimeout = null) {
	25	if (!Number.isFinite(timeoutMs) \|\| timeoutMs <= 0) return promise
	26
	27	return new Promise((resolve, reject) => {
	28	let done = false
	29
	30	const timer = setTimeout(() => {
	31	if (done) return
	32	done = true
	33	// Destroy the request FIRST, then reject: a caller that retries another upstream must not be
	34	// racing a socket we have already given up on.
	35	if (onTimeout) { try { onTimeout() } catch { /* a failed abort must not mask the timeout */ } }
	36	const err = new Error(`upstream timed out after ${timeoutMs}ms`)
	37	err.code = UPSTREAM_TIMEOUT
	38	err.timeoutMs = timeoutMs
	39	reject(err)
	40	}, timeoutMs)
	41	// NOT unref'd, deliberately. An unref'd deadline is skipped whenever nothing else holds the loop
	42	// open — i.e. it stops guaranteeing a settle in exactly the idle case. It is always cleared below,
	43	// so it outlives the call by nothing.
	44
	45	promise.then(
	46	(v) => { if (done) return; done = true; clearTimeout(timer); resolve(v) },
	47	// An upstream error is passed through VERBATIM. Masking ECONNREFUSED as a timeout would make
	48	// "the box is off" and "the process is down" indistinguishable — they need different responses.
	49	(e) => { if (done) return; done = true; clearTimeout(timer); reject(e) }
	50	)
	51	})
	52	}
	53
	54	export default withDeadline

pds/lib/favela-proxy.js +8 -2

…	…	@@ -31,7 +31,13 @@ export function buildMembraneDidDoc (serverDid, { favelaWebUpstream = null } = {
31	31	// Proxy one GET to favela-web. upstream = the gateway base URL (config.favelaWebUpstream, e.g.
32	32	// http://10.22.0.1:4041). doFetch(url,{method,headers}) → { status, contentType, body:Buffer } (the same
33	33	// injected fetch proxyToAppView uses). 501 when unconfigured; 502 on upstream failure.
34		export async function proxyToFavela (req, res, { nsid, search, cors, doFetch, upstream }) {
	34	// The lens is co-located or one VPN hop away, and bsky imposes its own (undocumented) ceiling on a
	35	// feed generator, so a lens read gets a much tighter deadline than a general upstream call. A blown
	36	// deadline is a 502 like any other upstream failure — the difference is that it now HAPPENS instead
	37	// of hanging the request forever when the lens host is unreachable.
	38	export const FAVELA_TIMEOUT_MS = 2000
	39
	40	export async function proxyToFavela (req, res, { nsid, search, cors, doFetch, upstream, timeoutMs = FAVELA_TIMEOUT_MS }) {
35	41	if (!upstream) {
36	42	const b = Buffer.from(JSON.stringify({ error: 'NotConfigured', message: 'favela-web upstream not configured' }))
37	43	res.writeHead(501, { 'content-type': 'application/json', 'content-length': b.length, ...cors })
…	…	@@ -39,7 +45,7 @@ export async function proxyToFavela (req, res, { nsid, search, cors, doFetch, up
39	45	}
40	46	let up
41	47	try {
42		up = await doFetch(`${String(upstream).replace(/\/$/, '')}/xrpc/${nsid}${search \|\| ''}`, { method: 'GET', headers: {} })
	48	up = await doFetch(`${String(upstream).replace(/\/$/, '')}/xrpc/${nsid}${search \|\| ''}`, { method: 'GET', headers: {}, timeoutMs })
43	49	} catch (e) {
44	50	const b = Buffer.from(JSON.stringify({ error: 'UpstreamFailure', message: `favela proxy failed: ${e.message}` }))
45	51	res.writeHead(502, { 'content-type': 'application/json', 'content-length': b.length, ...cors })

pds/lib/server-bare.js +16 -3

…	…	@@ -13,6 +13,11 @@ import { tidNow } from 'atproto-store/lib/tid.js'
13	13	import { buildRecordProofCar } from './sync-get-record.js'
14	14	import { proxyToAppView, isAppViewMethod } from './appview.js'
15	15	import { proxyToFavela, isFavelaMethod, buildMembraneDidDoc } from './favela-proxy.js'
	16	import { withDeadline } from './deadline.js'
	17
	18	// Ceiling for ANY upstream call (AppView proxy, describeRepo's PLC read). Generous on purpose — this
	19	// exists to kill hangs, not slow-but-live responses. The lens proxy sets its own, much tighter, value.
	20	const DEFAULT_UPSTREAM_MS = 20000
16	21
17	22	// The caller's bearer token (access/refresh JWT) — passed to the peer, which is the auth authority.
18	23	const bearer = (req) => (req.headers.authorization \|\| '').replace(/^Bearer\s+/i, '')
…	…	@@ -22,10 +27,17 @@ const jwtSub = (jwt) => { try { const p = String(jwt).split('.')[1].replace(/-/g
22	27	// Upstream client (bare): request → { status, contentType, body:Buffer }. Picks http vs https by the URL
23	28	// scheme — the AppView proxy targets https://api.bsky.app, but the favela lens-export proxy targets a plain
24	29	// http:// gateway over the VPN, and https.request on an http URL hangs on a TLS handshake (the proxy-hang bug).
25		function doFetch (url, { method, headers, body }) {
	30	// ⛔ The deadline is NOT optional. Without it this rejects only on the request's 'error' event, which
	31	// an unreachable HOST never produces: the agent times the socket out and destroys it, and
	32	// client-connection turns that into a bare `req.destroy()` with no error — so the promise dangles and
	33	// the proxied XRPC hangs until the caller gives up (measured: >30s against a blackholed host, vs ~5ms
	34	// to reject a REFUSED connection). A per-request `timeout` option cannot fix it either — the global
	35	// agent merges its own opts last and overwrites it.
	36	function doFetch (url, { method, headers, body, timeoutMs = DEFAULT_UPSTREAM_MS }) {
26	37	const mod = String(url).startsWith('http:') ? http : https
27		return new Promise((resolve, reject) => {
28		const req = mod.request(url, { method, headers }, (res) => {
	38	let req = null
	39	const call = new Promise((resolve, reject) => {
	40	req = mod.request(url, { method, headers }, (res) => {
29	41	const chunks = []
30	42	res.on('data', (c) => chunks.push(c))
31	43	res.on('end', () => resolve({ status: res.statusCode, contentType: res.headers['content-type'], body: Buffer.concat(chunks) }))
…	…	@@ -34,6 +46,7 @@ function doFetch (url, { method, headers, body }) {
34	46	if (body) req.write(body)
35	47	req.end()
36	48	})
	49	return withDeadline(call, timeoutMs, () => { try { req.destroy() } catch {} })
37	50	}
38	51
39	52	// bare-ws (CJS) exports { Server, Socket }; default-import the exports object, then take each.

pds/lib/server.js +14 -3

…	…	@@ -9,11 +9,22 @@ import { createServer } from 'node:http'
9	9	import { WebSocketServer } from 'ws'
10	10	import { proxyToAppView, isAppViewMethod } from './appview.js'
11	11	import { buildMembraneDidDoc } from './favela-proxy.js'
	12	import { withDeadline } from './deadline.js'
	13
	14	// Twin of the bare edition's ceiling — kills hangs, not slow responses.
	15	const DEFAULT_UPSTREAM_MS = 20000
12	16
13	17	// AppView upstream client (node): global fetch → { status, contentType, body:Buffer }.
14		async function doFetch (url, { method, headers, body }) {
15		const r = await fetch(url, { method, headers, body })
16		return { status: r.status, contentType: r.headers.get('content-type'), body: Buffer.from(await r.arrayBuffer()) }
	18	// Deadline for the same reason as the bare edition (see server-bare.js): global fetch has no default
	19	// timeout either, so an unreachable upstream hangs the proxied request instead of failing it. Twin of
	20	// the guard there — changing one and not the other is the "two surfaces" bug this file has hit before.
	21	async function doFetch (url, { method, headers, body, timeoutMs = DEFAULT_UPSTREAM_MS }) {
	22	const ac = new AbortController()
	23	const call = (async () => {
	24	const r = await fetch(url, { method, headers, body, signal: ac.signal })
	25	return { status: r.status, contentType: r.headers.get('content-type'), body: Buffer.from(await r.arrayBuffer()) }
	26	})()
	27	return withDeadline(call, timeoutMs, () => { try { ac.abort() } catch {} })
17	28	}
18	29
19	30	// Lexicon-correct JSON: top-level CIDs (format: cid) → strings; CIDs/bytes INSIDE a

pds/test/deadline.test.js +61 -0

…	…	@@ -0,0 +1,61 @@
	1	// withDeadline — the settle guarantee both doFetch editions were missing.
	2	//
	3	// WHY THIS EXISTS: `doFetch` rejects only on the request's 'error' event. When the upstream HOST is
	4	// gone (VPN blackhole — packets dropped, no RST), bare-http1's global agent times the socket out at
	5	// 5s and destroys it, which reaches the request as a plain `destroy()` with NO error. Nothing rejects,
	6	// so the promise dangles and the proxied XRPC hangs until the CLIENT gives up. Measured against a
	7	// blackholed host: still pending after 30s. A connection REFUSED rejects in ~5ms, so the failure is
	8	// specific to "host unreachable", which is exactly the tabertron-is-off case.
	9	//
	10	// The fix is a deadline that (a) rejects, and (b) hands back a hook so the caller can destroy the
	11	// in-flight request instead of leaking the socket.
	12	import test from 'brittle'
	13	import { withDeadline, UPSTREAM_TIMEOUT } from '../lib/deadline.js'
	14
	15	const never = () => new Promise(() => {})
	16	const after = (ms, v) => new Promise((resolve) => setTimeout(() => resolve(v), ms))
	17
	18	test('passes a value through when it settles inside the deadline', async (t) => {
	19	const r = await withDeadline(after(5, 'ok'), 200)
	20	t.is(r, 'ok', 'the resolved value is returned untouched')
	21	})
	22
	23	test('rejects with a typed timeout error when it does not', async (t) => {
	24	await t.exception(() => withDeadline(never(), 30), /timed out/i, 'rejects rather than dangling')
	25	try {
	26	await withDeadline(never(), 30)
	27	} catch (e) {
	28	t.is(e.code, UPSTREAM_TIMEOUT, 'carries a code the proxy can branch on (failover vs pass-through)')
	29	t.is(e.timeoutMs, 30, 'reports the deadline it blew')
	30	}
	31	})
	32
	33	test('calls the abort hook exactly once on timeout — the socket must not leak', async (t) => {
	34	let aborts = 0
	35	await t.exception(() => withDeadline(never(), 30, () => { aborts++ }), /timed out/i)
	36	await after(60)
	37	t.is(aborts, 1, 'aborted once, and not again after the rejection')
	38	})
	39
	40	test('does NOT abort when the call finishes in time', async (t) => {
	41	let aborts = 0
	42	const r = await withDeadline(after(5, 'ok'), 200, () => { aborts++ })
	43	await after(250) // outlive the original deadline — a cleared timer must not fire
	44	t.is(r, 'ok')
	45	t.is(aborts, 0, 'timer cleared on success')
	46	})
	47
	48	test('an upstream error passes through unchanged, never masked as a timeout', async (t) => {
	49	const boom = Object.assign(new Error('ECONNREFUSED'), { code: 'ECONNREFUSED' })
	50	try {
	51	await withDeadline(Promise.reject(boom), 200)
	52	t.fail('should have rejected')
	53	} catch (e) {
	54	t.is(e.code, 'ECONNREFUSED', 'the real cause survives — refused is not a timeout')
	55	}
	56	})
	57
	58	test('a non-positive deadline disables the guard (opt-out stays possible)', async (t) => {
	59	const r = await withDeadline(after(20, 'ok'), 0)
	60	t.is(r, 'ok', 'ms<=0 means no deadline')
	61	})

pds/test/favela-proxy.test.js +22 -1

…	…	@@ -1,6 +1,7 @@
1	1	// Favela lens-export proxy — the membrane forwards feed-gen + labeler service reads to favela-web.
2	2	import test from 'brittle'
3		import { isFavelaMethod, proxyToFavela } from '../lib/favela-proxy.js'
	3	import { isFavelaMethod, proxyToFavela, FAVELA_TIMEOUT_MS } from '../lib/favela-proxy.js'
	4	import { UPSTREAM_TIMEOUT } from '../lib/deadline.js'
4	5
5	6	function mockRes () {
6	7	const out = { status: 0, headers: null, body: null }
…	…	@@ -38,6 +39,26 @@ test('proxyToFavela → 501 when no upstream is configured', async (t) => {
38	39	t.is(JSON.parse(res.out.body).error, 'NotConfigured', 'error body')
39	40	})
40	41
	42	test('proxyToFavela passes a tight deadline down to doFetch', async (t) => {
	43	let opts = null
	44	const doFetch = async (url, o) => { opts = o; return { status: 200, contentType: 'application/json', body: Buffer.from('{}') } }
	45	await proxyToFavela({ headers: {} }, mockRes(), { nsid: 'app.bsky.feed.getFeedSkeleton', cors: CORS, doFetch, upstream: 'http://10.22.0.1:4041' })
	46	t.is(opts.timeoutMs, FAVELA_TIMEOUT_MS, 'lens reads use the lens ceiling, not the general one')
	47	t.ok(FAVELA_TIMEOUT_MS > 0 && FAVELA_TIMEOUT_MS <= 5000, 'and it is tight enough to fail INSIDE bsky\'s own patience')
	48	})
	49
	50	test('proxyToFavela → 502 when the lens host is unreachable (the hang case)', async (t) => {
	51	// What a blown deadline looks like coming out of doFetch. Before withDeadline this call never
	52	// settled at all, so this 502 is the whole point: an unreachable lens must FAIL, not hang.
	53	const timeout = Object.assign(new Error('upstream timed out after 2000ms'), { code: UPSTREAM_TIMEOUT })
	54	const res = mockRes()
	55	await proxyToFavela({ headers: {} }, res, {
	56	nsid: 'app.bsky.feed.getFeedSkeleton', cors: CORS, doFetch: async () => { throw timeout }, upstream: 'http://10.22.0.1:4041'
	57	})
	58	t.is(res.out.status, 502, 'answers instead of dangling')
	59	t.ok(JSON.parse(res.out.body).message.includes('timed out'), 'and says WHY — host unreachable, not a bad request')
	60	})
	61
41	62	test('proxyToFavela → 502 on upstream failure', async (t) => {
42	63	const res = mockRes()
43	64	await proxyToFavela({ headers: {} }, res, { nsid: 'app.bsky.feed.getFeedSkeleton', cors: CORS, doFetch: async () => { throw new Error('ECONNREFUSED') }, upstream: 'http://10.22.0.1:4041' })

atproto-bridge commits / 96a5740

Changed files (6) +175 -9