History

Make relay install lock more idiomatic Effect

Cursor Agent committed 5 months ago

Commit 5feb87d1 parent 3ea6adf2 files · +197 −54Browse files

Unified · Split

2 files
  1. packages/shared/src/relayClient.test.ts +118 −17
  2. packages/shared/src/relayClient.ts +79 −37
packages/shared/src/relayClient.test.ts+118−17
@@ -1,13 +1,17 @@
11 import { sha256 } from "@noble/hashes/sha2";
22 import * as NodeServices from "@effect/platform-node/NodeServices";
3−import { describe, expect, it } from "@effect/vitest";
3+import { assert, describe, it } from "@effect/vitest";
44 import * as ConfigProvider from "effect/ConfigProvider";
5+import * as Duration from "effect/Duration";
56 import * as Effect from "effect/Effect";
67 import * as Encoding from "effect/Encoding";
78 import * as FileSystem from "effect/FileSystem";
9+import * as Fiber from "effect/Fiber";
810 import * as Layer from "effect/Layer";
11+import * as Path from "effect/Path";
912 import * as Sink from "effect/Sink";
1013 import * as Stream from "effect/Stream";
14+import * as TestClock from "effect/testing/TestClock";
1115 import { HttpClient, HttpClientResponse } from "effect/unstable/http";
1216 import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
1317
@@ -47,15 +51,13 @@ const makeHttpClientLayer = (bytes: Uint8Array) => · 33 unmodified lines
4751 );
4852
4953 const makeSpawnerLayer = (commands: Array<string>) =>
50− Layer.succeed(
51− ChildProcessSpawner.ChildProcessSpawner,
52− ChildProcessSpawner.make((command) =>
54+ Layer.mock(ChildProcessSpawner.ChildProcessSpawner, {
55+ spawn: (command) =>
5356 Effect.sync(() => {
5457 commands.push(ChildProcess.isStandardCommand(command) ? command.command : "piped-command");
5558 return makeHandle();
5659 }),
57− ),
58− );
60+ });
5961
6062 describe("RelayClient", () => {
6163 it.effect("resolves explicit overrides before managed and PATH executables", () =>
@@ -80,7 +82,7 @@ describe("RelayClient", () => { · 18 unmodified lines
8082 }),
8183 });
8284
83− expect(yield* manager.resolve).toEqual({
85+ assert.deepStrictEqual(yield* manager.resolve, {
8486 status: "available",
8587 executablePath: overridePath,
8688 source: "override",
@@ -130,16 +132,17 @@ describe("RelayClient", () => { · 43 unmodified lines
130132 platform: "linux",
131133 arch: "x64",
132134 });
133− expect(installed).toEqual({
135+ assert.deepStrictEqual(installed, {
134136 status: "available",
135137 executablePath: managedPath,
136138 source: "managed",
137139 version: CLOUDFLARED_VERSION,
138140 });
139− expect(new TextDecoder().decode(yield* fileSystem.readFile(managedPath))).toBe(
141+ assert.equal(
142+ new TextDecoder().decode(yield* fileSystem.readFile(managedPath)),
140143 "test-cloudflared-binary",
141144 );
142− expect(progress).toEqual([
145+ assert.deepStrictEqual(progress, [
143146 "checking",
144147 "waiting_for_lock",
145148 "downloading",
@@ -148,7 +151,7 @@ describe("RelayClient", () => { · 2 unmodified lines
148151 "validating",
149152 "activating",
150153 ]);
151− expect(yield* manager.resolve).toEqual(installed);
154+ assert.deepStrictEqual(yield* manager.resolve, installed);
152155 }).pipe(
153156 Effect.scoped,
154157 Effect.provide(
@@ -180,8 +183,8 @@ describe("RelayClient", () => { · 25 unmodified lines
180183 });
181184
182185 const error = yield* manager.install.pipe(Effect.flip);
183− expect(error).toBeInstanceOf(RelayClientInstallError);
184− expect(error.reason).toBe("invalid_checksum");
186+ assert.ok(error instanceof RelayClientInstallError);
187+ assert.equal(error.reason, "invalid_checksum");
185188 }).pipe(
186189 Effect.scoped,
187190 Effect.provide(
@@ -217,8 +220,8 @@ describe("RelayClient", () => { · 29 unmodified lines
217220 const [first, second] = yield* Effect.all([manager.install, manager.install], {
218221 concurrency: "unbounded",
219222 });
220− expect(second).toEqual(first);
221− expect(commands).toHaveLength(1);
223+ assert.deepStrictEqual(second, first);
224+ assert.equal(commands.length, 1);
222225 }).pipe(
223226 Effect.scoped,
224227 Effect.provide(
@@ -243,7 +246,7 @@ describe("RelayClient", () => { · 18 unmodified lines
243246 configProvider: () => ConfigProvider.fromEnv({ env: { PATH: path } }),
244247 });
245248
246− expect(yield* manager.resolve).toEqual({
249+ assert.deepStrictEqual(yield* manager.resolve, {
247250 status: "missing",
248251 version: CLOUDFLARED_VERSION,
249252 });
@@ -253,7 +256,7 @@ describe("RelayClient", () => { · 3 unmodified lines
253256 yield* fileSystem.chmod(executablePath, 0o755);
254257 path = binDir;
255258
256− expect(yield* manager.resolve).toEqual({
259+ assert.deepStrictEqual(yield* manager.resolve, {
257260 status: "available",
258261 executablePath,
259262 source: "path",
@@ -270,4 +273,102 @@ describe("RelayClient", () => { · 10 unmodified lines
270273 ),
271274 ),
272275 );
276+
277+ it.effect("removes stale install locks before installing", () => {
278+ const commands: Array<string> = [];
279+ const bytes = new TextEncoder().encode("test-cloudflared-binary");
280+ return Effect.gen(function* () {
281+ const fileSystem = yield* FileSystem.FileSystem;
282+ const path = yield* Path.Path;
283+ const baseDir = yield* fileSystem.makeTempDirectoryScoped({
284+ prefix: "t3-cloudflared-test-",
285+ });
286+ const managedPath = resolveManagedCloudflaredPath({
287+ baseDir,
288+ platform: "linux",
289+ arch: "x64",
290+ });
291+ const lockPath = `${managedPath}.lock`;
292+ yield* fileSystem.makeDirectory(path.dirname(lockPath), { recursive: true });
293+ yield* fileSystem.writeFileString(lockPath, "stale");
294+ yield* fileSystem.utimes(lockPath, 0, 0);
295+ yield* TestClock.adjust(Duration.minutes(6));
296+
297+ const manager = yield* makeCloudflaredRelayClient({
298+ baseDir,
299+ platform: "linux",
300+ arch: "x64",
301+ releaseAsset: {
302+ url: "https://example.test/cloudflared",
303+ sha256: Encoding.encodeHex(sha256(bytes)),
304+ archive: "binary",
305+ },
306+ configProvider: emptyConfigProvider,
307+ });
308+
309+ const installed = yield* manager.install;
310+
311+ assert.deepStrictEqual(installed, {
312+ status: "available",
313+ executablePath: managedPath,
314+ source: "managed",
315+ version: CLOUDFLARED_VERSION,
316+ });
317+ assert.equal(yield* fileSystem.exists(lockPath), false);
318+ assert.equal(commands.length, 1);
319+ }).pipe(
320+ Effect.scoped,
321+ Effect.provide(
322+ Layer.mergeAll(NodeServices.layer, makeHttpClientLayer(bytes), makeSpawnerLayer(commands)),
323+ ),
324+ );
325+ });
326+
327+ it.effect("times out install lock waits with the test clock", () => {
328+ const bytes = new TextEncoder().encode("test-cloudflared-binary");
329+ return Effect.gen(function* () {
330+ const fileSystem = yield* FileSystem.FileSystem;
331+ const path = yield* Path.Path;
332+ const baseDir = yield* fileSystem.makeTempDirectoryScoped({
333+ prefix: "t3-cloudflared-test-",
334+ });
335+ const managedPath = resolveManagedCloudflaredPath({
336+ baseDir,
337+ platform: "linux",
338+ arch: "x64",
339+ });
340+ const lockPath = `${managedPath}.lock`;
341+ yield* fileSystem.makeDirectory(path.dirname(lockPath), { recursive: true });
342+ yield* fileSystem.writeFileString(lockPath, "locked");
343+
344+ const manager = yield* makeCloudflaredRelayClient({
345+ baseDir,
346+ platform: "linux",
347+ arch: "x64",
348+ releaseAsset: {
349+ url: "https://example.test/cloudflared",
350+ sha256: Encoding.encodeHex(sha256(bytes)),
351+ archive: "binary",
352+ },
353+ configProvider: emptyConfigProvider,
354+ });
355+
356+ const install = yield* manager.install.pipe(Effect.flip, Effect.fork);
357+ yield* TestClock.adjust(Duration.seconds(10));
358+ const error = yield* Fiber.join(install);
359+
360+ assert.ok(error instanceof RelayClientInstallError);
361+ assert.equal(error.reason, "install_locked");
362+ assert.equal(yield* fileSystem.exists(lockPath), true);
363+ }).pipe(
364+ Effect.scoped,
365+ Effect.provide(
366+ Layer.mergeAll(
367+ NodeServices.layer,
368+ makeHttpClientLayer(bytes),
369+ makeSpawnerLayer([]),
370+ ),
371+ ),
372+ );
373+ });
273374 });
packages/shared/src/relayClient.ts+79−37
@@ -1,4 +1,3 @@
1−import * as Clock from "effect/Clock";
21 import type {
32 RelayClientInstallProgressEvent,
43 RelayClientInstallProgressStage,
@@ -8,6 +7,8 @@ import * as ConfigProvider from "effect/ConfigProvider"; · 3 unmodified lines
87 import * as Context from "effect/Context";
98 import * as Crypto from "effect/Crypto";
109 import * as Data from "effect/Data";
10+import * as DateTime from "effect/DateTime";
11+import * as Duration from "effect/Duration";
1112 import * as Effect from "effect/Effect";
1213 import * as Encoding from "effect/Encoding";
1314 import * as FileSystem from "effect/FileSystem";
@@ -15,6 +16,7 @@ import * as Layer from "effect/Layer"; · 1 unmodified line
1516 import * as Option from "effect/Option";
1617 import * as Path from "effect/Path";
1718 import * as PlatformError from "effect/PlatformError";
19+import * as Schedule from "effect/Schedule";
1820 import * as Semaphore from "effect/Semaphore";
1921 import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http";
2022 import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
@@ -99,8 +101,18 @@ const CLOUDFLARED_RELEASE_ASSETS: Readonly< · 78 unmodified lines
99101 };
100102
101103 const INSTALL_LOCK_RETRY_COUNT = 100;
102−const INSTALL_LOCK_RETRY_DELAY = "100 millis";
103−const INSTALL_LOCK_STALE_MS = 5 * 60 * 1_000;
104+const INSTALL_LOCK_RETRY_DELAY = Duration.millis(100);
105+const INSTALL_LOCK_TIMEOUT = Duration.times(INSTALL_LOCK_RETRY_DELAY, INSTALL_LOCK_RETRY_COUNT);
106+const INSTALL_LOCK_STALE_AGE = Duration.minutes(5);
107+
108+class RelayClientInstallLockBusy extends Data.TaggedError("RelayClientInstallLockBusy")<{
109+ readonly lockPath: string;
110+}> {}
111+
112+const retryWhileInstallLockBusy = Schedule.spaced(INSTALL_LOCK_RETRY_DELAY).pipe(
113+ Schedule.setInputType<RelayClientInstallLockBusy | PlatformError.PlatformError>(),
114+ Schedule.while(({ input }) => input._tag === "RelayClientInstallLockBusy"),
115+);
104116
105117 const trimmedString = (name: string) =>
106118 Config.string(name).pipe(
@@ -161,8 +173,8 @@ export function resolveManagedCloudflaredPath(input: { · 54 unmodified lines
161173 function resolveReleaseAsset(
162174 platform: NodeJS.Platform,
163175 arch: string,
164−): CloudflaredReleaseAsset | null {
165− return CLOUDFLARED_RELEASE_ASSETS[`${platform}-${arch}`] ?? null;
176+): Option.Option<CloudflaredReleaseAsset> {
177+ return Option.fromUndefinedOr(CLOUDFLARED_RELEASE_ASSETS[`${platform}-${arch}`]);
166178 }
167179
168180 function isAlreadyExists(error: PlatformError.PlatformError): boolean {
@@ -207,7 +219,9 @@ export const makeCloudflaredRelayClient = Effect.fn("cloudflared.make")(function · 38 unmodified lines
207219 const installSemaphore = yield* Semaphore.make(1);
208220 const platform = options.platform ?? process.platform;
209221 const arch = options.arch ?? process.arch;
210− const releaseAsset = options.releaseAsset ?? resolveReleaseAsset(platform, arch);
222+ const releaseAsset = Option.fromUndefinedOr(options.releaseAsset).pipe(
223+ Option.orElse(() => resolveReleaseAsset(platform, arch)),
224+ );
211225 const loadCloudflaredConfig = Effect.suspend(() =>
212226 CloudflaredConfig.pipe(
213227 Effect.provideService(
@@ -236,15 +250,15 @@ export const makeCloudflaredRelayClient = Effect.fn("cloudflared.make")(function · 22 unmodified lines
236250 const resolvePathExecutable = Effect.gen(function* () {
237251 const config = yield* loadCloudflaredConfig;
238252 const pathValue = Option.getOrUndefined(config.path);
239− if (!pathValue) return null;
253+ if (!pathValue) return Option.none();
240254 const delimiter = platform === "win32" ? ";" : ":";
241255 for (const directory of pathValue.split(delimiter)) {
242256 const trimmed = directory.trim().replace(/^"|"$/gu, "");
243257 if (trimmed.length === 0) continue;
244258 const candidate = path.join(trimmed, executableFileName(platform));
245− if (yield* isExecutableFile(candidate)) return candidate;
259+ if (yield* isExecutableFile(candidate)) return Option.some(candidate);
246260 }
247− return null;
261+ return Option.none();
248262 });
249263
250264 const resolve: RelayClientShape["resolve"] = Effect.gen(function* () {
@@ -268,15 +282,15 @@ export const makeCloudflaredRelayClient = Effect.fn("cloudflared.make")(function · 17 unmodified lines
268282 };
269283 }
270284 const pathExecutable = yield* resolvePathExecutable;
271− if (pathExecutable) {
285+ if (Option.isSome(pathExecutable)) {
272286 return {
273287 status: "available",
274− executablePath: pathExecutable,
288+ executablePath: pathExecutable.value,
275289 source: "path",
276290 version: CLOUDFLARED_VERSION,
277291 };
278292 }
279− return releaseAsset
293+ return Option.isSome(releaseAsset)
280294 ? { status: "missing", version: CLOUDFLARED_VERSION }
281295 : {
282296 status: "unsupported",
@@ -351,31 +365,59 @@ export const makeCloudflaredRelayClient = Effect.fn("cloudflared.make")(function · 68 unmodified lines
351365 return bytes;
352366 });
353367
368+ const isInstallLockStale = Effect.fn("cloudflared.isInstallLockStale")(function* (
369+ lockPath: string,
370+ ) {
371+ const lockInfo = yield* fileSystem.stat(lockPath).pipe(Effect.option);
372+ const lockModifiedAt = Option.flatMap(lockInfo, (info) => info.mtime);
373+ if (Option.isNone(lockModifiedAt)) return false;
374+
375+ const now = yield* DateTime.now;
376+ const lockAge = DateTime.distance(DateTime.makeUnsafe(lockModifiedAt.value), now);
377+ return Duration.isGreaterThan(lockAge, INSTALL_LOCK_STALE_AGE);
378+ });
379+
380+ const attemptAcquireInstallLock = Effect.fn("cloudflared.attemptAcquireInstallLock")(function* (
381+ lockPath: string,
382+ ) {
383+ const acquired = yield* fileSystem.writeFileString(lockPath, "", { flag: "wx" }).pipe(
384+ Effect.as(true),
385+ Effect.catch((error) => (isAlreadyExists(error) ? Effect.succeed(false) : Effect.fail(error))),
386+ );
387+ if (acquired) return;
388+
389+ if (yield* isInstallLockStale(lockPath)) {
390+ yield* fileSystem.remove(lockPath, { force: true });
391+ return yield* attemptAcquireInstallLock(lockPath);
392+ }
393+
394+ return yield* new RelayClientInstallLockBusy({ lockPath });
395+ });
396+
354397 const acquireInstallLock = Effect.fn("cloudflared.acquireInstallLock")(function* (
355398 lockPath: string,
356399 ) {
357− for (let attempt = 0; attempt < INSTALL_LOCK_RETRY_COUNT; attempt += 1) {
358− const acquired = yield* fileSystem.writeFileString(lockPath, "", { flag: "wx" }).pipe(
359− Effect.as(true),
360− Effect.catch((error) =>
361− isAlreadyExists(error) ? Effect.succeed(false) : Effect.fail(error),
400+ return yield* attemptAcquireInstallLock(lockPath).pipe(
401+ Effect.retry(retryWhileInstallLockBusy),
402+ Effect.timeoutOrElse({
403+ duration: INSTALL_LOCK_TIMEOUT,
404+ orElse: () =>
405+ Effect.fail(
406+ new RelayClientInstallError({
407+ reason: "install_locked",
408+ message: "Another relay client installation is still in progress.",
409+ }),
410+ ),
411+ }),
412+ Effect.catchTag("RelayClientInstallLockBusy", () =>
413+ Effect.fail(
414+ new RelayClientInstallError({
415+ reason: "install_locked",
416+ message: "Another relay client installation is still in progress.",
417+ }),
362418 ),
363− );
364− if (acquired) return;
365−
366− const now = yield* Clock.currentTimeMillis;
367− const lockInfo = yield* fileSystem.stat(lockPath).pipe(Effect.option);
368− const mtime = Option.flatMap(lockInfo, (info) => info.mtime);
369− if (Option.isSome(mtime) && now - mtime.value.getTime() > INSTALL_LOCK_STALE_MS) {
370− yield* fileSystem.remove(lockPath, { force: true });
371− continue;
372− }
373− yield* Effect.sleep(INSTALL_LOCK_RETRY_DELAY);
374− }
375− return yield* new RelayClientInstallError({
376− reason: "install_locked",
377− message: "Another relay client installation is still in progress.",
378− });
419+ ),
420+ );
379421 });
380422
381423 const installUnlocked = Effect.fn("cloudflared.installUnlocked")(function* (
@@ -391,7 +433,7 @@ export const makeCloudflaredRelayClient = Effect.fn("cloudflared.make")(function · 9 unmodified lines
391433 message: `${CLOUDFLARED_PATH_ENV_NAME} does not point to an executable file.`,
392434 });
393435 }
394− if (!releaseAsset) {
436+ if (Option.isNone(releaseAsset)) {
395437 return yield* new RelayClientInstallError({
396438 reason: "unsupported_platform",
397439 message: `T3 Code does not provide a managed relay client binary for ${platform}-${arch}.`,
@@ -427,16 +469,16 @@ export const makeCloudflaredRelayClient = Effect.fn("cloudflared.make")(function · 29 unmodified lines
427469 });
428470 const archivePath = path.join(
429471 tempDirectory,
430− releaseAsset.archive === "tgz" ? "cloudflared.tgz" : executableFileName(platform),
472+ releaseAsset.value.archive === "tgz" ? "cloudflared.tgz" : executableFileName(platform),
431473 );
432− const download = yield* downloadAsset(releaseAsset, report);
474+ const download = yield* downloadAsset(releaseAsset.value, report);
433475 yield* report("installing");
434476 yield* fileSystem
435477 .writeFile(archivePath, download)
436478 .pipe(wrapInstallFailure("write_failed", "Could not write the relay client download."));
437479
438480 const executablePath = path.join(tempDirectory, executableFileName(platform));
439− if (releaseAsset.archive === "tgz") {
481+ if (releaseAsset.value.archive === "tgz") {
440482 yield* runCommand("tar", ["-xzf", archivePath, "-C", tempDirectory]).pipe(
441483 wrapInstallFailure("write_failed", "Could not extract the relay client."),
442484 );