atproto pds in zig pds.zat.dev
pds atproto
Something went wrong. Try again.
JavaScript
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071const [url, roundsText = "100", concurrencyText = "32"] = process.argv.slice(2);
if (!url) { console.error("usage: websocket_churn.mjs URL [ROUNDS] [CONCURRENCY]"); process.exit(2);}
const rounds = Number.parseInt(roundsText, 10);const concurrency = Number.parseInt(concurrencyText, 10);
if (!Number.isSafeInteger(rounds) || rounds < 1 || !Number.isSafeInteger(concurrency) || concurrency < 1) { console.error("rounds and concurrency must be positive integers"); process.exit(2);}
function connectOnce() { return new Promise((resolve, reject) => { const socket = new WebSocket(url); const timeout = setTimeout(() => { socket.close(); reject(new Error("websocket close timed out")); }, 5000);
socket.addEventListener("open", () => socket.close(), { once: true }); socket.addEventListener("close", () => { clearTimeout(timeout); resolve(); }, { once: true }); socket.addEventListener("error", () => { clearTimeout(timeout); reject(new Error("websocket connection failed")); }, { once: true }); });}
// Long-lived cursored subscribers must not exhaust the bounded backfill// slots: each slot is released once the subscriber catches up to live, so// more held connections than slots all handshake successfully.function connectAndHold(cursorUrl, sockets) { return new Promise((resolve, reject) => { const socket = new WebSocket(cursorUrl); socket.binaryType = "arraybuffer"; sockets.push(socket); const timeout = setTimeout(() => reject(new Error("held websocket open timed out")), 5000); socket.addEventListener("open", () => { clearTimeout(timeout); setTimeout(resolve, 300); }, { once: true }); socket.addEventListener("error", () => { clearTimeout(timeout); reject(new Error("held cursored websocket rejected")); }, { once: true }); });}
const heldText = process.argv[5];if (heldText) { const held = Number.parseInt(heldText, 10); const sockets = []; const cursorUrl = `${url}${url.includes("?") ? "&" : "?"}cursor=0`; for (let i = 0; i < held; i += 1) { await connectAndHold(cursorUrl, sockets); } for (const socket of sockets) socket.close();} else { for (let round = 0; round < rounds; round += 1) { await Promise.all(Array.from({ length: concurrency }, connectOnce)); }}