feat(grpc): hard-cut to v1 grpc interface

.gitignore

@@ -43,7 +43,7 @@ docs/superpowers/
 
 # gRPC bindings are generated from proto/**/*.proto before compile
 # -- see the :grpc_gen Mix compiler in mix.exs.
-/lib/hyper/grpc/v0/
+/lib/hyper/grpc/v1/
 /lib/hyper/agent/v1/
 
 # The expected suidhelper build identity (version + checksum) is captured from

AGENTS.md

@@ -49,7 +49,7 @@ run in CI's `integration` job, which provisions the GitHub-hosted runner via
 - `native/suidhelper/` — the privileged Rust helper. Source in `src/`, tests in
   `tests/` (see Rust rules below).
 - Generated, do not hand-edit: `lib/hyper/firecracker/api/{operations,schemas}`
-  (regen `mix firecracker.gen`) and `lib/hyper/grpc/v0/hyper.pb.ex`
+  (regen `mix firecracker.gen`) and `lib/hyper/grpc/v1/hyper.pb.ex`
   (regen `mix grpc.gen`). Both are gitignored and rebuilt by a Mix compiler.
 
 ## Testing philosophy — read this before writing any test

CHANGELOG.md

@@ -6,6 +6,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Changed
+- **Breaking:** the gRPC package was promoted `hyper.grpc.v0` -> `hyper.grpc.v1`.
+  `google.protobuf.Empty` is replaced by dedicated `StopVmResponse` and
+  `ListVmsRequest` messages; the `InstanceType` and `Architecture` enums gained
+  `*_UNSPECIFIED = 0` sentinels (existing values renumbered), and an unset
+  value is now rejected with `INVALID_ARGUMENT`; `ListVms` is now paginated
+  (`page_size` / `page_token` / `next_page_token`).
+
 ## [0.1.0]
 
 First public release: a distributed orchestrator for Firecracker microVMs on the

docs/grpc.md

@@ -5,17 +5,13 @@ The gRPC interface puts that same machine lifecycle behind a language-agnostic
 contract, so consumers in **any** language -- and off-BEAM services -- can
 create, stop, locate, and list microVMs.
 
-> **v0 -- unstable.** The contract may change without notice during early
-> development. Pin to a commit if you depend on it.
-
 > #### No authentication {: .warning}
 >
 > The gRPC API has **no authentication**. Any client that can reach the port can
 > create and stop VMs and load images. It is **off by default** (`grpc.enabled =
 > false`). When you enable it, bind it to loopback or a trusted network, or put
 > it behind a proxy that terminates TLS and authenticates callers. TLS (`cred`)
-> encrypts the channel but does not authenticate the client. Authentication is
-> planned for a later release; the `v0` contract is UNSTABLE.
+> encrypts the channel but does not authenticate the client.
 
 ## Configuration
 
@@ -61,8 +57,7 @@ language. We will be using _Python_ here:
 
 ```python
 import grpc
-from google.protobuf import empty_pb2
-from hyper.grpc.v0 import hyper_pb2, hyper_pb2_grpc
+from hyper.grpc.v1 import hyper_pb2, hyper_pb2_grpc
 
 # Plaintext. For TLS, pass grpc.ssl_channel_credentials(ca_pem) to
 # grpc.aio.secure_channel(...) instead.
@@ -104,13 +99,17 @@ print(created.vm_id, created.node)
 
 ### Listing VMs
 
-You can list running virtual machines with `ListVms`, which takes a
-`google.protobuf.Empty`:
+You can list running virtual machines with `ListVms`, which is paginated via
+`page_size`/`page_token` in and `next_page_token` out:
 
 ```python
-listed = await client.ListVms(empty_pb2.Empty())
+from hyper.grpc.v1 import hyper_pb2
+
+# List a page of VMs. Follow next_page_token until it is empty for the full set.
+listed = await client.ListVms(hyper_pb2.ListVmsRequest(page_size=100))
 for vm in listed.vms:
     print(vm.vm_id, vm.node)
+next_token = listed.next_page_token  # "" means this was the last page
 ```
 
 ### Getting VM Info
@@ -143,13 +142,13 @@ the `vm_usage` Postgres table (`vm_id`, `window_start`, `window_end`,
 
 ### Stopping a VM
 
-You can stop a running VM with `StopVm`, which returns a
-`google.protobuf.Empty`:
+You can stop a running VM with `StopVm`, which takes a request and returns an
+(empty) `StopVmResponse`:
 
 ```python
-await client.StopVm(hyper_pb2.StopVmRequest(vm_id=created.vm_id))
+await client.StopVm(hyper_pb2.StopVmRequest(vm_id=created.vm_id))  # -> StopVmResponse
 ```
 
 For full documentation, please read the documentation in the
-[`.proto`](https://github.com/harmont-dev/hyper/blob/main/proto/hyper/grpc/v0/hyper.proto)
+[`.proto`](https://github.com/harmont-dev/hyper/blob/main/proto/hyper/grpc/v1/hyper.proto)
 file.

lib/hyper/grpc.ex

@@ -2,11 +2,11 @@ defmodule Hyper.Grpc do
   @moduledoc """
   Public gRPC interface to a Hyper cluster.
 
-  The service contract is `hyper.grpc.v0.Hyper` (see
-  `proto/hyper/grpc/v0/hyper.proto`). Any gRPC client, in any language, can
+  The service contract is `hyper.grpc.v1.Hyper` (see
+  `proto/hyper/grpc/v1/hyper.proto`). Any gRPC client, in any language, can
   create, stop, locate, and list microVMs. Off-BEAM clients generate their own
   stubs from the `.proto`; BEAM clients can use the generated
-  `Hyper.Grpc.V0.Hyper.Stub` together with `connect/2`.
+  `Hyper.Grpc.V1.Hyper.Stub` together with `connect/2`.
   """
 
   alias Hyper.Cfg.Grpc, as: Config

lib/hyper/grpc/codec.ex

@@ -1,16 +1,14 @@
 defmodule Hyper.Grpc.Codec do
   @moduledoc """
-  Translation between the gRPC wire types (`Hyper.Grpc.V0.*`) and Hyper's domain
+  Translation between the gRPC wire types (`Hyper.Grpc.V1.*`) and Hyper's domain
   types. Two entry points, each dispatching by pattern match on the value's type:
 
     * `from_grpc/1` -- an inbound request message -> a domain value.
     * `to_grpc/1`   -- a domain result (or error) -> an outbound response message,
       or a `GRPC.RPCError` for the server to raise.
   """
 
-  alias Google.Protobuf.Empty
-
-  alias Hyper.Grpc.V0.{
+  alias Hyper.Grpc.V1.{
     CreateVmRequest,
     CreateVmResponse,
     ForkVmResponse,
@@ -19,6 +17,7 @@ defmodule Hyper.Grpc.Codec do
     ListVmsResponse,
     LoadImageRequest,
     LoadImageResponse,
+    StopVmResponse,
     Vm
   }
 
@@ -88,37 +87,52 @@ defmodule Hyper.Grpc.Codec do
   def to_grpc({:usage, vm_id, cpu_time}),
     do: %GetVmUsageResponse{vm_id: vm_id, cpu_usec: Unit.Time.as_us(cpu_time)}
 
-  @spec to_grpc({:vms, [{Hyper.Vm.Id.t(), node()}]}) :: ListVmsResponse.t()
-  def to_grpc({:vms, vms}),
-    do: %ListVmsResponse{vms: Enum.map(vms, &vm/1)}
+  @spec to_grpc({:vms, [{Hyper.Vm.Id.t(), node()}], String.t()}) :: ListVmsResponse.t()
+  def to_grpc({:vms, vms, next_page_token}),
+    do: %ListVmsResponse{vms: Enum.map(vms, &vm/1), next_page_token: next_page_token}
 
   @spec to_grpc({:loaded, Hyper.Img.id()}) :: LoadImageResponse.t()
   def to_grpc({:loaded, img_id}) when is_binary(img_id),
     do: %LoadImageResponse{img_id: img_id}
 
-  @spec to_grpc(:stopped) :: Empty.t()
-  def to_grpc(:stopped), do: %Empty{}
+  @spec to_grpc(:stopped) :: StopVmResponse.t()
+  def to_grpc(:stopped), do: %StopVmResponse{}
 
   @spec to_grpc({:error, term()}) :: GRPC.RPCError.t()
   def to_grpc({:error, reason}), do: rpc_error(reason)
 
   @spec vm({Hyper.Vm.Id.t(), node()}) :: Vm.t()
   defp vm({vm_id, node}), do: %Vm{vm_id: vm_id, node: to_string(node)}
 
-  @spec instance_type(term()) :: {:ok, Hyper.Vm.Instance.t()} | {:error, :bad_instance_type}
+  @spec instance_type(term()) ::
+          {:ok, Hyper.Vm.Instance.t()} | {:error, :missing_instance_type | :bad_instance_type}
+  defp instance_type(:INSTANCE_TYPE_UNSPECIFIED), do: {:error, :missing_instance_type}
+
   defp instance_type(enum) when is_map_key(@instance_types, enum),
     do: {:ok, @instance_types[enum]}
 
   defp instance_type(_unrecognised), do: {:error, :bad_instance_type}
 
-  @spec arch(term()) :: {:ok, Hyper.Vm.Instance.arch()} | {:error, :bad_arch}
+  @spec arch(term()) :: {:ok, Hyper.Vm.Instance.arch()} | {:error, :missing_arch | :bad_arch}
+  defp arch(:ARCHITECTURE_UNSPECIFIED), do: {:error, :missing_arch}
+
   defp arch(enum) when is_map_key(@arches, enum), do: {:ok, @arches[enum]}
+
   defp arch(_unrecognised), do: {:error, :bad_arch}
 
   @spec rpc_error(term()) :: GRPC.RPCError.t()
   defp rpc_error(:missing_img_id),
     do: GRPC.RPCError.exception(:invalid_argument, "img_id is required")
 
+  defp rpc_error(:missing_instance_type),
+    do: GRPC.RPCError.exception(:invalid_argument, "instance_type is required")
+
+  defp rpc_error(:missing_arch),
+    do: GRPC.RPCError.exception(:invalid_argument, "arch is required")
+
+  defp rpc_error(:bad_page_token),
+    do: GRPC.RPCError.exception(:invalid_argument, "page_token is malformed")
+
   defp rpc_error(:bad_instance_type),
     do: GRPC.RPCError.exception(:invalid_argument, "instance_type holds an unrecognised value")
 

lib/hyper/grpc/page.ex

@@ -0,0 +1,71 @@
+defmodule Hyper.Grpc.Page do
+  @moduledoc """
+  Generic keyset pagination over an in-memory collection. Pure: given the full
+  unordered list of resources, a requested page size, an opaque cursor token,
+  and a `Hyper.Grpc.Pageable` module describing how a resource is ordered,
+  returns one deterministic page plus the token for the next page.
+
+  Keyset (cursor-by-`Pageable.cursor/1`), not offset: elements appearing or
+  disappearing between calls cannot shift a page or skip/duplicate an entry,
+  because the cursor is the last key seen, not a position. Cursors are unique and
+  totally ordered as binaries, so the sort is stable across calls. An empty
+  `next_page_token` -- and only that -- signals end-of-collection (AIP-158).
+
+  Invariants are exercised in `test/hyper/grpc/page_properties_test.exs`.
+  """
+
+  @type resource :: term()
+
+  @doc """
+  One page of `entries` after the cursor in `page_token`, plus the next token.
+
+  `pageable` is a module implementing `Hyper.Grpc.Pageable`; its `cursor/1`
+  orders the collection and its `default_page_size/0` / `max_page_size/0` bound
+  the page size. `page_size <= 0` selects the default; larger than the max is
+  capped. A `page_token` that is not a token this module issued yields
+  `{:error, :bad_page_token}`.
+  """
+  @spec paginate([resource()], integer(), String.t(), module()) ::
+          {:ok, {[resource()], String.t()}} | {:error, :bad_page_token}
+  def paginate(entries, page_size, page_token, pageable) do
+    with {:ok, after_id} <- decode(page_token) do
+      remaining =
+        entries
+        |> Enum.sort_by(&pageable.cursor/1)
+        |> drop_through(after_id, pageable)
+
+      {page, rest} = Enum.split(remaining, clamp(page_size, pageable))
+      {:ok, {page, next_token(page, rest, pageable)}}
+    end
+  end
+
+  @spec decode(String.t()) :: {:ok, binary() | nil} | {:error, :bad_page_token}
+  defp decode(""), do: {:ok, nil}
+
+  defp decode(token) do
+    case Base.url_decode64(token, padding: false) do
+      {:ok, id} -> {:ok, id}
+      :error -> {:error, :bad_page_token}
+    end
+  end
+
+  @spec encode(binary()) :: String.t()
+  defp encode(id), do: Base.url_encode64(id, padding: false)
+
+  @spec clamp(integer(), module()) :: pos_integer()
+  defp clamp(size, pageable) when size <= 0, do: pageable.default_page_size()
+  defp clamp(size, pageable), do: min(size, pageable.max_page_size())
+
+  @spec drop_through([resource()], binary() | nil, module()) :: [resource()]
+  defp drop_through(sorted, nil, _pageable), do: sorted
+
+  defp drop_through(sorted, after_id, pageable),
+    do: Enum.drop_while(sorted, fn e -> pageable.cursor(e) <= after_id end)
+
+  @spec next_token([resource()], [resource()], module()) :: String.t()
+  defp next_token([], _rest, _pageable), do: ""
+  defp next_token(_page, [], _pageable), do: ""
+
+  defp next_token(page, _rest, pageable),
+    do: encode(pageable.cursor(List.last(page)))
+end

lib/hyper/grpc/pageable.ex

@@ -0,0 +1,36 @@
+defmodule Hyper.Grpc.Pageable do
+  @moduledoc """
+  Contract a resource must satisfy to be paginated by `Hyper.Grpc.Page`.
+
+  A conforming module declares how one element of a collection is ordered and
+  what page sizes are allowed. `Hyper.Grpc.Page` is generic over any such module,
+  so a new resource-listing RPC gets AIP-158 keyset pagination by implementing
+  this behaviour rather than copying the paginator.
+
+  ## Invariants a conforming module must uphold
+
+    * `cursor/1` returns a **unique** binary for every element in the collection.
+      Uniqueness is what makes keyset pagination stable: the page cursor is the
+      last key seen, not a position, so elements appearing or disappearing
+      between calls cannot shift a page or skip/duplicate an element.
+    * The cursor's **lexicographic (byte) order matches the resource's intended
+      total order**. `Hyper.Grpc.Page` sorts and compares cursors as binaries.
+      A key that is not already an order-preserving binary (e.g. an integer id)
+      must be encoded to one here -- zero-padded or fixed-width big-endian -- so
+      byte order matches the logical order. Returning, say, `Integer.to_string/1`
+      would sort `"10"` before `"2"` and page incorrectly.
+
+  The cursor is a binary (not an arbitrary term) so the page token stays a plain
+  `Base.url_encode64` of the cursor and decoding a token never runs the unsafe
+  `:erlang.binary_to_term`.
+  """
+
+  @doc "A unique, order-preserving binary key for one collection element."
+  @callback cursor(resource :: term()) :: binary()
+
+  @doc "Page size used when the request asks for a non-positive size."
+  @callback default_page_size() :: pos_integer()
+
+  @doc "Upper bound a requested page size is capped to."
+  @callback max_page_size() :: pos_integer()
+end

lib/hyper/grpc/pageable/vm.ex

@@ -0,0 +1,16 @@
+defmodule Hyper.Grpc.Pageable.Vm do
+  @moduledoc false
+  @behaviour Hyper.Grpc.Pageable
+
+  @impl true
+  @spec cursor({Hyper.Vm.Id.t(), node()}) :: binary()
+  def cursor({vm_id, _node}), do: vm_id
+
+  @impl true
+  @spec default_page_size() :: pos_integer()
+  def default_page_size, do: 100
+
+  @impl true
+  @spec max_page_size() :: pos_integer()
+  def max_page_size, do: 1000
+end

lib/hyper/grpc/server.ex

@@ -1,18 +1,17 @@
 defmodule Hyper.Grpc.Server do
   @moduledoc """
-  gRPC handler for `hyper.grpc.v0.Hyper`. A thin translation layer: each RPC
+  gRPC handler for `hyper.grpc.v1.Hyper`. A thin translation layer: each RPC
   maps its request to a domain value via `Hyper.Grpc.Codec.from_grpc/1`, calls
   the existing `Hyper` BEAM API, and maps the result back with
   `Hyper.Grpc.Codec.to_grpc/1` (raising the `GRPC.RPCError` it returns on error).
   """
 
-  use GRPC.Server, service: Hyper.Grpc.V0.Hyper.Service
+  use GRPC.Server, service: Hyper.Grpc.V1.Hyper.Service
   use OpenTelemetryDecorator
 
-  alias Google.Protobuf.Empty
   alias Hyper.Grpc.Codec
 
-  alias Hyper.Grpc.V0.{
+  alias Hyper.Grpc.V1.{
     CreateVmRequest,
     CreateVmResponse,
     ForkVmRequest,
@@ -21,10 +20,12 @@ defmodule Hyper.Grpc.Server do
     GetVmResponse,
     GetVmUsageRequest,
     GetVmUsageResponse,
+    ListVmsRequest,
     ListVmsResponse,
     LoadImageRequest,
     LoadImageResponse,
-    StopVmRequest
+    StopVmRequest,
+    StopVmResponse
   }
 
   @spec load_image(LoadImageRequest.t(), GRPC.Server.Stream.t()) :: LoadImageResponse.t()
@@ -66,7 +67,7 @@ defmodule Hyper.Grpc.Server do
     end
   end
 
-  @spec stop_vm(StopVmRequest.t(), GRPC.Server.Stream.t()) :: Empty.t()
+  @spec stop_vm(StopVmRequest.t(), GRPC.Server.Stream.t()) :: StopVmResponse.t()
   @decorate with_span("Hyper.Grpc.Server.stop_vm", include: [:vm_id])
   def stop_vm(%StopVmRequest{vm_id: vm_id}, _stream) do
     case Hyper.stop_vm(vm_id) do
@@ -93,9 +94,17 @@ defmodule Hyper.Grpc.Server do
     end
   end
 
-  @spec list_vms(Empty.t(), GRPC.Server.Stream.t()) :: ListVmsResponse.t()
+  @spec list_vms(ListVmsRequest.t(), GRPC.Server.Stream.t()) :: ListVmsResponse.t()
   @decorate with_span("Hyper.Grpc.Server.list_vms")
-  def list_vms(%Empty{}, _stream) do
-    Codec.to_grpc({:vms, Hyper.Cluster.Routing.all()})
+  def list_vms(%ListVmsRequest{page_size: page_size, page_token: page_token}, _stream) do
+    case Hyper.Grpc.Page.paginate(
+           Hyper.Cluster.Routing.all(),
+           page_size,
+           page_token,
+           Hyper.Grpc.Pageable.Vm
+         ) do
+      {:ok, {page, next}} -> Codec.to_grpc({:vms, page, next})
+      {:error, reason} -> raise Codec.to_grpc({:error, reason})
+    end
   end
 end