Skip to content

vllm.snapshot.file_control

Private, one-shot file control for an external startup capturer.

Classes:

  • Activation –

    Fresh deployment context, supplied after native CUDA restore/unlock.

  • FileSnapshotControl –

    The capturer provides a fresh private directory for every activation.

Activation

Bases: BaseModel

Fresh deployment context, supplied after native CUDA restore/unlock.

Source code in vllm/snapshot/file_control.py 
23
24
25
26
27
28
29
30
31
32
33
34
class Activation(BaseModel):
    """Fresh deployment context, supplied after native CUDA restore/unlock."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)

    version: Literal[1]
    capture_id: str
    activation_id: str = Field(min_length=1, max_length=128, pattern=r"^[\w.-]+$")
    kind: Literal["donor", "copy"]
    native_complete: bool
    host: str | None
    port: int = Field(ge=1, le=65535)

FileSnapshotControl

The capturer provides a fresh private directory for every activation.

It must capture before writing activation.json. A challenge generated afterward prevents a release saved by an earlier copy from being replayed. The capturer owns artifact compatibility, namespace/stdio restoration and native completion. This protocol does not invoke or retry native recovery.

Methods:

  • publish_ready –

    Publish only after all worker preparation and rehearsal complete.

  • wait_for_activation –

    Require fresh native completion and a response to a new challenge.

  • write_error –

    Retain the primary failure even if restored standard streams fail.

  • write_status –

    Record recovery progress; validated is not HTTP serving readiness.

Source code in vllm/snapshot/file_control.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class FileSnapshotControl:
    """The capturer provides a fresh private directory for every activation.

    It must capture before writing activation.json. A challenge generated
    afterward prevents a release saved by an earlier copy from being replayed.
    The capturer owns artifact compatibility, namespace/stdio restoration and
    native completion. This protocol does not invoke or retry native recovery.
    """

    def __init__(self, directory: Path, timeout_s: float):
        validate_artifact_root(directory, creating=False)
        if any(directory.iterdir()):
            raise ValueError("startup snapshot control directory must be empty")
        self.directory = directory
        self.timeout_s = timeout_s
        self.capture_id = uuid4().hex
        self.activation: Activation | None = None
        self._wait_started = False

    def publish_ready(self, oracle: Oracle) -> None:
        """Publish only after all worker preparation and rehearsal complete."""
        _write_json_atomic(
            self.directory / "capture-ready.json",
            {
                "version": 1,
                "capture_id": self.capture_id,
                "pid": os.getpid(),
                "oracle": asdict(oracle),
            },
        )

    def _read(self, name: str) -> dict:
        validate_artifact_root(self.directory, creating=False)
        path = self.directory / name
        if not stat.S_ISREG(path.lstat().st_mode):
            raise ValueError(f"snapshot control must be a regular file: {name}")
        with path.open("rb") as source:
            data = source.read(65537)
        if len(data) > 65536:
            raise ValueError(f"snapshot control is too large: {name}")
        payload = json.loads(data)
        if not isinstance(payload, dict):
            raise ValueError(f"snapshot control must be a JSON object: {name}")
        return payload

    async def _wait(self, name: str) -> dict:
        # Frozen time is owned by the capturer's deadline, not a deadline
        # captured in the event loop. Bound this instance's polling budget.
        remaining = self.timeout_s
        while True:
            try:
                return self._read(name)
            except FileNotFoundError:
                if remaining <= 0:
                    raise TimeoutError(
                        f"snapshot control timed out waiting for {name}"
                    ) from None
                delay = min(0.05, remaining)
                await asyncio.sleep(delay)
                remaining -= delay

    async def wait_for_activation(self) -> Activation:
        """Require fresh native completion and a response to a new challenge."""
        if self.activation is not None:
            return self.activation
        if self._wait_started:
            raise RuntimeError("snapshot activation wait already started")
        self._wait_started = True
        payload = await self._wait("activation.json")
        activation = Activation.model_validate(payload)
        if activation.capture_id != self.capture_id or not activation.native_complete:
            raise ValueError("snapshot activation does not match a completed capture")
        for name in ("challenge.json", "release.json", "status.json", "error.json"):
            if os.path.lexists(self.directory / name):
                raise ValueError(f"stale snapshot activation control: {name}")
        challenge = {
            "version": 1,
            "capture_id": self.capture_id,
            "activation_id": activation.activation_id,
            "nonce": uuid4().hex,
        }
        _write_json_atomic(self.directory / "challenge.json", challenge)
        release = await self._wait("release.json")
        if release != challenge or self._read("activation.json") != payload:
            raise ValueError(
                "snapshot release does not match this activation challenge"
            )
        self.activation = activation
        return activation

    def write_status(self, phase: str) -> None:
        """Record recovery progress; validated is not HTTP serving readiness."""
        assert self.activation is not None
        _write_json_atomic(
            self.directory / "status.json",
            {
                "capture_id": self.capture_id,
                "activation_id": self.activation.activation_id,
                "phase": phase,
            },
            overwrite=True,
        )

    def write_error(self, phase: str, error: BaseException) -> None:
        """Retain the primary failure even if restored standard streams fail."""
        try:
            _write_json_atomic(
                self.directory / "error.json",
                {
                    "capture_id": self.capture_id,
                    "activation_id": (
                        self.activation.activation_id if self.activation else None
                    ),
                    "phase": phase,
                    "error_type": type(error).__name__,
                    "error": str(error),
                },
                overwrite=True,
            )
        except Exception:
            logger.exception("Could not record startup snapshot failure")

publish_ready(oracle)

Publish only after all worker preparation and rehearsal complete.

Source code in vllm/snapshot/file_control.py 
56
57
58
59
60
61
62
63
64
65
66
def publish_ready(self, oracle: Oracle) -> None:
    """Publish only after all worker preparation and rehearsal complete."""
    _write_json_atomic(
        self.directory / "capture-ready.json",
        {
            "version": 1,
            "capture_id": self.capture_id,
            "pid": os.getpid(),
            "oracle": asdict(oracle),
        },
    )

wait_for_activation() async

Require fresh native completion and a response to a new challenge.

Source code in vllm/snapshot/file_control.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def wait_for_activation(self) -> Activation:
    """Require fresh native completion and a response to a new challenge."""
    if self.activation is not None:
        return self.activation
    if self._wait_started:
        raise RuntimeError("snapshot activation wait already started")
    self._wait_started = True
    payload = await self._wait("activation.json")
    activation = Activation.model_validate(payload)
    if activation.capture_id != self.capture_id or not activation.native_complete:
        raise ValueError("snapshot activation does not match a completed capture")
    for name in ("challenge.json", "release.json", "status.json", "error.json"):
        if os.path.lexists(self.directory / name):
            raise ValueError(f"stale snapshot activation control: {name}")
    challenge = {
        "version": 1,
        "capture_id": self.capture_id,
        "activation_id": activation.activation_id,
        "nonce": uuid4().hex,
    }
    _write_json_atomic(self.directory / "challenge.json", challenge)
    release = await self._wait("release.json")
    if release != challenge or self._read("activation.json") != payload:
        raise ValueError(
            "snapshot release does not match this activation challenge"
        )
    self.activation = activation
    return activation

write_error(phase, error)

Retain the primary failure even if restored standard streams fail.

Source code in vllm/snapshot/file_control.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def write_error(self, phase: str, error: BaseException) -> None:
    """Retain the primary failure even if restored standard streams fail."""
    try:
        _write_json_atomic(
            self.directory / "error.json",
            {
                "capture_id": self.capture_id,
                "activation_id": (
                    self.activation.activation_id if self.activation else None
                ),
                "phase": phase,
                "error_type": type(error).__name__,
                "error": str(error),
            },
            overwrite=True,
        )
    except Exception:
        logger.exception("Could not record startup snapshot failure")

write_status(phase)

Record recovery progress; validated is not HTTP serving readiness.

Source code in vllm/snapshot/file_control.py 
127
128
129
130
131
132
133
134
135
136
137
138
def write_status(self, phase: str) -> None:
    """Record recovery progress; validated is not HTTP serving readiness."""
    assert self.activation is not None
    _write_json_atomic(
        self.directory / "status.json",
        {
            "capture_id": self.capture_id,
            "activation_id": self.activation.activation_id,
            "phase": phase,
        },
        overwrite=True,
    )