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
16const MANAGED_ENDPOINT_GRACE_PERIOD_MINUTES = 5;
17// Never-connected tunnels may still be pairing; allow an hour.
18const MANAGED_ENDPOINT_INACTIVE_GRACE_PERIOD_MINUTES = 60;
19const MANAGED_ENDPOINT_SWEEP_PAGE_SIZE = 100;
20export const MANAGED_ENDPOINT_SWEEP_ATTEMPT_LIMIT = 250;
21const MANAGED_ENDPOINT_SWEEP_LIST_REQUEST_LIMIT = 10;
22const MANAGED_ENDPOINT_LEGACY_AGE_BUCKET_DAYS = [7, 30, 90] as const;
23// Hosts without recovery cannot replace deleted tunnels; allow a longer absence.
24const MANAGED_ENDPOINT_LEGACY_GRACE_PERIOD_DAYS = 7;
25// Each delete holds a Hyperdrive connection for ~2s; stay well below the 40 request handlers share.
26const MANAGED_ENDPOINT_SWEEP_DELETE_CONCURRENCY = 8;
27// Leave time under the cron's two-minute limit to finish deletions and record counters.
28const MANAGED_ENDPOINT_SWEEP_DELETE_BUDGET_MS = 90_000;
29
30export interface ManagedEndpointSweepResult {
31 readonly mode: RelayConfiguration.ManagedEndpointCleanupMode;
32 readonly legacyMode: RelayConfiguration.ManagedEndpointCleanupMode;
33 readonly listRequests: number;
34 readonly scanned: number;
35 readonly attempted: number;
36 readonly deleted: number;
37 readonly wouldDelete: number;
38 readonly wouldDeleteLegacy: number;
39 readonly deletedLegacy: number;
40 readonly skippedLegacy: number;
41 readonly legacyOver7Days: number;
42 readonly legacyOver30Days: number;
43 readonly legacyOver90Days: number;
44 readonly skippedOrphan: number;
45 readonly skippedReplaced: number;
46 readonly skippedUnrecorded: number;
47 readonly totalDown: number | null;
48 readonly totalInactive: number | null;
49 readonly failed: number;
50 readonly truncated: boolean;
51}
52
53export class ManagedEndpointReaper extends Context.Service<
54 ManagedEndpointReaper,
55 {
56 readonly sweep: Effect.Effect<
57 ManagedEndpointSweepResult,
58 | ManagedEndpointProvider.ManagedEndpointTunnelClientError
59 | ManagedEndpointAllocations.ManagedEndpointAllocationPersistenceError
60 >;
61 }
62>()("t3code-relay/environments/ManagedEndpointReaper") {}
63
64function isExpiredManagedTunnel(input: {
65 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel;
66 readonly status: "down" | "inactive";
67 readonly prefix: string;
68 readonly cutoff: DateTime.Utc;
69}): input is typeof input & {
70 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel & {
71 readonly id: string;
72 readonly name: string;
73 };
74} {
75 const { tunnel, status, prefix, cutoff } = input;
76 if (
77 typeof tunnel.id !== "string" ||
78 typeof tunnel.name !== "string" ||
79 tunnel.status !== status ||
80 !tunnel.name.startsWith(prefix) ||
81 !/^[a-f0-9]{16}$/u.test(tunnel.name.slice(prefix.length))
82 ) {
83 return false;
84 }
85 const inactiveAt = status === "down" ? tunnel.connsInactiveAt : tunnel.createdAt;
86 if (typeof inactiveAt !== "string") {
87 return false;
88 }
89 const timestamp = DateTime.make(inactiveAt);
90 return Option.isSome(timestamp) && timestamp.value.epochMilliseconds <= cutoff.epochMilliseconds;
91}
92
93function inactiveDaysAt(
94 tunnel: ManagedEndpointProvider.ManagedEndpointTunnel,
95 status: "down" | "inactive",
96 now: DateTime.Utc,
97): number | null {
98 const since = status === "down" ? tunnel.connsInactiveAt : tunnel.createdAt;
99 if (typeof since !== "string") return null;
100 const timestamp = DateTime.make(since);
101 if (Option.isNone(timestamp)) return null;
102 return (now.epochMilliseconds - timestamp.value.epochMilliseconds) / 86_400_000;
103}
104
105function rotatedPages(input: {
106 readonly totalCount: number | undefined;
107 readonly slot: number;
108 readonly limit: number;
109}): ReadonlyArray<number> {
110 if (input.limit <= 0) return [];
111 if (input.totalCount === undefined) {
112 return Array.from({ length: input.limit }, (_, index) => index + 2);
113 }
114 const laterPageCount = Math.max(
115 0,
116 Math.ceil(input.totalCount / MANAGED_ENDPOINT_SWEEP_PAGE_SIZE) - 1,
117 );
118 if (laterPageCount === 0) return [];
119 const count = Math.min(input.limit, laterPageCount);
120 const start = input.slot % laterPageCount;
121 return Array.from({ length: count }, (_, index) => 2 + ((start + index) % laterPageCount));
122}
123
124const emptyResult = (
125 mode: RelayConfiguration.ManagedEndpointCleanupMode,
126 legacyMode: RelayConfiguration.ManagedEndpointCleanupMode,
127): ManagedEndpointSweepResult => ({
128 mode,
129 legacyMode,
130 listRequests: 0,
131 scanned: 0,
132 attempted: 0,
133 deleted: 0,
134 wouldDelete: 0,
135 wouldDeleteLegacy: 0,
136 deletedLegacy: 0,
137 skippedLegacy: 0,
138 legacyOver7Days: 0,
139 legacyOver30Days: 0,
140 legacyOver90Days: 0,
141 skippedOrphan: 0,
142 skippedReplaced: 0,
143 skippedUnrecorded: 0,
144 totalDown: null,
145 totalInactive: null,
146 failed: 0,
147 truncated: false,
148});
149
150const make = Effect.gen(function* () {
151 const config = yield* RelayConfiguration.RelayConfiguration;
152 const tunnels = yield* ManagedEndpointProvider.ManagedEndpointTunnelClient;
153 const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
154 const provider = yield* ManagedEndpointProvider.ManagedEndpointProvider;
155
156 const sweep = Effect.gen(function* () {
157 const mode = config.managedEndpointCleanupMode ?? "off";
158 const legacyMode = config.legacyManagedEndpointCleanupMode ?? "off";
159 const namespace = config.managedEndpointNamespace;
160 if ((mode === "off" && legacyMode === "off") || !namespace) {
161 return emptyResult(mode, legacyMode);
162 }
163 const sweepStartedAtMillis = yield* Clock.currentTimeMillis;
164 const legacyGraceMinutes =
165 namespace !== MANAGED_ENDPOINT_ZONE_OWNER_STAGE &&
166 config.legacyTunnelGraceMinutes !== undefined
167 ? config.legacyTunnelGraceMinutes
168 : MANAGED_ENDPOINT_LEGACY_GRACE_PERIOD_DAYS * 24 * 60;
169 const legacyCutoff = DateTime.subtract(yield* DateTime.now, { minutes: legacyGraceMinutes });
170
171 const now = yield* DateTime.now;
172 const cutoffFor = (status: "down" | "inactive") =>
173 DateTime.subtract(now, {
174 minutes:
175 status === "down"
176 ? MANAGED_ENDPOINT_GRACE_PERIOD_MINUTES
177 : MANAGED_ENDPOINT_INACTIVE_GRACE_PERIOD_MINUTES,
178 });
179 const prefix = managedEndpointTunnelNamePrefix(namespace);
180 const slot = Math.floor(
181 now.epochMilliseconds / (MANAGED_ENDPOINT_GRACE_PERIOD_MINUTES * 60 * 1_000),
182 );
183 let listRequests = 0;
184 let truncated = false;
185 const totals: Record<"down" | "inactive", number | null> = { down: null, inactive: null };
186 const expired: Array<{
187 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel & {
188 readonly id: string;
189 readonly name: string;
190 };
191 readonly status: "down" | "inactive";
192 readonly cutoff: DateTime.Utc;
193 }> = [];
194
195 for (const status of ["down", "inactive"] as const) {
196 const cutoff = cutoffFor(status);
197 const cutoffIso = DateTime.formatIso(cutoff);
198 const listPage = (page: number) => {
199 listRequests += 1;
200 return tunnels.list({
201 isDeleted: false,
202 includePrefix: prefix,
203 status,
204 existedAt: cutoffIso,
205 ...(status === "down" ? { wasInactiveAt: cutoffIso } : {}),
206 page,
207 perPage: MANAGED_ENDPOINT_SWEEP_PAGE_SIZE,
208 });
209 };
210 const first = yield* listPage(1);
211 const totalCount =
212 typeof first.resultInfo?.totalCount === "number" ? first.resultInfo.totalCount : undefined;
213 totals[status] = totalCount ?? null;
214 const pages = rotatedPages({
215 totalCount,
216 slot,
217 limit: Math.floor(MANAGED_ENDPOINT_SWEEP_LIST_REQUEST_LIMIT / 2) - 1,
218 });
219 const responses = [first, ...(yield* Effect.forEach(pages, listPage, { concurrency: 1 }))];
220 if (
221 totalCount !== undefined &&
222 Math.ceil(totalCount / MANAGED_ENDPOINT_SWEEP_PAGE_SIZE) > responses.length
223 ) {
224 truncated = true;
225 } else if (
226 totalCount === undefined &&
227 responses.at(-1)?.result.length === MANAGED_ENDPOINT_SWEEP_PAGE_SIZE
228 ) {
229 truncated = true;
230 }
231 for (const response of responses) {
232 expired.push(
233 ...response.result
234 .map((tunnel) => ({ tunnel, status, prefix, cutoff }))
235 .filter(isExpiredManagedTunnel)
236 .map(({ tunnel }) => ({ tunnel, status, cutoff })),
237 );
238 }
239 }
240
241 const collected = [...new Map(expired.map((entry) => [entry.tunnel.id, entry])).values()];
242 // Rotate the start offset so repeatedly failing candidates cannot starve later ones.
243 const offset =
244 collected.length === 0 ? 0 : (slot * MANAGED_ENDPOINT_SWEEP_ATTEMPT_LIMIT) % collected.length;
245 const uniqueExpired = [...collected.slice(offset), ...collected.slice(0, offset)];
246 const recorded = yield* allocations.listByTunnelNames(
247 uniqueExpired.map(({ tunnel }) => tunnel.name),
248 );
249 const recordedByTunnelName = new Map(
250 recorded.map((allocation) => [allocation.tunnelName, allocation]),
251 );
252 let attempted = 0;
253 let deleted = 0;
254 let wouldDelete = 0;
255 const candidates: Array<{
256 readonly owner: ManagedEndpointAllocations.ManagedEndpointTunnelAllocation;
257 readonly tunnel: ManagedEndpointProvider.ManagedEndpointTunnel & {
258 readonly id: string;
259 readonly name: string;
260 };
261 readonly status: "down" | "inactive";
262 readonly legacy: boolean;
263 readonly inactiveBefore: string;
264 }> = [];
265 let wouldDeleteLegacy = 0;
266 let deletedLegacy = 0;
267 let skippedLegacy = 0;
268 const legacyOverDays = new Map<number, number>(
269 MANAGED_ENDPOINT_LEGACY_AGE_BUCKET_DAYS.map((days) => [days, 0]),
270 );
271 let skippedOrphan = 0;
272 let skippedReplaced = 0;
273 let skippedUnrecorded = 0;
274 let failed = 0;
275
276 for (const { tunnel, status, cutoff } of uniqueExpired) {
277 const allocation = recordedByTunnelName.get(tunnel.name);
278 if (
279 allocation !== undefined &&
280 allocation.tunnelId !== null &&
281 allocation.tunnelId !== tunnel.id
282 ) {
283 skippedReplaced += 1;
284 continue;
285 }
286 const owner = allocation?.tunnelId === tunnel.id ? allocation : undefined;
287 if (allocation !== undefined && owner === undefined) {
288 skippedUnrecorded += 1;
289 continue;
290 }
291 // Without an allocation claim, deletion can race relinking by name; leave orphans for manual sweeps.
292 if (owner === undefined) {
293 skippedOrphan += 1;
294 continue;
295 }
296 const legacy = !owner.recoveryEnabled;
297 if (legacy) {
298 skippedLegacy += 1;
299 const inactiveDays = inactiveDaysAt(tunnel, status, now);
300 for (const days of MANAGED_ENDPOINT_LEGACY_AGE_BUCKET_DAYS) {
301 if (inactiveDays !== null && inactiveDays > days) {
302 legacyOverDays.set(days, (legacyOverDays.get(days) ?? 0) + 1);
303 }
304 }
305 if (
306 legacyMode === "off" ||
307 !isExpiredManagedTunnel({ tunnel, status, prefix, cutoff: legacyCutoff })
308 ) {
309 continue;
310 }
311 wouldDeleteLegacy += 1;
312 if (legacyMode === "dry-run") continue;
313 } else {
314 if (mode === "off") continue;
315 wouldDelete += 1;
316 if (mode === "dry-run") continue;
317 }
318 if (candidates.length >= MANAGED_ENDPOINT_SWEEP_ATTEMPT_LIMIT) {
319 truncated = true;
320 break;
321 }
322 candidates.push({
323 owner,
324 tunnel,
325 status,
326 legacy,
327 inactiveBefore: DateTime.formatIso(legacy ? legacyCutoff : cutoff),
328 });
329 }
330
331 const deleteDeadline = sweepStartedAtMillis + MANAGED_ENDPOINT_SWEEP_DELETE_BUDGET_MS;
332 let stopDeleting = false;
333 yield* Effect.forEach(
334 candidates,
335 (candidate) =>
336 Effect.gen(function* () {
337 if (stopDeleting || (yield* Clock.currentTimeMillis) >= deleteDeadline) {
338 stopDeleting = true;
339 truncated = true;
340 return;
341 }
342 attempted += 1;
343 const result = yield* provider
344 .release({
345 userId: candidate.owner.userId,
346 environmentId: candidate.owner.environmentId,
347 expectedTunnelId: candidate.tunnel.id,
348 expectedInactiveBefore: candidate.inactiveBefore,
349 expectedStatus: candidate.status,
350 ...(candidate.legacy ? { markReleased: true } : {}),
351 })
352 .pipe(Effect.result);
353 if (result._tag === "Failure") {
354 failed += 1;
355 yield* Effect.logWarning("Failed to delete an inactive managed tunnel", {
356 tunnelId: candidate.tunnel.id,
357 tunnelName: candidate.tunnel.name,
358 legacy: candidate.legacy,
359 cause: result.failure,
360 });
361 if (ManagedEndpointProvider.isCloudflareRateLimited(result.failure)) {
362 stopDeleting = true;
363 truncated = true;
364 }
365 } else if (result.success) {
366 deleted += 1;
367 if (candidate.legacy) deletedLegacy += 1;
368 yield* Effect.logInfo("Deleted an inactive managed tunnel", {
369 tunnelId: candidate.tunnel.id,
370 tunnelName: candidate.tunnel.name,
371 status: candidate.status,
372 legacy: candidate.legacy,
373 });
374 }
375 }),
376 { concurrency: MANAGED_ENDPOINT_SWEEP_DELETE_CONCURRENCY, discard: true },
377 );
378
379 return {
380 mode,
381 legacyMode,
382 listRequests,
383 scanned: uniqueExpired.length,
384 attempted,
385 deleted,
386 wouldDelete,
387 wouldDeleteLegacy,
388 deletedLegacy,
389 skippedLegacy,
390 legacyOver7Days: legacyOverDays.get(7) ?? 0,
391 legacyOver30Days: legacyOverDays.get(30) ?? 0,
392 legacyOver90Days: legacyOverDays.get(90) ?? 0,
393 skippedOrphan,
394 skippedReplaced,
395 skippedUnrecorded,
396 totalDown: totals.down,
397 totalInactive: totals.inactive,
398 failed,
399 truncated,
400 };
401 }).pipe(
402 Effect.tap((result) =>
403 Effect.annotateCurrentSpan(
404 Object.fromEntries(
405 Object.entries(result).map(([key, value]) => [
406 `relay.managed_endpoint_reaper.${key}`,
407 value,
408 ]),
409 ),
410 ),
411 ),
412 Effect.withSpan("relay.managed_endpoint_reaper.sweep"),
413 );
414
415 return ManagedEndpointReaper.of({ sweep });
416});
417
418export const layer = Layer.effect(ManagedEndpointReaper, make);