Distributed Systems Fundamentals: CAP Theorem & Consistency Models

Classic Definition:
"A collection of independent computers that appears to its users as a single system"

Key Characteristics:
- No global clock
- Concurrency
- Independent failure
- Asynchronous communication
- Geographic distribution

                    CAP Triangle

                  +-----------+
                  | Consistency |
                  +-----------+
                  /           \
                 /             \
                /                   +-----------+               +-----------+
    | Availability |-----------| Partition |
    +-----------+    Tolerance  +-----------+

Consistency (C):
- All nodes see the same data at the same time
- Every read receives the most recent write or an error
- Linearizability guarantee

Availability (A):
- Every request receives a (non-error) response
- Without the guarantee that it contains the most recent write
- System remains operational

Partition Tolerance (P):
- System continues to operate despite network partitions
- Arbitrary message loss or failure
- Network divides the system

CP Systems (Consistency + Partition Tolerance):
Use: Financial systems, inventory management
Examples: MongoDB, HBase, Redis Cluster
Trade-off: May reject writes during partition

Example: Banking System
class BankTransfer {
  async transfer(from, to, amount) {
    // Strong consistency required
    const transaction = await db.beginTransaction();

    try {
      await transaction.debit(from, amount);
      // If partition occurs, fail rather than allow inconsistency
      await transaction.credit(to, amount);
      await transaction.commit();
    } catch (error) {
      await transaction.rollback();
      throw new Error('Transfer failed: System unavailable');
    }
  }
}

AP Systems (Availability + Partition Tolerance):
Use: Social media feeds, analytics, caching
Examples: Cassandra, DynamoDB, Cosmos DB
Trade-off: Eventual consistency, stale reads possible

Example: Social Media Feed
class FeedService {
  async publishPost(userId, content) {
    const postId = generateId();

    // Write to local replica immediately
    await localDB.posts.insert({
      id: postId,
      userId,
      content,
      timestamp: Date.now()
    });

    // Async replication to other nodes
    await replicator.queue({
      type: 'new_post',
      postId,
      userId,
      content
    });

    return postId; // Available immediately
  }

  async getFeed(userId) {
    // May return slightly stale data
    return await localDB.feed.find(userId);
  }
}

CA Systems (Consistency + Availability):
Use: Single-region applications, traditional databases
Examples: PostgreSQL, MySQL, Oracle
Trade-off: Cannot tolerate network partitions
Note: True CA systems are rare in distributed setups

Linearizability (Strong Consistency):
- Operations appear instantaneous
- All nodes see same data
- Real-time ordering

Implementation: Raft Consensus Algorithm
class RaftNode {
  constructor(nodeId) {
    this.nodeId = nodeId;
    this.state = 'follower';
    this.currentTerm = 0;
    this.votedFor = null;
    this.log = [];
    this.commitIndex = 0;
  }

  async requestVote(term, candidateId) {
    if (term > this.currentTerm) {
      this.currentTerm = term;
      this.state = 'follower';
      this.votedFor = null;
    }

    if (term === this.currentTerm &&
        (this.votedFor === null || this.votedFor === candidateId)) {
      this.votedFor = candidateId;
      return { term: this.currentTerm, voteGranted: true };
    }

    return { term: this.currentTerm, voteGranted: false };
  }

  async appendEntries(term, leaderId, prevLogIndex, entries, leaderCommit) {
    if (term < this.currentTerm) {
      return { term: this.currentTerm, success: false };
    }

    this.currentTerm = term;
    this.state = 'follower';

    // Append log entries
    if (this.log[prevLogIndex]?.term === term) {
      this.log = this.log.slice(0, prevLogIndex + 1);
      this.log.push(...entries);

      if (leaderCommit > this.commitIndex) {
        this.commitIndex = Math.min(leaderCommit, this.log.length - 1);
      }

      return { term: this.currentTerm, success: true };
    }

    return { term: this.currentTerm, success: false };
  }
}

Eventual Consistency Properties:
- If no new updates, all accesses return last updated value
- Guarantees convergence
- High availability
- Low latency

Implementation: Vector Clocks
class VectorClock {
  constructor() {
    this.clock = {};
  }

  increment(nodeId) {
    this.clock[nodeId] = (this.clock[nodeId] || 0) + 1;
  }

  merge(otherClock) {
    for (const [node, timestamp] of Object.entries(otherClock)) {
      this.clock[node] = Math.max(
        this.clock[node] || 0,
        timestamp
      );
    }
  }

  compare(otherClock) {
    let thisGreater = false;
    let otherGreater = false;

    for (const node of new Set([
      ...Object.keys(this.clock),
      ...Object.keys(otherClock)
    ])) {
      const thisTime = this.clock[node] || 0;
      const otherTime = otherClock[node] || 0;

      if (thisTime > otherTime) thisGreater = true;
      if (otherTime > thisTime) otherGreater = true;
    }

    if (thisGreater && !otherGreater) return 'after';
    if (otherGreater && !thisGreater) return 'before';
    if (thisGreater && otherGreater) return 'concurrent';
    return 'equal';
  }
}

// Usage in distributed database
class Document {
  constructor(id) {
    this.id = id;
    this.content = '';
    this.vectorClock = new VectorClock();
  }

  update(nodeId, newContent) {
    this.vectorClock.increment(nodeId);
    this.content = newContent;
  }

  merge(otherDoc) {
    const comparison = this.vectorClock.compare(otherDoc.vectorClock);

    if (comparison === 'before') {
      return otherDoc;
    } else if (comparison === 'after') {
      return this;
    } else if (comparison === 'concurrent') {
      // Conflict resolution
      return this.resolveConflict(otherDoc);
    }

    return this;
  }

  resolveConflict(otherDoc) {
    // Application-specific conflict resolution
    // Example: Last Write Wins (based on timestamp)
    // Example: Operational Transformation
    // Example: Conflict-Free Replicated Data Types (CRDTs)

    const merged = new Document(this.id);
    merged.vectorClock.merge(otherDoc.vectorClock);
    merged.content = this.content + '\n\n[Merged]\n\n' + otherDoc.content;

    return merged;
  }
}

Causal Consistency:
- Causally related operations seen in order by all nodes
- Concurrent operations can be in any order
- Weaker than strong consistency, stronger than eventual

Implementation: Causal Ordering
class CausalBroadcast {
  constructor() {
    this.vectorClock = new VectorClock();
    this.pending = [];
    this.delivered = new Set();
  }

  broadcast(message) {
    this.vectorClock.increment(this.nodeId);
    message.clock = { ...this.vectorClock.clock };

    // Send to all nodes
    this.nodes.forEach(node => {
      this.send(node, message);
    });
  }

  receive(message, from) {
    // Check if can deliver
    if (this.canDeliver(message)) {
      this.deliver(message);

      // Try to deliver pending messages
      this.tryDeliverPending();
    } else {
      this.pending.push(message);
    }
  }

  canDeliver(message) {
    // Check if all causal predecessors have been delivered
    for (const [node, timestamp] of Object.entries(message.clock)) {
      if (timestamp > this.vectorClock.clock[node]) {
        return false;
      }
    }
    return true;
  }

  deliver(message) {
    this.delivered.add(message.id);

    // Update vector clock
    for (const [node, timestamp] of Object.entries(message.clock)) {
      this.vectorClock.clock[node] = Math.max(
        this.vectorClock.clock[node] || 0,
        timestamp
      );
    }

    // Actually deliver to application
    this.onMessage(message);
  }
}

Classic 2PC Protocol:
                  Coordinator
                      |
          +-----------+-----------+
          |                       |
      Phase 1: Prepare          Phase 2: Commit
          |                       |
    +-----+-----+           +-----+-----+
    |           |           |           |
 Participant1 Participant2 Participant1 Participant2

    Can commit? <-----------        Commit/Vote
    Yes/No                  -------->    |
          |                             |
    All Yes?                        Commit!
          |
    Send commit!

class TwoPhaseCommit {
  private participants: string[];

  async execute(transaction: Transaction): Promise<boolean> {
    // Phase 1: Prepare
    const preparePromises = this.participants.map(p =>
      this.prepare(p, transaction)
    );

    const results = await Promise.all(preparePromises);

    // Check if all participants can commit
    const allCanCommit = results.every(r => r.canCommit);

    if (!allCanCommit) {
      // Phase 2: Abort
      await Promise.all(
        this.participants.map(p => this.abort(p, transaction))
      );
      return false;
    }

    // Phase 2: Commit
    await Promise.all(
      this.participants.map(p => this.commit(p, transaction))
    );

    return true;
  }

  private async prepare(
    participant: string,
    transaction: Transaction
  ): Promise<{ canCommit: boolean }> {
    try {
      // Lock resources and validate
      await this.lockResources(participant, transaction.resources);
      await this.validate(participant, transaction);

      return { canCommit: true };
    } catch (error) {
      return { canCommit: false };
    }
  }

  private async commit(
    participant: string,
    transaction: Transaction
  ): Promise<void> {
    // Apply changes and release locks
    await this.applyChanges(participant, transaction);
    await this.releaseLocks(participant, transaction.resources);
  }

  private async abort(
    participant: string,
    transaction: Transaction
  ): Promise<void> {
    // Rollback and release locks
    await this.rollback(participant, transaction);
    await this.releaseLocks(participant, transaction.resources);
  }
}

Saga Pattern:
- Break transaction into sequence of local transactions
- Each transaction publishes events
- Compensating transactions for rollback

Example: E-Commerce Order
class OrderSaga {
  async execute(orderId: string): Promise<void> {
    const saga = new Saga();

    // Step 1: Create order
    saga.addStep(
      async () => this.orderService.create(orderId),
      async () => this.orderService.cancel(orderId)
    );

    // Step 2: Reserve inventory
    saga.addStep(
      async () => this.inventory.reserve(orderId),
      async () => this.inventory.release(orderId)
    );

    // Step 3: Process payment
    saga.addStep(
      async () => this.payment.charge(orderId),
      async () => this.payment.refund(orderId)
    );

    // Step 4: Ship order
    saga.addStep(
      async () => this.shipping.ship(orderId),
      async () => this.shipping.cancelShipment(orderId)
    );

    await saga.execute();
  }
}

class Saga {
  private steps: SagaStep[] = [];
  private completedSteps: number = 0;

  addStep(execute: () => Promise<void>, compensate: () => Promise<void>) {
    this.steps.push({ execute, compensate });
  }

  async execute(): Promise<void> {
    try {
      for (let i = 0; i < this.steps.length; i++) {
        await this.steps[i].execute();
        this.completedSteps = i + 1;
      }
    } catch (error) {
      // Compensate completed steps in reverse order
      await this.compensate();
      throw error;
    }
  }

  private async compensate(): Promise<void> {
    for (let i = this.completedSteps - 1; i >= 0; i--) {
      try {
        await this.steps[i].compensate();
      } catch (error) {
        console.error(`Compensation failed for step ${i}:`, error);
        // Continue compensating other steps
      }
    }
  }
}

Leader-Follower Pattern:
              Writes
                |
            +-------+
            | Leader |
            +-------+
                |
        +-------+-------+
        |       |       |
    Read    Read    Read
        |       |       |
    +-------+ +-------+ +-------+
    |Follower| |Follower| |Follower|
    +-------+ +-------+ +-------+

Implementation:
class ReplicatedService {
  private leader: ServiceNode;
  private followers: ServiceNode[];

  constructor(leader: ServiceNode, followers: ServiceNode[]) {
    this.leader = leader;
    this.followers = followers;

    // Setup replication
    this.leader.on('write', (data) => this.replicate(data));
  }

  async write(data: any): Promise<void> {
    // Write to leader
    await this.leader.write(data);

    // Wait for replication to quorum
    await this.waitForReplication(data);
  }

  async read(): Promise<any> {
    // Read from any follower
    const follower = this.selectFollower();
    return await follower.read();
  }

  private async replicate(data: any): Promise<void> {
    // Parallel replication to all followers
    await Promise.all(
      this.followers.map(f => f.replicate(data))
    );
  }

  private async waitForReplication(data: any): Promise<void> {
    const quorum = Math.floor(this.followers.length / 2) + 1;
    let replicated = 0;

    return new Promise((resolve, reject) => {
      const checkReplication = setInterval(() => {
        replicated = this.followers.filter(
          f => f.hasData(data.id)
        ).length;

        if (replicated >= quorum) {
          clearInterval(checkReplication);
          resolve();
        }
      }, 100);

      setTimeout(() => {
        clearInterval(checkReplication);
        reject(new Error('Replication timeout'));
      }, 5000);
    });
  }
}

Multi-Leader Pattern:
    +-----------+     +-----------+
    |  Leader A  |<--->|  Leader B  |
    +-----------+     +-----------+
         |                   |
    +----+----+         +----+----+
    | Follower|         | Follower|
    +---------+         +---------+

Use Cases:
- Multi-data center deployment
- Offline-first applications
- High availability requirements

Implementation:
class MultiLeaderService {
  private leaders: Map<string, ServiceNode>;
  private conflictResolver: ConflictResolver;

  constructor(leaders: ServiceNode[]) {
    this.leaders = new Map(
      leaders.map(l => [l.id, l])
    );
    this.conflictResolver = new LastWriteWinsResolver();

    // Setup bi-directional replication
    this.setupReplication();
  }

  async write(leaderId: string, data: any): Promise<void> {
    const leader = this.leaders.get(leaderId);

    // Add metadata for conflict resolution
    const writeData = {
      ...data,
      id: generateId(),
      timestamp: Date.now(),
      source: leaderId,
      vectorClock: new VectorClock()
    };

    // Write to local leader
    await leader.write(writeData);

    // Async replication to other leaders
    this.replicateToOthers(leaderId, writeData);
  }

  private async replicateToOthers(
    sourceId: string,
    data: any
  ): Promise<void> {
    for (const [id, leader] of this.leaders) {
      if (id !== sourceId) {
        try {
          await leader.replicate(data);
        } catch (error) {
          // Queue for retry
          await this.queueReplication(id, data);
        }
      }
    }
  }

  private setupReplication(): void {
    for (const [id, leader] of this.leaders) {
      leader.on('conflict', async (local, remote) => {
        const resolved = await this.conflictResolver.resolve(
          local,
          remote
        );
        await leader.write(resolved);
      });
    }
  }
}

class LastWriteWinsResolver {
  async resolve(local: any, remote: any): Promise<any> {
    // Compare timestamps
    if (local.timestamp > remote.timestamp) {
      return local;
    } else if (remote.timestamp > local.timestamp) {
      return remote;
    } else {
      // Same timestamp, use source ID as tiebreaker
      return local.source > remote.source ? local : remote;
    }
  }
}

Edge Computing Architecture:
                    Cloud Region
                        |
            +-----------+-----------+
            |           |           |
        Edge Node   Edge Node   Edge Node
            |           |           |
        +---+---+   +---+---+   +---+---+
        | IoT   |   | IoT   |   | IoT   |
        +-------+   +-------+   +-------+

Benefits:
- Reduced latency (5-20ms vs 100-500ms)
- Bandwidth optimization
- Offline capability
- Data sovereignty

Implementation: Edge-Aware Data Sync
class EdgeDataService {
  private edgeNode: EdgeNode;
  private cloudService: CloudService;
  private syncQueue: SyncQueue;

  async write(data: any): Promise<void> {
    // Write to edge node immediately
    await this.edgeNode.write(data);

    // Queue for cloud sync
    await this.syncQueue.enqueue({
      type: 'write',
      data,
      priority: this.calculatePriority(data)
    });
  }

  async read(id: string): Promise<any> {
    // Try edge node first
    const local = await this.edgeNode.read(id);
    if (local) {
      return local;
    }

    // Fallback to cloud
    return await this.cloudService.read(id);
  }

  private calculatePriority(data: any): number {
    // Critical data syncs immediately
    if (data.critical) return 0;

    // Regular data syncs based on network conditions
    const bandwidth = this.edgeNode.getBandwidth();
    if (bandwidth > 100) return 1;  // High bandwidth
    if (bandwidth > 10) return 2;   // Medium bandwidth
    return 3;                       // Low bandwidth - batch sync
  }
}

Serverless Architecture Patterns:
API Gateway → Lambda Functions → Managed Services

Benefits:
- Auto-scaling
- Pay-per-use
- No server management
- High availability

Example: Distributed Workflow
class ServerlessWorkflow {
  private stepFunctions: AWS.StepFunctions;

  async execute(order: Order): Promise<void> {
    const stateMachine = {
      StartAt: 'ValidateOrder',
      States: {
        ValidateOrder: {
          Type: 'Task',
          Resource: 'arn:aws:lambda:us-east-1:123456789:function:validate',
          Next: 'ReserveInventory',
          Retry: [{
            ErrorEquals: ['States.TaskFailed'],
            IntervalSeconds: 2,
            MaxAttempts: 3,
            BackoffRate: 2.0
          }],
          Catch: [{
            ErrorEquals: ['States.ALL'],
            ResultPath: '$.error',
            Next: 'Fail'
          }]
        },
        ReserveInventory: {
          Type: 'Task',
          Resource: 'arn:aws:lambda:us-east-1:123456789:function:inventory',
          Next: 'ProcessPayment',
          Catch: [{
            ErrorEquals: ['States.ALL'],
            Next: 'CancelOrder'
          }]
        },
        ProcessPayment: {
          Type: 'Task',
          Resource: 'arn:aws:lambda:us-east-1:123456789:function:payment',
          Next: 'ShipOrder',
          Catch: [{
            ErrorEquals: ['States.ALL'],
            Next: 'ReleaseInventory'
          }]
        },
        ShipOrder: {
          Type: 'Task',
          Resource: 'arn:aws:lambda:us-east-1:123456789:function:shipping',
          End: true
        },
        ReleaseInventory: {
          Type: 'Task',
          Resource: 'arn:aws:lambda:us-east-1:123456789:function:release',
          Next: 'Fail'
        },
        CancelOrder: {
          Type: 'Task',
          Resource: 'arn:aws:lambda:us-east-1:123456789:function:cancel',
          Next: 'Fail'
        },
        Fail: {
          Type: 'Fail'
        }
      }
    };

    await this.stepFunctions.startExecution({
      stateMachineArn: process.env.WORKFLOW_ARN,
      input: JSON.stringify(order)
    });
  }
}

Decentralized Consensus:
P2P Network → Consensus Algorithm → Distributed Ledger

Example: Proof of Authority Consortium
class ConsortiumNode {
  private validators: string[];
  private privateKey: string;

  async proposeBlock(transactions: Transaction[]): Promise<Block> {
    const lastBlock = await this.getLastBlock();

    const block = {
      index: lastBlock.index + 1,
      timestamp: Date.now(),
      transactions,
      previousHash: lastBlock.hash,
      validator: this.getAddress(),
      signature: null
    };

    // Sign block
    block.signature = await this.signBlock(block);
    block.hash = this.calculateHash(block);

    return block;
  }

  async validateBlock(block: Block): Promise<boolean> {
    // Verify signature
    const isValid = await this.verifySignature(
      block.validator,
      block.signature,
      block
    );

    if (!isValid) return false;

    // Check if validator is authorized
    if (!this.validators.includes(block.validator)) {
      return false;
    }

    // Verify transactions
    for (const tx of block.transactions) {
      if (!(await this.validateTransaction(tx))) {
        return false;
      }
    }

    return true;
  }

  async achieveConsensus(block: Block): Promise<boolean> {
    const votes: Map<string, boolean> = new Map();

    // Request votes from validators
    const votePromises = this.validators.map(validator =>
      this.requestVote(validator, block)
    );

    const results = await Promise.allSettled(votePromises);

    // Count votes
    results.forEach((result, index) => {
      if (result.status === 'fulfilled') {
        votes.set(this.validators[index], result.value);
      }
    });

    // Check for supermajority (>2/3)
    const approveCount = Array.from(votes.values()).filter(v => v).length;
    return approveCount > (this.validators.length * 2 / 3);
  }
}

1. Design for Failure:
   - Assume network partitions will happen
   - Implement circuit breakers
   - Use retries with exponential backoff
   - Plan for degradation

2. Observe Everything:
   - Distributed tracing (OpenTelemetry)
   - Real-time metrics (Prometheus)
   - Structured logging (ELK stack)
   - Alert on anomalies

3. Embrace Event-Driven Architecture:
   - Loose coupling
   - Better scalability
   - Easier debugging
   - Natural audit trail

4. Use Managed Services:
   - Reduce operational overhead
   - Focus on business logic
   - Leverage cloud provider expertise
   - Automatic scaling and updates

5. Plan for Multi-Region:
   - Data sovereignty requirements
   - Low latency global access
   - Disaster recovery
   - Regulatory compliance