1import * as Clock from "effect/Clock";
2import * as Context from "effect/Context";
3import * as DateTime from "effect/DateTime";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7
8import * as RelayConfiguration from "../Config.ts";
9import {
10 MANAGED_ENDPOINT_ZONE_OWNER_STAGE,
11 managedEndpointTunnelNamePrefix,
12} from "../deploymentConfig.ts";
13import * as ManagedEndpointAllocations from "./ManagedEndpointAllocations.ts";
14import * as ManagedEndpointProvider from "./ManagedEndpointProvider.ts";
15
16export const MANAGED_ENDPOINT_GRACE_PERIOD_MINUTES = 5;
17// A tunnel that never connected is usually a link still being set up: a slow
18// cloudflared download or a user who walked away mid-pairing. Give it an hour.
19export const MANAGED_ENDPOINT_INACTIVE_GRACE_PERIOD_MINUTES = 60;
20export const MANAGED_ENDPOINT_SWEEP_PAGE_SIZE = 100;
21export const MANAGED_ENDPOINT_SWEEP_ATTEMPT_LIMIT = 100;
22export const MANAGED_ENDPOINT_SWEEP_LIST_REQUEST_LIMIT = 10;
23// Age buckets for legacy candidates, in days since the tunnel went down (or
24// was created, for one that never connected).
25const MANAGED_ENDPOINT_LEGACY_AGE_BUCKET_DAYS = [7, 30, 90] as const;
26// A host that never registered recovery cannot replace a deleted tunnel on
27// its own build. Only delete its tunnel after it has been gone this long. A
28// returning host that has updated recovers the tunnel at the same hostname;
29// one that has not sees the client's "update T3 Code" message instead.
30const MANAGED_ENDPOINT_LEGACY_GRACE_PERIOD_DAYS = 7;
31// Deletions run a few at a time: each one is a row-locked database
32// transaction plus two Cloudflare calls, so one at a time cannot finish a
33// full attempt budget inside the cron's timeout. Each holds a Hyperdrive
34// connection while it runs; keep this well under the 20-connection origin
35// limit that request handlers share.
36export const MANAGED_ENDPOINT_SWEEP_DELETE_CONCURRENCY = 4;
37// Stop starting deletions this long after the sweep starts, leaving room under
38// the cron's two-minute timeout to finish in-flight ones and record the
39// counters. Counted from sweep start so slow listing eats into it.
40const MANAGED_ENDPOINT_SWEEP_DELETE_BUDGET_MS = 90_000;
41
42export interface ManagedEndpointSweepResult {
43 readonly mode: RelayConfiguration.ManagedEndpointCleanupMode;
44 readonly legacyMode: RelayConfiguration.ManagedEndpointCleanupMode;
45 readonly listRequests: number;
46 readonly scanned: number;
47 readonly attempted: number;
48 readonly deleted: number;
49 readonly wouldDelete: number;
50 /** Legacy tunnels past the legacy grace period, and how many were deleted. */
51 readonly wouldDeleteLegacy: number;
52 readonly deletedLegacy: number;
53 readonly skippedLegacy: number;
54 /** Legacy candidates, by days since they went down: over 7, 30, and 90. */
55 readonly legacyOver7Days: number;
56 readonly legacyOver30Days: number;
57 readonly legacyOver90Days: number;
58 readonly skippedOrphan: number;
59 /** The allocation row records a different tunnel under this name. */
60 readonly skippedReplaced: number;
61 /** The allocation row has not recorded a tunnel yet. */
62 readonly skippedUnrecorded: number;
63 /** Cloudflare's count of matching tunnels, before page limits. */
64 readonly totalDown: number | null;
65 readonly totalInactive: number | null;
66 readonly failed: number;
67 readonly truncated: boolean;
68}
69
70export class ManagedEndpointReaper extends Context.Service<
71 ManagedEndpointReaper,
72 {
73 readonly sweep: Effect.Effect<
74 ManagedEndpointSweepResult,
75 | ManagedEndpointProvider.ManagedEndpointTunnelClientError
76 | ManagedEndpointAllocations.ManagedEndpointAllocationPersistenceError
77 >;
78 }
79>()("t3code-relay/environments/ManagedEndpointReaper") {}
80
81function isExpiredManagedTunnel(input: {
82 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel;
83 readonly status: "down" | "inactive";
84 readonly prefix: string;
85 readonly cutoff: DateTime.Utc;
86}): input is typeof input & {
87 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel & {
88 readonly id: string;
89 readonly name: string;
90 };
91} {
92 const { tunnel, status, prefix, cutoff } = input;
93 if (
94 typeof tunnel.id !== "string" ||
95 typeof tunnel.name !== "string" ||
96 tunnel.status !== status ||
97 !tunnel.name.startsWith(prefix) ||
98 !/^[a-f0-9]{16}$/u.test(tunnel.name.slice(prefix.length))
99 ) {
100 return false;
101 }
102 const inactiveAt = status === "down" ? tunnel.connsInactiveAt : tunnel.createdAt;
103 if (typeof inactiveAt !== "string") {
104 return false;
105 }
106 const timestamp = DateTime.make(inactiveAt);
107 return Option.isSome(timestamp) && timestamp.value.epochMilliseconds <= cutoff.epochMilliseconds;
108}
109
110/** Days (fractional) since the tunnel went down, or was created if it never connected. */
111function inactiveDaysAt(
112 tunnel: ManagedEndpointProvider.ManagedEndpointTunnel,
113 status: "down" | "inactive",
114 now: DateTime.Utc,
115): number | null {
116 const since = status === "down" ? tunnel.connsInactiveAt : tunnel.createdAt;
117 if (typeof since !== "string") return null;
118 const timestamp = DateTime.make(since);
119 if (Option.isNone(timestamp)) return null;
120 return (now.epochMilliseconds - timestamp.value.epochMilliseconds) / 86_400_000;
121}
122
123function isRateLimited(cause: unknown): boolean {
124 if (typeof cause !== "object" || cause === null) {
125 return false;
126 }
127 if ("_tag" in cause && cause._tag === "TooManyRequests") {
128 return true;
129 }
130 if ("status" in cause && cause.status === 429) {
131 return true;
132 }
133 return "cause" in cause && isRateLimited(cause.cause);
134}
135
136function rotatedPages(input: {
137 readonly totalCount: number | undefined;
138 readonly slot: number;
139 readonly limit: number;
140}): ReadonlyArray<number> {
141 if (input.limit <= 0) return [];
142 if (input.totalCount === undefined) {
143 return Array.from({ length: input.limit }, (_, index) => index + 2);
144 }
145 const laterPageCount = Math.max(
146 0,
147 Math.ceil(input.totalCount / MANAGED_ENDPOINT_SWEEP_PAGE_SIZE) - 1,
148 );
149 if (laterPageCount === 0) return [];
150 const count = Math.min(input.limit, laterPageCount);
151 const start = input.slot % laterPageCount;
152 return Array.from({ length: count }, (_, index) => 2 + ((start + index) % laterPageCount));
153}
154
155const emptyResult = (
156 mode: RelayConfiguration.ManagedEndpointCleanupMode,
157 legacyMode: RelayConfiguration.ManagedEndpointCleanupMode,
158): ManagedEndpointSweepResult => ({
159 mode,
160 legacyMode,
161 listRequests: 0,
162 scanned: 0,
163 attempted: 0,
164 deleted: 0,
165 wouldDelete: 0,
166 wouldDeleteLegacy: 0,
167 deletedLegacy: 0,
168 skippedLegacy: 0,
169 legacyOver7Days: 0,
170 legacyOver30Days: 0,
171 legacyOver90Days: 0,
172 skippedOrphan: 0,
173 skippedReplaced: 0,
174 skippedUnrecorded: 0,
175 totalDown: null,
176 totalInactive: null,
177 failed: 0,
178 truncated: false,
179});
180
181export const make = Effect.gen(function* () {
182 const config = yield* RelayConfiguration.RelayConfiguration;
183 const tunnels = yield* ManagedEndpointProvider.ManagedEndpointTunnelClient;
184 const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
185 const provider = yield* ManagedEndpointProvider.ManagedEndpointProvider;
186
187 const sweep = Effect.gen(function* () {
188 const mode = config.managedEndpointCleanupMode ?? "off";
189 const legacyMode = config.legacyManagedEndpointCleanupMode ?? "off";
190 const namespace = config.managedEndpointNamespace;
191 if ((mode === "off" && legacyMode === "off") || !namespace) {
192 return emptyResult(mode, legacyMode);
193 }
194 const sweepStartedAtMillis = yield* Clock.currentTimeMillis;
195 // The override exists for the disposable canary stage; prod always
196 // waits the full grace period.
197 const legacyGraceMinutes =
198 namespace !== MANAGED_ENDPOINT_ZONE_OWNER_STAGE &&
199 config.legacyTunnelGraceMinutes !== undefined
200 ? config.legacyTunnelGraceMinutes
201 : MANAGED_ENDPOINT_LEGACY_GRACE_PERIOD_DAYS * 24 * 60;
202 const legacyCutoff = DateTime.subtract(yield* DateTime.now, { minutes: legacyGraceMinutes });
203
204 const now = yield* DateTime.now;
205 const cutoffFor = (status: "down" | "inactive") =>
206 DateTime.subtract(now, {
207 minutes:
208 status === "down"
209 ? MANAGED_ENDPOINT_GRACE_PERIOD_MINUTES
210 : MANAGED_ENDPOINT_INACTIVE_GRACE_PERIOD_MINUTES,
211 });
212 const prefix = managedEndpointTunnelNamePrefix(namespace);
213 const slot = Math.floor(
214 now.epochMilliseconds / (MANAGED_ENDPOINT_GRACE_PERIOD_MINUTES * 60 * 1_000),
215 );
216 let listRequests = 0;
217 let truncated = false;
218 const totals: Record<"down" | "inactive", number | null> = { down: null, inactive: null };
219 const expired: Array<{
220 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel & {
221 readonly id: string;
222 readonly name: string;
223 };
224 readonly status: "down" | "inactive";
225 readonly cutoff: DateTime.Utc;
226 }> = [];
227
228 for (const status of ["down", "inactive"] as const) {
229 const cutoff = cutoffFor(status);
230 const cutoffIso = DateTime.formatIso(cutoff);
231 const listPage = (page: number) => {
232 listRequests += 1;
233 return tunnels.list({
234 isDeleted: false,
235 includePrefix: prefix,
236 status,
237 existedAt: cutoffIso,
238 ...(status === "down" ? { wasInactiveAt: cutoffIso } : {}),
239 page,
240 perPage: MANAGED_ENDPOINT_SWEEP_PAGE_SIZE,
241 });
242 };
243 const first = yield* listPage(1);
244 const totalCount =
245 typeof first.resultInfo?.totalCount === "number" ? first.resultInfo.totalCount : undefined;
246 totals[status] = totalCount ?? null;
247 const pages = rotatedPages({
248 totalCount,
249 slot,
250 limit: Math.floor(MANAGED_ENDPOINT_SWEEP_LIST_REQUEST_LIMIT / 2) - 1,
251 });
252 const responses = [first, ...(yield* Effect.forEach(pages, listPage, { concurrency: 1 }))];
253 if (
254 totalCount !== undefined &&
255 Math.ceil(totalCount / MANAGED_ENDPOINT_SWEEP_PAGE_SIZE) > responses.length
256 ) {
257 truncated = true;
258 } else if (
259 totalCount === undefined &&
260 responses.at(-1)?.result.length === MANAGED_ENDPOINT_SWEEP_PAGE_SIZE
261 ) {
262 truncated = true;
263 }
264 for (const response of responses) {
265 expired.push(
266 ...response.result
267 .map((tunnel) => ({ tunnel, status, prefix, cutoff }))
268 .filter(isExpiredManagedTunnel)
269 .map(({ tunnel }) => ({ tunnel, status, cutoff })),
270 );
271 }
272 }
273
274 const collected = [...new Map(expired.map((entry) => [entry.tunnel.id, entry])).values()];
275 // Start each sweep one attempt budget further along so a run of
276 // candidates whose deletes keep failing cannot hold the budget forever
277 // and starve everything listed after them.
278 const offset =
279 collected.length === 0 ? 0 : (slot * MANAGED_ENDPOINT_SWEEP_ATTEMPT_LIMIT) % collected.length;
280 const uniqueExpired = [...collected.slice(offset), ...collected.slice(0, offset)];
281 const recorded = yield* allocations.listByTunnelNames(
282 uniqueExpired.map(({ tunnel }) => tunnel.name),
283 );
284 const recordedByTunnelName = new Map(
285 recorded.map((allocation) => [allocation.tunnelName, allocation]),
286 );
287 let attempted = 0;
288 let deleted = 0;
289 let wouldDelete = 0;
290 const candidates: Array<{
291 readonly owner: ManagedEndpointAllocations.ManagedEndpointTunnelAllocation;
292 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel & {
293 readonly id: string;
294 readonly name: string;
295 };
296 readonly status: "down" | "inactive";
297 readonly legacy: boolean;
298 readonly inactiveBefore: string;
299 }> = [];
300 let wouldDeleteLegacy = 0;
301 let deletedLegacy = 0;
302 let skippedLegacy = 0;
303 const legacyOverDays = new Map<number, number>(
304 MANAGED_ENDPOINT_LEGACY_AGE_BUCKET_DAYS.map((days) => [days, 0]),
305 );
306 let skippedOrphan = 0;
307 let skippedReplaced = 0;
308 let skippedUnrecorded = 0;
309 let failed = 0;
310
311 for (const { tunnel, status, cutoff } of uniqueExpired) {
312 const allocation = recordedByTunnelName.get(tunnel.name);
313 if (
314 allocation !== undefined &&
315 allocation.tunnelId !== null &&
316 allocation.tunnelId !== tunnel.id
317 ) {
318 skippedReplaced += 1;
319 continue;
320 }
321 const owner = allocation?.tunnelId === tunnel.id ? allocation : undefined;
322 if (allocation !== undefined && owner === undefined) {
323 skippedUnrecorded += 1;
324 continue;
325 }
326 // A tunnel with no allocation row cannot be claimed, so a relink that
327 // adopts it by name races any delete here. Count it and leave it for a
328 // manual sweep instead.
329 if (owner === undefined) {
330 skippedOrphan += 1;
331 continue;
332 }
333 const legacy = !owner.recoveryEnabled;
334 if (legacy) {
335 skippedLegacy += 1;
336 const inactiveDays = inactiveDaysAt(tunnel, status, now);
337 for (const days of MANAGED_ENDPOINT_LEGACY_AGE_BUCKET_DAYS) {
338 if (inactiveDays !== null && inactiveDays > days) {
339 legacyOverDays.set(days, (legacyOverDays.get(days) ?? 0) + 1);
340 }
341 }
342 if (
343 legacyMode === "off" ||
344 !isExpiredManagedTunnel({ tunnel, status, prefix, cutoff: legacyCutoff })
345 ) {
346 continue;
347 }
348 wouldDeleteLegacy += 1;
349 if (legacyMode === "dry-run") continue;
350 } else {
351 if (mode === "off") continue;
352 wouldDelete += 1;
353 if (mode === "dry-run") continue;
354 }
355 if (candidates.length >= MANAGED_ENDPOINT_SWEEP_ATTEMPT_LIMIT) {
356 truncated = true;
357 break;
358 }
359 candidates.push({
360 owner,
361 tunnel,
362 status,
363 legacy,
364 // The release re-reads the tunnel and deletes only if it is still in
365 // this status and inactive since before this cutoff.
366 inactiveBefore: DateTime.formatIso(legacy ? legacyCutoff : cutoff),
367 });
368 }
369
370 const deleteDeadline = sweepStartedAtMillis + MANAGED_ENDPOINT_SWEEP_DELETE_BUDGET_MS;
371 let stopDeleting = false;
372 yield* Effect.forEach(
373 candidates,
374 (candidate) =>
375 Effect.gen(function* () {
376 if (stopDeleting || (yield* Clock.currentTimeMillis) >= deleteDeadline) {
377 stopDeleting = true;
378 truncated = true;
379 return;
380 }
381 attempted += 1;
382 const result = yield* provider
383 .release({
384 userId: candidate.owner.userId,
385 environmentId: candidate.owner.environmentId,
386 expectedTunnelId: candidate.tunnel.id,
387 expectedInactiveBefore: candidate.inactiveBefore,
388 expectedStatus: candidate.status,
389 // Only a legacy host needs an update to recover; tell its user so.
390 ...(candidate.legacy ? { markReleased: true } : {}),
391 })
392 .pipe(Effect.result);
393 if (result._tag === "Failure") {
394 failed += 1;
395 yield* Effect.logWarning("Failed to delete an inactive managed tunnel", {
396 tunnelId: candidate.tunnel.id,
397 tunnelName: candidate.tunnel.name,
398 legacy: candidate.legacy,
399 cause: result.failure,
400 });
401 if (isRateLimited(result.failure)) {
402 stopDeleting = true;
403 truncated = true;
404 }
405 } else if (result.success) {
406 deleted += 1;
407 if (candidate.legacy) deletedLegacy += 1;
408 yield* Effect.logInfo("Deleted an inactive managed tunnel", {
409 tunnelId: candidate.tunnel.id,
410 tunnelName: candidate.tunnel.name,
411 status: candidate.status,
412 legacy: candidate.legacy,
413 });
414 }
415 }),
416 { concurrency: MANAGED_ENDPOINT_SWEEP_DELETE_CONCURRENCY, discard: true },
417 );
418
419 return {
420 mode,
421 legacyMode,
422 listRequests,
423 scanned: uniqueExpired.length,
424 attempted,
425 deleted,
426 wouldDelete,
427 wouldDeleteLegacy,
428 deletedLegacy,
429 skippedLegacy,
430 legacyOver7Days: legacyOverDays.get(7) ?? 0,
431 legacyOver30Days: legacyOverDays.get(30) ?? 0,
432 legacyOver90Days: legacyOverDays.get(90) ?? 0,
433 skippedOrphan,
434 skippedReplaced,
435 skippedUnrecorded,
436 totalDown: totals.down,
437 totalInactive: totals.inactive,
438 failed,
439 truncated,
440 };
441 }).pipe(
442 // Dry-run rollout reads these counters from the exported span.
443 Effect.tap((result) =>
444 Effect.annotateCurrentSpan(
445 Object.fromEntries(
446 Object.entries(result).map(([key, value]) => [
447 `relay.managed_endpoint_reaper.${key}`,
448 value,
449 ]),
450 ),
451 ),
452 ),
453 Effect.withSpan("relay.managed_endpoint_reaper.sweep"),
454 );
455
456 return ManagedEndpointReaper.of({ sweep });
457});
458
459export const layer = Layer.effect(ManagedEndpointReaper, make);