update

Author: jaredmauch

Mailman/Queue/IncomingRunner.py

@@ -157,100 +157,55 @@ def _convert_message(self, msg):
         return Runner._convert_message(self, msg)
 
     def _dispose(self, mlist, msg, msgdata):
-        """Process an incoming message."""
-        msgid = msg.get('message-id', 'n/a')
-        filebase = msgdata.get('_filebase', 'unknown')
-        
+        # Try to get the list lock.
+        try:
+            mlist.Lock(timeout=mm_cfg.LIST_LOCK_TIMEOUT)
+        except LockFile.TimeOutError:
+            # Oh well, try again later
+            return 1
+        # Process the message through a handler pipeline.  The handler
+        # pipeline can actually come from one of three places: the message
+        # metadata, the mlist, or the global pipeline.
+        #
+        # If a message was requeued due to an uncaught exception, its metadata
+        # will contain the retry pipeline.  Use this above all else.
+        # Otherwise, if the mlist has a `pipeline' attribute, it should be
+        # used.  Final fallback is the global pipeline.
         try:
-            # Enhanced logging for message details
-            mailman_log('debug', 'IncomingRunner._dispose: Starting to process message %s (file: %s)', msgid, filebase)
-            mailman_log('debug', 'Message details:')
-            mailman_log('debug', '  Message ID: %s', msgid)
-            mailman_log('debug', '  From: %s', msg.get('from', 'unknown'))
-            mailman_log('debug', '  To: %s', msg.get('to', 'unknown'))
-            mailman_log('debug', '  Subject: %s', msg.get('subject', '(no subject)'))
-            mailman_log('debug', '  Message type: %s', type(msg).__name__)
-            mailman_log('debug', '  Message data: %s', str(msgdata))
-            mailman_log('debug', '  Pipeline: %s', msgdata.get('pipeline', 'No pipeline'))
-            
-            # Check if this is an administrative message
-            is_admin = msgdata.get('admin_type', False)
-            mailman_log('debug', '  Is admin message: %s', is_admin)
-            
-            # Check if this is a list post
-            is_list_post = msgdata.get('list_post', False)
-            mailman_log('debug', '  Is list post: %s', is_list_post)
-            
-            # Log recipients information
-            recipients = msgdata.get('recips', [])
-            mailman_log('debug', '  Recipients: %s', recipients)
-            if not recipients:
-                mailman_log('warning', 'IncomingRunner: No recipients found in msgdata for message %s, pipeline handlers may set them', msgid)
-                mailman_log('debug', '  Message data: %s', str(msgdata))
-                mailman_log('debug', '  To header: %s', msg.get('to', 'unknown'))
-                mailman_log('debug', '  Cc header: %s', msg.get('cc', 'unknown'))
-                mailman_log('debug', '  Resent-To: %s', msg.get('resent-to', 'unknown'))
-                mailman_log('debug', '  Resent-Cc: %s', msg.get('resent-cc', 'unknown'))
-            
-            # Convert Python's Message to Mailman's Message if needed
-            msg = self._convert_message(msg)
-            
-            # Check if this is a bounce message
-            if self._is_bounce(msg):
-                mailman_log('debug', 'IncomingRunner._dispose: Message %s is a bounce, routing to bounce queue', msgid)
-                # Route to bounce queue
-                bounce_queue = Switchboard(mm_cfg.BOUNCEQUEUE_DIR)
-                bounce_queue.enqueue(msg, msgdata)
-                return False
-            
-            # Get the pipeline
             pipeline = self._get_pipeline(mlist, msg, msgdata)
-            if not pipeline:
-                mailman_log('error', 'IncomingRunner._dispose: No pipeline found for message %s', msgid)
-                return False
-            
-            # Process the message through the pipeline
-            try:
-                more = self._dopipeline(mlist, msg, msgdata, pipeline)
-                if more:
-                    mailman_log('debug', 'IncomingRunner._dispose: Message %s needs more processing', msgid)
-                    return True
-            except Exception as e:
-                mailman_log('error', 'IncomingRunner._dispose: Error processing message %s: %s\nTraceback:\n%s',
-                           msgid, str(e), traceback.format_exc())
-                return False
-            
-            mailman_log('debug', 'IncomingRunner._dispose: Successfully processed message %s', msgid)
-            return True
-            
-        except Exception as e:
-            mailman_log('error', 'IncomingRunner._dispose: Error processing message %s: %s\nTraceback:\n%s',
-                       msgid, str(e), traceback.format_exc())
-            return False
+            msgdata['pipeline'] = pipeline
+            more = self._dopipeline(mlist, msg, msgdata, pipeline)
+            if not more:
+                del msgdata['pipeline']
+            mlist.Save()
+            return more
+        finally:
+            mlist.Unlock()
 
     def _get_pipeline(self, mlist, msg, msgdata):
-        """Get the pipeline for processing the message."""
         # We must return a copy of the list, otherwise, the first message that
         # flows through the pipeline will empty it out!
         return msgdata.get('pipeline',
-                          getattr(mlist, 'pipeline',
-                                 mm_cfg.GLOBAL_PIPELINE))[:]
+                           getattr(mlist, 'pipeline',
+                                   mm_cfg.GLOBAL_PIPELINE))[:]
 
     def _dopipeline(self, mlist, msg, msgdata, pipeline):
-        """Process the message through the pipeline of handlers."""
         while pipeline:
             handler = pipeline.pop(0)
             modname = 'Mailman.Handlers.' + handler
             __import__(modname)
             try:
                 pid = os.getpid()
                 sys.modules[modname].process(mlist, msg, msgdata)
-                # Failsafe -- a child may have leaked through
+                # Failsafe -- a child may have leaked through.
                 if pid != os.getpid():
                     mailman_log('error', 'Child process leaked through: %s', modname)
                     os._exit(1)
             except Errors.DiscardMessage:
-                # Throw the message away
+                # Throw the message away; we need do nothing else with it.
+                # We do need to push the current handler back in the pipeline
+                # just in case the syslog call throws an exception and the
+                # message is shunted.
                 pipeline.insert(0, handler)
                 mailman_log('vette', """Message discarded, msgid: %s
         list: %s,
@@ -259,10 +214,14 @@ def _dopipeline(self, mlist, msg, msgdata, pipeline):
                        mlist.real_name, handler)
                 return 0
             except Errors.HoldMessage:
-                # Let the approval process take it from here
+                # Let the approval process take it from here.  The message no
+                # longer needs to be queued.
                 return 0
             except Errors.RejectMessage as e:
-                # Log rejection and bounce message
+                # Log this.
+                # We do need to push the current handler back in the pipeline
+                # just in case the syslog call or BounceMessage throws an
+                # exception and the message is shunted.
                 pipeline.insert(0, handler)
                 mailman_log('vette', """Message rejected, msgid: %s
         list: %s,
@@ -272,8 +231,9 @@ def _dopipeline(self, mlist, msg, msgdata, pipeline):
                        mlist.real_name, handler, e.notice())
                 mlist.BounceMessage(msg, msgdata, e)
                 return 0
-            except Exception as e:
+            except:
                 # Push this pipeline module back on the stack, then re-raise
+                # the exception.
                 pipeline.insert(0, handler)
                 raise
         # We've successfully completed handling of this message

Mailman/Queue/OutgoingRunner.py

@@ -81,22 +81,22 @@ def __init__(self, slice=None, numslices=1):
             self._last_cleanup = time.time()
 
             # We look this function up only at startup time
-            self._modname = 'Mailman.Handlers.' + mm_cfg.DELIVERY_MODULE
-            mailman_log('debug', 'OutgoingRunner: Attempting to import delivery module: %s', self._modname)
+            modname = 'Mailman.Handlers.' + mm_cfg.DELIVERY_MODULE
+            mailman_log('debug', 'OutgoingRunner: Attempting to import delivery module: %s', modname)
 
             try:
-                mod = __import__(self._modname)
+                mod = __import__(modname)
                 mailman_log('debug', 'OutgoingRunner: Successfully imported delivery module')
             except ImportError as e:
-                mailman_log('error', 'OutgoingRunner: Failed to import delivery module %s: %s', self._modname, str(e))
+                mailman_log('error', 'OutgoingRunner: Failed to import delivery module %s: %s', modname, str(e))
                 mailman_log('error', 'OutgoingRunner: Traceback: %s', traceback.format_exc())
                 raise
 
             try:
-                self._func = getattr(sys.modules[self._modname], 'process')
+                self._func = getattr(sys.modules[modname], 'process')
                 mailman_log('debug', 'OutgoingRunner: Successfully got process function from module')
             except AttributeError as e:
-                mailman_log('error', 'OutgoingRunner: Failed to get process function from module %s: %s', self._modname, str(e))
+                mailman_log('error', 'OutgoingRunner: Failed to get process function from module %s: %s', modname, str(e))
                 mailman_log('error', 'OutgoingRunner: Traceback: %s', traceback.format_exc())
                 raise
 
@@ -242,100 +242,77 @@ def _validate_message(self, msg, msgdata):
             return msg, False
 
     def _dispose(self, mlist, msg, msgdata):
-        """Process an outgoing message."""
-        msgid = msg.get('message-id', 'n/a')
-        filebase = msgdata.get('_filebase', 'unknown')
-        
-        # Log the full msgdata at the start of processing
-        mailman_log('debug', 'OutgoingRunner._dispose: Full msgdata at start:\n%s', str(msgdata))
-        
-        # Ensure we have a MailList object
-        if isinstance(mlist, str):
-            try:
-                mlist = get_mail_list()(mlist, lock=0)
-                should_unlock = True
-            except Errors.MMUnknownListError:
-                mailman_log('error', 'OutgoingRunner: Unknown list %s', mlist)
-                self._shunt.enqueue(msg, msgdata)
-                return True
-        else:
-            should_unlock = False
-        
+        # See if we should retry delivery of this message again.
+        deliver_after = msgdata.get('deliver_after', 0)
+        if time.time() < deliver_after:
+            return True
+        # Make sure we have the most up-to-date state
+        mlist.Load()
         try:
-            mailman_log('debug', 'OutgoingRunner._dispose: Starting to process outgoing message %s (file: %s) for list %s',
-                       msgid, filebase, mlist.internal_name())
-            
-            # Check retry delay and duplicate processing
-            if not self._check_retry_delay(msgid, filebase):
-                mailman_log('debug', 'OutgoingRunner._dispose: Message %s failed retry delay check, skipping', msgid)
-                return False
-
-            # Make sure we have the most up-to-date state
-            try:
-                mlist.Load()
-                mailman_log('debug', 'OutgoingRunner._dispose: Successfully loaded list %s', mlist.internal_name())
-            except Errors.MMCorruptListDatabaseError as e:
-                mailman_log('error', 'OutgoingRunner._dispose: Failed to load list %s: %s\nTraceback:\n%s',
-                           mlist.internal_name(), str(e), traceback.format_exc())
-                self._unmark_message_processed(msgid)
-                return False
-            except Exception as e:
-                mailman_log('error', 'OutgoingRunner._dispose: Unexpected error loading list %s: %s\nTraceback:\n%s',
-                           mlist.internal_name(), str(e), traceback.format_exc())
-                self._unmark_message_processed(msgid)
-                return False
-
-            # Validate message type first
-            msg, success = self._validate_message(msg, msgdata)
-            if not success:
-                mailman_log('error', 'OutgoingRunner._dispose: Message validation failed for message %s', msgid)
-                self._unmark_message_processed(msgid)
-                return False
-
-            # Log the full msgdata after validation
-            mailman_log('debug', 'OutgoingRunner._dispose: Full msgdata after validation:\n%s', str(msgdata))
-
-            # Validate message headers
-            if not msg.get('message-id'):
-                mailman_log('error', 'OutgoingRunner._dispose: Message missing Message-ID header')
-                self._unmark_message_processed(msgid)
-                return False
-
-            # Process the outgoing message
-            try:
-                mailman_log('debug', 'OutgoingRunner._dispose: Processing outgoing message %s', msgid)
-                
-                # Get message type and recipient
-                msgtype = msgdata.get('_msgtype', 'unknown')
-                recipient = msgdata.get('recipient', 'unknown')
-                
-                mailman_log('debug', 'OutgoingRunner._dispose: Message %s is type %s for recipient %s',
-                           msgid, msgtype, recipient)
-                
-                # Process based on message type
-                if msgtype == 'bounce':
-                    success = self._process_bounce(mlist, msg, msgdata)
-                elif msgtype == 'admin':
-                    success = self._process_admin(mlist, msg, msgdata)
-                else:
-                    success = self._process_regular(mlist, msg, msgdata)
-                    
-                if success:
-                    mailman_log('debug', 'OutgoingRunner._dispose: Successfully processed outgoing message %s', msgid)
-                    return True
-                else:
-                    mailman_log('error', 'OutgoingRunner._dispose: Failed to process outgoing message %s', msgid)
-                    return False
-
-            except Exception as e:
-                mailman_log('error', 'OutgoingRunner._dispose: Error processing outgoing message %s: %s\nTraceback:\n%s',
-                           msgid, str(e), traceback.format_exc())
-                self._unmark_message_processed(msgid)
-                return False
-                
-        finally:
-            if should_unlock:
-                mlist.Unlock()
+            pid = os.getpid()
+            self._func(mlist, msg, msgdata)
+            # Failsafe -- a child may have leaked through.
+            if pid != os.getpid():
+                mailman_log('error', 'child process leaked thru: %s', mm_cfg.DELIVERY_MODULE)
+                os._exit(1)
+            self.__logged = False
+        except socket.error:
+            # There was a problem connecting to the SMTP server.  Log this
+            # once, but crank up our sleep time so we don't fill the error
+            # log.
+            port = mm_cfg.SMTPPORT
+            if port == 0:
+                port = 'smtp'
+            # Log this just once.
+            if not self.__logged:
+                mailman_log('error', 'Cannot connect to SMTP server %s on port %s',
+                           mm_cfg.SMTPHOST, port)
+                self.__logged = True
+            self._snooze(0)
+            return True
+        except Errors.SomeRecipientsFailed as e:
+            # Handle local rejects of probe messages differently.
+            if msgdata.get('probe_token') and e.permfailures:
+                self._probe_bounce(mlist, msgdata['probe_token'])
+            else:
+                # Delivery failed at SMTP time for some or all of the
+                # recipients.  Permanent failures are registered as bounces,
+                # but temporary failures are retried for later.
+                #
+                # BAW: msg is going to be the original message that failed
+                # delivery, not a bounce message.  This may be confusing if
+                # this is what's sent to the user in the probe message.  Maybe
+                # we should craft a bounce-like message containing information
+                # about the permanent SMTP failure?
+                if e.permfailures:
+                    self._queue_bounces(mlist.internal_name(), e.permfailures,
+                                        msg)
+                # Move temporary failures to the qfiles/retry queue which will
+                # occasionally move them back here for another shot at
+                # delivery.
+                if e.tempfailures:
+                    now = time.time()
+                    recips = e.tempfailures
+                    last_recip_count = msgdata.get('last_recip_count', 0)
+                    deliver_until = msgdata.get('deliver_until', now)
+                    if len(recips) == last_recip_count:
+                        # We didn't make any progress, so don't attempt
+                        # delivery any longer.  BAW: is this the best
+                        # disposition?
+                        if now > deliver_until:
+                            return False
+                    else:
+                        # Keep trying to delivery this message for a while
+                        deliver_until = now + mm_cfg.DELIVERY_RETRY_PERIOD
+                    # Don't retry delivery too soon.
+                    deliver_after = now + mm_cfg.DELIVERY_RETRY_WAIT
+                    msgdata['deliver_after'] = deliver_after
+                    msgdata['last_recip_count'] = len(recips)
+                    msgdata['deliver_until'] = deliver_until
+                    msgdata['recips'] = recips
+                    self.__retryq.enqueue(msg, msgdata)
+        # We've successfully completed handling of this message
+        return False
 
     def _process_bounce(self, mlist, msg, msgdata):
         """Process a bounce message."""