Improve batching logic

Author: martosaur

.sampo/changesets/haughty-duchess-tursas.md

@@ -0,0 +1,5 @@
+---
+hex/posthog: patch
+---
+
+Improve events batching logic and prevent SDK from sending empty batches

lib/posthog/sender.ex

@@ -8,6 +8,7 @@ defmodule PostHog.Sender do
     :api_client,
     :max_batch_time_ms,
     :max_batch_events,
+    :timer_ref,
     events: [],
     num_events: 0
   ]
@@ -74,24 +75,29 @@ defmodule PostHog.Sender do
   def handle_cast({:event, event}, state) do
     case state do
       %{num_events: n, events: events} when n + 1 >= state.max_batch_events ->
+        if state.timer_ref, do: Process.cancel_timer(state.timer_ref, async: true, info: false)
+
         {:noreply, %{state | events: [event | events], num_events: n + 1},
          {:continue, :send_batch}}
 
       %{num_events: 0, events: events} ->
-        Process.send_after(self(), :batch_time_reached, state.max_batch_time_ms)
+        ref = :erlang.start_timer(state.max_batch_time_ms, self(), :batch_time_reached)
 
-        {:noreply, %{state | events: [event | events], num_events: 1}}
+        {:noreply, %{state | events: [event | events], num_events: 1, timer_ref: ref}}
 
       %{num_events: n, events: events} ->
         {:noreply, %{state | events: [event | events], num_events: n + 1}}
     end
   end
 
   @impl GenServer
-  def handle_info(:batch_time_reached, state) do
+  def handle_info({:timeout, ref, :batch_time_reached}, %{num_events: n, timer_ref: ref} = state)
+      when n > 0 do
     {:noreply, state, {:continue, :send_batch}}
   end
 
+  def handle_info({:timeout, _ref, :batch_time_reached}, state), do: {:noreply, state}
+
   @impl GenServer
   def handle_continue(:send_batch, state) do
     # Before we initiate an HTTP request that might block the process
@@ -101,7 +107,7 @@ defmodule PostHog.Sender do
     Registry.update_value(state.registry, registry_key(state.index), fn _ -> :busy end)
     PostHog.API.batch(state.api_client, state.events)
     Registry.update_value(state.registry, registry_key(state.index), fn _ -> :available end)
-    {:noreply, %{state | events: [], num_events: 0}}
+    {:noreply, %{state | events: [], num_events: 0, timer_ref: nil}}
   end
 
   @impl GenServer

test/posthog/sender_test.exs

@@ -97,7 +97,7 @@ defmodule PostHog.SenderTest do
       assert %{events: ["my_event"]} = :sys.get_state(pid)
     end
 
-    test "immediately sends after reaching max_batch_events", %{
+    test "immediately sends after reaching max_batch_events and cancels timer", %{
       api_client: api_client,
       registry: registry
     } do
@@ -129,14 +129,18 @@ defmodule PostHog.SenderTest do
       end)
 
       Sender.send("foo", @supervisor_name)
+
+      %{timer_ref: ref} = :sys.get_state(pid)
+      assert is_reference(ref)
+
       Sender.send("bar", @supervisor_name)
 
       assert_receive :ready
 
       [{^pid, :busy}] = Registry.lookup(registry, {PostHog.Sender, 1})
       send(pid, :go)
 
-      assert %{events: []} = :sys.get_state(pid)
+      assert %{events: [], timer_ref: nil} = :sys.get_state(pid)
       [{^pid, :available}] = Registry.lookup(registry, {PostHog.Sender, 1})
     end
 
@@ -207,5 +211,49 @@ defmodule PostHog.SenderTest do
 
       assert :ok = GenServer.stop(pid)
     end
+
+    test "does not send empty batch", %{api_client: api_client, registry: registry} do
+      test_pid = self()
+
+      pid =
+        start_link_supervised!(
+          {Sender,
+           supervisor_name: @supervisor_name,
+           index: 1,
+           api_client: api_client,
+           max_batch_time_ms: 60_000,
+           max_batch_events: 2}
+        )
+
+      expect(API.Mock, :request, fn _client, method, url, opts ->
+        assert method == :post
+        assert url == "/batch"
+
+        assert opts[:json] == %{
+                 batch: ["bar", "foo"]
+               }
+
+        send(test_pid, :ready)
+
+        receive do
+          :go -> :ok
+        end
+
+        send(test_pid, :done)
+      end)
+
+      Sender.send("foo", @supervisor_name)
+      Sender.send("bar", @supervisor_name)
+
+      assert_receive :ready
+      [{^pid, :busy}] = Registry.lookup(registry, {PostHog.Sender, 1})
+      send(pid, :go)
+      assert_receive :done
+      %{timer_ref: ref} = :sys.get_state(pid)
+      [{^pid, :available}] = Registry.lookup(registry, {PostHog.Sender, 1})
+
+      send(pid, {:timeout, ref, :batch_time_reached})
+      refute_receive :ready
+    end
   end
 end