[DylanTinianov]

Description

Adds an opt-in byte budget to the chip ingress batch client. Bytes are reserved when a message is accepted and released after its batch callbacks run, or immediately if the enqueue fails.

WithMaxBufferedBytes(n) caps the total proto size of messages accepted by QueueMessage whose batch send and callbacks have not completed yet. The default of 0 means unlimited, so existing users are unchanged.

[github-actions]

✅ API Diff Results - github.com/smartcontractkit/chainlink-common/pkg/chipingress

✅ Compatible Changes (3)

batch (2)

• ErrBufferedBytesExceeded — ➕ Added

• WithMaxBufferedBytes — ➕ Added

batch.(*Client) (1)

• BufferedBytes — ➕ Added

---

📄 View full apidiff report

[pkcll]

Reviewed at f994eb04, and compared with #2436, which tries to do the same thing.

Verdict: approve with minor suggestions. None of these block merge. I'd take this over #2436.

What I checked:

• CI is green.
• Locally: go vet ./batch/ is clean, go test -race -count=3 ./batch/ passes, and the new tests pass 30× under -race.
• Every exit path releases its bytes: enqueue failure, the shutdown race, send error, an oversized split batch, the results path and the transaction path. The release is deferred inside the callback goroutine, so a panicking callback can't leak budget either.

Why I prefer this over #2436:

• Bytes are held until the batch's callbacks return, so in-flight and callback-pending messages count too. chipingress/batch: cap message buffer at 1 GiB live bytes #2436 frees bytes when a message leaves the channel, so batches waiting on the network or on callbacks fall outside its cap.
• It's opt-in and free when disabled. In a local benchmark with 1 KiB events, the limit-off path showed no measurable time change and no extra allocations. chipingress/batch: cap message buffer at 1 GiB live bytes #2436 costs about +25–65% per QueueMessage even with its cap set to 0, because it always computes the proto size and takes a mutex.
• Admission is a lock-free CAS that works with the existing admission lock.
• ErrBufferedBytesExceeded wraps ErrMessageBufferFull, so ErrorCodeFor and existing errors.Is callers work unchanged.

Worth porting from #2436

1. A buffered-bytes gauge. chipingress/batch: cap message buffer at 1 GiB live bytes #2436 adds chip_ingress.batch.buffer_live_bytes (unit By). Here, BufferedBytes() exists but nothing exports it, so once this is wired in nobody can see how close a node runs to the limit, or pick the value from data. Suggest an Int64ObservableGauge like chip_ingress.batch.buffered_bytes. Unlike chipingress/batch: cap message buffer at 1 GiB live bytes #2436:
   • register it on the client's meter, keep the metric.Registration, and call Unregister() in Stop. chipingress/batch: cap message buffer at 1 GiB live bytes #2436 registers on the global meter, ignores the error and never unregisters, so the closure keeps every stopped client alive.
   • handle or log the registration error.
2. The limit in config.info. chipingress/batch: cap message buffer at 1 GiB live bytes #2436 adds max_message_buffer_bytes to the config attributes. Adding attribute.Int64("max_buffered_bytes", c.maxBufferedBytes) next to max_grpc_request_size_bytes in recordConfig (around client.go:809 and :821) makes the configured value visible per client.
3. Two tests.
   • A concurrency test through QueueMessage itself, not only reserveBytes/releaseBytes (client_buffered_bytes_test.go:116). chipingress/batch: cap message buffer at 1 GiB live bytes #2436's "concurrent enqueue respects byte cap" has 20 goroutines queue against a small cap and asserts that some were dropped and the total stayed under the cap.
   • A config-metric assertion like chipingress/batch: cap message buffer at 1 GiB live bytes #2436's "records max_message_buffer_bytes in config metric", if you take (2).

Not worth porting: #2436's library-wide 1 GiB default, which changes behavior for every consumer (better set by each caller when wiring this in); its release at dequeue; its mutex; its separate error that doesn't wrap ErrMessageBufferFull; and the variadic onDequeue hook on the generic batcher.

Smaller suggestions

4. Check the budget before cloning. The reservation at client.go:330 runs after proto.Clone (:308), seqnum stamping and proto.Size. With the budget full, a rejected call cost about the same as an accepted one in the benchmark: roughly 2 µs, about 1.5 KiB and 11 allocations per 1 KiB event. That is exactly when memory is short. A cheap early exit would make shedding nearly free:

   if b.maxBufferedBytes > 0 && b.bufferedBytes.Load() >= b.maxBufferedBytes {
       return ErrBufferedBytesExceeded
   }

   The CAS in reserveBytes stays the authoritative check; this only skips work that is certain to be wasted. The rejection still takes a seqnum, which keeps the existing behavior that drops show up as detectable gaps.
5. Oversized messages (optional). A message larger than the whole budget (reserveBytes, :356) returns the same retryable ErrMessageBufferFull-wrapped error as a budget that is only temporarily full. A caller that retries on "buffer full" would retry it forever. A distinct error for the never-fits case, or at least a doc note, would make that clear.
6. Doc wording. The budget counts proto.Size of the event, not the message struct, the callback closure or map overhead, so real heap use is somewhat higher. Calling it a payload-bytes budget in the WithMaxBufferedBytes doc (:626) would set expectations.
7. Nit. The n > b.maxBufferedBytes early return in reserveBytes is already covered by the CAS loop's cur+n > max check. It's fine to keep for readability.

Wiring follow-up. Callbacks now hold budget until they return, so slow callbacks turn into producer backpressure. That's the right accounting, but worth a line in the WithMaxBufferedBytes doc so callers know to keep callbacks fast.