1import * as Effect from "effect/Effect";
2import * as FileSystem from "effect/FileSystem";
3import { tmpdir } from "node:os";
4import { join } from "node:path";
5
6// AWS permits only one OAM sink per account and Region. Files still run in
7// parallel globally, but the distributed and Lambda quota lanes are separate
8// Bun processes. Hold a service-local cross-process lease for each file's full
9// setup/assert/teardown lifecycle so Sink.test and Bindings.test cannot delete,
10// replace, or throttle each other's singleton sink.
11const profile = (process.env.ALCHEMY_PROFILE ?? "testing").replace(
12 /[^a-zA-Z0-9_.-]/g,
13 "-",
14);
15const uid = process.getuid?.() ?? 0;
16const lockDirectory = join(tmpdir(), `alchemy-test-oam-${uid}-${profile}.lock`);
17const ownerFile = join(lockDirectory, "owner");
18
19const processIsAlive = (pid: number) => {
20 try {
21 process.kill(pid, 0);
22 return true;
23 } catch (error) {
24 return (error as NodeJS.ErrnoException).code === "EPERM";
25 }
26};
27
28const pollInterval = "250 millis";
29// A freshly-acquired lock writes its owner file immediately after the atomic
30// mkdir, so an unreadable owner is normally a sub-second race. If it stays
31// unreadable for this many consecutive polls (~10s) the lock predates a crash
32// or reboot that corrupted the directory (reads can then fail with
33// EFAULT/ENOENT/EPERM in bun) and it is treated as stale.
34const unreadableStalePolls = 40;
35
36/**
37 * Cross-process test lease over a tmpdir lock directory.
38 *
39 * Robustness contract: NO filesystem failure may escape into the test run.
40 * A machine reboot can leave the lock directory in a state where even
41 * `rm`/`read` fail with EFAULT (observed bun edge case on stale tmp entries);
42 * every fs primitive below is therefore best-effort, and a stale lock whose
43 * removal keeps failing is stolen in place by overwriting the owner file.
44 */
45export const makeOamTestLease = () => {
46 let held = false;
47
48 const acquire = Effect.gen(function* () {
49 const fs = yield* FileSystem.FileSystem;
50
51 const tryMakeDir = fs.makeDirectory(lockDirectory).pipe(
52 Effect.as(true),
53 Effect.catchCause(() => Effect.succeed(false)),
54 );
55 const tryWriteOwner = fs
56 .writeFileString(ownerFile, String(process.pid))
57 .pipe(
58 Effect.as(true),
59 Effect.catchCause(() => Effect.succeed(false)),
60 );
61 const tryReadOwnerPid = fs.readFileString(ownerFile).pipe(
62 Effect.map((content) => Number.parseInt(content, 10)),
63 Effect.catchCause(() => Effect.succeed<number | undefined>(undefined)),
64 );
65 const tryRemoveLock = fs.remove(lockDirectory, { recursive: true }).pipe(
66 Effect.as(true),
67 Effect.catchCause(() => Effect.succeed(false)),
68 );
69
70 let unreadablePolls = 0;
71
72 while (true) {
73 if (yield* tryMakeDir) {
74 if (yield* tryWriteOwner) {
75 held = true;
76 return;
77 }
78 // We created the directory but could not record ownership. Back out
79 // (best-effort) so other processes are not stuck waiting on a
80 // headless lock, then retry.
81 yield* tryRemoveLock;
82 yield* Effect.sleep(pollInterval);
83 continue;
84 }
85
86 // The directory already exists — or mkdir itself failed abnormally
87 // (e.g. EFAULT on a reboot-stale path). Inspect the recorded owner.
88 const ownerPid = yield* tryReadOwnerPid;
89
90 if (
91 ownerPid !== undefined &&
92 Number.isFinite(ownerPid) &&
93 processIsAlive(ownerPid)
94 ) {
95 unreadablePolls = 0;
96 yield* Effect.sleep(pollInterval);
97 continue;
98 }
99
100 if (ownerPid === undefined) {
101 // Missing or unreadable owner file: give the current owner a grace
102 // window to finish writing it before declaring the lock stale.
103 unreadablePolls += 1;
104 if (unreadablePolls < unreadableStalePolls) {
105 yield* Effect.sleep(pollInterval);
106 continue;
107 }
108 }
109
110 // Stale lock: the owner is dead, invalid, or unreadable past the grace
111 // window. Prefer removing the directory so the atomic-mkdir path can
112 // re-acquire; if removal itself keeps failing (EFAULT on the stale
113 // directory), steal in place by overwriting the owner file and
114 // verifying we won any concurrent write race.
115 unreadablePolls = 0;
116 if (yield* tryRemoveLock) {
117 continue;
118 }
119 if (yield* tryWriteOwner) {
120 yield* Effect.sleep("100 millis");
121 if ((yield* tryReadOwnerPid) === process.pid) {
122 held = true;
123 return;
124 }
125 }
126 yield* Effect.sleep(pollInterval);
127 }
128 });
129
130 const release = Effect.gen(function* () {
131 if (!held) return;
132 held = false;
133 const fs = yield* FileSystem.FileSystem;
134 const removed = yield* fs.remove(lockDirectory, { recursive: true }).pipe(
135 Effect.as(true),
136 Effect.catchCause(() => Effect.succeed(false)),
137 );
138 if (!removed) {
139 // Could not remove the directory (stale-tmp EFAULT). Best-effort drop
140 // of the owner file so a successor's staleness detection steals the
141 // lock instead of waiting on this (soon-dead) pid. Never fail release.
142 yield* fs.remove(ownerFile).pipe(Effect.catchCause(() => Effect.void));
143 }
144 });
145
146 return { acquire, release };
147};