fix(pskr): handshake on passive socket, switch to active after SUBSCRIBE
`:gen_tcp.recv/3` requires a passive socket and returns `:einval` if
the socket is in any active mode. We were opening with
`active: :once`, sending CONNECT, then calling `:gen_tcp.recv` to
read CONNACK — which kernel-rejected every time.
The visible symptom was "Pskr.Client connect failed: :einval" every
30s on the elected leader, while a manual `:gen_tcp.connect` from
`bin/microwaveprop rpc` worked because it never tried to recv. Each
failed attempt also leaked a port (connect succeeded; recv failed
later), and we'd accumulated ~27 dangling sockets on the leader
process.
Open the socket in `active: false`, finish the synchronous
CONNECT/CONNACK/SUBSCRIBE handshake with `:gen_tcp.recv`, then
`:inet.setopts(active: :once)` so PUBLISHes arrive as
`{:tcp, socket, data}` messages going forward. Also close the
socket on any handshake failure so retries stop leaking ports.
This commit is contained in:
parent
1f021f42e6
commit
5521b0b153
1 changed files with 25 additions and 7 deletions
|
|
@ -178,22 +178,40 @@ defmodule Microwaveprop.Pskr.Client do
|
||||||
# ---------- Connection setup ----------
|
# ---------- Connection setup ----------
|
||||||
|
|
||||||
defp do_connect(state) do
|
defp do_connect(state) do
|
||||||
|
# Open in passive mode for the synchronous CONNECT/CONNACK
|
||||||
|
# handshake — `:gen_tcp.recv/3` only works on passive sockets
|
||||||
|
# and returns `:einval` if the socket is in any active mode.
|
||||||
|
# After CONNACK we flip to `active: :once` so PUBLISHes arrive
|
||||||
|
# as `{:tcp, socket, data}` messages.
|
||||||
|
#
|
||||||
# `:inet` forces IPv4 resolution + an IPv4 socket. Without
|
# `:inet` forces IPv4 resolution + an IPv4 socket. Without
|
||||||
# this, BEAM can pick the broker's AAAA record while our K8s
|
# this, BEAM can pick a broker AAAA record while our K8s pod
|
||||||
# pod has IPv4-only egress, and `:gen_tcp.connect/4` fails
|
# has IPv4-only egress.
|
||||||
# with `:einval` from the kernel `connect()` syscall.
|
tcp_opts = [:binary, :inet, active: false, packet: :raw, keepalive: true]
|
||||||
tcp_opts = [:binary, :inet, active: :once, packet: :raw, keepalive: true]
|
|
||||||
|
|
||||||
with {:ok, socket} <- :gen_tcp.connect(@broker_host, @broker_port, tcp_opts, 10_000),
|
case :gen_tcp.connect(@broker_host, @broker_port, tcp_opts, 10_000) do
|
||||||
:ok <- :gen_tcp.send(socket, Mqtt.connect(state.client_id, @keepalive_seconds)),
|
{:ok, socket} -> handshake(socket, state)
|
||||||
|
{:error, _} = err -> err
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp handshake(socket, state) do
|
||||||
|
with :ok <- :gen_tcp.send(socket, Mqtt.connect(state.client_id, @keepalive_seconds)),
|
||||||
{:ok, :connected, rest} <- await_connack(socket, <<>>),
|
{:ok, :connected, rest} <- await_connack(socket, <<>>),
|
||||||
:ok <- :gen_tcp.send(socket, subscribe_packet(state.bands)) do
|
:ok <- :gen_tcp.send(socket, subscribe_packet(state.bands)),
|
||||||
|
:ok <- :inet.setopts(socket, active: :once) do
|
||||||
Logger.info("Pskr.Client connected as #{state.client_id}, subscribed to #{length(state.bands)} bands")
|
Logger.info("Pskr.Client connected as #{state.client_id}, subscribed to #{length(state.bands)} bands")
|
||||||
# `rest` holds anything the broker sent piggybacked after
|
# `rest` holds anything the broker sent piggybacked after
|
||||||
# CONNACK — feed it back into the buffer so we don't lose
|
# CONNACK — feed it back into the buffer so we don't lose
|
||||||
# the SUBACK or an early PUBLISH.
|
# the SUBACK or an early PUBLISH.
|
||||||
send(self(), {:tcp, socket, rest})
|
send(self(), {:tcp, socket, rest})
|
||||||
{:ok, socket, schedule_ping()}
|
{:ok, socket, schedule_ping()}
|
||||||
|
else
|
||||||
|
{:error, _} = err ->
|
||||||
|
# Close the connected-but-not-handshaked socket so we
|
||||||
|
# don't leak ports across retries.
|
||||||
|
_ = :gen_tcp.close(socket)
|
||||||
|
err
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue