feat(dds_ingestion): tell the user when DDS cannot write to S3, in it…

src/eea_datalakehouse/dds_ingestion/__init__.py

@@ -27,7 +27,12 @@
 
 from __future__ import annotations
 
-from .client import IngestApiError, IngestClient, S3UploadError
+from .client import (
+    IngestApiError,
+    IngestClient,
+    S3UploadError,
+    StorageUnavailableError,
+)
 from .credentials import (
     DremioCreds,
     MissingCredentialsError,
@@ -77,6 +82,7 @@
     "S3UploadError",
     "StageResult",
     "StatusResult",
+    "StorageUnavailableError",
     "UploadPart",
     "UploadTarget",
     "ingest_folder",

src/eea_datalakehouse/dds_ingestion/client.py

@@ -72,6 +72,31 @@ def __init__(self, status_code: int, message: str, *, where: str | None = None)
         self.where = where
 
 
+class StorageUnavailableError(IngestApiError):
+    """DDS refused the transfer: it cannot write to its own S3 storage (503).
+
+    Not a fault in the folder, the target or the caller's rights — the service
+    verified, with its own credentials, that the object storage behind the
+    catalog rejects writes, and said so instead of handing out upload targets
+    that every file would fail against. Nothing has been uploaded; the fix is an
+    administrator's (credentials, bucket policy, endpoint), and the transfer can
+    simply be re-run once storage is working.
+
+    Raised only on DDS's own ``storage_unavailable`` answer, so a 503 from a
+    proxy or load balancer in front of the service stays a plain
+    :class:`IngestApiError` and is not mislabelled as a storage fault.
+    """
+
+    def __init__(self, status_code: int, message: str, *, where: str | None = None) -> None:
+        # The server's message is already written for the person reading this in
+        # a notebook; wrapping it in "DDS ingest API error 503 on POST ..." would
+        # bury the explanation behind plumbing. The class name carries the rest.
+        RuntimeError.__init__(self, message)
+        self.status_code = status_code
+        self.message = message
+        self.where = where
+
+
 class S3UploadError(IngestApiError):
     """A pre-signed upload to object storage failed — NOT a DDS API error.
 
@@ -381,12 +406,24 @@ def _error_message(resp: httpx.Response) -> str:
             message = message[:_MAX_ERROR_CHARS] + "… (truncated)"
         return message or f"(empty {resp.status_code} response body)"
 
+    @staticmethod
+    def _error_slug(resp: httpx.Response) -> str:
+        """The machine-readable ``error`` slug of a DDS error body, or ``""``."""
+        try:
+            payload = resp.json()
+        except ValueError:
+            return ""
+        return str(payload.get("error", "")) if isinstance(payload, dict) else ""
+
     @classmethod
     def _raise_for_status(cls, resp: httpx.Response, where: str) -> None:
         """Raise :class:`IngestApiError` naming ``where`` unless the call succeeded."""
         if resp.is_success:
             return
-        raise IngestApiError(resp.status_code, cls._error_message(resp), where=where)
+        message = cls._error_message(resp)
+        if cls._error_slug(resp) == "storage_unavailable":
+            raise StorageUnavailableError(resp.status_code, message, where=where)
+        raise IngestApiError(resp.status_code, message, where=where)
 
     @classmethod
     def _raise_for_upload(

tests/dds_ingestion/test_client.py

@@ -7,6 +7,7 @@
     IngestApiError,
     IngestClient,
     S3UploadError,
+    StorageUnavailableError,
 )
 from eea_datalakehouse.dds_ingestion.credentials import DremioCreds
 from eea_datalakehouse.dds_ingestion.models import FileSpec, UploadPart, UploadTarget
@@ -303,3 +304,70 @@ def test_commit_result_carries_the_physical_location() -> None:
         {"session_id": "s2", "status": "done", "table_path": "x/y", "record_count": 1}
     )
     assert staged.storage_path is None
+
+
+# --- storage the service itself cannot write to (DDS 503) ------------------
+
+_STORAGE_DOWN = (
+    "this transfer cannot start: DDS cannot write to the S3 storage behind the "
+    "catalog, so every file would fail to upload. This is a dependency problem "
+    "between DDS and S3 that has to be resolved by an administrator — it is not "
+    "caused by your data or your permissions, and nothing has been uploaded. "
+    "Verified at 2026-08-25T14:02:11+00:00 — catalog bucket (ingest): write: "
+    "ClientError: AccessDenied (HTTP 403): Access Denied."
+)
+
+
+@respx.mock
+def test_storage_unavailable_is_its_own_error(creds: DremioCreds) -> None:
+    """DDS refusing up front is not the caller's mistake, and says so.
+
+    The server has already written the explanation for whoever is reading it in
+    a notebook, so it must arrive intact rather than wrapped in plumbing.
+    """
+    respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock(
+        return_value=httpx.Response(
+            503,
+            json={
+                "error": "storage_unavailable",
+                "message": _STORAGE_DOWN,
+                "path": "bio.uploads",
+            },
+        )
+    )
+    with IngestClient(BASE_URL, creds) as client, pytest.raises(
+        StorageUnavailableError
+    ) as exc:
+        client.begin(
+            target_catalog_path="bio.uploads",
+            intent="read_only",
+            data_format="parquet",
+            conflict_mode="fail",
+            files=[FileSpec("a.parquet", 6)],
+        )
+
+    assert exc.value.status_code == 503
+    assert str(exc.value) == _STORAGE_DOWN
+    assert "DDS ingest API error" not in str(exc.value)
+    assert exc.value.where == "POST /api/v1/ingest/begin"
+    # …and still an IngestApiError, so existing handling keeps working.
+    assert isinstance(exc.value, IngestApiError)
+
+
+@respx.mock
+def test_a_503_from_a_proxy_is_not_called_a_storage_fault(creds: DremioCreds) -> None:
+    """Only DDS's own slug means storage; a gateway's 503 means the gateway."""
+    respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock(
+        return_value=httpx.Response(503, text="<html>503 Service Unavailable</html>")
+    )
+    with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc:
+        client.begin(
+            target_catalog_path="bio.uploads",
+            intent="read_only",
+            data_format="parquet",
+            conflict_mode="fail",
+            files=[FileSpec("a.parquet", 6)],
+        )
+
+    assert not isinstance(exc.value, StorageUnavailableError)
+    assert "POST /api/v1/ingest/begin" in str(exc.value)

tests/dds_ingestion/test_folder.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 
 import pytest
+from eea_datalakehouse.dds_ingestion.client import StorageUnavailableError
 from eea_datalakehouse.dds_ingestion.folder import (
     FolderIngest,
     IngestStateError,
@@ -296,3 +297,35 @@ def test_retry_still_starts_over_for_a_staged_transfer(tmp_path: Path) -> None:
     job._session_id = "sess-1"
     job.retry()
     assert len(client.begin_calls) == 1
+
+
+# --- storage DDS cannot write to -------------------------------------------
+
+
+class StorageDownClient(FakeClient):
+    """A server that refuses the transfer at ``begin``: its S3 rejects writes."""
+
+    def begin(self, *, files: list[FileSpec], **kwargs: object) -> BeginResult:
+        self.begin_calls.append({"files": files, **kwargs})
+        raise StorageUnavailableError(
+            503,
+            "this transfer cannot start: DDS cannot write to the S3 storage "
+            "behind the catalog … resolved by an administrator … nothing has "
+            "been uploaded.",
+            where="POST /api/v1/ingest/begin",
+        )
+
+
+def test_run_stops_before_uploading_when_dds_cannot_write_to_s3(
+    data_folder: Path,
+) -> None:
+    """The point of the server-side pre-flight: nothing is transferred."""
+    client = StorageDownClient()
+    job = _make_ingest(data_folder, client)
+
+    with pytest.raises(StorageUnavailableError) as exc:
+        job.run()
+
+    assert "administrator" in str(exc.value)
+    assert client.uploaded == []
+    assert client.commit_calls == []