Drop the whole connection on error in the proxy

Half-closed connections will get sources stuck.

node/proxy.go

@@ -150,11 +150,18 @@ func doIcecastProxy(rw http.ResponseWriter, req *http.Request, target *url.URL,
 // (>1.11), this is quite fast as io.CopyBuffer uses the splice()
 // system call internally (in exchange we lose the ability to figure
 // out which connection is the source of the error).
-func copyStream(out, in *net.TCPConn, promCounter prometheus.Counter, cntr *uint64) {
+func copyStream(tag string, out, in *net.TCPConn, promCounter prometheus.Counter, cntr *uint64) {
 	buf := getBuf()
 	defer releaseBuf(buf)
-	defer in.CloseRead()   //nolint
-	defer out.CloseWrite() //nolint
+
+	// We used to do this in order to support half-closed connections.
+	//defer in.CloseRead()   //nolint
+	//defer out.CloseWrite() //nolint
+
+	// Instead we do this and shut down the entire connection on error.
+	// We end up calling Close() twice but that's not a huge problem.
+	defer in.Close()  //nolint
+	defer out.Close() //nolint
 
 	for {
 		n, err := io.CopyBuffer(out, in, buf)
@@ -163,14 +170,21 @@ func copyStream(out, in *net.TCPConn, promCounter prometheus.Counter, cntr *uint
 			atomic.AddUint64(cntr, uint64(n))
 		}
 		if err != nil {
-			if err != io.EOF {
-				log.Printf("http: proxy error: %v", err)
+			if !isCloseError(err) {
+				log.Printf("http: proxy error (%s): %v", tag, err)
 			}
 			return
 		}
 	}
 }
 
+// This is a bad implementation (see https://github.com/golang/go/issues/4373
+// for some notes on why it is a layering violation), and we could replace it
+// with an atomic 'closing' flag.
+func isCloseError(err error) bool {
+	return strings.Contains(err.Error(), "use of closed network connection")
+}
+
 // Simple two-way TCP proxy that copies data in both directions and
 // can shutdown each direction of the connection independently.
 func handleProxy(conn *net.TCPConn, upstream *net.TCPConn, streamName string) {
@@ -181,11 +195,11 @@ func handleProxy(conn *net.TCPConn, upstream *net.TCPConn, streamName string) {
 	// Instrument both directions of the stream, but let the
 	// bandwidth estimator count only the bytes sent to the user.
 	go func() {
-		copyStream(conn, upstream, streamSentBytes.WithLabelValues(streamName), &bwBytesSent)
+		copyStream("upstream -> client", conn, upstream, streamSentBytes.WithLabelValues(streamName), &bwBytesSent)
 		wg.Done()
 	}()
 	go func() {
-		copyStream(upstream, conn, streamRcvdBytes.WithLabelValues(streamName), nil)
+		copyStream("client -> upstream", upstream, conn, streamRcvdBytes.WithLabelValues(streamName), nil)
 		wg.Done()
 	}()