wireprotov2: implement commands as a generator of objects

Previously, wire protocol version 2 inherited version 1's model of
having separate types to represent the results of different wire
protocol commands.

As I implemented more powerful commands in future commits, I found
I was using a common pattern of returning a special type to hold a
generator. This meant the command function required a closure to
do most of the work. That made logic flow more difficult to follow.
I also noticed that many commands were effectively a sequence of
objects to be CBOR encoded.

I think it makes sense to define version 2 commands as generators.
This way, commands can simply emit the data structures they wish to
send to the client. This eliminates the need for a closure in
command functions and removes encoding from the bodies of commands.

As part of this commit, the handling of response objects has been
moved into the serverreactor class. This puts the reactor in the
driver's seat with regards to CBOR encoding and error handling.
Having error handling in the function that emits frames is
particularly important because exceptions in that function can lead
to things getting in a bad state: I'm fairly certain that uncaught
exceptions in the frame generator were causing deadlocks.

I also introduced a dedicated error type for explicit error reporting
in command handlers. This will be used in subsequent commits.

There's still a bit of work to be done here, especially around
formalizing the error handling "protocol." I've added yet another
TODO to track this so we don't forget.

Test output changed because we're using generators and no longer know
we are at the end of the data until we hit the end of the generator.
This means we can't emit the end-of-stream flag until we've exhausted
the generator. Hence the introduction of 0-sized end-of-stream frames.

Differential Revision: https://phab.mercurial-scm.org/D4472

mercurial/error.py

@@ -308,3 +308,14 b' class PeerTransportError(Abort):'
 class InMemoryMergeConflictsError(Exception):
     """Exception raised when merge conflicts arose during an in-memory merge."""
     __bytes__ = _tobytes
+
+class WireprotoCommandError(Exception):
+    """Represents an error during execution of a wire protocol command.
+
+    Should only be thrown by wire protocol version 2 commands.
+
+    The error is a formatter string and an optional iterable of arguments.
+    """
+    def __init__(self, message, args=None):
+        self.message = message
+        self.messageargs = args

mercurial/help/internals/wireprotocolrpc.txt

@@ -514,3 +514,6 b' error (optional)'
    message
       (array of maps) A message describing the error. The message uses the
       same format as those in the ``Human Output Side-Channel`` frame.
+
+TODO formalize when error frames can be seen and how errors can be
+recognized midway through a command response.

mercurial/wireprotoframing.py

@@ -388,16 +388,14 b' def createcommandresponseframesfrombytes'
 
 def createbytesresponseframesfromgen(stream, requestid, gen,
                                      maxframesize=DEFAULT_MAX_FRAME_SIZE):
-    overall = b''.join(cborutil.streamencode({b'status': b'ok'}))
+    """Generator of frames from a generator of byte chunks.
 
-    yield stream.makeframe(requestid=requestid,
-                           typeid=FRAME_TYPE_COMMAND_RESPONSE,
-                           flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
-                           payload=overall)
-
+    This assumes that another frame will follow whatever this emits. i.e.
+    this always emits the continuation flag and never emits the end-of-stream
+    flag.
+    """
     cb = util.chunkbuffer(gen)
-
-    flags = 0
+    flags = FLAG_COMMAND_RESPONSE_CONTINUATION
 
     while True:
         chunk = cb.read(maxframesize)
@@ -411,11 +409,19 b' def createbytesresponseframesfromgen(str'
 
         flags |= FLAG_COMMAND_RESPONSE_CONTINUATION
 
-    flags ^= FLAG_COMMAND_RESPONSE_CONTINUATION
-    flags |= FLAG_COMMAND_RESPONSE_EOS
-    yield stream.makeframe(requestid=requestid,
+def createcommandresponseokframe(stream, requestid):
+    overall = b''.join(cborutil.streamencode({b'status': b'ok'}))
+
+    return stream.makeframe(requestid=requestid,
                            typeid=FRAME_TYPE_COMMAND_RESPONSE,
-                           flags=flags,
+                            flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
+                            payload=overall)
+
+def createcommandresponseeosframe(stream, requestid):
+    """Create an empty payload frame representing command end-of-stream."""
+    return stream.makeframe(requestid=requestid,
+                            typeid=FRAME_TYPE_COMMAND_RESPONSE,
+                            flags=FLAG_COMMAND_RESPONSE_EOS,
                            payload=b'')
 
 def createcommanderrorresponse(stream, requestid, message, args=None):
@@ -686,15 +692,70 b' class serverreactor(object):'
                 'framegen': result,
             }
 
-    def oncommandresponsereadygen(self, stream, requestid, gen):
-        """Signal that a bytes response is ready, with data as a generator."""
+    def oncommandresponsereadyobjects(self, stream, requestid, objs):
+        """Signal that objects are ready to be sent to the client.
+
+        ``objs`` is an iterable of objects (typically a generator) that will
+        be encoded via CBOR and added to frames, which will be sent to the
+        client.
+        """
         ensureserverstream(stream)
 
+        # We need to take care over exception handling. Uncaught exceptions
+        # when generating frames could lead to premature end of the frame
+        # stream and the possibility of the server or client process getting
+        # in a bad state.
+        #
+        # Keep in mind that if ``objs`` is a generator, advancing it could
+        # raise exceptions that originated in e.g. wire protocol command
+        # functions. That is why we differentiate between exceptions raised
+        # when iterating versus other exceptions that occur.
+        #
+        # In all cases, when the function finishes, the request is fully
+        # handled and no new frames for it should be seen.
+
         def sendframes():
-            for frame in createbytesresponseframesfromgen(stream, requestid,
-                                                          gen):
+            emitted = False
+            while True:
+                try:
+                    o = next(objs)
+                except StopIteration:
+                    if emitted:
+                        yield createcommandresponseeosframe(stream, requestid)
+                    break
+
+                except error.WireprotoCommandError as e:
+                    for frame in createcommanderrorresponse(
+                        stream, requestid, e.message, e.messageargs):
+                        yield frame
+                    break
+
+                except Exception as e:
+                    for frame in createerrorframe(stream, requestid,
+                                                  '%s' % e,
+                                                  errtype='server'):
                 yield frame
 
+                    break
+
+                try:
+                    if not emitted:
+                        yield createcommandresponseokframe(stream, requestid)
+                        emitted = True
+
+                    # TODO buffer chunks so emitted frame payloads can be
+                    # larger.
+                    for frame in createbytesresponseframesfromgen(
+                        stream, requestid, cborutil.streamencode(o)):
+                        yield frame
+                except Exception as e:
+                    for frame in createerrorframe(stream, requestid,
+                                                  '%s' % e,
+                                                  errtype='server'):
+                        yield frame
+
+                    break
+
             self._activecommands.remove(requestid)
 
         return self._handlesendframes(sendframes())

mercurial/wireprototypes.py

@@ -106,27 +106,6 b' class streamreslegacy(object):'
     def __init__(self, gen=None):
         self.gen = gen
 
-class cborresponse(object):
-    """Encode the response value as CBOR."""
-    def __init__(self, v):
-        self.value = v
-
-class v2errorresponse(object):
-    """Represents a command error for version 2 transports."""
-    def __init__(self, message, args=None):
-        self.message = message
-        self.args = args
-
-class v2streamingresponse(object):
-    """A response whose data is supplied by a generator.
-
-    The generator can either consist of data structures to CBOR
-    encode or a stream of already-encoded bytes.
-    """
-    def __init__(self, gen, compressible=True):
-        self.gen = gen
-        self.compressible = compressible
-
 # list of nodes encoding / decoding
 def decodelist(l, sep=' '):
     if l:

mercurial/wireprotov2peer.py

@@ -124,6 +124,8 b' class clienthandler(object):'
             else:
                 raise e
 
+            return
+
         if frame.requestid not in self._requests:
             raise error.ProgrammingError(
                 'received frame for unknown request; this is either a bug in '

mercurial/wireprotov2server.py

@@ -19,7 +19,6 b' from . import ('
     wireprototypes,
 )
 from .utils import (
-    cborutil,
     interfaceutil,
 )
 
@@ -295,31 +294,19 b' def _httpv2runcommand(ui, repo, req, res'
             res.setbodybytes(_('command in frame must match command in URL'))
             return True
 
-    rsp = dispatch(repo, proto, command['command'])
-
     res.status = b'200 OK'
     res.headers[b'Content-Type'] = FRAMINGTYPE
 
-    # TODO consider adding a type to represent an iterable of values to
-    # be CBOR encoded.
-    if isinstance(rsp, wireprototypes.cborresponse):
-        # TODO consider calling oncommandresponsereadygen().
-        encoded = b''.join(cborutil.streamencode(rsp.value))
-        action, meta = reactor.oncommandresponseready(outstream,
-                                                      command['requestid'],
-                                                      encoded)
-    elif isinstance(rsp, wireprototypes.v2streamingresponse):
-        action, meta = reactor.oncommandresponsereadygen(outstream,
-                                                         command['requestid'],
-                                                         rsp.gen)
-    elif isinstance(rsp, wireprototypes.v2errorresponse):
-        action, meta = reactor.oncommanderror(outstream,
-                                              command['requestid'],
-                                              rsp.message,
-                                              rsp.args)
-    else:
+    try:
+        objs = dispatch(repo, proto, command['command'])
+
+        action, meta = reactor.oncommandresponsereadyobjects(
+            outstream, command['requestid'], objs)
+
+    except Exception as e:
         action, meta = reactor.onservererror(
-            _('unhandled response type from wire proto command'))
+            outstream, command['requestid'],
+            _('exception when invoking command: %s') % e)
 
     if action == 'sendframes':
         res.setbodygen(meta['framegen'])
@@ -430,6 +417,12 b' def wireprotocommand(name, args=None, pe'
     respectively. Default is to assume command requires ``push`` permissions
     because otherwise commands not declaring their permissions could modify
     a repository that is supposed to be read-only.
+
+    Wire protocol commands are generators of objects to be serialized and
+    sent to the client.
+
+    If a command raises an uncaught exception, this will be translated into
+    a command error.
     """
     transports = {k for k, v in wireprototypes.TRANSPORTS.items()
                   if v['version'] == 2}
@@ -460,16 +453,12 b' def wireprotocommand(name, args=None, pe'
 
 @wireprotocommand('branchmap', permission='pull')
 def branchmapv2(repo, proto):
-    branchmap = {encoding.fromlocal(k): v
+    yield {encoding.fromlocal(k): v
                  for k, v in repo.branchmap().iteritems()}
 
-    return wireprototypes.cborresponse(branchmap)
-
 @wireprotocommand('capabilities', permission='pull')
 def capabilitiesv2(repo, proto):
-    caps = _capabilitiesv2(repo, proto)
-
-    return wireprototypes.cborresponse(caps)
+    yield _capabilitiesv2(repo, proto)
 
 @wireprotocommand('heads',
                   args={
@@ -480,7 +469,7 b' def headsv2(repo, proto, publiconly=Fals'
     if publiconly:
         repo = repo.filtered('immutable')
 
-    return wireprototypes.cborresponse(repo.heads())
+    yield repo.heads()
 
 @wireprotocommand('known',
                   args={
@@ -490,7 +479,7 b' def headsv2(repo, proto, publiconly=Fals'
 def knownv2(repo, proto, nodes=None):
     nodes = nodes or []
     result = b''.join(b'1' if n else b'0' for n in repo.known(nodes))
-    return wireprototypes.cborresponse(result)
+    yield result
 
 @wireprotocommand('listkeys',
                   args={
@@ -502,7 +491,7 b' def listkeysv2(repo, proto, namespace=No'
     keys = {encoding.fromlocal(k): encoding.fromlocal(v)
             for k, v in keys.iteritems()}
 
-    return wireprototypes.cborresponse(keys)
+    yield keys
 
 @wireprotocommand('lookup',
                   args={
@@ -515,7 +504,7 b' def lookupv2(repo, proto, key):'
     # TODO handle exception.
     node = repo.lookup(key)
 
-    return wireprototypes.cborresponse(node)
+    yield node
 
 @wireprotocommand('pushkey',
                   args={
@@ -527,9 +516,7 b' def lookupv2(repo, proto, key):'
                   permission='push')
 def pushkeyv2(repo, proto, namespace, key, old, new):
     # TODO handle ui output redirection
-    r = repo.pushkey(encoding.tolocal(namespace),
+    yield repo.pushkey(encoding.tolocal(namespace),
                      encoding.tolocal(key),
                      encoding.tolocal(old),
                      encoding.tolocal(new))
-
-    return wireprototypes.cborresponse(r)

tests/test-http-api-httpv2.t

@@ -176,8 +176,14 b' Request to read-only command works out o'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     32\r\n
-  s>     *\x00\x00\x01\x00\x02\x012\xa1FstatusBokX\x1dcustomreadonly bytes response
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011\xa1FstatusBok
+  s>     \r\n
+  s>     27\r\n
+  s>     \x1f\x00\x00\x01\x00\x02\x001X\x1dcustomreadonly bytes response
+  s>     \r\n
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
   s>     \r\n
   s>     0\r\n
   s>     \r\n
@@ -203,13 +209,22 b' Request to read-only command works out o'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     32\r\n
-  s>     *\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBokX\x1dcustomreadonly bytes response
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=42; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     27\r\n
+  s>     \x1f\x00\x00\x01\x00\x02\x001
+  s>     X\x1dcustomreadonly bytes response
+  s>     \r\n
+  received frame(size=31; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: [
     {
       b'status': b'ok'
@@ -322,8 +337,14 b' Authorized request for valid read-write '
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     32\r\n
-  s>     *\x00\x00\x01\x00\x02\x012\xa1FstatusBokX\x1dcustomreadonly bytes response
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011\xa1FstatusBok
+  s>     \r\n
+  s>     27\r\n
+  s>     \x1f\x00\x00\x01\x00\x02\x001X\x1dcustomreadonly bytes response
+  s>     \r\n
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
   s>     \r\n
   s>     0\r\n
   s>     \r\n
@@ -445,8 +466,14 b' Multiple requests to regular command URL'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     32\r\n
-  s>     *\x00\x00\x01\x00\x02\x012\xa1FstatusBokX\x1dcustomreadonly bytes response
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011\xa1FstatusBok
+  s>     \r\n
+  s>     27\r\n
+  s>     \x1f\x00\x00\x01\x00\x02\x001X\x1dcustomreadonly bytes response
+  s>     \r\n
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
   s>     \r\n
   s>     0\r\n
   s>     \r\n
@@ -478,11 +505,23 b' Multiple requests to "multirequest" URL '
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     32\r\n
-  s>     *\x00\x00\x01\x00\x02\x012\xa1FstatusBokX\x1dcustomreadonly bytes response
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011\xa1FstatusBok
+  s>     \r\n
+  s>     27\r\n
+  s>     \x1f\x00\x00\x01\x00\x02\x001X\x1dcustomreadonly bytes response
+  s>     \r\n
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
   s>     \r\n
-  s>     32\r\n
-  s>     *\x00\x00\x03\x00\x02\x002\xa1FstatusBokX\x1dcustomreadonly bytes response
+  s>     13\r\n
+  s>     \x0b\x00\x00\x03\x00\x02\x001\xa1FstatusBok
+  s>     \r\n
+  s>     27\r\n
+  s>     \x1f\x00\x00\x03\x00\x02\x001X\x1dcustomreadonly bytes response
+  s>     \r\n
+  s>     8\r\n
+  s>     \x00\x00\x00\x03\x00\x02\x002
   s>     \r\n
   s>     0\r\n
   s>     \r\n
@@ -516,11 +555,23 b' Interleaved requests to "multirequest" a'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     33\r\n
-  s>     +\x00\x00\x03\x00\x02\x012\xa1FstatusBok\xa3Ibookmarks@Jnamespaces@Fphases@
+  s>     13\r\n
+  s>     \x0b\x00\x00\x03\x00\x02\x011\xa1FstatusBok
+  s>     \r\n
+  s>     28\r\n
+  s>      \x00\x00\x03\x00\x02\x001\xa3Ibookmarks@Jnamespaces@Fphases@
+  s>     \r\n
+  s>     8\r\n
+  s>     \x00\x00\x00\x03\x00\x02\x002
   s>     \r\n
-  s>     14\r\n
-  s>     \x0c\x00\x00\x01\x00\x02\x002\xa1FstatusBok\xa0
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x001\xa1FstatusBok
+  s>     \r\n
+  s>     9\r\n
+  s>     \x01\x00\x00\x01\x00\x02\x001\xa0
+  s>     \r\n
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
   s>     \r\n
   s>     0\r\n
   s>     \r\n

tests/test-http-protocol.t

@@ -331,13 +331,22 b' Client with HTTPv2 enabled automatically'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     29\r\n
-  s>     !\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\x81T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=33; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     1e\r\n
+  s>     \x16\x00\x00\x01\x00\x02\x001
+  s>     \x81T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00
+  s>     \r\n
+  received frame(size=22; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: [
     b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
   ]

tests/test-wireproto-command-branchmap.t

@@ -59,14 +59,23 b' No arguments returns something reasonabl'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     83\r\n
-  s>     {\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\xa3Gbranch1\x81T\xb5\xfa\xac\xdf\xd2c7h\xcb1R3l\xc0\x953\x81&f\x88Gbranch2\x81T"Aa\xc7X\x9a\xa4\x8f\xa8:H\xfe\xff^\x95\xb5j\xe3\'\xfcGdefault\x82T&\x80Z\xba\x1e`\n
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
+  s>     \r\n
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     78\r\n
+  s>     p\x00\x00\x01\x00\x02\x001
+  s>     \xa3Gbranch1\x81T\xb5\xfa\xac\xdf\xd2c7h\xcb1R3l\xc0\x953\x81&f\x88Gbranch2\x81T"Aa\xc7X\x9a\xa4\x8f\xa8:H\xfe\xff^\x95\xb5j\xe3\'\xfcGdefault\x82T&\x80Z\xba\x1e`\n
   s>     \x82\xe96a\x14\x9f#\x13\x86j"\x1a{T\xbe\x0e\xf7<\x17\xad\xe3\xfc\x89\xdcAp\x1e\xb9\xfc:\x91\xb5\x82\x82
   s>     \r\n
-  received frame(size=123; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=112; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: {
     b'branch1': [
       b'\xb5\xfa\xac\xdf\xd2c7h\xcb1R3l\xc0\x953\x81&f\x88'

tests/test-wireproto-command-capabilities.t

@@ -333,13 +333,22 b' capabilities command returns expected in'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     1d7\r\n
-  s>     \xcf\x01\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\xa4Hcommands\xa7Ibranchmap\xa2Dargs\xa0Kpermissions\x81DpullLcapabilities\xa2Dargs\xa0Kpermissions\x81DpullEheads\xa2Dargs\xa1Jpubliconly\xf4Kpermissions\x81DpullEknown\xa2Dargs\xa1Enodes\x81HdeadbeefKpermissions\x81DpullHlistkeys\xa2Dargs\xa1InamespaceBnsKpermissions\x81DpullFlookup\xa2Dargs\xa1CkeyCfooKpermissions\x81DpullGpushkey\xa2Dargs\xa4CkeyCkeyInamespaceBnsCnewCnewColdColdKpermissions\x81DpushKcompression\x81\xa1DnameDzlibQframingmediatypes\x81X&application/mercurial-exp-framing-0005Nrawrepoformats\x82LgeneraldeltaHrevlogv1
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=463; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     1cc\r\n
+  s>     \xc4\x01\x00\x01\x00\x02\x001
+  s>     \xa4Hcommands\xa7Ibranchmap\xa2Dargs\xa0Kpermissions\x81DpullLcapabilities\xa2Dargs\xa0Kpermissions\x81DpullEheads\xa2Dargs\xa1Jpubliconly\xf4Kpermissions\x81DpullEknown\xa2Dargs\xa1Enodes\x81HdeadbeefKpermissions\x81DpullHlistkeys\xa2Dargs\xa1InamespaceBnsKpermissions\x81DpullFlookup\xa2Dargs\xa1CkeyCfooKpermissions\x81DpullGpushkey\xa2Dargs\xa4CkeyCkeyInamespaceBnsCnewCnewColdColdKpermissions\x81DpushKcompression\x81\xa1DnameDzlibQframingmediatypes\x81X&application/mercurial-exp-framing-0005Nrawrepoformats\x82LgeneraldeltaHrevlogv1
+  s>     \r\n
+  received frame(size=452; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: [
     {
       b'status': b'ok'

tests/test-wireproto-command-heads.t

@@ -51,13 +51,22 b' All non-secret heads returned by default'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     53\r\n
-  s>     K\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\x83T\x1dok\x91\xd4J\xab\xa6\xd5\xe5\x80\xbc0\xa9\x94\x850\xdb\xe0\x0bT\xaeI.6\xb0\xc83\x9f\xfa\xf3(\xd0\x0b\x85\xb4R]\xe1\x16^T)Dm-\xc5A\x9c_\x97Dz\x8b\xc0b\xe4\xcc2\x8b\xf2A
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=75; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     48\r\n
+  s>     @\x00\x00\x01\x00\x02\x001
+  s>     \x83T\x1dok\x91\xd4J\xab\xa6\xd5\xe5\x80\xbc0\xa9\x94\x850\xdb\xe0\x0bT\xaeI.6\xb0\xc83\x9f\xfa\xf3(\xd0\x0b\x85\xb4R]\xe1\x16^T)Dm-\xc5A\x9c_\x97Dz\x8b\xc0b\xe4\xcc2\x8b\xf2A
+  s>     \r\n
+  received frame(size=64; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: [
     b'\x1dok\x91\xd4J\xab\xa6\xd5\xe5\x80\xbc0\xa9\x94\x850\xdb\xe0\x0b',
     b'\xaeI.6\xb0\xc83\x9f\xfa\xf3(\xd0\x0b\x85\xb4R]\xe1\x16^',
@@ -88,13 +97,22 b' Requesting just the public heads works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     29\r\n
-  s>     !\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\x81Tx\xd2\xdc\xa46\xb2\xf5\xb1\x88\xac&~)\xb8\x1e\x07&m8\xfc
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=33; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     1e\r\n
+  s>     \x16\x00\x00\x01\x00\x02\x001
+  s>     \x81Tx\xd2\xdc\xa46\xb2\xf5\xb1\x88\xac&~)\xb8\x1e\x07&m8\xfc
+  s>     \r\n
+  received frame(size=22; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: [
     b'x\xd2\xdc\xa46\xb2\xf5\xb1\x88\xac&~)\xb8\x1e\x07&m8\xfc'
   ]

tests/test-wireproto-command-known.t

@@ -43,13 +43,22 b' No arguments returns something reasonabl'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     14\r\n
-  s>     \x0c\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok@
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=12; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     9\r\n
+  s>     \x01\x00\x00\x01\x00\x02\x001
+  s>     @
+  s>     \r\n
+  received frame(size=1; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: []
 
 Single known node works
@@ -76,13 +85,22 b' Single known node works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     15\r\n
-  s>     \r\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBokA1
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=13; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     a\r\n
+  s>     \x02\x00\x00\x01\x00\x02\x001
+  s>     A1
+  s>     \r\n
+  received frame(size=2; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: [
     True
   ]
@@ -111,13 +129,22 b' Multiple nodes works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     17\r\n
-  s>     \x0f\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBokC101
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=15; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     c\r\n
+  s>     \x04\x00\x00\x01\x00\x02\x001
+  s>     C101
+  s>     \r\n
+  received frame(size=4; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: [
     True,
     False,

tests/test-wireproto-command-listkeys.t

@@ -47,13 +47,22 b' Request for namespaces works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     33\r\n
-  s>     +\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\xa3Ibookmarks@Jnamespaces@Fphases@
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=43; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     28\r\n
+  s>      \x00\x00\x01\x00\x02\x001
+  s>     \xa3Ibookmarks@Jnamespaces@Fphases@
+  s>     \r\n
+  received frame(size=32; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: {
     b'bookmarks': b'',
     b'namespaces': b'',
@@ -84,13 +93,22 b' Request for phases works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     50\r\n
-  s>     H\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\xa2X(be0ef73c17ade3fc89dc41701eb9fc3a91b58282A1JpublishingDTrue
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=72; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     45\r\n
+  s>     =\x00\x00\x01\x00\x02\x001
+  s>     \xa2X(be0ef73c17ade3fc89dc41701eb9fc3a91b58282A1JpublishingDTrue
+  s>     \r\n
+  received frame(size=61; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: {
     b'be0ef73c17ade3fc89dc41701eb9fc3a91b58282': b'1',
     b'publishing': b'True'
@@ -120,13 +138,22 b' Request for bookmarks works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     40\r\n
-  s>     8\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\xa1A@X(26805aba1e600a82e93661149f2313866a221a7b
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=56; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     35\r\n
+  s>     -\x00\x00\x01\x00\x02\x001
+  s>     \xa1A@X(26805aba1e600a82e93661149f2313866a221a7b
+  s>     \r\n
+  received frame(size=45; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: {
     b'@': b'26805aba1e600a82e93661149f2313866a221a7b'
   }

tests/test-wireproto-command-lookup.t

@@ -43,13 +43,22 b' lookup for known node works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     28\r\n
-  s>      \x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBokTBk\xad\xa5\xc6u\x98\xcae\x03mW\xd9\xe4\xb6K\x0c\x1c\xe7\xa0
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=32; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     1d\r\n
+  s>     \x15\x00\x00\x01\x00\x02\x001
+  s>     TBk\xad\xa5\xc6u\x98\xcae\x03mW\xd9\xe4\xb6K\x0c\x1c\xe7\xa0
+  s>     \r\n
+  received frame(size=21; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: b'Bk\xad\xa5\xc6u\x98\xcae\x03mW\xd9\xe4\xb6K\x0c\x1c\xe7\xa0'
 
   $ cat error.log

tests/test-wireproto-command-pushkey.t

@@ -46,13 +46,22 b' pushkey for a bookmark works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     14\r\n
-  s>     \x0c\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\xf5
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=12; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     9\r\n
+  s>     \x01\x00\x00\x01\x00\x02\x001
+  s>     \xf5
+  s>     \r\n
+  received frame(size=1; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: True
 
   $ sendhttpv2peer << EOF
@@ -77,13 +86,22 b' pushkey for a bookmark works'
   s>     Content-Type: application/mercurial-exp-framing-0005\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
-  s>     40\r\n
-  s>     8\x00\x00\x01\x00\x02\x012
-  s>     \xa1FstatusBok\xa1A@X(426bada5c67598ca65036d57d9e4b64b0c1ce7a0
+  s>     13\r\n
+  s>     \x0b\x00\x00\x01\x00\x02\x011
+  s>     \xa1FstatusBok
   s>     \r\n
-  received frame(size=56; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=eos)
+  received frame(size=11; request=1; stream=2; streamflags=stream-begin; type=command-response; flags=continuation)
+  s>     35\r\n
+  s>     -\x00\x00\x01\x00\x02\x001
+  s>     \xa1A@X(426bada5c67598ca65036d57d9e4b64b0c1ce7a0
+  s>     \r\n
+  received frame(size=45; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
+  s>     8\r\n
+  s>     \x00\x00\x00\x01\x00\x02\x002
+  s>     \r\n
   s>     0\r\n
   s>     \r\n
+  received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
   response: {
     b'@': b'426bada5c67598ca65036d57d9e4b64b0c1ce7a0'
   }

tests/wireprotohelpers.sh

@@ -26,7 +26,7 b' def customreadonlyv1(repo, proto):'
 
 @wireprotov2server.wireprotocommand(b'customreadonly', permission=b'pull')
 def customreadonlyv2(repo, proto):
-    return wireprototypes.cborresponse(b'customreadonly bytes response')
+    yield b'customreadonly bytes response'
 
 @wireprotov1server.wireprotocommand(b'customreadwrite', permission=b'push')
 def customreadwrite(repo, proto):
@@ -34,7 +34,7 b' def customreadwrite(repo, proto):'
 
 @wireprotov2server.wireprotocommand(b'customreadwrite', permission=b'push')
 def customreadwritev2(repo, proto):
-    return wireprototypes.cborresponse(b'customreadwrite bytes response')
+    yield b'customreadwrite bytes response'
 EOF
 
 cat >> $HGRCPATH << EOF