Function: applyremotechange in multi-leader-replication/multi_leader.py

Date: 2026-05-29

Time: 14:16

applyremotechange — Multi-Leader Conflict Resolution

Purpose

This is the inbound replication handler for a multi-leader (multi-master) system. When a replica receives a write that originated on another node, this method decides whether the incoming change conflicts with local state and, if so, resolves the conflict using a pluggable strategy (last-write-wins or custom merge). It's the core of the convergence mechanism — without it, replicas would diverge permanently after concurrent writes.

Contract

Preconditions:

• change must contain keys "key", "timestamp", "nodeid", "value", and optionally "istombstone".
• If strategy is CUSTOMMERGE, mergefn must be non-None and callable with signature (key, localval, remoteval, localts, remotets) -> mergedvalue. The code does not guard against mergefn being None — it will raise TypeError at call time.
• Timestamps are Lamport clocks (monotonic integers), not wall-clock times.

Postconditions:

• The node's Lamport clock is advanced to at least max(localclock, remotets) + 1.
• The (remotets, remotenode) pair is recorded in _seen[key], guaranteeing idempotency on replay.
• If a conflict occurred, a ConflictRecord is appended to conflictlog and returned.
• If the change was applied (not deduplicated), it's queued in _pending for downstream propagation.

Invariant: Calling applyremotechange twice with the same change dict is a no-op after the first application — the seen set enforces exactly-once semantics per (key, timestamp, nodeid) triple.

Parameters

| Parameter | Type | Description |

|-----------|------|-------------|

| change | dict | Replication log entry from another node. Must have key, timestamp, nodeid, value; istombstone defaults to False. |

| strategy | ConflictStrategy | How to resolve conflicts. LASTWRITEWINS picks the higher (timestamp, nodeid) tuple. CUSTOMMERGE delegates to merge_fn. |

| mergefn | Optional[Callable] | Required when strategy is CUSTOMMERGE. Receives (key, localvalue, remotevalue, localts, remotets) and returns the merged value. Ignored otherwise. |

Edge cases on change: Tombstoned values arrive with istombstone: True and value: None. The method stores the sentinel TOMBSTONE object internally but reports None in ConflictRecord.remote_value for tombstones — the caller sees deletion as None, not the sentinel.

Return Value

• None — No conflict. Either the change was a duplicate (idempotency), the key didn't exist locally, or the remote change came from the same origin (sequential update, not concurrent write).
• ConflictRecord — A conflict was detected and resolved. The record captures both values, both timestamps, the resolved value, and which strategy resolved it. The caller can log this, surface it to the user, or ignore it — the store is already updated.

Algorithm

1. Dedup check. Look up (remotets, remotenode) in _seen[key]. If present, this is a replay — return None immediately. This is critical for ring topologies where changes circulate through every node.

2. Advance Lamport clock. Set clock = max(clock, remote_ts) + 1. This ensures causal ordering: any subsequent local write will have a timestamp strictly greater than the remote one.

3. Record seen. Mark this (ts, node) pair as processed for future dedup.

4. No local entry. If the key doesn't exist in store, accept unconditionally. Store the value (or TOMBSTONE), queue for propagation, return None.

5. Same-origin, same-timestamp. If local and remote have identical (origin, timestamp), this is the same write arriving via a different path — skip it.

6. Same origin, different timestamp. The remote node sent a newer version of its own write. This is a sequential update, not a conflict. Accept it if the remote (ts, node_id) tuple is lexicographically greater. Always propagate.

7. Different origins — conflict. Two nodes wrote to the same key concurrently. Resolve:

• LWW: Compare (timestamp, nodeid) tuples. Higher wins. The nodeid tiebreaker ensures deterministic resolution across all replicas even when timestamps collide.
• Custom merge: Call mergefn, assign a new timestamp (max + 1), pick a canonical origin (max(localorigin, remote_node) — deterministic), and propagate the merged result as a new change. This is the only strategy that synthesizes a new change rather than forwarding the original.

8. Log and return. Append the ConflictRecord to conflictlog. For LWW, forward the original change. For custom merge, the merged change was already appended in step 7. Return the record.

Side Effects

• Mutates _store: May overwrite the value for key.
• Mutates _clock: Always advances the Lamport clock.
• Mutates _seen: Always records the incoming (ts, node) pair.
• Mutates _pending: Appends the change (or a synthesized merged change) for downstream propagation. This is how changes ripple through ring topologies.
• Mutates conflictlog: Appends a record when a conflict is resolved.
• No I/O, no locking. This is a pure in-memory operation — thread safety is the caller's responsibility.

Error Handling

• ValueError: Raised if strategy is not a recognized ConflictStrategy variant. This is a programmer error, not a runtime condition.
• TypeError (implicit): If strategy is CUSTOMMERGE and mergefn is None, Python will raise TypeError: 'NoneType' object is not callable. The code trusts the caller to pair these correctly.
• KeyError (implicit): If change is missing required keys ("key", "timestamp", etc.), Python raises KeyError. No defensive checking — the dict schema is an assumed contract.

Usage Patterns

Typical caller is MultiLeaderCluster.sync(), which collects pending changes from all nodes, then applies them to destination nodes based on topology:

# All-to-all: every change goes to every other node
for change in pending_by_node[src_id]:
    for dst_id in self._node_order:
        if dst_id != src_id:
            self._nodes[dst_id].apply_remote_change(change, strategy, merge_fn)

Caller obligations:

• Must call getpendingchanges() before applyremotechange() in the same sync round to avoid infinite loops where a node's own propagated changes get re-applied.
• Must provide a valid mergefn when using CUSTOMMERGE.
• Should call sync() repeatedly (or use syncuntilconverged()) since ring topologies require multiple rounds — a change must hop through every node before convergence.

Dependencies

• _TOMBSTONE: Module-level sentinel object. Identity-compared, never exposed to callers — get() returns None for tombstoned keys.
• ConflictStrategy: Enum controlling resolution. Only two variants; adding a third requires updating this method.
• ConflictRecord: Dataclass for conflict metadata. Pure data, no behavior.
• Lamport clock (clock): Shared mutable state on the node. Both put() and applyremote_change() advance it, so interleaving local writes with remote applies maintains causal ordering.

Assumptions Not Enforced by Types

1. Timestamps are positive integers from Lamport clocks, not arbitrary values. Negative or zero timestamps would break the max + 1 advancement.

2. Node IDs are comparable strings — they're used in tuple comparison (ts, nodeid) for tiebreaking and in max(localorigin, remote_node) for canonical origin selection.

3. change dicts are not mutated after being passed in. The same dict may be forwarded to _pending and later applied to other nodes — mutation would corrupt the replication log.

4. Single-threaded execution. No synchronization on store, pending, seen, or clock.