update

Author: jaredmauch

Mailman/Queue/NewsRunner.py

@@ -128,29 +128,81 @@ def _dispose(self, mlist, msg, msgdata):
                    mlist.internal_name(), str(e))
             return True
 
+    def _onefile(self, msg, msgdata):
+        """Process a single news message.
+        
+        This method overrides the base class's _onefile to add news-specific
+        validation and processing.
+        
+        Args:
+            msg: The message to process
+            msgdata: Additional message metadata
+        """
+        try:
+            # Validate the message
+            if not self._validate_message(msg, msgdata):
+                syslog('error', 'NewsRunner._onefile: Message validation failed')
+                self._shunt.enqueue(msg, msgdata)
+                return
+                
+            # Get the list name from the message data
+            listname = msgdata.get('listname')
+            if not listname:
+                syslog('error', 'NewsRunner._onefile: No listname in message data')
+                self._shunt.enqueue(msg, msgdata)
+                return
+                
+            # Open the list
+            try:
+                mlist = self._open_list(listname)
+            except Exception as e:
+                self.log_error('list_open_error', str(e), listname=listname)
+                self._shunt.enqueue(msg, msgdata)
+                return
+                
+            # Process the message
+            try:
+                keepqueued = self._dispose(mlist, msg, msgdata)
+                if keepqueued:
+                    self._switchboard.enqueue(msg, msgdata)
+            except Exception as e:
+                self._handle_error(e, msg=msg, mlist=mlist)
+                
+        except Exception as e:
+            syslog('error', 'NewsRunner._onefile: Unexpected error: %s', str(e))
+            self._shunt.enqueue(msg, msgdata)
+
     def _oneloop(self):
-        # If NNTP is not enabled, sleep for a while before checking again
-        if not self._nntp:
-            # Check the stop flag every second during sleep
-            for _ in range(60):
-                if self._stop:
-                    return 0
-                time.sleep(1)
-            return 0
+        """Process one batch of messages from the news queue."""
+        try:
+            # Get the list of files to process
+            files = self._switchboard.files()
+            filecnt = len(files)
 
-        # Get one message from the queue
-        msg = self._switchboard.dequeue()
-        if msg is None:
+            # Process each file
+            for filebase in files:
+                try:
+                    # Dequeue the file
+                    msg, msgdata = self._switchboard.dequeue(filebase)
+                    if msg is None:
+                        continue
+                        
+                    # Process the message
+                    try:
+                        self._onefile(msg, msgdata)
+                    except Exception as e:
+                        syslog('error', 'NewsRunner._oneloop: Error processing message %s: %s', filebase, str(e))
+                        continue
+                        
+                except Exception as e:
+                    syslog('error', 'NewsRunner._oneloop: Error dequeuing file %s: %s', filebase, str(e))
+                    continue
+                    
+        except Exception as e:
+            syslog('error', 'NewsRunner._oneloop: Error in main loop: %s', str(e))
             return 0
 
-        # Process the message
-        try:
-            self._dopost(msg)
-        except Exception as e:
-            mailman_log('error', 'NewsRunner error: %s', str(e))
-            # Put the message back in the queue
-            self._switchboard.enqueue(msg, msgdata={})
-        return 1
+        return filecnt
 
     def _queue_news(self, listname, msg, msgdata):
         """Queue a news message for processing."""

Mailman/Queue/Runner.py

@@ -40,6 +40,7 @@
 class Runner:
     QDIR = None
     SLEEPTIME = mm_cfg.QRUNNER_SLEEP_TIME
+    MIN_RETRY_DELAY = 300  # 5 minutes minimum delay between retries
 
     # Message tracking configuration - can be overridden by subclasses
     _track_messages = False  # Whether to track processed messages
@@ -74,6 +75,10 @@ def __init__(self, slice=None, numslices=1):
             self._last_cleanup = time.time()
             self._cleanup_interval = 3600
 
+            # Initialize error tracking attributes
+            self._last_error_time = 0
+            self._error_count = 0
+            
             syslog('debug', 'Runner: Initialization complete')
         except Exception as e:
             syslog('error', 'Runner: Initialization failed: %s\nTraceback:\n%s',
@@ -112,17 +117,32 @@ def run(self):
             # subprocesses we've created and do any other necessary cleanups.
             self._cleanup()
 
-    def log_error(self, error_type, error, msg=None, mlist=None, **context):
-        """Structured error logging with context."""
-        context.update({
+    def log_error(self, error_type, error_msg, **kwargs):
+        """Log an error with the given type and message.
+        
+        Args:
+            error_type: A string identifying the type of error
+            error_msg: The error message to log
+            **kwargs: Additional context to include in the log message
+        """
+        context = {
             'runner': self.__class__.__name__,
-            'list': mlist.internal_name() if mlist else 'N/A',
-            'msg_id': msg.get('message-id', 'N/A') if msg else 'N/A',
             'error_type': error_type,
-            'error': str(error)
-        })
-        syslog('error', '%(runner)s: %(error_type)s - list: %(list)s, msg: %(msg_id)s, error: %(error)s',
-            context)
+            'error_msg': error_msg,
+        }
+        context.update(kwargs)
+        
+        # Format the error message
+        msg_parts = ['%s: %s' % (error_type, error_msg)]
+        if 'msg' in context:
+            msg_parts.append('Message-ID: %s' % context['msg'].get('message-id', 'unknown'))
+        if 'listname' in context:
+            msg_parts.append('List: %s' % context['listname'])
+        if 'traceback' in context:
+            msg_parts.append('Traceback:\n%s' % context['traceback'])
+            
+        # Log the error
+        syslog('error', ' '.join(msg_parts))
 
     def log_warning(self, warning_type, msg=None, mlist=None, **context):
         """Structured warning logging with context."""
@@ -196,7 +216,7 @@ def _oneloop(self):
 
                     # Process the message
                     try:
-                        self._dopost(msg, msgdata)
+                        self._onefile(msg, msgdata)
                     except Exception as e:
                         syslog('error', 'Runner._oneloop: Error processing message %s: %s', filebase, str(e))
                         continue