Reset HTTP parser for new transport generations

src/hio/core/http/clienting.py

@@ -858,9 +858,12 @@ def wind(self, tymth):
 
     def reopen(self):
         """
-        Return result of .connector.reopen()
+        Open a new transport generation and reset its response parser.
         """
-        return self.connector.reopen()
+        opened = self.connector.reopen()
+        if opened and self.respondent is not None:
+            self.respondent.reopen()
+        return opened
 
 
     def close(self):
@@ -1044,12 +1047,14 @@ def redirect(self):
 
                 self.secured = secured
                 self.connector = connector
-                self.connector.reopen()
+                opened = self.connector.reopen()
                 self.requester.reinit(hostname=hostname,
                                       port=port,
                                       scheme=scheme)
                 self.respondent.reinit(msg=self.connector.rxbs,
                                        method=method)
+                if opened:
+                    self.respondent.reopen()
 
             qargs = dict()
             qargs, query = httping.updateQargsQuery(qargs, query)
@@ -1151,7 +1156,7 @@ def service(self):
 
             if self.connector.reconnectable:  # useful for server sent event stream
                 if self.connector.tymeout > 0.0 and self.connector.tymer.expired:  # timed out
-                    self.connector.reopen()
+                    self.reopen()
                     if self.respondent.evented:
                         duration = float(self.respondent.retry) / 1000.0 # convert to seconds
                     else:

src/hio/core/http/httping.py

@@ -942,6 +942,19 @@ def close(self):
         """
         self.closed = True
 
+    def reopen(self):
+        """
+        Reset terminal parser state for a new transport generation.
+        """
+        self.started = False
+        self.headed = False
+        self.bodied = False
+        self.ended = False
+        self.closed = False
+        self.errored = False
+        self.error = None
+        self.makeParser()  # set up for new msg
+
     def checkPersisted(self):
         """
         Checks headers to determine if connection should be kept open until

tests/core/http/test_clienting.py

@@ -28,6 +28,125 @@
 certdirpath = os.path.join(tlsdirpath, 'tls', 'certs')
 
 
+def _service_client_connection(client, server):
+    """Drive a real nonblocking Client connection through server acceptance."""
+    for _ in range(100):
+        client.connector.serviceConnect()
+        server.serviceConnects()
+        if (client.connector.connected and
+                client.connector.ca in server.ixes):
+            return
+        time.sleep(0.01)
+    raise AssertionError("client did not connect")
+
+
+def test_respondent_reopen_resets_parser_generation():
+    """A new transport generation explicitly resets terminal parser state."""
+    respondent = clienting.Respondent(msg=bytearray())
+    old_parser = respondent.parser
+    respondent.started = True
+    respondent.headed = True
+    respondent.bodied = True
+    respondent.ended = True
+    respondent.closed = True
+    respondent.errored = True
+    respondent.error = "old transport failed"
+
+    respondent.reopen()
+
+    assert respondent.parser is not old_parser
+    assert not respondent.started
+    assert not respondent.headed
+    assert not respondent.bodied
+    assert not respondent.ended
+    assert not respondent.closed
+    assert not respondent.errored
+    assert respondent.error is None
+
+
+def test_client_reopen_resets_parser_generation():
+    """Manual reopen rearms parsing after opening a real new socket."""
+    with tcp.openServer(ha=("127.0.0.1", 0)) as server:
+        # Acceptor does not refresh .eha after binding an ephemeral port.
+        server.eha = server.ha
+        with clienting.openClient(hostname=server.ha[0],
+                                  port=server.ha[1]) as client:
+            _service_client_connection(client, server)
+            old_socket = client.connector.cs
+            old_parser = client.respondent.parser
+            client.respondent.close()
+
+            assert client.reopen()
+
+            assert old_socket.fileno() == -1
+            assert client.connector.cs is not old_socket
+            assert client.respondent.parser is not old_parser
+            assert not client.respondent.closed
+
+
+def test_client_automatic_reconnect_resets_parser_generation():
+    """A real cutoff/reconnect uses the parser-generation transition."""
+    tymist = tyming.Tymist(tyme=0.0)
+    with tcp.openServer(tymth=tymist.tymen(),
+                        ha=("127.0.0.1", 0)) as server:
+        server.eha = server.ha
+        with clienting.openClient(hostname=server.ha[0],
+                                  port=server.ha[1],
+                                  tymth=tymist.tymen(),
+                                  reconnectable=True,
+                                  tymeout=1.0) as client:
+            _service_client_connection(client, server)
+            old_parser = client.respondent.parser
+
+            server.removeIx(client.connector.ca)  # trigger server EOF; client eventually gets EOF
+            for _ in range(100):
+                client.connector.serviceReceives()
+                if client.connector.cutoff:
+                    break
+                time.sleep(0.01)
+            assert client.connector.cutoff  # we've received EOF and should be cutoff
+
+            tymist.tick(tock=1.1)
+            assert client.connector.tymer.expired
+            client.service()  # triggers new parser setup
+
+            assert client.respondent.parser is not old_parser  # should have new parser
+            assert not client.respondent.closed
+            assert not client.connector.cutoff
+
+
+def test_cross_origin_redirect_resets_parser_generation():
+    """A real cross-origin connector replacement rearms response parsing."""
+    with tcp.openServer(ha=("127.0.0.1", 0)) as origin, \
+         tcp.openServer(ha=("127.0.0.1", 0)) as target:
+        origin.eha = origin.ha
+        target.eha = target.ha
+        with clienting.openClient(hostname=origin.ha[0],
+                                  port=origin.ha[1]) as client:
+            _service_client_connection(client, origin)
+            old_connector = client.connector
+            old_socket = old_connector.cs
+            old_parser = client.respondent.parser
+            client.respondent.close()
+            client.redirects = [{
+                "headers": help.Hict({
+                    "Location": "http://{}:{}/next".format(*target.ha),
+                }),
+                "method": "GET",
+            }]
+
+            client.redirect()
+
+            assert old_socket.fileno() == -1
+            assert client.connector is not old_connector
+            assert isinstance(client.connector, tcp.Client)
+            assert client.connector.ha == target.ha
+            assert client.respondent.msg is client.connector.rxbs
+            assert client.respondent.parser is not old_parser
+            assert not client.respondent.closed
+            _service_client_connection(client, target)
+
+
 def mockEchoService(server):
     """
     mock echo server service for testing