29 files changed, 1615 insertions, 397 deletions

diff --git a/src/mailman/app/bounces.py b/src/mailman/app/bounces.py
index f216c959e..50a9f54d8 100644
--- a/src/mailman/app/bounces.py
+++ b/src/mailman/app/bounces.py
@@ -24,6 +24,7 @@ __all__ = [
     'ProbeVERP',
     'StandardVERP',
     'bounce_message',
+    'maybe_forward',
     'scan_message',
     'send_probe',
     ]
@@ -42,8 +43,9 @@ from zope.interface import implements
 from mailman.app.finder import find_components
 from mailman.config import config
 from mailman.core.i18n import _
-from mailman.email.message import UserNotification
-from mailman.interfaces.bounce import IBounceDetector
+from mailman.email.message import OwnerNotification, UserNotification
+from mailman.interfaces.bounce import (
+    IBounceDetector, Stop, UnrecognizedBounceDisposition)
 from mailman.interfaces.listmanager import IListManager
 from mailman.interfaces.membership import ISubscriptionService
 from mailman.interfaces.pending import IPendable, IPendings
@@ -53,6 +55,7 @@ from mailman.utilities.string import oneline
 
 log = logging.getLogger('mailman.config')
 elog = logging.getLogger('mailman.error')
+blog = logging.getLogger('mailman.bounce')
 
 DOT = '.'
 
@@ -104,13 +107,15 @@ def scan_message(mlist, msg):
         set will be empty if no addresses were found.
     :rtype: set
     """
+    fatal_addresses = set()
     for detector_class in find_components('mailman.bouncers', IBounceDetector):
         addresses = detector_class().process(msg)
-        # Detectors may return None or an empty sequence to signify that no
-        # addresses have been found.
-        if addresses:
-            return set(addresses)
-    return set()
+        # Detectors may return Stop to signify that no fatal errors were
+        # found, or a sequence of addresses.
+        if addresses is Stop:
+            return Stop
+        fatal_addresses.update(addresses)
+    return fatal_addresses
 
 
 
@@ -245,3 +250,43 @@ def send_probe(member, msg):
     probe.attach(MIMEMessage(msg))
     probe.send(mlist, envsender=probe_sender, verp=False, probe_token=token)
     return token
+
+
+
+def maybe_forward(mlist, msg):
+    """Possibly forward bounce messages with no recognizable addresses.
+
+    :param mlist: The mailing list.
+    :type mlist: `IMailingList`
+    :param msg: The bounce message to scan.
+    :type msg: `Message`
+    """
+    message_id = msg['message-id']
+    if (mlist.forward_unrecognized_bounces_to
+        == UnrecognizedBounceDisposition.discard):
+        blog.error('Discarding unrecognized bounce: {0}'.format(message_id))
+        return
+    # The notification is either going to go to the list's administrators
+    # (owners and moderators), or to the site administrators.  Most of the
+    # notification is exactly the same in either case.
+    adminurl = mlist.script_url('admin') + '/bounce'
+    subject=_('Uncaught bounce notification')
+    text = MIMEText(
+        make('unrecognized.txt', mlist, adminurl=adminurl),
+        _charset=mlist.preferred_language.charset)
+    attachment = MIMEMessage(msg)
+    if (mlist.forward_unrecognized_bounces_to
+        == UnrecognizedBounceDisposition.administrators):
+        keywords = dict(roster=mlist.administrators)
+    elif (mlist.forward_unrecognized_bounces_to
+        == UnrecognizedBounceDisposition.site_owner):
+        keywords = {}
+    else:
+        raise AssertionError('Invalid forwarding disposition: {0}'.format(
+                             mlist.forward_unrecognized_bounces_to))
+    # Create the notification and send it.
+    notice = OwnerNotification(mlist, subject, **keywords)
+    notice.set_type('multipart/mixed')
+    notice.attach(text)
+    notice.attach(attachment)
+    notice.send(mlist)
diff --git a/src/mailman/app/membership.py b/src/mailman/app/membership.py
index be2382a7f..e7c2f40e3 100644
--- a/src/mailman/app/membership.py
+++ b/src/mailman/app/membership.py
@@ -154,5 +154,6 @@ def delete_member(mlist, address, admin_notif=None, userack=None):
                     listname=mlist.real_name,
                     member=formataddr((realname, address)),
                     )
-        msg = OwnerNotification(mlist, subject, text)
+        msg = OwnerNotification(mlist, subject, text,
+                                roster=mlist.administrators)
         msg.send(mlist)
diff --git a/src/mailman/app/notifications.py b/src/mailman/app/notifications.py
index 8bfbd0934..72413845c 100644
--- a/src/mailman/app/notifications.py
+++ b/src/mailman/app/notifications.py
@@ -131,5 +131,5 @@ def send_admin_subscription_notice(mlist, address, full_name, language):
                 listname=mlist.real_name,
                 member=formataddr((full_name, address)),
                 )
-    msg = OwnerNotification(mlist, subject, text)
+    msg = OwnerNotification(mlist, subject, text, roster=mlist.administrators)
     msg.send(mlist)
diff --git a/src/mailman/app/tests/test_bounces.py b/src/mailman/app/tests/test_bounces.py
index a79b1524c..954b5fc05 100644
--- a/src/mailman/app/tests/test_bounces.py
+++ b/src/mailman/app/tests/test_bounces.py
@@ -32,14 +32,18 @@ import unittest
 
 from zope.component import getUtility
 
-from mailman.app.bounces import StandardVERP, ProbeVERP, send_probe
+from mailman.app.bounces import (
+    ProbeVERP, StandardVERP, maybe_forward, send_probe)
 from mailman.app.lifecycle import create_list
 from mailman.app.membership import add_member
 from mailman.config import config
+from mailman.interfaces.bounce import UnrecognizedBounceDisposition
 from mailman.interfaces.languages import ILanguageManager
-from mailman.interfaces.member import DeliveryMode
+from mailman.interfaces.member import DeliveryMode, MemberRole
 from mailman.interfaces.pending import IPendings
+from mailman.interfaces.usermanager import IUserManager
 from mailman.testing.helpers import (
+    LogFileMark,
     get_queue_messages,
     specialized_message_from_string as message_from_string)
 from mailman.testing.layers import ConfigLayer
@@ -365,8 +369,139 @@ From: mail-daemon@example.com
 
 
 
+class TestMaybeForward(unittest.TestCase):
+    """Test forwarding of unrecognized bounces."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        config.push('test config', """
+        [mailman]
+        site_owner: postmaster@example.com
+        """)
+        self._mlist = create_list('test@example.com')
+        self._msg = message_from_string("""\
+From: bouncer@example.com
+To: test-bounces@example.com
+Subject: You bounced
+Message-ID: <first>
+
+""")
+
+    def tearDown(self):
+        config.pop('test config')
+
+    def test_maybe_forward_discard(self):
+        # When forward_unrecognized_bounces_to is set to discard, no bounce
+        # messages are forwarded.
+        self._mlist.forward_unrecognized_bounces_to = (
+            UnrecognizedBounceDisposition.discard)
+        # The only artifact of this call is a log file entry.
+        mark = LogFileMark('mailman.bounce')
+        maybe_forward(self._mlist, self._msg)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 0)
+        line = mark.readline()
+        self.assertEqual(
+            line[-40:-1],
+            'Discarding unrecognized bounce: <first>')
+
+    def test_maybe_forward_list_owner(self):
+        # Set up some owner and moderator addresses.
+        user_manager = getUtility(IUserManager)
+        anne = user_manager.create_address('anne@example.com')
+        bart = user_manager.create_address('bart@example.com')
+        cris = user_manager.create_address('cris@example.com')
+        dave = user_manager.create_address('dave@example.com')
+        # Regular members.
+        elle = user_manager.create_address('elle@example.com')
+        fred = user_manager.create_address('fred@example.com')
+        self._mlist.subscribe(anne, MemberRole.owner)
+        self._mlist.subscribe(bart, MemberRole.owner)
+        self._mlist.subscribe(cris, MemberRole.moderator)
+        self._mlist.subscribe(dave, MemberRole.moderator)
+        self._mlist.subscribe(elle, MemberRole.member)
+        self._mlist.subscribe(fred, MemberRole.member)
+        # When forward_unrecognized_bounces_to is set to owners, the
+        # bounce is forwarded to the list owners and moderators.
+        self._mlist.forward_unrecognized_bounces_to = (
+            UnrecognizedBounceDisposition.administrators)
+        maybe_forward(self._mlist, self._msg)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 1)
+        msg = items[0].msg
+        self.assertEqual(msg['subject'], 'Uncaught bounce notification')
+        self.assertEqual(msg['from'], 'postmaster@example.com')
+        self.assertEqual(msg['to'], 'test-owner@example.com')
+        # The first attachment is a notification message with a url.
+        payload = msg.get_payload(0)
+        self.assertEqual(payload.get_content_type(), 'text/plain')
+        body = payload.get_payload()
+        self.assertEqual(
+            body.splitlines()[-1],
+            'http://lists.example.com/admin/test@example.com/bounce')
+        # The second attachment should be a message/rfc822 containing the
+        # original bounce message.
+        payload = msg.get_payload(1)
+        self.assertEqual(payload.get_content_type(), 'message/rfc822')
+        bounce = payload.get_payload(0)
+        self.assertEqual(bounce.as_string(), self._msg.as_string())
+        # All of the owners and moderators, but none of the members, should be
+        # recipients of this message.
+        self.assertEqual(items[0].msgdata['recipients'],
+                         set(['anne@example.com', 'bart@example.com',
+                              'cris@example.com', 'dave@example.com']))
+
+    def test_maybe_forward_site_owner(self):
+        # Set up some owner and moderator addresses.
+        user_manager = getUtility(IUserManager)
+        anne = user_manager.create_address('anne@example.com')
+        bart = user_manager.create_address('bart@example.com')
+        cris = user_manager.create_address('cris@example.com')
+        dave = user_manager.create_address('dave@example.com')
+        # Regular members.
+        elle = user_manager.create_address('elle@example.com')
+        fred = user_manager.create_address('fred@example.com')
+        self._mlist.subscribe(anne, MemberRole.owner)
+        self._mlist.subscribe(bart, MemberRole.owner)
+        self._mlist.subscribe(cris, MemberRole.moderator)
+        self._mlist.subscribe(dave, MemberRole.moderator)
+        self._mlist.subscribe(elle, MemberRole.member)
+        self._mlist.subscribe(fred, MemberRole.member)
+        # When forward_unrecognized_bounces_to is set to owners, the
+        # bounce is forwarded to the list owners and moderators.
+        self._mlist.forward_unrecognized_bounces_to = (
+            UnrecognizedBounceDisposition.site_owner)
+        maybe_forward(self._mlist, self._msg)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 1)
+        msg = items[0].msg
+        self.assertEqual(msg['subject'], 'Uncaught bounce notification')
+        self.assertEqual(msg['from'], 'postmaster@example.com')
+        self.assertEqual(msg['to'], 'postmaster@example.com')
+        # The first attachment is a notification message with a url.
+        payload = msg.get_payload(0)
+        self.assertEqual(payload.get_content_type(), 'text/plain')
+        body = payload.get_payload()
+        self.assertEqual(
+            body.splitlines()[-1],
+            'http://lists.example.com/admin/test@example.com/bounce')
+        # The second attachment should be a message/rfc822 containing the
+        # original bounce message.
+        payload = msg.get_payload(1)
+        self.assertEqual(payload.get_content_type(), 'message/rfc822')
+        bounce = payload.get_payload(0)
+        self.assertEqual(bounce.as_string(), self._msg.as_string())
+        # All of the owners and moderators, but none of the members, should be
+        # recipients of this message.
+        self.assertEqual(items[0].msgdata['recipients'],
+                         set(['postmaster@example.com',]))
+
+
+
 def test_suite():
     suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(TestMaybeForward))
     suite.addTest(unittest.makeSuite(TestProbe))
     suite.addTest(unittest.makeSuite(TestSendProbe))
     suite.addTest(unittest.makeSuite(TestSendProbeNonEnglish))
diff --git a/src/mailman/bouncers/dsn.py b/src/mailman/bouncers/dsn.py
index 49803ef88..c9d7803ee 100644
--- a/src/mailman/bouncers/dsn.py
+++ b/src/mailman/bouncers/dsn.py
@@ -37,84 +37,66 @@ from mailman.interfaces.bounce import IBounceDetector, Stop
 
 
 
-def check(msg):
-    # Iterate over each message/delivery-status subpart.
-    failed_addresses = []
-    delayed_addresses = []
-    for part in typed_subpart_iterator(msg, 'message', 'delivery-status'):
-        if not part.is_multipart():
-            # Huh?
-            continue
-        # Each message/delivery-status contains a list of Message objects
-        # which are the header blocks.  Iterate over those too.
-        for msgblock in part.get_payload():
-            address_set = None
-            # We try to dig out the Original-Recipient (which is optional) and
-            # Final-Recipient (which is mandatory, but may not exactly match
-            # an address on our list).  Some MTA's also use X-Actual-Recipient
-            # as a synonym for Original-Recipient, but some apparently use
-            # that for other purposes :(
-            #
-            # Also grok out Action so we can do something with that too.
-            action = msgblock.get('action', '').lower()
-            # Some MTAs have been observed that put comments on the action.
-            if action.startswith('delayed'):
-                address_set = delayed_addresses
-            elif action.startswith('fail'):
-                address_set = failed_addresses
-            else:
-                # Some non-permanent failure, so ignore this block.
-                continue
-            params = []
-            foundp = False
-            for header in ('original-recipient', 'final-recipient'):
-                for k, v in msgblock.get_params([], header):
-                    if k.lower() == 'rfc822':
-                        foundp = True
-                    else:
-                        params.append(k)
-                if foundp:
-                    # Note that params should already be unquoted.
-                    address_set.extend(params)
-                    break
-                else:
-                    # MAS: This is a kludge, but SMTP-GATEWAY01.intra.home.dk
-                    # has a final-recipient with an angle-addr and no
-                    # address-type parameter at all. Non-compliant, but ...
-                    for param in params:
-                        if param.startswith('<') and param.endswith('>'):
-                            address_set.append(param[1:-1])
-    # There may be both delayed and failed addresses.  If there are any failed
-    # addresses, return those, otherwise just stop processing.
-    if len(failed_addresses) == 0:
-        if len(delayed_addresses) == 0:
-            return set()
-        else:
-            return Stop
-    return set(parseaddr(address)[1] for address in failed_addresses
-               if address is not None)
-
-
-
 class DSN:
     """Parse RFC 3464 (i.e. DSN) bounce formats."""
 
     implements(IBounceDetector)
 
     def process(self, msg):
-        return check(msg)
-        ## # A DSN has been seen wrapped with a "legal disclaimer" by an outgoing
-        ## # MTA in a multipart/mixed outer part.
-        ## if msg.is_multipart() and msg.get_content_subtype() == 'mixed':
-        ##     msg = msg.get_payload()[0]
-        ## # The above will suffice if the original message 'parts' were wrapped
-        ## # with the disclaimer added, but the original DSN can be wrapped as a
-        ## # message/rfc822 part.  We need to test that too.
-        ## if msg.is_multipart() and msg.get_content_type() == 'message/rfc822':
-        ##     msg = msg.get_payload()[0]
-        ## # The report-type parameter should be "delivery-status", but it seems
-        ## # that some DSN generating MTAs don't include this on the
-        ## # Content-Type: header, so let's relax the test a bit.
-        ## if not msg.is_multipart() or msg.get_content_subtype() <> 'report':
-        ##     return set()
-        ## return check(msg)
+        """See `IBounceDetector`."""
+        # Iterate over each message/delivery-status subpart.
+        failed_addresses = []
+        delayed_addresses = []
+        for part in typed_subpart_iterator(msg, 'message', 'delivery-status'):
+            if not part.is_multipart():
+                # Huh?
+                continue
+            # Each message/delivery-status contains a list of Message objects
+            # which are the header blocks.  Iterate over those too.
+            for msgblock in part.get_payload():
+                address_set = None
+                # We try to dig out the Original-Recipient (which is optional)
+                # and Final-Recipient (which is mandatory, but may not exactly
+                # match an address on our list).  Some MTA's also use
+                # X-Actual-Recipient as a synonym for Original-Recipient, but
+                # some apparently use that for other purposes :(
+                #
+                # Also grok out Action so we can do something with that too.
+                action = msgblock.get('action', '').lower()
+                # Some MTAs have been observed that put comments on the action.
+                if action.startswith('delayed'):
+                    address_set = delayed_addresses
+                elif action.startswith('fail'):
+                    address_set = failed_addresses
+                else:
+                    # Some non-permanent failure, so ignore this block.
+                    continue
+                params = []
+                foundp = False
+                for header in ('original-recipient', 'final-recipient'):
+                    for k, v in msgblock.get_params([], header):
+                        if k.lower() == 'rfc822':
+                            foundp = True
+                        else:
+                            params.append(k)
+                    if foundp:
+                        # Note that params should already be unquoted.
+                        address_set.extend(params)
+                        break
+                    else:
+                        # MAS: This is a kludge, but
+                        # SMTP-GATEWAY01.intra.home.dk has a final-recipient
+                        # with an angle-addr and no address-type parameter at
+                        # all. Non-compliant, but ...
+                        for param in params:
+                            if param.startswith('<') and param.endswith('>'):
+                                address_set.append(param[1:-1])
+        # There may be both delayed and failed addresses.  If there are any
+        # failed addresses, return those, otherwise just stop processing.
+        if len(failed_addresses) == 0:
+            if len(delayed_addresses) == 0:
+                return set()
+            else:
+                return Stop
+        return set(parseaddr(address)[1] for address in failed_addresses
+                   if address is not None)
diff --git a/src/mailman/config/configure.zcml b/src/mailman/config/configure.zcml
index 3b4497ab8..299a0ce67 100644
--- a/src/mailman/config/configure.zcml
+++ b/src/mailman/config/configure.zcml
@@ -22,6 +22,11 @@
     />
 
   <utility
+    factory="mailman.model.bounce.BounceProcessor"
+    provides="mailman.interfaces.bounce.IBounceProcessor"
+    />
+
+  <utility
     factory="mailman.model.domain.DomainManager"
     provides="mailman.interfaces.domain.IDomainManager"
     />
@@ -37,11 +42,6 @@
     />
 
   <utility
-    factory="mailman.model.usermanager.UserManager"
-    provides="mailman.interfaces.usermanager.IUserManager"
-    />
-
-  <utility
     factory="mailman.model.messagestore.MessageStore"
     provides="mailman.interfaces.messages.IMessageStore"
     />
@@ -52,13 +52,13 @@
     />
 
   <utility
-    factory="mailman.model.requests.Requests"
-    provides="mailman.interfaces.requests.IRequests"
+    factory="mailman.app.registrar.Registrar"
+    provides="mailman.interfaces.registrar.IRegistrar"
     />
 
   <utility
-    factory="mailman.app.registrar.Registrar"
-    provides="mailman.interfaces.registrar.IRegistrar"
+    factory="mailman.model.requests.Requests"
+    provides="mailman.interfaces.requests.IRequests"
     />
 
   <utility
@@ -67,6 +67,11 @@
     />
 
   <utility
+    factory="mailman.model.usermanager.UserManager"
+    provides="mailman.interfaces.usermanager.IUserManager"
+    />
+
+  <utility
     factory="mailman.email.validate.Validator"
     provides="mailman.interfaces.address.IEmailValidator"
     />
diff --git a/src/mailman/core/logging.py b/src/mailman/core/logging.py
index e15924177..0ff129897 100644
--- a/src/mailman/core/logging.py
+++ b/src/mailman/core/logging.py
@@ -57,11 +57,11 @@ class ReopenableFileHandler(logging.Handler):
     def __init__(self, name, filename):
         logging.Handler.__init__(self)
         self.name = name
-        self._filename = filename
+        self.filename = filename
         self._stream = self._open()
 
     def _open(self):
-        return codecs.open(self._filename, 'a', 'utf-8')
+        return codecs.open(self.filename, 'a', 'utf-8')
 
     def flush(self):
         if self._stream:
@@ -98,7 +98,7 @@ class ReopenableFileHandler(logging.Handler):
         :type filename: string
         """
         if filename is not None:
-            self._filename = filename
+            self.filename = filename
         self._stream.close()
         self._stream = self._open()
 
diff --git a/src/mailman/database/mailman.sql b/src/mailman/database/mailman.sql
index 5d07c607a..7493f5d34 100644
--- a/src/mailman/database/mailman.sql
+++ b/src/mailman/database/mailman.sql
@@ -54,6 +54,17 @@ CREATE INDEX ix_autoresponserecord_address_id
 CREATE INDEX ix_autoresponserecord_mailing_list_id
     ON autoresponserecord (mailing_list_id);
 
+CREATE TABLE bounceevent (
+    id INTEGER NOT NULL,
+    list_name TEXT,
+    email TEXT,
+    'timestamp' TIMESTAMP,
+    message_id TEXT,
+    context TEXT,
+    processed BOOLEAN,
+    PRIMARY KEY (id)
+    );
+
 CREATE TABLE contentfilter (
     id INTEGER NOT NULL,
     mailing_list_id INTEGER,
@@ -116,13 +127,13 @@ CREATE TABLE mailinglist (
     autoresponse_request_text TEXT,
     autoresponse_grace_period TEXT,
     -- Bounces.
+    forward_unrecognized_bounces_to TEXT,
+    process_bounces BOOLEAN,
     bounce_info_stale_after TEXT,
     bounce_matching_headers TEXT,
     bounce_notify_owner_on_disable BOOLEAN,
     bounce_notify_owner_on_removal BOOLEAN,
-    bounce_processing BOOLEAN,
     bounce_score_threshold INTEGER,
-    bounce_unrecognized_goes_to_list_owner BOOLEAN,
     bounce_you_are_disabled_warnings INTEGER,
     bounce_you_are_disabled_warnings_interval TEXT,
     -- Content filtering.
diff --git a/src/mailman/database/types.py b/src/mailman/database/types.py
index 1195802ff..f126cc05a 100644
--- a/src/mailman/database/types.py
+++ b/src/mailman/database/types.py
@@ -34,7 +34,10 @@ from mailman.utilities.modules import find_name
 
 
 class _EnumVariable(Variable):
-    """Storm variable."""
+    """Storm variable.
+
+    To use this, make the database column a TEXT.
+    """
 
     def parse_set(self, value, from_db):
         if value is None:
diff --git a/src/mailman/email/message.py b/src/mailman/email/message.py
index 4eb049f17..546474fbe 100644
--- a/src/mailman/email/message.py
+++ b/src/mailman/email/message.py
@@ -162,7 +162,7 @@ class Message(email.message.Message):
 class UserNotification(Message):
     """Class for internally crafted messages."""
 
-    def __init__(self, recip, sender, subject=None, text=None, lang=None):
+    def __init__(self, recipients, sender, subject=None, text=None, lang=None):
         Message.__init__(self)
         charset = (lang.charset if lang is not None else 'us-ascii')
         subject = ('(no subject)' if subject is None else subject)
@@ -171,12 +171,12 @@ class UserNotification(Message):
         self['Subject'] = Header(subject.encode(charset), charset,
                                  header_name='Subject', errors='replace')
         self['From'] = sender
-        if isinstance(recip, list):
-            self['To'] = COMMASPACE.join(recip)
-            self.recips = recip
+        if isinstance(recipients, (list, set, tuple)):
+            self['To'] = COMMASPACE.join(recipients)
+            self.recipients = recipients
         else:
-            self['To'] = recip
-            self.recips = [recip]
+            self['To'] = recipients
+            self.recipients = set([recipients])
 
     def send(self, mlist, **_kws):
         """Sends the message by enqueuing it to the 'virgin' queue.
@@ -202,7 +202,7 @@ class UserNotification(Message):
         virginq = config.switchboards['virgin']
         # The message metadata better have a 'recip' attribute.
         enqueue_kws = dict(
-            recipients=self.recips,
+            recipients=self.recipients,
             nodecorate=True,
             reduced_list_headers=True,
             )
@@ -218,20 +218,21 @@ class UserNotification(Message):
 
 
 class OwnerNotification(UserNotification):
-    """Like user notifications, but this message goes to the list owners."""
+    """Like user notifications, but this message goes to some owners."""
 
-    def __init__(self, mlist, subject=None, text=None, tomoderators=True):
-        if tomoderators:
-            roster = mlist.moderators
+    def __init__(self, mlist, subject=None, text=None, roster=None):
+        if roster is None:
+            recipients = set([config.mailman.site_owner])
+            to = config.mailman.site_owner
         else:
-            roster = mlist.owners
-        recips = [address.address for address in roster.addresses]
+            recipients = set(address.email for address in roster.addresses)
+            to = mlist.owner_address
         sender = config.mailman.site_owner
-        UserNotification.__init__(self, recips, sender, subject,
+        UserNotification.__init__(self, recipients, sender, subject,
                                   text, mlist.preferred_language)
         # Hack the To header to look like it's going to the -owner address
         del self['to']
-        self['To'] = mlist.owner_address
+        self['To'] = to
         self._sender = sender
 
     def _enqueue(self, mlist, **_kws):
@@ -240,7 +241,7 @@ class OwnerNotification(UserNotification):
         # The message metadata better have a `recip' attribute
         virginq.enqueue(self,
                         listname=mlist.fqdn_listname,
-                        recipients=self.recips,
+                        recipients=self.recipients,
                         nodecorate=True,
                         reduced_list_headers=True,
                         envsender=self._sender,
diff --git a/src/mailman/interfaces/bounce.py b/src/mailman/interfaces/bounce.py
index 22e2467b8..99f6b50b7 100644
--- a/src/mailman/interfaces/bounce.py
+++ b/src/mailman/interfaces/bounce.py
@@ -21,13 +21,17 @@ from __future__ import absolute_import, unicode_literals
 
 __metaclass__ = type
 __all__ = [
+    'BounceContext',
     'IBounceDetector',
+    'IBounceEvent',
+    'IBounceProcessor',
     'Stop',
+    'UnrecognizedBounceDisposition',
     ]
 
 
 from flufl.enum import Enum
-from zope.interface import Interface
+from zope.interface import Attribute, Interface
 
 
 
@@ -39,6 +43,30 @@ Stop = object()
 
 
 
+class BounceContext(Enum):
+    """The context in which the bounce was detected."""
+
+    # This is a normal bounce detection.  IOW, Mailman received a bounce in
+    # response to a mailing list post.
+    normal = 1
+
+    # A probe email bounced.  This can be considered a bit more serious, since
+    # it occurred in response to a specific message to a specific user.
+    probe = 2
+
+
+
+class UnrecognizedBounceDisposition(Enum):
+    # Just throw the message away.
+    discard = 0
+    # Forward the message to the list administrators, which includes both the
+    # owners and the moderators.
+    administrators = 1
+    # Forward the message to the site owner.
+    site_owner = 2
+
+
+
 class IBounceDetector(Interface):
     """Detect a bounce in an email message."""
 
@@ -52,3 +80,54 @@ class IBounceDetector(Interface):
             returned to halt any bounce processing pipeline.
         :rtype: A set strings, or `Stop`
         """
+
+
+
+class IBounceEvent(Interface):
+    """Registration record for a single bounce event."""
+
+    list_name = Attribute(
+        """The name of the mailing list that received this bounce.""")
+
+    email = Attribute(
+        """The email address that bounced.""")
+
+    timestamp = Attribute(
+        """The timestamp for when the bounce was received.""")
+
+    message_id = Attribute(
+        """The Message-ID of the bounce message.""")
+
+    context = Attribute(
+        """Where was the bounce detected?""")
+
+    processed = Attribute(
+        """Has this bounce event been processed?""")
+
+
+
+class IBounceProcessor(Interface):
+    """Manager/processor of bounce events."""
+
+    def register(mlist, email, msg, context=None):
+        """Register a bounce event.
+
+        :param mlist: The mailing list that the bounce occurred on.
+        :type mlist: IMailingList
+        :param email: The email address that is bouncing.
+        :type email: str
+        :param msg: The bounce message.
+        :type msg: email.message.Message
+        :param context: In what context was the bounce detected?  The default
+            is 'normal' context (i.e. we received a normal bounce for the
+            address).
+        :type context: BounceContext
+        :return: The registered bounce event.
+        :rtype: IBounceEvent
+        """
+
+    events = Attribute(
+        """An iterator over all events.""")
+
+    unprocessed = Attribute(
+        """An iterator over all unprocessed bounce events.""")
diff --git a/src/mailman/interfaces/mailinglist.py b/src/mailman/interfaces/mailinglist.py
index 23d21cd34..44e015435 100644
--- a/src/mailman/interfaces/mailinglist.py
+++ b/src/mailman/interfaces/mailinglist.py
@@ -495,6 +495,19 @@ class IMailingList(Interface):
         without any other checks.
         """)
 
+    # Bounces.
+
+    forward_unrecognized_bounces_to = Attribute(
+        """What to do when a bounce contains no recognizable addresses.
+
+        This is an enumeration which specifies what to do with such bounce
+        messages.  They can be discarded, forward to the list owner, or
+        forwarded to the site owner.
+        """)
+
+    process_bounces = Attribute(
+        """Whether or not the mailing list processes bounces.""")
+
 
 
 class IAcceptableAlias(Interface):
diff --git a/src/mailman/model/bounce.py b/src/mailman/model/bounce.py
new file mode 100644
index 000000000..20953b0ff
--- /dev/null
+++ b/src/mailman/model/bounce.py
@@ -0,0 +1,82 @@
+# Copyright (C) 2011 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Bounce support."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'BounceEvent',
+    'BounceProcessor',
+    ]
+
+
+from storm.locals import Bool, Int, DateTime, Unicode
+from zope.interface import implements
+
+from mailman.config import config
+from mailman.database.model import Model
+from mailman.database.types import Enum
+from mailman.interfaces.bounce import (
+    BounceContext, IBounceEvent, IBounceProcessor)
+from mailman.utilities.datetime import now
+
+
+
+class BounceEvent(Model):
+    implements(IBounceEvent)
+
+    id = Int(primary=True)
+    list_name = Unicode()
+    email = Unicode()
+    timestamp = DateTime()
+    message_id = Unicode()
+    context = Enum()
+    processed = Bool()
+
+    def __init__(self, list_name, email, msg, context=None):
+        self.list_name = list_name
+        self.email = email
+        self.timestamp = now()
+        self.message_id = msg['message-id']
+        self.context = (BounceContext.normal if context is None else context)
+        self.processed = False
+
+
+
+class BounceProcessor:
+    implements(IBounceProcessor)
+
+    def register(self, mlist, email, msg, where=None):
+        """See `IBounceProcessor`."""
+        event = BounceEvent(mlist.fqdn_listname, email, msg, where)
+        config.db.store.add(event)
+        return event
+
+    @property
+    def events(self):
+        """See `IBounceProcessor`."""
+        for event in config.db.store.find(BounceEvent):
+            yield event
+
+    @property
+    def unprocessed(self):
+        """See `IBounceProcessor`."""
+        for event in config.db.store.find(BounceEvent,
+                                          BounceEvent.processed == False):
+            yield event
diff --git a/src/mailman/model/docs/bounce.rst b/src/mailman/model/docs/bounce.rst
new file mode 100644
index 000000000..41784cd9c
--- /dev/null
+++ b/src/mailman/model/docs/bounce.rst
@@ -0,0 +1,84 @@
+=======
+Bounces
+=======
+
+When a message to an email address bounces, Mailman's bounce runner will
+register a bounce event.  This registration is done through a utility.
+
+    >>> from zope.component import getUtility
+    >>> from zope.interface.verify import verifyObject
+    >>> from mailman.interfaces.bounce import IBounceProcessor
+    >>> processor = getUtility(IBounceProcessor)
+    >>> verifyObject(IBounceProcessor, processor)
+    True
+
+
+Registration
+============
+
+When a bounce occurs, it's always within the context of a specific mailing
+list.
+
+    >>> mlist = create_list('test@example.com')
+
+The bouncing email contains useful information that will be registered as
+well.  In particular, the Message-ID is a key piece of data that needs to be
+recorded.
+
+    >>> msg = message_from_string("""\
+    ... From: mail-daemon@example.org
+    ... To: test-bounces@example.com
+    ... Message-ID: <first>
+    ...
+    ... """)
+
+There is a suite of bounce detectors that are used to heuristically extract
+the bouncing email addresses.  Various techniques are employed including VERP,
+DSN, and magic.  It is the bounce queue's responsibility to extract the set of
+bouncing email addrsses.  These are passed one-by-one to the registration
+interface.
+
+    >>> event = processor.register(mlist, 'anne@example.com', msg)
+    >>> print event.list_name
+    test@example.com
+    >>> print event.email
+    anne@example.com
+    >>> print event.message_id
+    <first>
+
+Bounce events have a timestamp.
+
+    >>> print event.timestamp
+    2005-08-01 07:49:23
+
+Bounce events have a flag indicating whether they've been processed or not.
+
+    >>> event.processed
+    False
+
+When a bounce is registered, you can indicate the bounce context.
+
+    >>> msg = message_from_string("""\
+    ... From: mail-daemon@example.org
+    ... To: test-bounces@example.com
+    ... Message-ID: <second>
+    ...
+    ... """)
+
+If no context is given, then a default one is used.
+
+    >>> event = processor.register(mlist, 'bart@example.com', msg)
+    >>> print event.message_id
+    <second>
+    >>> print event.context
+    BounceContext.normal
+
+A probe bounce carries more weight than just a normal bounce.
+
+    >>> from mailman.interfaces.bounce import BounceContext
+    >>> event = processor.register(
+    ...     mlist, 'bart@example.com', msg, BounceContext.probe)
+    >>> print event.message_id
+    <second>
+    >>> print event.context
+    BounceContext.probe
diff --git a/src/mailman/model/docs/registration.txt b/src/mailman/model/docs/registration.txt
index d0827d37b..0e80bfa14 100644
--- a/src/mailman/model/docs/registration.txt
+++ b/src/mailman/model/docs/registration.txt
@@ -149,7 +149,7 @@ message is sent to the user in order to verify the registered address.
     _parsemsg           : False
     listname            : alpha@example.com
     nodecorate          : True
-    recipients          : [u'aperson@example.com']
+    recipients          : set([u'aperson@example.com'])
     reduced_list_headers: True
     version             : 3
 
diff --git a/src/mailman/model/docs/requests.txt b/src/mailman/model/docs/requests.txt
index bebb61259..812d25a43 100644
--- a/src/mailman/model/docs/requests.txt
+++ b/src/mailman/model/docs/requests.txt
@@ -302,7 +302,7 @@ The message can be rejected, meaning it is bounced back to the sender.
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'aperson@example.org']
+    recipients          : set([u'aperson@example.org'])
     reduced_list_headers: True
     version             : 3
 
@@ -479,7 +479,7 @@ queue when the message is held.
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'alist-owner@example.com']
+    recipients          : set([u'alist-owner@example.com'])
     reduced_list_headers: True
     tomoderators        : True
     version             : 3
@@ -534,7 +534,7 @@ subscriber.
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'cperson@example.org']
+    recipients          : set([u'cperson@example.org'])
     reduced_list_headers: True
     version             : 3
 
@@ -578,7 +578,7 @@ subscription and the fact that they may need to approve it.
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'alist-owner@example.com']
+    recipients          : set([u'alist-owner@example.com'])
     reduced_list_headers: True
     tomoderators        : True
     version             : 3
@@ -651,7 +651,7 @@ The welcome message is sent to the person who just subscribed.
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'fperson@example.org']
+    recipients          : set([u'fperson@example.org'])
     reduced_list_headers: True
     verp                : False
     version             : 3
@@ -677,7 +677,7 @@ The admin message is sent to the moderators.
     envsender           : changeme@example.com
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : []
+    recipients          : set([])
     reduced_list_headers: True
     version             : 3
 
@@ -759,7 +759,7 @@ notification.
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'alist-owner@example.com']
+    recipients          : set([u'alist-owner@example.com'])
     reduced_list_headers: True
     tomoderators        : True
     version             : 3
@@ -818,7 +818,7 @@ and the person remains a member of the mailing list.
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'hperson@example.com']
+    recipients          : set([u'hperson@example.com'])
     reduced_list_headers: True
     version             : 3
 
@@ -873,7 +873,7 @@ The goodbye message...
     _parsemsg           : False
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : [u'gperson@example.com']
+    recipients          : set([u'gperson@example.com'])
     reduced_list_headers: True
     verp                : False
     version             : 3
@@ -898,6 +898,6 @@ The goodbye message...
     envsender           : changeme@example.com
     listname            : alist@example.com
     nodecorate          : True
-    recipients          : []
+    recipients          : set([])
     reduced_list_headers: True
     version             : 3
diff --git a/src/mailman/model/mailinglist.py b/src/mailman/model/mailinglist.py
index 9294fe7cc..2d972a24a 100644
--- a/src/mailman/model/mailinglist.py
+++ b/src/mailman/model/mailinglist.py
@@ -118,11 +118,12 @@ class MailingList(Model):
     bounce_matching_headers = Unicode() # XXX
     bounce_notify_owner_on_disable = Bool() # XXX
     bounce_notify_owner_on_removal = Bool() # XXX
-    bounce_processing = Bool() # XXX
     bounce_score_threshold = Int() # XXX
-    bounce_unrecognized_goes_to_list_owner = Bool() # XXX
     bounce_you_are_disabled_warnings = Int() # XXX
     bounce_you_are_disabled_warnings_interval = TimeDelta() # XXX
+    forward_unrecognized_bounces_to = Enum()
+    process_bounces = Bool()
+    # Miscellaneous
     default_member_action = Enum()
     default_nonmember_action = Enum()
     description = Unicode()
@@ -195,7 +196,7 @@ class MailingList(Model):
         # For the pending database
         self.next_request_id = 1
         self._restore()
-        self.personalization = Personalization.none
+        self.personalize = Personalization.none
         self.real_name = string.capwords(
             SPACE.join(listname.split(UNDERSCORE)))
         makedirs(self.data_path)
diff --git a/src/mailman/model/tests/test_bounce.py b/src/mailman/model/tests/test_bounce.py
new file mode 100644
index 000000000..a232b37fd
--- /dev/null
+++ b/src/mailman/model/tests/test_bounce.py
@@ -0,0 +1,105 @@
+# Copyright (C) 2011 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test bounce model objects."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'test_suite',
+    ]
+
+
+import unittest
+
+from datetime import datetime
+from zope.component import getUtility
+
+from mailman.app.lifecycle import create_list
+from mailman.config import config
+from mailman.interfaces.bounce import BounceContext, IBounceProcessor
+from mailman.testing.helpers import (
+    specialized_message_from_string as message_from_string)
+from mailman.testing.layers import ConfigLayer
+
+
+
+class TestBounceEvents(unittest.TestCase):
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._processor = getUtility(IBounceProcessor)
+        self._mlist = create_list('test@example.com')
+        self._msg = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces@example.com
+Message-Id: <first>
+
+""")
+
+    def test_events_iterator(self):
+        self._processor.register(self._mlist, 'anne@example.com', self._msg)
+        config.db.commit()
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 1)
+        event = events[0]
+        self.assertEqual(event.list_name, 'test@example.com')
+        self.assertEqual(event.email, 'anne@example.com')
+        self.assertEqual(event.timestamp, datetime(2005, 8, 1, 7, 49, 23))
+        self.assertEqual(event.message_id, '<first>')
+        self.assertEqual(event.context, BounceContext.normal)
+        self.assertEqual(event.processed, False)
+        # The unprocessed list will be exactly the same right now.
+        unprocessed = list(self._processor.unprocessed)
+        self.assertEqual(len(unprocessed), 1)
+        event = unprocessed[0]
+        self.assertEqual(event.list_name, 'test@example.com')
+        self.assertEqual(event.email, 'anne@example.com')
+        self.assertEqual(event.timestamp, datetime(2005, 8, 1, 7, 49, 23))
+        self.assertEqual(event.message_id, '<first>')
+        self.assertEqual(event.context, BounceContext.normal)
+        self.assertEqual(event.processed, False)
+
+    def test_unprocessed_events_iterator(self):
+        self._processor.register(self._mlist, 'anne@example.com', self._msg)
+        self._processor.register(self._mlist, 'bart@example.com', self._msg)
+        config.db.commit()
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 2)
+        unprocessed = list(self._processor.unprocessed)
+        # The unprocessed list will be exactly the same right now.
+        self.assertEqual(len(unprocessed), 2)
+        # Process one of the events.
+        events[0].processed = True
+        config.db.commit()
+        # Now there will be only one unprocessed event.
+        unprocessed = list(self._processor.unprocessed)
+        self.assertEqual(len(unprocessed), 1)
+        # Process the other event.
+        events[1].processed = True
+        config.db.commit()
+        # Now there will be no unprocessed events.
+        unprocessed = list(self._processor.unprocessed)
+        self.assertEqual(len(unprocessed), 0)
+
+
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(TestBounceEvents))
+    return suite
diff --git a/src/mailman/pipeline/docs/acknowledge.txt b/src/mailman/pipeline/docs/acknowledge.txt
index b2ce11b7f..8c8552190 100644
--- a/src/mailman/pipeline/docs/acknowledge.txt
+++ b/src/mailman/pipeline/docs/acknowledge.txt
@@ -7,7 +7,7 @@ receive acknowledgments of their postings, Mailman will sent them such an
 acknowledgment.
 ::
 
-    >>> mlist = create_list('_xtest@example.com')
+    >>> mlist = create_list('test@example.com')
     >>> mlist.real_name = 'XTest'
     >>> mlist.preferred_language = 'en'
     >>> # XXX This will almost certainly change once we've worked out the web
@@ -30,7 +30,7 @@ Subscribe a user to the mailing list.
     >>> user_1 = user_manager.create_user('aperson@example.com')
     >>> address_1 = list(user_1.addresses)[0]
     >>> mlist.subscribe(address_1, MemberRole.member)
-    <Member: aperson@example.com on _xtest@example.com as MemberRole.member>
+    <Member: aperson@example.com on test@example.com as MemberRole.member>
 
 
 Non-member posts
@@ -83,7 +83,7 @@ will be sent.
     >>> user_2 = user_manager.create_user('dperson@example.com')
     >>> address_2 = list(user_2.addresses)[0]
     >>> mlist.subscribe(address_2, MemberRole.member)
-    <Member: dperson@example.com on _xtest@example.com as MemberRole.member>
+    <Member: dperson@example.com on test@example.com as MemberRole.member>
 
     >>> handler.process(mlist, msg,
     ...     dict(original_sender='dperson@example.com'))
@@ -112,14 +112,19 @@ The receipt will include the original message's subject in the response body,
     >>> qmsg, qdata = virginq.dequeue(virginq.files[0])
     >>> virginq.files
     []
-    >>> sorted(qdata.items())
-    [..., ('recipients', [u'aperson@example.com']), ...]
+    >>> dump_msgdata(qdata)
+    _parsemsg           : False
+    listname            : test@example.com
+    nodecorate          : True
+    recipients          : set([u'aperson@example.com'])
+    reduced_list_headers: True
+    ...
     >>> print qmsg.as_string()
     ...
     MIME-Version: 1.0
     ...
     Subject: XTest post acknowledgment
-    From: _xtest-bounces@example.com
+    From: test-bounces@example.com
     To: aperson@example.com
     ...
     Precedence: bulk
@@ -130,7 +135,7 @@ The receipt will include the original message's subject in the response body,
     <BLANKLINE>
     was successfully received by the XTest mailing list.
     <BLANKLINE>
-    List info page: http://lists.example.com/listinfo/_xtest@example.com
+    List info page: http://lists.example.com/listinfo/test@example.com
     Your preferences: http://example.com/aperson@example.com
     <BLANKLINE>
 
@@ -146,13 +151,18 @@ If there is no subject, then the receipt will use a generic message.
     >>> qmsg, qdata = virginq.dequeue(virginq.files[0])
     >>> virginq.files
     []
-    >>> sorted(qdata.items())
-    [..., ('recipients', [u'aperson@example.com']), ...]
+    >>> dump_msgdata(qdata)
+    _parsemsg           : False
+    listname            : test@example.com
+    nodecorate          : True
+    recipients          : set([u'aperson@example.com'])
+    reduced_list_headers: True
+    ...
     >>> print qmsg.as_string()
     MIME-Version: 1.0
     ...
     Subject: XTest post acknowledgment
-    From: _xtest-bounces@example.com
+    From: test-bounces@example.com
     To: aperson@example.com
     ...
     Precedence: bulk
@@ -163,6 +173,6 @@ If there is no subject, then the receipt will use a generic message.
     <BLANKLINE>
     was successfully received by the XTest mailing list.
     <BLANKLINE>
-    List info page: http://lists.example.com/listinfo/_xtest@example.com
+    List info page: http://lists.example.com/listinfo/test@example.com
     Your preferences: http://example.com/aperson@example.com
     <BLANKLINE>
diff --git a/src/mailman/pipeline/docs/replybot.txt b/src/mailman/pipeline/docs/replybot.txt
index 3a6d75499..208f6aae9 100644
--- a/src/mailman/pipeline/docs/replybot.txt
+++ b/src/mailman/pipeline/docs/replybot.txt
@@ -51,7 +51,7 @@ response.
     _parsemsg           : False
     listname            : _xtest@example.com
     nodecorate          : True
-    recipients          : [u'aperson@example.com']
+    recipients          : set([u'aperson@example.com'])
     reduced_list_headers: True
     version             : 3
 
@@ -143,7 +143,7 @@ Unless the ``X-Ack:`` header has a value of ``yes``, in which case, the
     _parsemsg           : False
     listname            : _xtest@example.com
     nodecorate          : True
-    recipients          : [u'asystem@example.com']
+    recipients          : set([u'asystem@example.com'])
     reduced_list_headers: True
     version             : 3
 
diff --git a/src/mailman/queue/bounce.py b/src/mailman/queue/bounce.py
index fb8b0124a..a714f2669 100644
--- a/src/mailman/queue/bounce.py
+++ b/src/mailman/queue/bounce.py
@@ -17,17 +17,13 @@
 
 """Bounce queue runner."""
 
-import os
-import cPickle
 import logging
-import datetime
 
-from lazr.config import as_timedelta
+from zope.component import getUtility
 
-from mailman.app.bounces import StandardVERP
-from mailman.config import config
-from mailman.core.i18n import _
-from mailman.interfaces.bounce import Stop
+from mailman.app.bounces import (
+    ProbeVERP, StandardVERP, maybe_forward, scan_message)
+from mailman.interfaces.bounce import BounceContext, IBounceProcessor, Stop
 from mailman.queue import Runner
 
 
@@ -37,214 +33,48 @@ log = logging.getLogger('mailman.bounce')
 elog = logging.getLogger('mailman.error')
 
 
-
-class BounceMixin:
-    def __init__(self):
-        # Registering a bounce means acquiring the list lock, and it would be
-        # too expensive to do this for each message.  Instead, each bounce
-        # runner maintains an event log which is essentially a file with
-        # multiple pickles.  Each bounce we receive gets appended to this file
-        # as a 4-tuple record: (listname, addr, today, msg)
-        #
-        # today is itself a 3-tuple of (year, month, day)
-        #
-        # Every once in a while (see _do_periodic()), the bounce runner cracks
-        # open the file, reads all the records and registers all the bounces.
-        # Then it truncates the file and continues on.  We don't need to lock
-        # the bounce event file because bounce qrunners are single threaded
-        # and each creates a uniquely named file to contain the events.
-        #
-        # XXX When Python 2.3 is minimal require, we can use the new
-        # tempfile.TemporaryFile() function.
-        #
-        # XXX We used to classify bounces to the site list as bounce events
-        # for every list, but this caused severe problems.  Here's the
-        # scenario: aperson@example.com is a member of 4 lists, and a list
-        # owner of the foo list.  example.com has an aggressive spam filter
-        # which rejects any message that is spam or contains spam as an
-        # attachment.  Now, a spambot sends a piece of spam to the foo list,
-        # but since that spambot is not a member, the list holds the message
-        # for approval, and sends a notification to aperson@example.com as
-        # list owner.  That notification contains a copy of the spam.  Now
-        # example.com rejects the message, causing a bounce to be sent to the
-        # site list's bounce address.  The bounce runner would then dutifully
-        # register a bounce for all 4 lists that aperson@example.com was a
-        # member of, and eventually that person would get disabled on all
-        # their lists.  So now we ignore site list bounces.  Ce La Vie for
-        # password reminder bounces.
-        self._bounce_events_file = os.path.join(
-            config.DATA_DIR, 'bounce-events-%05d.pck' % os.getpid())
-        self._bounce_events_fp = None
-        self._bouncecnt = 0
-        self._nextaction = (
-            datetime.datetime.now() +
-            as_timedelta(config.bounces.register_bounces_every))
 
-    def _queue_bounces(self, listname, addrs, msg):
-        today = datetime.date.today()
-        if self._bounce_events_fp is None:
-            self._bounce_events_fp = open(self._bounce_events_file, 'a+b')
-        for addr in addrs:
-            cPickle.dump((listname, addr, today, msg),
-                         self._bounce_events_fp, 1)
-        self._bounce_events_fp.flush()
-        os.fsync(self._bounce_events_fp.fileno())
-        self._bouncecnt += len(addrs)
-
-    def _register_bounces(self):
-        log.info('%s processing %s queued bounces', self, self._bouncecnt)
-        # Read all the records from the bounce file, then unlink it.  Sort the
-        # records by listname for more efficient processing.
-        events = {}
-        self._bounce_events_fp.seek(0)
-        while True:
-            try:
-                listname, addr, day, msg = cPickle.load(self._bounce_events_fp)
-            except ValueError, e:
-                log.error('Error reading bounce events: %s', e)
-            except EOFError:
-                break
-            events.setdefault(listname, []).append((addr, day, msg))
-        # Now register all events sorted by list
-        for listname in events.keys():
-            mlist = self._open_list(listname)
-            mlist.Lock()
-            try:
-                for addr, day, msg in events[listname]:
-                    mlist.registerBounce(addr, msg, day=day)
-                mlist.Save()
-            finally:
-                mlist.Unlock()
-        # Reset and free all the cached memory
-        self._bounce_events_fp.close()
-        self._bounce_events_fp = None
-        os.unlink(self._bounce_events_file)
-        self._bouncecnt = 0
-
-    def _clean_up(self):
-        if self._bouncecnt > 0:
-            self._register_bounces()
-
-    def _do_periodic(self):
-        now = datetime.datetime.now()
-        if self._nextaction > now or self._bouncecnt == 0:
-            return
-        # Let's go ahead and register the bounces we've got stored up
-        self._nextaction = now + as_timedelta(
-            config.bounces.register_bounces_every)
-        self._register_bounces()
-
-    def _probe_bounce(self, mlist, token):
-        locked = mlist.Locked()
-        if not locked:
-            mlist.Lock()
-        try:
-            op, addr, bmsg = mlist.pend_confirm(token)
-            info = mlist.getBounceInfo(addr)
-            mlist.disableBouncingMember(addr, info, bmsg)
-            # Only save the list if we're unlocking it
-            if not locked:
-                mlist.Save()
-        finally:
-            if not locked:
-                mlist.Unlock()
-
-
-
-class BounceRunner(Runner, BounceMixin):
+class BounceRunner(Runner):
     """The bounce runner."""
 
-    def __init__(self, slice=None, numslices=1):
-        Runner.__init__(self, slice, numslices)
-        BounceMixin.__init__(self)
+    def __init__(self, name, slice=None):
+        super(BounceRunner, self).__init__(name, slice)
+        self._processor = getUtility(IBounceProcessor)
 
     def _dispose(self, mlist, msg, msgdata):
-        # Make sure we have the most up-to-date state
-        mlist.Load()
-        # There are a few possibilities here:
-        #
-        # - the message could have been VERP'd in which case, we know exactly
-        #   who the message was destined for.  That make our job easy.
-        # - the message could have been originally destined for a list owner,
-        #   but a list owner address itself bounced.  That's bad, and for now
-        #   we'll simply log the problem and attempt to deliver the message to
-        #   the site owner.
-        #
-        # All messages sent to list owners have their sender set to the site
-        # owner address.  That way, if a list owner address bounces, at least
-        # some human has a chance to deal with it.  Is this a bounce for a
-        # message to a list owner, coming to the site owner?
-        if msg.get('to', '') == config.mailman.site_owner:
-            # Send it on to the site owners, but craft the envelope sender to
-            # be the noreply address, so if the site owner bounce, we won't
-            # get stuck in a bounce loop.
-            config.switchboards['out'].enqueue(
-                msg, msgdata,
-                recipients=[config.mailman.site_owner],
-                envsender=config.mailman.noreply_address,
-                )
         # List isn't doing bounce processing?
         if not mlist.bounce_processing:
-            return
+            return False
         # Try VERP detection first, since it's quick and easy
-        addrs = StandardVERP().get_verp(mlist, msg)
-        if addrs:
-            # We have an address, but check if the message is non-fatal.
-            if scan_messages(mlist, msg) is Stop:
-                return
+        context = BounceContext.normal
+        addresses = StandardVERP().get_verp(mlist, msg)
+        if addresses:
+            # We have an address, but check if the message is non-fatal.  It
+            # will be non-fatal if the bounce scanner returns Stop.  It will
+            # return a set of addresses when the bounce is fatal, but we don't
+            # care about those addresses, since we got it out of the VERP.
+            if scan_message(mlist, msg) is Stop:
+                return False
         else:
             # See if this was a probe message.
-            token = verp_probe(mlist, msg)
-            if token:
-                self._probe_bounce(mlist, token)
-                return
-            # That didn't give us anything useful, so try the old fashion
-            # bounce matching modules.
-            addrs = scan_messages(mlist, msg)
-            if addrs is Stop:
-                # This is a recognized, non-fatal notice. Ignore it.
-                return
+            addresses = ProbeVERP().get_verp(mlist, msg)
+            if addresses:
+                context = BounceContext.probe
+            else:
+                # That didn't give us anything useful, so try the old fashion
+                # bounce matching modules.
+                addresses = scan_message(mlist, msg)
+                if addresses is Stop:
+                    # This is a recognized, non-fatal notice. Ignore it.
+                    return False
         # If that still didn't return us any useful addresses, then send it on
         # or discard it.
-        if not addrs:
-            log.info('bounce message w/no discernable addresses: %s',
-                     msg.get('message-id'))
+        if len(addresses) > 0:
+            for address in addresses:
+                self._processor.register(mlist, address, msg, context)
+        else:
+            log.info('Bounce message w/no discernable addresses: %s',
+                     msg.get('message-id', 'n/a'))
             maybe_forward(mlist, msg)
-            return
-        # BAW: It's possible that there are None's in the list of addresses,
-        # although I'm unsure how that could happen.  Possibly scan_messages()
-        # can let None's sneak through.  In any event, this will kill them.
-        addrs = filter(None, addrs)
-        self._queue_bounces(mlist.fqdn_listname, addrs, msg)
-
-    _do_periodic = BounceMixin._do_periodic
-
-    def _clean_up(self):
-        BounceMixin._clean_up(self)
-        Runner._clean_up(self)
-
-
-
-def maybe_forward(mlist, msg):
-    # Does the list owner want to get non-matching bounce messages?
-    # If not, simply discard it.
-    if mlist.bounce_unrecognized_goes_to_list_owner:
-        adminurl = mlist.GetScriptURL('admin', absolute=1) + '/bounce'
-        mlist.ForwardMessage(msg,
-                             text=_("""\
-The attached message was received as a bounce, but either the bounce format
-was not recognized, or no member addresses could be extracted from it.  This
-mailing list has been configured to send all unrecognized bounce messages to
-the list administrator(s).
-
-For more information see:
-%(adminurl)s
-
-"""),
-                             subject=_('Uncaught bounce notification'),
-                             tomoderators=0)
-        log.error('forwarding unrecognized, message-id: %s',
-                  msg.get('message-id', 'n/a'))
-    else:
-        log.error('discarding unrecognized, message-id: %s',
-                  msg.get('message-id', 'n/a'))
+        # Dequeue this message.
+        return False
diff --git a/src/mailman/queue/docs/command.txt b/src/mailman/queue/docs/command.txt
index dfe6b8c19..c767e6f5f 100644
--- a/src/mailman/queue/docs/command.txt
+++ b/src/mailman/queue/docs/command.txt
@@ -63,7 +63,7 @@ And now the response is in the ``virgin`` queue.
     _parsemsg           : False
     listname            : test@example.com
     nodecorate          : True
-    recipients          : [u'aperson@example.com']
+    recipients          : set([u'aperson@example.com'])
     reduced_list_headers: True
     version             : ...
 
diff --git a/src/mailman/queue/outgoing.py b/src/mailman/queue/outgoing.py
index 7ff194219..ed27f014c 100644
--- a/src/mailman/queue/outgoing.py
+++ b/src/mailman/queue/outgoing.py
@@ -22,12 +22,16 @@ import logging
 
 from datetime import datetime
 from lazr.config import as_boolean, as_timedelta
+from zope.component import getUtility
 
 from mailman.config import config
+from mailman.interfaces.bounce import BounceContext, IBounceProcessor
 from mailman.interfaces.mailinglist import Personalization
+from mailman.interfaces.membership import ISubscriptionService
 from mailman.interfaces.mta import SomeRecipientsFailed
+from mailman.interfaces.pending import IPendings
 from mailman.queue import Runner
-from mailman.queue.bounce import BounceMixin
+from mailman.utilities.datetime import now
 from mailman.utilities.modules import find_name
 
 
@@ -36,15 +40,15 @@ from mailman.utilities.modules import find_name
 DEAL_WITH_PERMFAILURES_EVERY = 10
 
 log = logging.getLogger('mailman.error')
+smtp_log = logging.getLogger('mailman.smtp')
 
 
 
-class OutgoingRunner(Runner, BounceMixin):
+class OutgoingRunner(Runner):
     """The outgoing queue runner."""
 
     def __init__(self, slice=None, numslices=1):
-        Runner.__init__(self, slice, numslices)
-        BounceMixin.__init__(self)
+        super(OutgoingRunner, self).__init__(slice, numslices)
         # We look this function up only at startup time.
         self._func = find_name(config.mta.outgoing)
         # This prevents smtp server connection problems from filling up the
@@ -56,7 +60,7 @@ class OutgoingRunner(Runner, BounceMixin):
     def _dispose(self, mlist, msg, msgdata):
         # See if we should retry delivery of this message again.
         deliver_after = msgdata.get('deliver_after', datetime.fromtimestamp(0))
-        if datetime.now() < deliver_after:
+        if now() < deliver_after:
             return True
         # Calculate whether we should VERP this message or not.  The results of
         # this set the 'verp' key in the message metadata.
@@ -69,7 +73,7 @@ class OutgoingRunner(Runner, BounceMixin):
         # Also, if personalization is /not/ enabled, but
         # verp_delivery_interval is set (and we've hit this interval), then
         # again, this message should be VERP'd. Otherwise, no.
-        elif mlist.personalize <> Personalization.none:
+        elif mlist.personalize != Personalization.none:
             if as_boolean(config.mta.verp_personalized_deliveries):
                 msgdata['verp'] = True
         elif interval == 0:
@@ -88,59 +92,69 @@ class OutgoingRunner(Runner, BounceMixin):
             # 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 = int(config.mta.port)
+            port = int(config.mta.smtp_port)
             if port == 0:
-                port = 'smtp'
-            # Log this just once.
+                port = 'smtp'            # Log this just once.
             if not self._logged:
                 log.error('Cannot connect to SMTP server %s on port %s',
-                          config.mta.host, port)
+                          config.mta.smtp_host, port)
                 self._logged = True
             return True
         except SomeRecipientsFailed as error:
-            # Handle local rejects of probe messages differently.
-            if msgdata.get('probe_token') and error.permanent_failures:
-                self._probe_bounce(mlist, msgdata['probe_token'])
+            processor = getUtility(IBounceProcessor)
+            # BAW: msg is 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 'probe_token' in msgdata:
+                # This is a failure of our local MTA to deliver to a probe
+                # message recipient.  Register the bounce event for permanent
+                # failures.  Start by grabbing and confirming (i.e. removing)
+                # the pendable record associated with this bounce token,
+                # regardless of what address was actually failing.
+                if len(error.permanent_failures) > 0:
+                    pended = getUtility(IPendings).confirm(
+                        msgdata['probe_token'])
+                    # It's possible the token has been confirmed out of the
+                    # database.  Just ignore that.
+                    if pended is not None:
+                        member = getUtility(ISubscriptionService).get_member(
+                            pended['member_id'])
+                        processor.register(
+                            mlist, member.address.email, msg,
+                            BounceContext.probe)
             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 error.permanent_failures:
-                    self._queue_bounces(
-                        mlist.fqdn_listname, error.permanent_failures, msg)
+                for email in error.permanent_failures:
+                    processor.register(mlist, email, msg, BounceContext.normal)
                 # Move temporary failures to the qfiles/retry queue which will
                 # occasionally move them back here for another shot at
                 # delivery.
                 if error.temporary_failures:
-                    now = datetime.now()
-                    recips = error.temporary_failures
+                    current_time = now()
+                    recipients = error.temporary_failures
                     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:
+                    deliver_until = msgdata.get('deliver_until', current_time)
+                    if len(recipients) == last_recip_count:
+                        # We didn't make any progress.  If we've exceeded the
+                        # configured retry period, log this failure and
+                        # discard the message.
+                        if current_time > deliver_until:
+                            smtp_log.error('Discarding message with '
+                                           'persistent temporary failures: '
+                                           '{0}'.format(msg['message-id']))
                             return False
                     else:
-                        # Keep trying to delivery this message for a while
-                        deliver_until = now + as_timedelta(
+                        # We made some progress, so keep trying to delivery
+                        # this message for a while longer.
+                        deliver_until = current_time + as_timedelta(
                             config.mta.delivery_retry_period)
-                    msgdata['last_recip_count'] = len(recips)
+                    msgdata['last_recip_count'] = len(recipients)
                     msgdata['deliver_until'] = deliver_until
-                    msgdata['recipients'] = recips
+                    msgdata['recipients'] = recipients
                     self._retryq.enqueue(msg, msgdata)
-        # We've successfully completed handling of this message
+        # We've successfully completed handling of this message.
         return False
-
-    _do_periodic = BounceMixin._do_periodic
-
-    def _clean_up(self):
-        BounceMixin._clean_up(self)
-        Runner._clean_up(self)
diff --git a/src/mailman/queue/tests/__init__.py b/src/mailman/queue/tests/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/src/mailman/queue/tests/__init__.py
diff --git a/src/mailman/queue/tests/test_bounce.py b/src/mailman/queue/tests/test_bounce.py
new file mode 100644
index 000000000..1946df50c
--- /dev/null
+++ b/src/mailman/queue/tests/test_bounce.py
@@ -0,0 +1,236 @@
+# Copyright (C) 2011 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the bounce queue runner."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'test_suite',
+    ]
+
+
+import unittest
+
+from zope.component import getUtility
+
+from mailman.app.bounces import send_probe
+from mailman.app.lifecycle import create_list
+from mailman.config import config
+from mailman.interfaces.bounce import (
+    BounceContext, IBounceProcessor, UnrecognizedBounceDisposition)
+from mailman.interfaces.member import MemberRole
+from mailman.interfaces.usermanager import IUserManager
+from mailman.queue.bounce import BounceRunner
+from mailman.testing.helpers import (
+    LogFileMark,
+    get_queue_messages,
+    make_testable_runner,
+    specialized_message_from_string as message_from_string)
+from mailman.testing.layers import ConfigLayer
+
+
+
+class TestBounceQueue(unittest.TestCase):
+    """Test the bounce queue runner."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._mlist = create_list('test@example.com')
+        self._bounceq = config.switchboards['bounces']
+        self._runner = make_testable_runner(BounceRunner, 'bounces')
+        self._anne = getUtility(IUserManager).create_address(
+            'anne@example.com')
+        self._member = self._mlist.subscribe(self._anne, MemberRole.member)
+        self._msg = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces+anne=example.com@example.com
+Message-Id: <first>
+
+""")
+        self._msgdata = dict(listname='test@example.com')
+        self._processor = getUtility(IBounceProcessor)
+        config.push('site owner', """
+        [mailman]
+        site_owner: postmaster@example.com
+        """)
+
+    def tearDown(self):
+        config.pop('site owner')
+
+    def test_does_no_processing(self):
+        # If the mailing list does no bounce processing, the messages are
+        # simply discarded.
+        self._mlist.bounce_processing = False
+        self._bounceq.enqueue(self._msg, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        self.assertEqual(len(list(self._processor.events)), 0)
+
+    def test_verp_detection(self):
+        # When we get a VERPd bounce, and we're doing processing, a bounce
+        # event will be registered.
+        self._bounceq.enqueue(self._msg, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].list_name, 'test@example.com')
+        self.assertEqual(events[0].message_id, '<first>')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[0].processed, False)
+
+    def test_nonfatal_verp_detection(self):
+        # A VERPd bounce was received, but the error was nonfatal.
+        nonfatal = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces+anne=example.com@example.com
+Message-Id: <first>
+Content-Type: multipart/report; report-type=delivery-status; boundary=AAA
+MIME-Version: 1.0
+
+--AAA
+Content-Type: message/delivery-status
+
+Action: delayed
+Original-Recipient: rfc822; somebody@example.com
+
+--AAA--
+""")
+        self._bounceq.enqueue(nonfatal, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 0)
+
+    def test_verp_probe_bounce(self):
+        # A VERP probe bounced.  The primary difference here is that the
+        # registered bounce event will have a different context.  The
+        # Message-Id will be different too, because of the way we're
+        # simulating the probe bounce.
+        #
+        # Start be simulating a probe bounce.
+        send_probe(self._member, self._msg)
+        message = get_queue_messages('virgin')[0].msg
+        bounce = message_from_string("""\
+To: {0}
+From: mail-daemon@example.com
+Message-Id: <second>
+
+""".format(message['From']))
+        self._bounceq.enqueue(bounce, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].list_name, 'test@example.com')
+        self.assertEqual(events[0].message_id, '<second>')
+        self.assertEqual(events[0].context, BounceContext.probe)
+        self.assertEqual(events[0].processed, False)
+
+    def test_nonverp_detectable_fatal_bounce(self):
+        # Here's a bounce that is not VERPd, but which has a bouncing address
+        # that can be parsed from a known bounce format.  DSN is as good as
+        # any, but we'll make the parsed address different for the fun of it.
+        dsn = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces@example.com
+Message-Id: <first>
+Content-Type: multipart/report; report-type=delivery-status; boundary=AAA
+MIME-Version: 1.0
+
+--AAA
+Content-Type: message/delivery-status
+
+Action: fail
+Original-Recipient: rfc822; bart@example.com
+
+--AAA--
+""")
+        self._bounceq.enqueue(dsn, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'bart@example.com')
+        self.assertEqual(events[0].list_name, 'test@example.com')
+        self.assertEqual(events[0].message_id, '<first>')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[0].processed, False)
+
+    def test_nonverp_detectable_nonfatal_bounce(self):
+        # Here's a bounce that is not VERPd, but which has a bouncing address
+        # that can be parsed from a known bounce format.  The bounce is
+        # non-fatal so no bounce event is registered.
+        dsn = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces@example.com
+Message-Id: <first>
+Content-Type: multipart/report; report-type=delivery-status; boundary=AAA
+MIME-Version: 1.0
+
+--AAA
+Content-Type: message/delivery-status
+
+Action: delayed
+Original-Recipient: rfc822; bart@example.com
+
+--AAA--
+""")
+        self._bounceq.enqueue(dsn, self._msgdata)
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 0)
+
+    def test_no_detectable_bounce_addresses(self):
+        # A bounce message was received, but no addresses could be detected.
+        # A message will be logged in the bounce log though, and the message
+        # can be forwarded to someone who can do something about it.
+        self._mlist.forward_unrecognized_bounces_to = (
+            UnrecognizedBounceDisposition.site_owner)
+        bogus = message_from_string("""\
+From: mail-daemon@example.com
+To: test-bounces@example.com
+Message-Id: <third>
+
+""")
+        self._bounceq.enqueue(bogus, self._msgdata)
+        mark = LogFileMark('mailman.bounce')
+        self._runner.run()
+        self.assertEqual(len(get_queue_messages('bounces')), 0)
+        events = list(self._processor.events)
+        self.assertEqual(len(events), 0)
+        line = mark.readline()
+        self.assertEqual(
+            line[-51:-1],
+            'Bounce message w/no discernable addresses: <third>')
+        # Here's the forwarded message to the site owners.
+        forwards = get_queue_messages('virgin')
+        self.assertEqual(len(forwards), 1)
+        self.assertEqual(forwards[0].msg['to'], 'postmaster@example.com')
+
+
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(TestBounceQueue))
+    return suite
diff --git a/src/mailman/queue/tests/test_outgoing.py b/src/mailman/queue/tests/test_outgoing.py
new file mode 100644
index 000000000..a0fe407c8
--- /dev/null
+++ b/src/mailman/queue/tests/test_outgoing.py
@@ -0,0 +1,549 @@
+# Copyright (C) 2011 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the outgoing queue runner."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'test_suite',
+    ]
+
+
+import os
+import socket
+import logging
+import unittest
+
+from contextlib import contextmanager
+from datetime import datetime, timedelta
+from lazr.config import as_timedelta
+from zope.component import getUtility
+
+from mailman.app.bounces import send_probe
+from mailman.app.lifecycle import create_list
+from mailman.config import config
+from mailman.interfaces.bounce import BounceContext, IBounceProcessor
+from mailman.interfaces.mailinglist import Personalization
+from mailman.interfaces.member import MemberRole
+from mailman.interfaces.mta import SomeRecipientsFailed
+from mailman.interfaces.pending import IPendings
+from mailman.interfaces.usermanager import IUserManager
+from mailman.queue.outgoing import OutgoingRunner
+from mailman.testing.helpers import (
+    LogFileMark,
+    get_queue_messages,
+    make_testable_runner,
+    specialized_message_from_string as message_from_string)
+from mailman.testing.layers import ConfigLayer, SMTPLayer
+from mailman.utilities.datetime import factory, now
+
+
+
+def run_once(qrunner):
+    """Predicate for make_testable_runner().
+
+    Ensures that the queue runner only runs once.
+    """
+    return True
+
+
+@contextmanager
+def temporary_config(name, settings):
+    """Temporarily set a configuration (use in a with-statement)."""
+    config.push(name, settings)
+    try:
+        yield
+    finally:
+        config.pop(name)
+
+
+
+class TestOnce(unittest.TestCase):
+    """Test outgoing runner message disposition."""
+
+    layer = SMTPLayer
+
+    def setUp(self):
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out', run_once)
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+        self._msgdata = {}
+
+    def test_deliver_after(self):
+        # When the metadata has a deliver_after key in the future, the queue
+        # runner will re-enqueue the message rather than delivering it.
+        deliver_after = now() + timedelta(days=10)
+        self._msgdata['deliver_after'] = deliver_after
+        self._outq.enqueue(self._msg, self._msgdata,
+                           tolist=True, listname='test@example.com')
+        self._runner.run()
+        items = get_queue_messages('out')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['deliver_after'], deliver_after)
+        self.assertEqual(items[0].msg['message-id'], '<first>')
+
+
+
+captured_mlist = None
+captured_msg = None
+captured_msgdata = None
+
+def capture(mlist, msg, msgdata):
+    global captured_mlist, captured_msg, captured_msgdata
+    captured_mlist = mlist
+    captured_msg = msg
+    captured_msgdata = msgdata
+
+
+class TestVERPSettings(unittest.TestCase):
+    """Test the selection of VERP based on various criteria."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        global captured_mlist, captured_msg, captured_msgdata
+        # Push a config where actual delivery is handled by a dummy function.
+        # We generally don't care what this does, since we're just testing the
+        # setting of the 'verp' key in the metadata.
+        config.push('fake outgoing', """
+        [mta]
+        outgoing: mailman.queue.tests.test_outgoing.capture
+        """)
+        # Reset the captured data.
+        captured_mlist = None
+        captured_msg = None
+        captured_msgdata = None
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out')
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+
+    def tearDown(self):
+        config.pop('fake outgoing')
+
+    def test_delivery_callback(self):
+        # Test that the configuration variable calls the appropriate callback.
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        self.assertEqual(captured_mlist, self._mlist)
+        self.assertEqual(captured_msg.as_string(), self._msg.as_string())
+        # Of course, the message metadata will contain a bunch of keys added
+        # by the processing.  We don't really care about the details, so this
+        # test is a good enough stand-in.
+        self.assertEqual(captured_msgdata['listname'], 'test@example.com')
+
+    def test_verp_in_metadata(self):
+        # Test that if the metadata has a 'verp' key, it is unchanged.
+        marker = 'yepper'
+        msgdata = dict(verp=marker)
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], marker)
+
+    def test_personalized_individual_deliveries_verp(self):
+        # When deliveries are personalized, and the configuration setting
+        # indicates, messages will be VERP'd.
+        msgdata = {}
+        self._mlist.personalize = Personalization.individual
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_personalized_deliveries: yes
+        """):
+            self._runner.run()
+        self.assertTrue(captured_msgdata['verp'])
+
+    def test_personalized_full_deliveries_verp(self):
+        # When deliveries are personalized, and the configuration setting
+        # indicates, messages will be VERP'd.
+        msgdata = {}
+        self._mlist.personalize = Personalization.full
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_personalized_deliveries: yes
+        """):
+            self._runner.run()
+        self.assertTrue(captured_msgdata['verp'])
+
+    def test_personalized_deliveries_no_verp(self):
+        # When deliveries are personalized, but the configuration setting
+        # does not indicate, messages will not be VERP'd.
+        msgdata = {}
+        self._mlist.personalize = Personalization.full
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        self.assertFalse('verp' in captured_msgdata)
+
+    def test_verp_never(self):
+        # Never VERP when the interval is zero.
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_delivery_interval: 0
+        """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], False)
+
+    def test_verp_always(self):
+        # Always VERP when the interval is one.
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_delivery_interval: 1
+        """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], True)
+
+    def test_verp_on_interval_match(self):
+        # VERP every so often, when the post_id matches.
+        self._mlist.post_id = 5
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+        [mta]
+        verp_delivery_interval: 5
+        """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], True)
+
+    def test_no_verp_on_interval_miss(self):
+        # VERP every so often, when the post_id matches.
+        self._mlist.post_id = 4
+        msgdata = {}
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        with temporary_config('personalize', """
+            [mta]
+            verp_delivery_interval: 5
+            """):
+            self._runner.run()
+        self.assertEqual(captured_msgdata['verp'], False)
+
+
+
+def raise_socket_error(mlist, msg, msgdata):
+    raise socket.error
+
+
+class TestSocketError(unittest.TestCase):
+    """Test socket.error occurring in the delivery function."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        # Push a config where actual delivery is handled by a dummy function.
+        # We generally don't care what this does, since we're just testing the
+        # setting of the 'verp' key in the metadata.
+        config.push('fake outgoing', """
+        [mta]
+        outgoing: mailman.queue.tests.test_outgoing.raise_socket_error
+        """)
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out', run_once)
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+
+    def tearDown(self):
+        config.pop('fake outgoing')
+
+    def test_error_with_port_0(self):
+        # Test the code path where a socket.error is raised in the delivery
+        # function, and the MTA port is set to zero.  The only real effect of
+        # that is a log message.  Start by opening the error log and reading
+        # the current file position.
+        error_log = logging.getLogger('mailman.error')
+        filename = error_log.handlers[0].filename
+        filepos = os.stat(filename).st_size
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        with temporary_config('port 0', """
+            [mta]
+            smtp_port: 0
+            """):
+            self._runner.run()
+        with open(filename) as fp:
+            fp.seek(filepos)
+            line = fp.readline()
+        # The log line will contain a variable timestamp, the PID, and a
+        # trailing newline.  Ignore these.
+        self.assertEqual(
+            line[-53:-1],
+            'Cannot connect to SMTP server localhost on port smtp')
+
+    def test_error_with_numeric_port(self):
+        # Test the code path where a socket.error is raised in the delivery
+        # function, and the MTA port is set to zero.  The only real effect of
+        # that is a log message.  Start by opening the error log and reading
+        # the current file position.
+        mark = LogFileMark('mailman.error')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        with temporary_config('port 0', """
+            [mta]
+            smtp_port: 2112
+            """):
+            self._runner.run()
+        line = mark.readline()
+        # The log line will contain a variable timestamp, the PID, and a
+        # trailing newline.  Ignore these.
+        self.assertEqual(
+            line[-53:-1],
+            'Cannot connect to SMTP server localhost on port 2112')
+
+
+
+temporary_failures = []
+permanent_failures = []
+
+
+def raise_SomeRecipientsFailed(mlist, msg, msgdata):
+    raise SomeRecipientsFailed(temporary_failures, permanent_failures)
+
+
+class TestSomeRecipientsFailed(unittest.TestCase):
+    """Test socket.error occurring in the delivery function."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        global temporary_failures, permanent_failures
+        del temporary_failures[:]
+        del permanent_failures[:]
+        self._processor = getUtility(IBounceProcessor)
+        # Push a config where actual delivery is handled by a dummy function.
+        # We generally don't care what this does, since we're just testing the
+        # setting of the 'verp' key in the metadata.
+        config.push('fake outgoing', """
+        [mta]
+        outgoing: mailman.queue.tests.test_outgoing.raise_SomeRecipientsFailed
+        """)
+        self._mlist = create_list('test@example.com')
+        self._outq = config.switchboards['out']
+        self._runner = make_testable_runner(OutgoingRunner, 'out', run_once)
+        self._msg = message_from_string("""\
+From: anne@example.com
+To: test@example.com
+Message-Id: <first>
+
+""")
+
+    def tearDown(self):
+        config.pop('fake outgoing')
+
+    def test_probe_failure(self):
+        # When a probe message fails during SMTP, a bounce event is recorded
+        # with the proper bounce context.
+        anne = getUtility(IUserManager).create_address('anne@example.com')
+        member = self._mlist.subscribe(anne, MemberRole.member)
+        token = send_probe(member, self._msg)
+        msgdata = dict(probe_token=token)
+        permanent_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 1)
+        event = events[0]
+        self.assertEqual(event.list_name, 'test@example.com')
+        self.assertEqual(event.email, 'anne@example.com')
+        self.assertEqual(event.timestamp, datetime(2005, 8, 1, 7, 49, 23))
+        self.assertEqual(event.message_id, '<first>')
+        self.assertEqual(event.context, BounceContext.probe)
+        self.assertEqual(event.processed, False)
+
+    def test_confirmed_probe_failure(self):
+        # This time, a probe also fails, but for some reason the probe token
+        # has already been confirmed and no longer exists in the database.
+        anne = getUtility(IUserManager).create_address('anne@example.com')
+        member = self._mlist.subscribe(anne, MemberRole.member)
+        token = send_probe(member, self._msg)
+        getUtility(IPendings).confirm(token)
+        msgdata = dict(probe_token=token)
+        permanent_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+
+    def test_probe_temporary_failure(self):
+        # This time, a probe also fails, but the failures are temporary so
+        # they are not registered.
+        anne = getUtility(IUserManager).create_address('anne@example.com')
+        member = self._mlist.subscribe(anne, MemberRole.member)
+        token = send_probe(member, self._msg)
+        getUtility(IPendings).confirm(token)
+        msgdata = dict(probe_token=token)
+        temporary_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+
+    def test_one_permanent_failure(self):
+        # Normal (i.e. non-probe) permanent failures just get registered.
+        permanent_failures.append('anne@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].context, BounceContext.normal)
+
+    def test_two_permanent_failures(self):
+        # Two normal (i.e. non-probe) permanent failures just get registered.
+        permanent_failures.append('anne@example.com')
+        permanent_failures.append('bart@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 2)
+        self.assertEqual(events[0].email, 'anne@example.com')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[1].email, 'bart@example.com')
+        self.assertEqual(events[1].context, BounceContext.normal)
+
+    def test_one_temporary_failure(self):
+        # The first time there are temporary failures, the message just gets
+        # put in the retry queue, but with some metadata to prevent infinite
+        # retries.
+        temporary_failures.append('cris@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+        items = get_queue_messages('retry')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(self._msg.as_string(), items[0].msg.as_string())
+        # The metadata has three keys which are used two decide whether the
+        # next temporary failure should be retried.
+        self.assertEqual(items[0].msgdata['last_recip_count'], 1)
+        deliver_until = (datetime(2005, 8, 1, 7, 49, 23) +
+                         as_timedelta(config.mta.delivery_retry_period))
+        self.assertEqual(items[0].msgdata['deliver_until'], deliver_until)
+        self.assertEqual(items[0].msgdata['recipients'], ['cris@example.com'])
+
+    def test_two_temporary_failures(self):
+        # The first time there are temporary failures, the message just gets
+        # put in the retry queue, but with some metadata to prevent infinite
+        # retries.
+        temporary_failures.append('cris@example.com')
+        temporary_failures.append('dave@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 0)
+        items = get_queue_messages('retry')
+        # There's still only one item in the retry queue, but the metadata
+        # contains both temporary failures.
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['last_recip_count'], 2)
+        self.assertEqual(items[0].msgdata['recipients'],
+                         ['cris@example.com', 'dave@example.com'])
+
+    def test_mixed_failures(self):
+        # Some temporary and some permanent failures.
+        permanent_failures.append('elle@example.com')
+        permanent_failures.append('fred@example.com')
+        temporary_failures.append('gwen@example.com')
+        temporary_failures.append('herb@example.com')
+        self._outq.enqueue(self._msg, {}, listname='test@example.com')
+        self._runner.run()
+        # Let's look at the permanent failures.
+        events = list(self._processor.unprocessed)
+        self.assertEqual(len(events), 2)
+        self.assertEqual(events[0].email, 'elle@example.com')
+        self.assertEqual(events[0].context, BounceContext.normal)
+        self.assertEqual(events[1].email, 'fred@example.com')
+        self.assertEqual(events[1].context, BounceContext.normal)
+        # Let's look at the temporary failures.
+        items = get_queue_messages('retry')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['recipients'],
+                         ['gwen@example.com', 'herb@example.com'])
+
+    def test_no_progress_on_retries_within_retry_period(self):
+        # Temporary failures cause queuing for a retry later on, unless no
+        # progress is being made on the retries and we've tried for the
+        # specified delivery retry period.  This test ensures that even if no
+        # progress is made, if the retry period hasn't expired, the message
+        # will be requeued.
+        temporary_failures.append('iona@example.com')
+        temporary_failures.append('jeff@example.com')
+        deliver_until = (datetime(2005, 8, 1, 7, 49, 23) +
+                         as_timedelta(config.mta.delivery_retry_period))
+        msgdata = dict(last_recip_count=2,
+                       deliver_until=deliver_until)
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        self._runner.run()
+        # The retry queue should have our message waiting to be retried.
+        items = get_queue_messages('retry')
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0].msgdata['deliver_until'], deliver_until)
+        self.assertEqual(items[0].msgdata['recipients'],
+                         ['iona@example.com', 'jeff@example.com'])
+
+    def test_no_progress_on_retries_with_expired_retry_period(self):
+        # We've had temporary failures with no progress, and the retry period
+        # has expired.  In that case, a log entry is written and message is
+        # discarded.  There's nothing more that can be done.
+        temporary_failures.append('kira@example.com')
+        temporary_failures.append('lonn@example.com')
+        retry_period = as_timedelta(config.mta.delivery_retry_period)
+        deliver_until = datetime(2005, 8, 1, 7, 49, 23) + retry_period
+        msgdata = dict(last_recip_count=2,
+                       deliver_until=deliver_until)
+        self._outq.enqueue(self._msg, msgdata, listname='test@example.com')
+        # Before the queue runner runs, several days pass.
+        factory.fast_forward(retry_period.days + 1)
+        mark = LogFileMark('mailman.smtp')
+        self._runner.run()
+        # There should be no message in the retry or outgoing queues.
+        self.assertEqual(len(get_queue_messages('retry')), 0)
+        self.assertEqual(len(get_queue_messages('out')), 0)
+        # There should be a log message in the smtp log indicating that the
+        # message has been discarded.
+        line = mark.readline()
+        self.assertEqual(
+            line[-63:-1],
+            'Discarding message with persistent temporary failures: <first>')
+
+
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(TestOnce))
+    suite.addTest(unittest.makeSuite(TestVERPSettings))
+    suite.addTest(unittest.makeSuite(TestSocketError))
+    suite.addTest(unittest.makeSuite(TestSomeRecipientsFailed))
+    return suite
diff --git a/src/mailman/styles/default.py b/src/mailman/styles/default.py
index 4f2b3d7d5..68934e99c 100644
--- a/src/mailman/styles/default.py
+++ b/src/mailman/styles/default.py
@@ -24,6 +24,7 @@ __all__ = [
     'DefaultStyle',
     ]
 
+
 # XXX Styles need to be reconciled with lazr.config.
 
 import datetime
@@ -32,6 +33,7 @@ from zope.interface import implements
 
 from mailman.core.i18n import _
 from mailman.interfaces.action import Action
+from mailman.interfaces.bounce import UnrecognizedBounceDisposition
 from mailman.interfaces.digests import DigestFrequency
 from mailman.interfaces.autorespond import ResponseAction
 from mailman.interfaces.mailinglist import Personalization, ReplyToMunging
@@ -161,13 +163,14 @@ ${listinfo_page}
         mlist.autoresponse_request_text = ''
         mlist.autoresponse_grace_period = datetime.timedelta(days=90)
         # Bounces
+        mlist.forward_unrecognized_bounces_to = (
+            UnrecognizedBounceDisposition.administrators)
         mlist.bounce_processing = True
         mlist.bounce_score_threshold = 5.0
         mlist.bounce_info_stale_after = datetime.timedelta(days=7)
         mlist.bounce_you_are_disabled_warnings = 3
         mlist.bounce_you_are_disabled_warnings_interval = (
             datetime.timedelta(days=7))
-        mlist.bounce_unrecognized_goes_to_list_owner = True
         mlist.bounce_notify_owner_on_disable = True
         mlist.bounce_notify_owner_on_removal = True
         # This holds legacy member related information.  It's keyed by the
diff --git a/src/mailman/templates/en/unrecognized.txt b/src/mailman/templates/en/unrecognized.txt
new file mode 100644
index 000000000..16bd19e79
--- /dev/null
+++ b/src/mailman/templates/en/unrecognized.txt
@@ -0,0 +1,7 @@
+The attached message was received as a bounce, but either the bounce format
+was not recognized, or no member addresses could be extracted from it.  This
+mailing list has been configured to send all unrecognized bounce messages to
+the list administrator(s).
+
+For more information see:
+$adminurl
diff --git a/src/mailman/testing/helpers.py b/src/mailman/testing/helpers.py
index 71cddd0f4..ed579e39c 100644
--- a/src/mailman/testing/helpers.py
+++ b/src/mailman/testing/helpers.py
@@ -20,6 +20,7 @@ from __future__ import absolute_import, unicode_literals
 
 __metaclass__ = type
 __all__ = [
+    'LogFileMark',
     'TestableMaster',
     'call_api',
     'digest_mbox',
@@ -39,6 +40,7 @@ import time
 import errno
 import signal
 import socket
+import logging
 import smtplib
 import datetime
 import threading
@@ -63,7 +65,7 @@ from mailman.utilities.mailbox import Mailbox
 
 
 
-def make_testable_runner(runner_class, name=None):
+def make_testable_runner(runner_class, name=None, predicate=None):
     """Create a queue runner that runs until its queue is empty.
 
     :param runner_class: The queue runner's class.
@@ -71,6 +73,10 @@ def make_testable_runner(runner_class, name=None):
     :param name: Optional queue name; if not given, it is calculated from the
         class name.
     :type name: string or None
+    :param predicate: Optional alternative predicate for deciding when to stop
+        the queue runner.  When None (the default) it stops when the queue is
+        empty.
+    :type predicate: callable that gets one argument, the queue runner.
     :return: A runner instance.
     """
 
@@ -90,7 +96,10 @@ def make_testable_runner(runner_class, name=None):
 
         def _do_periodic(self):
             """Stop when the queue is empty."""
-            self._stop = (len(self.switchboard.files) == 0)
+            if predicate is None:
+                self._stop = (len(self.switchboard.files) == 0)
+            else:
+                self._stop = predicate(self)
 
     return EmptyingRunner(name)
 
@@ -364,7 +373,7 @@ def reset_the_world():
     config.db.commit()
     # Reset the global style manager.
     config.style_manager.populate()
-    
+
 
 
 def specialized_message_from_string(unicode_text):
@@ -384,3 +393,16 @@ def specialized_message_from_string(unicode_text):
     message = message_from_string(text, Message)
     message.original_size = original_size
     return message
+
+
+
+class LogFileMark:
+    def __init__(self, log_name):
+        self._log = logging.getLogger(log_name)
+        self._filename = self._log.handlers[0].filename
+        self._filepos = os.stat(self._filename).st_size
+
+    def readline(self):
+        with open(self._filename) as fp:
+            fp.seek(self._filepos)
+            return fp.readline()