1import { WS_METHODS } from "@t3tools/contracts";
2import { Atom } from "effect/reactivity";
3
4import { appAtomRegistry } from "./atomRegistry";
5
6const SLOW_RPC_ACK_THRESHOLD_MS = 15_000;
7/**
8 * Updates wait for package installs; VCS refreshes may wait for background work and network calls.
9 * Give these requests longer than the normal 15s warning threshold.
10 */
11const LONG_RUNNING_RPC_ACK_THRESHOLD_MS = 120_000;
12const MAX_TRACKED_RPC_ACK_REQUESTS = 256;
13let slowRpcAckThresholdMs = SLOW_RPC_ACK_THRESHOLD_MS;
14
15export interface SlowRpcAckRequest {
16 readonly requestId: string;
17 readonly startedAt: string;
18 readonly startedAtMs: number;
19 readonly tag: string;
20 readonly thresholdMs: number;
21}
22
23interface PendingRpcAckRequest {
24 readonly request: SlowRpcAckRequest;
25 readonly timeoutId: ReturnType<typeof setTimeout>;
26}
27
28const pendingRpcAckRequests = new Map<string, PendingRpcAckRequest>();
29// Background reports nobody waits on. A busy host delays them first, and warning
30// about a request the user never made does not say what is wrong. A sign-in wait
31// lasts as long as the user takes on the forge.
32const untrackedRpcAckMethods = new Set<string>([
33 WS_METHODS.serverGetUsageSummary,
34 WS_METHODS.serverReportClientActivity,
35 WS_METHODS.sourceControlSignInWait,
36]);
37const longRunningRpcAckMethods = new Set<string>([
38 WS_METHODS.serverUpdateProvider,
39 WS_METHODS.serverRefreshProviders,
40 WS_METHODS.serverUpdateServer,
41 WS_METHODS.vcsRefreshStatus,
42]);
43
44const slowRpcAckRequestsAtom = Atom.make<ReadonlyArray<SlowRpcAckRequest>>([]).pipe(
45 Atom.keepAlive,
46 Atom.withLabel("slow-rpc-ack-requests"),
47);
48
49function setSlowRpcAckRequests(requests: ReadonlyArray<SlowRpcAckRequest>) {
50 appAtomRegistry.set(slowRpcAckRequestsAtom, [...requests]);
51}
52
53function getSlowRpcAckRequestsValue(): ReadonlyArray<SlowRpcAckRequest> {
54 return appAtomRegistry.get(slowRpcAckRequestsAtom);
55}
56
57function shouldTrackRpcAck(method: string): boolean {
58 return (
59 !method.includes("subscribe") &&
60 !method.startsWith("pullRequests.") &&
61 !untrackedRpcAckMethods.has(method)
62 );
63}
64
65function rpcAckThresholdMs(method: string): number {
66 return longRunningRpcAckMethods.has(method)
67 ? Math.max(slowRpcAckThresholdMs, LONG_RUNNING_RPC_ACK_THRESHOLD_MS)
68 : slowRpcAckThresholdMs;
69}
70
71export function trackRpcRequestSent(requestId: string, method: string, tag = method): void {
72 if (!shouldTrackRpcAck(method)) {
73 return;
74 }
75
76 clearTrackedRpcRequest(requestId);
77 evictOldestPendingRpcRequestIfNeeded();
78
79 const startedAtMs = Date.now();
80 const thresholdMs = rpcAckThresholdMs(method);
81 const request: SlowRpcAckRequest = {
82 requestId,
83 startedAt: new Date(startedAtMs).toISOString(),
84 startedAtMs,
85 tag,
86 thresholdMs,
87 };
88 const timeoutId = setTimeout(() => {
89 pendingRpcAckRequests.delete(requestId);
90 appendSlowRpcAckRequest(request);
91 }, thresholdMs);
92
93 pendingRpcAckRequests.set(requestId, {
94 request,
95 timeoutId,
96 });
97}
98
99export function acknowledgeRpcRequest(requestId: string): void {
100 clearTrackedRpcRequest(requestId);
101 const slowRequests = getSlowRpcAckRequestsValue();
102 if (!slowRequests.some((request) => request.requestId === requestId)) {
103 return;
104 }
105
106 setSlowRpcAckRequests(slowRequests.filter((request) => request.requestId !== requestId));
107}
108
109function clearTrackedRpcRequest(requestId: string): void {
110 const pending = pendingRpcAckRequests.get(requestId);
111 if (!pending) {
112 return;
113 }
114
115 clearTimeout(pending.timeoutId);
116 pendingRpcAckRequests.delete(requestId);
117}
118
119function appendSlowRpcAckRequest(request: SlowRpcAckRequest): void {
120 const requests = [...getSlowRpcAckRequestsValue(), request];
121 if (requests.length <= MAX_TRACKED_RPC_ACK_REQUESTS) {
122 setSlowRpcAckRequests(requests);
123 return;
124 }
125
126 setSlowRpcAckRequests(requests.slice(-MAX_TRACKED_RPC_ACK_REQUESTS));
127}
128
129function evictOldestPendingRpcRequestIfNeeded(): void {
130 while (pendingRpcAckRequests.size >= MAX_TRACKED_RPC_ACK_REQUESTS) {
131 const oldestRequestId = pendingRpcAckRequests.keys().next().value;
132 if (oldestRequestId === undefined) {
133 return;
134 }
135
136 clearTrackedRpcRequest(oldestRequestId);
137 }
138}
139
140export function subscribeSlowRpcAckRequests(
141 listener: (requests: ReadonlyArray<SlowRpcAckRequest>) => void,
142): () => void {
143 return appAtomRegistry.subscribe(slowRpcAckRequestsAtom, listener);
144}