Following the leader: leadership election in XTDB 2.2

Hey folks! 👋

You might have spotted that XTDB 2.2 has been a while coming - today I can share some exciting details on what we’ve been up to.

The headline is that XT now does its own leadership election: we no longer rely on Kafka to elect our leaders for us, which is the first step towards removing XT’s (pretty much) mandatory dependency on Kafka for highly-available production deploys 🚀

There’s a beta out alongside this post if you’d like to kick the tyres - details at the bottom - and what follows is the longer version: how XT has worked since its inception, why 2.2 had to change that, the fix we shipped in a release candidate that I wasn’t happy with, and what we’ve replaced it with.

Every node, every transaction

Some background first - if you already know how XT’s indexing works, do skip ahead to the replica log.

Since its inception, XT has had an architecture where every node reads the whole log and resolves all of the transactions. This has been a remarkably simple architecture, in the main - even in hindsight I wouldn’t have changed that decision, it got us off the ground and we’d never have gotten this far without it.

It doesn’t come without trade-offs, though.

Firstly, that (completely understandably) not everyone wants Kafka in production, both for cost and operational reasons. Internally, it means that every transaction has to be deterministic - that every node has to arrive at the same state reading the same messages. This isn’t massively onerous for us - we’re Clojurians at heart, so functional purity came fairly easily to us - but it did mean that any kind of error had to restart all of the nodes, and we’ve spent time over the years increasing the number of errors that only roll back a transaction rather than putting the node into an ingestion-stopped mode.

So why change it now? Three reasons:

1. 2.1 brought multi-database support, so our users are already effectively giving us boundaries on what needs to be indexed in serial - but all the nodes still do all the work anyway.
2. 2.2 brings external sources (giving XT the ability to import data through Postgres CDC and external-format Kafka topics).
3. We’re currently in the process of extending XT to multi-partition topics, in order to take us to the next level in write scalability.

We’ve long had people question XT only having a single-threaded writer - indeed, I met with a potential client in my very first week on XT who asked us about it!

You can get a hell of a lot done in a single thread - locally, we’ve seen 300k docs/second written on a single thread when you take network latency out, there’s relatively little overhead in XT itself on that path. That said, though, the write performance you can unlock by distributing the load is potentially huge. We already distribute the load of the compaction, and that work is all shared,¹ but it’d be lovely to share the transaction resolution around too. To do any of that, though, we need to remove the constraint that every node does all of the indexing - and that’s what 2.2 is all about.

Enter the replica log

2.2.0-rc0 was the first release of our ‘single writer’ architecture: now, only one node in a cluster reads the transaction log (now the ‘source log’), resolves the transactions it finds there, and puts those resolutions onto a new ‘replica log’ topic for all of the other nodes to read. ‘Transaction resolution’ here means turning SQL (which may read existing data in order to determine what changes to make) into simpler put, delete and erase events (which don’t).² In rc0, Kafka chose the node that did this: the nodes joined a Kafka consumer group over the source topic, and whichever member Kafka assigned the partition to became the leader.

Which raises the obvious question: what stops two nodes both believing that they’re the writer?

In rc0, the answer was Kafka’s transactional producers. Using these means that Kafka only allows one producer to be writing to a given topic-partition at any one time. When a new producer comes along, Kafka ‘fences’ the old producer (denying them any further writes), and they then resign their leadership position. In practice, most of the time, Kafka had already revoked the partition from the old leader before the new leader came online, so the old leader had already gracefully resigned - the producer fencing was an extra safety measure.

However…

We then discovered (through an XT design partner, thanks!) that this was flawed: in Azure Event Hubs, the scale-to-zero standard tier doesn’t support Kafka transactional producers.

Uh oh.

To unblock them as quickly as possible, we released 2.2.0-rc1 - arguably back to beta rather than a release candidate - which removed the dependency.

In rc1, we still used Kafka’s consumer groups to elect the leader, but we now had to create our own producer fencing. We did this using the Kafka consumer generation id, which we called a ‘term’ - Raft parlance, and we’ll come back to that. The rules from there are short:

1. readers of the replica log keep track of the highest term they’ve seen so far and ignore any message carrying a term lower than that high-water mark
2. the leader reads its own messages back off the replica log to check that no other leader has come in behind it.

This worked, but the reliance on Kafka’s generation id turned out to have its drawbacks: Kafka can reset generation ids if the group id isn’t active, so we had to introduce specific handling for that case, as well as a few other workarounds. It was a quicker fix than we’d have liked for a relatively core part of the system, and I just felt uneasy about supporting that in production long term, so rather than releasing rc1 as a stable we’ve now fixed it properly: by taking responsibility for the terms ourselves.

Following the leader

Before getting into what XT does, it’s worth a quick word on how leadership election tends to work in general. Again, if you’re already familiar with the prior art, do skip ahead to how it now works in XT.

Most of what follows is heavily inspired by Raft - Diego Ongaro and John Ousterhout’s consensus algorithm, or, to give the paper its full title, ‘In Search of an Understandable Consensus Algorithm’. I’d say they’ve achieved what they set out to, as well: the summary of the core RPCs fits on a single page, and it’s a genuinely approachable paper even if you don’t habitually read papers. There are proofs in there too, of course, but you can come away with a good grasp of what Raft does and why without following those - so if you’ve twenty minutes spare, do go and read it.

I should be clear up front, though, that what we’ve built here isn’t Raft. It’s heavily inspired by Raft, and it borrows Raft’s vocabulary, but it’s still a fair way from Raft - and we’ll come to exactly where the two part company shortly.

Raft introduces distinct roles for nodes in the cluster: followers, candidates and leaders, and nodes transition between these roles throughout their lifetime. Ongaro and Ousterhout put the whole Raft state machine on one diagram:

Figure 4 of ‘In Search of an Understandable Consensus Algorithm’, Diego Ongaro and John Ousterhout.

If you’d asked me before I’d read any of the distributed consensus literature, I think I’d have guessed the mechanics of these transitions entirely wrong - I’d probably have assumed that becoming the leader meant taking some sort of lock: that there was a thing somewhere you had to acquire, and that holding it was what made you the leader. It’s very nearly the complete opposite.

Instead, every node starts out as a follower, quietly reading along. When a follower notices that there’s no active leader - a leadership vacuum, if you like - it can unilaterally decide to become the leader itself. Assuming it’s qualified to do so (i.e. it’s aware of all the latest transactions), it simply claims leadership - there’s no “lock” to acquire, as such. The incumbent, if there is one, doesn’t get any say in the handover - they must gracefully resign.

The other side of that, to avoid unnecessary leadership churn, is restraint. If any follower can take leadership at any time, they’d better only exercise that right when they feel they really have to. In practice, this means waiting long enough to be fairly confident that the leader really has gone, rather than merely being quiet.

Which leaves the harder question: if any node can claim leadership at any time, how does the cluster ever agree on who’s actually in charge?

The contribution of Raft is that only one node will ever consider themselves leader at any one time - and, just as importantly, that every node in the cluster agrees on which one it was. It gets there by dividing the cluster’s shared history into ‘terms’. A term is a stretch of that history with (at most) one leader, numbered with a monotonically increasing integer, and everything that leader writes is stamped with its term. A term ends when a follower decides the leader has gone and claims leadership of the next one, so terms are consecutive and never overlap.

That’s a subtle but significant shift: nobody has to agree on who the leader is right now, which is a genuinely hard question to answer in a distributed system. They only have to agree on who led term 4 - a question about the past, which the shared history itself can answer.

The paper has it in pictures - each term opens with an election (in blue), then runs as normal operation (in green) until the next one comes along:

Figure 5 of ‘In Search of an Understandable Consensus Algorithm’, Diego Ongaro and John Ousterhout.

Note t3 in there, incidentally - an election that produced no emerging leader at all, so that term passed without one. That’s why ‘at most one leader’ above, rather than ‘exactly one’, and it’s one of the places where we get to be a little simpler than ‘full’ Raft.

In XT, the log is the referee

While we have a replica log, though, our leadership election can be a little simpler. 🙂

In XT, every node tails the replica log anyway - that’s how it gets its resolutions in the first place:

The election falls out of that same read. Each follower polls the replica log for the length of its ‘election timeout’, which on Kafka we set to somewhere between 5 and 10 seconds: in other words, “give me whatever messages you’ve got as soon as you get them - but if you see nothing in that time, stop and let me know”. That empty poll is what tells the follower there’s a vacuum, and that it should now claim leadership.

The claim itself (in XT) is a no-op message carrying a term one higher than the highest the follower has seen, appended to the replica log. ‘Claim’ is one name for it - it’s pretty much a hostile takeover! As soon as the current leader reads that message back off the log and sees the higher term, it steps down.

Which raises the obvious objection: what if they all do it at once?

Well, they might - the timeout is randomised, so they’re spread out a little, but there’s nothing to stop every follower claiming term five simultaneously. It turns out not to impact correctness, because the log is the referee here. They all append their claim to the same totally-ordered log, and each of them knows the position of the message it wrote: if I claimed at message 12, and by the time I read message 12 back I haven’t seen another term-five message, then I know my claim succeeded. The node that claimed and got message 14 will read message 12 first, and by the time it reaches its own claim it knows perfectly well that it wasn’t the first to claim term five, and so isn’t the leader.

First write wins - and nobody needs telling, because the loser finds out by reading the record that beat it before they read their own claim back.

It also means that, in XT, a term always ends up with exactly one leader rather than at most one: where Raft can have a split vote that elects nobody and has to re-run the term, the log settles it first time.

While a claim is in flight, incidentally, nothing much changes for the claimant. It doesn’t pause, it doesn’t buffer messages, and it doesn’t stop answering queries - it remains a follower, processing the replica log exactly as it was, right up until it reads its own claim back and finds out whether it worked.

“I’m still here, folks”

The flip side of a follower claiming when the log is quiet is that a leader that has nothing else to send has to regularly assert its presence, more often than the followers’ timeouts. Any write a leader makes is implicitly an assertion that it’s still there, because every replica message carries the term that produced it - but a database that simply isn’t taking any writes at the moment would otherwise look exactly like a database whose leader has fallen over. So if a leader hasn’t put anything through the log for a while, it puts through a message with nothing in it but its term - “I’m still here, folks”.

That’s also why we derive the election timeout from the assert interval rather than setting it independently: Raft sets this as a ratio, the assert timeout being 5-10x as frequent as the election timeout.

And if the old leader wakes up, or comes back online, or was merely busy for a little too long? Then it’s just bad luck, I’m afraid - sorry, you were away. I’ll accept the messages it wrote before my claim, but as soon as my claim takes effect, everything it writes after that carries a lower term than the one every reader is now fencing on, and gets discarded. Its writes aren’t prevented from landing; they’re simply ignored by everyone, which is a good deal simpler than trying to stop them.

What’s next

This is obviously the first step of many. Taking the election and the term off Kafka means XT no longer needs a consumer group, but the source log and the replica log are both still Kafka topics, and XT still needs a durable, totally-ordered log to act as referee. Replacing that is where we’d go next in this particular stream of work - to be prioritised in amongst all of the other feature requests, performance and stability improvements and bug reports, of course!

We’ve made a very deliberate decision to release this incrementally, both to minimise the deployment risk and to minimise what you have to migrate. If we’ve done our jobs properly then, in the ideal case, this’ll be a series of blue/green deploys at the end of which you’ll simply be able to delete your XT Kafka topics 🙂

Beyond that, multi-partition is the one that unlocks the write scalability we started this post with - and ‘full’ Raft is quite possibly on the cards in time. That said, full Raft is a considerably larger undertaking, and this gets us a decent distance without it.

Give it a go

There’s a beta out with all of this in it - 2.2.0-beta2, where you’ll find the full release notes and the usual install instructions.

The headline feature it unlocks is external sources: if you’ve got existing Postgres tables or Kafka event streams, you can now import them into XT and get a bitemporal view over the data you already have, without moving your system of record to do it. That’s the bit I’d most like people to kick the tyres on, so if you give it a go, please do let us know how you get on. If anything’s unclear, or you’re unsure how to migrate, give us a shout: hello@xtdb.com, or come and join us on our Discord.

Until next time!

James

Footnotes

1. If you’d like the detail on how that works, it’s in part 3 of the bitemporal index series. ↩

2. Where we can, we pre-compile the SQL down to puts on the client before it ever reaches the single thread - INSERT INTO foo RECORDS ? and INSERT INTO foo (cols) VALUES (...) get translated into Arrow documents, which are then passed straight through as bytes. ↩