Client-side protocol V2 support for fetching

Make all transports request protocol V2 when fetching. Depending on
the transport, set the GIT_PROTOCOL environment variable (file and
ssh), pass the Git-Protocol header (http), or set the hidden
"\0version=2\0" (git anon). We'll fall back to V0 if the server
doesn't reply with a version 2 answer.

A user can control which protocol the client requests via the git
config protocol.version; if not set, JGit requests protocol V2 for
fetching. Pushing always uses protocol V0 still.

In the API, there is only a new Transport.openFetch() version that
takes a collection of RefSpecs plus additional patterns to construct
the Ref prefixes for the "ls-refs" command in protocol V2. If none
are given, the server will still advertise all refs, even in protocol
V2.

BasePackConnection.readAdvertisedRefs() handles falling back to
protocol V0. It newly returns true if V0 was used and the advertised
refs were read, and false if V2 is used and an explicit "ls-refs" is
needed. (This can't be done transparently inside readAdvertisedRefs()
because a "stateless RPC" transport like TransportHttp may need to
open a new connection for writing.)

BasePackFetchConnection implements the changes needed for the protocol
V2 "fetch" command (stateless protocol, simplified ACK handling,
delimiters, section headers).

In TransportHttp, change readSmartHeaders() to also recognize the
"version 2" packet line as a valid smart server indication.

Adapt tests, and run all the HTTP tests not only with both HTTP
connection factories (JDK and Apache HttpClient) but also with both
protocol V0 and V2. The SSH tests are much slower and much more
focused on the SSH protocol and SSH key handling. Factor out two
very simple cloning and pulling tests and make those run with
protocol V2.

Bug: 553083
Change-Id: I357c7f5daa7efb2872f1c64ee6f6d54229031ae1
Signed-off-by: Thomas Wolf <thomas.wolf@paranor.ch>

1 files changed, 270 insertions, 26 deletions

diff --git a/org.eclipse.jgit/src/org/eclipse/jgit/transport/BasePackFetchConnection.java b/org.eclipse.jgit/src/org/eclipse/jgit/transport/BasePackFetchConnection.java
index a2fb51f46d..fa0c0c6670 100644
--- a/org.eclipse.jgit/src/org/eclipse/jgit/transport/BasePackFetchConnection.java
+++ b/org.eclipse.jgit/src/org/eclipse/jgit/transport/BasePackFetchConnection.java
@@ -1,7 +1,7 @@
 /*
- * Copyright (C) 2008-2010, Google Inc.
+ * Copyright (C) 2008, 2010 Google Inc.
  * Copyright (C) 2008, Robin Rosenberg <robin.rosenberg@dewire.com>
- * Copyright (C) 2008, Shawn O. Pearce <spearce@spearce.org> and others
+ * Copyright (C) 2008, 2020 Shawn O. Pearce <spearce@spearce.org> and others
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Distribution License v. 1.0 which is available at
@@ -16,18 +16,21 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.text.MessageFormat;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.Set;
 
 import org.eclipse.jgit.errors.PackProtocolException;
+import org.eclipse.jgit.errors.RemoteRepositoryException;
 import org.eclipse.jgit.errors.TransportException;
 import org.eclipse.jgit.internal.JGitText;
 import org.eclipse.jgit.internal.storage.file.PackLock;
 import org.eclipse.jgit.lib.AnyObjectId;
 import org.eclipse.jgit.lib.Config;
-import org.eclipse.jgit.lib.Constants;
 import org.eclipse.jgit.lib.MutableObjectId;
 import org.eclipse.jgit.lib.NullProgressMonitor;
 import org.eclipse.jgit.lib.ObjectId;
@@ -44,6 +47,7 @@ import org.eclipse.jgit.revwalk.filter.CommitTimeRevFilter;
 import org.eclipse.jgit.revwalk.filter.RevFilter;
 import org.eclipse.jgit.transport.GitProtocolConstants.MultiAck;
 import org.eclipse.jgit.transport.PacketLineIn.AckNackResult;
+import org.eclipse.jgit.util.StringUtils;
 import org.eclipse.jgit.util.TemporaryBuffer;
 
 /**
@@ -207,7 +211,10 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 
 	private int maxHaves;
 
-	/** RPC state, if {@link BasePackConnection#statelessRPC} is true. */
+	/**
+	 * RPC state, if {@link BasePackConnection#statelessRPC} is true or protocol
+	 * V2 is used.
+	 */
 	private TemporaryBuffer.Heap state;
 
 	private PacketLineOut pckState;
@@ -321,6 +328,13 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 		return Collections.<PackLock> emptyList();
 	}
 
+	private void clearState() {
+		walk.dispose();
+		reachableCommits = null;
+		state = null;
+		pckState = null;
+	}
+
 	/**
 	 * Execute common ancestor negotiation and fetch the objects.
 	 *
@@ -349,18 +363,34 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 			markRefsAdvertised();
 			markReachable(have, maxTimeWanted(want));
 
+			if (TransferConfig.ProtocolVersion.V2
+					.equals(getProtocolVersion())) {
+				// Protocol V2 always is a "stateless" protocol, even over a
+				// bidirectional pipe: the server serves one "fetch" request and
+				// then forgets anything it has learned, so the next fetch
+				// request has to re-send all wants and previously determined
+				// common objects as "have"s again.
+				state = new TemporaryBuffer.Heap(Integer.MAX_VALUE);
+				pckState = new PacketLineOut(state);
+				try {
+					doFetchV2(monitor, want, outputStream);
+				} finally {
+					clearState();
+				}
+				return;
+			}
+			// Protocol V0/1
 			if (statelessRPC) {
 				state = new TemporaryBuffer.Heap(Integer.MAX_VALUE);
 				pckState = new PacketLineOut(state);
 			}
-
-			if (sendWants(want)) {
+			PacketLineOut output = statelessRPC ? pckState : pckOut;
+			if (sendWants(want, output)) {
+				output.end();
+				outNeedsEnd = false;
 				negotiate(monitor);
 
-				walk.dispose();
-				reachableCommits = null;
-				state = null;
-				pckState = null;
+				clearState();
 
 				receivePack(monitor, outputStream);
 			}
@@ -373,6 +403,180 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 		}
 	}
 
+	private void doFetchV2(ProgressMonitor monitor, Collection<Ref> want,
+			OutputStream outputStream) throws IOException, CancelledException {
+		sideband = true;
+		negotiateBegin();
+
+		pckState.writeString("command=" + GitProtocolConstants.COMMAND_FETCH); //$NON-NLS-1$
+		// Capabilities are sent as command arguments in protocol V2
+		String agent = UserAgent.get();
+		if (agent != null && isCapableOf(GitProtocolConstants.OPTION_AGENT)) {
+			pckState.writeString(
+					GitProtocolConstants.OPTION_AGENT + '=' + agent);
+		}
+		Set<String> capabilities = new HashSet<>();
+		String advertised = getCapability(GitProtocolConstants.COMMAND_FETCH);
+		if (!StringUtils.isEmptyOrNull(advertised)) {
+			capabilities.addAll(Arrays.asList(advertised.split("\\s+"))); //$NON-NLS-1$
+		}
+		// Arguments
+		pckState.writeDelim();
+		for (String capability : getCapabilitiesV2(capabilities)) {
+			pckState.writeString(capability);
+		}
+		if (!sendWants(want, pckState)) {
+			// We already have everything we wanted.
+			return;
+		}
+		// If we send something, we always close it properly ourselves.
+		outNeedsEnd = false;
+
+		FetchStateV2 fetchState = new FetchStateV2();
+		boolean sentDone = false;
+		for (;;) {
+			// The "state" buffer contains the full fetch request with all
+			// common objects found so far.
+			state.writeTo(out, monitor);
+			sentDone = sendNextHaveBatch(fetchState, pckOut, monitor);
+			if (sentDone) {
+				break;
+			}
+			if (readAcknowledgments(fetchState, pckIn, monitor)) {
+				// We got a "ready": next should be a patch file.
+				break;
+			}
+			// Note: C git reads and requires here (and after a packfile) a
+			// "0002" packet in stateless RPC transports (https). This "response
+			// end" packet is even mentioned in the protocol V2 technical
+			// documentation. However, it is not actually part of the public
+			// protocol; it occurs only in an internal protocol wrapper in the C
+			// git implementation.
+		}
+		clearState();
+		String line = pckIn.readString();
+		// If we sent a done, we may have an error reply here.
+		if (sentDone && line.startsWith("ERR ")) { //$NON-NLS-1$
+			throw new RemoteRepositoryException(uri, line.substring(4));
+		}
+		// "shallow-info", "wanted-refs", and "packfile-uris" would have to be
+		// handled here in that order.
+		if (!GitProtocolConstants.SECTION_PACKFILE.equals(line)) {
+			throw new PackProtocolException(
+					MessageFormat.format(JGitText.get().expectedGot,
+							GitProtocolConstants.SECTION_PACKFILE, line));
+		}
+		receivePack(monitor, outputStream);
+	}
+
+	/**
+	 * Sends the next batch of "have"s and terminates the {@code output}.
+	 *
+	 * @param fetchState
+	 *            is updated with information about the number of items written,
+	 *            and whether to expect a packfile next
+	 * @param output
+	 *            to write to
+	 * @param monitor
+	 *            for progress reporting and cancellation
+	 * @return {@code true} if a "done" was written and we should thus expect a
+	 *         packfile next
+	 * @throws IOException
+	 *             on errors
+	 * @throws CancelledException
+	 *             on cancellation
+	 */
+	private boolean sendNextHaveBatch(FetchStateV2 fetchState,
+			PacketLineOut output, ProgressMonitor monitor)
+			throws IOException, CancelledException {
+		long n = 0;
+		while (n < fetchState.havesToSend) {
+			final RevCommit c = walk.next();
+			if (c == null) {
+				break;
+			}
+			output.writeString("have " + c.getId().name() + '\n'); //$NON-NLS-1$
+			n++;
+			if (n % 10 == 0 && monitor.isCancelled()) {
+				throw new CancelledException();
+			}
+		}
+		fetchState.havesTotal += n;
+		if (n == 0
+				|| fetchState.havesWithoutAck > MAX_HAVES
+				|| fetchState.havesTotal > maxHaves) {
+			output.writeString("done\n"); //$NON-NLS-1$
+			output.end();
+			return true;
+		}
+		fetchState.havesWithoutAck += n;
+		output.end();
+		fetchState.incHavesToSend(statelessRPC);
+		return false;
+	}
+
+	/**
+	 * Reads and processes acknowledgments, adding ACKed objects as "have"s to
+	 * the global state {@link TemporaryBuffer}.
+	 *
+	 * @param fetchState
+	 *            to update
+	 * @param input
+	 *            to read from
+	 * @param monitor
+	 *            for progress reporting and cancellation
+	 * @return {@code true} if a "ready" was received and a packfile is expected
+	 *         next
+	 * @throws IOException
+	 *             on errors
+	 * @throws CancelledException
+	 *             on cancellation
+	 */
+	private boolean readAcknowledgments(FetchStateV2 fetchState,
+			PacketLineIn input, ProgressMonitor monitor)
+			throws IOException, CancelledException {
+		String line = input.readString();
+		if (!GitProtocolConstants.SECTION_ACKNOWLEDGMENTS.equals(line)) {
+			throw new PackProtocolException(MessageFormat.format(
+					JGitText.get().expectedGot,
+					GitProtocolConstants.SECTION_ACKNOWLEDGMENTS, line));
+		}
+		MutableObjectId returnedId = new MutableObjectId();
+		line = input.readString();
+		boolean gotReady = false;
+		long n = 0;
+		while (!PacketLineIn.isEnd(line) && !PacketLineIn.isDelimiter(line)) {
+			AckNackResult ack = PacketLineIn.parseACKv2(line, returnedId);
+			// If we got a "ready", we just skip the remaining lines after
+			// having checked them for being valid. (Normally, the "ready"
+			// should be the last line anyway.)
+			if (!gotReady) {
+				if (ack == AckNackResult.ACK_COMMON) {
+					// markCommon appends the object to the "state"
+					markCommon(walk.parseAny(returnedId), ack, true);
+					fetchState.havesWithoutAck = 0;
+				} else if (ack == AckNackResult.ACK_READY) {
+					gotReady = true;
+				}
+			}
+			n++;
+			if (n % 10 == 0 && monitor.isCancelled()) {
+				throw new CancelledException();
+			}
+			line = input.readString();
+		}
+		if (gotReady) {
+			if (!PacketLineIn.isDelimiter(line)) {
+				throw new PackProtocolException(MessageFormat
+						.format(JGitText.get().expectedGot, "0001", line)); //$NON-NLS-1$
+			}
+		} else if (!PacketLineIn.isEnd(line)) {
+			throw new PackProtocolException(MessageFormat
+					.format(JGitText.get().expectedGot, "0000", line)); //$NON-NLS-1$
+		}
+		return gotReady;
+	}
+
 	/** {@inheritDoc} */
 	@Override
 	public void close() {
@@ -456,8 +660,8 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 		}
 	}
 
-	private boolean sendWants(Collection<Ref> want) throws IOException {
-		final PacketLineOut p = statelessRPC ? pckState : pckOut;
+	private boolean sendWants(Collection<Ref> want, PacketLineOut p)
+			throws IOException {
 		boolean first = true;
 		for (Ref r : want) {
 			ObjectId objectId = r.getObjectId();
@@ -479,10 +683,11 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 			final StringBuilder line = new StringBuilder(46);
 			line.append("want "); //$NON-NLS-1$
 			line.append(objectId.name());
-			if (first) {
+			if (first && TransferConfig.ProtocolVersion.V0
+					.equals(getProtocolVersion())) {
 				line.append(enableCapabilities());
-				first = false;
 			}
+			first = false;
 			line.append('\n');
 			p.writeString(line.toString());
 		}
@@ -492,11 +697,34 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 		if (!filterSpec.isNoOp()) {
 			p.writeString(filterSpec.filterLine());
 		}
-		p.end();
-		outNeedsEnd = false;
 		return true;
 	}
 
+	private Set<String> getCapabilitiesV2(Set<String> advertisedCapabilities)
+			throws TransportException {
+		Set<String> capabilities = new LinkedHashSet<>();
+		// Protocol V2 is implicitly capable of all these.
+		if (noProgress) {
+			capabilities.add(OPTION_NO_PROGRESS);
+		}
+		if (includeTags) {
+			capabilities.add(OPTION_INCLUDE_TAG);
+		}
+		if (allowOfsDelta) {
+			capabilities.add(OPTION_OFS_DELTA);
+		}
+		if (thinPack) {
+			capabilities.add(OPTION_THIN_PACK);
+		}
+		if (!filterSpec.isNoOp()
+				&& !advertisedCapabilities.contains(OPTION_FILTER)) {
+			throw new PackProtocolException(uri,
+					JGitText.get().filterRequiresCapability);
+		}
+		// The FilterSpec will be added later in sendWants().
+		return capabilities;
+	}
+
 	private String enableCapabilities() throws TransportException {
 		final StringBuilder line = new StringBuilder();
 		if (noProgress)
@@ -622,7 +850,7 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 					// we need to continue to talk about other parts of
 					// our local history.
 					//
-					markCommon(walk.parseAny(ackId), anr);
+					markCommon(walk.parseAny(ackId), anr, statelessRPC);
 					receivedAck = true;
 					receivedContinue = true;
 					havesSinceLastContinue = 0;
@@ -757,16 +985,10 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 		}
 	}
 
-	private void markCommon(RevObject obj, AckNackResult anr)
+	private void markCommon(RevObject obj, AckNackResult anr, boolean useState)
 			throws IOException {
-		if (statelessRPC && anr == AckNackResult.ACK_COMMON && !obj.has(STATE)) {
-			StringBuilder s;
-
-			s = new StringBuilder(6 + Constants.OBJECT_ID_STRING_LENGTH);
-			s.append("have "); //$NON-NLS-1$
-			s.append(obj.name());
-			s.append('\n');
-			pckState.writeString(s.toString());
+		if (useState && anr == AckNackResult.ACK_COMMON && !obj.has(STATE)) {
+			pckState.writeString("have " + obj.name() + '\n'); //$NON-NLS-1$
 			obj.add(STATE);
 		}
 		obj.add(COMMON);
@@ -806,4 +1028,26 @@ public abstract class BasePackFetchConnection extends BasePackConnection
 	private static class CancelledException extends Exception {
 		private static final long serialVersionUID = 1L;
 	}
+
+	private static class FetchStateV2 {
+
+		long havesToSend = 32;
+
+		long havesTotal;
+
+		long havesWithoutAck;
+
+		void incHavesToSend(boolean statelessRPC) {
+			if (statelessRPC) {
+				// Increase this quicker since connection setup costs accumulate
+				if (havesToSend < 16384) {
+					havesToSend *= 2;
+				} else {
+					havesToSend = havesToSend * 11 / 10;
+				}
+			} else {
+				havesToSend += 32;
+			}
+		}
+	}
 }