Skip to content

feat: Implement DropOldest overflow policy with ring-buffer channel #271

Description

@m3ridian-zero

Summary

The DropOldest overflow policy was removed from 1.0 (485ad8b) because it could not be correctly implemented with tokio::sync::mpsc. The Sender has no access to the Receiver end, so the producer cannot evict the oldest item from the channel when the buffer is full.

Problem

For real-time market data subscriptions, slow consumers need a way to receive the freshest data rather than stale data from when they first fell behind. DropNewest (current default) keeps old data and discards new — exactly the wrong behavior for price-sensitive consumers.

Constraints

  • send_batch() is the hot path (~50K calls/sec at scale). Zero-overhead for DropNewest and Block policies must be preserved.
  • RecordBatch is Clone (cheap, Arc-backed columns).
  • BlpError is NOT Clone (contains Box<dyn Error>).
  • Multiple producers (one per topic) share a single consumer channel.

Rejected Approaches

Approach Why it fails
VecDeque alongside mpsc Evicts middle-age data, not oldest. The 256-item mpsc buffer is untouched.
Consumer-side drain Fresh batches rejected by try_send are already lost. Draining gives old data.
tokio::sync::broadcast BlpError is not Clone. Broadcast also clones on every recv, not just overflow.

Recommended Approach

Replace mpsc with a custom bounded ring channel (or use a crate like ringbuf / async-ringbuf) when DropOldest is requested:

  • Producer writes to a ring buffer. When full, oldest item is overwritten.
  • Consumer reads from the ring buffer.
  • DropNewest and Block continue using mpsc (zero change to existing paths).

The channel type selection can be done at subscription construction time based on the policy enum, behind a trait abstraction.

Acceptance Criteria

  • OverflowPolicy::DropOldest variant restored
  • Under sustained backpressure, consumer receives the N most recent batches (not the N oldest)
  • Zero overhead on DropNewest and Block code paths
  • Benchmark showing no regression on subscription throughput for default policy
  • Integration test: produce 2x capacity, verify consumer sees newest half

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or request

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions