Remove obsolete Straw queue limits

docker/version.txt

@@ -1 +1 @@
-nightly-dev-20260928a
+nightly-dev-20260930a

docs/en/advanced/straw.md

@@ -17,7 +17,7 @@ install it in an existing environment, run this on every rollout and training
 node, using the same version throughout the job:
 
 ```bash
-pip install straw-queue
+pip install 'straw-queue>=0.1.2'
 ```
 
 Add these arguments to your training command:
@@ -108,14 +108,8 @@ and ordinary fields still occupy manager memory.
 |---|---|---|
 | `--rollout-queue-segment-mib` | `256` | Pack rotation target in MiB |
 | `--rollout-io-concurrency` | `4` | Bound concurrent serialization and filesystem I/O submissions |
-| `--rollout-queue-max-pending` | `65536` | Pending-task admission limit |
-| `--rollout-queue-max-inflight` | `65536` | In-flight task limit |
 | `--rollout-queue-lease-seconds` | `300` | Worker lease duration; active readers renew it |
 
-Task limits do not bound retained bytes. Size storage for the workload and its
-checkpoint/archive retention. Shared-storage bandwidth, sync latency and the
-single queue coordinator can limit throughput.
-
 ## Online GC
 
 Online GC is disabled by default. Enable it with `--rollout-queue-online-gc`.
@@ -131,7 +125,7 @@ its storage ownership. Do not manually delete pack files from an active pool.
 For offline removal, stop all coordinators, writers and readers first.
 
 A GC failure is reported by subsequent queue operations and at shutdown.
-Capacity exhaustion requires freeing unneeded retention or increasing capacity;
+Disk exhaustion requires freeing unneeded retention or expanding storage;
 it does not reset the queue automatically. Live-pack and journal compaction are
 not implemented.
 
@@ -234,7 +228,12 @@ It shares existing tensors when saved with straw transport; otherwise it creates
 a `straw-data` pool alongside the index. There is one index per rollout, with
 samples packed together. Evaluation uses `eval_<id>` in place of the rollout ID.
 Copying only the JSON index does not copy the payloads. For train-only replay
-with straw transport, choose a separate writable training queue.
+with straw transport automatically reuses the archive’s storage pool and run,
+overriding `--rollout-data-dir` and `--rollout-queue-run-id`. Each replay creates
+an isolated queue in that writable pool; the source queue is left untouched.
+Full replay (without subsampling) reuses the archived Sample and tensor records
+without republishing them.
+All rollout archives in one replay must belong to the same pool and run.
 `--load-debug-rollout-data-subsample` also applies to archives.
 
 ```python

docs/en/get_started/usage.md

@@ -406,7 +406,7 @@ To enable distributed fully async rollout with straw, add:
 
 All nodes must mount the directory at the same absolute path and have
 `straw-queue` installed. The standard slime installation includes it; for an
-existing environment, run `pip install straw-queue`. Configure the JuiceFS
+existing environment, run `pip install 'straw-queue>=0.1.2'`. Configure the JuiceFS
 storage profile and deployment declaration as described in the
 [straw guide](../advanced/straw.md#enable-it). Selecting straw alone uses the
 synchronous rollout entrypoint. For a fresh run, omitting `--rollout-data-dir`
@@ -416,7 +416,7 @@ With straw, `--use-rollout-routing-replay` and `--use-score-centering` persist
 R3 and SC tensors with their samples, including partial continuations.
 `--rollout-queue-online-gc` optionally reclaims unused storage; it is disabled
 by default. Recovery uses `--load`, `--save` and optional `--ckpt-step`.
-See the [straw guide](../advanced/straw.md) for scheduling, storage limits,
+See the [straw guide](../advanced/straw.md) for scheduling,
 checkpoint recovery and debug replay, and [customization](customization.md)
 for custom rollout functions.
 

docs/zh/advanced/straw.md

@@ -9,7 +9,7 @@
 标准 slime 安装和 Docker 构建包含 `straw-queue`。已有环境可以在每个 rollout 和训练节点执行以下命令，同一任务使用相同版本：
 
 ```bash
-pip install straw-queue
+pip install 'straw-queue>=0.1.2'
 ```
 
 在训练命令中添加：
@@ -60,12 +60,8 @@ R3 使用 `--use-rollout-routing-replay`，SC 使用 `--use-score-centering`。
 |---|---|---|
 | `--rollout-queue-segment-mib` | `256` | Pack 轮转的目标大小，单位 MiB |
 | `--rollout-io-concurrency` | `4` | 限制并发序列化与文件系统 I/O 提交 |
-| `--rollout-queue-max-pending` | `65536` | 等待任务的接收上限 |
-| `--rollout-queue-max-inflight` | `65536` | 在途任务上限 |
 | `--rollout-queue-lease-seconds` | `300` | Worker 租期，活跃 reader 会续租 |
 
-任务数量上限不限制保留数据的字节数。应根据负载和 checkpoint/归档保留量规划容量。共享存储带宽、同步写入延迟和单个队列 coordinator 都可能限制吞吐。
-
 ## 在线 GC
 
 在线 GC 默认关闭，通过 `--rollout-queue-online-gc` 开启。slime 在任务和训练 batch 使用完成或被丢弃后通知 straw。只有所有任务、reader、checkpoint 和归档都释放引用后，straw 才会回收封存的 pack。一条存活记录就会保留整个 pack，活跃 writer 和日志历史也会占用空间。
@@ -133,7 +129,7 @@ Debug 保存和加载参数支持 `.pt` 与 `.straw.json` 两种格式：
 | `.pt` | Sample 数据和实际张量内容 | 独立文件 |
 | `.straw.json` | 带 sample/task key 的不可变索引 | 需要保留所引用的 straw 存储池 |
 
-索引归档独立保留数据，不受队列消费和 GC 影响。使用 straw 传输保存时复用已有张量，否则在索引旁创建 `straw-data` 存储池。每个 rollout 只有一个索引，sample 打包存储。Evaluation 使用 `eval_<id>` 替代 rollout ID。单独复制 JSON 索引不会复制载荷。使用 straw 做只训练回放时，应选择独立的可写训练队列。`--load-debug-rollout-data-subsample` 也适用于归档。
+索引归档独立保留数据，不受队列消费和 GC 影响。使用 straw 传输保存时复用已有张量，否则在索引旁创建 `straw-data` 存储池。每个 rollout 只有一个索引，sample 打包存储。Evaluation 使用 `eval_<id>` 替代 rollout ID。单独复制 JSON 索引不会复制载荷。使用 straw 做只训练回放时，自动使用归档的存储池和 run，覆盖 `--rollout-data-dir` 和 `--rollout-queue-run-id`，并在该可写存储池中新建独立队列，不改变原队列。完整回放（不抽样）直接复用已有 Sample 和张量记录，不重新写入。一次回放的所有归档必须属于同一存储池和 run。`--load-debug-rollout-data-subsample` 也适用于归档。
 
 ```python
 from slime.data.archive import RolloutArchive

docs/zh/get_started/usage.md

@@ -403,9 +403,9 @@ slime 支持不同程度的自定义数据生成（rollout）。
 --rollout-data-dir /shared/run/rollout_data
 ```
 
-所有节点必须以相同绝对路径挂载该目录，并安装 `straw-queue`。标准 slime 安装已包含此依赖；已有环境可以执行 `pip install straw-queue`。JuiceFS storage profile 和部署声明的配置见 [straw 指南](../advanced/straw.md#启用方式)。只选择 straw 时使用同步 rollout 入口。新任务已设置 `--save` 时，省略 `--rollout-data-dir` 会使用 `<save>/rollout_data`。
+所有节点必须以相同绝对路径挂载该目录，并安装 `straw-queue`。标准 slime 安装已包含此依赖；已有环境可以执行 `pip install 'straw-queue>=0.1.2'`。JuiceFS storage profile 和部署声明的配置见 [straw 指南](../advanced/straw.md#启用方式)。只选择 straw 时使用同步 rollout 入口。新任务已设置 `--save` 时，省略 `--rollout-data-dir` 会使用 `<save>/rollout_data`。
 
-使用 straw 时，`--use-rollout-routing-replay` 和 `--use-score-centering` 将 R3、SC 张量随 sample 持久化，也支持 partial continuation。`--rollout-queue-online-gc` 可开启未使用存储的回收，默认关闭。恢复使用 `--load`、`--save` 和可选的 `--ckpt-step`。调度、容量限制、checkpoint 恢复和 debug 回放见 [straw 指南](../advanced/straw.md)，自定义 rollout 函数见[自定义功能](customization.md)。
+使用 straw 时，`--use-rollout-routing-replay` 和 `--use-score-centering` 将 R3、SC 张量随 sample 持久化，也支持 partial continuation。`--rollout-queue-online-gc` 可开启未使用存储的回收，默认关闭。恢复使用 `--load`、`--save` 和可选的 `--ckpt-step`。调度、checkpoint 恢复和 debug 回放见 [straw 指南](../advanced/straw.md)，自定义 rollout 函数见[自定义功能](customization.md)。
 
 ## sglang 使用方法
 

requirements.txt

@@ -18,7 +18,7 @@ qwen_vl_utils # for VLM
 ray[default]
 safetensors
 sglang-router>=0.3.0
-straw-queue
+straw-queue>=0.1.2
 tensorboard
 transformers
 wandb

slime/backends/megatron_utils/actor.py

@@ -143,6 +143,12 @@ def init(
             hf_vocab = getattr(self.hf_config, "vocab_size", None)
             self.args.vocab_size = hf_vocab if hf_vocab is not None else self.tokenizer.vocab_size
 
+        # Model-only resumes keep the serving version aligned with the next
+        # rollout. Actor recreation can supply the latest version explicitly.
+        if not hasattr(args, "update_weight_start_version"):
+            args.update_weight_start_version = (
+                args.start_rollout_id if args.start_rollout_id is not None else start_rollout_id
+            )
         self.weight_updater = create_weight_updater(
             self.args,
             self.model,

slime/backends/megatron_utils/update_weight/update_weight_from_tensor.py

@@ -231,6 +231,14 @@ def _prepare_expert_weight_batch(
         for request in dist.batch_isend_irecv(p2p_ops) if p2p_ops else ():
             request.wait()
 
+        # NCCL Work.wait() only orders the current CUDA stream. A fast rank
+        # must not enter quantization's CUDA allocations/copies while peers
+        # are still issuing grouped P2P calls: those host calls can synchronize
+        # with the unfinished transfer and prevent its peers from launching.
+        # Include ranks without local transfers so every batch has one fence.
+        dist.barrier(group=get_gloo_group())
+        torch.cuda.synchronize()
+
         hf_named_tensors = []
         for expert_param, tensor in local_params:
             hf_named_tensors.extend(

slime/data/archive.py

@@ -37,7 +37,7 @@ def __init__(self, path, *, root=None):
         self.contents = self.codec.load(self.manifest)
 
     @classmethod
-    def save(cls, path, samples, *, rollout_id, evaluation=False, args=None):
+    def save(cls, path, samples, *, rollout_id, evaluation=False, args=None, reference=None):
         from straw.protocol import digest
         from straw.reporting import write_report
         from straw.store import SharedFilesystemStore
@@ -67,12 +67,30 @@ def save(cls, path, samples, *, rollout_id, evaluation=False, args=None):
         try:
             # Bounded chunks support indexed reads without a file per sample.
             # Lazy R3/SC references in the same pool share their existing bytes.
-            chunks, entries = [], []
+            from slime.data.sample_metadata import describe_sample
+
+            chunks, entries, sample_metadata = [], [], []
             with lock:
-                for start in range(0, len(samples), 64):
+                if reference is not None:
+                    if not shared or Path(reference.root).resolve() != store.backend.root:
+                        raise ValueError("An archive reference must belong to its storage pool")
+                    for offset, sample in enumerate(samples):
+                        sample_metadata.append(describe_sample(sample))
+                        context = getattr(sample, "_queue_receipt", None) or getattr(sample, "_queue_lease", None)
+                        entries.append(
+                            {
+                                "sample_key": (
+                                    f"sample:{sample.index}" if sample.index is not None else f"position:{offset}"
+                                ),
+                                "task_key": context["task_id"] if context else None,
+                                "offset": offset,
+                            }
+                        )
+                for start in range(0, len(samples) if reference is None else 0, 64):
                     chunk = samples[start : start + 64]
                     ref = codec.publish(chunk, submission_id=f"debug-chunk:{uuid.uuid4().hex}")
                     for offset, sample in enumerate(chunk):
+                        sample_metadata.append(describe_sample(sample))
                         context = getattr(sample, "_queue_receipt", None) or getattr(sample, "_queue_lease", None)
                         entries.append(
                             {
@@ -87,9 +105,12 @@ def save(cls, path, samples, *, rollout_id, evaluation=False, args=None):
                             }
                         )
                     chunks.append(DiskPayloadRef(ref, str(store.backend.root)))
-                manifest = codec.publish(
-                    {"chunks": chunks, "entries": entries}, submission_id=f"debug-index:{uuid.uuid4().hex}"
+                contents = (
+                    {"raw": reference, "entries": entries}
+                    if reference is not None
+                    else {"chunks": chunks, "entries": entries}
                 )
+                manifest = codec.publish(contents, submission_id=f"debug-index:{uuid.uuid4().hex}")
                 owner = f"debug:{path.resolve()}:{digest(asdict(manifest))}"
                 # Ownership must commit before the externally visible index.
                 store.retain(owner, [manifest])
@@ -104,12 +125,37 @@ def save(cls, path, samples, *, rollout_id, evaluation=False, args=None):
                         "root": str(store.backend.root),
                         "manifest": asdict(manifest),
                         "storage_owner": owner,
+                        "sample_metadata": {"version": 1, "samples": sample_metadata},
                     },
                 )
         finally:
             if not shared:
                 store.close()
 
+    @staticmethod
+    def check_metadata(path, args):
+        """Read only the small archive index, never the collection/tensor records.
+
+        Legacy archives can have an explicitly built metadata sidecar. The
+        manifest digest binds it to this exact immutable collection.
+        """
+        from slime.data.sample_metadata import validate_sample_metadata
+
+        path = Path(path)
+        index = json.loads(path.read_text())
+        if index.get("format") != "slime.straw-debug" or index.get("version") != 1:
+            raise ValueError("Unsupported rollout archive")
+        metadata = index.get("sample_metadata")
+        if metadata is None:
+            sidecar = json.loads(path.with_suffix(path.suffix + ".metadata.json").read_text())
+            if sidecar["manifest_digest"] != index["manifest"]["digest"]:
+                raise ValueError("Metadata sidecar belongs to a different archive")
+            metadata = sidecar["sample_metadata"]
+        if metadata.get("version") != 1:
+            raise ValueError("Unsupported sample metadata version")
+        validate_sample_metadata(metadata["samples"], args)
+        return metadata["samples"]
+
     def keys(self):
         """Return sample/task keys in archive order, including duplicate sample IDs."""
         return [(entry["sample_key"], entry["task_key"]) for entry in self.contents["entries"]]
@@ -129,8 +175,13 @@ def load_samples(self, *, sample_key=None, task_key=None):
         if not selected and (sample_key is not None or task_key is not None):
             raise KeyError((sample_key, task_key))
         loaded, result = {}, []
+        if "raw" in self.contents and selected:
+            from slime.data.transport import load_rollout_samples
+            from slime.rollout.base_types import iter_samples
+
+            loaded["raw"] = list(iter_samples(load_rollout_samples(self.contents["raw"])))
         for entry in selected:
-            chunk = entry["chunk"]
+            chunk = entry.get("chunk", "raw")
             if chunk not in loaded:
                 loaded[chunk] = self.codec.load(self.contents["chunks"][chunk].manifest)
             sample = copy.copy(loaded[chunk][entry["offset"]])

slime/data/batch_builder.py

@@ -12,7 +12,7 @@
 import torch
 
 from slime.data.tensor import TensorRef
-from slime.data.transport import DiskPayloadRef, RolloutGroupRef, TrainBatchRef, pack_rollout_payload
+from slime.data.transport import DiskPayloadRef, TrainBatchRef, pack_rollout_payload
 from slime.observability.rollout_data_utils import (
     tensorize_rollout_data_for_training,
     validate_rollout_routed_experts_for_replay,
@@ -102,12 +102,7 @@ def begin(self, samples):
                     receipt = ray.get(controller.result.remote(Lease(**lease)))
                     if receipt is not None:
                         positions.add(receipt.position)
-            value = self.raw_ref
-            while isinstance(value, DiskPayloadRef):
-                value = value.load()
-            for group in value:
-                if isinstance(group, RolloutGroupRef) and group.receipt:
-                    positions.add(group.receipt.position)
+            # Samples already carry group receipts/positions; do not reread raw.
             plan = {
                 **identity,
                 "digest": self.plan_digest,