INFOPLAT 3099 chip ingress batching

pkg/chipingress/batch/buffer.go

@@ -0,0 +1,73 @@
+// Package batch provides a thread-safe batching client for chip ingress messages.
+package batch
+
+import (
+	"sync"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+)
+
+// messageBatch is a thread-safe buffer for accumulating messages before sending as a batch
+type messageBatch struct {
+	messages []*messageWithCallback
+	mu       sync.Mutex
+}
+
+type messageWithCallback struct {
+	event    *chipingress.CloudEventPb
+	callback func(error)
+}
+
+// newMessageBatch creates a new messageBatch with the given initial capacity
+func newMessageBatch(capacity int) *messageBatch {
+	return &messageBatch{
+		messages: make([]*messageWithCallback, 0, capacity),
+	}
+}
+
+// Add appends a message to the batch
+func (b *messageBatch) Add(msg *messageWithCallback) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.messages = append(b.messages, msg)
+}
+
+// Len returns the current number of messages in the batch
+func (b *messageBatch) Len() int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return len(b.messages)
+}
+
+// Clear removes all messages from the batch and returns a copy of them
+func (b *messageBatch) Clear() []*messageWithCallback {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if len(b.messages) == 0 {
+		return nil
+	}
+
+	// Make a copy
+	result := make([]*messageWithCallback, len(b.messages))
+	copy(result, b.messages)
+
+	// Reset the internal slice
+	b.messages = b.messages[:0]
+
+	return result
+}
+
+// Values returns a copy of the current messages without clearing them
+func (b *messageBatch) Values() []*messageWithCallback {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if len(b.messages) == 0 {
+		return nil
+	}
+
+	result := make([]*messageWithCallback, len(b.messages))
+	copy(result, b.messages)
+	return result
+}

pkg/chipingress/batch/buffer_test.go

@@ -0,0 +1,183 @@
+package batch
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+)
+
+func TestMessageBatch(t *testing.T) {
+	t.Run("newMessageBatch creates empty batch", func(t *testing.T) {
+		batch := newMessageBatch(10)
+		require.NotNil(t, batch)
+		assert.Equal(t, 0, batch.Len())
+	})
+
+	t.Run("Add appends messages", func(t *testing.T) {
+		batch := newMessageBatch(10)
+
+		msg1 := &messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-1"},
+		}
+		msg2 := &messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-2"},
+		}
+
+		batch.Add(msg1)
+		assert.Equal(t, 1, batch.Len())
+
+		batch.Add(msg2)
+		assert.Equal(t, 2, batch.Len())
+	})
+
+	t.Run("Clear returns copy and empties batch", func(t *testing.T) {
+		batch := newMessageBatch(10)
+
+		msg1 := &messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-1"},
+		}
+		msg2 := &messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-2"},
+		}
+
+		batch.Add(msg1)
+		batch.Add(msg2)
+
+		result := batch.Clear()
+		require.NotNil(t, result)
+		assert.Len(t, result, 2)
+		assert.Equal(t, "test-1", result[0].event.Id)
+		assert.Equal(t, "test-2", result[1].event.Id)
+
+		assert.Equal(t, 0, batch.Len())
+	})
+
+	t.Run("Clear on empty batch returns nil", func(t *testing.T) {
+		batch := newMessageBatch(10)
+		result := batch.Clear()
+		assert.Nil(t, result)
+	})
+
+	t.Run("Values returns copy without clearing", func(t *testing.T) {
+		batch := newMessageBatch(10)
+
+		msg1 := &messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-1"},
+		}
+		msg2 := &messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-2"},
+		}
+
+		batch.Add(msg1)
+		batch.Add(msg2)
+
+		result := batch.Values()
+		require.NotNil(t, result)
+		assert.Len(t, result, 2)
+		assert.Equal(t, "test-1", result[0].event.Id)
+		assert.Equal(t, "test-2", result[1].event.Id)
+
+		// Batch should still have messages after Values
+		assert.Equal(t, 2, batch.Len())
+	})
+
+	t.Run("Values on empty batch returns nil", func(t *testing.T) {
+		batch := newMessageBatch(10)
+		result := batch.Values()
+		assert.Nil(t, result)
+	})
+
+	t.Run("Values returns slice copy with same pointers", func(t *testing.T) {
+		batch := newMessageBatch(10)
+
+		msg1 := &messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-1"},
+		}
+
+		batch.Add(msg1)
+		result := batch.Values()
+		require.Len(t, result, 1)
+
+		assert.Equal(t, msg1, result[0])
+		assert.Same(t, msg1.event, result[0].event)
+
+		batch.Add(&messageWithCallback{
+			event: &chipingress.CloudEventPb{Id: "test-2"},
+		})
+		assert.Len(t, result, 1)
+		assert.Equal(t, 2, batch.Len())
+	})
+
+	t.Run("concurrent Add operations are safe", func(t *testing.T) {
+		batch := newMessageBatch(100)
+		var wg sync.WaitGroup
+		numGoroutines := 10
+		messagesPerGoroutine := 10
+
+		for range numGoroutines {
+			wg.Go(func() {
+				for range messagesPerGoroutine {
+					batch.Add(&messageWithCallback{
+						event: &chipingress.CloudEventPb{Id: "test"},
+					})
+				}
+			})
+		}
+
+		wg.Wait()
+
+		assert.Equal(t, numGoroutines*messagesPerGoroutine, batch.Len())
+	})
+
+	t.Run("concurrent Add and Clear operations are safe", func(t *testing.T) {
+		batch := newMessageBatch(100)
+		var wg sync.WaitGroup
+		numAdders := 5
+		numClears := 3
+
+		for range numAdders {
+			wg.Go(func() {
+				for range 20 {
+					batch.Add(&messageWithCallback{
+						event: &chipingress.CloudEventPb{Id: "test"},
+					})
+				}
+			})
+		}
+
+		// Start clearers
+		for range numClears {
+			wg.Go(func() {
+				for range 10 {
+					batch.Clear()
+				}
+			})
+		}
+
+		wg.Wait()
+		assert.GreaterOrEqual(t, batch.Len(), 0)
+	})
+
+	t.Run("concurrent Len operations are safe", func(t *testing.T) {
+		batch := newMessageBatch(100)
+		var wg sync.WaitGroup
+		for range 10 {
+			batch.Add(&messageWithCallback{
+				event: &chipingress.CloudEventPb{Id: "test"},
+			})
+		}
+
+		for range 100 {
+			wg.Go(func() {
+				length := batch.Len()
+				assert.GreaterOrEqual(t, length, 0)
+			})
+		}
+
+		wg.Wait()
+	})
+}