enhancement(sinks): Add support for max_bytes for memory buffers

[graphcareful]

Summary

This PR adds support for memory buffers to be bound in terms of bytes allocated. This is an opt-in feature which is defaulted to false, meaning that the current implementation and its defaults will still be selected when not explicitly supplying a value for max_bytes.

At the core of this change is a new interface that allowes the selection of a different lock-free queue. This queue is at the center of the implementation of memory buffers. Today that queue is crossbeam_queue::ArrayQueue which is a fixed-sized lock-free data structure. This queue being fixed size is the reason that #8679 could not easily be implemented. The new interface allows to drop in a non-fixed sized queue. The crossbeam_queue::SegQueue was chosen, since it showed to be performant in initial testing and didn't require the inclusion of any new dependencies.

The main resource (queue) is already guarded by a semaphore. This semaphore currently bounds the queue by number of elements but there's no reason for why it couldn't guard against bytes allocated, therefore much of that existing code remains the same - which is positive as it is already battle tested and seems relatively stable as is.

Finally a new unit test was added and new benchmarks included in vector-buffers/benches.

Vector configuration

To any sink configuration try:

buffer:
  type: memory
  max_size: 123456

How did you test this PR?

Via the existing unit test and the developed benchmarks

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

• Closes: Support max_bytes for memory buffers #8679

Notes

• Please read our Vector contributor resources.
• Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
• Some CI checks run only after we manually approve them.
  • We recommend adding a pre-push hook, please see this template.
  • Alternatively, we recommend running the following locally before pushing to the remote branch:
    • cargo fmt --all
    • cargo clippy --workspace --all-targets -- -D warnings
    • cargo nextest run --workspace (alternatively, you can run cargo test --all)
• After a review is requested, please avoid force pushes to help us review incrementally.
  • Feel free to push as many commits as you want. They will be squashed into one before merging.
  • For example, you can run git merge origin master and git push.
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run cargo vdev build licenses to regenerate the license inventory and commit the changes (if any). More details here.

[Copilot]

Pull Request Overview

Adds support for byte-based limits for in-memory buffers by introducing a unified MemoryBufferSize enum and dynamically selecting between element-count and byte-size queues.

• Replace standalone max_events with a size object backed by MemoryBufferSize across configs and APIs
• Implement QueueImpl to choose between ArrayQueue (by events) and SegQueue (by bytes) using a semaphore guard
• Update all tests, examples, benchmarks, and documentation to use the new byte/event buffer sizing model

Reviewed Changes

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

Show a summary per file

File	Description
website/cue/reference/components/base/sinks.cue	Update CUE schema to use new size object with max_bytes & max_size
src/topology/test/backpressure.rs	Adapt backpressure tests to MemoryBufferSize
src/test_util/mock/sources/basic.rs	Use MemoryBufferSize in mock sources
src/source_sender/mod.rs	Initialize limited channel with MemoryBufferSize
lib/vector-buffers/src/variants/in_memory.rs	Refactor MemoryBuffer to accept MemoryBufferSize
lib/vector-buffers/src/topology/test_util.rs	Enhance Sample to account for heap-allocated bytes
lib/vector-buffers/src/topology/channel/limited_queue.rs	Add QueueImpl, SizeTerms, & dynamic queue selection
lib/vector-buffers/src/topology/builder.rs	Update topology builder to use MemoryBufferSize
lib/vector-buffers/src/test/variant.rs	Update variant tests to use MemoryBufferSize
lib/vector-buffers/src/lib.rs	Export MemoryBufferSize
lib/vector-buffers/src/config.rs	Implement serde de/serialization for MemoryBufferSize
lib/vector-buffers/examples/buffer_perf.rs	Adjust example to use new buffer size API
lib/vector-buffers/benches/sized_records.rs	Add benchmarks for byte-based buffers via a BoundBy helper
lib/vector-buffers/benches/common.rs	Extend Message to simulate heap allocation for size-based tests
changelog.d/8679_add_support_max_bytes_memory_buffers.feature.md	Add changelog entry for the new max_bytes feature

Comments suppressed due to low confidence (3)

lib/vector-buffers/src/config.rs:102

• [nitpick] The error message for invalid max_bytes is unclear and grammatically awkward; consider rephrasing to something like "max_bytes must fit within the platform's usize range" and include the actual bounds.

                            &"For memory buffers max_bytes expects an integer within the range of 268435488 and your architecture dependent usize",

lib/vector-buffers/src/config.rs:430

• Add a unit test in the config module to verify that a YAML or JSON config using max_bytes correctly deserializes into MemoryBufferSize::MaxSize under BufferType::Memory.

        for stage in self.stages() {

lib/vector-buffers/src/config.rs:207

• [nitpick] The variant name MaxSize is ambiguous; consider renaming it to MaxBytes to clearly indicate it represents a byte-based limit.

    MaxSize {

[bruceg]

A few comments around the memory buffer size enum variants, otherwise LGTM

[bruceg]

One issue with the sinks doc file, otherwise LGTM. I am approving it in advance of your resolution of that one, but please see if it can be fixed.

[pront]

A few comments, I still need to review the latter parts.

[pront]

Neat!