INFOPLAT 3099 chip ingress batching #1756

hendoxc · 2026-01-05T21:58:57Z

Adds chipingress batching client
Adds unit-tests

github-actions · 2026-01-05T21:59:07Z

👋 hendoxc, thanks for creating this pull request!

To help reviewers, please consider creating future PRs as drafts first. This allows you to self-review and make any final changes before notifying the team.

Once you're ready, you can mark it as "Ready for review" to request feedback. Thanks!

Copilot

Pull request overview

This PR implements a batching client for chipingress that collects CloudEvent messages and sends them in batches based on configurable size and timeout triggers. The implementation provides non-blocking message queuing with configurable concurrency control.

Key changes:

Implements a BatchClient with configurable batch size, timeout, buffer size, and concurrency limits
Provides asynchronous batch processing with proper context cancellation and shutdown handling
Includes comprehensive unit tests covering batch triggers, edge cases, and concurrent operations

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
pkg/chipingress/batch/client.go	Implements the core BatchClient with Start/Stop lifecycle, message queuing, and batch sending logic with configurable options
pkg/chipingress/batch/client_test.go	Provides comprehensive test coverage for client initialization, message queuing, batch sending, and various trigger scenarios

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

pkg/chipingress/batch/client.go

pkg/chipingress/batch/client_test.go

github-actions · 2026-01-05T21:59:56Z

✅ API Diff Results - No breaking changes

📄 View full apidiff report

FIxes test

4of9

approved with one small suggestion

4of9 · 2026-01-13T19:04:49Z

pkg/chipingress/batch/client.go

+					return
+				default:
+					if msg.callback != nil {
+						msg.callback(err)


Suggested change

msg.callback(err)

go msg.callback(err)

maybe each callback gets its own goroutine? this way no callback can block other callbacks

Copilot

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-01-14T23:58:31Z

pkg/chipingress/batch/client.go

+		go func() {
+			for _, msg := range messages {
+				select {
+				case <-ctx.Done():
+					return
+				case <-b.shutdownChan:
+					return
+				default:
+					if msg.callback != nil {
+						msg.callback(err)
+					}
+				}
+			}
+		}()


The nested goroutine for invoking callbacks creates a potential race condition. If ctx.Done() or b.shutdownChan are triggered after checking them in the select but before invoking the callback, some callbacks may not be invoked. Consider removing the nested goroutine and invoking callbacks directly in the parent goroutine, or ensuring all callbacks are invoked even during shutdown.

Suggested change

go func() {

for _, msg := range messages {

select {

case <-ctx.Done():

return

case <-b.shutdownChan:

return

default:

if msg.callback != nil {

msg.callback(err)

}

}

}

}()

for _, msg := range messages {

if msg.callback != nil {

msg.callback(err)

}

}

Copilot · 2026-01-14T23:58:31Z

pkg/chipingress/batch/client.go

+}
+
+func (b *Client) Stop() {
+	close(b.shutdownChan)


Calling Stop() multiple times will panic due to closing an already-closed channel. Add a sync.Once or check if the channel is already closed to make this method safe for concurrent calls.

@hendoxc ^^^

Copilot · 2026-01-14T23:58:32Z

pkg/chipingress/batch/client.go

+
+func WithCompressionType(compressionType string) Opt {
+	return func(c *Client) {
+		c.compressionType = compressionType


The compressionType field is set but never used in the client implementation. Either remove this field and option, or implement compression functionality in the sendBatch method.

Suggested change

c.compressionType = compressionType

// Compression is currently not implemented; this option is a no-op to

// avoid setting an unused field.

pkcll · 2026-01-15T00:06:23Z

pkg/chipingress/batch/client.go

+	callback func(error)
+}
+
+type Client struct {


Would be good to add metrics for batching logic here. Could be optional if otel MeterProvider is passed.

pkcll · 2026-01-15T00:17:08Z

pkg/chipingress/batch/client.go

+	callback func(error)
+}
+
+type Client struct {


Do we need retry mechanism here? Or that is expected to be configured on gRPC client level?

pkcll · 2026-01-15T00:25:00Z

pkg/chipingress/batch/client.go

+		client:             client,
+		batchSize:          1,
+		maxConcurrentSends: make(chan struct{}, 1),
+		messageBuffer:      make(chan *messageWithCallback, 1000),


Should buffer size be configurable?

pkcll · 2026-01-15T00:26:49Z

pkg/chipingress/batch/client.go

+	b.maxConcurrentSends <- struct{}{}
+
+	go func() {
+		defer func() { <-b.maxConcurrentSends }()


metric for this ?

pkcll · 2026-01-15T00:29:28Z

pkg/chipingress/batch/client.go

+	callback func(error)
+}
+
+type Client struct {


Do we need a backpressure mechanism, what should happen if ChIP Ingress is not available for X minutes?

pkcll · 2026-01-15T02:49:19Z

pkg/chipingress/batch/client.go

+			select {
+			case <-ctx.Done():
+				b.flush(batch)
+				close(b.shutdownChan)


Is closing shutdownChan needed here ?

pkcll · 2026-01-15T02:56:09Z

pkg/chipingress/batch/client.go

+				case <-b.shutdownChan:
+					return
+				default:
+					if msg.callback != nil {


Shall we add context to callback function so it can be canceled gracefully?

pkcll · 2026-01-15T02:57:19Z

pkg/chipingress/batch/client.go

+				select {
+				case <-ctx.Done():
+					return
+				case <-b.shutdownChan:


On shutdown or ctx cancelation should we attempt to execute callbacks and complete them with some timeout maybe ?

Maybe flush should take care of that ?

pkcll · 2026-01-15T03:00:36Z

pkg/chipingress/batch/client.go

+				if len(batch) >= b.batchSize {
+					batchToSend := batch
+					batch = make([]*messageWithCallback, 0, b.batchSize)
+					timer.Stop()


Suggested change

timer.Stop()

stopAndDrainTimer(timer)

pkcll · 2026-01-15T03:01:12Z

pkg/chipingress/batch/client.go

+				return
+			case msg := <-b.messageBuffer:
+				if len(batch) == 0 {
+					timer.Reset(b.batchInterval)


Suggested change

timer.Reset(b.batchInterval)

stopAndDrainTimer(timer)

timer.Reset(b.batchInterval)

pkcll · 2026-01-15T03:02:38Z

pkg/chipingress/batch/client.go

+	go func() {
+		batch := make([]*messageWithCallback, 0, b.batchSize)
+		timer := time.NewTimer(b.batchInterval)
+		timer.Stop()


Consider more safe flow: stop and drain timer channel

Suggested change

timer.Stop()

stopAndDrainTimer(timer)

func stopAndDrainTimer(timer *time.Timer) { if !timer.Stop() { select { case <-timer.C: default: } } }

Adds chipingress batching client

c50e461

Copilot AI review requested due to automatic review settings January 5, 2026 21:58

hendoxc requested a review from a team as a code owner January 5, 2026 21:58

product-security-plaid-production bot requested review from 4of9, engnke, fouadkada, kirqz23, pkcll and thomaska January 5, 2026 21:59

hendoxc temporarily deployed to integration January 5, 2026 21:59 — with GitHub Actions Inactive

Copilot AI reviewed Jan 5, 2026

View reviewed changes

pkg/chipingress/batch/client.go Outdated Show resolved Hide resolved

pkg/chipingress/batch/client_test.go Outdated Show resolved Hide resolved

hendoxc added 2 commits January 5, 2026 15:11

Adds unit-tests

92e5a61

FIxes test

Removes copying of slice

703498f

hendoxc force-pushed the INFOPLAT-3099-chip-ingress-batching branch from e705b9f to 703498f Compare January 5, 2026 22:12

hendoxc temporarily deployed to integration January 5, 2026 22:12 — with GitHub Actions Inactive

Adds zap logger

7ac8912

hendoxc had a problem deploying to integration January 5, 2026 22:21 — with GitHub Actions Failure

hendoxc temporarily deployed to integration January 5, 2026 22:21 — with GitHub Actions Inactive

Fixes mod file

24a80d6

hendoxc temporarily deployed to integration January 5, 2026 22:29 — with GitHub Actions Inactive

Merge branch 'main' into INFOPLAT-3099-chip-ingress-batching

a005109

hendoxc had a problem deploying to integration January 13, 2026 17:39 — with GitHub Actions Error

check for shutdown

6980605

hendoxc temporarily deployed to integration January 13, 2026 17:41 — with GitHub Actions Inactive

4of9 approved these changes Jan 13, 2026

View reviewed changes

Merge branch 'main' into INFOPLAT-3099-chip-ingress-batching

d476bab

hendoxc temporarily deployed to integration January 14, 2026 15:04 — with GitHub Actions Inactive

Merge branch 'main' into INFOPLAT-3099-chip-ingress-batching

5960cb9

hendoxc temporarily deployed to integration January 14, 2026 16:34 — with GitHub Actions Inactive

hendoxc had a problem deploying to integration January 14, 2026 16:34 — with GitHub Actions Failure

pkcll approved these changes Jan 14, 2026

View reviewed changes

pkcll requested a review from Copilot January 14, 2026 23:57

Copilot AI reviewed Jan 14, 2026

View reviewed changes

pkcll reviewed Jan 15, 2026

View reviewed changes

	c.compressionType = compressionType
	// Compression is currently not implemented; this option is a no-op to
	// avoid setting an unused field.

	timer.Reset(b.batchInterval)
	stopAndDrainTimer(timer)
	timer.Reset(b.batchInterval)

INFOPLAT 3099 chip ingress batching #1756

Are you sure you want to change the base?

INFOPLAT 3099 chip ingress batching #1756

Conversation

hendoxc commented Jan 5, 2026

Uh oh!

github-actions bot commented Jan 5, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Jan 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ API Diff Results - No breaking changes

Uh oh!

4of9 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

pkcll Jan 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pkcll Jan 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pkcll Jan 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

github-actions bot commented Jan 5, 2026 •

edited

Loading

pkcll Jan 15, 2026 •

edited

Loading

pkcll Jan 15, 2026 •

edited

Loading

pkcll Jan 15, 2026 •

edited

Loading