Test gaps at c46b500: four mutants of the claim error clause and the catch-up retry pass the whole suite #114

defmodule EventStore.Storage.MutantKillersLockTest do
  use EventStore.StorageCase

  alias EventStore.Storage.Lock

  @database "eventstore_test_xact_lock_denied"
  @role "eventstore_xact_lock_denied"

  setup %{conn: conn, postgrex_config: postgrex_config} do
    drop = fn ->
      Postgrex.query!(conn, ~s|DROP DATABASE IF EXISTS "#{@database}" WITH (FORCE)|, [])
      Postgrex.query!(conn, ~s|DROP ROLE IF EXISTS "#{@role}"|, [])
    end

    drop.()
    on_exit(fn -> drop.() end)

    Postgrex.query!(conn, ~s|CREATE ROLE "#{@role}" LOGIN PASSWORD 'secret'|, [])
    Postgrex.query!(conn, ~s|CREATE DATABASE "#{@database}" TEMPLATE template0|, [])

    {:ok, admin} = Postgrex.start_link(Keyword.put(postgrex_config, :database, @database))

    Postgrex.query!(
      admin,
      "CREATE TABLE public.subscriptions (subscription_id bigint PRIMARY KEY, " <>
        "lock_generation bigint NOT NULL DEFAULT 0, last_seen bigint)",
      []
    )

    Postgrex.query!(admin, "INSERT INTO public.subscriptions (subscription_id) VALUES (1)", [])
    Postgrex.query!(admin, ~s|GRANT SELECT, UPDATE ON public.subscriptions TO "#{@role}"|, [])

    Postgrex.query!(
      admin,
      "REVOKE EXECUTE ON FUNCTION pg_catalog.pg_try_advisory_xact_lock(integer, integer) FROM PUBLIC",
      []
    )

    GenServer.stop(admin)

    role_config =
      Keyword.merge(postgrex_config, database: @database, username: @role, password: "secret")

    [role_conn: start_supervised!({Postgrex, role_config}, id: :role_conn)]
  end

  test "a claim returns the error when only the advisory lock statement is refused", context do
    assert {:error, %Postgrex.Error{postgres: %{code: :insufficient_privilege} = postgres}} =
             Lock.claim_subscription_lock(context.role_conn, 1, schema: "public")

    assert postgres.message =~ "pg_try_advisory_xact_lock"
  end
end

defmodule EventStore.Subscriptions.MutantKillersFsmTest do
  use EventStore.StorageCase

  alias EventStore.{EventFactory, Storage, UUID}
  alias EventStore.Storage.Lock
  alias EventStore.Subscriptions.{SubscriptionFsm, SubscriptionState}

  setup context do
    stream_uuid = UUID.uuid4()
    subscription_name = UUID.uuid4()

    {:ok, %Storage.Subscription{subscription_id: subscription_id}} =
      Storage.subscribe_to_stream(context.conn, stream_uuid, subscription_name,
        schema: context.schema
      )

    {:ok, %Lock.Lease{token: fencing_token}} =
      Lock.claim_subscription_lock(context.conn, subscription_id, schema: context.schema)

    on_exit(fn -> :ok = Lock.unlock(context.conn, subscription_id, schema: context.schema) end)

    [
      stream_uuid: stream_uuid,
      subscription_name: subscription_name,
      subscription_id: subscription_id,
      fencing_token: fencing_token
    ]
  end

  test "a read that fails because the pool has no connection to give is retried", context do
    %{conn: conn} = context
    fsm = fsm(context, state: :request_catch_up, retry_interval: 20, query_timeout: 100)
    test_pid = self()

    holder =
      Task.async(fn ->
        Postgrex.transaction(conn, fn _held ->
          send(test_pid, :connection_held)
          receive do: (:release -> :ok)
        end)
      end)

    assert_receive :connection_held, 5_000

    assert %SubscriptionFsm{state: :catching_up} = SubscriptionFsm.catch_up(fsm)
    assert_receive :retry_catch_up, 1_000

    send(holder.pid, :release)
    assert {:ok, :ok} = Task.await(holder)
  end

  test "the wait before the next read is not shorter than the retry interval", context do
    fsm =
      fsm(context,
        state: :request_catch_up,
        schema: "no_such_schema",
        retry_interval: 5_000
      )

    assert %SubscriptionFsm{state: :catching_up} = SubscriptionFsm.catch_up(fsm)
    refute_receive :retry_catch_up, 300
  end

  for state <- [:subscribed, :max_capacity] do
    test "a stale retry leaves a #{state} subscription alone even with events to read", context do
      fsm = fsm(context, state: unquote(state))
      fsm = SubscriptionFsm.connect_subscriber(fsm, self(), buffer_size: 10)

      :ok = TestEventStore.append_to_stream(context.stream_uuid, 0, EventFactory.create_events(3))

      assert ^fsm = SubscriptionFsm.retry_catch_up(fsm)
      refute_received {:events, _events}
    end
  end

  defp fsm(context, overrides) do
    data = %SubscriptionState{
      conn: context.conn,
      schema: Keyword.get(overrides, :schema, context.schema),
      stream_uuid: context.stream_uuid,
      subscription_name: context.subscription_name,
      subscription_id: context.subscription_id,
      fencing_token: context.fencing_token,
      last_ack: 0,
      checkpoints_pending: 0,
      checkpoint_after: 0,
      checkpoint_threshold: 1,
      retry_interval: Keyword.get(overrides, :retry_interval, 60_000),
      query_timeout: Keyword.get(overrides, :query_timeout, 15_000),
      event_store: TestEventStore,
      serializer: context.serializer,
      correlation_id_type: context.correlation_id_type,
      causation_id_type: context.causation_id_type,
      max_size: 1_000,
      last_received: 0,
      last_sent: 0,
      transient: false
    }

    %SubscriptionFsm{state: Keyword.fetch!(overrides, :state), data: data}
  end
end