Read/Write Patterns

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

interface QuorumConfig {
    N: number;  // Total replicas
    W: number;  // Write quorum
    R: number;  // Read quorum
}
 
class QuorumCoordinator {
    private replicas: ReplicaNode[];
    private config: QuorumConfig;
    
    constructor(replicas: ReplicaNode[], config: QuorumConfig) {
        if (replicas.length !== config.N) {
            throw new Error(`Expected ${config.N} replicas, got ${replicas.length}`);
        }
        this.replicas = replicas;
        this.config = config;
    }
    
    async write(key: string, value: Value): Promise<WriteResult> {
        const timestamp = Date.now();
        const versionedValue = { ...value, timestamp };
        
        // Send write to all replicas, wait for W acknowledgments
        const results = await Promise.allSettled(
            this.replicas.map(r => r.write(key, versionedValue))
        );
        
        const successes = results.filter(r => r.status === 'fulfilled');
        
        if (successes.length >= this.config.W) {
            return { success: true, acknowledgments: successes.length };
        } else {
            throw new Error(`Write quorum not met: ${successes.length}/${this.config.W}`);
        }
    }
    
    async read(key: string): Promise<Value> {
        // Send read to all replicas, wait for R responses
        const results = await Promise.allSettled(
            this.replicas.map(r => r.read(key).then(value => ({ replica: r, value })))
        );
        
        const successes = results
            .filter(r => r.status === 'fulfilled')
            .map(r => (r as PromiseFulfilledResult<any>).value);
        
        if (successes.length < this.config.R) {
            throw new Error(`Read quorum not met: ${successes.length}/${this.config.R}`);
        }
        
        // Find the most recent value
        const mostRecent = successes.reduce((latest, current) =>
            current.value.timestamp > latest.value.timestamp ? current : latest
        );
        
        // Optionally trigger read repair for stale replicas
        this.triggerReadRepair(key, mostRecent.value, successes);
        
        return mostRecent.value;
    }
    
    private triggerReadRepair(
        key: string, 
        correctValue: Value, 
        responses: { replica: ReplicaNode; value: Value }[]
    ): void {
        for (const { replica, value } of responses) {
            if (value.timestamp < correctValue.timestamp) {
                // Fire-and-forget repair
                replica.write(key, correctValue).catch(console.error);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

// Different operations need different consistency levels
 
class UserService {
    private db: Database;
    
    // Password changes: Critical, need strong consistency
    async changePassword(userId: string, newHash: string): Promise<void> {
        await this.db.write(
            { table: 'users', key: userId, field: 'passwordHash', value: newHash },
            { consistency: 'QUORUM' }  // W=2, R=2 for N=3
        );
        
        // Verify the write completed by reading back
        const verified = await this.db.read(
            { table: 'users', key: userId, field: 'passwordHash' },
            { consistency: 'QUORUM' }
        );
        
        if (verified !== newHash) {
            throw new Error('Password update verification failed');
        }
    }
    
    // Profile view count: Non-critical, eventual consistency fine
    async incrementViewCount(userId: string): Promise<void> {
        await this.db.write(
            { table: 'user_stats', key: userId, field: 'viewCount', op: 'INCREMENT' },
            { consistency: 'ONE' }  // Fire and forget, eventually consistent
        );
    }
    
    // User display name: Read from local replica for speed
    async getDisplayName(userId: string): Promise<string> {
        return await this.db.read(
            { table: 'users', key: userId, field: 'displayName' },
            { consistency: 'LOCAL_ONE' }  // Fast local read, might be slightly stale
        );
    }
    
    // Balance check before purchase: Must be consistent
    async getBalance(userId: string): Promise<number> {
        return await this.db.read(
            { table: 'accounts', key: userId, field: 'balance' },
            { consistency: 'SERIAL' }  // Linearizable read (Cassandra's strongest)
        );
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

class SessionConsistentClient {
    private lastWriteTimestamp: Map<string, number> = new Map();
    private preferredReplica: ReplicaNode;
    
    async write(key: string, value: Value): Promise<void> {
        const timestamp = Date.now();
        
        // Write with timestamp
        await this.db.write(key, { ...value, timestamp });
        
        // Track when we last wrote this key
        this.lastWriteTimestamp.set(key, timestamp);
    }
    
    async read(key: string): Promise<Value> {
        const lastWrite = this.lastWriteTimestamp.get(key) || 0;
        
        // Strategy 1: Sticky sessions - read from same replica you wrote to
        if (this.preferredReplica) {
            const value = await this.preferredReplica.read(key);
            if (value.timestamp >= lastWrite) {
                return value;  // Replica has our write, we're good
            }
        }
        
        // Strategy 2: Read with minimum timestamp requirement
        const replicas = await this.db.getReplicas(key);
        
        for (const replica of replicas) {
            const value = await replica.read(key);
            if (value.timestamp >= lastWrite) {
                // Remember this replica for future reads
                this.preferredReplica = replica;
                return value;
            }
        }
        
        // Strategy 3: Block and wait for propagation
        return await this.waitForConsistency(key, lastWrite);
    }
    
    private async waitForConsistency(
        key: string, 
        minTimestamp: number,
        maxWaitMs: number = 5000
    ): Promise<Value> {
        const startTime = Date.now();
        
        while (Date.now() - startTime < maxWaitMs) {
            const value = await this.db.read(key);
            if (value.timestamp >= minTimestamp) {
                return value;
            }
            
            // Wait and retry
            await new Promise(r => setTimeout(r, 100));
        }
        
        throw new Error('Timed out waiting for read-your-writes consistency');
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

// React component example with stale data handling
 
interface UserProfileProps {
    userId: string;
}
 
function UserProfile({ userId }: UserProfileProps) {
    const [profile, setProfile] = useState<Profile | null>(null);
    const [isStale, setIsStale] = useState(false);
    const [isRefreshing, setIsRefreshing] = useState(false);
    
    useEffect(() => {
        // Initial fast fetch (eventual consistency)
        fetchProfile(userId, { consistency: 'ONE' })
            .then(data => {
                setProfile(data);
                
                // Check if data might be stale
                const ageMs = Date.now() - data.lastModified;
                if (ageMs > 5000) {
                    setIsStale(true);
                    // Trigger background refresh with stronger consistency
                    refreshInBackground();
                }
            });
    }, [userId]);
    
    const refreshInBackground = async () => {
        setIsRefreshing(true);
        try {
            const freshData = await fetchProfile(userId, { consistency: 'QUORUM' });
            setProfile(freshData);
            setIsStale(false);
        } finally {
            setIsRefreshing(false);
        }
    };
    
    if (!profile) return <LoadingSkeleton />;
    
    return (
        <div className="profile-card">
            <h1>{profile.name}</h1>
            <p>{profile.bio}</p>
            
            {/* Show staleness indicator */}
            {isStale && (
                <div className="stale-indicator">
                    <span>Data may be outdated</span>
                    {isRefreshing ? (
                        <Spinner size="small" />
                    ) : (
                        <button onClick={refreshInBackground}>Refresh</button>
                    )}
                </div>
            )}
            
            {/* Show when data was last updated */}
            <small>
                Last updated: {formatRelativeTime(profile.lastModified)}
            </small>
        </div>
    );
}
 
// Backend: Optimistic update with verification
async function updateProfile(userId: string, changes: Partial<Profile>) {
    // Optimistic write
    await db.write(userId, changes, { consistency: 'ONE' });
    
    // Background verification
    setTimeout(async () => {
        const current = await db.read(userId, { consistency: 'QUORUM' });
        if (!deepEqual(current, { ...current, ...changes })) {
            // Write didn't propagate correctly, retry or alert
            console.warn('Profile update may have been lost, retrying...');
            await db.write(userId, changes, { consistency: 'QUORUM' });
        }
    }, 1000);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

// PATTERN 1: Idempotent Writes with Unique Request IDs
async function idempotentWrite(
    key: string, 
    value: any, 
    requestId: string  // Client-generated unique ID
): Promise<WriteResult> {
    // Check if this request was already processed
    const existing = await db.get(`requests:${requestId}`);
    if (existing) {
        return existing.result;  // Return cached result
    }
    
    // Process the write
    await db.put(key, value);
    
    // Cache the result for future retries
    await db.put(`requests:${requestId}`, { result: 'success', timestamp: Date.now() });
    
    return { result: 'success' };
}
 
// PATTERN 2: Conditional Writes (Compare-and-Swap)
async function conditionalUpdate(
    key: string, 
    updateFn: (current: Value) => Value
): Promise<{ success: boolean; retries: number }> {
    const MAX_RETRIES = 5;
    
    for (let attempt = 0; attempt < MAX_RETRIES; attempt++) {
        // Read current value with version
        const { value, version } = await db.getWithVersion(key);
        
        // Compute new value
        const newValue = updateFn(value);
        
        // Attempt conditional write
        const result = await db.putIfVersion(key, newValue, version);
        
        if (result.success) {
            return { success: true, retries: attempt };
        }
        
        // Version mismatch - another write happened, retry
        await delay(Math.pow(2, attempt) * 10);  // Exponential backoff
    }
    
    return { success: false, retries: MAX_RETRIES };
}
 
// PATTERN 3: Append-Only Writes
interface Event {
    id: string;
    timestamp: number;
    type: string;
    payload: any;
}
 
async function appendEvent(streamId: string, event: Event): Promise<void> {
    // Events are immutable, no conflicts possible
    const eventKey = `${streamId}:${event.timestamp}:${event.id}`;
    await db.put(eventKey, event);
}
 
async function getState(streamId: string): Promise<State> {
    // Derive state by replaying events
    const events = await db.range(`${streamId}:`, `${streamId}~`);
    return events.reduce((state, event) => applyEvent(state, event), initialState);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

const consistencyConfig = {
    // User authentication: High consistency for security
    'auth.passwords': { write: 'QUORUM', read: 'QUORUM' },
    'auth.sessions': { write: 'QUORUM', read: 'LOCAL_QUORUM' },
    'auth.tokens': { write: 'QUORUM', read: 'QUORUM' },
    
    // Financial data: Maximum consistency
    'payments.transactions': { write: 'ALL', read: 'QUORUM' },
    'payments.balances': { write: 'QUORUM', read: 'SERIAL' },  // Linearizable
    
    // User profile: Balanced, humans are tolerant of slight staleness
    'users.profiles': { write: 'QUORUM', read: 'ONE' },
    'users.preferences': { write: 'ONE', read: 'ONE' },
    
    //  Social features: High availability, eventual consistency fine
    'social.followers': { write: 'ONE', read: 'ONE' },
    'social.likes': { write: 'ANY', read: 'ONE' },  // Can be lost, not critical
    'social.comments': { write: 'QUORUM', read: 'ONE' },  // Don't lose comments
    
    // Analytics: Maximum throughput
    'analytics.events': { write: 'ANY', read: 'ONE' },
    'analytics.aggregates': { write: 'ONE', read: 'ONE' },
    
    // System health: Fast access
    'health.heartbeats': { write: 'ONE', read: 'ONE' },
    'health.alerts': { write: 'QUORUM', read: 'ONE' },  // Don't miss alerts
};
 
function getConsistencyForKey(key: string): ConsistencyLevel {
    // Match key pattern to config
    for (const [pattern, config] of Object.entries(consistencyConfig)) {
        if (key.startsWith(pattern)) {
            return config;
        }
    }
    // Default: balanced
    return { write: 'QUORUM', read: 'ONE' };
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

// ❌ ANTI-PATTERN: Read-Modify-Write without versioning
async function transferBad(from: string, to: string, amount: number) {
    const fromBalance = await db.read(`balance:${from}`);
    const toBalance = await db.read(`balance:${to}`);
    
    // PROBLEM: Another transaction might have modified these in between
    await db.write(`balance:${from}`, fromBalance - amount);
    await db.write(`balance:${to}`, toBalance + amount);
}
 
// ✅ CORRECT: Use conditional writes with versioning
async function transferGood(from: string, to: string, amount: number) {
    let success = false;
    
    while (!success) {
        // Read with versions
        const { value: fromBalance, version: fromVersion } = 
            await db.getWithVersion(`balance:${from}`);
        const { value: toBalance, version: toVersion } = 
            await db.getWithVersion(`balance:${to}`);
        
        // Conditional writes - fail if version changed
        const results = await Promise.all([
            db.putIfVersion(`balance:${from}`, fromBalance - amount, fromVersion),
            db.putIfVersion(`balance:${to}`, toBalance + amount, toVersion),
        ]);
        
        success = results.every(r => r.success);
        
        if (!success) {
            // Exponential backoff before retry
            await delay(Math.random() * 100);
        }
    }
}
 
// ❌ ANTI-PATTERN: Assuming immediate visibility
async function createAndReadBad(data: any) {
    await db.write('key', data, { consistency: 'ONE' });
    // PROBLEM: This read might go to a different replica that doesn't have the write yet
    return await db.read('key', { consistency: 'ONE' });
}
 
// ✅ CORRECT: Use matching consistency or session guarantees
async function createAndReadGood(data: any) {
    await db.write('key', data, { consistency: 'QUORUM' });
    // Read with quorum ensures overlap with write
    return await db.read('key', { consistency: 'QUORUM' });
}