fix(smtp): bound outbound delivery timeouts

app/lib/smtp_client/endpoint.rb

@@ -1,6 +1,9 @@
 # frozen_string_literal: true
 
 module SMTPClient
+  class TimeoutError < StandardError
+  end
+
   class Endpoint
 
     class SMTPSessionNotStartedError < StandardError
@@ -55,6 +58,7 @@ def start_smtp_session(source_ip_address: nil, allow_ssl: true)
       @smtp_client = Net::SMTP.new(@ip_address, @server.port)
       @smtp_client.open_timeout = Postal::Config.smtp_client.open_timeout
       @smtp_client.read_timeout = Postal::Config.smtp_client.read_timeout
+      @smtp_client.write_timeout = Postal::Config.smtp_client.write_timeout if @smtp_client.respond_to?(:write_timeout=)
       @smtp_client.tls_hostname = @server.hostname
 
       if source_ip_address
@@ -133,6 +137,16 @@ def finish_smtp_session
       @smtp_client = nil
     end
 
+    # Close the underlying socket without sending SMTP commands. This is used after
+    # hard timeouts where RSET/QUIT may block on the same stuck connection.
+    def abort_smtp_session
+      @smtp_client&.instance_variable_get(:@socket)&.close
+    rescue StandardError
+      nil
+    ensure
+      @smtp_client = nil
+    end
+
     class << self
 
       # Return the default HELO hostname to present to SMTP servers that

app/lib/worker/jobs/process_queued_messages_job.rb

@@ -9,6 +9,7 @@ def call
         @locker = Postal.locker_name_with_suffix(SecureRandom.hex(8))
 
         find_ip_addresses
+        recover_stale_locks
         lock_message_for_processing
         obtain_locked_messages
         process_messages
@@ -39,6 +40,15 @@ def local_ip?(ip)
         !!(ip =~ /\A(127\.|fe80:|::)/)
       end
 
+      def recover_stale_locks
+        recovered = QueuedMessage.where(ip_address_id: [nil, @ip_addresses])
+                                 .where.not(locked_at: nil)
+                                 .where("locked_at < ?", Postal::Config.worker.queued_message_lock_timeout.seconds.ago)
+                                 .update_all(locked_by: nil, locked_at: nil)
+
+        logger.info "recovered #{recovered} stale queued message locks" if recovered.positive?
+      end
+
       # Obtain a queued message from the database for processing
       #
       # @return [void]

app/senders/smtp_sender.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require "timeout"
+
 class SMTPSender < BaseSender
 
   attr_reader :endpoints
@@ -26,10 +28,17 @@ def initialize(domain, source_ip_address = nil, servers: nil, log_id: nil, rcpt_
 
   def start
     servers = @servers || self.class.smtp_relays || resolve_mx_records_for_domain || []
+    deadline = smtp_start_deadline
 
     servers.each do |server|
       server.endpoints.each do |endpoint|
-        result = connect_to_endpoint(endpoint)
+        if smtp_start_timeout_exceeded?(deadline)
+          logger.error "SMTP session setup timed out after #{Postal::Config.smtp_client.start_timeout} seconds"
+          @connection_errors << "SMTP session setup timed out after #{Postal::Config.smtp_client.start_timeout} seconds"
+          return false
+        end
+
+        result = connect_to_endpoint(endpoint, deadline: deadline)
         return endpoint if result
       end
     end
@@ -86,14 +95,16 @@ def finish
   # @return [SendResult]
   def send_message_to_smtp_client(raw_message, mail_from, rcpt_to, retry_on_connection_error: true)
     start_time = Time.now
-    smtp_result = @current_endpoint.send_message(raw_message, mail_from, [rcpt_to])
+    smtp_result = with_smtp_timeout(Postal::Config.smtp_client.transaction_timeout) do
+      @current_endpoint.send_message(raw_message, mail_from, [rcpt_to])
+    end
     logger.info "Accepted by #{@current_endpoint} for #{rcpt_to}"
     create_result("Sent", start_time) do |r|
       r.details = "Message for #{rcpt_to} accepted by #{@current_endpoint}"
       r.details += " (from #{@current_endpoint.smtp_client.source_address})" if @current_endpoint.smtp_client.source_address
       r.output = smtp_result.string
     end
-  rescue Net::SMTPServerBusy, Net::SMTPAuthenticationError, Net::SMTPSyntaxError, Net::SMTPUnknownError, Net::ReadTimeout => e
+  rescue Net::SMTPServerBusy, Net::SMTPAuthenticationError, Net::SMTPSyntaxError, Net::SMTPUnknownError => e
     logger.error "#{e.class}: #{e.message}"
     @current_endpoint.reset_smtp_session
 
@@ -108,6 +119,17 @@ def send_message_to_smtp_client(raw_message, mail_from, rcpt_to, retry_on_connec
         r.retry = true
       end
     end
+  rescue SMTPClient::TimeoutError, Net::OpenTimeout, Net::ReadTimeout => e
+    logger.error "#{e.class}: #{e.message}"
+    timed_out_endpoint = @current_endpoint
+    timed_out_endpoint&.abort_smtp_session
+    @current_endpoint = nil
+
+    create_result("SoftFail", start_time) do |r|
+      r.details = "Temporary SMTP delivery timeout when sending to #{timed_out_endpoint}"
+      r.output = e.message
+      r.retry = true
+    end
   rescue Net::SMTPFatalError => e
     logger.error "#{e.class}: #{e.message}"
     @current_endpoint.reset_smtp_session
@@ -118,6 +140,19 @@ def send_message_to_smtp_client(raw_message, mail_from, rcpt_to, retry_on_connec
     end
   rescue StandardError => e
     logger.error "#{e.class}: #{e.message}"
+
+    if net_write_timeout?(e)
+      timed_out_endpoint = @current_endpoint
+      timed_out_endpoint&.abort_smtp_session
+      @current_endpoint = nil
+
+      return create_result("SoftFail", start_time) do |r|
+        r.details = "Temporary SMTP delivery timeout when sending to #{timed_out_endpoint}"
+        r.output = e.message
+        r.retry = true
+      end
+    end
+
     @current_endpoint.reset_smtp_session
 
     if defined?(Sentry)
@@ -176,7 +211,7 @@ def resolve_mx_records_for_domain
   #
   # @param endpoint [SMTPClient::Endpoint]
   # @return [Boolean]
-  def connect_to_endpoint(endpoint, allow_ssl: true)
+  def connect_to_endpoint(endpoint, allow_ssl: true, deadline: nil)
     if @source_ip_address && @source_ip_address.ipv6.blank? && endpoint.ipv6?
       # Don't try to use IPv6 if the IP address we're sending from doesn't support it.
       return false
@@ -185,11 +220,19 @@ def connect_to_endpoint(endpoint, allow_ssl: true)
     # Add this endpoint to the list of endpoints that we have attempted to connect to
     @endpoints << endpoint unless @endpoints.include?(endpoint)
 
-    endpoint.start_smtp_session(allow_ssl: allow_ssl, source_ip_address: @source_ip_address)
+    with_smtp_timeout(smtp_connection_timeout(deadline)) do
+      endpoint.start_smtp_session(allow_ssl: allow_ssl, source_ip_address: @source_ip_address)
+    end
     logger.info "Connected to #{endpoint}"
     @current_endpoint = endpoint
 
     true
+  rescue SMTPClient::TimeoutError => e
+    endpoint.abort_smtp_session
+    logger.error "Cannot connect to #{endpoint} (#{e.class}: #{e.message})"
+    @connection_errors << e.message unless @connection_errors.include?(e.message)
+
+    false
   rescue StandardError => e
     # Disconnect the SMTP client if we get any errors to avoid leaving
     # a connection around.
@@ -199,7 +242,7 @@ def connect_to_endpoint(endpoint, allow_ssl: true)
     # ssl.
     if e.is_a?(OpenSSL::SSL::SSLError) && endpoint.server.ssl_mode == "Auto"
       logger.error "SSL error (#{e.message}), retrying without SSL"
-      return connect_to_endpoint(endpoint, allow_ssl: false)
+      return connect_to_endpoint(endpoint, allow_ssl: false, deadline: deadline)
     end
 
     # Otherwise, just log the connection error and return false
@@ -209,6 +252,28 @@ def connect_to_endpoint(endpoint, allow_ssl: true)
     false
   end
 
+  def smtp_start_deadline
+    Process.clock_gettime(Process::CLOCK_MONOTONIC) + Postal::Config.smtp_client.start_timeout
+  end
+
+  def smtp_start_timeout_exceeded?(deadline)
+    Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
+  end
+
+  def smtp_connection_timeout(deadline)
+    [Postal::Config.smtp_client.connection_timeout, deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)].min
+  end
+
+  def with_smtp_timeout(seconds)
+    raise SMTPClient::TimeoutError, "SMTP operation timed out" if seconds <= 0
+
+    Timeout.timeout(seconds, SMTPClient::TimeoutError) { yield }
+  end
+
+  def net_write_timeout?(error)
+    defined?(Net::WriteTimeout) && error.is_a?(Net::WriteTimeout)
+  end
+
   # Create a new result object
   #
   # @param type [String] the type of result

doc/config/yaml.yml

@@ -49,6 +49,8 @@ worker:
   default_health_server_bind_address: 127.0.0.1
   # The number of threads to execute within each worker
   threads: 2
+  # The number of seconds after which a queued message processing lock can be recovered
+  queued_message_lock_timeout: 300
 
 main_db:
   # Hostname for the main MariaDB server
@@ -215,6 +217,14 @@ smtp_client:
   open_timeout: 30
   # The read timeout for outgoing SMTP connections
   read_timeout: 30
+  # The write timeout for outgoing SMTP connections
+  write_timeout: 30
+  # The hard timeout for a single outgoing SMTP connection attempt, including greeting and STARTTLS
+  connection_timeout: 10
+  # The hard timeout for finding a usable outgoing SMTP endpoint for one message
+  start_timeout: 60
+  # The hard timeout for an outgoing SMTP MAIL/RCPT/DATA transaction
+  transaction_timeout: 60
 
 migration_waiter:
   # Wait for all migrations to run before starting a process

lib/postal/config_schema.rb

@@ -135,6 +135,11 @@ module Postal
         description "The number of threads to execute within each worker"
         default 2
       end
+
+      integer :queued_message_lock_timeout do
+        description "The number of seconds after which a queued message processing lock can be recovered"
+        default 300
+      end
     end
 
     group :main_db do
@@ -500,6 +505,26 @@ module Postal
         description "The read timeout for outgoing SMTP connections"
         default 30
       end
+
+      integer :write_timeout do
+        description "The write timeout for outgoing SMTP connections"
+        default 30
+      end
+
+      integer :connection_timeout do
+        description "The hard timeout for a single outgoing SMTP connection attempt, including greeting and STARTTLS"
+        default 10
+      end
+
+      integer :start_timeout do
+        description "The hard timeout for finding a usable outgoing SMTP endpoint for one message"
+        default 60
+      end
+
+      integer :transaction_timeout do
+        description "The hard timeout for an outgoing SMTP MAIL/RCPT/DATA transaction"
+        default 60
+      end
     end
 
     group :migration_waiter do

spec/lib/smtp_client/endpoint_spec.rb

@@ -71,6 +71,7 @@ module SMTPClient
           client = endpoint.start_smtp_session
           expect(client.open_timeout).to eq Postal::Config.smtp_client.open_timeout
           expect(client.read_timeout).to eq Postal::Config.smtp_client.read_timeout
+          expect(client.write_timeout).to eq Postal::Config.smtp_client.write_timeout if client.respond_to?(:write_timeout)
         end
 
         it "does not set a source address" do

spec/lib/worker/jobs/process_queued_messages_job_spec.rb

@@ -63,7 +63,7 @@ module Jobs
 
         context "when there is a locked queued message without an IP address without a retry time" do
           it "does nothing" do
-            queued_message = create(:queued_message, :locked, ip_address: nil, retry_after: nil)
+            queued_message = create(:queued_message, :locked, ip_address: nil, retry_after: nil, locked_at: 1.minute.ago)
             job.call
             expect(MessageDequeuer).to_not have_received(:process)
             expect(queued_message.reload.locked?).to be true
@@ -79,6 +79,19 @@ module Jobs
           end
         end
 
+        context "when there is a stale locked queued message without an IP address" do
+          it "recovers the lock and processes the message" do
+            allow(Postal::Config.worker).to receive(:queued_message_lock_timeout).and_return(60)
+            queued_message = create(:queued_message, :locked, ip_address: nil, retry_after: nil, locked_at: 2.minutes.ago)
+
+            job.call
+
+            expect(MessageDequeuer).to have_received(:process).with(queued_message, logger: kind_of(Klogger::Logger))
+            expect(queued_message.reload.locked?).to be true
+            expect(queued_message.locked_by).to match(/\A#{Postal.locker_name} [a-f0-9]{16}\z/)
+          end
+        end
+
         context "when there is an unlocked queued message with an IP address that is ours without a retry time" do
           it "locks the message and calls the service" do
             ip_address = create(:ip_address, ipv4: "10.20.30.40")

spec/senders/smtp_sender_spec.rb

@@ -29,6 +29,7 @@
         smtp_send_message_result
       end
       allow(endpoint).to receive(:finish_smtp_session)
+      allow(endpoint).to receive(:abort_smtp_session)
       allow(endpoint).to receive(:reset_smtp_session)
       allow(endpoint).to receive(:smtp_client) do
         Net::SMTP.new(endpoint.ip_address, endpoint.server.port)
@@ -418,9 +419,28 @@
           )
         end
 
-        it "resets the endpoint SMTP sesssion" do
+        it "aborts the endpoint SMTP sesssion" do
           sender.send_message(message)
-          expect(sender.endpoints.last).to have_received(:reset_smtp_session)
+          expect(sender.endpoints.last).to have_received(:abort_smtp_session)
+        end
+      end
+
+      context "when the SMTP transaction exceeds the hard timeout" do
+        let(:smtp_send_message_error) { proc { SMTPClient::TimeoutError.new("SMTP operation timed out") } }
+
+        it "returns a SoftFail" do
+          result = sender.send_message(message)
+          expect(result).to be_a SendResult
+          expect(result).to have_attributes(
+            type: "SoftFail",
+            details: /Temporary SMTP delivery timeout when sending/
+          )
+        end
+
+        it "aborts the endpoint SMTP session without reset" do
+          sender.send_message(message)
+          expect(sender.endpoints.last).to have_received(:abort_smtp_session)
+          expect(sender.endpoints.last).to_not have_received(:reset_smtp_session)
         end
       end