Implement throttling based on participant

defaults.env

@@ -73,9 +73,12 @@ OPENSTREETMAP_CALLBACK=http://127.0.0.1:8537/on/openstreetmap/associate
 OPENSTREETMAP_API_URL=http://www.openstreetmap.org/api/0.6
 OPENSTREETMAP_AUTH_URL=http://www.openstreetmap.org
 
+EMAIL_QUEUE_FLUSH_EVERY=60
+EMAIL_QUEUE_SLEEP_FOR=1
+EMAIL_QUEUE_ALLOW_UP_TO=3
+
 UPDATE_CTA_EVERY=300
 CHECK_DB_EVERY=600
-FLUSH_EMAIL_QUEUE_EVERY=60
 OPTIMIZELY_ID=
 INCLUDE_PIWIK=no
 SENTRY_DSN=

gratipay/application.py

@@ -52,7 +52,7 @@ def install_periodic_jobs(self, website, env, db):
         cron = Cron(website)
         cron(env.update_cta_every, lambda: utils.update_cta(website))
         cron(env.check_db_every, db.self_check, True)
-        cron(env.flush_email_queue_every, self.email_queue.flush, True)
+        cron(env.email_queue_flush_every, self.email_queue.flush, True)
 
 
     def add_event(self, c, type, payload):

gratipay/billing/payday.py

@@ -477,6 +477,7 @@ def notify_participants(self):
                 exchange=dict(id=e.id, amount=e.amount, fee=e.fee, note=e.note),
                 nteams=nteams,
                 top_team=top_team,
+                _user_initiated=False
             )
 
 

gratipay/cli/queue_branch_email.py

@@ -86,4 +86,4 @@ def prompt(msg):
         _print( "{:>4} queuing for {} ({}={})".format(i, p.email_address, p.username, p.id)
               , file=sys.stderr
                )
-        app.email_queue.put(p, 'branch', include_unsubscribe=False)
+        app.email_queue.put(p, 'branch', include_unsubscribe=False, _user_initiated=False)

gratipay/email.py

@@ -13,6 +13,7 @@
 from jinja2 import Environment
 from markupsafe import escape as htmlescape
 
+from gratipay.exceptions import Throttled
 from gratipay.models.participant import Participant
 from gratipay.utils import find_files, i18n
 
@@ -35,6 +36,8 @@ def __init__(self, env, db, tell_sentry, root):
 
         self.db = db
         self.tell_sentry = tell_sentry
+        self.sleep_for = env.email_queue_sleep_for
+        self.allow_up_to = env.email_queue_allow_up_to
 
         templates = {}
         templates_dir = os.path.join(root, 'emails')
@@ -52,21 +55,35 @@ def _have_ses(self, env):
            and env.aws_ses_default_region
 
 
-    def put(self, to, template, **context):
+    def put(self, to, template, _user_initiated=True, **context):
         """Put an email message on the queue.
 
         :param Participant to: the participant to send the email message to
         :param unicode template: the name of the template to use when rendering
-          the email, corresponding to a filename in ``emails/`` without the file
-          extension
+            the email, corresponding to a filename in ``emails/`` without the
+            file extension
+        :param bool _user_initiated: user-initiated emails are throttled;
+            system-initiated messages don't count against throttling
         :param dict context: the values to use when rendering the template
 
+        :raise Throttled: if the participant already has a few messages in the
+            queue (that they put there); the specific number is tunable with
+            the ``EMAIL_QUEUE_ALLOW_UP_TO`` envvar.
+
+        :returns: ``None``
+
         """
-        self.db.run("""
-            INSERT INTO email_queue
-                        (participant, spt_name, context)
-                 VALUES (%s, %s, %s)
-        """, (to.id, template, pickle.dumps(context)))
+        with self.db.get_cursor() as cursor:
+            cursor.run("""
+                INSERT INTO email_queue
+                            (participant, spt_name, context, user_initiated)
+                     VALUES (%s, %s, %s, %s)
+            """, (to.id, template, pickle.dumps(context), _user_initiated))
+            if _user_initiated:
+                n = cursor.one('SELECT count(*) FROM email_queue '
+                               'WHERE participant=%s AND user_initiated', (to.id,))
+                if n > self.allow_up_to:
+                    raise Throttled()
 
 
     def flush(self):
@@ -87,7 +104,7 @@ def flush(self):
                 r = self._flush_one(rec)
                 self.db.run("DELETE FROM email_queue WHERE id = %s", (rec.id,))
                 if r == 1:
-                    sleep(1)
+                    sleep(self.sleep_for)
                 nsent += r
         return nsent
 

gratipay/exceptions.py

@@ -47,6 +47,10 @@ class TooManyEmailAddresses(ProblemChangingEmail):
     msg = "You've reached the maximum number of email addresses we allow."
 
 
+class Throttled(Exception):
+    msg = "You've initiated too many emails too quickly. Please try again in a minute or two."
+
+
 class ProblemChangingNumber(Exception):
     def __str__(self):
         return self.msg

gratipay/models/participant/email.py

@@ -55,6 +55,7 @@ def add_email(self, email):
             this participant
         :raises EmailTaken: if the email is verified for a different participant
         :raises TooManyEmailAddresses: if the participant already has 10 emails
+        :raises Throttled: if the participant adds too many emails too quickly
 
         """
 
@@ -113,6 +114,10 @@ def add_email(self, email):
                                     , 'verification_notice'
                                     , new_email=email
                                     , include_unsubscribe=False
+
+                                    # Don't count this one against their sending quota.
+                                    # It's going to their own verified address, anyway.
+                                    , _user_initiated=False
                                      )
 
 

gratipay/testing/email.py

@@ -8,6 +8,15 @@
 
 class _AbstractEmailHarness(Harness):
 
+    def setUp(self):
+        Harness.setUp(self)
+        self.__sleep_for = self.app.email_queue.sleep_for
+        self.app.email_queue.sleep_for = 0
+
+    def tearDown(self):
+        Harness.tearDown(self)
+        self.app.email_queue.sleep_for = self.__sleep_for
+
     def _get_last_email(self):
         raise NotImplementedError
 
@@ -45,7 +54,7 @@ class SentEmailHarness(_AbstractEmailHarness):
     """
 
     def setUp(self):
-        Harness.setUp(self)
+        _AbstractEmailHarness.setUp(self)
         self.mailer_patcher = mock.patch.object(self.app.email_queue._mailer, 'send_email')
         self.mailer = self.mailer_patcher.start()
         self.addCleanup(self.mailer_patcher.stop)

gratipay/wireup.py

@@ -386,7 +386,9 @@ def env():
         OPENSTREETMAP_AUTH_URL          = unicode,
         UPDATE_CTA_EVERY                = int,
         CHECK_DB_EVERY                  = int,
-        FLUSH_EMAIL_QUEUE_EVERY         = int,
+        EMAIL_QUEUE_FLUSH_EVERY         = int,
+        EMAIL_QUEUE_SLEEP_FOR           = int,
+        EMAIL_QUEUE_ALLOW_UP_TO         = int,
         OPTIMIZELY_ID                   = unicode,
         SENTRY_DSN                      = unicode,
         LOG_METRICS                     = is_yesish,

sql/branch.sql

@@ -0,0 +1,5 @@
+BEGIN;
+
+    ALTER TABLE email_queue ADD COLUMN user_initiated bool NOT NULL DEFAULT TRUE;
+
+END;

tests/py/test_email.py

@@ -3,8 +3,10 @@
 import json
 import sys
 
+from pytest import raises
+
 from gratipay.exceptions import CannotRemovePrimaryEmail, EmailTaken, EmailNotVerified
-from gratipay.exceptions import TooManyEmailAddresses
+from gratipay.exceptions import TooManyEmailAddresses, Throttled
 from gratipay.testing import P
 from gratipay.testing.email import QueuedEmailHarness, SentEmailHarness
 from gratipay.models.participant import email as _email
@@ -34,11 +36,13 @@ def verify_email(self, email, nonce, username='alice', should_fail=False):
         f = self.client.GxT if should_fail else self.client.GET
         return f(url, auth_as=username)
 
-    def verify_and_change_email(self, old_email, new_email, username='alice'):
+    def verify_and_change_email(self, old_email, new_email, username='alice', _flush=True):
         self.hit_email_spt('add-email', old_email)
         nonce = P(username).get_email(old_email).nonce
         self.verify_email(old_email, nonce)
         self.hit_email_spt('add-email', new_email)
+        if _flush:
+            self.app.email_queue.flush()
 
     def test_participant_can_add_email(self):
         response = self.hit_email_spt('add-email', '[email protected]')
@@ -66,9 +70,10 @@ def test_verification_email_doesnt_contain_unsubscribe(self):
         assert "To stop receiving" not in last_email['body_text']
 
     def test_adding_second_email_sends_verification_notice(self):
-        self.verify_and_change_email('[email protected]', '[email protected]')
+        self.verify_and_change_email('[email protected]', '[email protected]', _flush=False)
         assert self.count_email_messages() == 3
         last_email = self.get_last_email()
+        self.app.email_queue.flush()
         assert last_email['to'] == 'alice <[email protected]>'
         expected = "We are connecting [email protected] to the alice account on Gratipay"
         assert expected in last_email['body_text']
@@ -206,12 +211,15 @@ def test_remove_email(self):
 
 class TestFunctions(Alice):
 
+    def add_and_verify(self, participant, address):
+        participant.add_email('[email protected]')
+        nonce = participant.get_email('[email protected]').nonce
+        r = participant.verify_email('[email protected]', nonce)
+        assert r == _email.VERIFICATION_SUCCEEDED
+
     def test_cannot_update_email_to_already_verified(self):
         bob = self.make_participant('bob', claimed_time='now')
-        self.alice.add_email('[email protected]')
-        nonce = self.alice.get_email('[email protected]').nonce
-        r = self.alice.verify_email('[email protected]', nonce)
-        assert r == _email.VERIFICATION_SUCCEEDED
+        self.add_and_verify(self.alice, '[email protected]')
 
         with self.assertRaises(EmailTaken):
             bob.add_email('[email protected]')
@@ -225,12 +233,15 @@ def test_cannot_add_too_many_emails(self):
         self.alice.add_email('[email protected]')
         self.alice.add_email('[email protected]')
         self.alice.add_email('[email protected]')
+        self.app.email_queue.flush()
         self.alice.add_email('[email protected]')
         self.alice.add_email('[email protected]')
         self.alice.add_email('[email protected]')
+        self.app.email_queue.flush()
         self.alice.add_email('[email protected]')
         self.alice.add_email('[email protected]')
         self.alice.add_email('[email protected]')
+        self.app.email_queue.flush()
         self.alice.add_email('[email protected]')
         with self.assertRaises(TooManyEmailAddresses):
             self.alice.add_email('[email protected]')
@@ -241,6 +252,30 @@ def test_html_escaping(self):
         assert 'foo&#39;bar' in last_email['body_html']
         assert '&#39;' not in last_email['body_text']
 
+    def test_queueing_email_is_throttled(self):
+        self.app.email_queue.put(self.alice, "verification")
+        self.app.email_queue.put(self.alice, "branch")
+        self.app.email_queue.put(self.alice, "verification_notice")
+        raises(Throttled, self.app.email_queue.put, self.alice, "branch")
+
+    def test_only_user_initiated_messages_count_towards_throttling(self):
+        self.app.email_queue.put(self.alice, "verification")
+        self.app.email_queue.put(self.alice, "verification", _user_initiated=False)
+        self.app.email_queue.put(self.alice, "branch")
+        self.app.email_queue.put(self.alice, "branch", _user_initiated=False)
+        self.app.email_queue.put(self.alice, "verification_notice")
+        self.app.email_queue.put(self.alice, "verification_notice", _user_initiated=False)
+        raises(Throttled, self.app.email_queue.put, self.alice, "branch")
+
+    def test_flushing_queue_resets_throttling(self):
+        self.add_and_verify(self.alice, '[email protected]')
+        assert self.app.email_queue.flush() == 1
+        self.app.email_queue.put(self.alice, "verification")
+        self.app.email_queue.put(self.alice, "branch")
+        self.app.email_queue.put(self.alice, "verification_notice")
+        assert self.app.email_queue.flush() == 3
+        self.app.email_queue.put(self.alice, "verification_notice")
+
 
 class FlushEmailQueue(SentEmailHarness):
 

www/%team/set-status.json.spt

@@ -43,6 +43,7 @@ with website.db.get_cursor() as c:
                                        , team_name=team.name
                                        , review_url=team.review_url
                                        , include_unsubscribe=False
+                                       , _user_initiated=False
                                         )
 
 [---] application/json via json_dump

www/~/%username/identities/%country.spt

@@ -44,6 +44,7 @@ if identity is not None and participant != user.participant:
                                , country_name=country_name
                                , country_code=country.code
                                , include_unsubscribe=False
+                               , _user_initiated=False
                                 )
 
 # handle POST requests

www/~/%username/payout-status.spt

@@ -40,10 +40,18 @@ with website.db.get_cursor() as c:
         action='set', values=dict(status_of_1_0_payout=new_status)
     ))
 
-    if new_status == 'rejected':
-        website.app.email_queue.put(participant, 'rejected-1.0', include_unsubscribe=False)
-    elif new_status == 'pending-payout':
-        website.app.email_queue.put(participant, 'approved-1.0', include_unsubscribe=False)
+template = None
+if new_status == 'rejected':
+    template = 'rejected-1.0'
+elif new_status == 'pending-payout':
+    template = 'approved-1.0'
+
+if template:
+    website.app.email_queue.put( participant
+                               , template
+                               , include_unsubscribe=False
+                               , _user_initiated=False
+                                )
 
 out = {"status": new_status}
 [---] application/json via json_dump