MVP based on LWW and RESTORE

[muelstefamzn]

OSS-AA MVP - Key-based LWW

More complex conflict-free data types (CRDT) can be implemented on top of this at a later stage.

This approach allows us to focus more on the infrastructure challenges of active-active (AA) replication and increases our chances to be ready for Valkey 10.

This document consists of two parts. The first part discusses the proposed method and the tradeoffs the users will experience. The second part discusses the infrastructure components required for an implementation.

---

Active-Active Cluster Topology

With active-active replication, a global cluster is formed by multiple local clusters.

Each local cluster has one primary and zero or more replicas connected to that primary.

The primaries perform inter-cluster-replication (ICR) between them. The replicas do not directly participate in ICR, but have to be ready to take over if they are promoted.

For cluster-mode enabled clusters, each shard operates mostly independently; within each shard the primaries perform ICR.

---

Last-Writer-Wins

LWW is the simplest method to resolve write conflicts between local clusters.

Each key in the main dictionary stores a sequence number (seqno).

Each write command is assigned a seqno. When the command is replicated, the seqno is sent along.

Primaries only apply mutations if a command's sequence number is higher than the seqno locally stored for the modified key.

Delete operations are discussed in the tombstone section.

---

Why LWW needs Blind-Writes

Write commands can be categorized into two types:

• Blind-Writes: Commands such as SET, where the content of the key after the operation does not depend on the content it had before; it's fully replaced with information provided in the command itself.
• Read-Modify-Write: Commands such as APPEND, where the key's value depends on the old value and the command.

Example with blind-write:

• Local Cluster A executes SET key hello with seqno 100. Local Cluster B executes SET key world with seqno 101 at about the same time.
• Both commands are ICR'd to the other cluster.
  • Local Cluster A will execute SET key world since 101 is higher than 100.
  • Local Cluster B will skip SET key hello since 100 is lower than 101.
• Both clusters end up with key=world.

Example with read-modify-write:

• Local Cluster A executes APPEND key hello with seqno 100. Local Cluster B executes APPEND key world with seqno 101 at about the same time.
• Both commands are ICR'd to the other cluster.
  • Local Cluster A will execute APPEND key world since 101 is higher than 100. The result is key=helloworld.
  • Local Cluster B will skip APPEND key hello since 100 is lower than 101. The result is key=world.
• The clusters fail to converge. ❌

---

Converting Read-Modify-Write to Blind-Write

Many Valkey commands are read-modify-write.

We can convert a read-modify-write to a blind-write by replicating the entire value.

For example, if APPEND is executed, we don't replicate APPEND but instead replicate SET with the result of the APPEND:

• In both clusters key="greeting:".
• Local Cluster A executes APPEND key hello with seqno 100. Replicated as SET key "greeting:hello".
• Local Cluster B executes APPEND key hi with seqno 101 at about the same time. Replicated as SET key "greeting:hi".
• Both commands are ICR'd to the other cluster.
  • Local Cluster A will execute SET key "greeting:hi" since 101 is higher than 100.
  • Local Cluster B will skip SET key "greeting:hello" since 100 is lower than 101.
• The clusters converge on key="greeting:hi".

The main drawback is that the replicated command can be significantly larger than the original command, since it includes the entire key/value pair even for small incremental changes.

Restore-based Conversion

For strings, SET is an obvious choice to replicate the blind-write. However, most collection types do not have an existing blind-write command that would replace the entire collection.

Valkey has a generic method to serialize all value types into a series of bytes. This is used by the DUMP command but also during snapshot creation. It also has the code to deserialize such a serialized key individually with RESTORE.

We translate read-modify-write to blind-writes by replicating them as RESTORE commands (or a new internal-only variant, similar to RESTORE-ASKING).

This works for all data types, including third-party data types provided in Valkey Modules, since such data types implement (de)serialization callbacks.

---

Tombstones

Normal operations cause the seqno of the key to be updated to that of the last operation.

Delete operations are a challenge: If we delete the entire key, we also delete its seqno. If later a SET writes to this key, we have no way to decide if the SET or the DEL should win. We don't even know if there ever was a DEL or if the key never existed.

Proposed solution — tombstones: When a key is deleted, we free the value but keep a placeholder entry with only the seqno. Tombstones are not visible to the user (GET returns nil).

Tombstones solve the LWW problem but introduce a new issue: they use memory even after the key is deleted.

Tombstones can be cleaned up in a background process: Once a local cluster has processed all commands from its peer clusters with a lower seqno than the tombstone, it's impossible for a future write to lose the LWW check. The tombstone can then be removed.

As long as two local clusters are isolated from each other, they cannot clean up tombstones.

---

Supported Commands / Operations

• Single-key commands (initial scope)
• Multi-key commands — can be translated into multiple single-key commands; support added later
• Transaction support (MULTI/EXEC, Lua) — can be added later; atomicity only holds in the cluster of origin
• TTL support — initial version only
• Slot migration — adds considerable complexity; can be added later

Pros/Cons

Advantages of LWW with Restore:

• Supports all data types and all single-key commands (the majority).
• Does not require per-data-type or per-command implementation effort.
• Does not require complex CRDT logic.
• Single seqno per key as new metadata — no metadata within the collections.

Significant shortcomings:

• Write amplification: A small command can cause a much larger RESTORE to be replicated.
  • No upper bound on amplification factor (no upper bound on data stored within a single key).
  • Extra resource consumption (CPU, network, disk) can be significant and surprising.
  • Limits AA to workloads using small collections only.
• Key-level LWW causes more write loss: e.g., if two local clusters add distinct items to the same set (SADD) at about the same time, items from one cluster will be lost.
• Tombstones accumulate memory usage when local clusters are isolated.

Recommendation: Start with this approach and later refine with modular CRDTs. If a CRDT is loaded for a certain data type, it would be used instead of key-level LWW. The approach discussed here serves as the fallback for types without a CRDT module loaded.

---

Part 2: Infrastructure

Local Replication Streams

In general, each local cluster executes commands in a different order — a consequence of asynchronous ICR. Commands that originate from the same cluster are always processed in order.

We call the sequence of write commands that originate from one local cluster its local replication stream. All local streams combined form the combined replication stream. Each local cluster has the same commands in the combined stream, but the order can differ. The combined stream is what is transmitted between a primary and its replicas.

---

Local Stream Replication Offset

We need to track which commands have been ICR'd. This is needed for PSYNC, resuming after a full-sync, resuming after recovery from snapshot and AOF file, etc.

The global stream is a poor choice for tracking since command ordering differs in each local cluster. Instead, we track local streams.

Valkey already uses repl-offsets to track replication stream progress by counting total bytes transmitted. We use the same idea:

Each local stream has a local-repl-offset that counts the bytes of the commands in that stream.

---

AA PSYNC

When the ICR replication link is established between the primaries of two local clusters, they use the local stream offset:

AAPSYNC [local-cluster-id] [replication-id] [local-stream-offset]

When receiving this command, the primary checks its replication buffer to see if the required local offset is still available. If so, a partial sync is possible; otherwise a full-sync is required.

The replication link established this way will only forward commands of the specified local stream.

---

AA FULL SYNC

A full sync for a specific local cluster is similar to a regular full-sync, with one key difference:

An AA full-sync includes only a subset of keys — specifically, only the keys with a sequence number belonging to that local cluster.

When loading the snapshot:

• The main dictionary is not flushed initially.
• LWW is applied while loading the keys.

Once the snapshot is loaded, replication begins from the point where the snapshot was taken.

If a new local cluster is added, it performs an AA full-sync with all existing local clusters.

---

Replication Data

Valkey replicates commands by sending them over the wire using the RESP protocol. AA requires transmitting additional data:

• The seqno of each command
• The local stream the command belongs to

The simplest approach is a custom AARESTORE command:

AARESTORE [local-stream-id] [seqno] [key] [serialized-data]

A more flexible format may be desired later to support more complex CRDTs.

---

Failover

A failover within a local cluster poses an interesting problem:

Without durability, a few writes can be lost due to replication lag between the old primary and the promoted replica. If these writes have already been replicated to other clusters, the newly promoted primary has lost writes that other clusters have applied.

Options to deal with this:

1. Make AA dependent on durability.
2. Only replicate writes to other clusters once all replicas within the local cluster have applied the write.
3. After promotion, the new primary fetches any lost writes from the other local regions by performing an AASYNC for its own local cluster.

[hpatro]

I've been exploring on the same lines and would like us to ship a working AA solution for 10.0.

I would like us to scope it to

• Standalone (Primary with replicas)
• All commands supported. I would prefer us supporting the existing commands instead of introducing new commands. It wouldn't be practical for users to onboard to these APIs.

Questions/Thoughts:

• Avoid dependance other WIP features on Valkey.
• I'm trying to devise a mechanism to avoid the read-modify-write to blind-write for all of the operations on the collections. It won't scale much. It would be nice to only pull the full dump payload on conflict discovery. Without this mechanism the above proposal, would make it rather unusable for majority of workloads. Still deep diving on this.
• I would prefer keeping tombstones independently to avoid eviction flow to have better sampling hit. And also allow garbage collection to be streamlined.
• I would like us to add a section about additional metadata to be stored with snapshot (RDB).
• I would like us to add how full sync performs a merge instead of flush + load.
• How do we tackle flushdb/flushall/swapdb?

[hpatro]

Putting the LWW + Lazy restore design which I mentioned over the call yesterday.

The idea is to pull the RESTORE payload only on conflict. The above proposal converts every read-modify-write into a blind-write, which is what makes it correct but also would produce very high write amplification. An alternative is to keep replicating the original command and send the key's previous seqno alongside its new one. The receiver compares that predecessor against the seqno it holds locally and if they match, no concurrent write happened and the command is applied as-is. This cause no write amplification when there is no conflict. If they differ, the two local clusters wrote the key concurrently; LWW still picks the winner by seqno, but the loser cannot replay a read-modify-write against a value that no longer exists or the winner can drop it, and the loser pulls the full value for that one key as a RESTORE-style dump.

[deepakrn]

@hpatro
I like the idea of transferring full value only when there are conflicts. If we expect such concurrent writes to be rare, it should reduce the write amplification. As we discussed in the call, retrieving the full value on conflict needs to be asynchronous otherwise we will be trading off availability for consistency?

To further generalize the strategy, when a write happens, the default stance would be to propagate that write with the least expensive option (delta for most types?) and if the other node detects a case of conflict, it could trigger a mechanism to resolve conflict via a more expensive mechanism. Potentially, the least expensive option and the more expensive mechanism of resolution during conflicts might be configurable per data type?

@muelstefamzn
I understand that this proposal doesn't have all the low level details. But I had a comment about AA PSYNC and AA FULL SYNC. I don't think a single replication offset is going to be sufficient to understand what operations need to be replicated. Let's take the below example with multiple primaries:

1. Let A, B, C be three primaries from different Valkey instances
2. To begin with A, B, C are at offset 0.
3. Say a client writes concurrently to all the three nodes independently and say X, Y, Z be the three commands.
4. After executing the commands in step 3, let's say that A, B, C are all now at offset 5 (say that X, Y, Z are all the same byte size)
5. If any of the nodes want to sync with the rest of the nodes, they will already be at the same offset and there will be nothing to transfer but they all have diverged at offset 0.

I think this is why we probably need something like an array of replication IDs and offsets - each index representing a stream of commands from each primary in the AA setup.

[muelstefamzn]

@hpatro
Regarding reploffsets, correct, we need to track them per origin. This table needs to be included in the snapshot.

If we always transfer the full value or only if needed is probably the next design questions we should to nail down. Doing it on demand is superior in many scenarios at the cost of more complexity. We'll have to decide if we want this in the MVP. We can discuss this in the next meeting.