Data Persistence
Distributed Database
Peer-to-peer Architecture
Gossip Protocol

Gossip Protocol

We’ll explore the first approach to maintaining a peer-to-peer cluster: Gossip Protocol .

Eager Reliable Broadcast Protocol

Eager Reliable Broadcast is a communication protocol in distributed systems that ensures messages are broadcast to all participants. In this protocol, each peer continuously exchanges information with every other peer.

ClusterServer 1Server 2Server 3Server 4

This protocol follows an eager delivery model, where messages are immediately sent to all peers upon generation. While effective, this approach becomes inefficient as the number of peers grows because it consumes substantial bandwidth and system resources due to redundant message exchanges.

To address these inefficiencies, the Gossip Protocol was introduced to reduce resource usage by limiting the number of exchanges.

Gossip Protocol

The gossip protocol works much like rumors spreading in an office. A peer starts by sharing a message with a few randomly selected peers. These peers then forward the message to others, eventually reaching all nodes in the cluster.

For example, Server 1 first informs Server 2 and Server 3 about a piece of information. They then propagate it to Server 4. Eventually, all servers in the cluster acknowledge the information.

ClusterClusterClusterServer 1Server 2Server 3Server 4Server 1Server 2Server 3Server 4Server 1Server 2Server 3Server 4

Maintaining Cluster Membership

Each server maintains information about other nodes, such as their addresses, states, and shard allocations.

For example:

ClusterA (1.1.1.1)B (2.2.2.2)B:  State: UP  Address: 2.2.2.2B:  State: UP  Address: 2.2.2.2A:  State: UP  Address: 1.1.1.1A:  State: UP  Address: 1.1.1.1

Adding a Node

Bootstrap nodes serve as entry points for new nodes, typically accessed via static IP addresses or DNS.

In this example, A acts as a bootstrap node. When node C joins, it sends its information to A and receives the current cluster metadata.

ClusterC (3.3.3.3)A (1.1.1.1)B (2.2.2.2)B:  State: UP  Address: 2.2.2.2B:  State: UP  Address: 2.2.2.2A:  State: UP  Address: 1.1.1.1A:  State: UP  Address: 1.1.1.1Ask to joinClusterC (3.3.3.3)A (1.1.1.1)B (2.2.2.2)A:  State: UP  Address: 1.1.1.1B:  State: UP  Address: 2.2.2.2A:  State: UP  Address: 1.1.1.1B:  State: UP  Address: 2.2.2.2B:  State: UP  Address: 2.2.2.2C:  State: UP (Newly added)  Address: 3.3.3.3B:  State: UP  Address: 2.2.2.2C:  State: UP (Newly added)  Address: 3.3.3.3A:  State: UP  Address: 1.1.1.1A:  State: UP  Address: 1.1.1.1Cluster information

A then informs B of C’s arrival, allowing B to update its view of the cluster.

ClusterC (3.3.3.3)A (1.1.1.1)B (2.2.2.2)A:  State: UP  Address: 1.1.1.1B:  State: UP  Address: 2.2.2.2A:  State: UP  Address: 1.1.1.1B:  State: UP  Address: 2.2.2.2B:  State: UP  Address: 2.2.2.2C:  State: UP  Address: 3.3.3.3B:  State: UP  Address: 2.2.2.2C:  State: UP  Address: 3.3.3.3A:  State: UP  Address: 1.1.1.1C:  State: UP (Newly added)  Address: 3.3.3.3A:  State: UP  Address: 1.1.1.1C:  State: UP (Newly added)  Address: 3.3.3.3Inform of C

All members are now aware of C, meaning it has successfully joined the cluster. Based on this information, the system can redistribute data if necessary.

ClusterA (1.1.1.1)B (2.2.2.2)C (3.3.3.3)B:  State: UP  Address: 2.2.2.2C:  State: UP  Address: 3.3.3.3B:  State: UP  Address: 2.2.2.2C:  State: UP  Address: 3.3.3.3A:  State: UP  Address: 1.1.1.1C:  State: UP  Address: 3.3.3.3A:  State: UP  Address: 1.1.1.1C:  State: UP  Address: 3.3.3.3A:  State: UP  Address: 1.1.1.1B:  State: UP  Address: 2.2.2.2A:  State: UP  Address: 1.1.1.1B:  State: UP  Address: 2.2.2.2

Forwarding

With consistent hashing and shared metadata, any node can act as an interface:

  • Serve read requests directly or forward them to the corresponding replicas.
  • Route write requests to the owner node.
ClientClusterServer 1Server 21. Write a record2. Calculate appropriate node3. Forward

Gossip Rounds

Gossip rounds form the backbone of the protocol. Periodically, each node randomly selects peers and shares its current state (e.g., a heartbeat), progressively spreading updates throughout the cluster.

For example, at time 3, A gossips its state. B receives it and forwards it to C.

Cluster (00:03)Cluster (A gossips at 00:03)ABCABCB:  State: UP  Heartbeat: 1C:  State: UP  Heartbeat: 1B:  State: UP  Heartbeat: 1C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 1C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 1C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 1B:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 1B:  State: UP  Heartbeat: 1B:  State: UP  Heartbeat: 1C:  State: UP  Heartbeat: 1B:  State: UP  Heartbeat: 1C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 3 (New)C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 3 (New)C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 3 (New)B:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 3 (New)B:  State: UP  Heartbeat: 1A: UPA: UP

Later, B also gossips its state to random nodes.

Cluster (B gossips at 00:03)ABCB:  State: UP  Heartbeat: 3 (New)C:  State: UP  Heartbeat: 1B:  State: UP  Heartbeat: 3 (New)C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 3C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 3C:  State: UP  Heartbeat: 1A:  State: UP  Heartbeat: 3B:  State: UP  Heartbeat: 3 (New)A:  State: UP  Heartbeat: 3B:  State: UP  Heartbeat: 3 (New)B: UPB: UP

Failure Detection

One key advantage of Gossip rounds is failure detection. If a node doesn’t receive a heartbeat from another node within a defined time window, it marks that peer as DOWN.

If the heartbeat timeout is 3 seconds, by time 5, both A and B consider C down due to the absence of updates.

Cluster (HeartbeatLifetime = 3, Time = 5)ABCB:  State: UP  Heartbeat: 3C:  State: DOWN  Heartbeat: 1 (Expired)B:  State: UP  Heartbeat: 3C:  State: DOWN  Heartbeat: 1 (Expired)A:  State: UP  Heartbeat: 3C:  State: DOWN  Heartbeat: 1 (Expired)A:  State: UP  Heartbeat: 3C:  State: DOWN  Heartbeat: 1 (Expired)Mark DOWNMark DOWN

Temporary Promotion

When a node is deemed unreachable, the detector stops forwarding data to it and switches to a replica. Once the original node recovers, it synchronizes with the replica to restore data.

Node failureRecoveryServer AServer C (Primary)Server B (Replica)Server AServer C (Primary)Server B (Replica)Stop forwardingFailoverBack to primaryPull data to recover

Data Conflicts

This architecture emphasizes Availability over Consistency (AP). During network partitions, replicas can be promoted to serve writes temporarily. However, once partitions heal, conflict resolution becomes necessary.

Two effective approaches to conflict resolution are:

Last Write Wins

This method uses timestamps to resolve conflicts. Each record tracks the latest update time, and the most recent version wins during a merge.

Server 1Server 2Id: 123Name: MikeId: 123Name: MikeId: 123Name: JohnUpdated: 00:01Id: 123Name: JohnUpdated: 00:01Id: 123Name: MikeUpdated: 00:03Id: 123Name: MikeUpdated: 00:03Server 2 wins

This strategy relies on clock synchronization. If clocks are skewed, incorrect records might be selected.

Vector Clocks

A more robust alternative, vector clocks, avoids dependence on synchronized clocks. Instead, each record tracks a version vector in the format [(Server, Version)].

For example, a record initially receives its version number from the shard that owns it. Other servers then replicate the record from the owner, preserving this version for consistency.

ClusterServer 1Server 2Server 3Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: JohnVL:- Server 1: 3

Then, a network partition occurs, dividing the cluster into three partitions. Each group may update the record independently, expanding the vector clock:

  • Server 1 updates the record and increments its version number.
  • Server 3 updates the record and creates its own version.
ClusterClient 1Client 2Partition 1Partition 2Partition 3Server 1Server 2Server 3Id: 123Name: John DoeVL:- Server 1: 4Id: 123Name: John DoeVL:- Server 1: 4Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: John WickVL:- Server 1: 3- Server 3: 1Id: 123Name: John WickVL:- Server 1: 3- Server 3: 1SET Name = John DoeSET Name = John Wick

A conflict is identified when vectors cannot be merged deterministically:

  • [(Peer 1, 4)] is a clear successor of [(Peer 1, 3)].
  • [(Peer 1, 4)] and [(Peer 1, 3), (Peer 3, 1)] clearly conflict.

The system maintains both versions:

ClusterServer 1Server 2Server 3Id: 123Name: John DoeVL:- Server 1: 4Id: 123Name: John DoeVL:- Server 1: 4Id: 123Version 1:  Name: John Doe  VL:  - Server 1: 4Version 2:  Name: John Wick  VL:  - Server 1: 3  - Server 3: 1Id: 123Version 1:  Name: John Doe  VL:  - Server 1: 4Version 2:  Name: John Wick  VL:  - Server 1: 3  - Server 3: 1Id: 123Name: John DoeVL:- Server 1: 4Id: 123Name: John DoeVL:- Server 1: 4Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: JohnVL:- Server 1: 3Id: 123Name: John WickVL:- Server 1: 3- Server 3: 1Id: 123Name: John WickVL:- Server 1: 3- Server 3: 1OverrideConflictConflict

Application-Level Resolution

The application is responsible for resolving conflicts. It receives all conflicting versions and decides how to merge them.

For instance, it may choose to accept the first version:

DatabaseId: 123Version 1:  Name: John Doe  VL:  - Server 1: 4Version 2:  Name: John Wick  VL:  - Server 1: 3  - Server 3: 1Id: 123Version 1:  Name: John Doe  VL:  - Server 1: 4Version 2:  Name: John Wick  VL:  - Server 1: 3  - Server 3: 1ApplicationId: 123Name: John DoeVL:- Server 1: 4Id: 123Name: John DoeVL:- Server 1: 4Accept first version

Since the database lacks context about the business logic, deferring resolution to the application ensures safer and more flexible conflict handling. Applications should therefore include additional metadata in their records to support this process.

Last updated on
Peer-to-peer ArchitectureConsensus Protocol