feat: add simulation clocks and VM transitions

app/simulation/transitions.py

@@ -0,0 +1,63 @@
+"""Explicit virtual-machine state machine."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+from app.simulation.clock import Clock
+
+
+class VmState(StrEnum):
+    STOPPED = "stopped"
+    STARTING = "starting"
+    RUNNING = "running"
+    PAUSING = "pausing"
+    PAUSED = "paused"
+    RESUMING = "resuming"
+    STOPPING = "stopping"
+    MIGRATING = "migrating"
+    SNAPSHOTTING = "snapshotting"
+    BACKING_UP = "backing_up"
+    ERROR = "error"
+
+
+class InvalidTransitionError(ValueError):
+    pass
+
+
+TRANSITIONS: dict[tuple[VmState, str], tuple[VmState, VmState]] = {
+    (VmState.STOPPED, "start"): (VmState.STARTING, VmState.RUNNING),
+    (VmState.RUNNING, "stop"): (VmState.STOPPING, VmState.STOPPED),
+    (VmState.RUNNING, "pause"): (VmState.PAUSING, VmState.PAUSED),
+    (VmState.PAUSED, "resume"): (VmState.RESUMING, VmState.RUNNING),
+    (VmState.RUNNING, "migrate"): (VmState.MIGRATING, VmState.RUNNING),
+    (VmState.STOPPED, "migrate"): (VmState.MIGRATING, VmState.STOPPED),
+    (VmState.RUNNING, "snapshot"): (VmState.SNAPSHOTTING, VmState.RUNNING),
+    (VmState.STOPPED, "snapshot"): (VmState.SNAPSHOTTING, VmState.STOPPED),
+    (VmState.RUNNING, "backup"): (VmState.BACKING_UP, VmState.RUNNING),
+    (VmState.STOPPED, "backup"): (VmState.BACKING_UP, VmState.STOPPED),
+}
+
+
+@dataclass(frozen=True, slots=True)
+class Transition:
+    operation: str
+    before: VmState
+    intermediate: VmState
+    after: VmState
+
+
+def plan_transition(state: VmState, operation: str) -> Transition:
+    states = TRANSITIONS.get((state, operation))
+    if states is None:
+        raise InvalidTransitionError(f"cannot {operation} VM while it is {state}")
+    return Transition(operation, state, states[0], states[1])
+
+
+async def execute_transition(
+    state: VmState, operation: str, clock: Clock, duration_seconds: float
+) -> tuple[VmState, VmState]:
+    transition = plan_transition(state, operation)
+    await clock.sleep(duration_seconds)
+    return transition.intermediate, transition.after