Repository: skypilot-org/skypilot
Pull request #6256 (2025-07-14 13:26:56)

[API server] Rebuild broken pool
<!-- Describe the changes in this PR -->

This PR fixes an issue that a disrupted executor process will break the entire process pool:

- **Previously**: when a executor process exited (e.g. OOM killed), the entire process will be broken and neither user requests nor background daemons can proceed;
- **PR**: when the pool is broken, rebuild a new pool to replace the broken pool. Background daemon and retryable request will be retried.

User will see the following logs when this issue happens:

```
Error processing request: [...] [concurrent.futures.process.BrokenProcessPool[] A child process terminated abruptly, the process pool is not usable anymore
```

Test:

```bash
# Simulate an OOM kill
$ kill -9 14749
# Verify the daemons are retried and pool is rebuilt
$ cat  ~/.sky/api_server/server.log
...
Process SpawnProcess-442:
Process SpawnProcess-448:
Process SpawnProcess-444:
KeyboardInterrupt
KeyboardInterrupt
Traceback (most recent call last):
  File "/opt/homebrew/anaconda3/envs/sky/lib/python3.10/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/opt/homebrew/anaconda3/envs/sky/lib/python3.10/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/homebrew/anaconda3/envs/sky/lib/python3.10/concurrent/futures/process.py", line 240, in _process_worker
    call_item = call_queue.get(block=True)
  File "/opt/homebrew/anaconda3/envs/sky/lib/python3.10/multiprocessing/queues.py", line 102, in get
    with self._rlock:
  File "/opt/homebrew/anaconda3/envs/sky/lib/python3.10/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
  File "/Users/aylei/repo/skypilot-org/skypilot/sky/server/requests/executor.py", line 342, in _sigterm_handler
    raise KeyboardInterrupt
KeyboardInterrupt
E 07-14 21:26:25 executor.py:216] Request skypilot-status-refresh-daemon failed to get processed [concurrent.futures.process.BrokenProcessPool] A process in the process pool was terminated abruptly while the future was running or pending.
E 07-14 21:26:25 executor.py:216] Request skypilot-volume-status-refresh-daemon failed to get processed [concurrent.futures.process.BrokenProcessPool] A process in the process pool was terminated abruptly while the future was running or pending.
I 07-14 21:26:25 executor.py:184] [Worker(schedule_type=short)] Submitting request: skypilot-status-refresh-daemon
W 07-14 21:26:25 process.py:279] The guaranteed pool is broken, replacing the pool and retrying. Error: A child process terminated abruptly, the process pool is not usable anymore
I 07-14 21:26:25 executor.py:198] [Worker(schedule_type=short)] Submitted request: skypilot-status-refresh-daemon
I 07-14 21:26:25 executor.py:184] [Worker(schedule_type=short)] Submitting request: skypilot-volume-status-refresh-daemon
I 07-14 21:26:25 executor.py:198] [Worker(schedule_type=short)] Submitted request: skypilot-volume-status-refresh-daemon
I 07-14 21:26:26 executor.py:361] Running request skypilot-status-refresh-daemon with pid 19948
```

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [ ] Relevant individual tests: `/smoke-test -k test_name` (CI) or `pytest tests/test_smoke.py::test_name` (local)
- [ ] Backward compatibility: `/quicktest-core` (CI) or `pytest tests/smoke_tests/test_backward_compat.py` (local)

<!-- CI commands (/-prefixed) can only be triggered by repo members -->



Patch:
diff --git a/docs/source/_static/custom.js b/docs/source/_static/custom.js
index f86673056..39628c51c 100644
--- a/docs/source/_static/custom.js
+++ b/docs/source/_static/custom.js
@@ -21,6 +21,7 @@ document.addEventListener('DOMContentLoaded', function () {
        r.src=t+h._hjSettings.hjid+j+h._hjSettings.hjsv;
        a.appendChild(r);
 })(window,document,'https://static.hotjar.com/c/hotjar-','.js?sv=');
+!function(t){var k="ko",i=(window.globalKoalaKey=window.globalKoalaKey||k);if(window[i])return;var ko=(window[i]=[]);["identify","track","removeListeners","on","off","qualify","ready"].forEach(function(t){ko[t]=function(){var n=[].slice.call(arguments);return n.unshift(t),ko.push(n),ko}});var n=document.createElement("script");n.async=!0,n.setAttribute("src","https://cdn.getkoala.com/v1/pk_d9bb6290ccb8a01b2d181fc0c8cf0dbb9836/sdk.js"),(document.body || document.head).appendChild(n)}();
 
 // New items: add 'new-item' class for for new items.
 document.addEventListener('DOMContentLoaded', () => {
diff --git a/sky/client/cli/command.py b/sky/client/cli/command.py
index 745c2a19c..c8185bff0 100644
--- a/sky/client/cli/command.py
+++ b/sky/client/cli/command.py
@@ -151,13 +151,17 @@ def _get_cluster_records_and_set_ssh_config(
         if isinstance(handle.launched_resources.cloud, clouds.Kubernetes):
             # Replace the proxy command to proxy through the SkyPilot API
             # server with websocket.
-            key_path = (cluster_utils.SSHConfigHelper.generate_local_key_file(
-                handle.cluster_name, credentials))
+            escaped_key_path = shlex.quote(
+                (cluster_utils.SSHConfigHelper.generate_local_key_file(
+                    handle.cluster_name, credentials)))
+            escaped_executable_path = shlex.quote(sys.executable)
+            escaped_websocket_proxy_path = shlex.quote(
+                f'{sky.__root_dir__}/templates/websocket_proxy.py')
             # Instead of directly use websocket_proxy.py, we add an
             # additional proxy, so that ssh can use the head pod in the
             # cluster to jump to worker pods.
             proxy_command = (
-                f'ssh -tt -i {key_path} '
+                f'ssh -tt -i {escaped_key_path} '
                 '-o StrictHostKeyChecking=no '
                 '-o UserKnownHostsFile=/dev/null '
                 '-o IdentitiesOnly=yes '
@@ -167,10 +171,10 @@ def _get_cluster_records_and_set_ssh_config(
                 # TODO(zhwu): write the template to a temp file, don't use
                 # the one in skypilot repo, to avoid changing the file when
                 # updating skypilot.
-                f'\'{sys.executable} {sky.__root_dir__}/templates/'
-                f'websocket_proxy.py '
+                f'\"{escaped_executable_path} '
+                f'{escaped_websocket_proxy_path} '
                 f'{server_common.get_server_url()} '
-                f'{handle.cluster_name}\'')
+                f'{handle.cluster_name}\"')
             credentials['ssh_proxy_command'] = proxy_command
 
         cluster_utils.SSHConfigHelper.add_cluster(
diff --git a/sky/dashboard/src/components/clusters.jsx b/sky/dashboard/src/components/clusters.jsx
index 93d5be93f..c4e71acfc 100755
--- a/sky/dashboard/src/components/clusters.jsx
+++ b/sky/dashboard/src/components/clusters.jsx
@@ -80,14 +80,14 @@ const PROPERTY_OPTIONS = [
     label: 'User',
     value: 'user',
   },
-  {
-    label: 'Infra',
-    value: 'infra',
-  },
   {
     label: 'Workspace',
     value: 'workspace',
   },
+  {
+    label: 'Infra',
+    value: 'infra',
+  },
 ];
 
 // Helper function to format autostop information, similar to _get_autostop in CLI utils
@@ -683,7 +683,7 @@ export function ClusterTable({
   return (
     <div>
       <Card>
-        <div className="overflow-x-auto">
+        <div className="overflow-x-auto rounded-lg">
           <Table className="min-w-full">
             <TableHeader>
               <TableRow>
@@ -1073,45 +1073,64 @@ const FilterDropdown = ({
   placeholder = 'Filter clusters',
 }) => {
   const inputRef = useRef(null);
+  const dropdownRef = useRef(null);
 
   const [isOpen, setIsOpen] = useState(false);
   const [value, setValue] = useState('');
   const [propertyValue, setPropertValue] = useState('status');
   const [valueOptions, setValueOptions] = useState([]);
 
-  let filteredOptions = [];
-
+  // Handle clicks outside the dropdown
   useEffect(() => {
-    let updatedValueOptions = [];
-
-    switch (propertyValue) {
-      case 'status':
-        updatedValueOptions = valueList.status;
-        break;
-
-      case 'user':
-        updatedValueOptions = valueList.user;
-        break;
-
-      case 'cluster':
-        updatedValueOptions = valueList.cluster;
-        break;
+    const handleClickOutside = (event) => {
+      if (
+        dropdownRef.current &&
+        !dropdownRef.current.contains(event.target) &&
+        inputRef.current &&
+        !inputRef.current.contains(event.target)
+      ) {
+        setIsOpen(false);
+      }
+    };
 
-      case 'workspace':
-        updatedValueOptions = valueList.workspace;
-        break;
+    document.addEventListener('mousedown', handleClickOutside);
+    return () => {
+      document.removeEventListener('mousedown', handleClickOutside);
+    };
+  }, []);
 
-      case 'infra':
-        updatedValueOptions = valueList.infra;
-        break;
+  useEffect(() => {
+    let updatedValueOptions = [];
 
-      default:
-        break;
+    if (valueList && typeof valueList === 'object') {
+      switch (propertyValue) {
+        case 'status':
+          updatedValueOptions = valueList.status || [];
+          break;
+        case 'user':
+          updatedValueOptions = valueList.user || [];
+          break;
+        case 'cluster':
+          updatedValueOptions = valueList.cluster || [];
+          break;
+        case 'workspace':
+          updatedValueOptions = valueList.workspace || [];
+          break;
+        case 'infra':
+          updatedValueOptions = valueList.infra || [];
+          break;
+        default:
+          break;
+      }
     }
 
-    updatedValueOptions = updatedValueOptions.filter((item) =>
-      item.includes(value.toLowerCase())
-    );
+    // Filter options based on current input value
+    if (value.trim() !== '') {
+      updatedValueOptions = updatedValueOptions.filter(
+        (item) =>
+          item && item.toString().toLowerCase().includes(value.toLowerCase())
+      );
+    }
 
     setValueOptions(updatedValueOptions);
   }, [propertyValue, valueList, value]);
@@ -1126,123 +1145,136 @@ const FilterDropdown = ({
 
   const handleValueChange = (e) => {
     setValue(e.target.value);
+    if (!isOpen) {
+      setIsOpen(true);
+    }
+  };
+
+  const handleInputFocus = () => {
+    setIsOpen(true);
+  };
+
+  const handleOptionSelect = (option) => {
+    setFilters((prevFilters) => {
+      const updatedFilters = [
+        ...prevFilters,
+        {
+          property: getPropertyLabel(propertyValue),
+          operator: ':',
+          value: option,
+        },
+      ];
+
+      updateURLParams(updatedFilters);
+      return updatedFilters;
+    });
+    setIsOpen(false);
+    setValue('');
+    inputRef.current.focus();
+  };
+
+  const handleKeyDown = (e) => {
+    if (e.key === 'Enter' && value.trim() !== '') {
+      setFilters((prevFilters) => {
+        const updatedFilters = [
+          ...prevFilters,
+          {
+            property: getPropertyLabel(propertyValue),
+            operator: ':',
+            value: value,
+          },
+        ];
+
+        updateURLParams(updatedFilters);
+        return updatedFilters;
+      });
+      setValue('');
+      setIsOpen(false);
+    } else if (e.key === 'Escape') {
+      setIsOpen(false);
+      inputRef.current.blur();
+    }
   };
 
   return (
-    <>
-      <div className="flex flex-row ml-4 mr-2 border border-gray-300 rounded-md overflow-hidden">
-        <div className="border-r border-gray-300">
-          <Select
-            onValueChange={(value) => setPropertValue(value)}
-            value={propertyValue}
+    <div className="flex flex-row ml-4 mr-2 border border-gray-300 rounded-md overflow-visible">
+      <div className="border-r border-gray-300">
+        <Select onValueChange={setPropertValue} value={propertyValue}>
+          <SelectTrigger
+            aria-label="Filter Property"
+            className="focus:ring-0 focus:ring-offset-0 border-none rounded-l-md rounded-r-none w-32 h-8"
           >
-            <SelectTrigger
-              aria-label="Node"
-              className="focus:ring-0 focus:ring-offset-0 border-none rounded-none w-32 h-8"
-            >
-              <SelectValue placeholder="Select Property" />
-            </SelectTrigger>
-            <SelectContent>
-              {propertyList.map((item, index) => (
-                <SelectItem key={`property-item-${index}`} value={item.value}>
-                  {item.label}
-                </SelectItem>
-              ))}
-            </SelectContent>
-          </Select>
-        </div>
-        <div className="relative flex-1">
-          <input
-            type="text"
-            ref={inputRef}
-            placeholder={placeholder}
-            value={value}
-            onChange={(e) => handleValueChange(e)}
-            onFocus={() => setIsOpen(true)}
-            onBlur={() => setIsOpen(false)}
-            onKeyDown={(e) => {
-              if (e.key === 'Enter' && value !== '') {
-                setFilters((prevFilters) => {
-                  const updatedFilters = [
-                    ...prevFilters,
-                    {
-                      property: getPropertyLabel(propertyValue),
-                      operator: ':',
-                      value: value,
-                    },
-                  ];
-
-                  updateURLParams(updatedFilters);
-
-                  return updatedFilters;
-                });
-                setValue('');
-                setIsOpen(false);
-                inputRef.current.blur();
-              }
+            <SelectValue placeholder="Select Property" />
+          </SelectTrigger>
+          <SelectContent>
+            {propertyList.map((item, index) => (
+              <SelectItem key={`property-item-${index}`} value={item.value}>
+                {item.label}
+              </SelectItem>
+            ))}
+          </SelectContent>
+        </Select>
+      </div>
+      <div className="relative flex-1">
+        <input
+          type="text"
+          ref={inputRef}
+          placeholder={placeholder}
+          value={value}
+          onChange={handleValueChange}
+          onFocus={handleInputFocus}
+          onKeyDown={handleKeyDown}
+          className="h-8 w-32 sm:w-96 px-3 pr-8 text-sm border-none rounded-l-none rounded-r-md focus:ring-0 focus:outline-none"
+          autoComplete="off"
+        />
+        {value && (
+          <button
+            onClick={() => {
+              setValue('');
+              setIsOpen(false);
             }}
-            className="h-8 w-32 sm:w-96 px-3 pr-8 text-sm border-none rounded-none focus:ring-0 focus:outline-none"
-            autoComplete="off"
-          />
-          {value && (
-            <button
-              onClick={() => {
-                setValue('');
-              }}
-              className="absolute right-2 top-1/2 transform -translate-y-1/2 text-gray-400 hover:text-gray-600"
-              title="Clear filter"
-              tabIndex={-1}
+            className="absolute right-2 top-1/2 transform -translate-y-1/2 text-gray-400 hover:text-gray-600"
+            title="Clear filter"
+            tabIndex={-1}
+          >
+            <svg
+              className="h-4 w-4"
+              fill="none"
+              stroke="currentColor"
+              viewBox="0 0 24 24"
             >
-              <svg
-                className="h-4 w-4"
-                fill="none"
-                stroke="currentColor"
-                viewBox="0 0 24 24"
+              <path
+                strokeLinecap="round"
+                strokeLinejoin="round"
+                strokeWidth={2}
+                d="M6 18L18 6M6 6l12 12"
+              />
+            </svg>
+          </button>
+        )}
+        {isOpen && valueOptions.length > 0 && (
+          <div
+            ref={dropdownRef}
+            className="absolute z-50 mt-1 w-full bg-white border border-gray-200 rounded-md shadow-lg max-h-60 overflow-y-auto"
+            style={{ zIndex: 9999 }}
+          >
+            {valueOptions.map((option, index) => (
+              <div
+                key={`${option}-${index}`}
+                className={`px-3 py-2 cursor-pointer hover:bg-gray-50 text-sm ${
+                  index !== valueOptions.length - 1
+                    ? 'border-b border-gray-100'
+                    : ''
+                }`}
+                onClick={() => handleOptionSelect(option)}
               >
-                <path
-                  strokeLinecap="round"
-                  strokeLinejoin="round"
-                  strokeWidth={2}
-                  d="M6 18L18 6M6 6l12 12"
-                />
-              </svg>
-            </button>
-          )}
-          {isOpen && valueOptions.length > 0 && (
-            <div className="flex flex-col absolute z-10 mt-1 w-full bg-white border border-gray-200 rounded-md shadow-lg">
-              {valueOptions.map((option, index) => (
-                <div
-                  key={option}
-                  className={`flex flex-col pl-7 py-2 cursor-pointer hover:bg-sky-50 text-sm ${index != filteredOptions.length - 1} && border-b`}
-                  onMouseDown={(e) => {
-                    e.preventDefault();
-                    setFilters((prevFilters) => {
-                      const updatedFilters = [
-                        ...prevFilters,
-                        {
-                          property: getPropertyLabel(propertyValue),
-                          operator: '=',
-                          value: option,
-                        },
-                      ];
-
-                      updateURLParams(updatedFilters);
-
-                      return updatedFilters;
-                    });
-                    setIsOpen(false);
-                    setValue('');
-                    inputRef.current.blur();
-                  }}
-                >
-                  <span className="text-sm text-gray-600">{option}</span>
-                </div>
-              ))}
-            </div>
-          )}
-        </div>
+                <span className="text-sm text-gray-700">{option}</span>
+              </div>
+            ))}
+          </div>
+        )}
       </div>
-    </>
+    </div>
   );
 };
 
@@ -1266,7 +1298,7 @@ const Filters = ({ filters = [], setFilters, updateURLParams }) => {
 
   return (
     <>
-      <div className="flex items-center gap-4 py-1 px-2">
+      <div className="flex items-center gap-4 py-2 px-2">
         <div className="flex flex-wrap items-content gap-2">
           {filters.map((filter, _index) => (
             <FilterItem
@@ -1295,8 +1327,8 @@ const Filters = ({ filters = [], setFilters, updateURLParams }) => {
 const FilterItem = ({ filter, onRemove }) => {
   return (
     <>
-      <div className="flex items-center gap-1 text-blue-600 bg-blue-100 px-1 py-1 rounded-full text-sm">
-        <div className="flex items-center gap-1 px-2 rounded-l-lg">
+      <div className="flex items-center text-blue-600 bg-blue-100 px-1 py-1 rounded-full text-sm">
+        <div className="flex items-center gap-1 px-2">
           <span>{`${filter.property} `}</span>
           <span>{`${filter.operator} `}</span>
           <span>{` ${filter.value}`}</span>
@@ -1304,7 +1336,7 @@ const FilterItem = ({ filter, onRemove }) => {
 
         <button
           onClick={() => onRemove()}
-          className="p-0.5 transform text-gray-400 hover:text-gray-600 bg-blue-500 hover:bg-blue-600 rounded-full flex flex-col items-center"
+          className="p-0.5 ml-1 transform text-gray-400 hover:text-gray-600 bg-blue-500 hover:bg-blue-600 rounded-full flex flex-col items-center"
           title="Clear filter"
         >
           <svg
diff --git a/sky/server/requests/executor.py b/sky/server/requests/executor.py
index ace00920a..9a8bb1b93 100644
--- a/sky/server/requests/executor.py
+++ b/sky/server/requests/executor.py
@@ -176,7 +176,7 @@ class RequestWorker:
             if request_element is None:
                 time.sleep(0.1)
                 return
-            request_id, ignore_return_value, retryable = request_element
+            request_id, ignore_return_value, _ = request_element
             request = api_requests.get_request(request_id)
             assert request is not None, f'Request with ID {request_id} is None'
             if request.status == api_requests.RequestStatus.CANCELLED:
@@ -190,12 +190,10 @@ class RequestWorker:
             # the process, such as subprocess_daemon.py.
             fut = executor.submit_until_success(_request_execution_wrapper,
                                                 request_id, ignore_return_value)
-            if retryable:
-                # If the task might fail and be retried, start a thread to
-                # monitor the future and process retry.
-                threading.Thread(target=self.handle_task_result,
-                                 args=(fut, request_element),
-                                 daemon=True).start()
+            # Monitor the result of the request execution.
+            threading.Thread(target=self.handle_task_result,
+                             args=(fut, request_element),
+                             daemon=True).start()
 
             logger.info(f'[{self}] Submitted request: {request_id}')
         except (Exception, SystemExit) as e:  # pylint: disable=broad-except
@@ -209,6 +207,21 @@ class RequestWorker:
                            request_element: Tuple[str, bool, bool]) -> None:
         try:
             fut.result()
+        except concurrent.futures.process.BrokenProcessPool as e:
+            # Happens when the worker process dies unexpectedly, e.g. OOM
+            # killed.
+            request_id, _, retryable = request_element
+            # Ensure the request status.
+            api_requests.set_request_failed(request_id, e)
+            logger.error(
+                f'Request {request_id} failed to get processed '
+                f'{common_utils.format_exception(e, use_bracket=True)}')
+            if retryable:
+                # If the request is retryable and disrupted by broken
+                # process pool, reschedule it immediately to get it
+                # retried in the new process pool.
+                queue = _get_queue(self.schedule_type)
+                queue.put(request_element)
         except exceptions.ExecutionRetryableError as e:
             time.sleep(e.retry_wait_seconds)
             # Reschedule the request.
diff --git a/sky/server/requests/process.py b/sky/server/requests/process.py
index a8a17eb2e..1c0a74258 100644
--- a/sky/server/requests/process.py
+++ b/sky/server/requests/process.py
@@ -202,6 +202,9 @@ class BurstableExecutor:
                  burst_workers: int = 0,
                  **kwargs):
         if garanteed_workers > 0:
+            self._guaranteed_workers = garanteed_workers
+            self._guaranteed_pool_kwargs = kwargs
+            self._guaranteed_pool_lock = threading.Lock()
             self._executor = PoolExecutor(max_workers=garanteed_workers,
                                           **kwargs)
         if burst_workers > 0:
@@ -225,8 +228,7 @@ class BurstableExecutor:
 
         while True:
             if self._executor is not None and self._executor.has_idle_workers():
-                logger.info('Submitting to the guaranteed pool')
-                return self._executor.submit(fn, *args, **kwargs)
+                return self._submit_to_guaranteed_pool(fn, *args, **kwargs)
             if (self._burst_executor is not None and
                     self._burst_executor.has_idle_workers()):
                 try:
@@ -238,11 +240,52 @@ class BurstableExecutor:
             if self._executor is not None:
                 # No idle workers in either pool, still queue the request
                 # to the guaranteed pool to keep behavior consistent.
-                return self._executor.submit(fn, *args, **kwargs)
+                return self._submit_to_guaranteed_pool(fn, *args, **kwargs)
             logger.debug('No guaranteed pool set and the burst pool is full, '
                          'retry later.')
             time.sleep(0.1)
 
+    def _submit_to_guaranteed_pool(self, fn, *args,
+                                   **kwargs) -> concurrent.futures.Future:
+        """Submit to the guaranteed pool with retry.
+
+        The concurrent.futures.ProcessPoolExecutor will terminate all child
+        processes and reject new tasks if any of the child process died, e.g.
+        OOM killed. So when the pool is broken, we have to rebuild a new pool
+        to execute tasks.
+
+        Alternatives considered:
+        - multiprocessing.Pool: individual process failure does not affect the
+            pool, but it lacks of lazy-init support and does not handle process
+            failure gracefully.
+        - inherit from concurrent.futures.ProcessPoolExecutor: If a child
+            process dies when executing a task, we have to set an exception
+            in the Future object of the task. But there is a race condition
+            that the child process may dies exactly after get the task from
+            queue so there is no (obviously) safe way to record process PID
+            to task mapping in the current architecture of
+            ProcessPoolExecutor. It is more feasible to implement a custom
+            ProcessPoolExecutor with an new architecture.
+        - a custom ProcessPoolExecutor: non-trivial to implement, keep for
+            future improvement as BrokenProcessPool is not expected to be
+            a common case, i.e. it usually indicates there a bug elsewhere
+            when this happens.
+        """
+        with self._guaranteed_pool_lock:
+            assert self._executor is not None
+            try:
+                return self._executor.submit(fn, *args, **kwargs)
+            except concurrent.futures.process.BrokenProcessPool as e:
+                logger.warning('The guaranteed pool is broken, '
+                               f'replacing the pool and retrying. Error: {e}')
+                broken_pool = self._executor
+                threading.Thread(target=broken_pool.shutdown,
+                                 daemon=True).start()
+                self._executor = PoolExecutor(
+                    max_workers=self._guaranteed_workers,
+                    **self._guaranteed_pool_kwargs)
+        return self._submit_to_guaranteed_pool(fn, *args, **kwargs)
+
     def shutdown(self) -> None:
         """Shutdown the executor."""
 
diff --git a/sky/server/server.py b/sky/server/server.py
index 6294af62e..b0a1f8d4f 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -491,6 +491,9 @@ async def lifespan(app: fastapi.FastAPI):  # pylint: disable=redefined-outer-nam
                 func=event.run_event,
                 schedule_type=requests_lib.ScheduleType.SHORT,
                 is_skypilot_system=True,
+                # Request deamon should be retried if the process pool is
+                # broken.
+                retryable=True,
             )
         except exceptions.RequestAlreadyExistsError:
             # Lifespan will be executed in each uvicorn worker process, we
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index 70a80ea1e..9b9a90e36 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -956,8 +956,9 @@ class KubernetesCommandRunner(CommandRunner):
             return remote_home_dir
 
         # Build command.
-        helper_path = os.path.join(os.path.abspath(os.path.dirname(__file__)),
-                                   'kubernetes', 'rsync_helper.sh')
+        helper_path = shlex.quote(
+            os.path.join(os.path.abspath(os.path.dirname(__file__)),
+                         'kubernetes', 'rsync_helper.sh'))
         namespace_context = f'{self.namespace}+{self.context}'
         # Avoid rsync interpreting :, /, and + in namespace_context as the
         # default delimiter for options and arguments.


Test patch:
diff --git a/tests/unit_tests/test_sky/server/requests/test_process.py b/tests/unit_tests/test_sky/server/requests/test_process.py
index 494cc810d..591beefca 100644
--- a/tests/unit_tests/test_sky/server/requests/test_process.py
+++ b/tests/unit_tests/test_sky/server/requests/test_process.py
@@ -1,6 +1,8 @@
 """Unit tests for sky/server/requests/process.py."""
 from concurrent.futures import Future
+import concurrent.futures.process
 import time
+import unittest.mock
 
 import pytest
 
@@ -153,3 +155,50 @@ def test_burstable_executor_no_burst():
         executor.submit_until_success(dummy_task)
     finally:
         executor.shutdown()
+
+
+def test_burstable_executor_pool_recovery():
+    """Test BurstableExecutor recovery from BrokenProcessPool exception."""
+
+    executor = BurstableExecutor(garanteed_workers=1, burst_workers=0)
+
+    try:
+        # Store reference to original executor
+        original_executor = executor._executor
+        submit_call_count = 0
+
+        # Mock the PoolExecutor.submit method at class level to control
+        # behavior across all instances
+        original_submit = PoolExecutor.submit
+
+        def mock_submit(self, fn, *args, **kwargs):
+            nonlocal submit_call_count
+            submit_call_count += 1
+            if submit_call_count == 1:
+                # First call raises BrokenProcessPool to simulate pool failure
+                raise concurrent.futures.process.BrokenProcessPool(
+                    "Simulated process pool failure")
+            else:
+                # Subsequent calls should work normally
+                # Call the original submit method
+                return original_submit(self, fn, *args, **kwargs)
+
+        with unittest.mock.patch.object(PoolExecutor, 'submit',
+                                        new=mock_submit):
+            # This should trigger the pool recovery logic in
+            # _submit_to_guaranteed_pool
+            future = executor.submit_until_success(dummy_task)
+            result = future.result(timeout=5.0)
+
+            # Verify the task completed successfully despite initial failure
+            assert result is True
+
+            # Verify that submit was called at least twice
+            # (initial failure + successful retry)
+            assert submit_call_count >= 2
+
+            # Verify that a new executor was created after the failure
+            assert executor._executor is not original_executor
+
+    finally:
+        executor.shutdown()