Internals of clustering

The Core instances used as Primary servers in a cluster use the Raft protocol to ensure consistency and safety. See Advanced Causal Clustering for more information on the Raft protocol. An implementation detail of Raft is that it uses a Leader role to impose an ordering on an underlying log with other instances acting as Followers which replicate the leader’s state. Specifically in Neo4j, this means that writes to the database are ordered by the Core instance currently playing the Leader role for the respective database. If a Neo4j DBMS cluster contains multiple databases, each one of those databases operates within a logically separate Raft group, and therefore each has an individual leader. This means that a Core instance may act both as Leader for some databases, and as Follower for other databases.

If a follower has not heard from the leader for a while, then it can initiate an election and attempt to become the new leader. The follower makes itself a Candidate and asks other Cores to vote for it. If it can get a majority of the votes, then it assumes the leader role. Cores will not vote for a candidate which is less up-to-date than itself. There can only be one leader at any time per database, and that leader is guaranteed to have the most up-to-date log.

Elections are expected to occur during the normal running of a cluster and they do not pose an issue in and of itself. If you are experiencing frequent re-elections and they are disturbing the operation of the cluster then you should try to figure out what is causing them. Some common causes are environmental issues (e.g. a flaky networking) and work overload conditions (e.g. more concurrent queries and transactions than the hardware can handle).

Write transactions will always be routed to the leader for the respective database. As a result, unevenly distributed leaderships may cause write queries to be disproportionately directed to a subset of instances. By default, Neo4j avoids this by automatically transferring database leaderships so that they are evenly distributed throughout the cluster. Additionally, Neo4j will automatically transfer database leaderships away from instances where those databases are configured to be read-only using dbms.databases.read_only or similar.

Databases operate as independent entities in a Neo4j DBMS, both in standalone and in a cluster. Since a cluster can consist of multiple independent server instances, the effects of administrative operations like creating a new database happen asynchronously and independently for each server. However, the immediate effect of an administrative operation is to safely commit the desired state in the system database.

The desired state committed in the system database gets replicated and is picked up by an internal component called the reconciler. It runs on every instance and takes the appropriate actions required locally on that instance for reaching the desired state; creating, starting, stopping, and dropping databases.

Every database runs in an independent Raft group and since there are two databases in a fresh cluster, system and neo4j, this means that it also has two Raft groups. Every Raft group also has an independent leader and thus a particular Core instance could be the leader for one database and a follower for another.

Server-side routing is a complement to the client-side routing.

In a Causal Cluster deployment of Neo4j, Cypher queries may be directed to a cluster member that is unable to run the given query. With server-side routing enabled, such queries will be rerouted internally to a cluster member that is expected to be able to run it. This situation can occur for write-transaction queries when they address a database for which the receiving cluster member is not the leader.

The cluster role for core cluster members is per database. Thus, if a write-transaction query is sent to a cluster member that is not the leader for the specified database (specified either via the Bolt Protocol or with the Cypher USE clause), server-side routing will be performed if properly configured.

Server-side routing is enabled by the DBMS, by setting dbms.routing.enabled=true for each cluster member. The listen address (dbms.routing.listen_address) and advertised address (dbms.routing.advertised_address) also need to be configured for server-side routing communication.

Client connections need to state that server-side routing should be used and this is available for Neo4j Drivers and HTTP API.

The table below shows the criteria by which server-side routing is performed:

Store copies are initiated when an instance does not have an up-to-date copy of the database. For example, this is the case when a new instance is joining a cluster (without a seed). It can also happen as a consequence of falling behind the rest of the cluster, for reasons such as connectivity issues or having been shut down. Upon re-establishing connection with the cluster, an instance recognizes that it is too far behind and fetches a new copy from the rest of the cluster.

A store copy is a major operation, which may disrupt the availability of instances in the cluster. Store copies should not be a frequent occurrence in a well-functioning cluster, but rather be an exceptional operation that happens due to specific causes, e.g. network outages or planned maintenance outages. If store copies happen during regular operation, then the configuration of the cluster, or the workload directed at it, might have to be reviewed so that all instances can keep up, and that there is enough of a buffer of Raft logs and transaction logs to handle smaller transient issues.

The protocol used for store copies is robust and configurable. The network requests are directed at an upstream member according to configuration and they are retried despite transient failures. The maximum amount of time to retry every request can be configured with causal_clustering.store_copy_max_retry_time_per_request. If a request fails and the maximum retry time has elapsed then it stops retrying and the store copy fails.

Use causal_clustering.catch_up_client_inactivity_timeout to configure the inactivity timeout for any particular request.

The default upstream strategy is not applicable to Single instances and it differs for Core and Read Replica instances. Core instances always send the initial request to the leader to get the most up-to-date information about the store. The strategy for the file and index requests for Core instances is to vary every other request to a random Read Replica instance and every other to a random Core instance.

Read Replica instances use the same strategy for store copies as it uses for pulling transactions. The default is to pull from a random Core instance.

If you are running a multi-datacenter cluster, then upstream strategies for both Core and Read Replica instances can be configured. Remember that for Read Replica instances, this also affects from where transactions are pulled. See more in Configure for multi-data center operations.

The on-disk state of cluster instances is different from that of standalone instances. The biggest difference is the existence of an additional cluster state. Most of the files there are relatively small, but the Raft logs can become quite large depending on the configuration and workload.

It is important to understand that once a database has been extracted from a cluster and used in a standalone deployment, it must not be put back into an operational cluster.