SCIENCE-247 - reset lock on fork

docs/source/changelog.rst

@@ -4,6 +4,16 @@ Changelog
 Note: in some releases, there are no changes, because we always guarantee relasing in step
 with fsspec.
 
+2024.6.1
+--------
+
+no changes
+
+2024.6.0
+--------
+
+* Add seek(0) to request data to prevent issues on retries (#624)
+
 2024.5.0
 --------
 

environment_gcsfs.yaml

@@ -9,7 +9,7 @@ dependencies:
   - fusepy<3
   - google-api-core
   - google-api-python-client
-  - google-auth
+  - google-auth>=1.2, <2.33
   - google-auth-oauthlib
   - google-cloud-core
   - google-cloud-storage

gcsfs/core.py

@@ -27,6 +27,10 @@
 from .inventory_report import InventoryReport
 from .retry import errs, retry_request, validate_response
 
+os.register_at_fork(
+    after_in_child=asyn.reset_lock,
+)
+
 logger = logging.getLogger("gcsfs")
 
 
@@ -421,6 +425,8 @@ async def _request(
         self, method, path, *args, headers=None, json=None, data=None, **kwargs
     ):
         await self._set_session()
+        if hasattr(data, "seek"):
+            data.seek(0)
         async with self.session.request(
             method=method,
             url=self._format_path(path, args),

gcsfs/retry.py

@@ -99,7 +99,11 @@ def validate_response(status, content, path, args=None):
             content = content.decode()
         try:
             error = json.loads(content)["error"]
-            msg = error["message"]
+            # Sometimes the error message is a string.
+            if isinstance(error, str):
+                msg = error
+            else:
+                msg = error["message"]
         except json.decoder.JSONDecodeError:
             msg = content
 
@@ -109,7 +113,7 @@ def validate_response(status, content, path, args=None):
             raise requests.exceptions.ProxyError()
         elif "invalid" in str(msg):
             raise ValueError(f"Bad Request: {path}\n{msg}")
-        elif error:
+        elif error and not isinstance(error, str):
             raise HttpError(error)
         elif status:
             raise HttpError({"code": status, "message": msg})  # text-like

gcsfs/tests/test_retry.py

@@ -103,6 +103,15 @@ def test_validate_response():
         validate_response(502, b"", "/path")
 
 
+def test_validate_response_error_is_string():
+    # HttpError with JSON body
+    j = '{"error": "Too Many Requests"}'
+    with pytest.raises(HttpError) as e:
+        validate_response(429, j, "/path")
+    assert e.value.code == 429
+    assert e.value.message == "Too Many Requests, 429"
+
+
 @pytest.mark.parametrize(
     ["file_path", "validate_get_error", "validate_list_error", "expected_error"],
     [

requirements.txt

@@ -1,7 +1,7 @@
 aiohttp!=4.0.0a0, !=4.0.0a1
 decorator>4.1.2
-fsspec==2024.5.0
-google-auth>=1.2
+fsspec==2024.6.1
+google-auth>=1.2, <2.33
 google-auth-oauthlib
 google-cloud-storage
 requests