diff --git a/app/controllers/mailings_controller.rb b/app/controllers/mailings_controller.rb index 0796498..d057bdb 100644 --- a/app/controllers/mailings_controller.rb +++ b/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) - CampaignMailer.outreach(recipient, html: html, subject: subject).deliver_now + 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 diff --git a/app/jobs/drain_outbound_job.rb b/app/jobs/drain_outbound_job.rb new file mode 100644 index 0000000..aad71ea --- /dev/null +++ b/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 diff --git a/app/jobs/resume_mailings_job.rb b/app/jobs/resume_mailings_job.rb index ab3cc4d..7ccb33b 100644 --- a/app/jobs/resume_mailings_job.rb +++ b/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 diff --git a/app/jobs/send_mailing_recipient_job.rb b/app/jobs/send_mailing_recipient_job.rb index 295ebff..0aaff9c 100644 --- a/app/jobs/send_mailing_recipient_job.rb +++ b/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? + 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 - nxt = mailing.mailing_recipients.pending.order(:id).first - mailing.enqueue_next_recipient!(nxt, from: Time.current + mailing.interval_seconds.seconds) + DrainOutboundJob.perform_later end end diff --git a/app/models/mail_identity.rb b/app/models/mail_identity.rb index ad9c495..7331a88 100644 --- a/app/models/mail_identity.rb +++ b/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 diff --git a/app/models/mailing_recipient.rb b/app/models/mailing_recipient.rb index 3272f3b..6839073 100644 --- a/app/models/mailing_recipient.rb +++ b/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 - CampaignMailer.outreach(self, html: html, subject: subject_line).deliver_now + 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) diff --git a/app/services/mailings/outbound_queue.rb b/app/services/mailings/outbound_queue.rb new file mode 100644 index 0000000..3713bb1 --- /dev/null +++ b/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 diff --git a/app/services/mailings/smtp_gate.rb b/app/services/mailings/smtp_gate.rb new file mode 100644 index 0000000..a8f713a --- /dev/null +++ b/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 diff --git a/config/initializers/mailing_poller.rb b/config/initializers/mailing_poller.rb index a6c9e36..89fa022 100644 --- a/config/initializers/mailing_poller.rb +++ b/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") diff --git a/test/models/mail_identity_test.rb b/test/models/mail_identity_test.rb index c2dcba2..5863c7f 100644 --- a/test/models/mail_identity_test.rb +++ b/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 diff --git a/test/services/mailings/outbound_queue_test.rb b/test/services/mailings/outbound_queue_test.rb new file mode 100644 index 0000000..e83a2a8 --- /dev/null +++ b/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 diff --git a/test/services/mailings/smtp_gate_test.rb b/test/services/mailings/smtp_gate_test.rb new file mode 100644 index 0000000..cdee586 --- /dev/null +++ b/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