Merge bounce cleanup branch. This is a major cleansing and refactoring of the

bounce processor.  Note that one thing this does not do is implement any of
the policy around bounce scores.  It "merely" cleans up all the crud in the
BounceRunner, and all the logic around registering bounce events.  This means
the bounce runner can actually be enabled again.  More code needs to be
written to process bounces events occasionally.

Details:

* maybe_forward() moved to app/bounces.py, and it can now discard the
  message, forward it to the list administrators (moderators + owners), or
  site owners.  See UnrecognizedBounceDisposition.
* scan_message() always returns a set of addresses.
* The DSN bounce detector is cleaned up.
* An IBounceProcessor utility is added.
* IBounceEvents are added, with database support.
* bounce_unrecognized_goes_to_list_owner -> forward_unrecognized_bounces_to
* bounce_processing -> process_bounces
* ReopenableFileHandler.filename is exposed as a public attribute.  This aids
  in testing.
* Fix the signature of UserNotification.__init__() to be more PEP 8 compliant.
* Change the OwnerNotification.__init__() signature to take a roster object
  instead of `tomoderators`.  When the roster is None, the site owner is used
  instead.
* Fix the default style setting; s/personalization/personalize/
* BounceMixin is gone, baby gone.
* Add tests for the OutgoingRunner.
* make_testable_runner() can now take a predicate object, which can change how
  often the runner goes through its main loop.  Use this e.g. to go through
  only once when a message does not get dequeued.
* A new LogFileMark helper for testing messages to a log file.

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()