fix: bound Common Crawl orchestration fan-out

python/commonCrawlData/.funcignore

@@ -4,4 +4,5 @@ __azurite_db*__.json
 __blobstorage__
 __queuestorage__
 local.settings.json
-test
+test
+tests

python/commonCrawlData/README.md

@@ -41,7 +41,90 @@ add overhead. Confirm sizing against representative records and the intended
 worker concurrency before production deployment. These limits do not bound the
 orchestrator's total activity or result count.
 
-## Offline extraction tests
+## Crawl counts and activity scheduling
+
+The HTTP starter, orchestrator, and initial index activity share the policy in
+`crawl_limits.py`. The HTTP endpoint remains function-key protected.
+
+| Setting                          | Default             | Allowed range or budget                                                                                |
+| -------------------------------- | ------------------- | ------------------------------------------------------------------------------------------------------ |
+| `limit`                          | 1,000               | 1-1,000 accepted index records                                                                         |
+| `batch_size`                     | 50                  | 1-50 input records per processing activity                                                             |
+| `crawl_id`                       | `2025-13`           | Nonempty collection identifier without whitespace, control characters, or `/`, `\`, `?`, `#`, `&`, `%` |
+| `url_prefix`                     | Required            | Nonblank search string; existing prefix patterns are supported                                         |
+| Direct `get_urls` activity limit | 100                 | Same 1-1,000 range; orchestration calls pass their explicit limit                                      |
+| Outstanding activities           | 4 per orchestration | Processing and batch saves use separate windows and wait before starting the next window               |
+| Total logical activity calls     | At most 2,002       | One index call, up to 1,000 processing calls and 1,000 batch saves, and one combined save              |
+
+For `N` accepted records and batch size `B`, there are `ceil(N / B)` batches,
+with at most `2 * ceil(N / B) + 2` logical activity calls. The default 1,000/50
+crawl has 20 batches and at most 42 calls. These counts are not HTTP-attempt
+budgets or exactly-once execution guarantees; the existing per-record network
+retries remain unchanged.
+
+Missing or empty numeric HTTP query values retain their defaults. Other values
+must be decimal integers within the documented range; surrounding whitespace,
+an optional `+`, and leading zeros are accepted. Zero, negatives, fractions,
+exponents, and malformed values are rejected. Missing direct-orchestration
+fields retain defaults, but explicit null/empty/boolean/fractional values do not.
+Invalid starter requests return HTTP 400 JSON, for example:
+
+```json
+{
+  "status": "error",
+  "message": "batch_size must be an integer between 1 and 50"
+}
+```
+
+No orchestration is started for invalid HTTP input. Invalid direct orchestration
+input returns an explicit error payload before scheduling any activity; Durable
+may report the instance as **Completed**, with `status: error` in its output,
+rather than **Failed**. Activity exceptions still propagate.
+
+The initial index request sends the validated limit and counts streamed JSON
+records locally. More records than requested cause an explicit error, not silent
+truncation. The orchestrator independently rejects excess/malformed index output
+before processing. The index connect/read timeouts are 5/30 seconds; a read timeout
+limits inactivity, not the entire download duration. The existing logged non-200
+index response still returns an empty result, so it remains indistinguishable
+from no captures to the orchestrator.
+
+Rejected archive records may yield fewer extraction results, including empty
+batches. Empty batches keep their batch saves and original numbering. The combined
+save runs only when some data was extracted. `total_urls_processed` counts accepted
+input records, not successful downloads; `restaurants_extracted` counts actual
+results. Result formats, combined blob merging, and the starter's Durable status
+response are unchanged.
+
+`host.json` initially limits loaded activities and orchestrators to one each per
+worker. Four outstanding tasks can include queued tasks; this is not a promise of
+four simultaneous executions. Do not increase activity concurrency without
+reviewing worker threads/processes and the save activity's shared container state.
+Neither these host settings nor per-orchestration windows are a global rate limit.
+The numeric limits are conservative starting policy, not measured worker capacity.
+Record counts do not impose byte limits on initial-index lines, extraction objects,
+or retained workflow history. Per-record byte caps also do not establish a total
+worker-memory ceiling.
+
+This example retains the Python v1 (`function.json`) programming model and requires
+`azure-functions-durable<2`. The Durable Python SDK 2.x does not support that model;
+do not upgrade the SDK major version without a separately reviewed migration.
+The Requests, urllib3, and warcio pins from the per-record fix are unchanged.
+
+### Replay and rollout
+
+Output names use one recorded Durable UTC timestamp for all saves in a crawl.
+Domain naming uses tldextract's bundled suffix list without network fetching or
+disk caching. Its snapshot can classify rare/new suffixes differently from an
+older downloaded list; verify representative domains before rollout.
+
+This change alters activity scheduling history. Stop new starts and drain active
+instances on their old code before deployment. If draining is not possible, use
+a separately versioned deployment/task hub. Do not change the policy constants,
+dependencies, or bundled suffix data underneath active instances without the same
+replay-compatibility review.
+
+## Offline tests
 
 Use Python 3.11 or later. From the repository root on Windows:
 
@@ -50,13 +133,16 @@ Set-Location python\commonCrawlData
 python -m venv .venv
 .\.venv\Scripts\python.exe -m pip install -r requirements.txt
 .\.venv\Scripts\python.exe -m unittest discover -s tests -p "test_*.py"
-.\.venv\Scripts\python.exe -m compileall -q process_url_batch tests
+.\.venv\Scripts\python.exe -m compileall -q crawl_limits.py http_start get_urls orchestrator process_url_batch save_results tests
 ```
 
 The tests use mocked HTTP adapters and small synthetic WARC/gzip records, not
 live Common Crawl, Azure storage, or credentials. They also exercise real HTTP
 session behavior and real schema extraction, check cleanup and batch continuation,
-and measure bounded decoder allocation in an isolated process.
+and measure bounded decoder allocation in an isolated process. The crawl tests also
+cover parameter boundaries, zero scheduling on rejection, both activity windows,
+total calls, partial/empty extraction, output contracts, and replay-safe naming.
+Run the suites together to check the record-size and crawl-count fixes coexist.
 
 ## Trademarks
 

python/commonCrawlData/crawl_limits.py

@@ -0,0 +1,73 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import re
+
+
+DEFAULT_LIMIT = 1000
+DEFAULT_BATCH_SIZE = 50
+DEFAULT_INDEX_LIMIT = 100
+DEFAULT_CRAWL_ID = "2025-13"
+MAX_URLS = 1000
+MAX_BATCH_SIZE = 50
+MAX_IN_FLIGHT_ACTIVITIES = 4
+MAX_BATCHES = MAX_URLS
+MAX_ACTIVITY_CALLS = 2 * MAX_BATCHES + 2
+
+
+class CrawlValidationError(ValueError):
+    """Invalid crawl input or an activity result outside the crawl budget."""
+
+
+def bounded_positive_integer(value, name, maximum):
+    message = f"{name} must be an integer between 1 and {maximum}"
+    if type(value) is int:
+        number = value
+    elif isinstance(value, str):
+        value = value.strip()
+        if not re.fullmatch(r"[+-]?[0-9]+", value) or value.startswith("-"):
+            raise CrawlValidationError(message)
+        digits = value.lstrip("+").lstrip("0") or "0"
+        ceiling = str(maximum)
+        if len(digits) > len(ceiling) or (
+            len(digits) == len(ceiling) and digits > ceiling
+        ):
+            raise CrawlValidationError(message)
+        number = int(digits)
+    else:
+        raise CrawlValidationError(message)
+    if not 1 <= number <= maximum:
+        raise CrawlValidationError(message)
+    return number
+
+
+def validate_search_params(params, default_limit=DEFAULT_LIMIT):
+    if not isinstance(params, dict):
+        raise CrawlValidationError("Crawl input must be an object")
+    limit = bounded_positive_integer(
+        params.get("limit", default_limit), "limit", MAX_URLS
+    )
+    url_prefix = params.get("url_prefix")
+    if not isinstance(url_prefix, str) or not url_prefix.strip():
+        raise CrawlValidationError("url_prefix must be a nonempty string")
+    crawl_id = params.get("crawl_id", DEFAULT_CRAWL_ID)
+    if (
+        not isinstance(crawl_id, str)
+        or not crawl_id
+        or any(
+            not char.isprintable() or char.isspace() or char in "/\\?#&%"
+            for char in crawl_id
+        )
+    ):
+        raise CrawlValidationError(
+            "crawl_id must be a nonempty string without whitespace or URL delimiters"
+        )
+    return {"limit": limit, "crawl_id": crawl_id, "url_prefix": url_prefix}
+
+
+def validate_crawl_params(params):
+    validated = validate_search_params(params)
+    validated["batch_size"] = bounded_positive_integer(
+        params.get("batch_size", DEFAULT_BATCH_SIZE), "batch_size", MAX_BATCH_SIZE
+    )
+    return validated

python/commonCrawlData/get_urls/__init__.py

@@ -1,49 +1,76 @@
 # Copyright (c) Microsoft Corporation and Henry Lucco.
 # Licensed under the MIT License.
 
-import logging
-import azure.functions as func
 import json
+import logging
+
 import requests
+from crawl_limits import (
+    DEFAULT_CRAWL_ID,
+    DEFAULT_INDEX_LIMIT,
+    CrawlValidationError,
+    validate_search_params,
+)
+
+
+INDEX_TIMEOUT = (5, 30)
 
 
 def main(params: str) -> list:
     """Activity function to get URLs from Common Crawl"""
-    if isinstance(params, str):
-        try:
+    try:
+        if isinstance(params, str):
             params = json.loads(params)
-        except json.JSONDecodeError:
-            params = {}
-
-    limit = params.get("limit", 100)
-    crawl_id = params.get("crawl_id", "2025-13")
-    url_prefix = params.get("url_prefix", "")
-
-    if not url_prefix:
-        return {"status": "error", "message": "No URL search query provided"}
-
-    logging.info(f"Getting up to {limit} URLs from crawl {crawl_id}")
-    return get_urls(limit, crawl_id, url_prefix)
+        if not isinstance(params, dict):
+            raise CrawlValidationError("Index input must be an object")
+    except (ValueError, RecursionError) as exc:
+        logging.error("Invalid Common Crawl index input: expected an object")
+        raise CrawlValidationError(
+            "Index input must be an object or JSON object"
+        ) from exc
+    return get_urls(
+        params.get("limit", DEFAULT_INDEX_LIMIT),
+        params.get("crawl_id", DEFAULT_CRAWL_ID),
+        params.get("url_prefix", ""),
+    )
 
 
 def get_urls(limit, crawl_id, url_prefix):
     """Query Common Crawl index for restaurant URLs"""
-    # Common Crawl index URL pattern
-    cc_index_url = f"https://index.commoncrawl.org/CC-MAIN-{crawl_id}-index"
-    query = f"{url_prefix}*"
-
-    logging.info(f"Querying Common Crawl index: {cc_index_url}")
-    logging.info(f"Query: {query}")
-
-    response = requests.get(
-        cc_index_url + "?" + "url=" + query + "&output=json&limit=" + str(limit)
-    )
-
-    if response.status_code != 200:
-        logging.error(f"Error querying index: {response.status_code}")
-        return []
-
-    cc_data = [json.loads(line) for line in response.text.strip().split("\n")]
-    logging.info(f"Found {len(cc_data)} matching URLs")
-
-    return cc_data
+    try:
+        params = validate_search_params(
+            {"limit": limit, "crawl_id": crawl_id, "url_prefix": url_prefix}
+        )
+        limit = params["limit"]
+        cc_index_url = f"https://index.commoncrawl.org/CC-MAIN-{crawl_id}-index"
+        with requests.get(
+            cc_index_url,
+            params={"url": f"{url_prefix}*", "output": "json", "limit": limit},
+            timeout=INDEX_TIMEOUT,
+            stream=True,
+        ) as response:
+            if response.status_code != 200:
+                logging.error("Error querying index: %s", response.status_code)
+                return []
+            records = []
+            for line in response.iter_lines():
+                if not line.strip():
+                    continue
+                if len(records) == limit:
+                    raise CrawlValidationError(
+                        f"Index returned more records than the requested limit of {limit}"
+                    )
+                try:
+                    record = json.loads(line)
+                except (ValueError, RecursionError) as exc:
+                    raise CrawlValidationError(
+                        "Invalid JSON record in index response"
+                    ) from exc
+                if not isinstance(record, dict):
+                    raise CrawlValidationError("Index records must be objects")
+                records.append(record)
+        logging.info("Found %s matching URLs", len(records))
+        return records
+    except (CrawlValidationError, requests.exceptions.RequestException) as exc:
+        logging.error("Common Crawl index request failed: %.200s", exc)
+        raise

python/commonCrawlData/host.json

@@ -1,5 +1,11 @@
 {
   "version": "2.0",
+  "extensions": {
+    "durableTask": {
+      "maxConcurrentActivityFunctions": 1,
+      "maxConcurrentOrchestratorFunctions": 1
+    }
+  },
   "logging": {
     "applicationInsights": {
       "samplingSettings": {

python/commonCrawlData/http_start/__init__.py

@@ -1,45 +1,54 @@
 # Copyright (c) Microsoft Corporation and Henry Lucco.
 # Licensed under the MIT License.
 
+import json
 import logging
 import azure.functions as func
 import azure.durable_functions as df
+from crawl_limits import (
+    DEFAULT_BATCH_SIZE,
+    DEFAULT_CRAWL_ID,
+    DEFAULT_LIMIT,
+    CrawlValidationError,
+    validate_crawl_params,
+)
 
 
 async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
     """HTTP trigger to start the orchestration"""
-    client = df.DurableOrchestrationClient(starter)
-
-    # Get parameters from the request
-    limit_param = req.params.get("limit")
-    limit = int(limit_param) if limit_param else 1000
-
-    crawl_id_param = req.params.get("crawl_id")
-    crawl_id = crawl_id_param if crawl_id_param else "2025-13"
-
-    batch_size_param = req.params.get("batch_size")
-    batch_size = int(batch_size_param) if batch_size_param else 50
-
-    url_prefix_param = req.params.get("url_prefix")
-    if not url_prefix_param:
-        return {"status": "error", "message": "No URL search query provided"}
-
-    url_prefix = url_prefix_param
+    params = {
+        name: req.params.get(name)
+        for name in ("limit", "batch_size", "crawl_id", "url_prefix")
+    }
+    for name, default in (
+        ("limit", DEFAULT_LIMIT),
+        ("batch_size", DEFAULT_BATCH_SIZE),
+        ("crawl_id", DEFAULT_CRAWL_ID),
+    ):
+        if params[name] is None or params[name] == "":
+            params[name] = default
+    try:
+        params = validate_crawl_params(params)
+    except CrawlValidationError as exc:
+        logging.warning("Rejected Common Crawl request: %s", exc)
+        return func.HttpResponse(
+            json.dumps({"status": "error", "message": str(exc)}),
+            status_code=400,
+            mimetype="application/json",
+        )
 
     logging.info(
-        f"Starting orchestration with limit={limit}, crawl_id={crawl_id}, batch_size={batch_size}"
+        "Starting orchestration with limit=%s, crawl_id=%s, batch_size=%s",
+        params["limit"],
+        params["crawl_id"],
+        params["batch_size"],
     )
 
-    # Start the orchestration
+    client = df.DurableOrchestrationClient(starter)
     instance_id = await client.start_new(
         "orchestrator",
         None,
-        {
-            "limit": limit,
-            "crawl_id": crawl_id,
-            "batch_size": batch_size,
-            "url_prefix": url_prefix,
-        },
+        params,
     )
 
     logging.info(f"Started orchestration with ID: {instance_id}")