Upgrade SplitHTTP Transport (#3462)

* move to paths instead of querystrings

* permit early data on serverside

* early data for the client, fix context cancellation

transport/internet/splithttp/hub.go

@@ -7,6 +7,7 @@ import (
 	gonet "net"
 	"net/http"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -28,20 +29,65 @@ type requestHandler struct {
 	localAddr gonet.TCPAddr
 }
 
+type httpSession struct {
+	uploadQueue *UploadQueue
+	// for as long as the GET request is not opened by the client, this will be
+	// open ("undone"), and the session may be expired within a certain TTL.
+	// after the client connects, this becomes "done" and the session lives as
+	// long as the GET request.
+	isFullyConnected *done.Instance
+}
+
+func (h *requestHandler) maybeReapSession(isFullyConnected *done.Instance, sessionId string) {
+	shouldReap := done.New()
+	go func() {
+		time.Sleep(30 * time.Second)
+		shouldReap.Close()
+	}()
+
+	select {
+	case <-isFullyConnected.Wait():
+		return
+	case <-shouldReap.Wait():
+		h.sessions.Delete(sessionId)
+	}
+}
+
+func (h *requestHandler) upsertSession(sessionId string) *httpSession {
+	currentSessionAny, ok := h.sessions.Load(sessionId)
+	if ok {
+		return currentSessionAny.(*httpSession)
+	}
+
+	s := &httpSession{
+		uploadQueue:      NewUploadQueue(int(2 * h.ln.config.GetNormalizedMaxConcurrentUploads())),
+		isFullyConnected: done.New(),
+	}
+
+	h.sessions.Store(sessionId, s)
+	go h.maybeReapSession(s.isFullyConnected, sessionId)
+	return s
+}
+
 func (h *requestHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	if len(h.host) > 0 && request.Host != h.host {
 		newError("failed to validate host, request:", request.Host, ", config:", h.host).WriteToLog()
 		writer.WriteHeader(http.StatusNotFound)
 		return
 	}
-	if request.URL.Path != h.path {
+
+	if !strings.HasPrefix(request.URL.Path, h.path) {
 		newError("failed to validate path, request:", request.URL.Path, ", config:", h.path).WriteToLog()
 		writer.WriteHeader(http.StatusNotFound)
 		return
 	}
 
-	queryString := request.URL.Query()
-	sessionId := queryString.Get("session")
+	sessionId := ""
+	subpath := strings.Split(request.URL.Path[len(h.path):], "/")
+	if len(subpath) > 0 {
+		sessionId = subpath[0]
+	}
+
 	if sessionId == "" {
 		newError("no sessionid on request:", request.URL.Path).WriteToLog()
 		writer.WriteHeader(http.StatusBadRequest)
@@ -60,15 +106,14 @@ func (h *requestHandler) ServeHTTP(writer http.ResponseWriter, request *http.Req
 		}
 	}
 
+	currentSession := h.upsertSession(sessionId)
+
 	if request.Method == "POST" {
-		uploadQueue, ok := h.sessions.Load(sessionId)
-		if !ok {
-			newError("sessionid does not exist").WriteToLog()
-			writer.WriteHeader(http.StatusBadRequest)
-			return
+		seq := ""
+		if len(subpath) > 1 {
+			seq = subpath[1]
 		}
 
-		seq := queryString.Get("seq")
 		if seq == "" {
 			newError("no seq on request:", request.URL.Path).WriteToLog()
 			writer.WriteHeader(http.StatusBadRequest)
@@ -89,7 +134,7 @@ func (h *requestHandler) ServeHTTP(writer http.ResponseWriter, request *http.Req
 			return
 		}
 
-		err = uploadQueue.(*UploadQueue).Push(Packet{
+		err = currentSession.uploadQueue.Push(Packet{
 			Payload: payload,
 			Seq:     seqInt,
 		})
@@ -107,10 +152,9 @@ func (h *requestHandler) ServeHTTP(writer http.ResponseWriter, request *http.Req
 			panic("expected http.ResponseWriter to be an http.Flusher")
 		}
 
-		uploadQueue := NewUploadQueue(int(2 * h.ln.config.GetNormalizedMaxConcurrentUploads()))
-
-		h.sessions.Store(sessionId, uploadQueue)
-		// the connection is finished, clean up map
+		// after GET is done, the connection is finished. disable automatic
+		// session reaping, and handle it in defer
+		currentSession.isFullyConnected.Close()
 		defer h.sessions.Delete(sessionId)
 
 		// magic header instructs nginx + apache to not buffer response body
@@ -130,7 +174,7 @@ func (h *requestHandler) ServeHTTP(writer http.ResponseWriter, request *http.Req
 				downloadDone:    downloadDone,
 				responseFlusher: responseFlusher,
 			},
-			reader:     uploadQueue,
+			reader:     currentSession.uploadQueue,
 			remoteAddr: remoteAddr,
 		}