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

@@ -1,310 +1,321 b''
 # error.py - Mercurial exceptions
 #
 # Copyright 2005-2008 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 """Mercurial exceptions.
 
 This allows us to catch exceptions at higher levels without forcing
 imports.
 """
 
 from __future__ import absolute_import
 
 # Do not import anything but pycompat here, please
 from . import pycompat
 
 def _tobytes(exc):
     """Byte-stringify exception in the same way as BaseException_str()"""
     if not exc.args:
         return b''
     if len(exc.args) == 1:
         return pycompat.bytestr(exc.args[0])
     return b'(%s)' % b', '.join(b"'%s'" % pycompat.bytestr(a) for a in exc.args)
 
 class Hint(object):
     """Mix-in to provide a hint of an error
 
     This should come first in the inheritance list to consume a hint and
     pass remaining arguments to the exception class.
     """
     def __init__(self, *args, **kw):
         self.hint = kw.pop(r'hint', None)
         super(Hint, self).__init__(*args, **kw)
 
 class RevlogError(Hint, Exception):
     __bytes__ = _tobytes
 
 class FilteredIndexError(IndexError):
     __bytes__ = _tobytes
 
 class LookupError(RevlogError, KeyError):
     def __init__(self, name, index, message):
         self.name = name
         self.index = index
         # this can't be called 'message' because at least some installs of
         # Python 2.6+ complain about the 'message' property being deprecated
         self.lookupmessage = message
         if isinstance(name, bytes) and len(name) == 20:
             from .node import short
             name = short(name)
         RevlogError.__init__(self, '%s@%s: %s' % (index, name, message))
 
     def __bytes__(self):
         return RevlogError.__bytes__(self)
 
     def __str__(self):
         return RevlogError.__str__(self)
 
 class AmbiguousPrefixLookupError(LookupError):
     pass
 
 class FilteredLookupError(LookupError):
     pass
 
 class ManifestLookupError(LookupError):
     pass
 
 class CommandError(Exception):
     """Exception raised on errors in parsing the command line."""
     __bytes__ = _tobytes
 
 class InterventionRequired(Hint, Exception):
     """Exception raised when a command requires human intervention."""
     __bytes__ = _tobytes
 
 class Abort(Hint, Exception):
     """Raised if a command needs to print an error and exit."""
     __bytes__ = _tobytes
 
 class HookLoadError(Abort):
     """raised when loading a hook fails, aborting an operation
 
     Exists to allow more specialized catching."""
 
 class HookAbort(Abort):
     """raised when a validation hook fails, aborting an operation
 
     Exists to allow more specialized catching."""
 
 class ConfigError(Abort):
     """Exception raised when parsing config files"""
 
 class UpdateAbort(Abort):
     """Raised when an update is aborted for destination issue"""
 
 class MergeDestAbort(Abort):
     """Raised when an update is aborted for destination issues"""
 
 class NoMergeDestAbort(MergeDestAbort):
     """Raised when an update is aborted because there is nothing to merge"""
 
 class ManyMergeDestAbort(MergeDestAbort):
     """Raised when an update is aborted because destination is ambiguous"""
 
 class ResponseExpected(Abort):
     """Raised when an EOF is received for a prompt"""
     def __init__(self):
         from .i18n import _
         Abort.__init__(self, _('response expected'))
 
 class OutOfBandError(Hint, Exception):
     """Exception raised when a remote repo reports failure"""
     __bytes__ = _tobytes
 
 class ParseError(Hint, Exception):
     """Raised when parsing config files and {rev,file}sets (msg[, pos])"""
     __bytes__ = _tobytes
 
 class PatchError(Exception):
     __bytes__ = _tobytes
 
 class UnknownIdentifier(ParseError):
     """Exception raised when a {rev,file}set references an unknown identifier"""
 
     def __init__(self, function, symbols):
         from .i18n import _
         ParseError.__init__(self, _("unknown identifier: %s") % function)
         self.function = function
         self.symbols = symbols
 
 class RepoError(Hint, Exception):
     __bytes__ = _tobytes
 
 class RepoLookupError(RepoError):
     pass
 
 class FilteredRepoLookupError(RepoLookupError):
     pass
 
 class CapabilityError(RepoError):
     pass
 
 class RequirementError(RepoError):
     """Exception raised if .hg/requires has an unknown entry."""
 
 class StdioError(IOError):
     """Raised if I/O to stdout or stderr fails"""
 
     def __init__(self, err):
         IOError.__init__(self, err.errno, err.strerror)
 
     # no __bytes__() because error message is derived from the standard IOError
 
 class UnsupportedMergeRecords(Abort):
     def __init__(self, recordtypes):
         from .i18n import _
         self.recordtypes = sorted(recordtypes)
         s = ' '.join(self.recordtypes)
         Abort.__init__(
             self, _('unsupported merge state records: %s') % s,
             hint=_('see https://mercurial-scm.org/wiki/MergeStateRecords for '
                    'more information'))
 
 class UnknownVersion(Abort):
     """generic exception for aborting from an encounter with an unknown version
     """
 
     def __init__(self, msg, hint=None, version=None):
         self.version = version
         super(UnknownVersion, self).__init__(msg, hint=hint)
 
 class LockError(IOError):
     def __init__(self, errno, strerror, filename, desc):
         IOError.__init__(self, errno, strerror, filename)
         self.desc = desc
 
     # no __bytes__() because error message is derived from the standard IOError
 
 class LockHeld(LockError):
     def __init__(self, errno, filename, desc, locker):
         LockError.__init__(self, errno, 'Lock held', filename, desc)
         self.locker = locker
 
 class LockUnavailable(LockError):
     pass
 
 # LockError is for errors while acquiring the lock -- this is unrelated
 class LockInheritanceContractViolation(RuntimeError):
     __bytes__ = _tobytes
 
 class ResponseError(Exception):
     """Raised to print an error with part of output and exit."""
     __bytes__ = _tobytes
 
 class UnknownCommand(Exception):
     """Exception raised if command is not in the command table."""
     __bytes__ = _tobytes
 
 class AmbiguousCommand(Exception):
     """Exception raised if command shortcut matches more than one command."""
     __bytes__ = _tobytes
 
 # derived from KeyboardInterrupt to simplify some breakout code
 class SignalInterrupt(KeyboardInterrupt):
     """Exception raised on SIGTERM and SIGHUP."""
 
 class SignatureError(Exception):
     __bytes__ = _tobytes
 
 class PushRaced(RuntimeError):
     """An exception raised during unbundling that indicate a push race"""
     __bytes__ = _tobytes
 
 class ProgrammingError(Hint, RuntimeError):
     """Raised if a mercurial (core or extension) developer made a mistake"""
     __bytes__ = _tobytes
 
 class WdirUnsupported(Exception):
     """An exception which is raised when 'wdir()' is not supported"""
     __bytes__ = _tobytes
 
 # bundle2 related errors
 class BundleValueError(ValueError):
     """error raised when bundle2 cannot be processed"""
     __bytes__ = _tobytes
 
 class BundleUnknownFeatureError(BundleValueError):
     def __init__(self, parttype=None, params=(), values=()):
         self.parttype = parttype
         self.params = params
         self.values = values
         if self.parttype is None:
             msg = 'Stream Parameter'
         else:
             msg = parttype
         entries = self.params
         if self.params and self.values:
             assert len(self.params) == len(self.values)
             entries = []
             for idx, par in enumerate(self.params):
                 val = self.values[idx]
                 if val is None:
                     entries.append(val)
                 else:
                     entries.append("%s=%r" % (par, pycompat.maybebytestr(val)))
         if entries:
             msg = '%s - %s' % (msg, ', '.join(entries))
         ValueError.__init__(self, msg)
 
 class ReadOnlyPartError(RuntimeError):
     """error raised when code tries to alter a part being generated"""
     __bytes__ = _tobytes
 
 class PushkeyFailed(Abort):
     """error raised when a pushkey part failed to update a value"""
 
     def __init__(self, partid, namespace=None, key=None, new=None, old=None,
                  ret=None):
         self.partid = partid
         self.namespace = namespace
         self.key = key
         self.new = new
         self.old = old
         self.ret = ret
         # no i18n expected to be processed into a better message
         Abort.__init__(self, 'failed to update value for "%s/%s"'
                        % (namespace, key))
 
 class CensoredNodeError(RevlogError):
     """error raised when content verification fails on a censored node
 
     Also contains the tombstone data substituted for the uncensored data.
     """
 
     def __init__(self, filename, node, tombstone):
         from .node import short
         RevlogError.__init__(self, '%s:%s' % (filename, short(node)))
         self.tombstone = tombstone
 
 class CensoredBaseError(RevlogError):
     """error raised when a delta is rejected because its base is censored
 
     A delta based on a censored revision must be formed as single patch
     operation which replaces the entire base with new content. This ensures
     the delta may be applied by clones which have not censored the base.
     """
 
 class InvalidBundleSpecification(Exception):
     """error raised when a bundle specification is invalid.
 
     This is used for syntax errors as opposed to support errors.
     """
     __bytes__ = _tobytes
 
 class UnsupportedBundleSpecification(Exception):
     """error raised when a bundle specification is not supported."""
     __bytes__ = _tobytes
 
 class CorruptedState(Exception):
     """error raised when a command is not able to read its state from file"""
     __bytes__ = _tobytes
 
 class PeerTransportError(Abort):
     """Transport-level I/O error when communicating with a peer repo."""
 
 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

@@ -1,516 +1,519 b''
 **Experimental and under development**
 
 This document describe's Mercurial's transport-agnostic remote procedure
 call (RPC) protocol which is used to perform interactions with remote
 servers. This protocol is also referred to as ``hgrpc``.
 
 The protocol has the following high-level features:
 
 * Concurrent request and response support (multiple commands can be issued
   simultaneously and responses can be streamed simultaneously).
 * Supports half-duplex and full-duplex connections.
 * All data is transmitted within *frames*, which have a well-defined
   header and encode their length.
 * Side-channels for sending progress updates and printing output. Text
   output from the remote can be localized locally.
 * Support for simultaneous and long-lived compression streams, even across
   requests.
 * Uses CBOR for data exchange.
 
 The protocol is not specific to Mercurial and could be used by other
 applications.
 
 High-level Overview
 ===================
 
 To operate the protocol, a bi-directional, half-duplex pipe supporting
 ordered sends and receives is required. That is, each peer has one pipe
 for sending data and another for receiving. Full-duplex pipes are also
 supported.
 
 All data is read and written in atomic units called *frames*. These
 are conceptually similar to TCP packets. Higher-level functionality
 is built on the exchange and processing of frames.
 
 All frames are associated with a *stream*. A *stream* provides a
 unidirectional grouping of frames. Streams facilitate two goals:
 content encoding and parallelism. There is a dedicated section on
 streams below.
 
 The protocol is request-response based: the client issues requests to
 the server, which issues replies to those requests. Server-initiated
 messaging is not currently supported, but this specification carves
 out room to implement it.
 
 All frames are associated with a numbered request. Frames can thus
 be logically grouped by their request ID.
 
 Frames
 ======
 
 Frames begin with an 8 octet header followed by a variable length
 payload::
 
     +------------------------------------------------+
     |                 Length (24)                    |
     +--------------------------------+---------------+
     |         Request ID (16)        | Stream ID (8) |
     +------------------+-------------+---------------+
     | Stream Flags (8) |
     +-----------+------+
     | Type (4)  |
     +-----------+
     | Flags (4) |
     +===========+===================================================|
     |                     Frame Payload (0...)                    ...
     +---------------------------------------------------------------+
 
 The length of the frame payload is expressed as an unsigned 24 bit
 little endian integer. Values larger than 65535 MUST NOT be used unless
 given permission by the server as part of the negotiated capabilities
 during the handshake. The frame header is not part of the advertised
 frame length. The payload length is the over-the-wire length. If there
 is content encoding applied to the payload as part of the frame's stream,
 the length is the output of that content encoding, not the input.
 
 The 16-bit ``Request ID`` field denotes the integer request identifier,
 stored as an unsigned little endian integer. Odd numbered requests are
 client-initiated. Even numbered requests are server-initiated. This
 refers to where the *request* was initiated - not where the *frame* was
 initiated, so servers will send frames with odd ``Request ID`` in
 response to client-initiated requests. Implementations are advised to
 start ordering request identifiers at ``1`` and ``0``, increment by
 ``2``, and wrap around if all available numbers have been exhausted.
 
 The 8-bit ``Stream ID`` field denotes the stream that the frame is
 associated with. Frames belonging to a stream may have content
 encoding applied and the receiver may need to decode the raw frame
 payload to obtain the original data. Odd numbered IDs are
 client-initiated. Even numbered IDs are server-initiated.
 
 The 8-bit ``Stream Flags`` field defines stream processing semantics.
 See the section on streams below.
 
 The 4-bit ``Type`` field denotes the type of frame being sent.
 
 The 4-bit ``Flags`` field defines special, per-type attributes for
 the frame.
 
 The sections below define the frame types and their behavior.
 
 Command Request (``0x01``)
 --------------------------
 
 This frame contains a request to run a command.
 
 The payload consists of a CBOR map defining the command request. The
 bytestring keys of that map are:
 
 name
    Name of the command that should be executed (bytestring).
 args
    Map of bytestring keys to various value types containing the named
    arguments to this command.
 
    Each command defines its own set of argument names and their expected
    types.
 
 This frame type MUST ONLY be sent from clients to servers: it is illegal
 for a server to send this frame to a client.
 
 The following flag values are defined for this type:
 
 0x01
    New command request. When set, this frame represents the beginning
    of a new request to run a command. The ``Request ID`` attached to this
    frame MUST NOT be active.
 0x02
    Command request continuation. When set, this frame is a continuation
    from a previous command request frame for its ``Request ID``. This
    flag is set when the CBOR data for a command request does not fit
    in a single frame.
 0x04
    Additional frames expected. When set, the command request didn't fit
    into a single frame and additional CBOR data follows in a subsequent
    frame.
 0x08
    Command data frames expected. When set, command data frames are
    expected to follow the final command request frame for this request.
 
 ``0x01`` MUST be set on the initial command request frame for a
 ``Request ID``.
 
 ``0x01`` or ``0x02`` MUST be set to indicate this frame's role in
 a series of command request frames.
 
 If command data frames are to be sent, ``0x08`` MUST be set on ALL
 command request frames.
 
 Command Data (``0x02``)
 -----------------------
 
 This frame contains raw data for a command.
 
 Most commands can be executed by specifying arguments. However,
 arguments have an upper bound to their length. For commands that
 accept data that is beyond this length or whose length isn't known
 when the command is initially sent, they will need to stream
 arbitrary data to the server. This frame type facilitates the sending
 of this data.
 
 The payload of this frame type consists of a stream of raw data to be
 consumed by the command handler on the server. The format of the data
 is command specific.
 
 The following flag values are defined for this type:
 
 0x01
    Command data continuation. When set, the data for this command
    continues into a subsequent frame.
 
 0x02
    End of data. When set, command data has been fully sent to the
    server. The command has been fully issued and no new data for this
    command will be sent. The next frame will belong to a new command.
 
 Command Response Data (``0x03``)
 --------------------------------
 
 This frame contains response data to an issued command.
 
 Response data ALWAYS consists of a series of 1 or more CBOR encoded
 values. A CBOR value may be using indefinite length encoding. And the
 bytes constituting the value may span several frames.
 
 The following flag values are defined for this type:
 
 0x01
    Data continuation. When set, an additional frame containing response data
    will follow.
 0x02
    End of data. When set, the response data has been fully sent and
    no additional frames for this response will be sent.
 
 The ``0x01`` flag is mutually exclusive with the ``0x02`` flag.
 
 Error Occurred (``0x05``)
 -------------------------
 
 Some kind of error occurred.
 
 There are 3 general kinds of failures that can occur:
 
 * Command error encountered before any response issued
 * Command error encountered after a response was issued
 * Protocol or stream level error
 
 This frame type is used to capture the latter cases. (The general
 command error case is handled by the leading CBOR map in
 ``Command Response`` frames.)
 
 The payload of this frame contains a CBOR map detailing the error. That
 map has the following bytestring keys:
 
 type
    (bytestring) The overall type of error encountered. Can be one of the
    following values:
 
    protocol
       A protocol-level error occurred. This typically means someone
       is violating the framing protocol semantics and the server is
       refusing to proceed.
 
    server
       A server-level error occurred. This typically indicates some kind of
       logic error on the server, likely the fault of the server.
 
    command
       A command-level error, likely the fault of the client.
 
 message
    (array of maps) A richly formatted message that is intended for
    human consumption. See the ``Human Output Side-Channel`` frame
    section for a description of the format of this data structure.
 
 Human Output Side-Channel (``0x06``)
 ------------------------------------
 
 This frame contains a message that is intended to be displayed to
 people. Whereas most frames communicate machine readable data, this
 frame communicates textual data that is intended to be shown to
 humans.
 
 The frame consists of a series of *formatting requests*. Each formatting
 request consists of a formatting string, arguments for that formatting
 string, and labels to apply to that formatting string.
 
 A formatting string is a printf()-like string that allows variable
 substitution within the string. Labels allow the rendered text to be
 *decorated*. Assuming use of the canonical Mercurial code base, a
 formatting string can be the input to the ``i18n._`` function. This
 allows messages emitted from the server to be localized. So even if
 the server has different i18n settings, people could see messages in
 their *native* settings. Similarly, the use of labels allows
 decorations like coloring and underlining to be applied using the
 client's configured rendering settings.
 
 Formatting strings are similar to ``printf()`` strings or how
 Python's ``%`` operator works. The only supported formatting sequences
 are ``%s`` and ``%%``. ``%s`` will be replaced by whatever the string
 at that position resolves to. ``%%`` will be replaced by ``%``. All
 other 2-byte sequences beginning with ``%`` represent a literal
 ``%`` followed by that character. However, future versions of the
 wire protocol reserve the right to allow clients to opt in to receiving
 formatting strings with additional formatters, hence why ``%%`` is
 required to represent the literal ``%``.
 
 The frame payload consists of a CBOR array of CBOR maps. Each map
 defines an *atom* of text data to print. Each *atom* has the following
 bytestring keys:
 
 msg
    (bytestring) The formatting string. Content MUST be ASCII.
 args (optional)
    Array of bytestrings defining arguments to the formatting string.
 labels (optional)
    Array of bytestrings defining labels to apply to this atom.
 
 All data to be printed MUST be encoded into a single frame: this frame
 does not support spanning data across multiple frames.
 
 All textual data encoded in these frames is assumed to be line delimited.
 The last atom in the frame SHOULD end with a newline (``\n``). If it
 doesn't, clients MAY add a newline to facilitate immediate printing.
 
 Progress Update (``0x07``)
 --------------------------
 
 This frame holds the progress of an operation on the peer. Consumption
 of these frames allows clients to display progress bars, estimated
 completion times, etc.
 
 Each frame defines the progress of a single operation on the peer. The
 payload consists of a CBOR map with the following bytestring keys:
 
 topic
    Topic name (string)
 pos
    Current numeric position within the topic (integer)
 total
    Total/end numeric position of this topic (unsigned integer)
 label (optional)
    Unit label (string)
 item (optional)
    Item name (string)
 
 Progress state is created when a frame is received referencing a
 *topic* that isn't currently tracked. Progress tracking for that
 *topic* is finished when a frame is received reporting the current
 position of that topic as ``-1``.
 
 Multiple *topics* may be active at any given time.
 
 Rendering of progress information is not mandated or governed by this
 specification: implementations MAY render progress information however
 they see fit, including not at all.
 
 The string data describing the topic SHOULD be static strings to
 facilitate receivers localizing that string data. The emitter
 MUST normalize all string data to valid UTF-8 and receivers SHOULD
 validate that received data conforms to UTF-8. The topic name
 SHOULD be ASCII.
 
 Stream Encoding Settings (``0x08``)
 -----------------------------------
 
 This frame type holds information defining the content encoding
 settings for a *stream*.
 
 This frame type is likely consumed by the protocol layer and is not
 passed on to applications.
 
 This frame type MUST ONLY occur on frames having the *Beginning of Stream*
 ``Stream Flag`` set.
 
 The payload of this frame defines what content encoding has (possibly)
 been applied to the payloads of subsequent frames in this stream.
 
 The payload begins with an 8-bit integer defining the length of the
 encoding *profile*, followed by the string name of that profile, which
 must be an ASCII string. All bytes that follow can be used by that
 profile for supplemental settings definitions. See the section below
 on defined encoding profiles.
 
 Stream States and Flags
 =======================
 
 Streams can be in two states: *open* and *closed*. An *open* stream
 is active and frames attached to that stream could arrive at any time.
 A *closed* stream is not active. If a frame attached to a *closed*
 stream arrives, that frame MUST have an appropriate stream flag
 set indicating beginning of stream. All streams are in the *closed*
 state by default.
 
 The ``Stream Flags`` field denotes a set of bit flags for defining
 the relationship of this frame within a stream. The following flags
 are defined:
 
 0x01
    Beginning of stream. The first frame in the stream MUST set this
    flag. When received, the ``Stream ID`` this frame is attached to
    becomes ``open``.
 
 0x02
    End of stream. The last frame in a stream MUST set this flag. When
    received, the ``Stream ID`` this frame is attached to becomes
    ``closed``. Any content encoding context associated with this stream
    can be destroyed after processing the payload of this frame.
 
 0x04
    Apply content encoding. When set, any content encoding settings
    defined by the stream should be applied when attempting to read
    the frame. When not set, the frame payload isn't encoded.
 
 Streams
 =======
 
 Streams - along with ``Request IDs`` - facilitate grouping of frames.
 But the purpose of each is quite different and the groupings they
 constitute are independent.
 
 A ``Request ID`` is essentially a tag. It tells you which logical
 request a frame is associated with.
 
 A *stream* is a sequence of frames grouped for the express purpose
 of applying a stateful encoding or for denoting sub-groups of frames.
 
 Unlike ``Request ID``s which span the request and response, a stream
 is unidirectional and stream IDs are independent from client to
 server.
 
 There is no strict hierarchical relationship between ``Request IDs``
 and *streams*. A stream can contain frames having multiple
 ``Request IDs``. Frames belonging to the same ``Request ID`` can
 span multiple streams.
 
 One goal of streams is to facilitate content encoding. A stream can
 define an encoding to be applied to frame payloads. For example, the
 payload transmitted over the wire may contain output from a
 zstandard compression operation and the receiving end may decompress
 that payload to obtain the original data.
 
 The other goal of streams is to facilitate concurrent execution. For
 example, a server could spawn 4 threads to service a request that can
 be easily parallelized. Each of those 4 threads could write into its
 own stream. Those streams could then in turn be delivered to 4 threads
 on the receiving end, with each thread consuming its stream in near
 isolation. The *main* thread on both ends merely does I/O and
 encodes/decodes frame headers: the bulk of the work is done by worker
 threads.
 
 In addition, since content encoding is defined per stream, each
 *worker thread* could perform potentially CPU bound work concurrently
 with other threads. This approach of applying encoding at the
 sub-protocol / stream level eliminates a potential resource constraint
 on the protocol stream as a whole (it is common for the throughput of
 a compression engine to be smaller than the throughput of a network).
 
 Having multiple streams - each with their own encoding settings - also
 facilitates the use of advanced data compression techniques. For
 example, a transmitter could see that it is generating data faster
 and slower than the receiving end is consuming it and adjust its
 compression settings to trade CPU for compression ratio accordingly.
 
 While streams can define a content encoding, not all frames within
 that stream must use that content encoding. This can be useful when
 data is being served from caches and being derived dynamically. A
 cache could pre-compressed data so the server doesn't have to
 recompress it. The ability to pick and choose which frames are
 compressed allows servers to easily send data to the wire without
 involving potentially expensive encoding overhead.
 
 Content Encoding Profiles
 =========================
 
 Streams can have named content encoding *profiles* associated with
 them. A profile defines a shared understanding of content encoding
 settings and behavior.
 
 The following profiles are defined:
 
 TBD
 
 Command Protocol
 ================
 
 A client can request that a remote run a command by sending it
 frames defining that command. This logical stream is composed of
 1 or more ``Command Request`` frames and and 0 or more ``Command Data``
 frames.
 
 All frames composing a single command request MUST be associated with
 the same ``Request ID``.
 
 Clients MAY send additional command requests without waiting on the
 response to a previous command request. If they do so, they MUST ensure
 that the ``Request ID`` field of outbound frames does not conflict
 with that of an active ``Request ID`` whose response has not yet been
 fully received.
 
 Servers MAY respond to commands in a different order than they were
 sent over the wire. Clients MUST be prepared to deal with this. Servers
 also MAY start executing commands in a different order than they were
 received, or MAY execute multiple commands concurrently.
 
 If there is a dependency between commands or a race condition between
 commands executing (e.g. a read-only command that depends on the results
 of a command that mutates the repository), then clients MUST NOT send
 frames issuing a command until a response to all dependent commands has
 been received.
 TODO think about whether we should express dependencies between commands
 to avoid roundtrip latency.
 
 A command is defined by a command name, 0 or more command arguments,
 and optional command data.
 
 Arguments are the recommended mechanism for transferring fixed sets of
 parameters to a command. Data is appropriate for transferring variable
 data. Thinking in terms of HTTP, arguments would be headers and data
 would be the message body.
 
 It is recommended for servers to delay the dispatch of a command
 until all argument have been received. Servers MAY impose limits on the
 maximum argument size.
 TODO define failure mechanism.
 
 Servers MAY dispatch to commands immediately once argument data
 is available or delay until command data is received in full.
 
 Once a ``Command Request`` frame is sent, a client must be prepared to
 receive any of the following frames associated with that request:
 ``Command Response``, ``Error Response``, ``Human Output Side-Channel``,
 ``Progress Update``.
 
 The *main* response for a command will be in ``Command Response`` frames.
 The payloads of these frames consist of 1 or more CBOR encoded values.
 The first CBOR value on the first ``Command Response`` frame is special
 and denotes the overall status of the command. This CBOR map contains
 the following bytestring keys:
 
 status
    (bytestring) A well-defined message containing the overall status of
    this command request. The following values are defined:
 
    ok
       The command was received successfully and its response follows.
    error
       There was an error processing the command. More details about the
       error are encoded in the ``error`` key.
 
 error (optional)
    A map containing information about an encountered error. The map has the
    following keys:
 
    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

@@ -1,1169 +1,1230 b''
 # wireprotoframing.py - unified framing protocol for wire protocol
 #
 # Copyright 2018 Gregory Szorc <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 # This file contains functionality to support the unified frame-based wire
 # protocol. For details about the protocol, see
 # `hg help internals.wireprotocol`.
 
 from __future__ import absolute_import
 
 import collections
 import struct
 
 from .i18n import _
 from .thirdparty import (
     attr,
 )
 from . import (
     encoding,
     error,
     util,
 )
 from .utils import (
     cborutil,
     stringutil,
 )
 
 FRAME_HEADER_SIZE = 8
 DEFAULT_MAX_FRAME_SIZE = 32768
 
 STREAM_FLAG_BEGIN_STREAM = 0x01
 STREAM_FLAG_END_STREAM = 0x02
 STREAM_FLAG_ENCODING_APPLIED = 0x04
 
 STREAM_FLAGS = {
     b'stream-begin': STREAM_FLAG_BEGIN_STREAM,
     b'stream-end': STREAM_FLAG_END_STREAM,
     b'encoded': STREAM_FLAG_ENCODING_APPLIED,
 }
 
 FRAME_TYPE_COMMAND_REQUEST = 0x01
 FRAME_TYPE_COMMAND_DATA = 0x02
 FRAME_TYPE_COMMAND_RESPONSE = 0x03
 FRAME_TYPE_ERROR_RESPONSE = 0x05
 FRAME_TYPE_TEXT_OUTPUT = 0x06
 FRAME_TYPE_PROGRESS = 0x07
 FRAME_TYPE_STREAM_SETTINGS = 0x08
 
 FRAME_TYPES = {
     b'command-request': FRAME_TYPE_COMMAND_REQUEST,
     b'command-data': FRAME_TYPE_COMMAND_DATA,
     b'command-response': FRAME_TYPE_COMMAND_RESPONSE,
     b'error-response': FRAME_TYPE_ERROR_RESPONSE,
     b'text-output': FRAME_TYPE_TEXT_OUTPUT,
     b'progress': FRAME_TYPE_PROGRESS,
     b'stream-settings': FRAME_TYPE_STREAM_SETTINGS,
 }
 
 FLAG_COMMAND_REQUEST_NEW = 0x01
 FLAG_COMMAND_REQUEST_CONTINUATION = 0x02
 FLAG_COMMAND_REQUEST_MORE_FRAMES = 0x04
 FLAG_COMMAND_REQUEST_EXPECT_DATA = 0x08
 
 FLAGS_COMMAND_REQUEST = {
     b'new': FLAG_COMMAND_REQUEST_NEW,
     b'continuation': FLAG_COMMAND_REQUEST_CONTINUATION,
     b'more': FLAG_COMMAND_REQUEST_MORE_FRAMES,
     b'have-data': FLAG_COMMAND_REQUEST_EXPECT_DATA,
 }
 
 FLAG_COMMAND_DATA_CONTINUATION = 0x01
 FLAG_COMMAND_DATA_EOS = 0x02
 
 FLAGS_COMMAND_DATA = {
     b'continuation': FLAG_COMMAND_DATA_CONTINUATION,
     b'eos': FLAG_COMMAND_DATA_EOS,
 }
 
 FLAG_COMMAND_RESPONSE_CONTINUATION = 0x01
 FLAG_COMMAND_RESPONSE_EOS = 0x02
 
 FLAGS_COMMAND_RESPONSE = {
     b'continuation': FLAG_COMMAND_RESPONSE_CONTINUATION,
     b'eos': FLAG_COMMAND_RESPONSE_EOS,
 }
 
 # Maps frame types to their available flags.
 FRAME_TYPE_FLAGS = {
     FRAME_TYPE_COMMAND_REQUEST: FLAGS_COMMAND_REQUEST,
     FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA,
     FRAME_TYPE_COMMAND_RESPONSE: FLAGS_COMMAND_RESPONSE,
     FRAME_TYPE_ERROR_RESPONSE: {},
     FRAME_TYPE_TEXT_OUTPUT: {},
     FRAME_TYPE_PROGRESS: {},
     FRAME_TYPE_STREAM_SETTINGS: {},
 }
 
 ARGUMENT_RECORD_HEADER = struct.Struct(r'<HH')
 
 def humanflags(mapping, value):
     """Convert a numeric flags value to a human value, using a mapping table."""
     namemap = {v: k for k, v in mapping.iteritems()}
     flags = []
     val = 1
     while value >= val:
         if value & val:
             flags.append(namemap.get(val, '<unknown 0x%02x>' % val))
         val <<= 1
 
     return b'|'.join(flags)
 
 @attr.s(slots=True)
 class frameheader(object):
     """Represents the data in a frame header."""
 
     length = attr.ib()
     requestid = attr.ib()
     streamid = attr.ib()
     streamflags = attr.ib()
     typeid = attr.ib()
     flags = attr.ib()
 
 @attr.s(slots=True, repr=False)
 class frame(object):
     """Represents a parsed frame."""
 
     requestid = attr.ib()
     streamid = attr.ib()
     streamflags = attr.ib()
     typeid = attr.ib()
     flags = attr.ib()
     payload = attr.ib()
 
     @encoding.strmethod
     def __repr__(self):
         typename = '<unknown 0x%02x>' % self.typeid
         for name, value in FRAME_TYPES.iteritems():
             if value == self.typeid:
                 typename = name
                 break
 
         return ('frame(size=%d; request=%d; stream=%d; streamflags=%s; '
                 'type=%s; flags=%s)' % (
             len(self.payload), self.requestid, self.streamid,
             humanflags(STREAM_FLAGS, self.streamflags), typename,
             humanflags(FRAME_TYPE_FLAGS.get(self.typeid, {}), self.flags)))
 
 def makeframe(requestid, streamid, streamflags, typeid, flags, payload):
     """Assemble a frame into a byte array."""
     # TODO assert size of payload.
     frame = bytearray(FRAME_HEADER_SIZE + len(payload))
 
     # 24 bits length
     # 16 bits request id
     # 8 bits stream id
     # 8 bits stream flags
     # 4 bits type
     # 4 bits flags
 
     l = struct.pack(r'<I', len(payload))
     frame[0:3] = l[0:3]
     struct.pack_into(r'<HBB', frame, 3, requestid, streamid, streamflags)
     frame[7] = (typeid << 4) | flags
     frame[8:] = payload
 
     return frame
 
 def makeframefromhumanstring(s):
     """Create a frame from a human readable string
 
     Strings have the form:
 
         <request-id> <stream-id> <stream-flags> <type> <flags> <payload>
 
     This can be used by user-facing applications and tests for creating
     frames easily without having to type out a bunch of constants.
 
     Request ID and stream IDs are integers.
 
     Stream flags, frame type, and flags can be specified by integer or
     named constant.
 
     Flags can be delimited by `|` to bitwise OR them together.
 
     If the payload begins with ``cbor:``, the following string will be
     evaluated as Python literal and the resulting object will be fed into
     a CBOR encoder. Otherwise, the payload is interpreted as a Python
     byte string literal.
     """
     fields = s.split(b' ', 5)
     requestid, streamid, streamflags, frametype, frameflags, payload = fields
 
     requestid = int(requestid)
     streamid = int(streamid)
 
     finalstreamflags = 0
     for flag in streamflags.split(b'|'):
         if flag in STREAM_FLAGS:
             finalstreamflags |= STREAM_FLAGS[flag]
         else:
             finalstreamflags |= int(flag)
 
     if frametype in FRAME_TYPES:
         frametype = FRAME_TYPES[frametype]
     else:
         frametype = int(frametype)
 
     finalflags = 0
     validflags = FRAME_TYPE_FLAGS[frametype]
     for flag in frameflags.split(b'|'):
         if flag in validflags:
             finalflags |= validflags[flag]
         else:
             finalflags |= int(flag)
 
     if payload.startswith(b'cbor:'):
         payload = b''.join(cborutil.streamencode(
             stringutil.evalpythonliteral(payload[5:])))
 
     else:
         payload = stringutil.unescapestr(payload)
 
     return makeframe(requestid=requestid, streamid=streamid,
                      streamflags=finalstreamflags, typeid=frametype,
                      flags=finalflags, payload=payload)
 
 def parseheader(data):
     """Parse a unified framing protocol frame header from a buffer.
 
     The header is expected to be in the buffer at offset 0 and the
     buffer is expected to be large enough to hold a full header.
     """
     # 24 bits payload length (little endian)
     # 16 bits request ID
     # 8 bits stream ID
     # 8 bits stream flags
     # 4 bits frame type
     # 4 bits frame flags
     # ... payload
     framelength = data[0] + 256 * data[1] + 16384 * data[2]
     requestid, streamid, streamflags = struct.unpack_from(r'<HBB', data, 3)
     typeflags = data[7]
 
     frametype = (typeflags & 0xf0) >> 4
     frameflags = typeflags & 0x0f
 
     return frameheader(framelength, requestid, streamid, streamflags,
                        frametype, frameflags)
 
 def readframe(fh):
     """Read a unified framing protocol frame from a file object.
 
     Returns a 3-tuple of (type, flags, payload) for the decoded frame or
     None if no frame is available. May raise if a malformed frame is
     seen.
     """
     header = bytearray(FRAME_HEADER_SIZE)
 
     readcount = fh.readinto(header)
 
     if readcount == 0:
         return None
 
     if readcount != FRAME_HEADER_SIZE:
         raise error.Abort(_('received incomplete frame: got %d bytes: %s') %
                           (readcount, header))
 
     h = parseheader(header)
 
     payload = fh.read(h.length)
     if len(payload) != h.length:
         raise error.Abort(_('frame length error: expected %d; got %d') %
                           (h.length, len(payload)))
 
     return frame(h.requestid, h.streamid, h.streamflags, h.typeid, h.flags,
                  payload)
 
 def createcommandframes(stream, requestid, cmd, args, datafh=None,
                         maxframesize=DEFAULT_MAX_FRAME_SIZE):
     """Create frames necessary to transmit a request to run a command.
 
     This is a generator of bytearrays. Each item represents a frame
     ready to be sent over the wire to a peer.
     """
     data = {b'name': cmd}
     if args:
         data[b'args'] = args
 
     data = b''.join(cborutil.streamencode(data))
 
     offset = 0
 
     while True:
         flags = 0
 
         # Must set new or continuation flag.
         if not offset:
             flags |= FLAG_COMMAND_REQUEST_NEW
         else:
             flags |= FLAG_COMMAND_REQUEST_CONTINUATION
 
         # Data frames is set on all frames.
         if datafh:
             flags |= FLAG_COMMAND_REQUEST_EXPECT_DATA
 
         payload = data[offset:offset + maxframesize]
         offset += len(payload)
 
         if len(payload) == maxframesize and offset < len(data):
             flags |= FLAG_COMMAND_REQUEST_MORE_FRAMES
 
         yield stream.makeframe(requestid=requestid,
                                typeid=FRAME_TYPE_COMMAND_REQUEST,
                                flags=flags,
                                payload=payload)
 
         if not (flags & FLAG_COMMAND_REQUEST_MORE_FRAMES):
             break
 
     if datafh:
         while True:
             data = datafh.read(DEFAULT_MAX_FRAME_SIZE)
 
             done = False
             if len(data) == DEFAULT_MAX_FRAME_SIZE:
                 flags = FLAG_COMMAND_DATA_CONTINUATION
             else:
                 flags = FLAG_COMMAND_DATA_EOS
                 assert datafh.read(1) == b''
                 done = True
 
             yield stream.makeframe(requestid=requestid,
                                    typeid=FRAME_TYPE_COMMAND_DATA,
                                    flags=flags,
                                    payload=data)
 
             if done:
                 break
 
 def createcommandresponseframesfrombytes(stream, requestid, data,
                                          maxframesize=DEFAULT_MAX_FRAME_SIZE):
     """Create a raw frame to send a bytes response from static bytes input.
 
     Returns a generator of bytearrays.
     """
     # Automatically send the overall CBOR response map.
     overall = b''.join(cborutil.streamencode({b'status': b'ok'}))
     if len(overall) > maxframesize:
         raise error.ProgrammingError('not yet implemented')
 
     # Simple case where we can fit the full response in a single frame.
     if len(overall) + len(data) <= maxframesize:
         flags = FLAG_COMMAND_RESPONSE_EOS
         yield stream.makeframe(requestid=requestid,
                                typeid=FRAME_TYPE_COMMAND_RESPONSE,
                                flags=flags,
                                payload=overall + data)
         return
 
     # It's easier to send the overall CBOR map in its own frame than to track
     # offsets.
     yield stream.makeframe(requestid=requestid,
                            typeid=FRAME_TYPE_COMMAND_RESPONSE,
                            flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
                            payload=overall)
 
     offset = 0
     while True:
         chunk = data[offset:offset + maxframesize]
         offset += len(chunk)
         done = offset == len(data)
 
         if done:
             flags = FLAG_COMMAND_RESPONSE_EOS
         else:
             flags = FLAG_COMMAND_RESPONSE_CONTINUATION
 
         yield stream.makeframe(requestid=requestid,
                                typeid=FRAME_TYPE_COMMAND_RESPONSE,
                                flags=flags,
                                payload=chunk)
 
         if done:
             break
 
 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)
         if not chunk:
             break
 
         yield stream.makeframe(requestid=requestid,
                                typeid=FRAME_TYPE_COMMAND_RESPONSE,
                                flags=flags,
                                payload=chunk)
 
         flags |= FLAG_COMMAND_RESPONSE_CONTINUATION
 
-    flags ^= FLAG_COMMAND_RESPONSE_CONTINUATION
-    flags |= FLAG_COMMAND_RESPONSE_EOS
-    yield stream.makeframe(requestid=requestid,
-                           typeid=FRAME_TYPE_COMMAND_RESPONSE,
-                           flags=flags,
-                           payload=b'')
+def createcommandresponseokframe(stream, requestid):
+    overall = b''.join(cborutil.streamencode({b'status': b'ok'}))
+
+    return stream.makeframe(requestid=requestid,
+                            typeid=FRAME_TYPE_COMMAND_RESPONSE,
+                            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):
     # TODO should this be using a list of {'msg': ..., 'args': {}} so atom
     # formatting works consistently?
     m = {
         b'status': b'error',
         b'error': {
             b'message': message,
         }
     }
 
     if args:
         m[b'error'][b'args'] = args
 
     overall = b''.join(cborutil.streamencode(m))
 
     yield stream.makeframe(requestid=requestid,
                            typeid=FRAME_TYPE_COMMAND_RESPONSE,
                            flags=FLAG_COMMAND_RESPONSE_EOS,
                            payload=overall)
 
 def createerrorframe(stream, requestid, msg, errtype):
     # TODO properly handle frame size limits.
     assert len(msg) <= DEFAULT_MAX_FRAME_SIZE
 
     payload = b''.join(cborutil.streamencode({
         b'type': errtype,
         b'message': [{b'msg': msg}],
     }))
 
     yield stream.makeframe(requestid=requestid,
                            typeid=FRAME_TYPE_ERROR_RESPONSE,
                            flags=0,
                            payload=payload)
 
 def createtextoutputframe(stream, requestid, atoms,
                           maxframesize=DEFAULT_MAX_FRAME_SIZE):
     """Create a text output frame to render text to people.
 
     ``atoms`` is a 3-tuple of (formatting string, args, labels).
 
     The formatting string contains ``%s`` tokens to be replaced by the
     corresponding indexed entry in ``args``. ``labels`` is an iterable of
     formatters to be applied at rendering time. In terms of the ``ui``
     class, each atom corresponds to a ``ui.write()``.
     """
     atomdicts = []
 
     for (formatting, args, labels) in atoms:
         # TODO look for localstr, other types here?
 
         if not isinstance(formatting, bytes):
             raise ValueError('must use bytes formatting strings')
         for arg in args:
             if not isinstance(arg, bytes):
                 raise ValueError('must use bytes for arguments')
         for label in labels:
             if not isinstance(label, bytes):
                 raise ValueError('must use bytes for labels')
 
         # Formatting string must be ASCII.
         formatting = formatting.decode(r'ascii', r'replace').encode(r'ascii')
 
         # Arguments must be UTF-8.
         args = [a.decode(r'utf-8', r'replace').encode(r'utf-8') for a in args]
 
         # Labels must be ASCII.
         labels = [l.decode(r'ascii', r'strict').encode(r'ascii')
                   for l in labels]
 
         atom = {b'msg': formatting}
         if args:
             atom[b'args'] = args
         if labels:
             atom[b'labels'] = labels
 
         atomdicts.append(atom)
 
     payload = b''.join(cborutil.streamencode(atomdicts))
 
     if len(payload) > maxframesize:
         raise ValueError('cannot encode data in a single frame')
 
     yield stream.makeframe(requestid=requestid,
                            typeid=FRAME_TYPE_TEXT_OUTPUT,
                            flags=0,
                            payload=payload)
 
 class stream(object):
     """Represents a logical unidirectional series of frames."""
 
     def __init__(self, streamid, active=False):
         self.streamid = streamid
         self._active = active
 
     def makeframe(self, requestid, typeid, flags, payload):
         """Create a frame to be sent out over this stream.
 
         Only returns the frame instance. Does not actually send it.
         """
         streamflags = 0
         if not self._active:
             streamflags |= STREAM_FLAG_BEGIN_STREAM
             self._active = True
 
         return makeframe(requestid, self.streamid, streamflags, typeid, flags,
                          payload)
 
 def ensureserverstream(stream):
     if stream.streamid % 2:
         raise error.ProgrammingError('server should only write to even '
                                      'numbered streams; %d is not even' %
                                      stream.streamid)
 
 class serverreactor(object):
     """Holds state of a server handling frame-based protocol requests.
 
     This class is the "brain" of the unified frame-based protocol server
     component. While the protocol is stateless from the perspective of
     requests/commands, something needs to track which frames have been
     received, what frames to expect, etc. This class is that thing.
 
     Instances are modeled as a state machine of sorts. Instances are also
     reactionary to external events. The point of this class is to encapsulate
     the state of the connection and the exchange of frames, not to perform
     work. Instead, callers tell this class when something occurs, like a
     frame arriving. If that activity is worthy of a follow-up action (say
     *run a command*), the return value of that handler will say so.
 
     I/O and CPU intensive operations are purposefully delegated outside of
     this class.
 
     Consumers are expected to tell instances when events occur. They do so by
     calling the various ``on*`` methods. These methods return a 2-tuple
     describing any follow-up action(s) to take. The first element is the
     name of an action to perform. The second is a data structure (usually
     a dict) specific to that action that contains more information. e.g.
     if the server wants to send frames back to the client, the data structure
     will contain a reference to those frames.
 
     Valid actions that consumers can be instructed to take are:
 
     sendframes
        Indicates that frames should be sent to the client. The ``framegen``
        key contains a generator of frames that should be sent. The server
        assumes that all frames are sent to the client.
 
     error
        Indicates that an error occurred. Consumer should probably abort.
 
     runcommand
        Indicates that the consumer should run a wire protocol command. Details
        of the command to run are given in the data structure.
 
     wantframe
        Indicates that nothing of interest happened and the server is waiting on
        more frames from the client before anything interesting can be done.
 
     noop
        Indicates no additional action is required.
 
     Known Issues
     ------------
 
     There are no limits to the number of partially received commands or their
     size. A malicious client could stream command request data and exhaust the
     server's memory.
 
     Partially received commands are not acted upon when end of input is
     reached. Should the server error if it receives a partial request?
     Should the client send a message to abort a partially transmitted request
     to facilitate graceful shutdown?
 
     Active requests that haven't been responded to aren't tracked. This means
     that if we receive a command and instruct its dispatch, another command
     with its request ID can come in over the wire and there will be a race
     between who responds to what.
     """
 
     def __init__(self, deferoutput=False):
         """Construct a new server reactor.
 
         ``deferoutput`` can be used to indicate that no output frames should be
         instructed to be sent until input has been exhausted. In this mode,
         events that would normally generate output frames (such as a command
         response being ready) will instead defer instructing the consumer to
         send those frames. This is useful for half-duplex transports where the
         sender cannot receive until all data has been transmitted.
         """
         self._deferoutput = deferoutput
         self._state = 'idle'
         self._nextoutgoingstreamid = 2
         self._bufferedframegens = []
         # stream id -> stream instance for all active streams from the client.
         self._incomingstreams = {}
         self._outgoingstreams = {}
         # request id -> dict of commands that are actively being received.
         self._receivingcommands = {}
         # Request IDs that have been received and are actively being processed.
         # Once all output for a request has been sent, it is removed from this
         # set.
         self._activecommands = set()
 
     def onframerecv(self, frame):
         """Process a frame that has been received off the wire.
 
         Returns a dict with an ``action`` key that details what action,
         if any, the consumer should take next.
         """
         if not frame.streamid % 2:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('received frame with even numbered stream ID: %d') %
                   frame.streamid)
 
         if frame.streamid not in self._incomingstreams:
             if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM:
                 self._state = 'errored'
                 return self._makeerrorresult(
                     _('received frame on unknown inactive stream without '
                       'beginning of stream flag set'))
 
             self._incomingstreams[frame.streamid] = stream(frame.streamid)
 
         if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED:
             # TODO handle decoding frames
             self._state = 'errored'
             raise error.ProgrammingError('support for decoding stream payloads '
                                          'not yet implemented')
 
         if frame.streamflags & STREAM_FLAG_END_STREAM:
             del self._incomingstreams[frame.streamid]
 
         handlers = {
             'idle': self._onframeidle,
             'command-receiving': self._onframecommandreceiving,
             'errored': self._onframeerrored,
         }
 
         meth = handlers.get(self._state)
         if not meth:
             raise error.ProgrammingError('unhandled state: %s' % self._state)
 
         return meth(frame)
 
     def oncommandresponseready(self, stream, requestid, data):
         """Signal that a bytes response is ready to be sent to the client.
 
         The raw bytes response is passed as an argument.
         """
         ensureserverstream(stream)
 
         def sendframes():
             for frame in createcommandresponseframesfrombytes(stream, requestid,
                                                               data):
                 yield frame
 
             self._activecommands.remove(requestid)
 
         result = sendframes()
 
         if self._deferoutput:
             self._bufferedframegens.append(result)
             return 'noop', {}
         else:
             return 'sendframes', {
                 '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):
-                yield frame
+            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())
 
     def oninputeof(self):
         """Signals that end of input has been received.
 
         No more frames will be received. All pending activity should be
         completed.
         """
         # TODO should we do anything about in-flight commands?
 
         if not self._deferoutput or not self._bufferedframegens:
             return 'noop', {}
 
         # If we buffered all our responses, emit those.
         def makegen():
             for gen in self._bufferedframegens:
                 for frame in gen:
                     yield frame
 
         return 'sendframes', {
             'framegen': makegen(),
         }
 
     def _handlesendframes(self, framegen):
         if self._deferoutput:
             self._bufferedframegens.append(framegen)
             return 'noop', {}
         else:
             return 'sendframes', {
                 'framegen': framegen,
             }
 
     def onservererror(self, stream, requestid, msg):
         ensureserverstream(stream)
 
         def sendframes():
             for frame in createerrorframe(stream, requestid, msg,
                                           errtype='server'):
                 yield frame
 
             self._activecommands.remove(requestid)
 
         return self._handlesendframes(sendframes())
 
     def oncommanderror(self, stream, requestid, message, args=None):
         """Called when a command encountered an error before sending output."""
         ensureserverstream(stream)
 
         def sendframes():
             for frame in createcommanderrorresponse(stream, requestid, message,
                                                     args):
                 yield frame
 
             self._activecommands.remove(requestid)
 
         return self._handlesendframes(sendframes())
 
     def makeoutputstream(self):
         """Create a stream to be used for sending data to the client."""
         streamid = self._nextoutgoingstreamid
         self._nextoutgoingstreamid += 2
 
         s = stream(streamid)
         self._outgoingstreams[streamid] = s
 
         return s
 
     def _makeerrorresult(self, msg):
         return 'error', {
             'message': msg,
         }
 
     def _makeruncommandresult(self, requestid):
         entry = self._receivingcommands[requestid]
 
         if not entry['requestdone']:
             self._state = 'errored'
             raise error.ProgrammingError('should not be called without '
                                          'requestdone set')
 
         del self._receivingcommands[requestid]
 
         if self._receivingcommands:
             self._state = 'command-receiving'
         else:
             self._state = 'idle'
 
         # Decode the payloads as CBOR.
         entry['payload'].seek(0)
         request = cborutil.decodeall(entry['payload'].getvalue())[0]
 
         if b'name' not in request:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('command request missing "name" field'))
 
         if b'args' not in request:
             request[b'args'] = {}
 
         assert requestid not in self._activecommands
         self._activecommands.add(requestid)
 
         return 'runcommand', {
             'requestid': requestid,
             'command': request[b'name'],
             'args': request[b'args'],
             'data': entry['data'].getvalue() if entry['data'] else None,
         }
 
     def _makewantframeresult(self):
         return 'wantframe', {
             'state': self._state,
         }
 
     def _validatecommandrequestframe(self, frame):
         new = frame.flags & FLAG_COMMAND_REQUEST_NEW
         continuation = frame.flags & FLAG_COMMAND_REQUEST_CONTINUATION
 
         if new and continuation:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('received command request frame with both new and '
                   'continuation flags set'))
 
         if not new and not continuation:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('received command request frame with neither new nor '
                   'continuation flags set'))
 
     def _onframeidle(self, frame):
         # The only frame type that should be received in this state is a
         # command request.
         if frame.typeid != FRAME_TYPE_COMMAND_REQUEST:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('expected command request frame; got %d') % frame.typeid)
 
         res = self._validatecommandrequestframe(frame)
         if res:
             return res
 
         if frame.requestid in self._receivingcommands:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('request with ID %d already received') % frame.requestid)
 
         if frame.requestid in self._activecommands:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('request with ID %d is already active') % frame.requestid)
 
         new = frame.flags & FLAG_COMMAND_REQUEST_NEW
         moreframes = frame.flags & FLAG_COMMAND_REQUEST_MORE_FRAMES
         expectingdata = frame.flags & FLAG_COMMAND_REQUEST_EXPECT_DATA
 
         if not new:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('received command request frame without new flag set'))
 
         payload = util.bytesio()
         payload.write(frame.payload)
 
         self._receivingcommands[frame.requestid] = {
             'payload': payload,
             'data': None,
             'requestdone': not moreframes,
             'expectingdata': bool(expectingdata),
         }
 
         # This is the final frame for this request. Dispatch it.
         if not moreframes and not expectingdata:
             return self._makeruncommandresult(frame.requestid)
 
         assert moreframes or expectingdata
         self._state = 'command-receiving'
         return self._makewantframeresult()
 
     def _onframecommandreceiving(self, frame):
         if frame.typeid == FRAME_TYPE_COMMAND_REQUEST:
             # Process new command requests as such.
             if frame.flags & FLAG_COMMAND_REQUEST_NEW:
                 return self._onframeidle(frame)
 
             res = self._validatecommandrequestframe(frame)
             if res:
                 return res
 
         # All other frames should be related to a command that is currently
         # receiving but is not active.
         if frame.requestid in self._activecommands:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('received frame for request that is still active: %d') %
                 frame.requestid)
 
         if frame.requestid not in self._receivingcommands:
             self._state = 'errored'
             return self._makeerrorresult(
                 _('received frame for request that is not receiving: %d') %
                   frame.requestid)
 
         entry = self._receivingcommands[frame.requestid]
 
         if frame.typeid == FRAME_TYPE_COMMAND_REQUEST:
             moreframes = frame.flags & FLAG_COMMAND_REQUEST_MORE_FRAMES
             expectingdata = bool(frame.flags & FLAG_COMMAND_REQUEST_EXPECT_DATA)
 
             if entry['requestdone']:
                 self._state = 'errored'
                 return self._makeerrorresult(
                     _('received command request frame when request frames '
                       'were supposedly done'))
 
             if expectingdata != entry['expectingdata']:
                 self._state = 'errored'
                 return self._makeerrorresult(
                     _('mismatch between expect data flag and previous frame'))
 
             entry['payload'].write(frame.payload)
 
             if not moreframes:
                 entry['requestdone'] = True
 
             if not moreframes and not expectingdata:
                 return self._makeruncommandresult(frame.requestid)
 
             return self._makewantframeresult()
 
         elif frame.typeid == FRAME_TYPE_COMMAND_DATA:
             if not entry['expectingdata']:
                 self._state = 'errored'
                 return self._makeerrorresult(_(
                     'received command data frame for request that is not '
                     'expecting data: %d') % frame.requestid)
 
             if entry['data'] is None:
                 entry['data'] = util.bytesio()
 
             return self._handlecommanddataframe(frame, entry)
         else:
             self._state = 'errored'
             return self._makeerrorresult(_(
                 'received unexpected frame type: %d') % frame.typeid)
 
     def _handlecommanddataframe(self, frame, entry):
         assert frame.typeid == FRAME_TYPE_COMMAND_DATA
 
         # TODO support streaming data instead of buffering it.
         entry['data'].write(frame.payload)
 
         if frame.flags & FLAG_COMMAND_DATA_CONTINUATION:
             return self._makewantframeresult()
         elif frame.flags & FLAG_COMMAND_DATA_EOS:
             entry['data'].seek(0)
             return self._makeruncommandresult(frame.requestid)
         else:
             self._state = 'errored'
             return self._makeerrorresult(_('command data frame without '
                                            'flags'))
 
     def _onframeerrored(self, frame):
         return self._makeerrorresult(_('server already errored'))
 
 class commandrequest(object):
     """Represents a request to run a command."""
 
     def __init__(self, requestid, name, args, datafh=None):
         self.requestid = requestid
         self.name = name
         self.args = args
         self.datafh = datafh
         self.state = 'pending'
 
 class clientreactor(object):
     """Holds state of a client issuing frame-based protocol requests.
 
     This is like ``serverreactor`` but for client-side state.
 
     Each instance is bound to the lifetime of a connection. For persistent
     connection transports using e.g. TCP sockets and speaking the raw
     framing protocol, there will be a single instance for the lifetime of
     the TCP socket. For transports where there are multiple discrete
     interactions (say tunneled within in HTTP request), there will be a
     separate instance for each distinct interaction.
     """
     def __init__(self, hasmultiplesend=False, buffersends=True):
         """Create a new instance.
 
         ``hasmultiplesend`` indicates whether multiple sends are supported
         by the transport. When True, it is possible to send commands immediately
         instead of buffering until the caller signals an intent to finish a
         send operation.
 
         ``buffercommands`` indicates whether sends should be buffered until the
         last request has been issued.
         """
         self._hasmultiplesend = hasmultiplesend
         self._buffersends = buffersends
 
         self._canissuecommands = True
         self._cansend = True
 
         self._nextrequestid = 1
         # We only support a single outgoing stream for now.
         self._outgoingstream = stream(1)
         self._pendingrequests = collections.deque()
         self._activerequests = {}
         self._incomingstreams = {}
 
     def callcommand(self, name, args, datafh=None):
         """Request that a command be executed.
 
         Receives the command name, a dict of arguments to pass to the command,
         and an optional file object containing the raw data for the command.
 
         Returns a 3-tuple of (request, action, action data).
         """
         if not self._canissuecommands:
             raise error.ProgrammingError('cannot issue new commands')
 
         requestid = self._nextrequestid
         self._nextrequestid += 2
 
         request = commandrequest(requestid, name, args, datafh=datafh)
 
         if self._buffersends:
             self._pendingrequests.append(request)
             return request, 'noop', {}
         else:
             if not self._cansend:
                 raise error.ProgrammingError('sends cannot be performed on '
                                              'this instance')
 
             if not self._hasmultiplesend:
                 self._cansend = False
                 self._canissuecommands = False
 
             return request, 'sendframes', {
                 'framegen': self._makecommandframes(request),
             }
 
     def flushcommands(self):
         """Request that all queued commands be sent.
 
         If any commands are buffered, this will instruct the caller to send
         them over the wire. If no commands are buffered it instructs the client
         to no-op.
 
         If instances aren't configured for multiple sends, no new command
         requests are allowed after this is called.
         """
         if not self._pendingrequests:
             return 'noop', {}
 
         if not self._cansend:
             raise error.ProgrammingError('sends cannot be performed on this '
                                          'instance')
 
         # If the instance only allows sending once, mark that we have fired
         # our one shot.
         if not self._hasmultiplesend:
             self._canissuecommands = False
             self._cansend = False
 
         def makeframes():
             while self._pendingrequests:
                 request = self._pendingrequests.popleft()
                 for frame in self._makecommandframes(request):
                     yield frame
 
         return 'sendframes', {
             'framegen': makeframes(),
         }
 
     def _makecommandframes(self, request):
         """Emit frames to issue a command request.
 
         As a side-effect, update request accounting to reflect its changed
         state.
         """
         self._activerequests[request.requestid] = request
         request.state = 'sending'
 
         res = createcommandframes(self._outgoingstream,
                                   request.requestid,
                                   request.name,
                                   request.args,
                                   request.datafh)
 
         for frame in res:
             yield frame
 
         request.state = 'sent'
 
     def onframerecv(self, frame):
         """Process a frame that has been received off the wire.
 
         Returns a 2-tuple of (action, meta) describing further action the
         caller needs to take as a result of receiving this frame.
         """
         if frame.streamid % 2:
             return 'error', {
                 'message': (
                     _('received frame with odd numbered stream ID: %d') %
                     frame.streamid),
             }
 
         if frame.streamid not in self._incomingstreams:
             if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM:
                 return 'error', {
                     'message': _('received frame on unknown stream '
                                  'without beginning of stream flag set'),
                 }
 
             self._incomingstreams[frame.streamid] = stream(frame.streamid)
 
         if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED:
             raise error.ProgrammingError('support for decoding stream '
                                          'payloads not yet implemneted')
 
         if frame.streamflags & STREAM_FLAG_END_STREAM:
             del self._incomingstreams[frame.streamid]
 
         if frame.requestid not in self._activerequests:
             return 'error', {
                 'message': (_('received frame for inactive request ID: %d') %
                             frame.requestid),
             }
 
         request = self._activerequests[frame.requestid]
         request.state = 'receiving'
 
         handlers = {
             FRAME_TYPE_COMMAND_RESPONSE: self._oncommandresponseframe,
             FRAME_TYPE_ERROR_RESPONSE: self._onerrorresponseframe,
         }
 
         meth = handlers.get(frame.typeid)
         if not meth:
             raise error.ProgrammingError('unhandled frame type: %d' %
                                          frame.typeid)
 
         return meth(request, frame)
 
     def _oncommandresponseframe(self, request, frame):
         if frame.flags & FLAG_COMMAND_RESPONSE_EOS:
             request.state = 'received'
             del self._activerequests[request.requestid]
 
         return 'responsedata', {
             'request': request,
             'expectmore': frame.flags & FLAG_COMMAND_RESPONSE_CONTINUATION,
             'eos': frame.flags & FLAG_COMMAND_RESPONSE_EOS,
             'data': frame.payload,
         }
 
     def _onerrorresponseframe(self, request, frame):
         request.state = 'errored'
         del self._activerequests[request.requestid]
 
         # The payload should be a CBOR map.
         m = cborutil.decodeall(frame.payload)[0]
 
         return 'error', {
             'request': request,
             'type': m['type'],
             'message': m['message'],
         }

mercurial/wireprototypes.py

@@ -1,375 +1,354 b''
 # Copyright 2018 Gregory Szorc <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 from .node import (
     bin,
     hex,
 )
 from .i18n import _
 from . import (
     error,
     util,
 )
 from .utils import (
     interfaceutil,
 )
 
 # Names of the SSH protocol implementations.
 SSHV1 = 'ssh-v1'
 # These are advertised over the wire. Increment the counters at the end
 # to reflect BC breakages.
 SSHV2 = 'exp-ssh-v2-0001'
 HTTP_WIREPROTO_V2 = 'exp-http-v2-0001'
 
 # All available wire protocol transports.
 TRANSPORTS = {
     SSHV1: {
         'transport': 'ssh',
         'version': 1,
     },
     SSHV2: {
         'transport': 'ssh',
         # TODO mark as version 2 once all commands are implemented.
         'version': 1,
     },
     'http-v1': {
         'transport': 'http',
         'version': 1,
     },
     HTTP_WIREPROTO_V2: {
         'transport': 'http',
         'version': 2,
     }
 }
 
 class bytesresponse(object):
     """A wire protocol response consisting of raw bytes."""
     def __init__(self, data):
         self.data = data
 
 class ooberror(object):
     """wireproto reply: failure of a batch of operation
 
     Something failed during a batch call. The error message is stored in
     `self.message`.
     """
     def __init__(self, message):
         self.message = message
 
 class pushres(object):
     """wireproto reply: success with simple integer return
 
     The call was successful and returned an integer contained in `self.res`.
     """
     def __init__(self, res, output):
         self.res = res
         self.output = output
 
 class pusherr(object):
     """wireproto reply: failure
 
     The call failed. The `self.res` attribute contains the error message.
     """
     def __init__(self, res, output):
         self.res = res
         self.output = output
 
 class streamres(object):
     """wireproto reply: binary stream
 
     The call was successful and the result is a stream.
 
     Accepts a generator containing chunks of data to be sent to the client.
 
     ``prefer_uncompressed`` indicates that the data is expected to be
     uncompressable and that the stream should therefore use the ``none``
     engine.
     """
     def __init__(self, gen=None, prefer_uncompressed=False):
         self.gen = gen
         self.prefer_uncompressed = prefer_uncompressed
 
 class streamreslegacy(object):
     """wireproto reply: uncompressed binary stream
 
     The call was successful and the result is a stream.
 
     Accepts a generator containing chunks of data to be sent to the client.
 
     Like ``streamres``, but sends an uncompressed data for "version 1" clients
     using the application/mercurial-0.1 media type.
     """
     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:
         return [bin(v) for v in  l.split(sep)]
     return []
 
 def encodelist(l, sep=' '):
     try:
         return sep.join(map(hex, l))
     except TypeError:
         raise
 
 # batched call argument encoding
 
 def escapebatcharg(plain):
     return (plain
             .replace(':', ':c')
             .replace(',', ':o')
             .replace(';', ':s')
             .replace('=', ':e'))
 
 def unescapebatcharg(escaped):
     return (escaped
             .replace(':e', '=')
             .replace(':s', ';')
             .replace(':o', ',')
             .replace(':c', ':'))
 
 # mapping of options accepted by getbundle and their types
 #
 # Meant to be extended by extensions. It is extensions responsibility to ensure
 # such options are properly processed in exchange.getbundle.
 #
 # supported types are:
 #
 # :nodes: list of binary nodes
 # :csv:   list of comma-separated values
 # :scsv:  list of comma-separated values return as set
 # :plain: string with no transformation needed.
 GETBUNDLE_ARGUMENTS = {
     'heads':  'nodes',
     'bookmarks': 'boolean',
     'common': 'nodes',
     'obsmarkers': 'boolean',
     'phases': 'boolean',
     'bundlecaps': 'scsv',
     'listkeys': 'csv',
     'cg': 'boolean',
     'cbattempted': 'boolean',
     'stream': 'boolean',
 }
 
 class baseprotocolhandler(interfaceutil.Interface):
     """Abstract base class for wire protocol handlers.
 
     A wire protocol handler serves as an interface between protocol command
     handlers and the wire protocol transport layer. Protocol handlers provide
     methods to read command arguments, redirect stdio for the duration of
     the request, handle response types, etc.
     """
 
     name = interfaceutil.Attribute(
         """The name of the protocol implementation.
 
         Used for uniquely identifying the transport type.
         """)
 
     def getargs(args):
         """return the value for arguments in <args>
 
         For version 1 transports, returns a list of values in the same
         order they appear in ``args``. For version 2 transports, returns
         a dict mapping argument name to value.
         """
 
     def getprotocaps():
         """Returns the list of protocol-level capabilities of client
 
         Returns a list of capabilities as declared by the client for
         the current request (or connection for stateful protocol handlers)."""
 
     def getpayload():
         """Provide a generator for the raw payload.
 
         The caller is responsible for ensuring that the full payload is
         processed.
         """
 
     def mayberedirectstdio():
         """Context manager to possibly redirect stdio.
 
         The context manager yields a file-object like object that receives
         stdout and stderr output when the context manager is active. Or it
         yields ``None`` if no I/O redirection occurs.
 
         The intent of this context manager is to capture stdio output
         so it may be sent in the response. Some transports support streaming
         stdio to the client in real time. For these transports, stdio output
         won't be captured.
         """
 
     def client():
         """Returns a string representation of this client (as bytes)."""
 
     def addcapabilities(repo, caps):
         """Adds advertised capabilities specific to this protocol.
 
         Receives the list of capabilities collected so far.
 
         Returns a list of capabilities. The passed in argument can be returned.
         """
 
     def checkperm(perm):
         """Validate that the client has permissions to perform a request.
 
         The argument is the permission required to proceed. If the client
         doesn't have that permission, the exception should raise or abort
         in a protocol specific manner.
         """
 
 class commandentry(object):
     """Represents a declared wire protocol command."""
     def __init__(self, func, args='', transports=None,
                  permission='push'):
         self.func = func
         self.args = args
         self.transports = transports or set()
         self.permission = permission
 
     def _merge(self, func, args):
         """Merge this instance with an incoming 2-tuple.
 
         This is called when a caller using the old 2-tuple API attempts
         to replace an instance. The incoming values are merged with
         data not captured by the 2-tuple and a new instance containing
         the union of the two objects is returned.
         """
         return commandentry(func, args=args, transports=set(self.transports),
                             permission=self.permission)
 
     # Old code treats instances as 2-tuples. So expose that interface.
     def __iter__(self):
         yield self.func
         yield self.args
 
     def __getitem__(self, i):
         if i == 0:
             return self.func
         elif i == 1:
             return self.args
         else:
             raise IndexError('can only access elements 0 and 1')
 
 class commanddict(dict):
     """Container for registered wire protocol commands.
 
     It behaves like a dict. But __setitem__ is overwritten to allow silent
     coercion of values from 2-tuples for API compatibility.
     """
     def __setitem__(self, k, v):
         if isinstance(v, commandentry):
             pass
         # Cast 2-tuples to commandentry instances.
         elif isinstance(v, tuple):
             if len(v) != 2:
                 raise ValueError('command tuples must have exactly 2 elements')
 
             # It is common for extensions to wrap wire protocol commands via
             # e.g. ``wireproto.commands[x] = (newfn, args)``. Because callers
             # doing this aren't aware of the new API that uses objects to store
             # command entries, we automatically merge old state with new.
             if k in self:
                 v = self[k]._merge(v[0], v[1])
             else:
                 # Use default values from @wireprotocommand.
                 v = commandentry(v[0], args=v[1],
                                  transports=set(TRANSPORTS),
                                  permission='push')
         else:
             raise ValueError('command entries must be commandentry instances '
                              'or 2-tuples')
 
         return super(commanddict, self).__setitem__(k, v)
 
     def commandavailable(self, command, proto):
         """Determine if a command is available for the requested protocol."""
         assert proto.name in TRANSPORTS
 
         entry = self.get(command)
 
         if not entry:
             return False
 
         if proto.name not in entry.transports:
             return False
 
         return True
 
 def supportedcompengines(ui, role):
     """Obtain the list of supported compression engines for a request."""
     assert role in (util.CLIENTROLE, util.SERVERROLE)
 
     compengines = util.compengines.supportedwireengines(role)
 
     # Allow config to override default list and ordering.
     if role == util.SERVERROLE:
         configengines = ui.configlist('server', 'compressionengines')
         config = 'server.compressionengines'
     else:
         # This is currently implemented mainly to facilitate testing. In most
         # cases, the server should be in charge of choosing a compression engine
         # because a server has the most to lose from a sub-optimal choice. (e.g.
         # CPU DoS due to an expensive engine or a network DoS due to poor
         # compression ratio).
         configengines = ui.configlist('experimental',
                                       'clientcompressionengines')
         config = 'experimental.clientcompressionengines'
 
     # No explicit config. Filter out the ones that aren't supposed to be
     # advertised and return default ordering.
     if not configengines:
         attr = 'serverpriority' if role == util.SERVERROLE else 'clientpriority'
         return [e for e in compengines
                 if getattr(e.wireprotosupport(), attr) > 0]
 
     # If compression engines are listed in the config, assume there is a good
     # reason for it (like server operators wanting to achieve specific
     # performance characteristics). So fail fast if the config references
     # unusable compression engines.
     validnames = set(e.name() for e in compengines)
     invalidnames = set(e for e in configengines if e not in validnames)
     if invalidnames:
         raise error.Abort(_('invalid compression engine defined in %s: %s') %
                           (config, ', '.join(sorted(invalidnames))))
 
     compengines = [e for e in compengines if e.name() in configengines]
     compengines = sorted(compengines,
                          key=lambda e: configengines.index(e.name()))
 
     if not compengines:
         raise error.Abort(_('%s config option does not specify any known '
                             'compression engines') % config,
                           hint=_('usable compression engines: %s') %
                           ', '.sorted(validnames))
 
     return compengines

mercurial/wireprotov2peer.py

@@ -1,210 +1,212 b''
 # wireprotov2peer.py - client side code for wire protocol version 2
 #
 # Copyright 2018 Gregory Szorc <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 from .i18n import _
 from . import (
     encoding,
     error,
     util,
     wireprotoframing,
 )
 from .utils import (
     cborutil,
 )
 
 def formatrichmessage(atoms):
     """Format an encoded message from the framing protocol."""
 
     chunks = []
 
     for atom in atoms:
         msg = _(atom[b'msg'])
 
         if b'args' in atom:
             msg = msg % tuple(atom[b'args'])
 
         chunks.append(msg)
 
     return b''.join(chunks)
 
 class commandresponse(object):
     """Represents the response to a command request."""
 
     def __init__(self, requestid, command):
         self.requestid = requestid
         self.command = command
 
         self.b = util.bytesio()
 
     def cborobjects(self):
         """Obtain decoded CBOR objects from this response."""
         self.b.seek(0)
 
         for v in cborutil.decodeall(self.b.getvalue()):
             yield v
 
 class clienthandler(object):
     """Object to handle higher-level client activities.
 
     The ``clientreactor`` is used to hold low-level state about the frame-based
     protocol, such as which requests and streams are active. This type is used
     for higher-level operations, such as reading frames from a socket, exposing
     and managing a higher-level primitive for representing command responses,
     etc. This class is what peers should probably use to bridge wire activity
     with the higher-level peer API.
     """
 
     def __init__(self, ui, clientreactor):
         self._ui = ui
         self._reactor = clientreactor
         self._requests = {}
         self._futures = {}
         self._responses = {}
 
     def callcommand(self, command, args, f):
         """Register a request to call a command.
 
         Returns an iterable of frames that should be sent over the wire.
         """
         request, action, meta = self._reactor.callcommand(command, args)
 
         if action != 'noop':
             raise error.ProgrammingError('%s not yet supported' % action)
 
         rid = request.requestid
         self._requests[rid] = request
         self._futures[rid] = f
         self._responses[rid] = commandresponse(rid, command)
 
         return iter(())
 
     def flushcommands(self):
         """Flush all queued commands.
 
         Returns an iterable of frames that should be sent over the wire.
         """
         action, meta = self._reactor.flushcommands()
 
         if action != 'sendframes':
             raise error.ProgrammingError('%s not yet supported' % action)
 
         return meta['framegen']
 
     def readframe(self, fh):
         """Attempt to read and process a frame.
 
         Returns None if no frame was read. Presumably this means EOF.
         """
         frame = wireprotoframing.readframe(fh)
         if frame is None:
             # TODO tell reactor?
             return
 
         self._ui.note(_('received %r\n') % frame)
         self._processframe(frame)
 
         return True
 
     def _processframe(self, frame):
         """Process a single read frame."""
 
         action, meta = self._reactor.onframerecv(frame)
 
         if action == 'error':
             e = error.RepoError(meta['message'])
 
             if frame.requestid in self._futures:
                 self._futures[frame.requestid].set_exception(e)
             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 '
                 'the clientreactor not screening for this or this instance was '
                 'never told about this request: %r' % frame)
 
         response = self._responses[frame.requestid]
 
         if action == 'responsedata':
             # Any failures processing this frame should bubble up to the
             # future tracking the request.
             try:
                 self._processresponsedata(frame, meta, response)
             except BaseException as e:
                 self._futures[frame.requestid].set_exception(e)
         else:
             raise error.ProgrammingError(
                 'unhandled action from clientreactor: %s' % action)
 
     def _processresponsedata(self, frame, meta, response):
         # This buffers all data until end of stream is received. This
         # is bad for performance.
         # TODO make response data streamable
         response.b.write(meta['data'])
 
         if meta['eos']:
             # If the command has a decoder, resolve the future to the
             # decoded value. Otherwise resolve to the rich response object.
             decoder = COMMAND_DECODERS.get(response.command)
 
             # TODO consider always resolving the overall status map.
             if decoder:
                 objs = response.cborobjects()
 
                 overall = next(objs)
 
                 if overall['status'] == 'ok':
                     self._futures[frame.requestid].set_result(decoder(objs))
                 else:
                     atoms = [{'msg': overall['error']['message']}]
                     if 'args' in overall['error']:
                         atoms[0]['args'] = overall['error']['args']
                     e = error.RepoError(formatrichmessage(atoms))
                     self._futures[frame.requestid].set_exception(e)
             else:
                 self._futures[frame.requestid].set_result(response)
 
             del self._requests[frame.requestid]
             del self._futures[frame.requestid]
 
 def decodebranchmap(objs):
     # Response should be a single CBOR map of branch name to array of nodes.
     bm = next(objs)
 
     return {encoding.tolocal(k): v for k, v in bm.items()}
 
 def decodeheads(objs):
     # Array of node bytestrings.
     return next(objs)
 
 def decodeknown(objs):
     # Bytestring where each byte is a 0 or 1.
     raw = next(objs)
 
     return [True if c == '1' else False for c in raw]
 
 def decodelistkeys(objs):
     # Map with bytestring keys and values.
     return next(objs)
 
 def decodelookup(objs):
     return next(objs)
 
 def decodepushkey(objs):
     return next(objs)
 
 COMMAND_DECODERS = {
     'branchmap': decodebranchmap,
     'heads': decodeheads,
     'known': decodeknown,
     'listkeys': decodelistkeys,
     'lookup': decodelookup,
     'pushkey': decodepushkey,
 }

mercurial/wireprotov2server.py

@@ -1,535 +1,522 b''
 # Copyright 21 May 2005 - (c) 2005 Jake Edge <[e-mail]>
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import contextlib
 
 from .i18n import _
 from . import (
     encoding,
     error,
     pycompat,
     streamclone,
     util,
     wireprotoframing,
     wireprototypes,
 )
 from .utils import (
-    cborutil,
     interfaceutil,
 )
 
 FRAMINGTYPE = b'application/mercurial-exp-framing-0005'
 
 HTTP_WIREPROTO_V2 = wireprototypes.HTTP_WIREPROTO_V2
 
 COMMANDS = wireprototypes.commanddict()
 
 def handlehttpv2request(rctx, req, res, checkperm, urlparts):
     from .hgweb import common as hgwebcommon
 
     # URL space looks like: <permissions>/<command>, where <permission> can
     # be ``ro`` or ``rw`` to signal read-only or read-write, respectively.
 
     # Root URL does nothing meaningful... yet.
     if not urlparts:
         res.status = b'200 OK'
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('HTTP version 2 API handler'))
         return
 
     if len(urlparts) == 1:
         res.status = b'404 Not Found'
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('do not know how to process %s\n') %
                          req.dispatchpath)
         return
 
     permission, command = urlparts[0:2]
 
     if permission not in (b'ro', b'rw'):
         res.status = b'404 Not Found'
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('unknown permission: %s') % permission)
         return
 
     if req.method != 'POST':
         res.status = b'405 Method Not Allowed'
         res.headers[b'Allow'] = b'POST'
         res.setbodybytes(_('commands require POST requests'))
         return
 
     # At some point we'll want to use our own API instead of recycling the
     # behavior of version 1 of the wire protocol...
     # TODO return reasonable responses - not responses that overload the
     # HTTP status line message for error reporting.
     try:
         checkperm(rctx, req, 'pull' if permission == b'ro' else 'push')
     except hgwebcommon.ErrorResponse as e:
         res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e))
         for k, v in e.headers:
             res.headers[k] = v
         res.setbodybytes('permission denied')
         return
 
     # We have a special endpoint to reflect the request back at the client.
     if command == b'debugreflect':
         _processhttpv2reflectrequest(rctx.repo.ui, rctx.repo, req, res)
         return
 
     # Extra commands that we handle that aren't really wire protocol
     # commands. Think extra hard before making this hackery available to
     # extension.
     extracommands = {'multirequest'}
 
     if command not in COMMANDS and command not in extracommands:
         res.status = b'404 Not Found'
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('unknown wire protocol command: %s\n') % command)
         return
 
     repo = rctx.repo
     ui = repo.ui
 
     proto = httpv2protocolhandler(req, ui)
 
     if (not COMMANDS.commandavailable(command, proto)
         and command not in extracommands):
         res.status = b'404 Not Found'
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('invalid wire protocol command: %s') % command)
         return
 
     # TODO consider cases where proxies may add additional Accept headers.
     if req.headers.get(b'Accept') != FRAMINGTYPE:
         res.status = b'406 Not Acceptable'
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('client MUST specify Accept header with value: %s\n')
                            % FRAMINGTYPE)
         return
 
     if req.headers.get(b'Content-Type') != FRAMINGTYPE:
         res.status = b'415 Unsupported Media Type'
         # TODO we should send a response with appropriate media type,
         # since client does Accept it.
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('client MUST send Content-Type header with '
                            'value: %s\n') % FRAMINGTYPE)
         return
 
     _processhttpv2request(ui, repo, req, res, permission, command, proto)
 
 def _processhttpv2reflectrequest(ui, repo, req, res):
     """Reads unified frame protocol request and dumps out state to client.
 
     This special endpoint can be used to help debug the wire protocol.
 
     Instead of routing the request through the normal dispatch mechanism,
     we instead read all frames, decode them, and feed them into our state
     tracker. We then dump the log of all that activity back out to the
     client.
     """
     import json
 
     # Reflection APIs have a history of being abused, accidentally disclosing
     # sensitive data, etc. So we have a config knob.
     if not ui.configbool('experimental', 'web.api.debugreflect'):
         res.status = b'404 Not Found'
         res.headers[b'Content-Type'] = b'text/plain'
         res.setbodybytes(_('debugreflect service not available'))
         return
 
     # We assume we have a unified framing protocol request body.
 
     reactor = wireprotoframing.serverreactor()
     states = []
 
     while True:
         frame = wireprotoframing.readframe(req.bodyfh)
 
         if not frame:
             states.append(b'received: <no frame>')
             break
 
         states.append(b'received: %d %d %d %s' % (frame.typeid, frame.flags,
                                                   frame.requestid,
                                                   frame.payload))
 
         action, meta = reactor.onframerecv(frame)
         states.append(json.dumps((action, meta), sort_keys=True,
                                  separators=(', ', ': ')))
 
     action, meta = reactor.oninputeof()
     meta['action'] = action
     states.append(json.dumps(meta, sort_keys=True, separators=(', ',': ')))
 
     res.status = b'200 OK'
     res.headers[b'Content-Type'] = b'text/plain'
     res.setbodybytes(b'\n'.join(states))
 
 def _processhttpv2request(ui, repo, req, res, authedperm, reqcommand, proto):
     """Post-validation handler for HTTPv2 requests.
 
     Called when the HTTP request contains unified frame-based protocol
     frames for evaluation.
     """
     # TODO Some HTTP clients are full duplex and can receive data before
     # the entire request is transmitted. Figure out a way to indicate support
     # for that so we can opt into full duplex mode.
     reactor = wireprotoframing.serverreactor(deferoutput=True)
     seencommand = False
 
     outstream = reactor.makeoutputstream()
 
     while True:
         frame = wireprotoframing.readframe(req.bodyfh)
         if not frame:
             break
 
         action, meta = reactor.onframerecv(frame)
 
         if action == 'wantframe':
             # Need more data before we can do anything.
             continue
         elif action == 'runcommand':
             sentoutput = _httpv2runcommand(ui, repo, req, res, authedperm,
                                            reqcommand, reactor, outstream,
                                            meta, issubsequent=seencommand)
 
             if sentoutput:
                 return
 
             seencommand = True
 
         elif action == 'error':
             # TODO define proper error mechanism.
             res.status = b'200 OK'
             res.headers[b'Content-Type'] = b'text/plain'
             res.setbodybytes(meta['message'] + b'\n')
             return
         else:
             raise error.ProgrammingError(
                 'unhandled action from frame processor: %s' % action)
 
     action, meta = reactor.oninputeof()
     if action == 'sendframes':
         # We assume we haven't started sending the response yet. If we're
         # wrong, the response type will raise an exception.
         res.status = b'200 OK'
         res.headers[b'Content-Type'] = FRAMINGTYPE
         res.setbodygen(meta['framegen'])
     elif action == 'noop':
         pass
     else:
         raise error.ProgrammingError('unhandled action from frame processor: %s'
                                      % action)
 
 def _httpv2runcommand(ui, repo, req, res, authedperm, reqcommand, reactor,
                       outstream, command, issubsequent):
     """Dispatch a wire protocol command made from HTTPv2 requests.
 
     The authenticated permission (``authedperm``) along with the original
     command from the URL (``reqcommand``) are passed in.
     """
     # We already validated that the session has permissions to perform the
     # actions in ``authedperm``. In the unified frame protocol, the canonical
     # command to run is expressed in a frame. However, the URL also requested
     # to run a specific command. We need to be careful that the command we
     # run doesn't have permissions requirements greater than what was granted
     # by ``authedperm``.
     #
     # Our rule for this is we only allow one command per HTTP request and
     # that command must match the command in the URL. However, we make
     # an exception for the ``multirequest`` URL. This URL is allowed to
     # execute multiple commands. We double check permissions of each command
     # as it is invoked to ensure there is no privilege escalation.
     # TODO consider allowing multiple commands to regular command URLs
     # iff each command is the same.
 
     proto = httpv2protocolhandler(req, ui, args=command['args'])
 
     if reqcommand == b'multirequest':
         if not COMMANDS.commandavailable(command['command'], proto):
             # TODO proper error mechanism
             res.status = b'200 OK'
             res.headers[b'Content-Type'] = b'text/plain'
             res.setbodybytes(_('wire protocol command not available: %s') %
                              command['command'])
             return True
 
         # TODO don't use assert here, since it may be elided by -O.
         assert authedperm in (b'ro', b'rw')
         wirecommand = COMMANDS[command['command']]
         assert wirecommand.permission in ('push', 'pull')
 
         if authedperm == b'ro' and wirecommand.permission != 'pull':
             # TODO proper error mechanism
             res.status = b'403 Forbidden'
             res.headers[b'Content-Type'] = b'text/plain'
             res.setbodybytes(_('insufficient permissions to execute '
                                'command: %s') % command['command'])
             return True
 
         # TODO should we also call checkperm() here? Maybe not if we're going
         # to overhaul that API. The granted scope from the URL check should
         # be good enough.
 
     else:
         # Don't allow multiple commands outside of ``multirequest`` URL.
         if issubsequent:
             # TODO proper error mechanism
             res.status = b'200 OK'
             res.headers[b'Content-Type'] = b'text/plain'
             res.setbodybytes(_('multiple commands cannot be issued to this '
                                'URL'))
             return True
 
         if reqcommand != command['command']:
             # TODO define proper error mechanism
             res.status = b'200 OK'
             res.headers[b'Content-Type'] = b'text/plain'
             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'])
         return True
     elif action == 'noop':
         return False
     else:
         raise error.ProgrammingError('unhandled event from reactor: %s' %
                                      action)
 
 def getdispatchrepo(repo, proto, command):
     return repo.filtered('served')
 
 def dispatch(repo, proto, command):
     repo = getdispatchrepo(repo, proto, command)
 
     func, spec = COMMANDS[command]
     args = proto.getargs(spec)
 
     return func(repo, proto, **args)
 
 @interfaceutil.implementer(wireprototypes.baseprotocolhandler)
 class httpv2protocolhandler(object):
     def __init__(self, req, ui, args=None):
         self._req = req
         self._ui = ui
         self._args = args
 
     @property
     def name(self):
         return HTTP_WIREPROTO_V2
 
     def getargs(self, args):
         data = {}
         for k, typ in args.items():
             if k == '*':
                 raise NotImplementedError('do not support * args')
             elif k in self._args:
                 # TODO consider validating value types.
                 data[k] = self._args[k]
 
         return data
 
     def getprotocaps(self):
         # Protocol capabilities are currently not implemented for HTTP V2.
         return set()
 
     def getpayload(self):
         raise NotImplementedError
 
     @contextlib.contextmanager
     def mayberedirectstdio(self):
         raise NotImplementedError
 
     def client(self):
         raise NotImplementedError
 
     def addcapabilities(self, repo, caps):
         return caps
 
     def checkperm(self, perm):
         raise NotImplementedError
 
 def httpv2apidescriptor(req, repo):
     proto = httpv2protocolhandler(req, repo.ui)
 
     return _capabilitiesv2(repo, proto)
 
 def _capabilitiesv2(repo, proto):
     """Obtain the set of capabilities for version 2 transports.
 
     These capabilities are distinct from the capabilities for version 1
     transports.
     """
     compression = []
     for engine in wireprototypes.supportedcompengines(repo.ui, util.SERVERROLE):
         compression.append({
             b'name': engine.wireprotosupport().name,
         })
 
     caps = {
         'commands': {},
         'compression': compression,
         'framingmediatypes': [FRAMINGTYPE],
     }
 
     for command, entry in COMMANDS.items():
         caps['commands'][command] = {
             'args': entry.args,
             'permissions': [entry.permission],
         }
 
     if streamclone.allowservergeneration(repo):
         caps['rawrepoformats'] = sorted(repo.requirements &
                                         repo.supportedformats)
 
     return proto.addcapabilities(repo, caps)
 
 def wireprotocommand(name, args=None, permission='push'):
     """Decorator to declare a wire protocol command.
 
     ``name`` is the name of the wire protocol command being provided.
 
     ``args`` is a dict of argument names to example values.
 
     ``permission`` defines the permission type needed to run this command.
     Can be ``push`` or ``pull``. These roughly map to read-write and read-only,
     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}
 
     if permission not in ('push', 'pull'):
         raise error.ProgrammingError('invalid wire protocol permission; '
                                      'got %s; expected "push" or "pull"' %
                                      permission)
 
     if args is None:
         args = {}
 
     if not isinstance(args, dict):
         raise error.ProgrammingError('arguments for version 2 commands '
                                      'must be declared as dicts')
 
     def register(func):
         if name in COMMANDS:
             raise error.ProgrammingError('%s command already registered '
                                          'for version 2' % name)
 
         COMMANDS[name] = wireprototypes.commandentry(
             func, args=args, transports=transports, permission=permission)
 
         return func
 
     return register
 
 @wireprotocommand('branchmap', permission='pull')
 def branchmapv2(repo, proto):
-    branchmap = {encoding.fromlocal(k): v
-                 for k, v in repo.branchmap().iteritems()}
-
-    return wireprototypes.cborresponse(branchmap)
+    yield {encoding.fromlocal(k): v
+           for k, v in repo.branchmap().iteritems()}
 
 @wireprotocommand('capabilities', permission='pull')
 def capabilitiesv2(repo, proto):
-    caps = _capabilitiesv2(repo, proto)
-
-    return wireprototypes.cborresponse(caps)
+    yield _capabilitiesv2(repo, proto)
 
 @wireprotocommand('heads',
                   args={
                       'publiconly': False,
                   },
                   permission='pull')
 def headsv2(repo, proto, publiconly=False):
     if publiconly:
         repo = repo.filtered('immutable')
 
-    return wireprototypes.cborresponse(repo.heads())
+    yield repo.heads()
 
 @wireprotocommand('known',
                   args={
                       'nodes': [b'deadbeef'],
                   },
                   permission='pull')
 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={
                       'namespace': b'ns',
                   },
                   permission='pull')
 def listkeysv2(repo, proto, namespace=None):
     keys = repo.listkeys(encoding.tolocal(namespace))
     keys = {encoding.fromlocal(k): encoding.fromlocal(v)
             for k, v in keys.iteritems()}
 
-    return wireprototypes.cborresponse(keys)
+    yield keys
 
 @wireprotocommand('lookup',
                   args={
                       'key': b'foo',
                   },
                   permission='pull')
 def lookupv2(repo, proto, key):
     key = encoding.tolocal(key)
 
     # TODO handle exception.
     node = repo.lookup(key)
 
-    return wireprototypes.cborresponse(node)
+    yield node
 
 @wireprotocommand('pushkey',
                   args={
                       'namespace': b'ns',
                       'key': b'key',
                       'old': b'old',
                       'new': b'new',
                   },
                   permission='push')
 def pushkeyv2(repo, proto, namespace, key, old, new):
     # TODO handle ui output redirection
-    r = repo.pushkey(encoding.tolocal(namespace),
-                     encoding.tolocal(key),
-                     encoding.tolocal(old),
-                     encoding.tolocal(new))
-
-    return wireprototypes.cborresponse(r)
+    yield repo.pushkey(encoding.tolocal(namespace),
+                       encoding.tolocal(key),
+                       encoding.tolocal(old),
+                       encoding.tolocal(new))

tests/test-http-api-httpv2.t

@@ -1,571 +1,622 b''
 #require no-chg
 
   $ . $TESTDIR/wireprotohelpers.sh
   $ enabledummycommands
 
   $ hg init server
   $ cat > server/.hg/hgrc << EOF
   > [experimental]
   > web.apiserver = true
   > EOF
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
 
 HTTP v2 protocol not enabled by default
 
   $ sendhttpraw << EOF
   > httprequest GET api/$HTTPV2
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     GET /api/exp-http-v2-0001 HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 404 Not Found\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 33\r\n
   s>     \r\n
   s>     API exp-http-v2-0001 not enabled\n
 
 Restart server with support for HTTP v2 API
 
   $ killdaemons.py
   $ enablehttpv2 server
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
 
 Request to unknown command yields 404
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/badcommand
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/badcommand HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 404 Not Found\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 42\r\n
   s>     \r\n
   s>     unknown wire protocol command: badcommand\n
 
 GET to read-only command yields a 405
 
   $ sendhttpraw << EOF
   > httprequest GET api/$HTTPV2/ro/customreadonly
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     GET /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 405 Method Not Allowed\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Allow: POST\r\n
   s>     Content-Length: 30\r\n
   s>     \r\n
   s>     commands require POST requests
 
 Missing Accept header results in 406
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/customreadonly
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 406 Not Acceptable\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 85\r\n
   s>     \r\n
   s>     client MUST specify Accept header with value: application/mercurial-exp-framing-0005\n
 
 Bad Accept header results in 406
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/customreadonly
   >     accept: invalid
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: invalid\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 406 Not Acceptable\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 85\r\n
   s>     \r\n
   s>     client MUST specify Accept header with value: application/mercurial-exp-framing-0005\n
 
 Bad Content-Type header results in 415
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/customreadonly
   >     accept: $MEDIATYPE
   >     user-agent: test
   >     content-type: badmedia
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: badmedia\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 415 Unsupported Media Type\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 88\r\n
   s>     \r\n
   s>     client MUST send Content-Type header with value: application/mercurial-exp-framing-0005\n
 
 Request to read-only command works out of the box
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/customreadonly
   >     accept: $MEDIATYPE
   >     content-type: $MEDIATYPE
   >     user-agent: test
   >     frame 1 1 stream-begin command-request new cbor:{b'name': b'customreadonly'}
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     *\r\n (glob)
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     user-agent: test\r\n
   s>     content-length: 29\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s>     \x15\x00\x00\x01\x00\x01\x01\x11\xa1DnameNcustomreadonly
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
 
   $ sendhttpv2peer << EOF
   > command customreadonly
   > EOF
   creating http peer for wire protocol version 2
   sending customreadonly command
   s>     POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 29\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     \x15\x00\x00\x01\x00\x01\x01\x11\xa1DnameNcustomreadonly
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
     },
     b'customreadonly bytes response'
   ]
 
 Request to read-write command fails because server is read-only by default
 
 GET to read-write request yields 405
 
   $ sendhttpraw << EOF
   > httprequest GET api/$HTTPV2/rw/customreadonly
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     GET /api/exp-http-v2-0001/rw/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 405 Method Not Allowed\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Allow: POST\r\n
   s>     Content-Length: 30\r\n
   s>     \r\n
   s>     commands require POST requests
 
 Even for unknown commands
 
   $ sendhttpraw << EOF
   > httprequest GET api/$HTTPV2/rw/badcommand
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     GET /api/exp-http-v2-0001/rw/badcommand HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 405 Method Not Allowed\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Allow: POST\r\n
   s>     Content-Length: 30\r\n
   s>     \r\n
   s>     commands require POST requests
 
 SSL required by default
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/rw/customreadonly
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/rw/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 403 ssl required\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Length: 17\r\n
   s>     \r\n
   s>     permission denied
 
 Restart server to allow non-ssl read-write operations
 
   $ killdaemons.py
   $ cat > server/.hg/hgrc << EOF
   > [experimental]
   > web.apiserver = true
   > web.api.http-v2 = true
   > [web]
   > push_ssl = false
   > allow-push = *
   > EOF
 
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 Authorized request for valid read-write command works
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/rw/customreadonly
   >     user-agent: test
   >     accept: $MEDIATYPE
   >     content-type: $MEDIATYPE
   >     frame 1 1 stream-begin command-request new cbor:{b'name': b'customreadonly'}
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/rw/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     user-agent: test\r\n
   s>     content-length: 29\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s>     \x15\x00\x00\x01\x00\x01\x01\x11\xa1DnameNcustomreadonly
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
 
 Authorized request for unknown command is rejected
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/rw/badcommand
   >     user-agent: test
   >     accept: $MEDIATYPE
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/rw/badcommand HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 404 Not Found\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 42\r\n
   s>     \r\n
   s>     unknown wire protocol command: badcommand\n
 
 debugreflect isn't enabled by default
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/debugreflect
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/debugreflect HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 404 Not Found\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 34\r\n
   s>     \r\n
   s>     debugreflect service not available
 
 Restart server to get debugreflect endpoint
 
   $ killdaemons.py
   $ cat > server/.hg/hgrc << EOF
   > [experimental]
   > web.apiserver = true
   > web.api.debugreflect = true
   > web.api.http-v2 = true
   > [web]
   > push_ssl = false
   > allow-push = *
   > EOF
 
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 Command frames can be reflected via debugreflect
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/debugreflect
   >     accept: $MEDIATYPE
   >     content-type: $MEDIATYPE
   >     user-agent: test
   >     frame 1 1 stream-begin command-request new cbor:{b'name': b'command1', b'args': {b'foo': b'val1', b'bar1': b'val'}}
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/debugreflect HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     user-agent: test\r\n
   s>     content-length: 47\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s>     \'\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa2Dbar1CvalCfooDval1DnameHcommand1
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 205\r\n
   s>     \r\n
   s>     received: 1 1 1 \xa2Dargs\xa2Dbar1CvalCfooDval1DnameHcommand1\n
   s>     ["runcommand", {"args": {"bar1": "val", "foo": "val1"}, "command": "command1", "data": null, "requestid": 1}]\n
   s>     received: <no frame>\n
   s>     {"action": "noop"}
 
 Multiple requests to regular command URL are not allowed
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/customreadonly
   >     accept: $MEDIATYPE
   >     content-type: $MEDIATYPE
   >     user-agent: test
   >     frame 1 1 stream-begin command-request new cbor:{b'name': b'customreadonly'}
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     user-agent: test\r\n
   s>     content-length: 29\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s>     \x15\x00\x00\x01\x00\x01\x01\x11\xa1DnameNcustomreadonly
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
 
 Multiple requests to "multirequest" URL are allowed
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/multirequest
   >     accept: $MEDIATYPE
   >     content-type: $MEDIATYPE
   >     user-agent: test
   >     frame 1 1 stream-begin command-request new cbor:{b'name': b'customreadonly'}
   >     frame 3 1 0 command-request new cbor:{b'name': b'customreadonly'}
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     *\r\n (glob)
   s>     *\r\n (glob)
   s>     user-agent: test\r\n
   s>     content-length: 58\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s>     \x15\x00\x00\x01\x00\x01\x01\x11\xa1DnameNcustomreadonly\x15\x00\x00\x03\x00\x01\x00\x11\xa1DnameNcustomreadonly
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
 
 Interleaved requests to "multirequest" are processed
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/multirequest
   >     accept: $MEDIATYPE
   >     content-type: $MEDIATYPE
   >     user-agent: test
   >     frame 1 1 stream-begin command-request new|more \xa2Dargs\xa1Inamespace
   >     frame 3 1 0 command-request new|more \xa2Dargs\xa1Inamespace
   >     frame 3 1 0 command-request continuation JnamespacesDnameHlistkeys
   >     frame 1 1 0 command-request continuation IbookmarksDnameHlistkeys
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     user-agent: test\r\n
   s>     content-length: 115\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s>     \x11\x00\x00\x01\x00\x01\x01\x15\xa2Dargs\xa1Inamespace\x11\x00\x00\x03\x00\x01\x00\x15\xa2Dargs\xa1Inamespace\x19\x00\x00\x03\x00\x01\x00\x12JnamespacesDnameHlistkeys\x18\x00\x00\x01\x00\x01\x00\x12IbookmarksDnameHlistkeys
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
 
 Restart server to disable read-write access
 
   $ killdaemons.py
   $ cat > server/.hg/hgrc << EOF
   > [experimental]
   > web.apiserver = true
   > web.api.debugreflect = true
   > web.api.http-v2 = true
   > [web]
   > push_ssl = false
   > EOF
 
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 Attempting to run a read-write command via multirequest on read-only URL is not allowed
 
   $ sendhttpraw << EOF
   > httprequest POST api/$HTTPV2/ro/multirequest
   >     accept: $MEDIATYPE
   >     content-type: $MEDIATYPE
   >     user-agent: test
   >     frame 1 1 stream-begin command-request new cbor:{b'name': b'pushkey'}
   > EOF
   using raw connection to peer
   s>     POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     user-agent: test\r\n
   s>     content-length: 22\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s>     \x0e\x00\x00\x01\x00\x01\x01\x11\xa1DnameGpushkey
   s> makefile('rb', None)
   s>     HTTP/1.1 403 Forbidden\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 52\r\n
   s>     \r\n
   s>     insufficient permissions to execute command: pushkey
 
   $ cat error.log

tests/test-http-protocol.t

@@ -1,740 +1,749 b''
 #require no-chg
 
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ cat >> $HGRCPATH << EOF
   > [web]
   > push_ssl = false
   > allow_push = *
   > EOF
 
   $ hg init server
   $ cd server
   $ touch a
   $ hg -q commit -A -m initial
   $ cd ..
 
   $ hg serve -R server -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid >> $DAEMON_PIDS
 
 compression formats are advertised in compression capability
 
 #if zstd
   $ get-with-headers.py $LOCALIP:$HGPORT '?cmd=capabilities' | tr ' ' '\n' | grep '^compression=zstd,zlib$' > /dev/null
 #else
   $ get-with-headers.py $LOCALIP:$HGPORT '?cmd=capabilities' | tr ' ' '\n' | grep '^compression=zlib$' > /dev/null
 #endif
 
   $ killdaemons.py
 
 server.compressionengines can replace engines list wholesale
 
   $ hg serve --config server.compressionengines=none -R server -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
   $ get-with-headers.py $LOCALIP:$HGPORT '?cmd=capabilities' | tr ' ' '\n' | grep '^compression=none$' > /dev/null
 
   $ killdaemons.py
 
 Order of engines can also change
 
   $ hg serve --config server.compressionengines=none,zlib -R server -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
   $ get-with-headers.py $LOCALIP:$HGPORT '?cmd=capabilities' | tr ' ' '\n' | grep '^compression=none,zlib$' > /dev/null
 
   $ killdaemons.py
 
 Start a default server again
 
   $ hg serve -R server -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
 
 Server should send application/mercurial-0.1 to clients if no Accept is used
 
   $ get-with-headers.py --headeronly $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000' -
   200 Script output follows
   content-type: application/mercurial-0.1
   date: $HTTP_DATE$
   server: testing stub value
   transfer-encoding: chunked
 
 Server should send application/mercurial-0.1 when client says it wants it
 
   $ get-with-headers.py --hgproto '0.1' --headeronly $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000' -
   200 Script output follows
   content-type: application/mercurial-0.1
   date: $HTTP_DATE$
   server: testing stub value
   transfer-encoding: chunked
 
 Server should send application/mercurial-0.2 when client says it wants it
 
   $ get-with-headers.py --hgproto '0.2' --headeronly $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000' -
   200 Script output follows
   content-type: application/mercurial-0.2
   date: $HTTP_DATE$
   server: testing stub value
   transfer-encoding: chunked
 
   $ get-with-headers.py --hgproto '0.1 0.2' --headeronly $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000' -
   200 Script output follows
   content-type: application/mercurial-0.2
   date: $HTTP_DATE$
   server: testing stub value
   transfer-encoding: chunked
 
 Requesting a compression format that server doesn't support results will fall back to 0.1
 
   $ get-with-headers.py --hgproto '0.2 comp=aa' --headeronly $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000' -
   200 Script output follows
   content-type: application/mercurial-0.1
   date: $HTTP_DATE$
   server: testing stub value
   transfer-encoding: chunked
 
 #if zstd
 zstd is used if available
 
   $ get-with-headers.py --hgproto '0.2 comp=zstd' $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000' > resp
   $ f --size --hexdump --bytes 36 --sha1 resp
   resp: size=248, sha1=4d8d8f87fb82bd542ce52881fdc94f850748
   0000: 32 30 30 20 53 63 72 69 70 74 20 6f 75 74 70 75 |200 Script outpu|
   0010: 74 20 66 6f 6c 6c 6f 77 73 0a 0a 04 7a 73 74 64 |t follows...zstd|
   0020: 28 b5 2f fd                                     |(./.|
 
 #endif
 
 application/mercurial-0.2 is not yet used on non-streaming responses
 
   $ get-with-headers.py --hgproto '0.2' $LOCALIP:$HGPORT '?cmd=heads' -
   200 Script output follows
   content-length: 41
   content-type: application/mercurial-0.1
   date: $HTTP_DATE$
   server: testing stub value
   
   e93700bd72895c5addab234c56d4024b487a362f
 
 Now test protocol preference usage
 
   $ killdaemons.py
   $ hg serve --config server.compressionengines=none,zlib -R server -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
 
 No Accept will send 0.1+zlib, even though "none" is preferred b/c "none" isn't supported on 0.1
 
   $ get-with-headers.py --headeronly $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000' Content-Type
   200 Script output follows
   content-type: application/mercurial-0.1
 
   $ get-with-headers.py $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000'  > resp
   $ f --size --hexdump --bytes 28 --sha1 resp
   resp: size=227, sha1=35a4c074da74f32f5440da3cbf04
   0000: 32 30 30 20 53 63 72 69 70 74 20 6f 75 74 70 75 |200 Script outpu|
   0010: 74 20 66 6f 6c 6c 6f 77 73 0a 0a 78             |t follows..x|
 
 Explicit 0.1 will send zlib because "none" isn't supported on 0.1
 
   $ get-with-headers.py --hgproto '0.1' $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000'  > resp
   $ f --size --hexdump --bytes 28 --sha1 resp
   resp: size=227, sha1=35a4c074da74f32f5440da3cbf04
   0000: 32 30 30 20 53 63 72 69 70 74 20 6f 75 74 70 75 |200 Script outpu|
   0010: 74 20 66 6f 6c 6c 6f 77 73 0a 0a 78             |t follows..x|
 
 0.2 with no compression will get "none" because that is server's preference
 (spec says ZL and UN are implicitly supported)
 
   $ get-with-headers.py --hgproto '0.2' $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000'  > resp
   $ f --size --hexdump --bytes 32 --sha1 resp
   resp: size=432, sha1=ac931b412ec185a02e0e5bcff98dac83
   0000: 32 30 30 20 53 63 72 69 70 74 20 6f 75 74 70 75 |200 Script outpu|
   0010: 74 20 66 6f 6c 6c 6f 77 73 0a 0a 04 6e 6f 6e 65 |t follows...none|
 
 Client receives server preference even if local order doesn't match
 
   $ get-with-headers.py --hgproto '0.2 comp=zlib,none' $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000'  > resp
   $ f --size --hexdump --bytes 32 --sha1 resp
   resp: size=432, sha1=ac931b412ec185a02e0e5bcff98dac83
   0000: 32 30 30 20 53 63 72 69 70 74 20 6f 75 74 70 75 |200 Script outpu|
   0010: 74 20 66 6f 6c 6c 6f 77 73 0a 0a 04 6e 6f 6e 65 |t follows...none|
 
 Client receives only supported format even if not server preferred format
 
   $ get-with-headers.py --hgproto '0.2 comp=zlib' $LOCALIP:$HGPORT '?cmd=getbundle&heads=e93700bd72895c5addab234c56d4024b487a362f&common=0000000000000000000000000000000000000000'  > resp
   $ f --size --hexdump --bytes 33 --sha1 resp
   resp: size=232, sha1=a1c727f0c9693ca15742a75c30419bc36
   0000: 32 30 30 20 53 63 72 69 70 74 20 6f 75 74 70 75 |200 Script outpu|
   0010: 74 20 66 6f 6c 6c 6f 77 73 0a 0a 04 7a 6c 69 62 |t follows...zlib|
   0020: 78                                              |x|
 
   $ killdaemons.py
   $ cd ..
 
 Test listkeys for listing namespaces
 
   $ hg init empty
   $ hg -R empty serve -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
 
   $ hg --verbose debugwireproto http://$LOCALIP:$HGPORT << EOF
   > command listkeys
   >     namespace namespaces
   > EOF
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   sending listkeys command
   s>     GET /?cmd=listkeys HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     vary: X-HgArg-1,X-HgProto-1\r\n
   s>     x-hgarg-1: namespace=namespaces\r\n
   s>     x-hgproto-1: 0.1 0.2 comp=$USUAL_COMPRESSIONS$ partial-pull\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 30\r\n
   s>     \r\n
   s>     bookmarks\t\n
   s>     namespaces\t\n
   s>     phases\t
   response: {
     b'bookmarks': b'',
     b'namespaces': b'',
     b'phases': b''
   }
 
 Same thing, but with "httprequest" command
 
   $ hg --verbose debugwireproto --peer raw http://$LOCALIP:$HGPORT << EOF
   > httprequest GET ?cmd=listkeys
   >     user-agent: test
   >     x-hgarg-1: namespace=namespaces
   > EOF
   using raw connection to peer
   s>     GET /?cmd=listkeys HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     x-hgarg-1: namespace=namespaces\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 30\r\n
   s>     \r\n
   s>     bookmarks\t\n
   s>     namespaces\t\n
   s>     phases\t
 
 Client with HTTPv2 enabled advertises that and gets old capabilities response from old server
 
   $ hg --config experimental.httppeer.advertise-v2=true --verbose debugwireproto http://$LOCALIP:$HGPORT << EOF
   > command heads
   > EOF
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     vary: X-HgProto-1,X-HgUpgrade-1\r\n
   s>     x-hgproto-1: cbor\r\n
   s>     x-hgupgrade-1: exp-http-v2-0001\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   sending heads command
   s>     GET /?cmd=heads HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     vary: X-HgProto-1\r\n
   s>     x-hgproto-1: 0.1 0.2 comp=$USUAL_COMPRESSIONS$ partial-pull\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 41\r\n
   s>     \r\n
   s>     0000000000000000000000000000000000000000\n
   response: [
     b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
   ]
 
   $ killdaemons.py
   $ enablehttpv2 empty
   $ hg --config server.compressionengines=zlib -R empty serve -p $HGPORT -d --pid-file hg.pid
   $ cat hg.pid > $DAEMON_PIDS
 
 Client with HTTPv2 enabled automatically upgrades if the server supports it
 
   $ hg --config experimental.httppeer.advertise-v2=true --verbose debugwireproto http://$LOCALIP:$HGPORT << EOF
   > command heads
   > EOF
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     vary: X-HgProto-1,X-HgUpgrade-1\r\n
   s>     x-hgproto-1: cbor\r\n
   s>     x-hgupgrade-1: exp-http-v2-0001\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-cbor\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     \xa3GapibaseDapi/Dapis\xa1Pexp-http-v2-0001\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\x82LgeneraldeltaHrevlogv1Nv1capabilitiesY\x01\xc5batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   sending heads command
   s>     POST /api/exp-http-v2-0001/ro/heads HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 20\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     \x0c\x00\x00\x01\x00\x01\x01\x11\xa1DnameEheads
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
   ]
 
   $ killdaemons.py
 
 HTTP client follows HTTP redirect on handshake to new repo
 
   $ cd $TESTTMP
 
   $ hg init redirector
   $ hg init redirected
   $ cd redirected
   $ touch foo
   $ hg -q commit -A -m initial
   $ cd ..
 
   $ cat > paths.conf << EOF
   > [paths]
   > / = $TESTTMP/*
   > EOF
 
   $ cat > redirectext.py << EOF
   > from mercurial import extensions, wireprotoserver
   > def wrappedcallhttp(orig, repo, req, res, proto, cmd):
   >     path = req.advertisedurl[len(req.advertisedbaseurl):]
   >     if not path.startswith(b'/redirector'):
   >         return orig(repo, req, res, proto, cmd)
   >     relpath = path[len(b'/redirector'):]
   >     res.status = b'301 Redirect'
   >     newurl = b'%s/redirected%s' % (req.baseurl, relpath)
   >     if not repo.ui.configbool('testing', 'redirectqs', True) and b'?' in newurl:
   >         newurl = newurl[0:newurl.index(b'?')]
   >     res.headers[b'Location'] = newurl
   >     res.headers[b'Content-Type'] = b'text/plain'
   >     res.setbodybytes(b'redirected')
   >     return True
   > 
   > extensions.wrapfunction(wireprotoserver, '_callhttp', wrappedcallhttp)
   > EOF
 
   $ hg --config extensions.redirect=$TESTTMP/redirectext.py \
   >    --config server.compressionengines=zlib \
   >     serve --web-conf paths.conf --pid-file hg.pid -p $HGPORT -d
   $ cat hg.pid > $DAEMON_PIDS
 
 Verify our HTTP 301 is served properly
 
   $ hg --verbose debugwireproto --peer raw http://$LOCALIP:$HGPORT << EOF
   > httprequest GET /redirector?cmd=capabilities
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     GET /redirector?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 301 Redirect\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Location: http://$LOCALIP:$HGPORT/redirected?cmd=capabilities\r\n (glob)
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 10\r\n
   s>     \r\n
   s>     redirected
   s>     GET /redirected?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 453\r\n
   s>     \r\n
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
 
 Test with the HTTP peer
 
   $ hg --verbose debugwireproto http://$LOCALIP:$HGPORT/redirector << EOF
   > command heads
   > EOF
   s>     GET /redirector?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 301 Redirect\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Location: http://$LOCALIP:$HGPORT/redirected?cmd=capabilities\r\n (glob)
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 10\r\n
   s>     \r\n
   s>     redirected
   s>     GET /redirected?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 453\r\n
   s>     \r\n
   real URL is http://$LOCALIP:$HGPORT/redirected (glob)
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   sending heads command
   s>     GET /redirected?cmd=heads HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     vary: X-HgProto-1\r\n
   s>     x-hgproto-1: 0.1 0.2 comp=$USUAL_COMPRESSIONS$ partial-pull\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 41\r\n
   s>     \r\n
   s>     96ee1d7354c4ad7372047672c36a1f561e3a6a4c\n
   response: [
     b'\x96\xee\x1dsT\xc4\xadsr\x04vr\xc3j\x1fV\x1e:jL'
   ]
 
   $ killdaemons.py
 
 Now test a variation where we strip the query string from the redirect URL.
 (SCM Manager apparently did this and clients would recover from it)
 
   $ hg --config extensions.redirect=$TESTTMP/redirectext.py \
   >    --config server.compressionengines=zlib \
   >    --config testing.redirectqs=false \
   >     serve --web-conf paths.conf --pid-file hg.pid -p $HGPORT -d
   $ cat hg.pid > $DAEMON_PIDS
 
   $ hg --verbose debugwireproto --peer raw http://$LOCALIP:$HGPORT << EOF
   > httprequest GET /redirector?cmd=capabilities
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     GET /redirector?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 301 Redirect\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Location: http://$LOCALIP:$HGPORT/redirected\r\n (glob)
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 10\r\n
   s>     \r\n
   s>     redirected
   s>     GET /redirected HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     ETag: W/"*"\r\n (glob)
   s>     Content-Type: text/html; charset=ascii\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
   s>     414\r\n
   s>     <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN" "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">\n
   s>     <html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en-US">\n
   s>     <head>\n
   s>     <link rel="icon" href="/redirected/static/hgicon.png" type="image/png" />\n
   s>     <meta name="robots" content="index, nofollow" />\n
   s>     <link rel="stylesheet" href="/redirected/static/style-paper.css" type="text/css" />\n
   s>     <script type="text/javascript" src="/redirected/static/mercurial.js"></script>\n
   s>     \n
   s>     <title>redirected: log</title>\n
   s>     <link rel="alternate" type="application/atom+xml"\n
   s>        href="/redirected/atom-log" title="Atom feed for redirected" />\n
   s>     <link rel="alternate" type="application/rss+xml"\n
   s>        href="/redirected/rss-log" title="RSS feed for redirected" />\n
   s>     </head>\n
   s>     <body>\n
   s>     \n
   s>     <div class="container">\n
   s>     <div class="menu">\n
   s>     <div class="logo">\n
   s>     <a href="https://mercurial-scm.org/">\n
   s>     <img src="/redirected/static/hglogo.png" alt="mercurial" /></a>\n
   s>     </div>\n
   s>     <ul>\n
   s>     <li class="active">log</li>\n
   s>     <li><a href="/redirected/graph/tip">graph</a></li>\n
   s>     <li><a href="/redirected/tags">tags</a></li>\n
   s>     <li><a href="
   s>     \r\n
   s>     810\r\n
   s>     /redirected/bookmarks">bookmarks</a></li>\n
   s>     <li><a href="/redirected/branches">branches</a></li>\n
   s>     </ul>\n
   s>     <ul>\n
   s>     <li><a href="/redirected/rev/tip">changeset</a></li>\n
   s>     <li><a href="/redirected/file/tip">browse</a></li>\n
   s>     </ul>\n
   s>     <ul>\n
   s>     \n
   s>     </ul>\n
   s>     <ul>\n
   s>      <li><a href="/redirected/help">help</a></li>\n
   s>     </ul>\n
   s>     <div class="atom-logo">\n
   s>     <a href="/redirected/atom-log" title="subscribe to atom feed">\n
   s>     <img class="atom-logo" src="/redirected/static/feed-icon-14x14.png" alt="atom feed" />\n
   s>     </a>\n
   s>     </div>\n
   s>     </div>\n
   s>     \n
   s>     <div class="main">\n
   s>     <h2 class="breadcrumb"><a href="/">Mercurial</a> &gt; <a href="/redirected">redirected</a> </h2>\n
   s>     <h3>log</h3>\n
   s>     \n
   s>     \n
   s>     <form class="search" action="/redirected/log">\n
   s>     \n
   s>     <p><input name="rev" id="search1" type="text" size="30" value="" /></p>\n
   s>     <div id="hint">Find changesets by keywords (author, files, the commit message), revision\n
   s>     number or hash, or <a href="/redirected/help/revsets">revset expression</a>.</div>\n
   s>     </form>\n
   s>     \n
   s>     <div class="navigate">\n
   s>     <a href="/redirected/shortlog/tip?revcount=30">less</a>\n
   s>     <a href="/redirected/shortlog/tip?revcount=120">more</a>\n
   s>     | rev 0: <a href="/redirected/shortlog/96ee1d7354c4">(0)</a> <a href="/redirected/shortlog/tip">tip</a> \n
   s>     </div>\n
   s>     \n
   s>     <table class="bigtable">\n
   s>     <thead>\n
   s>      <tr>\n
   s>       <th class="age">age</th>\n
   s>       <th class="author">author</th>\n
   s>       <th class="description">description</th>\n
   s>      </tr>\n
   s>     </thead>\n
   s>     <tbody class="stripes2">\n
   s>      <tr>\n
   s>       <td class="age">Thu, 01 Jan 1970 00:00:00 +0000</td>\n
   s>       <td class="author">test</td>\n
   s>       <td class="description">\n
   s>        <a href="/redirected/rev/96ee1d7354c4">initial</a>\n
   s>        <span class="phase">draft</span> <span class="branchhead">default</span> <span class="tag">tip</span> \n
   s>       </td>\n
   s>      </tr>\n
   s>     \n
   s>     </tbody>\n
   s>     </table>\n
   s>     \n
   s>     <div class="navigate">\n
   s>     <a href="/redirected/shortlog/tip?revcount=30">less</a>\n
   s>     <a href="/redirected/shortlog/tip?revcount=120">more</a>\n
   s>     | rev 0: <a href="/redirected/shortlog/96ee1d7354c4">(0)</a> <a href="/redirected/shortlog/tip">tip</a> \n
   s>     </div>\n
   s>     \n
   s>     <script type="text/javascript">\n
   s>         ajaxScrollInit(\n
   s>                 \'/redirected/shortlog/%next%\',\n
   s>                 \'\', <!-- NEXTHASH\n
   s>                 function (htmlText) {
   s>     \r\n
   s>     14a\r\n
   s>     \n
   s>                     var m = htmlText.match(/\'(\\w+)\', <!-- NEXTHASH/);\n
   s>                     return m ? m[1] : null;\n
   s>                 },\n
   s>                 \'.bigtable > tbody\',\n
   s>                 \'<tr class="%class%">\\\n
   s>                 <td colspan="3" style="text-align: center;">%text%</td>\\\n
   s>                 </tr>\'\n
   s>         );\n
   s>     </script>\n
   s>     \n
   s>     </div>\n
   s>     </div>\n
   s>     \n
   s>     \n
   s>     \n
   s>     </body>\n
   s>     </html>\n
   s>     \n
   s>     \r\n
   s>     0\r\n
   s>     \r\n
 
   $ hg --verbose debugwireproto http://$LOCALIP:$HGPORT/redirector << EOF
   > command heads
   > EOF
   s>     GET /redirector?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 301 Redirect\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Location: http://$LOCALIP:$HGPORT/redirected\r\n (glob)
   s>     Content-Type: text/plain\r\n
   s>     Content-Length: 10\r\n
   s>     \r\n
   s>     redirected
   s>     GET /redirected HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     ETag: W/"*"\r\n (glob)
   s>     Content-Type: text/html; charset=ascii\r\n
   s>     Transfer-Encoding: chunked\r\n
   s>     \r\n
   real URL is http://$LOCALIP:$HGPORT/redirected (glob)
   s>     414\r\n
   s>     <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN" "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">\n
   s>     <html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en-US">\n
   s>     <head>\n
   s>     <link rel="icon" href="/redirected/static/hgicon.png" type="image/png" />\n
   s>     <meta name="robots" content="index, nofollow" />\n
   s>     <link rel="stylesheet" href="/redirected/static/style-paper.css" type="text/css" />\n
   s>     <script type="text/javascript" src="/redirected/static/mercurial.js"></script>\n
   s>     \n
   s>     <title>redirected: log</title>\n
   s>     <link rel="alternate" type="application/atom+xml"\n
   s>        href="/redirected/atom-log" title="Atom feed for redirected" />\n
   s>     <link rel="alternate" type="application/rss+xml"\n
   s>        href="/redirected/rss-log" title="RSS feed for redirected" />\n
   s>     </head>\n
   s>     <body>\n
   s>     \n
   s>     <div class="container">\n
   s>     <div class="menu">\n
   s>     <div class="logo">\n
   s>     <a href="https://mercurial-scm.org/">\n
   s>     <img src="/redirected/static/hglogo.png" alt="mercurial" /></a>\n
   s>     </div>\n
   s>     <ul>\n
   s>     <li class="active">log</li>\n
   s>     <li><a href="/redirected/graph/tip">graph</a></li>\n
   s>     <li><a href="/redirected/tags">tags</a
   s>     GET /redirected?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 453\r\n
   s>     \r\n
   real URL is http://$LOCALIP:$HGPORT/redirected (glob)
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   sending heads command
   s>     GET /redirected?cmd=heads HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     vary: X-HgProto-1\r\n
   s>     x-hgproto-1: 0.1 0.2 comp=$USUAL_COMPRESSIONS$ partial-pull\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: 41\r\n
   s>     \r\n
   s>     96ee1d7354c4ad7372047672c36a1f561e3a6a4c\n
   response: [
     b'\x96\xee\x1dsT\xc4\xadsr\x04vr\xc3j\x1fV\x1e:jL'
   ]

tests/test-wireproto-command-branchmap.t

@@ -1,83 +1,92 b''
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ hg init server
   $ enablehttpv2 server
   $ cd server
   $ hg debugdrawdag << EOF
   > C D
   > |/
   > B
   > |
   > A
   > EOF
 
   $ hg up B
   2 files updated, 0 files merged, 0 files removed, 0 files unresolved
   $ hg branch branch1
   marked working directory as branch branch1
   (branches are permanent and global, did you want a bookmark?)
   $ echo b1 > foo
   $ hg -q commit -A -m 'branch 1'
   $ hg up B
   0 files updated, 0 files merged, 1 files removed, 0 files unresolved
   $ hg branch branch2
   marked working directory as branch branch2
   $ echo b2 > foo
   $ hg -q commit -A -m 'branch 2'
 
   $ hg log -T '{rev}:{node} {branch} {desc}\n'
   5:224161c7589aa48fa83a48feff5e95b56ae327fc branch2 branch 2
   4:b5faacdfd2633768cb3152336cc0953381266688 branch1 branch 1
   3:be0ef73c17ade3fc89dc41701eb9fc3a91b58282 default D
   2:26805aba1e600a82e93661149f2313866a221a7b default C
   1:112478962961147124edd43549aedd1a335e44bf default B
   0:426bada5c67598ca65036d57d9e4b64b0c1ce7a0 default A
 
   $ hg serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 No arguments returns something reasonable
 
   $ sendhttpv2peer << EOF
   > command branchmap
   > EOF
   creating http peer for wire protocol version 2
   sending branchmap command
   s>     POST /api/exp-http-v2-0001/ro/branchmap HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 24\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     \x10\x00\x00\x01\x00\x01\x01\x11\xa1DnameIbranchmap
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
     ],
     b'branch2': [
       b'"Aa\xc7X\x9a\xa4\x8f\xa8:H\xfe\xff^\x95\xb5j\xe3\'\xfc'
     ],
     b'default': [
       b'&\x80Z\xba\x1e`\n\x82\xe96a\x14\x9f#\x13\x86j"\x1a{',
       b'\xbe\x0e\xf7<\x17\xad\xe3\xfc\x89\xdcAp\x1e\xb9\xfc:\x91\xb5\x82\x82'
     ]
   }
 
   $ cat error.log

tests/test-wireproto-command-capabilities.t

@@ -1,422 +1,431 b''
 #require no-chg
 
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ hg init server
 
 zstd isn't present in plain builds. Make tests easier by removing
 zstd from the equation.
 
   $ cat >> server/.hg/hgrc << EOF
   > [server]
   > compressionengines = zlib
   > EOF
 
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 A normal capabilities request is serviced for version 1
 
   $ sendhttpraw << EOF
   > httprequest GET ?cmd=capabilities
   >     user-agent: test
   > EOF
   using raw connection to peer
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
 
 A proper request without the API server enabled returns the legacy response
 
   $ sendhttpraw << EOF
   > httprequest GET ?cmd=capabilities
   >    user-agent: test
   >    x-hgupgrade-1: foo
   >    x-hgproto-1: cbor
   > EOF
   using raw connection to peer
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     x-hgproto-1: cbor\r\n
   s>     x-hgupgrade-1: foo\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
 
 Restart with just API server enabled. This enables serving the new format.
 
   $ killdaemons.py
   $ cat error.log
 
   $ cat >> server/.hg/hgrc << EOF
   > [experimental]
   > web.apiserver = true
   > EOF
 
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 X-HgUpgrade-<N> without CBOR advertisement uses legacy response
 
   $ sendhttpraw << EOF
   > httprequest GET ?cmd=capabilities
   >    user-agent: test
   >    x-hgupgrade-1: foo bar
   > EOF
   using raw connection to peer
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     x-hgupgrade-1: foo bar\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
 
 X-HgUpgrade-<N> without known serialization in X-HgProto-<N> uses legacy response
 
   $ sendhttpraw << EOF
   > httprequest GET ?cmd=capabilities
   >    user-agent: test
   >    x-hgupgrade-1: foo bar
   >    x-hgproto-1: some value
   > EOF
   using raw connection to peer
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     x-hgproto-1: some value\r\n
   s>     x-hgupgrade-1: foo bar\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 Script output follows\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-0.1\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
 
 X-HgUpgrade-<N> + X-HgProto-<N> headers trigger new response format
 
   $ sendhttpraw << EOF
   > httprequest GET ?cmd=capabilities
   >    user-agent: test
   >    x-hgupgrade-1: foo bar
   >    x-hgproto-1: cbor
   > EOF
   using raw connection to peer
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     x-hgproto-1: cbor\r\n
   s>     x-hgupgrade-1: foo bar\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-cbor\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     \xa3GapibaseDapi/Dapis\xa0Nv1capabilitiesY\x01\xc5batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   cbor> {
     b'apibase': b'api/',
     b'apis': {},
     b'v1capabilities': b'batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash'
   }
 
 Restart server to enable HTTPv2
 
   $ killdaemons.py
   $ enablehttpv2 server
   $ hg -R server serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 Only requested API services are returned
 
   $ sendhttpraw << EOF
   > httprequest GET ?cmd=capabilities
   >    user-agent: test
   >    x-hgupgrade-1: foo bar
   >    x-hgproto-1: cbor
   > EOF
   using raw connection to peer
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     x-hgproto-1: cbor\r\n
   s>     x-hgupgrade-1: foo bar\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-cbor\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     \xa3GapibaseDapi/Dapis\xa0Nv1capabilitiesY\x01\xc5batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   cbor> {
     b'apibase': b'api/',
     b'apis': {},
     b'v1capabilities': b'batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash'
   }
 
 Request for HTTPv2 service returns information about it
 
   $ sendhttpraw << EOF
   > httprequest GET ?cmd=capabilities
   >    user-agent: test
   >    x-hgupgrade-1: exp-http-v2-0001 foo bar
   >    x-hgproto-1: cbor
   > EOF
   using raw connection to peer
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     user-agent: test\r\n
   s>     x-hgproto-1: cbor\r\n
   s>     x-hgupgrade-1: exp-http-v2-0001 foo bar\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-cbor\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     \xa3GapibaseDapi/Dapis\xa1Pexp-http-v2-0001\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\x82LgeneraldeltaHrevlogv1Nv1capabilitiesY\x01\xc5batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   cbor> {
     b'apibase': b'api/',
     b'apis': {
       b'exp-http-v2-0001': {
         b'commands': {
           b'branchmap': {
             b'args': {},
             b'permissions': [
               b'pull'
             ]
           },
           b'capabilities': {
             b'args': {},
             b'permissions': [
               b'pull'
             ]
           },
           b'heads': {
             b'args': {
               b'publiconly': False
             },
             b'permissions': [
               b'pull'
             ]
           },
           b'known': {
             b'args': {
               b'nodes': [
                 b'deadbeef'
               ]
             },
             b'permissions': [
               b'pull'
             ]
           },
           b'listkeys': {
             b'args': {
               b'namespace': b'ns'
             },
             b'permissions': [
               b'pull'
             ]
           },
           b'lookup': {
             b'args': {
               b'key': b'foo'
             },
             b'permissions': [
               b'pull'
             ]
           },
           b'pushkey': {
             b'args': {
               b'key': b'key',
               b'namespace': b'ns',
               b'new': b'new',
               b'old': b'old'
             },
             b'permissions': [
               b'push'
             ]
           }
         },
         b'compression': [
           {
             b'name': b'zlib'
           }
         ],
         b'framingmediatypes': [
           b'application/mercurial-exp-framing-0005'
         ],
         b'rawrepoformats': [
           b'generaldelta',
           b'revlogv1'
         ]
       }
     },
     b'v1capabilities': b'batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash'
   }
 
 capabilities command returns expected info
 
   $ sendhttpv2peerhandshake << EOF
   > command capabilities
   > EOF
   creating http peer for wire protocol version 2
   s>     GET /?cmd=capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     vary: X-HgProto-1,X-HgUpgrade-1\r\n
   s>     x-hgproto-1: cbor\r\n
   s>     x-hgupgrade-1: exp-http-v2-0001\r\n
   s>     accept: application/mercurial-0.1\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   s>     Content-Type: application/mercurial-cbor\r\n
   s>     Content-Length: *\r\n (glob)
   s>     \r\n
   s>     \xa3GapibaseDapi/Dapis\xa1Pexp-http-v2-0001\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\x82LgeneraldeltaHrevlogv1Nv1capabilitiesY\x01\xc5batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
   sending capabilities command
   s>     POST /api/exp-http-v2-0001/ro/capabilities HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     *\r\n (glob)
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 27\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     \x13\x00\x00\x01\x00\x01\x01\x11\xa1DnameLcapabilities
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
     },
     {
       b'commands': {
         b'branchmap': {
           b'args': {},
           b'permissions': [
             b'pull'
           ]
         },
         b'capabilities': {
           b'args': {},
           b'permissions': [
             b'pull'
           ]
         },
         b'heads': {
           b'args': {
             b'publiconly': False
           },
           b'permissions': [
             b'pull'
           ]
         },
         b'known': {
           b'args': {
             b'nodes': [
               b'deadbeef'
             ]
           },
           b'permissions': [
             b'pull'
           ]
         },
         b'listkeys': {
           b'args': {
             b'namespace': b'ns'
           },
           b'permissions': [
             b'pull'
           ]
         },
         b'lookup': {
           b'args': {
             b'key': b'foo'
           },
           b'permissions': [
             b'pull'
           ]
         },
         b'pushkey': {
           b'args': {
             b'key': b'key',
             b'namespace': b'ns',
             b'new': b'new',
             b'old': b'old'
           },
           b'permissions': [
             b'push'
           ]
         }
       },
       b'compression': [
         {
           b'name': b'zlib'
         }
       ],
       b'framingmediatypes': [
         b'application/mercurial-exp-framing-0005'
       ],
       b'rawrepoformats': [
         b'generaldelta',
         b'revlogv1'
       ]
     }
   ]
 
   $ cat error.log

tests/test-wireproto-command-heads.t

@@ -1,102 +1,120 b''
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ hg init server
   $ enablehttpv2 server
   $ cd server
   $ hg debugdrawdag << EOF
   > H I J
   > | | |
   > E F G
   > | |/
   > C D
   > |/
   > B
   > |
   > A
   > EOF
 
   $ hg phase --force --secret J
   $ hg phase --public E
 
   $ hg log -r 'E + H + I + G + J' -T '{rev}:{node} {desc} {phase}\n'
   4:78d2dca436b2f5b188ac267e29b81e07266d38fc E public
   7:ae492e36b0c8339ffaf328d00b85b4525de1165e H draft
   8:1d6f6b91d44aaba6d5e580bc30a9948530dbe00b I draft
   6:29446d2dc5419c5f97447a8bc062e4cc328bf241 G draft
   9:dec04b246d7cbb670c6689806c05ad17c835284e J secret
 
   $ hg serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 All non-secret heads returned by default
 
   $ sendhttpv2peer << EOF
   > command heads
   > EOF
   creating http peer for wire protocol version 2
   sending heads command
   s>     POST /api/exp-http-v2-0001/ro/heads HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 20\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     \x0c\x00\x00\x01\x00\x01\x01\x11\xa1DnameEheads
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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^',
     b')Dm-\xc5A\x9c_\x97Dz\x8b\xc0b\xe4\xcc2\x8b\xf2A'
   ]
 
 Requesting just the public heads works
 
   $ sendhttpv2peer << EOF
   > command heads
   >     publiconly 1
   > EOF
   creating http peer for wire protocol version 2
   sending heads command
   s>     POST /api/exp-http-v2-0001/ro/heads HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 39\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     \x1f\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1JpubliconlyA1DnameEheads
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
   ]
 
   $ cat error.log

tests/test-wireproto-command-known.t

@@ -1,127 +1,154 b''
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ hg init server
   $ enablehttpv2 server
   $ cd server
   $ hg debugdrawdag << EOF
   > C D
   > |/
   > B
   > |
   > A
   > EOF
 
   $ hg log -T '{rev}:{node} {desc}\n'
   3:be0ef73c17ade3fc89dc41701eb9fc3a91b58282 D
   2:26805aba1e600a82e93661149f2313866a221a7b C
   1:112478962961147124edd43549aedd1a335e44bf B
   0:426bada5c67598ca65036d57d9e4b64b0c1ce7a0 A
 
   $ hg serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 No arguments returns something reasonable
 
   $ sendhttpv2peer << EOF
   > command known
   > EOF
   creating http peer for wire protocol version 2
   sending known command
   s>     POST /api/exp-http-v2-0001/ro/known HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 20\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     \x0c\x00\x00\x01\x00\x01\x01\x11\xa1DnameEknown
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
 
   $ sendhttpv2peer << EOF
   > command known
   >     nodes eval:[b'\x42\x6b\xad\xa5\xc6\x75\x98\xca\x65\x03\x6d\x57\xd9\xe4\xb6\x4b\x0c\x1c\xe7\xa0']
   > EOF
   creating http peer for wire protocol version 2
   sending known command
   s>     POST /api/exp-http-v2-0001/ro/known HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 54\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     .\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1Enodes\x81TBk\xad\xa5\xc6u\x98\xcae\x03mW\xd9\xe4\xb6K\x0c\x1c\xe7\xa0DnameEknown
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
   ]
 
 Multiple nodes works
 
   $ sendhttpv2peer << EOF
   > command known
   >     nodes eval:[b'\x42\x6b\xad\xa5\xc6\x75\x98\xca\x65\x03\x6d\x57\xd9\xe4\xb6\x4b\x0c\x1c\xe7\xa0', b'00000000000000000000', b'\x11\x24\x78\x96\x29\x61\x14\x71\x24\xed\xd4\x35\x49\xae\xdd\x1a\x33\x5e\x44\xbf']
   > EOF
   creating http peer for wire protocol version 2
   sending known command
   s>     POST /api/exp-http-v2-0001/ro/known HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 96\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     X\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1Enodes\x83TBk\xad\xa5\xc6u\x98\xcae\x03mW\xd9\xe4\xb6K\x0c\x1c\xe7\xa0T00000000000000000000T\x11$x\x96)a\x14q$\xed\xd45I\xae\xdd\x1a3^D\xbfDnameEknown
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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,
     True
   ]
 
   $ cat error.log

tests/test-wireproto-command-listkeys.t

@@ -1,134 +1,161 b''
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ hg init server
   $ enablehttpv2 server
   $ cd server
   $ hg debugdrawdag << EOF
   > C D
   > |/
   > B
   > |
   > A
   > EOF
 
   $ hg phase --public -r C
   $ hg book -r C @
 
   $ hg log -T '{rev}:{node} {desc}\n'
   3:be0ef73c17ade3fc89dc41701eb9fc3a91b58282 D
   2:26805aba1e600a82e93661149f2313866a221a7b C
   1:112478962961147124edd43549aedd1a335e44bf B
   0:426bada5c67598ca65036d57d9e4b64b0c1ce7a0 A
 
   $ hg serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 Request for namespaces works
 
   $ sendhttpv2peer << EOF
   > command listkeys
   >     namespace namespaces
   > EOF
   creating http peer for wire protocol version 2
   sending listkeys command
   s>     POST /api/exp-http-v2-0001/ro/listkeys HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 50\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     *\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1InamespaceJnamespacesDnameHlistkeys
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'',
     b'phases': b''
   }
 
 Request for phases works
 
   $ sendhttpv2peer << EOF
   > command listkeys
   >     namespace phases
   > EOF
   creating http peer for wire protocol version 2
   sending listkeys command
   s>     POST /api/exp-http-v2-0001/ro/listkeys HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 46\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     &\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1InamespaceFphasesDnameHlistkeys
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
   }
 
 Request for bookmarks works
 
   $ sendhttpv2peer << EOF
   > command listkeys
   >     namespace bookmarks
   > EOF
   creating http peer for wire protocol version 2
   sending listkeys command
   s>     POST /api/exp-http-v2-0001/ro/listkeys HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 49\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     )\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1InamespaceIbookmarksDnameHlistkeys
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
   }
 
   $ cat error.log

tests/test-wireproto-command-lookup.t

@@ -1,55 +1,64 b''
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ hg init server
   $ enablehttpv2 server
   $ cd server
   $ cat >> .hg/hgrc << EOF
   > [web]
   > push_ssl = false
   > allow-push = *
   > EOF
   $ hg debugdrawdag << EOF
   > C D
   > |/
   > B
   > |
   > A
   > EOF
 
   $ hg serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 lookup for known node works
 
   $ sendhttpv2peer << EOF
   > command lookup
   >     key 426bada5c67598ca65036d57d9e4b64b0c1ce7a0
   > EOF
   creating http peer for wire protocol version 2
   sending lookup command
   s>     *\r\n (glob)
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 73\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     A\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1CkeyX(426bada5c67598ca65036d57d9e4b64b0c1ce7a0DnameFlookup
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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

@@ -1,91 +1,109 b''
   $ . $TESTDIR/wireprotohelpers.sh
 
   $ hg init server
   $ enablehttpv2 server
   $ cd server
   $ cat >> .hg/hgrc << EOF
   > [web]
   > push_ssl = false
   > allow-push = *
   > EOF
   $ hg debugdrawdag << EOF
   > C D
   > |/
   > B
   > |
   > A
   > EOF
 
   $ hg serve -p $HGPORT -d --pid-file hg.pid -E error.log
   $ cat hg.pid > $DAEMON_PIDS
 
 pushkey for a bookmark works
 
   $ sendhttpv2peer << EOF
   > command pushkey
   >     namespace bookmarks
   >     key @
   >     old
   >     new 426bada5c67598ca65036d57d9e4b64b0c1ce7a0
   > EOF
   creating http peer for wire protocol version 2
   sending pushkey command
   s>     *\r\n (glob)
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 105\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     a\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa4CkeyA@InamespaceIbookmarksCnewX(426bada5c67598ca65036d57d9e4b64b0c1ce7a0Cold@DnameGpushkey
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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
   > command listkeys
   >     namespace bookmarks
   > EOF
   creating http peer for wire protocol version 2
   sending listkeys command
   s>     POST /api/exp-http-v2-0001/ro/listkeys HTTP/1.1\r\n
   s>     Accept-Encoding: identity\r\n
   s>     accept: application/mercurial-exp-framing-0005\r\n
   s>     content-type: application/mercurial-exp-framing-0005\r\n
   s>     content-length: 49\r\n
   s>     host: $LOCALIP:$HGPORT\r\n (glob)
   s>     user-agent: Mercurial debugwireproto\r\n
   s>     \r\n
   s>     )\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1InamespaceIbookmarksDnameHlistkeys
   s> makefile('rb', None)
   s>     HTTP/1.1 200 OK\r\n
   s>     Server: testing stub value\r\n
   s>     Date: $HTTP_DATE$\r\n
   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'
   }
 
   $ cat error.log

tests/wireprotohelpers.sh

@@ -1,58 +1,58 b''
 HTTPV2=exp-http-v2-0001
 MEDIATYPE=application/mercurial-exp-framing-0005
 
 sendhttpraw() {
   hg --verbose debugwireproto --peer raw http://$LOCALIP:$HGPORT/
 }
 
 sendhttpv2peer() {
   hg --verbose debugwireproto --nologhandshake --peer http2 http://$LOCALIP:$HGPORT/
 }
 
 sendhttpv2peerhandshake() {
   hg --verbose debugwireproto --peer http2 http://$LOCALIP:$HGPORT/
 }
 
 cat > dummycommands.py << EOF
 from mercurial import (
     wireprototypes,
     wireprotov1server,
     wireprotov2server,
 )
 
 @wireprotov1server.wireprotocommand(b'customreadonly', permission=b'pull')
 def customreadonlyv1(repo, proto):
     return wireprototypes.bytesresponse(b'customreadonly bytes response')
 
 @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):
     return wireprototypes.bytesresponse(b'customreadwrite bytes response')
 
 @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
 [extensions]
 drawdag = $TESTDIR/drawdag.py
 EOF
 
 enabledummycommands() {
   cat >> $HGRCPATH << EOF
 [extensions]
 dummycommands = $TESTTMP/dummycommands.py
 EOF
 }
 
 enablehttpv2() {
   cat >> $1/.hg/hgrc << EOF
 [experimental]
 web.apiserver = true
 web.api.http-v2 = true
 EOF
 }