Repository: skypilot-org/skypilot

Issue:
decode_rich_status can chunk in the middle of utf-8 sequences
On some occasions my logs are not retrievable and they end with:
```
File "/home/user/.keybotic/venvs/skypilot/bin/sky", line 8, in <module>
    sys.exit(cli())
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/click/core.py", line 1161, in __call__
    return self.main(*args, **kwargs)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/click/core.py", line 1082, in main
    rv = self.invoke(ctx)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/utils/common_utils.py", line 434, in _record
    return f(*args, **kwargs)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/cli.py", line 875, in invoke
    return super().invoke(ctx)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/click/core.py", line 1697, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/utils/common_utils.py", line 434, in _record
    return f(*args, **kwargs)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/cli.py", line 875, in invoke
    return super().invoke(ctx)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/click/core.py", line 1697, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/click/core.py", line 1443, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/click/core.py", line 788, in invoke
    return __callback(*args, **kwargs)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/utils/common_utils.py", line 454, in _record
    return f(*args, **kwargs)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/cli.py", line 4092, in jobs_logs
    managed_jobs.tail_logs(name=name,
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/utils/common_utils.py", line 454, in _record
    return f(*args, **kwargs)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/server/common.py", line 285, in wrapper
    return func(*args, **kwargs)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/jobs/client/sdk.py", line 214, in tail_logs
    sdk.stream_response(request_id, response, output_stream)
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/client/sdk.py", line 72, in stream_response
    for line in rich_utils.decode_rich_status(response):
  File "/home/user/.keybotic/venvs/skypilot/lib/python3.8/site-packages/sky/utils/rich_utils.py", line 239, in decode_rich_status
    encoded_msg = encoded_msg.decode('utf-8')
UnicodeDecodeError: 'utf-8' codec can't decode bytes in position 3962-3963: unexpected end of data
```

Printing the offending log buffer:
```
b'  0.9255      1.133      3.474      1.101         22        640:  20%|\xe2\x96\x88\xe2\x96\x89        | 115/581 [00:12<00:41, 11.20it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9236      1.129      3.474        1.1         17        640:  20%|\xe2\x96\x88\xe2\x96\x89        | 115/581 [00:12<00:41, 11.20it/s]\n       1/11      2.75G     0.9209      1.126       3.47      1.098         23        640:  20%|\xe2\x96\x88\xe2\x96\x89        | 115/581 [00:13<00:41, 11.20it/s]\n       1/11      2.75G     0.9209      1.126       3.47      1.098         23        640:  20%|\xe2\x96\x88\xe2\x96\x88        | 117/581 [00:13<00:52,  8.81it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9185       1.12      3.469      1.097         16        640:  20%|\xe2\x96\x88\xe2\x96\x88        | 117/581 [00:13<00:52,  8.81it/s]\n       1/11      2.75G     0.9161      1.118      3.466      1.096         27        640:  20%|\xe2\x96\x88\xe2\x96\x88        | 117/581 [00:13<00:52,  8.81it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9161      1.118      3.466      1.096         27        640:  20%|\xe2\x96\x88\xe2\x96\x88        | 119/581 [00:13<00:47,  9.83it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9141      1.115      3.462      1.094         27        640:  20%|\xe2\x96\x88\xe2\x96\x88        | 119/581 [00:13<00:47,  9.83it/s]\n       1/11      2.75G     0.9114       1.11      3.457      1.092         28        640:  20%|\xe2\x96\x88\xe2\x96\x88        | 119/581 [00:13<00:47,  9.83it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9114       1.11      3.457      1.092         28        640:  21%|\xe2\x96\x88\xe2\x96\x88        | 121/581 [00:13<00:42, 10.70it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9075      1.106      3.453       1.09         20        640:  21%|\xe2\x96\x88\xe2\x96\x88        | 121/581 [00:13<00:42, 10.70it/s]\n       1/11      2.75G     0.9053      1.103      3.448      1.089         19        640:  21%|\xe2\x96\x88\xe2\x96\x88        | 121/581 [00:13<00:42, 10.70it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9053      1.103      3.448      1.089         19        640:  21%|\xe2\x96\x88\xe2\x96\x88        | 123/581 [00:13<00:41, 11.09it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9025      1.098      3.444      1.088         18        640:  21%|\xe2\x96\x88\xe2\x96\x88        | 123/581 [00:13<00:41, 11.09it/s]\n       1/11      2.75G     0.9001      1.102       3.44      1.086         22        640:  21%|\xe2\x96\x88\xe2\x96\x88        | 123/581 [00:14<00:41, 11.09it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.9001      1.102       3.44      1.086         22        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 125/581 [00:14<00:57,  7.99it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.8983      1.097      3.438      1.084         22        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 125/581 [00:14<00:57,  7.99it/s]\n       1/11      2.75G     0.8974      1.095      3.433      1.083         21        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 125/581 [00:14<00:57,  7.99it/s]\n       1/11      2.75G     0.8974      1.095      3.433      1.083         21        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 127/581 [00:14<00:49,  9.12it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.8968      1.095       3.43      1.081         25        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 127/581 [00:14<00:49,  9.12it/s]\n       1/11      2.75G     0.8946      1.092      3.426       1.08         23        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 127/581 [00:14<00:49,  9.12it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.8946      1.092      3.426       1.08         23        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 129/581 [00:14<00:45, 10.04it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.8923      1.088      3.422      1.078         25        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 129/581 [00:14<00:45, 10.04it/s]\n       1/11      2.75G     0.8907      1.086      3.417      1.077         33        640:  22%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x8f       | 129/581 [00:14<00:45, 10.04it/s]\n\x1b[36m(TestSky, pid=2952)\x1b[0m        1/11      2.75G     0.8907      1.086      3.417      1.077         33        640:  23%|\xe2\x96'
```

I believe the chunking is splitting a multibyte utf-8 character or something.

Discussion:

Patch:
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index 4db8c612..c786ce35 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -183,6 +183,8 @@ Supported bucket types:
 
 Configure resources for the managed jobs controller.
 
+For more details about tuning the jobs controller resources, see :ref:`jobs-controller-sizing`.
+
 Example:
 
 .. code-block:: yaml
@@ -190,10 +192,13 @@ Example:
   jobs:
     controller:
       resources:  # same spec as 'resources' in a task YAML
+        # optionally set specific cloud/region
         cloud: gcp
         region: us-central1
-        cpus: 4+  # number of vCPUs, max concurrent spot jobs = 2 * cpus
-        disk_size: 100
+        # default resources:
+        cpus: 4+
+        memory: 8x
+        disk_size: 50
 
 .. _config-yaml-allowed-clouds:
 
diff --git a/sky/jobs/server/core.py b/sky/jobs/server/core.py
index 12db6c27..6a0666b8 100644
--- a/sky/jobs/server/core.py
+++ b/sky/jobs/server/core.py
@@ -157,6 +157,7 @@ def launch(
             'modified_catalogs':
                 service_catalog_common.get_modified_catalog_file_mounts(),
             'dashboard_setup_cmd': managed_job_constants.DASHBOARD_SETUP_CMD,
+            'dashboard_user_id': common.SERVER_ID,
             **controller_utils.shared_controller_vars_to_fill(
                 controller_utils.Controllers.JOBS_CONTROLLER,
                 remote_user_config_path=remote_user_config_path,
@@ -305,10 +306,9 @@ def _maybe_restart_controller(
     with rich_utils.safe_status(
             ux_utils.spinner_message('Starting dashboard...')):
         runner = handle.get_command_runners()[0]
-        user_hash = common_utils.get_user_hash()
         runner.run(
             f'export '
-            f'{skylet_constants.USER_ID_ENV_VAR}={user_hash!r}; '
+            f'{skylet_constants.USER_ID_ENV_VAR}={common.SERVER_ID!r}; '
             f'{managed_job_constants.DASHBOARD_SETUP_CMD}',
             stream_logs=True,
         )
diff --git a/sky/jobs/server/server.py b/sky/jobs/server/server.py
index b95ca66d..18d309b4 100644
--- a/sky/jobs/server/server.py
+++ b/sky/jobs/server/server.py
@@ -159,6 +159,9 @@ async def dashboard(request: fastapi.Request,
             except Exception as e:  # pylint: disable=broad-except
                 # We catch all exceptions to gracefully handle unknown
                 # errors and retry or raise an HTTPException to the client.
+                # Assume an exception indicates that the dashboard connection
+                # is stale - remove it so that a new one is created.
+                dashboard_utils.remove_dashboard_session(user_hash)
                 msg = (
                     f'Dashboard connection attempt {attempt + 1} failed with '
                     f'{common_utils.format_exception(e, use_bracket=True)}')
diff --git a/sky/templates/jobs-controller.yaml.j2 b/sky/templates/jobs-controller.yaml.j2
index 7f78a397..4f154ba6 100644
--- a/sky/templates/jobs-controller.yaml.j2
+++ b/sky/templates/jobs-controller.yaml.j2
@@ -39,9 +39,7 @@ setup: |
   After=network.target
 
   [Service]
-  Environment="PATH={{ sky_python_env_path }}:\$PATH"
-  Environment="SKYPILOT_USER_ID={{controller_envs.SKYPILOT_USER_ID}}"
-  Environment="SKYPILOT_USER={{controller_envs.SKYPILOT_USER}}"
+  Environment="SKYPILOT_USER_ID={{ dashboard_user_id }}"
   Restart=always
   StandardOutput=append:/home/$USER/.sky/job-dashboard.log
   StandardError=append:/home/$USER/.sky/job-dashboard.log
@@ -51,6 +49,7 @@ setup: |
   WantedBy=default.target
   EOF
 
+  export SKYPILOT_USER_ID="{{ dashboard_user_id }}"
   {{ dashboard_setup_cmd }}
 
 run: |
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index b30f4a52..be2e710c 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -531,8 +531,8 @@ class SSHCommandRunner(CommandRunner):
         if port_forward is not None:
             for local, remote in port_forward:
                 logger.info(
-                    f'Forwarding port {local} to port {remote} on localhost.')
-                ssh += ['-NL', f'{remote}:localhost:{local}']
+                    f'Forwarding local port {local} to remote port {remote}.')
+                ssh += ['-NL', f'{local}:localhost:{remote}']
         if self._docker_ssh_proxy_command is not None:
             docker_ssh_proxy_command = self._docker_ssh_proxy_command(ssh)
         else:
diff --git a/sky/utils/rich_utils.py b/sky/utils/rich_utils.py
index 54e9672f..2f4dae30 100644
--- a/sky/utils/rich_utils.py
+++ b/sky/utils/rich_utils.py
@@ -230,15 +230,53 @@ def decode_rich_status(
     decoding_status = None
     try:
         last_line = ''
+        # Buffer to store incomplete UTF-8 bytes between chunks
+        undecoded_buffer = b''
+
         # Iterate over the response content in chunks. We do not use iter_lines
         # because it will strip the trailing newline characters, causing the
         # progress bar ending with `\r` becomes a pyramid.
-        for encoded_msg in response.iter_content(chunk_size=None):
-            if encoded_msg is None:
+        for chunk in response.iter_content(chunk_size=None):
+            if chunk is None:
                 return
-            encoded_msg = encoded_msg.decode('utf-8')
+
+            # Append the new chunk to any leftover bytes from previous iteration
+            current_bytes = undecoded_buffer + chunk
+            undecoded_buffer = b''
+
+            # Try to decode the combined bytes
+            try:
+                encoded_msg = current_bytes.decode('utf-8')
+            except UnicodeDecodeError as e:
+                # Check if this is potentially an incomplete sequence at the end
+                if e.start > 0:
+                    # Decode the valid part
+                    encoded_msg = current_bytes[:e.start].decode('utf-8')
+
+                    # Check if the remaining bytes are likely a partial char
+                    # or actually invalid UTF-8
+                    remaining_bytes = current_bytes[e.start:]
+                    if len(remaining_bytes) < 4:  # Max UTF-8 char is 4 bytes
+                        # Likely incomplete - save for next chunk
+                        undecoded_buffer = remaining_bytes
+                    else:
+                        # Likely invalid - replace with replacement character
+                        encoded_msg += remaining_bytes.decode('utf-8',
+                                                              errors='replace')
+                        undecoded_buffer = b''
+                else:
+                    # Error at the very beginning of the buffer - invalid UTF-8
+                    encoded_msg = current_bytes.decode('utf-8',
+                                                       errors='replace')
+                    undecoded_buffer = b''
+
             lines = encoded_msg.splitlines(keepends=True)
 
+            # Skip processing if lines is empty to avoid IndexError
+            if not lines:
+                continue
+
+            # Append any leftover text from previous chunk to first line
             lines[0] = last_line + lines[0]
             last_line = lines[-1]
             # If the last line is not ended with `\r` or `\n` (with ending


Test patch:
diff --git a/tests/unit_tests/sky/utils/test_rich_utils.py b/tests/unit_tests/sky/utils/test_rich_utils.py
new file mode 100644
index 00000000..36239cb3
--- /dev/null
+++ b/tests/unit_tests/sky/utils/test_rich_utils.py
@@ -0,0 +1,83 @@
+import unittest
+from unittest import mock
+
+import requests
+
+from sky.utils import rich_utils
+
+
+class RichUtilsTest(unittest.TestCase):
+
+    def test_decode_rich_status_with_split_utf8(self):
+        """Test that decode_rich_status handles UTF-8 characters split across chunks."""
+        # This is a progress bar with UTF-8 box drawing characters
+        # The character █ is a multibyte UTF-8 character (0xE2 0x96 0x88)
+        progress_bar = "20%|████        | 115/581 [00:12<00:41, 11.20it/s]\n"
+
+        # Create a response object with a custom content iterator
+        mock_response = mock.MagicMock(spec=requests.Response)
+
+        # Convert to bytes and deliberately split in the middle of a multibyte character
+        # The character █ is encoded as three bytes: \xe2\x96\x88
+        progress_bytes = progress_bar.encode('utf-8')
+
+        # Find position of the first █ character and split between its bytes
+        pos = progress_bytes.find(b'\xe2\x96\x88')
+
+        # Split the bytes so that \xe2 is in one chunk and \x96\x88 is in the next
+        chunk1 = progress_bytes[:pos + 1]  # Include the first byte of █ (\xe2)
+        chunk2 = progress_bytes[pos +
+                                1:]  # Rest of the bytes including \x96\x88
+
+        # Configure the mock to return these chunks when iterated
+        mock_response.iter_content.return_value = [chunk1, chunk2]
+
+        # The function should handle the split character correctly
+        result = list(rich_utils.decode_rich_status(mock_response))
+
+        # Verify the output contains the correctly decoded progress bar
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0], progress_bar)
+
+    def test_decode_rich_status_with_multiple_split_characters(self):
+        """Test handling multiple UTF-8 characters split across several chunks."""
+        # Progress bar with multiple multibyte characters
+        progress_bar = "50%|█████████████████████████████░░░░░░░░░░░░░| 250/500 [01:15<01:15, 3.33it/s]\n"
+        progress_bytes = progress_bar.encode('utf-8')
+
+        # Create chunks that split at several different points
+        # Find positions of several █ characters and split between their bytes
+        multibyte_char = '█'.encode('utf-8')  # \xe2\x96\x88
+        positions = []
+        start = 0
+        while True:
+            pos = progress_bytes.find(multibyte_char, start)
+            if pos == -1 or len(positions) >= 3:
+                break
+            positions.append(pos)
+            start = pos + len(multibyte_char)
+
+        # Create chunks that split at different multibyte character boundaries
+        chunks = []
+        last_pos = 0
+        for i, pos in enumerate(positions):
+            if i % 2 == 0:  # For even indices, split after first byte of multibyte char
+                chunks.append(progress_bytes[last_pos:pos + 1])
+                last_pos = pos + 1
+            else:  # For odd indices, split normally
+                chunks.append(progress_bytes[last_pos:pos])
+                last_pos = pos
+
+        # Add the final chunk
+        chunks.append(progress_bytes[last_pos:])
+
+        # Configure mock response
+        mock_response = mock.MagicMock(spec=requests.Response)
+        mock_response.iter_content.return_value = chunks
+
+        # Should handle multiple split characters correctly
+        result = list(rich_utils.decode_rich_status(mock_response))
+
+        # Verify the output
+        joined_result = ''.join(r for r in result if r is not None)
+        self.assertEqual(joined_result, progress_bar)