refine error handling in retry logic

app/proxyman/outbound/handler.go

@@ -66,10 +66,13 @@ func NewHandler(ctx context.Context, config *proxyman.OutboundHandlerConfig) (*H
 func (h *Handler) Dispatch(ctx context.Context, outboundRay ray.OutboundRay) {
 	err := h.proxy.Process(ctx, outboundRay, h)
 	// Ensure outbound ray is properly closed.
-	if err != nil && errors.Cause(err) != io.EOF {
+	if err != nil {
-		outboundRay.OutboundOutput().CloseError()
+		log.Warning("Proxyman|OutboundHandler: Failed to process outbound traffic.")
-	} else {
+		if errors.Cause(err) != io.EOF {
-		outboundRay.OutboundOutput().Close()
+			outboundRay.OutboundOutput().CloseError()
+		} else {
+			outboundRay.OutboundOutput().Close()
+		}
 	}
 	outboundRay.OutboundInput().CloseError()
 }

common/retry/retry.go

@@ -2,11 +2,12 @@ package retry
 
 import (
 	"time"
+
 	"v2ray.com/core/common/errors"
 )
 
 var (
-	ErrRetryFailed = errors.New("All retry attempts failed.")
+	ErrRetryFailed = errors.New("Retry: All retry attempts failed.")
 )
 
 // Strategy is a way to retry on a specific function.
@@ -23,16 +24,21 @@ type retryer struct {
 // On implements Strategy.On.
 func (r *retryer) On(method func() error) error {
 	attempt := 0
+	accumulatedError := make([]error, 0, r.totalAttempt)
 	for attempt < r.totalAttempt {
 		err := method()
 		if err == nil {
 			return nil
 		}
+		numErrors := len(accumulatedError)
+		if numErrors == 0 || err.Error() != accumulatedError[numErrors-1].Error() {
+			accumulatedError = append(accumulatedError, err)
+		}
 		delay := r.nextDelay()
 		<-time.After(time.Duration(delay) * time.Millisecond)
 		attempt++
 	}
-	return ErrRetryFailed
+	return errors.Base(ErrRetryFailed).Message(accumulatedError)
 }
 
 // Timed returns a retry strategy with fixed interval.

common/retry/retry_test.go

@@ -73,7 +73,7 @@ func TestRetryExhausted(t *testing.T) {
 	})
 	duration := time.Since(startTime)
 
-	assert.Error(err).Equals(ErrRetryFailed)
+	assert.Error(errors.Cause(err)).Equals(ErrRetryFailed)
 	assert.Int64(int64(duration / time.Millisecond)).AtLeast(1900)
 }
 
@@ -88,6 +88,6 @@ func TestExponentialBackoff(t *testing.T) {
 	})
 	duration := time.Since(startTime)
 
-	assert.Error(err).Equals(ErrRetryFailed)
+	assert.Error(errors.Cause(err)).Equals(ErrRetryFailed)
 	assert.Int64(int64(duration / time.Millisecond)).AtLeast(4000)
 }

proxy/blackhole/blackhole.go

@@ -3,7 +3,6 @@ package blackhole
 
 import (
 	"context"
-	"errors"
 	"time"
 
 	"v2ray.com/core/common"
@@ -11,10 +10,6 @@ import (
 	"v2ray.com/core/transport/ray"
 )
 
-var (
-	errConnectionBlocked = errors.New("Blackhole: connection blocked.")
-)
-
 // Handler is an outbound connection that sliently swallow the entire payload.
 type Handler struct {
 	response ResponseConfig
@@ -36,7 +31,9 @@ func (v *Handler) Process(ctx context.Context, outboundRay ray.OutboundRay, dial
 	v.response.WriteTo(outboundRay.OutboundOutput())
 	// Sleep a little here to make sure the response is sent to client.
 	time.Sleep(time.Second)
-	return errConnectionBlocked
+	outboundRay.OutboundOutput().Close()
+	outboundRay.OutboundInput().CloseError()
+	return nil
 }
 
 func init() {

proxy/freedom/freedom.go

@@ -102,8 +102,7 @@ func (v *Handler) Process(ctx context.Context, outboundRay ray.OutboundRay, dial
 		return nil
 	})
 	if err != nil {
-		log.Warning("Freedom: Failed to open connection to ", destination, ": ", err)
+		return errors.Base(err).Message("Freedom: Failed to open connection to ", destination)
-		return err
 	}
 	defer conn.Close()
 

proxy/shadowsocks/client.go

@@ -2,7 +2,6 @@ package shadowsocks
 
 import (
 	"context"
-	"errors"
 	"runtime"
 	"time"
 
@@ -10,6 +9,7 @@ import (
 	"v2ray.com/core/common"
 	"v2ray.com/core/common/buf"
 	"v2ray.com/core/common/bufio"
+	"v2ray.com/core/common/errors"
 	"v2ray.com/core/common/net"
 	"v2ray.com/core/common/protocol"
 	"v2ray.com/core/common/retry"
@@ -61,8 +61,7 @@ func (v *Client) Process(ctx context.Context, outboundRay ray.OutboundRay, diale
 		return nil
 	})
 	if err != nil {
-		log.Warning("Shadowsocks|Client: Failed to find an available destination:", err)
+		return errors.Base(err).Message("Shadowsocks|Client: Failed to find an available destination.")
-		return err
 	}
 	log.Info("Shadowsocks|Client: Tunneling request to ", destination, " via ", server.Destination())
 

proxy/socks/client.go

@@ -2,13 +2,13 @@ package socks
 
 import (
 	"context"
-	"errors"
 	"runtime"
 	"time"
 
 	"v2ray.com/core/app/log"
 	"v2ray.com/core/common"
 	"v2ray.com/core/common/buf"
+	"v2ray.com/core/common/errors"
 	"v2ray.com/core/common/net"
 	"v2ray.com/core/common/protocol"
 	"v2ray.com/core/common/retry"
@@ -56,8 +56,7 @@ func (c *Client) Process(ctx context.Context, ray ray.OutboundRay, dialer proxy.
 	})
 
 	if err != nil {
-		log.Warning("Socks|Client: Failed to find an available destination.")
+		return errors.Base(err).Message("Socks|Client: Failed to find an available destination.")
-		return err
 	}
 
 	defer conn.Close()

proxy/vmess/outbound/outbound.go

@@ -51,7 +51,7 @@ func (v *VMessOutboundHandler) Process(ctx context.Context, outboundRay ray.Outb
 	var rec *protocol.ServerSpec
 	var conn internet.Connection
 
-	err := retry.ExponentialBackoff(5, 100).On(func() error {
+	err := retry.ExponentialBackoff(5, 200).On(func() error {
 		rec = v.serverPicker.PickServer()
 		rawConn, err := dialer.Dial(ctx, rec.Destination())
 		if err != nil {
@@ -62,8 +62,7 @@ func (v *VMessOutboundHandler) Process(ctx context.Context, outboundRay ray.Outb
 		return nil
 	})
 	if err != nil {
-		log.Warning("VMess|Outbound: Failed to find an available destination:", err)
+		return errors.Base(err).Message("VMess|Outbound: Failed to find an available destination.")
-		return err
 	}
 	defer conn.Close()
 

transport/internet/tcp_hub.go

@@ -5,7 +5,6 @@ import (
 	"sync"
 
 	"v2ray.com/core/common/errors"
-	"v2ray.com/core/app/log"
 	v2net "v2ray.com/core/common/net"
 	"v2ray.com/core/common/retry"
 )
@@ -73,16 +72,13 @@ func (v *TCPHub) start() {
 	v.accepting = true
 	for v.accepting {
 		var newConn Connection
-		err := retry.ExponentialBackoff(10, 200).On(func() error {
+		err := retry.ExponentialBackoff(10, 500).On(func() error {
 			if !v.accepting {
 				return nil
 			}
 			conn, err := v.listener.Accept()
 			if err != nil {
-				if v.accepting {
+				return errors.Base(err).Message("Internet|Listener: Failed to accept new TCP connection.")
-					log.Warning("Internet|Listener: Failed to accept new TCP connection: ", err)
-				}
-				return err
 			}
 			newConn = conn
 			return nil

transport/internet/websocket/dialer.go

@@ -2,12 +2,12 @@ package websocket
 
 import (
 	"context"
-	"io/ioutil"
 	"net"
 
 	"github.com/gorilla/websocket"
 	"v2ray.com/core/app/log"
 	"v2ray.com/core/common"
+	"v2ray.com/core/common/errors"
 	v2net "v2ray.com/core/common/net"
 	"v2ray.com/core/transport/internet"
 	"v2ray.com/core/transport/internet/internal"
@@ -78,11 +78,11 @@ func wsDial(ctx context.Context, dest v2net.Destination) (net.Conn, error) {
 
 	conn, resp, err := dialer.Dial(uri, nil)
 	if err != nil {
+		var reason string
 		if resp != nil {
-			reason, reasonerr := ioutil.ReadAll(resp.Body)
+			reason = resp.Status
-			log.Info(string(reason), reasonerr)
 		}
-		return nil, err
+		return nil, errors.Base(err).Format("WebSocket|Dialer: Failed to dial to (", uri, "): ", reason)
 	}
 
 	return &wsconn{