Move task status updates onto the RPC loop.

Author: rjpower

lib/finelog/src/finelog/deploy/cli.py

@@ -157,6 +157,7 @@ def status_cmd(name: str) -> None:
 class OutputFormat(StrEnum):
     TABLE = "table"
     JSON = "json"
+    JSONL = "jsonl"
     CSV = "csv"
 
 
@@ -192,6 +193,12 @@ def _print_json(table: pa.Table) -> None:
     sys.stdout.write("\n")
 
 
+def _print_jsonl(table: pa.Table) -> None:
+    for row in table.to_pylist():
+        json.dump(row, sys.stdout, default=str)
+        sys.stdout.write("\n")
+
+
 def _print_csv(table: pa.Table) -> None:
     writer = csv.writer(sys.stdout)
     writer.writerow(table.schema.names)
@@ -202,6 +209,7 @@ def _print_csv(table: pa.Table) -> None:
 _PRINTERS = {
     OutputFormat.TABLE: _print_table,
     OutputFormat.JSON: _print_json,
+    OutputFormat.JSONL: _print_jsonl,
     OutputFormat.CSV: _print_csv,
 }
 
@@ -213,7 +221,7 @@ def _print_csv(table: pa.Table) -> None:
     "--format",
     "output_format",
     type=click.Choice([f.value for f in OutputFormat]),
-    default=OutputFormat.TABLE.value,
+    default=OutputFormat.JSONL.value,
     show_default=True,
     help="Output format for the result.",
 )
@@ -399,7 +407,7 @@ def _register_namespace_views(
     "--format",
     "output_format",
     type=click.Choice([f.value for f in OutputFormat]),
-    default=OutputFormat.TABLE.value,
+    default=OutputFormat.JSONL.value,
     show_default=True,
     help="Output format for the result.",
 )

lib/iris/src/iris/cluster/controller/controller.py

@@ -1091,7 +1091,7 @@ class ControllerConfig:
     remote_state_dir: str = ""
     """Remote URI for controller checkpoints and worker profiles (e.g. gs://bucket/iris/state)."""
 
-    scheduler_min_interval: Duration = field(default_factory=lambda: Duration.from_seconds(1.0))
+    scheduler_min_interval: Duration = field(default_factory=lambda: Duration.from_seconds(10.0))
     """Minimum scheduling loop interval (used when cluster is active)."""
 
     scheduler_max_interval: Duration = field(default_factory=lambda: Duration.from_seconds(10.0))

lib/iris/src/iris/cluster/controller/dashboard.py

@@ -426,6 +426,8 @@ def _create_app(self) -> ASGIApp:
             # when present but treat everything as anonymous/admin.
             auth_interceptor = NullAuthInterceptor(verifier=self._auth_verifier)
         controller_interceptors = [auth_interceptor, controller_timing]
+        # @on_loop handlers run inline on the event loop; everything else
+        # is dispatched to a thread by AsyncServiceAdapter.
         rpc_asgi_app = ControllerServiceASGIApplication(
             service=AsyncServiceAdapter(self._service),
             interceptors=controller_interceptors,

lib/iris/src/iris/cluster/controller/service.py

@@ -90,6 +90,7 @@
     is_job_finished,
 )
 from iris.rpc import controller_pb2, job_pb2, query_pb2, vm_pb2, worker_pb2
+from iris.rpc.async_adapter import on_loop
 from iris.rpc.auth import (
     AuthzAction,
     authorize,
@@ -2589,6 +2590,7 @@ def update_task_status(
 
     # --- Task Status Text Push ---
 
+    @on_loop
     def set_task_status_text(
         self,
         request: job_pb2.SetTaskStatusTextRequest,

lib/iris/src/iris/rpc/async_adapter.py

@@ -6,9 +6,14 @@
 
 The ASGI application invokes ``await endpoint.function(...)`` for every
 unary RPC, so each handler must be a coroutine function.
-``AsyncServiceAdapter`` exposes a sync service's methods as async by
-wrapping each sync method in ``asyncio.to_thread``. Methods that are
-already coroutine functions pass through untouched.
+``AsyncServiceAdapter`` exposes a sync service's methods as async:
+
+- methods marked with :func:`on_loop` are wrapped in an ``async def`` that
+  calls the sync body inline on the event loop. Use this only for short,
+  non-blocking handlers (in-memory dicts, very cheap SQL reads) — a long
+  handler running inline blocks every other RPC.
+- every other sync method gets an ``asyncio.to_thread`` wrapper.
+- methods that are already coroutine functions pass through untouched.
 
 Interceptors are not adapted here — each interceptor that participates in
 an ASGI chain implements ``async intercept_unary`` directly.
@@ -19,7 +24,26 @@
 import asyncio
 import functools
 import inspect
-from typing import Any
+from collections.abc import Callable
+from typing import Any, TypeVar
+
+F = TypeVar("F", bound=Callable[..., Any])
+
+_ON_LOOP_ATTR = "__iris_rpc_on_loop__"
+
+
+def on_loop(fn: F) -> F:
+    """Mark a sync RPC handler as safe to run directly on the event loop.
+
+    The handler must be short and non-blocking. A handler that blocks the
+    loop for tens of milliseconds will queue every other RPC behind it.
+    """
+    setattr(fn, _ON_LOOP_ATTR, True)
+    return fn
+
+
+def _is_on_loop(fn: Callable[..., Any]) -> bool:
+    return getattr(fn, _ON_LOOP_ATTR, False)
 
 
 class AsyncServiceAdapter:
@@ -36,6 +60,13 @@ def __getattr__(self, name: str) -> Any:
             return attr
         if inspect.iscoroutinefunction(attr):
             return attr
+        if _is_on_loop(attr):
+
+            @functools.wraps(attr)
+            async def _on_loop_call(*args: Any, **kwargs: Any) -> Any:
+                return attr(*args, **kwargs)
+
+            return _on_loop_call
 
         @functools.wraps(attr)
         async def _threaded_call(*args: Any, **kwargs: Any) -> Any: