refactor: change batch collector to use blocking channel writes

Replace non-blocking channel writes with blocking behavior in BatchCollector.
This ensures tasks wait for available channel space rather than returning
errors when the channel is full, providing better backpressure handling.

- Remove ErrChannelFull error type
- Update SubmitBatch to accept context and block on channel writes
- Update tests to verify blocking behavior with timeouts
- Simplify error handling by removing channel full fallback logic

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: Savid

pkg/processor/transaction/structlog/batch_collector.go

@@ -2,7 +2,6 @@ package structlog
 
 import (
 	"context"
-	"fmt"
 	"runtime"
 	"sync"
 	"time"
@@ -76,23 +75,22 @@ func (bc *BatchCollector) Stop(ctx context.Context) error {
 }
 
 // SubmitBatch submits a task batch for processing
-func (bc *BatchCollector) SubmitBatch(taskBatch TaskBatch) error {
+func (bc *BatchCollector) SubmitBatch(ctx context.Context, taskBatch TaskBatch) error {
 	// Check shutdown first
 	select {
 	case <-bc.shutdown:
 		return context.Canceled
 	default:
 	}
 
-	// Then try to submit
+	// Block and wait for space in the channel
 	select {
 	case bc.taskChannel <- taskBatch:
 		return nil
 	case <-bc.shutdown:
 		return context.Canceled
-	default:
-		// Channel is full, caller should fallback to direct insert
-		return ErrChannelFull
+	case <-ctx.Done():
+		return ctx.Err()
 	}
 }
 
@@ -344,6 +342,3 @@ func (bc *BatchCollector) flushRemaining() {
 		bc.flushBatch(ctx)
 	}
 }
-
-// Custom error for when channel is full
-var ErrChannelFull = fmt.Errorf("batch collector channel is full")

pkg/processor/transaction/structlog/batch_collector_test.go

@@ -131,8 +131,8 @@ func TestBatchCollector_ProcessLargeTask(t *testing.T) {
 	assert.True(t, len(largeTask.Rows) > bc.maxBatchSize)
 }
 
-func TestBatchCollector_ChannelFull(t *testing.T) {
-	// Test behavior when channel is full
+func TestBatchCollector_ChannelBlocking(t *testing.T) {
+	// Test blocking behavior when channel is full
 	config := BatchConfig{
 		Enabled:           true,
 		MaxRows:           1000,
@@ -155,18 +155,23 @@ func TestBatchCollector_ChannelFull(t *testing.T) {
 			ResponseChan: make(chan error, 1),
 			TaskID:       "test-" + string(rune(i)),
 		}
-		err := bc.SubmitBatch(task)
+		err := bc.SubmitBatch(context.Background(), task)
 		assert.NoError(t, err)
 	}
 
-	// Next submit should fail with channel full
+	// Next submit should block, test with timeout
 	task := TaskBatch{
 		Rows:         createTestStructlogs(100),
 		ResponseChan: make(chan error, 1),
-		TaskID:       "test-full",
+		TaskID:       "test-blocking",
 	}
-	err := bc.SubmitBatch(task)
-	assert.Equal(t, ErrChannelFull, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	err := bc.SubmitBatch(ctx, task)
+	assert.Error(t, err)
+	assert.Equal(t, context.DeadlineExceeded, err)
 }
 
 func TestBatchCollector_StartStop(t *testing.T) {
@@ -196,14 +201,14 @@ func TestBatchCollector_StartStop(t *testing.T) {
 	}
 
 	// Can submit before shutdown
-	err := bc.SubmitBatch(task)
+	err := bc.SubmitBatch(context.Background(), task)
 	assert.NoError(t, err)
 
 	// Close shutdown channel
 	close(bc.shutdown)
 
 	// Should not accept new tasks after shutdown
-	err = bc.SubmitBatch(task)
+	err = bc.SubmitBatch(context.Background(), task)
 	assert.Equal(t, context.Canceled, err)
 }
 
@@ -352,7 +357,7 @@ func TestBatchCollector_ConcurrentAccess(t *testing.T) {
 				TaskID:       "concurrent-" + string(rune(id)),
 			}
 
-			if err := bc.SubmitBatch(task); err != nil && err != ErrChannelFull {
+			if err := bc.SubmitBatch(context.Background(), task); err != nil {
 				errors <- err
 			}
 			done <- true

pkg/processor/transaction/structlog/processor.go

@@ -250,7 +250,7 @@ func (p *Processor) sendToBatchCollector(ctx context.Context, structlogs []Struc
 	}()
 
 	// Submit to batch collector
-	if err := p.batchCollector.SubmitBatch(taskBatch); err != nil {
+	if err := p.batchCollector.SubmitBatch(ctx, taskBatch); err != nil {
 		// Let Asynq handle retries - don't fallback to direct insert
 		p.log.WithFields(logrus.Fields{
 			"task_id": taskBatch.TaskID,