release 0.0.301 (#32)

* customizable exception page with @app.error(500)

* implement app_handler_timeout

* more efficient ASGILifespan impl.

---------

Co-authored-by: nggit <[email protected]>

setup.py

@@ -7,7 +7,7 @@
 
 setup(
     name='tremolo',
-    version='0.0.300',
+    version='0.0.301',
     license='MIT',
     author='nggit',
     author_email='[email protected]',

tests/http_server.py

@@ -270,6 +270,8 @@ async def timeouts(request=None, **_):
         # attempt to read body on a GET request
         # should raise a TimeoutError and ended up with a RequestTimeout
         await request.recv(100)
+    elif request.query_string == b'handler':
+        await asyncio.sleep(10)
 
 
 @app.route('/reload')
@@ -284,7 +286,7 @@ async def reload(request=None, **_):
 
 # test multiple ports
 app.listen(HTTP_PORT + 1, request_timeout=2, keepalive_timeout=2)
-app.listen(HTTP_PORT + 2)
+app.listen(HTTP_PORT + 2, app_handler_timeout=1)
 
 # test unix socket
 # 'tremolo-test.sock'

tests/test_http_client.py

@@ -501,6 +501,18 @@ def test_recvtimeout(self):
                          b'HTTP/1.1 408 Request Timeout')
         self.assertEqual(body, b'Request Timeout')
 
+    def test_handlertimeout(self):
+        header, body = getcontents(
+            host=HTTP_HOST,
+            port=HTTP_PORT + 2,
+            raw=b'GET /timeouts?handler HTTP/1.1\r\n'
+                b'Host: localhost:%d\r\n\r\n' % (HTTP_PORT + 2)
+        )
+
+        self.assertEqual(header[:header.find(b'\r\n')],
+                         b'HTTP/1.1 500 Internal Server Error')
+        self.assertEqual(body, b'Internal Server Error')
+
     def test_download_10(self):
         header, body = getcontents(host=HTTP_HOST,
                                    port=HTTP_PORT + 2,

tremolo/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '0.0.300'
+__version__ = '0.0.301'
 
 from .tremolo import Tremolo  # noqa: E402
 from . import exceptions  # noqa: E402,F401

tremolo/__main__.py

@@ -50,6 +50,12 @@
         print('  --keepalive-timeout       Defaults to 30 (seconds)')
         print('  --keepalive-connections   Maximum number of keep-alive connections')  # noqa: E501
         print('                            Defaults to 512 (connections/worker)')  # noqa: E501
+        print('  --app-handler-timeout     Kill the app if it takes too long to finish')  # noqa: E501
+        print('                            Upgraded connection/scope will not be affected')  # noqa: E501
+        print('                            Defaults to 120 (seconds)')
+        print('  --app-close-timeout       Kill the app if it does not exit within this timeframe,')  # noqa: E501
+        print('                            from when the client is disconnected')  # noqa: E501
+        print('                            Defaults to 30 (seconds)')
         print('  --server-name             Set the "Server" field in the response header')  # noqa: E501
         print('  --root-path               Set the ASGI root_path. Defaults to ""')  # noqa: E501
         print('  --help                    Show this help and exit')
@@ -73,7 +79,9 @@
                              '--client-max-header-size',
                              '--request-timeout',
                              '--keepalive-timeout',
-                             '--keepalive-connections'):
+                             '--keepalive-connections',
+                             '--app-handler-timeout',
+                             '--app-close-timeout'):
         try:
             options[sys.argv[i - 1].lstrip('-').replace('-', '_')] = int(sys.argv[i])  # noqa: E501
         except ValueError:

tremolo/asgi_lifespan.py

@@ -10,26 +10,26 @@ def __init__(self, app, **kwargs):
         self._loop = kwargs['loop']
         self._logger = kwargs['logger']
 
-        scope = {
-            'type': 'lifespan',
-            'asgi': {'version': '3.0'}
-        }
-
         self._queue = asyncio.Queue()
-        self._task = self._loop.create_task(
-            app(scope, self.receive, self.send)
-        )
-        self._complete = False
+        self._waiter = self._loop.create_future()
+        self._task = self._loop.create_task(self.main(app))
 
-    def startup(self):
-        self._complete = False
+    async def main(self, app):
+        try:
+            scope = {
+                'type': 'lifespan',
+                'asgi': {'version': '3.0'}
+            }
 
+            await app(scope, self.receive, self.send)
+        finally:
+            self._waiter.cancel()
+
+    def startup(self):
         self._queue.put_nowait({'type': 'lifespan.startup'})
         self._logger.info('lifespan: startup')
 
     def shutdown(self):
-        self._complete = False
-
         self._queue.put_nowait({'type': 'lifespan.shutdown'})
         self._logger.info('lifespan: shutdown')
 
@@ -42,7 +42,7 @@ async def receive(self):
     async def send(self, data):
         if data['type'] in ('lifespan.startup.complete',
                             'lifespan.shutdown.complete'):
-            self._complete = True
+            self._waiter.set_result(None)
             self._logger.info(data['type'])
         elif data['type'] in ('lifespan.startup.failed',
                               'lifespan.shutdown.failed'):
@@ -56,10 +56,14 @@ async def send(self, data):
             raise LifespanProtocolUnsupported
 
     async def exception(self, timeout=30):
-        for _ in range(timeout):
-            if self._complete:
-                return
+        timer = self._loop.call_at(self._loop.time() + timeout,
+                                   self._waiter.cancel)
+
+        try:
+            await self._waiter
 
+            self._waiter = self._loop.create_future()
+        except asyncio.CancelledError:
             try:
                 exc = self._task.exception()
 
@@ -72,12 +76,9 @@ async def exception(self, timeout=30):
                     else:
                         self._logger.info(
                             '%s: %s' % (LifespanProtocolUnsupported.message,
-                                        str(exc))
+                                        str(exc) or repr(exc))
                         )
-
-                return
             except asyncio.InvalidStateError:
-                await asyncio.sleep(1)
-
-        if not self._complete:
-            self._logger.warning('lifespan: timeout after %gs' % timeout)
+                self._logger.warning('lifespan: timeout after %gs' % timeout)
+        finally:
+            timer.cancel()

tremolo/asgi_server.py

@@ -26,22 +26,16 @@
 
 
 class ASGIServer(HTTPProtocol):
-    __slots__ = ('_app',
-                 '_scope',
+    __slots__ = ('_scope',
                  '_read',
-                 '_task',
                  '_timer',
-                 '_timeout',
                  '_websocket',
                  '_http_chunked')
 
-    def __init__(self, _app=None, **kwargs):
-        self._app = _app
+    def __init__(self, **kwargs):
         self._scope = None
         self._read = None
-        self._task = None
         self._timer = None
-        self._timeout = 30
         self._websocket = None
         self._http_chunked = None
 
@@ -88,37 +82,32 @@ async def header_received(self):
             await self._handle_http()
             self._read = self.request.stream()
 
-        self._task = self.loop.create_task(self.app())
+        self.handler = self.loop.create_task(self.main())
 
     def connection_lost(self, exc):
-        if (self._task is not None and not self._task.done() and
-                self._timer is None):
-            self._timer = self.loop.call_at(self.loop.time() + self._timeout,
-                                            self._task.cancel)
+        if self.handler is not None and not self.handler.done():
+            self._set_app_close_timeout()
 
         super().connection_lost(exc)
 
-    async def app(self):
+    async def main(self):
         try:
-            await self._app(self._scope, self.receive, self.send)
+            await self.options['_app'](self._scope, self.receive, self.send)
 
             if self._timer is not None:
                 self._timer.cancel()
-        except asyncio.CancelledError:
-            self.logger.warning(
-                'task: ASGI application is cancelled due to timeout'
-            )
-        except Exception as exc:
+        except (asyncio.CancelledError, Exception) as exc:
             if (self.request is not None and self.request.upgraded and
                     self._websocket is not None):
                 exc = WebSocketServerClosed(cause=exc)
 
             await self.handle_exception(exc)
 
-    def _set_app_timeout(self):
+    def _set_app_close_timeout(self):
         if self._timer is None:
             self._timer = self.loop.call_at(
-                self.loop.time() + self._timeout, self._task.cancel
+                self.loop.time() + self.options['_app_close_timeout'],
+                self.handler.cancel
             )
 
     async def receive(self):
@@ -152,7 +141,7 @@ async def receive(self):
                 if self.request is not None:
                     self.print_exception(exc)
 
-                self._set_app_timeout()
+                self._set_app_close_timeout()
                 return {
                     'type': 'websocket.disconnect',
                     'code': code
@@ -176,12 +165,12 @@ async def receive(self):
                     self.request.body_size < self.request.content_length
                 )
             }
-        except Exception as exc:
+        except (asyncio.CancelledError, Exception) as exc:
             if not (self.request is None or
                     isinstance(exc, StopAsyncIteration)):
                 self.print_exception(exc)
 
-            self._set_app_timeout()
+            self._set_app_close_timeout()
             return {'type': 'http.disconnect'}
 
     async def send(self, data):

tremolo/handlers.py

@@ -1,5 +1,7 @@
 # Copyright (c) 2023 nggit
 
+import traceback
+
 from .exceptions import BadRequest
 from .utils import html_escape
 
@@ -27,3 +29,12 @@ async def error_404(request=None, **_):
         b'<address title="Powered by Tremolo">%s</address>'
         b'</body></html>' % request.protocol.options['server_info']['name']
     )
+
+
+async def error_500(request=None, exc=None, **_):
+    if request.protocol.options['debug']:
+        return '<ul><li>%s</li></ul>' % '</li><li>'.join(
+            traceback.TracebackException.from_exception(exc).format()
+        )
+
+    return str(exc)