fix: resolve race condition in VMQ device deletion acknowledgment

Device deletion in Astarte involves a distributed coordination mechanism where
the VMQ plugin must acknowledge deletion by setting vmq_ack=true in the database
and sending a /f message to trigger final cleanup. The current implementation has
a critical race condition that can cause device deletions to stall forever.

The root cause is that the ack_device_deletion/2 function executes
operations in the wrong order:
1. Sends /f message to AMQP
2. Writes vmq_ack=true to database

This creates a race condition where:
- /f message is published successfully
- Database write fails (timeout, connection error, etc.)
- DUP processes /f message and sets dup_end_ack=true
- vmq_ack remains false, causing all_ack?() to return false
- Device deletion stalls permanently with no retry mechanism

Additionally, the RPC server always returns :ok regardless of actual operation
success, masking errors from callers.

The involves:
1. Reordering operations to ensure database write completes before
message publication
2. Using proper error handling with early return on database failures
3. Propagating errors correctly through RPC server to enable caller
retry logic

This ensures atomic behavior: either both operations succeed or both fail,
eliminating the race condition that caused permanent deletion stalls.

Breaking change: RPC callers will now receive error responses for failed
operations instead of false success indicators. However, error cases are
already handled in Astarte since v1.2:
https://github.com/astarte-platform/astarte/blob/35c877efeece31a66576982f9fa30c00b4b801ea/apps/astarte_data_updater_plant/lib/astarte_data_updater_plant/data_updater/impl.ex#L2340

Signed-off-by: Davide Briani <davide.briani@secomind.com>

Author: davidebriani

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 - Add the option to enable keepalive for scylladb connections, using the environment variable
   `DOCKER_VERNEMQ_ASTARTE_VMQ_PLUGIN__CASSANDRA__ENABLE_KEEPALIVE`. Defaults to `true`
+### Fixed
+- Corner case in device deletion acknowledgment that could lead to deletion stalling permanently.
 
 ## [1.2.1-rc.0] 2025-08-22
 ### Changed

lib/astarte_vmq_plugin.ex

@@ -182,12 +182,16 @@ defmodule Astarte.VMQ.Plugin do
     end
   end
 
+  @spec ack_device_deletion(String.t(), Astarte.Core.Device.encoded_device_id()) ::
+          :ok | {:error, term()}
   def ack_device_deletion(realm_name, encoded_device_id) do
-    timestamp = now_us_x10_timestamp()
-    publish_internal_message(realm_name, encoded_device_id, "/f", "", timestamp)
-    {:ok, decoded_device_id} = Device.decode_device_id(encoded_device_id)
-    {:ok, _} = Queries.ack_device_deletion(realm_name, decoded_device_id)
-    :ok
+    with {:ok, decoded_device_id} <- Device.decode_device_id(encoded_device_id),
+         {:ok, _} <- Queries.ack_device_deletion(realm_name, decoded_device_id) do
+      # Only send /f message after successful database write
+      timestamp = now_us_x10_timestamp()
+      publish_internal_message(realm_name, encoded_device_id, "/f", "", timestamp)
+      :ok
+    end
   end
 
   defp setup_heartbeat_timer(realm, device_id, session_pid) do

lib/astarte_vmq_plugin/rpc/server.ex

@@ -70,12 +70,23 @@ defmodule Astarte.VMQ.Plugin.RPC.Server do
   @impl GenServer
   def handle_call({:delete, %{realm_name: realm, device_id: device}}, _from, state) do
     client_id = "#{realm}/#{device}"
-    # Either the client has been deleted or it is :not_found,
-    # which means that there is no session anyway.
+
+    # Plugin.disconnect_client/2 return :ok or {:error, :not_found}, which
+    # means that there is no session anyway.
     Plugin.disconnect_client(client_id, true)
-    Plugin.ack_device_deletion(realm, device)
 
-    {:reply, :ok, state}
+    case Plugin.ack_device_deletion(realm, device) do
+      :ok ->
+        {:reply, :ok, state}
+
+      {:error, reason} ->
+        Logger.error(
+          "Failed to acknowledge device deletion for #{realm}/#{device}: #{inspect(reason)}",
+          tag: "vmq_ack_deletion_failed"
+        )
+
+        {:reply, {:error, reason}, state}
+    end
   end
 
   @impl GenServer