feat(server): server execution guarantees (time bounds + loud failures + contract base)

cookbook/client/async_rl/server_config.yaml

@@ -42,9 +42,6 @@ applications:
           target_ongoing_requests: 128
         ray_actor_options:
           num_cpus: 0.1
-          runtime_env:
-            env_vars:
-              TWINKLE_FAIL_FAST: "0"
 
   # TransferQueue-backed DataRef service.
   - name: data-plane
@@ -95,7 +92,6 @@ applications:
           runtime_env:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "1"
-              TWINKLE_FAIL_FAST: "0"
 
   # A second GPU hosts vLLM and loads the same local base model.
   - name: sampler-Qwen3.5-4B
@@ -133,7 +129,6 @@ applications:
           runtime_env:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "1"
-              TWINKLE_FAIL_FAST: "0"
 
   - name: processor
     route_prefix: /api/v1/processor
@@ -155,6 +150,3 @@ applications:
           target_ongoing_requests: 128
         ray_actor_options:
           num_cpus: 0.1
-          runtime_env:
-            env_vars:
-              TWINKLE_FAIL_FAST: "0"

cookbook/client/server/megatron/server_config.yaml

@@ -54,7 +54,6 @@ applications:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "0"
               TWINKLE_LONG_POLL_TIMEOUT: "120"
-              TWINKLE_FAIL_FAST: "0"
 
   # 3. Sampler Service - Runs inference / sampling using vLLM engine
   #    Used for generating text from the model (e.g., evaluating LoRA results).
@@ -98,7 +97,6 @@ applications:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "0"
               TWINKLE_LONG_POLL_TIMEOUT: "120"
-              TWINKLE_FAIL_FAST: "0"
 
   # 2. Model Service - Hosts the base model for training.
   #    Config: PP=2 x DP=2 on 4 GPUs, ~27GB weights/GPU, comfortable for LoRA training
@@ -139,4 +137,3 @@ applications:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "0"
               TWINKLE_LONG_POLL_TIMEOUT: "120"
-              TWINKLE_FAIL_FAST: "0"

cookbook/client/server/megatron/server_config_4b.yaml

@@ -31,9 +31,6 @@ applications:
           target_ongoing_requests: 128   # Target concurrent requests per replica
         ray_actor_options:
           num_cpus: 0.1                  # CPU resources allocated to this actor
-          runtime_env:
-            env_vars:
-              TWINKLE_FAIL_FAST: "0"
 
   # 2. Model Service (commented out) - Would host the base model for training.
   #    Uncomment and configure if you need a training model worker.
@@ -71,7 +68,6 @@ applications:
           runtime_env:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "0"
-              TWINKLE_FAIL_FAST: "0"
 
   # 3. Sampler Service - Runs inference / sampling using vLLM engine
   #    Used for generating text from the model (e.g., evaluating LoRA results).
@@ -109,7 +105,6 @@ applications:
           runtime_env:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "0"
-              TWINKLE_FAIL_FAST: "0"
 
   # 4. Processor Service
   - name: processor
@@ -132,6 +127,3 @@ applications:
           target_ongoing_requests: 128
         ray_actor_options:
           num_cpus: 0.1
-          runtime_env:
-            env_vars:
-              TWINKLE_FAIL_FAST: "0"

cookbook/client/server/transformer/server_config.yaml

@@ -49,9 +49,6 @@ applications:
           target_ongoing_requests: 128   # Target concurrent requests per replica
         ray_actor_options:
           num_cpus: 0.1                  # CPU resources allocated to this actor
-          runtime_env:
-            env_vars:
-              TWINKLE_FAIL_FAST: "0"
 
   # 2. Model Service - Hosts the base model for training.
   - name: models-Qwen3.5-4B
@@ -85,7 +82,6 @@ applications:
           runtime_env:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "1"
-              TWINKLE_FAIL_FAST: "0"
 
   # 3. Sampler Service - Runs inference / sampling using vLLM engine
   #    Used for generating text from the model (e.g., evaluating LoRA results).
@@ -122,7 +118,6 @@ applications:
           runtime_env:
             env_vars:
               TWINKLE_TRUST_REMOTE_CODE: "1"
-              TWINKLE_FAIL_FAST: "0"
 
   # 4. Processor Service
   - name: processor
@@ -145,6 +140,3 @@ applications:
           target_ongoing_requests: 128
         ray_actor_options:
           num_cpus: 0.1
-          runtime_env:
-            env_vars:
-              TWINKLE_FAIL_FAST: "0"

docs/source_en/Usage Guide/Server and Client/Server.md

@@ -450,3 +450,33 @@ twinkle-server check-config -c server_config.yaml
 | `use_megatron: false` | `backend: transformers` |
 
 Additionally, this refactor introduces two new top-level fields — `telemetry` and `persistence` — which did not exist before. Add them as needed.
+
+## Execution time bounds
+
+Every backend call has a finite time bound. `T` is the effective task execution
+timeout: it equals `execution_timeout`, or `3600s` when that setting is `0`.
+`asyncio.wait_for` uses `T`. The Ray wait uses `R`, which is a method's explicit
+constant timeout when present and otherwise `T`. The default `T` is `1800s`.
+
+Two distinct bounds follow, and they must not be collapsed into one number:
+
+| Bound | Expression | Meaning |
+|-------|------------|---------|
+| Record-terminal bound | `queue_timeout + T` | After this, a task's future record is guaranteed to be in a terminal state (`completed`/`failed`). Use it for alerting thresholds and client polling total-timeout. |
+| Resource-release bound | `Collect_Width × R` from execution start, or `queue_timeout + Collect_Width × R` from submission | After this, the executor thread and the in-flight model-actor call for that task are guaranteed to have finished. Use it for capacity planning. |
+
+`Collect_Width = len(self._actors) = world_size = tp × pp × dp` — the number of
+futures each `remote_function` collection waits on per call. Evidence:
+`LazyCollect._get_result` iterates `self._futures`, which come from
+`_get_workers(self._actors, execute)` (`infra/__init__.py`), covering every actor —
+not just the data-parallel width. On a `tp=8` deployment the execution-start
+resource-release bound is therefore `8 × R`, not `R`.
+
+After the task record becomes terminal, the per-replica Admission_Gate can remain
+closed for at most `max(0, Collect_Width × R − T)`: the record is already terminal,
+but a leaked executor thread may still hold the gate until its `ray.get` returns or
+raises. During that window newly arriving tasks fail fast with a `server`/503 error.
+
+Each persisted future stores its immutable `absolute_deadline` when it is created.
+Cleanup therefore reaches the same decision regardless of which deployment process
+holds the cleanup lease.

docs/source_zh/使用指引/服务端和客户端/服务端.md

@@ -450,3 +450,20 @@ twinkle-server check-config -c server_config.yaml
 | `use_megatron: false` | `backend: transformers` |
 
 此外本次重构新增了 `telemetry` 和 `persistence` 两个顶层字段（旧版本中不存在），可按需添加。
+
+## 执行时间上界
+
+每一次 backend 调用都存在有限时间上界。`T` 是任务的有效 execution timeout：等于 task-queue 配置中的 `execution_timeout`；配置为 `0` 时取 `3600` 秒。`asyncio.wait_for` 使用 `T`。Ray 等待使用 `R`：方法显式声明 timeout 时取该常量，否则取 `T`。`T` 的默认值为 `1800` 秒。
+
+由此派生出两个**不同**的上界，二者不得合成一个数：
+
+| 上界 | 表达式 | 含义 |
+|------|--------|------|
+| 记录终态上界 | `queue_timeout + T` | 超过它后，任务的 future 记录必处于终态（`completed`/`failed`）。用于设置告警阈值与客户端轮询总超时。 |
+| 资源释放上界 | 从执行开始为 `Collect_Width × R`；从提交开始为 `queue_timeout + Collect_Width × R` | 超过它后，该任务占用的 executor 线程与 model actor 在飞调用必已结束。用于容量规划。 |
+
+`Collect_Width = len(self._actors) = world_size = tp × pp × dp`——即每次 `remote_function` 结果收集所等待的 future 个数。证据：`LazyCollect._get_result` 遍历的 `self._futures` 来自 `_get_workers(self._actors, execute)`（`infra/__init__.py`），覆盖全部 actor，而非 data-parallel 宽度。因此在 `tp=8` 的部署上，从执行开始的资源释放上界是 `8 × R` 而非 `R`。
+
+任务记录进入终态后，per-replica 准入闸门额外保持关闭的最长时长为 `max(0, Collect_Width × R − T)`：此时记录已是终态，但泄漏的 executor 线程可能仍持有闸门，直到其 `ray.get` 返回或抛出。在该窗口内新到达的任务会以 `server`/503 错误快速失败。
+
+每条持久化 future 在创建时写入不可变的 `absolute_deadline`，因此无论哪个 deployment 进程持有 cleanup lease，清理结果都由任务自身契约决定。

src/twinkle/infra/__init__.py

@@ -10,7 +10,7 @@
 
 from twinkle.notifier import Notifier, notify_exception
 from twinkle.utils import DeviceGroup, DeviceMesh, Platform, check_unsafe, framework_util, get_logger, requires
-from .collectors import collect_tensor_dict
+from .collectors import collect_tensor_dict as collect_tensor_dict
 
 logger = get_logger()
 
@@ -530,7 +530,7 @@ def _run_continous_work(self, func_name: str, execute_method, workers, args, kwa
     try:
         ordered: List[Any] = [None] * batch_len
         for _, indices, ref in submitted:
-            part = ray.get(ref, timeout=ray_get_timeout) if ray_get_timeout else ray.get(ref)
+            part = ray.get(ref, timeout=ray_get_timeout) if ray_get_timeout is not None else ray.get(ref)
             if not isinstance(part, (list, tuple)) or len(part) != len(indices):
                 raise TypeError(f'{func_name}: enable_continous_work needs one result per request, but a worker given '
                                 f'{len(indices)} request(s) returned {type(part).__name__} of length '
@@ -740,7 +740,6 @@ def _get_device_mesh_param(args, kwargs):
 def _prepare_lazy_collect(args, kwargs):
     # if a worker received an actor handle,
     # lazy collect should be false to prevent any outer function receives an object ref
-    from ._ray import RayHelper
     if not os.environ.get('WORKER_NAME'):
         # If this is a driver
         return args, kwargs
@@ -996,7 +995,10 @@ def remote_function(dispatch: Union[Literal['slice', 'all', 'slice_dp', 'last_pp
         sync: If True, use synchronous execution (execute_all_sync) instead of async.
             Required for methods with NCCL collective operations (e.g., Megatron forward_backward).
         lazy_collect: Do lazy collect, this boolean value decides whether this function needs lazy collect. If setting to None, it will follow the global setting.
-        timeout: Timeout in seconds for ray.get() when collecting results. Instance attribute ``_ray_get_timeout`` overrides this.
+        timeout: Timeout in seconds for ray.get() when collecting results. The decorator's
+            explicitly declared value takes priority; the instance attribute ``_ray_get_timeout``
+            is the fallback for methods that declare none (``timeout if timeout is not None
+            else instance``).
         enable_continous_work: Route each request to the least busy worker instead
             of slicing the batch over all of them, and return the results in the
             caller's order. This is what lets a batch smaller than the worker
@@ -1044,7 +1046,14 @@ def wrapper(self, *args, **kwargs) -> T1:
                     else:
                         # This is the driver
                         from ._ray import RayHelper
-                        execute_method = RayHelper.execute_all_async if not sync else RayHelper.execute_all_sync
+
+                        # Resolve the effective ray.get timeout before choosing execute_method:
+                        # the decorator's explicit value wins, the instance attribute is the
+                        # fallback. ``is not None`` (not ``or``) so that a decorator ``timeout=0``
+                        # is honored instead of falling back to unbounded waiting.
+                        _rgt = timeout if timeout is not None else getattr(self, '_ray_get_timeout', None)
+                        execute_method = RayHelper.execute_all_async if not sync else functools.partial(
+                            RayHelper.execute_all_sync, timeout=_rgt)
                         # Only classes whose workers run methods side by side need
                         # this; elsewhere Ray already orders calls per actor.
                         _concurrent_actor = bool(getattr(self, '_max_concurrency', None))
@@ -1060,8 +1069,7 @@ def wrapper(self, *args, **kwargs) -> T1:
                             _batch_len = _cw_batch_len(args, kwargs)
                             if _batch_len:
                                 return _run_continous_work(self, func.__name__, execute_method, _workers, args, kwargs,
-                                                           _batch_len,
-                                                           getattr(self, '_ray_get_timeout', None) or timeout)
+                                                           _batch_len, _rgt)
                         if RayHelper.has_ref(args, kwargs):
                             # If has any object-ref, dispatch in worker, because we don't know the structure in the ref.
                             # for example, dataloader returns any data list.
@@ -1079,7 +1087,6 @@ def wrapper(self, *args, **kwargs) -> T1:
                         # busy.
                         _tracked_refs = _cw_register(self, func.__name__, result) if _concurrent_actor else []
                         # This is a result future, call it to get the actual result
-                        _rgt = getattr(self, '_ray_get_timeout', None) or timeout
                         result_func = RayHelper.do_get_and_collect_func(
                             _collect_func, collect, result, device_mesh, timeout=_rgt)
                         _local_lazy_collect = _lazy_collect
@@ -1090,13 +1097,13 @@ def wrapper(self, *args, **kwargs) -> T1:
                         if func.__name__ == '__len__':
                             # Get the first result and ignore the `lazy_collect`
                             import ray
-                            return ray.get(result[0])
+                            return ray.get(result[0], timeout=_rgt)
 
                         if func.__name__ == '__next__':
                             import ray
                             for _res in result:
                                 # raise when any worker raises StopIteration
-                                stop = ray.get(_res[1])
+                                stop = ray.get(_res[1], timeout=_rgt)
                                 if stop:
                                     raise StopIteration()
                             result = [_res[0] for _res in result]

src/twinkle/infra/_ray/ray_helper.py

@@ -137,10 +137,17 @@ def is_worker():
         return RayHelper.ray_inited() and ray._private.worker.global_worker.mode == ray._private.worker.WORKER_MODE
 
     @staticmethod
-    def execute_all_sync(method_name: str, workers_and_args: List[Tuple[Any, List[Any], Dict[str, Any]]]):
-        """Execute method and return results."""
+    def execute_all_sync(method_name: str, workers_and_args: List[Tuple[Any, List[Any], Dict[str, Any]]], timeout=None):
+        """Execute method and return results.
+
+        ``timeout`` is passed to ``ray.get(list, timeout=)``, whose semantics are
+        the **total** wall-clock time to collect the whole list -- different from
+        ``LazyCollect``'s per-future timing (see ``do_get_and_collect_func``).
+        The two paths are each bounded on their own; the total-time semantics here
+        are strictly tighter.
+        """
         import ray
-        return ray.get(RayHelper.execute_all_async(method_name, workers_and_args))
+        return ray.get(RayHelper.execute_all_async(method_name, workers_and_args), timeout=timeout)
 
     @staticmethod
     def execute_all_async(method_name: str, workers_and_args: List[Tuple[Any, List[Any], Dict[str, Any]]]):

src/twinkle/loss/grpo.py

@@ -202,14 +202,16 @@ def _pad_and_align_to_batch(
             elif n_sample == n_pos:
                 # Response-only form (e.g. old_logps from vLLM).
                 result[i, pos] = sample
-            elif n_sample >= seq_len:
-                # Full-sequence form (e.g. ref_logps right-padded with ignore-value).
-                result[i, pos] = sample[:seq_len][mask[i]]
+            elif n_pos == 0 or (n_sample > 0 and pos[-1].item() < n_sample):
+                # Variable-length full-sequence form. The processor right-pads the
+                # batch, but per-sample RL fields from Tinker remain unpadded. They
+                # are valid when every selected mask position exists in this row.
+                result[i, pos] = sample[pos]
             else:
                 raise AssertionError(f'data/mask length mismatch at sample {i}: '
                                      f'n_pos={n_pos}, n_sample={n_sample}, seq_len={seq_len} '
-                                     '(expected n_sample == n_pos for response-only form, '
-                                     'or n_sample >= seq_len for full-sequence form)')
+                                     '(expected n_sample == n_pos for response-only form, or all masked positions '
+                                     'to exist in the per-sample full-sequence form)')
 
         return result