{"version":3,"file":"ports.cjs","names":["Data","Context","Effect","createServer","Layer","claimedPorts"],"sources":["../../src/infra/ports.ts"],"sourcesContent":["import { createServer } from \"node:net\";\nimport { Context, Data, Effect, Layer } from \"effect\";\nimport { claimedPorts } from \"../services/processes/registry\";\n\nconst PROBE_TIMEOUT_MS = 250;\nconst MAX_PORT_SCAN_STEPS = 1000;\nconst PARALLEL_PROBE_WINDOW = 8;\nconst BLOCK_STEP = 100;\n\nexport type PortBudget = { min: number; max: number };\n\nexport class PortAllocationError extends Data.TaggedError(\"PortAllocationError\")<{\n preferred: number;\n budget?: PortBudget;\n cause?: unknown;\n}> {}\n\nexport interface PortBlockEntry {\n key: string;\n preferred: number;\n pinned: boolean;\n}\n\nexport interface PortBlockRequest {\n base: number;\n step?: number;\n entries: PortBlockEntry[];\n}\n\nexport interface PortBlockAllocation {\n base: number;\n ports: Record;\n conflicts: Array<{ key: string; port: number; claimed: boolean }>;\n}\n\nexport class PortAllocator extends Context.Service<\n PortAllocator,\n {\n pickAvailable: (\n preferred: number,\n budget?: PortBudget,\n ) => Effect.Effect;\n acquireBlock: (\n request: PortBlockRequest,\n ) => Effect.Effect;\n }\n>()(\"PortAllocator\") {}\n\nconst scheduleTimeout = (ms: number, fn: () => void): ReturnType =>\n setTimeout(fn, ms);\n\nexport function probePortBindable(port: number): Effect.Effect {\n return Effect.callback((resume) => {\n const server = createServer();\n\n server.once(\"listening\", () => {\n server.close(() => {\n resume(Effect.succeed(true));\n });\n });\n\n server.once(\"error\", () => {\n server.removeAllListeners();\n // EADDRINUSE, EACCES, or any other bind error → not available\n resume(Effect.succeed(false));\n });\n\n server.listen(port, \"127.0.0.1\");\n\n const timer = scheduleTimeout(PROBE_TIMEOUT_MS, () => {\n server.removeAllListeners();\n try {\n server.close();\n } catch {\n // ignore\n }\n resume(Effect.succeed(false));\n });\n\n server.once(\"listening\", () => clearTimeout(timer));\n server.once(\"error\", () => clearTimeout(timer));\n });\n}\n\nfunction pickAvailablePort(\n preferred: number,\n usedPorts: Set,\n budget?: PortBudget,\n): Effect.Effect {\n return Effect.gen(function* () {\n const within = (candidate: number): boolean =>\n !budget || (candidate >= budget.min && candidate <= budget.max);\n\n let port = preferred;\n if (!within(port)) {\n port = budget ? budget.min : port;\n }\n\n const ceiling = budget ? budget.max + 1 : Number.MAX_SAFE_INTEGER;\n let steps = 0;\n\n const fail = () =>\n Effect.fail(\n new PortAllocationError({\n preferred,\n budget,\n cause: budget\n ? `No free port in budget [${budget.min}, ${budget.max}] starting from ${preferred}`\n : `No free port found starting from ${preferred} within ${MAX_PORT_SCAN_STEPS} steps`,\n }),\n );\n\n while (true) {\n if (port >= ceiling || steps > MAX_PORT_SCAN_STEPS) {\n return yield* fail();\n }\n\n const candidates: number[] = [];\n for (let i = 0; i < PARALLEL_PROBE_WINDOW && port + i < ceiling; i++) {\n const candidate = port + i;\n if (!usedPorts.has(candidate)) {\n candidates.push(candidate);\n }\n }\n\n if (candidates.length === 0) {\n port += PARALLEL_PROBE_WINDOW;\n steps += PARALLEL_PROBE_WINDOW;\n continue;\n }\n\n const results = yield* Effect.forEach(\n candidates,\n (c) => probePortBindable(c).pipe(Effect.map((free) => ({ port: c, free }))),\n { concurrency: \"unbounded\" },\n );\n\n const firstFree = results.find((r) => r.free);\n if (firstFree) {\n usedPorts.add(firstFree.port);\n return firstFree.port;\n }\n\n port += PARALLEL_PROBE_WINDOW;\n steps += PARALLEL_PROBE_WINDOW;\n }\n });\n}\n\nexport const PortAllocatorLive: Layer.Layer = Layer.sync(PortAllocator, () => {\n const usedPorts = claimedPorts();\n return {\n pickAvailable: (preferred, budget) => pickAvailablePort(preferred, usedPorts, budget),\n acquireBlock: (request) => acquireBlockPort(request, usedPorts),\n };\n});\n\n// acquireBlockPort validates and acquires a whole port block atomically: every\n// entry is probed before any is yielded; on any conflict the entire block\n// steps (base += step) and retries. Pinned entries (explicit CLI flags) are\n// absolute — an actual listener on one fails loudly instead of silently\n// drifting a port the user named; a registry claim without a listener is\n// stale (PID reuse across container restarts) and never wedges allocation.\nfunction acquireBlockPort(\n request: PortBlockRequest,\n usedPorts: Set,\n): Effect.Effect {\n const step = request.step ?? BLOCK_STEP;\n const fail = (cause: string) =>\n Effect.fail(\n new PortAllocationError({\n preferred: request.base,\n cause,\n }),\n );\n\n return Effect.gen(function* () {\n let base = request.base;\n let steps = 0;\n let firstConflicts: PortBlockAllocation[\"conflicts\"] = [];\n while (steps <= MAX_PORT_SCAN_STEPS) {\n const shift = base - request.base;\n const candidates = request.entries.map((entry) => ({\n ...entry,\n port: entry.pinned ? entry.preferred : entry.preferred + shift,\n }));\n\n // Pinned entries are absolute (ADR 0012): an actual listener on one\n // fails loudly instead of drifting a port the user named. Verified\n // against the real socket every iteration — a registry claim without\n // a listener is stale (PID reuse across container restarts) and must\n // never wedge a pinned allocation.\n let pinnedHeldByListener: (typeof candidates)[number] | undefined;\n for (const candidate of candidates.filter((c) => c.pinned)) {\n const bindable = yield* probePortBindable(candidate.port);\n if (!bindable) {\n pinnedHeldByListener = candidate;\n break;\n }\n usedPorts.delete(candidate.port);\n }\n if (pinnedHeldByListener) {\n return yield* fail(\n `explicitly-requested port ${pinnedHeldByListener.port} (${pinnedHeldByListener.key}) is occupied by a listener`,\n );\n }\n\n const unpinned = candidates.filter((c) => !c.pinned);\n const claimConflicts = unpinned\n .filter((c) => usedPorts.has(c.port))\n .map((c) => ({ key: c.key, port: c.port, claimed: true }));\n if (shift === 0) {\n firstConflicts = claimConflicts;\n }\n\n if (claimConflicts.length === 0) {\n const results = yield* Effect.forEach(\n unpinned,\n (c) => probePortBindable(c.port).pipe(Effect.map((free) => ({ ...c, free }))),\n { concurrency: \"unbounded\" },\n );\n const busy = results.filter((r) => !r.free);\n if (shift === 0) {\n firstConflicts = busy.map((b) => ({ key: b.key, port: b.port, claimed: false }));\n }\n if (busy.length === 0) {\n for (const candidate of candidates) usedPorts.add(candidate.port);\n return {\n base,\n ports: Object.fromEntries(candidates.map((c) => [c.key, c.port])),\n conflicts: firstConflicts,\n };\n }\n }\n\n base += step;\n steps += 1;\n }\n return yield* fail(\n `no free port block within ${MAX_PORT_SCAN_STEPS} steps of base ${request.base} (step ${step})`,\n );\n });\n}\n"],"mappings":";;;;;AAIA,MAAM,mBAAmB;AACzB,MAAM,sBAAsB;AAC5B,MAAM,wBAAwB;AAC9B,MAAM,aAAa;AAInB,IAAa,sBAAb,cAAyCA,YAAK,YAAY,qBAAqB,CAAC,CAI7E,CAAC;AAoBJ,IAAa,gBAAb,cAAmCC,eAAQ,QAWzC,CAAC,CAAC,eAAe,CAAC,CAAC,CAAC;AAEtB,MAAM,mBAAmB,IAAY,OACnC,WAAW,IAAI,EAAE;AAEnB,SAAgB,kBAAkB,MAAsC;CACtE,OAAOC,cAAO,UAAmB,WAAW;EAC1C,MAAM,aAASC,uBAAa;EAE5B,OAAO,KAAK,mBAAmB;GAC7B,OAAO,YAAY;IACjB,OAAOD,cAAO,QAAQ,IAAI,CAAC;GAC7B,CAAC;EACH,CAAC;EAED,OAAO,KAAK,eAAe;GACzB,OAAO,mBAAmB;GAE1B,OAAOA,cAAO,QAAQ,KAAK,CAAC;EAC9B,CAAC;EAED,OAAO,OAAO,MAAM,WAAW;EAE/B,MAAM,QAAQ,gBAAgB,wBAAwB;GACpD,OAAO,mBAAmB;GAC1B,IAAI;IACF,OAAO,MAAM;GACf,QAAQ,CAER;GACA,OAAOA,cAAO,QAAQ,KAAK,CAAC;EAC9B,CAAC;EAED,OAAO,KAAK,mBAAmB,aAAa,KAAK,CAAC;EAClD,OAAO,KAAK,eAAe,aAAa,KAAK,CAAC;CAChD,CAAC;AACH;AAEA,SAAS,kBACP,WACA,WACA,QAC4C;CAC5C,OAAOA,cAAO,IAAI,aAAa;EAC7B,MAAM,UAAU,cACd,CAAC,UAAW,aAAa,OAAO,OAAO,aAAa,OAAO;EAE7D,IAAI,OAAO;EACX,IAAI,CAAC,OAAO,IAAI,GACd,OAAO,SAAS,OAAO,MAAM;EAG/B,MAAM,UAAU,SAAS,OAAO,MAAM,IAAI,OAAO;EACjD,IAAI,QAAQ;EAEZ,MAAM,aACJA,cAAO,KACL,IAAI,oBAAoB;GACtB;GACA;GACA,OAAO,SACH,2BAA2B,OAAO,IAAI,IAAI,OAAO,IAAI,kBAAkB,cACvE,oCAAoC,UAAU,UAAU,oBAAoB;EAClF,CAAC,CACH;EAEF,OAAO,MAAM;GACX,IAAI,QAAQ,WAAW,QAAQ,qBAC7B,OAAO,OAAO,KAAK;GAGrB,MAAM,aAAuB,CAAC;GAC9B,KAAK,IAAI,IAAI,GAAG,IAAI,yBAAyB,OAAO,IAAI,SAAS,KAAK;IACpE,MAAM,YAAY,OAAO;IACzB,IAAI,CAAC,UAAU,IAAI,SAAS,GAC1B,WAAW,KAAK,SAAS;GAE7B;GAEA,IAAI,WAAW,WAAW,GAAG;IAC3B,QAAQ;IACR,SAAS;IACT;GACF;GAQA,MAAM,aAAY,OANKA,cAAO,QAC5B,aACC,MAAM,kBAAkB,CAAC,CAAC,CAAC,KAAKA,cAAO,KAAK,UAAU;IAAE,MAAM;IAAG;GAAK,EAAE,CAAC,GAC1E,EAAE,aAAa,YAAY,CAC7B,EAEyB,CAAC,MAAM,MAAM,EAAE,IAAI;GAC5C,IAAI,WAAW;IACb,UAAU,IAAI,UAAU,IAAI;IAC5B,OAAO,UAAU;GACnB;GAEA,QAAQ;GACR,SAAS;EACX;CACF,CAAC;AACH;AAEA,MAAa,oBAAgDE,aAAM,KAAK,qBAAqB;CAC3F,MAAM,YAAYC,8BAAa;CAC/B,OAAO;EACL,gBAAgB,WAAW,WAAW,kBAAkB,WAAW,WAAW,MAAM;EACpF,eAAe,YAAY,iBAAiB,SAAS,SAAS;CAChE;AACF,CAAC;AAQD,SAAS,iBACP,SACA,WACyD;CACzD,MAAM,OAAO,QAAQ,QAAQ;CAC7B,MAAM,QAAQ,UACZH,cAAO,KACL,IAAI,oBAAoB;EACtB,WAAW,QAAQ;EACnB;CACF,CAAC,CACH;CAEF,OAAOA,cAAO,IAAI,aAAa;EAC7B,IAAI,OAAO,QAAQ;EACnB,IAAI,QAAQ;EACZ,IAAI,iBAAmD,CAAC;EACxD,OAAO,SAAS,qBAAqB;GACnC,MAAM,QAAQ,OAAO,QAAQ;GAC7B,MAAM,aAAa,QAAQ,QAAQ,KAAK,WAAW;IACjD,GAAG;IACH,MAAM,MAAM,SAAS,MAAM,YAAY,MAAM,YAAY;GAC3D,EAAE;GAOF,IAAI;GACJ,KAAK,MAAM,aAAa,WAAW,QAAQ,MAAM,EAAE,MAAM,GAAG;IAE1D,IAAI,EAAC,OADmB,kBAAkB,UAAU,IAAI,IACzC;KACb,uBAAuB;KACvB;IACF;IACA,UAAU,OAAO,UAAU,IAAI;GACjC;GACA,IAAI,sBACF,OAAO,OAAO,KACZ,6BAA6B,qBAAqB,KAAK,IAAI,qBAAqB,IAAI,4BACtF;GAGF,MAAM,WAAW,WAAW,QAAQ,MAAM,CAAC,EAAE,MAAM;GACnD,MAAM,iBAAiB,SACpB,QAAQ,MAAM,UAAU,IAAI,EAAE,IAAI,CAAC,CAAC,CACpC,KAAK,OAAO;IAAE,KAAK,EAAE;IAAK,MAAM,EAAE;IAAM,SAAS;GAAK,EAAE;GAC3D,IAAI,UAAU,GACZ,iBAAiB;GAGnB,IAAI,eAAe,WAAW,GAAG;IAM/B,MAAM,QAAO,OALUA,cAAO,QAC5B,WACC,MAAM,kBAAkB,EAAE,IAAI,CAAC,CAAC,KAAKA,cAAO,KAAK,UAAU;KAAE,GAAG;KAAG;IAAK,EAAE,CAAC,GAC5E,EAAE,aAAa,YAAY,CAC7B,EACoB,CAAC,QAAQ,MAAM,CAAC,EAAE,IAAI;IAC1C,IAAI,UAAU,GACZ,iBAAiB,KAAK,KAAK,OAAO;KAAE,KAAK,EAAE;KAAK,MAAM,EAAE;KAAM,SAAS;IAAM,EAAE;IAEjF,IAAI,KAAK,WAAW,GAAG;KACrB,KAAK,MAAM,aAAa,YAAY,UAAU,IAAI,UAAU,IAAI;KAChE,OAAO;MACL;MACA,OAAO,OAAO,YAAY,WAAW,KAAK,MAAM,CAAC,EAAE,KAAK,EAAE,IAAI,CAAC,CAAC;MAChE,WAAW;KACb;IACF;GACF;GAEA,QAAQ;GACR,SAAS;EACX;EACA,OAAO,OAAO,KACZ,6BAA6B,oBAAoB,iBAAiB,QAAQ,KAAK,SAAS,KAAK,EAC/F;CACF,CAAC;AACH"}