comparison mercurial/wireprotov2peer.py @ 40724:15a643304728 stable

wireprotov2peer: wait for initial object before resolving future

As part of rolling out wireprotov2 with redirect support, I
encountered an edge case with regards to future resolution.

Essentially, the initial response frame from the server did not
fully decode the initial CBOR object. The frame wasn't marked as
EOS. In the previous code, we resolved the future for the request
to response.objects(), which mapped to the commandresponse instance
which would eventually produce a redirect. Upon receiving
subsequent data, the initial CBOR object containing the redirect
would be decoded and we'd process the redirect. However, the
future would already have been resolved with the initial
commandresponse.objects() and the client iterating over the
objects wouldn't receive any objects from the redirect because
the redirect was populating a different commandresponse instance!

This commit changes the logic so we don't resolve futures until
the initial CBOR response object is fully decoded or until EOS
occurs. In cases where there is an empty or partial frame
associated with a redirect, the future will now resolve with the
commandresponse containing the proper series of decoded objects.

40723:94b0d0f996e1

    def _processresponsedata(self, frame, meta, response):
        # This can raise. The caller can handle it.
        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']:
            response._oninputcomplete()
            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
        # received before resolving the future. Otherwise we resolve the
        # future immediately.

40724:15a643304728

    def _processresponsedata(self, frame, meta, response):
        # This can raise. The caller can handle it.
        response._onresponsedata(meta['data'])

        # We need to be careful about resolving futures prematurely. If a
        # response is a redirect response, resolving the future before the
        # redirect is processed would result in the consumer seeing an
        # empty stream of objects, since they'd be consuming our
        # response.objects() instead of the redirect's response.objects().
        #
        # Our strategy is to not resolve/finish the request until either
        # EOS occurs or until the initial response object is fully received.

        # Always react to eos.
        if meta['eos']:
            response._oninputcomplete()
            del self._requests[frame.requestid]

        # Not EOS but we haven't decoded the initial response object yet.
        # Return and wait for more data.
        elif not response._seeninitial:
            return

        # The specification says no objects should follow the initial/redirect
        # object. So it should be safe to handle the redirect object if one is
        # decoded, without having to wait for EOS.
        if response._redirect:
            self._followredirect(frame.requestid, response._redirect)
            return

        # If the command has a decoder, we wait until all input has been
        # received before resolving the future. Otherwise we resolve the
        # future immediately.