Persistent Plugin Questions

[HoneyryderChuck]

From email:

have a question regarding HTTPX. We use it in some adapter gems to communicate with other systems.

The way our adapter works is that it creates a long lived HTTPX connection (with the Persistent plugin), authenticates itsef with a request (some custom auth protocol), and keeps it around long as the connection works (peer responds to periodic requests) for other requests that the adapter needs to perform. If a request ever fails, the connection object is discarded and a new one is created.

Due to how the Persistent plugin works, this leads to too many open files if the host is not reachable or its name can't be resolved.

I'm not sure what I should do. I can't use HTTPX.wrap due to the async nature of the requests that need to be performed. Is there a way to use the Persistent plugin but still have HTTPX clean up old connections when the corresponding Ruby object goes out of scope?

Right now I don't have access to a test system where the server side for this adapter would be running. So for now I'm just brainstorming.

Should I ...

• pass in my own IO object so the connection stays open (without the Persistent plugin)?
• use a connection pool of size 1?
• just use the Persistent plugin and never discard the connection object, trusting HTTPX to reconnect whenever necessary? It would still have to perform that authentication request for any new connection though.

Let me know what you think. Thanks

[HoneyryderChuck]

If a request ever fails, the connection object is discarded and a new one is created.

Can you post here an example of what the code looks like? I think what mean by "connection" is what httpx calls the "session". If so, you shouldn't need to discard it, a :persistent session already takes care of reconnecting for you. This may be the reason you you're seeing the EMFILE issue.

there a way to use the Persistent plugin but still have HTTPX clean up old connections when the corresponding Ruby object goes out of scope?

If the :persistent plugin enabled session is going out of scope, it's probably not what you should use. Ideally, you initiate that session once, and put it somewhere (global registry, static variable) where you can access it always:

PERSISTENT = HTTPX.plugin(:persistent)

If the variable is going out of scope, then HTTPX.wrap is more appropriate. You did mention you can't use it though. can you post a code snippet explaining the issue?

just use the Persistent plugin and never discard the connection object, trusting HTTPX to reconnect whenever necessary? It would still have to perform that authentication request for any new connection though.

This is what I would recommend. That "must reauthenticate on reconnect" requirement may be an impediment though. You could perhaps hack it by using the Callbacks plugin and the on_connection_opened callback. However, a custom plugin may be your best choice (you could then do some sort of similar logic to what the digest auth plugin does).

[paddor]

Thanks for putting this here. For now I'm using the third option.

Good point on the Callbacks plugin. If necessary, I'll use it.

Here's a rough example. Super proprietary, over-engineered protocol, you know the drill:

require 'async'
require 'httpx'


class MyClient
  attr_reader :credentials


  def initialize(uri)
    @uri             = uri
    @connection      = nil
    @connection_task = nil
    @polling_task    = nil
  end


  def start
    @connection_task = open_connection
    @polling_task    = start_polling
  end


  def open_connection
    Async do
      loop do
        @connection = Connection.new(self)
        @connection.start
        @connection.wait
      rescue => e
        warn "Error ..."
      end
    end
  end


  def start_polling
    Async do
      loop do
        sleep 10
        get_status
      end
    end
  end


  def heartbeat!
    # register peer heartbeat
  end


  def get_status
    response = @connection.get_status

    # do something with response
  rescue
    @connection&.close
  end


  def set_foo
    response = @connection.set_foo

    # do something with response
  rescue
    @connection&.close
  end
end


class Connection
  class << self
    def http_sessions_by_uri
      @http_sessions_by_uri ||= {}
    end
  end


  def initialize(adapter)
    @adapter      = adapter
    @auth         = Auth.new
    @http         = new_http
    @disconnected = Async::Condition.new
  end


  def new_http
    if existing_session = self.class.http_sessions_by_uri[@adapter.uri]
      return existing_session
    end

    options = {
      origin: @adapter.uri,
      timeout: {
        connect_timeout: 10,
        request_timeout: 5,
      },
    }

    HTTPX.plugin(:persistent).with(**options)
  end


  def start
    connect
    login

    @adapter.heartbeat!
  rescue
    warn "Failed to connect and login ..."
  end


  def connect
    send_request Requests::Connect.new(self)
  end


  def disconnect
    send_request Requests::Disconnect.new(self)
  end


  def login
    request = Requests::Login.new(self).tap do |req|
      req.username = @auth.encrypt @adapter.credentials.fetch(:username)
      req.password = @auth.encrypt @adapter.credentials.fetch(:password)
    end

    send_request request
  end


  def get_status
    send_request Requests::GetStatus.new(self)
  end


  def set_foo
    send_request Requests::SetFoo.new(self)
  end


  def close
    disconnect
  rescue HTTPX::ConnectTimeoutError
    # noop
  ensure
    @disconnected.signal
  end


  def wait
    @disconnected.wait
  end

  private

  def send_request(request)
    @http.post request.path, body: request.body
  end
end


class Auth
  # ...
end


module Requests
  class Connect
    # ...
  end

  class Disconnect
    # ...
  end

  class Login
    # ...
  end

  class GetStatus
    # ...
  end

  class SetFoo
    # ...
  end
end


Async do |task|
  client = MyClient.new MY_URI
  client.start

  # simulate other async requests, maybe triggered by a user
  task.async do
    loop do
      sleep 10
      client.set_foo
    end
  end
end

[HoneyryderChuck]

Looking at your snippet, a few ideas pop in my head:

I find it interesting you've used it in this way, but the use case for it was more for the type of "single service SDK" static session. I'd recommend against caching all sessions by "session/origin" at the class level though, for the reasons below.

First thing, accessing / storing it does not seem to be thread safe (you didn't present code for storing, not sure you're pre-allocating "out-of-band"). Another reason, the connection pools aren't shared across sessions; while you're logically separating them per pool, and perhaps in the end that'd have the same outcome, if you'd use a single session for all requests, you could benefit from connection coalescing when/if applicable (and it'd fix your potential thread-safety issues mentioned in the previous point).

class Connection
  HTTP = HTTPX.plugin(:persistent).with(timeout: {
    connect_timeout: 10,
    request_timeout: 5,
  })

  def initialize(adapter)
    @adapter      = adapter
    @auth         = Auth.new
    @disconnected = Async::Condition.new
  end

  #...
  private

  def send_request(request)
    HTTP.post @adapter.merge(request.path), body: request.body
  end

That authentication protocol is interesting. If I understood it correctly, you want to, on each new established connection, "connect-and-authenticate" before it can be used for requests, is that it? That's certainly unusual compared to other authentication handshakes supported already by httpx (basic: "authenticate all requests via header", digest: "generate nonce on every response to use on the next request"; oauth: "handshake to geneate access token, then use it in header for every other request"). I feel you're working around it due to a missing feature in the persistent plugin though: a callback on established connection. It may be that you can get it to work by using the callbacks plugin though:

class Connection
  HTTP = HTTPX..plugin(:callbacks).plugin(:persistent).with(timeout: {
    connect_timeout: 10,
    request_timeout: 5,
  }).on_connection_opened(&method(:authenticate))
 
  def self.authenticate(new_conn)
    connect(new_conn.origin)
    authenticate(new_conn.origin)
  end
 

lmk if I misunderstood it though.

As a last note, I see you're using httpx on top of async. While it definitely works OOTB, you maybe be creating a connection per fiber when you should instead reusing them. You can verify this by enabling debug logs. I've been testing compatibility with async 2 locally though, so I can share what an async plugin would look like:

module AsyncPlugin
  module InstanceMethods

    private

    def get_current_selector
      selector_store[self] || begin
        return unless block_given?

        default = yield

        set_current_selector(default)

        default
      end
    end
  end
end

HTTPX.plugin(AsyncPlugin).plugin(:persistent)...

still a WIP, but if it helps you, LMK.