1import { useAtomValue } from "@effect/atom-react";
2import { WS_METHODS } from "@t3tools/contracts";
3import { Atom } from "effect/reactivity";
4
5import { appAtomRegistry } from "./atomRegistry";
6
7export const SLOW_RPC_ACK_THRESHOLD_MS = 15_000;
8/**
9 * Some requests are slow by design — they shell out to a package manager on the
10 * server and only respond once the install finishes. Warning about those after
11 * 15s is noise, so they get a much longer leash.
12 */
13export const LONG_RUNNING_RPC_ACK_THRESHOLD_MS = 120_000;
14export const MAX_TRACKED_RPC_ACK_REQUESTS = 256;
15let slowRpcAckThresholdMs = SLOW_RPC_ACK_THRESHOLD_MS;
16
17export interface SlowRpcAckRequest {
18 readonly requestId: string;
19 readonly startedAt: string;
20 readonly startedAtMs: number;
21 readonly tag: string;
22 readonly thresholdMs: number;
23}
24
25interface PendingRpcAckRequest {
26 readonly request: SlowRpcAckRequest;
27 readonly timeoutId: ReturnType<typeof setTimeout>;
28}
29
30const pendingRpcAckRequests = new Map<string, PendingRpcAckRequest>();
31const untrackedRpcAckMethods = new Set<string>([WS_METHODS.serverGetUsageSummary]);
32const longRunningRpcAckMethods = new Set<string>([
33 WS_METHODS.serverUpdateProvider,
34 WS_METHODS.serverRefreshProviders,
35 WS_METHODS.serverUpdateServer,
36]);
37
38const slowRpcAckRequestsAtom = Atom.make<ReadonlyArray<SlowRpcAckRequest>>([]).pipe(
39 Atom.keepAlive,
40 Atom.withLabel("slow-rpc-ack-requests"),
41);
42
43function setSlowRpcAckRequests(requests: ReadonlyArray<SlowRpcAckRequest>) {
44 appAtomRegistry.set(slowRpcAckRequestsAtom, [...requests]);
45}
46
47function getSlowRpcAckRequestsValue(): ReadonlyArray<SlowRpcAckRequest> {
48 return appAtomRegistry.get(slowRpcAckRequestsAtom);
49}
50
51function shouldTrackRpcAck(method: string): boolean {
52 return (
53 !method.includes("subscribe") &&
54 !method.startsWith("pullRequests.") &&
55 !untrackedRpcAckMethods.has(method)
56 );
57}
58
59function rpcAckThresholdMs(method: string): number {
60 return longRunningRpcAckMethods.has(method)
61 ? Math.max(slowRpcAckThresholdMs, LONG_RUNNING_RPC_ACK_THRESHOLD_MS)
62 : slowRpcAckThresholdMs;
63}
64
65export function getSlowRpcAckRequests(): ReadonlyArray<SlowRpcAckRequest> {
66 return getSlowRpcAckRequestsValue();
67}
68
69/**
70 * Starts the slow-request timer for one in-flight unary RPC. `method` is the
71 * bare WS method (used to decide whether and how long to wait); `tag` is the
72 * human-readable label shown in the toast, which defaults to the method.
73 */
74export function trackRpcRequestSent(requestId: string, method: string, tag = method): void {
75 if (!shouldTrackRpcAck(method)) {
76 return;
77 }
78
79 clearTrackedRpcRequest(requestId);
80 evictOldestPendingRpcRequestIfNeeded();
81
82 const startedAtMs = Date.now();
83 const thresholdMs = rpcAckThresholdMs(method);
84 const request: SlowRpcAckRequest = {
85 requestId,
86 startedAt: new Date(startedAtMs).toISOString(),
87 startedAtMs,
88 tag,
89 thresholdMs,
90 };
91 const timeoutId = setTimeout(() => {
92 pendingRpcAckRequests.delete(requestId);
93 appendSlowRpcAckRequest(request);
94 }, thresholdMs);
95
96 pendingRpcAckRequests.set(requestId, {
97 request,
98 timeoutId,
99 });
100}
101
102export function acknowledgeRpcRequest(requestId: string): void {
103 clearTrackedRpcRequest(requestId);
104 const slowRequests = getSlowRpcAckRequestsValue();
105 if (!slowRequests.some((request) => request.requestId === requestId)) {
106 return;
107 }
108
109 setSlowRpcAckRequests(slowRequests.filter((request) => request.requestId !== requestId));
110}
111
112function clearAllTrackedRpcRequests(): void {
113 for (const pending of pendingRpcAckRequests.values()) {
114 clearTimeout(pending.timeoutId);
115 }
116 pendingRpcAckRequests.clear();
117 setSlowRpcAckRequests([]);
118}
119
120function clearTrackedRpcRequest(requestId: string): void {
121 const pending = pendingRpcAckRequests.get(requestId);
122 if (!pending) {
123 return;
124 }
125
126 clearTimeout(pending.timeoutId);
127 pendingRpcAckRequests.delete(requestId);
128}
129
130function appendSlowRpcAckRequest(request: SlowRpcAckRequest): void {
131 const requests = [...getSlowRpcAckRequestsValue(), request];
132 if (requests.length <= MAX_TRACKED_RPC_ACK_REQUESTS) {
133 setSlowRpcAckRequests(requests);
134 return;
135 }
136
137 setSlowRpcAckRequests(requests.slice(-MAX_TRACKED_RPC_ACK_REQUESTS));
138}
139
140function evictOldestPendingRpcRequestIfNeeded(): void {
141 while (pendingRpcAckRequests.size >= MAX_TRACKED_RPC_ACK_REQUESTS) {
142 const oldestRequestId = pendingRpcAckRequests.keys().next().value;
143 if (oldestRequestId === undefined) {
144 return;
145 }
146
147 clearTrackedRpcRequest(oldestRequestId);
148 }
149}
150
151export function resetRequestLatencyStateForTests(): void {
152 slowRpcAckThresholdMs = SLOW_RPC_ACK_THRESHOLD_MS;
153 clearAllTrackedRpcRequests();
154}
155
156export function useSlowRpcAckRequests(): ReadonlyArray<SlowRpcAckRequest> {
157 return useAtomValue(slowRpcAckRequestsAtom);
158}