# Multigres Architecture Overview (https://multigres.com/docs/architecture)
## Principles [#principles]
Multigres will follow a set of principles suited for large scale distributed systems. They are as follows:
### Scalability [#scalability]
In a distributed system, scalability is mainly achieved by removing all possible bottlenecks. Among them, the most challenging one is the database. Multigres will be designed to scale the database horizontally by sharding it across multiple Postgres instances. Multigres will also provide an additional scalability option by managing read replicas.
### High Availability [#high-availability]
Multigres strives for enterprise grade availability. To achieve this, it will use a combination of techniques:
* A consensus protocol for leader election and failover management.
* Fully automated cluster management.
* No disruption of service during upgrades or maintenance.
### Data Durability [#data-durability]
Multigres will ensure that data is durable using a consensus protocol. It will guarantee that a write that has been acknowledged as success to a client must not be lost.
Additionally, it will provide a backup and restore mechanism to ensure that data can be recovered in case of catastrophic failures.
All other metadata will be stored in a distributed key-value store like etcd, which can also be backed up regularly, or can be manually reconstructed.
### Resilience [#resilience]
Multigres will provide resilience against spikes and overloads by employing queuing and load shedding mechanisms.
To protect from cascading failures, it will implement adaptive timeouts and exponential backoffs on retries.
### Observability [#observability]
In spite of all precautions, incidents can happen. Multigres will provide a comprehensive set of metrics and logs to help diagnose issues.
## Features [#features]
A primary goal of Multigres is to provide full Postgres compatibility while enhancing scalability, availability, and performance. Its key features will include:
* Proxy layer and Connection pooling
* Performance and High Availability
* Cluster management across multiple zones
* Indefinite scaling through sharding
Multigres can be deployed to suit different needs. We will introduce you to the key components as we illustrate various deployment scenarios.
## Single database deployment [#single-database-deployment]
In a single database deployment, Multigres will act as a proxy layer in front of a single PostgreSQL instance. This setup will be ideal for small applications or development environments where simplicity is key.
The main components involved will be:
* **Multigateway**: Multigateway will speak the Postgres protocol and route queries to Multipooler through a single multiplexed gRPC connection.
* **Multipooler**: The Multipooler will be connected to a single Postgres server, and will manage a pool of connections to the database. They will both run on the same host, which will typically be a Kubernetes pod.
In this scenario, Multigres will not address the durability of the underlying data. Therefore, it is recommended to use a resilient form of cloud storage to ensure data safety.
For a Multigres cluster to operate, two other components will be required:
* **Provisioner**: This will typically be a Kubernetes operator that handles provisioning of resources for the cluster. For example, a `CREATE DATABASE` command will be redirected to the provisioner that will allocate the necessary resources and launch the Multipooler along with its associated Postgres instance.
* **Topo Server**: This will typically be an etcd cluster. The Provisioner will store the existence of the newly created database in the Topo Server. The Multipooler will also register itself in the Topo Server to allow Multigateway to discover it.
## Multiple database deployment [#multiple-database-deployment]
Unlike a traditional Postgres server, every Multigres database will be created in a brand new Postgres instance coupled with its own Multipooler.
The Multigateways will be scalable independently based on resource needs. The application will connect to any Multigateway, which will route the queries to the appropriate Multipooler based on the database name.
This deployment style will allow for a large number of databases to be deployed under a single Multigres cluster.
The figure does not show the Topo Server and Provisioner components, but they will still be required for the cluster to operate.
## Performance and High Availability [#performance-and-high-availability]
Multigres can be configured to add replicas as standbys. In this setup, we introduce the `Multiorch` component, which manages the health of replication across replicas. It monitors replication, repairs broken streams, and coordinates failover, ensuring replicas remain in sync with the primary database.
Multiorch will implement a distributed consensus algorithm that will provide the following benefits:
* **High Availability**: by promoting one of the replicas to be the new primary in case of a failure.
* **Data Durability**: by ensuring that all writes are acknowledged by a quorum of replicas before being considered successful.
* **Performance**: because the data can be stored on a local NVMe for faster access.
Multiorch will operate on an unmodified Postgres engine by using full sync replication. For a better experience, we recommend using the two-phase sync plug in (more details on this later).
Multiorch will be configurable to use a Raft style majority quorum. It will also be configurable to support more advanced durability policies that don't depend on the quorum size. This will be achieved by using a new generalized consensus approach (covered later).
Multigateway will make use of the replicas to scale reads for situations where the application can tolerate eventual consistency. It will also be configurable to support consistent reads from replicas at the cost of waiting for writes to finish transmitting the data to the replicas.
## Cluster management [#cluster-management]
A Multigres cluster will be deployable across multiple zones or geographical regions. In the previous examples, the components will be deployed in a single zone. Under the covers, they will be deployed in the `default` cell, which will be implicitly created for every new database. In the case of a multi-zone deployment, you will have to explicitly create cells and deploy the components in those cells. You will not have to preserve the original default cell.
In Multigres parlance, a cell will be a user-defined grouping of components. It will represent a zone or a region.
### Topo Servers [#topo-servers]
In a multi-cell deployment, the Topo Server will be splittable into multiple instances. It is recommended that the Global Topo Server be deployed with nodes in multiple cells. For every cell, a cell-specific topo server will be deployable.
The Global Topo Server will contain the list of databases and the cells in which replicas are deployed. This information will be used more sparingly.
The cell-specific topo servers will contain the list of components deployed in that cell, such as Multigateways and Multipoolers. The purpose of this design is to ensure that a cell that is partitioned from the rest of the system can continue to operate independently for as long as the data is not stale.
### Single Primary [#single-primary]
Irrespective of the number of cells, there will exist only one primary database at any given time. The Multigateways will route all requests meant for the primary to the current primary even if it is not in the same cell.
However, read traffic directed at replicas will be served from the local cell.
### Multiorch [#multiorch]
It will be recommended that one Multiorch be deployed per cell to ensure that failovers can be successfully performed even if the network is partitioned.
The consensus protocol will ensure safety even if the Multiorchs are not able to communicate with each other.
The durability policies will be settable to survive network partitions. For example, you may request a cross-cell durability policy that will require an acknowledgment from a replica in a different cell before considering a write successful.
You will also be able to request Multiorch to prefer appointing a primary within the same cell as the previous primary to avoid unnecessary churn.
### Backup and Restore [#backup-and-restore]
Multigres will perform regular backups of the databases. These backups will be restored when new replicas are brought online.
## Sharding [#sharding]
In the previous examples where the databases were unsharded, all the tables would have been stored on a single Postgres database. In this situation, there will be a one-to-one mapping between a Multigres database and the Postgres instance.
In reality, a Multigres database will be distributable across multiple Postgres instances. These will be known as TableGroups. Additionally, each TableGroup will be shardable independently, which will result in more Postgres instances within a TableGroup. When a Multigres database is created, a `default TableGroup` will be created, which will be an unsharded Postgres instance. This will be where all the initial tables are created. When you decide to shard a set of tables, you will be able to create a new sharded TableGroup and migrate those tables to it. From the application's perspective, the tables will appear as if they are part of a single database.
You will also be able to create separate unsharded TableGroups.
In the above example:
* `t1` is stored in the original `default` unsharded TableGroup. The single shard in this TableGroup is named `default:0-inf`.
* `t2` is split in the two shards of `tg1`. The shards are named `tg1:0-8` and `tg1:8-inf`. Note that the digits are hexadecimal.
* `t3` is stored in TableGroup `tg2`. The single shard in this TableGroup is named `tg2:0-inf`.
The rest of the Multigres features like cluster management, HA, etc. will be designed to work seamlessly with these sharded TableGroups.
We will cover the Multigres sharding model in more detail in a separate document.
## Combining Shards and Cells [#combining-shards-and-cells]
If cells were one axis of a Multigres cluster, then shards would be another axis. The two multiply with each other to produce a matrix of components. This is illustrated in Figure 5 below:
In a fully deployed cluster, the components will function in the following ways:
* **Multigateway**: There can be multiple instances of Multigateways in each cell. A Multigateway uses Multipoolers within the current cell to serve traffic. If necessary, a Multigateway would go cross-cell to access a primary if it's not in the current cell. The user or application can connect to any Multigateway to run their queries.
* **Multipooler**: There will be one instance of Multipooler per Postgres instance. Such an instance contains the data for one shard. Among all Multipoolers, one of them will be the primary and the others will be standbies or replicas.
* **Multiorch**: A Multiorch watches over a single shard across all cells. For a given shard, it is recommended that a Multiorch be provisioned for each cell. This allows for resilience against network partitions. If one Multiorch does not have the connectivity to perform a failover, another one in a different can take over. The Multiorchs are capable of operating without interfering with each other.
* **Local Toposerver**: One local toposerver per cell is required. For smaller deployments, the global toposerver could also be reused for this purpose. However, it is recommended that separate local toposervers be provisioned for each cell in case of larger deployments. This server is used for components within a cell to discover each other. For example, a Multipooler will publish itself through the toposerver, which will allow Multigateway to discover its existence.
* **Global Toposerver**: One global toposerver is needed to store the list of cells, the list of databases, and their backup locations. The global toposerver is typically deployed across multiple cells to survive network partitions.
## Upcoming Topics [#upcoming-topics]
We will cover the following topics in more detail in the future:
* Two-phase sync replication
* Generalized consensus for durability policies
* Multigres sharding model
# Credits (https://multigres.com/docs/consensus/part-00)
**Sugu Sougoumarane** — Creator of Multigres, Vitess
Many people from the community have reviewed this series and provided valuable feedback. It will be hard to name all of them. Heartfelt thanks go to the following members of the multigres maintainer team:
* [Deepthi Sigireddi](https://github.com/deepthi) (@deepthi)
* [Rafael Chacon](https://github.com/rafael) (@rafael)
* [Manan Gupta](https://github.com/GuptaManan100) (@GuptaManan100)
* [David Weitzman](https://github.com/dweitzman) (@dweitzman)
* [Cuong Do](https://github.com/cuongdo) (@cuongdo)
# Part 1: Defining the Problem (https://multigres.com/docs/consensus/part-01)
In this blog series, I have the following goals:
* Propose an alternate and more approachable definition of consensus.
* Expand the definition into concrete requirements.
* Break the problem down into goal-oriented rules.
* Provide algorithms and approaches to satisfy the rules with adequate explanations to prove correctness and safety.
* Show that existing algorithms are special cases of this generalization.
The first research paper that gained popularity was [Paxos](https://lamport.azurewebsites.net/pubs/lamport-paxos.pdf), and it was intimidating. Most people still don't fully understand it. Around the same time, another paper called [Viewstamped Replication](http://pmg.csail.mit.edu/papers/vr.pdf) was published, but it didn't achieve as much popularity. Later, [Raft](https://raft.github.io/) was introduced, providing an alternative approach that was easier to understand. It also included practical improvements that made it more usable in real-world scenarios. Specifically, it added failure detection and an enhancement to support log replication instead of the single-decree algorithm used by Paxos.
However, Raft remains a monolithic algorithm and is mostly used as a black box these days. Making changes to it is risky because you don't know what rules you might break. This fear has halted most progress in this area.
There are two reasons why consensus has remained a mystery for most:
1. The problem is not well-defined.
2. Previous research has focused on proving the correctness and safety of specific algorithms, rather than conceptualization.
Let's conceptualize instead. If we succeed, verifying the correctness of existing algorithms will become easier. More importantly, we can be bolder about modifying them to meet our needs better or creating entirely new ones.
There is a paper by Heidi Howard on [Generalized Consensus](https://arxiv.org/abs/1902.06776). I have read it, but I cannot claim to fully understand it. The paper is too theoretical for me, and I couldn't find an easy way to adapt it to real-world problems. It's quite possible that, if translated, it would be even more generic than what I intend to propose. However, I believe the goals differ: the paper focuses on a unified algorithm that can accommodate all existing consensus protocols. My goal is to develop a conceptual framework that enables the adaptation of consensus systems across diverse environments. Still, I did notice some overlaps between the topics discussed here and the paper. The concepts of revocation and flexible durability rules are definitely present in that paper.
I've made a previous attempt at this in my earlier [blog series](https://planetscale.com/blog/consensus-algorithms-at-scale-part-1), but it was incomplete. The series also had a bias because I wanted to demonstrate how to achieve this in [Vitess](http://vitess.io), despite its constraints. This time, I intend to be more precise and provide a foundation for something that can lead to a formal proof.
## Why are we even doing this? [#why-are-we-even-doing-this]
Above all, it never hurts to gain a better understanding of a system we depend so much on.
Additionally, the existing implementations are based on a majority quorum, which is too rigid. We are continuing to live with them because we don't have better options. [FlexPaxos](https://fpaxos.github.io/) proved that you don't need a majority quorum. However, no implementation has yet adopted those learnings.
We are also stuck with implementations that cannot be separated into meaningful concerns. This makes it hard to adapt them to other systems.
For this reason, there is still no native consensus protocol in Postgres. The few commercial organizations that offer solutions appear to have utilized Raft, but the details are not publicly known. Anecdotal information seems to imply that they used Raft as a black box.
Instead, we should ask how to make consensus work for the Postgres WAL replication. In Multigres, we plan to do precisely this. The additions we will add to Postgres will enable the implementation of many consensus protocols, including Raft.
## Redefining the problem [#redefining-the-problem]
I've asked people about what they think consensus is. I've heard a variety of answers:
* An algorithm to make a group of nodes agree on a value
* Consistency
* Majority quorum
There is some truth to all those answers. But there is a more appropriate definition:
:::tip Key Definition
Consensus solves the problem of Distributed Durability.
:::
If you look back at all the places where consensus has been used, you'll realize that durability is the primary reason why it gets used.
Beyond durability, we want the system to recover and resume operation quickly in case of a failure. For this, we need automation that detects and responds to such failures. From a theoretical viewpoint, failure detection isn't in scope. However, we can't build a usable system without this capability. Therefore, we should make it a requirement:
:::tip Key Definition
Consensus also solves the problem of High Availability.
:::
Of course, we also want to ensure that nodes don't diverge while fulfilling the above two requirements. In a way, this is an implicit requirement, because a system that diverges has essentially failed at durability.
To restate in simple words:
*A consensus system must ensure that every request is saved elsewhere before it is completed and acknowledged. If there is a failure after the acknowledgment, the system must have the ability to find the saved requests, complete them, and resume operations from that point.*
With the problem defined this way, we will work on a focused solution. As we progress, we will learn concepts and establish rules. We will also explore different implementation options. Then, we can verify the current algorithms against these rules.
# Part 2: Building the Foundation (https://multigres.com/docs/consensus/part-02)
In our previous post, we came up with an informal definition :
*A consensus system must ensure that every request is saved elsewhere before it is completed and acknowledged. If there is a failure after the acknowledgment, the system must have the ability to find the saved requests, complete them, and resume operations from that point.*
Let us stick to this definition and expand on some of these rules.
# Single value vs log replication [#single-value-vs-log-replication]
The original [Paxos](https://lamport.azurewebsites.net/pubs/lamport-paxos.pdf) paper was for a set of nodes to accept a single value. Although not practical, it is foundational. Understanding the single-value behavior will help us extend it for multiple values.
If we ask a Paxos system to accept a value and it succeeds, subsequent attempts to set a different value will fail. If the first attempt had an ambiguous outcome, the system might still finalize it later. A subsequent attempt may succeed or fail depending on the outcome of the first. This is shown in Figure 1 below.
Most practical systems need to accept multiple requests. To accommodate this, we have to modify this rule a bit: If the first attempt (A) succeeds, then a subsequent attempt (B) must be accepted and recorded as having occurred after A. If the outcome of A was ambiguous, then B requires the system to make a final resolution on A. If A is recovered and accepted, B is recorded after A. Otherwise, A is discarded, and only B is accepted. The system changes into one that consistently orders a series of requests. This is illustrated in Figure 2 below.
This was well understood by Raft, which is why it redefined this as a log replication problem. Since this is more practical, we will adopt Raft’s approach of replicating a log.
Depending on the type of system being implemented, these attempts can mean different things. For a key-value store, it may be a `SetKey`. For a database, it may be a `transaction`. For the sake of uniformity, we will generalize these as `requests`. Also, the data needed to persist a request may be physically different from the request sent by the application. For simplicity, we will treat them as equivalent.
### Consensus state diagram [#consensus-state-diagram]
Figure 3 above shows the state diagram for a request.
* A node can crash as soon as a request is received. This results in an abandonment.
* A received request could have been logged, but might not have met the durability criteria. If there is a failure, the request may not be discovered by the recovery process. If so, it will be abandoned.
* A request that has not yet become durable might get discovered by the recovery process. The process will replicate the request to make it durable.
* A request that has become durable will not be abandoned. This gives confidence for every node in the system to apply the request.
If a request gets applied without experiencing any failures, it will be acknowledged as a success to the requester. Otherwise, its outcome will be resolved later by a recovery process.
### Rejections and failures [#rejections-and-failures]
The system can reject an invalid request. If so, the application can assume that it was not accepted. However, if a failure occurs due to a timeout or a node failing, the outcome would be unknown. The application must reconnect to the system and verify if the previous request was accepted or not. It is the application’s responsibility to know the difference between these two errors.
Many of us would have experienced this when we click on the “Pay” button while shopping online, and it spins and times out 😂.
# Durability Requirements [#durability-requirements]
The problem definition states: “*every request is saved elsewhere”.*
This requirement is open-ended because durability requirements are user-defined. We want to accommodate all reasonable use cases.
Today’s cloud environments have complex architectures with nodes, racks, zones, and regions. They have pricing structures that may encourage specific layouts. Additionally, enterprises often bring in their own policies. Combining these could result in complex requirements.
Here are some examples:
* We want X nodes to receive the data before a request is deemed durable.
* We need Y total nodes to ensure availability when there’s a failure.
* Something more sophisticated: We want to deploy eight nodes across four zones, with two nodes in each zone. Our durability requirement is that at least one node in a zone other than the primary must hold the data. This ensures protection against a zone failure and a network partition between zones. We choose two nodes per zone to prevent leadership from switching zones during routine maintenance.
These requirements do not necessarily fit the pattern of a majority quorum. What ends up happening is that we configure a majority quorum system in such a way that these requirements are met. Sometimes, the configurations end up being sub-optimal.
We need a design that can accommodate these types of complexities.
### Pluggable Durability [#pluggable-durability]
Since such durability requirements can be arbitrarily complex, let’s make these rules pluggable, but add some restrictions:
* The rules must depend on the current set of nodes.
* Properties of nodes (like AZ) can be used, as long as they are static.
* The rules cannot depend on external variables, such as time.
* Each leader can have different rules.
Additionally, the rules must be sensible for the system to function effectively. If not, it may lose data, not perform well, or stall.
The ruleset data structure would conceptually look like this:
* List of participants
* List of eligible primaries. For every primary:
* A list of node groups, where each node group is a valid durability combination
This could be further generalized by removing leadership from the picture and specifying durability as a set of acceptable node combinations. This approach would be more theoretically pure, but it would not improve the flexibility of the system. On the other hand, a leader-based approach is easier to reason about.
This sounds ambitious, but it is possible to build such a system.
# Orders of Magnitude [#orders-of-magnitude]
In real-life scenarios, a leader is expected to fulfill a large volume of requests, in the range of thousands of requests per second. A leadership term also lasts a long time, typically many days, and sometimes longer. The durability policies can be tuned to take this into account.
For example, you can choose to have a five-node system, but require the leader to reach only one other node for durability. This configuration will give you the performance benefit of a three-node cluster. At the same time, a node crashing will cause less anxiety because you still have four other nodes running. The trade-off is that a leadership change will require the coordination of more nodes.
You might find this hard to believe: Vitess operated a consensus system at YouTube with over fifty replicas worldwide. We mainly depended on the fact that a neighboring replica is likely to have received the transaction before the distant ones. There was one incident when a transaction somehow reached a single node at a remote location. Fortunately, the system detected this and still managed to preserve the transaction.
Although I wouldn't recommend something this audacious, it shows that you can run a system with an unusually large number of nodes without sacrificing performance and safety.
# Leader-Based Consensus [#leader-based-consensus]
We will focus on leader-based consensus systems. I am aware of the existence of some leaderless algorithms, but I am not familiar with how they operate. I also don’t know if the principles we discover during this design will cover those approaches.
# Part 3: Governing Rules (https://multigres.com/docs/consensus/part-03)
As we solve the problem of durability, we will realize that there is a simple set of governing rules that we will be applying repetitively. We will develop these as we progress in our design. However, we will share their entirety upfront.
If you followed these rules, you should actually be able to implement any kind of consensus system. Here are some definitions and rules:
### Definitions [#definitions]
* A consensus system executes a series of consistent distributed decisions made by multiple agents.
* A `decision` is an intent to make a change to the state of the system.
* An `agent` fulfills decisions.
### Rules [#rules]
1. Durability: Every decision is a distributed decision.
1. A distributed decision must be made durable.
2. A decision that has been made durable can be applied.
2. Consistency: Decisions must be applied sequentially.
1. Every agent must revoke the ability of all previous agents to make further progress before taking any action.
1. Inference: Every agent must provide a way for future agents to revoke its ability to make progress.
2. Every agent must discover decisions that were previously made durable, but not applied, and honor them. Clarifications:
1. There are situations where it will be impossible to know if a decision met the durability criteria. If so, the agent must honor such decisions because they might have been applied.
2. Decisions that get honored must be made durable and applied as a new decision made by the current agent (rule 1).
3. Inference: If an agent discovers multiple conflicting timelines, the newest one must be chosen.
These rules have a hierarchy. If you can satisfy the top-level rule, you do not have to follow the sub-parts. To accommodate all possible algorithms, the rules also avoid dictating any approach or implementation.
For a leader-based system, there are three types of `decisions`:
* Fulfilling requests
* Changing leadership
* Changing durability rules
In the next few posts, we will discuss implementation strategies for these decisions.
In Raft, a `leader` is an `agent`. In our analysis, we will introduce one other agent: the `coordinator`.
### Questions [#questions]
Are we claiming that algorithms like Paxos and Raft follow these rules?
*Yes. We’ll validate this as we expand on rules.*
If one were to implement a system that followed these rules, but didn’t follow anything like what Paxos or Raft did, would it still work?
*Yes.*
What do these generalizations allow that previous algorithms didn’t?
* *Durability rules can be arbitrarily complex.*
* *The number of nodes need not dictate the durability rules. This was already demonstrated in FlexPaxos. This generalization includes this flexibility.*
* *The rules don’t dictate implementation: We have the flexibility to separate concerns in an implementation or implement them differently. We can also reuse existing parts of other systems to compose a full system.*
# Durability vs Discoverability [#durability-vs-discoverability]
Durability and discoverability are two sides of the same coin. We need to define durability rules for two purposes:
1. Data must survive node failures.
2. Data must be discoverable if there are network partitions.
For a majority quorum, if there is a single network partition, data that reached durability can always be discovered. This is because one side of the partition will have a majority and one of those nodes will have the data. However, more than one network partition can cause the data to not be discoverable.
In real life, network partitions are not totally random. So, you can craft durability rules based on expected failure patterns.
If there is a failure, the agent that performs the discovery can compute the minimum set of nodes that need to be visited to ensure that it discovers all completed requests. If one of those nodes is not reachable, the recovery will stall. People will get paged, and everyone can panic.
In other words, if the durability criteria do not take discovery into account, the system can stall. For all practical purposes, it is equivalent to a data loss. This is because production systems are required to meet specific availability requirements. The business priorities may force us to abandon the unreachable node in favor of serving new requests.
# Meaning of Apply [#meaning-of-apply]
The meaning of ‘apply’ depends on the system that implements the protocol. For example, in the case of a database, a `commit` would count as an apply. For a file system, an `fsync` would count as an apply.
An apply is considered to be an irreversible process. It should be done only when we are certain that a request will not be abandoned.
The consensus system is not concerned with the semantics of apply. However, the request stored in the log should be such that the outcome of the apply is deterministic.
# Missing terminology [#missing-terminology]
You’ll notice that there are some expected terms that are missing:
* **Leader, Follower, Candidate**: These are states that agents go through during the process of fulfilling their decisions. We will introduce these as needed.
* **Proposal number/term**: These are implementation details, used to enforce the ordering of decisions. There are other options.
* **Majority quorum**: We’ve already covered this. This is not a necessity.
* **Intersecting quorums**: This concept was introduced by FlexPaxos in place of majority quorums. We will instead discuss discovery, revocation, and candidacy.
* **Voting** is also not used, because it is misleading. There is no election either. A leader is appointed, not elected.
With all the groundwork laid out, it’s time to jump into the actual algorithms.
# Part 4: Fulfilling Requests (https://multigres.com/docs/consensus/part-04)
Let’s restate the subset of rules that are relevant to this section:
1. Durability: Every decision is a distributed decision.
1. A distributed decision must be made durable.
2. A decision that has been made durable can be applied.
# Definitions [#definitions]
* The `cohort` is the full set of nodes that are responsible for fulfilling the durability requirements of the system. In other words, these nodes are responsible for persisting their logs.
* A `quorum` is any combination of nodes that are needed to meet the `durability criteria`. We will use these terms interchangeably depending on the context.
### Roles [#roles]
* A `leader` is a designated node in the cohort that is empowered to accept and complete requests. It continues to serve requests until its leadership is revoked.
* The rest of the nodes in the cohort are `followers`. Their role is to assist the leader in its workflow to make requests durable.
* `Observers` are nodes that are not part of the cohort. They are replicas that only accept requests that are ready to be applied by the leader.
# Sample use case [#sample-use-case]
For better understanding, we will use the following example setup:
* A six-node cohort: N1-N6.
* Only N1 and N4 are eligible leaders.
* Durability criteria for N1: Data must reach both N2 and N3
* Durability criteria for N4: Data must reach either N5 or N6
This is an impractical configuration. However, it has some unique properties that will help us demonstrate that a system can work with arbitrary rules:
* Eligible leaders have different durability rules.
* Not all nodes are eligible leaders.
* It has an even number of nodes.
The role of a leader is well understood in practical terms: It is a node that is authorized to accept requests from the application and fulfill them while also making the requests durable. As covered before, this is achieved by replicating a log to other followers.
# Initial State [#initial-state]
Let us start with the initial state as follows:
* N1 is the leader
* Nodes N2-N6 are followers
# Processing requests [#processing-requests]
The algorithm explained below is very similar to the way Raft fulfills requests. The only difference is in the method used to determine if the durability criteria is met. For Raft, durability is determined by counting the number of followers that have acked a request. If it’s a majority, the request has become durable. In the generalized approach, the specific criteria has to be met. For N1’s leadership, acks from N2 and N3 must be received. No other acks count.
In Raft, each node must have a log of the requests being processed. Requests get appended to the log, and there is a trailing commit index (aka applied index) that determines the point up to which the events of the log have been applied.
The figure below shows a step-by-step animation of how requests get fulfilled by a generalized consensus system.
When a leader receives a request, it appends it to the log and also sends it out in sequential order to all the nodes of the cohort. Every node that receives the event appends the request to its own log and responds with an acknowledgement (ack) stating that the event has been received. At this point, nothing has been applied.
The leader may receive other requests while waiting for an ack. If so, it can continue to append them to the log and transmit them to the followers.
The leader (N1) must wait until it receives the necessary acks to reach quorum. In this case, the acks must come from nodes N2 and N3. Once both those acks are received, N1 is allowed to move the applied index forward and apply the event. At this point, N1 can also return to the caller with a success response. At the same time, it must send an apply message (update applied index) to all the followers to apply the event.
We call this method of replication “Two-Phase Sync”.
### Additional Observations [#additional-observations]
* Acks from any nodes other than N2 and N3 do not count towards the durability criteria, and must be ignored by N1.
* If N4 were the leader, a single ack from either N5 or N6 would be sufficient.
* Other nodes are still required to apply the logs as they receive the apply messages.
* While N1 is the leader, acks from N4, N5, and N6 do not count. They could optionally be configured as observers for as long as N1 is the leader. However, as we will see much later, there are some advantages to them continuing to act as followers, and have N1 ignore their acks instead.
### Followers [#followers]
A follower can be in one of these states:
* It might have just been rebuilt from a backup, or it may be lagging in replication: In this state, the follower’s latest logs would be behind the commit index of the primary. If so, the leader sends the committed logs as final until the commit index is reached.
* Caught up: In this state, the follower’s latest logs are past the commit index of the primary. The primary sends events in two-phase mode, and the follower responds with corresponding acks.
* Conflicting entries: In case of a conflict where a follower’s unapplied log does not match that of the leader, the follower should accept the leader’s logs as authoritative and discard conflicting entries.
### Observers [#observers]
Observers only receive finalized apply requests.
# Validation [#validation]
Let us now validate if the above algorithm follows Rule 1. In this scenario, the fulfillment of a request is a decision. From this perspective:
* Sending the request to all followers and waiting for the necessary acks is rule 1a.
* Moving the commit index forward and applying the event is rule 1b.
There are other replication modes, but they don’t follow rule 1:
* Async replication: The leader appends and applies the request, and asynchronously sends the events to the followers. This breaks rules 1a and 1b. This can lead to data loss if the leader node crashes.
* Synchronous replication: The leader sends the request to the followers as final. The followers apply the change and send an ack to the leader. The leader applies the change when the ack is received. This follows rule 1a, but breaks rule 1b. This is because the followers apply the change before the request has met the durability criteria. This can lead to inconsistent “split-brain” states.
*Postgres can rewind transactions. When a split-brain scenario happens, it is possible to identify the transactions that must be rewound to restore system consistency. Using this approach, it is possible to design a system that meets the necessary durability criteria. Details about this are covered in our earlier [blog post](https://multigres.com/blog/postgres-ha-full-sync#existing-replication-pitfalls).*
### Rule 2 [#rule-2]
During this explanation, we did not pay attention to rule 2: Consistency: Decisions must be applied sequentially.
Because this is the initial state, there is no previous agent. So, there is no need to revoke anything or honor any previous work.
However, rule 2 does apply to subsequent requests that follow the first one. In this case, we follow the title of rule 2: “Decisions must be applied sequentially”. As long as we append to the log, we are following the entire rule, which is sufficient.
# Roles [#roles-1]
In the above scenarios, nodes have taken on two roles: Leader and Follower. Of these, the Leader is the active agent making decisions. The followers support the leader by making requests durable and by responding with acks.
### An alternate type of leadership [#an-alternate-type-of-leadership]
Reviewing our cohort setup again, we had instinctively linked a log to each node. However, the rules do not require this; they only specify that decisions must be made durable.
For example, we could detach the leader from its log and make it an independent node. ~~It~~ This detached leader will still need to send events to all nodes required for a quorum. In this case, it would be N1, N2, and N3, where all three nodes are followers. This is also valid because it satisfies the same rules.
This is how systems like Aurora and Neon achieve distributed durability even though they don’t resemble Paxos or Raft.
In the interest of focus, we will not expand on this scenario.
# Steady State [#steady-state]
A leader can continue to serve a large number of requests in the current state. This is usually interrupted if there is a need to update the software or if a failure occurs.
When the leader has to change, we have new decisions to make. We will talk about those in the next post.
# How is this different from traditional consensus? [#how-is-this-different-from-traditional-consensus]
The two-phase mechanism of sending out the requests, waiting for the necessary acks, and then sending out messages to apply the requests is nothing new. This is how Raft also works. The part that differs is that the rules for what constitutes durable can be arbitrary.
If we provided a plugin mechanism for the rules, these acks would be handled by the plugin, which would validate them against the durability rules. This would allow the main algorithm to remain agnostic of the durability policy.
We have also shown that the rules allow for an alternate way to meet the durability criteria with a leader that is detached from its logs.
# Part 5: Before and After (https://multigres.com/docs/consensus/part-05)
Let us reiterate the relevant rules from part 3.
### Rules [#rules]
1. (not needed for this section)
2. Consistency: Decisions must be applied sequentially.
1. Every agent must revoke the ability of all previous agents to make further progress before taking any action.
1. Inference: Every agent must provide a way for future agents to revoke its ability to make progress.
There will come a time when a new Leader must be chosen. It may be due to a planned event, such as a software rollout, or a failure.
A leadership change is essentially a new decision, but it differs from a traditional request because it involves a change in roles. This requires applying the complete ruleset to execute a new decision. The outcome will be a leadership change. In this post, we will lay the groundwork for the approach we will use to satisfy rule 2a.
Before diving into the details, let us introduce some concepts.
# Failure detection [#failure-detection]
Failure detection is a necessary component for high availability in consensus systems. However, it is not a necessity for reasoning about safety. Therefore, we will cover this topic after we have completed the analysis of the core algorithms. For now, we can assume that failures can trigger an action to change leadership.
# The coordinator [#the-coordinator]
Majority-based systems, such as Raft, typically have either three or five nodes in their cohort. A larger number becomes inefficient because the number of acks needed to make a request durable becomes too high. Due to this limited number, the nodes also take on the tasks of health checking and performing leadership changes.
However, in a generalized setup, the number of nodes can be much larger. You might have ten or twenty nodes in the cohort. In that case, it isn't practical for all of them to perform health checks or coordinate leadership changes.
This task is logically separate and can be performed by agents that are not part of the cohort. We will name them the `coordinators`. Figure 1 illustrates an example setup of six nodes deployed across three availability zones, each with its own coordinator.
To summarize, the role of a coordinator is as follows:
* Perform health checks on all the nodes of the cohort.
* In case of a failure, perform a failover by appointing a new leader and ensuring that requests that the previous leader might have applied are honored.
* A coordinator can optionally provide the functionality to perform a “planned leader change”.
A smaller number of these coordinators can be strategically placed in different availability zones so that at least one of them has the necessary connectivity to appoint a new leader.
This does not preclude a cohort node from acting as a coordinator. We are highlighting that it is an independent role.
# A detour [#a-detour]
One way to satisfy Rule 2a is to ensure that no two coordinators act simultaneously.
For example, a coordinator could obtain a distributed lock with exclusive rights to take action until a timeout, and then act. The [Redis distributed](https://redis.io/docs/latest/develop/clients/patterns/distributed-locks/) lock is one such example. The coordinator that obtains the lock must ensure that it finishes its work before the timeout.
The advantage of this approach is that it eliminates races, thereby simplifying implementation. As we will see below, an algorithm that allows coordinators to race will be substantially more complex.
Unfortunately, this approach cannot be used for a theoretical proof due to the following reasons:
* It is impossible to guarantee how long a process will take to complete its work. While it is in the middle of taking a critical action, the timeout may pass, and a new coordinator may start to act, thereby violating sequentiality.
* Clocks are imperfect: Clock skews could cause the coordinator to think that it still has time to finish its work, while the time might have elapsed for the other clocks. Another coordinator may then start to act, and again, violate sequentiality.
We shouldn’t dismiss this approach. After all, real-life systems rely on clocks and timing. Even High Availability, which is essential for consensus protocols, depends on timing. From a practical standpoint, using locks and timeouts remains viable as long as we understand the trade-offs and implement safeguards against potential issues. In fact, Vitess employs this approach.
The bigger point we’d like to make here is that you can approach this problem in radically different ways.
# Elapsed time [#elapsed-time]
A theoretically correct solution should not depend on elapsed time: there should be no reliance on a clock or assumptions about how long actions take. For example, our reasoning should consider that an action can take one microsecond or one year. The same assumption applies to observing a previous action also: it might have occurred a few seconds ago or many weeks ago.
On the other hand, multiple coordinators could decide to act simultaneously and compete with each other. If so, we must ensure a consistent outcome.
The intuitive approach to solving a race condition is to favor the first coordinator. However, if the first coordinator takes too long, or even crashes before completing its work, no other coordinator can ever supersede it. In other words, this works only if we set a time limit for the completion of the task. This is the same as the lock-based approach described above.
We are now left with the alternate approach where a newer coordinator must be able to supersede an older one. This is why rule 2a uses the terms “current” and “previous” agents. It’s a lock-free algorithm and, therefore, naturally more complex than a lock-based algorithm.
# Ordering [#ordering]
When two coordinators decide to act and are expected to race, we need a way to ensure that their actions are serialized. This means that the system must assign an order to those decisions. In a distributed system, there are two types of ordering:
### Time ordering [#time-ordering]
Time ordering is the use of timestamps to determine the order in which coordinators make their decisions. The problem with timestamps is that clocks are unreliable.
In other words, time ordering is inaccurate.
### Encounter ordering [#encounter-ordering]
Encounter ordering refers to the physical sequence in which coordinators interact with a common node. This is also equivalent to causal ordering. It is accurate.
However, encounter ordering is unpredictable.
This unpredictability is acute because a coordinator can crash and never finish. Per the [FLP theorem](https://groups.csail.mit.edu/tds/papers/Lynch/jacm85.pdf), this is theoretically indistinguishable from a slow coordinator.
*The lock-based approach is an attempt to control this unpredictability.*
### Choosing the order [#choosing-the-order]
The unpredictability of encounter ordering is unavoidable because coordinators need to ask the cohort nodes to do work. If they race against each other, their actions are likely to be interleaved.
We need an algorithm that can withstand this unpredictability. The best approach is to assign an order to these coordinators in advance and set rules that cover actions occurring out of order.
Assigning an approximate timestamp when a coordinator decides to act can satisfy these requirements as long as we can ensure that the timestamps don’t collide. Additionally, we need to consider rogue clocks.
Raft offers a better approach: have the coordinators visit a set of overlapping nodes and use that information to determine a sequence. The benefit of this method is that it is precise due to the usage of encounter ordering. The clever part of this approach is that it does this before taking any action, meeting the above constraints. We will explain this in part 10.
# The term number [#the-term-number]
The assignment of an order between two independent nodes deciding to act is what Paxos calls a proposal number, and Raft calls a term number. This number must be universally unique, and is expected to increase monotonically. For clarity, we will use the Raft terminology and refer to it as the term number. The rules around term numbers apply to all agents. This includes coordinators as well as leaders.
To handle agents acting out of sequence, we’ll specify that a newer agent always supersedes an older one. This is a prerequisite for Rule 2a(i). To achieve this, we will make agents `recruit` nodes into their term:
* An existing agent is expected to give instructions to a node using its current term number as authority.
* A newer agent can use its term number as authority and instruct those nodes to stop accepting further requests from the existing agent.
For this to work correctly, the nodes should obey the following rules, also shown in Figure 2:
* Every node in the cohort must have a persistent term number.
* A node must honor requests from an agent with a matching term number.
* A node must reject requests from an agent with a lower term number.
* A node can be recruited into a term whose number is higher than the current one.
In Figure 2, the last example shows an agent implicitly recruiting a node that is from a lower term. This is allowed because it is equivalent to a recruitment followed by a request.
*The term number must be persisted to survive restarts. Otherwise, a restarted node that does not remember the term number it last agreed to may accept requests from a coordinator with a lower term and break rule 2a.*
# Reinterpreting Leadership [#reinterpreting-leadership]
In the previous post, we talked about a leader being able to fulfill multiple requests. We now need to define how the term numbers interact with these actions.
One approach would be to assign a term number for each request. This has a disadvantage: a new coordinator that intends to change the leadership must come up with a number that is not only greater than the current term but also greater than other terms the leader could be generating as it fulfills more requests.
An alternate approach is to treat these requests as sub-terms. So, if a leader started under term 5, its requests would have the terms 5-1, 5-2, etc., or alternatively, a log position under term 5. This way, a coordinator that starts a term 6 is guaranteed to supersede the current leadership.
This is the reason why we chose the term “term”: It implies that it is long-lived and can fulfill multiple requests.
This also simplifies our reasoning: Within a term, we only pay attention to rule 1. To start a new term, we have to follow the entire ruleset.
***Do we need a term number once leadership is established? Yes.** If the system becomes chaotic with multiple coordinators and leaderships partially succeeding and failing, we need the ability to know the order in which events took place. In these situations, the term number can be used as an authoritative source to determine this order.*
Essentially, every leadership starts under a term that is newer than the previous one, until a newer term replaces it.
In the next section, we will see how to safely perform these leadership changes by following rule 2a.
# Part 6: Revocation and Candidacy (https://multigres.com/docs/consensus/part-06)
Reiterating the relevant part of the rules.
### Rules [#rules]
1. Durability: Every decision is a distributed decision.
1. A distributed decision must be made durable.
2. A decision that has been made durable can be applied.
2. Consistency: Decisions must be applied sequentially.
1. Every agent must revoke the ability of all previous agents to make further progress before taking any action.
1. Inference: Every agent must provide a way for future agents to revoke its ability to make progress.
2. (skipped)
In the previous post:
* We covered the need for term numbers and some guidelines about how they should be generated.
* We discussed the need for nodes to participate in terms, and also covered the governing rules about what they can and cannot do.
* We also concluded that a leader can execute multiple requests within its current term.
In this post, we will conclude Rule 2a by focusing on Revocation along with its counterpart: Candidacy.
# Recruitment [#recruitment]
For a coordinator to successfully give instructions to a node, their term numbers must match. To enable this, the coordinator should first recruit the node to participate in its term. If the node’s own term number is lower, it will accept the recruitment and update its term number to match the coordinator’s term. Otherwise, it will reject the recruitment.
The coordinator does not need not specify a reason at the time of recruitment. It can choose what to do with the recruited nodes at a later time.
# Leader revocation [#leader-revocation]
To revoke an existing leadership, a coordinator can:
1. Directly recruit the leader. This will make it relinquish its leadership and wait for further requests.
2. Recruit its quorum nodes. This will stop them from accepting requests from the current leader.
Performing one of these actions will satisfy rule 2a.
Recruiting the leader, if reachable, gives us the advantage of a clean shutdown. The leader can ensure that in-flight requests are completed. It could inform its callers of an impending change in leadership, among other things.
The advantage of the second method is that it can succeed when the leader may be unreachable. This works even if it is still attempting to process requests on the other side of a partition. In our use case, if N1 is the leader, recruiting N2 or N3 is sufficient to revoke its leadership.
Both these examples are illustrated in Figure 1.
# Coordinator revocation [#coordinator-revocation]
Rule 1 states that every decision is distributed and must be made durable. This rule applies to coordinators also.
We also noted that durability rules depend on which node is the leader. The coordinator’s role is to appoint a new leader, hereafter called the `candidate`. The coordinator is expected to interact with the candidate and the nodes it relies on for its quorum. The specifics of these changes will be explained in the next post. For now, we can assume that it will need to:
* Recruit the candidate.
* Recruit the minimum number of nodes necessary for the candidate to fulfill requests successfully.
These recruitment actions with the intent to establish leadership are what constitute the `Candidacy`. This satisfies the rule 2a(i) requirement, because this candidacy can be revoked by requesting these nodes to participate in a newer term.
The revocation action for the candidacy is the same as the revocation action for leadership. This is not a coincidence because revocation is achieved by disrupting the ability for decisions to be made durable, and the only durability rules that exist in the system apply to leaderships.
***What if the coordinator completes its work of appointing a leader before the revocation process begins?** The answer to this question depends on whether we want the appointed leader to start a term that is newer than the process that is performing the revocation. Deciding to go this route makes things more complicated: As a newer agent, it must also follow rule 2a, which will be a repetition of the coordinator’s work.*
It is simpler for the established leadership to inherit the same term as the coordinator. The reasoning is that, under the coordinator's assigned term, the goal is to follow all the steps needed to appoint a leader, which involves rules 2a, 2b, and 1. Once this is accomplished, the term is delegated to the leader. Because this is a delegation, the leader does not have to revoke anything. It can therefore fulfill requests by continuing to iterate on rule 1. This is also consistent with Raft, where the candidate acts as the coordinator and eventually becomes the leader, all within the same term.
A newer agent will be capable of revoking the progress of the above term at any stage, even long after the leadership is established.
This sounds suboptimal for the use case we are trying to address: A newer coordinator may unnecessarily disrupt a leadership that was just established. Since our rules do not allow the usage of elapsed time, we have to accept this as a possibility. However, there are other ways to avoid such disruptions, and we will cover those options later.
# All possible leaders [#all-possible-leaders]
As discussed earlier, a coordinator is unlikely to know upfront whether other coordinators are active and, if so, who their candidate is. For this reason, a coordinator must assume that there may be multiple other coordinators racing with it, and they could be aiming to promote any of the eligible leaders. Therefore, it must revoke all possible leaderships in the cohort. We will cover how to do this with an example.
# Overlapping nodes [#overlapping-nodes]
Can there be an overlap between the nodes that are needed for revocation and the nodes that are needed for the candidate?
The answer is yes. In fact, it is likely the case for most practical scenarios. Fortunately, the act of recruitment does not have to differentiate between these two intents. This is also simpler and more efficient because a single recruitment message can be sent to all the nodes in parallel.
Once recruited, the nodes will be asked to do different things depending on their role in the new Candidacy.
# Example [#example]
In this example, we will first illustrate targeted revocations and then outline the requirements for a general revocation.
As a reminder, the example config is as follows:
* The cohort has six nodes: N1-N6.
* Only N1 and N4 are eligible leaders.
* Durability criteria for N1: Data must reach both N2 and N3
* Durability criteria for N4: Data must reach either N5 or N6
Let us assume that the current leader is N1 at term 5.
## Revocation [#revocation]
A coordinator decides to appoint a new leader and begins term 6, now called C6. Method 1 revocation requires recruiting N1 into term 6, which will cause N1 to step down from leadership. For method 2, recruiting the quorum nodes N2 or N3 into term 6 is sufficient. This will cause them to reject requests from N1, which is still on term 5. Both actions meet the requirements of rule 2a for the current leader.
This is illustrated in Figure 1. Ideally, the coordinator would try to recruit all the nodes. However, the two examples shown are sufficient for the revocation.
## Candidacy [#candidacy]
For the following scenarios, we will assume that N3 has been recruited by C6 for the sake of revoking N1’s leadership.
### Scenario 1: no race [#scenario-1-no-race]
C6 must now satisfy 2a(i) by recruiting the nodes needed for candidacy. Let us assume that it chose N4 to be the candidate. Then it must recruit N4 and N5, or N4 and N6, or all three. After the recruitment, those nodes will be on term 6. In the animation below, C6 recruits N4 and N5, which is sufficient for candidacy. This action is sufficient even if the other three nodes, N1, N2, and N6, are not reachable.
If the network partition was what caused C6 to act, then N1 might not have known that N3 was recruited and may still think that it is the leader. But it would not be able to fulfill any requests.
### Scenario 2: newer term steals the nodes [#scenario-2-newer-term-steals-the-nodes]
If a different coordinator decides on a newer term 7 (C7), it must attempt to revoke both terms 5 and 6. For revoking term 5, it has the same goal as C6, but does not have to follow the same method. For revoking term 6, it must recruit N4, or both N5 and N6.
If this happens before C6 reaches these nodes, then C6 will fail to recruit them due to them being on a higher term.
In the above example, C7 revokes N1’s leadership by recruiting N2, which is different from what C6 recruited. This is acceptable because it is still a successful revocation of N1’s leadership. C7 also revokes the candidacy for N4 by recruiting N5 and N6, which is different from what C6 recruited. This is also sufficient because C6 will fail to make progress. After all, N5, which it recruited, is now in term 7.
In other words, coordinators can each recruit a different set of nodes for revocation and candidacy, and they will still preserve safety.
### Scenario 3: newer term starts after scenario 1 [#scenario-3-newer-term-starts-after-scenario-1]
If C7 started after scenario 1 finishes, it will still end up recruiting the nodes that were recruited by C6, which will prevent C6 from making further progress.
C6 could have completed the rest of the actions needed to establish the new leadership. If so, C7 will end up revoking that leadership.
The result of scenario 3 would look the same as the result of scenario 2.
### All possible leaders [#all-possible-leaders-1]
So far, we targeted specific nodes for revocation and candidacy. This was mainly to illustrate the logic. As explained before, a coordinator must actually attempt to revoke all possible leaderships in the cohort. To achieve this, it must recruit a combination from each group:
For N1:
* N1
* N2
* N3
For N4:
* N4
* N5, N6
For example, N1, N4 is a valid combination. N1, N5, N6 is also a valid combination, etc.
To recruit for leadership:
* For N1, it must recruit N1, N2, N3.
* For N4, it must recruit N4, N5 or N4, N6.
To perform a leadership change to N4, a coordinator must recruit for both revocation and candidacy. This would be any combination from the first set and a combination needed for N4’s leadership. A valid set would be: N3, N4, and N5, which is illustrated in scenario 1. The animation below shows a few examples of valid combinations:
# Summarizing the rules [#summarizing-the-rules]
The summarized rules are more straightforward than the explanation: The coordinator must try to recruit all reachable nodes to participate in the new term. After the recruitment, the following criteria must be met among the nodes that were successfully recruited:
* No leader of an older term must be able to complete any requests.
* They must contain a candidate (or the intended candidate) with a sufficient set of nodes needed for its quorum.
# Which parts of Paxos or Raft do this? [#which-parts-of-paxos-or-raft-do-this]
For Paxos, this is the `prepare` message where it sends a proposal number to all nodes. For Raft, it is the `RequestForVote` message.
For both algorithms, the requirement is that the candidate recruit a majority of the nodes. This is sufficient because a majority satisfies both the requirements of revocation and candidacy.
Suppose a majority is not needed for quorum, like in the case of FlexPaxos. In that case, the nodes required for revocation will be different from those that are necessary for candidacy. FlexPaxos used an approach of intersecting quorums to ensure safety. However, it was essentially implementing Rule 2a without being explicit about it.
It took a lot of explanation to unravel the concepts behind such a simple action. But without this understanding, we can't safely modify these algorithms. Additionally, this understanding will help us when discussing rule changes in a later post.
# Part 7: Discovery and Propagation (https://multigres.com/docs/consensus/part-07)
Rules covered in this section:
### Rules [#rules]
1. Durability: Every decision is a distributed decision.
1. A distributed decision must be made durable.
2. A decision that has been made durable can be applied.
2. Consistency: Decisions must be applied sequentially.
1. (skipped)
2. Every agent must discover decisions that were previously made durable, but not applied, and honor them. Clarifications:
1. There are situations where it will be impossible to know if a decision met the durability criteria. If so, the agent must honor such decisions because they might have been applied.
2. Decisions that get honored must be made durable and applied as a new decision made by the current agent (rule 1).
3. Inference: If an agent discovers multiple conflicting timelines, the newest one must be chosen.
In the previous section, we covered how new coordinators ensured that they followed rule 2a, essentially ensuring that only one was able to take action at a given point of time. We discussed revocation and candidacy.
In this post, we will discuss:
* Discovery of timelines
* Propagation
* Establishment of leadership
# Discovery [#discovery]
The act of revocation has a serendipitous side effect: It also lets you discover all completed requests. The nodes that were recruited were necessary for the leader to complete its requests. By definition, it means that one of those nodes must have all the requests that were completed.
Beyond the completed requests, some of those nodes may also contain requests that were attempted.
Let us take the example in Figure 1. The durability rules are the same as the previous examples: N1 requires requests to reach both N2 and N3 for completion. In the above scenario:
* N1 has completed A. This request must not be lost.
* B has met the durability criteria. This request must also not be lost.
* C and D have not met the durability criteria.
If the coordinator manages to recruit all three nodes, it will know the whole truth: Requests A and B must be completed. C and D can be discarded. This is the hard requirement from rule 2b.
The follow-up question is: Is there harm in also completing C and D? There is no harm. After all, they were valid requests that the leader was trying to complete. We will need and use this flexibility in other failure scenarios.
Suppose there is a network partition, and the coordinator is only able to recruit N3, with no visibility into N1 or N2. Based on the log information, all it can infer is that A and B might have been applied. This is where we use the inference 2b(i): We honor A and B.
If the coordinator recruits N2, the same logic applies. But in this case, all it can infer is that A, B, and C might have been applied. Here, we honor A, B, and C.
If the coordinator recruits N2 and N3, it knows that C was not complete, and it has the option of discarding it. In this case, we can choose to honor just A and B, or A, B, and C. However, a general rule to honor the most progressed timeline is safe and simpler.
*The outcome of what will be honored after a failure is non-deterministic: If N3 were the only discovered node, C would be abandoned. If N2 were discovered, C would be included in the recovery. However, B will not be abandoned because it has already met the durability criteria.*
This algorithm would be simple if a coordinator always succeeded in establishing leadership. However, multiple failures can occur during propagation. If that happens, newer coordinators may see conflicting timelines. We will discuss these scenarios after analyzing propagation.
# Propagation [#propagation]
For a leader, the `decisions` it was fulfilling were `requests`.
A coordinator’s intent is not to fulfill requests. The `decision` it needs to fulfill is to establish a leadership using the `timeline` it has selected.
If the rules from the previous post were followed, the coordinator would have already recruited the candidacy nodes into the current term number. The goal now is to propagate this timeline to those nodes. Once this occurs, it can delegate its term to the candidate. This will establish its leadership, allowing the new leader to begin accepting external requests.
Since a timeline includes multiple requests, the standard action performed by a leader cannot be used for propagation; A leader has the right to apply each request individually. According to Rule 1, the entire timeline must be made durable before it is applied.
Before discussing implementation options, let's briefly review Rule 2b(ii). It states that propagation must be made durable as a new decision. This means that decisions should be versioned and their sequence should be known. We need to do this because we assume these attempts may fail. If they do, we must know the order in which these propagations were attempted. Without this, we cannot apply Rule 2b(iii).
This gives rise to a few implementation choices:
### The Paxos way [#the-paxos-way]
Paxos is a protocol meant for finalizing a single value. The way to reconcile this with logs that have multiple values is to treat each timeline as a composite value. Our goal will be to finalize a chosen timeline.
For those that are not familiar with Paxos, we actually need to track three variables:
1. The proposal number the node has agreed to participate in
2. The current value
3. The proposal number that was used to accept the value
The proposal number for the value is stored in a variable different from the proposal number that was accepted from the `prepare` request. We will now explain why it should be tracked separately.
Figure 2 shows an illustration of how this works.
As mentioned above, we will treat each timeline as a composite value like T0, T1, and T2, as shown in Figure 2.
In this scenario, let us start with N6, which contains timeline T1 which has two requests in its log.
The node has the following variables:
* Node name: N6
* Node’s term: 5
* Value: T1 (two requests)
* Value’s term: 5. This is the new variable we are introducing, which stores the term number when the value was accepted.
When C6 recruits N6, this will update the node’s term to 6. However, the value T1 and the `value’s term` 5 stay the same. At this point, if a new coordinator asks about N6, it will see the node’s term as 6 and the value as T1. But that timeline was set under term 5 and is not the correct value for term 6. That’s why we need to add a new variable to track the value’s term.
When C6 requests the node to change its timeline to T2, then the timeline and the value’s terms are updated. This, in essence, is rule 2b(ii). The decision made by C6 to change N6’s timeline from T0 to T1 is executed as a new decision under term number 6.
**This change must be atomic**: If C6 crashes while still writing T2, then no change should happen. It would not be acceptable for part of T2 to overwrite T1. In this state, it would have destroyed the previous timeline and replaced it with an incomplete portion of itself. This can lead to data loss.
**The change is authoritative**: The previous timeline may conflict with the new one. No matter, the new timeline must completely overwrite the previous one.
An extension of this rule is that regular leadership requests also have the value’s term associated with them. But they don’t change during the completion of requests because they are all under the same term.
*The value’s term must be persisted for the same reasons why the node’s term must be persisted.*
### The Raft way [#the-raft-way]
Raft has a different approach.
Raft does not have the term as a separate variable. Instead, each request includes a term number, which is part of the log. When it comes to completing requests, Raft and Paxos are equivalent. One could say that Paxos is more storage-efficient than Raft because storing the single value is functionally equivalent to what Raft achieves by storing the term number for every request.
But they differ on how the timeline is propagated. Let us take the following animated example:
Let us assume that term 7 is trying to propagate N1’s timeline ABCD to N6, which initially has AB in its log.
In Raft, the log is propagated non-atomically. When N1 has two additional entries, it could propagate to N6 in two steps (steps 1 and 2). Additionally, the term number associated with those log entries remains the old term 5. This appears to violate rule 2b(ii). According to this rule, propagation must use the latest term number.
However, the rule is valid, and Raft follows it. The reason: Raft has an addendum to how it implements durability. The updated rule is as follows:
*For a request to be durable, it must reach quorum. Additionally, the term number of the request must match the current term.*
In other words, from a new term’s perspective, events from all previous terms are considered non-durable. They only become durable when a new event using the current term is appended to the logs. This requires the entire timeline to become durable under the new term before it can be applied.
On Step 3, a new request with term 7 is created and replicated. This is what makes the timeline meet the term number matching requirement for durability. Once the necessary followers have also received the amended timeline, it can be safely applied. This behavior meets the requirements of rule 2b(ii).
We intentionally left a gap to accommodate a complication that term 6 might have brought in, which term 7 is not unaware of. We will cover this in a later section.
### The timestamp way [#the-timestamp-way]
Sometimes, you might not have control over the data you can add to the log replication. The specific use case is Postgres WAL replication: There is no simple way to add extra metadata, like a term number, to that log.
However, WAL commit events have timestamps. Assuming that clock skews are within tolerable limits, these timestamps can serve the same purpose as term numbers: they mark the order in which decisions are made.
This means that algorithms that resolve conflicting timelines will work equally well if we use event timestamps instead of term numbers.
In part 5 of our series, we discussed an alternate way of using locks and timeouts to enforce the sequencing of coordinators. Combining this timestamp method with the locks and timeouts approach creates a complete system that satisfies all our rules. This combination eliminates the need for term numbers entirely.
# Discovery revisited [#discovery-revisited]
### Selecting Timelines [#selecting-timelines]
In a network with intermittent failures, multiple coordinator attempts can fail, and each time, a coordinator may get to see only a subset of the nodes. Over time, a coordinator may see variety of timelines, and has to ensure that it chooses a safe one that does not violate the requirements of durability and consistency.
The rules for selecting a safe timeline are simple:
* The coordinator must recruit enough nodes to ensure that all possible leaderships are revoked. If this is not possible, then no progress can be made.
* Among the recruited nodes, the timeline with the latest decision (term) is always safe.
* If there are multiple timelines with the same term, then the most progressed timeline is always safe.
The reasoning is as follows:
* Every previous decision that was made was a safe one for that term. This applies recursively back to the oldest decision.
* The last discovered decision might have reached durability. It is even possible that some nodes have started applying that decision. This possibility makes all decisions previous to the last one unsafe.
What if we encounter a timeline that is more progressed than a newer decision? This only means that the timeline did not reach durability. Otherwise, the newer decision would have honored it. But now, we have to discard that progressed timeline because there is a chance that the newer decision has been applied already.
What if there exists a decision that is newer, but we don’t see it among the recruited nodes? If so, the decision did not reach durability, and need not be honored. We can choose the most appropriate timeline among those we discovered, and make sure to propagate it as the newest decision.
### Failure scenarios [#failure-scenarios]
We will now cover the following failure scenarios:
1. A coordinator may not be able to reach enough nodes to make any progress.
2. A coordinator may attempt to propagate a timeline and fail before making it durable.
3. A coordinator may attempt a timeline that differs from the previous one, try to propagate it, and fail.
4. A coordinator may succeed at propagating a timeline, but fail before promoting the leader.
5. A final coordinator may see all these attempts and must make a decision that does not compromise safety.
In the above sequence of failures, the most critical requirement is that attempt 5 must successfully discover attempt 4 and honor it.
We will analyze these scenarios assuming that we are using the Raft method of propagation. However, the strategy will work for all methods.
Let us restart with the example shown in the Raft section:
* N1 is the primary at term 5. It has received requests ABCD.
* N2 is a quorum requirement for N1. It has received requests ABC.
* N3 is a quorum requirement for N1. It has received requests AB.
* N5 is not a quorum requirement of N1 and has received A.
* N4 and N6 are not quorum requirements of N1 and have both received AB.
### Scenario 1 [#scenario-1]
*Scenario 1 is a no op. The coordinator cannot make any progress.*
### Scenario 2 [#scenario-2]
*A coordinator may attempt to propagate a timeline and fail before making it durable.*
C6 recruits N3, N4 and N5:
* N3 & N4 for revocation
* N4 & N5 for candidacy of N4
C6 crashes after propagating N3 to N5.
### Scenario 3 [#scenario-3]
*A coordinator may attempt a timeline that is different from the previous one, try to propagate it, and fail.*
C7 recruits N1, N4 and N6:
* N1 & N4 for revocation
* N4 & N6 for candidacy of N4
In this scenario, C7 did not discover any of C6’s activity. Based on what it discovered, it decides to propagate N1 to N6.
C7 crashes at this point.
### Scenario 4 [#scenario-4]
*A coordinator may succeed at propagating a timeline, but fail before promoting the leader.*
Let us now assume that Coordinator 8 (C8) attempts another leadership change. Let us also assume that it recruits the same nodes that C6 recruited. It will discover the following terms in the timeline:
* N3: 556
* N5: 556
* N4: 55
From this, C8 infers that C6 tried to propagate timeline AB (55), which makes it a legitimate decision. It propagates `6:ok` to N4. Following this, it appends `8:ok` to N5, and propagates it to N3 and N4.
This action makes the timeline durable. C8 can delegate leadership to N4, which can then apply this timeline and request that N5 and N3 apply it as well.
But let us assume that C8 crashes at this point.
### Scenario 5 [#scenario-5]
*A final coordinator may see all these attempts and must make a decision that does not compromise safety.*
After scenario 4, the cluster’s state is as shown in the animation above, which shows three distinct timelines.
In this particular scenario, the coordinator C9 sees all the nodes. It can see that N6 has a more progressed timeline. However, its term is lower than the highest term so far, which is 8.
It could make a "smart" inference and choose N6's timeline. However, the most safe decision would be to choose the timeline with the highest term. This is because choosing the highest term can never be wrong.
The animation shows the outcome of C9 choosing the timeline of N4 that is on term 8. You will also notice that the propagation overwrites any conflicting timelines by truncating the logs of the targets as needed.
Let us repeat the reasoning from above using this specific example:
* When C6 made its decision, that decision was based on its visibility. Even though it did not discover the most progressed timeline, its decision was valid because it satisfied the requirements of revocation and candidacy, which transitively satisfies the durability requirements.
* C7 also made a decision, but it did not discover the actions of C6. That means that C6 failed at reaching quorum. C7 had the authority to choose the most progressed timeline among the nodes it recruited, which it propagated to N6.
* C8 discovered artifacts of C6, but not of C7. That only means that C7 also failed at reaching quorum. However, C8 does not know that there was even a C7. From its point of view, it sees the work by C6. For safety, it has to assume that C6 might have reached quorum. So, it must honor every action taken by term 6. This time, C8 succeeds at reaching quorum.
* We finally come to C9, which may discover any combination of the above nodes. However, every combination is guaranteed to include the work done by C8.
In other words, we expect each term to make a safe decision. This remains true even if the decision conflicts with a previous term’s decision. For a new term, the only safe option is to honor the actions of the most recent term among those discovered.
This, in essence, is rule 2b(iii).
The timeline selection priority is as shown in Figure 6. If timeline 5568 was not discovered, 55557 would be chosen, and so on.
If N4 had applied its timeline before C9 intervened, the system would stay consistent, and the end result would remain the same. The only difference is that the applied indexes would be at different points.
As mentioned earlier, the action that supersedes these timelines must be either non-destructive or atomic. In other words, events A and B should not be deleted before accepting the new timeline. Instead, anything following A and B should be truncated, and any remaining events from the source should be appended after the truncation.
At this point, C9 can delegate its term to N4, allowing it to accept new requests.
# Intermission [#intermission]
This completes the expected parts of consensus systems that are traditionally required to prove correctness. However, we will discuss a few more points in upcoming blog posts. These are necessary for a consensus system to work effectively.
# Part 8: Changing the Rules (https://multigres.com/docs/consensus/part-08)
So far, we have analyzed fulfillment of requests and leadership changes for a consensus system. In reality, these two actions alone are not sufficient to maintain long-running clusters. In addition to these, we also need the following capabilities:
* Adding and removing nodes to the cohort.
* Changing the durability rules
* Adding and removing agents
The ability to add and remove agents is already satisfied since the proposed approach has no explicit constraints on them. However, there was an implicit assumption that the agents knew the current durability rules. If these rules are going to change, we need to discuss how the agents will learn about these changes and maintain the cluster's safety.
Conceptually, adding and removing nodes to the cohort is a change in the durability rules. We wanted to list them out separately because they are different use cases. Otherwise, the general approach of changing durability rules should work equally well for adding and removing nodes.
We will present two approaches for changing the durability rules.
# Policy vs Rules [#policy-vs-rules]
So far, we have not explicitly distinguished between the terms 'policy' and 'rules.' They are subtly different: A policy is an abstract requirement. For example, “I want my data to be in more than one AZ” is a policy. When the policy is combined with the list of nodes in the cohort, it results in a set of rules.
A change of policy may require you to install a new plugin. This type of change will be out of scope for this discussion. Any type of rule change that a single plugin can handle is in scope.
We will call this the `ruleset`. This ruleset must be known and understood by all agents. Additionally, since changes to rulesets are treated as distributed decisions, they must also reach quorum, which means that each cohort node must store the ruleset.
This also makes the cohort nodes the authoritative source for rulesets.
A coordinator can be initialized by pointing it at one of the nodes of the cohort. From that node, it can fetch the ruleset and the current term. Using this information, it can discover the rest of the nodes in the cohort.
# Coordinator method [#coordinator-method]
We can use the coordinator to modify the ruleset. For this, we have to interpret and apply the rules for the type of change we are making.
In this cluster, let us assume that we want to change the leadership rules for N1 from “both N2 and N3” to “either N2 or N3”. We will call them rs1 and rs2, respectively.
The coordinator performs the same actions as a leader change, but validates the recruited nodes for revocation and candidacy against both rulesets, as shown in Figure 1. Additionally, instead of inserting a standard `completion` event, it inserts a special `ruleset change` event.
Every node that receives and applies this ruleset change event updates its ruleset accordingly. If N1 is the new leader, it changes its behavior to “either N2 or N3” for all subsequent requests.
*It is actually sufficient if the coordinator satisfies rs1. However, recovery from subsequent failures will need to satisfy both rulesets. For uniformity, it is preferable to apply both rulesets to all situations.*
Figure 2 above shows an example where a coordinator in term 6 propagates N3 to N1 and N2, thereby satisfying rs1 and rs2 for N1’s candidacy.
### Corner case [#corner-case]
Suppose the coordinator made the ruleset change durable and delegated leadership to N1. This allows N1 to apply the change and proceed. At this stage, if N3 gets partitioned, N1 can still complete requests because it now uses rs2, which can be satisfied with an ack from N2.
Let's consider a scenario where a different coordinator (C7) assumes the system is still using rs1 and attempts to change leadership. From its perspective, recruiting N3 is enough to revoke N1’s leadership. This recruitment leads to the discovery of a pending ruleset change in the logs. This discovery informs the coordinator that it needs to recruit N2 to revoke N1’s leadership successfully.
There are two possibilities here:
#### Scenario 1 [#scenario-1]
After the new ruleset rs2 became durable, N1 gets promoted and completes additional requests. But it just uses N2 for its acks, which is sufficient to satisfy rs2.
C7 assumes rs1 is currently active. It recruits N3, which it thinks is sufficient to revoke N1’s leadership. However, it notices the ruleset change in the unapplied logs. Therefore, it must continue its revocation and also recruit N2. Recruitment of N2 leads to the discovery of a more progressed timeline. It must therefore propagate N2's timeline instead of N3’s timeline. In this case, it can use rs2 because the ruleset change has already been applied. At this point, it will realize that the minimum conditions are already met, and it could delegate leadership of the 7th term back to N1. N1 will eventually propagate the changes to N3. This scenario is shown in the animation below:
#### Scenario 2 [#scenario-2]
In this scenario, let us assume that no further progress was made after rs2 became durable.
The story starts off the same as scenario 1: C7 recruits N3, discovers the ruleset change, which makes it recruit N2. This time, it discovers the same timeline as N3. This allows it to append a completion event for the 7th term. However, the log now contains a mix of events from both rulesets. Therefore, its propagation must satisfy both rulesets: the requests must reach N1, N2 and N3. This scenario is shown in the animation below:
In reality, the coordinator would try to recruit all nodes. We presented it as a two-step process to demonstrate safety. If it could only recruit N3 and not N2, it would mean N2 was unreachable, which would cause the attempt to fail.
### Summary of rules [#summary-of-rules]
A coordinator that intends to change leadership must perform an initial discovery using its last known ruleset.
Among the discovered nodes, it must obtain the ruleset of the most advanced node. It must also inspect the logs for any changes to the ruleset. If changes are present, then its recruitment and propagation must satisfy the ruleset of the current node as well as the rulesets present in the log.
For this to work, we need to make one change to the node’s behavior: upon recruitment, the node should return the current term number *as well as the current ruleset*. The coordinator must correspondingly preserve the last known ruleset.
# Leader Method [#leader-method]
One problem with the coordinator method is that it is disruptive because it requires revoking the previous leadership. However, there is a way to implement this exact ruleset change with no disruption in traffic.
For this, we issue a request to the leader for the ruleset change. The leader fulfills this like any other request. The only difference is that the quorum rules for this specific request are expanded to include both rulesets. This is the same rule that was followed by the coordinator method. Once this is applied, the leader can switch to the new ruleset.
In this case, there is no change in the term number. Other than the different quorum rules, there is nothing special about this request.
If a failure occurs during this process, the above coordinator method can be used to appoint a new leader safely.
### Planned leadership change [#planned-leadership-change]
The request-based approach of changing rulesets can also be used to make planned leadership changes. In this case, we create a special request called `leadership change`, and the quorum rules are expanded to include those of the intended leader.
Once the request is successfully applied, the current leader can step down to be a follower. The intended leader will observe this event and promote itself as the leader. The followers will also start expecting requests from the new leader as they see this event.
Again, there is no need to start a new term number for this method.
### Adding and removing cohort nodes [#adding-and-removing-cohort-nodes]
Adding and removing cohort nodes are, in fact, a special case of a ruleset change. This is because the ruleset contains the list of cohort nodes. There are policies where the addition or removal of a node changes the quorum rules of a leader. A majority quorum is one such example. If so, that has to be taken into account while applying this special event.
# Part 9: Consistent Reads (https://multigres.com/docs/consensus/part-09)
Most official publications of consensus protocols have paid lip service to the issue of consistent reads. Implementors of these protocols have each developed their own methods for achieving consistency, and they all involve trade-offs.
The reason for this avoidance is that no solution is both perfect and performant. These properties determine the trade-offs.
* Consensus is a replicated system. There is no guarantee that a follower has the latest data.
* The current leader is guaranteed to have the latest data, but there is no guarantee that you know who the current leader is.
One important factor to keep in mind is that leader terms are expected to last a long time in the order of days. A planned leader change typically happens once a week. Unplanned leader changes might be even less common. This is a key factor to consider when choosing your solution.
At this point, we have the opportunity to reintroduce `observers`. These nodes are not part of the cohort, but they receive completed requests and can be used for reads. We will refer to the combination of followers and observers as replicas.
Here are a few approaches:
# Leader lease [#leader-lease]
The lease approach involves giving a leader a lease once appointed. During this period, the system will not revoke its leadership. The leader can renew its lease either by completing requests or through heartbeats. If a leader cannot renew its lease, it will stop serving reads before the lease expires.
There are a few disadvantages:
* We trust the clock.
* If the leader becomes unreachable, you have to wait till the lease expires before appointing another leader.
* We lose the opportunity to distribute reads across the replicas.
*For reference, Spanner supposedly uses this approach with a lease period of ten seconds.*
# Leader heartbeat read [#leader-heartbeat-read]
In this approach, the leader sends out heartbeats for every read. If a valid quorum of followers respond with the same term number, then it knows the leadership has not been revoked yet. It can respond to the read request.
Downsides:
* The cost of a read is as high as the network cost of completing requests.
* We lose the opportunity to distribute reads across the replicas.
# Log index based read [#log-index-based-read]
This method works for a single client. For each successful write request, the leader returns the log position of the request. The client can request a read from any replica by requiring it to wait until it reaches that position before serving the read.
An advantage of this approach is that reads can be load-balanced across multiple replicas.
Downsides:
* Replica lag or network partitions can impact read performance.
* Only the client that wrote the last request knows the latest position of its request.
# Replica heartbeat read [#replica-heartbeat-read]
This is a combination of the leader heartbeat read and the log index-based read. In this case, the read is sent to a replica, which sends out a heartbeat to the current leader and its quorum nodes. For its response, it must receive matching term numbers as well as the latest apply index from the leader. The replica waits until its own apply index reaches that of the leader, and then it can serve the read.
This allows reads to be load-balanced across multiple replicas. Also, this read works even if the client did not perform the last write.
Downsides:
* The cost of a read is as high as the network cost of completing requests.
* Replica lag or network partitions can impact read performance.
# Eventually consistent reads [#eventually-consistent-reads]
If the application can tolerate stale reads, those reads can be directed to any replica. There are many use cases where a certain level of staleness is acceptable. Based on this, we recommend setting a staleness tolerance and having the system reject reads that exceed this limit.
# Part 10: Addenda (https://multigres.com/docs/consensus/part-10)
# Health checks [#health-checks]
In a distributed system, there are no accurate methods of detecting failure. When a node becomes unreachable, it could be one of the following problems:
* It could be a network partition
* The node could have crashed
* The network could be too slow
Attempting to make decisions based on failure with an incorrect diagnosis may actually end up disrupting an already healthy system.
However, we must do the best we can.
We have previously stated that coordinators will perform health checks on all nodes in the cohort. We also assume that the coordinators are strategically positioned to handle expected failure scenarios. This approach offers several advantages because it allows us to draw reliable inferences.
### Responsibilities [#responsibilities]
Each coordinator must connect to all cohort nodes and perform regular health checks. This can be achieved either through polling or by having the nodes stream their health status at regular intervals.
During health checks, the coordinator can keep the current leader, term, and ruleset up to date.
Each leader must send regular heartbeats to all nodes in the cohort.
### Failure detection [#failure-detection]
This is a topic that requires its own study. However, Raft’s simple approach seems to have satisfied most deployments. The coordinator performing health checks is slightly better than Raft because it checks the health of all nodes before making a decision. In Raft, a follower makes a decision simply because it hasn’t received a heartbeat from the leader.
When the coordinator detects a failure, it must answer these two questions:
1. Is the leader able to complete requests? We determine the answer to this question using the following data:
* Is the leader itself reachable?
* Among reachable nodes, are they receiving heartbeats from the leader?
* Among those receiving heartbeats, are they enough for the leader to complete its requests?
* Are the nodes still completing requests from the leader?
* How long has this been going on?
2. Can the coordinator reach enough nodes to perform a leader change?
Answers to these questions should lead us through a decision tree where the outcome is either a decision to perform a leadership change or not to take any action.
# Term numbers [#term-numbers]
We previously promised that we would cover ways to generate term numbers. Here are some options:
### The Raft approach [#the-raft-approach]
Raft uses a clever method that lets nodes compete by using the same term number. The first coordinator to reach a majority of nodes wins that term number and gets permission to change leadership.
Those who do not win must wait for a timeout period and then try again using a higher term number. If the cluster is healed by that time, they have no action to take. This approach provides a mitigation for the livelock problem, where nodes can continuously race with each other, preventing anyone from succeeding.
The animation above is a reproduction of the one from the section on Revocation and Candidacy.
Applying the same approach to our pluggable durability, the coordinators do not need to reach a majority. If you examine the rightmost recruitment options, you will see that each option shares at least one node with every other option. This is a necessary property of recruitment.
We can utilize this property, similar to Raft, to have the coordinators compete against each other to recruit the necessary nodes for a leadership change. Whoever succeeds first wins the term. By definition, others must fail.
### Time [#time]
The current time can be used as a term number. There are a couple of risk factors associated with this:
* Timestamps can theoretically collide. Adding extra bits, such as a unique coordinator ID, may be necessary to ensure collision avoidance.
* Rogue clocks can accelerate by a vast margin. Such incidents will require human intervention to reset the system.
### etcd [#etcd]
You can use an external system, such as etcd, to acquire a lock and generate a unique, monotonically increasing number. This method also solves the livelock problem. Some might say that this is impure. But it is still a wise engineering choice.
# Alternate durability [#alternate-durability]
We previously discussed the need to revoke all possible leaderships for a safe leadership change. With this assumption, it is sufficient that a request reach any leader’s quorum. The current leader can consider that the request is durable and apply it.
If there is a failure, the act of global revocation will also discover any unapplied logs from the alternate group of nodes.
# Part 11: Recap (https://multigres.com/docs/consensus/part-11)
We covered a lot of ground in this series. We started with the following objectives:
* Propose an alternate and more approachable definition of consensus.
* Expand the definition into concrete requirements.
* Break the problem down into goal-oriented rules.
* Provide algorithms and approaches to satisfy the rules with adequate explanations to prove correctness and safety.
* Show that existing algorithms are special cases of this generalization.
Below is a summary of the topics we covered.
# Definition [#definition]
We introduced an alternate informal definition for a consensus system:
*A consensus system must ensure that every request is saved elsewhere before it is completed and acknowledged. If there is a failure after the acknowledgment, the system must have the ability to find the saved requests, complete them, and resume operations from that point.*
# Durability Policy [#durability-policy]
We declared that durability policies can be specified externally, such as a plugin. The algorithm should not have to change if the rules change. The rules have the following restrictions:
* The rules must depend on the current nodes in the cohort.
* Properties of cohort nodes (like AZ) can be used, as long as they are static.
* The rules cannot depend on external variables, such as time.
* Each leader can have different rules.
The ruleset data structure would conceptually look like this:
* List of cohort nodes
* List of eligible primaries, each containing:
* A list of node groups, where each node group is a valid durability combination
It may be possible to add more flexibility to the rules, but we think this is sufficient for most of today’s requirements.
# Governing Rules [#governing-rules]
We introduced a set of governing rules.
Using these rules as a foundation, we proposed multiple ways to achieve consensus by focusing on different sections of the rules. We also included existing approaches and explained how they adhered to the governing rules.
The rules are as follows:
### Definitions [#definitions]
* A consensus system executes a series of consistent distributed decisions made by multiple agents.
* A `decision` is an intent to make a change to the state of the system.
* An `agent` fulfills decisions.
### Rules [#rules]
1. Durability: Every decision is a distributed decision.
1. A distributed decision must be made durable.
2. A decision that has been made durable can be applied.
2. Consistency: Decisions must be applied sequentially.
1. Every agent must revoke the ability of all previous agents to make further progress before taking any action.
1. Inference: Every agent must provide a way for future agents to revoke its ability to make progress.
2. Every agent must discover decisions that were previously made durable, but not applied, and honor them.
1. There are situations where it will be impossible to know if a decision met the durability criteria. If so, the agent must honor such decisions because they might have been applied.
2. Decisions that get honored must be made durable and applied as a new decision made by the current agent (rule 1).
3. Inference: If an agent discovers multiple conflicting timelines, the newest one must be chosen.
These rules have a hierarchy. If you can satisfy the top-level rule, you do not have to follow the sub-parts. To accommodate all possible algorithms, the rules also avoid dictating any approach or implementation.
We demonstrated that these rules could be applied to the three types of `decisions` that a leader-based system would make:
* Fulfilling requests
* Changing leadership
* Changing durability rules
# Scoping [#scoping]
For the sake of practicality, we narrowed down the scope of our analysis when exploring solutions:
* We adopted Raft’s log replication as a requirement.
* We assumed a leader-based approach.
We re-introduced the following terminology from existing consensus protocols:
* A `leader` is an `agent`. It is a designated node in the cohort that is empowered to accept and complete requests. It continues to serve requests until its leadership is revoked.
* The rest of the nodes in the cohort are `followers`. Their role is to assist the leader in its workflow to make requests durable.
* `Observers` are nodes that are not part of the cohort. They are replicas that only accept requests that the leader completes.
# Coordinator [#coordinator]
We introduced a specialized agent called the `coordinator`. This separate role is necessary because a generalized approach allows for a large number of nodes in the cohort, making it impractical for every node to health check and respond to failures.
The coordinator is responsible for the following actions:
* Perform health checks on all the nodes of the cohort.
* In case of a failure, perform a failover by appointing a new leader and ensuring that requests that the previous leader might have applied are honored.
* A coordinator can optionally provide the functionality to perform a “planned leader change”.
Coordinators are not part of the cohort. Multiple coordinators can be deployed, and they do not need to be aware of each other’s existence.
In Raft, `leaders` are agents that fulfill requests, and `followers` act as agents when they choose to become candidates. In our approach, the task of changing leadership is taken on by the `coordinators` instead.
For small cohort sizes, nodes can take on the role of coordinators, just like in Raft.
# Term Numbers [#term-numbers]
We analyzed the problem of ordering in a distributed system. We concluded that assigning monotonically increasing and unique term numbers to each decision resulted in safer solutions.
The agents would use this term number to recruit nodes from older terms. If they succeed at recruiting a sufficient number of them to execute a leadership change, they move forward with the rest of the actions.
As a counterpoint, we demonstrated an engineering approach that utilized locks and timeouts to achieve ordering without relying on term numbers. However, it has trade-offs due to the reliance on clocks and execution time.
# A Raft inspired approach [#a-raft-inspired-approach]
We will now cover an example inspired by Raft that demonstrates one approach to implementing a system that can accommodate an externally specified durability policy.
As a bonus, we will also show how a change in durability rules can be trivially included as part of this algorithm.
This approach assumes that you are familiar with Raft. For brevity, we will skip over the common parts.
## Components [#components]
### Coordinator [#coordinator-1]
The coordinator does not need to persist any information. However, it needs either the current ruleset or a way to discover the existing cohort nodes to initialize itself. While active, it needs the following information:
* Term number
* Ruleset
### Node [#node]
A node needs to persist the following:
* Term number
* Ruleset
* A log that allows requests to be appended. You can also truncate the log at a specified point, which will cause all entries up to the end of the log to be deleted.
* Every log entry contains the term under which the request was made.
* An “applied index” that trails behind the end of the log. It is the point up to which it is safe to apply requests that are present in the log.
Term number rules for cohort nodes:
* A node must honor requests from an agent with a matching term number.
* A node must reject requests from an agent with a lower term number.
* In response to a recruitment, a node must agree to participate in a term number that is higher than the current one.
* A node responds to only one agent for a new term number. If another agent attempts to recruit the node with the same term number, it is rejected.
When recruited, each node returns the following information:
* The current log index
* The term number of the last log entry
* The current ruleset
* The list of ruleset changes in the unapplied parts of the log
### Ruleset [#ruleset]
The Ruleset is a data structure that is embedded in the node and persisted by it. Functionally, the ruleset must answer questions about durability, revocation, and candidacy. The following is an example of what a ruleset could look like.
* Name
* List of cohort nodes
* List of eligible primaries, each containing:
* A list of node groups, where each node group is a valid durability combination
## Completing requests [#completing-requests]
Unlike Raft, which validates durability by waiting for acks from a majority of followers, the generalized approach validates the acks against the ruleset. Aside from this, the entire algorithm remains the same.
If the request is a ruleset change, the durability rules must satisfy both the previous and the new ruleset. After the ruleset is applied, the leader can proceed with the newer ruleset.
## Leadership Change [#leadership-change]
In Raft, failure detection and leadership change are handled by individual nodes. In our generalized approach, separate coordinators perform these tasks. However, the actions taken by a coordinator still closely resemble those performed by the candidate in Raft. We do want to highlight how it “thinks” differently.
A coordinator that has decided to change leadership has the following goals:
* Obtain a term number
* Revocation
* Candidacy
* Discovery
* Propagation
* Establishment
Note that these are a restatement of rule 2, except that they are goal-oriented.
### Obtaining a term number, Revocation, Candidacy, and Discovery [#obtaining-a-term-number-revocation-candidacy-and-discovery]
A single step achieves the above four goals: The coordinator increments its current term number and sends a message to recruit all the nodes in the cohort.
For the nodes that were successfully recruited, it discovers the most progressed node:
* The log with the highest term number is the most progressed.
* For logs with identical term numbers, the one with the highest index is the most progressed.
From the most progressed node:
* It saves the ruleset returned by that node for subsequent attempts.
* If additional rulesets were returned, they are stored in a temporary list.
It validates revocation and candidacy by ensuring that the recruited nodes satisfy all the rulesets, which include the one returned by the node and the secondary list of in-flight ruleset changes:
* No leader of an older term must be able to complete any more requests.
* The nodes must contain a candidate (or the intended candidate) with a sufficient set of nodes needed to reach quorum.
If these criteria are not met, then the change of leadership cannot proceed. This can happen either because the coordinator was unable to reach all the necessary nodes or because a different coordinator recruited those nodes under the same term.
### Propagation [#propagation]
By now, the coordinator should have identified a candidate and the most progressed node. At this stage, the propagation mechanism can be the same as how Raft’s `AppendEntries` works. However, if there are multiple rulesets, the propagation must satisfy the quorum rules for all of them.
Raft requires that you can commit only when the log's term number matches the current term. We interpret this as an additional constraint on the durability requirements. This requires the entire timeline to become durable under the new term before it can be applied. It is an indirect way to satisfy Rule 2b(ii).
Propagation succeeds when the quorum rules for the candidate are met.
### Establishment [#establishment]
The coordinator moves the applied index to the end of the log and delegates its term number to the candidate. At this point, the candidate becomes the new leader.
# Conclusion [#conclusion]
We believe that we have satisfied our goal of generalizing consensus in the following ways:
* Accommodating arbitrarily complex durability requirements.
* Providing a set of governing rules that can be used for different approaches and implementations.
We also have a goal of implementing this approach in Multigres.
If you have any feedback or questions, please create a [Multigres discussion](https://github.com/multigres/multigres/discussions).
# Contributing (https://multigres.com/docs/contributing)
We are focused on getting the project to a stable state and then we will open it up to contributions.
For smaller contributions, such as documentation fixes, please open a pull request. You will need to follow instructions in the [GitHub Workflow](/docs/github-workflow) document.
## Ideas and Feature Requests [#ideas-and-feature-requests]
Use [GitHub Discussions](https://github.com/multigres/multigres/discussions) to share your ideas and feature requests.
## Bug Reports [#bug-reports]
[Open an issue](https://github.com/multigres/multigres/issues) on GitHub. Include reproducible information and any relevant error messages.
# Site contributions [#site-contributions]
You can make site contributions at this [Github repository](https://github.com/multigres/site). As mentioned above, please follow the [GitHub Workflow](/docs/github-workflow) document.
### Install dependencies [#install-dependencies]
```bash
pnpm i
```
### Local Development [#local-development]
```bash
pnpm start
```
This command starts a local development server and opens up a browser window. Most changes are reflected live without having to restart the server.
### Build [#build]
```bash
pnpm build
```
This command generates static content into the `build` directory and can be served using any static contents hosting service.
### Deployment [#deployment]
The site is automatically deployed when a Pull Request is merged into the `main` branch.
# GitHub Workflow (https://multigres.com/docs/github-workflow)
If you are new to Git and GitHub, we recommend to read this page. Otherwise, you may skip it.
Our GitHub workflow is a so called triangular workflow:
The Multigres code is [hosted on GitHub](https://github.com/multigres/multigres).
This repository is called *upstream*.
You develop and commit your changes in a clone of our upstream repository (shown as *local* in the image above).
Then you push your changes to your forked repository (*origin*) and send us a pull request.
Eventually, we will merge your pull request back into the *upstream* repository.
## Remotes [#remotes]
Since you should have cloned the repository from your fork, the `origin` remote
should look like this:
```
$ git remote -v
origin git@github.com:/multigres.git (fetch)
origin git@github.com:/multigres.git (push)
```
To help you keep your fork in sync with the main repo, add an `upstream` remote:
```
$ git remote add upstream git@github.com:multigres/multigres.git
$ git remote -v
origin git@github.com:/multigres.git (fetch)
origin git@github.com:/multigres.git (push)
upstream git@github.com:multigres/multigres.git (fetch)
upstream git@github.com:multigres/multigres.git (push)
```
Now to sync your local `main` branch, do this:
```
$ git checkout main
(main) $ git pull upstream main
```
Note: In the example output above we prefixed the prompt with `(main)` to
stress the fact that the command must be run from the branch `main`.
You can omit the `upstream main` from the `git pull` command when you let your
`main` branch always track the main `multigres/multigres` repository. To achieve
this, run this command once:
```
(main) $ git branch --set-upstream-to=upstream/main
```
Now the following command syncs your local `main` branch as well:
```
(main) $ git pull
```
## Topic Branches [#topic-branches]
Before you start working on changes, create a topic branch:
```
$ git checkout main
(main) $ git pull
(main) $ git checkout -b new-feature
(new-feature) $ # You are now in the new-feature branch.
```
As you work in a package, you can run just
the unit tests for that package by running `go test` from within that package.
When you're ready to test the whole system, run the full test suite with `make
test` from the root of the Git tree.
If you haven't installed all dependencies for `make test`, you can rely on the CI test results as well.
These results will be present on your pull request.
## Committing your work [#committing-your-work]
When running `git commit` use the `-s` option to add a Signed-off-by line.
This is needed for [the Developer Certificate of Origin](https://github.com/apps/dco).
## Sending Pull Requests [#sending-pull-requests]
Push your branch to the repository (and set it to track with `-u`):
```
(new-feature) $ git push -u origin new-feature
```
You can omit `origin` and `-u new-feature` parameters from the `git push`
command with the following two Git configuration changes:
```
$ git config remote.pushdefault origin
$ git config push.default current
```
The first setting saves you from typing `origin` every time. And with the second
setting, Git assumes that the remote branch on the GitHub side will have the
same name as your local branch.
After this change, you can run `git push` without arguments:
```
(new-feature) $ git push
```
Then go to the [repository page](https://github.com/multigres/multigres) and it
should prompt you to create a Pull Request from a branch you recently pushed.
You can also [choose a branch manually](https://github.com/multigres/multigres/compare).
## Addressing Changes [#addressing-changes]
If you need to make changes in response to the reviewer's comments, just make
another commit on your branch and then push it again:
```
$ git checkout new-feature
(new-feature) $ git commit
(new-feature) $ git push
```
That is because a pull request always mirrors all commits from your topic branch which are not in the `main` branch.
Once your pull request is merged:
* close the GitHub issue (if it wasn't automatically closed)
* delete your local topic branch (`git branch -d new-feature`)
# Multigres Upcoming features (https://multigres.com/docs)
Multigres is a project to build an adaptation of Vitess for Postgres.
This document is a loosely ordered tentative list of features we intend to build or import from Vitess.
## Proxy layer and connection pooling [#proxy-layer-and-connection-pooling]
Multigres will have a two-level proxy layer. In the case of a single small server, the primary benefit of these two layers would be connection pooling.
### Multigateway [#multigateway]
Multigateway is the top layer. In its simplest form, it will masquerade as a Postgres server. When it receives a query, it will forward the request to the next (Multipooler) layer. In the case of a single Postgres server, its primary usefulness is to shield the clients from restarts or failovers that may happen in the underlying layers due to software rollouts and failures.
In the case of primary-replica configurations, Multigateway can be configured to send read traffic to the replicas.
The Multigateway layer could be conceptualized as the compute layer of Multigres. It can be horizontally scaled as needed.
### Multipooler [#multipooler]
There will be one Multipooler per Postgres instance. Multipooler's primary function is to provide connection pooling. It will be aware of transaction state and connection specific changes of state, and will preserve the correct behavior to the clients connected at the Multigateway level.
Each Multipooler, its associated Postgres instance, and its storage are treated as one unit. This trio can be conceptualized as one node in the storage layer. In this layer, data can be replicated and/or sharded across other (trio) nodes as needed. Traditional storage layers typically implement a file or object store API. In the case of Multigres, the storage layer API is that of a database.
## Sharding [#sharding]
As the data grows, you will soon encounter the need to split some tables into shards. When this need arises, Multigres will manage the workflows to split your monolithic Postgres instance into smaller parts. Typically, the smaller tables will remain in the original Postgres database, and the bigger tables will be migrated out into multiple (sharded) databases.
Multigres will provide a powerful relational sharding model that will help you keep tables along with their related rows in the same shard, providing optimum efficiency and performance as your data grows.
TODO: Doc on Vitess sharding model
### Multigateway (Sharding) [#multigateway-sharding]
In a sharded setup, Multigateway's functionality will expand to present this distributed cluster as if it was a single Postgres server. For simpler queries, it will just act as a routing layer by sending it to where the data is. For more complex queries, it will act as a database engine while maximally leveraging the capabilities of the individual databases underneath.
Multigateway will have the ability to push entire join queries into an underlying shard if it determines that all data for that join is within that shard. Similarly, Multigateway will "scatter" entire joins across all shards if it determines that the related rows of a join are within their respective shards.
### Multipooler (Sharding) [#multipooler-sharding]
The query serving functionality of Multipooler will have no awareness of sharding. It will just use the underlying Postgres instance to serve the requested queries. However, Multipooler will be the workhorse behind facilitating all the resharding efforts.
When the need to reshard arises, new (target) Multipoolers will be created to receive filtered data from the original unsharded table. This data will be streamed by the source Multipooler. Once these tables are populated and up-to-date, a failover will be performed to move traffic to the sharded Multipoolers. For safety, the replication will be reversed. If any issue is found after the cut-over, it can be undone by switching traffic back to the source tables.
This failover and fail-back can be repeatedly performed without requiring any change in the application.
### 2PC [#2pc]
Due to the flexible sharding scheme of Multigres, you should be able to minimize or completely eliminate the need for distributed transactions by a careful selection of an optimal sharding scheme. However, if the need arises, Multigateway and Multipooler will work together and use the two-phase commit protocol supported by Postgres to complete transactions that span across multiple instances.
TODO: Doc on 2PC atomicity and isolation
## Multiorch: NVME performance, durability and High Availability [#multiorch-nvme-performance-durability-and-high-availability]
Multigres allows for the Postgres data files to be stored on the local NVME. For an OLTP system like Multigres, a local NVME has substantial advantages over a mounted drive:
* IOPS are free
* Lower latency, by one order of magnitude
These advantages translate into higher performance and reduced cost.
Multigres will implement a consensus protocol to ensure that transactions satisfy the required durability policy. This will be achieved using a two-phase sync to replicate the WAL to the required number of replicas in the quorum. This functionality eliminates the need to rely on a mounted (and replicated) file system like EBS to safeguard from data loss.
In case of a primary node failure, Multiorch (Multigres Orchestrator) will promote an existing replica to primary, ensuring that it contains all the committed transactions of the previous primary. Following this, Multigres will resume serving of traffic using the new primary. Essentially, this solves the problem of durability and high availability.
The above approach aligns with the conceptual view of the Multipooler+Postgres+Data trio as being part of the data layer. Hence, there is no need to rely on yet another (mounted) data layer underneath Postgres.
It is certainly possible to run Multigres on a mounted drive. It is just not necessary.
## Cluster management [#cluster-management]
Multigres will come equipped with a variety of tools and automation to facilitate the management of your cluster. This tooling is capable of scaling to tens of thousands of nodes, and can be spread across multiple regions worldwide.
### Automated Backups [#automated-backups]
Multigres can automatically perform backups of your Postgres instances on a regular basis to a centralized location.
### New Multipoolers [#new-multipoolers]
Adding more replicas to a Postgres instance will be as trivial as launching a Multipooler with the right command line arguments. The Multipooler will seek the latest available backup, restore it, and point the Postgres instance to the current primary. Once the replication has caught up, it will open itself up to serve read traffic.
Multipoolers can be of different types: They can be quorum participants, in which case they will follow the two-phase sync protocol during replication. If not, they will just be "observers", to just serve additional read traffic.
Bringing up a Multipooler as a quorum participant will automatically make it join the quorum after it has fully caught up on replication.
Multipoolers will have configurable user-defined tags that will allow Multigateways to split traffic based on differing workloads. For example, you may tag a group of Multipoolers as "olap". Workloads that would like to perform analytics queries would specify "dbname\@olap" as the database to connect to. This will inform the Multigateways to redirect these reads to only Multipoolers that export the "olap" tag.
### Replication [#replication]
Beyond managing the consensus protocol for Multipoolers in the quorum, Multiorch will also monitor all Multipoolers and ensure that they are replicating from their respective primaries. If a connection is lost for any reason, Multiorch will restore it.
### Cross-zone clusters [#cross-zone-clusters]
Multigres can be deployed across a large number of nodes distributed across the world. This is architecturally achieved by a star configuration in the topology. There will be a global topology server (etcd). It will contain information that changes infrequently, like sharding info, etc. This information will be propagated into cell-specific topology servers (local etcds), one for each cell. Multigateways and Multipoolers will be deployed in a cell, and they'll use the local topology to serve traffic for that cell. In case of a network partition, a cell will be capable of continuing to serve traffic for as long as the replication lag is tolerable.
Of course, there can exist only one Primary per shard. Any cell that does not host the primary database is meant to serve read-only traffic that is replicated from the primary.
### Cloud-Native [#cloud-native]
Multigres will be cloud-native. This means that it can run in a cloud framework like Kubernetes:
* All components are restartable without causing disruption to the overall system. This needs to be within a reasonable disruption budget, which is configurable.
* Components can be migrated to different nodes.
* Components can use mounted persistent storage, and will automatically reattach if moved.
* New components can be launched to increase capacity. They will initialize themselves correctly, and join the system to serve traffic. Conversely, components can be removed in order to scale down.
* Specific to databases: Primary failures and restarts will be handled automatically by a watchdog process (Multiorch) that can perform a failover to an up-to-date replica.
* Scale to zero: You can shut down all components. This will result in just the metadata and backups being preserved, with no active components to serve any traffic. Adding the serving components back to the system will bootstrap the cluster to an operational state.
Multigres will come with a Kubernetes Operator that will translate components described using Multigres terminology like cells, shards and replicas into corresponding Kubernetes components.
We intend to develop a way for Multigres to use local storage in order to leverage the proximity of Postgres and its data files.
TODO: Doc to elaborate on local NVME
## Materializer [#materializer]
The next few features described below will all be powered by a primitive called Materializer. This primitive is essentially a stream processor capable of materializing a source of tables into a target set of tables. The rules of materialization can be expressed as an SQL statement. The only limitation is that the expression is stream processable. In other words, we should be able to incrementally and independently apply every change event to the target. For example, a `count(*)` is stream processable, whereas a `max(col)` is not.
This materialization can happen without causing any downtime to the source database. Once the target tables are caught up to the sources, Materializer will verify correctness of the target tables by performing a diff.
If the materialization expression is non-lossy (reversible), you can atomically switch traffic from the source tables to the target. At this point, Materializer can reverse the replication to keep the source tables up-to-date. You can go back and forth indefinitely. This ability allows you to undo a migration if problems are encountered after a cutover.
Materializer streams operate orthogonally to each other. For example, resharding uses Materializer, but a simultaneous table materialization will correctly migrate to use the new shards after resharding.
Materializer works in conjunction with Multigateway routing rules that allow you to smoothly and safely transition traffic from source to target.
## Migrations [#migrations]
### Migrate tables from anywhere to anywhere [#migrate-tables-from-anywhere-to-anywhere]
MoveTables will be a workflow built using Materializer. This will allow you to migrate a group of tables from any source to any target. This can be used to split a database into smaller parts, or merge two databases into one. Let us assume that you want to migrate table `t` from database `a` to database `b`.
* Initially, the Multigateways will have a rule to redirect traffic intended for `a.t` and `b.t` to `a.t`. MoveTables will setup these routing rules automatically before starting Materializer.
* Once this rule is setup, you can start refactoring your application to write to `b.t` instead of `a.t`. Writing to any of these tables will get redirected to `a.t`.
* Once the data is verified to be correct, we can switch the Multigateway routing rules to send traffic to `b.t` instead. MoveTables will have a subcommand to do this safely. If the application has not finished refactoring all the code, it's ok, because all traffic will flow to `b.t`.
* At this time, Materializer will reverse the replication to keep `a.t` up-to-date. If any problem is detected after the cutover, you can fall back to `a.t`. This back and forth can be repeated as often as necessary.
* After we are certain that all problems are resolved, and we have verified that the application refactor is complete, we can use the MoveTables clean up command to drop the reverse replication, routing rules, and source table `a.t`.
Of course, this workflow would require Multigateways to have access to both `a` and `b`.
### Migrate across Postgres versions [#migrate-across-postgres-versions]
Materializer will rely on Postgres logical replication. Due to this design, we can also use it to safely migrate from one version of Postgres to another without incurring downtime, and with the ability to revert a migration, just like the case of a table migration.
This will essentially be a MoveTables, but with the source and target running different versions of Postgres.
### Resharding [#resharding]
Materializer will be used for resharding. For this purpose, Multigres will use the filtering ability of Materializer to split sharded (or unsharded) tables into their target shards based on the target's sharding key ranges.
Just like the other cases, the ability to verify correctness and revert will be available.
### Change sharding key [#change-sharding-key]
It may happen that the original sharding key you chose for the table was suboptimal, or it may be possible that the application workload has changed substantially. In such cases, you can use Materializer to change the sharding key of a table to one that is more optimal to the current workload.
### Migrate from MySQL [#migrate-from-mysql]
If you are currently running on a MySQL database, or Vitess, Multigres will allow you to migrate from that source. In this situation, additional work may have to be done by you in the area of how you switch traffic. This is because Multigres will only support the Postgres protocol, and will not be able to orchestrate the traffic meant for the source MySQL.
### Exotic Migrations [#exotic-migrations]
Materializer is versatile enough that it can be used to address migration use cases that have not been thought of yet. For example, you could script it to merge tables from different databases into one. This can be a necessity if you decide to merge many multi-tenant databases into a single sharded one, or vice-versa.
## Materialized Views [#materialized-views]
Multigres materialized views will be built using Materializer. Such views are physical tables that are updated in real-time by processing events from the WAL. Consequently, the complexity of the SQL expression for these views is limited to operations that can be incrementally processed from the change events. For example, `count(*)` will be supported, while `max(col)` will not be. Another trade-off of materialization is that it is eventually consistent.
In a sharded environment, materialized views can greatly enhance the performance of certain join operations, if the eventual consistency trade-off is acceptable:
* For a table that has foreign keys into two different tables, you can have the source table be sharded by the first foreign key, and you can materialize the target table using the second sharding key. This approach will allow you to efficiently join this table with either of the two other tables, because both those joins will be in-shard joins.
* Reference tables: Many databases have smaller tables that have slow changing data. However, they may need to be joined with other massive tables that need to be sharded. In this situation, you can materialize the small table into all the shards of the bigger table. Such a join will then become an in-shard join.
## Schema Deployment [#schema-deployment]
Some schema changes are inherently time-consuming, along with the downside of locking the table from other changes. In such situations, Multigres will use Materializer to materialize the post-schema version of the table within the same database. Once the materialization is complete, it will swap the old table with the new one, thereby achieving a non-blocking, zero-downtime deployment.
Additionally, the materialization will be reversed. If the schema deployment causes problems, you have the option to instantly revert with no data loss.
## Change Data Capture [#change-data-capture]
Materializer will be made up of two parts: the part that streams from the source (MStream), and the part that writes to the target (MPlayer). MStream will also speak the Postgres logical replication protocol. This will allow you to integrate multigres with any tooling that can consume such a stream.
If you use MStream, you do not need to perform a dump and restore. Instead, you can initiate a stream "from the beginning of time". With this directive, MStream will provide all the events necessary to completely materialize the source into the target.
As your database grows into multiple shards, you can run individual MStreams for each shard for better parallelization.
## Observability [#observability]
Apart from exporting real-time metrics, Multigres will also have an extensive toolset to facilitate troubleshooting when the system exhibits unexpected behavior. These will be exported by Multigateway and Multipooler:
* Standard metrics like QPS, latencies, and error rates.
* Per-table metrics.
* Normalized per-query metrics. In this case, each query is normalized by stripping out values and it becomes a key to its own metric. This helps identify the worst performing queries. This can lead to an explosion of values. To help make this optional, this metric is served in a separate human readable end point.
* On-demand real-time query or transaction logs. This is useful to get a snapshot of everything that's currently in progress, useful if there's an active incident.
* Error logs.
* Trace points.
## Messaging [#messaging]
The Messaging feature will implement transactional message queues that can guarantee that every message is processed. The message queue will keep resending unacknowledged messages while exponentially backing off, until an acknowledgement is received.
Rows can be transactionally inserted into a message table as part of a larger transaction that involves other tables.
## Database Protection [#database-protection]
Multigres will implement the ability to enable various database protection features:
* Query Killer: Terminate a query if it takes too long.
* Transaction Killer: Rollback a transaction if it takes too long.
* Result limiter: Return an error if the number of rows in a query exceed a threshold.
* Result consolidator: If multiple identical read queries are simultaneously sent to the database, only one query is executed, and the results are shared across all other requests.
If any of these features are already available in Postgres, Multigres will utilize them under the covers.
## Tooling [#tooling]
### Multiadmin [#multiadmin]
Multiadmin will be a dashboard that allows you to view the various components of a live cluster, like the currently running Multipoolers, browse to their status pages, etc.
# Multigres Operator (https://multigres.com/docs/multigres-operator)
## What is an operator? [#what-is-an-operator]
Multigres is deployed using a Kubernetes operator. An operator is a Kubernetes controller that runs inside your cluster and manages the lifecycle of a complex application on your behalf. Instead of manually applying configuration changes, scaling resources, or handling failures, you declare the state you want and the operator continuously works to make it so.
The Multigres operator is what turns a single YAML declaration into a fully orchestrated, multi-zone Postgres cluster.
## What the operator manages [#what-the-operator-manages]
The operator manages pods — the individual containers running each component — directly, rather than delegating to Kubernetes StatefulSets. This means it can be primary-aware: it knows which pod holds the active Postgres primary at any given moment.
Primary-awareness ensures the operator always acts on standbys first and the primary last when restarting pods for a config change or scaling event, guaranteeing safe operations without interrupting writes.
A Multigres cluster is made up of several components that work together across availability zones. The operator provisions all of the following from a single `MultigresCluster` manifest:
* **GlobalTopoServer** — A managed etcd cluster that records topology state: which databases exist, which cells they live in, and where every component is registered. In a multi-zone deployment, the topology splits into a global server for cluster-wide state and per-cell servers for local discovery, so a partitioned cell keeps operating against its own local view.
* **Multiadmin** — The management plane for the cluster, including a web UI.
* **Multigateway** — Speaks the Postgres wire protocol. Applications connect to a gateway, which forwards queries to the right Multipooler over gRPC. Adding more gateways scales connection capacity horizontally.
* **Multiorch** — The orchestrator. One set of Multiorch instances per shard, running across cells. Watches replication health, appoints leaders through a consensus protocol, runs failovers, and coordinates bootstrap. When you apply the manifest, Multiorch runs bootstrap as a consensus-backed election — the same code path as every later failover — so there is no separate provisioning script that can race with itself.
* **Pools** — The Postgres pods themselves, managed by pgctld (which owns the local Postgres process) and Multipooler (which owns the connection pool). One pooler per Postgres instance.
### Prerequisites [#prerequisites]
Before deploying a Multigres cluster on AWS EKS, ensure the following are in place:
* **AWS CLI v2** — installed and configured
* **kubectl** — configured for the target EKS cluster
* **Permissions** — to create namespaces, CRDs, pods, services, PVCs, secrets, and to install the Multigres operator
***
# Getting Started on EKS [#getting-started-on-eks]
This guide does not cover creating an [EKS cluster](https://docs.aws.amazon.com/eks/latest/userguide/getting-started.html), installing the [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html), or configuring [IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/introduction.html) from scratch.
## Deployment Shape [#deployment-shape]
Multigres can run on a single Availability Zone or across multiple Availability Zones, as long as the cluster satisfies the minimum quorum requirements for the configured topology.
> **Note:** For EKS deployments, begin with the smallest topology that satisfies your availability requirements and validate operational workflows before expanding the deployment. At minimum, validate cluster creation, backup and restore, failover behavior, scaling, and cleanup in your own EKS environment.
## Setup Checklist [#setup-checklist]
Complete these steps before applying the `MultigresCluster` manifest:
| Step | Command or check |
| -------------------------------------- | ------------------------------------------------------------------------------------------------------------------ |
| Install the Multigres operator | `kubectl apply -f https://github.com/multigres/multigres-operator/releases/download/v0.1.0/install.yaml` |
| Choose a namespace | `kubectl create namespace multigres-demo` |
| Confirm persistent volume provisioning | `kubectl get storageclass` |
| Confirm EKS zone labels | `kubectl get nodes -L topology.k8s.aws/zone-id,topology.kubernetes.io/zone` |
| Prepare an S3 backup bucket and prefix | Create or select a bucket, choose a unique `keyPrefix`, and grant the backup service account access to that prefix |
| Create the PostgreSQL password secret | `kubectl create secret generic multigres-admin-password --from-literal=password=''` |
| Apply the `MultigresCluster` manifest | `kubectl apply -f multigres-demo.yaml` |
| Verify the cluster with a SQL query | Run the `psql` check in [Step 4: Verify the Cluster](#4-verify-the-cluster) |
## Step 1: Install the Operator [#step-1-install-the-operator]
Install the operator version that matches the Multigres release you want to run. For v0.1.0 Multigres Alpha:
```bash
kubectl apply -f https://github.com/multigres/multigres-operator/releases/download/v0.1.0/install.yaml
```
To use the latest available release:
```bash
kubectl apply -f https://github.com/multigres/multigres-operator/releases/latest/download/install.yaml
```
> **Note:** For reproducible deployments, prefer a versioned release URL over `latest`.
Wait for the operator to become ready:
```bash
kubectl get pods -n multigres-operator
```
### Image Selection [#image-selection]
The `MultigresCluster` example below intentionally does not set `spec.images`. When image fields are omitted, the installed operator uses its default component images for that operator release.
Use the matching operator release for the Multigres release you want to test. Only set `spec.images` when you deliberately need to override the release defaults, for example while testing a custom build.
Optional image override shape:
```yaml
spec:
images:
multigateway: ghcr.io/multigres/multigres:
multiorch: ghcr.io/multigres/multigres:
multipooler: ghcr.io/multigres/multigres:
multiadmin: ghcr.io/multigres/multigres:
multiadminWeb: ghcr.io/multigres/multiadmin-web:
postgres: ghcr.io/multigres/pgctld:
```
After the cluster is created, you can inspect the actual images selected by the operator:
```bash
kubectl get pods -o jsonpath='{range .items[*]}{.metadata.name}{" "}{range .spec.containers[*]}{.name}{"="}{.image}{" "}{end}{"\n"}{end}'
```
## Step 2: Choose a Namespace [#step-2-choose-a-namespace]
Create a namespace for the Multigres cluster:
```bash
kubectl create namespace multigres-demo
```
Use the same namespace for the PostgreSQL password secret, backup service account, and `MultigresCluster` resource. Set it as your default context namespace:
```bash
kubectl config set-context --current --namespace=multigres-demo
```
## Step 3: Configure Storage [#step-3-configure-storage]
Multigres poolers require persistent volumes. On EKS, install and configure the AWS EBS CSI driver before creating a Multigres cluster. Verify that dynamic volume provisioning is available:
```bash
kubectl get storageclass
```
The reference manifest uses a `gp3` StorageClass named `multigres-gp3`. If your cluster already has a suitable default StorageClass, you can omit the explicit storage class from the manifest.
## Step 4: Confirm Zone Labels [#step-4-confirm-zone-labels]
Multigres cells map to Kubernetes topology labels. On EKS, verify that nodes expose AWS zone IDs:
```bash
kubectl get nodes -L topology.k8s.aws/zone-id,topology.kubernetes.io/zone
```
Use the `topology.k8s.aws/zone-id` values for the Multigres `zoneId` fields in the manifest. The supported zones for the shared demo cluster are:
| Zone ID | AWS AZ |
| -------- | ---------- |
| use1-az1 | us-east-1a |
| use1-az2 | us-east-1b |
| use1-az6 | us-east-1d |
## Step 5: Prepare S3 Backups [#step-5-prepare-s3-backups]
Create or select an S3 bucket for backups. Use a unique `keyPrefix` for each cluster if you plan to recreate clusters or run multiple clusters in the same bucket.
The backup service account must be able to read, write, list, and delete objects under the configured backup prefix. On EKS, use IAM Roles for Service Accounts (IRSA) to grant the Kubernetes service account access to S3.
Create the service account:
```bash
kubectl create serviceaccount multigres-backup
```
If using IRSA, annotate the service account with the IAM role that has S3 access:
```bash
kubectl annotate serviceaccount multigres-backup \
eks.amazonaws.com/role-arn=arn:aws:iam:::role/
```
At minimum, the backup identity needs the following S3 permissions:
* `s3:ListBucket` and `s3:GetBucketLocation` on the bucket, scoped to the configured prefix
* `s3:GetObject`, `s3:PutObject`, `s3:DeleteObject` on objects under the prefix
Example IAM policy:
```json
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": ["s3:ListBucket", "s3:GetBucketLocation"],
"Resource": "arn:aws:s3:::your-backup-bucket",
"Condition": {
"StringLike": {
"s3:prefix": ["multigres-demo/", "multigres-demo/*"]
}
}
},
{
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
"Resource": "arn:aws:s3:::your-backup-bucket/multigres-demo/*"
}
]
}
```
## Step 6: Create the PostgreSQL Password Secret [#step-6-create-the-postgresql-password-secret]
Create a Kubernetes secret containing the PostgreSQL superuser password:
```bash
kubectl create secret generic multigres-admin-password \
--from-literal=password='change-this-password'
```
Replace `change-this-password` with the password you want Multigres to use for the `postgres` user during cluster initialization. The manifest references this secret via:
```yaml
postgresPasswordSecretRef:
name: multigres-admin-password
key: password
```
***
# Deploying the Operator on EKS [#deploying-the-operator-on-eks]
[](https://www.youtube.com/watch?v=ds0bdNlaAoQ)
## 1. Configure Kubernetes Access [#1-configure-kubernetes-access]
Update your kubeconfig to point at the EKS cluster:
```bash
aws eks update-kubeconfig \
--name multigres-dev \
--region us-east-1 \
--profile eks-demo
```
Set your default namespace:
```bash
kubectl config set-context --current --namespace=eks-demo
```
Verify access:
```bash
kubectl get pods
kubectl get multigresclusters
kubectl get coretemplates,celltemplates,shardtemplates
```
You should see no running pods, no clusters, and the default templates present. The operator is installed and ready.
## 2. Create the Cluster Manifest [#2-create-the-cluster-manifest]
Create a file named `demo-multi-az.yaml` with the following contents. Replace placeholder values as indicated:
```yaml
apiVersion: multigres.com/v1alpha1
kind: MultigresCluster
metadata:
name:
namespace:
spec:
pvcDeletionPolicy:
whenDeleted: Delete
whenScaled: Delete
durabilityPolicy: AT_LEAST_2
templateDefaults:
coreTemplate:
cellTemplate:
shardTemplate:
postgresPasswordSecretRef:
name:
key: password
backup:
type: filesystem
filesystem:
path: /backups
storage:
size:
images:
imagePullPolicy: Always
# Same Multigres runtime image used by all Multigres components.
multiadmin: /multigres:
multigateway: /multigres:
multiorch: /multigres:
multipooler: /multigres:
# pgctld/Postgres image.
postgres: /pgctld:
# Web UI image.
multiadminWeb: ghcr.io/multigres/multiadmin-web:
cells:
- name:
zoneId:
- name:
zoneId:
- name:
zoneId:
databases:
- name: postgres
default: true
backup:
type: s3
s3:
bucket:
region:
keyPrefix:
serviceAccountName:
tablegroups:
- name: default
default: true
shards:
- name: 0-inf
spec:
pools:
default:
replicasPerCell:
storage: {}
multipooler:
resources: {}
postgres:
resources: {}
multiorch:
resources: {}
```
The manifest defines three cells, each mapped to a real AWS availability zone. `AT_LEAST_2` durability means every committed write is acknowledged by at least one standby before the client receives confirmation. The operator infers resource limits, gateway configuration, and topo server sizing from the referenced templates.
## 3. Apply the Manifest [#3-apply-the-manifest]
```bash
kubectl apply -f demo-multi-az.yaml
```
Watch the pods come up:
```bash
kubectl get pods -w
```
You should eventually see pods for:
* `global-topo`
* `multiadmin`
* `multigateway` (one per zone)
* `multiorch` (one per zone)
* pool pods (one per zone)
Within a minute the cluster should be fully running.
## 4. Verify the Cluster [#4-verify-the-cluster]
Check the Multigres resources:
```bash
kubectl get multigresclusters
kubectl get shards
```
Check the pods:
```bash
kubectl get pods
```
A healthy cluster with minimum size should have one primary and two replicas. Run this to check directly:
```bash
PW=$(kubectl get secret -o jsonpath='{.data.password}' | base64 -d)
for p in $(kubectl get pods -o name | grep '-postgres-default-0-inf'); do
echo "== $p =="
kubectl exec "$p" -c postgres -- env PGPASSWORD="$PW" \
psql -h /var/lib/pooler/pg_sockets \
-U postgres -d postgres -Atc \
"select case when pg_is_in_recovery() then 'replica' else 'primary' end"
done
```
Expected result:
```
one primary
two replicas
```
### Verify Query Serving via Multigateway [#verify-query-serving-via-multigateway]
```bash
kubectl run psql-client \
--rm -it --restart=Never \
--image=postgres:17 \
--env="PGPASSWORD=$(kubectl get secret -o jsonpath='{.data.password}' | base64 -d)" \
-- psql -h -multigateway \
-U postgres -d postgres \
-c "select 1, current_database(), current_user;"
```
Expected result:
```
?column? | current_database | current_user
----------+------------------+--------------
1 | postgres | postgres
```
## 5. Delete and Recreate a Cluster [#5-delete-and-recreate-a-cluster]
Use this if you want to start over.
Delete the cluster:
```bash
kubectl delete multigrescluster --ignore-not-found
```
Wait until the old pods disappear:
```bash
kubectl get pods | grep || true
```
Delete all PVCs for the cluster, including Postgres and topo/etcd data:
```bash
kubectl get pvc -o name \
| grep 'persistentvolumeclaim/data--' \
| xargs -r kubectl delete
```
Verify no PVCs remain:
```bash
kubectl get pvc | grep || true
```
Clear the S3 backup prefix before recreating a cluster with the same name and prefix:
```bash
aws s3 rm s3:/// --recursive
```
Verify the S3 prefix is empty:
```bash
aws s3 ls s3:/// --recursive
```
Apply the manifest again:
```bash
kubectl apply -f .yaml
```
## 6. How Applications Connect [#6-how-applications-connect]
Applications running inside the Kubernetes cluster connect to the Multigres gateway Service:
```
postgresql://postgres:@demo-multi-az-multigateway:5432/postgres
```
If the application runs in another namespace, use the full Service name:
```
postgresql://postgres:@demo-multi-az-multigateway.eks-demo.svc.cluster.local:5432/postgres
```
> **Note:** `*.svc.cluster.local` names are Kubernetes-internal DNS names and will not resolve from your laptop shell.
## 7. Scaling [#7-scaling]
To start:
```yaml
replicasPerCell: 1
```
This creates one pooler per cell — three poolers total across three AZs.
To scale, edit the `MultigresCluster` directly. Do not edit the generated `Shard`.
Example:
```bash
kubectl patch multigrescluster \
--type=json \
-p='[
{
"op": "replace",
"path": "/spec/databases/0/tablegroups/0/shards/0/spec/pools/default/replicasPerCell",
"value": 2
}
]'
```
Set the value back to `1` for the basic demo shape.
# MVP Project Plan (https://multigres.com/docs/mvp)
## Approach [#approach]
The first goal of Multigres is to build an MVP. For this, there are three approaches:
1. Retrofit Postgres into a fork of Vitess
2. Build from the ground up using Vitess as a reference
3. Hybrid: Build from scratch, and copy what you can from Vitess
We'll be using approach 3. Reasons:
* Approach 1 will likely get us to an MVP the fastest. However, we'll be continuously fighting against mysql-isms in the code. We think this will eventually result in a product that is subpar.
* Approach 2 could likely be the cleanest, but also the slowest. We'll be reinventing the wheel for many major features for which solutions already exist in Vitess.
* Approach 3 will take a bit longer than approach 1, but it will be clean. It will be faster than approach 2 because there are many substantially large parts of Vitess that are designed to be database agnostic. Those parts can be copied as is into Multigres.
For approach 3, we intend to leave behind:
* Anything that is MySQL specific
* Any legacy features that are not needed any more
* Anything that was not well implemented
Multigres will have the following project tracks, and they will be worked on in parallel to the extent possible. The tracks are listed in the expected order of MVP completion. These parts are independently useful and can be deployed into production as they become ready.
## Proxy [#proxy]
* Multipooler: Connection pooling
* Multigateway: Postgres and PostgREST protocols, discovery of Multipoolers, route traffic to primary or replica Multipoolers, load balance replica traffic
The MVP must be able to scale to support tens of thousands of connections per Multigateway.
Each pool of Multigateways should be able to support up to ten thousand databases.
## Cluster management [#cluster-management]
The purpose of cluster management is to minimize human intervention by automating mundane tasks.
* Initialize a new database
* Automated backups of databases and WALs
* Add (and remove) replicas, will use backups and WALs to make them catch up to the cluster, and publish to Multigateway about readiness to serve traffic
* Deactivate a database (scale to zero) and bring up a previously deactivated database
* Kubernetes Operator
Planned cluster management operations should have no impact to user traffic.
## Performance, Durability and HA [#performance-durability-and-ha]
These three features go hand in hand. Storing the data in a local NVME drive yields the best performance for an OLTP system like Postgres. However, this can lead to data loss in case of a node failure. This can be solved by implementing a consensus protocol. The side benefit of a consensus protocol is that it also helps us address the problem of High Availability, because there will always be an up-to-date replica if the primary node fails.
### Postgres [#postgres]
The existing Postgres primitives are insufficient to build a robust consensus protocol. Multigres will make changes to Postgres to implement a two-phase sync mechanism. A number of consensus protocols can be built once this functionality is in place. We will also work towards getting this change accepted upstream.
### Multigres [#multigres]
Multigres will build the coordination part of the consensus protocol using a brand new Multiorch. This will not be ported from Vitess, because the Multiorch from Vitess has a large amount of legacy code that was inherited from the MySQL Orchestrator.
Multiorch will operate as a coordinated cluster across failure zones to ensure that at least one of them can perform a successful failover if there is a network partition.
Functionally, Multiorch will be the same as its Vitess counterpart:
* Elect a primary if none exist
* Perform smooth primary changes if requested
* Detect failures and failover as needed
* Rewire replicas and observers if they lose their connection to the primary, or if their connection to the primary needs to be updated
* Cooperate with other Multiorchs to ensure that they don't step on each others' actions.
* Honor a variety of durability policies for each cluster.
## Materializer [#materializer]
The Postgres logical replication is functionally very close to MySQL's row-based binlog replication. Hence, most of Materializer will be copied from Vitess, and changes will be made to address any mismatch.
The MVP will support:
* Resharding
* MoveTables
* Migrations
* Materialization
## Sharded query serving [#sharded-query-serving]
Sharded query serving has an extensive list of constructs to support. The MVP will support the following:
* SELECT, INSERT, UPDATE and DELETE that can be executed within a shard. This should cover all multi-tenant use cases.
* 2PC for transactions that span across shards.
* Stored procedures that can be executed within a shard.
The above functionality already exists in Vitess. The primary challenge will be to reconcile the Vitess parser and the associated data structures against the Postgres syntax.