⚡️ Remove State.pop_blocked_messages

This behavior is O(messages) whenever a message fails. We don't need this anymore, as we automatically flush stale in-memory messages on an interval.

Author: acco

lib/sequin/runtime/slot_message_store.ex

@@ -667,11 +667,8 @@ defmodule Sequin.Runtime.SlotMessageStore do
 
       state = State.put_persisted_messages(state, messages)
 
-      with {newly_blocked_messages, state} <- State.pop_blocked_messages(state),
-           :ok <- handle_discarded_messages(state, discarded_messages),
-           # Now put the blocked messages into persisted_messages
-           state = State.put_persisted_messages(state, newly_blocked_messages),
-           :ok <- upsert_messages(state, messages ++ newly_blocked_messages) do
+      with :ok <- handle_discarded_messages(state, discarded_messages),
+           :ok <- upsert_messages(state, messages) do
         maybe_finish_table_reader_batch(prev_state, state)
         {:reply, :ok, state}
       else

lib/sequin/runtime/slot_message_store_state.ex

@@ -207,22 +207,6 @@ defmodule Sequin.Runtime.SlotMessageStore.State do
      }}
   end
 
-  @spec pop_blocked_messages(State.t()) :: {list(message()), State.t()}
-  def pop_blocked_messages(%State{} = state) do
-    execute_timed(:pop_blocked_messages, fn ->
-      blocked_cursor_tuples =
-        state.messages
-        |> Stream.filter(fn {cursor_tuple, msg} ->
-          is_message_group_persisted?(state, msg.group_id) and
-            not Multiset.value_member?(state.persisted_message_groups, msg.group_id, cursor_tuple)
-        end)
-        |> Stream.map(fn {cursor_tuple, _msg} -> cursor_tuple end)
-        |> Enum.to_list()
-
-      pop_messages(state, blocked_cursor_tuples)
-    end)
-  end
-
   @spec pop_all_messages(State.t()) :: {list(message()), State.t()}
   def pop_all_messages(%State{} = state) do
     pop_messages(state, Map.keys(state.messages))
@@ -539,16 +523,4 @@ defmodule Sequin.Runtime.SlotMessageStore.State do
     end)
     |> Enum.to_list()
   end
-
-  defp incr_counter(name, amount \\ 1) do
-    current = Process.get(name, 0)
-    Process.put(name, current + amount)
-  end
-
-  defp execute_timed(name, fun) do
-    {time, result} = :timer.tc(fun, :millisecond)
-    incr_counter(:"#{name}_total_ms", time)
-    incr_counter(:"#{name}_count")
-    result
-  end
 end

test/sequin/slot_message_store_state_test.exs

@@ -196,32 +196,6 @@ defmodule Sequin.Runtime.SlotMessageStoreStateTest do
     end
   end
 
-  describe "pop_blocked_messages/2" do
-    test "pops messages that should be moved to persisted_message_groups in persisted_message_groups", %{state: state} do
-      # Add two messages with different group_ids
-      msg1 = ConsumersFactory.consumer_message(group_id: "group1")
-      msg2 = ConsumersFactory.consumer_message(group_id: "group2")
-
-      {:ok, state} = State.put_messages(state, [msg1, msg2])
-
-      # Add a persisted message that shares group_id with msg1
-      persisted_msg = ConsumersFactory.consumer_message(group_id: "group1")
-      state = State.put_persisted_messages(state, [persisted_msg])
-
-      # Call pop_blocked_messages/2
-      {popped_messages, updated_state} = State.pop_blocked_messages(state)
-
-      # Verify we get back only msg1 which shares group_id with the persisted message
-      assert length(popped_messages) == 1
-      [popped_msg] = popped_messages
-      assert_cursor_tuple_matches(msg1, popped_msg)
-      assert popped_msg.group_id == "group1"
-
-      # Verify msg2 is still in state since it wasn't blocked
-      assert_message_in_state(msg2, updated_state)
-    end
-  end
-
   describe "end to end test with put and pop" do
     test "popped messages are removed from state", %{state: state} do
       now = DateTime.utc_now()

test/sequin/slot_message_store_test.exs

@@ -221,50 +221,6 @@ defmodule Sequin.SlotMessageStoreTest do
       assert Enum.all?(messages, fn msg -> msg.not_visible_until == not_visible_until end)
     end
 
-    test "when a message fails, non-persisted messages of the same group_id are persisted", %{consumer: consumer} do
-      # Create two messages with same group_id
-      group_id = "test-group"
-
-      message1 =
-        ConsumersFactory.consumer_message(
-          group_id: group_id,
-          message_kind: consumer.message_kind,
-          consumer_id: consumer.id
-        )
-
-      message2 =
-        ConsumersFactory.consumer_message(
-          group_id: group_id,
-          message_kind: consumer.message_kind,
-          consumer_id: consumer.id
-        )
-
-      # Put both messages in store
-      :ok = SlotMessageStore.put_messages(consumer, [message1, message2])
-
-      # Get and fail first message
-      {:ok, [delivered]} = SlotMessageStore.produce(consumer, 1, self())
-
-      meta = %{
-        ack_id: delivered.ack_id,
-        deliver_count: 1,
-        group_id: delivered.group_id,
-        last_delivered_at: DateTime.utc_now(),
-        not_visible_until: DateTime.add(DateTime.utc_now(), 60)
-      }
-
-      :ok = SlotMessageStore.messages_failed(consumer, [meta])
-
-      # Verify both messages are now persisted
-      messages = SlotMessageStore.peek_messages(consumer, 2)
-      assert length(messages) == 2
-      assert Enum.all?(messages, &(&1.group_id == group_id))
-
-      persisted_messages = Consumers.list_consumer_messages_for_consumer(consumer)
-      assert length(persisted_messages) == 2
-      assert Enum.all?(persisted_messages, &(&1.group_id == group_id))
-    end
-
     test "if the pid changes between calls of produce_messages, produced_messages are available for deliver", %{
       consumer: consumer
     } do
@@ -389,12 +345,6 @@ defmodule Sequin.SlotMessageStoreTest do
 
       # Put second message (nil group_id)
       :ok = SlotMessageStore.put_messages(consumer, [message2])
-
-      # Verify only messages with non-nil group_ids are blocked
-      Enum.each(SlotMessageStore.peek(consumer), fn state ->
-        {blocked_messages, _state} = State.pop_blocked_messages(state)
-        assert length(blocked_messages) == 0
-      end)
     end
 
     test "duplicate messages don't accumulate payload size", %{consumer: consumer} do