feat: add debugging to feed download process

docs/OperationsAPI.yaml

@@ -205,6 +205,33 @@ paths:
           description: GTFS feed not found.
         500:
           description: Internal server error.
+  /v1/operations/gtfs_feeds/{id}/download_debug:
+    parameters:
+      - $ref: "#/components/parameters/feed_id_path_param"
+    post:
+      description: >
+        Downloads the GTFS feed through the same code path used by the nightly dataset download and returns the request and response detail of that attempt. Credential values in headers and URLs are redacted. Nothing is stored.
+      tags:
+        - "operations"
+      operationId: debugGtfsFeedDownload
+      security:
+        - Authentication: []
+      parameters:
+        - $ref: "#/components/parameters/download_debug_max_bytes"
+        - $ref: "#/components/parameters/download_debug_timeout_seconds"
+      responses:
+        200:
+          description: Diagnostics for the download attempt.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/GtfsFeedDownloadDebugResponse"
+        400:
+          description: Invalid request parameters.
+        404:
+          description: GTFS feed not found.
+        500:
+          description: Internal server error.
   /v1/operations/gtfs_rt_feeds/{id}:
     parameters:
       - $ref: "#/components/parameters/feed_id_path_param"
@@ -3931,6 +3958,30 @@ components:
         minimum: 0
         default: 0
         example: 0
+    download_debug_max_bytes:
+      x-operation: true
+      name: max_bytes
+      in: query
+      description: Stop reading the feed after this many bytes. Keeps large feeds from exhausting the instance.
+      required: false
+      schema:
+        type: integer
+        minimum: 1024
+        maximum: 52428800
+        default: 52428800
+        example: 1048576
+    download_debug_timeout_seconds:
+      x-operation: true
+      name: timeout_seconds
+      in: query
+      description: Connect and read timeout for the download, in seconds.
+      required: false
+      schema:
+        type: integer
+        minimum: 1
+        maximum: 300
+        default: 60
+        example: 30
     limit_query_param_availability_endpoint:
       x-operation: true
       name: limit

functions-python/batch_process_dataset/src/main.py

@@ -42,6 +42,8 @@
 from shared.dataset_service.main import DatasetTraceService, DatasetTrace, Status
 from shared.helpers.logger import init_logger, get_logger
 from shared.helpers.utils import (
+    FeedDownloadError,
+    build_availability_check,
     download_and_get_hash,
     get_hash_from_file,
     download_from_gcs,
@@ -139,19 +141,40 @@ def download_content(self, temporary_file_path, feed_id):
         """
         Downloads the content of a URL and return the hash of the file
         """
-        file_hash = download_and_get_hash(
-            self.producer_url,
-            file_path=temporary_file_path,
-            feed_id=feed_id,
-            authentication_type=self.authentication_type,
-            api_key_parameter_name=self.api_key_parameter_name,
-            credentials=self.feed_credentials,
-            logger=self.logger,
-        )
+        try:
+            file_hash = download_and_get_hash(
+                self.producer_url,
+                file_path=temporary_file_path,
+                feed_id=feed_id,
+                authentication_type=self.authentication_type,
+                api_key_parameter_name=self.api_key_parameter_name,
+                credentials=self.feed_credentials,
+                logger=self.logger,
+            )
+        except FeedDownloadError as exc:
+            if feed_id:
+                self.record_download_failure(feed_id, exc.diagnostics)
+            raise
         self.logger.info("hash is: %s", file_hash)
         is_zip = zipfile.is_zipfile(temporary_file_path)
         return file_hash, is_zip
 
+    @with_db_session
+    def record_download_failure(self, feed_id, diagnostics, db_session: Session = None):
+        """
+        Stores the diagnostics of a failed download in gtfs_feed_availability_check.
+        Never raises: a storage failure must not mask the original download error.
+        """
+        try:
+            db_session.add(
+                build_availability_check(
+                    diagnostics, feed_id=feed_id, source="dataset_download"
+                )
+            )
+            db_session.commit()
+        except Exception as exc:
+            self.logger.error("Could not store download diagnostics: %s", exc)
+
     def upload_dataset_zip_to_storage(
         self,
         source_file_path,

functions-python/batch_process_dataset/src/scripts/download_verifier.py

@@ -1,28 +1,32 @@
+import argparse
+import json
 import logging
 import os
 
 from main import DatasetProcessor
 from gcp_storage_emulator.server import create_server
+from shared.helpers.utils import FeedDownloadError
 
 HOST = "localhost"
 PORT = 9023
 BUCKET_NAME = "verifier"
-PRODUCER_URL = "https://example.com/dataset.zip"  # Replace with actual producer URL
 
 
-def verify_download_content(producer_url: str):
+def verify_download_content(producer_url: str, feed_id=None):
     """
-    Verifies the download_content is able to retrieve the file
-    This is useful to simulate the download code locally and test issues related with user-agent and downloaded content.
-    Not supported authenticated feeds currently.
-    """
-    logging.info("Verifying downloaded content... (not implemented)")
+    Verifies the download_content is able to retrieve the file.
+    This is useful to simulate the download code locally and test issues related with
+    user-agent and downloaded content.
+    Authenticated feeds are not supported currently.
 
-    logging.info(f"Producer URL: {producer_url}")
+    Pass feed_id to apply the per-feed feed_download/http_headers override from the
+    config database, as the nightly run does.
+    """
+    logging.info("Producer URL: %s", producer_url)
 
     processor = DatasetProcessor(
         producer_url=producer_url,
-        feed_id=None,
+        feed_id=feed_id,
         feed_stable_id=None,
         execution_id=None,
         latest_hash=None,
@@ -32,20 +36,26 @@ def verify_download_content(producer_url: str):
         public_hosted_datasets_url=None,
     )
     tempfile = processor.generate_temp_filename()
-    logging.info(f"Temp filename: {tempfile}")
-    file_hash, is_zip = processor.download_content(tempfile, "feed_id")
-    logging.info(f"Downloaded file from {producer_url} is a valid ZIP file: {is_zip}")
-    logging.info(f"File hash: {file_hash}")
+    logging.info("Temp filename: %s", tempfile)
+    try:
+        file_hash, is_zip = processor.download_content(tempfile, feed_id)
+    except FeedDownloadError as exc:
+        logging.error("Download failed")
+        logging.error(json.dumps(exc.diagnostics.as_dict(), indent=2, default=str))
+        raise
+    logging.info(
+        "Downloaded file from %s is a valid ZIP file: %s", producer_url, is_zip
+    )
+    logging.info("File hash: %s", file_hash)
 
 
 def verify_upload_dataset(producer_url: str):
     """
     Verifies the upload_dataset is able to upload the dataset to the GCP storage emulator.
-    This is useful to simulate the upload code locally and test issues related with user-agent and uploaded content.
-    This function also tests the DatasetProcessor class methods for generating a temporary filename
-    and uploading the dataset.
-    :param producer_url:
-    :return:
+    This is useful to simulate the upload code locally and test issues related with
+    user-agent and uploaded content.
+    This function also tests the DatasetProcessor class methods for generating a temporary
+    filename and uploading the dataset.
     """
     processor = DatasetProcessor(
         producer_url=producer_url,
@@ -59,31 +69,55 @@ def verify_upload_dataset(producer_url: str):
         public_hosted_datasets_url=None,
     )
     tempfile = processor.generate_temp_filename()
-    logging.info(f"Temp filename: {tempfile}")
+    logging.info("Temp filename: %s", tempfile)
     dataset_file = processor.transfer_dataset("feed_id_2126", False)
-    logging.info(f"Dataset File: {dataset_file}")
+    logging.info("Dataset File: %s", dataset_file)
 
 
-if __name__ == "__main__":
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Download a GTFS feed through the shared download code and print the "
+        "request and response detail of the attempt."
+    )
+    parser.add_argument("--url", required=True, help="Producer URL of the feed.")
+    parser.add_argument(
+        "--feed_id",
+        default=None,
+        help="Feed UUID, to apply the per-feed http_headers config override.",
+    )
+    parser.add_argument(
+        "--skip-upload",
+        action="store_true",
+        help="Only verify the download, skipping the GCP storage emulator.",
+    )
+    return parser.parse_args()
+
+
+def main():
     logging.basicConfig(level=logging.INFO)
-    # Replace with actual producer URL
+    args = parse_args()
+
+    os.environ["WORKING_DIR"] = "/tmp/verifier"
+    os.makedirs(os.environ["WORKING_DIR"], exist_ok=True)
+
+    if args.skip_upload:
+        verify_download_content(producer_url=args.url, feed_id=args.feed_id)
+        logging.info("Download content verification completed successfully.")
+        return
+
+    os.environ["STORAGE_EMULATOR_HOST"] = f"http://{HOST}:{PORT}"
+    server = create_server(
+        host=HOST, port=PORT, in_memory=False, default_bucket=BUCKET_NAME
+    )
+    server.start()
     try:
-        os.environ["STORAGE_EMULATOR_HOST"] = f"http://{HOST}:{PORT}"
-        os.environ["WORKING_DIR"] = "/tmp/verifier"
-        # create working dir if not exists
-        if not os.path.exists(os.environ["WORKING_DIR"]):
-            os.makedirs(os.environ["WORKING_DIR"])
-
-        server = create_server(
-            host=HOST, port=PORT, in_memory=False, default_bucket=BUCKET_NAME
-        )
-        server.start()
-
-        verify_download_content(producer_url=PRODUCER_URL)
+        verify_download_content(producer_url=args.url, feed_id=args.feed_id)
         logging.info("Download content verification completed successfully.")
-        verify_upload_dataset(producer_url=PRODUCER_URL)
-    except Exception as e:
-        logging.error(f"Error verifying download content: {e}")
+        verify_upload_dataset(producer_url=args.url)
     finally:
         server.stop()
         logging.info("Verification completed.")
+
+
+if __name__ == "__main__":
+    main()

functions-python/batch_process_dataset/tests/test_download_failure_recording.py

@@ -0,0 +1,113 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from main import DatasetProcessor
+from shared.helpers.utils import FeedDownloadError, FeedRequestDiagnostics
+
+
+def _processor(feed_id="feed-1"):
+    return DatasetProcessor(
+        producer_url="https://example.com/gtfs.zip",
+        feed_id=feed_id,
+        feed_stable_id="mdb-1",
+        execution_id=None,
+        latest_hash=None,
+        bucket_name=None,
+        authentication_type=0,
+        api_key_parameter_name=None,
+        public_hosted_datasets_url=None,
+    )
+
+
+def _diagnostics():
+    return FeedRequestDiagnostics(
+        request_url="https://example.com/gtfs.zip",
+        resolved_url="https://example.com/gtfs.zip",
+        request_headers={"User-Agent": "agent"},
+        response_headers={"Server": "cloudflare"},
+        status_code=403,
+        latency_ms=1840,
+        redirect_urls=[],
+        external_ip="34.1.2.3",
+        content_type="text/html",
+        is_zip=False,
+        error_type="ValueError",
+        error_message="Invalid HTTP response code: [403]",
+    )
+
+
+class TestDownloadFailureRecording(unittest.TestCase):
+    def test_failure_records_one_row_and_reraises(self):
+        processor = _processor()
+        diagnostics = _diagnostics()
+
+        with patch(
+            "main.download_and_get_hash",
+            side_effect=FeedDownloadError("boom", diagnostics),
+        ):
+            with patch.object(processor, "record_download_failure") as mock_record:
+                with pytest.raises(FeedDownloadError):
+                    processor.download_content("/tmp/does-not-matter", "feed-1")
+
+        mock_record.assert_called_once()
+        args = mock_record.call_args[0]
+        self.assertEqual(args[0], "feed-1")
+        self.assertIs(args[1], diagnostics)
+
+    def test_success_records_nothing(self):
+        processor = _processor()
+
+        with patch("main.download_and_get_hash", return_value="abc123"):
+            with patch("main.zipfile.is_zipfile", return_value=True):
+                with patch.object(processor, "record_download_failure") as mock_record:
+                    file_hash, is_zip = processor.download_content(
+                        "/tmp/does-not-matter", "feed-1"
+                    )
+
+        self.assertEqual(file_hash, "abc123")
+        self.assertTrue(is_zip)
+        mock_record.assert_not_called()
+
+    def test_no_feed_id_skips_recording(self):
+        """The local verifier runs without a feed id, so there is no row to attach."""
+        processor = _processor(feed_id=None)
+
+        with patch(
+            "main.download_and_get_hash",
+            side_effect=FeedDownloadError("boom", _diagnostics()),
+        ):
+            with patch.object(processor, "record_download_failure") as mock_record:
+                with pytest.raises(FeedDownloadError):
+                    processor.download_content("/tmp/does-not-matter", None)
+
+        mock_record.assert_not_called()
+
+    def test_storage_failure_does_not_mask_download_error(self):
+        processor = _processor()
+
+        session = MagicMock()
+        session.commit.side_effect = Exception("db down")
+
+        with patch(
+            "main.download_and_get_hash",
+            side_effect=FeedDownloadError("boom", _diagnostics()),
+        ):
+            with patch("main.build_availability_check", return_value=MagicMock()):
+                with pytest.raises(FeedDownloadError):
+                    processor.download_content("/tmp/does-not-matter", "feed-1")
+
+    def test_recorded_row_is_tagged_as_dataset_download(self):
+        processor = _processor()
+        diagnostics = _diagnostics()
+        session = MagicMock()
+
+        with patch("main.build_availability_check") as mock_build:
+            processor.record_download_failure("feed-1", diagnostics, db_session=session)
+
+        mock_build.assert_called_once_with(
+            diagnostics, feed_id="feed-1", source="dataset_download"
+        )
+        session.add.assert_called_once()
+        session.commit.assert_called_once()