Persistent subscriptions and clustered eventstore.

[henriiik]

Hello!

I am trying to use spear create/connect to persistent subscriptions in a clustered eventstoredb setup.

If the Spear.Connection is connected to a non-leader node an error like this is returned, which contains the info needed to re-connect to the leader node:

%Spear.Connection.Response{
  data: "",
  headers: [
    {"content-type", "application/grpc"},
    {"date", "Thu, 02 Mar 2023 08:08:04 GMT"},
    {"server", "Kestrel"},
    {"content-length", "0"},
    {"exception", "not-leader"},
    {"leader-endpoint-host", "127.0.0.1"},
    {"leader-endpoint-port", "2111"},
    {"grpc-status", "5"},
    {"grpc-message", "Leader info available"}
  ],
  status: 200,
  type: {:event_store_db_gpb_protobufs_persistent,
   :"event_store.client.persistent_subscriptions.CreateResp"}
}

The call to Spear.create_persistent_subscription returns an error like this, which was very confusing to me at first:

%Spear.Grpc.Response{
  data: "",
  message: "Leader info available",
  status: :not_found,
  status_code: 5
}

I was able to reproduce the issue by switching the docker setup in the spear repo to use a clustered eventstoredb, and with some additional logging i get this:

❯ mix test test/spear_test.exs:310
Compiling 2 files (.ex)
Excluding tags: [:test]
Including tags: [line: "310"]

%Spear.Connection.Response{
  data: <<0, 0, 0, 0, 18, 10, 16, 8, 6, 26, 12, 8, 237, 169, 129, 128, 1, 16,
    237, 169, 129, 128, 1>>,
  headers: [
    {"content-type", "application/grpc"},
    {"date", "Thu, 02 Mar 2023 08:13:01 GMT"},
    {"server", "Kestrel"},
    {"grpc-status", "0"}
  ],
  status: 200,
  type: {:event_store_db_gpb_protobufs_streams,
   :"event_store.client.streams.AppendResp"}
}
%Spear.Connection.Response{
  data: "",
  headers: [
    {"content-type", "application/grpc"},
    {"date", "Thu, 02 Mar 2023 08:13:01 GMT"},
    {"server", "Kestrel"},
    {"content-length", "0"},
    {"exception", "not-leader"},
    {"leader-endpoint-host", "127.0.0.1"},
    {"leader-endpoint-port", "2111"},
    {"grpc-status", "5"},
    {"grpc-message", "Leader info available"}
  ],
  status: 200,
  type: {:event_store_db_gpb_protobufs_persistent,
   :"event_store.client.persistent_subscriptions.CreateResp"}
}
%Spear.Grpc.Response{
  data: "",
  message: "Leader info available",
  status: :not_found,
  status_code: 5
}


  1) test given a stream contains events info about a psub to :all can be fetched (SpearTest)
     test/spear_test.exs:310
     Assertion with == failed
     code:  assert Spear.create_persistent_subscription(c.conn, :all, group, settings) == :ok
     left:  {:error, %Spear.Grpc.Response{data: "", message: "Leader info available", status: :not_found, status_code: 5}}
     right: :ok
     stacktrace:
       test/spear_test.exs:314: (test)



Finished in 1.3 seconds (1.3s async, 0.00s sync)
64 tests, 1 failure, 63 excluded

Randomized with seed 439068

edit:

I accidentally submitted this issue before i was done writing it.

Is there recomended way to handle this? Look up leader before creating the subscription? or is this a bug?

[the-mikedavis]

I haven't tried this locally yet but my guess is that EventStore doesn't let you create persistent subscriptions from follower members of a cluster. You could inspect the list of cluster members with Spear.cluster_info/2 and find the %Spear.ClusterMember{} which is the leader (the :state key should have a value :Leader) and start a connection to that member for operations like creating persistent subscriptions and appending events. Spear focuses on single-server connections so there aren't conveniences built into Spear for clustering/gossip actions like switching from a follower to a leader currently.

[henriiik]

Thank you for your response!

I haven't tried this locally yet but my guess is that EventStore doesn't let you create persistent subscriptions from follower members of a cluster. You could inspect the list of cluster members with Spear.cluster_info/2 and find the %Spear.ClusterMember{} which is the leader (the :state key should have a value :Leader) and start a connection to that member for operations like creating persistent subscriptions and appending events.

That is correct and indeed what i ended up doing.

But since the leader information is available in the headers of the error response it would save some api calls if they was exposed, would you be willing to accept a PR that exposes them? :)

[the-mikedavis]

Yeah, avoiding a round trip for this sounds like a nice improvement! Go ahead and open up a PR 👍

[jake-fermented-io]

I ran into this same issue while using Commanded with the Spear adapter. I'm an elixir noob, so there could be a better way to do this, but I start a Spear.Connection, call cluster_info to determine the leader, then rewrite the Commanded.Application configuration using the correct host and port.

defp connect_to_leader(module) do
    app_conf = module.config()
    spear_conf = get_in(app_conf, [:event_store, :spear])
    {:ok, spear} = Spear.Connection.start_link(Keyword.put(spear_conf, :name, LeaderNodeLocator))
    {:ok, nodes} = Spear.cluster_info(spear)
    [%Spear.ClusterMember{address: address, port: port}] = Enum.filter(nodes, fn node -> node.state == :Leader end)
    leader_conn_str = Regex.replace(~r/@.*\/?\?/, spear_conf |> Keyword.get(:connection_string), "@#{address}:#{port}?")
    leader_app_conf =
        Keyword.update!(app_conf, :event_store, fn event_store_conf ->
          Keyword.update!(event_store_conf, :spear, fn spear_conf ->
            Keyword.put(spear_conf, :connection_string, leader_conn_str)
          end)
        end)
    GenServer.call(spear, :close)
    GenServer.stop(spear)
    {module, leader_app_conf}
  end