[WIP] server: having issues with POST partial reads

Author: sebastien

src/py/extra/http/model.py

@@ -19,6 +19,7 @@
 from dataclasses import dataclass
 from pathlib import Path
 from enum import Enum
+from ..utils.logging import warning
 from ..utils.primitives import TPrimitive
 from .status import HTTP_STATUS
 from ..utils.io import DEFAULT_ENCODING, asWritable
@@ -139,26 +140,58 @@ def __init__(
 
 
 class HTTPBodyIO:
-    __slots__ = ["reader", "read", "expected", "remaining"]
+    __slots__ = ["reader", "read", "expected", "remaining", "existing"]
     """Represents a body that is loaded from a reader IO."""
 
-    def __init__(self, reader: "HTTPBodyReader", expected: int | None = None):
+    def __init__(
+        self,
+        reader: "HTTPBodyReader",
+        expected: int | None = None,
+        existing: bytes | None = None,
+    ):
         self.reader: HTTPBodyReader = reader
         self.read: int = 0
         self.expected: int | None = expected
         self.remaining: int | None = expected
+        self.existing: bytes | None = existing
 
-    async def load(
+    async def _read(
         self,
     ) -> bytes | None:
-        """Loads all the data and returns a list of bodies."""
-        payload = await self.reader.load()
-        if payload:
-            n = len(payload)
-            self.read += n
-            if self.remaining is not None:
+        """Reads the next available bytes"""
+        if self.existing and self.read == 0:
+            self.read += len(self.existing)
+            return self.existing
+        elif self.remaining:
+            # FIXME: We should probably have a timeout there
+            try:
+                payload = await self.reader.load()
+            except TimeoutError:
+                warning(
+                    "Request body loading timed out",
+                    Remaining=self.remaining,
+                    Read=self.read,
+                )
+                return None
+            if payload:
+                n = len(payload)
+                self.read += n
                 self.remaining -= n
-        return payload
+            return payload
+        else:
+            return None
+
+    async def load(self) -> bytes:
+        """Fully loads the body."""
+        res = bytearray()
+        # FIXME: This would read other requests as well if there is no
+        # remaining -- there should be at least a delimiter.
+        while True:
+            chunk = await self._read()
+            if chunk:
+                res += chunk
+            else:
+                return bytes(res)
 
 
 class HTTPBodyBlob(NamedTuple):
@@ -457,10 +490,17 @@ def contentType(self) -> str | None:
 
     @property
     def body(self) -> HTTPBodyIO | HTTPBodyBlob:
+        b = self._body
         if self._body is None:
             if not self._reader:
                 raise RuntimeError("Request has no reader, can't read body")
             self._body = HTTPBodyIO(self._reader)
+        elif isinstance(self._body, HTTPBodyBlob) and self._body.remaining:
+            if not self._reader:
+                raise RuntimeError("Request has no reader, can't read body")
+            self._body = HTTPBodyIO(
+                self._reader, expected=self._body.remaining, existing=self._body.payload
+            )
         return self._body
 
     @property

src/py/extra/http/parser.py

@@ -324,13 +324,19 @@ def feed(self, chunk: bytes) -> Iterator[HTTPAtom]:
                     if self.requestLine is None or self.requestHeaders is None:
                         yield HTTPProcessingStatus.BadFormat
                     else:
+                        # FIXME: In some circumstances (for POST requests),
+                        # reading the body will time out, and then the request
+                        # will be duplicated.
                         headers = self.requestHeaders
                         line = self.requestLine
                         # NOTE: This is an awkward dance around the type checker
                         if self.parser is self.bodyRest:
                             self.parser.feed(chunk[offset:])
                             offset = len(chunk)
                         body = self.parser.flush()
+                        # NOTE: Careful here as we may create a request that has
+                        # a body with remaining data. The HTTP request will need
+                        # to make sure it can continue reading the body.
                         yield (
                             HTTPRequest(
                                 method=line.method,

src/py/extra/server.py

@@ -100,6 +100,12 @@ def __init__(
     async def _read(
         self, timeout: float = 1.0, size: int | None = None
     ) -> bytes | None:
+        info(
+            "Reading Body",
+            Client=f"{id(self.socket):x}",
+            Size=size,
+            Timeout=timeout,
+        )
         return await asyncio.wait_for(
             self.loop.sock_recv(self.socket, size or self.size),
             timeout=timeout,
@@ -213,15 +219,31 @@ async def OnRequest(
                 # request in the same payload, so we need to be prepared
                 # to answer more than one request.
                 stream = parser.feed(buffer[:n] if n != size else buffer)
+                # TODO: Should be debug
+                info(
+                    "Reading Requests(s)",
+                    Client=f"{id(client):x}",
+                    Read=n,
+                    Iteration=iteration,
+                    Count=req_count,
+                )
                 while True:
+
                     try:
                         atom = next(stream)
+                        # TODO: Should be debug
+                        info("Request Atom", Atom=atom.__class__.__name__)
                     except StopIteration:
+                        # TODO: Should be debug
+                        info("Request End")
                         break
                     if atom is HTTPProcessingStatus.Complete:
                         status = atom
                     elif isinstance(atom, HTTPRequest):
                         req = atom
+                        # We pass the reader to the request, as for instance
+                        # the request may need more than what was available
+                        # from the socket.
                         req._reader = reader
                         # Logs the request method
                         if options.logRequests:
@@ -234,8 +256,19 @@ async def OnRequest(
                             keep_alive = False
                         if await cls.SendResponse(req, app, writer):
                             res_count += 1
+                            info("Request Sent", Iteration=iteration, Count=res_count)
+                        else:
+                            warning(
+                                "Sending Response Failed",
+                                Iteration=iteration,
+                                Count=res_count,
+                            )
                 # We clear what we've read from the buffer
-                del buffer[:n]
+                if n != size:
+                    del buffer[:n]
+                else:
+                    # FIXME: Or is it buffer.clear()?
+                    del buffer[:]
                 iteration += 1
 
             # NOTE: We'll need to think about the loading of the body, which