Serializza gli invii SMTP su una coda unica per non far chiudere Aruba.

Co-authored-by: Cursor <cursoragent@cursor.com>

app/controllers/mailings_controller.rb

@@ -205,7 +205,12 @@ class MailingsController < ApplicationController
     html = @test_preview_html
     subject = "[TEST#{@mailing.ab_test? ? " #{@test_variant}" : ""}] #{@test_preview_subject}"
     recipient = Mailings::TestRecipient.new(mailing: @mailing, email: to)
+    Mailings::OutboundQueue.exclusive do
+      Mailings::SmtpGate.deliver do
         CampaignMailer.outreach(recipient, html: html, subject: subject).deliver_now
+      end
+      Mailings::OutboundQueue.stamp!
+    end
     @mailing.update!(test_sent_at: Time.current, test_sent_to: to)
     redirect_to @mailing, notice: "Prova inviata subito a #{to}. Controlla l'output e, se è corretto, autorizza l'invio di massa."
   rescue StandardError => e

app/jobs/drain_outbound_job.rb

@@ -0,0 +1,16 @@
+class DrainOutboundJob < ApplicationJob
+  queue_as :mailers
+
+  def perform
+    result = Mailings::OutboundQueue.drain_one!
+    return unless result.in?(%i[sent deferred wait])
+    return unless Mailings::OutboundQueue.queued_scope.exists?
+
+    wait = Mailings::OutboundQueue.seconds_until_next_slot
+    wait = 0 if wait.blank? || wait.negative?
+    max_wait = [ Mailings::OutboundQueue.min_interval.to_f, 2.minutes ].max + 5.seconds
+    return if wait > max_wait
+
+    self.class.set(wait: wait).perform_later
+  end
+end

app/jobs/resume_mailings_job.rb

@@ -1,28 +1,16 @@
 class ResumeMailingsJob < ApplicationJob
   queue_as :mailers
 
-  STALE_QUEUED_AFTER = 10.minutes
-
   def perform
-    recover_stale_queued!
     Mailing.sending.due_to_send.find_each do |mailing|
       next unless Mailings::SendClock.new(mailing).open?
-      next if mailing.mailing_recipients.queued.exists?
 
       recipient = mailing.mailing_recipients.pending.order(:id).first
       next unless recipient
 
       SendMailingRecipientJob.perform_later(recipient.id)
     end
-  end
 
-  private
-
-  def recover_stale_queued!
-    MailingRecipient.queued.where("updated_at < ?", STALE_QUEUED_AFTER.ago).find_each do |recipient|
-      next unless recipient.mailing.sending?
-
-      recipient.update!(status: "pending", error_message: nil)
-    end
+    DrainOutboundJob.perform_later if Mailings::OutboundQueue.queued_scope.exists?
   end
 end

app/jobs/send_mailing_recipient_job.rb

@@ -21,14 +21,11 @@ class SendMailingRecipientJob < ApplicationJob
       return
     end
 
-    begin
-      CampaignMailer.raise_delivery_errors = true
-      recipient.deliver!
-    ensure
-      CampaignMailer.raise_delivery_errors = false if Rails.env.development?
-    end
-
+    if recipient.pending? && Mailings::OutboundQueue.enqueue(recipient)
       nxt = mailing.mailing_recipients.pending.order(:id).first
       mailing.enqueue_next_recipient!(nxt, from: Time.current + mailing.interval_seconds.seconds)
     end
+
+    DrainOutboundJob.perform_later
+  end
 end

app/models/mail_identity.rb

@@ -30,7 +30,9 @@ class MailIdentity < ApplicationRecord
       port: smtp_port,
       enable_starttls_auto: encryption == "starttls",
       ssl: encryption == "tls",
-      openssl_verify_mode: verify_ssl? ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE
+      openssl_verify_mode: verify_ssl? ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE,
+      open_timeout: 15,
+      read_timeout: 60
     }
     if smtp_username.present?
       settings[:user_name] = smtp_username

app/models/mailing_recipient.rb

@@ -71,32 +71,70 @@ class MailingRecipient < ApplicationRecord
     MailMerge.render(mailing.subject_for(ab_variant), **merge_context)
   end
 
-  def deliver!
+  def queued?
+    status == "queued"
+  end
+
+  def enqueue_for_delivery!
+    Mailings::OutboundQueue.enqueue(self)
+  end
+
+  def deliver_queued!
     with_lock do
       return if status.in?(%w[sent skipped])
       return abort_delivery!("invio annullato") unless mailing.sending?
-
-      update!(status: "queued")
+      return unless queued?
     end
 
-    return unless mailing.sending?
-
     html = rendered_html
     subject_line = rendered_subject_line
+    Mailings::SmtpGate.deliver do
       CampaignMailer.outreach(self, html: html, subject: subject_line).deliver_now
+    end
     record_success!(subject_line)
+    :sent
+  rescue *Mailings::SmtpGate::RETRYABLE => e
+    defer_or_fail!(e)
   rescue StandardError => e
     update!(status: "failed", error_message: e.message.to_s.truncate(500))
+    :failed
   ensure
     mailing.mark_finished_if_done!
   end
 
+  def deliver!
+    enqueue_for_delivery!
+    deliver_queued! if queued?
+  end
+
   private
 
   def abort_delivery!(reason)
     update!(status: "skipped", skip_reason: reason)
   end
 
+  MAX_SMTP_DEFERS = 8
+  DEFER_PATTERN = /tentativo (\d+)\//
+
+  def defer_or_fail!(error)
+    attempt = smtp_defer_count + 1
+    message = error.message.to_s.truncate(400)
+    if attempt >= MAX_SMTP_DEFERS
+      update!(status: "failed", error_message: message)
+      :failed
+    else
+      update!(
+        status: "queued",
+        error_message: "Da ritentare (#{attempt}/#{MAX_SMTP_DEFERS}): #{message}"
+      )
+      :deferred
+    end
+  end
+
+  def smtp_defer_count
+    error_message.to_s[DEFER_PATTERN, 1].to_i
+  end
+
   def record_success!(subject_line)
     transaction do
       update!(status: "sent", sent_at: Time.current, rendered_subject: subject_line, error_message: nil)

app/services/mailings/outbound_queue.rb

@@ -0,0 +1,98 @@
+class Mailings::OutboundQueue
+  LOCK_PATH = Rails.root.join("tmp/outbound_smtp.lock")
+  STAMP_PATH = Rails.root.join("tmp/outbound_last_sent")
+
+  class << self
+    attr_accessor :min_interval
+
+    def enqueue(recipient)
+      recipient.with_lock do
+        return false unless recipient.mailing.sending?
+        return false unless recipient.pending?
+
+        recipient.update!(status: "queued", error_message: nil)
+        true
+      end
+    end
+
+    def drain_one!
+      File.open(lock_path, File::RDWR | File::CREAT, 0o644) do |file|
+        return :busy unless file.flock(File::LOCK_EX | File::LOCK_NB)
+
+        wait = seconds_until_next_slot
+        return :wait if wait.positive?
+
+        recipient = next_queued
+        return :empty if recipient.nil?
+        return :closed unless Mailings::SendClock.new(recipient.mailing).open?
+
+        begin
+          CampaignMailer.raise_delivery_errors = true
+          outcome = recipient.deliver_queued!
+          stamp!
+          outcome == :sent ? :sent : :deferred
+        ensure
+          CampaignMailer.raise_delivery_errors = false if Rails.env.development?
+        end
+      end
+    end
+
+    def exclusive
+      File.open(lock_path, File::RDWR | File::CREAT, 0o644) do |file|
+        file.flock(File::LOCK_EX)
+        yield
+      end
+    end
+
+    def next_queued
+      queued_scope.order(:updated_at, :id).first
+    end
+
+    def queued_scope
+      MailingRecipient.queued.joins(:mailing).merge(Mailing.sending)
+    end
+
+    def seconds_until_next_slot
+      return 0 if min_interval.to_f <= 0
+
+      last = last_sent_at
+      return 0 if last.nil?
+
+      remaining = min_interval - (Time.current - last)
+      remaining.positive? ? remaining : 0
+    end
+
+    def reset!
+      @last_sent_at = nil
+      File.delete(stamp_path) if File.exist?(stamp_path)
+    rescue Errno::ENOENT
+      nil
+    end
+
+    def stamp!
+      @last_sent_at = Time.current
+      File.write(stamp_path, @last_sent_at.to_f.to_s)
+    end
+
+    def last_sent_at
+      return @last_sent_at if @last_sent_at
+      return unless File.exist?(stamp_path)
+
+      @last_sent_at = Time.zone.at(Float(File.read(stamp_path)))
+    rescue ArgumentError, TypeError, Errno::ENOENT
+      nil
+    end
+
+    private
+
+    def lock_path
+      LOCK_PATH
+    end
+
+    def stamp_path
+      STAMP_PATH
+    end
+  end
+
+  self.min_interval = Rails.env.test? ? 0.0 : 180.0
+end

app/services/mailings/smtp_gate.rb

@@ -0,0 +1,73 @@
+require "net/smtp"
+require "openssl"
+
+class Mailings::SmtpGate
+  MUTEX = Mutex.new
+  RETRYABLE = [
+    EOFError,
+    IOError,
+    Errno::ECONNRESET,
+    Errno::EPIPE,
+    Errno::ETIMEDOUT,
+    Net::OpenTimeout,
+    Net::ReadTimeout,
+    OpenSSL::SSL::SSLError,
+    Net::SMTPServerBusy
+  ].freeze
+
+  class << self
+    attr_accessor :min_gap, :max_attempts, :backoff_base
+
+    def deliver
+      MUTEX.synchronize { deliver_locked { yield } }
+    end
+
+    def reset!
+      MUTEX.synchronize { @last_monotonic = nil }
+    end
+
+    private
+
+    def deliver_locked
+      attempts = 0
+      begin
+        wait_min_gap
+        result = yield
+        stamp!
+        result
+      rescue *RETRYABLE => e
+        attempts += 1
+        stamp!
+        Rails.logger.warn("[smtp-gate] #{e.class}: #{e.message} attempt=#{attempts}/#{max_attempts}")
+        raise if attempts >= max_attempts
+
+        sleep_wait(backoff_base * (2**(attempts - 1)))
+        retry
+      end
+    end
+
+    def wait_min_gap
+      return if min_gap.to_f <= 0 || @last_monotonic.nil?
+
+      elapsed = now - @last_monotonic
+      remaining = min_gap - elapsed
+      sleep_wait(remaining) if remaining.positive?
+    end
+
+    def stamp!
+      @last_monotonic = now
+    end
+
+    def now
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+
+    def sleep_wait(seconds)
+      sleep(seconds) if seconds.to_f.positive?
+    end
+  end
+
+  self.min_gap = 0.0
+  self.max_attempts = 1
+  self.backoff_base = 0.0
+end

config/initializers/mailing_poller.rb

@@ -2,7 +2,7 @@
 # o dopo un riavvio del processo. L'adapter :async non persiste i job.
 Rails.application.config.after_initialize do
   next if Rails.env.test?
-  next unless defined?(Rails::Server) || ENV["MAILING_POLLER"] == "true"
+  next unless defined?(Rails::Server)
 
   Thread.new do
     Rails.logger.info("[mailing-poller] avviato")

test/models/mail_identity_test.rb

@@ -33,6 +33,8 @@ class MailIdentityTest < ActiveSupport::TestCase
     settings = identity.smtp_settings
     assert settings[:ssl]
     assert_not settings[:enable_starttls_auto]
+    assert_equal 15, settings[:open_timeout]
+    assert_equal 60, settings[:read_timeout]
   end
 
   test "rejects ssl encryption on port 587" do

test/services/mailings/outbound_queue_test.rb

@@ -0,0 +1,106 @@
+require "test_helper"
+
+class Mailings::OutboundQueueTest < ActiveSupport::TestCase
+  include ActionMailer::TestHelper
+
+  setup do
+    Mailings::OutboundQueue.min_interval = 0
+    Mailings::OutboundQueue.reset!
+    Mailings::SmtpGate.min_gap = 0
+    Mailings::SmtpGate.backoff_base = 0
+    Mailings::SmtpGate.reset!
+  end
+
+  teardown do
+    Mailings::OutboundQueue.min_interval = 0
+    Mailings::OutboundQueue.reset!
+  end
+
+  test "campaign jobs only enqueue, drain sends one email at a time" do
+    first = create_campaign_org(name: "Club Uno", email: "uno@example.com")
+    second = create_campaign_org(name: "Club Due", email: "due@example.com")
+    mailing_a = create_sending_mailing(name: "Campagna A")
+    mailing_b = create_sending_mailing(name: "Campagna B")
+    recipient_a = enqueue_org(mailing_a, first)
+    recipient_b = enqueue_org(mailing_b, second)
+
+    assert_equal "queued", recipient_a.reload.status
+    assert_equal "queued", recipient_b.status
+
+    assert_emails 1 do
+      assert_equal :sent, Mailings::OutboundQueue.drain_one!
+    end
+    assert_equal "sent", recipient_a.reload.status
+    assert_equal "queued", recipient_b.reload.status
+
+    assert_emails 1 do
+      assert_equal :sent, Mailings::OutboundQueue.drain_one!
+    end
+    assert_equal "sent", recipient_b.reload.status
+    assert_equal :empty, Mailings::OutboundQueue.drain_one!
+  end
+
+  test "does not start a second SMTP session while the lock is held" do
+    org = create_campaign_org(name: "Club Lock", email: "lock@example.com")
+    mailing = create_sending_mailing(name: "Campagna lock")
+    enqueue_org(mailing, org)
+
+    started = Queue.new
+    release = Queue.new
+    result = nil
+    thread = Thread.new do
+      Mailings::OutboundQueue.exclusive do
+        started << true
+        release.pop
+      end
+    end
+    started.pop
+    result = Mailings::OutboundQueue.drain_one!
+    release << true
+    thread.join
+
+    assert_equal :busy, result
+    assert_equal "queued", mailing.mailing_recipients.first.reload.status
+  end
+
+  test "transient SMTP errors stay queued and the next drain tries another recipient" do
+    first = create_campaign_org(name: "Club EOF", email: "eof@example.com")
+    second = create_campaign_org(name: "Club Ok", email: "ok@example.com")
+    mailing = create_sending_mailing(name: "Campagna eof")
+    recipient_a = enqueue_org(mailing, first)
+    recipient_b = enqueue_org(mailing, second)
+    gate = Mailings::SmtpGate
+    original = gate.method(:deliver)
+    gate.define_singleton_method(:deliver) { raise EOFError, "end of file reached" }
+
+    begin
+      assert_equal :deferred, Mailings::OutboundQueue.drain_one!
+    ensure
+      gate.define_singleton_method(:deliver, original)
+    end
+
+    recipient_a.reload
+    assert_equal "queued", recipient_a.status
+    assert_match(/Da ritentare \(1\/8\)/, recipient_a.error_message)
+    assert_equal recipient_b.id, Mailings::OutboundQueue.next_queued.id
+  end
+
+  private
+
+  def create_sending_mailing(name:)
+    mailing = create_mailing(name: name, audience: "to_send", interval_seconds: 0)
+    mailing.update!(status: "sending", queued_at: Time.current, test_sent_at: Time.current, test_sent_to: "test@example.com")
+    mailing
+  end
+
+  def enqueue_org(mailing, org)
+    recipient = mailing.mailing_recipients.create!(
+      organization: org,
+      contact: org.contacts.first,
+      email: org.email,
+      status: "pending"
+    )
+    assert Mailings::OutboundQueue.enqueue(recipient.reload)
+    recipient
+  end
+end

test/services/mailings/smtp_gate_test.rb

@@ -0,0 +1,69 @@
+require "test_helper"
+
+class Mailings::SmtpGateTest < ActiveSupport::TestCase
+  setup do
+    Mailings::SmtpGate.min_gap = 0
+    Mailings::SmtpGate.backoff_base = 0
+    Mailings::SmtpGate.max_attempts = 4
+    Mailings::SmtpGate.reset!
+  end
+
+  teardown do
+    Mailings::SmtpGate.min_gap = 0
+    Mailings::SmtpGate.backoff_base = 0
+    Mailings::SmtpGate.max_attempts = 4
+    Mailings::SmtpGate.reset!
+  end
+
+  test "retries EOFError then succeeds" do
+    hits = 0
+    result = Mailings::SmtpGate.deliver do
+      hits += 1
+      raise EOFError, "end of file reached" if hits < 3
+
+      :sent
+    end
+
+    assert_equal :sent, result
+    assert_equal 3, hits
+  end
+
+  test "gives up after max attempts" do
+    hits = 0
+    error = assert_raises(EOFError) do
+      Mailings::SmtpGate.deliver do
+        hits += 1
+        raise EOFError, "end of file reached"
+      end
+    end
+
+    assert_equal "end of file reached", error.message
+    assert_equal Mailings::SmtpGate.max_attempts, hits
+  end
+
+  test "does not retry permanent SMTP errors" do
+    hits = 0
+    assert_raises(Net::SMTPAuthenticationError) do
+      Mailings::SmtpGate.deliver do
+        hits += 1
+        raise Net::SMTPAuthenticationError, "535 authentication failed"
+      end
+    end
+    assert_equal 1, hits
+  end
+
+  test "serializes concurrent deliveries" do
+    order = Queue.new
+    threads = 2.times.map do |i|
+      Thread.new do
+        Mailings::SmtpGate.deliver do
+          order << i
+          sleep 0.05
+        end
+      end
+    end
+    threads.each(&:join)
+
+    assert_equal 2, order.size
+  end
+end