Track firmware delta generation in more detail (#2235)

Add status field to firmware deltas:

- Starting the creation of a firmware delta always creates an entry with
status "processing"
- Successfully creating the delta set status "completed"
- Failure sets status "failed"

Add firmware delta generation timeout worker

Fetching firmware deltas can now fall back to full because a delta is
not done.

Author: lawik

config/config.exs

@@ -63,6 +63,7 @@ config :nerves_hub, Oban,
     delete_firmware: 1,
     device: 1,
     firmware_delta_builder: 2,
+    firmware_delta_timeout: 1,
     truncate: 1,
     # temporary, will remove in November
     truncation: 1
@@ -74,6 +75,7 @@ config :nerves_hub, Oban,
      crontab: [
        {"0 * * * *", NervesHub.Workers.ScheduleOrgAuditLogTruncation},
        {"*/1 * * * *", NervesHub.Workers.CleanStaleDeviceConnections},
+       {"* * * * *", NervesHub.Workers.FirmwareDeltaTimeout},
        {"1,16,31,46 * * * *", NervesHub.Workers.DeleteOldDeviceConnections},
        {"*/5 * * * *", NervesHub.Workers.ExpireInflightUpdates},
        {"*/15 * * * *", NervesHub.Workers.DeviceHealthTruncation}

lib/nerves_hub/devices.ex

@@ -1734,7 +1734,7 @@ defmodule NervesHub.Devices do
            {:delta_updatable, delta_updatable?(device, target)},
          {:firmware, {:ok, source}} <-
            {:firmware, Firmwares.get_firmware_by_product_id_and_uuid(product_id, source_uuid)},
-         {:delta, {:ok, delta}} <-
+         {:delta, {:ok, %{status: :completed} = delta}} <-
            {:delta, Firmwares.get_firmware_delta_by_source_and_target(source, target)} do
       Logger.info(
         "Delivering firmware delta",
@@ -1766,6 +1766,17 @@ defmodule NervesHub.Devices do
 
         Firmwares.get_firmware_url(target)
 
+      {:delta, {:ok, %{status: status} = delta}} ->
+        Logger.info(
+          "Delivering full firmware as delta had status #{status}",
+          device_id: device.id,
+          source_firmware: source_uuid,
+          target_firmware: target.uuid,
+          delta: delta.id
+        )
+
+        Firmwares.get_firmware_url(target)
+
       {:delta, {:error, :not_found}} ->
         Logger.info(
           "Delivering full firmware as no delta can be found",

lib/nerves_hub/firmwares.ex

@@ -13,6 +13,7 @@ defmodule NervesHub.Firmwares do
   alias NervesHub.Products
   alias NervesHub.Products.Product
   alias NervesHub.Workers.DeleteFirmware
+  alias NervesHub.Workers.FirmwareDeltaBuilder
 
   alias NervesHub.Repo
 
@@ -391,10 +392,12 @@ defmodule NervesHub.Firmwares do
     FirmwareDelta
     |> where([fd], source_id: ^source_id)
     |> where([fd], target_id: ^target_id)
-    |> Repo.one()
+    |> order_by(desc: :inserted_at)
+    |> limit(1)
+    |> Repo.all()
     |> case do
-      nil -> {:error, :not_found}
-      firmware_delta -> {:ok, firmware_delta}
+      [] -> {:error, :not_found}
+      [firmware_delta] -> {:ok, firmware_delta}
     end
   end
 
@@ -405,11 +408,11 @@ defmodule NervesHub.Firmwares do
     firmware_upload_config().download_file(fw_or_delta)
   end
 
-  @spec create_firmware_delta(Firmware.t(), Firmware.t()) ::
+  @spec generate_firmware_delta(FirmwareDelta.t(), Firmware.t(), Firmware.t()) ::
           :ok
           | {:error, Changeset.t()}
 
-  def create_firmware_delta(source_firmware, target_firmware) do
+  def generate_firmware_delta(firmware_delta, source_firmware, target_firmware) do
     Logger.info(
       "Creating firmware delta between #{source_firmware.uuid} and #{target_firmware.uuid}."
     )
@@ -428,16 +431,15 @@ defmodule NervesHub.Firmwares do
               with upload_metadata <-
                      firmware_upload_config().metadata(org.id, firmware_delta_filename),
                    {:ok, firmware_delta} <-
-                     insert_firmware_delta(%{
-                       source_id: source_firmware.id,
-                       target_id: target_firmware.id,
-                       tool: created.tool,
-                       tool_metadata: created.tool_metadata,
-                       size: created.size,
-                       source_size: created.source_size,
-                       target_size: created.target_size,
-                       upload_metadata: upload_metadata
-                     }),
+                     complete_firmware_delta(
+                       firmware_delta,
+                       created.tool,
+                       created.size,
+                       created.source_size,
+                       created.target_size,
+                       created.tool_metadata,
+                       upload_metadata
+                     ),
                    {:ok, firmware_delta} <- get_firmware_delta(firmware_delta.id),
                    :ok <-
                      firmware_upload_config().upload_file(created.filepath, upload_metadata),
@@ -468,8 +470,12 @@ defmodule NervesHub.Firmwares do
           )
 
         case result do
-          {:ok, _delta} -> :ok
-          {:error, err} -> {:error, err}
+          {:ok, _delta} ->
+            :ok
+
+          {:error, err} ->
+            _ = fail_firmware_delta(firmware_delta)
+            {:error, err}
         end
 
       {:error, :no_delta_support_in_firmware} ->
@@ -496,9 +502,82 @@ defmodule NervesHub.Firmwares do
     |> preload([d, p], product: p)
   end
 
+  @spec attempt_firmware_delta(
+          source_id :: non_neg_integer(),
+          target_id :: non_neg_integer()
+        ) :: {:ok, FirmwareDelta.t()} | {:error, Ecto.Changeset.t()}
+  def attempt_firmware_delta(source_id, target_id) do
+    Repo.transaction(fn ->
+      with {:error, :not_found} <- get_firmware_delta_by_source_and_target(source_id, target_id),
+           {:ok, firmware_delta} <- start_firmware_delta(source_id, target_id) do
+        FirmwareDeltaBuilder.start(source_id, target_id)
+        firmware_delta
+      end
+    end)
+  end
+
+  @spec start_firmware_delta(
+          source :: Firmware.t() | non_neg_integer(),
+          target :: Firmware.t() | non_neg_integer()
+        ) :: {:ok, FirmwareDelta.t()} | {:error, Ecto.Changeset.t()}
+  def start_firmware_delta(%Firmware{id: source_id}, %Firmware{id: target_id}) do
+    start_firmware_delta(source_id, target_id)
+  end
+
+  def start_firmware_delta(source_id, target_id) do
+    FirmwareDelta.start_changeset(source_id, target_id)
+    |> Repo.insert()
+  end
+
+  @spec complete_firmware_delta(
+          firmware_delta :: FirmwareDelta.t(),
+          tool :: String.t(),
+          size :: non_neg_integer(),
+          source_size :: non_neg_integer(),
+          target_size :: non_neg_integer(),
+          tool_metadata :: map(),
+          upload_metadata :: map()
+        ) :: {:ok, FirmwareDelta.t()} | {:error, Ecto.Changeset.t()}
+  def complete_firmware_delta(
+        %FirmwareDelta{} = firmware_delta,
+        tool,
+        size,
+        source_size,
+        target_size,
+        tool_metadata,
+        upload_metadata
+      ) do
+    firmware_delta
+    |> FirmwareDelta.complete_changeset(
+      tool,
+      size,
+      source_size,
+      target_size,
+      tool_metadata,
+      upload_metadata
+    )
+    |> Repo.update()
+  end
+
+  @spec fail_firmware_delta(FirmwareDelta.t()) ::
+          {:ok, FirmwareDelta.t()} | {:error, Ecto.Changeset.t()}
+  def fail_firmware_delta(%FirmwareDelta{} = firmware_delta) do
+    firmware_delta
+    |> FirmwareDelta.fail_changeset()
+    |> Repo.update()
+  end
+
+  @spec time_out_firmware_delta(FirmwareDelta.t()) ::
+          {:ok, FirmwareDelta.t()} | {:error, Ecto.Changeset.t()}
+  def time_out_firmware_delta(%FirmwareDelta{} = firmware_delta) do
+    firmware_delta
+    |> FirmwareDelta.time_out_changeset()
+    |> Repo.update()
+  end
+
   def insert_firmware_delta(params) do
     %FirmwareDelta{}
-    |> FirmwareDelta.changeset(params)
+    |> FirmwareDelta.create_changeset(params)
     |> Repo.insert()
   end
 
@@ -521,6 +600,21 @@ defmodule NervesHub.Firmwares do
     end
   end
 
+  @spec time_out_firmware_delta_generations(
+          age :: non_neg_integer(),
+          unit :: :second | :millisecond | :minute
+        ) ::
+          :ok | {:error, any()}
+  def time_out_firmware_delta_generations(age_seconds, unit) do
+    cutoff = DateTime.add(DateTime.utc_now(), -age_seconds, unit)
+
+    from(fd in FirmwareDelta,
+      where: fd.status == :processing,
+      where: fd.inserted_at < ^cutoff
+    )
+    |> Repo.update_all(set: [status: :timed_out])
+  end
+
   @spec build_firmware_params(Org.t(), Path.t()) :: {:ok, map()} | {:error, any()}
   defp build_firmware_params(%{id: org_id} = org, filepath) do
     org = NervesHub.Repo.preload(org, :org_keys)

lib/nerves_hub/firmwares/firmware_delta.ex

@@ -8,22 +8,12 @@ defmodule NervesHub.Firmwares.FirmwareDelta do
   alias __MODULE__
 
   @type t :: %__MODULE__{}
-  @optional_params []
-  @required_params [
-    :source_id,
-    :target_id,
-    :upload_metadata,
-    :tool,
-    :tool_metadata,
-    :size,
-    :source_size,
-    :target_size
-  ]
 
   schema "firmware_deltas" do
     belongs_to(:source, Firmware)
     belongs_to(:target, Firmware)
 
+    field(:status, Ecto.Enum, values: [:processing, :completed, :failed, :timed_out])
     field(:tool, :string)
     # Metadata about the delta that the update tool needs to operate
     field(:tool_metadata, :map)
@@ -35,10 +25,124 @@ defmodule NervesHub.Firmwares.FirmwareDelta do
     timestamps()
   end
 
-  def changeset(%FirmwareDelta{} = firmware_delta, params) do
+  @spec start_changeset(source_id :: integer(), target_id :: integer()) :: Ecto.Changeset.t()
+  def start_changeset(source_id, target_id) do
+    params = %{
+      status: :processing,
+      source_id: source_id,
+      target_id: target_id,
+      tool: "pending",
+      tool_metadata: %{},
+      upload_metadata: %{}
+    }
+
+    %FirmwareDelta{}
+    |> cast(params, [:status, :source_id, :target_id, :tool, :tool_metadata, :upload_metadata])
+    |> validate_required([
+      :status,
+      :source_id,
+      :target_id,
+      :tool,
+      :tool_metadata,
+      :upload_metadata
+    ])
+    |> unique_constraint(:unique_firmware_delta, name: :source_id_target_id_unique_index)
+    |> foreign_key_constraint(:source_id, name: :firmware_deltas_source_id_fkey)
+    |> foreign_key_constraint(:target_id, name: :firmware_deltas_target_id_fkey)
+  end
+
+  @spec complete_changeset(
+          firmware_delta :: FirmwareDelta.t(),
+          tool :: String.t(),
+          size :: non_neg_integer(),
+          source_size :: non_neg_integer(),
+          target_size :: non_neg_integer(),
+          tool_metadata :: map(),
+          upload_metadata :: map()
+        ) :: Ecto.Changeset.t()
+  def complete_changeset(
+        %FirmwareDelta{} = firmware_delta,
+        tool,
+        size,
+        source_size,
+        target_size,
+        tool_metadata,
+        upload_metadata
+      ) do
+    firmware_delta
+    |> cast(
+      %{
+        status: :completed,
+        tool: tool,
+        size: size,
+        source_size: source_size,
+        target_size: target_size,
+        tool_metadata: tool_metadata,
+        upload_metadata: upload_metadata
+      },
+      [
+        :status,
+        :tool,
+        :size,
+        :source_size,
+        :target_size,
+        :tool_metadata,
+        :upload_metadata
+      ]
+    )
+    |> validate_required([
+      :status,
+      :tool,
+      :size,
+      :source_size,
+      :target_size,
+      :tool_metadata,
+      :upload_metadata
+    ])
+    |> unique_constraint(:unique_firmware_delta, name: :source_id_target_id_unique_index)
+    |> foreign_key_constraint(:source_id, name: :firmware_deltas_source_id_fkey)
+    |> foreign_key_constraint(:target_id, name: :firmware_deltas_target_id_fkey)
+  end
+
+  @spec fail_changeset(FirmwareDelta.t()) :: Ecto.Changeset.t()
+  def fail_changeset(%FirmwareDelta{} = firmware_delta) do
+    firmware_delta
+    |> cast(%{status: :failed}, [:status])
+    |> validate_required([:status])
+  end
+
+  @spec time_out_changeset(FirmwareDelta.t()) :: Ecto.Changeset.t()
+  def time_out_changeset(%FirmwareDelta{} = firmware_delta) do
+    firmware_delta
+    |> cast(%{status: :timed_out}, [:status])
+    |> validate_required([:status])
+  end
+
+  @spec create_changeset(FirmwareDelta.t(), map()) :: Ecto.Changeset.t()
+  def create_changeset(%FirmwareDelta{} = firmware_delta, params) do
     firmware_delta
-    |> cast(params, @required_params ++ @optional_params)
-    |> validate_required(@required_params)
+    |> cast(params, [
+      :source_id,
+      :target_id,
+      :status,
+      :upload_metadata,
+      :tool,
+      :tool_metadata,
+      :size,
+      :source_size,
+      :target_size
+    ])
+    |> validate_required([
+      :source_id,
+      :target_id,
+      :status,
+      :upload_metadata,
+      :tool,
+      :tool_metadata,
+      :size,
+      :source_size,
+      :target_size
+    ])
     |> unique_constraint(:unique_firmware_delta, name: :source_id_target_id_unique_index)
     |> foreign_key_constraint(:source_id, name: :firmware_deltas_source_id_fkey)
     |> foreign_key_constraint(:target_id, name: :firmware_deltas_target_id_fkey)

lib/nerves_hub/firmwares/update_tool/fwup.ex

@@ -301,9 +301,10 @@ defmodule NervesHub.Firmwares.UpdateTool.Fwup do
         |> Unzip.LocalFile.open()
         |> Unzip.new()
 
-      unzip
-      |> Unzip.file_stream!("meta.conf")
-      |> Enum.into(stream, &IO.iodata_to_binary/1)
+      _ =
+        unzip
+        |> Unzip.file_stream!("meta.conf")
+        |> Enum.into(stream, &IO.iodata_to_binary/1)
 
       {:ok, path}
     rescue