Introduction

I recently worked through the Gossip Glomers challenges, a series of distributed systems problems created by Fly.io and Kyle Kingsbury. As someone who has worked with distributed systems professionally, I was curious to see how these challenges would test my understanding of the fundamental concepts.

Spoiler alert: This post covers my approach to solving these challenges. If you prefer figuring them out yourself first, skip to What I Learned section.

Having prior experience with distributed systems helped me work through these challenges more easily. If you are new to distributed systems, expect to spend more time on the later challenges. I also found Martin Kleppmann's distributed systems lecture series helpful for understanding the underlying concepts.

Setup

The challenges use Maelstrom, a tool that lets us build and test distributed systems in any programming language. It simulates real network conditions like delays, network partitions, and failures, and verifies if our implementation works correctly through automated tests and visualizations.

Setting up Maelstrom was easy with their documentation.

I chose Rust for my implementation. As a Rust developer professionally, it was a natural choice for me. The challenges communicate via JSON messages over stdin/stdout, so I built a simple Maelstrom client myself using serde for serialization and tokio for async runtime support. This gave me full control over the implementation and helped me understand the protocol better.

Challenge #1: Echo

The first challenge is a warm-up to get familiar with the Maelstrom protocol. We need to implement a simple echo server that responds to messages with the same payload.

The implementation is straightforward: read a message from stdin, parse it, and send back a response with the echoed payload. This helped me establish the basic structure of my Maelstrom client and verify the setup was working correctly.

Challenge #2: Unique ID Generation

The second challenge requires implementing a globally-unique ID generation system.

My solution uses a monotonically increasing counter. Each node maintains its own counter, and the final ID combines the node_id with the counter value. Since node IDs are unique, this guarantees unique IDs across the cluster.

struct UniqueIdsApp {
    counter: AtomicU64,
}

impl UniqueIdsApp {
    fn generate_id(&self, node_id: &str) -> String {
        let counter = self.counter.fetch_add(1, Ordering::Relaxed);
        format!("{}-{}", node_id, counter)
    }
}

This approach is simple and efficient. No network communication is needed to generate IDs, making it highly available.

Other common approaches include UUID (random unique identifiers), Snowflake ID (Twitter's timestamp-based system), and centralized sequence generators. Each has its own tradeoffs in terms of performance, coordination, and guarantees.

Challenge #3: Broadcast

This challenge was to implement a broadcast system that gossips messages between all nodes in the cluster. I explored two different approaches - Immediate Broadcast and Periodic Batch Broadcast.

Immediate Broadcast

In this approach, messages are sent to all neighbors immediately upon receipt. Each node keeps track of messages it has already seen and forwards new messages to its neighbors.

struct BroadcastApp {
    neighbours: OnceCell<Vec<String>>,
    messages: Mutex<HashSet<i64>>,
}

impl BroadcastApp {
    async fn handle_broadcast(&self, request: &Request, message: i64) {
        let mut data = self.messages.lock().await;
        if !data.contains(&message) {
            data.insert(message);
            drop(data); // release the lock

            let neighbours = self.neighbours.get().unwrap().to_owned();
            let body = MessageBody::with_type(MessageType::Broadcast { message });

            // broadcast message to all neighbours except src
            for neighbour in neighbours {
                if neighbour.ne(&request.src) {
                    maelstrom.rpc(neighbour, body.clone());
                }
            }
        }
    }
}

This ensures low latency since messages spread through the network quickly. However, it creates a lot of network traffic.

Periodic Batch Broadcast

The second approach collects messages and periodically sends them to neighbours in batches. Instead of sending individual messages, nodes exchange their entire message sets every 500ms.

struct BroadcastApp {
    neighbours: OnceCell<Vec<String>>,
    // holds all messages the app received through broadcast
    messages: Mutex<HashSet<i64>>,
    // holds pending messages that need to be broadcasted
    neighbours_meta: OnceCell<HashMap<String, Mutex<HashSet<i64>>>>,
}

async fn gossip_broadcast(maelstrom: Arc<Maelstrom>, app: Arc<BroadcastApp>) {
    loop {
        std::thread::sleep(std::time::Duration::from_millis(500));
        let neighbours_meta = app.neighbours_meta.get().unwrap();

        // get pending messages that need to be broadcasted to each neighbour
        for (dest, meta) in neighbours_meta.iter() {
            let mut data = meta.lock().await; // acquire lock to access data
            let messages = data.clone();
            data.clear(); // clear data to hold only pending messages
            drop(data); // release lock

            // broadcast messages if not empty
            if !messages.is_empty() {
                let body = MessageBody::with_type(MessageType::BroadcastMany { messages });
                maelstrom.rpc(dest.to_owned(), body);
            }
        }
    }
}

This approach uses less bandwidth by batching messages together, but messages take longer to spread across the network. I experimented with different intervals and found 500ms to be a good balance between latency and bandwidth for these tests.

Challenge #4: Grow-Only Counter

The fourth challenge was to implement a stateless, grow-only counter using a sequentially-consistent key/value store service provided by Maelstrom.

This challenge introduced me to CRDTs (Conflict-free Replicated Data Types). These are data structures designed to be replicated across multiple nodes without conflicts.

Stateless Approach using seq-kv

In this approach, each node stores its counter value in Maelstrom's seq-kv service. When a node receives an add request, it reads its current value, adds the delta, and writes it back. For read requests, nodes fetch all counter values and sum them up.

async fn read(&self) -> io::Result<i64> {
    // read and add counter values of all nodes
    let mut sum = 0;
    for node_id in maelstrom.node_ids() {
        sum += self
            .seq_kv_read(node_id)
            .await?
            .as_int()
            .unwrap_or_default();
    }
    Ok(sum)
}

async fn add(&self) -> io::Result<()> {
    let current_value = self
        .seq_kv_read(maelstrom.node_id())
        .await?
        .as_int()
        .unwrap_or_default();

    self.seq_kv_write(maelstrom.node_id(), Value::Int(current_value + *delta))
        .await;
    Ok(())
}

While simple to implement, this approach relies on an external service. I wanted to try a stateful approach that better demonstrates CRDT principles.

Stateful Approach

Each node maintains separate counters for all nodes in the network. When a node receives an add request, it updates its own counter and broadcasts the update to other nodes. When nodes receive broadcasts, they take the maximum value between the received and stored counter. On read requests, nodes sum all counter values.

struct GrowOnlyCounterApp {
    counters: OnceCell<HashMap<String, AtomicI64>>,
}

impl GrowOnlyCounterApp {
    fn read(&self) -> i64 {
        self.counters.values().map(|a| a.load(Ordering::Relaxed)).sum()
    }

    fn add(&mut self, delta: i64) {
        // update counter of the current node
        let counter = self.counters.get(maelstrom.node_id()).unwrap();
        let new_value = counter.fetch_add(*delta, Ordering::Relaxed) + delta;

        // broadcast current node value to other nodes in the network
        let body = MessageBody::with_type(MessageType::Broadcast { new_value });
        for dest in maelstrom.node_ids() {
            if dest.ne(maelstrom.node_id()) {
                maelstrom.send(dest.to_owned(), body.clone());
            }
        }
    }

    fn handle_broadcast(&mut self, request: Request, message: i64) {
        // update counter of the node which sent this broadcast
        self.counters
            .get(&request.src)
            .unwrap()
            .fetch_max(message, Ordering::Relaxed);
    }
}

By taking the maximum value when updating counters, all nodes eventually agree on the same numbers. This is the key idea behind CRDTs. Nodes can receive updates in any order and still converge to the same result without any coordination.

Challenge #5: Kafka-Style Log

The next challenge was to implement a replicated log service similar to Kafka using the linearizable key/value store provided by Maelstrom.

I used distributed locking for write operations to maintain log ordering. Read operations don't need locks, which keeps them fast.

async fn distributed_lock(acquire_lock: bool) -> io::Result<()> {
    // The `CAS`(Compare-And-Swap) message type changes the value for a key using `from` and `to` fields
    let (from, to) = if acquire_lock {
        ("", maelstrom.node_id())
    } else {
        (maelstrom.node_id(), "")
    };

    loop {
        let body = MessageBody::with_type(MessageType::Cas {
            key: "lock", from, to, create_if_not_exists: true,
        });
        let response = maelstrom.rpc("lin-kv".to_owned(), body).await?;
        match response.body.msg_type {
            MessageType::CasOk => break,
            _ => {}
        };
    }

    Ok(())
}

async fn send() -> io::Result<()> {
    // acquire distributed lock
    self.distributed_lock(true).await?;

    // read data for key from lin-kv, append new msg to key and write back to lin-kv store
    // offset will be index of new msg in the list
    let mut data = self
        .lin_kv_read(key)
        .await?
        .as_vec()
        .unwrap_or_default();
    let offset = data.len() as i64;
    data.push(*msg as i64);
    self.lin_kv_write(key, data).await?;

    // release distributed lock
    self.distributed_lock(false).await
}

async fn poll(offsets: HashMap<String, i64>) -> io::Result<HashMap<String, Vec<[i64; 2]>>> {
    let mut msgs = HashMap::new();

    // read data for each key from lin-kv store and convert the data to required format
    for (key, offset) in offsets {
        if let Some(data) = self.lin_kv_read(key).await?.as_vec() {
            let data: Vec<[i64; 2]> = data
                .into_iter()
                .enumerate()
                .filter(|(idx, _)| *idx as i64 >= *offset)
                .map(|(idx, value)| [idx as i64, value])
                .collect();

            msgs.insert(key, data);
        }
    }
    Ok(msgs)
}

async fn commit_offsets(offsets: HashMap<String, i64>) -> io::Result<()> {
    // acquire distributed lock
    self.distributed_lock(true).await?;

    // read committed offset for each key from lin-kv and update if the new offset is greater
    for (key, offset) in offsets {
        let key = format!("{key}-committed");
        let last_committed_offset =
            self.lin_kv_read(&key).await?.as_int().unwrap_or(-1);

        if last_committed_offset < offset {
            self.lin_kv_write(key, offset)
                .await?;
        }
    }

    // release distributed lock
    self.distributed_lock(false).await
}

async fn list_committed_offsets(keys: Vec<String>) -> io::Result<HashMap<String, i64>> {
    let mut offsets = HashMap::new();

    // read committed offset for each key from lin-kv store
    for key in keys {
        let key = format!("{key}-committed");
        if let Some(offset) = self.lin_kv_read(&key).await?.as_int() {
            offsets.insert(key, offset);
        }
    }
    Ok(offsets)
}

The code uses Compare-And-Swap (CAS) operation to implement locking. When appending a message, a node locks, reads the message list, adds the new message, writes it back, and unlocks. Reading messages (polling) needs no locks since it just fetches data. Committing offsets uses locks to safely update the last processed offset for each key.

Challenge #6: Totally-Available Transactions

The final challenge was to implement a key-value store that handles transactions containing read and write operations, supporting weak consistency while also being totally available.

Built on Maelstrom's lin-kv service, I used distributed locking to ensure transaction integrity. Each transaction acquires the lock before performing any operations and releases it after completing all operations in the transaction.

async fn transaction_handler(mut txn: Vec<Transaction>) -> io::Result<Vec<Transaction>> {
    // acquire distributed lock
    distributed_lock(true).await?;

    for t in txn.iter_mut() {
        match t {
            Transaction::Read { key, val } => {
                let body = MessageBody::with_type(MessageType::Read {
                    key: Some(key.to_string()),
                });
                let response = maelstrom.rpc("lin-kv".to_owned(), body).await?;

                let value = match response.body.msg_type {
                    MessageType::ReadOk { value, .. } => value.unwrap(),
                    _ => Value::None,
                };

                *val = value;
            }
            Transaction::Write { key, value } => {
                let body = MessageBody::with_type(MessageType::Write {
                    key: key.to_string(),
                    value: Value::Int(*value),
                });
                maelstrom.rpc("lin-kv".to_owned(), body).await?;
            }
            _ => {}
        }
    }

    // release distributed lock
    distributed_lock(false).await?;

    Ok(txn)
}

The code processes each operation in the transaction one by one. Read operations fetch values from lin-kv, write operations store values. The lock ensures no other transaction runs at the same time, making all operations atomic.

What I Learned

Working through these challenges helped me understand several important concepts in distributed systems.

1. Gossip protocols: They are simple but effective. Nodes just share messages to their neighbors, who then forward them to others. Messages spread across the entire network without any central coordinator.

2. CRDTs: The grow-only counter showed how we can design data structures that avoid conflicts. By using operations like "take the maximum", nodes can merge updates in any order and still end up with the same result without coordinating.

3. Trade-offs everywhere: Each challenge had multiple solutions with different trade-offs. Immediate vs batched broadcast (speed vs bandwidth). Stateful vs stateless counters (independent service vs simplicity). Locks vs lock-free (consistency vs performance). Picking the right trade-off depends on what matters most for your system.

4. Simple solutions work: The unique ID generation was just combining node_id with a counter. It didn't need complex algorithms. Sometimes understanding the problem constraints is enough.

5. Testing distributed systems is tough: Maelstrom simulates network partitions and delays, which caught issues I wouldn't have found with regular testing. Building reliable distributed systems requires dedicated testing tools.

Conclusion

The Gossip Glomers challenges are a great way to learn distributed systems concepts. Whether you are new to distributed systems or looking to deepen your understanding, they let you experiment with ideas and see what works and what doesn't in a controlled environment.

Thanks to Fly.io and Kyle Kingsbury for creating these challenges. They provide hands-on experience with concepts that are often only discussed theoretically.

The complete code is available on GitHub.

Happy coding! 😊