Synapse should limit parallel connections to send email to a SMTP server

This issue has been migrated from [#5612](https://github.com/matrix-org/synapse/issues/5612).

---

## Description

When sending lots of emails, some mailservers apply rate limiting or parallel connection limits.

eg, mail logs include:
```
2019-07-03 06:55:30 Connection from [10.0.0.0.1] refused: too many connections
```

This seems to kick in due to a number of parallel smtp connections being opened by synapse, generally due to pushes attempting to update lots of users about a message in a previously idle room (or an `@room` notification, i suppose)

Synapse errors with this:

```
2019-07-03 06:55:30,494 - synapse.push.mailer - 259 - INFO - emailpush.process-352663 - Sending email notification to
2019-07-03 06:55:30,541 - synapse.push.emailpusher - 156 - ERROR - emailpush.process-352663 - Exception processing notifs
Capture point (most recent call last):
File "/usr/local/lib/python3.7/runpy.py", line 193, in _run_module_as_main
"__main__", mod_spec)
File "/usr/local/lib/python3.7/runpy.py", line 85, in _run_code
exec(code, run_globals)
File "/home/synapse/src/synapse/app/pusher.py", line 227, in
ps = start(sys.argv[1:])
File "/home/synapse/src/synapse/app/pusher.py", line 222, in start
_base.start_worker_reactor("synapse-pusher", config)
File "/home/synapse/src/synapse/app/_base.py", line 71, in start_worker_reactor
logger=logger,
File "/home/synapse/src/synapse/app/_base.py", line 120, in start_reactor
daemon.start()
File "/home/synapse/env-py37/lib/python3.7/site-packages/daemonize.py", line 248, in start
self.action(*privileged_action_result)
File "/home/synapse/src/synapse/app/_base.py", line 106, in run
reactor.run()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/base.py", line 1272, in run
self.mainLoop()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/base.py", line 1284, in mainLoop
self.doIteration(t)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/epollreactor.py", line 235, in doPoll
log.callWithLogger(selectable, _drdw, selectable, fd, event)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/python/log.py", line 103, in callWithLogger
return callWithContext({"system": lp}, func, *args, **kw)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/python/log.py", line 86, in callWithContext
return context.call({ILogContext: newCtx}, func, *args, **kw)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/python/context.py", line 122, in callWithContext
return self.currentContext().callWithContext(ctx, func, *args, **kw)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/python/context.py", line 85, in callWithContext
return func(*args,**kw)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/posixbase.py", line 614, in _doReadOrWrite
why = selectable.doRead()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/tcp.py", line 243, in doRead
return self._dataReceived(data)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/tcp.py", line 249, in _dataReceived
rval = self.protocol.dataReceived(data)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/protocols/basic.py", line 572, in dataReceived
why = self.lineReceived(line)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/mail/smtp.py", line 998, in lineReceived
why = self._failresponse(self.code, b'\n'.join(self.resp))
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/mail/smtp.py", line 1006, in smtpConnectionFailed
self.sendError(SMTPConnectError(code, resp, self.log.str()))
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/mail/smtp.py", line 1819, in sendError
self.factory.result.errback(exc)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 501, in errback
self._startRunCallbacks(fail)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 568, in _startRunCallbacks
self._runCallbacks()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 654, in _runCallbacks
current.result = callback(current.result, *args, **kw)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 1475, in gotResult
_inlineCallbacks(r, g, status)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 1464, in _inlineCallbacks
status.deferred.errback()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 501, in errback
self._startRunCallbacks(fail)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 568, in _startRunCallbacks
self._runCallbacks()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 654, in _runCallbacks
current.result = callback(current.result, *args, **kw)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 1475, in gotResult
_inlineCallbacks(r, g, status)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 1464, in _inlineCallbacks
status.deferred.errback()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 501, in errback
self._startRunCallbacks(fail)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 568, in _startRunCallbacks
self._runCallbacks()
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 654, in _runCallbacks
current.result = callback(current.result, *args, **kw)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 1475, in gotResult
_inlineCallbacks(r, g, status)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/internet/defer.py", line 1416, in _inlineCallbacks
result = result.throwExceptionIntoGenerator(g)
File "/home/synapse/env-py37/lib/python3.7/site-packages/twisted/python/failure.py", line 512, in throwExceptionIntoGenerator
return g.throw(self.type, self.value, self.tb)
Traceback (most recent call last):
File "/home/synapse/src/synapse/push/emailpusher.py", line 154, in _process
yield self._unsafe_process()
twisted.mail._except.SMTPConnectError: 421 b'Too many concurrent SMTP connections; please try again later.'
<<< 421 Too many concurrent SMTP connections; please try again later.
```

I think we would do well with a limit on the count of SMTP connections running in parallel, even if this means that a temporary queue of emails builds up.

We definitely shouldn't fire and forget, especially for things like password resets, which seem to go through similar code, and while are unlikely to themselves hit a parallel connection limit, would fail if they were to be triggered within the same time window as a large email push is going on.

### Version information

- **Homeserver**: matrix.org