requester: defer execution of request.Item generation in closure (#723)

* requester: defer execution of request.Item generation in closure.

* bithumb: fix issue

* coinut/itbit: fix linter issues

* binance: fix bug on recvWindow setting

* requester: standardize sendpayload + add readme update

* nonce: remove inc() function

* request: defer unlockiflocked

* binance: revert changes for open orders

* btcmarkets: defer auth generation functionality, rm context deadline as this will be created just before sending HTTP request.

* binance: move const to top

* exmo: remove debug output as its generated in the requester function

* ftx: defer auth functionality

* requester: move error to top

* bittrex: defer auth functionality

* bitmex: defer auth functionality and remove deadline as generation occurs after rate limiting.

* btse: defer auth functionality

* coinbasepro: defer auth functionality and removed context deadline as this is generated after rate limiting

* coinbene: defer auth functionality and remove context deadline as this is generated after rate limiting

* huobi: defer auth functionality and remove context deadline as this is generated after rate limiting

* huobi-futures: defer auth functionality and remove context deadline as this is generated after rate limiting

* kraken: defer auth functionality and remove context deadline as this is generated after rate limiting

* kraken: remove deadline protection for timestamp generation

* okgroup: defer auth functionality and remove context deadline as this is generated after rate limiting

* poloniex: defer auth functionality

* zb: defer auth functionality and remove context deadline as this is generated after rate limiting

* exchanges: clean up log output which are done and inspected in the requester package

* binance: fix path bug on every retry, rm timeout context as this is not needed

* coinbene: fix path bug on retry

* binance: consolidate functionality

* coinbene: fix linter issues

* poloniex: linter fix

* kraken: change add -> set

* bitstamp: fix path bug for retry

* BTSE: fix retry path bug

* coinbene: fix path bug whoopsie by me

* gateio: fix bug where on retry it does not reset reader

* localbitcoins: fix path bug on retry

* zb: change domain to land

* exchanges: make sure io.Reader is generated every request

* exchanges: move reader generation into function scope

* wrapper_issues: setup exchange manager

* engine: expand withdraw manager test

* engine: dont look for environment

* bitstamp: fix pathing bug (@thrasher-)

* engine/withdraw_manager: purge tests as this is covered in repository withdraw

exchanges/request/request.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -20,6 +19,17 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/log"
 )
 
+var (
+	errRequestSystemIsNil     = errors.New("request system is nil")
+	errMaxRequestJobs         = errors.New("max request jobs reached")
+	errRequestFunctionIsNil   = errors.New("request function is nil")
+	errServiceNameUnset       = errors.New("service name unset")
+	errRequestItemNil         = errors.New("request item is nil")
+	errInvalidPath            = errors.New("invalid path")
+	errHeaderResponseMapIsNil = errors.New("header response map is nil")
+	errFailedToRetryRequest   = errors.New("failed to retry request")
+)
+
 // New returns a new Requester
 func New(name string, httpRequester *http.Client, opts ...RequesterOption) *Requester {
 	r := &Requester{
@@ -39,15 +49,47 @@ func New(name string, httpRequester *http.Client, opts ...RequesterOption) *Requ
 }
 
 // SendPayload handles sending HTTP/HTTPS requests
-func (r *Requester) SendPayload(ctx context.Context, i *Item) error {
+func (r *Requester) SendPayload(ctx context.Context, ep EndpointLimit, newRequest Generate) error {
+	if r == nil {
+		return errRequestSystemIsNil
+	}
+
+	defer r.timedLock.UnlockIfLocked()
+
+	if newRequest == nil {
+		return errRequestFunctionIsNil
+	}
+
+	if atomic.LoadInt32(&r.jobs) >= MaxRequestJobs {
+		return errMaxRequestJobs
+	}
+
+	atomic.AddInt32(&r.jobs, 1)
+	err := r.doRequest(ctx, ep, newRequest)
+	atomic.AddInt32(&r.jobs, -1)
+	return err
+}
+
+// validateRequest validates the requester item fields
+func (i *Item) validateRequest(ctx context.Context, r *Requester) (*http.Request, error) {
+	if i == nil {
+		return nil, errRequestItemNil
+	}
+
+	if i.Path == "" {
+		return nil, errInvalidPath
+	}
+
+	if i.HeaderResponse != nil && *i.HeaderResponse == nil {
+		return nil, errHeaderResponseMapIsNil
+	}
+
 	if !i.NonceEnabled {
 		r.timedLock.LockForDuration()
 	}
-
-	req, err := i.validateRequest(ctx, r)
+	req, err := http.NewRequestWithContext(ctx, i.Method, i.Path, i.Body)
 	if err != nil {
-		r.timedLock.UnlockIfLocked()
-		return err
+		return nil, err
 	}
 
 	if i.HTTPDebugging {
@@ -56,43 +98,6 @@ func (r *Requester) SendPayload(ctx context.Context, i *Item) error {
 		log.Debugf(log.RequestSys, "DumpRequest:\n%s", dump)
 	}
 
-	if atomic.LoadInt32(&r.jobs) >= MaxRequestJobs {
-		r.timedLock.UnlockIfLocked()
-		return errors.New("max request jobs reached")
-	}
-
-	atomic.AddInt32(&r.jobs, 1)
-	err = r.doRequest(req, i)
-	atomic.AddInt32(&r.jobs, -1)
-	r.timedLock.UnlockIfLocked()
-
-	return err
-}
-
-// validateRequest validates the requester item fields
-func (i *Item) validateRequest(ctx context.Context, r *Requester) (*http.Request, error) {
-	if r == nil || r.Name == "" {
-		return nil, errors.New("not initialised, SetDefaults() called before making request?")
-	}
-
-	if i == nil {
-		return nil, errors.New("request item cannot be nil")
-	}
-
-	if i.Path == "" {
-		return nil, errors.New("invalid path")
-	}
-
-	if i.HeaderResponse != nil {
-		if *i.HeaderResponse == nil {
-			return nil, errors.New("header response is nil")
-		}
-	}
-	req, err := http.NewRequestWithContext(ctx, i.Method, i.Path, i.Body)
-	if err != nil {
-		return nil, err
-	}
-
 	for k, v := range i.Headers {
 		req.Header.Add(k, v)
 	}
@@ -105,43 +110,35 @@ func (i *Item) validateRequest(ctx context.Context, r *Requester) (*http.Request
 }
 
 // DoRequest performs a HTTP/HTTPS request with the supplied params
-func (r *Requester) doRequest(req *http.Request, p *Item) error {
-	if p == nil {
-		return errors.New("request item cannot be nil")
-	}
-	if p.Verbose {
-		log.Debugf(log.RequestSys,
-			"%s request path: %s",
-			r.Name,
-			p.Path)
-
-		for k, d := range req.Header {
-			log.Debugf(log.RequestSys,
-				"%s request header [%s]: %s",
-				r.Name,
-				k,
-				d)
-		}
-		log.Debugf(log.RequestSys,
-			"%s request type: %s",
-			r.Name,
-			req.Method)
-
-		if p.Body != nil {
-			log.Debugf(log.RequestSys,
-				"%s request body: %v",
-				r.Name,
-				p.Body)
-		}
-	}
-
+func (r *Requester) doRequest(ctx context.Context, endpoint EndpointLimit, newRequest Generate) error {
 	for attempt := 1; ; attempt++ {
 		// Initiate a rate limit reservation and sleep on requested endpoint
-		err := r.InitiateRateLimit(p.Endpoint)
+		err := r.InitiateRateLimit(endpoint)
 		if err != nil {
 			return err
 		}
 
+		p, err := newRequest()
+		if err != nil {
+			return err
+		}
+
+		req, err := p.validateRequest(ctx, r)
+		if err != nil {
+			return err
+		}
+
+		if p.Verbose {
+			log.Debugf(log.RequestSys, "%s attempt %d request path: %s", r.Name, attempt, p.Path)
+			for k, d := range req.Header {
+				log.Debugf(log.RequestSys, "%s request header [%s]: %s", r.Name, k, d)
+			}
+			log.Debugf(log.RequestSys, "%s request type: %s", r.Name, p.Method)
+			if p.Body != nil {
+				log.Debugf(log.RequestSys, "%s request body: %v", r.Name, p.Body)
+			}
+		}
+
 		resp, err := r.HTTPClient.Do(req)
 		if retry, checkErr := r.retryPolicy(resp, err); checkErr != nil {
 			return checkErr
@@ -151,18 +148,11 @@ func (r *Requester) doRequest(req *http.Request, p *Item) error {
 				r.drainBody(resp.Body)
 			}
 
-			// Can't currently regenerate nonce and signatures with fresh values for retries, so for now, we must not retry
-			if p.NonceEnabled {
-				if timeoutErr, ok := err.(net.Error); !ok || !timeoutErr.Timeout() {
-					return fmt.Errorf("unable to retry request using nonce, err: %v", err)
-				}
-			}
-
 			if attempt > r.maxRetries {
 				if err != nil {
-					return fmt.Errorf("failed to retry request, err: %v", err)
+					return fmt.Errorf("%w, err: %v", errFailedToRetryRequest, err)
 				}
-				return fmt.Errorf("failed to retry request, status: %s", resp.Status)
+				return fmt.Errorf("%w, status: %s", errFailedToRetryRequest, resp.Status)
 			}
 
 			after := RetryAfter(resp, time.Now())
@@ -262,8 +252,7 @@ func (r *Requester) GetNonce(isNano bool) nonce.Value {
 		}
 		return r.Nonce.Get()
 	}
-	r.Nonce.Inc()
-	return r.Nonce.Get()
+	return r.Nonce.GetInc()
 }
 
 // GetNonceMilli returns a nonce for requests. This locks and enforces concurrent
@@ -274,8 +263,7 @@ func (r *Requester) GetNonceMilli() nonce.Value {
 		r.Nonce.Set(time.Now().UnixNano() / int64(time.Millisecond))
 		return r.Nonce.Get()
 	}
-	r.Nonce.Inc()
-	return r.Nonce.Get()
+	return r.Nonce.GetInc()
 }
 
 // SetProxy sets a proxy address to the client transport

exchanges/request/request_test.go

@@ -178,6 +178,8 @@ type GlobalLimitTest struct {
 	UnAuth *rate.Limiter
 }
 
+var errEndpointLimitNotFound = errors.New("endpoint limit not found")
+
 func (g *GlobalLimitTest) Limit(e EndpointLimit) error {
 	switch e {
 	case Auth:
@@ -193,7 +195,9 @@ func (g *GlobalLimitTest) Limit(e EndpointLimit) error {
 		time.Sleep(g.UnAuth.Reserve().Delay())
 		return nil
 	default:
-		return fmt.Errorf("cannot execute functionality: %d not found", e)
+		return fmt.Errorf("cannot execute functionality: %d %w",
+			e,
+			errEndpointLimitNotFound)
 	}
 }
 
@@ -203,81 +207,92 @@ var globalshell = GlobalLimitTest{
 
 func TestDoRequest(t *testing.T) {
 	t.Parallel()
-	r := New("test",
-		new(http.Client),
-		WithLimiter(&globalshell))
+	r := New("test", new(http.Client), WithLimiter(&globalshell))
+
 	ctx := context.Background()
-
-	err := r.SendPayload(ctx, &Item{})
-	if err == nil {
-		t.Fatal(unexpected)
+	err := (*Requester)(nil).SendPayload(ctx, Unset, nil)
+	if !errors.Is(errRequestSystemIsNil, err) {
+		t.Fatalf("expected: %v but received: %v", errRequestSystemIsNil, err)
 	}
-	if !strings.Contains(err.Error(), "invalid path") {
-		t.Fatal(err)
+	err = r.SendPayload(ctx, Unset, nil)
+	if !errors.Is(errRequestFunctionIsNil, err) {
+		t.Fatalf("expected: %v but received: %v", errRequestFunctionIsNil, err)
 	}
 
-	err = r.SendPayload(ctx, &Item{Method: http.MethodGet})
-	if err == nil {
-		t.Fatal(unexpected)
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) { return nil, nil })
+	if !errors.Is(errRequestItemNil, err) {
+		t.Fatalf("expected: %v but received: %v", errRequestItemNil, err)
 	}
-	if !strings.Contains(err.Error(), "invalid path") {
-		t.Fatal(err)
+
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) { return &Item{}, nil })
+	if !errors.Is(errInvalidPath, err) {
+		t.Fatalf("expected: %v but received: %v", errInvalidPath, err)
+	}
+
+	var nilHeader http.Header
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) {
+		return &Item{
+			Path:           testURL,
+			HeaderResponse: &nilHeader,
+		}, nil
+	})
+	if !errors.Is(errHeaderResponseMapIsNil, err) {
+		t.Fatalf("expected: %v but received: %v", errHeaderResponseMapIsNil, err)
 	}
 
 	// Invalid/missing endpoint limit
-	err = r.SendPayload(ctx, &Item{
-		Method: http.MethodGet,
-		Path:   testURL,
+	err = r.SendPayload(ctx, Unset, func() (*Item, error) {
+		return &Item{
+			Path: testURL,
+		}, nil
 	})
-	if err == nil {
-		t.Fatal(unexpected)
-	}
-	if !strings.Contains(err.Error(), "cannot execute functionality") {
-		t.Fatal(err)
+	if !errors.Is(err, errEndpointLimitNotFound) {
+		t.Fatalf("expected: %v but received: %v", errEndpointLimitNotFound, err)
 	}
 
-	// force debug
-	err = r.SendPayload(ctx, &Item{
-		Method:        http.MethodGet,
-		Path:          testURL,
-		HTTPDebugging: true,
-		Verbose:       true,
+	// Force debug
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) {
+		return &Item{
+			Path: testURL,
+			Headers: map[string]string{
+				"test": "supertest",
+			},
+			Body:          strings.NewReader("test"),
+			HTTPDebugging: true,
+			Verbose:       true,
+		}, nil
 	})
-	if err == nil {
-		t.Fatal(unexpected)
+	if !errors.Is(err, nil) {
+		t.Fatalf("received: %v but expected: %v", err, nil)
 	}
-	if !strings.Contains(err.Error(), "cannot execute functionality") {
-		t.Fatal(err)
+
+	// Fail new request call
+	newError := errors.New("request item failure")
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) {
+		return nil, newError
+	})
+	if !errors.Is(err, newError) {
+		t.Fatalf("received: %v but expected: %v", err, newError)
 	}
 
 	// max request job ceiling
 	r.jobs = MaxRequestJobs
-	err = r.SendPayload(ctx, &Item{
-		Method:   http.MethodGet,
-		Path:     testURL,
-		Endpoint: UnAuth,
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) {
+		return &Item{Path: testURL}, nil
 	})
-	if err == nil {
-		t.Fatal(unexpected)
-	}
-	if !strings.Contains(err.Error(), "max request jobs reached") {
-		t.Fatal(err)
+	if !errors.Is(err, errMaxRequestJobs) {
+		t.Fatalf("received: %v but expected: %v", err, errMaxRequestJobs)
 	}
 	// reset jobs
 	r.jobs = 0
 
 	// timeout checker
 	r.HTTPClient.Timeout = time.Millisecond * 50
-	err = r.SendPayload(ctx, &Item{
-		Method:   http.MethodGet,
-		Path:     testURL + "/timeout",
-		Endpoint: UnAuth,
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) {
+		return &Item{Path: testURL + "/timeout"}, nil
 	})
-	if err == nil {
-		t.Fatal(unexpected)
-	}
-	if !strings.Contains(err.Error(), "failed to retry request") {
-		t.Fatal(err)
+	if !errors.Is(err, errFailedToRetryRequest) {
+		t.Fatalf("received: %v but expected: %v", err, errFailedToRetryRequest)
 	}
 	// reset timeout
 	r.HTTPClient.Timeout = 0
@@ -289,18 +304,16 @@ func TestDoRequest(t *testing.T) {
 
 	// Check header contents
 	var passback = http.Header{}
-	err = r.SendPayload(ctx, &Item{
-		Method:         http.MethodGet,
-		Path:           testURL,
-		Result:         &resp,
-		Endpoint:       UnAuth,
-		HeaderResponse: &passback,
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) {
+		return &Item{
+			Method:         http.MethodGet,
+			Path:           testURL,
+			Result:         &resp,
+			HeaderResponse: &passback,
+		}, nil
 	})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if !resp.Response {
-		t.Fatal(unexpected)
+	if !errors.Is(err, nil) {
+		t.Fatalf("received: %v but expected: %v", err, nil)
 	}
 
 	if passback.Get("Content-Length") != "17" {
@@ -315,17 +328,19 @@ func TestDoRequest(t *testing.T) {
 	var respErr struct {
 		Error bool `json:"error"`
 	}
-	err = r.SendPayload(ctx, &Item{
-		Method:   http.MethodGet,
-		Path:     testURL,
-		Result:   &respErr,
-		Endpoint: UnAuth,
+	err = r.SendPayload(ctx, UnAuth, func() (*Item, error) {
+		return &Item{
+			Method: http.MethodGet,
+			Path:   testURL,
+			Result: &respErr,
+		}, nil
 	})
-	if err != nil {
-		t.Fatal(err)
+	if !errors.Is(err, nil) {
+		t.Fatalf("received: %v but expected: %v", err, nil)
 	}
-	if !resp.Response {
-		t.Fatal(unexpected)
+
+	if respErr.Error {
+		t.Fatal("unexpected value")
 	}
 
 	// Check client side rate limit
@@ -337,12 +352,13 @@ func TestDoRequest(t *testing.T) {
 			var resp struct {
 				Response bool `json:"response"`
 			}
-			payloadError := r.SendPayload(ctx, &Item{
-				Method:      http.MethodGet,
-				Path:        testURL + "/rate",
-				Result:      &resp,
-				AuthRequest: true,
-				Endpoint:    Auth,
+			payloadError := r.SendPayload(ctx, Auth, func() (*Item, error) {
+				return &Item{
+					Method:      http.MethodGet,
+					Path:        testURL + "/rate",
+					Result:      &resp,
+					AuthRequest: true,
+				}, nil
 			})
 			wg.Done()
 			if payloadError != nil {
@@ -378,12 +394,13 @@ func TestDoRequest_Retries(t *testing.T) {
 			var resp struct {
 				Response bool `json:"response"`
 			}
-			payloadError := r.SendPayload(context.Background(), &Item{
-				Method:      http.MethodGet,
-				Path:        testURL + "/rate-retry",
-				Result:      &resp,
-				AuthRequest: true,
-				Endpoint:    Auth,
+			payloadError := r.SendPayload(context.Background(), Auth, func() (*Item, error) {
+				return &Item{
+					Method:      http.MethodGet,
+					Path:        testURL + "/rate-retry",
+					Result:      &resp,
+					AuthRequest: true,
+				}, nil
 			})
 			if payloadError != nil {
 				atomic.StoreInt32(&failed, 1)
@@ -409,31 +426,36 @@ func TestDoRequest_RetryNonRecoverable(t *testing.T) {
 		return 0
 	}
 	r := New("test", new(http.Client), WithBackoff(backoff))
-	payloadError := r.SendPayload(context.Background(), &Item{
-		Method: http.MethodGet,
-		Path:   testURL + "/always-retry",
+	err := r.SendPayload(context.Background(), Unset, func() (*Item, error) {
+		return &Item{
+			Method: http.MethodGet,
+			Path:   testURL + "/always-retry",
+		}, nil
 	})
-	if payloadError == nil {
-		t.Fatal("expected an error")
+	if !errors.Is(err, errFailedToRetryRequest) {
+		t.Fatalf("received: %v but expected: %v", err, errFailedToRetryRequest)
 	}
 }
 
 func TestDoRequest_NotRetryable(t *testing.T) {
 	t.Parallel()
 
+	notRetryErr := errors.New("not retryable")
 	retry := func(resp *http.Response, err error) (bool, error) {
-		return false, errors.New("not retryable")
+		return false, notRetryErr
 	}
 	backoff := func(n int) time.Duration {
 		return time.Duration(n) * time.Millisecond
 	}
 	r := New("test", new(http.Client), WithRetryPolicy(retry), WithBackoff(backoff))
-	payloadError := r.SendPayload(context.Background(), &Item{
-		Method: http.MethodGet,
-		Path:   testURL + "/always-retry",
+	err := r.SendPayload(context.Background(), Unset, func() (*Item, error) {
+		return &Item{
+			Method: http.MethodGet,
+			Path:   testURL + "/always-retry",
+		}, nil
 	})
-	if payloadError == nil {
-		t.Fatal("expected an error")
+	if !errors.Is(err, notRetryErr) {
+		t.Fatalf("received: %v but expected: %v", err, notRetryErr)
 	}
 }
 
@@ -505,8 +527,8 @@ func TestBasicLimiter(t *testing.T) {
 	ctx := context.Background()
 
 	tn := time.Now()
-	_ = r.SendPayload(ctx, &i)
-	_ = r.SendPayload(ctx, &i)
+	_ = r.SendPayload(ctx, Unset, func() (*Item, error) { return &i, nil })
+	_ = r.SendPayload(ctx, Unset, func() (*Item, error) { return &i, nil })
 	if time.Since(tn) < time.Second {
 		t.Error("rate limit issues")
 	}
@@ -519,12 +541,13 @@ func TestEnableDisableRateLimit(t *testing.T) {
 	ctx := context.Background()
 
 	var resp interface{}
-	err := r.SendPayload(ctx, &Item{
-		Method:      http.MethodGet,
-		Path:        testURL,
-		Result:      &resp,
-		AuthRequest: true,
-		Endpoint:    Auth,
+	err := r.SendPayload(ctx, Auth, func() (*Item, error) {
+		return &Item{
+			Method:      http.MethodGet,
+			Path:        testURL,
+			Result:      &resp,
+			AuthRequest: true,
+		}, nil
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -540,12 +563,13 @@ func TestEnableDisableRateLimit(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = r.SendPayload(ctx, &Item{
-		Method:      http.MethodGet,
-		Path:        testURL,
-		Result:      &resp,
-		AuthRequest: true,
-		Endpoint:    Auth,
+	err = r.SendPayload(ctx, Auth, func() (*Item, error) {
+		return &Item{
+			Method:      http.MethodGet,
+			Path:        testURL,
+			Result:      &resp,
+			AuthRequest: true,
+		}, nil
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -564,12 +588,13 @@ func TestEnableDisableRateLimit(t *testing.T) {
 	ti := time.NewTicker(time.Second)
 	c := make(chan struct{})
 	go func(c chan struct{}) {
-		err = r.SendPayload(ctx, &Item{
-			Method:      http.MethodGet,
-			Path:        testURL,
-			Result:      &resp,
-			AuthRequest: true,
-			Endpoint:    Auth,
+		err = r.SendPayload(ctx, Auth, func() (*Item, error) {
+			return &Item{
+				Method:      http.MethodGet,
+				Path:        testURL,
+				Result:      &resp,
+				AuthRequest: true,
+			}, nil
 		})
 		if err != nil {
 			log.Fatal(err)

exchanges/request/request_types.go

@@ -56,7 +56,6 @@ type Item struct {
 	// HeaderResponse for inspection of header contents package side useful for
 	// pagination
 	HeaderResponse *http.Header
-	Endpoint       EndpointLimit
 }
 
 // Backoff determines how long to wait between request attempts.
@@ -67,3 +66,9 @@ type RetryPolicy func(resp *http.Response, err error) (bool, error)
 
 // RequesterOption is a function option that can be applied to configure a Requester when creating it.
 type RequesterOption func(*Requester)
+
+// Generate defines a closure for functionality outside of the requester to
+// to generate new *http.Request on every attempt. This minimizes the chance of
+// being outside of receive window if application rate limiting reduces outbound
+// requests.
+type Generate func() (*Item, error)