| ... | ... | @@ -1,7 +1,10 @@ |
| 1 | | // @effect-diagnostics nodeBuiltinImport:off -- Local socket ownership checks need lstat uid and an atomic stale-socket unlink at the Node adapter boundary. |
| 1 | // @effect-diagnostics nodeBuiltinImport:off -- Local socket ownership checks need lstat, an atomic rename, and a directory watch at the Node adapter boundary. |
| 2 | import * as NodeCrypto from "node:crypto"; |
| 3 | import * as NodeFS from "node:fs"; |
| 2 | 4 | import * as NodeFSP from "node:fs/promises"; |
| 3 | 5 | import * as NodeNet from "node:net"; |
| 4 | 6 | import * as NodeOS from "node:os"; |
| 7 | import * as NodePath from "node:path"; |
| 5 | 8 | |
| 6 | 9 | import { |
| 7 | 10 | DESKTOP_APP_ACTIVATION_PROTOCOL_VERSION, |
| ... | ... | @@ -44,6 +47,8 @@ export class DesktopAppActivationStartError extends Schema.TaggedError<DesktopAp |
| 44 | 47 | } |
| 45 | 48 | |
| 46 | 49 | interface RunningControlServer { |
| 50 | /** Binds the address again if its socket file is gone. Directory changes run this too. */ |
| 51 | readonly reclaim: () => Promise<void>; |
| 47 | 52 | readonly close: () => Promise<void>; |
| 48 | 53 | } |
| 49 | 54 | |
| ... | ... | @@ -70,12 +75,12 @@ function requestIdFromUnknown(value: unknown): string { |
| 70 | 75 | return "invalid-request"; |
| 71 | 76 | } |
| 72 | 77 | |
| 73 | | async function prepareUnixSocket(input: { |
| 74 | | readonly address: string; |
| 78 | /** Makes sure the socket directory is safe to use. Returns true when it had to create it. */ |
| 79 | async function prepareUnixDirectory(input: { |
| 75 | 80 | readonly directory: string; |
| 76 | 81 | readonly userId: number | undefined; |
| 77 | | }): Promise<void> { |
| 78 | | await NodeFSP.mkdir(input.directory, { recursive: true, mode: 0o700 }); |
| 82 | }): Promise<boolean> { |
| 83 | const created = await NodeFSP.mkdir(input.directory, { recursive: true, mode: 0o700 }); |
| 79 | 84 | const stat = await NodeFSP.lstat(input.directory); |
| 80 | 85 | if (!stat.isDirectory() || stat.isSymbolicLink()) { |
| 81 | 86 | throw new Error(`${input.directory} is not a directory.`); |
| ... | ... | @@ -84,28 +89,43 @@ async function prepareUnixSocket(input: { |
| 84 | 89 | throw new Error(`${input.directory} is owned by another user.`); |
| 85 | 90 | } |
| 86 | 91 | await NodeFSP.chmod(input.directory, 0o700); |
| 87 | | await NodeFSP.unlink(input.address).catch((error: NodeJS.ErrnoException) => { |
| 88 | | if (error.code !== "ENOENT") throw error; |
| 89 | | }); |
| 92 | return created !== undefined; |
| 93 | } |
| 94 | |
| 95 | async function inodeAt(path: string): Promise<number | null> { |
| 96 | return NodeFSP.lstat(path).then( |
| 97 | (stat) => stat.ino, |
| 98 | (error: NodeJS.ErrnoException) => { |
| 99 | if (error.code === "ENOENT") return null; |
| 100 | throw error; |
| 101 | }, |
| 102 | ); |
| 103 | } |
| 104 | |
| 105 | function closeServer(server: NodeNet.Server): Promise<void> { |
| 106 | return new Promise((resolve) => server.close(() => resolve())); |
| 90 | 107 | } |
| 91 | 108 | |
| 109 | /** |
| 110 | * Serves `t3 app` requests on the local control address until `close`. |
| 111 | * |
| 112 | * Two desktop apps can share one state dir, for example nightly and a preview |
| 113 | * build. They share one socket path, so on Unix: |
| 114 | * - The newest app takes the path over. |
| 115 | * - `close` removes the socket file only while it is still this app's socket. |
| 116 | * - An app binds the path again when it is gone, for example after the app that |
| 117 | * took it over quits. |
| 118 | */ |
| 92 | 119 | export async function startDesktopAppControlServer(input: { |
| 93 | 120 | readonly address: string; |
| 94 | 121 | readonly directory: string | null; |
| 95 | 122 | readonly userId: number | undefined; |
| 96 | 123 | readonly handle: (request: DesktopAppActivationRequest) => Promise<DesktopAppActivationResponse>; |
| 97 | 124 | readonly cancel: (requestId: string) => void; |
| 125 | readonly onReclaimError: (error: unknown) => void; |
| 98 | 126 | }): Promise<RunningControlServer> { |
| 99 | | if (input.directory !== null) { |
| 100 | | await prepareUnixSocket({ |
| 101 | | address: input.address, |
| 102 | | directory: input.directory, |
| 103 | | userId: input.userId, |
| 104 | | }); |
| 105 | | } |
| 106 | | |
| 107 | 127 | const sockets = new Set<NodeNet.Socket>(); |
| 108 | | const server = NodeNet.createServer((socket) => { |
| 128 | const handleConnection = (socket: NodeNet.Socket) => { |
| 109 | 129 | sockets.add(socket); |
| 110 | 130 | socket.setEncoding("utf8"); |
| 111 | 131 | let buffer = ""; |
| ... | ... | @@ -160,40 +180,116 @@ export async function startDesktopAppControlServer(input: { |
| 160 | 180 | sockets.delete(socket); |
| 161 | 181 | if (!responseSent && activeRequestId !== null) input.cancel(activeRequestId); |
| 162 | 182 | }); |
| 163 | | }); |
| 183 | }; |
| 164 | 184 | |
| 165 | | await new Promise<void>((resolve, reject) => { |
| 166 | | const onError = (error: Error) => { |
| 167 | | server.removeListener("listening", onListening); |
| 168 | | reject(error); |
| 169 | | }; |
| 170 | | const onListening = () => { |
| 171 | | server.removeListener("error", onError); |
| 172 | | resolve(); |
| 173 | | }; |
| 174 | | server.once("error", onError); |
| 175 | | server.once("listening", onListening); |
| 176 | | server.listen(input.address); |
| 177 | | }); |
| 185 | const listen = (address: string) => |
| 186 | new Promise<NodeNet.Server>((resolve, reject) => { |
| 187 | const server = NodeNet.createServer(handleConnection); |
| 188 | const onError = (error: Error) => { |
| 189 | server.removeListener("listening", onListening); |
| 190 | reject(error); |
| 191 | }; |
| 192 | const onListening = () => { |
| 193 | server.removeListener("error", onError); |
| 194 | resolve(server); |
| 195 | }; |
| 196 | server.once("error", onError); |
| 197 | server.once("listening", onListening); |
| 198 | server.listen(address); |
| 199 | }); |
| 178 | 200 | |
| 179 | | try { |
| 180 | | if (input.directory !== null) { |
| 181 | | await NodeFSP.chmod(input.address, 0o600); |
| 201 | // Closing a Unix socket server unlinks the path it was bound to, even when |
| 202 | // another app's socket lives there now. Bind a staging path and move it onto |
| 203 | // the address instead, so a later close only unlinks the staging path, which |
| 204 | // is already gone. `rename` takes the address over in one step. `link` claims |
| 205 | // it only while it is free, and fails with EEXIST otherwise. |
| 206 | const bindUnix = async (directory: string, mode: "take-over" | "claim-free") => { |
| 207 | const staging = NodePath.join(directory, `${NodeCrypto.randomBytes(6).toString("hex")}.tmp`); |
| 208 | const server = await listen(staging); |
| 209 | try { |
| 210 | await NodeFSP.chmod(staging, 0o600); |
| 211 | const inode = await inodeAt(staging); |
| 212 | if (mode === "take-over") { |
| 213 | await NodeFSP.rename(staging, input.address); |
| 214 | } else { |
| 215 | await NodeFSP.link(staging, input.address); |
| 216 | await NodeFSP.unlink(staging); |
| 217 | } |
| 218 | return { server, inode }; |
| 219 | } catch (error) { |
| 220 | await closeServer(server); |
| 221 | throw error; |
| 182 | 222 | } |
| 183 | | } catch (error) { |
| 184 | | await new Promise<void>((resolve) => server.close(() => resolve())); |
| 185 | | throw error; |
| 223 | }; |
| 224 | |
| 225 | let server: NodeNet.Server; |
| 226 | let inode: number | null = null; |
| 227 | if (input.directory === null) { |
| 228 | // Named pipes close with the app that owns them, so no other app can remove this one. |
| 229 | server = await listen(input.address); |
| 230 | } else { |
| 231 | await prepareUnixDirectory({ directory: input.directory, userId: input.userId }); |
| 232 | ({ server, inode } = await bindUnix(input.directory, "take-over")); |
| 186 | 233 | } |
| 187 | 234 | |
| 188 | 235 | let closed = false; |
| 236 | const reclaimOnce = async () => { |
| 237 | // Never replace a socket that exists, so two apps cannot trade the path back and forth. |
| 238 | if (closed || input.directory === null || (await inodeAt(input.address)) !== null) return; |
| 239 | if (await prepareUnixDirectory({ directory: input.directory, userId: input.userId })) { |
| 240 | // A watch follows the directory's inode, so a recreated directory needs a new one. |
| 241 | watchDirectory(input.directory); |
| 242 | } |
| 243 | const next = await bindUnix(input.directory, "claim-free").catch( |
| 244 | (error: NodeJS.ErrnoException) => { |
| 245 | // Another app bound the address first. |
| 246 | if (error.code === "EEXIST") return null; |
| 247 | throw error; |
| 248 | }, |
| 249 | ); |
| 250 | if (next === null) return; |
| 251 | const previous = server; |
| 252 | ({ server, inode } = next); |
| 253 | previous.close(); |
| 254 | }; |
| 255 | let pendingReclaim = Promise.resolve(); |
| 256 | const reclaim = () => { |
| 257 | const run = pendingReclaim.then(reclaimOnce); |
| 258 | pendingReclaim = run.catch(() => undefined); |
| 259 | return run; |
| 260 | }; |
| 261 | let watcher: NodeFS.FSWatcher | null = null; |
| 262 | const watchDirectory = (directory: string) => { |
| 263 | watcher?.close(); |
| 264 | watcher = null; |
| 265 | try { |
| 266 | watcher = NodeFS.watch(directory, { persistent: false }, () => { |
| 267 | reclaim().catch(input.onReclaimError); |
| 268 | }); |
| 269 | watcher.on("error", input.onReclaimError); |
| 270 | } catch (error) { |
| 271 | // The socket still works without a watcher. It only cannot recover after removal. |
| 272 | input.onReclaimError(error); |
| 273 | } |
| 274 | }; |
| 275 | if (input.directory !== null) { |
| 276 | watchDirectory(input.directory); |
| 277 | // Catch a removal that happened before the watcher started. |
| 278 | reclaim().catch(input.onReclaimError); |
| 279 | } |
| 280 | |
| 189 | 281 | return { |
| 282 | reclaim, |
| 190 | 283 | close: async () => { |
| 191 | 284 | if (closed) return; |
| 192 | 285 | closed = true; |
| 286 | // A running reclaim can replace the watcher, so close the watcher after it. |
| 287 | await pendingReclaim; |
| 288 | watcher?.close(); |
| 193 | 289 | for (const socket of sockets) socket.destroy(); |
| 194 | | await new Promise<void>((resolve) => server.close(() => resolve())); |
| 290 | await closeServer(server); |
| 195 | 291 | server.removeAllListeners(); |
| 196 | | if (input.directory !== null) { |
| 292 | if (inode !== null && (await inodeAt(input.address)) === inode) { |
| 197 | 293 | await NodeFSP.unlink(input.address).catch((error: NodeJS.ErrnoException) => { |
| 198 | 294 | if (error.code !== "ENOENT") throw error; |
| 199 | 295 | }); |
| ... | ... | @@ -258,6 +354,10 @@ export const make = Effect.gen(function* () { |
| 258 | 354 | userId, |
| 259 | 355 | handle: (request) => broker.request(request), |
| 260 | 356 | cancel: (requestId) => broker.cancel(requestId), |
| 357 | onReclaimError: (cause) => |
| 358 | void runPromise( |
| 359 | logWarning("failed to restore the desktop app control socket", { cause }), |
| 360 | ), |
| 261 | 361 | }), |
| 262 | 362 | catch: (cause) => new DesktopAppActivationStartError({ address: address.address, cause }), |
| 263 | 363 | }), |