In the last 4 years of my day job, I've been working on the next evolution of the chat system for the largest second hand marketplaces in Europe. The lessons learned are enough to fill several blog posts but if there's one takeaway it’s that getting an event driven system right is hard.
Why event driven systems are hard
By the nature of event driven systems you will always deal with an exponential number of states. Even in the happy path, when reading from multiple topics or when your message queue doesn’t have order guarantees, you can receive the events related to a user or entity in an arbitrary order.
Add the unhappy path where an event failed to process and got put into a dead letter queue or ignored and you’re in for a headache keeping all different scenarios in mind.
Let's say you're reading from 2 kafka topics and have 6 different events that can affect 1 isolated entity in your system. If you have 3 on one topic and 3 on the other, that still means you can have 20 different possibilities (binom(6,3)) in which order they will be processed, not even accounting for race conditions in case they are processed at the same time.
Your Assumptions are wrong
Let’s say you have a pretty good grasp on all of the services in your system, great separation between components so the state in one system doesn’t affect the other. Chances are that you’re still implicitly making some assumptions that are wrong!
A huge advantage of your event driven system is that you can potentially roll back to a previous state of your event queue (if your message bus supports it). Or put it in a dead letter queue and reprocess later.
But that means that you can’t rely on any of the nice properties that your queue potentially supports.
Event order? Get your thoughts in order!
Events processed only once? Consider that once more!
Events processed at least once? You better be prepared to put some work into dead letter queues, halt your entire pipeline if a message fails or write infallible side effect free code (like that's ever going to happen...) if you need that guarantee.
You will roll back your topic because you mess up.
Your service will have failures, connection issues or plain misconfigurations and requests will fail.
Your mom will yank the cable from your server because you didn’t clean your room.
Better to let go of the notion that your events will be processed in order and only once.
That means your service has to be idempotent and resilient to out of order events or topic consumer group resets if your message bus/broker supports that is easier and will save you and your shareholders some tears.
Fuzz Testing is Your Best Friend
Fuzz testing or property based testing (footnote: the names seem to be used interchangeably in literature) is different from unit or integration testing. Instead of writing a test that validates a known flow, you write a generator that generates test cases and an oracle that verifies that the output is correct.
It turns out that event driven systems are a pretty good fit for that. For the generator you generate a stream of events. The oracle depends a lot on what you want to test. But one standard approach is that after consuming all events you want your system to reach a final state. When generating your event you can adapt your expected final state based on that.
The final ingredient is generating a few hundred runs (or however many you like) and see if your assumptions hold.
The big capitalized word in here is assumptions. I’ve overused it in this blog post but that just shows how important it is to think about them. The oracle is a fantastic tool to allow you to think about them specifically. Because so far we only talked about the underlying message bus. But equally important are your domain assumptions and the contract that your service fulfills. When you write that oracle (or tell your favourite llm to do so) you have to think deeply about what those assumptions are and what guarantees your service can realistically make.
Writing fuzz tests I’ve never not found bugs, even in running production systems.
The cherry on top is that you can include those wrong assumptions that everyone makes into your test cases generically. Add a function that simulates event duplication, reprocessing something out of order events and apply that to all of your event driven fuzz tests. Since the final state should always be the same you can use the same oracle on permutations/duplications of your event stream.
Example
The arguably most important microservice is the one displaying messages to our users. It’s got a few seemingly simple rules. When listing conversations, keep track of the latest message and show it as part of the response When a user archives a conversation and receives a message afterwards that conversation should be unarchived
We use kafka and partition the topics by conversation id, so we don’t even have to worry about the order of the messages coming in.
Simple, right? Had you not read this article, your approach to implementing it would be to take message added events and update the latest message in that conversation list and mark conversations as unarchived in case they’re archived.
Let’s look at the model itself
The model
Message Added Event:
#[derive(Debug, Clone, PartialEq, Eq)]
struct ParticipantData {
// ...
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct BotData {
// ...
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct Message {
id: MessageId,
author_id: AuthorId,
message_sent_timestamp: Timestamp,
payload: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct Conversation {
id: ConversationId,
created_timestamp: Timestamp,
participants: HashMap<UserId, ParticipantData>,
bots: HashMap<BotId, BotData>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ConversationEvent {
id: EventId,
event_timestamp: Timestamp,
conversation: Conversation,
event_data: ConversationEventData,
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum ConversationEventData {
MessageAddedEvent(Message),
ConversationArchivedEvent(UserId),
}Writing the generator we have to think about a few things. Each of the fields by itself can in theory take any valid value. The ids are random UUIDs themselves, but what guarantees do we have about the timestamps, payload conversation participants, bots etc? As it turns out, not many. A payload can be any valid utf8 string, the number of participants and bots are in theory unlimited, in practice though there aren't more than 10. Just thinking about the generator already made us consider potential usage limits we might want to enforce.
The timestamps are particularly interesting and also the source of one of the discovered flaws in the production system. Is the message sent timestamp always before the event timestamp? Not necessarily! While in theory the message sent timestamp should always come after the conversation has been created, in practice, the request for sending a message and creating a conversation can be answered by two different servers with clocks that are out of sync. To not rely on assumptions we can't guarantee, for the generator we can generate a random timestamp between epoch and the end of time and call it a day.
The author id itself can be either a bot id or a user id. To find out which, one would have to inspect the conversations participants and bots. The ids of participants are strings. Are there any limits imposed by our system or guarantees given? Now would be a good way to find out or enforce them.
The generator
There are quite a few frameworks to do property based testing (footnote: quickcheck etc). But often you can get away with just a random generator.
One thing to define beforehand is what properties of your service you want to validate with your test. You can generate completely random events, introduce arbitrary failures into your methods but doing that will make your oracle as complex as your service itself. Intuition can guide you in selecting good properties to check.
For this example, we are verifying that for a simple user, the state of their conversation overview for one single conversation is correct.
fn random_event(
rng: &mut impl Rng,
conversation: &Conversation,
tested_user: &UserId,
) -> ConversationEvent {
let data = if rng.random_ratio(2, 7) {
ConversationEventData::ConversationArchivedEvent(tested_user.clone())
} else {
let author = random_author(rng, conversation);
ConversationEventData::MessageAddedEvent(random_message(rng, author))
};
return ConversationEvent {
id: Uuid::new_v4(),
event_timestamp: random_timestamp(rng),
conversation: conversation.clone(),
event_data: data,
};
}A generator for a random message added event is relatively simple. With some probability, we either archive the conversation or add messages. The event timestamp is completely random since we want to avoid relying on event order.
Likewise the message is random:
fn random_message(rng: &mut impl Rng, author_id: AuthorId) -> Message {
Message {
id: Uuid::new_v4(),
author_id,
message_sent_timestamp: random_timestamp(rng),
payload: random_string(rng, 0, 10_000),
}
}For archiving the conversation, we only archive the actively tested user, since archiving other users conversations is unlikely to influence our use case.
Finally, to generate the full test case:
fn generate_test_case(seed: [u8;32]) -> TestCase {
let mut rng = StdRng::from_seed(seed);
let conversation = random_conversation(&mut rng);
let tested_user = conversation
.participants
.keys()
.choose(&mut rng)
.unwrap()
.clone();
let mut test_case = TestCase {
events: Vec::new(),
tested_user,
expected_final_state: empty_user_inbox_conversation(conversation.id),
seed,
};
let num_messages = rng.random_range(1..50);
for _ in 0..num_messages {
let event = random_event(&mut rng, &conversation, &test_case.tested_user);
reconcile_final_state(&mut test_case.expected_final_state, &event);
test_case.events.push(event);
}
test_case
}Notably, we're always passing around the random number generator and initialize the test case generator with the seed. If we fail a test case on a rare condition, it's useful to be able to run through it with a debugger. Logging the seed can make your life a lot easier in that regard.
Running the Test Case
Lastly, you run the test case and compare against the output:
#[test]
fn fuzzy_test() {
let mut service = initialize_test_service();
for _ in 0..1000 {
run_fuzzy_test(&mut service)
}
}
fn run_fuzzy_test_case(service: &mut impl UserInboxService) {
let test_case = generate_test_case(rand::random());
for event in &test_case.events {
service
.process_conversation_event(event)
.unwrap_or_else(|e| {
panic!(
"Failed to process event for case {:?}: {:?}",
test_case.seed, e
)
});
}
let conversations = service
.get_user_inbox(&test_case.tested_user)
.unwrap_or_else(|e| {
panic!(
"Failed to load inbox for case {:?}: {:?}",
test_case.seed, e
)
});
let inbox_conversation = conversations
.iter()
.find(|c| c.conversation_id == test_case.expected_final_state.conversation_id)
.unwrap_or_else(|| {
panic!(
"Failed find conversation in inbox for case {:?}",
test_case.seed
)
});
assert_eq!(
inbox_conversation, &test_case.expected_final_state,
"The expected state for case {:?} is different from the actual state. Expected: {:?}, Actual {:?}",
test_case.seed, test_case.expected_final_state, inbox_conversation
);
}Variations
The example above already covers a lot of different behaviour. But depending on the correctness requirements of your system, you can add more variations to your system.
One easy way to cover more cases is by randomly shuffling or duplicating events. Furthermore, we don't verify any potential concurrency bugs which could be done by processing the events in multiple threads in parallel.
Lastly, even when things go wrong, you will still want certain properties to hold. In our example for instance we never want the get conversations to crash. Injecting for instance random failures into your repositories or generating even invalid events and feeding them into your system could be a good way to ensure correctness in the face of adversary.
As many things in software it's a trade off between time investment and confidence in your system.
Outro
As we've seen, implementing a fuzz test is not that difficult. Writing the example code took me about 2 hours.
What's fundamentally different from unit tests