Stripe CLI transfers across relay and direct paths; fix 10s relay deadline

cmd/cdx/direct.go

@@ -44,6 +44,8 @@ const (
 	maxSDPBytes = 64 * 1024
 )
 
+var errDirectStalled = errors.New("direct connection stalled")
+
 // directICEServers is a variable so tests can run without STUN.
 var directICEServers = []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302", "stun:global.stun.twilio.com:3478"}}}
 
@@ -230,7 +232,7 @@ func (path *directPath) send(record []byte) error {
 		case <-path.lost:
 			return errors.New("direct connection lost")
 		case <-time.After(transferIdle):
-			return errors.New("direct connection stalled")
+			return errDirectStalled
 		}
 	}
 	for start := 0; start < len(record); start += directPieceSize {

cmd/cdx/link.go

@@ -44,6 +44,13 @@ type recordLink struct {
 	// drops whichever copy arrives second.
 	retained []retainedRecord
 	switched bool
+	// striping, set by senders, keeps the relay carrying data records
+	// alongside the direct path once it opens (see stripe.go); stripe is
+	// its state from then on.
+	striping bool
+	stripe   *striper
+	// ackedBytes is the peer's last acknowledged byte count.
+	ackedBytes uint64
 }
 
 type retainedRecord struct {
@@ -89,6 +96,10 @@ func newRecordLink(connection *websocket.Conn, sealer *recordSealer, opener *rec
 // start begins reading the relay WebSocket; call it once the caller no
 // longer reads the socket itself.
 func (link *recordLink) start() {
+	// Clear the deadline waitRelayEvent left behind (10s on a receiver):
+	// next and poll time out on their own, and a deadline here would end
+	// the relay reader mid-transfer.
+	_ = link.connection.SetReadDeadline(time.Time{})
 	go link.readRelay()
 	go link.pingRelay()
 }
@@ -174,18 +185,30 @@ func (link *recordLink) sendReleasable(kind messageKind, payload []byte, release
 	if err != nil {
 		return err
 	}
+	if link.stripe != nil {
+		// The writer goroutines own both paths now.
+		return link.sendReleasableStriped(record, release)
+	}
 	path := link.direct.Load()
 	if path != nil && path.isOpen() {
 		link.directUsed.Store(true)
 		if !link.switched {
 			link.switched = true
+			if link.striping {
+				link.startStriping(path)
+			}
 			for _, earlier := range link.retained {
-				if err := path.send(earlier.record); err != nil {
+				if link.stripe != nil {
+					link.enqueueControl(earlier.record)
+				} else if err := path.send(earlier.record); err != nil {
 					return err
 				}
 			}
 			link.retained = nil
 		}
+		if link.stripe != nil {
+			return link.sendReleasableStriped(record, release)
+		}
 		return path.send(record)
 	}
 	if path != nil && !link.switched && !path.isLost() {
@@ -210,6 +233,10 @@ func (link *recordLink) acknowledged(bytes uint64) {
 		}
 	}
 	link.retained = kept
+	if link.stripe != nil {
+		link.stripe.acknowledged(bytes, link.ackedBytes, time.Now())
+	}
+	link.ackedBytes = bytes
 }
 
 // next returns the next record in sequence order, waiting up to timeout.
@@ -274,7 +301,10 @@ func (link *recordLink) accept(item linkItem) error {
 			return nil // an unused direct path failing changes nothing
 		}
 		if !item.direct && link.directUsed.Load() {
-			return nil // the relay is optional once the direct path carries the transfer
+			// The relay is optional once the direct path carries the
+			// transfer; records striped onto it go out again there.
+			link.relayFailed()
+			return nil
 		}
 		return item.err
 	}

cmd/cdx/sender.go

@@ -350,6 +350,7 @@ func sendFile(ctx context.Context, paths []string, linkMode bool, hooks sendHook
 		return err
 	}
 	link := newRecordLink(connection, sealer, opener, errors.New("receiver disconnected (they closed the tab or lost network)"))
+	link.striping = stripeEnabled()
 	defer link.close()
 	// Gather direct-path candidates while waiting for the receiver.
 	var offerSDP chan string