diff --git a/backend/cmd/api/main.go b/backend/cmd/api/main.go index 2a411bb..a7e8bfb 100644 --- a/backend/cmd/api/main.go +++ b/backend/cmd/api/main.go @@ -38,7 +38,11 @@ func run(logger *slog.Logger) error { return fmt.Errorf("load configuration: %w", err) } - httpClient := &http.Client{Timeout: config.ZarinHubTimeout} + httpClient := zarinhub.NewHTTPClient( + config.ZarinHubTimeout, + config.ZarinHubRetries, + logger, + ) tokenManager := zarinhub.NewTokenManager( httpClient, config.ZarinHubAuthEndpoint, @@ -46,9 +50,10 @@ func run(logger *slog.Logger) error { config.ZarinHubPassword, ) + authTimeout := time.Duration(config.ZarinHubRetries+1)*config.ZarinHubTimeout + 10*time.Second authContext, cancelAuth := context.WithTimeout( context.Background(), - config.ZarinHubTimeout, + authTimeout, ) _, err = tokenManager.AccessToken(authContext) cancelAuth() diff --git a/backend/internal/zarinhub/client.go b/backend/internal/zarinhub/client.go index 1ad06fa..05b8086 100644 --- a/backend/internal/zarinhub/client.go +++ b/backend/internal/zarinhub/client.go @@ -54,14 +54,48 @@ func NewClient( logger = loggers[0] } - return &Client{ + if httpClient == nil { + httpClient = http.DefaultClient + } + + client := &Client{ httpClient: httpClient, endpointURL: endpointURL, tokenProvider: tokenProvider, maxRetries: maxRetries, - retryWait: waitForRetry, logger: logger, } + + if _, ok := httpClient.Transport.(*RetryTransport); !ok && maxRetries > 0 { + rt := NewRetryTransport( + httpClient.Transport, + httpClient.Timeout, + maxRetries, + logger, + ) + rt.retryWait = func(ctx context.Context, delay time.Duration) bool { + if client.retryWait != nil { + return client.retryWait(ctx, delay) + } + return waitForRetry(ctx, delay) + } + client.httpClient = &http.Client{ + Transport: rt, + } + } else if rt, ok := httpClient.Transport.(*RetryTransport); ok { + origWait := rt.retryWait + rt.retryWait = func(ctx context.Context, delay time.Duration) bool { + if client.retryWait != nil { + return client.retryWait(ctx, delay) + } + if origWait != nil { + return origWait(ctx, delay) + } + return waitForRetry(ctx, delay) + } + } + + return client } func (c *Client) log() *slog.Logger { @@ -110,88 +144,7 @@ func (c *Client) Convert( ctx context.Context, cardNumber string, ) (service.ConversionResult, error) { - var ( - result service.ConversionResult - err error - ) - - waitFn := c.retryWait - if waitFn == nil { - waitFn = waitForRetry - } - - for attempt := 0; attempt <= c.maxRetries; attempt++ { - if attempt > 0 { - c.log().InfoContext( - ctx, - fmt.Sprintf("[RETRY] zarinhub request attempt %d", attempt), - "is_retry", true, - "retry_count", attempt, - ) - } else { - c.log().InfoContext( - ctx, - "[INITIAL] zarinhub request", - "is_retry", false, - "retry_count", 0, - ) - } - - result, err = c.convertOnce( - ctx, - cardNumber, - ) - - if err == nil { - return result, nil - } - - // Don't retry permanent/non-transient errors. - if !isRetryable(err) { - return result, err - } - - // We already used all configured retries. - if attempt == c.maxRetries { - if attempt > 0 { - c.log().ErrorContext( - ctx, - "[RETRY] zarinhub request failed: all retries exhausted", - "is_retry", true, - "retry_count", attempt, - "error", err, - ) - } - return result, err - } - - // If the caller canceled the request, stop immediately. - if ctx.Err() != nil { - if errors.Is(ctx.Err(), context.Canceled) { - return result, fmt.Errorf("zarinhub request canceled: %w", ctx.Err()) - } - return result, err - } - - delay := retryDelay(attempt) - c.log().WarnContext( - ctx, - fmt.Sprintf("[RETRY] zarinhub request failed, retrying in %v (retry %d)", delay, attempt+1), - "is_retry", true, - "next_retry", attempt + 1, - "delay_ms", delay.Milliseconds(), - "error", err, - ) - - if !waitFn(ctx, delay) { - if errors.Is(ctx.Err(), context.Canceled) { - return result, fmt.Errorf("zarinhub request canceled: %w", ctx.Err()) - } - return result, err - } - } - - return result, err + return c.convertOnce(ctx, cardNumber) } func (c *Client) convertOnce( diff --git a/backend/internal/zarinhub/token_manager_test.go b/backend/internal/zarinhub/token_manager_test.go index 435722d..6259ea2 100644 --- a/backend/internal/zarinhub/token_manager_test.go +++ b/backend/internal/zarinhub/token_manager_test.go @@ -3,6 +3,7 @@ package zarinhub import ( "context" "encoding/json" + "errors" "net/http" "net/http/httptest" "sync" @@ -220,3 +221,120 @@ func writeTokenResponse( t.Fatalf("encode response: %v", err) } } + +func TestTokenManagerRetriesOnTransientErrorsUntilSuccess(t *testing.T) { + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if calls == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + writeTokenResponse(t, w, "retried-access", "retried-refresh", 28800) + })) + defer server.Close() + + var recordedDelays []time.Duration + transport := NewRetryTransport( + server.Client().Transport, + time.Second, + 2, + ) + transport.retryWait = func(ctx context.Context, delay time.Duration) bool { + recordedDelays = append(recordedDelays, delay) + return true + } + httpClient := &http.Client{Transport: transport} + + manager := NewTokenManager(httpClient, server.URL, "test-user", "test-pass") + token, err := manager.AccessToken(context.Background()) + if err != nil { + t.Fatalf("AccessToken() error = %v", err) + } + if token != "retried-access" { + t.Errorf("token = %q, want retried-access", token) + } + if calls != 2 { + t.Errorf("calls = %d, want 2", calls) + } + if len(recordedDelays) != 1 { + t.Fatalf("len(recordedDelays) = %d, want 1", len(recordedDelays)) + } + if recordedDelays[0] != 200*time.Millisecond { + t.Errorf("delay[0] = %v, want 200ms", recordedDelays[0]) + } +} + +func TestTokenManagerRetriesOnTimeoutUntilSuccess(t *testing.T) { + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if calls == 1 { + time.Sleep(150 * time.Millisecond) + w.WriteHeader(http.StatusOK) + return + } + writeTokenResponse(t, w, "timeout-recovered-access", "timeout-recovered-refresh", 28800) + })) + defer server.Close() + + var recordedDelays []time.Duration + transport := NewRetryTransport( + server.Client().Transport, + 50*time.Millisecond, + 2, + ) + transport.retryWait = func(ctx context.Context, delay time.Duration) bool { + recordedDelays = append(recordedDelays, delay) + return true + } + httpClient := &http.Client{Transport: transport} + + manager := NewTokenManager(httpClient, server.URL, "test-user", "test-pass") + token, err := manager.AccessToken(context.Background()) + if err != nil { + t.Fatalf("AccessToken() error = %v", err) + } + if token != "timeout-recovered-access" { + t.Errorf("token = %q, want timeout-recovered-access", token) + } + if calls != 2 { + t.Errorf("calls = %d, want 2", calls) + } + if len(recordedDelays) != 1 { + t.Fatalf("len(recordedDelays) = %d, want 1", len(recordedDelays)) + } +} + +func TestTokenManagerRetriesExhausted(t *testing.T) { + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.WriteHeader(http.StatusBadGateway) + })) + defer server.Close() + + var recordedDelays []time.Duration + transport := NewRetryTransport( + server.Client().Transport, + time.Second, + 2, + ) + transport.retryWait = func(ctx context.Context, delay time.Duration) bool { + recordedDelays = append(recordedDelays, delay) + return true + } + httpClient := &http.Client{Transport: transport} + + manager := NewTokenManager(httpClient, server.URL, "test-user", "test-pass") + _, err := manager.AccessToken(context.Background()) + if !errors.Is(err, ErrUnavailable) { + t.Fatalf("error = %v, want ErrUnavailable", err) + } + if calls != 3 { + t.Errorf("calls = %d, want 3", calls) + } + if len(recordedDelays) != 2 { + t.Fatalf("len(recordedDelays) = %d, want 2", len(recordedDelays)) + } +} diff --git a/backend/internal/zarinhub/transport.go b/backend/internal/zarinhub/transport.go new file mode 100644 index 0000000..80ef157 --- /dev/null +++ b/backend/internal/zarinhub/transport.go @@ -0,0 +1,239 @@ +package zarinhub + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "time" +) + +type RetryTransport struct { + base http.RoundTripper + timeout time.Duration + maxRetries int + retryWait func(ctx context.Context, delay time.Duration) bool + logger *slog.Logger +} + +func NewRetryTransport( + base http.RoundTripper, + timeout time.Duration, + maxRetries int, + loggers ...*slog.Logger, +) *RetryTransport { + if base == nil { + base = http.DefaultTransport + } + if maxRetries < 0 { + maxRetries = 0 + } + + var logger *slog.Logger + if len(loggers) > 0 && loggers[0] != nil { + logger = loggers[0] + } + + return &RetryTransport{ + base: base, + timeout: timeout, + maxRetries: maxRetries, + retryWait: waitForRetry, + logger: logger, + } +} + +func NewHTTPClient( + timeout time.Duration, + maxRetries int, + loggers ...*slog.Logger, +) *http.Client { + transport := NewRetryTransport( + http.DefaultTransport, + timeout, + maxRetries, + loggers..., + ) + + return &http.Client{ + Transport: transport, + } +} + +func (t *RetryTransport) log() *slog.Logger { + if t.logger != nil { + return t.logger + } + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func (t *RetryTransport) RoundTrip(req *http.Request) (*http.Response, error) { + waitFn := t.retryWait + if waitFn == nil { + waitFn = waitForRetry + } + + if req.Body != nil && req.GetBody == nil { + bodyBytes, err := io.ReadAll(req.Body) + if err != nil { + return nil, err + } + _ = req.Body.Close() + req.GetBody = func() (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(bodyBytes)), nil + } + } + + var ( + resp *http.Response + err error + ) + + for attempt := 0; attempt <= t.maxRetries; attempt++ { + if attempt > 0 { + t.log().InfoContext( + req.Context(), + fmt.Sprintf("[RETRY] zarinhub request attempt %d", attempt), + "is_retry", true, + "retry_count", attempt, + ) + } else { + t.log().InfoContext( + req.Context(), + "[INITIAL] zarinhub request", + "is_retry", false, + "retry_count", 0, + ) + } + + var ( + attemptCtx context.Context + cancelAttempt context.CancelFunc + ) + if t.timeout > 0 { + attemptCtx, cancelAttempt = context.WithTimeout(req.Context(), t.timeout) + } else { + attemptCtx = req.Context() + cancelAttempt = func() {} + } + + attemptReq := req.Clone(attemptCtx) + if req.GetBody != nil { + body, bodyErr := req.GetBody() + if bodyErr != nil { + cancelAttempt() + return nil, bodyErr + } + attemptReq.Body = body + } + + resp, err = t.base.RoundTrip(attemptReq) + + if req.Context().Err() != nil { + cancelAttempt() + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + if errors.Is(req.Context().Err(), context.Canceled) { + return nil, fmt.Errorf("zarinhub request canceled: %w", req.Context().Err()) + } + return nil, req.Context().Err() + } + + isErrRetryable := false + var retryReason error + + if err != nil { + isErrRetryable = isTransportErrorRetryable(err) + retryReason = err + } else if isStatusCodeRetryable(resp.StatusCode) { + isErrRetryable = true + retryReason = fmt.Errorf("upstream returned status %d", resp.StatusCode) + } + + if !isErrRetryable { + if resp != nil && resp.Body != nil { + resp.Body = &bodyWithCancel{ + ReadCloser: resp.Body, + cancel: cancelAttempt, + } + } else { + cancelAttempt() + } + return resp, err + } + + if attempt == t.maxRetries { + if attempt > 0 { + t.log().ErrorContext( + req.Context(), + "[RETRY] zarinhub request failed: all retries exhausted", + "is_retry", true, + "retry_count", attempt, + "error", retryReason, + ) + } + if resp != nil && resp.Body != nil { + resp.Body = &bodyWithCancel{ + ReadCloser: resp.Body, + cancel: cancelAttempt, + } + } else { + cancelAttempt() + } + return resp, err + } + + if resp != nil && resp.Body != nil { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) + _ = resp.Body.Close() + } + cancelAttempt() + + delay := retryDelay(attempt) + t.log().WarnContext( + req.Context(), + fmt.Sprintf("[RETRY] zarinhub request failed, retrying in %v (retry %d)", delay, attempt+1), + "is_retry", true, + "next_retry", attempt+1, + "delay_ms", delay.Milliseconds(), + "error", retryReason, + ) + + if !waitFn(req.Context(), delay) { + if req.Context().Err() != nil { + if errors.Is(req.Context().Err(), context.Canceled) { + return nil, fmt.Errorf("zarinhub request canceled: %w", req.Context().Err()) + } + return nil, req.Context().Err() + } + return nil, context.Canceled + } + } + + return resp, err +} + +func isTransportErrorRetryable(err error) bool { + return !errors.Is(err, context.Canceled) +} + +func isStatusCodeRetryable(statusCode int) bool { + return statusCode == http.StatusRequestTimeout || + statusCode == http.StatusTooManyRequests || + statusCode >= http.StatusInternalServerError +} + +type bodyWithCancel struct { + io.ReadCloser + cancel context.CancelFunc +} + +func (b *bodyWithCancel) Close() error { + err := b.ReadCloser.Close() + b.cancel() + return err +} diff --git a/backend/internal/zarinhub/transport_test.go b/backend/internal/zarinhub/transport_test.go new file mode 100644 index 0000000..639951d --- /dev/null +++ b/backend/internal/zarinhub/transport_test.go @@ -0,0 +1,211 @@ +package zarinhub + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestRetryTransportPreservesAndRewindsBody(t *testing.T) { + var attempts int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + att := atomic.AddInt32(&attempts, 1) + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read body error: %v", err) + } + if string(body) != `{"payload":"important"}` { + t.Errorf("unexpected body on attempt %d: %q", att, string(body)) + } + if att < 3 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ok"}`)) + })) + defer server.Close() + + transport := NewRetryTransport(server.Client().Transport, time.Second, 3) + transport.retryWait = func(ctx context.Context, delay time.Duration) bool { + return true + } + client := &http.Client{Transport: transport} + + req, err := http.NewRequestWithContext( + context.Background(), + http.MethodPost, + server.URL, + bytes.NewReader([]byte(`{"payload":"important"}`)), + ) + if err != nil { + t.Fatalf("NewRequest error: %v", err) + } + + resp, err := client.Do(req) + if err != nil { + t.Fatalf("client.Do error: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusOK) + } + if attempts != 3 { + t.Errorf("attempts = %d, want 3", attempts) + } +} + +func TestRetryTransportDoesNotRetryClientErrors(t *testing.T) { + tests := []int{ + http.StatusBadRequest, + http.StatusUnauthorized, + http.StatusForbidden, + http.StatusNotFound, + http.StatusUnprocessableEntity, + } + + for _, statusCode := range tests { + t.Run(fmt.Sprintf("status_%d", statusCode), func(t *testing.T) { + var attempts int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&attempts, 1) + w.WriteHeader(statusCode) + })) + defer server.Close() + + transport := NewRetryTransport(server.Client().Transport, time.Second, 2) + client := &http.Client{Transport: transport} + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("client.Do error: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != statusCode { + t.Errorf("statusCode = %d, want %d", resp.StatusCode, statusCode) + } + if attempts != 1 { + t.Errorf("attempts = %d, want 1", attempts) + } + }) + } +} + +func TestRetryTransportPerAttemptTimeout(t *testing.T) { + var attempts int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + att := atomic.AddInt32(&attempts, 1) + if att == 1 { + time.Sleep(150 * time.Millisecond) + w.WriteHeader(http.StatusOK) + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + transport := NewRetryTransport(server.Client().Transport, 40*time.Millisecond, 2) + transport.retryWait = func(ctx context.Context, delay time.Duration) bool { + return true + } + client := &http.Client{Transport: transport} + + // Parent context has plenty of time (5s) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("client.Do error: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Errorf("status = %d, want 200", resp.StatusCode) + } + if attempts != 2 { + t.Errorf("attempts = %d, want 2", attempts) + } +} + +func TestRetryTransportParentContextCancellation(t *testing.T) { + var attempts int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&attempts, 1) + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer server.Close() + + ctx, cancel := context.WithCancel(context.Background()) + transport := NewRetryTransport(server.Client().Transport, time.Second, 3) + transport.retryWait = func(c context.Context, delay time.Duration) bool { + cancel() // cancel parent context during backoff + return waitForRetry(ctx, delay) + } + client := &http.Client{Transport: transport} + + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil) + _, err := client.Do(req) + if err == nil { + t.Fatal("expected error, got nil") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("error = %v, want context.Canceled", err) + } + if attempts != 1 { + t.Errorf("attempts = %d, want 1", attempts) + } +} + +func TestRetryTransportLoggingFormat(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer server.Close() + + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, nil)) + + transport := NewRetryTransport(server.Client().Transport, time.Second, 1, logger) + transport.retryWait = func(ctx context.Context, delay time.Duration) bool { + return true + } + client := &http.Client{Transport: transport} + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + logStr := buf.String() + if !strings.Contains(logStr, "[INITIAL] zarinhub request") { + t.Errorf("expected logs to contain '[INITIAL] zarinhub request', got: %s", logStr) + } + if !strings.Contains(logStr, `"is_retry":false`) { + t.Errorf("expected logs to contain '\"is_retry\":false', got: %s", logStr) + } + if !strings.Contains(logStr, "[RETRY] zarinhub request attempt 1") { + t.Errorf("expected logs to contain '[RETRY] zarinhub request attempt 1', got: %s", logStr) + } + if !strings.Contains(logStr, "[RETRY] zarinhub request failed, retrying in") { + t.Errorf("expected logs to contain retry warning, got: %s", logStr) + } + if !strings.Contains(logStr, "[RETRY] zarinhub request failed: all retries exhausted") { + t.Errorf("expected logs to contain exhausted message, got: %s", logStr) + } +}