✨ AtLeastOnceVerification redis module

Author: RTLS

.gitignore

@@ -62,3 +62,6 @@ sequin.yaml
 .aider*
 .env
 .iex.local.exs
+
+# Cursor shenanigans
+*plan.md

lib/sequin/consumers_runtime/at_least_once_verification.ex

@@ -0,0 +1,109 @@
+defmodule Sequin.ConsumersRuntime.AtLeastOnceVerification do
+  @moduledoc """
+  Verifies at-least-once delivery guarantees by tracking commit tuples in Redis.
+
+  As messages are fanned out to SlotMessageStores, commit tuples are written to a Redis sorted set
+  with the commit timestamp as the score. When messages are acknowledged, the corresponding commit
+  tuples are removed.
+
+  Remaining commit tuples in the sorted set that are older than a certain threshold may indicate
+  missed deliveries, which can be verified against the ConsumerEvent/ConsumerRecord tables.
+  """
+
+  alias Sequin.Error
+  alias Sequin.Redis
+
+  require Logger
+
+  @type consumer_id :: String.t()
+  # {xid, lsn}
+  @type commit_tuple :: {integer(), integer()}
+  @type commit_timestamp :: integer()
+
+  @spec record_commit(consumer_id(), commit_tuple(), commit_timestamp()) :: :ok | {:error, Error.t()}
+  def record_commit(consumer_id, {xid, lsn}, timestamp) do
+    key = commit_key(consumer_id)
+    member = "#{xid}:#{lsn}"
+
+    case Redis.command(["ZADD", key, timestamp, member]) do
+      {:ok, _} -> :ok
+      {:error, error} -> {:error, error}
+    end
+  end
+
+  @spec remove_commit(consumer_id(), commit_tuple()) :: :ok | {:error, Error.t()}
+  def remove_commit(consumer_id, {xid, lsn}) do
+    key = commit_key(consumer_id)
+    member = "#{xid}:#{lsn}"
+
+    case Redis.command(["ZREM", key, member]) do
+      {:ok, _} -> :ok
+      {:error, error} -> {:error, error}
+    end
+  end
+
+  @spec get_unverified_commits(consumer_id(), commit_timestamp()) ::
+          {:ok, [{commit_tuple(), commit_timestamp()}]} | {:error, Error.t()}
+  def get_unverified_commits(consumer_id, older_than_timestamp) do
+    key = commit_key(consumer_id)
+
+    with {:ok, results} <- Redis.command(["ZRANGEBYSCORE", key, "-inf", older_than_timestamp, "WITHSCORES"]) do
+      commits =
+        results
+        |> Enum.chunk_every(2)
+        |> Enum.map(fn [member, score] ->
+          [xid, lsn] = String.split(member, ":")
+          {{String.to_integer(xid), String.to_integer(lsn)}, String.to_integer(score)}
+        end)
+
+      {:ok, commits}
+    end
+  end
+
+  @spec trim_commits(consumer_id(), commit_timestamp()) :: :ok | {:error, Error.t()}
+  def trim_commits(consumer_id, older_than_timestamp) do
+    key = commit_key(consumer_id)
+
+    with {:ok, initial_size} <- Redis.command(["ZCARD", key]),
+         {:ok, trimmed} <- Redis.command(["ZREMRANGEBYSCORE", key, "-inf", older_than_timestamp]),
+         {:ok, final_size} <- Redis.command(["ZCARD", key]) do
+      Logger.info("[AtLeastOnceVerification] Trimmed commits older than #{older_than_timestamp}",
+        consumer_id: consumer_id,
+        initial_size: initial_size,
+        records_removed: trimmed,
+        final_size: final_size
+      )
+
+      :ok
+    else
+      {:error, error} when is_exception(error) ->
+        Logger.error("[AtLeastOnceVerification] Error trimming commits", error: error)
+        {:error, error}
+    end
+  end
+
+  @spec all_commits(consumer_id()) :: {:ok, [{commit_tuple(), commit_timestamp()}]} | {:error, Error.t()}
+  def all_commits(consumer_id) do
+    key = commit_key(consumer_id)
+
+    with {:ok, results} <- Redis.command(["ZRANGE", key, 0, -1, "WITHSCORES"]) do
+      commits =
+        results
+        |> Enum.chunk_every(2)
+        |> Enum.map(fn [member, score] ->
+          [xid, lsn] = String.split(member, ":")
+          {{String.to_integer(xid), String.to_integer(lsn)}, String.to_integer(score)}
+        end)
+
+      {:ok, commits}
+    end
+  end
+
+  @spec count_commits(consumer_id()) :: {:ok, non_neg_integer()} | {:error, Error.t()}
+  def count_commits(consumer_id) do
+    key = commit_key(consumer_id)
+    Redis.command(["ZCARD", key])
+  end
+
+  defp commit_key(consumer_id), do: "consumer:#{consumer_id}:commit_verification"
+end

test/sequin/at_least_once_verification_test.exs

@@ -0,0 +1,131 @@
+defmodule Sequin.ConsumersRuntime.AtLeastOnceVerificationTest do
+  use ExUnit.Case, async: true
+
+  alias Sequin.ConsumersRuntime.AtLeastOnceVerification
+  alias Sequin.Factory
+  alias Sequin.Redis
+
+  setup do
+    consumer_id = Factory.uuid()
+    commit_key = "consumer:#{consumer_id}:commit_verification"
+
+    on_exit(fn ->
+      Redis.command(["DEL", commit_key])
+    end)
+
+    %{consumer_id: consumer_id}
+  end
+
+  describe "record_commit/3" do
+    test "records a commit tuple with timestamp", %{consumer_id: consumer_id} do
+      commit = {123, 456}
+      timestamp = System.system_time(:second)
+
+      assert :ok = AtLeastOnceVerification.record_commit(consumer_id, commit, timestamp)
+
+      # Verify the commit was recorded with correct score
+      {:ok, commits} = AtLeastOnceVerification.all_commits(consumer_id)
+      assert length(commits) == 1
+      assert List.first(commits) == {commit, timestamp}
+    end
+
+    test "can record multiple commits", %{consumer_id: consumer_id} do
+      commits = [
+        {{100, 200}, 1000},
+        {{101, 201}, 1001},
+        {{102, 202}, 1002}
+      ]
+
+      Enum.each(commits, fn {commit, ts} ->
+        assert :ok = AtLeastOnceVerification.record_commit(consumer_id, commit, ts)
+      end)
+
+      # Verify all commits were recorded
+      {:ok, count} = AtLeastOnceVerification.count_commits(consumer_id)
+      assert String.to_integer(count) == 3
+    end
+  end
+
+  describe "remove_commit/2" do
+    test "removes a specific commit tuple", %{consumer_id: consumer_id} do
+      commit = {123, 456}
+      timestamp = System.system_time(:second)
+
+      :ok = AtLeastOnceVerification.record_commit(consumer_id, commit, timestamp)
+      assert :ok = AtLeastOnceVerification.remove_commit(consumer_id, commit)
+
+      # Verify the commit was removed
+      {:ok, members} = Redis.command(["ZRANGE", "consumer:#{consumer_id}:commit_verification", 0, -1])
+      assert members == []
+    end
+
+    test "returns ok when removing non-existent commit", %{consumer_id: consumer_id} do
+      assert :ok = AtLeastOnceVerification.remove_commit(consumer_id, {999, 999})
+    end
+  end
+
+  describe "get_unverified_commits/2" do
+    test "returns commits older than specified timestamp", %{consumer_id: consumer_id} do
+      now = System.system_time(:second)
+
+      old_commits = [
+        {{100, 200}, now - 100},
+        {{101, 201}, now - 50}
+      ]
+
+      new_commit = {{102, 202}, now}
+
+      # Record all commits
+      Enum.each(old_commits, fn {commit, ts} ->
+        :ok = AtLeastOnceVerification.record_commit(consumer_id, commit, ts)
+      end)
+
+      :ok = AtLeastOnceVerification.record_commit(consumer_id, elem(new_commit, 0), elem(new_commit, 1))
+
+      # Get commits older than (now - 25)
+      {:ok, unverified} = AtLeastOnceVerification.get_unverified_commits(consumer_id, now - 25)
+
+      assert length(unverified) == 2
+      assert Enum.all?(unverified, fn {_commit, ts} -> ts < now - 25 end)
+    end
+
+    test "returns empty list when no commits exist", %{consumer_id: consumer_id} do
+      {:ok, unverified} = AtLeastOnceVerification.get_unverified_commits(consumer_id, System.system_time(:second))
+      assert unverified == []
+    end
+  end
+
+  describe "trim_commits/2" do
+    test "removes commits older than specified timestamp", %{consumer_id: consumer_id} do
+      now = System.system_time(:second)
+
+      commits = [
+        # old
+        {{100, 200}, now - 100},
+        # old
+        {{101, 201}, now - 50},
+        # current
+        {{102, 202}, now}
+      ]
+
+      # Record all commits
+      Enum.each(commits, fn {commit, ts} ->
+        :ok = AtLeastOnceVerification.record_commit(consumer_id, commit, ts)
+      end)
+
+      # Trim commits older than (now - 25)
+      assert :ok = AtLeastOnceVerification.trim_commits(consumer_id, now - 25)
+
+      # Verify only newer commit remains
+      {:ok, remaining} = AtLeastOnceVerification.count_commits(consumer_id)
+      assert String.to_integer(remaining) == 1
+
+      {:ok, [member]} = AtLeastOnceVerification.all_commits(consumer_id)
+      assert member == {{102, 202}, now}
+    end
+
+    test "handles empty set gracefully", %{consumer_id: consumer_id} do
+      assert :ok = AtLeastOnceVerification.trim_commits(consumer_id, System.system_time(:second))
+    end
+  end
+end