Geo-Distributed Caching: Multi-Region Cache Topologies with Redis

import redis
import time
import json
import hashlib

class RegionRouter:
    def __init__(self):
        self.regions = {
            "us-east-1": {"host": "cache-us-east-1.example.com", "port": 6379, "latency_ms": 5},
            "us-west-2": {"host": "cache-us-west-2.example.com", "port": 6379, "latency_ms": 65},
            "eu-west-1": {"host": "cache-eu-west-1.example.com", "port": 6379, "latency_ms": 80},
            "ap-southeast-1": {"host": "cache-ap-southeast-1.example.com", "port": 6379, "latency_ms": 180},
        }
        self.connections = {}

    def get_connection(self, region):
        """Get or create a connection for a region."""
        if region not in self.connections:
            cfg = self.regions[region]
            self.connections[region] = redis.Redis(
                host=cfg["host"], port=cfg["port"], decode_responses=True
            )
        return self.connections[region]

    def nearest_region(self, user_region):
        """Get the nearest cache region for a user."""
        user_latency = {r: abs(cfg["latency_ms"]) for r, cfg in self.regions.items()}
        return min(user_latency, key=user_latency.get)

    def get(self, key, user_region):
        """Get a key from the nearest cache region."""
        region = self.nearest_region(user_region)
        conn = self.get_connection(region)
        try:
            value = conn.get(key)
            return {"value": value, "region": region, "hit": value is not None}
        except redis.ConnectionError:
            for fallback_region in self.regions:
                if fallback_region != region:
                    try:
                        conn = self.get_connection(fallback_region)
                        value = conn.get(key)
                        return {"value": value, "region": fallback_region, "hit": value is not None}
                    except redis.ConnectionError:
                        continue
            return {"value": None, "region": None, "hit": False}

router = RegionRouter()

user_sydney = "ap-southeast-1"
nearest = router.nearest_region(user_sydney)
print(f"User in Sydney -> nearest cache: {nearest}")

user_ireland = "eu-west-1"
nearest = router.nearest_region(user_ireland)
print(f"User in Ireland -> nearest cache: {nearest}")

print(f"\nRouting simulation for users in different regions:")
for user_region in ["us-east-1", "ap-southeast-1", "eu-west-1"]:
    result = router.get("geo:test", user_region)
    print(f"  {user_region:15s} -> {result['region']:15s} hit={result['hit']}")

import redis
import time
import json
import threading

class CrossRegionReplicator:
    def __init__(self):
        self.regions = {}
        self.replication_lag = {
            ("us-east-1", "us-west-2"): 0.05,
            ("us-east-1", "eu-west-1"): 0.08,
            ("us-east-1", "ap-southeast-1"): 0.15,
        }

    def add_region(self, name, conn):
        """Register a region's Redis connection."""
        self.regions[name] = conn

    def write_to_primary(self, key, value, ttl=3600, primary="us-east-1"):
        """Write to primary and schedule async replication."""
        conn = self.regions[primary]
        conn.setex(key, ttl, json.dumps(value))

        replicated = []
        for region in self.regions:
            if region == primary:
                continue
            lag = self.replication_lag.get((primary, region), 0.1)
            t = threading.Timer(lag, self._replicate, args=(primary, region, key, value, ttl))
            t.daemon = True
            t.start()
            replicated.append({"region": region, "estimated_lag_s": lag})

        return {"primary": primary, "key": key, "replicating_to": replicated}

    def _replicate(self, from_region, to_region, key, value, ttl):
        """Replicate a key to a target region."""
        try:
            conn = self.regions[to_region]
            conn.setex(key, ttl, json.dumps(value))
            print(f"  Replicated {key} from {from_region} to {to_region}")
        except Exception as e:
            print(f"  Replication failed {from_region}->{to_region}: {e}")

    def read_local(self, region, key):
        """Read from the local region (may be stale)."""
        conn = self.regions[region]
        value = conn.get(key)
        return {
            "region": region,
            "value": json.loads(value) if value else None,
            "hit": value is not None,
        }

import redis as r_lib
r_primary = r_lib.Redis(decode_responses=True)
r_west = r_lib.Redis(decode_responses=True)
r_eu = r_lib.Redis(decode_responses=True)

replicator = CrossRegionReplicator()
replicator.add_region("us-east-1", r_primary)
replicator.add_region("us-west-2", r_west)
replicator.add_region("eu-west-1", r_eu)

result = replicator.write_to_primary("geo:config", {"theme": "dark"}, ttl=300)
print(f"Write to primary: {result['key']}")

time.sleep(0.2)

for region in ["us-east-1", "us-west-2", "eu-west-1"]:
    read = replicator.read_local(region, "geo:config")
    print(f"Read from {region:15s}: hit={read['hit']}, value={read['value']}")

import redis
import time
import json

r = redis.Redis(decode_responses=True)

class ConflictResolver:
    def __init__(self, redis_client):
        self.r = redis_client

    def write_with_version(self, key, value, region, ttl=3600):
        """Write with version tracking for conflict resolution."""
        version_key = f"{key}:version"
        current_version = int(self.r.get(version_key) or 0)
        new_version = current_version + 1

        entry = {
            "value": value,
            "version": new_version,
            "region": region,
            "timestamp": time.time(),
        }
        self.r.setex(key, ttl, json.dumps(entry))
        self.r.set(version_key, new_version)
        return entry

    def resolve_conflict(self, key, entries):
        """Resolve conflicts using last-writer-wins with version check."""
        resolved = max(entries, key=lambda e: (e["version"], e["timestamp"]))
        return {
            "key": key,
            "resolved_value": resolved["value"],
            "winner_region": resolved["region"],
            "winner_version": resolved["version"],
        }

resolver = ConflictResolver(r)

entry_us = resolver.write_with_version("geo:counter", {"count": 1}, "us-east-1")
print(f"Write from us-east-1: v{entry_us['version']}")

entry_eu = resolver.write_with_version("geo:counter", {"count": 2}, "eu-west-1")
print(f"Write from eu-west-1: v{entry_eu['version']}")

entry_ap = resolver.write_with_version("geo:counter", {"count": 3}, "ap-southeast-1")
print(f"Write from ap-southeast-1: v{entry_ap['version']}")

current = json.loads(r.get("geo:counter"))
resolution = resolver.resolve_conflict("geo:counter", [current])
print(f"\nConflict resolution: {resolution}")