This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4778

wireprotov2: client support for following content redirects
ClosedPublic

Authored by indygreg on Sep 26 2018, 9:09 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG7e807b8a9e56: wireprotov2: client support for following content redirects
Summary

And with the server actually sending content redirects, it is finally
time to implement client support for following them!

When a redirect response is seen, we wait until all data for that
request has been received (it should be nearly immediate since no
data is expected to follow the redirect message). Then we use
a URL opener to make a request. We stuff that response into the
client handler and construct a new response object to track it.

When readdata() is called for servicing requests, we attempt to
read data from the first redirected response. During data reading,
data is processed similarly to as if it came from a frame payload.

The existing test for the functionality demonstrates the client
transparently following the redirect and obtaining the command
response data from an alternate URL!

There is still plenty of work to do here, including shoring up
testing. I'm not convinced things will work in the presence of
multiple redirect responses. And we don't yet implement support
for integrity verification or configuring server certificates
to validate the connection. But it's a start. And it should enable
us to start experimenting with "real" caches.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Sep 26 2018, 9:09 PM
Closed by commit rHG7e807b8a9e56: wireprotov2: client support for following content redirects (authored by indygreg). · Explain WhyOct 3 2018, 11:30 AM
This revision was automatically updated to reflect the committed changes.
mharbison72 added a subscriber: mharbison72.Oct 5 2018, 12:12 AM

Windows doesn't like this test, and gets a connection refused. Under the glob is the machine name instead of $LOCALIP that is used above it. Obviously this isn't the right fix, but this makes the test run on Windows to illustrate it's not something else:

diff --git a/mercurial/wireprotov2peer.py b/mercurial/wireprotov2peer.py
--- a/mercurial/wireprotov2peer.py
+++ b/mercurial/wireprotov2peer.py
@@ -388,6 +388,9 @@ class clienthandler(object):

     def _followredirect(self, requestid, redirect):
         """Called to initiate redirect following for a request."""
+        url = util.url(redirect.url)
+        url.host = '127.0.0.1'
+        redirect.url = str(url)
         self._ui.note(_('(following redirect to %s)\n') % redirect.url)

         # TODO handle framed responses.

This seems like the problem I mentioned in D3433 on May 2 that you thought boiled down to a gethostname(). Not sure what to do about it.

Revision Contents
Changeset List

Diff 11449

mercurial/httppeer.py

def _abort(self, exception): def _abort(self, exception):
raise exception raise exception
def sendv2request(ui, opener, requestbuilder, apiurl, permission, requests, def sendv2request(ui, opener, requestbuilder, apiurl, permission, requests,
redirect): redirect):
reactor = wireprotoframing.clientreactor(hasmultiplesend=False, reactor = wireprotoframing.clientreactor(hasmultiplesend=False,
buffersends=True) buffersends=True)
handler = wireprotov2peer.clienthandler(ui, reactor) handler = wireprotov2peer.clienthandler(ui, reactor,
opener=opener,
requestbuilder=requestbuilder)
url = '%s/%s' % (apiurl, permission) url = '%s/%s' % (apiurl, permission)
if len(requests) > 1: if len(requests) > 1:
url += '/multirequest' url += '/multirequest'
else: else:
url += '/%s' % requests[0][0] url += '/%s' % requests[0][0]

mercurial/wireprotov2peer.py

# wireprotov2peer.py - client side code for wire protocol version 2 # wireprotov2peer.py - client side code for wire protocol version 2
# #
# Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com> # Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import threading import threading
from .i18n import _ from .i18n import _
from . import ( from . import (
encoding, encoding,
error, error,
pycompat,
sslutil, sslutil,
url as urlmod,
util,
wireprotoframing, wireprotoframing,
wireprototypes,
) )
from .utils import ( from .utils import (
cborutil, cborutil,
) )
def formatrichmessage(atoms): def formatrichmessage(atoms):
"""Format an encoded message from the framing protocol.""" """Format an encoded message from the framing protocol."""
"""Represents the response to a command request. """Represents the response to a command request.
Instances track the state of the command and hold its results. Instances track the state of the command and hold its results.
An external entity is required to update the state of the object when An external entity is required to update the state of the object when
events occur. events occur.
""" """
def __init__(self, requestid, command): def __init__(self, requestid, command, fromredirect=False):
self.requestid = requestid self.requestid = requestid
self.command = command self.command = command
self.fromredirect = fromredirect
# Whether all remote input related to this command has been # Whether all remote input related to this command has been
# received. # received.
self._inputcomplete = False self._inputcomplete = False
# We have a lock that is acquired when important object state is # We have a lock that is acquired when important object state is
# mutated. This is to prevent race conditions between 1 thread # mutated. This is to prevent race conditions between 1 thread
# sending us new data and another consuming it. # sending us new data and another consuming it.
self._lock = threading.RLock() self._lock = threading.RLock()
# An event is set when state of the object changes. This event # An event is set when state of the object changes. This event
# is waited on by the generator emitting objects. # is waited on by the generator emitting objects.
self._serviceable = threading.Event() self._serviceable = threading.Event()
self._pendingevents = [] self._pendingevents = []
self._decoder = cborutil.bufferingdecoder() self._decoder = cborutil.bufferingdecoder()
self._seeninitial = False self._seeninitial = False
self._redirect = None
def _oninputcomplete(self): def _oninputcomplete(self):
with self._lock: with self._lock:
self._inputcomplete = True self._inputcomplete = True
self._serviceable.set() self._serviceable.set()
def _onresponsedata(self, data): def _onresponsedata(self, data):
available, readcount, wanted = self._decoder.decode(data) available, readcount, wanted = self._decoder.decode(data)
if not available: if not available:
return return
with self._lock: with self._lock:
for o in self._decoder.getavailable(): for o in self._decoder.getavailable():
if not self._seeninitial: if not self._seeninitial and not self.fromredirect:
self._handleinitial(o) self._handleinitial(o)
continue continue
# We should never see an object after a content redirect,
# as the spec says the main status object containing the
# content redirect is the only object in the stream. Fail
# if we see a misbehaving server.
if self._redirect:
raise error.Abort(_('received unexpected response data '
'after content redirect; the remote is '
'buggy'))
self._pendingevents.append(o) self._pendingevents.append(o)
self._serviceable.set() self._serviceable.set()
def _handleinitial(self, o): def _handleinitial(self, o):
self._seeninitial = True self._seeninitial = True
if o[b'status'] == b'ok': if o[b'status'] == b'ok':
return return
elif o[b'status'] == b'redirect': elif o[b'status'] == b'redirect':
raise error.Abort(_('redirect responses not yet supported')) l = o[b'location']
self._redirect = wireprototypes.alternatelocationresponse(
url=l[b'url'],
mediatype=l[b'mediatype'],
size=l.get(b'size'),
fullhashes=l.get(b'fullhashes'),
fullhashseed=l.get(b'fullhashseed'),
serverdercerts=l.get(b'serverdercerts'),
servercadercerts=l.get(b'servercadercerts'))
return
atoms = [{'msg': o[b'error'][b'message']}] atoms = [{'msg': o[b'error'][b'message']}]
if b'args' in o[b'error']: if b'args' in o[b'error']:
atoms[0]['args'] = o[b'error'][b'args'] atoms[0]['args'] = o[b'error'][b'args']
raise error.RepoError(formatrichmessage(atoms)) raise error.RepoError(formatrichmessage(atoms))
def objects(self): def objects(self):
The ``clientreactor`` is used to hold low-level state about the frame-based 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 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 for higher-level operations, such as reading frames from a socket, exposing
and managing a higher-level primitive for representing command responses, and managing a higher-level primitive for representing command responses,
etc. This class is what peers should probably use to bridge wire activity etc. This class is what peers should probably use to bridge wire activity
with the higher-level peer API. with the higher-level peer API.
""" """
def __init__(self, ui, clientreactor): def __init__(self, ui, clientreactor, opener=None,
requestbuilder=util.urlreq.request):
self._ui = ui self._ui = ui
self._reactor = clientreactor self._reactor = clientreactor
self._requests = {} self._requests = {}
self._futures = {} self._futures = {}
self._responses = {} self._responses = {}
self._redirects = []
self._frameseof = False self._frameseof = False
self._opener = opener or urlmod.opener(ui)
self._requestbuilder = requestbuilder
def callcommand(self, command, args, f, redirect=None): def callcommand(self, command, args, f, redirect=None):
"""Register a request to call a command. """Register a request to call a command.
Returns an iterable of frames that should be sent over the wire. Returns an iterable of frames that should be sent over the wire.
""" """
request, action, meta = self._reactor.callcommand(command, args, request, action, meta = self._reactor.callcommand(command, args,
redirect=redirect) redirect=redirect)
frame = wireprotoframing.readframe(framefh) frame = wireprotoframing.readframe(framefh)
if frame is None: if frame is None:
# TODO tell reactor? # TODO tell reactor?
self._frameseof = True self._frameseof = True
else: else:
self._ui.note(_('received %r\n') % frame) self._ui.note(_('received %r\n') % frame)
self._processframe(frame) self._processframe(frame)
if self._frameseof: # Also try to read the first redirect.
if self._redirects:
if not self._processredirect(*self._redirects[0]):
self._redirects.pop(0)
if self._frameseof and not self._redirects:
return None return None
return True return True
def _processframe(self, frame): def _processframe(self, frame):
"""Process a single read frame.""" """Process a single read frame."""
action, meta = self._reactor.onframerecv(frame) action, meta = self._reactor.onframerecv(frame)
else: else:
raise error.ProgrammingError( raise error.ProgrammingError(
'unhandled action from clientreactor: %s' % action) 'unhandled action from clientreactor: %s' % action)
def _processresponsedata(self, frame, meta, response): def _processresponsedata(self, frame, meta, response):
# This can raise. The caller can handle it. # This can raise. The caller can handle it.
response._onresponsedata(meta['data']) response._onresponsedata(meta['data'])
# If we got a content redirect response, we want to fetch it and
# expose the data as if we received it inline. But we also want to
# keep our internal request accounting in order. Our strategy is to
# basically put meaningful response handling on pause until EOS occurs
# and the stream accounting is in a good state. At that point, we follow
# the redirect and replace the response object with its data.
redirect = response._redirect
handlefuture = False if redirect else True
if meta['eos']: if meta['eos']:
response._oninputcomplete() response._oninputcomplete()
del self._requests[frame.requestid] del self._requests[frame.requestid]
if redirect:
self._followredirect(frame.requestid, redirect)
return
if not handlefuture:
return
# If the command has a decoder, we wait until all input has been # If the command has a decoder, we wait until all input has been
# received before resolving the future. Otherwise we resolve the # received before resolving the future. Otherwise we resolve the
# future immediately. # future immediately.
if frame.requestid not in self._futures: if frame.requestid not in self._futures:
return return
if response.command not in COMMAND_DECODERS: if response.command not in COMMAND_DECODERS:
self._futures[frame.requestid].set_result(response.objects()) self._futures[frame.requestid].set_result(response.objects())
del self._futures[frame.requestid] del self._futures[frame.requestid]
elif response._inputcomplete: elif response._inputcomplete:
decoded = COMMAND_DECODERS[response.command](response.objects()) decoded = COMMAND_DECODERS[response.command](response.objects())
self._futures[frame.requestid].set_result(decoded) self._futures[frame.requestid].set_result(decoded)
del self._futures[frame.requestid] del self._futures[frame.requestid]
def _followredirect(self, requestid, redirect):
"""Called to initiate redirect following for a request."""
self._ui.note(_('(following redirect to %s)\n') % redirect.url)
# TODO handle framed responses.
if redirect.mediatype != b'application/mercurial-cbor':
raise error.Abort(_('cannot handle redirects for the %s media type')
% redirect.mediatype)
if redirect.fullhashes:
self._ui.warn(_('(support for validating hashes on content '
'redirects not supported)\n'))
if redirect.serverdercerts or redirect.servercadercerts:
self._ui.warn(_('(support for pinning server certificates on '
'content redirects not supported)\n'))
headers = {
r'Accept': redirect.mediatype,
}
req = self._requestbuilder(pycompat.strurl(redirect.url), None, headers)
try:
res = self._opener.open(req)
except util.urlerr.httperror as e:
if e.code == 401:
raise error.Abort(_('authorization failed'))
raise
except util.httplib.HTTPException as e:
self._ui.debug('http error requesting %s\n' % req.get_full_url())
self._ui.traceback()
raise IOError(None, e)
urlmod.wrapresponse(res)
# The existing response object is associated with frame data. Rather
# than try to normalize its state, just create a new object.
oldresponse = self._responses[requestid]
self._responses[requestid] = commandresponse(requestid,
oldresponse.command,
fromredirect=True)
self._redirects.append((requestid, res))
def _processredirect(self, rid, res):
"""Called to continue processing a response from a redirect."""
response = self._responses[rid]
try:
data = res.read(32768)
response._onresponsedata(data)
# We're at end of stream.
if not data:
response._oninputcomplete()
if rid not in self._futures:
return
if response.command not in COMMAND_DECODERS:
self._futures[rid].set_result(response.objects())
del self._futures[rid]
elif response._inputcomplete:
decoded = COMMAND_DECODERS[response.command](response.objects())
self._futures[rid].set_result(decoded)
del self._futures[rid]
return bool(data)
except BaseException as e:
self._futures[rid].set_exception(e)
del self._futures[rid]
response._oninputcomplete()
return False
def decodebranchmap(objs): def decodebranchmap(objs):
# Response should be a single CBOR map of branch name to array of nodes. # Response should be a single CBOR map of branch name to array of nodes.
bm = next(objs) bm = next(objs)
return {encoding.tolocal(k): v for k, v in bm.items()} return {encoding.tolocal(k): v for k, v in bm.items()}
def decodeheads(objs): def decodeheads(objs):
# Array of node bytestrings. # Array of node bytestrings.

tests/test-wireproto-content-redirects.t

s> \r\n s> \r\n
s> 8\r\n s> 8\r\n
received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=continuation) received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=continuation)
s> \x00\x00\x00\x01\x00\x02\x002 s> \x00\x00\x00\x01\x00\x02\x002
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos) received frame(size=0; request=1; stream=2; streamflags=; type=command-response; flags=eos)
abort: redirect responses not yet supported (following redirect to http://*:$HGPORT/api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0) (glob)
[255] s> GET /api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0 HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-cbor\r\n
s> host: *:$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: 91\r\n
s> \r\n
s> \xa1Jtotalitems\x01\xa2DnodeT\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&AGparents\x82T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00
response: gen[
{
b'totalitems': 1
},
{
b'node': b'\x99/Gy\x02\x9a=\xf8\xd0fm\x00\xbb\x92OicN&A',
b'parents': [
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
]
}
]
$ cat error.log $ cat error.log
$ killdaemons.py $ killdaemons.py
$ cat .hg/blackbox.log $ cat .hg/blackbox.log
*> cacher constructed for manifestdata (glob) *> cacher constructed for manifestdata (glob)
*> cache miss for c045a581599d58608efd3d93d8129841f2af04a0 (glob) *> cache miss for c045a581599d58608efd3d93d8129841f2af04a0 (glob)
*> storing cache entry for c045a581599d58608efd3d93d8129841f2af04a0 (glob) *> storing cache entry for c045a581599d58608efd3d93d8129841f2af04a0 (glob)
*> cacher constructed for manifestdata (glob) *> cacher constructed for manifestdata (glob)
*> cache hit for c045a581599d58608efd3d93d8129841f2af04a0 (glob) *> cache hit for c045a581599d58608efd3d93d8129841f2af04a0 (glob)
*> sending content redirect for c045a581599d58608efd3d93d8129841f2af04a0 to http://*:$HGPORT/api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0 (glob) *> sending content redirect for c045a581599d58608efd3d93d8129841f2af04a0 to http://*:$HGPORT/api/simplecache/c045a581599d58608efd3d93d8129841f2af04a0 (glob)