🐛 Clean up Metrics for acked messages

* This resolves a bug where the Posthog Telemetry Reporter detaches due
  to `key :bytes_processed not found`
* This also resolves a bug where messages stored in Postgres are counted
  twice towards message throughput / count

Author: RTLS

lib/sequin/consumers/consumers.ex

@@ -639,36 +639,12 @@ defmodule Sequin.Consumers do
         :record -> ConsumerRecord
       end
 
-    {count, msgs} =
+    {count, _} =
       consumer.id
       |> msg_module.where_consumer_id()
       |> msg_module.where_ack_ids(ack_ids)
-      |> select([ce], ce)
       |> Repo.delete_all()
 
-    :telemetry.execute(
-      [:sequin, :posthog, :event],
-      %{event: "consumer_ack"},
-      %{
-        distinct_id: "00000000-0000-0000-0000-000000000000",
-        properties: %{
-          consumer_id: consumer.id,
-          consumer_name: consumer.name,
-          message_count: count,
-          message_kind: consumer.message_kind,
-          "$groups": %{account: consumer.account_id}
-        }
-      }
-    )
-
-    Health.put_event(consumer, %Event{slug: :messages_delivered, status: :success})
-    Metrics.incr_consumer_messages_processed_count(consumer, count)
-    Metrics.incr_consumer_messages_processed_throughput(consumer, count)
-
-    TracerServer.messages_acked(consumer, ack_ids)
-
-    AcknowledgedMessages.store_messages(consumer.id, msgs)
-
     {:ok, count}
   end
 
@@ -804,9 +780,6 @@ defmodule Sequin.Consumers do
 
     AcknowledgedMessages.store_messages(consumer.id, acked_messages)
 
-    Metrics.incr_consumer_messages_processed_count(consumer, count)
-    Metrics.incr_consumer_messages_processed_throughput(consumer, count)
-
     bytes_processed =
       Enum.sum_by(
         acked_messages,
@@ -815,6 +788,8 @@ defmodule Sequin.Consumers do
         end
       )
 
+    Metrics.incr_consumer_messages_processed_count(consumer, count)
+    Metrics.incr_consumer_messages_processed_throughput(consumer, count)
     Metrics.incr_consumer_messages_processed_bytes(consumer, bytes_processed)
 
     :telemetry.execute(
@@ -833,6 +808,9 @@ defmodule Sequin.Consumers do
       }
     )
 
+    ack_ids = Enum.map(acked_messages, & &1.ack_id)
+    TracerServer.messages_acked(consumer, ack_ids)
+
     {:ok, count}
   end
 

lib/sequin/tracer/server.ex

@@ -52,7 +52,7 @@ defmodule Sequin.Tracer.Server do
 
   def messages_acked(_consumer, []), do: :ok
 
-  def messages_acked(consumer, ack_ids) do
+  def messages_acked(consumer, ack_ids) when is_list(ack_ids) do
     GenServer.cast(via_tuple(consumer.account_id), {:acked, consumer, ack_ids})
   end
 

test/sequin/consumers_test.exs

@@ -95,7 +95,11 @@ defmodule Sequin.ConsumersTest do
       record1 = ConsumersFactory.insert_consumer_record!(consumer_id: consumer.id, state: :delivered)
       record2 = ConsumersFactory.insert_consumer_record!(consumer_id: consumer.id, state: :delivered)
 
+      record1 = %ConsumerRecord{record1 | payload_size_bytes: Size.bytes(100)}
+      record2 = %ConsumerRecord{record2 | payload_size_bytes: Size.bytes(200)}
+
       assert {:ok, 2} = Consumers.ack_messages(consumer, [record1.ack_id, record2.ack_id])
+      assert {:ok, 2} = Consumers.after_messages_acked(consumer, [record1, record2])
 
       assert {:ok, messages} = AcknowledgedMessages.fetch_messages(consumer.id)
       assert length(messages) == 2