Merge pull request #1048 from dandi/gh-1033

Retry Zarr entry uploads that fail with "header ... implies functionality ... not implemented"

dandi/dandiapi.py

@@ -141,6 +141,7 @@ def request(
         headers: Optional[dict] = None,
         json_resp: bool = True,
         retry_statuses: Sequence[int] = (),
+        retry_if: Optional[Callable[[requests.Response], Any]] = None,
         **kwargs: Any,
     ) -> Any:
         """
@@ -184,6 +185,8 @@ def request(
         :type json_resp: bool
         :param retry_statuses: a sequence of HTTP response status codes to
             retry in addition to `dandi.consts.RETRY_STATUSES`
+        :param retry_if: an optional predicate applied to a failed HTTP
+            response to test whether to retry
         """
 
         url = self.get_url(path)
@@ -223,7 +226,9 @@ def request(
                         headers=headers,
                         **kwargs,
                     )
-                    if result.status_code in [*RETRY_STATUSES, *retry_statuses]:
+                    if result.status_code in [*RETRY_STATUSES, *retry_statuses] or (
+                        retry_if is not None and retry_if(result)
+                    ):
                         result.raise_for_status()
         except Exception:
             lgr.exception("HTTP connection failed")

dandi/files.py

@@ -1213,10 +1213,23 @@ def _upload_zarr_file(
     storage_session: RESTFullAPIClient, path: Path, upload_url: str
 ) -> int:
     with path.open("rb") as fp:
-        storage_session.put(upload_url, data=fp, json_resp=False)
+        storage_session.put(
+            upload_url, data=fp, json_resp=False, retry_if=_retry_zarr_file
+        )
     return path.stat().st_size
 
 
+def _retry_zarr_file(r: requests.Response) -> bool:
+    # Some sort of filesystem hiccup can cause requests to be unable to get the
+    # filesize, leading to it falling back to "chunked" transfer encoding,
+    # which S3 doesn't support.
+    return (
+        r.status_code == 501
+        and "header you provided implies functionality that is not implemented"
+        in r.text
+    )
+
+
 def _check_required_fields(d: dict, required: List[str]) -> List[str]:
     errors: List[str] = []
     for f in required: