From e7489ecd52e67b3855bf5e98791b80147de701fa Mon Sep 17 00:00:00 2001 From: RickZ <121943721+rickisba@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:15:50 +0800 Subject: [PATCH 1/3] Fix lifecycle immutability and legacy replica health --- core/__init__.py | 8 +- core/state_models.py | 156 +++++++++++++++++++++++++++ kdn_server/legacy_state.py | 71 ++++++++++++ test/test_legacy_kv_state_mapping.py | 66 ++++++++++++ test/test_state_models.py | 43 ++++++++ 5 files changed, 343 insertions(+), 1 deletion(-) create mode 100644 core/state_models.py create mode 100644 kdn_server/legacy_state.py create mode 100644 test/test_legacy_kv_state_mapping.py create mode 100644 test/test_state_models.py diff --git a/core/__init__.py b/core/__init__.py index 974f5ab..9810e1f 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -3,4 +3,10 @@ from .model_calculation import MLAmodel from .request import Request, Prompt, Service, Task from .tokenizer_registry import TokenizerRegistry -from .fwd import forward_request + + +def forward_request(*args, **kwargs): + """Import the optional HTTP forwarding dependency only when it is used.""" + from .fwd import forward_request as implementation + + return implementation(*args, **kwargs) diff --git a/core/state_models.py b/core/state_models.py new file mode 100644 index 0000000..9252d16 --- /dev/null +++ b/core/state_models.py @@ -0,0 +1,156 @@ +"""Shared lifecycle contracts for cache and data-plane work.""" + +from __future__ import annotations + +import hashlib +from enum import Enum +from typing import Any, ClassVar, Mapping + +from pydantic import BaseModel, ConfigDict + + +class ArtifactState(str, Enum): + PENDING = "pending" + BUILDING = "building" + READY = "ready" + FAILED = "failed" + DELETING = "deleting" + DELETED = "deleted" + + +class ReplicaState(str, Enum): + PENDING = "pending" + COPYING = "copying" + READY = "ready" + FAILED = "failed" + DELETING = "deleting" + DELETED = "deleted" + + +class DataPlaneTaskState(str, Enum): + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + EXPIRED = "expired" + + +class QueueWorkState(str, Enum): + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + SKIPPED = "skipped" + + +class TransitionResult(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + previous_state: str + state: str + changed: bool + + +class InvalidStateTransition(ValueError): + """Raised when a lifecycle transition is outside the declared contract.""" + + def __init__(self, current_state: Enum, target_state: Enum, allowed_targets: set[Enum]): + self.detail = { + "code": "invalid_state_transition", + "current_state": current_state.value, + "target_state": target_state.value, + "allowed_targets": sorted(state.value for state in allowed_targets), + } + super().__init__(str(self.detail)) + + +class _StateModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + _transitions: ClassVar[Mapping[Enum, set[Enum]]] + + def transition_to(self, target_state: Enum) -> tuple["_StateModel", TransitionResult]: + state = self.state + target = type(state)(target_state) + if target == state: + return self, TransitionResult(previous_state=state.value, state=state.value, changed=False) + allowed = self._transitions.get(state, set()) + if target not in allowed: + raise InvalidStateTransition(state, target, allowed) + changed = self.model_copy(update={"state": target}) + return changed, TransitionResult(previous_state=state.value, state=target.value, changed=True) + + +ARTIFACT_TRANSITIONS = { + ArtifactState.PENDING: {ArtifactState.BUILDING, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.BUILDING: {ArtifactState.READY, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.READY: {ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.FAILED: {ArtifactState.BUILDING, ArtifactState.DELETING}, + ArtifactState.DELETING: {ArtifactState.DELETED, ArtifactState.FAILED}, + ArtifactState.DELETED: set(), +} + +REPLICA_TRANSITIONS = { + ReplicaState.PENDING: {ReplicaState.COPYING, ReplicaState.FAILED, ReplicaState.DELETING}, + ReplicaState.COPYING: {ReplicaState.READY, ReplicaState.FAILED, ReplicaState.DELETING}, + ReplicaState.READY: {ReplicaState.FAILED, ReplicaState.DELETING}, + ReplicaState.FAILED: {ReplicaState.COPYING, ReplicaState.DELETING}, + ReplicaState.DELETING: {ReplicaState.DELETED, ReplicaState.FAILED}, + ReplicaState.DELETED: set(), +} + +TASK_TRANSITIONS = { + DataPlaneTaskState.PENDING: {DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.RUNNING: {DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.FAILED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.FAILED: {DataPlaneTaskState.PENDING, DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.SUCCEEDED: set(), DataPlaneTaskState.CANCELLED: set(), DataPlaneTaskState.EXPIRED: set(), +} + +WORK_TRANSITIONS = { + QueueWorkState.PENDING: {QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.RUNNING: {QueueWorkState.SUCCEEDED, QueueWorkState.FAILED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.FAILED: {QueueWorkState.PENDING, QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.SUCCEEDED: set(), QueueWorkState.CANCELLED: set(), QueueWorkState.SKIPPED: set(), +} + + +class CacheArtifact(_StateModel): + artifact_id: str + kid: str + state: ArtifactState = ArtifactState.PENDING + _transitions = ARTIFACT_TRANSITIONS + + +class CacheReplica(_StateModel): + replica_id: str + artifact_id: str + location_key: str + state: ReplicaState = ReplicaState.PENDING + _transitions = REPLICA_TRANSITIONS + + +class DataPlaneTask(_StateModel): + task_id: str + state: DataPlaneTaskState = DataPlaneTaskState.PENDING + _transitions = TASK_TRANSITIONS + + +class QueueWork(_StateModel): + work_id: str + state: QueueWorkState = QueueWorkState.PENDING + _transitions = WORK_TRANSITIONS + + +def stable_id(kind: str, *identity: Any) -> str: + """Return a deterministic identifier from an explicitly ordered identity tuple.""" + material = "\x1f".join([kind, *(str(value) for value in identity)]) + return f"{kind}_{hashlib.sha256(material.encode()).hexdigest()[:24]}" + + +def artifact_id(kid: str) -> str: + return stable_id("artifact", kid) + + +def replica_id(artifact: str, location_key: str) -> str: + return stable_id("replica", artifact, location_key) diff --git a/kdn_server/legacy_state.py b/kdn_server/legacy_state.py new file mode 100644 index 0000000..60ee10f --- /dev/null +++ b/kdn_server/legacy_state.py @@ -0,0 +1,71 @@ +"""Read-only projection of Legacy ``kv_ready`` rows onto lifecycle models.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, ConfigDict + +from core.state_models import ( + ArtifactState, CacheArtifact, CacheReplica, ReplicaState, artifact_id, replica_id, +) + + +class LegacyStateWarning(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + code: str + message: str + + +class LegacyKVStateView(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + compatibility_status: str = "unknown" + artifact: CacheArtifact + replica: CacheReplica + replica_directory_exists: bool + warnings: tuple[LegacyStateWarning, ...] = () + + +def _safe_component(value: Any, field: str, *, allow_dot: bool = False) -> str: + component = str(value or "").strip() + if component == "." and allow_dot: + return component + path = Path(component) + if not component or component == ".." or (component == "." and not allow_dot) or path.is_absolute() or len(path.parts) != 1: + raise ValueError(f"{field} must be a non-empty single path component") + return component + + +def map_legacy_kv_state(row: Any, kv_root: str | Path) -> LegacyKVStateView: + """Map a Legacy database row without mutating either storage or the filesystem.""" + get = row.get if hasattr(row, "get") else lambda key, default=None: getattr(row, key, default) + kid = _safe_component(get("kid"), "kid") + raw_rel_dir = get("kv_rel_dir") + rel_dir = None + if raw_rel_dir is not None and str(raw_rel_dir).strip(): + rel_dir = _safe_component(raw_rel_dir, "kv_rel_dir", allow_dot=True) + + root = Path(kv_root).resolve(strict=False) + runtime_directory = (root / kid).resolve(strict=False) + if runtime_directory.parent != root: + raise ValueError("kid escapes the configured KV root") + directory_exists = runtime_directory.is_dir() + kv_ready = bool(get("kv_ready", False)) + artifact_state = ArtifactState.READY if kv_ready else ArtifactState.PENDING + replica_state = ReplicaState.READY if kv_ready and directory_exists else ReplicaState.FAILED + artifact = CacheArtifact(artifact_id=artifact_id(kid), kid=kid, state=artifact_state) + replica = CacheReplica( + replica_id=replica_id(artifact.artifact_id, kid), artifact_id=artifact.artifact_id, + location_key=kid, state=replica_state, + ) + warnings = () + if rel_dir is not None and rel_dir != kid: + warnings = (LegacyStateWarning( + code="legacy_kv_rel_dir_mismatch", + message="Legacy kv_rel_dir differs from the runtime kid directory.", + ),) + return LegacyKVStateView( + artifact=artifact, replica=replica, replica_directory_exists=directory_exists, + warnings=warnings, + ) diff --git a/test/test_legacy_kv_state_mapping.py b/test/test_legacy_kv_state_mapping.py new file mode 100644 index 0000000..8a0486e --- /dev/null +++ b/test/test_legacy_kv_state_mapping.py @@ -0,0 +1,66 @@ +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from core.state_models import ReplicaState +from kdn_server.legacy_state import map_legacy_kv_state + + +def row(kid="kid", kv_rel_dir="kid"): + return {"kid": kid, "kv_ready": 1, "kv_rel_dir": kv_rel_dir} + + +def warning_codes(view): + return {warning.code for warning in view.warnings} + + +def test_runtime_kid_directory_is_healthy_and_matching_metadata_has_no_warning(tmp_path): + (tmp_path / "kid").mkdir() + view = map_legacy_kv_state(row(), tmp_path) + assert view.replica.state == ReplicaState.READY + assert view.replica_directory_exists is True + assert view.replica.location_key == "kid" + assert "legacy_kv_rel_dir_mismatch" not in warning_codes(view) + + +def test_stale_existing_metadata_directory_does_not_make_replica_healthy(tmp_path): + (tmp_path / "stale").mkdir() + view = map_legacy_kv_state(row(kv_rel_dir="stale"), tmp_path) + assert view.replica.state == ReplicaState.FAILED + assert view.replica_directory_exists is False + assert "legacy_kv_rel_dir_mismatch" in warning_codes(view) + + +def test_dot_metadata_does_not_treat_root_as_replica(tmp_path): + view = map_legacy_kv_state(row(kv_rel_dir="."), tmp_path) + assert view.replica.state == ReplicaState.FAILED + assert view.replica_directory_exists is False + assert "legacy_kv_rel_dir_mismatch" in warning_codes(view) + + +def test_escaping_metadata_is_rejected(tmp_path): + with pytest.raises(ValueError, match="kv_rel_dir"): + map_legacy_kv_state(row(kv_rel_dir="../outside"), tmp_path) + + +def test_projection_and_nested_models_are_frozen(tmp_path): + (tmp_path / "kid").mkdir() + view = map_legacy_kv_state(row(kv_rel_dir="other"), tmp_path) + with pytest.raises(ValidationError, match="Instance is frozen"): + view.compatibility_status = "compatible" + with pytest.raises(ValidationError, match="Instance is frozen"): + view.replica.state = ReplicaState.FAILED + with pytest.raises(ValidationError, match="Instance is frozen"): + view.warnings[0].code = "changed" + + +def test_mapping_does_not_create_or_modify_files(tmp_path): + marker = tmp_path / "marker" + marker.write_text("unchanged") + before = {path.relative_to(tmp_path): (path.stat().st_mtime_ns, path.read_bytes()) + for path in tmp_path.iterdir() if path.is_file()} + map_legacy_kv_state(row(), tmp_path) + after = {path.relative_to(tmp_path): (path.stat().st_mtime_ns, path.read_bytes()) + for path in tmp_path.iterdir() if path.is_file()} + assert after == before diff --git a/test/test_state_models.py b/test/test_state_models.py new file mode 100644 index 0000000..c6c5c80 --- /dev/null +++ b/test/test_state_models.py @@ -0,0 +1,43 @@ +import pytest +from pydantic import ValidationError + +from core.state_models import ( + ArtifactState, CacheArtifact, CacheReplica, DataPlaneTask, DataPlaneTaskState, + InvalidStateTransition, QueueWork, QueueWorkState, ReplicaState, +) + + +@pytest.mark.parametrize("model,target", [ + (CacheArtifact(artifact_id="a", kid="kid"), ArtifactState.READY), + (CacheReplica(replica_id="r", artifact_id="a", location_key="kid"), ReplicaState.READY), + (DataPlaneTask(task_id="t"), DataPlaneTaskState.RUNNING), + (QueueWork(work_id="w"), QueueWorkState.RUNNING), +]) +def test_lifecycle_state_is_frozen(model, target): + with pytest.raises(ValidationError, match="Instance is frozen"): + model.state = target + + +def test_legal_transition_returns_changed_copy_without_mutating_source(): + source = CacheArtifact(artifact_id="a", kid="kid") + building, result = source.transition_to(ArtifactState.BUILDING) + assert source.state == ArtifactState.PENDING + assert building.state == ArtifactState.BUILDING + assert result.changed is True + + +def test_same_state_transition_is_idempotent(): + source = DataPlaneTask(task_id="t") + unchanged, result = source.transition_to(DataPlaneTaskState.PENDING) + assert unchanged is source + assert result.changed is False + + +def test_terminal_transition_has_structured_sorted_detail(): + source = QueueWork(work_id="w", state=QueueWorkState.SUCCEEDED) + with pytest.raises(InvalidStateTransition) as caught: + source.transition_to(QueueWorkState.RUNNING) + assert caught.value.detail == { + "code": "invalid_state_transition", "current_state": "succeeded", + "target_state": "running", "allowed_targets": [], + } From 0fb9ecfb403b5d758f854887398a9bee633bdb79 Mon Sep 17 00:00:00 2001 From: RickZ <121943721+rickisba@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:32:49 +0800 Subject: [PATCH 2/3] Complete cache lifecycle state contract --- README.md | 1 + core/__init__.py | 8 +- core/state_models.py | 267 +++++++++++++++++++-------- doc/state-models.md | 74 ++++++++ kdn_server/legacy_state.py | 93 +++++++--- test/README.md | 7 + test/test_legacy_kv_state_mapping.py | 106 +++++++---- test/test_state_models.py | 136 +++++++++++--- 8 files changed, 528 insertions(+), 164 deletions(-) create mode 100644 doc/state-models.md diff --git a/README.md b/README.md index 8c384fd..9d8284d 100644 --- a/README.md +++ b/README.md @@ -647,6 +647,7 @@ curl -s http://127.0.0.1:7001/debug/strategy | Document | Description | |---|---| +| [`doc/state-models.md`](doc/state-models.md) | Immutable cache lifecycle, task states, stable IDs, transitions, and Legacy mapping. | | [`core/README.md`](core/README.md) | Shared configuration, request model, and multi-machine deployment settings. | | [`scheduler/README.md`](scheduler/README.md) | Global routing, KDN / Proxy pool management, and Scheduler control plane. | | [`proxy/README.md`](proxy/README.md) | Local Instance pool, capability-aware registration, heartbeat recovery, prepare / ready queues, injection strategy, and Proxy resource APIs. | diff --git a/core/__init__.py b/core/__init__.py index 9810e1f..974f5ab 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -3,10 +3,4 @@ from .model_calculation import MLAmodel from .request import Request, Prompt, Service, Task from .tokenizer_registry import TokenizerRegistry - - -def forward_request(*args, **kwargs): - """Import the optional HTTP forwarding dependency only when it is used.""" - from .fwd import forward_request as implementation - - return implementation(*args, **kwargs) +from .fwd import forward_request diff --git a/core/state_models.py b/core/state_models.py index 9252d16..cef5224 100644 --- a/core/state_models.py +++ b/core/state_models.py @@ -1,17 +1,22 @@ -"""Shared lifecycle contracts for cache and data-plane work.""" +"""Versioned, immutable lifecycle contracts for cache and data-plane work.""" from __future__ import annotations import hashlib +import json +import uuid from enum import Enum -from typing import Any, ClassVar, Mapping +from typing import ClassVar, Dict, Mapping, Optional, Set, Tuple, Type, TypeVar -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, Field, field_validator + +SCHEMA_VERSION = "1" class ArtifactState(str, Enum): PENDING = "pending" BUILDING = "building" + STAGING = "staging" READY = "ready" FAILED = "failed" DELETING = "deleting" @@ -20,15 +25,24 @@ class ArtifactState(str, Enum): class ReplicaState(str, Enum): PENDING = "pending" - COPYING = "copying" + STAGING = "staging" READY = "ready" FAILED = "failed" - DELETING = "deleting" + EVICTING = "evicting" DELETED = "deleted" +class ReplicaHealth(str, Enum): + UNKNOWN = "unknown" + HEALTHY = "healthy" + DEGRADED = "degraded" + UNHEALTHY = "unhealthy" + + class DataPlaneTaskState(str, Enum): PENDING = "pending" + QUEUED = "queued" + LEASED = "leased" RUNNING = "running" SUCCEEDED = "succeeded" FAILED = "failed" @@ -38,6 +52,9 @@ class DataPlaneTaskState(str, Enum): class QueueWorkState(str, Enum): PENDING = "pending" + BLOCKED = "blocked" + READY = "ready" + QUEUED = "queued" RUNNING = "running" SUCCEEDED = "succeeded" FAILED = "failed" @@ -45,112 +62,216 @@ class QueueWorkState(str, Enum): SKIPPED = "skipped" -class TransitionResult(BaseModel): - model_config = ConfigDict(extra="forbid", frozen=True) - - previous_state: str - state: str - changed: bool - - -class InvalidStateTransition(ValueError): - """Raised when a lifecycle transition is outside the declared contract.""" - - def __init__(self, current_state: Enum, target_state: Enum, allowed_targets: set[Enum]): - self.detail = { - "code": "invalid_state_transition", - "current_state": current_state.value, - "target_state": target_state.value, - "allowed_targets": sorted(state.value for state in allowed_targets), - } - super().__init__(str(self.detail)) - - -class _StateModel(BaseModel): - model_config = ConfigDict(extra="forbid", frozen=True) - _transitions: ClassVar[Mapping[Enum, set[Enum]]] - - def transition_to(self, target_state: Enum) -> tuple["_StateModel", TransitionResult]: - state = self.state - target = type(state)(target_state) - if target == state: - return self, TransitionResult(previous_state=state.value, state=state.value, changed=False) - allowed = self._transitions.get(state, set()) - if target not in allowed: - raise InvalidStateTransition(state, target, allowed) - changed = self.model_copy(update={"state": target}) - return changed, TransitionResult(previous_state=state.value, state=target.value, changed=True) - +LifecycleState = ArtifactState | ReplicaState | DataPlaneTaskState | QueueWorkState ARTIFACT_TRANSITIONS = { ArtifactState.PENDING: {ArtifactState.BUILDING, ArtifactState.FAILED, ArtifactState.DELETING}, - ArtifactState.BUILDING: {ArtifactState.READY, ArtifactState.FAILED, ArtifactState.DELETING}, - ArtifactState.READY: {ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.BUILDING: {ArtifactState.STAGING, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.STAGING: {ArtifactState.READY, ArtifactState.FAILED, ArtifactState.DELETING}, + ArtifactState.READY: {ArtifactState.BUILDING, ArtifactState.FAILED, ArtifactState.DELETING}, ArtifactState.FAILED: {ArtifactState.BUILDING, ArtifactState.DELETING}, ArtifactState.DELETING: {ArtifactState.DELETED, ArtifactState.FAILED}, ArtifactState.DELETED: set(), } - REPLICA_TRANSITIONS = { - ReplicaState.PENDING: {ReplicaState.COPYING, ReplicaState.FAILED, ReplicaState.DELETING}, - ReplicaState.COPYING: {ReplicaState.READY, ReplicaState.FAILED, ReplicaState.DELETING}, - ReplicaState.READY: {ReplicaState.FAILED, ReplicaState.DELETING}, - ReplicaState.FAILED: {ReplicaState.COPYING, ReplicaState.DELETING}, - ReplicaState.DELETING: {ReplicaState.DELETED, ReplicaState.FAILED}, + ReplicaState.PENDING: {ReplicaState.STAGING, ReplicaState.FAILED, ReplicaState.EVICTING}, + ReplicaState.STAGING: {ReplicaState.READY, ReplicaState.FAILED, ReplicaState.EVICTING}, + ReplicaState.READY: {ReplicaState.STAGING, ReplicaState.FAILED, ReplicaState.EVICTING}, + ReplicaState.FAILED: {ReplicaState.STAGING, ReplicaState.EVICTING}, + ReplicaState.EVICTING: {ReplicaState.DELETED, ReplicaState.FAILED}, ReplicaState.DELETED: set(), } - TASK_TRANSITIONS = { - DataPlaneTaskState.PENDING: {DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, - DataPlaneTaskState.RUNNING: {DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.FAILED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, - DataPlaneTaskState.FAILED: {DataPlaneTaskState.PENDING, DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.PENDING: {DataPlaneTaskState.QUEUED, DataPlaneTaskState.CANCELLED}, + DataPlaneTaskState.QUEUED: {DataPlaneTaskState.LEASED, DataPlaneTaskState.RUNNING, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.LEASED: {DataPlaneTaskState.RUNNING, DataPlaneTaskState.QUEUED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED}, + DataPlaneTaskState.RUNNING: {DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.FAILED, DataPlaneTaskState.CANCELLED}, + DataPlaneTaskState.FAILED: {DataPlaneTaskState.QUEUED, DataPlaneTaskState.CANCELLED}, DataPlaneTaskState.SUCCEEDED: set(), DataPlaneTaskState.CANCELLED: set(), DataPlaneTaskState.EXPIRED: set(), } - WORK_TRANSITIONS = { - QueueWorkState.PENDING: {QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, - QueueWorkState.RUNNING: {QueueWorkState.SUCCEEDED, QueueWorkState.FAILED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, - QueueWorkState.FAILED: {QueueWorkState.PENDING, QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.PENDING: {QueueWorkState.BLOCKED, QueueWorkState.READY, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.BLOCKED: {QueueWorkState.READY, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.READY: {QueueWorkState.QUEUED, QueueWorkState.RUNNING, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, + QueueWorkState.QUEUED: {QueueWorkState.RUNNING, QueueWorkState.CANCELLED}, + QueueWorkState.RUNNING: {QueueWorkState.SUCCEEDED, QueueWorkState.FAILED, QueueWorkState.CANCELLED}, + QueueWorkState.FAILED: {QueueWorkState.READY, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED}, QueueWorkState.SUCCEEDED: set(), QueueWorkState.CANCELLED: set(), QueueWorkState.SKIPPED: set(), } +TRANSITIONS: Mapping[Type[Enum], Mapping[Enum, Set[Enum]]] = { + ArtifactState: ARTIFACT_TRANSITIONS, ReplicaState: REPLICA_TRANSITIONS, + DataPlaneTaskState: TASK_TRANSITIONS, QueueWorkState: WORK_TRANSITIONS, +} +TERMINAL_STATES = { + ArtifactState.DELETED, ReplicaState.DELETED, + DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED, + QueueWorkState.SUCCEEDED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED, +} +RETRYABLE_STATES = { + ArtifactState.FAILED, ReplicaState.FAILED, DataPlaneTaskState.FAILED, QueueWorkState.FAILED, +} + + +def allowed_target_states(state: LifecycleState) -> Tuple[LifecycleState, ...]: + table = TRANSITIONS.get(type(state)) + if table is None: + raise TypeError("state must be a lifecycle state enum") + return tuple(sorted(table[state], key=lambda item: item.value)) + + +def is_terminal_state(state: LifecycleState) -> bool: + return state in TERMINAL_STATES + + +def is_retryable_state(state: LifecycleState) -> bool: + return state in RETRYABLE_STATES + + +class InvalidStateTransitionDetail(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + code: str = "invalid_state_transition" + entity_type: str + entity_id: str + current_state: str + target_state: str + allowed_targets: Tuple[str, ...] + reason: str + terminal: bool + retryable: bool + + +class InvalidStateTransition(ValueError): + def __init__(self, detail: InvalidStateTransitionDetail): + self.detail = detail + super().__init__(detail.model_dump_json()) + + +def validate_state_transition( + current: LifecycleState, target: LifecycleState, *, entity_type: str, entity_id: str, +) -> bool: + allowed = allowed_target_states(current) + if type(target) is not type(current) or target not in allowed: + target_value = target.value if isinstance(target, Enum) else str(target) + raise InvalidStateTransition(InvalidStateTransitionDetail( + entity_type=entity_type, entity_id=entity_id, current_state=current.value, + target_state=target_value, allowed_targets=tuple(item.value for item in allowed), + reason="terminal_state" if is_terminal_state(current) else "transition_not_allowed", + terminal=is_terminal_state(current), retryable=is_retryable_state(current), + )) + return True + + +class TransitionResult(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + previous_state: str + state: str + changed: bool + + +class _StateModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + schema_version: str = SCHEMA_VERSION + _entity_type: ClassVar[str] + _id_field: ClassVar[str] + + @field_validator("schema_version") + @classmethod + def validate_schema_version(cls, value: str) -> str: + if value != SCHEMA_VERSION: + raise ValueError("unsupported schema_version") + return value + + def transition_to(self, target_state: LifecycleState): + current = self.state + if type(target_state) is type(current) and target_state == current: + return self, TransitionResult(previous_state=current.value, state=current.value, changed=False) + validate_state_transition(current, target_state, entity_type=self._entity_type, entity_id=getattr(self, self._id_field)) + return self.model_copy(update={"state": target_state}), TransitionResult( + previous_state=current.value, state=target_state.value, changed=True, + ) + + +def _non_empty(value: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError("identifier must not be empty") + return value class CacheArtifact(_StateModel): artifact_id: str - kid: str + knowledge_id: str + capability_fingerprint: Optional[str] = None + artifact_variant: str = "default" state: ArtifactState = ArtifactState.PENDING - _transitions = ARTIFACT_TRANSITIONS + _entity_type = "cache_artifact" + _id_field = "artifact_id" + _ids = field_validator("artifact_id", "knowledge_id", "artifact_variant")(_non_empty) class CacheReplica(_StateModel): replica_id: str artifact_id: str - location_key: str + data_plane_id: Optional[str] = None + backend_type: Optional[str] = None + location_key: Optional[str] = None state: ReplicaState = ReplicaState.PENDING - _transitions = REPLICA_TRANSITIONS + health: ReplicaHealth = ReplicaHealth.UNKNOWN + _entity_type = "cache_replica" + _id_field = "replica_id" + _ids = field_validator("replica_id", "artifact_id")(_non_empty) + + +def _task_id() -> str: + return f"dpt:{uuid.uuid4().hex}" + + +def _work_id() -> str: + return f"qwork:{uuid.uuid4().hex}" class DataPlaneTask(_StateModel): - task_id: str + task_id: str = Field(default_factory=_task_id) + operation: Optional[str] = None + artifact_id: Optional[str] = None + source_replica_id: Optional[str] = None + target_replica_id: Optional[str] = None state: DataPlaneTaskState = DataPlaneTaskState.PENDING - _transitions = TASK_TRANSITIONS + _entity_type = "data_plane_task" + _id_field = "task_id" + _id = field_validator("task_id")(_non_empty) class QueueWork(_StateModel): - work_id: str + work_id: str = Field(default_factory=_work_id) + work_type: Optional[str] = None + resource_class: Optional[str] = None state: QueueWorkState = QueueWorkState.PENDING - _transitions = WORK_TRANSITIONS + _entity_type = "queue_work" + _id_field = "work_id" + _id = field_validator("work_id")(_non_empty) + + +def _digest(prefix: str, identity: Dict[str, Optional[str]]) -> str: + canonical = json.dumps(identity, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return f"{prefix}:sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}" -def stable_id(kind: str, *identity: Any) -> str: - """Return a deterministic identifier from an explicitly ordered identity tuple.""" - material = "\x1f".join([kind, *(str(value) for value in identity)]) - return f"{kind}_{hashlib.sha256(material.encode()).hexdigest()[:24]}" +def artifact_id( + knowledge_id: str, capability_fingerprint: Optional[str] = None, + artifact_variant: str = "default", schema_version: str = SCHEMA_VERSION, +) -> str: + normalized = _non_empty(knowledge_id).strip() + return _digest("artifact", {"schema_version": schema_version, "knowledge_id": normalized, + "capability_fingerprint": capability_fingerprint, "artifact_variant": _non_empty(artifact_variant)}) -def artifact_id(kid: str) -> str: - return stable_id("artifact", kid) +def legacy_artifact_id(knowledge_id: str) -> str: + return artifact_id(knowledge_id, capability_fingerprint=None) -def replica_id(artifact: str, location_key: str) -> str: - return stable_id("replica", artifact, location_key) +def replica_id( + artifact_id_value: str, data_plane_id: Optional[str] = None, + backend_type: Optional[str] = None, location_key: Optional[str] = None, +) -> str: + return _digest("replica", {"artifact_id": _non_empty(artifact_id_value), + "data_plane_id": data_plane_id, "backend_type": backend_type, "location_key": location_key}) diff --git a/doc/state-models.md b/doc/state-models.md new file mode 100644 index 0000000..30a1405 --- /dev/null +++ b/doc/state-models.md @@ -0,0 +1,74 @@ +# Cache lifecycle and task state models + +Issue #139 defines immutable, schema-version `"1"` mechanism contracts for CacheArtifact, CacheReplica, DataPlaneTask, and QueueWork. Policy engines may consume these objects, but policy is outside this contract. + +## Wire values and purposes + +| Object | Purpose | Values | +|---|---|---| +| Artifact | capability-specific knowledge materialization | `pending`, `building`, `staging`, `ready`, `failed`, `deleting`, `deleted` | +| Replica | data-plane placement of an Artifact | `pending`, `staging`, `ready`, `failed`, `evicting`, `deleted` | +| Replica health | observation independent of lifecycle | `unknown`, `healthy`, `degraded`, `unhealthy` | +| DataPlaneTask | data creation, movement, or removal work | `pending`, `queued`, `leased`, `running`, `succeeded`, `failed`, `cancelled`, `expired` | +| QueueWork | scheduler-local work | `pending`, `blocked`, `ready`, `queued`, `running`, `succeeded`, `failed`, `cancelled`, `skipped` | + +## Complete transitions + +| Lifecycle | Source | Targets | +|---|---|---| +| Artifact | pending | building, failed, deleting | +| Artifact | building | staging, failed, deleting | +| Artifact | staging | ready, failed, deleting | +| Artifact | ready | building, failed, deleting | +| Artifact | failed | building, deleting | +| Artifact | deleting | deleted, failed | +| Artifact | deleted | none | +| Replica | pending | staging, failed, evicting | +| Replica | staging | ready, failed, evicting | +| Replica | ready | staging, failed, evicting | +| Replica | failed | staging, evicting | +| Replica | evicting | deleted, failed | +| Replica | deleted | none | +| DataPlaneTask | pending | queued, cancelled | +| DataPlaneTask | queued | leased, running, cancelled, expired | +| DataPlaneTask | leased | running, queued, cancelled, expired | +| DataPlaneTask | running | succeeded, failed, cancelled | +| DataPlaneTask | failed | queued, cancelled | +| DataPlaneTask | succeeded, cancelled, expired | none | +| QueueWork | pending | blocked, ready, cancelled, skipped | +| QueueWork | blocked | ready, cancelled, skipped | +| QueueWork | ready | queued, running, cancelled, skipped | +| QueueWork | queued | running, cancelled | +| QueueWork | running | succeeded, failed, cancelled | +| QueueWork | failed | ready, cancelled, skipped | +| QueueWork | succeeded, cancelled, skipped | none | + +Same-state transitions are idempotent. Terminal states are Artifact `deleted`, Replica `deleted`, DataPlaneTask `succeeded`/`cancelled`/`expired`, and QueueWork `succeeded`/`cancelled`/`skipped`. Every `failed` state is retryable and non-terminal. + +Invalid transitions expose typed detail: + +```json +{"code":"invalid_state_transition","entity_type":"cache_artifact","entity_id":"artifact:...","current_state":"deleted","target_state":"pending","allowed_targets":[],"reason":"terminal_state","terminal":true,"retryable":false} +``` + +## Stable IDs + +Artifact IDs have format `artifact:sha256:<64 lowercase hex>` and hash canonical schema version, normalized knowledge ID, nullable capability fingerprint, and Artifact variant. The Legacy helper uses a null capability fingerprint. Replica IDs have format `replica:sha256:<64 lowercase hex>` and hash Artifact ID, nullable data-plane ID, nullable backend type, and nullable opaque non-secret location key. Generated IDs are `dpt:<32 lowercase UUID4 hex>` and `qwork:<32 lowercase UUID4 hex>`; restored non-empty IDs are accepted. Credentials, passwords, secret URLs, and raw Redis keys are not identity inputs. + +## Legacy mapping + +Compatibility is always `unknown`: Legacy metadata cannot establish capability compatibility. When checked, only `/` is authoritative; `kv_rel_dir` is validated metadata. + +| kv_ready | Runtime directory | Artifact | Replica | Health | Warning | +|---|---|---|---|---|---| +| false | missing | pending | pending | unknown | none | +| false | exists | staging | staging | unknown | `legacy_files_without_ready_confirmation` | +| true | exists | ready | ready | healthy | none | +| true | missing | ready | failed | unhealthy | `legacy_replica_directory_missing` | +| true | not checked | ready | ready | unknown | none | + +Non-empty `kv_rel_dir` unequal to `kid` adds `legacy_kv_rel_dir_mismatch`. Mapping changes neither SQLite nor files. + +## Boundary and non-goals + +This is lifecycle mechanism, not scheduling policy. Issue #139 does not add issue #140 protocol APIs, a persistent Artifact catalog, ExecutionGraph scheduling, routing/injection changes, database tables, or final operation and resource-class enums. diff --git a/kdn_server/legacy_state.py b/kdn_server/legacy_state.py index 60ee10f..24fd1be 100644 --- a/kdn_server/legacy_state.py +++ b/kdn_server/legacy_state.py @@ -3,12 +3,13 @@ from __future__ import annotations from pathlib import Path -from typing import Any +from typing import Any, Literal, Optional, Tuple from pydantic import BaseModel, ConfigDict from core.state_models import ( - ArtifactState, CacheArtifact, CacheReplica, ReplicaState, artifact_id, replica_id, + ArtifactState, CacheArtifact, CacheReplica, ReplicaHealth, ReplicaState, + legacy_artifact_id, replica_id, ) @@ -20,11 +21,11 @@ class LegacyStateWarning(BaseModel): class LegacyKVStateView(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) - compatibility_status: str = "unknown" + source: Literal["legacy_kv_ready"] = "legacy_kv_ready" + compatibility_status: Literal["unknown"] = "unknown" artifact: CacheArtifact replica: CacheReplica - replica_directory_exists: bool - warnings: tuple[LegacyStateWarning, ...] = () + warnings: Tuple[LegacyStateWarning, ...] = () def _safe_component(value: Any, field: str, *, allow_dot: bool = False) -> str: @@ -32,40 +33,78 @@ def _safe_component(value: Any, field: str, *, allow_dot: bool = False) -> str: if component == "." and allow_dot: return component path = Path(component) - if not component or component == ".." or (component == "." and not allow_dot) or path.is_absolute() or len(path.parts) != 1: + if not component or component in {".", ".."} or path.is_absolute() or len(path.parts) != 1: raise ValueError(f"{field} must be a non-empty single path component") return component -def map_legacy_kv_state(row: Any, kv_root: str | Path) -> LegacyKVStateView: - """Map a Legacy database row without mutating either storage or the filesystem.""" +def _normalize_kv_ready(value: Any) -> bool: + if value is False or (type(value) is int and value == 0) or (type(value) is str and value == "0"): + return False + if value is True or (type(value) is int and value == 1) or (type(value) is str and value == "1"): + return True + raise ValueError("kv_ready must be one of False, 0, '0', True, 1, or '1'") + + +def map_legacy_kv_state(row: Any, kv_root: Optional[str | Path] = None) -> LegacyKVStateView: + """Map a Legacy row without changing SQLite or the filesystem. + + Passing ``kv_root=None`` explicitly means that no filesystem check was performed. + """ get = row.get if hasattr(row, "get") else lambda key, default=None: getattr(row, key, default) kid = _safe_component(get("kid"), "kid") + ready = _normalize_kv_ready(get("kv_ready", 0)) raw_rel_dir = get("kv_rel_dir") rel_dir = None if raw_rel_dir is not None and str(raw_rel_dir).strip(): rel_dir = _safe_component(raw_rel_dir, "kv_rel_dir", allow_dot=True) - root = Path(kv_root).resolve(strict=False) - runtime_directory = (root / kid).resolve(strict=False) - if runtime_directory.parent != root: - raise ValueError("kid escapes the configured KV root") - directory_exists = runtime_directory.is_dir() - kv_ready = bool(get("kv_ready", False)) - artifact_state = ArtifactState.READY if kv_ready else ArtifactState.PENDING - replica_state = ReplicaState.READY if kv_ready and directory_exists else ReplicaState.FAILED - artifact = CacheArtifact(artifact_id=artifact_id(kid), kid=kid, state=artifact_state) - replica = CacheReplica( - replica_id=replica_id(artifact.artifact_id, kid), artifact_id=artifact.artifact_id, - location_key=kid, state=replica_state, - ) - warnings = () + exists: Optional[bool] = None + if kv_root is not None: + root = Path(kv_root).resolve(strict=False) + runtime_directory = (root / kid).resolve(strict=False) + if runtime_directory.parent != root: + raise ValueError("kid escapes the configured KV root") + exists = runtime_directory.is_dir() + + warnings = [] if rel_dir is not None and rel_dir != kid: - warnings = (LegacyStateWarning( + warnings.append(LegacyStateWarning( code="legacy_kv_rel_dir_mismatch", message="Legacy kv_rel_dir differs from the runtime kid directory.", - ),) - return LegacyKVStateView( - artifact=artifact, replica=replica, replica_directory_exists=directory_exists, - warnings=warnings, + )) + + if ready: + artifact_state = ArtifactState.READY + if exists is False: + replica_state, health = ReplicaState.FAILED, ReplicaHealth.UNHEALTHY + warnings.append(LegacyStateWarning( + code="legacy_replica_directory_missing", + message="Legacy kv_ready is set but the runtime kid directory is missing.", + )) + else: + replica_state = ReplicaState.READY + health = ReplicaHealth.HEALTHY if exists is True else ReplicaHealth.UNKNOWN + elif exists is True: + artifact_state, replica_state = ArtifactState.STAGING, ReplicaState.STAGING + health = ReplicaHealth.UNKNOWN + warnings.append(LegacyStateWarning( + code="legacy_files_without_ready_confirmation", + message="Runtime kid directory exists without Legacy kv_ready confirmation.", + )) + else: + artifact_state, replica_state, health = ( + ArtifactState.PENDING, ReplicaState.PENDING, ReplicaHealth.UNKNOWN, + ) + + artifact_identifier = legacy_artifact_id(kid) + artifact = CacheArtifact( + artifact_id=artifact_identifier, knowledge_id=kid, + capability_fingerprint=None, state=artifact_state, + ) + replica = CacheReplica( + replica_id=replica_id(artifact_identifier, location_key=kid), + artifact_id=artifact_identifier, location_key=kid, + state=replica_state, health=health, ) + return LegacyKVStateView(artifact=artifact, replica=replica, warnings=tuple(warnings)) diff --git a/test/README.md b/test/README.md index efb0fc2..030f88d 100644 --- a/test/README.md +++ b/test/README.md @@ -2,6 +2,13 @@ `test/` contains local entrypoints, smoke scripts, and historical utility tests for CacheRoute development. The main `demo_*.py` entrypoints add the repository root to Python's module search path, so the documented commands can be launched directly from the `test` directory without manually setting `PYTHONPATH`. +Focused CPU-only lifecycle validation from the repository root: + +```bash +pytest -q test/test_state_models.py +pytest -q test/test_legacy_kv_state_mapping.py +``` + Install the complete application and development dependency set from the repository root before running the commands in this document: ```bash diff --git a/test/test_legacy_kv_state_mapping.py b/test/test_legacy_kv_state_mapping.py index 8a0486e..00ee961 100644 --- a/test/test_legacy_kv_state_mapping.py +++ b/test/test_legacy_kv_state_mapping.py @@ -1,66 +1,108 @@ +import importlib.util from pathlib import Path import pytest from pydantic import ValidationError -from core.state_models import ReplicaState +from core.state_models import ArtifactState, ReplicaHealth, ReplicaState from kdn_server.legacy_state import map_legacy_kv_state +_spec = importlib.util.spec_from_file_location("proxy_queue_knowledge", Path(__file__).parents[1] / "proxy/queue/knowledge.py") +_knowledge = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_knowledge) +classify_kdn_items = _knowledge.classify_kdn_items -def row(kid="kid", kv_rel_dir="kid"): - return {"kid": kid, "kv_ready": 1, "kv_rel_dir": kv_rel_dir} +def row(kid="kid", kv_ready=1, kv_rel_dir="kid"): + return {"kid": kid, "kv_ready": kv_ready, "kv_rel_dir": kv_rel_dir} -def warning_codes(view): +def codes(view): return {warning.code for warning in view.warnings} -def test_runtime_kid_directory_is_healthy_and_matching_metadata_has_no_warning(tmp_path): - (tmp_path / "kid").mkdir() - view = map_legacy_kv_state(row(), tmp_path) +@pytest.mark.parametrize("ready,create,artifact,replica,health,warning", [ + (0, False, ArtifactState.PENDING, ReplicaState.PENDING, ReplicaHealth.UNKNOWN, None), + (0, True, ArtifactState.STAGING, ReplicaState.STAGING, ReplicaHealth.UNKNOWN, "legacy_files_without_ready_confirmation"), + (1, True, ArtifactState.READY, ReplicaState.READY, ReplicaHealth.HEALTHY, None), + (1, False, ArtifactState.READY, ReplicaState.FAILED, ReplicaHealth.UNHEALTHY, "legacy_replica_directory_missing"), +]) +def test_complete_filesystem_mapping(tmp_path, ready, create, artifact, replica, health, warning): + if create: + (tmp_path / "kid").mkdir() + view = map_legacy_kv_state(row(kv_ready=ready), tmp_path) + assert (view.artifact.state, view.replica.state, view.replica.health) == (artifact, replica, health) + assert warning is None or warning in codes(view) + + +def test_ready_without_filesystem_check_has_unknown_health(): + view = map_legacy_kv_state(row(), None) + assert view.artifact.state == ArtifactState.READY assert view.replica.state == ReplicaState.READY - assert view.replica_directory_exists is True - assert view.replica.location_key == "kid" - assert "legacy_kv_rel_dir_mismatch" not in warning_codes(view) + assert view.replica.health == ReplicaHealth.UNKNOWN + + +@pytest.mark.parametrize("value,expected", [(False, False), (0, False), ("0", False), (True, True), (1, True), ("1", True)]) +def test_explicit_kv_ready_normalization(tmp_path, value, expected): + view = map_legacy_kv_state(row(kv_ready=value), tmp_path) + assert (view.artifact.state == ArtifactState.READY) is expected + + +@pytest.mark.parametrize("value", [None, "", "true", 2, -1, [], {}]) +def test_invalid_kv_ready_rejected(tmp_path, value): + with pytest.raises(ValueError, match="kv_ready"): + map_legacy_kv_state(row(kv_ready=value), tmp_path) -def test_stale_existing_metadata_directory_does_not_make_replica_healthy(tmp_path): +def test_runtime_kid_is_authoritative_and_mismatch_warns(tmp_path): (tmp_path / "stale").mkdir() view = map_legacy_kv_state(row(kv_rel_dir="stale"), tmp_path) assert view.replica.state == ReplicaState.FAILED - assert view.replica_directory_exists is False - assert "legacy_kv_rel_dir_mismatch" in warning_codes(view) + assert view.replica.location_key == "kid" + assert "legacy_kv_rel_dir_mismatch" in codes(view) + + +def test_matching_metadata_has_no_mismatch_warning(tmp_path): + (tmp_path / "kid").mkdir() + assert "legacy_kv_rel_dir_mismatch" not in codes(map_legacy_kv_state(row(), tmp_path)) -def test_dot_metadata_does_not_treat_root_as_replica(tmp_path): +def test_dot_cannot_make_root_healthy(tmp_path): view = map_legacy_kv_state(row(kv_rel_dir="."), tmp_path) assert view.replica.state == ReplicaState.FAILED - assert view.replica_directory_exists is False - assert "legacy_kv_rel_dir_mismatch" in warning_codes(view) + assert "legacy_kv_rel_dir_mismatch" in codes(view) -def test_escaping_metadata_is_rejected(tmp_path): - with pytest.raises(ValueError, match="kv_rel_dir"): - map_legacy_kv_state(row(kv_rel_dir="../outside"), tmp_path) +@pytest.mark.parametrize("field,value", [("kid", "../outside"), ("kv_rel_dir", "../outside")]) +def test_escaped_paths_rejected(tmp_path, field, value): + data = row() + data[field] = value + with pytest.raises(ValueError, match=field): + map_legacy_kv_state(data, tmp_path) -def test_projection_and_nested_models_are_frozen(tmp_path): - (tmp_path / "kid").mkdir() - view = map_legacy_kv_state(row(kv_rel_dir="other"), tmp_path) - with pytest.raises(ValidationError, match="Instance is frozen"): - view.compatibility_status = "compatible" +def test_ids_are_stable_and_projection_is_deeply_frozen(tmp_path): + first, second = map_legacy_kv_state(row(), tmp_path), map_legacy_kv_state(row(), tmp_path) + assert first.artifact.artifact_id == second.artifact.artifact_id + assert first.replica.replica_id == second.replica.replica_id + assert first.source == "legacy_kv_ready" and first.compatibility_status == "unknown" with pytest.raises(ValidationError, match="Instance is frozen"): - view.replica.state = ReplicaState.FAILED + first.compatibility_status = "compatible" with pytest.raises(ValidationError, match="Instance is frozen"): - view.warnings[0].code = "changed" + first.replica.state = ReplicaState.READY -def test_mapping_does_not_create_or_modify_files(tmp_path): +def test_mapping_does_not_write(tmp_path): marker = tmp_path / "marker" - marker.write_text("unchanged") - before = {path.relative_to(tmp_path): (path.stat().st_mtime_ns, path.read_bytes()) - for path in tmp_path.iterdir() if path.is_file()} + marker.write_bytes(b"unchanged") + before = [(path.name, path.stat().st_mtime_ns, path.read_bytes()) for path in tmp_path.iterdir()] map_legacy_kv_state(row(), tmp_path) - after = {path.relative_to(tmp_path): (path.stat().st_mtime_ns, path.read_bytes()) - for path in tmp_path.iterdir() if path.is_file()} + after = [(path.name, path.stat().st_mtime_ns, path.read_bytes()) for path in tmp_path.iterdir()] assert after == before + + +def test_existing_proxy_classification_is_unchanged(): + items = [{"kid": "ready", "kv_ready": 1}, {"kid": "text", "kv_ready": 0}] + result = classify_kdn_items(["ready", "text", "miss"], items, ["miss"]) + assert [item["kid"] for item in result["kv_ready_items"]] == ["ready"] + assert [item["kid"] for item in result["text_only_items"]] == ["text"] + assert result["miss_ids"] == ["miss"] diff --git a/test/test_state_models.py b/test/test_state_models.py index c6c5c80..c3f7e41 100644 --- a/test/test_state_models.py +++ b/test/test_state_models.py @@ -1,43 +1,129 @@ +import re + import pytest from pydantic import ValidationError from core.state_models import ( + ARTIFACT_TRANSITIONS, REPLICA_TRANSITIONS, TASK_TRANSITIONS, WORK_TRANSITIONS, ArtifactState, CacheArtifact, CacheReplica, DataPlaneTask, DataPlaneTaskState, - InvalidStateTransition, QueueWork, QueueWorkState, ReplicaState, + InvalidStateTransition, QueueWork, QueueWorkState, ReplicaHealth, ReplicaState, + allowed_target_states, artifact_id, is_retryable_state, is_terminal_state, replica_id, ) +ENUM_VALUES = { + ArtifactState: ["pending", "building", "staging", "ready", "failed", "deleting", "deleted"], + ReplicaState: ["pending", "staging", "ready", "failed", "evicting", "deleted"], + ReplicaHealth: ["unknown", "healthy", "degraded", "unhealthy"], + DataPlaneTaskState: ["pending", "queued", "leased", "running", "succeeded", "failed", "cancelled", "expired"], + QueueWorkState: ["pending", "blocked", "ready", "queued", "running", "succeeded", "failed", "cancelled", "skipped"], +} + + +def artifact(state=ArtifactState.PENDING): + return CacheArtifact(artifact_id="artifact:restored", knowledge_id="kid", state=state) + + +def replica(state=ReplicaState.PENDING): + return CacheReplica(replica_id="replica:restored", artifact_id="artifact:restored", state=state) + + +def test_exact_enum_wire_values(): + for enum_type, values in ENUM_VALUES.items(): + assert [item.value for item in enum_type] == values + + +@pytest.mark.parametrize("model", [artifact(), replica(), DataPlaneTask(), QueueWork()]) +def test_json_round_trip(model): + assert type(model).model_validate_json(model.model_dump_json()) == model + + +def test_unknown_enum_and_extra_field_rejected(): + with pytest.raises(ValidationError): + artifact().model_copy(update={"state": "unknown"}).model_validate( + {**artifact().model_dump(), "state": "unknown"}) + with pytest.raises(ValidationError, match="Extra inputs"): + CacheArtifact(artifact_id="a", knowledge_id="k", unexpected=True) + + @pytest.mark.parametrize("model,target", [ - (CacheArtifact(artifact_id="a", kid="kid"), ArtifactState.READY), - (CacheReplica(replica_id="r", artifact_id="a", location_key="kid"), ReplicaState.READY), - (DataPlaneTask(task_id="t"), DataPlaneTaskState.RUNNING), - (QueueWork(work_id="w"), QueueWorkState.RUNNING), + (artifact(), ArtifactState.READY), (replica(), ReplicaState.READY), + (DataPlaneTask(), DataPlaneTaskState.RUNNING), (QueueWork(), QueueWorkState.RUNNING), ]) -def test_lifecycle_state_is_frozen(model, target): +def test_lifecycle_models_are_frozen(model, target): with pytest.raises(ValidationError, match="Instance is frozen"): model.state = target -def test_legal_transition_returns_changed_copy_without_mutating_source(): - source = CacheArtifact(artifact_id="a", kid="kid") - building, result = source.transition_to(ArtifactState.BUILDING) - assert source.state == ArtifactState.PENDING - assert building.state == ArtifactState.BUILDING - assert result.changed is True +@pytest.mark.parametrize("factory,table", [ + (artifact, ARTIFACT_TRANSITIONS), (replica, REPLICA_TRANSITIONS), + (lambda state: DataPlaneTask(task_id="task", state=state), TASK_TRANSITIONS), + (lambda state: QueueWork(work_id="work", state=state), WORK_TRANSITIONS), +]) +def test_every_declared_transition_and_same_state(factory, table): + for source_state, targets in table.items(): + source = factory(source_state) + unchanged, result = source.transition_to(source_state) + assert unchanged is source and result.changed is False + for target in targets: + changed, result = source.transition_to(target) + assert source.state == source_state + assert changed.state == target and result.changed is True + + +def test_invalid_terminal_and_cross_enum_transition_details(): + with pytest.raises(InvalidStateTransition) as terminal: + artifact(ArtifactState.DELETED).transition_to(ArtifactState.PENDING) + assert terminal.value.detail.reason == "terminal_state" + assert terminal.value.detail.terminal is True + assert terminal.value.detail.allowed_targets == () + with pytest.raises(InvalidStateTransition) as cross_enum: + artifact().transition_to(ReplicaState.STAGING) + assert cross_enum.value.detail.reason == "transition_not_allowed" + assert cross_enum.value.detail.target_state == "staging" + + +def test_allowed_targets_are_sorted_and_traits_are_centralized(): + for enum_type, values in ENUM_VALUES.items(): + if enum_type is ReplicaHealth: + continue + for state in enum_type: + assert [item.value for item in allowed_target_states(state)] == sorted(item.value for item in allowed_target_states(state)) + assert is_retryable_state(state) is (state.value == "failed") + assert is_terminal_state(ArtifactState.DELETED) + assert is_terminal_state(ReplicaState.DELETED) + for state in (DataPlaneTaskState.SUCCEEDED, DataPlaneTaskState.CANCELLED, DataPlaneTaskState.EXPIRED, + QueueWorkState.SUCCEEDED, QueueWorkState.CANCELLED, QueueWorkState.SKIPPED): + assert is_terminal_state(state) + assert not is_terminal_state(ArtifactState.FAILED) + + +def test_stable_artifact_and_replica_ids_include_all_identity_inputs(): + first = artifact_id(" kid ", "cap", "variant") + assert first == artifact_id("kid", "cap", "variant") + assert re.fullmatch(r"artifact:sha256:[0-9a-f]{64}", first) + assert first != artifact_id("kid", "other", "variant") + base = replica_id(first, "plane", "disk", "location") + assert re.fullmatch(r"replica:sha256:[0-9a-f]{64}", base) + assert len({base, replica_id(first, "other", "disk", "location"), + replica_id(first, "plane", "other", "location"), + replica_id(first, "plane", "disk", "other"), + replica_id(artifact_id("other"), "plane", "disk", "location")}) == 5 -def test_same_state_transition_is_idempotent(): - source = DataPlaneTask(task_id="t") - unchanged, result = source.transition_to(DataPlaneTaskState.PENDING) - assert unchanged is source - assert result.changed is False +def test_generated_and_restored_ids_and_empty_rejection(): + tasks = {DataPlaneTask().task_id for _ in range(3)} + works = {QueueWork().work_id for _ in range(3)} + assert len(tasks) == 3 and all(re.fullmatch(r"dpt:[0-9a-f]{32}", item) for item in tasks) + assert len(works) == 3 and all(re.fullmatch(r"qwork:[0-9a-f]{32}", item) for item in works) + assert DataPlaneTask(task_id="restored").task_id == "restored" + assert QueueWork(work_id="restored").work_id == "restored" + for constructor, kwargs in [(DataPlaneTask, {"task_id": ""}), (QueueWork, {"work_id": ""}), + (CacheArtifact, {"artifact_id": "", "knowledge_id": "kid"}), + (CacheReplica, {"replica_id": "", "artifact_id": "a"})]: + with pytest.raises(ValidationError): + constructor(**kwargs) -def test_terminal_transition_has_structured_sorted_detail(): - source = QueueWork(work_id="w", state=QueueWorkState.SUCCEEDED) - with pytest.raises(InvalidStateTransition) as caught: - source.transition_to(QueueWorkState.RUNNING) - assert caught.value.detail == { - "code": "invalid_state_transition", "current_state": "succeeded", - "target_state": "running", "allowed_targets": [], - } +def test_model_defaults_are_independent(): + assert DataPlaneTask().model_dump() is not DataPlaneTask().model_dump() From 16f34882c421e9ba1029298b313ac88d87466dda Mon Sep 17 00:00:00 2001 From: RickZ <121943721+rickisba@users.noreply.github.com> Date: Sun, 26 Jul 2026 18:42:11 +0800 Subject: [PATCH 3/3] Harden lifecycle compatibility boundaries --- core/state_models.py | 25 +++++++++++++++-- doc/state-models.md | 2 +- kdn_server/legacy_state.py | 18 +++++++++--- test/test_legacy_kv_state_mapping.py | 20 +++++++++++++ test/test_state_models.py | 42 ++++++++++++++++++++++++++++ 5 files changed, 100 insertions(+), 7 deletions(-) diff --git a/core/state_models.py b/core/state_models.py index cef5224..fa9ad75 100644 --- a/core/state_models.py +++ b/core/state_models.py @@ -4,9 +4,10 @@ import hashlib import json +import re import uuid from enum import Enum -from typing import ClassVar, Dict, Mapping, Optional, Set, Tuple, Type, TypeVar +from typing import ClassVar, Dict, Mapping, Optional, Set, Tuple, Type from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -150,6 +151,8 @@ def validate_state_transition( current: LifecycleState, target: LifecycleState, *, entity_type: str, entity_id: str, ) -> bool: allowed = allowed_target_states(current) + if type(target) is type(current) and target == current: + return True if type(target) is not type(current) or target not in allowed: target_value = target.value if isinstance(target, Enum) else str(target) raise InvalidStateTransition(InvalidStateTransitionDetail( @@ -183,9 +186,9 @@ def validate_schema_version(cls, value: str) -> str: def transition_to(self, target_state: LifecycleState): current = self.state + validate_state_transition(current, target_state, entity_type=self._entity_type, entity_id=getattr(self, self._id_field)) if type(target_state) is type(current) and target_state == current: return self, TransitionResult(previous_state=current.value, state=current.value, changed=False) - validate_state_transition(current, target_state, entity_type=self._entity_type, entity_id=getattr(self, self._id_field)) return self.model_copy(update={"state": target_state}), TransitionResult( previous_state=current.value, state=target_state.value, changed=True, ) @@ -197,6 +200,22 @@ def _non_empty(value: str) -> str: return value +_USER_INFO_PATTERN = re.compile(r"[^/@\s:]+:[^/@\s]+@") +_SECRET_LOCATION_MARKERS = ("://", "password=", "credential=", "token=") + + +def validate_location_key(value: Optional[str]) -> Optional[str]: + """Reject connection strings and obvious secret-bearing Replica identities.""" + if value is None: + return None + if not isinstance(value, str) or not value.strip(): + raise ValueError("location_key must be a non-empty opaque value") + lowered = value.lower() + if any(marker in lowered for marker in _SECRET_LOCATION_MARKERS) or _USER_INFO_PATTERN.search(value): + raise ValueError("location_key must not contain credentials or connection strings") + return value + + class CacheArtifact(_StateModel): artifact_id: str knowledge_id: str @@ -219,6 +238,7 @@ class CacheReplica(_StateModel): _entity_type = "cache_replica" _id_field = "replica_id" _ids = field_validator("replica_id", "artifact_id")(_non_empty) + _location = field_validator("location_key")(validate_location_key) def _task_id() -> str: @@ -273,5 +293,6 @@ def replica_id( artifact_id_value: str, data_plane_id: Optional[str] = None, backend_type: Optional[str] = None, location_key: Optional[str] = None, ) -> str: + location_key = validate_location_key(location_key) return _digest("replica", {"artifact_id": _non_empty(artifact_id_value), "data_plane_id": data_plane_id, "backend_type": backend_type, "location_key": location_key}) diff --git a/doc/state-models.md b/doc/state-models.md index 30a1405..1fb4498 100644 --- a/doc/state-models.md +++ b/doc/state-models.md @@ -53,7 +53,7 @@ Invalid transitions expose typed detail: ## Stable IDs -Artifact IDs have format `artifact:sha256:<64 lowercase hex>` and hash canonical schema version, normalized knowledge ID, nullable capability fingerprint, and Artifact variant. The Legacy helper uses a null capability fingerprint. Replica IDs have format `replica:sha256:<64 lowercase hex>` and hash Artifact ID, nullable data-plane ID, nullable backend type, and nullable opaque non-secret location key. Generated IDs are `dpt:<32 lowercase UUID4 hex>` and `qwork:<32 lowercase UUID4 hex>`; restored non-empty IDs are accepted. Credentials, passwords, secret URLs, and raw Redis keys are not identity inputs. +Artifact IDs have format `artifact:sha256:<64 lowercase hex>` and hash canonical schema version, normalized knowledge ID, nullable capability fingerprint, and Artifact variant. The Legacy helper uses a null capability fingerprint. Replica IDs have format `replica:sha256:<64 lowercase hex>` and hash Artifact ID, nullable data-plane ID, nullable backend type, and nullable opaque non-secret location key. Opaque locations may use values such as `kid` or `node-a/cache-1`; connection strings, embedded user-info, and `password=`, `credential=`, or `token=` values are rejected. Generated IDs are `dpt:<32 lowercase UUID4 hex>` and `qwork:<32 lowercase UUID4 hex>`; restored non-empty IDs are accepted. Credentials, passwords, secret URLs, and raw Redis keys are not identity inputs. ## Legacy mapping diff --git a/kdn_server/legacy_state.py b/kdn_server/legacy_state.py index 24fd1be..b6f2ae9 100644 --- a/kdn_server/legacy_state.py +++ b/kdn_server/legacy_state.py @@ -13,6 +13,17 @@ ) +def _row_value(row: Any, key: str, default: Any = None) -> Any: + """Read dictionaries, sqlite3.Row objects, and attribute-backed records safely.""" + getter = getattr(row, "get", None) + if callable(getter): + return getter(key, default) + try: + return row[key] + except (KeyError, IndexError, TypeError): + return getattr(row, key, default) + + class LegacyStateWarning(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) code: str @@ -51,10 +62,9 @@ def map_legacy_kv_state(row: Any, kv_root: Optional[str | Path] = None) -> Legac Passing ``kv_root=None`` explicitly means that no filesystem check was performed. """ - get = row.get if hasattr(row, "get") else lambda key, default=None: getattr(row, key, default) - kid = _safe_component(get("kid"), "kid") - ready = _normalize_kv_ready(get("kv_ready", 0)) - raw_rel_dir = get("kv_rel_dir") + kid = _safe_component(_row_value(row, "kid"), "kid") + ready = _normalize_kv_ready(_row_value(row, "kv_ready", 0)) + raw_rel_dir = _row_value(row, "kv_rel_dir") rel_dir = None if raw_rel_dir is not None and str(raw_rel_dir).strip(): rel_dir = _safe_component(raw_rel_dir, "kv_rel_dir", allow_dot=True) diff --git a/test/test_legacy_kv_state_mapping.py b/test/test_legacy_kv_state_mapping.py index 00ee961..6943624 100644 --- a/test/test_legacy_kv_state_mapping.py +++ b/test/test_legacy_kv_state_mapping.py @@ -1,4 +1,5 @@ import importlib.util +import sqlite3 from pathlib import Path import pytest @@ -66,6 +67,25 @@ def test_matching_metadata_has_no_mismatch_warning(tmp_path): assert "legacy_kv_rel_dir_mismatch" not in codes(map_legacy_kv_state(row(), tmp_path)) +def test_sqlite_row_is_mapped_without_conversion_or_mutation(tmp_path): + (tmp_path / "kid").mkdir() + connection = sqlite3.connect(":memory:") + connection.row_factory = sqlite3.Row + connection.execute("CREATE TABLE legacy (kid TEXT, kv_ready INTEGER, kv_rel_dir TEXT)") + connection.execute("INSERT INTO legacy VALUES (?, ?, ?)", ("kid", 1, "kid")) + sqlite_row = connection.execute("SELECT kid, kv_ready, kv_rel_dir FROM legacy").fetchone() + view = map_legacy_kv_state(sqlite_row, tmp_path) + assert view.artifact.knowledge_id == "kid" + assert view.artifact.state == ArtifactState.READY + assert view.replica.state == ReplicaState.READY + assert view.replica.health == ReplicaHealth.HEALTHY + assert view.replica.location_key == "kid" + assert view.artifact.artifact_id and view.replica.replica_id + assert codes(view) == set() + assert connection.execute("SELECT COUNT(*) FROM legacy").fetchone()[0] == 1 + connection.close() + + def test_dot_cannot_make_root_healthy(tmp_path): view = map_legacy_kv_state(row(kv_rel_dir="."), tmp_path) assert view.replica.state == ReplicaState.FAILED diff --git a/test/test_state_models.py b/test/test_state_models.py index c3f7e41..fe39395 100644 --- a/test/test_state_models.py +++ b/test/test_state_models.py @@ -8,6 +8,7 @@ ArtifactState, CacheArtifact, CacheReplica, DataPlaneTask, DataPlaneTaskState, InvalidStateTransition, QueueWork, QueueWorkState, ReplicaHealth, ReplicaState, allowed_target_states, artifact_id, is_retryable_state, is_terminal_state, replica_id, + validate_state_transition, ) @@ -83,6 +84,21 @@ def test_invalid_terminal_and_cross_enum_transition_details(): assert cross_enum.value.detail.target_state == "staging" +def test_direct_validator_accepts_same_state_but_rejects_cross_enum_same_wire_value(): + assert validate_state_transition( + ArtifactState.PENDING, ArtifactState.PENDING, + entity_type="cache_artifact", entity_id="artifact:restored", + ) is True + with pytest.raises(InvalidStateTransition): + validate_state_transition( + ArtifactState.PENDING, ReplicaState.PENDING, + entity_type="cache_artifact", entity_id="artifact:restored", + ) + assert [item.value for item in allowed_target_states(ArtifactState.PENDING)] == [ + "building", "deleting", "failed", + ] + + def test_allowed_targets_are_sorted_and_traits_are_centralized(): for enum_type, values in ENUM_VALUES.items(): if enum_type is ReplicaHealth: @@ -111,6 +127,32 @@ def test_stable_artifact_and_replica_ids_include_all_identity_inputs(): replica_id(artifact_id("other"), "plane", "disk", "location")}) == 5 +@pytest.mark.parametrize("location", ["kid", "node-a/cache-1", "opaque-location"]) +def test_opaque_replica_locations_are_accepted_and_deterministic(location): + identifier = artifact_id("kid") + assert replica_id(identifier, location_key=location) == replica_id(identifier, location_key=location) + assert CacheReplica(replica_id="restored", artifact_id=identifier, location_key=location).location_key == location + + +@pytest.mark.parametrize("location", [ + "redis://host/key", "password=secret", "credential=value", "token=value", + "user:password@host", "", " ", +]) +def test_secret_bearing_replica_locations_are_rejected(location): + identifier = artifact_id("kid") + with pytest.raises(ValueError, match="location_key"): + replica_id(identifier, location_key=location) + with pytest.raises(ValidationError, match="location_key"): + CacheReplica(replica_id="restored", artifact_id=identifier, location_key=location) + safe_serialized = CacheReplica( + replica_id=replica_id(identifier, location_key="kid"), + artifact_id=identifier, location_key="kid", + ).model_dump_json() + assert all(secret not in safe_serialized for secret in ( + "redis://", "password=", "credential=", "token=", "user:password@host", + )) + + def test_generated_and_restored_ids_and_empty_rejection(): tasks = {DataPlaneTask().task_id for _ in range(3)} works = {QueueWork().work_id for _ in range(3)}