diff --git a/docs/OperationsAPI.yaml b/docs/OperationsAPI.yaml
index d8f4c10a7..d299dc3a0 100644
--- a/docs/OperationsAPI.yaml
+++ b/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
diff --git a/functions-python/batch_process_dataset/src/main.py b/functions-python/batch_process_dataset/src/main.py
index 2ff117741..f305e6b1d 100644
--- a/functions-python/batch_process_dataset/src/main.py
+++ b/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,
diff --git a/functions-python/batch_process_dataset/src/scripts/download_verifier.py b/functions-python/batch_process_dataset/src/scripts/download_verifier.py
index 08a8be08c..db2a0a534 100644
--- a/functions-python/batch_process_dataset/src/scripts/download_verifier.py
+++ b/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()
diff --git a/functions-python/batch_process_dataset/tests/test_download_failure_recording.py b/functions-python/batch_process_dataset/tests/test_download_failure_recording.py
new file mode 100644
index 000000000..f3582831d
--- /dev/null
+++ b/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()
diff --git a/functions-python/helpers/tests/test_download_debug.py b/functions-python/helpers/tests/test_download_debug.py
new file mode 100644
index 000000000..997eefe4c
--- /dev/null
+++ b/functions-python/helpers/tests/test_download_debug.py
@@ -0,0 +1,326 @@
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import utils
+from utils import (
+    REDACTED,
+    FeedDownloadError,
+    FeedRequestDiagnostics,
+    build_availability_check,
+    download_feed,
+    get_external_ip,
+    sanitize_headers,
+    sanitize_url,
+)
+
+
+class TestSanitizeHeaders(unittest.TestCase):
+    def test_redacts_known_secret_headers(self):
+        result = sanitize_headers(
+            {
+                "User-Agent": "agent",
+                "Authorization": "Bearer abc",
+                "Cookie": "session=1",
+                "X-Api-Key": "secret",
+            }
+        )
+        self.assertEqual(result["User-Agent"], "agent")
+        self.assertEqual(result["Authorization"], REDACTED)
+        self.assertEqual(result["Cookie"], REDACTED)
+        self.assertEqual(result["X-Api-Key"], REDACTED)
+
+    def test_redacts_api_key_parameter_name_header(self):
+        result = sanitize_headers(
+            {"User-Agent": "agent", "my-feed-token": "secret"},
+            api_key_parameter_name="my-feed-token",
+        )
+        self.assertEqual(result["my-feed-token"], REDACTED)
+        self.assertEqual(result["User-Agent"], "agent")
+
+    def test_header_names_are_preserved(self):
+        result = sanitize_headers({"Authorization": "Bearer abc"})
+        self.assertIn("Authorization", result)
+
+    def test_none_returns_none(self):
+        self.assertIsNone(sanitize_headers(None))
+
+
+class TestSanitizeUrl(unittest.TestCase):
+    def test_redacts_api_key_query_parameter(self):
+        result = sanitize_url(
+            "https://example.com/gtfs.zip?operator=CE&token=secret",
+            api_key_parameter_name="token",
+        )
+        self.assertIn("operator=CE", result)
+        self.assertNotIn("secret", result)
+
+    def test_redacts_known_secret_query_parameters(self):
+        result = sanitize_url("https://example.com/gtfs.zip?api_key=secret&x=1")
+        self.assertIn("x=1", result)
+        self.assertNotIn("secret", result)
+
+    def test_strips_userinfo(self):
+        result = sanitize_url("https://user:password@example.com/gtfs.zip")
+        self.assertNotIn("password", result)
+        self.assertIn("example.com", result)
+
+    def test_url_without_credentials_is_unchanged(self):
+        url = "https://example.com/gtfs.zip"
+        self.assertEqual(sanitize_url(url), url)
+
+    def test_none_returns_none(self):
+        self.assertIsNone(sanitize_url(None))
+
+
+class TestGetExternalIp(unittest.TestCase):
+    def setUp(self):
+        utils._external_ip_cache = None
+
+    def tearDown(self):
+        utils._external_ip_cache = None
+
+    def test_returns_ip_on_success(self):
+        response = MagicMock(status=200, data=b"34.1.2.3\n")
+        pool = MagicMock()
+        pool.request.return_value = response
+        pool.__enter__ = lambda s: pool
+        pool.__exit__ = MagicMock(return_value=False)
+
+        with patch("utils.urllib3.PoolManager", return_value=pool):
+            self.assertEqual(get_external_ip(), "34.1.2.3")
+
+    def test_returns_none_when_lookup_fails(self):
+        with patch("utils.urllib3.PoolManager", side_effect=Exception("boom")):
+            self.assertIsNone(get_external_ip())
+
+    def test_result_is_cached(self):
+        response = MagicMock(status=200, data=b"34.1.2.3")
+        pool = MagicMock()
+        pool.request.return_value = response
+        pool.__enter__ = lambda s: pool
+        pool.__exit__ = MagicMock(return_value=False)
+
+        with patch("utils.urllib3.PoolManager", return_value=pool) as mock_pool:
+            get_external_ip()
+            get_external_ip()
+            self.assertEqual(mock_pool.call_count, 1)
+
+
+def _mock_response(status=200, headers=None, chunks=None):
+    response = MagicMock()
+    response.status = status
+    response.headers = headers if headers is not None else {}
+    response.retries = MagicMock(history=[])
+    response.__enter__ = lambda s: response
+    response.__exit__ = MagicMock(return_value=False)
+    data = list(chunks if chunks is not None else [b""])
+
+    def read(_size):
+        return data.pop(0) if data else b""
+
+    response.read.side_effect = read
+    return response
+
+
+def _mock_pool(response):
+    pool = MagicMock()
+    pool.request.return_value = response
+    pool.__enter__ = lambda s: pool
+    pool.__exit__ = MagicMock(return_value=False)
+    return pool
+
+
+@patch("shared.common.config_reader.get_config_value", return_value=None)
+class TestDownloadFeed(unittest.TestCase):
+    def setUp(self):
+        self.file_path = os.path.join(
+            os.path.dirname(__file__), "download_feed_test.bin"
+        )
+        utils._external_ip_cache = None
+
+    def tearDown(self):
+        if os.path.exists(self.file_path):
+            os.remove(self.file_path)
+        utils._external_ip_cache = None
+
+    def test_diagnostics_populated_on_success(self, _mock_config):
+        response = _mock_response(
+            status=200,
+            headers={"Content-Type": "application/zip", "Server": "nginx"},
+            chunks=[b"PK\x03\x04data", b""],
+        )
+        with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+            result = download_feed(
+                "https://example.com/gtfs.zip", self.file_path, feed_id="feed-1"
+            )
+
+        diagnostics = result.diagnostics
+        self.assertTrue(diagnostics.success)
+        self.assertEqual(diagnostics.status_code, 200)
+        self.assertEqual(diagnostics.content_type, "application/zip")
+        self.assertEqual(diagnostics.downloaded_bytes, 8)
+        self.assertEqual(diagnostics.response_headers["Server"], "nginx")
+        self.assertIn("User-Agent", diagnostics.request_headers)
+        self.assertIsNotNone(diagnostics.latency_ms)
+        self.assertIsNone(diagnostics.error_type)
+        self.assertIsNotNone(result.file_hash)
+
+    def test_external_ip_not_looked_up_on_success(self, _mock_config):
+        response = _mock_response(chunks=[b"data", b""])
+        with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+            with patch("utils.get_external_ip") as mock_ip:
+                download_feed("https://example.com/gtfs.zip", self.file_path)
+                mock_ip.assert_not_called()
+
+    def test_non_2xx_raises_with_diagnostics(self, _mock_config):
+        response = _mock_response(
+            status=403, headers={"Content-Type": "text/html", "Server": "cloudflare"}
+        )
+        with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+            with patch("utils.get_external_ip", return_value="34.1.2.3"):
+                with pytest.raises(FeedDownloadError) as exc_info:
+                    download_feed("https://example.com/gtfs.zip", self.file_path)
+
+        diagnostics = exc_info.value.diagnostics
+        self.assertFalse(diagnostics.success)
+        self.assertEqual(diagnostics.status_code, 403)
+        self.assertEqual(diagnostics.content_type, "text/html")
+        self.assertEqual(diagnostics.response_headers["Server"], "cloudflare")
+        self.assertEqual(diagnostics.external_ip, "34.1.2.3")
+        self.assertEqual(diagnostics.error_type, "ValueError")
+
+    def test_network_error_raises_with_diagnostics(self, _mock_config):
+        with patch("utils.urllib3.PoolManager", side_effect=Exception("Network error")):
+            with patch("utils.get_external_ip", return_value=None):
+                with pytest.raises(FeedDownloadError) as exc_info:
+                    download_feed("https://example.com/gtfs.zip", self.file_path)
+
+        diagnostics = exc_info.value.diagnostics
+        self.assertFalse(diagnostics.success)
+        self.assertIsNone(diagnostics.status_code)
+        self.assertEqual(diagnostics.error_type, "Exception")
+        self.assertEqual(diagnostics.error_message, "Network error")
+
+    def test_credentials_never_appear_in_diagnostics(self, _mock_config):
+        response = _mock_response(status=403)
+        with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+            with patch("utils.get_external_ip", return_value=None):
+                with pytest.raises(FeedDownloadError) as exc_info:
+                    download_feed(
+                        "https://example.com/gtfs.zip",
+                        self.file_path,
+                        authentication_type=1,
+                        api_key_parameter_name="token",
+                        credentials="super-secret",
+                    )
+
+        serialized = str(exc_info.value.diagnostics.as_dict())
+        self.assertNotIn("super-secret", serialized)
+        self.assertIn(REDACTED, serialized)
+
+    def test_max_bytes_truncates_and_returns_no_hash(self, _mock_config):
+        response = _mock_response(chunks=[b"a" * 10, b"a" * 10, b""])
+        with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+            result = download_feed(
+                "https://example.com/gtfs.zip",
+                self.file_path,
+                chunk_size=10,
+                max_bytes=10,
+            )
+
+        self.assertIsNone(result.file_hash)
+        self.assertEqual(result.diagnostics.downloaded_bytes, 10)
+
+    def test_redirect_urls_are_recorded(self, _mock_config):
+        response = _mock_response(chunks=[b"data", b""])
+        response.retries = MagicMock(
+            history=[MagicMock(redirect_location="https://cdn.example.com/gtfs.zip")]
+        )
+        with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+            result = download_feed("https://example.com/gtfs.zip", self.file_path)
+
+        self.assertEqual(
+            result.diagnostics.redirect_urls, ["https://cdn.example.com/gtfs.zip"]
+        )
+
+
+class TestBuildAvailabilityCheck(unittest.TestCase):
+    def test_maps_diagnostics_onto_row(self):
+        diagnostics = 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=["https://cdn.example.com/gtfs.zip"],
+            external_ip="34.1.2.3",
+            content_type="text/html",
+            is_zip=False,
+            error_type="ValueError",
+            error_message="Invalid HTTP response code: [403]",
+        )
+
+        row = build_availability_check(
+            diagnostics, feed_id="feed-1", source="dataset_download"
+        )
+
+        self.assertEqual(row.feed_id, "feed-1")
+        self.assertEqual(row.source, "dataset_download")
+        self.assertEqual(row.request_type, "http_get")
+        self.assertEqual(row.status_code, 403)
+        self.assertEqual(row.external_ip, "34.1.2.3")
+        self.assertEqual(row.response_headers, {"Server": "cloudflare"})
+        self.assertFalse(row.success)
+
+    def test_success_flag_follows_status_code(self):
+        diagnostics = FeedRequestDiagnostics(status_code=200)
+        row = build_availability_check(
+            diagnostics, feed_id="feed-1", source="dataset_download"
+        )
+        self.assertTrue(row.success)
+
+
+@patch("shared.common.config_reader.get_config_value", return_value=None)
+class TestErrorMessageSanitization(unittest.TestCase):
+    def setUp(self):
+        self.file_path = os.path.join(os.path.dirname(__file__), "err_test.bin")
+        utils._external_ip_cache = None
+
+    def tearDown(self):
+        if os.path.exists(self.file_path):
+            os.remove(self.file_path)
+        utils._external_ip_cache = None
+
+    def test_userinfo_in_producer_url_is_redacted_in_error(self, _mock_config):
+        url = "https://user:password@example.com/gtfs.zip"
+        with patch(
+            "utils.urllib3.PoolManager", side_effect=Exception(f"failed for {url}")
+        ):
+            with patch("utils.get_external_ip", return_value=None):
+                with pytest.raises(FeedDownloadError) as exc_info:
+                    download_feed(url, self.file_path)
+
+        self.assertNotIn("password", exc_info.value.diagnostics.error_message)
+
+    def test_credentials_in_resolved_url_are_redacted_in_error(self, _mock_config):
+        def boom(*_args, **_kwargs):
+            raise Exception(
+                "failed for https://example.com/gtfs.zip?token=super-secret"
+            )
+
+        with patch("utils.urllib3.PoolManager", side_effect=boom):
+            with patch("utils.get_external_ip", return_value=None):
+                with pytest.raises(FeedDownloadError) as exc_info:
+                    download_feed(
+                        "https://example.com/gtfs.zip",
+                        self.file_path,
+                        authentication_type=1,
+                        api_key_parameter_name="token",
+                        credentials="super-secret",
+                    )
+
+        self.assertNotIn("super-secret", exc_info.value.diagnostics.error_message)
diff --git a/functions-python/helpers/utils.py b/functions-python/helpers/utils.py
index 8d9f53dc3..ce138940f 100644
--- a/functions-python/helpers/utils.py
+++ b/functions-python/helpers/utils.py
@@ -18,10 +18,13 @@
 import os
 import ssl
 import time
+import zipfile
 import urllib3.exceptions
+from dataclasses import asdict, dataclass
 from datetime import date, datetime, timezone
 from logging import Logger
 from typing import Optional, TYPE_CHECKING
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
 if TYPE_CHECKING:  # pragma: no cover
     from shared.helpers.sizing import Size
@@ -145,6 +148,41 @@ def create_feed_ssl_context(trusted_certs: bool = False):
     return ctx
 
 
+def get_feed_credentials(stable_id: str) -> Optional[str]:
+    """Return the API credential for a feed from the FEEDS_CREDENTIALS env var, or None."""
+    try:
+        import json
+
+        feeds_credentials = json.loads(os.getenv("FEEDS_CREDENTIALS", "{}"))
+        return feeds_credentials.get(stable_id, None)
+    except Exception as exc:
+        logging.warning("Could not parse FEEDS_CREDENTIALS: %s", exc)
+        return None
+
+
+def _get_http_headers_override(feed_id: Optional[str]) -> Optional[dict]:
+    """Return the feed_download/http_headers config override, or None.
+
+    Returns None when the config database is unreachable so that callers fall back to
+    the default headers instead of failing. This keeps the local verifier script usable
+    without a database.
+    """
+    try:
+        from shared.common.config_reader import get_config_value
+
+        return get_config_value(
+            namespace="feed_download", key="http_headers", feed_id=feed_id
+        )
+    except Exception as exc:
+        logging.warning(
+            "Could not read feed_download/http_headers for feed %s, "
+            "falling back to default headers: %s",
+            feed_id,
+            exc,
+        )
+        return None
+
+
 def build_feed_request_params(
     url: str,
     feed_id: Optional[str] = None,
@@ -164,11 +202,7 @@ def build_feed_request_params(
     Returns:
         (headers, resolved_url) ready to pass to any HTTP method.
     """
-    from shared.common.config_reader import get_config_value
-
-    headers = get_config_value(
-        namespace="feed_download", key="http_headers", feed_id=feed_id
-    )
+    headers = _get_http_headers_override(feed_id)
     if headers is None:
         headers = {
             "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) "
@@ -252,6 +286,201 @@ def _sanitize_error_message(
     return message
 
 
+REDACTED = "***REDACTED***"
+
+_SECRET_HEADER_NAMES = frozenset(
+    {
+        "authorization",
+        "proxy-authorization",
+        "cookie",
+        "set-cookie",
+        "x-api-key",
+        "api-key",
+        "apikey",
+        "token",
+        "access-token",
+        "x-access-token",
+        "x-auth-token",
+    }
+)
+
+
+def _is_secret_name(name: str, api_key_parameter_name: Optional[str] = None) -> bool:
+    """Return True when a header or query parameter name carries a credential.
+
+    Underscores and hyphens are treated as equivalent so that both api_key and api-key
+    are matched.
+    """
+    lowered = name.strip().lower()
+    if lowered.replace("_", "-") in _SECRET_HEADER_NAMES:
+        return True
+    if not api_key_parameter_name:
+        return False
+    return lowered.replace("_", "-") == api_key_parameter_name.strip().lower().replace(
+        "_", "-"
+    )
+
+
+def sanitize_headers(
+    headers: Optional[dict], api_key_parameter_name: Optional[str] = None
+) -> Optional[dict]:
+    """Return a copy of headers with credential values replaced by REDACTED.
+
+    Names are always preserved, only values are redacted.
+    """
+    if headers is None:
+        return None
+    return {
+        name: (
+            REDACTED if _is_secret_name(name, api_key_parameter_name) else str(value)
+        )
+        for name, value in headers.items()
+    }
+
+
+def sanitize_url(
+    url: Optional[str], api_key_parameter_name: Optional[str] = None
+) -> Optional[str]:
+    """Return url with credential query-parameter values and userinfo redacted."""
+    if not url:
+        return url
+    try:
+        parsed = urlsplit(url)
+    except ValueError:
+        return REDACTED
+
+    netloc = parsed.netloc
+    if "@" in netloc:
+        netloc = f"{REDACTED}@{netloc.rsplit('@', 1)[1]}"
+
+    query = parsed.query
+    if query:
+        query = urlencode(
+            [
+                (
+                    name,
+                    (
+                        REDACTED
+                        if _is_secret_name(name, api_key_parameter_name)
+                        else value
+                    ),
+                )
+                for name, value in parse_qsl(query, keep_blank_values=True)
+            ],
+            safe="*",
+        )
+
+    return urlunsplit((parsed.scheme, netloc, parsed.path, query, parsed.fragment))
+
+
+_EXTERNAL_IP_URL = "https://checkip.amazonaws.com"
+_external_ip_cache: Optional[str] = None
+
+
+def get_external_ip(timeout_seconds: int = 3) -> Optional[str]:
+    """Return the egress IP address of this process, or None if it cannot be determined.
+
+    The result is cached for the lifetime of the process. Never raises.
+    """
+    global _external_ip_cache
+    if _external_ip_cache is not None:
+        return _external_ip_cache
+    try:
+        with urllib3.PoolManager() as http:
+            response = http.request(
+                "GET",
+                _EXTERNAL_IP_URL,
+                timeout=urllib3.Timeout(connect=timeout_seconds, read=timeout_seconds),
+                retries=False,
+            )
+            if response.status == 200:
+                _external_ip_cache = response.data.decode("utf-8").strip() or None
+    except Exception as exc:
+        logging.warning("Could not determine external IP address: %s", exc)
+    return _external_ip_cache
+
+
+@dataclass
+class FeedRequestDiagnostics:
+    """Request and response detail for a feed download attempt, safe to log and store."""
+
+    request_url: Optional[str] = None
+    resolved_url: Optional[str] = None
+    request_headers: Optional[dict] = None
+    response_headers: Optional[dict] = None
+    status_code: Optional[int] = None
+    latency_ms: Optional[int] = None
+    redirect_urls: Optional[list] = None
+    external_ip: Optional[str] = None
+    content_type: Optional[str] = None
+    is_zip: Optional[bool] = None
+    downloaded_bytes: Optional[int] = None
+    error_type: Optional[str] = None
+    error_message: Optional[str] = None
+
+    @property
+    def success(self) -> bool:
+        return (
+            self.error_type is None
+            and self.status_code is not None
+            and 200 <= self.status_code < 300
+        )
+
+    def as_dict(self) -> dict:
+        return asdict(self)
+
+
+class FeedDownloadError(Exception):
+    """Raised when a feed download fails, carrying the diagnostics of the attempt."""
+
+    def __init__(self, message: str, diagnostics: FeedRequestDiagnostics):
+        super().__init__(message)
+        self.diagnostics = diagnostics
+
+
+@dataclass
+class FeedDownloadResult:
+    """Outcome of a successful feed download."""
+
+    file_hash: Optional[str]
+    is_zip: Optional[bool]
+    diagnostics: FeedRequestDiagnostics
+
+
+def build_availability_check(
+    diagnostics: FeedRequestDiagnostics,
+    feed_id: str,
+    source: str,
+    request_type: str = "http_get",
+    checked_at: Optional[datetime] = None,
+):
+    """Map diagnostics onto a GtfsFeedAvailabilityCheck row.
+
+    source is one of 'availability_check' or 'dataset_download'.
+    """
+    from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck
+
+    return GtfsFeedAvailabilityCheck(
+        feed_id=feed_id,
+        checked_at=checked_at or datetime.now(timezone.utc),
+        request_url=diagnostics.request_url,
+        resolved_url=diagnostics.resolved_url,
+        request_type=request_type,
+        status_code=diagnostics.status_code,
+        latency_ms=diagnostics.latency_ms,
+        error_message=diagnostics.error_message,
+        error_type=diagnostics.error_type,
+        success=diagnostics.success,
+        content_type=diagnostics.content_type,
+        is_zip=diagnostics.is_zip,
+        source=source,
+        request_headers=diagnostics.request_headers,
+        response_headers=diagnostics.response_headers,
+        redirect_urls=diagnostics.redirect_urls,
+        external_ip=diagnostics.external_ip,
+    )
+
+
 def _execute_http_request(
     method: str,
     url: str,
@@ -413,6 +642,7 @@ def perform_request(
         feed_id=feed_id,
         checked_at=checked_at,
         request_url=producer_url,
+        resolved_url=sanitize_url(resolved_url, api_key_parameter_name),
         request_type=request_type,
         status_code=status_code,
         latency_ms=latency_ms,
@@ -421,10 +651,16 @@ def perform_request(
         success=success,
         content_type=content_type,
         is_zip=is_zip,
+        source="availability_check",
+        request_headers=sanitize_headers(headers, api_key_parameter_name),
+        response_headers=sanitize_headers(dict(resp_headers) if resp_headers else None),
+        redirect_urls=[
+            sanitize_url(u, api_key_parameter_name) for u in (redirect_urls or [])
+        ],
     )
 
 
-def download_and_get_hash(
+def download_feed(
     url,
     file_path,
     hash_algorithm="sha256",
@@ -435,48 +671,143 @@ def download_and_get_hash(
     credentials=None,
     logger=None,
     trusted_certs=False,  # If True, disables SSL verification
-):
-    """
-    Downloads the content of a URL and stores it in a file and returns the hash of the file
+    max_bytes: Optional[int] = None,
+    timeout_seconds: Optional[int] = None,
+) -> FeedDownloadResult:
+    """Download a feed to file_path and return its hash along with request diagnostics.
+
+    max_bytes stops the download once that many bytes have been read, which lets callers
+    inspect the response without storing the whole dataset. The returned hash is None in
+    that case because the file is incomplete.
+
+    Raises FeedDownloadError carrying a FeedRequestDiagnostics on any failure.
     """
     logger = logger or logging.getLogger(__name__)
-    try:
-        hash_object = hashlib.new(hash_algorithm)
+    producer_url = url
+    resolved_url = url
+    hash_object = hashlib.new(hash_algorithm)
+    diagnostics = FeedRequestDiagnostics(
+        request_url=sanitize_url(producer_url, api_key_parameter_name)
+    )
+    truncated = False
 
+    try:
         ctx = create_feed_ssl_context(trusted_certs=trusted_certs)
 
-        headers, url = build_feed_request_params(
-            url,
+        headers, resolved_url = build_feed_request_params(
+            producer_url,
             feed_id=feed_id,
             authentication_type=authentication_type,
             api_key_parameter_name=api_key_parameter_name,
             credentials=credentials,
         )
+        diagnostics.resolved_url = sanitize_url(resolved_url, api_key_parameter_name)
+        diagnostics.request_headers = sanitize_headers(headers, api_key_parameter_name)
+
+        timeout = (
+            urllib3.Timeout(connect=timeout_seconds, read=timeout_seconds)
+            if timeout_seconds
+            else None
+        )
+        request_kwargs = {"timeout": timeout} if timeout else {}
 
         with urllib3.PoolManager(ssl_context=ctx) as http:
+            start = time.monotonic()
             with http.request(
-                "GET", url, preload_content=False, headers=headers, redirect=True
+                "GET",
+                resolved_url,
+                preload_content=False,
+                headers=headers,
+                redirect=True,
+                **request_kwargs,
             ) as r, open(file_path, "wb") as out_file:
-                if 200 <= r.status < 300:
-                    logger.info(f"HTTP response code: [{r.status}]")
-                    while True:
-                        data = r.read(chunk_size)
-                        if not data:
-                            break
-                        hash_object.update(data)
-                        out_file.write(data)
-                    r.release_conn()
-                else:
+                diagnostics.latency_ms = int((time.monotonic() - start) * 1000)
+                diagnostics.status_code = r.status
+                diagnostics.response_headers = sanitize_headers(dict(r.headers))
+                diagnostics.redirect_urls = [
+                    sanitize_url(h.redirect_location, api_key_parameter_name)
+                    for h in (r.retries.history or [])
+                    if h.redirect_location
+                ]
+                diagnostics.content_type = _parse_content_type(
+                    r.headers.get("Content-Type")
+                )
+
+                if not 200 <= r.status < 300:
                     raise ValueError(f"Invalid HTTP response code: [{r.status}]")
-        return hash_object.hexdigest()
-    except Exception as e:
+
+                logger.info(f"HTTP response code: [{r.status}]")
+                total = 0
+                while True:
+                    data = r.read(chunk_size)
+                    if not data:
+                        break
+                    hash_object.update(data)
+                    out_file.write(data)
+                    total += len(data)
+                    if max_bytes is not None and total >= max_bytes:
+                        truncated = True
+                        break
+                r.release_conn()
+                diagnostics.downloaded_bytes = total
+
+        diagnostics.is_zip = zipfile.is_zipfile(file_path) or (
+            _is_zip_from_content_type(diagnostics.content_type) or False
+        )
+        return FeedDownloadResult(
+            file_hash=None if truncated else hash_object.hexdigest(),
+            is_zip=diagnostics.is_zip,
+            diagnostics=diagnostics,
+        )
+    except Exception as exc:
+        diagnostics.error_type = type(exc).__name__
+        # Collapse the credential-bearing resolved URL onto the producer URL, then the
+        # producer URL onto its sanitized form, so neither can leak through the message.
+        message = _sanitize_error_message(str(exc), resolved_url, producer_url)
+        if message and producer_url != diagnostics.request_url:
+            message = message.replace(producer_url, diagnostics.request_url)
+        diagnostics.error_message = message
+        diagnostics.external_ip = get_external_ip()
+        logger.warning(
+            "Feed download failed for %s: %s",
+            diagnostics.request_url,
+            diagnostics.as_dict(),
+        )
         if os.path.exists(file_path):
             try:
                 os.remove(file_path)
             except Exception:
                 logger.error(f"Delete file: [{file_path}]")
+        raise FeedDownloadError(str(exc), diagnostics) from exc
 
-        raise e
+
+def download_and_get_hash(
+    url,
+    file_path,
+    hash_algorithm="sha256",
+    chunk_size=8192,
+    feed_id=None,
+    authentication_type=0,
+    api_key_parameter_name=None,
+    credentials=None,
+    logger=None,
+    trusted_certs=False,  # If True, disables SSL verification
+):
+    """
+    Downloads the content of a URL and stores it in a file and returns the hash of the file
+    """
+    return download_feed(
+        url,
+        file_path,
+        hash_algorithm=hash_algorithm,
+        chunk_size=chunk_size,
+        feed_id=feed_id,
+        authentication_type=authentication_type,
+        api_key_parameter_name=api_key_parameter_name,
+        credentials=credentials,
+        logger=logger,
+        trusted_certs=trusted_certs,
+    ).file_hash
 
 
 def create_http_task(
diff --git a/functions-python/operations_api/.openapi-generator/FILES b/functions-python/operations_api/.openapi-generator/FILES
index c30bf8a8e..fccbd0a4b 100644
--- a/functions-python/operations_api/.openapi-generator/FILES
+++ b/functions-python/operations_api/.openapi-generator/FILES
@@ -49,6 +49,7 @@ src/feeds_gen/models/gtfs_feed_continuous_coverage.py
 src/feeds_gen/models/gtfs_feed_continuous_coverage_boundary.py
 src/feeds_gen/models/gtfs_feed_continuous_coverage_file.py
 src/feeds_gen/models/gtfs_feed_continuous_coverage_response.py
+src/feeds_gen/models/gtfs_feed_download_debug_response.py
 src/feeds_gen/models/gtfs_feed_validation_notice.py
 src/feeds_gen/models/gtfs_feed_validation_report.py
 src/feeds_gen/models/gtfs_feed_validation_reports_response.py
diff --git a/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py b/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py
index 31c462abf..0be24e953 100644
--- a/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py
+++ b/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py
@@ -18,6 +18,9 @@
 from fastapi.responses import JSONResponse
 
 import logging
+import os
+import tempfile
+import uuid
 from datetime import datetime
 from typing import Annotated, Final, Optional
 
@@ -35,6 +38,9 @@
 from feeds_gen.models.gtfs_feed_availability_response import (
     GtfsFeedAvailabilityResponse,
 )
+from feeds_gen.models.gtfs_feed_download_debug_response import (
+    GtfsFeedDownloadDebugResponse,
+)
 from feeds_gen.models.operation_create_request_gtfs_feed import (
     OperationCreateRequestGtfsFeed,
 )
@@ -73,6 +79,11 @@
     GtfsFeedAvailabilityCheckImpl,
 )
 from shared.helpers.pub_sub import get_execution_id, trigger_dataset_download
+from shared.helpers.utils import (
+    FeedDownloadError,
+    download_feed,
+    get_feed_credentials,
+)
 from shared.helpers.query_helper import (
     query_feed_by_stable_id,
     get_feeds_query,
@@ -432,6 +443,71 @@ def get_gtfs_feed_execution_attempts(
             attempts=[_execution_attempt(row) for row in attempts],
         )
 
+    @with_db_session
+    def debug_gtfs_feed_download(
+        self,
+        id: Annotated[
+            StrictStr, Field(description="The feed ID of the requested feed.")
+        ],
+        max_bytes: Optional[int] = 52428800,
+        timeout_seconds: Optional[int] = 60,
+        db_session: Session = None,
+    ) -> GtfsFeedDownloadDebugResponse:
+        """Download the feed through the shared download path and return the diagnostics."""
+        gtfs_feed = (
+            db_session.query(Gtfsfeed).filter(Gtfsfeed.stable_id == id).one_or_none()
+        )
+        if gtfs_feed is None:
+            raise HTTPException(status_code=404, detail="GTFS feed not found")
+        if not gtfs_feed.producer_url:
+            raise HTTPException(status_code=400, detail="GTFS feed has no producer URL")
+
+        temp_file = os.path.join(
+            tempfile.gettempdir(), f"download_debug_{uuid.uuid4()}.zip"
+        )
+        truncated = False
+        try:
+            try:
+                result = download_feed(
+                    gtfs_feed.producer_url,
+                    file_path=temp_file,
+                    feed_id=gtfs_feed.id,
+                    authentication_type=gtfs_feed.authentication_type or 0,
+                    api_key_parameter_name=gtfs_feed.api_key_parameter_name,
+                    credentials=get_feed_credentials(gtfs_feed.stable_id),
+                    max_bytes=max_bytes,
+                    timeout_seconds=timeout_seconds,
+                )
+                diagnostics = result.diagnostics
+                truncated = result.file_hash is None
+            except FeedDownloadError as exc:
+                diagnostics = exc.diagnostics
+        finally:
+            if os.path.exists(temp_file):
+                try:
+                    os.remove(temp_file)
+                except OSError as exc:
+                    logging.warning("Could not remove %s: %s", temp_file, exc)
+
+        return GtfsFeedDownloadDebugResponse(
+            feed_id=gtfs_feed.stable_id,
+            producer_url=diagnostics.request_url,
+            resolved_url=diagnostics.resolved_url,
+            success=diagnostics.success,
+            status_code=diagnostics.status_code,
+            latency_ms=diagnostics.latency_ms,
+            request_headers=diagnostics.request_headers,
+            response_headers=diagnostics.response_headers,
+            redirect_urls=diagnostics.redirect_urls,
+            external_ip=diagnostics.external_ip,
+            content_type=diagnostics.content_type,
+            is_zip=diagnostics.is_zip,
+            downloaded_bytes=diagnostics.downloaded_bytes,
+            truncated=truncated,
+            error_type=diagnostics.error_type,
+            error_message=diagnostics.error_message,
+        )
+
     @with_db_session
     def get_gtfs_rt_feed(
         self,
diff --git a/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py b/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py
new file mode 100644
index 000000000..7afc517e7
--- /dev/null
+++ b/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py
@@ -0,0 +1,130 @@
+from unittest.mock import patch
+
+import pytest
+from fastapi import HTTPException
+
+from conftest import feed_mdb_40
+from feeds_operations.impl.feeds_operations_impl import OperationsApiImpl
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck
+from shared.helpers.utils import FeedDownloadError, FeedDownloadResult
+from shared.helpers.utils import FeedRequestDiagnostics
+
+
+def _diagnostics(**overrides):
+    defaults = dict(
+        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=200,
+        latency_ms=120,
+        redirect_urls=[],
+        content_type="application/zip",
+        is_zip=True,
+        downloaded_bytes=2038,
+    )
+    defaults.update(overrides)
+    return FeedRequestDiagnostics(**defaults)
+
+
+@pytest.mark.asyncio
+async def test_download_debug_success():
+    """Returns the diagnostics of a successful download."""
+    api = OperationsApiImpl()
+    diagnostics = _diagnostics()
+    result = FeedDownloadResult(
+        file_hash="abc123", is_zip=True, diagnostics=diagnostics
+    )
+
+    with patch(
+        "feeds_operations.impl.feeds_operations_impl.download_feed",
+        return_value=result,
+    ):
+        response = api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id)
+
+    assert response.feed_id == feed_mdb_40.stable_id
+    assert response.success is True
+    assert response.status_code == 200
+    assert response.is_zip is True
+    assert response.downloaded_bytes == 2038
+    assert response.truncated is False
+    assert response.response_headers == {"Server": "cloudflare"}
+
+
+@pytest.mark.asyncio
+async def test_download_debug_failure_returns_diagnostics():
+    """A failed download returns 200 with the failure detail rather than raising."""
+    api = OperationsApiImpl()
+    diagnostics = _diagnostics(
+        status_code=403,
+        content_type="text/html",
+        is_zip=False,
+        downloaded_bytes=None,
+        external_ip="34.1.2.3",
+        error_type="ValueError",
+        error_message="Invalid HTTP response code: [403]",
+    )
+
+    with patch(
+        "feeds_operations.impl.feeds_operations_impl.download_feed",
+        side_effect=FeedDownloadError("boom", diagnostics),
+    ):
+        response = api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id)
+
+    assert response.success is False
+    assert response.status_code == 403
+    assert response.error_type == "ValueError"
+    assert response.external_ip == "34.1.2.3"
+
+
+@pytest.mark.asyncio
+async def test_download_debug_truncated_when_capped():
+    """A download stopped at the byte cap is reported as truncated."""
+    api = OperationsApiImpl()
+    result = FeedDownloadResult(
+        file_hash=None, is_zip=None, diagnostics=_diagnostics(downloaded_bytes=1024)
+    )
+
+    with patch(
+        "feeds_operations.impl.feeds_operations_impl.download_feed",
+        return_value=result,
+    ):
+        response = api.debug_gtfs_feed_download(
+            id=feed_mdb_40.stable_id, max_bytes=1024
+        )
+
+    assert response.truncated is True
+
+
+@pytest.mark.asyncio
+async def test_download_debug_not_found():
+    """Returns 404 for an unknown feed ID."""
+    api = OperationsApiImpl()
+    with pytest.raises(HTTPException) as exc_info:
+        api.debug_gtfs_feed_download(id="mdb-9999")
+    assert exc_info.value.status_code == 404
+
+
+@with_db_session
+def _count_availability_checks(db_session=None) -> int:
+    return db_session.query(GtfsFeedAvailabilityCheck).count()
+
+
+@pytest.mark.asyncio
+async def test_download_debug_does_not_persist():
+    """The endpoint stores nothing in gtfs_feed_availability_check."""
+    api = OperationsApiImpl()
+    result = FeedDownloadResult(
+        file_hash="abc123", is_zip=True, diagnostics=_diagnostics()
+    )
+
+    before = _count_availability_checks()
+
+    with patch(
+        "feeds_operations.impl.feeds_operations_impl.download_feed",
+        return_value=result,
+    ):
+        api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id)
+
+    assert _count_availability_checks() == before
diff --git a/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py b/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py
index 8e54d6577..371f52f51 100644
--- a/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py
+++ b/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py
@@ -14,7 +14,6 @@
 #  limitations under the License.
 #
 import logging
-import os
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
@@ -28,7 +27,7 @@
     Gtfsfeed,
     GtfsFeedAvailabilityCheck,
 )
-from shared.helpers.utils import perform_request
+from shared.helpers.utils import get_feed_credentials, perform_request
 
 DEFAULT_CONCURRENCY: int = 15
 DEFAULT_TIMEOUT_SECONDS: int = 20
@@ -36,18 +35,6 @@
 DEFAULT_FALLBACK_TO_GET: bool = True
 
 
-def get_feed_credentials(stable_id: str) -> Optional[str]:
-    """Return the API credential for a feed from the FEEDS_CREDENTIALS env var, or None."""
-    try:
-        import json
-
-        feeds_credentials = json.loads(os.getenv("FEEDS_CREDENTIALS", "{}"))
-        return feeds_credentials.get(stable_id, None)
-    except Exception as exc:
-        logging.warning("Could not parse FEEDS_CREDENTIALS: %s", exc)
-        return None
-
-
 def get_parameters(payload: dict):
     dry_run = payload.get("dry_run", True)
     skip_db_update = payload.get("skip_db_update", False)
@@ -286,6 +273,7 @@ def check_gtfs_feed_availability(
                     error_message=str(exc),
                     error_type=type(exc).__name__,
                     success=False,
+                    source="availability_check",
                 )
             results.append(check)
             batch.append(check)
diff --git a/liquibase/changelog.xml b/liquibase/changelog.xml
index aea96aac0..e7a7d9f34 100644
--- a/liquibase/changelog.xml
+++ b/liquibase/changelog.xml
@@ -154,6 +154,8 @@
     <include file="changes/feat_task_execution_attempt.sql" relativeToChangelogFile="true"/>
     <!-- Register the per-feed Parquet worker size key in config_key. -->
     <include file="changes/feat_parquet_builder_size.sql" relativeToChangelogFile="true"/>
+    <!-- Request/response debugging columns on gtfs_feed_availability_check. -->
+    <include file="changes/feat_download_debug.sql" relativeToChangelogFile="true"/>
     <!-- Keep this after all source table and schema changes so materialized views
          are recreated from the final source schema. -->
     <include file="materialized_views/materialized_views.xml" relativeToChangelogFile="true"/>
diff --git a/liquibase/changes/feat_download_debug.sql b/liquibase/changes/feat_download_debug.sql
new file mode 100644
index 000000000..55a9893f5
--- /dev/null
+++ b/liquibase/changes/feat_download_debug.sql
@@ -0,0 +1,20 @@
+-- Add request/response debugging columns to gtfs_feed_availability_check.
+-- source distinguishes rows written by the daily availability task from rows written
+-- when a dataset download fails in batch_process_dataset.
+-- request_headers, response_headers and resolved_url are sanitized before being stored:
+-- credential-bearing header values and query parameters are replaced with a redaction marker.
+-- external_ip is the egress IP observed for the request, captured best-effort on failures only.
+ALTER TABLE gtfs_feed_availability_check
+    ADD COLUMN IF NOT EXISTS source TEXT,
+    ADD COLUMN IF NOT EXISTS request_headers JSONB,
+    ADD COLUMN IF NOT EXISTS response_headers JSONB,
+    ADD COLUMN IF NOT EXISTS redirect_urls JSONB,
+    ADD COLUMN IF NOT EXISTS external_ip TEXT,
+    ADD COLUMN IF NOT EXISTS resolved_url TEXT;
+
+UPDATE gtfs_feed_availability_check
+SET source = 'availability_check'
+WHERE source IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_gtfs_feed_availability_check_source_checked_at
+    ON gtfs_feed_availability_check (source, checked_at);
diff --git a/scripts/download-verifier.sh b/scripts/download-verifier.sh
new file mode 100755
index 000000000..2a129a0fd
--- /dev/null
+++ b/scripts/download-verifier.sh
@@ -0,0 +1,124 @@
+#!/bin/bash
+#
+#
+#  MobilityData 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+
+# Downloads a GTFS feed using the same shared code as the nightly dataset download and
+# prints the request and response detail of the attempt, with credentials redacted.
+# No database is required: when the config database is unreachable the download falls back
+# to the default headers.
+
+# Usage:
+#   download-verifier.sh --url <producer url> [--feed_id <uuid>] [--with_upload] [--no_install_venv]
+# Example:
+#   download-verifier.sh --url https://example.com/gtfs.zip
+
+# relative path
+SCRIPT_PATH="$(dirname -- "${BASH_SOURCE[0]}")"
+FUNCTION_PATH="$SCRIPT_PATH/../functions-python/batch_process_dataset"
+
+# function printing usage
+display_usage() {
+  printf "\nThis script downloads a GTFS feed and prints the request and response detail"
+  printf "\nScript Usage:\n"
+  echo "Usage: $0 [options]"
+  echo "Options:"
+  echo "  -h|--help                 Display help content."
+  echo "  --url <URL>               Producer URL of the feed to download. Required."
+  echo "  --feed_id <UUID>          Feed UUID, to apply the per-feed http_headers config override."
+  echo "  --with_upload             Also verify the upload path using the GCP storage emulator."
+  echo "  --no_install_venv         Do not reinstall the python virtual environment."
+  exit 1
+}
+
+url=''
+feed_id=''
+with_upload=false
+no_install_venv=false
+
+while [[ $# -gt 0 ]]; do
+  key="$1"
+
+  case $key in
+  -h | --help)
+    display_usage
+    exit 0
+    ;;
+  --url)
+    url="$2"
+    shift # past argument
+    shift # past value
+    ;;
+  --feed_id)
+    feed_id="$2"
+    shift # past argument
+    shift # past value
+    ;;
+  --with_upload)
+    with_upload=true
+    shift # past argument
+    ;;
+  --no_install_venv)
+    no_install_venv=true
+    shift # past argument
+    ;;
+  *) # unknown option
+    shift # past argument
+    ;;
+  esac
+done
+
+if [ -z "$url" ]; then
+  printf "\nERROR: --url is required\n"
+  display_usage
+  exit 1
+fi
+
+if [ ! -d "$FUNCTION_PATH/src" ]; then
+  printf "\nERROR: function's folder not found at location: %s\n" "$FUNCTION_PATH/src"
+  exit 1
+fi
+
+# Link the shared code into src/shared so the function's imports resolve.
+printf "\nINFO: linking shared code\n"
+"$SCRIPT_PATH"/function-python-setup.sh --function_name batch_process_dataset
+
+if [ "$no_install_venv" != "true" ]; then
+  printf "\nINFO: installing python virtual environment\n"
+  pushd "$FUNCTION_PATH" >/dev/null
+  rm -rf venv
+  pip3 install --disable-pip-version-check virtualenv >/dev/null
+  python3 -m virtualenv venv >/dev/null
+  venv/bin/python -m pip install --disable-pip-version-check \
+    -r requirements.txt -r requirements_dev.txt >/dev/null
+  popd >/dev/null
+else
+  printf "\nINFO: skipping python virtual environment installation\n"
+fi
+
+export PYTHONPATH="$FUNCTION_PATH/src"
+
+args=(--url "$url")
+if [ -n "$feed_id" ]; then
+  args+=(--feed_id "$feed_id")
+fi
+if [ "$with_upload" != "true" ]; then
+  args+=(--skip-upload)
+fi
+
+printf "\nINFO: downloading %s\n" "$url"
+"$FUNCTION_PATH"/venv/bin/python "$FUNCTION_PATH/src/scripts/download_verifier.py" "${args[@]}"