feat(health): деплоить probe-Worker из CFDM и опрашивать цели с edge

Worker сам ходит на origin по Cron Trigger; CFDM кладёт цели в KV и забирает результаты без POST /probe.

Co-authored-by: Cursor <[email protected]>

workers/health-probe/README.md

@@ -1,37 +1,17 @@
 # CFDM health-probe Worker
 
-Stateless edge probe for CFDM. **Not** Cloudflare Health Checks API (unavailable on Free).
-Cron stays in CFDM — this Worker has no Cron Trigger.
+Edge probe for CFDM. **Not** Cloudflare Health Checks API (unavailable on Free).
 
-## Deploy
+Production: CFDM creates this Worker via the Cloudflare API (KV mailbox + Cron Trigger).
+You do not need `wrangler deploy`. Token needs **Account**: Workers Scripts Write and Workers KV Storage Write.
+
+Local debug:
 
 ```powershell
 cd workers/health-probe
-npx wrangler login
-npx wrangler secret put PROBE_TOKEN
-npx wrangler deploy
+npx wrangler dev
 ```
 
-Paste the Worker URL (`https://cfdm-health-probe.<account>.workers.dev`) and the same token into **Настройки → Health-check**.
+Worker reads KV `targets`, probes TCP (`cloudflare:sockets` + `opened`) or HTTP (`fetch` to IP + `Host`), writes KV `results`. Batch ≤ 48, concurrency 5.
 
-Free Workers ≈ 100k requests/day. CFDM cron every 2 minutes × number of IPs must fit.
-
-## API
-
-`POST /probe` + `Authorization: Bearer <PROBE_TOKEN>`
-
-```json
-{
-  "type": "tcp",
-  "ip": "1.2.3.4",
-  "hostname": "app.example.com",
-  "port": 443,
-  "path": "/",
-  "expected_status": 200,
-  "timeout_ms": 3000,
-  "verify_tls": true,
-  "method": "GET"
-}
-```
-
-Response: `{ "ok": true, "latencyMs": 42, "error": null, "colo": "AMS" }`.
+`GET /` — liveness.

workers/health-probe/src/index.mjs

@@ -0,0 +1,228 @@
+/**
+ * CFDM health-probe Worker. Cron Trigger reads KV `targets`, probes TCP/HTTP
+ * from the edge (UptimeFlare-style), writes KV `results`. CFDM is SoT in SQLite.
+ */
+
+const TARGETS_KEY = "targets";
+const RESULTS_KEY = "results";
+const CURSOR_KEY = "cursor";
+const BATCH = 48;
+const CONCURRENCY = 5;
+const COOLDOWN_MS = 3 * 60 * 1000;
+const UA = "CFDM-health-probe/1.0";
+
+export default {
+  async fetch() {
+    return new Response(JSON.stringify({ ok: true, service: "cfdm-health-probe" }), {
+      headers: { "content-type": "application/json" },
+    });
+  },
+
+  async scheduled(_event, env) {
+    await probeBatch(env);
+  },
+};
+
+async function probeBatch(env) {
+  const raw = await env.HEALTH_KV.get(TARGETS_KEY);
+  if (!raw) return;
+  let doc;
+  try {
+    doc = JSON.parse(raw);
+  } catch {
+    return;
+  }
+  const items = Array.isArray(doc.items) ? doc.items : [];
+  if (items.length === 0) return;
+
+  let offset = 0;
+  const cursorRaw = await env.HEALTH_KV.get(CURSOR_KEY);
+  if (cursorRaw) {
+    try {
+      const cursor = JSON.parse(cursorRaw);
+      if (Number.isFinite(cursor.offset) && cursor.offset >= 0) {
+        offset = cursor.offset % items.length;
+      }
+    } catch {
+      offset = 0;
+    }
+  }
+
+  const slice = items.slice(offset, offset + BATCH);
+  const nextOffset = offset + slice.length >= items.length ? 0 : offset + slice.length;
+
+  const colo = await readColo();
+  const probed = await mapPool(slice, CONCURRENCY, async (target) => {
+    const type = target.type === "http" ? "http" : "tcp";
+    const port = Number(target.port) || (type === "http" ? 80 : 80);
+    const timeoutMs = Math.min(Math.max(Number(target.timeoutMs) || 3000, 100), 25_000);
+    const hostname = String(target.hostname ?? "").trim() || target.ip;
+    try {
+      const result =
+        type === "http"
+          ? await httpProbe({
+              ip: target.ip,
+              hostname,
+              port,
+              path: target.path || "/",
+              expectedStatus: target.expectedStatus ?? 200,
+              timeoutMs,
+              verifyTls: Boolean(target.verifyTls),
+            })
+          : await tcpProbe(target.ip, port, timeoutMs);
+      return { key: target.key, ...result };
+    } catch (err) {
+      return {
+        key: target.key,
+        ok: false,
+        latencyMs: 0,
+        error: err instanceof Error ? err.message : "probe failed",
+      };
+    }
+  });
+
+  const fingerprint = resultFingerprint(probed);
+  const previousRaw = await env.HEALTH_KV.get(RESULTS_KEY);
+  let skipWrite = false;
+  if (previousRaw) {
+    try {
+      const prev = JSON.parse(previousRaw);
+      const age = Date.now() - Date.parse(prev.probedAt);
+      if (prev.fingerprint === fingerprint && Number.isFinite(age) && age < COOLDOWN_MS) {
+        skipWrite = true;
+      }
+    } catch {
+      skipWrite = false;
+    }
+  }
+
+  if (!skipWrite) {
+    const results = {
+      probedAt: new Date().toISOString(),
+      colo,
+      fingerprint,
+      items: probed,
+    };
+    await env.HEALTH_KV.put(RESULTS_KEY, JSON.stringify(results));
+  }
+  if (items.length > BATCH || offset !== 0) {
+    await env.HEALTH_KV.put(CURSOR_KEY, JSON.stringify({ offset: nextOffset }));
+  }
+}
+
+function resultFingerprint(items) {
+  return items
+    .map((item) => `${item.key}:${item.ok ? "1" : "0"}:${item.error ?? ""}`)
+    .sort()
+    .join("|");
+}
+
+async function mapPool(items, concurrency, fn) {
+  if (items.length === 0) return [];
+  const results = new Array(items.length);
+  let next = 0;
+  async function worker() {
+    while (next < items.length) {
+      const idx = next;
+      next += 1;
+      results[idx] = await fn(items[idx]);
+    }
+  }
+  const n = Math.min(concurrency, items.length);
+  await Promise.all(Array.from({ length: n }, () => worker()));
+  return results;
+}
+
+async function readColo() {
+  try {
+    const res = await fetch("https://www.cloudflare.com/cdn-cgi/trace", {
+      cf: { cacheTtlByStatus: { "100-599": -1 } },
+    });
+    const text = await res.text();
+    const line = text.split("\n").find((row) => row.startsWith("colo="));
+    return line ? line.slice(5).trim() || null : null;
+  } catch {
+    return null;
+  }
+}
+
+function withTimeout(promise, timeoutMs, label) {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => reject(new Error(`${label} timeout`)), timeoutMs);
+    promise.then(
+      (value) => {
+        clearTimeout(timer);
+        resolve(value);
+      },
+      (err) => {
+        clearTimeout(timer);
+        reject(err);
+      },
+    );
+  });
+}
+
+async function tcpProbe(ip, port, timeoutMs) {
+  const started = Date.now();
+  const { connect } = await import("cloudflare:sockets");
+  const socket = connect({ hostname: ip, port });
+  try {
+    await withTimeout(socket.opened, timeoutMs, "tcp");
+    return { ok: true, latencyMs: Date.now() - started, error: null };
+  } catch (err) {
+    const message = err instanceof Error ? err.message : "tcp failed";
+    return { ok: false, latencyMs: Date.now() - started, error: message };
+  } finally {
+    try {
+      socket.close();
+    } catch {
+      // ignore
+    }
+  }
+}
+
+async function httpProbe(opts) {
+  const started = Date.now();
+  const useTls = opts.verifyTls || opts.port === 443;
+  const host = opts.ip.includes(":") ? `[${opts.ip}]` : opts.ip;
+  const path = opts.path.startsWith("/") ? opts.path : `/${opts.path}`;
+  const url = `${useTls ? "https" : "http"}://${host}:${opts.port}${path}`;
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), opts.timeoutMs);
+  try {
+    const res = await fetch(url, {
+      method: "GET",
+      headers: {
+        Host: opts.hostname,
+        "User-Agent": UA,
+      },
+      signal: controller.signal,
+      redirect: "manual",
+      cf: { cacheTtlByStatus: { "100-599": -1 } },
+    });
+    try {
+      await res.body?.cancel();
+    } catch {
+      // ignore
+    }
+    const latencyMs = Date.now() - started;
+    if (res.status !== opts.expectedStatus) {
+      return {
+        ok: false,
+        latencyMs,
+        error: `HTTP ${res.status} (ожидали ${opts.expectedStatus})`,
+      };
+    }
+    return { ok: true, latencyMs, error: null };
+  } catch (err) {
+    const message =
+      err instanceof Error
+        ? err.name === "AbortError"
+          ? "http timeout"
+          : err.message
+        : "http failed";
+    return { ok: false, latencyMs: Date.now() - started, error: message };
+  } finally {
+    clearTimeout(timer);
+  }
+}

workers/health-probe/src/index.ts

@@ -1,181 +0,0 @@
-/**
- * Stateless CFDM health probe. Cron lives in CFDM API — this Worker only
- * answers POST /probe. Deploy: wrangler deploy; paste URL + token into
- * Настройки → Health-check.
- */
-
-export interface Env {
-  PROBE_TOKEN: string;
-}
-
-type ProbeType = "tcp" | "http";
-
-interface ProbeRequest {
-  type?: ProbeType;
-  ip?: string;
-  hostname?: string;
-  port?: number;
-  path?: string;
-  expected_status?: number | null;
-  timeout_ms?: number;
-  verify_tls?: boolean;
-  method?: string;
-}
-
-interface ProbeResponse {
-  ok: boolean;
-  latencyMs: number;
-  error: string | null;
-  colo: string | null;
-}
-
-export default {
-  async fetch(request: Request, env: Env): Promise<Response> {
-    const colo =
-      (request as Request & { cf?: { colo?: string } }).cf?.colo ?? null;
-    if (request.method !== "POST") {
-      return json({ ok: false, latencyMs: 0, error: "method not allowed", colo }, 405);
-    }
-    const pathname = new URL(request.url).pathname.replace(/\/$/, "") || "/";
-    if (pathname !== "/probe") {
-      return json({ ok: false, latencyMs: 0, error: "not found", colo }, 404);
-    }
-    const token = bearer(request);
-    if (!env.PROBE_TOKEN || token !== env.PROBE_TOKEN) {
-      return json({ ok: false, latencyMs: 0, error: "unauthorized", colo }, 401);
-    }
-
-    let body: ProbeRequest;
-    try {
-      body = (await request.json()) as ProbeRequest;
-    } catch {
-      return json({ ok: false, latencyMs: 0, error: "invalid json", colo }, 400);
-    }
-
-    const ip = String(body.ip ?? "").trim();
-    if (!ip) {
-      return json({ ok: false, latencyMs: 0, error: "ip required", colo }, 400);
-    }
-    const type: ProbeType = body.type === "http" ? "http" : "tcp";
-    const port = Number(body.port) || (type === "http" ? 80 : 80);
-    const timeoutMs = Math.min(Math.max(Number(body.timeout_ms) || 3000, 100), 25_000);
-    const hostname = String(body.hostname ?? "").trim() || ip;
-
-    try {
-      const result =
-        type === "http"
-          ? await httpProbe({
-              ip,
-              hostname,
-              port,
-              path: body.path || "/",
-              expectedStatus: body.expected_status ?? 200,
-              timeoutMs,
-              verifyTls: Boolean(body.verify_tls),
-              method: (body.method || "GET").toUpperCase(),
-            })
-          : await tcpProbe(ip, port, timeoutMs);
-      return json({ ...result, colo });
-    } catch (err) {
-      const message = err instanceof Error ? err.message : "probe failed";
-      return json({ ok: false, latencyMs: 0, error: message, colo });
-    }
-  },
-};
-
-function bearer(request: Request): string {
-  const header = request.headers.get("Authorization") ?? "";
-  return header.startsWith("Bearer ") ? header.slice(7) : "";
-}
-
-function json(body: ProbeResponse, status = 200): Response {
-  return new Response(JSON.stringify(body), {
-    status,
-    headers: { "content-type": "application/json" },
-  });
-}
-
-function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
-  return new Promise((resolve, reject) => {
-    const timer = setTimeout(() => reject(new Error(`${label} timeout`)), timeoutMs);
-    promise.then(
-      (value) => {
-        clearTimeout(timer);
-        resolve(value);
-      },
-      (err) => {
-        clearTimeout(timer);
-        reject(err);
-      },
-    );
-  });
-}
-
-async function tcpProbe(
-  ip: string,
-  port: number,
-  timeoutMs: number,
-): Promise<Omit<ProbeResponse, "colo">> {
-  const started = Date.now();
-  const { connect } = await import("cloudflare:sockets");
-  const socket = connect({ hostname: ip, port });
-  try {
-    await withTimeout(socket.opened, timeoutMs, "tcp");
-    return { ok: true, latencyMs: Date.now() - started, error: null };
-  } catch (err) {
-    const message = err instanceof Error ? err.message : "tcp failed";
-    return { ok: false, latencyMs: Date.now() - started, error: message };
-  } finally {
-    try {
-      socket.close();
-    } catch {
-      // ignore
-    }
-  }
-}
-
-async function httpProbe(opts: {
-  ip: string;
-  hostname: string;
-  port: number;
-  path: string;
-  expectedStatus: number;
-  timeoutMs: number;
-  verifyTls: boolean;
-  method: string;
-}): Promise<Omit<ProbeResponse, "colo">> {
-  const started = Date.now();
-  const useTls = opts.verifyTls || opts.port === 443;
-  const host = opts.ip.includes(":") ? `[${opts.ip}]` : opts.ip;
-  const path = opts.path.startsWith("/") ? opts.path : `/${opts.path}`;
-  const url = `${useTls ? "https" : "http"}://${host}:${opts.port}${path}`;
-  const controller = new AbortController();
-  const timer = setTimeout(() => controller.abort(), opts.timeoutMs);
-  try {
-    const res = await fetch(url, {
-      method: opts.method === "HEAD" ? "HEAD" : "GET",
-      headers: { Host: opts.hostname },
-      signal: controller.signal,
-      redirect: "manual",
-    });
-    const latencyMs = Date.now() - started;
-    if (res.status !== opts.expectedStatus) {
-      return {
-        ok: false,
-        latencyMs,
-        error: `HTTP ${res.status} (ожидали ${opts.expectedStatus})`,
-      };
-    }
-    return { ok: true, latencyMs, error: null };
-  } catch (err) {
-    const message =
-      err instanceof Error
-        ? err.name === "AbortError"
-          ? "http timeout"
-          : err.message
-        : "http failed";
-    return { ok: false, latencyMs: Date.now() - started, error: message };
-  } finally {
-    clearTimeout(timer);
-  }
-}

workers/health-probe/wrangler.toml

@@ -1,6 +1,10 @@
 name = "cfdm-health-probe"
-main = "src/index.ts"
+main = "src/index.mjs"
 compatibility_date = "2025-04-01"
 
-# Set the shared secret: wrangler secret put PROBE_TOKEN
-# Then paste the Worker URL + token into CFDM → Настройки → Health-check.
+# Production Worker is created by CFDM (Workers Scripts API + KV + Cron Trigger).
+# This file is for local `wrangler dev` only.
+
+[[kv_namespaces]]
+binding = "HEALTH_KV"
+id = "00000000-0000-0000-0000-000000000000"