[Examples & benchmarks](https://cdpfleet.com/docs/cases) / #86 · 2026-11-07 · Example · Medium · for [web scraping](https://cdpfleet.com/web-scraping), [llm data](https://cdpfleet.com/llm-data)

# Crawl a whole site as a job: map, crawl, stop, resume, results, zip and a signed webhook

Map books.toscrape.com, crawl 100 pages of it as a background job, stop it part-way and resume it, then read the results as JSON lines and as a zip archive, and check the signed \`job.done\` and \`job.archive\` webhooks. Pages per minute and thread-seconds per page from the job's own bill, in five languages.

[Camoufox](https://cdpfleet.com/docs/browsers/camoufox)

Run on production on 2026-11-10: ✓ Node.js ✓ Python ✓ Java ✓ C# ✓ Go

## The problem

One page is one call. A whole site is hundreds or thousands of pages, minutes or hours of work, and you don't want to hold a connection open for it, re-fetch what you already have after a stop, or poll forever to learn it's done. A [crawl job](https://cdpfleet.com/docs/jobs) runs in the background on your threads: you start it with one request, follow it, stop it and resume it where it left off, and get told by a signed webhook when it's finished. This case does the whole round trip on a site made for scraping practice, with a bounded crawl, and reads what it cost from the job itself.

## What we used, and why

| What | Why |
| --- | --- |
| `POST /v1/functions/map` | The instant map: the URLs a site lists in its sitemaps plus the links on its start page, in one call — a quick look at how big the site is before crawling it. |
| `POST /v1/functions/crawl` | `max_depth: 2`, `max_pages: 100`, `concurrency: 4`, `delay_ms: 250`, `formats: ["markdown"]` and an `extract` of each page's `h1` and price. Returns `202 {job_id}` at once; the job runs in pooled Camoufox tabs. |
| `deliver: {webhook: {url, events: ["done", "archive"]}, format: "archive"}` | A zip of every result built at the end, a `job.done` POST to your URL when the job ends, and a `job.archive` POST with the zip's `bytes` and download `url` when it is ready (only when `events` includes `"archive"`). The receiver in the programs is cdpfleet.com's own test inbox, which keeps deliveries an hour and reads them back with GET; use your own public https endpoint. |
| `GET /v1/jobs/{id}` | Polled every 3 s: `status`, `progress {done, failed, queued, total}`, `runs`, `concurrency`, `billed_seconds`, `results`, `archive`. |
| `DELETE /v1/jobs/{id}` → `POST /v1/jobs/{id}/resume` | Stop once 30 pages are done (`stopping`, then `stopped`, results kept), then continue as a second run from the job's checkpoint. |
| `GET /v1/jobs/{id}/results?cursor=` and `/results.jsonl` | The stored results at the stop, part by part; at the end, everything as one JSON-lines download. |
| `GET /v1/jobs/{id}/archive` | `{status, bytes, url}`: the `url` downloads the zip straight from storage, without your key. Read with the standard zip library in each language (Node: a short central-directory reader and `zlib`). |
| `GET /v1/me/webhook-secret` | The account's `whsec_…`; each delivery's `x-cdpfleet-signature: t=…,v1=…` must equal HMAC-SHA256(secret, `"<t>.<raw body>"`). |

## How it works

1.  Map the site with one instant call and count the URLs it found and where they came from.
2.  Start the crawl job with a webhook and `format: "archive"`; poll it until 30 pages are done, then stop it and time how long it takes to reach `stopped`. Read the results stored so far and what the webhook inbox has received.
3.  Resume it and poll until the second run has ended.
4.  Download `results.jsonl`: count the lines, the distinct URLs, the pages with Markdown, the book pages with a price, the depths, and the URLs the second run added.
5.  Wait for the archive, download the zip and list it: the Markdown and `extract.json` files per page, `index.jsonl`, `errors.jsonl`, `job.json`.
6.  Read the deliveries from the webhook inbox, check every signature with the account's secret, and download the zip again from the link in `job.archive`.
7.  Work out pages per minute and thread-seconds per page from `billed_seconds` and `concurrency`.

## The code

The same program in five languages (also [on GitHub](https://github.com/cdpfleet/cdpfleet/tree/main/cases/86-crawl-a-site-as-a-job), with the raw output). Set these environment variables first:

-   `CDPFLEET_API_KEY` — your API key (dashboard → API keys)

 

```node
// Node.js 18+, no packages. env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
import crypto from 'node:crypto';
import zlib from 'node:zlib';

const KEY = process.env.CDPFLEET_API_KEY;
const API = 'https://cdpfleet.com';
const SITE = 'https://books.toscrape.com/'; // a sandbox made for scraping practice
const CRAWL = { max_depth: 2, max_pages: 100, concurrency: 4, delay_ms: 250, formats: ['markdown'] };
const EXTRACT = { title: 'h1', price: '.product_main .price_color' };
const STOP_AT = 30; // stop the job once this many pages are done, then resume it
const ENDED = ['done', 'failed', 'stopped', 'interrupted'];
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const secs = (ms) => Math.round(ms / 100) / 10;

async function api(method, path, body) {
  const res = await fetch(`${API}${path}`, {
    method,
    headers: { 'x-api-key': KEY, ...(body ? { 'content-type': 'application/json' } : {}) },
    body: body ? JSON.stringify(body) : undefined,
  });
  const text = await res.text();
  let json = null;
  try { json = JSON.parse(text); } catch { /* not JSON */ }
  return { status: res.status, json, text, headers: res.headers };
}
// Poll the job every 3 s until `until(job)` holds.
async function waitFor(id, until, ms) {
  let job;
  for (const end = Date.now() + ms; Date.now() < end; await sleep(3000)) {
    job = (await api('GET', `/v1/jobs/${id}`)).json;
    if (job && until(job)) return job;
  }
  throw new Error(`job ${id}: still ${job?.status} after ${ms / 1000} s`);
}
async function stored(id) { // results stored so far, part by part
  const rows = [];
  for (let cursor = 0; cursor !== null;) {
    const r = (await api('GET', `/v1/jobs/${id}/results?cursor=${cursor}`)).json;
    rows.push(...r.results);
    if (r.next_cursor === cursor) break; // nothing new yet
    cursor = r.next_cursor;
  }
  return rows;
}

// A zip's file names, and one file's bytes (Node has no unzip of its own; zip64-aware).
function unzip(buf, want) {
  const eocd = buf.lastIndexOf(Buffer.from([0x50, 0x4b, 0x05, 0x06]));
  let count = buf.readUInt16LE(eocd + 10);
  let at = buf.readUInt32LE(eocd + 16);
  if (count === 0xffff || at === 0xffffffff) {
    const z = Number(buf.readBigUInt64LE(eocd - 20 + 8)); // zip64 end record
    count = Number(buf.readBigUInt64LE(z + 32));
    at = Number(buf.readBigUInt64LE(z + 48));
  }
  const names = [];
  let data = null;
  for (let i = 0; i < count; i++) {
    let comp = buf.readUInt32LE(at + 20);
    let local = buf.readUInt32LE(at + 42);
    const method = buf.readUInt16LE(at + 10);
    const nameLen = buf.readUInt16LE(at + 28);
    const extraLen = buf.readUInt16LE(at + 30);
    const name = buf.toString('utf8', at + 46, at + 46 + nameLen);
    for (let x = at + 46 + nameLen; x < at + 46 + nameLen + extraLen; x += 4 + buf.readUInt16LE(x + 2)) {
      if (buf.readUInt16LE(x) !== 1) continue; // zip64 sizes and offset
      let p = x + 4;
      if (buf.readUInt32LE(at + 24) === 0xffffffff) p += 8;
      if (comp === 0xffffffff) { comp = Number(buf.readBigUInt64LE(p)); p += 8; }
      if (local === 0xffffffff) local = Number(buf.readBigUInt64LE(p));
    }
    names.push(name);
    if (name === want) {
      const start = local + 30 + buf.readUInt16LE(local + 26) + buf.readUInt16LE(local + 28);
      const raw = buf.subarray(start, start + comp);
      data = method === 8 ? zlib.inflateRawSync(raw) : raw;
    }
    at += 46 + nameLen + extraLen + buf.readUInt16LE(at + 32);
  }
  return { names, data };
}

// 1. Map: the site's URLs in one instant call (sitemaps + the start page's links).
let t = Date.now();
const map = await api('POST', '/v1/functions/map', { url: SITE, max_urls: 5000 });
const mapOut = {
  status: map.status, seconds: secs(Date.now() - t), urls: map.json?.urls?.length ?? null,
  sources: map.json?.sources ?? null, partial: map.headers.get('x-cdpfleet-partial'),
  categories: (map.json?.urls || []).filter((u) => u.includes('/catalogue/category/')).length,
};

// 2. Start the crawl as a job: a zip archive at the end, and webhooks when it's done and when the zip is ready.
// The receiver here is cdpfleet.com's own test inbox (keeps deliveries an hour, read back with GET);
// use your own public https endpoint.
const hook = `${API}/e2e/hook/${crypto.randomBytes(18).toString('base64url')}`;
const tStart = Date.now();
const start = await api('POST', '/v1/functions/crawl', {
  url: SITE, ...CRAWL, extract: EXTRACT,
  deliver: { webhook: { url: hook, events: ['done', 'archive'] }, format: 'archive' },
});
if (start.status !== 202) throw new Error(`crawl ${start.status} ${start.text}`);
const id = start.json.job_id;

// 3. Watch it, and stop it part-way.
const before = await waitFor(id, (j) => (j.progress?.done || 0) >= STOP_AT || ENDED.includes(j.status), 600000);
const stopReply = await api('DELETE', `/v1/jobs/${id}`);
const tStop = Date.now();
const stopped = await waitFor(id, (j) => ENDED.includes(j.status), 300000);
const tStopped = Date.now();
const atStop = await stored(id);
const hooksAtStop = (await (await fetch(hook)).json()).items;
const archiveAtStop = (await api('GET', `/v1/jobs/${id}/archive`)).json;

// 4. Resume: a new run of the same job, from its checkpoint.
const tResume = Date.now();
const resume = await api('POST', `/v1/jobs/${id}/resume`);
// 5. Wait for the second run to end.
const done = await waitFor(id, (j) => j.runs >= 2 && ENDED.includes(j.status), 900000);
const tDone = Date.now();

// 6a. Every result as one JSON-lines file.
const jsonl = await (await fetch(`${API}/v1/jobs/${id}/results.jsonl`, { headers: { 'x-api-key': KEY } })).text();
const rows = jsonl.trim().split('\n').map((l) => JSON.parse(l));
const urls = new Set(rows.map((r) => r.url));
const firstRun = new Set(atStop.map((r) => r.url));

// 6b. The same results as a zip (built when the job ended).
const archive = await waitFor(id, (j) => ['ready', 'failed'].includes(j.archive?.status), 300000)
  .then(() => api('GET', `/v1/jobs/${id}/archive`)).then((r) => r.json);
const zip = Buffer.from(await (await fetch(archive.url)).arrayBuffer()); // a storage link, no key
const { names, data: index } = unzip(zip, 'index.jsonl');

// 7. The webhooks: every delivery is signed with the account's secret, as HMAC-SHA256(secret, "<t>.<raw body>").
const secret = (await api('GET', '/v1/me/webhook-secret')).json.secret;
const check = (d) => {
  const { t: ts, v1 } = Object.fromEntries((d.signature || '').split(',').map((kv) => kv.split('=')));
  const body = JSON.parse(d.body);
  return {
    event: d.event, status: body.data?.status ?? null, job_id_matches: body.job_id === id,
    delivery_id_is_uuid: /^[0-9a-f-]{36}$/.test(d.delivery || ''),
    signature_valid: Boolean(v1) && v1 === crypto.createHmac('sha256', secret).update(`${ts}.${d.body}`).digest('hex'),
    signed_at: Number(ts), data: body.data,
  };
};
let deliveries = [];
for (const end = Date.now() + 90000; Date.now() < end; await sleep(3000)) {
  deliveries = (await (await fetch(hook)).json()).items.map(check);
  if (deliveries.some((d) => d.event === 'job.done' && d.status === 'done') && deliveries.some((d) => d.event === 'job.archive')) break;
}
const final = deliveries.find((d) => d.event === 'job.done' && d.status === 'done');
// job.archive carries the zip's size and a download link, like GET /v1/jobs/{id}/archive.
const archiveHook = deliveries.find((d) => d.event === 'job.archive');
const hookZip = typeof archiveHook?.data?.url === 'string' ? await fetch(archiveHook.data.url).then((r) => r.arrayBuffer()) : null;

// 8. What it cost: a run holds `concurrency` threads, billed_seconds is the total in thread-seconds.
const runSeconds = done.billed_seconds / done.concurrency;

console.log(JSON.stringify({
  language: 'node',
  site: SITE,
  map: mapOut,
  crawl: { ...CRAWL, extract: Object.keys(EXTRACT), start_status: start.status },
  stop: {
    requested_at_done: before.progress?.done ?? 0, reply: stopReply.json?.status ?? stopReply.status,
    seconds_to_stopped: secs(tStopped - tStop), status: stopped.status, progress: stopped.progress,
    stored_results: atStop.length, webhooks_by_then: hooksAtStop.map((d) => `${d.event} ${JSON.parse(d.body).data?.status}`), archive: archiveAtStop?.status ?? null,
  },
  resume: { status: resume.status, reply: resume.json?.status ?? null },
  job: {
    status: done.status, runs: done.runs, progress: done.progress, concurrency: done.concurrency,
    billed_seconds: done.billed_seconds, results: { parts: done.results?.parts, bytes: done.results?.bytes },
    started_at: done.started_at, ended_at: done.ended_at, expires_at: done.expires_at,
  },
  results_jsonl: {
    lines: rows.length, distinct_urls: urls.size, bytes: Buffer.byteLength(jsonl),
    status_200: rows.filter((r) => r.status === 200).length,
    statuses: Object.fromEntries([...new Set(rows.map((r) => String(r.status)))].sort().map((s) => [s, rows.filter((r) => String(r.status) === s).length])),
    errors: rows.filter((r) => r.error).map((r) => ({ url: r.url, status: r.status ?? null, error: String(r.error).slice(0, 120) })),
    with_markdown: rows.filter((r) => typeof r.markdown === 'string' && r.markdown.length > 0).length,
    books_with_price: rows.filter((r) => typeof r.extract?.price === 'string').length,
    category_pages: rows.filter((r) => r.url.includes('/catalogue/category/')).length,
    depths: Object.fromEntries([...new Set(rows.map((r) => r.depth))].sort().map((d) => [d, rows.filter((r) => r.depth === d).length])),
    first_run: firstRun.size, second_run_new: [...urls].filter((u) => !firstRun.has(u)).length,
    sample: rows.filter((r) => r.extract?.price).slice(0, 2).map((r) => ({ url: r.url, title: r.extract.title, price: r.extract.price })),
  },
  archive: {
    status: archive.status, bytes: archive.bytes, downloaded_bytes: zip.length, files: names.length,
    markdown_files: names.filter((n) => n.endsWith('/index.md')).length,
    extract_files: names.filter((n) => n.endsWith('/extract.json')).length,
    index_lines: index ? index.toString('utf8').trim().split('\n').length : 0,
    has: ['index.jsonl', 'errors.jsonl', 'job.json'].filter((n) => names.includes(n)),
  },
  webhook: {
    deliveries: deliveries.map(({ data, signed_at, ...d }) => d),
    final_done: final ? { data: final.data, signed_seconds_after_end: final.signed_at - Math.floor(Date.parse(done.ended_at) / 1000) } : null,
    archive: archiveHook ? {
      status: archiveHook.data?.status ?? null, bytes: archiveHook.data?.bytes ?? null, has_url: typeof archiveHook.data?.url === 'string',
      url_downloads_bytes: hookZip ? hookZip.byteLength : null, signed_seconds_after_end: archiveHook.signed_at - Math.floor(Date.parse(done.ended_at) / 1000),
    } : null,
  },
  cost: {
    billed_thread_seconds: done.billed_seconds, pages: rows.length,
    run_seconds: Math.round(runSeconds * 10) / 10,
    pages_per_minute: Math.round(rows.length / (runSeconds / 60) * 10) / 10,
    thread_seconds_per_page: Math.round(done.billed_seconds / rows.length * 10) / 10,
    client_running_seconds: secs(tStopped - tStart + tDone - tResume),
  },
  measured_at: new Date().toISOString().slice(0, 10),
}, null, 2));
```

```python
# pip install requests
# env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
import hashlib
import hmac
import io
import json
import os
import re
import secrets
import time
import zipfile
from datetime import datetime, timezone

import requests

KEY = os.environ["CDPFLEET_API_KEY"]
API = "https://cdpfleet.com"
SITE = "https://books.toscrape.com/"  # a sandbox made for scraping practice
CRAWL = {"max_depth": 2, "max_pages": 100, "concurrency": 4, "delay_ms": 250, "formats": ["markdown"]}
EXTRACT = {"title": "h1", "price": ".product_main .price_color"}
STOP_AT = 30  # stop the job once this many pages are done, then resume it
ENDED = ["done", "failed", "stopped", "interrupted"]
HEADERS = {"x-api-key": KEY}

def secs(s):
    return round(s, 1)

def api(method, path, body=None):
    return requests.request(method, API + path, headers=HEADERS, json=body, timeout=90)

def wait_for(job_id, until, seconds):
    end = time.time() + seconds
    job = None
    while time.time() < end:
        job = api("GET", f"/v1/jobs/{job_id}").json()
        if until(job):
            return job
        time.sleep(3)
    raise RuntimeError(f"job {job_id}: still {job and job.get('status')} after {seconds} s")

def stored(job_id):  # results stored so far, part by part
    rows, cursor = [], 0
    while cursor is not None:
        r = api("GET", f"/v1/jobs/{job_id}/results?cursor={cursor}").json()
        rows += r["results"]
        if r["next_cursor"] == cursor:  # nothing new yet
            break
        cursor = r["next_cursor"]
    return rows

def hook_items(hook):
    return requests.get(hook, timeout=30).json()["items"]

# 1. Map: the site's URLs in one instant call (sitemaps + the start page's links).
t = time.time()
m = api("POST", "/v1/functions/map", {"url": SITE, "max_urls": 5000})
mj = m.json() if m.ok else {}
map_out = {
    "status": m.status_code, "seconds": secs(time.time() - t), "urls": len(mj.get("urls", [])) if m.ok else None,
    "sources": mj.get("sources"), "partial": m.headers.get("x-cdpfleet-partial"),
    "categories": sum("/catalogue/category/" in u for u in mj.get("urls", [])),
}

# 2. Start the crawl as a job: a zip archive at the end, and webhooks when it's done and when the zip is ready.
# The receiver here is cdpfleet.com's own test inbox (keeps deliveries an hour, read back with GET);
# use your own public https endpoint.
hook = f"{API}/e2e/hook/{secrets.token_urlsafe(18)}"
t_start = time.time()
start = api("POST", "/v1/functions/crawl", {
    "url": SITE, **CRAWL, "extract": EXTRACT,
    "deliver": {"webhook": {"url": hook, "events": ["done", "archive"]}, "format": "archive"},
})
if start.status_code != 202:
    raise RuntimeError(f"crawl {start.status_code} {start.text}")
job_id = start.json()["job_id"]

# 3. Watch it, and stop it part-way.
before = wait_for(job_id, lambda j: (j.get("progress") or {}).get("done", 0) >= STOP_AT or j["status"] in ENDED, 600)
stop_reply = api("DELETE", f"/v1/jobs/{job_id}")
t_stop = time.time()
stopped = wait_for(job_id, lambda j: j["status"] in ENDED, 300)
t_stopped = time.time()
at_stop = stored(job_id)
hooks_at_stop = hook_items(hook)
archive_at_stop = api("GET", f"/v1/jobs/{job_id}/archive").json()

# 4. Resume: a new run of the same job, from its checkpoint.
t_resume = time.time()
resume = api("POST", f"/v1/jobs/{job_id}/resume")
# 5. Wait for the second run to end.
done = wait_for(job_id, lambda j: j.get("runs", 0) >= 2 and j["status"] in ENDED, 900)
t_done = time.time()

# 6a. Every result as one JSON-lines file.
jsonl = api("GET", f"/v1/jobs/{job_id}/results.jsonl").content.decode("utf-8")
rows = [json.loads(line) for line in jsonl.strip().split("\n")]
urls = {r["url"] for r in rows}
first_run = {r["url"] for r in at_stop}

# 6b. The same results as a zip (built when the job ended).
wait_for(job_id, lambda j: (j.get("archive") or {}).get("status") in ("ready", "failed"), 300)
archive = api("GET", f"/v1/jobs/{job_id}/archive").json()
zip_bytes = requests.get(archive["url"], timeout=120).content  # a storage link, no key
with zipfile.ZipFile(io.BytesIO(zip_bytes)) as z:
    names = z.namelist()
    index = z.read("index.jsonl").decode("utf-8") if "index.jsonl" in names else ""

# 7. The webhooks: every delivery is signed with the account's secret, as HMAC-SHA256(secret, "<t>.<raw body>").
secret = api("GET", "/v1/me/webhook-secret").json()["secret"]

def check(d):
    parts = dict(kv.split("=", 1) for kv in (d.get("signature") or "").split(",") if "=" in kv)
    body = json.loads(d["body"])
    want = hmac.new(secret.encode(), f"{parts.get('t')}.{d['body']}".encode(), hashlib.sha256).hexdigest()
    return {
        "event": d["event"], "status": (body.get("data") or {}).get("status"), "job_id_matches": body.get("job_id") == job_id,
        "delivery_id_is_uuid": bool(re.fullmatch(r"[0-9a-f-]{36}", d.get("delivery") or "")),
        "signature_valid": bool(parts.get("v1")) and hmac.compare_digest(parts["v1"], want),
        "signed_at": int(parts.get("t") or 0), "data": body.get("data"),
    }

deliveries = []
end = time.time() + 90
while time.time() < end:
    deliveries = [check(d) for d in hook_items(hook)]
    if any(d["event"] == "job.done" and d["status"] == "done" for d in deliveries) and any(d["event"] == "job.archive" for d in deliveries):
        break
    time.sleep(3)
final = next((d for d in deliveries if d["event"] == "job.done" and d["status"] == "done"), None)
# job.archive carries the zip's size and a download link, like GET /v1/jobs/{id}/archive.
archive_hook = next((d for d in deliveries if d["event"] == "job.archive"), None)
archive_hook_data = (archive_hook or {}).get("data") or {}
hook_zip = requests.get(archive_hook_data["url"], timeout=120).content if isinstance(archive_hook_data.get("url"), str) else None

# 8. What it cost: a run holds `concurrency` threads, billed_seconds is the total in thread-seconds.
run_seconds = done["billed_seconds"] / done["concurrency"]
ended_at = datetime.fromisoformat(done["ended_at"].replace("Z", "+00:00")).timestamp()
depths = sorted({r.get("depth") for r in rows})

print(json.dumps({
    "language": "python",
    "site": SITE,
    "map": map_out,
    "crawl": {**CRAWL, "extract": list(EXTRACT), "start_status": start.status_code},
    "stop": {
        "requested_at_done": (before.get("progress") or {}).get("done", 0),
        "reply": (stop_reply.json() or {}).get("status", stop_reply.status_code),
        "seconds_to_stopped": secs(t_stopped - t_stop), "status": stopped["status"], "progress": stopped.get("progress"),
        "stored_results": len(at_stop),
        "webhooks_by_then": [f"{d['event']} {(json.loads(d['body']).get('data') or {}).get('status')}" for d in hooks_at_stop],
        "archive": archive_at_stop.get("status"),
    },
    "resume": {"status": resume.status_code, "reply": (resume.json() or {}).get("status")},
    "job": {
        "status": done["status"], "runs": done["runs"], "progress": done.get("progress"), "concurrency": done["concurrency"],
        "billed_seconds": done["billed_seconds"],
        "results": {"parts": done["results"]["parts"], "bytes": done["results"]["bytes"]},
        "started_at": done["started_at"], "ended_at": done["ended_at"], "expires_at": done["expires_at"],
    },
    "results_jsonl": {
        "lines": len(rows), "distinct_urls": len(urls), "bytes": len(jsonl.encode("utf-8")),
        "status_200": sum(r.get("status") == 200 for r in rows),
        "statuses": {s: sum(json.dumps(r.get("status")) == s for r in rows) for s in sorted({json.dumps(r.get("status")) for r in rows})},
        "errors": [{"url": r["url"], "status": r.get("status"), "error": str(r["error"])[:120]} for r in rows if r.get("error")],
        "with_markdown": sum(isinstance(r.get("markdown"), str) and len(r["markdown"]) > 0 for r in rows),
        "books_with_price": sum(isinstance((r.get("extract") or {}).get("price"), str) for r in rows),
        "category_pages": sum("/catalogue/category/" in r["url"] for r in rows),
        "depths": {str(d): sum(r.get("depth") == d for r in rows) for d in depths},
        "first_run": len(first_run), "second_run_new": len(urls - first_run),
        "sample": [{"url": r["url"], "title": r["extract"]["title"], "price": r["extract"]["price"]}
                   for r in rows if (r.get("extract") or {}).get("price")][:2],
    },
    "archive": {
        "status": archive["status"], "bytes": archive["bytes"], "downloaded_bytes": len(zip_bytes), "files": len(names),
        "markdown_files": sum(n.endswith("/index.md") for n in names),
        "extract_files": sum(n.endswith("/extract.json") for n in names),
        "index_lines": len(index.strip().split("\n")) if index.strip() else 0,
        "has": [n for n in ["index.jsonl", "errors.jsonl", "job.json"] if n in names],
    },
    "webhook": {
        "deliveries": [{k: v for k, v in d.items() if k not in ("data", "signed_at")} for d in deliveries],
        "final_done": {"data": final["data"], "signed_seconds_after_end": final["signed_at"] - int(ended_at)} if final else None,
        "archive": {
            "status": archive_hook_data.get("status"), "bytes": archive_hook_data.get("bytes"),
            "has_url": isinstance(archive_hook_data.get("url"), str),
            "url_downloads_bytes": len(hook_zip) if hook_zip is not None else None,
            "signed_seconds_after_end": archive_hook["signed_at"] - int(ended_at),
        } if archive_hook else None,
    },
    "cost": {
        "billed_thread_seconds": done["billed_seconds"], "pages": len(rows),
        "run_seconds": round(run_seconds, 1),
        "pages_per_minute": round(len(rows) / (run_seconds / 60), 1),
        "thread_seconds_per_page": round(done["billed_seconds"] / len(rows), 1),
        "client_running_seconds": secs(t_stopped - t_start + t_done - t_resume),
    },
    "measured_at": datetime.now(timezone.utc).date().isoformat(),
}, indent=2, ensure_ascii=False))
```

```java
// Java 21. Maven: com.google.code.gson:gson:2.11.0
// env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
import com.google.gson.*;
import java.net.URI;
import java.net.http.*;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.security.SecureRandom;
import java.time.*;
import java.util.*;
import java.util.function.Predicate;
import java.util.zip.ZipFile;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;

public class Main {
  static final String KEY = System.getenv("CDPFLEET_API_KEY");
  static final String API = "https://cdpfleet.com";
  static final String SITE = "https://books.toscrape.com/"; // a sandbox made for scraping practice
  static final int MAX_DEPTH = 2, MAX_PAGES = 100, CONCURRENCY = 4, DELAY_MS = 250;
  static final int STOP_AT = 30; // stop the job once this many pages are done, then resume it
  static final Set<String> ENDED = Set.of("done", "failed", "stopped", "interrupted");
  static final HttpClient HTTP = HttpClient.newBuilder().followRedirects(HttpClient.Redirect.NORMAL).build();
  static final Gson GSON = new GsonBuilder().serializeNulls().setPrettyPrinting().disableHtmlEscaping().create();

  static HttpResponse<byte[]> send(String method, String url, JsonObject body, boolean withKey) throws Exception {
    HttpRequest.Builder b = HttpRequest.newBuilder(URI.create(url)).timeout(Duration.ofSeconds(120));
    if (withKey) b.header("x-api-key", KEY);
    if (body != null) b.header("content-type", "application/json").method(method, HttpRequest.BodyPublishers.ofString(body.toString()));
    else b.method(method, HttpRequest.BodyPublishers.noBody());
    return HTTP.send(b.build(), HttpResponse.BodyHandlers.ofByteArray());
  }
  static JsonObject json(HttpResponse<byte[]> r) {
    try { return JsonParser.parseString(new String(r.body(), StandardCharsets.UTF_8)).getAsJsonObject(); } catch (Exception e) { return new JsonObject(); }
  }
  static JsonObject api(String method, String path, JsonObject body) throws Exception { return json(send(method, API + path, body, true)); }
  static double num(JsonObject o, String k) { return o != null && o.has(k) && !o.get(k).isJsonNull() ? o.get(k).getAsDouble() : 0; }
  static String str(JsonObject o, String k) { return o != null && o.has(k) && !o.get(k).isJsonNull() ? o.get(k).getAsString() : null; }
  static JsonObject obj(JsonObject o, String k) { return o != null && o.has(k) && o.get(k).isJsonObject() ? o.getAsJsonObject(k) : new JsonObject(); }
  static double round1(double v) { return Math.round(v * 10) / 10.0; }

  // Poll the job every 3 s until `until` holds.
  static JsonObject waitFor(String id, Predicate<JsonObject> until, int seconds) throws Exception {
    JsonObject job = null;
    for (long end = System.currentTimeMillis() + seconds * 1000L; System.currentTimeMillis() < end; Thread.sleep(3000)) {
      job = api("GET", "/v1/jobs/" + id, null);
      if (until.test(job)) return job;
    }
    throw new RuntimeException("job " + id + ": still " + str(job, "status") + " after " + seconds + " s");
  }
  static List<JsonObject> stored(String id) throws Exception { // results stored so far, part by part
    List<JsonObject> rows = new ArrayList<>();
    JsonElement cursor = new JsonPrimitive(0);
    while (!cursor.isJsonNull()) {
      JsonObject r = api("GET", "/v1/jobs/" + id + "/results?cursor=" + cursor.getAsLong(), null);
      for (JsonElement e : r.getAsJsonArray("results")) rows.add(e.getAsJsonObject());
      if (r.get("next_cursor").equals(cursor)) break; // nothing new yet
      cursor = r.get("next_cursor");
    }
    return rows;
  }
  static JsonArray hookItems(String hook) throws Exception { return json(send("GET", hook, null, false)).getAsJsonArray("items"); }
  static String hmac(String secret, String data) throws Exception {
    Mac mac = Mac.getInstance("HmacSHA256");
    mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
    return HexFormat.of().formatHex(mac.doFinal(data.getBytes(StandardCharsets.UTF_8)));
  }

  public static void main(String[] args) throws Exception {
    // 1. Map: the site's URLs in one instant call (sitemaps + the start page's links).
    long t = System.currentTimeMillis();
    JsonObject mapBody = new JsonObject();
    mapBody.addProperty("url", SITE);
    mapBody.addProperty("max_urls", 5000);
    HttpResponse<byte[]> mapRes = send("POST", API + "/v1/functions/map", mapBody, true);
    JsonObject m = json(mapRes);
    JsonObject mapOut = new JsonObject();
    mapOut.addProperty("status", mapRes.statusCode());
    mapOut.addProperty("seconds", round1((System.currentTimeMillis() - t) / 1000.0));
    int categories = 0;
    if (m.has("urls")) for (JsonElement u : m.getAsJsonArray("urls")) if (u.getAsString().contains("/catalogue/category/")) categories++;
    mapOut.addProperty("urls", m.has("urls") ? m.getAsJsonArray("urls").size() : 0);
    mapOut.add("sources", m.get("sources"));
    mapOut.addProperty("partial", mapRes.headers().firstValue("x-cdpfleet-partial").orElse(null));
    mapOut.addProperty("categories", categories);

    // 2. Start the crawl as a job: a zip archive at the end, and webhooks when it's done and when the zip is ready.
    // The receiver here is cdpfleet.com's own test inbox (keeps deliveries an hour, read back with GET);
    // use your own public https endpoint.
    byte[] token = new byte[18];
    new SecureRandom().nextBytes(token);
    String hook = API + "/e2e/hook/" + Base64.getUrlEncoder().withoutPadding().encodeToString(token);
    JsonObject crawl = new JsonObject();
    crawl.addProperty("max_depth", MAX_DEPTH);
    crawl.addProperty("max_pages", MAX_PAGES);
    crawl.addProperty("concurrency", CONCURRENCY);
    crawl.addProperty("delay_ms", DELAY_MS);
    crawl.add("formats", GSON.toJsonTree(List.of("markdown")));
    JsonObject body = crawl.deepCopy();
    body.addProperty("url", SITE);
    body.add("extract", GSON.toJsonTree(Map.of("title", "h1", "price", ".product_main .price_color")));
    body.add("deliver", JsonParser.parseString("{\"webhook\":{\"url\":\"" + hook + "\",\"events\":[\"done\",\"archive\"]},\"format\":\"archive\"}"));
    long tStart = System.currentTimeMillis();
    HttpResponse<byte[]> startRes = send("POST", API + "/v1/functions/crawl", body, true);
    if (startRes.statusCode() != 202) throw new RuntimeException("crawl " + startRes.statusCode() + " " + new String(startRes.body()));
    String id = str(json(startRes), "job_id");

    // 3. Watch it, and stop it part-way.
    JsonObject before = waitFor(id, j -> num(obj(j, "progress"), "done") >= STOP_AT || ENDED.contains(str(j, "status")), 600);
    JsonObject stopReply = api("DELETE", "/v1/jobs/" + id, null);
    long tStop = System.currentTimeMillis();
    JsonObject stopped = waitFor(id, j -> ENDED.contains(str(j, "status")), 300);
    long tStopped = System.currentTimeMillis();
    List<JsonObject> atStop = stored(id);
    JsonArray hooksAtStop = new JsonArray();
    for (JsonElement d : hookItems(hook)) {
      JsonObject b = JsonParser.parseString(str(d.getAsJsonObject(), "body")).getAsJsonObject();
      hooksAtStop.add(str(d.getAsJsonObject(), "event") + " " + str(obj(b, "data"), "status"));
    }
    JsonObject archiveAtStop = api("GET", "/v1/jobs/" + id + "/archive", null);

    // 4. Resume: a new run of the same job, from its checkpoint.
    long tResume = System.currentTimeMillis();
    HttpResponse<byte[]> resumeRes = send("POST", API + "/v1/jobs/" + id + "/resume", null, true);
    // 5. Wait for the second run to end.
    JsonObject done = waitFor(id, j -> num(j, "runs") >= 2 && ENDED.contains(str(j, "status")), 900);
    long tDone = System.currentTimeMillis();

    // 6a. Every result as one JSON-lines file.
    byte[] jsonl = send("GET", API + "/v1/jobs/" + id + "/results.jsonl", null, true).body();
    List<JsonObject> rows = new ArrayList<>();
    for (String line : new String(jsonl, StandardCharsets.UTF_8).strip().split("\n")) rows.add(JsonParser.parseString(line).getAsJsonObject());
    Set<String> urls = new LinkedHashSet<>(), firstRun = new HashSet<>();
    for (JsonObject r : rows) urls.add(str(r, "url"));
    for (JsonObject r : atStop) firstRun.add(str(r, "url"));
    int status200 = 0, withMarkdown = 0, booksWithPrice = 0, categoryPages = 0;
    JsonArray errors = new JsonArray(), sample = new JsonArray();
    TreeMap<String, Integer> depthCount = new TreeMap<>(), statuses = new TreeMap<>();
    for (JsonObject r : rows) {
      if (num(r, "status") == 200) status200++;
      statuses.merge(r.has("status") && !r.get("status").isJsonNull() ? String.valueOf(r.get("status").getAsInt()) : "null", 1, Integer::sum);
      if (r.has("error") && !r.get("error").isJsonNull()) {
        JsonObject e = new JsonObject();
        e.addProperty("url", str(r, "url"));
        e.add("status", r.has("status") ? r.get("status") : JsonNull.INSTANCE);
        String msg = r.get("error").isJsonPrimitive() ? r.get("error").getAsString() : r.get("error").toString();
        e.addProperty("error", msg.substring(0, Math.min(120, msg.length())));
        errors.add(e);
      }
      String md = str(r, "markdown");
      if (md != null && !md.isEmpty()) withMarkdown++;
      String price = str(obj(r, "extract"), "price");
      if (price != null) {
        booksWithPrice++;
        if (sample.size() < 2) {
          JsonObject s = new JsonObject();
          s.addProperty("url", str(r, "url"));
          s.addProperty("title", str(obj(r, "extract"), "title"));
          s.addProperty("price", price);
          sample.add(s);
        }
      }
      if (str(r, "url").contains("/catalogue/category/")) categoryPages++;
      depthCount.merge(r.has("depth") ? String.valueOf(r.get("depth").getAsInt()) : "null", 1, Integer::sum);
    }
    long secondNew = urls.stream().filter(u -> !firstRun.contains(u)).count();

    // 6b. The same results as a zip (built when the job ended).
    waitFor(id, j -> Set.of("ready", "failed").contains(String.valueOf(str(obj(j, "archive"), "status"))), 300);
    JsonObject archive = api("GET", "/v1/jobs/" + id + "/archive", null);
    byte[] zipBytes = send("GET", str(archive, "url"), null, false).body(); // a storage link, no key
    Path tmp = Files.createTempFile("crawl", ".zip");
    Files.write(tmp, zipBytes);
    List<String> names = new ArrayList<>();
    int indexLines = 0;
    try (ZipFile zf = new ZipFile(tmp.toFile())) {
      zf.stream().forEach(e -> names.add(e.getName()));
      var entry = zf.getEntry("index.jsonl");
      if (entry != null) {
        String text = new String(zf.getInputStream(entry).readAllBytes(), StandardCharsets.UTF_8).strip();
        indexLines = text.isEmpty() ? 0 : text.split("\n").length;
      }
    }
    Files.delete(tmp);

    // 7. The webhooks: every delivery is signed with the account's secret, as HMAC-SHA256(secret, "<t>.<raw body>").
    String secret = str(api("GET", "/v1/me/webhook-secret", null), "secret");
    JsonArray deliveries = new JsonArray();
    JsonObject finalData = null, archiveData = null;
    long finalSignedAt = 0, archiveSignedAt = 0;
    for (long end = System.currentTimeMillis() + 90000; System.currentTimeMillis() < end && (finalData == null || archiveData == null);) {
      deliveries = new JsonArray();
      finalData = null;
      archiveData = null;
      for (JsonElement el : hookItems(hook)) {
        JsonObject d = el.getAsJsonObject();
        Map<String, String> parts = new HashMap<>();
        for (String kv : String.valueOf(str(d, "signature")).split(",")) { String[] p = kv.split("=", 2); if (p.length == 2) parts.put(p[0], p[1]); }
        String raw = str(d, "body");
        JsonObject b = JsonParser.parseString(raw).getAsJsonObject();
        String want = hmac(secret, parts.get("t") + "." + raw);
        JsonObject o = new JsonObject();
        o.addProperty("event", str(d, "event"));
        o.addProperty("status", str(obj(b, "data"), "status"));
        o.addProperty("job_id_matches", id.equals(str(b, "job_id")));
        o.addProperty("delivery_id_is_uuid", String.valueOf(str(d, "delivery")).matches("[0-9a-f-]{36}"));
        o.addProperty("signature_valid", parts.containsKey("v1") && java.security.MessageDigest.isEqual(parts.get("v1").getBytes(), want.getBytes()));
        deliveries.add(o);
        if (finalData == null && "job.done".equals(str(d, "event")) && "done".equals(str(obj(b, "data"), "status"))) {
          finalData = obj(b, "data");
          finalSignedAt = Long.parseLong(parts.get("t"));
        }
        if (archiveData == null && "job.archive".equals(str(d, "event"))) {
          archiveData = obj(b, "data");
          archiveSignedAt = Long.parseLong(parts.get("t"));
        }
      }
      if (finalData == null || archiveData == null) Thread.sleep(3000);
    }
    JsonElement finalOut = JsonNull.INSTANCE;
    if (finalData != null) {
      JsonObject f = new JsonObject();
      f.add("data", finalData);
      f.addProperty("signed_seconds_after_end", finalSignedAt - Instant.parse(str(done, "ended_at")).getEpochSecond());
      finalOut = f;
    }
    // job.archive carries the zip's size and a download link, like GET /v1/jobs/{id}/archive.
    JsonElement archiveHookOut = JsonNull.INSTANCE;
    if (archiveData != null) {
      JsonObject a = new JsonObject();
      String hookZipUrl = str(archiveData, "url");
      a.addProperty("status", str(archiveData, "status"));
      a.add("bytes", archiveData.has("bytes") ? archiveData.get("bytes") : JsonNull.INSTANCE);
      a.addProperty("has_url", hookZipUrl != null);
      a.addProperty("url_downloads_bytes", hookZipUrl != null ? send("GET", hookZipUrl, null, false).body().length : null);
      a.addProperty("signed_seconds_after_end", archiveSignedAt - Instant.parse(str(done, "ended_at")).getEpochSecond());
      archiveHookOut = a;
    }

    // 8. What it cost: a run holds `concurrency` threads, billed_seconds is the total in thread-seconds.
    double billed = num(done, "billed_seconds");
    double runSeconds = billed / num(done, "concurrency");

    JsonObject out = new JsonObject();
    out.addProperty("language", "java");
    out.addProperty("site", SITE);
    out.add("map", mapOut);
    JsonObject crawlOut = crawl.deepCopy();
    crawlOut.add("extract", GSON.toJsonTree(List.of("title", "price")));
    crawlOut.addProperty("start_status", startRes.statusCode());
    out.add("crawl", crawlOut);
    JsonObject stop = new JsonObject();
    stop.addProperty("requested_at_done", (int) num(obj(before, "progress"), "done"));
    stop.addProperty("reply", str(stopReply, "status"));
    stop.addProperty("seconds_to_stopped", round1((tStopped - tStop) / 1000.0));
    stop.addProperty("status", str(stopped, "status"));
    stop.add("progress", stopped.get("progress"));
    stop.addProperty("stored_results", atStop.size());
    stop.add("webhooks_by_then", hooksAtStop);
    stop.addProperty("archive", str(archiveAtStop, "status"));
    out.add("stop", stop);
    JsonObject resume = new JsonObject();
    resume.addProperty("status", resumeRes.statusCode());
    resume.addProperty("reply", str(json(resumeRes), "status"));
    out.add("resume", resume);
    JsonObject job = new JsonObject();
    for (String k : List.of("status", "runs", "progress", "concurrency", "billed_seconds")) job.add(k, done.get(k));
    JsonObject res = new JsonObject();
    res.add("parts", obj(done, "results").get("parts"));
    res.add("bytes", obj(done, "results").get("bytes"));
    job.add("results", res);
    for (String k : List.of("started_at", "ended_at", "expires_at")) job.add(k, done.get(k));
    out.add("job", job);
    JsonObject rj = new JsonObject();
    rj.addProperty("lines", rows.size());
    rj.addProperty("distinct_urls", urls.size());
    rj.addProperty("bytes", jsonl.length);
    rj.addProperty("status_200", status200);
    rj.add("statuses", GSON.toJsonTree(statuses));
    rj.add("errors", errors);
    rj.addProperty("with_markdown", withMarkdown);
    rj.addProperty("books_with_price", booksWithPrice);
    rj.addProperty("category_pages", categoryPages);
    rj.add("depths", GSON.toJsonTree(depthCount));
    rj.addProperty("first_run", firstRun.size());
    rj.addProperty("second_run_new", secondNew);
    rj.add("sample", sample);
    out.add("results_jsonl", rj);
    JsonObject ar = new JsonObject();
    ar.addProperty("status", str(archive, "status"));
    ar.add("bytes", archive.get("bytes"));
    ar.addProperty("downloaded_bytes", zipBytes.length);
    ar.addProperty("files", names.size());
    ar.addProperty("markdown_files", names.stream().filter(x -> x.endsWith("/index.md")).count());
    ar.addProperty("extract_files", names.stream().filter(x -> x.endsWith("/extract.json")).count());
    ar.addProperty("index_lines", indexLines);
    ar.add("has", GSON.toJsonTree(List.of("index.jsonl", "errors.jsonl", "job.json").stream().filter(names::contains).toList()));
    out.add("archive", ar);
    JsonObject wh = new JsonObject();
    wh.add("deliveries", deliveries);
    wh.add("final_done", finalOut);
    wh.add("archive", archiveHookOut);
    out.add("webhook", wh);
    JsonObject cost = new JsonObject();
    cost.addProperty("billed_thread_seconds", billed);
    cost.addProperty("pages", rows.size());
    cost.addProperty("run_seconds", round1(runSeconds));
    cost.addProperty("pages_per_minute", round1(rows.size() / (runSeconds / 60)));
    cost.addProperty("thread_seconds_per_page", round1(billed / rows.size()));
    cost.addProperty("client_running_seconds", round1((tStopped - tStart + tDone - tResume) / 1000.0));
    out.add("cost", cost);
    out.addProperty("measured_at", LocalDate.now(ZoneOffset.UTC).toString());
    System.out.println(GSON.toJson(out));
  }
}
```

```csharp
// .NET 9, no packages. env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
using System.IO.Compression;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using System.Text.RegularExpressions;

var key = Environment.GetEnvironmentVariable("CDPFLEET_API_KEY")!;
const string Api = "https://cdpfleet.com";
const string Site = "https://books.toscrape.com/"; // a sandbox made for scraping practice
const int StopAt = 30; // stop the job once this many pages are done, then resume it
var crawl = new JsonObject { ["max_depth"] = 2, ["max_pages"] = 100, ["concurrency"] = 4, ["delay_ms"] = 250, ["formats"] = new JsonArray("markdown") };
var ended = new HashSet<string> { "done", "failed", "stopped", "interrupted" };
var http = new HttpClient { Timeout = TimeSpan.FromSeconds(120) };

async Task<HttpResponseMessage> Send(HttpMethod method, string url, JsonNode? body = null, bool withKey = true)
{
    var req = new HttpRequestMessage(method, url);
    if (withKey) req.Headers.Add("x-api-key", key);
    if (body != null) req.Content = new StringContent(body.ToJsonString(), Encoding.UTF8, "application/json");
    return await http.SendAsync(req);
}
async Task<JsonObject> Json(HttpResponseMessage r)
{
    try { return JsonNode.Parse(await r.Content.ReadAsStringAsync())!.AsObject(); } catch { return new JsonObject(); }
}
async Task<JsonObject> CallApi(HttpMethod method, string path, JsonNode? body = null) => await Json(await Send(method, Api + path, body));
double Num(JsonNode? n) => n is JsonValue v && v.TryGetValue<double>(out var d) ? d : 0;
string? Str(JsonNode? n) => n is JsonValue v && v.TryGetValue<string>(out var s) ? s : null;
double Round1(double v) => Math.Round(v, 1, MidpointRounding.AwayFromZero);
JsonNode? Copy(JsonNode? n) => n?.DeepClone();

// Poll the job every 3 s until `until` holds.
async Task<JsonObject> WaitFor(string id, Func<JsonObject, bool> until, int seconds)
{
    JsonObject? job = null;
    for (var end = DateTime.UtcNow.AddSeconds(seconds); DateTime.UtcNow < end; await Task.Delay(3000))
    {
        job = await CallApi(HttpMethod.Get, $"/v1/jobs/{id}");
        if (until(job)) return job;
    }
    throw new Exception($"job {id}: still {Str(job?["status"])} after {seconds} s");
}
async Task<List<JsonObject>> Stored(string id) // results stored so far, part by part
{
    var rows = new List<JsonObject>();
    JsonNode? cursor = 0;
    while (cursor != null)
    {
        var r = await CallApi(HttpMethod.Get, $"/v1/jobs/{id}/results?cursor={cursor.ToJsonString()}");
        rows.AddRange(r["results"]!.AsArray().Select(x => x!.AsObject()));
        if (r["next_cursor"]?.ToJsonString() == cursor.ToJsonString()) break; // nothing new yet
        cursor = Copy(r["next_cursor"]);
    }
    return rows;
}
async Task<JsonArray> HookItems(string hook) => (await Json(await Send(HttpMethod.Get, hook, withKey: false)))["items"]!.AsArray();

// 1. Map: the site's URLs in one instant call (sitemaps + the start page's links).
var t = DateTime.UtcNow;
var mapRes = await Send(HttpMethod.Post, $"{Api}/v1/functions/map", new JsonObject { ["url"] = Site, ["max_urls"] = 5000 });
var m = await Json(mapRes);
var mapUrls = m["urls"]?.AsArray().Select(u => (string)u!).ToList() ?? new List<string>();
var mapOut = new JsonObject
{
    ["status"] = (int)mapRes.StatusCode, ["seconds"] = Round1((DateTime.UtcNow - t).TotalSeconds), ["urls"] = mapUrls.Count,
    ["sources"] = Copy(m["sources"]), ["partial"] = mapRes.Headers.TryGetValues("x-cdpfleet-partial", out var p) ? p.First() : null,
    ["categories"] = mapUrls.Count(u => u.Contains("/catalogue/category/")),
};

// 2. Start the crawl as a job: a zip archive at the end, and webhooks when it's done and when the zip is ready.
// The receiver here is cdpfleet.com's own test inbox (keeps deliveries an hour, read back with GET);
// use your own public https endpoint.
var hook = $"{Api}/e2e/hook/{Convert.ToBase64String(RandomNumberGenerator.GetBytes(18)).Replace('+', '-').Replace('/', '_').TrimEnd('=')}";
var body = crawl.DeepClone().AsObject();
body["url"] = Site;
body["extract"] = new JsonObject { ["title"] = "h1", ["price"] = ".product_main .price_color" };
body["deliver"] = new JsonObject { ["webhook"] = new JsonObject { ["url"] = hook, ["events"] = new JsonArray("done", "archive") }, ["format"] = "archive" };
var tStart = DateTime.UtcNow;
var startRes = await Send(HttpMethod.Post, $"{Api}/v1/functions/crawl", body);
if ((int)startRes.StatusCode != 202) throw new Exception($"crawl {(int)startRes.StatusCode} {await startRes.Content.ReadAsStringAsync()}");
var id = (string)(await Json(startRes))["job_id"]!;

// 3. Watch it, and stop it part-way.
var before = await WaitFor(id, j => Num(j["progress"]?["done"]) >= StopAt || ended.Contains(Str(j["status"]) ?? ""), 600);
var stopReply = await CallApi(HttpMethod.Delete, $"/v1/jobs/{id}");
var tStop = DateTime.UtcNow;
var stopped = await WaitFor(id, j => ended.Contains(Str(j["status"]) ?? ""), 300);
var tStopped = DateTime.UtcNow;
var atStop = await Stored(id);
var hooksAtStop = new JsonArray((await HookItems(hook)).Select(d => (JsonNode?)$"{Str(d!["event"])} {Str(JsonNode.Parse(Str(d["body"])!)!["data"]?["status"])}").ToArray());
var archiveAtStop = await CallApi(HttpMethod.Get, $"/v1/jobs/{id}/archive");

// 4. Resume: a new run of the same job, from its checkpoint.
var tResume = DateTime.UtcNow;
var resumeRes = await Send(HttpMethod.Post, $"{Api}/v1/jobs/{id}/resume");
var resume = await Json(resumeRes);
// 5. Wait for the second run to end.
var done = await WaitFor(id, j => Num(j["runs"]) >= 2 && ended.Contains(Str(j["status"]) ?? ""), 900);
var tDone = DateTime.UtcNow;

// 6a. Every result as one JSON-lines file.
var jsonl = await (await Send(HttpMethod.Get, $"{Api}/v1/jobs/{id}/results.jsonl")).Content.ReadAsByteArrayAsync();
var rows = Encoding.UTF8.GetString(jsonl).Trim().Split('\n').Select(l => JsonNode.Parse(l)!.AsObject()).ToList();
var urls = rows.Select(r => Str(r["url"])!).ToHashSet();
var firstRun = atStop.Select(r => Str(r["url"])!).ToHashSet();
var depths = new JsonObject();
foreach (var g in rows.GroupBy(r => r["depth"]?.ToJsonString() ?? "null").OrderBy(g => g.Key)) depths[g.Key] = g.Count();

// 6b. The same results as a zip (built when the job ended).
await WaitFor(id, j => Str(j["archive"]?["status"]) is "ready" or "failed", 300);
var archive = await CallApi(HttpMethod.Get, $"/v1/jobs/{id}/archive");
var zipBytes = await (await Send(HttpMethod.Get, Str(archive["url"])!, withKey: false)).Content.ReadAsByteArrayAsync(); // a storage link, no key
List<string> names;
var indexLines = 0;
using (var zip = new ZipArchive(new MemoryStream(zipBytes), ZipArchiveMode.Read))
{
    names = zip.Entries.Select(e => e.FullName).ToList();
    var entry = zip.GetEntry("index.jsonl");
    if (entry != null)
    {
        using var reader = new StreamReader(entry.Open(), Encoding.UTF8);
        var text = (await reader.ReadToEndAsync()).Trim();
        indexLines = text.Length == 0 ? 0 : text.Split('\n').Length;
    }
}

// 7. The webhooks: every delivery is signed with the account's secret, as HMAC-SHA256(secret, "<t>.<raw body>").
var secret = Str((await CallApi(HttpMethod.Get, "/v1/me/webhook-secret"))["secret"])!;
var deliveries = new JsonArray();
JsonNode? finalData = null, archiveData = null;
long finalSignedAt = 0, archiveSignedAt = 0;
for (var end = DateTime.UtcNow.AddSeconds(90); DateTime.UtcNow < end && (finalData == null || archiveData == null);)
{
    deliveries = new JsonArray();
    finalData = null;
    archiveData = null;
    foreach (var d in await HookItems(hook))
    {
        var parts = (Str(d!["signature"]) ?? "").Split(',').Select(kv => kv.Split('=', 2)).Where(a => a.Length == 2).ToDictionary(a => a[0], a => a[1]);
        var raw = Str(d["body"])!;
        var b = JsonNode.Parse(raw)!;
        var want = Convert.ToHexStringLower(HMACSHA256.HashData(Encoding.UTF8.GetBytes(secret), Encoding.UTF8.GetBytes($"{parts.GetValueOrDefault("t")}.{raw}")));
        deliveries.Add(new JsonObject
        {
            ["event"] = Str(d["event"]), ["status"] = Str(b["data"]?["status"]), ["job_id_matches"] = Str(b["job_id"]) == id,
            ["delivery_id_is_uuid"] = Regex.IsMatch(Str(d["delivery"]) ?? "", "^[0-9a-f-]{36}$"),
            ["signature_valid"] = parts.TryGetValue("v1", out var v1) && CryptographicOperations.FixedTimeEquals(Encoding.ASCII.GetBytes(v1), Encoding.ASCII.GetBytes(want)),
        });
        if (finalData == null && Str(d["event"]) == "job.done" && Str(b["data"]?["status"]) == "done")
        {
            finalData = Copy(b["data"]);
            finalSignedAt = long.Parse(parts["t"]);
        }
        if (archiveData == null && Str(d["event"]) == "job.archive")
        {
            archiveData = Copy(b["data"]);
            archiveSignedAt = long.Parse(parts["t"]);
        }
    }
    if (finalData == null || archiveData == null) await Task.Delay(3000);
}
// job.archive carries the zip's size and a download link, like GET /v1/jobs/{id}/archive.
var hookZipUrl = Str(archiveData?["url"]);
int? hookZipBytes = hookZipUrl == null ? null : (await (await Send(HttpMethod.Get, hookZipUrl, withKey: false)).Content.ReadAsByteArrayAsync()).Length;

// 8. What it cost: a run holds `concurrency` threads, billed_seconds is the total in thread-seconds.
var billed = Num(done["billed_seconds"]);
var runSeconds = billed / Num(done["concurrency"]);
var crawlOut = crawl.DeepClone().AsObject();
crawlOut["extract"] = new JsonArray("title", "price");
crawlOut["start_status"] = (int)startRes.StatusCode;

Console.WriteLine(new JsonObject
{
    ["language"] = "csharp",
    ["site"] = Site,
    ["map"] = mapOut,
    ["crawl"] = crawlOut,
    ["stop"] = new JsonObject
    {
        ["requested_at_done"] = Num(before["progress"]?["done"]), ["reply"] = Str(stopReply["status"]),
        ["seconds_to_stopped"] = Round1((tStopped - tStop).TotalSeconds), ["status"] = Str(stopped["status"]), ["progress"] = Copy(stopped["progress"]),
        ["stored_results"] = atStop.Count, ["webhooks_by_then"] = hooksAtStop, ["archive"] = Str(archiveAtStop["status"]),
    },
    ["resume"] = new JsonObject { ["status"] = (int)resumeRes.StatusCode, ["reply"] = Str(resume["status"]) },
    ["job"] = new JsonObject
    {
        ["status"] = Copy(done["status"]), ["runs"] = Copy(done["runs"]), ["progress"] = Copy(done["progress"]), ["concurrency"] = Copy(done["concurrency"]),
        ["billed_seconds"] = Copy(done["billed_seconds"]), ["results"] = new JsonObject { ["parts"] = Copy(done["results"]?["parts"]), ["bytes"] = Copy(done["results"]?["bytes"]) },
        ["started_at"] = Copy(done["started_at"]), ["ended_at"] = Copy(done["ended_at"]), ["expires_at"] = Copy(done["expires_at"]),
    },
    ["results_jsonl"] = new JsonObject
    {
        ["lines"] = rows.Count, ["distinct_urls"] = urls.Count, ["bytes"] = jsonl.Length,
        ["status_200"] = rows.Count(r => Num(r["status"]) == 200),
        ["statuses"] = new JsonObject(rows.GroupBy(r => r["status"]?.ToJsonString() ?? "null").OrderBy(g => g.Key, StringComparer.Ordinal).Select(g => KeyValuePair.Create(g.Key, (JsonNode?)g.Count()))),
        ["errors"] = new JsonArray(rows.Where(r => r["error"] != null).Select(r =>
        {
            var msg = r["error"] is JsonValue ev && ev.TryGetValue<string>(out var es) ? es : r["error"]!.ToJsonString();
            return (JsonNode?)new JsonObject { ["url"] = Str(r["url"]), ["status"] = Copy(r["status"]), ["error"] = msg[..Math.Min(120, msg.Length)] };
        }).ToArray()),
        ["with_markdown"] = rows.Count(r => !string.IsNullOrEmpty(Str(r["markdown"]))),
        ["books_with_price"] = rows.Count(r => Str(r["extract"]?["price"]) != null),
        ["category_pages"] = rows.Count(r => Str(r["url"])!.Contains("/catalogue/category/")),
        ["depths"] = depths,
        ["first_run"] = firstRun.Count, ["second_run_new"] = urls.Count(u => !firstRun.Contains(u)),
        ["sample"] = new JsonArray(rows.Where(r => Str(r["extract"]?["price"]) != null).Take(2)
            .Select(r => (JsonNode?)new JsonObject { ["url"] = Str(r["url"]), ["title"] = Str(r["extract"]!["title"]), ["price"] = Str(r["extract"]!["price"]) }).ToArray()),
    },
    ["archive"] = new JsonObject
    {
        ["status"] = Str(archive["status"]), ["bytes"] = Copy(archive["bytes"]), ["downloaded_bytes"] = zipBytes.Length, ["files"] = names.Count,
        ["markdown_files"] = names.Count(x => x.EndsWith("/index.md")), ["extract_files"] = names.Count(x => x.EndsWith("/extract.json")),
        ["index_lines"] = indexLines,
        ["has"] = new JsonArray(new[] { "index.jsonl", "errors.jsonl", "job.json" }.Where(names.Contains).Select(x => (JsonNode?)x).ToArray()),
    },
    ["webhook"] = new JsonObject
    {
        ["deliveries"] = deliveries,
        ["final_done"] = finalData == null ? null : new JsonObject
        {
            ["data"] = finalData, ["signed_seconds_after_end"] = finalSignedAt - DateTimeOffset.Parse(Str(done["ended_at"])!).ToUnixTimeSeconds(),
        },
        ["archive"] = archiveData == null ? null : new JsonObject
        {
            ["status"] = Str(archiveData["status"]), ["bytes"] = Copy(archiveData["bytes"]), ["has_url"] = hookZipUrl != null,
            ["url_downloads_bytes"] = hookZipBytes,
            ["signed_seconds_after_end"] = archiveSignedAt - DateTimeOffset.Parse(Str(done["ended_at"])!).ToUnixTimeSeconds(),
        },
    },
    ["cost"] = new JsonObject
    {
        ["billed_thread_seconds"] = billed, ["pages"] = rows.Count, ["run_seconds"] = Round1(runSeconds),
        ["pages_per_minute"] = Round1(rows.Count / (runSeconds / 60)),
        ["thread_seconds_per_page"] = Round1(billed / rows.Count),
        ["client_running_seconds"] = Round1((tStopped - tStart + (tDone - tResume)).TotalSeconds),
    },
    ["measured_at"] = DateTime.UtcNow.ToString("yyyy-MM-dd"),
}.ToJsonString(new JsonSerializerOptions { WriteIndented = true, Encoder = System.Text.Encodings.Web.JavaScriptEncoder.UnsafeRelaxedJsonEscaping }));
```

```go
// Go 1.22, standard library only. env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
package main

import (
	"archive/zip"
	"bytes"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"os"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"time"
)

const api = "https://cdpfleet.com"
const site = "https://books.toscrape.com/" // a sandbox made for scraping practice
const stopAt = 30                          // stop the job once this many pages are done, then resume it

var key = os.Getenv("CDPFLEET_API_KEY")
var crawl = map[string]any{"max_depth": 2, "max_pages": 100, "concurrency": 4, "delay_ms": 250, "formats": []string{"markdown"}}
var extract = map[string]string{"title": "h1", "price": ".product_main .price_color"}
var ended = map[string]bool{"done": true, "failed": true, "stopped": true, "interrupted": true}
var client = &http.Client{Timeout: 120 * time.Second}

type M = map[string]any

func must[T any](v T, err error) T {
	if err != nil {
		log.Fatal(err)
	}
	return v
}

func round1(f float64) float64 { return math.Round(f*10) / 10 }

// call sends a request with the API key (when withKey) and returns the status, headers and body.
func call(method, url string, body any, withKey bool) (int, http.Header, []byte) {
	var rd io.Reader
	if body != nil {
		rd = bytes.NewReader(must(json.Marshal(body)))
	}
	req := must(http.NewRequest(method, url, rd))
	if withKey {
		req.Header.Set("x-api-key", key)
	}
	if body != nil {
		req.Header.Set("content-type", "application/json")
	}
	res := must(client.Do(req))
	defer res.Body.Close()
	return res.StatusCode, res.Header, must(io.ReadAll(res.Body))
}

func apiJSON(method, path string, body any) (int, http.Header, M) {
	status, h, b := call(method, api+path, body, true)
	var m M
	_ = json.Unmarshal(b, &m)
	return status, h, m
}

func num(v any) float64 {
	f, _ := v.(float64)
	return f
}

func str(v any) string {
	s, _ := v.(string)
	return s
}

func obj(v any) M {
	m, _ := v.(M)
	return m
}

// waitFor polls the job every 3 s until until(job) holds.
func waitFor(id string, until func(M) bool, d time.Duration) M {
	var job M
	for end := time.Now().Add(d); time.Now().Before(end); time.Sleep(3 * time.Second) {
		_, _, job = apiJSON("GET", "/v1/jobs/"+id, nil)
		if job != nil && until(job) {
			return job
		}
	}
	log.Fatalf("job %s: still %v after %v", id, job["status"], d)
	return nil
}

// stored returns the results stored so far, part by part.
func stored(id string) []M {
	var rows []M
	cursor := any(0.0)
	for cursor != nil {
		_, _, r := apiJSON("GET", fmt.Sprintf("/v1/jobs/%s/results?cursor=%v", id, cursor), nil)
		for _, x := range r["results"].([]any) {
			rows = append(rows, x.(M))
		}
		if r["next_cursor"] == cursor { // nothing new yet
			break
		}
		cursor = r["next_cursor"]
	}
	return rows
}

func hookItems(hook string) []M {
	_, _, b := call("GET", hook, nil, false)
	var r struct{ Items []M }
	must(0, json.Unmarshal(b, &r))
	return r.Items
}

func main() {
	// 1. Map: the site's URLs in one instant call (sitemaps + the start page's links).
	t := time.Now()
	mapStatus, mapHeaders, m := apiJSON("POST", "/v1/functions/map", M{"url": site, "max_urls": 5000})
	urlsFound, categories := 0, 0
	if list, ok := m["urls"].([]any); ok {
		urlsFound = len(list)
		for _, u := range list {
			if strings.Contains(str(u), "/catalogue/category/") {
				categories++
			}
		}
	}
	var partial any
	if p := mapHeaders.Get("x-cdpfleet-partial"); p != "" {
		partial = p
	}
	mapOut := M{"status": mapStatus, "seconds": round1(time.Since(t).Seconds()), "urls": urlsFound, "sources": m["sources"], "partial": partial, "categories": categories}

	// 2. Start the crawl as a job: a zip archive at the end, and webhooks when it's done and when the zip is ready.
	// The receiver here is cdpfleet.com's own test inbox (keeps deliveries an hour, read back with GET);
	// use your own public https endpoint.
	token := make([]byte, 18)
	must(rand.Read(token))
	hook := api + "/e2e/hook/" + base64.RawURLEncoding.EncodeToString(token)
	body := M{"url": site, "extract": extract, "deliver": M{"webhook": M{"url": hook, "events": []string{"done", "archive"}}, "format": "archive"}}
	for k, v := range crawl {
		body[k] = v
	}
	tStart := time.Now()
	startStatus, _, start := apiJSON("POST", "/v1/functions/crawl", body)
	if startStatus != 202 {
		log.Fatalf("crawl %d %v", startStatus, start)
	}
	id := str(start["job_id"])

	// 3. Watch it, and stop it part-way.
	before := waitFor(id, func(j M) bool { return num(obj(j["progress"])["done"]) >= stopAt || ended[str(j["status"])] }, 10*time.Minute)
	_, _, stopReply := apiJSON("DELETE", "/v1/jobs/"+id, nil)
	tStop := time.Now()
	stopped := waitFor(id, func(j M) bool { return ended[str(j["status"])] }, 5*time.Minute)
	tStopped := time.Now()
	atStop := stored(id)
	hooksAtStop := []string{}
	for _, d := range hookItems(hook) {
		var b M
		_ = json.Unmarshal([]byte(str(d["body"])), &b)
		hooksAtStop = append(hooksAtStop, fmt.Sprintf("%s %v", str(d["event"]), obj(b["data"])["status"]))
	}
	_, _, archiveAtStop := apiJSON("GET", "/v1/jobs/"+id+"/archive", nil)

	// 4. Resume: a new run of the same job, from its checkpoint.
	tResume := time.Now()
	resumeStatus, _, resume := apiJSON("POST", "/v1/jobs/"+id+"/resume", nil)
	// 5. Wait for the second run to end.
	done := waitFor(id, func(j M) bool { return num(j["runs"]) >= 2 && ended[str(j["status"])] }, 15*time.Minute)
	tDone := time.Now()

	// 6a. Every result as one JSON-lines file.
	_, _, jsonl := call("GET", api+"/v1/jobs/"+id+"/results.jsonl", nil, true)
	var rows []M
	for _, line := range strings.Split(strings.TrimSpace(string(jsonl)), "\n") {
		var r M
		must(0, json.Unmarshal([]byte(line), &r))
		rows = append(rows, r)
	}
	urls, firstRun := map[string]bool{}, map[string]bool{}
	for _, r := range rows {
		urls[str(r["url"])] = true
	}
	for _, r := range atStop {
		firstRun[str(r["url"])] = true
	}
	secondNew := 0
	for u := range urls {
		if !firstRun[u] {
			secondNew++
		}
	}
	status200, withMarkdown, booksWithPrice, categoryPages := 0, 0, 0, 0
	errors, sample := []M{}, []M{}
	depthCount, statuses := map[string]int{}, map[string]int{}
	for _, r := range rows {
		if num(r["status"]) == 200 {
			status200++
		}
		if r["status"] == nil {
			statuses["null"]++
		} else {
			statuses[fmt.Sprint(r["status"])]++
		}
		if e, ok := r["error"]; ok && e != nil {
			msg := fmt.Sprint(e)
			if len(msg) > 120 {
				msg = msg[:120]
			}
			errors = append(errors, M{"url": r["url"], "status": r["status"], "error": msg})
		}
		if md, ok := r["markdown"].(string); ok && md != "" {
			withMarkdown++
		}
		if price, ok := obj(r["extract"])["price"].(string); ok {
			booksWithPrice++
			if len(sample) < 2 {
				sample = append(sample, M{"url": r["url"], "title": obj(r["extract"])["title"], "price": price})
			}
		}
		if strings.Contains(str(r["url"]), "/catalogue/category/") {
			categoryPages++
		}
		depthCount[fmt.Sprint(r["depth"])]++
	}

	// 6b. The same results as a zip (built when the job ended).
	waitFor(id, func(j M) bool { s := str(obj(j["archive"])["status"]); return s == "ready" || s == "failed" }, 5*time.Minute)
	_, _, archive := apiJSON("GET", "/v1/jobs/"+id+"/archive", nil)
	_, _, zipBytes := call("GET", str(archive["url"]), nil, false) // a storage link, no key
	zr := must(zip.NewReader(bytes.NewReader(zipBytes), int64(len(zipBytes))))
	names := map[string]bool{}
	markdownFiles, extractFiles, indexLines := 0, 0, 0
	for _, f := range zr.File {
		names[f.Name] = true
		if strings.HasSuffix(f.Name, "/index.md") {
			markdownFiles++
		}
		if strings.HasSuffix(f.Name, "/extract.json") {
			extractFiles++
		}
		if f.Name == "index.jsonl" {
			rc := must(f.Open())
			text := strings.TrimSpace(string(must(io.ReadAll(rc))))
			rc.Close()
			if text != "" {
				indexLines = len(strings.Split(text, "\n"))
			}
		}
	}
	has := []string{}
	for _, n := range []string{"index.jsonl", "errors.jsonl", "job.json"} {
		if names[n] {
			has = append(has, n)
		}
	}

	// 7. The webhooks: every delivery is signed with the account's secret, as HMAC-SHA256(secret, "<t>.<raw body>").
	_, _, sec := apiJSON("GET", "/v1/me/webhook-secret", nil)
	secret := str(sec["secret"])
	uuid := regexp.MustCompile(`^[0-9a-f-]{36}$`)
	type delivery struct {
		out      M
		signedAt int64
		data     any
	}
	check := func(d M) delivery {
		parts := map[string]string{}
		for _, kv := range strings.Split(str(d["signature"]), ",") {
			if k, v, ok := strings.Cut(kv, "="); ok {
				parts[k] = v
			}
		}
		raw := str(d["body"])
		var b M
		_ = json.Unmarshal([]byte(raw), &b)
		mac := hmac.New(sha256.New, []byte(secret))
		mac.Write([]byte(parts["t"] + "." + raw))
		want := hex.EncodeToString(mac.Sum(nil))
		signedAt, _ := strconv.ParseInt(parts["t"], 10, 64)
		return delivery{
			out: M{
				"event": d["event"], "status": obj(b["data"])["status"], "job_id_matches": str(b["job_id"]) == id,
				"delivery_id_is_uuid": uuid.MatchString(str(d["delivery"])),
				"signature_valid":     parts["v1"] != "" && hmac.Equal([]byte(parts["v1"]), []byte(want)),
			},
			signedAt: signedAt, data: b["data"],
		}
	}
	var deliveries []delivery
	var final, archiveHook *delivery
	for end := time.Now().Add(90 * time.Second); time.Now().Before(end); time.Sleep(3 * time.Second) {
		deliveries = nil
		for _, d := range hookItems(hook) {
			deliveries = append(deliveries, check(d))
		}
		final, archiveHook = nil, nil
		for i := range deliveries {
			if deliveries[i].out["event"] == "job.done" && deliveries[i].out["status"] == "done" {
				final = &deliveries[i]
			}
			if deliveries[i].out["event"] == "job.archive" {
				archiveHook = &deliveries[i]
			}
		}
		if final != nil && archiveHook != nil {
			break
		}
	}
	deliveryOut := []M{}
	for _, d := range deliveries {
		deliveryOut = append(deliveryOut, d.out)
	}
	endedAt := must(time.Parse(time.RFC3339Nano, str(done["ended_at"])))
	var finalOut any
	if final != nil {
		finalOut = M{"data": final.data, "signed_seconds_after_end": final.signedAt - endedAt.Unix()}
	}
	// job.archive carries the zip's size and a download link, like GET /v1/jobs/{id}/archive.
	var archiveHookOut any
	if archiveHook != nil {
		data := obj(archiveHook.data)
		url, hasURL := data["url"].(string)
		var downloaded any
		if hasURL {
			_, _, b := call("GET", url, nil, false)
			downloaded = len(b)
		}
		archiveHookOut = M{
			"status": data["status"], "bytes": data["bytes"], "has_url": hasURL, "url_downloads_bytes": downloaded,
			"signed_seconds_after_end": archiveHook.signedAt - endedAt.Unix(),
		}
	}

	// 8. What it cost: a run holds `concurrency` threads, billed_seconds is the total in thread-seconds.
	billed := num(done["billed_seconds"])
	runSeconds := billed / num(done["concurrency"])
	pages := float64(len(rows))
	depths := M{}
	keys := make([]string, 0, len(depthCount))
	for k := range depthCount {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		depths[k] = depthCount[k]
	}
	results := obj(done["results"])
	stopReplyStatus := stopReply["status"]

	out := M{
		"language": "go",
		"site":     site,
		"map":      mapOut,
		"crawl":    M{"max_depth": 2, "max_pages": 100, "concurrency": 4, "delay_ms": 250, "formats": crawl["formats"], "extract": []string{"title", "price"}, "start_status": startStatus},
		"stop": M{
			"requested_at_done": num(obj(before["progress"])["done"]), "reply": stopReplyStatus,
			"seconds_to_stopped": round1(tStopped.Sub(tStop).Seconds()), "status": stopped["status"], "progress": stopped["progress"],
			"stored_results": len(atStop), "webhooks_by_then": hooksAtStop, "archive": archiveAtStop["status"],
		},
		"resume": M{"status": resumeStatus, "reply": resume["status"]},
		"job": M{
			"status": done["status"], "runs": done["runs"], "progress": done["progress"], "concurrency": done["concurrency"],
			"billed_seconds": done["billed_seconds"], "results": M{"parts": results["parts"], "bytes": results["bytes"]},
			"started_at": done["started_at"], "ended_at": done["ended_at"], "expires_at": done["expires_at"],
		},
		"results_jsonl": M{
			"lines": len(rows), "distinct_urls": len(urls), "bytes": len(jsonl),
			"status_200": status200, "statuses": statuses, "errors": errors, "with_markdown": withMarkdown, "books_with_price": booksWithPrice,
			"category_pages": categoryPages, "depths": depths, "first_run": len(firstRun), "second_run_new": secondNew, "sample": sample,
		},
		"archive": M{
			"status": archive["status"], "bytes": archive["bytes"], "downloaded_bytes": len(zipBytes), "files": len(zr.File),
			"markdown_files": markdownFiles, "extract_files": extractFiles, "index_lines": indexLines, "has": has,
		},
		"webhook": M{"deliveries": deliveryOut, "final_done": finalOut, "archive": archiveHookOut},
		"cost": M{
			"billed_thread_seconds": billed, "pages": len(rows), "run_seconds": round1(runSeconds),
			"pages_per_minute":        round1(pages / (runSeconds / 60)),
			"thread_seconds_per_page": round1(billed / pages),
			"client_running_seconds":  round1(tStopped.Sub(tStart).Seconds() + tDone.Sub(tResume).Seconds()),
		},
		"measured_at": time.Now().UTC().Format("2006-01-02"),
	}
	enc := json.NewEncoder(os.Stdout)
	enc.SetEscapeHTML(false)
	enc.SetIndent("", "  ")
	must(0, enc.Encode(out))
}
```

## What we got

| Step | What happened | Measured |
| --- | --- | --- |
| Map | 73 URLs in 6.4 s (sitemap 0, links 73); 51 category pages | 2026-11-10 |
| Crawl job | 202: max\_depth 2, max\_pages 100, concurrency 4, delay 250 ms, Markdown + extract | 2026-11-10 |
| Stop | DELETE at 30 pages → stopping; stopped 3.1 s later with 33 results stored; webhooks by then: job.done stopped | 2026-11-10 |
| Resume | 202 running → done after 2 runs; the second run added 67 new pages | 2026-11-10 |
| results.jsonl | 100 lines, 100 distinct URLs, 811 KB; 100 pages with Markdown (status 200 × 96, 407 × 4), 44 books with title and price, 0 failed | 2026-11-10 |
| Zip archive | 252 KB: 203 files, 100 index.md + 100 extract.json, index.jsonl 100 lines | 2026-11-10 |
| Webhooks | job.done (stopped) signature ok, job.done (done) signature ok, job.archive (ready) signature ok | 2026-11-10 |
| Speed | 100 pages in 246 s of running time: 24.4 pages/min | 2026-11-10 |
| Cost | 984 thread-seconds (9.8 per page) at concurrency 4 | 2026-11-10 |

From the Node.js run on 2026-11-10. IP addresses are replaced with placeholders (203.0.113.x); equal addresses stay equal. The other languages produced the same findings.

Raw output (Node.js)

```
{
  "language": "node",
  "site": "https://books.toscrape.com/",
  "map": {
    "status": 200,
    "seconds": 6.4,
    "urls": 73,
    "sources": {
      "sitemap": 0,
      "links": 73
    },
    "partial": "fetch_errors",
    "categories": 51
  },
  "crawl": {
    "max_depth": 2,
    "max_pages": 100,
    "concurrency": 4,
    "delay_ms": 250,
    "formats": [
      "markdown"
    ],
    "extract": [
      "title",
      "price"
    ],
    "start_status": 202
  },
  "stop": {
    "requested_at_done": 30,
    "reply": "stopping",
    "seconds_to_stopped": 3.1,
    "status": "stopped",
    "progress": {
      "done": 33,
      "total": 100,
      "failed": 0,
      "queued": 67
    },
    "stored_results": 33,
    "webhooks_by_then": [
      "job.done stopped"
    ],
    "archive": "none"
  },
  "resume": {
    "status": 202,
    "reply": "running"
  },
  "job": {
    "status": "done",
    "runs": 2,
    "progress": {
      "done": 100,
      "total": 100,
      "failed": 0,
      "queued": 0
    },
    "concurrency": 4,
    "billed_seconds": 984,
    "results": {
      "parts": 2,
      "bytes": 830120
    },
    "started_at": "2026-11-10T10:17:45.403Z",
    "ended_at": "2026-11-10T10:21:53.632Z",
    "expires_at": "2026-11-24T10:21:53.632Z"
  },
  "results_jsonl": {
    "lines": 100,
    "distinct_urls": 100,
    "bytes": 830120,
    "status_200": 96,
    "statuses": {
      "200": 96,
      "407": 4
    },
    "errors": [],
    "with_markdown": 100,
    "books_with_price": 44,
    "category_pages": 53,
    "depths": {
      "0": 1,
      "1": 73,
      "2": 26
    },
    "first_run": 33,
    "second_run_new": 67,
    "sample": [
      {
        "url": "https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html",
        "title": "A Light in the Attic",
        "price": "£51.77"
      },
      {
        "url": "https://books.toscrape.com/catalogue/tipping-the-velvet_999/index.html",
        "title": "Tipping the Velvet",
        "price": "£53.74"
      }
    ]
  },
  "archive": {
    "status": "ready",
    "bytes": 257662,
    "downloaded_bytes": 257662,
    "files": 203,
    "markdown_files": 100,
    "extract_files": 100,
    "index_lines": 100,
    "has": [
      "index.jsonl",
      "errors.jsonl",
      "job.json"
    ]
  },
  "webhook": {
    "deliveries": [
      {
        "event": "job.done",
        "status": "stopped",
        "job_id_matches": true,
        "delivery_id_is_uuid": true,
        "signature_valid": true
      },
      {
        "event": "job.done",
        "status": "done",
        "job_id_matches": true,
        "delivery_id_is_uuid": true,
        "signature_valid": true
      },
      {
        "event": "job.archive",
        "status": "ready",
        "job_id_matches": true,
        "delivery_id_is_uuid": true,
        "signature_valid": true
      }
    ],
    "final_done": {
      "data": {
        "status": "done",
        "progress": {
          "done": 100,
          "failed": 0,
          "queued": 0,
          "total": 100
        }
      },
      "signed_seconds_after_end": 0
    },
    "archive": {
      "status": "ready",
      "bytes": 257662,
      "has_url": true,
      "url_downloads_bytes": 257662,
      "signed_seconds_after_end": 0
    }
  },
  "cost": {
    "billed_thread_seconds": 984,
    "pages": 100,
    "run_seconds": 246,
    "pages_per_minute": 24.4,
    "thread_seconds_per_page": 9.8,
    "client_running_seconds": 249.8
  },
  "measured_at": "2026-11-10"
}
```

## Takeaways

-   **The whole round trip works, in all five languages.** On 2026-11-10 (re-run after the Functions fixes of that day) each run mapped the site, started the crawl, stopped it, resumed it and finished with exactly 100 pages: 100 lines in `results.jsonl`, 100 distinct URLs, every one with Markdown, none fetched twice. The depths were the same every time (1 start page, 73 at depth 1, 26 at depth 2); which 26 differed, so the book pages with a title and price came out 44 or 45.
-   **Map first: 73 URLs in 4.6–10.6 s.** books.toscrape.com has no sitemap, so the instant map is the start page's links (`sources: {sitemap: 0, links: 73}`, and `x-cdpfleet-partial: fetch_errors` for the missing robots/sitemap files): 51 category pages, 20 books, the home page and page 2. For a site with sitemaps the same call lists up to 5,000 URLs; `"job": true` maps deeper.
-   **Stop is quick and loses nothing; resume continues, it doesn't restart.** `DELETE` at 30–38 pages answered `stopping`; the job was `stopped` 3.1–12.5 s later with 33–45 results stored (the pages already in flight finish, at most 7 more). `POST …/resume` answered `202 running`, the second run fetched only URLs the first hadn't (55–67 of them) and the job ended `done` with `runs: 2`. No archive is built at a stop; it is built once, at the end.
-   **`progress` counts against `max_pages`.** At the stop `progress` read `total: 100` with 55–67 `queued`, and at the end `{done: 100, failed: 0, queued: 0, total: 100}`: a progress bar can be drawn straight from it. (Until 2026-11-10 `total` counted every link discovered, 1,600–1,800 here.)
-   **The zip is about a third of the JSON lines.** `results.jsonl` was 801–811 KB (Markdown inside JSON strings); the archive 250–252 KB with 203 files: an `index.md` and an `extract.json` per page, plus `index.jsonl` (100 lines), `errors.jsonl` and `job.json`. Use the JSON lines to stream results into code, the zip to hand a folder of Markdown to an LLM pipeline or keep it.
-   **Three signed webhooks, each saying what happened.** Every run received exactly three deliveries, all with a valid `x-cdpfleet-signature`, the job's id and a uuid delivery id: `job.done` with `data.status: "stopped"` at the stop, `job.done` with `status: "done"` and the final progress (signed in the same second as `ended_at`), and `job.archive` 0–1 s later with the zip's `bytes` and a download `url` — the same size as `GET …/archive`, and the link downloaded the same zip without a key. `job.archive` comes only because `events` asks for `"archive"`; with the default `["done"]` you get the two `job.done`. (Before the 2026-11-10 fix the stop's `job.done` said `interrupted`, and `job.archive` came unasked with only `{status}`.)
-   **Check `markdown`, not only `status`.** 2–4 of the 100 pages in three of the five runs came back with `status: 407` (Proxy Authentication Required — not what the site answered) while their Markdown and extracted price were complete (reported to Functions). Treat a page as fetched when it has its Markdown, and look at `error` for the ones that weren't.
-   **24–29 pages a minute at concurrency 4, 8.4–9.8 thread-seconds a page.** The two runs together ran 210–246 s for 100 pages and were billed 840–984 thread-seconds — `billed_seconds` is exactly `concurrency` × running time (98–99% of the time the client saw the job running; the pause between stop and resume costs nothing). At this pace 1,000 pages take about 2.3–2.7 thread-hours of your plan; the Community proxy (the default) adds nothing. A page took 8–10 s in a pooled tab including the 250 ms delay, so the way to go faster is more `concurrency`, not a shorter delay.

---
Source: https://cdpfleet.com/docs/cases/crawl-a-site-as-a-job · Markdown: https://cdpfleet.com/docs/cases/crawl-a-site-as-a-job.md · All docs for AI tools: https://cdpfleet.com/llms.txt