Using RoaringBitmap for Message Broker Consumer State in Rust

Updated at July 2026

Imagine we are building a small message broker.

Apps publish messages to topics, and consumers subscribe to those topics. Each consumer pulls messages independently and has its own progress.

The topic log stores messages once:

0 -> message A
1 -> message B
2 -> message C
3 -> message D
4 -> message E

But each consumer has its own view of what still needs to be processed.

For example, email-worker may still need:

0, 1, 2, 3, 4

while analytics-worker may only need:

2, 3, 4

So pending delivery state should not live on the message itself.

The topic owns the messages.

The consumer owns the pending state.

That pending state is basically a set of message indexes:

pending = messages this consumer still needs to process

This is where RoaringBitmap is useful.

The Rust roaring crate provides RoaringBitmap, a compressed set of u32 integers with operations such as insert, remove, contains, min, iter, and serialized_size.

For a message broker, this fits naturally:

pending message indexes = set of u32

Basic message model

We start with a simple message type:

#[derive(Debug, Clone)]
struct Message {
    payload: String,
}

#[derive(Debug, Clone)]
struct PulledMessage {
    index: u32,
    payload: String,
}

The topic stores messages in append-only order:

#[derive(Default)]
struct Topic {
    messages: Vec<Message>,
}

A message does not know which consumers have processed it.

That information belongs to each consumer.

1. Naive Vec implementation

The simplest implementation is a Vec<u32> containing pending message indexes.

#[derive(Default)]
struct ConsumerVecState {
    pending: Vec<u32>,
}

impl Topic {
    fn publish_for_vec(&mut self, consumer: &mut ConsumerVecState, payload: String) {
        let index = self.messages.len() as u32;

        self.messages.push(Message { payload });
        consumer.pending.push(index);
    }
}

impl ConsumerVecState {
    fn pull_next(&mut self, topic: &Topic) -> Option<PulledMessage> {
        let next_index = *self.pending.iter().min()?;

        self.pending.retain(|index| *index != next_index);

        let message = topic.messages.get(next_index as usize)?;

        Some(PulledMessage {
            index: next_index,
            payload: message.payload.clone(),
        })
    }

    fn ack(&mut self, _index: u32) {
        // The message was already removed from pending when pulled.
    }

    fn nack(&mut self, index: u32) {
        if !self.pending.contains(&index) {
            self.pending.push(index);
        }
    }
}

Example usage:

fn main() {
    let mut topic = Topic::default();
    let mut consumer = ConsumerVecState::default();

    topic.publish_for_vec(&mut consumer, "send email".to_string());
    topic.publish_for_vec(&mut consumer, "resize image".to_string());

    let message = consumer.pull_next(&topic).unwrap();

    println!("Pulled message: {:?}", message);

    consumer.ack(message.index);
}

This works, but the hot path is expensive.

Finding the next message scans the whole list:

self.pending.iter().min()

Removing the message also scans the list:

self.pending.retain(...)

So the performance shape looks like this:

0
1
2

Find next pending message: O(n)
Remove pending message:    O(n)
Reinsert on nack:          O(n), because of contains

For a small queue, that is fine.

For many pending messages per consumer, it becomes expensive.

2. HashSet implementation

A HashSet<u32> improves insertion, removal, and membership checks.

use std::collections::HashSet;

#[derive(Default)]
struct ConsumerHashSetState {
    pending: HashSet<u32>,
}

impl Topic {
    fn publish_for_hashset(&mut self, consumer: &mut ConsumerHashSetState, payload: String) {
        let index = self.messages.len() as u32;

        self.messages.push(Message { payload });
        consumer.pending.insert(index);
    }
}

impl ConsumerHashSetState {
    fn pull_next(&mut self, topic: &Topic) -> Option<PulledMessage> {
        let next_index = *self.pending.iter().min()?;

        self.pending.remove(&next_index);

        let message = topic.messages.get(next_index as usize)?;

        Some(PulledMessage {
            index: next_index,
            payload: message.payload.clone(),
        })
    }

    fn ack(&mut self, _index: u32) {
        // The message was already removed from pending when pulled.
    }

    fn nack(&mut self, index: u32) {
        self.pending.insert(index);
    }
}

This improves several operations:

0
1
2

Remove pending message: average O(1)
Reinsert on nack:       average O(1)
Check membership:       average O(1)

But FIFO pulling is still awkward:

self.pending.iter().min()

A HashSet is unordered.

So finding the smallest pending index still requires scanning the whole set:

Find next pending message: O(n)

That is the main problem.

For a broker, pull_next() is the hot path. We do not want every pull to scan all pending messages.

3. RoaringBitmap implementation

Now we store pending message indexes in a RoaringBitmap.

0
1

[dependencies]
roaring = "0.11"

A RoaringBitmap is a compressed set of u32 integers.

That is a good fit for message indexes because indexes are already integers and are often dense or range-like.

use roaring::RoaringBitmap;

#[derive(Default)]
struct ConsumerRoaringState {
    pending: RoaringBitmap,
}

impl Topic {
    fn publish_for_roaring(&mut self, consumer: &mut ConsumerRoaringState, payload: String) {
        let index = self.messages.len() as u32;

        self.messages.push(Message { payload });
        consumer.pending.insert(index);
    }
}

impl ConsumerRoaringState {
    fn pull_next_fifo(&mut self, topic: &Topic) -> Option<PulledMessage> {
        let next_index = self.pending.min()?;

        self.pending.remove(next_index);

        let message = topic.messages.get(next_index as usize)?;

        Some(PulledMessage {
            index: next_index,
            payload: message.payload.clone(),
        })
    }

    fn pull_any(&mut self, topic: &Topic) -> Option<PulledMessage> {
        let next_index = self.pending.iter().next()?;

        self.pending.remove(next_index);

        let message = topic.messages.get(next_index as usize)?;

        Some(PulledMessage {
            index: next_index,
            payload: message.payload.clone(),
        })
    }

    fn ack(&mut self, _index: u32) {
        // The message was already removed from pending when pulled.
    }

    fn nack(&mut self, index: u32) {
        self.pending.insert(index);
    }
}

The important operation is this:

let next_index = self.pending.min()?;

That asks:

What is the smallest pending message index for this consumer?

Then this:

self.pending.remove(next_index);

means:

The consumer has pulled this message, so it is no longer pending.

If the consumer acks the message, nothing else has to happen to the pending set:

0
1
2

fn ack(&mut self, _index: u32) {
    // Already removed from pending.
}

If the consumer nacks the message, it goes back into pending:

self.pending.insert(index);

That gives us a simple lifecycle:

publish -> insert into pending
pull    -> remove from pending
ack     -> leave removed
nack    -> insert back into pending

Full runnable example

Cargo.toml:

[package]
name = "roaring-message-broker-demo"
version = "0.1.0"
edition = "2021"

[dependencies]
roaring = "0.11"

src/main.rs:

use roaring::RoaringBitmap;

#[derive(Debug, Clone)]
struct Message {
    payload: String,
}

#[derive(Debug, Clone)]
struct PulledMessage {
    index: u32,
    payload: String,
}

#[derive(Default)]
struct Topic {
    messages: Vec<Message>,
}

#[derive(Default)]
struct ConsumerState {
    pending: RoaringBitmap,
}

impl Topic {
    fn publish(&mut self, consumer: &mut ConsumerState, payload: String) {
        let index = self.messages.len() as u32;

        self.messages.push(Message { payload });
        consumer.pending.insert(index);
    }
}

impl ConsumerState {
    fn pull_next_fifo(&mut self, topic: &Topic) -> Option<PulledMessage> {
        let index = self.pending.min()?;

        self.pending.remove(index);

        let message = topic.messages.get(index as usize)?;

        Some(PulledMessage {
            index,
            payload: message.payload.clone(),
        })
    }

    fn ack(&mut self, _index: u32) {
        // Already removed from pending.
    }

    fn nack(&mut self, index: u32) {
        self.pending.insert(index);
    }

    fn pending_count(&self) -> u64 {
        self.pending.len()
    }
}

fn main() {
    let mut topic = Topic::default();
    let mut consumer = ConsumerState::default();

    topic.publish(&mut consumer, "send email".to_string());
    topic.publish(&mut consumer, "resize image".to_string());
    topic.publish(&mut consumer, "generate invoice".to_string());

    let first = consumer.pull_next_fifo(&topic).unwrap();

    println!("Pulled first: {:?}", first);

    consumer.ack(first.index);

    let second = consumer.pull_next_fifo(&topic).unwrap();

    println!("Pulled second: {:?}", second);

    consumer.nack(second.index);

    let redelivered = consumer.pull_next_fifo(&topic).unwrap();

    println!("Redelivered: {:?}", redelivered);

    consumer.ack(redelivered.index);

    println!("Pending messages: {}", consumer.pending_count());
}

Example output:

Pulled first: PulledMessage { index: 0, payload: "send email" }
Pulled second: PulledMessage { index: 1, payload: "resize image" }
Redelivered: PulledMessage { index: 1, payload: "resize image" }
Pending messages: 1

Why this is faster than scanning the log

Without a pending set, the consumer might need to scan the whole topic log:

for index in 0..topic.messages.len() {
    if !acked.contains(index) {
        return Some(index);
    }
}

That becomes expensive because every pull may walk through old messages.

With a pending set, the consumer state already knows what is still available:

let index = pending.min()?;

So the broker does not ask:

Which message in the entire log has not been consumed yet?

It asks:

What is the next item in this consumer's pending set?

That is the key optimization.

The broker avoids repeatedly scanning irrelevant old messages.

Performance comparison

Let:

0
1

n = number of pending indexes
c = number of Roaring containers

A Roaring container groups part of the integer space. For u32 values, the high bits select the container and the low bits select the value inside that container.

Structure	Pull FIFO	Ack for pending state	Nack / redelivery	Main issue
`Vec<u32>`	`O(n)`	`O(1)`	`O(n)`	Simple, but scans often
`HashSet<u32>`	`O(n)`	`O(1)`	average `O(1)`	Fast lookup, but unordered
`RoaringBitmap`	`O(1)` for `min()` + remove cost	`O(1)`	`O(log c + container cost)`	Best fit for integer index sets

For RoaringBitmap, the operation is not compared against all pending messages.

It works through compressed integer containers.

For this use case, that matters because pulling the next message is usually:

0
1

let index = pending.min()?;
pending.remove(index);

The min() part does not scan all pending indexes.

The remove(index) part finds the relevant container and removes the value inside it.

So a more precise way to think about the pull cost is:

0
1

pull_next_fifo = min() + remove(index)
               = O(1) + O(log c + container cost)

This is usually much better than scanning n pending messages.

Space complexity

Time complexity is only half of the story.

For a broker, memory matters too because every consumer may have its own pending-message state.

Imagine:

0
1
2

10 topics
100 consumers per topic
1,000,000 pending indexes per consumer

That is a lot of per-consumer state.

Space complexity comparison

Structure	Space complexity	Practical memory behavior
`Vec<u32>`	`O(n)`	Compact per value, but slow operations
`HashSet<u32>`	`O(n)`	Higher overhead due to hash table storage
`RoaringBitmap`	`O(c + compressed data)`	Often compact for dense integer indexes

A Vec<u32> stores every pending index directly:

[0, 1, 2, 3, 4, 5, ...]

That is memory-efficient, but operations like min, remove, and contains require scanning.

A HashSet<u32> also stores every pending index, but with hash table overhead. It improves lookup and removal, but usually uses more memory per value.

A RoaringBitmap stores integer sets in compressed containers.

That is especially useful when indexes are dense or range-like, for example:

0, 1, 2, 3, 4, 5, 6, 7, ...

or:

0
1
2

0..100000
120000..150000
200000..210000

Message indexes often have exactly this shape because they are generated sequentially.

That makes Roaring a strong fit for consumer pending state.

Measuring serialized size

The roaring crate exposes serialized_size().

This does not measure exact live heap memory, but it is useful for estimating how compact the bitmap is when stored on disk or sent over the network.

Example:

use roaring::RoaringBitmap;
use std::collections::HashSet;
use std::mem::size_of;

fn main() {
    let pending_indexes: Vec<u32> = (0..1_000_000).collect();

    let vec_pending = pending_indexes.clone();

    let hashset_pending: HashSet<u32> = pending_indexes
        .iter()
        .copied()
        .collect();

    let roaring_pending: RoaringBitmap = pending_indexes
        .iter()
        .copied()
        .collect();

    let vec_bytes = vec_pending.capacity() * size_of::<u32>();

    println!("Vec approx bytes:       {vec_bytes}");
    println!("HashSet len:            {}", hashset_pending.len());
    println!("HashSet capacity:       {}", hashset_pending.capacity());
    println!("Roaring serialized:     {} bytes", roaring_pending.serialized_size());
}

Example output shape:

Vec approx bytes:       4000000
HashSet len:            1000000
HashSet capacity:       ...
Roaring serialized:     ...

The HashSet capacity is not a full memory measurement. It only shows the number of buckets the set can hold before resizing. Actual memory usage is higher because hash tables also need metadata.

The practical takeaway is:

Vec:
  low memory
  slow lookup/removal

HashSet:
  fast lookup/removal
  higher memory overhead
  no natural ordering

RoaringBitmap:
  compact for many integer sets
  ordered iteration
  good fit for sequential message indexes

What about in-flight messages?

This article only tracks pending because that is enough to explain the RoaringBitmap use case.

A real broker usually also needs to track pulled-but-not-acked messages.

For example, if a consumer pulls a message and then crashes, the broker needs to know that the message should eventually become available again.

That state is usually called something like:

in_flight
locked
leased
processing

But it does not have to be another RoaringBitmap.

In a real broker, it may be better represented as:

HashMap<u32, LockDeadline>

because in-flight messages usually need timeout metadata.

So the simplified model is:

0
1

pending:
  what still needs to be delivered

The real broker model may be:

pending:
  what is available to deliver

in_flight:
  what has been delivered but not acked yet

The RoaringBitmap optimization is mainly about pending.

When not to use RoaringBitmap

RoaringBitmap is not always the best choice.

For very small consumers with only a few pending messages, a Vec<u32> may be simpler and good enough.

If you only need membership checks and do not care about ordering, a HashSet<u32> may also be fine.

Roaring becomes more interesting when:

- you have many integer indexes
- indexes are often sequential or range-like
- you need ordered iteration
- you need the smallest pending index
- you want compact serialized state
- each consumer has its own pending set

That is why it fits this message broker use case well.

Final takeaway

The bitmap should not track what is inside the message.

It should track the consumer's pending work.

A useful mental model is:

Topic log:
  owns messages

Consumer state:
  pending bitmap = messages this consumer still needs

Publish:
  append message to topic log
  insert message index into pending

Pull:
  get the smallest pending index
  remove it from pending

Ack:
  keep it removed

Nack:
  insert it back into pending

This turns:

Scan the whole log to find work.

into:

Ask the consumer's pending set for the next index.

That is the core optimization.

RoaringBitmap works well here because message indexes are integer IDs, usually sequential, and naturally represented as a compact set.