test(e2e): slow-fork fallback publishes then surfaces cluster verdict

Also pins the dead-handle contract: fork/fast_fork of an unknown VM is
:not_found and is never misrouted into the capacity fallback.

Author: markovejnovic

test/e2e/fork_test.exs

@@ -11,7 +11,13 @@ defmodule Hyper.E2e.ForkTest do
   - a VM created from the derived image composes that chain (dm-snapshot over
     the published COW store) and boots with the parent's pre-publish state —
     including the parent's writes, excluding the child's;
-  - all forked writable volumes are reclaimed on stop (no dm leak).
+  - all forked writable volumes are reclaimed on stop (no dm leak);
+  - `Hyper.Vm.fork/1`'s slow-fork fallback: when a real node-admission refusal
+    stops `fast_fork/1`, `fork/1` publishes exactly one derived image before
+    re-placing cluster-wide, and the cluster-wide capacity verdict surfaces
+    unmasked when no other node can take the child;
+  - `fast_fork/1` and `fork/1` on a dead VM handle are `{:error, :not_found}`,
+    never misclassified as a capacity refusal.
 
   Runs only under `--only integration` / `--include integration` on a host
   provisioned per docs/cookbook/install.md (CI: the `integration` job).
@@ -27,6 +33,17 @@ defmodule Hyper.E2e.ForkTest do
   # limits, which shared GHA egress IPs routinely exhaust.
   @image System.get_env("HYPER_E2E_IMAGE", "public.ecr.aws/docker/library/alpine:3.19")
 
+  alias Hyper.Img.Db.{Image, Repo}
+
+  # Boot :micro fillers until cluster placement refuses, so the next
+  # fast_fork must hit a node-admission refusal.
+  defp saturate(img_id) do
+    case Hyper.create_vm(%Hyper.Vm.Spec{img_id: img_id, type: :micro}) do
+      {:ok, vm} -> [vm | saturate(img_id)]
+      {:error, reason} when reason in [:no_capacity, :exhausted] -> []
+    end
+  end
+
   test "fast_fork isolates writes; published delta composes and boots" do
     assert {:ok, img_id} = Hyper.Img.OciLoader.load(@image)
 
@@ -108,4 +125,47 @@ defmodule Hyper.E2e.ForkTest do
 
     assert cousin_sees_childfile != 0, "child's writes leaked into the published delta"
   end
+
+  test "fork/1 falls back to publish-and-reschedule when the node refuses admission" do
+    assert {:ok, img_id} = Hyper.Img.OciLoader.load(@image)
+
+    assert {:ok, parent} = Hyper.create_vm(%Hyper.Vm.Spec{img_id: img_id, type: :micro})
+    on_exit(fn -> Hyper.Node.stop_image_vm(parent) end)
+
+    # Unique disk content so the published delta (and thus the derived image
+    # id) cannot collide with any image published by an earlier test.
+    assert {:ok, %{exit_code: 0}} =
+             await_exec(
+               parent,
+               ["/bin/sh", "-c", "head -c 32 /dev/urandom > /marker && sync"],
+               :timer.minutes(3)
+             )
+
+    fillers = saturate(img_id)
+    on_exit(fn -> Enum.each(fillers, &Hyper.Node.stop_image_vm/1) end)
+
+    images_before = Repo.aggregate(Image, :count)
+
+    # Single-node cluster: fast_fork must refuse with a capacity verdict,
+    # fork/1 must then publish the parent's disk state (the observable slow-
+    # fork side effect) and re-place cluster-wide — where the same saturated
+    # node is the only candidate, so the cluster verdict surfaces unmasked.
+    # fork/1's doc names both cluster-wide verdicts; either proves the point.
+    assert {:error, reason} = Hyper.Vm.fork(parent)
+
+    assert reason in [:no_capacity, :exhausted],
+           "expected the cluster-wide capacity verdict, got #{inspect(reason)}"
+
+    assert Repo.aggregate(Image, :count) == images_before + 1,
+           "slow fork must publish exactly one derived image before re-placing"
+  end
+
+  test "forking a dead VM handle is :not_found, with no fallback attempted" do
+    dead = spawn(fn -> :ok end)
+    ref = Process.monitor(dead)
+    assert_receive {:DOWN, ^ref, :process, ^dead, _}
+
+    assert Hyper.Vm.fast_fork(dead) == {:error, :not_found}
+    assert Hyper.Vm.fork(dead) == {:error, :not_found}
+  end
 end