feat: add retry config for UNAVAILABLE errors

projects/fal/src/fal/sdk.py

@@ -39,8 +39,9 @@ def to_grpc(self) -> grpc.ChannelCredentials:
         raise NotImplementedError
 
     @property
-    def extra_options(self) -> list[tuple[str, str]]:
-        return GRPC_OPTIONS
+    def base_options(self) -> dict[str, str]:
+        grpc_ops: list[tuple[str, str]] = GRPC_OPTIONS
+        return dict(grpc_ops)
 
 
 class LocalCredentials(ServerCredentials):
@@ -341,15 +342,68 @@ def __exit__(self, *exc_info):
     def close(self):
         self._stack.close()
 
+    def grpc_service_config(self, existing_service_config: str | None) -> str:
+        # To be used for the `"grpc.service_config"` option in the gRPC channel.
+        import json
+
+        existing_service_config = existing_service_config or "{}"
+
+        service_config = {}
+        try:
+            service_config = json.loads(existing_service_config or "{}")
+            if not isinstance(service_config, dict):
+                raise json.JSONDecodeError("Invalid service config type", "", 0)
+
+            existing_method_config: list[dict] = service_config.get("methodConfig", [])
+            if not isinstance(existing_method_config, list):
+                raise json.JSONDecodeError("Invalid 'methodConfig' type", "", 0)
+        except json.JSONDecodeError:
+            log.warning(
+                f"Invalid service config: {existing_service_config}. Overwriting."
+            )
+            existing_method_config = []
+
+        method_config = {
+            # List all services this applies to
+            "name": [
+                {"service": "controller.IsolateController"},
+            ],
+            "retryPolicy": {
+                "maxAttempts": 5,
+                "initialBackoff": "0.1s",
+                "maxBackoff": "10s",
+                "backoffMultiplier": 2,
+                "retryableStatusCodes": ["UNAVAILABLE"],
+            },
+        }
+
+        existing_method_config.insert(0, method_config)
+        service_config["methodConfig"] = existing_method_config
+
+        try:
+            return json.dumps(service_config)
+        except json.JSONDecodeError:
+            log.warning(f"Failed to serialize service config: {service_config}")
+            return existing_service_config
+
     @property
     def stub(self) -> isolate_proto.IsolateControllerStub:
         if self._stub:
             return self._stub
 
-        options = self.credentials.server_credentials.extra_options
+        options: dict[str, str] = self.credentials.server_credentials.base_options
+        # Add the service config to the options
+        options["grpc.service_config"] = self.grpc_service_config(
+            options.get("grpc.service_config")
+        )
+
         channel_creds = self.credentials.to_grpc()
         channel = self._stack.enter_context(
-            grpc.secure_channel(self.hostname, channel_creds, options)
+            grpc.secure_channel(
+                target=self.hostname,
+                credentials=channel_creds,
+                options=list(options.items()),
+            )
         )
         channel = grpc.intercept_channel(channel, TraceContextInterceptor())
         self._stub = isolate_proto.IsolateControllerStub(channel)