From e5deff1f177e9d62a3365ec3f0c91e67e90cbddc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hac=C4=B1=20Mert=20G=C3=B6khan?= Date: Thu, 14 May 2026 11:37:47 +0300 Subject: [PATCH] fix(smtp): bound outbound delivery timeouts --- app/lib/smtp_client/endpoint.rb | 14 ++++ .../jobs/process_queued_messages_job.rb | 10 +++ app/senders/smtp_sender.rb | 77 +++++++++++++++++-- doc/config/yaml.yml | 10 +++ lib/postal/config_schema.rb | 25 ++++++ spec/lib/smtp_client/endpoint_spec.rb | 1 + .../jobs/process_queued_messages_job_spec.rb | 15 +++- spec/senders/smtp_sender_spec.rb | 24 +++++- 8 files changed, 167 insertions(+), 9 deletions(-) diff --git a/app/lib/smtp_client/endpoint.rb b/app/lib/smtp_client/endpoint.rb index 5449e695a..c030abbbc 100644 --- a/app/lib/smtp_client/endpoint.rb +++ b/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 diff --git a/app/lib/worker/jobs/process_queued_messages_job.rb b/app/lib/worker/jobs/process_queued_messages_job.rb index 18489b578..ede753ddd 100644 --- a/app/lib/worker/jobs/process_queued_messages_job.rb +++ b/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] diff --git a/app/senders/smtp_sender.rb b/app/senders/smtp_sender.rb index 1f3c863fd..ee8274527 100644 --- a/app/senders/smtp_sender.rb +++ b/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 diff --git a/doc/config/yaml.yml b/doc/config/yaml.yml index f3a735a9f..1a6594986 100644 --- a/doc/config/yaml.yml +++ b/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 diff --git a/lib/postal/config_schema.rb b/lib/postal/config_schema.rb index e3c6415d5..5d5beae4d 100644 --- a/lib/postal/config_schema.rb +++ b/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 diff --git a/spec/lib/smtp_client/endpoint_spec.rb b/spec/lib/smtp_client/endpoint_spec.rb index 5af326735..0939047c1 100644 --- a/spec/lib/smtp_client/endpoint_spec.rb +++ b/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 diff --git a/spec/lib/worker/jobs/process_queued_messages_job_spec.rb b/spec/lib/worker/jobs/process_queued_messages_job_spec.rb index 46ed1efd9..12a1e0b58 100644 --- a/spec/lib/worker/jobs/process_queued_messages_job_spec.rb +++ b/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") diff --git a/spec/senders/smtp_sender_spec.rb b/spec/senders/smtp_sender_spec.rb index 6008a2655..0e330ce87 100644 --- a/spec/senders/smtp_sender_spec.rb +++ b/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