feat: implement durable QEMU snapshots

app/tasks/qemu.py

@@ -23,6 +23,10 @@ def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler:
             return await _update(repository, task, resource_id)
         if operation == "delete":
             return await _delete(repository, task, resource_id)
+        if operation.startswith("snapshot-"):
+            return await _snapshot(
+                repository, task, resource_id, operation.removeprefix("snapshot-")
+            )
         async with repository.pool.acquire() as connection:
             row = await connection.fetchrow("SELECT state FROM resources WHERE id=$1", resource_id)
             if row is None:
@@ -129,5 +133,73 @@ async def _delete(repository: TaskRepository, task: Task, resource_id: uuid.UUID
     return {"deleted": True}
 
 
+async def _snapshot(
+    repository: TaskRepository,
+    task: Task,
+    resource_id: uuid.UUID,
+    operation: str,
+) -> dict[str, Any]:
+    name = str(task.payload["snapname"])
+    async with repository.pool.acquire() as connection:
+        async with connection.transaction():
+            if operation == "create":
+                row = await connection.fetchrow(
+                    """SELECT r.state, v.config FROM resources r
+                    JOIN virtual_machines v ON v.resource_id=r.id WHERE r.id=$1""",
+                    resource_id,
+                )
+                if row is None:
+                    raise ValueError("resource disappeared")
+                captured = {
+                    "resource_state": _object(row["state"]),
+                    "config": _object(row["config"]),
+                    "vmstate": bool(task.payload.get("vmstate", False)),
+                }
+                await connection.execute(
+                    """INSERT INTO snapshots(id, resource_id, name, description, state)
+                    VALUES($1, $2, $3, $4, $5::jsonb)""",
+                    uuid.uuid4(),
+                    resource_id,
+                    name,
+                    str(task.payload.get("description", "")),
+                    json.dumps(captured, sort_keys=True),
+                )
+            elif operation == "delete":
+                status = await connection.execute(
+                    "DELETE FROM snapshots WHERE resource_id=$1 AND name=$2",
+                    resource_id,
+                    name,
+                )
+                if status != "DELETE 1":
+                    raise ValueError("snapshot disappeared")
+            elif operation == "rollback":
+                row = await connection.fetchrow(
+                    "SELECT state FROM snapshots WHERE resource_id=$1 AND name=$2",
+                    resource_id,
+                    name,
+                )
+                if row is None:
+                    raise ValueError("snapshot disappeared")
+                captured = _object(row["state"])
+                state = dict(cast(Mapping[str, Any], captured["resource_state"]))
+                if bool(task.payload.get("start", False)):
+                    state["status"] = "running"
+                await connection.execute(
+                    """UPDATE resources SET state=$2::jsonb, version=version+1,
+                    updated_at=now() WHERE id=$1""",
+                    resource_id,
+                    json.dumps(state, sort_keys=True),
+                )
+                await connection.execute(
+                    "UPDATE virtual_machines SET config=$2::jsonb WHERE resource_id=$1",
+                    resource_id,
+                    json.dumps(captured["config"], sort_keys=True),
+                )
+            else:
+                raise ValueError(f"unsupported snapshot operation: {operation}")
+    await repository.append_log(task.id, f"snapshot {name} {operation} completed")
+    return {"snapshot": name, "operation": operation}
+
+
 def _object(value: object) -> dict[str, Any]:
     return json.loads(value) if isinstance(value, str) else dict(cast(Mapping[str, Any], value))