Requirements for Transaction Replication

Neo4j Causal Clusters communicate and synchronise using the Raft Protocol, which also governs the various states held by cluster members throughout the duration of their cluster membership.

Applications submit transaction requests to the cluster, which are served by one of the cluster members, selected based on a host of criterion, such as mode (core/primary, read-replica/secondary), role, (leader, follower), bookmarks, connection protocol used, etc.

Data Files transferred during Catchup:

Files that contain actual entity records (as opposed to meta-data around those records, e.g. files such as index files) constitute one of the key factors determining effective load-balancing, high availability and fault-tolerance, as well as the secured availability of the required data-set, which is essentially made up of:
 

1- Data-Store files (files under $NEO4J_HOME/data/databases/<database_name>) e.g. neostore.nodestore.db. These contain actual data records in an encoded form, for long-term persistence on disk.
 

2- Transaction Log files (files under $NEO4J_HOME/data/transactions/<database_name>) e.g. neostore.transaction.db.x. These contain transaction metadata and data records, temporarily persisted on disk. Transaction logs are flagged for pruning, following checkpoint events and are eventually pruned, having met additional criteria configured via the db.tx_log.rotation.retention_policy setting (in 4.x this is under a different name, dbms.tx_log.rotation.retention_policy ).
 

A note on Raft Logs and Transaction Logs:

Raft logs are the log files under $NEO4J_HOME/data/cluster-state/db/<database_name>/raft-log e.g. raft.log.x. They contain metadata around each raft replication event, received and recorded by the particular cluster member. Raft logs are pruned once all transactions within are successfully written to local transaction logs and the criteria configured prune_strategy setting is met.

Transaction logs contain a subset of information, initially written to raft logs (via raft replication). A transaction is considered committed by a leader cluster instance once it is committed to the local raft logs of a simple majority of cluster members. Transactions are then copied over from raft logs to transaction logs, primarily to aid intra-cluster catchup and incremental backups (see documentation regarding transaction logs for further details).

The Store-Copy Process

The term "store-copy" is used for synchronising store files across a Neo4j Causal Cluster (as well as backups). The process, conducted and controlled by the "CATCHUP" protocol. Store-Copy can be triggered by different sources: a backup client, a primary falling behind in raft, a secondary falling behind in transaction pulling, a member starting up with or without a seed. In the context of raft replication amongst core members, store-copy gets triggered by the Raft Protocol, which continuously compares transaction states on each cluster member, to ensure causal consistency and minimising transaction lag between cluster members.

A store copy process is performed in the following way:

1- Lagging Cluster member, broadcasts a transaction pull request, to other cluster members, identifying transaction ID of the last check-pointed (and successfully committed) transaction locally available to it and the transaction ID of the latest transaction available on leader.

org.neo4j.kernel.impl.transaction.log.NoSuchTransactionException: 
Unable to find transaction 1234567 
in any of my logical logs: 
Couldn't find any log containing 1234567

2- Cluster members containing transaction logs corresponding to the requested transaction ID differential, respond, confirming such availability, at which point a dedicated "data" channel is established (by default, using the configured causal_clustering.transaction_listen_address and causal_clustering.transaction_advertised_address, addresses.

INFO [c.n.c.p.h.InboundChannelLogger] [catchup-server] Initializing server channel [id: 0x4295eaef, L:/127.0.0.1:6001 - R:/127.0.0.1:53822]
INFO [c.n.c.p.h.HandshakeClientInitializer] Handshake completed on channel [id: 0x9ffed860, L:/127.0.0.1:53828 - R:localhost/127.0.0.1:6002]. Installing: Catchup version:1.0

3- Download begins towards the lagging member, of either the transaction logs (delta store-copy, also called transaction log shipping) or, store-files (also called "full store-copy" or "snapshot download"), If the delta count of missing transactions, is considered too large.

INFO [c.n.c.c.t.TxPullRequestBatchingLogger] New TxPullRequest from /127.0.0.1:64006 for DatabaseId{0d5c30c8} (above transaction id: 3). 
TxPullRequests for this database and address will be batched the next 5 minutes

INFO","category":"c.n.c.c.s.TxPuller","message":"[mydatabase/1c2eb495] 
Pulling transactions from 10.34.200.109:6000 starting with txId: 2505321"}

4- If a snapshot download occurs, upon completion, the lagging member compares the latest transaction ID in it's local transaction logs, to those available on the remote/sender instance and if required, invokes further transaction log shipping to copy over any transactions not present in it's local transaction logs.

INFO [c.n.c.c.t.TxPullRequestBatchingLogger] 
Handling GetPrepareStoreCopyRequest for DatabaseId{0d5c30c8} [From: /127.0.0.1:64006]
2023-06-12 15:31:29.874+0000 INFO [c.n.c.c.t.TxPullRequestBatchingLogger] 
Handling GetStoreFileRequest for DatabaseId{0d5c30c8}. 
Requesting file: neostore.nodestore.db.labels with required minimum transaction id: 4 [From: /127.0.0.1:64007]

5- Recover the pulled transactions within the store files received. This is a costly process. Often taking much longer than the download and depends on the amount and type of transactions pulled in.

6- Lastly, the lagging member compares the current available latest transaction ID in it's local raft logs, to those reported by the current leader. This is done to synchronise with the most recent transactions that were not appended to the lagging member, during the preceding store-copy process, which had placed the store-copied database in an offline state, until completion of the store-copy process. The lagging member then pulls/downloads all append-entries from the remote cluster member to its local raft log.

Cyclic/Repetitive Store-Copy Events - underlying causes and impact

Repeated frequent store-copy events, whether involving the same, or, different databases, same or different cluster members, may occur. Here are some causes of such events:

1- The lag, or transaction delta between cluster members, occurs frequently, or is ever increasing.

2- If requirements for "catchup" aren't met by locally persisted logs (e.g. due to catchup timeouts)

3- If there are abrupt catchup channel terminations. These may occur due to ongoing GC pauses, or other server or network resource exhaustion en-route amongst cluster members. Such abrupt connection terminations are usually reported in debug.log as:

"ERROR","category":"c.n.c.p.i.ServerChannelInitializer",
"message":"[catchup-server] Exception in outbound for channel: [id: 0xc9922767, L:/10.112.117.21:6000 ! R:/10.114.203.149:50226]",
"stacktrace":" io.netty.channel.StacklessClosedChannelException\n\tat io.netty.channel.AbstractChannel.close(ChannelPromise)(Unknown Source)\n"}

4- High raft replication pressure. The following example of a log entry indicates inbound raft pressure higher than the ability of the concerned cluster instances (read iops at leader, network throughput, write iops at followers) to write raft requests to local raft logs:

"WARN","category":"c.n.c.c.b.BatchingMessageHandler",
"message":"[mydatabase/1c2eb495] Raft in-queue dropping messages after: E_COUNT_EXCEEDED"}

The above warning reports that the follower instance had dropped some of the append-entry requests. Such drops have a transient impact, since any dropped raft entries remain uncommitted and are automatically retried to ensure eventual consistency. The warnings, however, highlight the inability of the core, at that point in time, to cope with the incoming workload.

Impact:

The persistence of any of the above underlying issues will likely result in cyclic store-copy events, which are undesirable due to:

Impact to the particular database: unavailability of the impacted Database, as it is kept in an offline state, whilst the store-copy process completes and the database is subsequently brought online. A more acute impact might be observed in very high workload situations, since, in the time it takes for store-copy complete, there may have already been significant transaction lag built-up, requiring further transaction log and raft log catch-ups.

Overall impact on the cluster: necessitating persisted increased network activity for copying store-files in addition to any ongoing replication that must also occur over the same network, simultaneously. Additionally, there is increased disk read and write pressure on all members involved in the catchup, again, in addition to serving simultaneous replication.

Missing/Pruned Logs (an edge case):

Transaction logs as well as raft logs are persisted on disk. One best-practice is to host the $NEO4J_HOME/data and $NEO4J_HOME/transactions directories, on separate disks, to reduce IOPs contention and possible impact of disk space exhaustion. Further details on this subject are beyond scope of this article.

Transaction and raft logs however grow proportionally to transactional and replication workloads and thus, the need for curtailing their disk utilisation. Referring to the five stages of the store-copy process, stated earlier in this article, if transaction lag grows quickly, insufficient log retention (for transaction logs and rarely, raft logs) may lead to frequent store-copies. Given the aforementioned impact, log retention must therefore be optimised in accordance with:
 

• Rates of write transactions (both counts and sizes contribute, towards total bit rate). The following metrics may help identify spikes or otherwise variations from expected traffic patterns:
  • bolt.messages_received
  • bolt.accumulated_queue_time
  • bolt.accumulated_processing_time
  • transaction.committed_write
  • transaction.active_write

• Minimum guaranteed intra-cluster network throughput and stability. Network layer is beyond Neo4j server's operations and thus, relevant metrics are the ones available on the host machine's OS as well as any VM management platform. Key factors to monitor are: 
  • Packet drops
  • Average network throughput

• Store file sizes. Referring to the data file types described earlier in this article, if one or more of the store files, is significantly large (determined by the total time taken for the read IOPS on the remote instance, the intra-cluster network and write disk IOPS on the target instance, to successfully transfer the file) then a full/snapshot store-copy event may take too long for a delta of transaction and raft logs to be subsequently applied, thus necessitating the need for another store-copy. This may only occur when large store file sizes are coupled with high ongoing write workloads.

Optimising log retention

Transaction log retention:

Debug.log under $NEO4J_HOME/logs, records new transaction pull requests as e.g.:

INFO [c.n.c.c.t.TxPullRequestBatchingLogger] New TxPullRequest from /127.0.0.1:64006 
for DatabaseId{0d5c30c8} (above transaction id: 3). 
TxPullRequests for this database and address will be batched the next 5 minutes

Failure to download these, generally due to the remote server having pruned them, results in log entries in debug.log, similar to:

INFO [c.n.c.c.s.TxPuller] [neo4j3/abcd123] Transaction pulling attempt 
failed with error: E_STORE_UNAVAILABLE
WARN [c.n.c.c.p.m.g.RawTxPullRequestHandler] Error obtaining txLogChannel, must be pruned
java.lang.IllegalArgumentException: Transaction id 305620045 not found in transaction logs
Caused by: org.neo4j.kernel.impl.transaction.log.NoSuchTransactionException: Unable to find transaction 305620045 in any of my logical logs: Couldn't find any log containing 305620045

WARN [Store Copy] [neo4j/abcd123] [Event] Failed to catch up store [requiredTxId=-1, error=com.neo4j.causalclustering.catchup.storecopy.StoreCopyFailedException: com.neo4j.causalclustering.catchup.storecopy.StoreCopyFailedException: Pulling tx failed consecutively without progress]
WARN [c.n.c.s.StoreCopyProcess] [neo4j/abcd123] Failed to catch up
com.neo4j.causalclustering.catchup.storecopy.StoreCopyFailedException: com.neo4j.causalclustering.catchup.storecopy.StoreCopyFailedException: Pulling tx failed consecutively without progress

Above reported failure to pull transaction logs due to such logs having been pruned away on remote servers, consequently triggers replacement of the existing local store, with a newly copied data store from one of the other cluster members, which is reported similar to:

INFO [Store Copy] [neo4j/abcd123] [Event] Unable to incrementally bring store up to date. Start replacing store
INFO [c.n.c.s.DatabaseStoreCopier] Deleted old store

Frequent occurrence of above log entries, indicate the need to increase transaction log retention in an attempt to prevent frequent store-copy events. Transaction log retention policy is set via the db.tx_log.rotation.retention_policy setting (in 4.x it is instead dbms.tx_log.rotation.retention_policy). As discussed earlier, the optimum value largely depends upon ingress writes workload, network conditions and disk iops (both for reads and writes).

As an example, consider peak workloads of 10 transactions (of 1G each) per second. This means 10G of data written per second to raft and then transaction logs. This would quickly (within 15s) fill up a log retention set at e.g. 150G. Hence, a key best-practice for such high workloads, is to keep transaction sizes short (few MBs if possible), which is beneficial for various reasons, but require a longer discussion. Subsequently, if the cluster then encounter a situation where, say core2 needs to catchup with the leader core1 with the newest 100k transactions due to an ever increasing leader-follower lag, yet, core1 has already pruned away those 100k transactions in its local transaction logs, because those 100k transactions were large enough to quickly fill up transaction logs worth 150G.
 
One way to avoid such situation, is to use a time-based instead of size-based retention, however, it might increase the likelihood of disk-space exhaustion. 

In Neo4j 5.9 onwards, the rotation.retention_policy can be configured with both a period based restriction as well as a size restriction, such as 2 days 2G, in which case this is an OR condition: logs would be pruned that only contain transactions older than 2 days, or that are larger than 2G.

There is, however, no alternative to ensuring adequate network and disk performance to meet and surpass the peak ingress workload requirements. Key take-away, is to monitor the average disk space growth in X hours/days and then set a time-based transaction log retention (with transaction logs ideally hosted on a separate disk) of a duration that ensures that such average plus peak variation, doesn't take us beyond the max available disk space. 

A useful indicator showing transaction log growth over time, is "log.rotation_events" (see Neo4j Metrics reference for details). Higher counts indicate high transaction workload and a possible need for increased transaction log retention.

Note: Transaction log retention may be dynamically configured , whilst persisting a new value beyond restarts requires updating the configuration within $NEO4J_HOME/conf/neo4j.conf file.

Note: It is advisable to set the rotation.retention_policy to the same value, on each cluster member.

Raft log retention:

Observing one or more of the following, indicates possible need for increasing raft log retention:

1- Repeated raft replication attempt for the same "RaftMemberId" and "sequenceNumber", e.g.:

INFO [c.n.c.c.r.RaftReplicator] [system/00000000] 
Replication attempt 2 to leader RaftMemberId{8171a880}: 
DistributedOperation{content=ReplicatedLeaseRequest{owner=RaftMemberId{8171a880},
 leaseId=0}, globalSession=GlobalSession{sessionId=c56d559d-b6e7-4bfd-85fe-b87f624f7918, owner=RaftMemberId{8171a880}}, operationId=LocalOperationId{localSessionId=0, sequenceNumber=0}}

2- Repeated occurrences of "download because of SnapshotRequirement" despite having completed store copy.

In 4.x, it is logged similar to:

# database finished the store copy 
Database 'neo4j' transition is complete from STORE_COPYING{db=neo4j/429e0bcd} 
to STARTED{db=neo4j/429e0bcd} (request by StoreCopy:3)
#But then we find that we need to store copy again
 because the remote doesn't have the raft log entries we need anymore
[neo4j/429e0bcd] Scheduling download because of SnapshotRequirement{leaderPrevIndex=670930593, localAppendIndex=665976865}

in 5.x:

Database 'neo4j' transition is complete from STORE_COPYING{db=neo4j/429e0bcd} 
to STARTED{db=neo4j/429e0bcd} (request by StoreCopy:3)
Scheduling download of snapshot

3- A spike in raft replication attempts and more directly, a spike in raft replication failures, reported by the following metrics, for a given database:

• causal_clustering.core.replication_attempt
• causal_clustering.core.replication_fail

Raft log prune strategy and pruning frequency is controlled via the following settings:

For Neo4j 5.x:

db.cluster.raft.log.prune_strategy (hereafter referred to as prune_strategy)

dbms.cluster.raft.log.pruning_frequency (hereafter referred to as pruning_frequency)

For Neo4j 4.x:

causal.clustering.raft_log_prune_strategy

causal_clustering.raft_log_pruning_frequency

Please refer to the links for details and defaults. Depending on available disk space, small to medium increments, e.g. 2G, or 5G, should usually suffice in most high workload situations, however, an optimal value is best estimated from monitoring the indicators above.

Note: Raft log retention is non-dynamic and thus requires a Neo4j service restart to take effect.

Note: It is advisable to set the prune_strategy to the same value, on each cluster member

Recovering from ongoing cyclic store-copy events

One approach to recover from ongoing cyclic store-copy events is listed by the steps below:

1- Ensure sufficient disk-space availability and accordingly increase transaction log retention dynamically first, then within neo4j.conf

2- If the need for increased raft log retention is evident from the aforementioned indicators, stop any ongoing data ingestion/write workloads, or wait for an off-peak workload time to tune the prune_strategy as described above. Changing this requires a restart, which should be done in a rolling fashion: followers first, leader last.

Recovering from store-copy in a "stuck" non-progressive state

In rare cases observed in Neo4j 4.x and 5.x, store-copies requested by read-replicas (or secondary servers in 5.x) may appear "stuck", never completing, nor reporting failures. The underlying cause is the requesting replica (or secondary cluster member) lagging too far behind that a full store-copy gets triggered, yet simultaneously transactions continue being applied asynchronously, resulting in such non-progressive scenario, observable within debug.log as follows:
 

ERROR [o.n.m.DatabaseHealth] [mydatabase/002e9g9f] 
Database panic: The database has encountered a critical error, and needs to be restarted. 
Please see database logs for more details.
org.neo4j.internal.kernel.api.exceptions.TransactionFailureException: 
Could not append transaction representation to log
Caused by: org.neo4j.storageengine.StoreFileClosedException: 
Store for file '/neo4j/data/databases/mydatabase/neostore' is closed
ERROR [c.n.d.e.DefaultPanicService] Components for 'DatabaseId{002e9g9f[mydatabase]}' 
have encountered a critical error: Catchup failed
org.neo4j.internal.kernel.api.exceptions.TransactionFailureException: 
Could not append transaction representation to log
Caused by: org.neo4j.storageengine.StoreFileClosedException: 
Store for file '/neo4j/data/databases/mydatabase/neostore' is closed

The previous log represents an already discovered product issue, fixed in versions 4.4.26 and 5.12. Encountering this issue indicates a need for an upgrade to later versions containing the fix, however, a temporary workaround might be to set causal_clustering.pull_interval in 4.x, or db.cluster.catchup.pull_interval in Neo4j 5.x, to a high value, e.g. 2h (replacing the default of 1s), followed by restarting the Neo4j service. A subsequent store-copy should then avoid attempting to pull transaction logs separately over a different channel.