package main import ( "context" "fmt" "net/http" "net/http/httptest" "strings" "sync" "sync/atomic" "time" "github.com/gofiber/fiber/v2" libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache" libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring" "github.com/sony/gobreaker" "github.com/valyala/fasthttp" ) // Integration tests that test the interactions between different components // TestCachingAndCircuitBreakerInteraction tests the interaction between // caching system and circuit breaker func (suite *Tests) TestCachingAndCircuitBreakerInteraction() { // Original values to restore later originalCircuitBreaker := cfg.CircuitBreaker originalCache := cfg.Cache originalClient := cfg.Client.FastProxyClient // Restore after test defer func() { cfg.CircuitBreaker = originalCircuitBreaker cfg.Cache = originalCache cfg.Client.FastProxyClient = originalClient // Reset the circuit breaker cbMutex.Lock() cb = nil cbMetrics = nil cbMutex.Unlock() }() // Ensure cache is enabled cfg.Cache.CacheEnable = true cfg.Cache.CacheTTL = 60 // 60 seconds // Configure circuit breaker cfg.CircuitBreaker.Enable = true cfg.CircuitBreaker.MaxFailures = 3 cfg.CircuitBreaker.Timeout = 5 // 5 seconds to half-open cfg.CircuitBreaker.ReturnCachedOnOpen = true cfg.CircuitBreaker.TripOn5xx = true // Initialize circuit breaker initCircuitBreaker(cfg) // Set up test server with variable behavior responseStatus := http.StatusOK responseBody := `{"data":{"test":"original"}}` responseDelay := time.Duration(0) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // Apply configured delay time.Sleep(responseDelay) // Return configured response w.Header().Set("Content-Type", "application/json") w.WriteHeader(responseStatus) _, _ = w.Write([]byte(responseBody)) })) defer server.Close() // Configure client cfg.Client.ClientTimeout = 2 // 2 seconds (shorter than server delay for timeout tests) cfg.Client.FastProxyClient = createFasthttpClient(cfg) // Configure server URL cfg.Server.HostGraphQL = server.URL // Track metrics trackedMetrics := []string{ libpack_monitoring.MetricsCacheHit, libpack_monitoring.MetricsCacheMiss, libpack_monitoring.MetricsCircuitFallbackSuccess, libpack_monitoring.MetricsCircuitFallbackFailed, } metricCounts := make(map[string]int, len(trackedMetrics)) // Capture initial metric values for _, metric := range trackedMetrics { metricCounts[metric] = getMetricValue(metric) } // Test Case 1: Initial request is successful and cached t := suite.T() // Create request context reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqBody := `{"query": "query { test }"}` reqCtx.Request.SetBody([]byte(reqBody)) // Initialize the cache libpack_cache.EnableCache(&libpack_cache.CacheConfig{ Logger: cfg.Logger, TTL: cfg.Cache.CacheTTL, }) // First request: should succeed and be cached ctx := suite.app.AcquireCtx(reqCtx) err := proxyTheRequest(ctx, cfg.Server.HostGraphQL) // Save response before releasing context firstResponseBody := string(ctx.Response().Body()) suite.Nil(err, "First request should succeed") suite.Equal(responseBody, firstResponseBody, "Response body should match server response") // Calculate hash the same way the system does, before releasing context // Use default user context ("-", "-") since no auth headers are set in this test cacheKey := libpack_cache.CalculateHash(ctx, "-", "-") // Store in cache directly for test libpack_cache.CacheStore(cacheKey, []byte(responseBody)) suite.app.ReleaseCtx(ctx) // Verify cache was populated cachedResponse := libpack_cache.CacheLookup(cacheKey) suite.NotNil(cachedResponse, "Response should be cached") suite.Equal(responseBody, string(cachedResponse), "Cached response should match server response") // Test Case 2: Server begins failing, trips circuit breaker, fallback to cache // Update server to fail with 500 errors responseStatus = http.StatusInternalServerError responseBody = `{"errors":[{"message":"Server error"}]}` // Make enough failing requests to trip the circuit for i := 0; i < cfg.CircuitBreaker.MaxFailures; i++ { ctx = suite.app.AcquireCtx(reqCtx) _ = proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) } // Verify circuit is now open suite.Equal(gobreaker.StateOpen.String(), cb.State().String(), "Circuit should be open after failures") // Update server to return success again (but circuit is open, so this shouldn't be called) responseStatus = http.StatusOK responseBody = `{"data":{"test":"updated"}}` // Next request should use cache fallback ctx = suite.app.AcquireCtx(reqCtx) err = proxyTheRequest(ctx, cfg.Server.HostGraphQL) // Save response before releasing context fallbackResponseBody := "" if ctx.Response() != nil { fallbackResponseBody = string(ctx.Response().Body()) } suite.app.ReleaseCtx(ctx) // Verify request succeeded via cache fallback suite.Nil(err, "Request with open circuit should succeed with cache fallback") suite.Equal(`{"data":{"test":"original"}}`, fallbackResponseBody, "Response should match cached version, not updated server response") // Verify metrics were incremented newCacheHitCount := getMetricValue(libpack_monitoring.MetricsCacheHit) newFallbackSuccessCount := getMetricValue(libpack_monitoring.MetricsCircuitFallbackSuccess) suite.Greater(newCacheHitCount, metricCounts[libpack_monitoring.MetricsCacheHit], "Cache hit metric should be incremented") suite.Greater(newFallbackSuccessCount, metricCounts[libpack_monitoring.MetricsCircuitFallbackSuccess], "Circuit fallback success metric should be incremented") // Test Case 3: Request with different query missing in cache while circuit is open // Create new request with different query reqCtx = &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") newReqBody := `{"query": "query { differentQuery }"}` reqCtx.Request.SetBody([]byte(newReqBody)) // Capture metrics before request fallbackFailedBefore := getMetricValue(libpack_monitoring.MetricsCircuitFallbackFailed) // Request should fail as circuit is open and cache has no matching entry ctx = suite.app.AcquireCtx(reqCtx) err = proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) // Verify request failed with circuit open error suite.NotNil(err, "Request with open circuit and no cache should fail") suite.Equal(ErrCircuitOpen.Error(), err.Error(), "Error should be ErrCircuitOpen") // Verify metrics were incremented fallbackFailedAfter := getMetricValue(libpack_monitoring.MetricsCircuitFallbackFailed) suite.Greater(fallbackFailedAfter, fallbackFailedBefore, "Circuit fallback failed metric should be incremented") // Test Case 4: Circuit timeout and transition to half-open state t.Log("Waiting for circuit timeout to transition to half-open state...") // Wait for the circuit timeout plus a bit more time.Sleep(time.Duration(cfg.CircuitBreaker.Timeout+1) * time.Second) // Reset server to success again for when the circuit allows a probe request responseStatus = http.StatusOK responseBody = `{"data":{"test":"after recovery"}}` // The first request will transition circuit to half-open and probe the server // We don't need to check the actual response here, just that the circuit // has properly transitioned from open reqCtx = &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(reqBody)) ctx = suite.app.AcquireCtx(reqCtx) _ = proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) // Allow time for circuit state to fully update time.Sleep(100 * time.Millisecond) // Just verify circuit state changed - don't try to test the actual half-open behavior // as it's timing sensitive and can lead to flaky tests t.Logf("Final circuit state: %s", cb.State().String()) suite.NotEqual(gobreaker.StateOpen.String(), cb.State().String(), "Circuit should no longer be fully open after recovery") } // TestGzipHandlingAndCachingInteraction tests the interaction between // the gzip handling and caching system func (suite *Tests) TestGzipHandlingAndCachingInteraction() { // Original values to restore later originalCache := cfg.Cache originalClient := cfg.Client.FastProxyClient // Restore after test defer func() { cfg.Cache = originalCache cfg.Client.FastProxyClient = originalClient }() // Ensure cache is enabled cfg.Cache.CacheEnable = true cfg.Cache.CacheTTL = 60 // 60 seconds // Initialize monitoring - re-initialize from scratch for testing cfg.Monitoring = libpack_monitoring.NewMonitoring(&libpack_monitoring.InitConfig{}) // Initialize cache - must be done after initializing monitoring libpack_cache.EnableCache(&libpack_cache.CacheConfig{ Logger: cfg.Logger, TTL: cfg.Cache.CacheTTL, }) // Make sure old cache entries are cleared libpack_cache.CacheClear() // Create a test server that returns gzipped content server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // Set the Content-Encoding header to indicate gzipped content w.Header().Set("Content-Encoding", "gzip") // Create a gzipped response with query-specific data reqBody := make([]byte, r.ContentLength) _, _ = r.Body.Read(reqBody) var queryStr string if strings.Contains(string(reqBody), "query1") { queryStr = "query1" } else if strings.Contains(string(reqBody), "query2") { queryStr = "query2" } else { queryStr = "unknown" } payload := fmt.Sprintf(`{"data":{"test":"%s response"}}`, queryStr) gzipped := createGzippedData([]byte(payload)) // Send the gzipped data w.WriteHeader(http.StatusOK) _, _ = w.Write(gzipped) })) defer server.Close() // Configure client cfg.Client.ClientTimeout = 5 cfg.Client.FastProxyClient = createFasthttpClient(cfg) // Configure server URL cfg.Server.HostGraphQL = server.URL // Instead of using metrics, we'll manually track cache hits and misses cacheHits := 0 cacheMisses := 0 // First request - query1, should be a cache miss reqCtx1 := &fasthttp.RequestCtx{} reqCtx1.Request.SetRequestURI("/graphql") reqCtx1.Request.Header.SetMethod("POST") reqCtx1.Request.Header.Set("Content-Type", "application/json") reqCtx1.Request.SetBody([]byte(`{"query": "query { query1 }"}`)) ctx := suite.app.AcquireCtx(reqCtx1) err := proxyTheRequest(ctx, cfg.Server.HostGraphQL) // Save response data before releasing context firstResponseStatus := ctx.Response().StatusCode() firstResponseBody := string(ctx.Response().Body()) firstResponseHeaders := string(ctx.Response().Header.Peek("Content-Encoding")) suite.app.ReleaseCtx(ctx) // First request is a cache miss cacheMisses++ // Check response suite.Nil(err, "First request should succeed") suite.Equal(fiber.StatusOK, firstResponseStatus, "Status should be 200 OK") suite.Contains(firstResponseBody, "query1 response", "Response should contain uncompressed query1 content") // Content-Encoding header should be removed after decompression suite.Equal("", firstResponseHeaders, "Content-Encoding header should be removed") // Verify cache metrics - should have one miss, no hits yet suite.Equal(1, cacheMisses, "Should have one cache miss") suite.Equal(0, cacheHits, "Should have no cache hits yet") // Second request - repeat query1, should be a cache hit reqCtx2 := &fasthttp.RequestCtx{} reqCtx2.Request.SetRequestURI("/graphql") reqCtx2.Request.Header.SetMethod("POST") reqCtx2.Request.Header.Set("Content-Type", "application/json") reqCtx2.Request.SetBody([]byte(`{"query": "query { query1 }"}`)) ctx = suite.app.AcquireCtx(reqCtx2) err = proxyTheRequest(ctx, cfg.Server.HostGraphQL) // Save response data before releasing context secondResponseStatus := ctx.Response().StatusCode() secondResponseBody := string(ctx.Response().Body()) suite.app.ReleaseCtx(ctx) // Second request is a cache hit cacheHits++ suite.Nil(err, "Second request should succeed") suite.Equal(fiber.StatusOK, secondResponseStatus, "Status should be 200 OK") suite.Contains(secondResponseBody, "query1 response", "Response should contain correct content") // Verify cache metrics - should have one hit now suite.Equal(1, cacheHits, "Should have one cache hit") // Third request - different query, should be a cache miss reqCtx3 := &fasthttp.RequestCtx{} reqCtx3.Request.SetRequestURI("/graphql") reqCtx3.Request.Header.SetMethod("POST") reqCtx3.Request.Header.Set("Content-Type", "application/json") reqCtx3.Request.SetBody([]byte(`{"query": "query { query2 }"}`)) ctx = suite.app.AcquireCtx(reqCtx3) err = proxyTheRequest(ctx, cfg.Server.HostGraphQL) // Save response data before releasing context thirdResponseStatus := ctx.Response().StatusCode() thirdResponseBody := string(ctx.Response().Body()) suite.app.ReleaseCtx(ctx) // Third request is a cache miss cacheMisses++ suite.Nil(err, "Third request should succeed") suite.Equal(fiber.StatusOK, thirdResponseStatus, "Status should be 200 OK") suite.Contains(thirdResponseBody, "query2 response", "Response should contain query2 content") // Verify cache metrics - should have one hit and two misses suite.Equal(2, cacheMisses, "Should have two cache misses total") suite.Equal(1, cacheHits, "Should have one cache hit total") } // TestGraphQLQueryParsing tests GraphQL parsing with various query types func (suite *Tests) TestGraphQLQueryParsing() { testCases := []struct { name string query string expectEndpoint string expectParseErr bool expectReadOnly bool }{ { name: "simple_query", query: `{"query": "query { users { id name } }"}`, expectParseErr: false, expectReadOnly: true, }, { name: "mutation", query: `{"query": "mutation { createUser(name: \"Test\") { id } }"}`, expectParseErr: false, expectReadOnly: false, }, { name: "query_with_variables", query: `{"query": "query($id: ID!) { user(id: $id) { name } }", "variables": {"id": "123"}}`, expectParseErr: false, expectReadOnly: true, }, { name: "malformed_query", query: `{"query": "query { unclosed }"}`, expectParseErr: false, // Should handle malformed queries gracefully expectReadOnly: true, // Default to read-only for safety }, { name: "subscription", query: `{"query": "subscription { userUpdated { id name } }"}`, expectParseErr: false, expectReadOnly: true, // Subscriptions are read-only }, { name: "mixed_query_and_mutation", query: `{"query": "query { users { id } } mutation { createUser(name: \"Test\") { id } }"}`, expectParseErr: false, expectReadOnly: false, // Should detect mutation }, { name: "introspection_query", query: `{"query": "query { __schema { types { name } } }"}`, expectParseErr: false, expectReadOnly: true, // Introspection is read-only }, } // Setup test environment originalHost := cfg.Server.HostGraphQL originalHostRO := cfg.Server.HostGraphQLReadOnly defer func() { cfg.Server.HostGraphQL = originalHost cfg.Server.HostGraphQLReadOnly = originalHostRO }() // Set distinct endpoints for clear testing cfg.Server.HostGraphQL = "https://write.example.com" cfg.Server.HostGraphQLReadOnly = "https://read.example.com" for _, tc := range testCases { suite.Run(tc.name, func() { // Create request context reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(tc.query)) // Create fiber context ctx := suite.app.AcquireCtx(reqCtx) defer suite.app.ReleaseCtx(ctx) // Parse GraphQL query result := parseGraphQLQuery(ctx) // Verify parsing result if tc.expectParseErr { suite.True(result.shouldIgnore, "Should report parse error via shouldIgnore") } else { suite.False(result.shouldIgnore, "Should not report parse error via shouldIgnore") } if tc.expectReadOnly { suite.Equal(cfg.Server.HostGraphQLReadOnly, result.activeEndpoint, "Should use read-only endpoint") } else { suite.Equal(cfg.Server.HostGraphQL, result.activeEndpoint, "Should use write endpoint") } }) } } // Helper function to get current metric value func getMetricValue(metricName string) int { counter := cfg.Monitoring.RegisterMetricsCounter(metricName, nil) if counter == nil { return 0 } return int(counter.Get()) } // TestRequestCoalescingIntegration tests that request coalescing works end-to-end // through the proxy layer, ensuring concurrent identical requests result in only // one backend call while all clients receive the correct response. func (suite *Tests) TestRequestCoalescingIntegration() { // Save original config originalCoalescing := cfg.RequestCoalescing originalClient := cfg.Client.FastProxyClient originalHostGraphQL := cfg.Server.HostGraphQL // Restore after test defer func() { cfg.RequestCoalescing = originalCoalescing cfg.Client.FastProxyClient = originalClient cfg.Server.HostGraphQL = originalHostGraphQL }() // Track backend calls var backendCallCount atomic.Int32 var requestDelay = 100 * time.Millisecond // Create test server that counts requests and introduces delay server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { backendCallCount.Add(1) time.Sleep(requestDelay) // Delay to allow concurrent requests to coalesce w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(`{"data":{"users":[{"id":"1","name":"Test User"}]}}`)) })) defer server.Close() // Configure for test cfg.Server.HostGraphQL = server.URL cfg.Client.ClientTimeout = 5 cfg.Client.FastProxyClient = createFasthttpClient(cfg) cfg.RequestCoalescing.Enable = true // Initialize request coalescer for this test // Reset the global coalescer by creating a new one testCoalescer := NewRequestCoalescer(true, cfg.Logger, cfg.Monitoring) // Temporarily replace global coalescer originalCoalescer := requestCoalescer requestCoalescer = testCoalescer defer func() { requestCoalescer = originalCoalescer }() // Test Case 1: Concurrent identical requests should coalesce suite.Run("concurrent_identical_requests_coalesce", func() { backendCallCount.Store(0) testCoalescer.Reset() concurrentRequests := 10 var wg sync.WaitGroup wg.Add(concurrentRequests) responses := make([]string, concurrentRequests) errors := make([]error, concurrentRequests) // Launch concurrent requests with identical query for i := 0; i < concurrentRequests; i++ { go func(index int) { defer wg.Done() reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { users { id name } }"}`)) ctx := suite.app.AcquireCtx(reqCtx) err := proxyTheRequest(ctx, cfg.Server.HostGraphQL) errors[index] = err responses[index] = string(ctx.Response().Body()) suite.app.ReleaseCtx(ctx) }(i) } wg.Wait() // Verify only 1 backend call was made suite.Equal(int32(1), backendCallCount.Load(), "Should make only 1 backend call for %d concurrent identical requests", concurrentRequests) // Verify all requests succeeded with same response expectedResponse := `{"data":{"users":[{"id":"1","name":"Test User"}]}}` for i := 0; i < concurrentRequests; i++ { suite.Nil(errors[i], "Request %d should succeed", i) suite.Equal(expectedResponse, responses[i], "Request %d should have correct response", i) } // Verify coalescing stats stats := testCoalescer.GetStats() suite.Equal(int64(concurrentRequests), stats["total_requests"], "Total requests should match") suite.Equal(int64(1), stats["primary_requests"], "Should have 1 primary request") suite.Equal(int64(concurrentRequests-1), stats["coalesced_requests"], "Should have %d coalesced requests", concurrentRequests-1) }) // Test Case 2: Different queries should NOT coalesce suite.Run("different_queries_not_coalesced", func() { backendCallCount.Store(0) testCoalescer.Reset() // Create server that returns query-specific responses server2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { backendCallCount.Add(1) time.Sleep(50 * time.Millisecond) body := make([]byte, r.ContentLength) _, _ = r.Body.Read(body) var response string if strings.Contains(string(body), "query1") { response = `{"data":{"result":"query1"}}` } else if strings.Contains(string(body), "query2") { response = `{"data":{"result":"query2"}}` } else { response = `{"data":{"result":"unknown"}}` } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(response)) })) defer server2.Close() cfg.Server.HostGraphQL = server2.URL cfg.Client.FastProxyClient = createFasthttpClient(cfg) var wg sync.WaitGroup wg.Add(2) var response1, response2 string var err1, err2 error // Launch two requests with different queries concurrently go func() { defer wg.Done() reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { query1 }"}`)) ctx := suite.app.AcquireCtx(reqCtx) err1 = proxyTheRequest(ctx, cfg.Server.HostGraphQL) response1 = string(ctx.Response().Body()) suite.app.ReleaseCtx(ctx) }() go func() { defer wg.Done() reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { query2 }"}`)) ctx := suite.app.AcquireCtx(reqCtx) err2 = proxyTheRequest(ctx, cfg.Server.HostGraphQL) response2 = string(ctx.Response().Body()) suite.app.ReleaseCtx(ctx) }() wg.Wait() // Both requests should succeed suite.Nil(err1, "Query1 should succeed") suite.Nil(err2, "Query2 should succeed") // Should have made 2 backend calls (no coalescing for different queries) suite.Equal(int32(2), backendCallCount.Load(), "Should make 2 backend calls for 2 different queries") // Responses should be different suite.Contains(response1, "query1", "Response1 should be for query1") suite.Contains(response2, "query2", "Response2 should be for query2") }) // Test Case 3: Coalescing disabled should make separate calls suite.Run("coalescing_disabled", func() { // Create a fresh server for this test var disabledCallCount atomic.Int32 serverDisabled := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { disabledCallCount.Add(1) time.Sleep(50 * time.Millisecond) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(`{"data":{"users":[{"id":"1"}]}}`)) })) defer serverDisabled.Close() cfg.Server.HostGraphQL = serverDisabled.URL cfg.Client.FastProxyClient = createFasthttpClient(cfg) // Disable coalescing cfg.RequestCoalescing.Enable = false concurrentRequests := 5 var wg sync.WaitGroup wg.Add(concurrentRequests) // Launch concurrent identical requests for i := 0; i < concurrentRequests; i++ { go func() { defer wg.Done() reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { users { id } }"}`)) ctx := suite.app.AcquireCtx(reqCtx) _ = proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) }() } wg.Wait() // Should make separate backend calls when coalescing is disabled suite.Equal(int32(concurrentRequests), disabledCallCount.Load(), "Should make %d backend calls when coalescing is disabled", concurrentRequests) // Re-enable for subsequent tests cfg.RequestCoalescing.Enable = true }) // Test Case 4: Error responses should be shared correctly suite.Run("error_responses_coalesced", func() { backendCallCount.Store(0) testCoalescer.Reset() // Create server that returns errors serverError := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { backendCallCount.Add(1) time.Sleep(50 * time.Millisecond) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusInternalServerError) _, _ = w.Write([]byte(`{"errors":[{"message":"Internal server error"}]}`)) })) defer serverError.Close() cfg.Server.HostGraphQL = serverError.URL cfg.Client.FastProxyClient = createFasthttpClient(cfg) concurrentRequests := 5 var wg sync.WaitGroup wg.Add(concurrentRequests) errors := make([]error, concurrentRequests) for i := 0; i < concurrentRequests; i++ { go func(index int) { defer wg.Done() reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { fail }"}`)) ctx := suite.app.AcquireCtx(reqCtx) errors[index] = proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) }(i) } wg.Wait() // Should still only make 1 backend call suite.Equal(int32(1), backendCallCount.Load(), "Should make only 1 backend call even for error responses") // All requests should receive the same error for i := 0; i < concurrentRequests; i++ { suite.NotNil(errors[i], "Request %d should have error", i) } }) } // TestRetryBudgetIntegration tests that retry budget correctly limits retry attempts func (suite *Tests) TestRetryBudgetIntegration() { // Initialize a retry budget with limited tokens for testing budgetCtx := context.Background() testBudget := NewRetryBudgetWithContext(budgetCtx, RetryBudgetConfig{ MaxTokens: 3, // Only allow 3 retries total TokensPerSecond: 0, // Don't refill during test Enabled: true, }, cfg.Logger) // Replace global retry budget originalBudget := retryBudget retryBudget = testBudget defer func() { testBudget.Shutdown() retryBudget = originalBudget }() suite.Run("retry_budget_limits_retries", func() { testBudget.Reset() // Verify retry budget is set and works correctly rb := GetRetryBudget() suite.NotNil(rb, "Retry budget should be set") suite.True(rb.enabled, "Retry budget should be enabled") suite.T().Logf("Retry budget: enabled=%v, tokens=%d", rb.enabled, rb.currentTokens.Load()) // Test that AllowRetry consumes tokens correctly initialTokens := rb.currentTokens.Load() suite.Equal(int64(3), initialTokens, "Should start with 3 tokens") // First 3 retries should be allowed suite.True(rb.AllowRetry(), "First retry should be allowed") suite.True(rb.AllowRetry(), "Second retry should be allowed") suite.True(rb.AllowRetry(), "Third retry should be allowed") // Fourth retry should be denied (tokens exhausted) suite.False(rb.AllowRetry(), "Fourth retry should be denied - budget exhausted") // Verify stats stats := rb.GetStats() suite.Equal(int64(4), stats["total_attempts"].(int64), "Should have 4 total attempts") suite.Equal(int64(3), stats["allowed_retries"].(int64), "Should have 3 allowed retries") suite.Equal(int64(1), stats["denied_retries"].(int64), "Should have 1 denied retry") suite.T().Logf("Retry budget stats: total=%d, allowed=%d, denied=%d", stats["total_attempts"], stats["allowed_retries"], stats["denied_retries"]) }) suite.Run("retry_budget_exhaustion", func() { // Create a new budget with only 1 token testBudget.Shutdown() budgetCtx2 := context.Background() testBudget2 := NewRetryBudgetWithContext(budgetCtx2, RetryBudgetConfig{ MaxTokens: 1, // Only allow 1 retry TokensPerSecond: 0, // Don't refill Enabled: true, }, cfg.Logger) retryBudget = testBudget2 defer func() { testBudget2.Shutdown() }() // Test budget exhaustion with 1 token rb := GetRetryBudget() suite.NotNil(rb, "Retry budget should be set") suite.Equal(int64(1), rb.currentTokens.Load(), "Should start with 1 token") // First retry should be allowed suite.True(rb.AllowRetry(), "First retry should be allowed") // Second retry should be denied (only 1 token available) suite.False(rb.AllowRetry(), "Second retry should be denied - budget exhausted") // Verify stats stats := rb.GetStats() suite.Equal(int64(2), stats["total_attempts"].(int64), "Should have 2 total attempts") suite.Equal(int64(1), stats["allowed_retries"].(int64), "Should have 1 allowed retry") suite.Equal(int64(1), stats["denied_retries"].(int64), "Should have 1 denied retry") suite.T().Logf("Retry budget stats: total=%d, allowed=%d, denied=%d", stats["total_attempts"], stats["allowed_retries"], stats["denied_retries"]) }) } // TestConnectionPoolStatsIntegration tests that connection pool stats are tracked func (suite *Tests) TestConnectionPoolStatsIntegration() { // Save original config originalClient := cfg.Client.FastProxyClient originalHostGraphQL := cfg.Server.HostGraphQL originalCoalescing := cfg.RequestCoalescing.Enable // Restore after test defer func() { cfg.Client.FastProxyClient = originalClient cfg.Server.HostGraphQL = originalHostGraphQL cfg.RequestCoalescing.Enable = originalCoalescing }() // Disable request coalescing for accurate tracking cfg.RequestCoalescing.Enable = false suite.Run("connection_success_tracked", func() { // Create test server that succeeds server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(`{"data":{"test":"success"}}`)) })) defer server.Close() cfg.Server.HostGraphQL = server.URL cfg.Client.ClientTimeout = 5 cfg.Client.FastProxyClient = createFasthttpClient(cfg) // Initialize connection pool InitializeConnectionPool(cfg.Client.FastProxyClient) defer ShutdownConnectionPool() poolMgr := GetConnectionPoolManager() suite.NotNil(poolMgr, "Connection pool manager should be initialized") // Get stats before statsBefore := poolMgr.GetConnectionStats() successBefore := statsBefore["total_connections"].(int64) // Make a successful request reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { test }"}`)) ctx := suite.app.AcquireCtx(reqCtx) err := proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) suite.Nil(err, "Request should succeed") // Get stats after statsAfter := poolMgr.GetConnectionStats() successAfter := statsAfter["total_connections"].(int64) suite.Greater(successAfter, successBefore, "Total connections should increase after successful request") }) suite.Run("connection_failure_tracked_on_5xx", func() { // Create test server that returns 503 // Note: 503 triggers retry which records failures server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) _, _ = w.Write([]byte(`{"errors":[{"message":"Service unavailable"}]}`)) })) defer server.Close() cfg.Server.HostGraphQL = server.URL cfg.Client.ClientTimeout = 2 cfg.Client.FastProxyClient = createFasthttpClient(cfg) // Initialize connection pool InitializeConnectionPool(cfg.Client.FastProxyClient) defer ShutdownConnectionPool() poolMgr := GetConnectionPoolManager() suite.NotNil(poolMgr, "Connection pool manager should be initialized") // Get stats before statsBefore := poolMgr.GetConnectionStats() failuresBefore := statsBefore["connection_failures"].(int64) // Make a failing request (503 is retryable, so it will retry and track failures) reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { fail }"}`)) ctx := suite.app.AcquireCtx(reqCtx) _ = proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) // Get stats after - should have failures from retry attempts statsAfter := poolMgr.GetConnectionStats() failuresAfter := statsAfter["connection_failures"].(int64) suite.Greater(failuresAfter, failuresBefore, "Connection failures should increase after 5xx responses that trigger retries") suite.T().Logf("Connection failures: before=%d, after=%d", failuresBefore, failuresAfter) }) suite.Run("stats_reflect_request_outcomes", func() { // This test verifies that connection stats properly reflect the // combination of successes and failures over multiple requests // Start with a fresh server var requestCount atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { count := requestCount.Add(1) // First 2 requests succeed, rest fail if count <= 2 { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(`{"data":{"test":"success"}}`)) } else { w.WriteHeader(http.StatusInternalServerError) _, _ = w.Write([]byte(`{"errors":[{"message":"Error"}]}`)) } })) defer server.Close() cfg.Server.HostGraphQL = server.URL cfg.Client.ClientTimeout = 2 cfg.Client.FastProxyClient = createFasthttpClient(cfg) // Initialize connection pool InitializeConnectionPool(cfg.Client.FastProxyClient) defer ShutdownConnectionPool() poolMgr := GetConnectionPoolManager() suite.NotNil(poolMgr, "Connection pool manager should be initialized") // Make 2 successful requests for i := 0; i < 2; i++ { reqCtx := &fasthttp.RequestCtx{} reqCtx.Request.SetRequestURI("/graphql") reqCtx.Request.Header.SetMethod("POST") reqCtx.Request.Header.Set("Content-Type", "application/json") reqCtx.Request.SetBody([]byte(`{"query": "query { test }"}`)) ctx := suite.app.AcquireCtx(reqCtx) _ = proxyTheRequest(ctx, cfg.Server.HostGraphQL) suite.app.ReleaseCtx(ctx) } // Get stats after successes statsAfterSuccess := poolMgr.GetConnectionStats() totalConnections := statsAfterSuccess["total_connections"].(int64) suite.GreaterOrEqual(totalConnections, int64(2), "Should have at least 2 successful connections tracked") suite.T().Logf("Total connections after 2 successful requests: %d", totalConnections) }) }