Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
339761045f | ||
|
|
4c773109b1 |
No files matched your search
@@ -24,6 +24,7 @@ V1 includes:
|
|||||||
- coordinated local lookup and one-producer publication;
|
- coordinated local lookup and one-producer publication;
|
||||||
- descriptor-backed safetensors mmap leases;
|
- descriptor-backed safetensors mmap leases;
|
||||||
- preferred and required resolution policy;
|
- preferred and required resolution policy;
|
||||||
|
- explicit, separately reported post-load publication;
|
||||||
- validation, deny, quarantine, cleanup, and capacity controls; and
|
- validation, deny, quarantine, cleanup, and capacity controls; and
|
||||||
- typed reports for every terminal resolution outcome.
|
- typed reports for every terminal resolution outcome.
|
||||||
|
|
||||||
@@ -33,7 +34,6 @@ V1 does not include:
|
|||||||
- a generic BF16, FP8, quantized, or model-specific producer;
|
- a generic BF16, FP8, quantized, or model-specific producer;
|
||||||
- CUDA registration, pinned staging, H2D scheduling, or GPU kernels;
|
- CUDA registration, pinned staging, H2D scheduling, or GPU kernels;
|
||||||
- a remote artifact provider or cross-node coordination;
|
- a remote artifact provider or cross-node coordination;
|
||||||
- post-load publication (V1 accepts only pre-load-safe producers);
|
|
||||||
- automatic eviction; or
|
- automatic eviction; or
|
||||||
- a change to DLO AllGather or no-AllGather behavior.
|
- a change to DLO AllGather or no-AllGather behavior.
|
||||||
|
|
||||||
@@ -84,6 +84,12 @@ flowchart TD
|
|||||||
F -->|"preferred"| C
|
F -->|"preferred"| C
|
||||||
F -->|"required"| E["Fail startup"]
|
F -->|"required"| E["Fail startup"]
|
||||||
P -->|"nonretryable failure"| E
|
P -->|"nonretryable failure"| E
|
||||||
|
C --> W{"Post-load publication enabled?"}
|
||||||
|
W -->|"yes"| B2["Publish final model through POST_LOAD_ONLY producer"]
|
||||||
|
B2 --> CL["Close validated publication lease"]
|
||||||
|
CL --> R2["Return separate publication report"]
|
||||||
|
R2 --> D
|
||||||
|
W -->|"no"| D["Keep canonical model"]
|
||||||
```
|
```
|
||||||
|
|
||||||
The modes are:
|
The modes are:
|
||||||
@@ -94,10 +100,12 @@ The modes are:
|
|||||||
| `preferred` | Return an exact lease when available; otherwise use canonical fallback only for a miss, invalid cache entry, or typed retryable failure. |
|
| `preferred` | Return an exact lease when available; otherwise use canonical fallback only for a miss, invalid cache entry, or typed retryable failure. |
|
||||||
| `required` | Fail when an exact lease cannot be acquired. |
|
| `required` | Fail when an exact lease cannot be acquired. |
|
||||||
|
|
||||||
An unsupported capability, semantic identity collision, producer failure, or
|
During pre-load resolution, an unsupported capability, semantic identity
|
||||||
publication failure is nonretryable and remains visible even in preferred
|
collision, producer failure, or publication failure is nonretryable and remains
|
||||||
mode. Storage policy must never disguise a semantic or configuration error as
|
visible even in preferred mode. A post-load publication failure is likewise
|
||||||
a cache miss.
|
visible in its own report, but cannot revise the canonical-fallback outcome.
|
||||||
|
Storage policy must never disguise a semantic or configuration error as a cache
|
||||||
|
miss.
|
||||||
|
|
||||||
## Loader and restoration sequence
|
## Loader and restoration sequence
|
||||||
|
|
||||||
@@ -118,24 +126,35 @@ sequenceDiagram
|
|||||||
R->>S: lookup(exact identity)
|
R->>S: lookup(exact identity)
|
||||||
alt validated hit
|
alt validated hit
|
||||||
S-->>R: HostWeightLease
|
S-->>R: HostWeightLease
|
||||||
|
R-->>L: lease and LOCAL_HIT report
|
||||||
|
L->>X: plan_restore(model, lease)
|
||||||
|
X-->>L: validation-only restore plan
|
||||||
|
L->>X: commit() once
|
||||||
else miss with allowed producer
|
else miss with allowed producer
|
||||||
R->>S: get_or_build(identity, producer)
|
R->>S: get_or_build(identity, producer)
|
||||||
S->>P: produce(store-scoped writer)
|
S->>P: produce(store-scoped writer)
|
||||||
P-->>S: final-layout tensors and metadata
|
P-->>S: final-layout tensors and metadata
|
||||||
S-->>R: validated HostWeightLease
|
S-->>R: validated HostWeightLease
|
||||||
|
R-->>L: lease and LOCAL_PRODUCTION report
|
||||||
|
L->>X: plan_restore(model, lease)
|
||||||
|
X-->>L: validation-only restore plan
|
||||||
|
L->>X: commit() once
|
||||||
else policy permits canonical fallback
|
else policy permits canonical fallback
|
||||||
R-->>L: CANONICAL_FALLBACK
|
R-->>L: CANONICAL_FALLBACK
|
||||||
L->>L: Run canonical loader
|
L->>L: Run canonical loader
|
||||||
|
opt explicit post-load publication enabled
|
||||||
|
L->>R: publish_after_load(identity, POST_LOAD_ONLY producer)
|
||||||
|
R->>S: get_or_build(identity, producer)
|
||||||
|
S-->>R: validated HostWeightLease or typed failure
|
||||||
|
R->>R: close publication lease
|
||||||
|
R-->>L: separate publication report
|
||||||
|
end
|
||||||
else required or nonretryable failure
|
else required or nonretryable failure
|
||||||
R-->>L: FAILED report
|
R-->>L: FAILED report
|
||||||
end
|
end
|
||||||
R-->>L: lease and terminal report
|
L->>T: consume final CPU tensors and any lease
|
||||||
L->>X: plan_restore(model, lease)
|
|
||||||
X-->>L: validation-only restore plan
|
|
||||||
L->>X: commit() once
|
|
||||||
L->>T: consume restored CPU tensors
|
|
||||||
T-->>L: transfer teardown complete
|
T-->>L: transfer teardown complete
|
||||||
L->>L: close lease
|
L->>L: close lease when one was acquired
|
||||||
```
|
```
|
||||||
|
|
||||||
`plan_restore()` must not mutate the model or lease. `commit() -> None` is the
|
`plan_restore()` must not mutate the model or lease. `commit() -> None` is the
|
||||||
@@ -143,6 +162,15 @@ sole one-shot model mutation. If planning fails, canonical fallback may reuse
|
|||||||
the untouched model. If commit begins and fails, the partially hydrated model
|
the untouched model. If commit begins and fails, the partially hydrated model
|
||||||
must be discarded and canonical fallback must construct a fresh model.
|
must be discarded and canonical fallback must construct a fresh model.
|
||||||
|
|
||||||
|
`publish_after_load()` is synchronous in V1 and accepts only a
|
||||||
|
`POST_LOAD_ONLY` producer. Its report is separate from the terminal resolution,
|
||||||
|
so a publication failure cannot rewrite a successful canonical fallback. A
|
||||||
|
successful publication closes the store-returned lease inside the runtime and
|
||||||
|
warms only future startups. It does not restore, rebind, or otherwise mutate the
|
||||||
|
canonically loaded model serving the current startup.
|
||||||
|
`allow_local_build` gates producers during pre-load resolution, while
|
||||||
|
`allow_post_load_publish` independently gates this explicit post-load path.
|
||||||
|
|
||||||
## Exact representation identity
|
## Exact representation identity
|
||||||
|
|
||||||
The store performs exact lookup. It never silently converts one representation
|
The store performs exact lookup. It never silently converts one representation
|
||||||
@@ -252,7 +280,8 @@ removing live mappings.
|
|||||||
| Retryable lock, domain, or capacity failure | Canonical fallback | Fail |
|
| Retryable lock, domain, or capacity failure | Canonical fallback | Fail |
|
||||||
| Unsupported producer or backend | Fail | Fail |
|
| Unsupported producer or backend | Fail | Fail |
|
||||||
| Identity collision | Fail | Fail |
|
| Identity collision | Fail | Fail |
|
||||||
| Producer or publication failure | Fail | Fail |
|
| Pre-load producer or atomic publication failure | Fail | Fail |
|
||||||
|
| Post-load publication failure after canonical fallback | Keep canonical model; report publication failure | Not reached through required resolution |
|
||||||
| Restore planning failure | Canonical fallback may reuse untouched model | Fail |
|
| Restore planning failure | Canonical fallback may reuse untouched model | Fail |
|
||||||
| Restore commit failure | Discard model; fallback requires a fresh instance | Fail |
|
| Restore commit failure | Discard model; fallback requires a fresh instance | Fail |
|
||||||
|
|
||||||
@@ -275,7 +304,8 @@ A consumer PR must:
|
|||||||
a failed restore commit.
|
a failed restore commit.
|
||||||
7. Retain the lease until every transport operation that may access mapped
|
7. Retain the lease until every transport operation that may access mapped
|
||||||
memory has completed.
|
memory has completed.
|
||||||
8. Emit and test the terminal resolution report.
|
8. Emit and test the terminal resolution report and any separate post-load
|
||||||
|
publication report.
|
||||||
|
|
||||||
Consumer validation must prove:
|
Consumer validation must prove:
|
||||||
|
|
||||||
|
|||||||
@@ -55,7 +55,9 @@ flowchart LR
|
|||||||
The contracts have deliberately narrow ownership:
|
The contracts have deliberately narrow ownership:
|
||||||
|
|
||||||
- `HostWeightRuntime` applies resolution and fallback policy and emits one
|
- `HostWeightRuntime` applies resolution and fallback policy and emits one
|
||||||
terminal report. It does not own a model or call the canonical loader.
|
terminal resolution report. Explicit post-load publication emits a separate
|
||||||
|
report and may return a lease, but does not revise the completed resolution.
|
||||||
|
The runtime does not own a model or call the canonical loader.
|
||||||
- `HostWeightStore` owns artifact lookup, coordinated construction,
|
- `HostWeightStore` owns artifact lookup, coordinated construction,
|
||||||
publication, validation, quarantine, and lifecycle.
|
publication, validation, quarantine, and lifecycle.
|
||||||
- `HostWeightLease` owns process-local tensor views, mapped-file resources, and
|
- `HostWeightLease` owns process-local tensor views, mapped-file resources, and
|
||||||
@@ -135,9 +137,23 @@ producer cancellation requires a future process-isolated producer contract. A
|
|||||||
waiter timing out never cancels another process's valid build.
|
waiter timing out never cancels another process's valid build.
|
||||||
|
|
||||||
Remote providers are represented by protocols but are explicitly unsupported
|
Remote providers are represented by protocols but are explicitly unsupported
|
||||||
in the first implementation. V1 also invokes only `PRE_LOAD_SAFE` producers;
|
in the first implementation. Pre-load `resolve()` invokes only
|
||||||
enabling post-load publication is rejected during configuration until a
|
`PRE_LOAD_SAFE` producers. A loader that completed canonical materialization
|
||||||
separate post-load entry point and model-lifetime contract exist.
|
may explicitly call `publish_after_load()` with a `POST_LOAD_ONLY` producer
|
||||||
|
when `allow_post_load_publish` is enabled.
|
||||||
|
|
||||||
|
`allow_local_build` controls producer use during pre-load resolution;
|
||||||
|
`allow_post_load_publish` independently controls the explicit post-load entry
|
||||||
|
point. This permits lookup-only warm starts that populate a missing artifact
|
||||||
|
only after the canonical loader has established the current model.
|
||||||
|
|
||||||
|
Post-load publication is synchronous in V1, but its outcome is independent of
|
||||||
|
the already completed resolution. A store failure is returned in a separate
|
||||||
|
publication report and cannot turn a valid canonical model into a failed
|
||||||
|
resolution. On success, the runtime closes the validated store lease before
|
||||||
|
returning the report. Post-load publication only warms future startups; it does
|
||||||
|
not restore, rebind, or otherwise mutate the canonically loaded model serving
|
||||||
|
the current startup.
|
||||||
|
|
||||||
## Restoration transaction boundary
|
## Restoration transaction boundary
|
||||||
|
|
||||||
|
|||||||
@@ -142,9 +142,11 @@ def test_restoration_contract_has_a_validation_only_one_shot_commit_boundary() -
|
|||||||
assert "without mutating" in (WeightRestorer.__doc__ or "")
|
assert "without mutating" in (WeightRestorer.__doc__ or "")
|
||||||
|
|
||||||
|
|
||||||
def test_post_load_publication_policy_is_explicitly_unsupported() -> None:
|
def test_post_load_publication_policy_is_explicitly_enabled() -> None:
|
||||||
with pytest.raises(ValueError, match="post-load publication is not implemented"):
|
policy = ProductionPolicy(allow_post_load_publish=True)
|
||||||
ProductionPolicy(allow_post_load_publish=True)
|
|
||||||
|
assert policy.allow_local_build
|
||||||
|
assert policy.allow_post_load_publish
|
||||||
|
|
||||||
|
|
||||||
def test_manifest_and_publication_marker_round_trip_exactly() -> None:
|
def test_manifest_and_publication_marker_round_trip_exactly() -> None:
|
||||||
|
|||||||
+247
-2
@@ -1,10 +1,11 @@
|
|||||||
# SPDX-License-Identifier: Apache-2.0
|
# SPDX-License-Identifier: Apache-2.0
|
||||||
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
|
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
|
||||||
"""Resolution-policy tests for the loader-adjacent Host Weight Runtime."""
|
"""Runtime resolution and publication-policy tests for Host Weight Runtime."""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import cast
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
import torch
|
import torch
|
||||||
@@ -15,12 +16,16 @@ from vllm_omni.host_weight_runtime import (
|
|||||||
ComponentIdentity,
|
ComponentIdentity,
|
||||||
FailureCode,
|
FailureCode,
|
||||||
HostWeightFailure,
|
HostWeightFailure,
|
||||||
|
HostWeightLease,
|
||||||
HostWeightRuntime,
|
HostWeightRuntime,
|
||||||
HostWeightRuntimeConfig,
|
HostWeightRuntimeConfig,
|
||||||
IntegrityPolicy,
|
IntegrityPolicy,
|
||||||
LookupPhase,
|
LookupPhase,
|
||||||
|
PostLoadPublicationOutcome,
|
||||||
|
PostLoadPublicationReport,
|
||||||
ProducerIdentity,
|
ProducerIdentity,
|
||||||
ProductionMetadata,
|
ProductionMetadata,
|
||||||
|
ProductionPolicy,
|
||||||
ProductionSourceMode,
|
ProductionSourceMode,
|
||||||
RemoteImportPolicy,
|
RemoteImportPolicy,
|
||||||
RemoteOnMiss,
|
RemoteOnMiss,
|
||||||
@@ -39,7 +44,7 @@ from vllm_omni.host_weight_runtime import (
|
|||||||
WeightRepresentation,
|
WeightRepresentation,
|
||||||
WeightSourceIdentity,
|
WeightSourceIdentity,
|
||||||
)
|
)
|
||||||
from vllm_omni.host_weight_runtime.filesystem import detect_storage_class
|
from vllm_omni.host_weight_runtime.filesystem import FilesystemHostWeightStore, detect_storage_class
|
||||||
|
|
||||||
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
|
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
|
||||||
|
|
||||||
@@ -224,6 +229,246 @@ def test_preload_resolution_does_not_run_postload_only_producer(tmp_path: Path)
|
|||||||
assert resolution.report.attempts[0].action.value == "canonical_fallback"
|
assert resolution.report.attempts[0].action.value == "canonical_fallback"
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_closes_lease_and_only_warms_future_resolution(
|
||||||
|
tmp_path: Path,
|
||||||
|
) -> None:
|
||||||
|
publications: list[PostLoadPublicationReport] = []
|
||||||
|
identity = _identity()
|
||||||
|
producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.PREFERRED,
|
||||||
|
domain=_domain(tmp_path / "store"),
|
||||||
|
production=ProductionPolicy(
|
||||||
|
allow_local_build=False,
|
||||||
|
allow_post_load_publish=True,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
publication_observer=publications.append,
|
||||||
|
)
|
||||||
|
|
||||||
|
preload = runtime.resolve(identity, producer=producer)
|
||||||
|
assert preload.report.outcome is ResolutionOutcome.CANONICAL_FALLBACK
|
||||||
|
assert producer.calls == 0
|
||||||
|
|
||||||
|
publication = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.PUBLISHED
|
||||||
|
assert publication.identity_digest == identity.key
|
||||||
|
|
||||||
|
already_present = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
assert already_present.outcome is PostLoadPublicationOutcome.ALREADY_PRESENT
|
||||||
|
|
||||||
|
warm = runtime.resolve(identity)
|
||||||
|
assert warm.report.outcome is ResolutionOutcome.LOCAL_HIT
|
||||||
|
assert warm.lease is not None
|
||||||
|
assert torch.equal(
|
||||||
|
warm.lease.tensors["weight"],
|
||||||
|
torch.arange(4, dtype=torch.float32).to(torch.bfloat16).reshape(2, 2),
|
||||||
|
)
|
||||||
|
warm.lease.close()
|
||||||
|
assert producer.calls == 1
|
||||||
|
assert publications == [publication, already_present]
|
||||||
|
assert isinstance(runtime.store, FilesystemHostWeightStore)
|
||||||
|
assert runtime.store.cleanup(identity) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_policy_skip_does_not_run_producer(tmp_path: Path) -> None:
|
||||||
|
identity = _identity()
|
||||||
|
producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(mode=RuntimeMode.PREFERRED, domain=_domain(tmp_path / "store"))
|
||||||
|
)
|
||||||
|
|
||||||
|
publication = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.POLICY_DISABLED
|
||||||
|
assert producer.calls == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_reports_lease_cleanup_failure(
|
||||||
|
tmp_path: Path,
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
identity = _identity()
|
||||||
|
producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.PREFERRED,
|
||||||
|
domain=_domain(tmp_path / "store"),
|
||||||
|
production=ProductionPolicy(allow_post_load_publish=True),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert runtime.store is not None
|
||||||
|
|
||||||
|
class FailingLease:
|
||||||
|
def close(self) -> None:
|
||||||
|
raise RuntimeError("injected lease cleanup failure")
|
||||||
|
|
||||||
|
def return_failing_lease(*_args: object, **_kwargs: object) -> StoreResult:
|
||||||
|
return StoreResult(StoreStatus.BUILT, lease=cast(HostWeightLease, FailingLease()))
|
||||||
|
|
||||||
|
monkeypatch.setattr(runtime.store, "get_or_build", return_failing_lease)
|
||||||
|
|
||||||
|
publication = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.FAILED
|
||||||
|
assert publication.failure is not None
|
||||||
|
assert publication.failure.stage is ResolutionStage.LIFECYCLE
|
||||||
|
assert publication.failure.code is FailureCode.PUBLICATION_FAILED
|
||||||
|
assert publication.failure.details.to_value() == {
|
||||||
|
"exception_type": "RuntimeError",
|
||||||
|
"store_outcome": "built",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_disabled_postload_publication_does_not_probe_configured_root(tmp_path: Path) -> None:
|
||||||
|
root = tmp_path / "must-not-exist"
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.DISABLED,
|
||||||
|
domain=_domain(root),
|
||||||
|
production=ProductionPolicy(allow_post_load_publish=True),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
publication = runtime.publish_after_load()
|
||||||
|
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.RUNTIME_DISABLED
|
||||||
|
assert not root.exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_rejects_preload_producer(tmp_path: Path) -> None:
|
||||||
|
identity = _identity()
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.PREFERRED,
|
||||||
|
domain=_domain(tmp_path / "store"),
|
||||||
|
production=ProductionPolicy(allow_post_load_publish=True),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="POST_LOAD_ONLY"):
|
||||||
|
runtime.publish_after_load(identity, producer=CountingProducer(identity))
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_failure_is_reported_without_raising(
|
||||||
|
tmp_path: Path,
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
identity = _identity()
|
||||||
|
producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.PREFERRED,
|
||||||
|
domain=_domain(tmp_path / "store"),
|
||||||
|
production=ProductionPolicy(allow_post_load_publish=True),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert runtime.store is not None
|
||||||
|
failure = HostWeightFailure(
|
||||||
|
stage=ResolutionStage.CAPACITY,
|
||||||
|
code=FailureCode.ENOSPC,
|
||||||
|
retryable=True,
|
||||||
|
message="injected post-load publication failure",
|
||||||
|
details=CanonicalJson.empty(),
|
||||||
|
)
|
||||||
|
|
||||||
|
def fail_publication(*_args: object, **_kwargs: object) -> StoreResult:
|
||||||
|
return StoreResult(StoreStatus.FAILED, failure=failure)
|
||||||
|
|
||||||
|
monkeypatch.setattr(runtime.store, "get_or_build", fail_publication)
|
||||||
|
publication = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.FAILED
|
||||||
|
assert publication.failure is failure
|
||||||
|
assert producer.calls == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_reports_store_initialization_failure(tmp_path: Path) -> None:
|
||||||
|
root = tmp_path / "not-a-directory"
|
||||||
|
root.write_text("occupied", encoding="utf-8")
|
||||||
|
identity = _identity()
|
||||||
|
producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.PREFERRED,
|
||||||
|
domain=_domain(root),
|
||||||
|
production=ProductionPolicy(allow_post_load_publish=True),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
publication = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.FAILED
|
||||||
|
assert publication.failure is not None
|
||||||
|
assert publication.failure.code is FailureCode.DOMAIN_UNAVAILABLE
|
||||||
|
assert publication.failure.retryable
|
||||||
|
assert producer.calls == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_maps_joined_and_closes_lease(
|
||||||
|
tmp_path: Path,
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
identity = _identity()
|
||||||
|
producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.PREFERRED,
|
||||||
|
domain=_domain(tmp_path / "store"),
|
||||||
|
production=ProductionPolicy(allow_post_load_publish=True),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert runtime.store is not None
|
||||||
|
|
||||||
|
class ClosingLease:
|
||||||
|
closed = False
|
||||||
|
|
||||||
|
def close(self) -> None:
|
||||||
|
self.closed = True
|
||||||
|
|
||||||
|
lease = ClosingLease()
|
||||||
|
|
||||||
|
def return_joined_lease(*_args: object, **_kwargs: object) -> StoreResult:
|
||||||
|
return StoreResult(StoreStatus.JOINED, lease=cast(HostWeightLease, lease))
|
||||||
|
|
||||||
|
monkeypatch.setattr(runtime.store, "get_or_build", return_joined_lease)
|
||||||
|
|
||||||
|
publication = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.JOINED
|
||||||
|
assert lease.closed
|
||||||
|
|
||||||
|
|
||||||
|
def test_postload_publication_synthesizes_failure_for_unexpected_store_status(
|
||||||
|
tmp_path: Path,
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
identity = _identity()
|
||||||
|
producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
|
||||||
|
runtime = HostWeightRuntime.from_config(
|
||||||
|
HostWeightRuntimeConfig(
|
||||||
|
mode=RuntimeMode.PREFERRED,
|
||||||
|
domain=_domain(tmp_path / "store"),
|
||||||
|
production=ProductionPolicy(allow_post_load_publish=True),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert runtime.store is not None
|
||||||
|
|
||||||
|
def return_unexpected_miss(*_args: object, **_kwargs: object) -> StoreResult:
|
||||||
|
return StoreResult(StoreStatus.MISS)
|
||||||
|
|
||||||
|
monkeypatch.setattr(runtime.store, "get_or_build", return_unexpected_miss)
|
||||||
|
|
||||||
|
publication = runtime.publish_after_load(identity, producer=producer)
|
||||||
|
|
||||||
|
assert publication.outcome is PostLoadPublicationOutcome.FAILED
|
||||||
|
assert publication.failure is not None
|
||||||
|
assert publication.failure.stage is ResolutionStage.PRODUCTION
|
||||||
|
assert publication.failure.code is FailureCode.PRODUCER_FAILED
|
||||||
|
assert not publication.failure.retryable
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
("mode", "expected"),
|
("mode", "expected"),
|
||||||
[
|
[
|
||||||
@@ -47,6 +47,8 @@ from .manifest import (
|
|||||||
from .outcomes import (
|
from .outcomes import (
|
||||||
AttemptResult,
|
AttemptResult,
|
||||||
HostWeightResolution,
|
HostWeightResolution,
|
||||||
|
PostLoadPublicationOutcome,
|
||||||
|
PostLoadPublicationReport,
|
||||||
ResolutionAttempt,
|
ResolutionAttempt,
|
||||||
ResolutionOutcome,
|
ResolutionOutcome,
|
||||||
ResolutionReport,
|
ResolutionReport,
|
||||||
@@ -102,6 +104,8 @@ __all__ = [
|
|||||||
"ManifestValidationError",
|
"ManifestValidationError",
|
||||||
"MappedHostRegion",
|
"MappedHostRegion",
|
||||||
"ProducerIdentity",
|
"ProducerIdentity",
|
||||||
|
"PostLoadPublicationOutcome",
|
||||||
|
"PostLoadPublicationReport",
|
||||||
"ProductionMetadata",
|
"ProductionMetadata",
|
||||||
"ProductionPolicy",
|
"ProductionPolicy",
|
||||||
"ProductionSourceMode",
|
"ProductionSourceMode",
|
||||||
|
|||||||
@@ -117,8 +117,6 @@ class ProductionPolicy:
|
|||||||
def __post_init__(self) -> None:
|
def __post_init__(self) -> None:
|
||||||
if not isinstance(self.allow_local_build, bool) or not isinstance(self.allow_post_load_publish, bool):
|
if not isinstance(self.allow_local_build, bool) or not isinstance(self.allow_post_load_publish, bool):
|
||||||
raise ValueError("production switches must be booleans")
|
raise ValueError("production switches must be booleans")
|
||||||
if self.allow_post_load_publish:
|
|
||||||
raise ValueError("post-load publication is not implemented")
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
# SPDX-License-Identifier: Apache-2.0
|
# SPDX-License-Identifier: Apache-2.0
|
||||||
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
|
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
|
||||||
"""Typed store and end-to-end resolution outcomes."""
|
"""Typed store, resolution, and post-load publication outcomes."""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
@@ -60,6 +60,17 @@ class ResolutionOutcome(str, Enum):
|
|||||||
FAILED = "failed"
|
FAILED = "failed"
|
||||||
|
|
||||||
|
|
||||||
|
class PostLoadPublicationOutcome(str, Enum):
|
||||||
|
"""Terminal outcome for one explicit post-load publication attempt."""
|
||||||
|
|
||||||
|
RUNTIME_DISABLED = "runtime_disabled"
|
||||||
|
POLICY_DISABLED = "policy_disabled"
|
||||||
|
ALREADY_PRESENT = "already_present"
|
||||||
|
PUBLISHED = "published"
|
||||||
|
JOINED = "joined"
|
||||||
|
FAILED = "failed"
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class ResolutionAttempt:
|
class ResolutionAttempt:
|
||||||
stage: ResolutionStage
|
stage: ResolutionStage
|
||||||
@@ -108,9 +119,38 @@ class HostWeightResolution:
|
|||||||
raise ValueError("resolution outcome has inconsistent lease ownership")
|
raise ValueError("resolution outcome has inconsistent lease ownership")
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class PostLoadPublicationReport:
|
||||||
|
operation_id: str
|
||||||
|
outcome: PostLoadPublicationOutcome
|
||||||
|
identity_digest: str | None
|
||||||
|
elapsed_seconds: float
|
||||||
|
failure: HostWeightFailure | None = None
|
||||||
|
|
||||||
|
def __post_init__(self) -> None:
|
||||||
|
if not self.operation_id:
|
||||||
|
raise ValueError("post-load publication report requires an operation_id")
|
||||||
|
if not isinstance(self.outcome, PostLoadPublicationOutcome):
|
||||||
|
raise ValueError("post-load publication outcome must use PostLoadPublicationOutcome")
|
||||||
|
if self.identity_digest is not None and not self.identity_digest:
|
||||||
|
raise ValueError("post-load publication identity digest must not be empty")
|
||||||
|
requires_identity = self.outcome not in {
|
||||||
|
PostLoadPublicationOutcome.RUNTIME_DISABLED,
|
||||||
|
PostLoadPublicationOutcome.POLICY_DISABLED,
|
||||||
|
}
|
||||||
|
if requires_identity != (self.identity_digest is not None):
|
||||||
|
raise ValueError("post-load publication identity ownership is inconsistent")
|
||||||
|
if not math.isfinite(self.elapsed_seconds) or self.elapsed_seconds < 0:
|
||||||
|
raise ValueError("post-load publication elapsed_seconds must be finite and non-negative")
|
||||||
|
if (self.outcome is PostLoadPublicationOutcome.FAILED) != (self.failure is not None):
|
||||||
|
raise ValueError("post-load publication failure ownership is inconsistent")
|
||||||
|
|
||||||
|
|
||||||
__all__ = [
|
__all__ = [
|
||||||
"AttemptResult",
|
"AttemptResult",
|
||||||
"HostWeightResolution",
|
"HostWeightResolution",
|
||||||
|
"PostLoadPublicationOutcome",
|
||||||
|
"PostLoadPublicationReport",
|
||||||
"ResolutionAttempt",
|
"ResolutionAttempt",
|
||||||
"ResolutionOutcome",
|
"ResolutionOutcome",
|
||||||
"ResolutionReport",
|
"ResolutionReport",
|
||||||
|
|||||||
@@ -16,6 +16,8 @@ from .lease import HostWeightLease
|
|||||||
from .outcomes import (
|
from .outcomes import (
|
||||||
AttemptResult,
|
AttemptResult,
|
||||||
HostWeightResolution,
|
HostWeightResolution,
|
||||||
|
PostLoadPublicationOutcome,
|
||||||
|
PostLoadPublicationReport,
|
||||||
ResolutionAttempt,
|
ResolutionAttempt,
|
||||||
ResolutionOutcome,
|
ResolutionOutcome,
|
||||||
ResolutionReport,
|
ResolutionReport,
|
||||||
@@ -28,7 +30,7 @@ logger = logging.getLogger(__name__)
|
|||||||
|
|
||||||
|
|
||||||
class HostWeightRuntime:
|
class HostWeightRuntime:
|
||||||
"""Resolve one exact representation without owning model restoration."""
|
"""Resolve or publish one exact representation without owning restoration."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@@ -36,10 +38,13 @@ class HostWeightRuntime:
|
|||||||
store: HostWeightStore | None = None,
|
store: HostWeightStore | None = None,
|
||||||
*,
|
*,
|
||||||
observer: Callable[[ResolutionReport], None] | None = None,
|
observer: Callable[[ResolutionReport], None] | None = None,
|
||||||
|
publication_observer: Callable[[PostLoadPublicationReport], None] | None = None,
|
||||||
_initialization_failure: HostWeightFailure | None = None,
|
_initialization_failure: HostWeightFailure | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
if observer is not None and not callable(observer):
|
if observer is not None and not callable(observer):
|
||||||
raise ValueError("Host Weight Runtime observer must be callable")
|
raise ValueError("Host Weight Runtime observer must be callable")
|
||||||
|
if publication_observer is not None and not callable(publication_observer):
|
||||||
|
raise ValueError("Host Weight Runtime publication observer must be callable")
|
||||||
if config.mode is RuntimeMode.DISABLED and (store is not None or _initialization_failure is not None):
|
if config.mode is RuntimeMode.DISABLED and (store is not None or _initialization_failure is not None):
|
||||||
raise ValueError("disabled Host Weight Runtime must not initialize a store")
|
raise ValueError("disabled Host Weight Runtime must not initialize a store")
|
||||||
if config.mode is not RuntimeMode.DISABLED and (store is None) == (_initialization_failure is None):
|
if config.mode is not RuntimeMode.DISABLED and (store is None) == (_initialization_failure is None):
|
||||||
@@ -48,6 +53,7 @@ class HostWeightRuntime:
|
|||||||
self.store = store
|
self.store = store
|
||||||
self._initialization_failure = _initialization_failure
|
self._initialization_failure = _initialization_failure
|
||||||
self._observer = observer
|
self._observer = observer
|
||||||
|
self._publication_observer = publication_observer
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_config(
|
def from_config(
|
||||||
@@ -55,18 +61,24 @@ class HostWeightRuntime:
|
|||||||
config: HostWeightRuntimeConfig,
|
config: HostWeightRuntimeConfig,
|
||||||
*,
|
*,
|
||||||
observer: Callable[[ResolutionReport], None] | None = None,
|
observer: Callable[[ResolutionReport], None] | None = None,
|
||||||
|
publication_observer: Callable[[PostLoadPublicationReport], None] | None = None,
|
||||||
) -> HostWeightRuntime:
|
) -> HostWeightRuntime:
|
||||||
"""Construct lazily so disabled mode performs no filesystem probe."""
|
"""Construct lazily so disabled mode performs no filesystem probe."""
|
||||||
if config.mode is RuntimeMode.DISABLED:
|
if config.mode is RuntimeMode.DISABLED:
|
||||||
return cls(config, observer=observer)
|
return cls(config, observer=observer, publication_observer=publication_observer)
|
||||||
assert config.domain is not None
|
assert config.domain is not None
|
||||||
from .filesystem import FilesystemHostWeightStore # noqa: PLC0415
|
from .filesystem import FilesystemHostWeightStore # noqa: PLC0415
|
||||||
|
|
||||||
try:
|
try:
|
||||||
store = FilesystemHostWeightStore(config.domain, config.capacity, config.integrity)
|
store = FilesystemHostWeightStore(config.domain, config.capacity, config.integrity)
|
||||||
except HostWeightError as exc:
|
except HostWeightError as exc:
|
||||||
return cls(config, observer=observer, _initialization_failure=exc.failure)
|
return cls(
|
||||||
return cls(config, store, observer=observer)
|
config,
|
||||||
|
observer=observer,
|
||||||
|
publication_observer=publication_observer,
|
||||||
|
_initialization_failure=exc.failure,
|
||||||
|
)
|
||||||
|
return cls(config, store, observer=observer, publication_observer=publication_observer)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def enabled(self) -> bool:
|
def enabled(self) -> bool:
|
||||||
@@ -223,6 +235,123 @@ class HostWeightRuntime:
|
|||||||
|
|
||||||
return self._finish_terminal(tuple(attempts), started)
|
return self._finish_terminal(tuple(attempts), started)
|
||||||
|
|
||||||
|
def publish_after_load(
|
||||||
|
self,
|
||||||
|
identity: WeightArtifactIdentity | None = None,
|
||||||
|
*,
|
||||||
|
producer: WeightProducer | None = None,
|
||||||
|
) -> PostLoadPublicationReport:
|
||||||
|
"""Synchronously warm a future startup without revising resolution.
|
||||||
|
|
||||||
|
A successful store operation returns a validated lease. The runtime
|
||||||
|
closes that lease before returning because POST_LOAD_ONLY production
|
||||||
|
must not mutate or rebind the model serving the current startup.
|
||||||
|
"""
|
||||||
|
started = time.monotonic()
|
||||||
|
operation_id = uuid.uuid4().hex
|
||||||
|
if not self.enabled:
|
||||||
|
return self._finish_publication(
|
||||||
|
PostLoadPublicationReport(
|
||||||
|
operation_id=operation_id,
|
||||||
|
outcome=PostLoadPublicationOutcome.RUNTIME_DISABLED,
|
||||||
|
identity_digest=None,
|
||||||
|
elapsed_seconds=time.monotonic() - started,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
if not self.config.production.allow_post_load_publish:
|
||||||
|
return self._finish_publication(
|
||||||
|
PostLoadPublicationReport(
|
||||||
|
operation_id=operation_id,
|
||||||
|
outcome=PostLoadPublicationOutcome.POLICY_DISABLED,
|
||||||
|
identity_digest=None,
|
||||||
|
elapsed_seconds=time.monotonic() - started,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
if identity is None:
|
||||||
|
raise ValueError("enabled post-load publication requires an exact artifact identity")
|
||||||
|
if producer is None:
|
||||||
|
raise ValueError("enabled post-load publication requires one producer")
|
||||||
|
if producer.spec.lookup_phase is not LookupPhase.POST_LOAD_ONLY:
|
||||||
|
raise ValueError("publish_after_load requires a POST_LOAD_ONLY producer")
|
||||||
|
|
||||||
|
identity_digest = identity.key
|
||||||
|
if self._initialization_failure is not None:
|
||||||
|
return self._finish_publication(
|
||||||
|
PostLoadPublicationReport(
|
||||||
|
operation_id=operation_id,
|
||||||
|
outcome=PostLoadPublicationOutcome.FAILED,
|
||||||
|
identity_digest=identity_digest,
|
||||||
|
elapsed_seconds=time.monotonic() - started,
|
||||||
|
failure=self._initialization_failure,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
assert self.store is not None
|
||||||
|
result = self.store.get_or_build(
|
||||||
|
BuildRequest(identity),
|
||||||
|
producer,
|
||||||
|
validation=self.config.integrity.local_lookup,
|
||||||
|
deadline=started + self.config.wait.coordination_timeout_seconds,
|
||||||
|
)
|
||||||
|
successful_outcomes = {
|
||||||
|
StoreStatus.HIT: PostLoadPublicationOutcome.ALREADY_PRESENT,
|
||||||
|
StoreStatus.BUILT: PostLoadPublicationOutcome.PUBLISHED,
|
||||||
|
StoreStatus.JOINED: PostLoadPublicationOutcome.JOINED,
|
||||||
|
}
|
||||||
|
outcome = successful_outcomes.get(result.status)
|
||||||
|
if outcome is not None:
|
||||||
|
assert result.lease is not None
|
||||||
|
try:
|
||||||
|
result.lease.close()
|
||||||
|
except Exception as exc:
|
||||||
|
return self._finish_publication(
|
||||||
|
PostLoadPublicationReport(
|
||||||
|
operation_id=operation_id,
|
||||||
|
outcome=PostLoadPublicationOutcome.FAILED,
|
||||||
|
identity_digest=identity_digest,
|
||||||
|
elapsed_seconds=time.monotonic() - started,
|
||||||
|
failure=HostWeightFailure(
|
||||||
|
stage=ResolutionStage.LIFECYCLE,
|
||||||
|
code=FailureCode.PUBLICATION_FAILED,
|
||||||
|
retryable=False,
|
||||||
|
message="post-load publication lease cleanup failed",
|
||||||
|
details=CanonicalJson.from_value(
|
||||||
|
{
|
||||||
|
"exception_type": type(exc).__name__,
|
||||||
|
"store_outcome": result.status.value,
|
||||||
|
}
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return self._finish_publication(
|
||||||
|
PostLoadPublicationReport(
|
||||||
|
operation_id=operation_id,
|
||||||
|
outcome=outcome,
|
||||||
|
identity_digest=identity_digest,
|
||||||
|
elapsed_seconds=time.monotonic() - started,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
failure = result.failure
|
||||||
|
if failure is None:
|
||||||
|
failure = HostWeightFailure(
|
||||||
|
stage=ResolutionStage.PRODUCTION,
|
||||||
|
code=FailureCode.PRODUCER_FAILED,
|
||||||
|
retryable=False,
|
||||||
|
message=f"post-load publication returned unexpected store status {result.status.value}",
|
||||||
|
details=CanonicalJson.empty(),
|
||||||
|
)
|
||||||
|
return self._finish_publication(
|
||||||
|
PostLoadPublicationReport(
|
||||||
|
operation_id=operation_id,
|
||||||
|
outcome=PostLoadPublicationOutcome.FAILED,
|
||||||
|
identity_digest=identity_digest,
|
||||||
|
elapsed_seconds=time.monotonic() - started,
|
||||||
|
failure=failure,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
def _mode_terminal_action(self) -> ResolutionAction:
|
def _mode_terminal_action(self) -> ResolutionAction:
|
||||||
if self.config.mode is RuntimeMode.PREFERRED:
|
if self.config.mode is RuntimeMode.PREFERRED:
|
||||||
return ResolutionAction.CANONICAL_FALLBACK
|
return ResolutionAction.CANONICAL_FALLBACK
|
||||||
@@ -328,5 +457,19 @@ class HostWeightRuntime:
|
|||||||
)
|
)
|
||||||
return HostWeightResolution(report=report, lease=lease)
|
return HostWeightResolution(report=report, lease=lease)
|
||||||
|
|
||||||
|
def _finish_publication(
|
||||||
|
self,
|
||||||
|
report: PostLoadPublicationReport,
|
||||||
|
) -> PostLoadPublicationReport:
|
||||||
|
if self._publication_observer is not None:
|
||||||
|
try:
|
||||||
|
self._publication_observer(report)
|
||||||
|
except Exception:
|
||||||
|
logger.exception(
|
||||||
|
"Host Weight Runtime publication observer failed",
|
||||||
|
extra={"operation_id": report.operation_id},
|
||||||
|
)
|
||||||
|
return report
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["HostWeightRuntime"]
|
__all__ = ["HostWeightRuntime"]
|
||||||
Reference in new issue
Block a user