Open History to see the last commit
1import { newWebSocketRpcSession, type RpcStub } from "capnweb";
2import * as Config from "effect/Config";
3import * as Context from "effect/Context";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import * as HttpBody from "effect/http/HttpBody";
7import * as HttpClient from "effect/http/HttpClient";
8import * as Layer from "effect/Layer";
9import { AlchemyContext } from "../AlchemyContext.ts";
10import type { ProviderService } from "../Provider.ts";
11import type { ResourceLike } from "../Resource.ts";
12import { Stack } from "../Stack.ts";
13import { unwrapRpcHandlers } from "./RpcSerialization.ts";
14import type { RpcProxyApi, SessionProviderCounts } from "./RpcServer.ts";
15import { encodeSessionEnvironment, SESSION_ENV_PARAM } from "./RpcServerEnvironment.ts";
16import type { RpcSpawnPayload } from "./RpcSpawner.ts";
17
18export class RpcProviderProxyOperationFailed extends Data.TaggedError(
19 "RpcProviderProxyOperationFailed",
20)<{
21 readonly message: string;
22 readonly cause?: unknown;
23}> {}
24
25export class RpcProviderProxyUnexpectedResponse extends Data.TaggedError(
26 "RpcProviderProxyUnexpectedResponse",
27)<{
28 readonly message: string;
29 readonly cause?: unknown;
30}> {}
31
32/** A test file's leases on the sessions it actually uses. */
33export interface SessionOwner {
34 closed: boolean;
35 readonly sessions: Map<object, Effect.Effect<void>>;
36}
37
38export const makeSessionOwner = (): SessionOwner => ({
39 closed: false,
40 sessions: new Map(),
41});
42
43export const closeSessionOwner = (owner: SessionOwner) =>
44 Effect.suspend(() => {
45 if (owner.closed) return Effect.void;
46 owner.closed = true;
47 const releases = [...owner.sessions.values()];
48 owner.sessions.clear();
49 return Effect.forEach(releases, (release) => release, {
50 discard: true,
51 });
52 }).pipe(Effect.uninterruptible);
53
54export interface SessionCounts {
55 readonly sessions: number;
56 readonly testSessions: number;
57 readonly owners: number;
58 readonly connections: number;
59}
60
61export class RpcProviderProxy extends Context.Service<
62 RpcProviderProxy,
63 {
64 /**
65 * The provider for `providerName`, served by the dev sidecar. `providersUrl`
66 * is the URL of the module whose default export is the provider group's
67 * layer (see `Local/Sidecar.ts`); the sidecar imports it on first use.
68 */
69 readonly get: <R extends ResourceLike>(
70 providersUrl: string,
71 providerName: R["Type"],
72 owner?: SessionOwner,
73 ) => Effect.Effect<ProviderService<R>, never, AlchemyContext | Stack>;
74 /** Counts only; does not connect to or start the sidecar. */
75 readonly diagnostics: Effect.Effect<SessionCounts>;
76 /** Inspect an existing connection only; never spawn or reconnect. */
77 readonly serverDiagnostics: Effect.Effect<SessionProviderCounts | undefined, unknown>;
78 }
79>()("alchemy/Local/RpcProviderProxy") {}
80
81export const SPAWNER_URL_ENV_KEY = "ALCHEMY_RPC_SPAWNER_URL" as const;
82
83/**
84 * The one sidecar entry every RPC-backed provider is served from.
85 * Resolve through package exports. The active export conditions select `src/`
86 * under Bun or the dev loader and `lib/` in a published Node install.
87 */
88export const SIDECAR_ENTRY_URL = import.meta.resolve("alchemy/Local/Sidecar");
89
90interface ClientSession {
91 readonly rpc: RpcStub<RpcProxyApi>;
92 readonly socket: WebSocket;
93}
94
95interface Connection {
96 readonly pending: Promise<ClientSession>;
97}
98
99interface SessionEntry {
100 readonly owners: Set<SessionOwner>;
101 readonly testOwned: boolean;
102 readonly connections: Set<Connection>;
103 connection?: Connection;
104 closed: boolean;
105}
106
107const make = Effect.fn(function* (spawnerUrl: string) {
108 const client = yield* HttpClient.HttpClient;
109
110 const getSession = Effect.fn(
111 function* (sessionEnv: string) {
112 const payload: RpcSpawnPayload = { serverEntryUrl: SIDECAR_ENTRY_URL };
113 const response = yield* client.post(spawnerUrl, {
114 body: yield* HttpBody.json(payload),
115 });
116 const body = yield* response.text;
117 if (response.status !== 200) {
118 return yield* new RpcProviderProxyOperationFailed({
119 message: `RPC spawner POST ${spawnerUrl} returned ${response.status}: ${body.slice(0, 300)}`,
120 });
121 }
122 if (!URL.canParse(body)) {
123 return yield* new RpcProviderProxyUnexpectedResponse({
124 message: `RPC spawner POST ${spawnerUrl} did not return a websocket URL (got ${JSON.stringify(body.slice(0, 200))})`,
125 });
126 }
127 const websocketUrl = new URL(body);
128 if (websocketUrl.protocol !== "ws:" && websocketUrl.protocol !== "wss:") {
129 return yield* new RpcProviderProxyUnexpectedResponse({
130 message: `RPC spawner POST ${spawnerUrl} returned a non-websocket URL: ${websocketUrl.toString()}`,
131 });
132 }
133 websocketUrl.searchParams.set(SESSION_ENV_PARAM, sessionEnv);
134 const socket = new WebSocket(websocketUrl.toString());
135 return { rpc: newWebSocketRpcSession<RpcProxyApi>(socket), socket };
136 },
137 (effect) =>
138 Effect.catch(effect, (error) =>
139 Effect.die(
140 new Error("Failed to create a provider RPC session with the sidecar", {
141 cause: error,
142 }),
143 ),
144 ),
145 );
146
147 const sessions = new Map<string, SessionEntry>();
148 const dispose = (connection: Connection, release: boolean) =>
149 Effect.tryPromise(() => connection.pending).pipe(
150 Effect.flatMap((session) =>
151 (release
152 ? Effect.tryPromise(() => session.rpc.releaseSession()).pipe(Effect.ignore)
153 : Effect.void
154 ).pipe(
155 Effect.ensuring(
156 Effect.sync(() => {
157 session.rpc[Symbol.dispose]();
158 session.socket.close();
159 }),
160 ),
161 ),
162 ),
163 Effect.ignore,
164 );
165
166 const releaseEntry = (key: string, entry: SessionEntry) =>
167 Effect.suspend(() => {
168 entry.closed = true;
169 if (sessions.get(key) === entry) sessions.delete(key);
170 const connections = [...entry.connections];
171 entry.connections.clear();
172 entry.connection = undefined;
173 return Effect.forEach(connections, (connection) => dispose(connection, true), {
174 discard: true,
175 });
176 });
177
178 return RpcProviderProxy.of({
179 diagnostics: Effect.sync(() => {
180 let testSessions = 0;
181 let owners = 0;
182 let connections = 0;
183 for (const entry of sessions.values()) {
184 if (entry.testOwned) testSessions++;
185 owners += entry.owners.size;
186 connections += entry.connections.size;
187 }
188 return { sessions: sessions.size, testSessions, owners, connections };
189 }),
190 serverDiagnostics: Effect.suspend(() => {
191 const connection = [...sessions.values()].find(
192 (entry) => entry.connection !== undefined,
193 )?.connection;
194 return connection === undefined
195 ? Effect.undefined
196 : Effect.tryPromise(async () => (await connection.pending).rpc.getDiagnostics()).pipe(
197 Effect.map(
198 ({ sessions, testSessions, contexts, building, artifactBags, artifacts }) => ({
199 sessions,
200 testSessions,
201 contexts,
202 building,
203 artifactBags,
204 artifacts,
205 }),
206 ),
207 Effect.timeout("5 seconds"),
208 );
209 }),
210 get: Effect.fn(function* (providersUrl, providerName, owner) {
211 if (owner?.closed) return yield* Effect.die("Test sidecar handle is closed");
212 const alchemyContext = yield* AlchemyContext;
213 const stack = yield* Stack;
214 const environment = encodeSessionEnvironment({
215 alchemyContext,
216 stack: { name: stack.name, stage: stack.stage },
217 });
218 if (owner?.closed) return yield* Effect.die("Test sidecar handle is closed");
219 const key = `${owner === undefined ? "dev" : "test"}:${environment}`;
220 let entry = sessions.get(key);
221 if (entry === undefined) {
222 entry = {
223 owners: new Set(),
224 testOwned: owner !== undefined,
225 connections: new Set(),
226 closed: false,
227 };
228 sessions.set(key, entry);
229 }
230 const current = entry;
231 if (owner !== undefined && !current.owners.has(owner)) {
232 current.owners.add(owner);
233 owner.sessions.set(
234 current,
235 Effect.suspend(() => {
236 current.owners.delete(owner);
237 return current.owners.size === 0 ? releaseEntry(key, current) : Effect.void;
238 }),
239 );
240 }
241
242 const fetchProvider = Effect.gen(function* () {
243 if (current.closed || owner?.closed) {
244 return yield* Effect.die("Test sidecar handle is closed");
245 }
246 let connection = current.connection;
247 if (connection === undefined) {
248 const open = getSession(environment);
249 const context = yield* Effect.context();
250 connection = {
251 pending: Effect.runPromiseWith(context)(
252 current.testOwned ? open.pipe(Effect.timeout("30 seconds")) : open,
253 ),
254 };
255 current.connection = connection;
256 current.connections.add(connection);
257 const generation = connection;
258 connection.pending.then(
259 (session) => {
260 session.rpc.onRpcBroken(() => {
261 if (current.connection === generation) {
262 current.connection = undefined;
263 }
264 if (!current.testOwned) current.connections.delete(generation);
265 });
266 },
267 () => {
268 if (current.connection === generation) current.connection = undefined;
269 current.connections.delete(generation);
270 },
271 );
272 }
273 const generation = connection;
274 const session = yield* Effect.tryPromise(() => generation.pending);
275 if (current.closed || owner?.closed) {
276 return yield* Effect.die("Test sidecar handle is closed");
277 }
278 return yield* Effect.tryPromise(
279 () =>
280 session.rpc.getProvider(providerName, providersUrl, current.testOwned) as ReturnType<
281 RpcProxyApi["getProvider"]
282 >,
283 ).pipe(
284 Effect.tapError(() =>
285 Effect.sync(() => {
286 // A rejected lookup is not evidence that its shared transport died.
287 if (
288 current.connection === generation &&
289 (session.socket.readyState === WebSocket.CLOSING ||
290 session.socket.readyState === WebSocket.CLOSED)
291 ) {
292 current.connection = undefined;
293 }
294 }),
295 ),
296 );
297 });
298 // Retry once, without letting an old disconnect evict its successor.
299 const provider = yield* fetchProvider.pipe(
300 Effect.catch(() => fetchProvider),
301 Effect.orDie,
302 );
303 if (current.closed || owner?.closed) {
304 return yield* Effect.die("Test sidecar handle is closed");
305 }
306 return unwrapRpcHandlers(provider, ["tail"]) as ProviderService<any>;
307 }),
308 });
309});
310
311export const layer = (url: string) => Layer.effect(RpcProviderProxy, make(url));
312
313export const fromEnv = () =>
314 Layer.effect(RpcProviderProxy, Config.String(SPAWNER_URL_ENV_KEY).pipe(Effect.flatMap(make)));