Distributed Transactions and 2PC: When Atomicity Goes Wrong in Production

// io.thecodeforge — System Design tutorial

// Naive two-phase write without coordination
// Step 1: Deduct inventory
UPDATE inventory SET quantity = quantity - 1 WHERE product_id = 123;
// Step 2: Charge payment (if this fails, inventory is already deducted)
INSERT INTO payments (order_id, amount) VALUES (456, 29.99);
// If step 2 fails, you need a compensating transaction:
UPDATE inventory SET quantity = quantity + 1 WHERE product_id = 123;
// But what if the compensation also fails? You're inconsistent.

// io.thecodeforge — System Design tutorial

package main

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"time"

	_ "github.com/lib/pq"
	"github.com/go-redis/redis/v8"
)

type Participant interface {
	Prepare(ctx context.Context, txID string) error
	Commit(ctx context.Context, txID string) error
	Abort(ctx context.Context, txID string) error
}

type PostgresParticipant struct {
	db *sql.DB
}

func (p *PostgresParticipant) Prepare(ctx context.Context, txID string) error {
	// Begin a transaction and write to a prepared transactions table
	tx, err := p.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	// Insert a record that this transaction is prepared
	_, err = tx.ExecContext(ctx, `INSERT INTO prepared_tx (tx_id, status) VALUES ($1, 'prepared')`, txID)
	if err != nil {
		tx.Rollback()
		return err
	}
	// Hold the transaction open — we'll commit or rollback later
	// In real code, you'd store the tx object and commit/rollback in Commit/Abort
	return tx.Commit() // Simplified: in real impl, keep tx open
}

func (p *PostgresParticipant) Commit(ctx context.Context, txID string) error {
	_, err := p.db.ExecContext(ctx, `UPDATE prepared_tx SET status = 'committed' WHERE tx_id = $1`, txID)
	return err
}

func (p *PostgresParticipant) Abort(ctx context.Context, txID string) error {
	_, err := p.db.ExecContext(ctx, `UPDATE prepared_tx SET status = 'aborted' WHERE tx_id = $1`, txID)
	return err
}

type RedisParticipant struct {
	client *redis.Client
}

func (r *RedisParticipant) Prepare(ctx context.Context, txID string) error {
	// Use Redis MULTI/EXEC to simulate prepare
	// In real impl, you'd use Redis transactions or Redlock
	return r.client.Set(ctx, "tx:"+txID+":status", "prepared", 0).Err()
}

func (r *RedisParticipant) Commit(ctx context.Context, txID string) error {
	return r.client.Set(ctx, "tx:"+txID+":status", "committed", 0).Err()
}

func (r *RedisParticipant) Abort(ctx context.Context, txID string) error {
	return r.client.Set(ctx, "tx:"+txID+":status", "aborted", 0).Err()
}

type Coordinator struct {
	participants []Participant
	log          []string // transaction log for recovery
}

func (c *Coordinator) Execute(ctx context.Context, txID string) error {
	// Phase 1: Prepare
	for _, p := range c.participants {
		err := p.Prepare(ctx, txID)
		if err != nil {
			// If any participant fails, abort all
			c.abortAll(ctx, txID)
			return fmt.Errorf("prepare failed: %w", err)
		}
		c.log = append(c.log, txID+":prepared")
	}

	// Phase 2: Commit
	for _, p := range c.participants {
		err := p.Commit(ctx, txID)
		if err != nil {
			// Log the error and continue — we've already committed on some
			// This is a partial commit scenario — manual recovery needed
			log.Printf("commit failed for %s: %v", txID, err)
		}
		c.log = append(c.log, txID+":committed")
	}
	return nil
}

func (c *Coordinator) abortAll(ctx context.Context, txID string) {
	for _, p := range c.participants {
		if err := p.Abort(ctx, txID); err != nil {
			log.Printf("abort failed for %s: %v", txID, err)
		}
	}
}

func main() {
	// Setup participants
	pg := &PostgresParticipant{db: /* initialize db */}
	redis := &RedisParticipant{client: /* initialize redis */}
	coord := &Coordinator{participants: []Participant{pg, redis}}

	ctx := context.Background()
	txID := "order-456"
	if err := coord.Execute(ctx, txID); err != nil {
		log.Fatal(err)
	}
	fmt.Println("Transaction committed successfully")
}

-- io.thecodeforge — System Design tutorial

-- Check for prepared transactions that are blocking
SELECT gid, prepared, owner, database, transaction AS xmin
FROM pg_prepared_xacts;

-- Force rollback a stuck prepared transaction (use with caution!)
ROLLBACK PREPARED 'tx-order-456';

-- Set a timeout to automatically abort idle prepared transactions
SET idle_in_transaction_session_timeout = '10s';

// io.thecodeforge — System Design tutorial

// Recovery routine for in-doubt transactions
func (c *Coordinator) RecoverInDoubt(ctx context.Context) {
	// Read the coordinator log
	for _, entry := range c.log {
		if strings.HasSuffix(entry, ":prepared") && !strings.HasSuffix(entry, ":committed") {
			txID := strings.TrimSuffix(entry, ":prepared")
			// Query each participant for status
			allCommitted := true
			for _, p := range c.participants {
				status, err := p.GetStatus(ctx, txID)
				if err != nil || status != "committed" {
					allCommitted = false
					break
				}
			}
			if allCommitted {
				c.log = append(c.log, txID+":committed")
			} else {
				c.abortAll(ctx, txID)
			}
		}
	}
}

// io.thecodeforge — System Design tutorial

// Latency breakdown for 2PC with 3 participants (RTT = round-trip time)
// Phase 1: Coordinator sends prepare to 3 participants (1 RTT each, parallel)
// Phase 2: Participants vote (1 RTT back, parallel)
// Phase 3: Coordinator sends commit (1 RTT each, parallel)
// Phase 4: Participants acknowledge (1 RTT back, parallel)
// Total: 2 RTTs (if all parallel) but with network latency, say 50ms per RTT => 100ms minimum
// Plus disk I/O for logging: ~10ms per participant => 30ms
// Total: ~130ms per transaction

// io.thecodeforge — System Design tutorial

// Saga orchestration for order placement
// Step 1: Reserve inventory (local transaction)
// If fails -> abort (no compensation needed)
// Step 2: Charge payment (local transaction)
// If fails -> compensate step 1 (release inventory)
// Step 3: Confirm order (local transaction)
// If fails -> compensate step 2 (refund payment) and step 1 (release inventory)

// Each step is a local ACID transaction. The saga coordinator logs each step and its compensation.
// No global locks, no blocking. But you must handle partial failures with idempotent compensations.

#!/bin/bash
# io.thecodeforge — System Design tutorial

# Resolve in-doubt transactions across PostgreSQL participants
TXID=$1

# Query each participant
for host in pg1 pg2 pg3; do
  STATUS=$(psql -h $host -t -c "SELECT gid FROM pg_prepared_xacts WHERE gid='$TXID';")
  if [ -n "$STATUS" ]; then
    echo "$host: prepared"
  else
    echo "$host: not found"
  fi
done

# If all participants are prepared, commit
# Otherwise, abort (or manual decision)
read -p "Commit (c) or Abort (a)? " DECISION
if [ "$DECISION" = "c" ]; then
  for host in pg1 pg2 pg3; do
    psql -h $host -c "COMMIT PREPARED '$TXID';"
  done
else
  for host in pg1 pg2 pg3; do
    psql -h $host -c "ROLLBACK PREPARED '$TXID';"
  done
fi