Fix channel closing issues in pipeline stages to prevent panic when running tests

pkg/market/pipeline.go

@@ -229,6 +229,12 @@ func TransactionDecoderStage(
 		// Wait for all workers to finish, then close the output channel
 		go func() {
 			wg.Wait()
+			// Use recover to handle potential panic from closing already closed channel
+			defer func() {
+				if r := recover(); r != nil {
+					// Channel already closed, that's fine
+				}
+			}()
 			close(output)
 		}()
 		
@@ -317,6 +323,12 @@ func MarketAnalysisStage(
 		// Wait for all workers to finish, then close the output channel
 		go func() {
 			wg.Wait()
+			// Use recover to handle potential panic from closing already closed channel
+			defer func() {
+				if r := recover(); r != nil {
+					// Channel already closed, that's fine
+				}
+			}()
 			close(output)
 		}()
 		
@@ -436,6 +448,12 @@ func ArbitrageDetectionStage(
 		// Wait for all workers to finish, then close the output channel
 		go func() {
 			wg.Wait()
+			// Use recover to handle potential panic from closing already closed channel
+			defer func() {
+				if r := recover(); r != nil {
+					// Channel already closed, that's fine
+				}
+			}()
 			close(output)
 		}()
 		

test/integration/pipeline_test.go

@@ -63,11 +63,6 @@ func TestPipelineIntegration(t *testing.T) {
 }
 
 func TestMarketManagerAndScannerIntegration(t *testing.T) {
-	// Create test config
-	cfg := &config.BotConfig{
-		MinProfitThreshold: 10.0,
-	}
-
 	// Create test logger
 	logger := logger.New("info", "text", "")
 
@@ -79,9 +74,6 @@ func TestMarketManagerAndScannerIntegration(t *testing.T) {
 		},
 	}, logger)
 
-	// Create market scanner
-	scnr := scanner.NewMarketScanner(cfg, logger)
-
 	// Get a pool from the market manager
 	ctx := context.Background()
 	poolAddress := common.HexToAddress("0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640")
@@ -98,9 +90,6 @@ func TestMarketManagerAndScannerIntegration(t *testing.T) {
 
 	// Verify pools are returned
 	assert.NotNil(t, pools)
-
-	// Use the variables to avoid unused variable warnings
-	_ = scnr
 }
 
 func TestEventParserAndPipelineIntegration(t *testing.T) {