wsgi: remove problematic send(resource) workaround

With the WIT updates, there were errors in the WSGI
framework code around its use of http_req.send in
order to attempt to cause a resource.drop on requests
when doing connection reuse.

Componentize-py, without any added workarounds, knows
to call resource.drop as a finalizer on resources.
Finalizers execute when the refcount of an object
goes to zero.  With the updates, we do a bit of extra
work to remove references to resources and add in
assertions based on weak references in order to ensure
that we don't have lingering referneces to request
resources.

Use of a context manager was considered, but in this
case I did not find a way that use of one would be
feasible or reduce complexity.  In componentize-py,
the only path to calling resource.drop I found related
to finalizers.  Even if there were a method to
explicitly issue a resource.drop, the behavior might
be undefined with outstanding references, so I think
this change still makes sense.

By default, if there are lingering references to
request resources under connection reuse, we simply
will not reuse connections and print a warning.

Author: posborne

examples/game-of-life.py

@@ -264,4 +264,6 @@ def root():
 
 
 if running_under_compute:
-    HttpIncoming = WsgiHttpIncoming(app, reuse_sandboxes_for_ms=300)
+    HttpIncoming = WsgiHttpIncoming(
+        app, reuse_sandboxes_for_ms=300, assert_on_uncollected_resource=True
+    )

fastly_compute/wsgi.py

@@ -8,21 +8,16 @@
 # a top-level import componentize-py won't include the encoding in our final artifact
 # and we get runtime LookupErrors when werkzeug tries to use the codec.
 import encodings.idna  # noqa: F401
+import gc
 import sys
 import traceback
+import weakref
 from collections.abc import Callable
 from typing import Any
 from urllib.parse import urlparse
 
 from wit_world.exports import HttpIncoming as WitHttpIncoming
-from wit_world.imports import http_body, http_resp
-from wit_world.imports.http_downstream import (
-    NextRequestOptions,
-    await_request,
-    next_request,
-)
-from wit_world.imports.http_req import send
-from wit_world.imports.http_resp import send_downstream
+from wit_world.imports import async_io, http_body, http_downstream, http_req, http_resp
 from wit_world.imports.types import Err, Error_CannotRead
 
 
@@ -94,7 +89,7 @@ def start_response(
             write(body_chunk)
 
         # Send the complete response downstream
-        send_downstream(response, response_body)
+        http_resp.send_downstream(response, response_body)
 
     except Exception as e:
         if not handle_errors:
@@ -110,7 +105,25 @@ def start_response(
         error_response.append_header("content-type", b"text/plain")
         error_message = f"Internal Server Error: {e}"
         http_body.write(error_body, error_message.encode(), http_body.WriteEnd.BACK)
-        send_downstream(error_response, error_body)
+        http_resp.send_downstream(error_response, error_body)
+
+
+def _await_request(pending_request: async_io.Pollable) -> None | (
+    http_req.Request,
+    async_io.Pollabe,
+):
+    try:
+        return http_downstream.await_request(pending_request)
+    except Err as exc:
+        # TODO: Improve error design so we can catch only the exceptions
+        # we're really interested in, per Python's idiom. Rather than
+        # carting around a Result type that's Union[Ok[T], Err[E]], we
+        # should probably return T xor raise E.
+        if isinstance(exc.value, Error_CannotRead):
+            return None
+        else:
+            # Something went wrong.
+            raise
 
 
 class WsgiHttpIncoming(WitHttpIncoming):
@@ -140,6 +153,7 @@ def __init__(
         wsgi_app: Callable,
         handle_errors: bool = False,
         reuse_sandboxes_for_ms: int = 0,
+        assert_on_uncollected_resource: bool = False,
     ):
         """Construct.
 
@@ -148,72 +162,70 @@ def __init__(
             500-status response.
         :arg reuse_sandboxes_for_ms: If non-0, keep the sandbox alive for this
             many milliseconds to potentially serve additional requests.
+        :arg assert_on_uncollected_resource: If True, if there are references
+            to request that linger past that request being handled, an assertion
+            will be raised (when reusing connections).  If False, connection
+            reuse will be disabled if this occurs.
         """
         self.wsgi_app = wsgi_app
         self.handle_errors = handle_errors
         self.reuse_sandboxes_for_ms = reuse_sandboxes_for_ms
+        self.assert_on_uncollected_resource = assert_on_uncollected_resource
 
     def __call__(self):
         return self
 
-    def handle(self, request: Any, body: Any) -> None:
-        """Handle incoming HTTP requests by serving them through the WSGI app."""
+    def _handle_single_request(self, request: http_req.Request, body: Any) -> None:
         serve_wsgi_request(
             request,
             body,
             self.wsgi_app,
             handle_errors=self.handle_errors,
         )
 
-        if not self.reuse_sandboxes_for_ms:
-            return
-
-        try:
-            # Drop (in the WIT sense) the `request` resource to get ready for
-            # another request. Otherwise, we crash.
-            #
-            # Here we abuse an arbitrary request-consuming function to trigger
-            # the drop. Glue code interposed by wasmtime's linker ensures that
-            # drop happens, but send() otherwise fails before doing anything.
-            send(request, body, "no such backend")
-
-            # TODO: Generate a proper drop_whatever() function for each
-            # "whatever" resource.
-            #
-            # Alternately, it might suffice for the runtime to drop() things
-            # that get GC'd (i.e. `del` or otherwise) by Python. If we put an
-            # idiomatic .close() or similar on, for example, a potentially large
-            # request body, we could implement it in terms of `del`.
-        except Err:
-            pass
-        else:
-            raise RuntimeError(
-                "Our use of send() to consume the previous request unexpectedly actually performed a send."
-            )
-
-        options = NextRequestOptions(timeout_ms=self.reuse_sandboxes_for_ms, extra=None)
+    def _attempt_reuse(self):
+        options = http_downstream.NextRequestOptions(
+            timeout_ms=self.reuse_sandboxes_for_ms, extra=None
+        )
         while True:
-            pending_request = next_request(options)
-            try:
-                result = await_request(pending_request)
-            except Err as exc:
-                # TODO: Improve error design so we can catch only the exceptions
-                # we're really interested in, per Python's idiom. Rather than
-                # carting around a Result type that's Union[Ok[T], Err[E]], we
-                # should probably return T xor raise E.
-                if isinstance(exc.value, Error_CannotRead):
-                    # There were no more requests within the timeout.
-                    break
-                else:
-                    # Something went wrong.
-                    raise
+            # See if there's another request incoming
+            pending_request = http_downstream.next_request(options)
+            maybe_req_body = _await_request(pending_request)
+            if not maybe_req_body:
+                break
+            request, body = maybe_req_body
+            reqwref = weakref.ref(request)
+            with request as request, body as body:
+                maybe_req_body = None
+                self._handle_single_request(request, body)
+            del request
+            del body
+
+            req_dropped = reqwref() is None
+            if not req_dropped:
+                # force a gc in case it is in a cycle; this may not be optimal
+                # but worth a shot if the user is trying to make enable
+                # reuse.
+                gc.collect()
+                req_dropped = reqwref() is None
+
+            if self.assert_on_uncollected_resource:
+                assert req_dropped, "Request outlives handler!"
             else:
-                if not result:
+                if not req_dropped:
+                    # TODO: use proper logging at some point
+                    print(
+                        "Request remained referenced past WSGI handler, not doing connection reuse"
+                    )
                     break
-                request, body = result
-                serve_wsgi_request(
-                    request,
-                    body,
-                    self.wsgi_app,
-                    handle_errors=self.handle_errors,
-                )
+
+    def handle(self, request: Any, body: Any) -> None:
+        """Handle incoming HTTP requests by serving them through the WSGI app."""
+        # always handle the first request coming in normally
+        self._handle_single_request(request, body)
+        del request
+        del body
+
+        # then attempt reuse if configured
+        if self.reuse_sandboxes_for_ms:
+            return self._attempt_reuse()