Shahzad Bhatti Welcome to my ramblings and rants!

// This looks innocent but is fundamentally broken
public class OrderService {
    public void processOrder(Order order) {
        orderRepository.save(order);           // What if this succeeds...
        paymentService.chargeCard(order);      // ...but this fails?
        inventoryService.allocate(order);      // Now we have inconsistent state
        emailService.sendConfirmation(order);  // And this might never happen
    }
}

// The double spending problem with default isolation
@Service
public class VulnerableAccountService {
    
    // This uses READ COMMITTED by default - DANGEROUS for financial operations!
    @Transactional
    public void withdrawFunds(String accountId, BigDecimal amount) {
        Account account = accountRepository.findById(accountId);
        
        // RACE CONDITION: Another transaction can modify balance here!
        if (account.getBalance().compareTo(amount) >= 0) {
            account.setBalance(account.getBalance().subtract(amount));
            accountRepository.save(account);
        } else {
            throw new InsufficientFundsException();
        }
    }
}

// What happens with concurrent requests:
// Thread 1: Read balance = $100, check passes
// Thread 2: Read balance = $100, check passes  
// Thread 1: Withdraw $100, balance = $0
// Thread 2: Withdraw $100, balance = -$100 (DOUBLE SPENDING!)

// Method 1: Database constraints (fastest)
@Entity
@Table(name = "accounts")
public class Account {
    @Id
    private String accountId;
    
    @Column(nullable = false)
    @Check(constraints = "balance >= 0") // Database prevents negative balance
    private BigDecimal balance;
    
    @Version
    private Long version;
}

@Service
public class SafeAccountService {
    
    // Let database constraint handle the race condition
    @Transactional
    public void withdrawFundsWithConstraint(String accountId, BigDecimal amount) {
        try {
            Account account = accountRepository.findById(accountId);
            account.setBalance(account.getBalance().subtract(amount));
            accountRepository.save(account); // Database throws exception if balance < 0
        } catch (DataIntegrityViolationException e) {
            throw new InsufficientFundsException("Withdrawal would result in negative balance");
        }
    }
    
    // Method 2: SERIALIZABLE isolation (most secure)
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public void withdrawFundsSerializable(String accountId, BigDecimal amount) {
        Account account = accountRepository.findById(accountId);
        if (account.getBalance().compareTo(amount) >= 0) {
            account.setBalance(account.getBalance().subtract(amount));
            accountRepository.save(account);
        } else {
            throw new InsufficientFundsException();
        }
        // SERIALIZABLE guarantees no other transaction can interfere
    }
    
    // Method 3: Optimistic locking (good performance)
    @Transactional
    @Retryable(value = {OptimisticLockingFailureException.class}, maxAttempts = 3)
    public void withdrawFundsOptimistic(String accountId, BigDecimal amount) {
        Account account = accountRepository.findById(accountId);
        if (account.getBalance().compareTo(amount) >= 0) {
            account.setBalance(account.getBalance().subtract(amount));
            accountRepository.save(account); // Version check prevents race conditions
        } else {
            throw new InsufficientFundsException();
        }
    }
}

// What actually happens under MVCC
@Transactional(isolation = Isolation.REPEATABLE_READ)
public void demonstrateMVCC() {
    // T1: Transaction starts, gets snapshot at time=100
    Account account = accountRepository.findById("123"); // Reads version at time=100
    
    // T2: Another transaction modifies the account (creates version at time=101)
    
    // T1: Reads same account again
    Account sameAccount = accountRepository.findById("123"); // Still reads version at time=100!
    
    assert account.getBalance().equals(sameAccount.getBalance()); // MVCC guarantees this
}

-- Traditional locking approach (not MVCC)
BEGIN TRANSACTION;
SELECT * FROM accounts WHERE id = '123' FOR SHARE; -- Acquires shared lock
-- Other transactions blocked from writing until this transaction ends

-- MVCC approach (PostgreSQL, MySQL InnoDB)
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT * FROM accounts WHERE id = '123'; -- No locks, reads from snapshot
-- Other transactions can write freely, creating new versions

// MVCC write conflict example
@Transactional
@Retryable(value = {OptimisticLockingFailureException.class})
public void updateAccountMVCC(String accountId, BigDecimal newBalance) {
    Account account = accountRepository.findById(accountId);
    
    // If another transaction modified this account between our read
    // and write, MVCC will detect the conflict and retry
    account.setBalance(newBalance);
    accountRepository.save(account); // May throw OptimisticLockingFailureException
}

@Service
@Transactional
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private PaymentService paymentService;
    
    @Autowired
    private InventoryService inventoryService;
    
    // All operations within this method are atomic
    public Order processOrder(CreateOrderRequest request) {
        Order order = new Order(request);
        order = orderRepository.save(order);
        
        // If any of these fail, everything rolls back
        Payment payment = paymentService.processPayment(
            order.getCustomerId(), 
            order.getTotalAmount()
        );
        
        inventoryService.reserveItems(order.getItems());
        order.setPaymentId(payment.getId());
        order.setStatus(OrderStatus.CONFIRMED);
        
        return orderRepository.save(order);
    }
}

@Service
public class OrderService {
    
    // Read-only transactions can be optimized by the database
    @Transactional(readOnly = true)
    public List<Order> getOrderHistory(String customerId) {
        return orderRepository.findByCustomerId(customerId);
    }
    
    // Long-running operations need higher timeout
    @Transactional(timeout = 300) // 5 minutes
    public void processBulkOrders(List<CreateOrderRequest> requests) {
        for (CreateOrderRequest request : requests) {
            processOrder(request);
        }
    }
    
    // Critical operations need strict isolation
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public void transferInventory(String fromLocation, String toLocation, 
                                String itemId, int quantity) {
        Item fromItem = inventoryRepository.findByLocationAndItem(fromLocation, itemId);
        Item toItem = inventoryRepository.findByLocationAndItem(toLocation, itemId);
        
        if (fromItem.getQuantity() < quantity) {
            throw new InsufficientInventoryException();
        }
        
        fromItem.setQuantity(fromItem.getQuantity() - quantity);
        toItem.setQuantity(toItem.getQuantity() + quantity);
        
        inventoryRepository.save(fromItem);
        inventoryRepository.save(toItem);
    }
    
    // Some operations should create new transactions
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void logAuditEvent(String event, String details) {
        AuditLog log = new AuditLog(event, details, Instant.now());
        auditRepository.save(log);
        // This commits immediately, independent of calling transaction
    }
    
    // Handle specific rollback conditions
    @Transactional(rollbackFor = {BusinessException.class, ValidationException.class})
    public void processComplexOrder(ComplexOrderRequest request) {
        // Business logic that might throw business exceptions
        validateOrderRules(request);
        Order order = createOrder(request);
        processPayment(order);
    }
}

@Service
public class OrderProcessingService {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private NotificationService notificationService;
    
    @Transactional
    public void processOrderWithNotification(CreateOrderRequest request) {
        // This participates in the existing transaction
        Order order = orderService.processOrder(request);
        
        // This creates a new transaction that commits independently
        notificationService.sendOrderConfirmation(order);
        
        // If something fails here, the order transaction can still commit
        // but the notification might not be sent
    }
}

@Service
public class NotificationService {
    
    // Creates a new transaction - notifications are sent even if 
    // the main order processing fails later
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void sendOrderConfirmation(Order order) {
        NotificationRecord record = new NotificationRecord(
            order.getCustomerId(),
            "Order confirmed: " + order.getId(),
            NotificationType.ORDER_CONFIRMATION
        );
        notificationRepository.save(record);
        
        // Send actual notification asynchronously
        emailService.sendAsync(order.getCustomerEmail(), 
                              "Order Confirmation", 
                              generateOrderEmail(order));
    }
}

package services

import (
    "context"
    "fmt"
    "gorm.io/gorm"
)

type OrderService struct {
    db *gorm.DB
}

type Order struct {
    ID          uint   `gorm:"primarykey"`
    CustomerID  string
    TotalAmount int64
    Status      string
    PaymentID   string
    Items       []OrderItem `gorm:"foreignKey:OrderID"`
}

type OrderItem struct {
    ID       uint   `gorm:"primarykey"`
    OrderID  uint
    SKU      string
    Quantity int
    Price    int64
}

// Basic transaction with explicit rollback handling
func (s *OrderService) ProcessOrder(ctx context.Context, request CreateOrderRequest) (*Order, error) {
    tx := s.db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
            panic(r)
        }
    }()

    order := &Order{
        CustomerID:  request.CustomerID,
        TotalAmount: request.TotalAmount,
        Status:      "PENDING",
    }

    // Save the order
    if err := tx.Create(order).Error; err != nil {
        tx.Rollback()
        return nil, fmt.Errorf("failed to create order: %w", err)
    }

    // Process payment
    paymentID, err := s.processPayment(ctx, tx, order)
    if err != nil {
        tx.Rollback()
        return nil, fmt.Errorf("payment failed: %w", err)
    }

    // Reserve inventory
    if err := s.reserveInventory(ctx, tx, request.Items); err != nil {
        tx.Rollback()
        return nil, fmt.Errorf("inventory reservation failed: %w", err)
    }

    // Update order with payment info
    order.PaymentID = paymentID
    order.Status = "CONFIRMED"
    if err := tx.Save(order).Error; err != nil {
        tx.Rollback()
        return nil, fmt.Errorf("failed to update order: %w", err)
    }

    if err := tx.Commit().Error; err != nil {
        return nil, fmt.Errorf("failed to commit transaction: %w", err)
    }

    return order, nil
}

// TransactionFunc represents a function that runs within a transaction
type TransactionFunc func(tx *gorm.DB) error

// WithTransaction wraps a function in a database transaction
func (s *OrderService) WithTransaction(fn TransactionFunc) error {
    tx := s.db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
            panic(r)
        }
    }()

    if err := fn(tx); err != nil {
        tx.Rollback()
        return err
    }

    return tx.Commit().Error
}

// Now our business logic becomes cleaner
func (s *OrderService) ProcessOrderClean(ctx context.Context, request CreateOrderRequest) (*Order, error) {
    var order *Order
    
    err := s.WithTransaction(func(tx *gorm.DB) error {
        order = &Order{
            CustomerID:  request.CustomerID,
            TotalAmount: request.TotalAmount,
            Status:      "PENDING",
        }

        if err := tx.Create(order).Error; err != nil {
            return fmt.Errorf("failed to create order: %w", err)
        }

        paymentID, err := s.processPaymentInTx(ctx, tx, order)
        if err != nil {
            return fmt.Errorf("payment failed: %w", err)
        }

        if err := s.reserveInventoryInTx(ctx, tx, request.Items); err != nil {
            return fmt.Errorf("inventory reservation failed: %w", err)
        }

        order.PaymentID = paymentID
        order.Status = "CONFIRMED"
        
        return tx.Save(order).Error
    })

    return order, err
}

type contextKey string

const txKey contextKey = "transaction"

// WithTransactionContext creates a new context with a transaction
func WithTransactionContext(ctx context.Context, tx *gorm.DB) context.Context {
    return context.WithValue(ctx, txKey, tx)
}

// TxFromContext retrieves a transaction from context
func TxFromContext(ctx context.Context) (*gorm.DB, bool) {
    tx, ok := ctx.Value(txKey).(*gorm.DB)
    return tx, ok
}

// GetDB returns either the transaction from context or the main DB
func (s *OrderService) GetDB(ctx context.Context) *gorm.DB {
    if tx, ok := TxFromContext(ctx); ok {
        return tx
    }
    return s.db
}

// Now services can automatically use transactions when available
func (s *PaymentService) ProcessPayment(ctx context.Context, customerID string, amount int64) (string, error) {
    db := s.GetDB(ctx) // Uses transaction if available
    
    payment := &Payment{
        CustomerID: customerID,
        Amount:     amount,
        Status:     "PROCESSING",
    }
    
    if err := db.Create(payment).Error; err != nil {
        return "", err
    }
    
    // Simulate payment processing
    if amount > 100000 { // Reject large amounts for demo
        payment.Status = "FAILED"
        db.Save(payment)
        return "", fmt.Errorf("payment amount too large")
    }
    
    payment.Status = "COMPLETED"
    payment.TransactionID = generatePaymentID()
    
    if err := db.Save(payment).Error; err != nil {
        return "", err
    }
    
    return payment.TransactionID, nil
}

// Usage with context-based transactions
func (s *OrderService) ProcessOrderWithContext(ctx context.Context, request CreateOrderRequest) (*Order, error) {
    var order *Order
    
    return order, s.WithTransaction(func(tx *gorm.DB) error {
        // Create context with transaction
        txCtx := WithTransactionContext(ctx, tx)
        
        order = &Order{
            CustomerID:  request.CustomerID,
            TotalAmount: request.TotalAmount,
            Status:      "PENDING",
        }

        if err := tx.Create(order).Error; err != nil {
            return err
        }

        // These services will automatically use the transaction
        paymentID, err := s.paymentService.ProcessPayment(txCtx, order.CustomerID, order.TotalAmount)
        if err != nil {
            return err
        }

        if err := s.inventoryService.ReserveItems(txCtx, request.Items); err != nil {
            return err
        }

        order.PaymentID = paymentID
        order.Status = "CONFIRMED"
        
        return tx.Save(order).Error
    })
}

// Read-only operations can be optimized
func (s *OrderService) GetOrderHistory(ctx context.Context, customerID string) ([]Order, error) {
    var orders []Order
    
    // Use read-only transaction for consistency
    err := s.db.Transaction(func(tx *gorm.DB) error {
        return tx.Raw("SELECT * FROM orders WHERE customer_id = ? ORDER BY created_at DESC", 
                     customerID).Scan(&orders).Error
    }, &sql.TxOptions{ReadOnly: true})
    
    return orders, err
}

// Operations requiring specific isolation levels
func (s *InventoryService) TransferStock(ctx context.Context, fromSKU, toSKU string, quantity int) error {
    return s.db.Transaction(func(tx *gorm.DB) error {
        var fromItem, toItem InventoryItem
        
        // Lock rows to prevent concurrent modifications
        if err := tx.Set("gorm:query_option", "FOR UPDATE").
            Where("sku = ?", fromSKU).First(&fromItem).Error; err != nil {
            return err
        }
        
        if err := tx.Set("gorm:query_option", "FOR UPDATE").
            Where("sku = ?", toSKU).First(&toItem).Error; err != nil {
            return err
        }
        
        if fromItem.Quantity < quantity {
            return fmt.Errorf("insufficient inventory")
        }
        
        fromItem.Quantity -= quantity
        toItem.Quantity += quantity
        
        if err := tx.Save(&fromItem).Error; err != nil {
            return err
        }
        
        return tx.Save(&toItem).Error
        
    }, &sql.TxOptions{Isolation: sql.LevelSerializable})
}

// src/transaction/mod.rs
use diesel::prelude::*;
use diesel::result::Error as DieselError;
use std::fmt;

#[derive(Debug)]
pub enum TransactionError {
    Database(DieselError),
    Business(String),
    Validation(String),
}

impl fmt::Display for TransactionError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            TransactionError::Database(e) => write!(f, "Database error: {}", e),
            TransactionError::Business(e) => write!(f, "Business error: {}", e),
            TransactionError::Validation(e) => write!(f, "Validation error: {}", e),
        }
    }
}

impl std::error::Error for TransactionError {}

pub type TransactionResult<T> = Result<T, TransactionError>;

// Macro for creating transactional functions
#[macro_export]
macro_rules! transactional {
    (
        fn $name:ident($($param:ident: $param_type:ty),*) -> $return_type:ty {
            $($body:tt)*
        }
    ) => {
        fn $name(conn: &mut PgConnection, $($param: $param_type),*) -> TransactionResult<$return_type> {
            conn.transaction::<$return_type, TransactionError, _>(|conn| {
                $($body)*
            })
        }
    };
}

// Macro for read-only transactions
#[macro_export]
macro_rules! read_only {
    (
        fn $name:ident($($param:ident: $param_type:ty),*) -> $return_type:ty {
            $($body:tt)*
        }
    ) => {
        fn $name(conn: &mut PgConnection, $($param: $param_type),*) -> TransactionResult<$return_type> {
            // In a real implementation, we'd set READ ONLY mode
            conn.transaction::<$return_type, TransactionError, _>(|conn| {
                $($body)*
            })
        }
    };
}

// src/services/order_service.rs
use diesel::prelude::*;
use crate::transaction::*;
use crate::models::*;
use crate::schema::orders::dsl::*;

pub struct OrderService;

impl OrderService {
    // Transactional order processing with automatic rollback
    transactional! {
        fn process_order(request: CreateOrderRequest) -> Order {
            // Create the order
            let new_order = NewOrder {
                customer_id: &request.customer_id,
                total_amount: request.total_amount,
                status: "PENDING",
            };
            
            let order: Order = diesel::insert_into(orders)
                .values(&new_order)
                .get_result(conn)
                .map_err(TransactionError::Database)?;

            // Process payment
            let payment_id = Self::process_payment_internal(conn, &order)
                .map_err(|e| TransactionError::Business(format!("Payment failed: {}", e)))?;

            // Reserve inventory
            Self::reserve_inventory_internal(conn, &request.items)
                .map_err(|e| TransactionError::Business(format!("Inventory reservation failed: {}", e)))?;

            // Update order with payment info
            let updated_order = diesel::update(orders.filter(id.eq(order.id)))
                .set((
                    payment_id.eq(&payment_id),
                    status.eq("CONFIRMED"),
                ))
                .get_result(conn)
                .map_err(TransactionError::Database)?;

            Ok(updated_order)
        }
    }

    // Read-only transaction for queries
    read_only! {
        fn get_order_history(customer_id: String) -> Vec<Order> {
            let order_list = orders
                .filter(customer_id.eq(&customer_id))
                .order(created_at.desc())
                .load::<Order>(conn)
                .map_err(TransactionError::Database)?;
            
            Ok(order_list)
        }
    }

    // Helper functions that work within existing transactions
    fn process_payment_internal(conn: &mut PgConnection, order: &Order) -> Result<String, String> {
        use crate::schema::payments::dsl::*;
        
        let new_payment = NewPayment {
            customer_id: &order.customer_id,
            order_id: order.id,
            amount: order.total_amount,
            status: "PROCESSING",
        };
        
        let payment: Payment = diesel::insert_into(payments)
            .values(&new_payment)
            .get_result(conn)
            .map_err(|e| format!("Payment creation failed: {}", e))?;
        
        // Simulate payment processing logic
        if order.total_amount > 100000 {
            diesel::update(payments.filter(id.eq(payment.id)))
                .set(status.eq("FAILED"))
                .execute(conn)
                .map_err(|e| format!("Payment update failed: {}", e))?;
            
            return Err("Payment amount too large".to_string());
        }
        
        let transaction_id = format!("txn_{}", uuid::Uuid::new_v4());
        
        diesel::update(payments.filter(id.eq(payment.id)))
            .set((
                status.eq("COMPLETED"),
                transaction_id.eq(&transaction_id),
            ))
            .execute(conn)
            .map_err(|e| format!("Payment finalization failed: {}", e))?;
        
        Ok(transaction_id)
    }

    fn reserve_inventory_internal(conn: &mut PgConnection, items: &[OrderItemRequest]) -> Result<(), String> {
        use crate::schema::inventory::dsl::*;
        
        for item in items {
            // Lock the inventory row for update
            let mut inventory_item: InventoryItem = inventory
                .filter(sku.eq(&item.sku))
                .for_update()
                .first(conn)
                .map_err(|e| format!("Inventory lookup failed: {}", e))?;
            
            if inventory_item.quantity < item.quantity {
                return Err(format!("Insufficient inventory for SKU: {}", item.sku));
            }
            
            inventory_item.quantity -= item.quantity;
            
            diesel::update(inventory.filter(sku.eq(&item.sku)))
                .set(quantity.eq(inventory_item.quantity))
                .execute(conn)
                .map_err(|e| format!("Inventory update failed: {}", e))?;
        }
        
        Ok(())
    }
}

// More sophisticated transaction management with isolation levels
#[macro_export]
macro_rules! serializable_transaction {
    (
        fn $name:ident($($param:ident: $param_type:ty),*) -> $return_type:ty {
            $($body:tt)*
        }
    ) => {
        fn $name(conn: &mut PgConnection, $($param: $param_type),*) -> TransactionResult<$return_type> {
            // Set serializable isolation level
            conn.batch_execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
                .map_err(TransactionError::Database)?;
                
            conn.transaction::<$return_type, TransactionError, _>(|conn| {
                $($body)*
            })
        }
    };
}

// Usage for operations requiring strict consistency
impl InventoryService {
    serializable_transaction! {
        fn transfer_stock(from_sku: String, to_sku: String, quantity: i32) -> (InventoryItem, InventoryItem) {
            use crate::schema::inventory::dsl::*;
            
            // Lock both items in consistent order to prevent deadlocks
            let (first_sku, second_sku) = if from_sku < to_sku {
                (&from_sku, &to_sku)
            } else {
                (&to_sku, &from_sku)
            };
            
            let mut from_item: InventoryItem = inventory
                .filter(sku.eq(first_sku))
                .for_update()
                .first(conn)
                .map_err(TransactionError::Database)?;
                
            let mut to_item: InventoryItem = inventory
                .filter(sku.eq(second_sku))
                .for_update()
                .first(conn)
                .map_err(TransactionError::Database)?;
            
            // Ensure we have the right items
            if from_item.sku != from_sku {
                std::mem::swap(&mut from_item, &mut to_item);
            }
            
            if from_item.quantity < quantity {
                return Err(TransactionError::Business(
                    "Insufficient inventory for transfer".to_string()
                ));
            }
            
            from_item.quantity -= quantity;
            to_item.quantity += quantity;
            
            let updated_from = diesel::update(inventory.filter(sku.eq(&from_sku)))
                .set(quantity.eq(from_item.quantity))
                .get_result(conn)
                .map_err(TransactionError::Database)?;
                
            let updated_to = diesel::update(inventory.filter(sku.eq(&to_sku)))
                .set(quantity.eq(to_item.quantity))
                .get_result(conn)
                .map_err(TransactionError::Database)?;
            
            Ok((updated_from, updated_to))
        }
    }
}

// src/transaction/async_transaction.rs
use diesel_async::{AsyncPgConnection, AsyncConnection};
use diesel_async::pooled_connection::bb8::Pool;

#[macro_export]
macro_rules! async_transactional {
    (
        async fn $name:ident($($param:ident: $param_type:ty),*) -> $return_type:ty {
            $($body:tt)*
        }
    ) => {
        async fn $name(pool: &Pool<AsyncPgConnection>, $($param: $param_type),*) -> TransactionResult<$return_type> {
            let mut conn = pool.get().await
                .map_err(|e| TransactionError::Database(e.into()))?;
                
            conn.transaction::<$return_type, TransactionError, _>(|conn| {
                Box::pin(async move {
                    $($body)*
                })
            }).await
        }
    };
}

// Usage with async operations
impl OrderService {
    async_transactional! {
        async fn process_order_async(request: CreateOrderRequest) -> Order {
            // All the same logic as before, but with async/await support
            let new_order = NewOrder {
                customer_id: &request.customer_id,
                total_amount: request.total_amount,
                status: "PENDING",
            };
            
            let order: Order = diesel::insert_into(orders)
                .values(&new_order)
                .get_result(conn)
                .await
                .map_err(TransactionError::Database)?;

            // Process payment asynchronously
            let payment_id = Self::process_payment_async(conn, &order).await
                .map_err(|e| TransactionError::Business(format!("Payment failed: {}", e)))?;

            // Continue with order processing...
            
            Ok(order)
        }
    }
}

@Configuration
@EnableTransactionManagement
public class XATransactionConfig {
    
    @Bean
    @Primary
    public DataSource orderDataSource() {
        MysqlXADataSource xaDataSource = new MysqlXADataSource();
        xaDataSource.setURL("jdbc:mysql://localhost:3306/orders");
        xaDataSource.setUser("orders_user");
        xaDataSource.setPassword("orders_pass");
        return xaDataSource;
    }
    
    @Bean
    public DataSource inventoryDataSource() {
        MysqlXADataSource xaDataSource = new MysqlXADataSource();
        xaDataSource.setURL("jdbc:mysql://localhost:3306/inventory");
        xaDataSource.setUser("inventory_user");
        xaDataSource.setPassword("inventory_pass");
        return xaDataSource;
    }
    
    @Bean
    public JtaTransactionManager jtaTransactionManager() {
        JtaTransactionManager jtaTransactionManager = new JtaTransactionManager();
        jtaTransactionManager.setTransactionManager(atomikosTransactionManager());
        jtaTransactionManager.setUserTransaction(atomikosUserTransaction());
        return jtaTransactionManager;
    }
    
    @Bean(initMethod = "init", destroyMethod = "close")
    public UserTransactionManager atomikosTransactionManager() {
        UserTransactionManager transactionManager = new UserTransactionManager();
        transactionManager.setForceShutdown(false);
        return transactionManager;
    }
    
    @Bean
    public UserTransactionImp atomikosUserTransaction() throws SystemException {
        UserTransactionImp userTransactionImp = new UserTransactionImp();
        userTransactionImp.setTransactionTimeout(300);
        return userTransactionImp;
    }
}

@Service
public class DistributedOrderService {
    
    @Autowired
    @Qualifier("orderDataSource")
    private DataSource orderDataSource;
    
    @Autowired
    @Qualifier("inventoryDataSource")
    private DataSource inventoryDataSource;
    
    // XA transaction spans both databases
    @Transactional
    public void processDistributedOrder(CreateOrderRequest request) {
        // Operations on orders database
        try (Connection orderConn = orderDataSource.getConnection()) {
            PreparedStatement orderStmt = orderConn.prepareStatement(
                "INSERT INTO orders (customer_id, total_amount, status) VALUES (?, ?, ?)"
            );
            orderStmt.setString(1, request.getCustomerId());
            orderStmt.setBigDecimal(2, request.getTotalAmount());
            orderStmt.setString(3, "PENDING");
            orderStmt.executeUpdate();
        }
        
        // Operations on inventory database
        try (Connection inventoryConn = inventoryDataSource.getConnection()) {
            for (OrderItem item : request.getItems()) {
                PreparedStatement inventoryStmt = inventoryConn.prepareStatement(
                    "UPDATE inventory SET quantity = quantity - ? WHERE sku = ? AND quantity >= ?"
                );
                inventoryStmt.setInt(1, item.getQuantity());
                inventoryStmt.setString(2, item.getSku());
                inventoryStmt.setInt(3, item.getQuantity());
                
                int updatedRows = inventoryStmt.executeUpdate();
                if (updatedRows == 0) {
                    throw new InsufficientInventoryException("Not enough inventory for " + item.getSku());
                }
            }
        }
        
        // If we get here, both database operations succeeded
        // The XA transaction manager will coordinate the commit across both databases
    }
}

package distributed

import (
    "context"
    "database/sql"
    "fmt"
    "log"
    "time"
    
    "github.com/google/uuid"
)

type TransactionManager struct {
    resources []XAResource
}

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

type DatabaseResource struct {
    db   *sql.DB
    name string
}

func (r *DatabaseResource) Prepare(ctx context.Context, txID string) error {
    tx, err := r.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    
    // Store transaction for later commit/rollback
    // In production, you'd need a proper transaction store
    transactionStore[txID+"-"+r.name] = tx
    
    return nil
}

func (r *DatabaseResource) Commit(ctx context.Context, txID string) error {
    tx, exists := transactionStore[txID+"-"+r.name]
    if !exists {
        return fmt.Errorf("transaction not found: %s", txID)
    }
    
    err := tx.Commit()
    delete(transactionStore, txID+"-"+r.name)
    return err
}

func (r *DatabaseResource) Rollback(ctx context.Context, txID string) error {
    tx, exists := transactionStore[txID+"-"+r.name]
    if !exists {
        return nil // Already rolled back
    }
    
    err := tx.Rollback()
    delete(transactionStore, txID+"-"+r.name)
    return err
}

// Global transaction store (in production, use Redis or similar)
var transactionStore = make(map[string]*sql.Tx)

func (tm *TransactionManager) ExecuteDistributedTransaction(ctx context.Context, fn func() error) error {
    txID := uuid.New().String()
    
    // Phase 1: Prepare all resources
    for _, resource := range tm.resources {
        if err := resource.Prepare(ctx, txID); err != nil {
            // Rollback all prepared resources
            tm.rollbackAll(ctx, txID)
            return fmt.Errorf("prepare failed: %w", err)
        }
    }
    
    // Execute business logic
    if err := fn(); err != nil {
        tm.rollbackAll(ctx, txID)
        return fmt.Errorf("business logic failed: %w", err)
    }
    
    // Phase 2: Commit all resources
    for _, resource := range tm.resources {
        if err := resource.Commit(ctx, txID); err != nil {
            log.Printf("Commit failed for txID %s: %v", txID, err)
            // In production, you'd need a recovery mechanism here
            return fmt.Errorf("commit failed: %w", err)
        }
    }
    
    return nil
}

func (tm *TransactionManager) rollbackAll(ctx context.Context, txID string) {
    for _, resource := range tm.resources {
        if err := resource.Rollback(ctx, txID); err != nil {
            log.Printf("Rollback failed for txID %s: %v", txID, err)
        }
    }
}

// Usage example
func ProcessDistributedOrder(ctx context.Context, request CreateOrderRequest) error {
    orderDB, _ := sql.Open("mysql", "orders_connection_string")
    inventoryDB, _ := sql.Open("mysql", "inventory_connection_string")
    
    tm := &TransactionManager{
        resources: []XAResource{
            &DatabaseResource{db: orderDB, name: "orders"},
            &DatabaseResource{db: inventoryDB, name: "inventory"},
        },
    }
    
    return tm.ExecuteDistributedTransaction(ctx, func() error {
        // Business logic goes here - use the prepared transactions
        orderTx := transactionStore[txID+"-orders"]
        inventoryTx := transactionStore[txID+"-inventory"]
        
        // Create order
        _, err := orderTx.Exec(
            "INSERT INTO orders (customer_id, total_amount, status) VALUES (?, ?, ?)",
            request.CustomerID, request.TotalAmount, "PENDING",
        )
        if err != nil {
            return err
        }
        
        // Update inventory
        for _, item := range request.Items {
            result, err := inventoryTx.Exec(
                "UPDATE inventory SET quantity = quantity - ? WHERE sku = ? AND quantity >= ?",
                item.Quantity, item.SKU, item.Quantity,
            )
            if err != nil {
                return err
            }
            
            rowsAffected, _ := result.RowsAffected()
            if rowsAffected == 0 {
                return fmt.Errorf("insufficient inventory for %s", item.SKU)
            }
        }
        
        return nil
    })
}

// Java/JPA pessimistic locking
@Service
public class AccountService {
    
    @Transactional
    public void transferFunds(String fromAccountId, String toAccountId, BigDecimal amount) {
        // Lock accounts in consistent order to prevent deadlocks
        String firstId = fromAccountId.compareTo(toAccountId) < 0 ? fromAccountId : toAccountId;
        String secondId = fromAccountId.compareTo(toAccountId) < 0 ? toAccountId : fromAccountId;
        
        Account firstAccount = accountRepository.findById(firstId, LockModeType.PESSIMISTIC_WRITE);
        Account secondAccount = accountRepository.findById(secondId, LockModeType.PESSIMISTIC_WRITE);
        
        Account fromAccount = fromAccountId.equals(firstId) ? firstAccount : secondAccount;
        Account toAccount = fromAccountId.equals(firstId) ? secondAccount : firstAccount;
        
        if (fromAccount.getBalance().compareTo(amount) < 0) {
            throw new InsufficientFundsException();
        }
        
        fromAccount.setBalance(fromAccount.getBalance().subtract(amount));
        toAccount.setBalance(toAccount.getBalance().add(amount));
        
        accountRepository.save(fromAccount);
        accountRepository.save(toAccount);
    }
}

// Go pessimistic locking with GORM
func (s *AccountService) TransferFunds(ctx context.Context, fromAccountID, toAccountID string, amount int64) error {
    return s.WithTransaction(func(tx *gorm.DB) error {
        var fromAccount, toAccount Account
        
        // Lock accounts in consistent order
        firstID, secondID := fromAccountID, toAccountID
        if fromAccountID > toAccountID {
            firstID, secondID = toAccountID, fromAccountID
        }
        
        // Lock first account
        if err := tx.Set("gorm:query_option", "FOR UPDATE").
            Where("id = ?", firstID).First(&fromAccount).Error; err != nil {
            return err
        }
        
        // Lock second account
        if err := tx.Set("gorm:query_option", "FOR UPDATE").
            Where("id = ?", secondID).First(&toAccount).Error; err != nil {
            return err
        }
        
        // Ensure we have the correct accounts
        if fromAccount.ID != fromAccountID {
            fromAccount, toAccount = toAccount, fromAccount
        }
        
        if fromAccount.Balance < amount {
            return fmt.Errorf("insufficient funds")
        }
        
        fromAccount.Balance -= amount
        toAccount.Balance += amount
        
        if err := tx.Save(&fromAccount).Error; err != nil {
            return err
        }
        
        return tx.Save(&toAccount).Error
    })
}

// JPA optimistic locking with version fields
@Entity
public class Account {
    @Id
    private String id;
    
    private BigDecimal balance;
    
    @Version
    private Long version; // JPA automatically manages this
    
    // getters and setters...
}

@Service
public class OptimisticAccountService {
    
    @Transactional
    @Retryable(value = {OptimisticLockingFailureException.class}, maxAttempts = 3)
    public void transferFunds(String fromAccountId, String toAccountId, BigDecimal amount) {
        Account fromAccount = accountRepository.findById(fromAccountId);
        Account toAccount = accountRepository.findById(toAccountId);
        
        if (fromAccount.getBalance().compareTo(amount) < 0) {
            throw new InsufficientFundsException();
        }
        
        fromAccount.setBalance(fromAccount.getBalance().subtract(amount));
        toAccount.setBalance(toAccount.getBalance().add(amount));
        
        // If either account was modified by another transaction,
        // OptimisticLockingFailureException will be thrown
        accountRepository.save(fromAccount);
        accountRepository.save(toAccount);
    }
}

// Rust optimistic locking with version fields
#[derive(Queryable, Identifiable, AsChangeset)]
#[diesel(table_name = accounts)]
pub struct Account {
    pub id: String,
    pub balance: i64,
    pub version: i32,
}

impl AccountService {
    transactional! {
        fn transfer_funds_optimistic(from_account_id: String, to_account_id: String, amount: i64) -> () {
            use crate::schema::accounts::dsl::*;
            
            // Read current versions
            let from_account: Account = accounts
                .filter(id.eq(&from_account_id))
                .first(conn)
                .map_err(TransactionError::Database)?;
                
            let to_account: Account = accounts
                .filter(id.eq(&to_account_id))
                .first(conn)
                .map_err(TransactionError::Database)?;
            
            if from_account.balance < amount {
                return Err(TransactionError::Business("Insufficient funds".to_string()));
            }
            
            // Update with version check
            let from_updated = diesel::update(
                accounts
                    .filter(id.eq(&from_account_id))
                    .filter(version.eq(from_account.version))
            )
            .set((
                balance.eq(from_account.balance - amount),
                version.eq(from_account.version + 1)
            ))
            .execute(conn)
            .map_err(TransactionError::Database)?;
            
            if from_updated == 0 {
                return Err(TransactionError::Business("Concurrent modification detected".to_string()));
            }
            
            let to_updated = diesel::update(
                accounts
                    .filter(id.eq(&to_account_id))
                    .filter(version.eq(to_account.version))
            )
            .set((
                balance.eq(to_account.balance + amount),
                version.eq(to_account.version + 1)
            ))
            .execute(conn)
            .map_err(TransactionError::Database)?;
            
            if to_updated == 0 {
                return Err(TransactionError::Business("Concurrent modification detected".to_string()));
            }
            
            Ok(())
        }
    }
}

// Base interfaces for SAGA operations
public interface SagaCommand<T> {
    T execute() throws Exception;
    void compensate(T result) throws Exception;
}

public class SagaOrchestrator {
    
    public class SagaExecution<T> {
        private final SagaCommand<T> command;
        private T result;
        private boolean executed = false;
        
        public SagaExecution(SagaCommand<T> command) {
            this.command = command;
        }
        
        public T execute() throws Exception {
            result = command.execute();
            executed = true;
            return result;
        }
        
        public void compensate() throws Exception {
            if (executed && result != null) {
                command.compensate(result);
            }
        }
    }
    
    private final List<SagaExecution<?>> executions = new ArrayList<>();
    
    public <T> T execute(SagaCommand<T> command) throws Exception {
        SagaExecution<T> execution = new SagaExecution<>(command);
        executions.add(execution);
        return execution.execute();
    }
    
    public void compensateAll() {
        // Compensate in reverse order
        for (int i = executions.size() - 1; i >= 0; i--) {
            try {
                executions.get(i).compensate();
            } catch (Exception e) {
                log.error("Compensation failed", e);
                // In production, you'd need dead letter queue handling
            }
        }
    }
}

// Concrete command implementations
public class CreateOrderCommand implements SagaCommand<Order> {
    private final OrderService orderService;
    private final CreateOrderRequest request;
    
    public CreateOrderCommand(OrderService orderService, CreateOrderRequest request) {
        this.orderService = orderService;
        this.request = request;
    }
    
    @Override
    public Order execute() throws Exception {
        return orderService.createOrder(request);
    }
    
    @Override
    public void compensate(Order order) throws Exception {
        orderService.cancelOrder(order.getId());
    }
}

public class ProcessPaymentCommand implements SagaCommand<Payment> {
    private final PaymentService paymentService;
    private final String customerId;
    private final BigDecimal amount;
    
    @Override
    public Payment execute() throws Exception {
        return paymentService.processPayment(customerId, amount);
    }
    
    @Override
    public void compensate(Payment payment) throws Exception {
        paymentService.refundPayment(payment.getId());
    }
}

public class ReserveInventoryCommand implements SagaCommand<List<InventoryReservation>> {
    private final InventoryService inventoryService;
    private final List<OrderItem> items;
    
    @Override
    public List<InventoryReservation> execute() throws Exception {
        return inventoryService.reserveItems(items);
    }
    
    @Override
    public void compensate(List<InventoryReservation> reservations) throws Exception {
        for (InventoryReservation reservation : reservations) {
            inventoryService.releaseReservation(reservation.getId());
        }
    }
}

// Usage in service
@Service
public class SagaOrderService {
    
    public void processOrderWithSaga(CreateOrderRequest request) throws Exception {
        SagaOrchestrator saga = new SagaOrchestrator();
        
        try {
            // Execute commands in sequence
            Order order = saga.execute(new CreateOrderCommand(orderService, request));
            
            Payment payment = saga.execute(new ProcessPaymentCommand(
                paymentService, order.getCustomerId(), order.getTotalAmount()
            ));
            
            List<InventoryReservation> reservations = saga.execute(
                new ReserveInventoryCommand(inventoryService, request.getItems())
            );
            
            // If we get here, everything succeeded
            orderService.confirmOrder(order.getId());
            
        } catch (Exception e) {
            // Compensate all executed commands
            saga.compensateAll();
            throw e;
        }
    }
}

// SAGA state management
@Entity
public class SagaTransaction {
    @Id
    private String id;
    
    @Enumerated(EnumType.STRING)
    private SagaStatus status;
    
    private String currentStep;
    
    @ElementCollection
    private List<String> completedSteps = new ArrayList<>();
    
    @ElementCollection
    private List<String> compensatedSteps = new ArrayList<>();
    
    private String contextData; // JSON serialized context
    
    // getters/setters...
}

public enum SagaStatus {
    STARTED, IN_PROGRESS, COMPLETED, COMPENSATING, COMPENSATED, FAILED
}

@Component
public class PersistentSagaOrchestrator {
    
    @Autowired
    private SagaTransactionRepository sagaRepo;
    
    @Transactional
    public void executeSaga(String sagaId, List<SagaStep> steps) {
        SagaTransaction saga = sagaRepo.findById(sagaId)
            .orElse(new SagaTransaction(sagaId));
        
        try {
            for (SagaStep step : steps) {
                if (saga.getCompletedSteps().contains(step.getName())) {
                    continue; // Already completed
                }
                
                saga.setCurrentStep(step.getName());
                saga.setStatus(SagaStatus.IN_PROGRESS);
                sagaRepo.save(saga);
                
                // Execute step
                step.execute();
                
                saga.getCompletedSteps().add(step.getName());
                sagaRepo.save(saga);
            }
            
            saga.setStatus(SagaStatus.COMPLETED);
            sagaRepo.save(saga);
            
        } catch (Exception e) {
            compensateSaga(sagaId);
            throw e;
        }
    }
    
    @Transactional
    public void compensateSaga(String sagaId) {
        SagaTransaction saga = sagaRepo.findById(sagaId)
            .orElseThrow(() -> new IllegalArgumentException("SAGA not found"));
        
        saga.setStatus(SagaStatus.COMPENSATING);
        sagaRepo.save(saga);
        
        // Compensate in reverse order
        List<String> stepsToCompensate = new ArrayList<>(saga.getCompletedSteps());
        Collections.reverse(stepsToCompensate);
        
        for (String stepName : stepsToCompensate) {
            if (saga.getCompensatedSteps().contains(stepName)) {
                continue;
            }
            
            try {
                SagaStep step = findStepByName(stepName);
                step.compensate();
                saga.getCompensatedSteps().add(stepName);
                sagaRepo.save(saga);
            } catch (Exception e) {
                log.error("Compensation failed for step: " + stepName, e);
                saga.setStatus(SagaStatus.FAILED);
                sagaRepo.save(saga);
                return;
            }
        }
        
        saga.setStatus(SagaStatus.COMPENSATED);
        sagaRepo.save(saga);
    }
}

// SAGA state machine in Go
package saga

import (
    "context"
    "encoding/json"
    "fmt"
    "time"
    
    "gorm.io/gorm"
)

type SagaStatus string

const (
    StatusStarted     SagaStatus = "STARTED"
    StatusInProgress  SagaStatus = "IN_PROGRESS"
    StatusCompleted   SagaStatus = "COMPLETED"
    StatusCompensating SagaStatus = "COMPENSATING"
    StatusCompensated SagaStatus = "COMPENSATED"
    StatusFailed      SagaStatus = "FAILED"
)

type SagaTransaction struct {
    ID               string     `gorm:"primarykey"`
    Status           SagaStatus
    CurrentStep      string
    CompletedSteps   string // JSON array
    CompensatedSteps string // JSON array
    ContextData      string // JSON context
    CreatedAt        time.Time
    UpdatedAt        time.Time
}

type SagaStep interface {
    Name() string
    Execute(ctx context.Context, sagaContext map[string]interface{}) error
    Compensate(ctx context.Context, sagaContext map[string]interface{}) error
}

type SagaOrchestrator struct {
    db *gorm.DB
}

func NewSagaOrchestrator(db *gorm.DB) *SagaOrchestrator {
    return &SagaOrchestrator{db: db}
}

func (o *SagaOrchestrator) ExecuteSaga(ctx context.Context, sagaID string, steps []SagaStep, context map[string]interface{}) error {
    return o.db.Transaction(func(tx *gorm.DB) error {
        var saga SagaTransaction
        if err := tx.First(&saga, "id = ?", sagaID).Error; err != nil {
            if err == gorm.ErrRecordNotFound {
                // Create new saga
                contextJSON, _ := json.Marshal(context)
                saga = SagaTransaction{
                    ID:          sagaID,
                    Status:      StatusStarted,
                    ContextData: string(contextJSON),
                }
                if err := tx.Create(&saga).Error; err != nil {
                    return err
                }
            } else {
                return err
            }
        }
        
        // Parse completed steps
        var completedSteps []string
        if saga.CompletedSteps != "" {
            json.Unmarshal([]byte(saga.CompletedSteps), &completedSteps)
        }
        
        completedMap := make(map[string]bool)
        for _, step := range completedSteps {
            completedMap[step] = true
        }
        
        // Execute steps
        for _, step := range steps {
            if completedMap[step.Name()] {
                continue // Already completed
            }
            
            saga.CurrentStep = step.Name()
            saga.Status = StatusInProgress
            if err := tx.Save(&saga).Error; err != nil {
                return err
            }
            
            // Parse saga context
            var sagaContext map[string]interface{}
            json.Unmarshal([]byte(saga.ContextData), &sagaContext)
            
            // Execute step
            if err := step.Execute(ctx, sagaContext); err != nil {
                // Start compensation
                return o.compensateSaga(ctx, tx, sagaID, steps[:len(completedSteps)])
            }
            
            // Mark step as completed
            completedSteps = append(completedSteps, step.Name())
            completedJSON, _ := json.Marshal(completedSteps)
            saga.CompletedSteps = string(completedJSON)
            
            // Update context if modified
            updatedContext, _ := json.Marshal(sagaContext)
            saga.ContextData = string(updatedContext)
            
            if err := tx.Save(&saga).Error; err != nil {
                return err
            }
        }
        
        saga.Status = StatusCompleted
        return tx.Save(&saga).Error
    })
}

func (o *SagaOrchestrator) compensateSaga(ctx context.Context, tx *gorm.DB, sagaID string, completedSteps []SagaStep) error {
    var saga SagaTransaction
    if err := tx.First(&saga, "id = ?", sagaID).Error; err != nil {
        return err
    }
    
    saga.Status = StatusCompensating
    if err := tx.Save(&saga).Error; err != nil {
        return err
    }
    
    // Parse compensated steps
    var compensatedSteps []string
    if saga.CompensatedSteps != "" {
        json.Unmarshal([]byte(saga.CompensatedSteps), &compensatedSteps)
    }
    
    compensatedMap := make(map[string]bool)
    for _, step := range compensatedSteps {
        compensatedMap[step] = true
    }
    
    // Compensate in reverse order
    for i := len(completedSteps) - 1; i >= 0; i-- {
        step := completedSteps[i]
        
        if compensatedMap[step.Name()] {
            continue
        }
        
        var sagaContext map[string]interface{}
        json.Unmarshal([]byte(saga.ContextData), &sagaContext)
        
        if err := step.Compensate(ctx, sagaContext); err != nil {
            saga.Status = StatusFailed
            tx.Save(&saga)
            return fmt.Errorf("compensation failed for step %s: %w", step.Name(), err)
        }
        
        compensatedSteps = append(compensatedSteps, step.Name())
        compensatedJSON, _ := json.Marshal(compensatedSteps)
        saga.CompensatedSteps = string(compensatedJSON)
        
        if err := tx.Save(&saga).Error; err != nil {
            return err
        }
    }
    
    saga.Status = StatusCompensated
    return tx.Save(&saga).Error
}

// Concrete step implementations
type CreateOrderStep struct {
    orderService *OrderService
    request      CreateOrderRequest
}

func (s *CreateOrderStep) Name() string {
    return "CREATE_ORDER"
}

func (s *CreateOrderStep) Execute(ctx context.Context, sagaContext map[string]interface{}) error {
    order, err := s.orderService.CreateOrder(ctx, s.request)
    if err != nil {
        return err
    }
    
    // Store order ID in context for later steps
    sagaContext["orderId"] = order.ID
    return nil
}

func (s *CreateOrderStep) Compensate(ctx context.Context, sagaContext map[string]interface{}) error {
    if orderID, exists := sagaContext["orderId"]; exists {
        return s.orderService.CancelOrder(ctx, orderID.(uint))
    }
    return nil
}

// Usage
func ProcessOrderWithSaga(ctx context.Context, orchestrator *SagaOrchestrator, request CreateOrderRequest) error {
    sagaID := uuid.New().String()
    
    steps := []SagaStep{
        &CreateOrderStep{orderService, request},
        &ProcessPaymentStep{paymentService, request.CustomerID, request.TotalAmount},
        &ReserveInventoryStep{inventoryService, request.Items},
    }
    
    context := map[string]interface{}{
        "customerId": request.CustomerID,
        "requestId":  request.RequestID,
    }
    
    return orchestrator.ExecuteSaga(ctx, sagaID, steps, context)
}

// THIS IS FUNDAMENTALLY BROKEN - DON'T DO THIS
@Service
public class BrokenOrderService {
    
    @Transactional
    public void processOrder(CreateOrderRequest request) {
        Order order = orderRepository.save(new Order(request));
        
        // DANGER: Event published outside transaction boundary
        eventPublisher.publishEvent(new OrderCreatedEvent(order));
        
        // What if this line throws an exception?
        // Event is already published but transaction will rollback!
    }
    
    // ALSO BROKEN: Event first, then database
    @Transactional  
    public void processOrderEventFirst(CreateOrderRequest request) {
        Order order = new Order(request);
        
        // DANGER: Event published before persistence
        eventPublisher.publishEvent(new OrderCreatedEvent(order));
        
        // What if database save fails?
        // Event consumers will query for order that doesn't exist!
        orderRepository.save(order);
    }
}

// Outbox event entity
@Entity
@Table(name = "outbox_events")
public class OutboxEvent {
    @Id
    private String id;
    
    @Column(name = "event_type")
    private String eventType;
    
    @Column(name = "payload", columnDefinition = "TEXT")
    private String payload;
    
    @Column(name = "created_at")
    private Instant createdAt;
    
    @Column(name = "processed_at")
    private Instant processedAt;
    
    @Enumerated(EnumType.STRING)
    private OutboxStatus status;
    
    // constructors, getters, setters...
}

public enum OutboxStatus {
    PENDING, PROCESSED, FAILED
}

// Outbox repository
@Repository
public interface OutboxEventRepository extends JpaRepository<OutboxEvent, String> {
    @Query("SELECT e FROM OutboxEvent e WHERE e.status = :status ORDER BY e.createdAt ASC")
    List<OutboxEvent> findByStatusOrderByCreatedAt(@Param("status") OutboxStatus status);
    
    @Modifying
    @Query("UPDATE OutboxEvent e SET e.status = :status, e.processedAt = :processedAt WHERE e.id = :id")
    void updateStatus(@Param("id") String id, @Param("status") OutboxStatus status, @Param("processedAt") Instant processedAt);
}

// Corrected order service using outbox
@Service
public class TransactionalOrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private OutboxEventRepository outboxRepository;
    
    @Transactional
    public void processOrder(CreateOrderRequest request) {
        // 1. Process business logic
        Order order = new Order(request);
        order = orderRepository.save(order);
        
        // 2. Store event in same transaction
        OutboxEvent event = new OutboxEvent(
            UUID.randomUUID().toString(),
            "OrderCreated",
            serializeEvent(new OrderCreatedEvent(order)),
            Instant.now(),
            OutboxStatus.PENDING
        );
        outboxRepository.save(event);
        
        // Both order and event are committed atomically!
    }
    
    private String serializeEvent(Object event) {
        try {
            return objectMapper.writeValueAsString(event);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Event serialization failed", e);
        }
    }
}

// Event relay service
@Component
public class OutboxEventRelay {
    
    @Autowired
    private OutboxEventRepository outboxRepository;
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @Scheduled(fixedDelay = 1000) // Poll every second
    @Transactional
    public void processOutboxEvents() {
        List<OutboxEvent> pendingEvents = outboxRepository
            .findByStatusOrderByCreatedAt(OutboxStatus.PENDING);
        
        for (OutboxEvent outboxEvent : pendingEvents) {
            try {
                // Deserialize and publish the event
                Object event = deserializeEvent(outboxEvent.getEventType(), outboxEvent.getPayload());
                eventPublisher.publishEvent(event);
                
                // Mark as processed
                outboxRepository.updateStatus(
                    outboxEvent.getId(), 
                    OutboxStatus.PROCESSED, 
                    Instant.now()
                );
                
            } catch (Exception e) {
                log.error("Failed to process outbox event: " + outboxEvent.getId(), e);
                outboxRepository.updateStatus(
                    outboxEvent.getId(), 
                    OutboxStatus.FAILED, 
                    Instant.now()
                );
            }
        }
    }
}

// Outbox event model
type OutboxEvent struct {
    ID          string     `gorm:"primarykey"`
    EventType   string     `gorm:"not null"`
    Payload     string     `gorm:"type:text;not null"`
    CreatedAt   time.Time
    ProcessedAt *time.Time
    Status      OutboxStatus `gorm:"type:varchar(20);default:'PENDING'"`
}

type OutboxStatus string

const (
    OutboxStatusPending   OutboxStatus = "PENDING"
    OutboxStatusProcessed OutboxStatus = "PROCESSED" 
    OutboxStatusFailed    OutboxStatus = "FAILED"
)

// Service with outbox pattern
type OrderService struct {
    db           *gorm.DB
    eventRelay   *OutboxEventRelay
}

func (s *OrderService) ProcessOrder(ctx context.Context, request CreateOrderRequest) (*Order, error) {
    var order *Order
    
    err := s.db.Transaction(func(tx *gorm.DB) error {
        // 1. Create order
        order = &Order{
            CustomerID:  request.CustomerID,
            TotalAmount: request.TotalAmount,
            Status:      "CONFIRMED",
        }
        
        if err := tx.Create(order).Error; err != nil {
            return fmt.Errorf("failed to create order: %w", err)
        }
        
        // 2. Store event in same transaction
        eventPayload, err := json.Marshal(OrderCreatedEvent{
            OrderID:     order.ID,
            CustomerID:  order.CustomerID,
            TotalAmount: order.TotalAmount,
        })
        if err != nil {
            return fmt.Errorf("failed to serialize event: %w", err)
        }
        
        outboxEvent := &OutboxEvent{
            ID:        uuid.New().String(),
            EventType: "OrderCreated",
            Payload:   string(eventPayload),
            CreatedAt: time.Now(),
            Status:    OutboxStatusPending,
        }
        
        if err := tx.Create(outboxEvent).Error; err != nil {
            return fmt.Errorf("failed to store outbox event: %w", err)
        }
        
        return nil
    })
    
    return order, err
}

// Event relay service
type OutboxEventRelay struct {
    db            *gorm.DB
    eventPublisher EventPublisher
    ticker         *time.Ticker
    done           chan bool
}

func NewOutboxEventRelay(db *gorm.DB, publisher EventPublisher) *OutboxEventRelay {
    return &OutboxEventRelay{
        db:             db,
        eventPublisher: publisher,
        ticker:         time.NewTicker(1 * time.Second),
        done:           make(chan bool),
    }
}

func (r *OutboxEventRelay) Start(ctx context.Context) {
    go func() {
        for {
            select {
            case <-r.ticker.C:
                r.processOutboxEvents(ctx)
            case <-r.done:
                return
            case <-ctx.Done():
                return
            }
        }
    }()
}

func (r *OutboxEventRelay) processOutboxEvents(ctx context.Context) {
    var events []OutboxEvent
    
    // Find pending events
    if err := r.db.Where("status = ?", OutboxStatusPending).
        Order("created_at ASC").
        Limit(100).
        Find(&events).Error; err != nil {
        log.Printf("Failed to fetch outbox events: %v", err)
        return
    }
    
    for _, event := range events {
        if err := r.processEvent(ctx, event); err != nil {
            log.Printf("Failed to process event %s: %v", event.ID, err)
            
            // Mark as failed
            now := time.Now()
            r.db.Model(&event).Updates(OutboxEvent{
                Status:      OutboxStatusFailed,
                ProcessedAt: &now,
            })
        } else {
            // Mark as processed
            now := time.Now()
            r.db.Model(&event).Updates(OutboxEvent{
                Status:      OutboxStatusProcessed,
                ProcessedAt: &now,
            })
        }
    }
}

func (r *OutboxEventRelay) processEvent(ctx context.Context, event OutboxEvent) error {
    // Deserialize and publish the event
    switch event.EventType {
    case "OrderCreated":
        var orderEvent OrderCreatedEvent
        if err := json.Unmarshal([]byte(event.Payload), &orderEvent); err != nil {
            return err
        }
        return r.eventPublisher.Publish(ctx, orderEvent)
        
    default:
        return fmt.Errorf("unknown event type: %s", event.EventType)
    }
}

// Outbox event model
use diesel::prelude::*;
use serde::{Deserialize, Serialize};
use chrono::{DateTime, Utc};

#[derive(Debug, Clone, Queryable, Insertable, Serialize, Deserialize)]
#[diesel(table_name = outbox_events)]
pub struct OutboxEvent {
    pub id: String,
    pub event_type: String,
    pub payload: String,
    pub created_at: DateTime<Utc>,
    pub processed_at: Option<DateTime<Utc>>,
    pub status: OutboxStatus,
}

#[derive(Debug, Clone, Serialize, Deserialize, diesel_derive_enum::DbEnum)]
#[ExistingTypePath = "crate::schema::sql_types::OutboxStatus"]
pub enum OutboxStatus {
    Pending,
    Processed,
    Failed,
}

// Service with outbox
impl OrderService {
    transactional! {
        fn process_order_with_outbox(request: CreateOrderRequest) -> Order {
            use crate::schema::orders::dsl::*;
            use crate::schema::outbox_events::dsl::*;
            
            // 1. Create order
            let new_order = NewOrder {
                customer_id: &request.customer_id,
                total_amount: request.total_amount,
                status: "CONFIRMED",
            };
            
            let order: Order = diesel::insert_into(orders)
                .values(&new_order)
                .get_result(conn)
                .map_err(TransactionError::Database)?;
            
            // 2. Create event in same transaction
            let event_payload = serde_json::to_string(&OrderCreatedEvent {
                order_id: order.id,
                customer_id: order.customer_id.clone(),
                total_amount: order.total_amount,
            }).map_err(|e| TransactionError::Business(format!("Event serialization failed: {}", e)))?;
            
            let outbox_event = OutboxEvent {
                id: uuid::Uuid::new_v4().to_string(),
                event_type: "OrderCreated".to_string(),
                payload: event_payload,
                created_at: Utc::now(),
                processed_at: None,
                status: OutboxStatus::Pending,
            };
            
            diesel::insert_into(outbox_events)
                .values(&outbox_event)
                .execute(conn)
                .map_err(TransactionError::Database)?;
            
            Ok(order)
        }
    }
}

// Event relay service
pub struct OutboxEventRelay {
    pool: Pool<ConnectionManager<PgConnection>>,
    event_publisher: Arc<dyn EventPublisher>,
}

impl OutboxEventRelay {
    pub async fn start(&self, mut shutdown: tokio::sync::broadcast::Receiver<()>) {
        let mut interval = tokio::time::interval(Duration::from_secs(1));
        
        loop {
            tokio::select! {
                _ = interval.tick() => {
                    if let Err(e) = self.process_outbox_events().await {
                        tracing::error!("Failed to process outbox events: {}", e);
                    }
                }
                _ = shutdown.recv() => {
                    tracing::info!("Outbox relay shutting down");
                    break;
                }
            }
        }
    }
    
    async fn process_outbox_events(&self) -> Result<(), Box<dyn std::error::Error>> {
        use crate::schema::outbox_events::dsl::*;
        
        let mut conn = self.pool.get()?;
        
        let pending_events: Vec<OutboxEvent> = outbox_events
            .filter(status.eq(OutboxStatus::Pending))
            .order(created_at.asc())
            .limit(100)
            .load(&mut conn)?;
        
        for event in pending_events {
            match self.process_single_event(&event).await {
                Ok(_) => {
                    // Mark as processed
                    let now = Utc::now();
                    diesel::update(outbox_events.filter(id.eq(&event.id)))
                        .set((
                            status.eq(OutboxStatus::Processed),
                            processed_at.eq(Some(now))
                        ))
                        .execute(&mut conn)?;
                }
                Err(e) => {
                    tracing::error!("Failed to process event {}: {}", event.id, e);
                    let now = Utc::now();
                    diesel::update(outbox_events.filter(id.eq(&event.id)))
                        .set((
                            status.eq(OutboxStatus::Failed),
                            processed_at.eq(Some(now))
                        ))
                        .execute(&mut conn)?;
                }
            }
        }
        
        Ok(())
    }
}

// CDC-based event publisher using Debezium
@Component
public class CDCEventHandler {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @KafkaListener(topics = "dbserver.public.orders")
    public void handleOrderChange(ConsumerRecord<String, String> record) {
        try {
            // Parse CDC event
            JsonNode changeEvent = objectMapper.readTree(record.value());
            String operation = changeEvent.get("op").asText(); // c=create, u=update, d=delete
            
            if ("c".equals(operation)) {
                JsonNode after = changeEvent.get("after");
                
                OrderCreatedEvent event = OrderCreatedEvent.builder()
                    .orderId(after.get("id").asLong())
                    .customerId(after.get("customer_id").asText())
                    .totalAmount(after.get("total_amount").asLong())
                    .build();
                
                // Publish to business event topic
                kafkaTemplate.send("order-events", 
                    event.getOrderId().toString(), 
                    objectMapper.writeValueAsString(event));
            }
            
        } catch (Exception e) {
            log.error("Failed to process CDC event", e);
        }
    }
}

// Event store as the primary persistence
@Entity
public class EventStore {
    @Id
    private String eventId;
    
    private String aggregateId;
    private String eventType;
    private String eventData;
    private Long version;
    private Instant timestamp;
    
    // getters, setters...
}

@Service
public class EventSourcedOrderService {
    
    @Transactional
    public void processOrder(CreateOrderRequest request) {
        String orderId = UUID.randomUUID().toString();
        
        // Store event - this IS the transaction
        OrderCreatedEvent event = new OrderCreatedEvent(orderId, request);
        
        EventStore eventRecord = new EventStore(
            UUID.randomUUID().toString(),
            orderId,
            "OrderCreated",
            objectMapper.writeValueAsString(event),
            1L,
            Instant.now()
        );
        
        eventStoreRepository.save(eventRecord);
        
        // Publish immediately - event is already persisted
        eventPublisher.publishEvent(event);
        
        // Read model is updated asynchronously via event handlers
    }
}

// DANGEROUS CONFIGURATION - This will cause deadlocks
@Configuration
public class ProblematicDataSourceConfig {
    
    @Bean
    public DataSource dataSource() {
        HikariConfig config = new HikariConfig();
        config.setMaximumPoolSize(10);        // Only 10 connections
        config.setConnectionTimeout(30000);   // 30 second timeout
        return new HikariDataSource(config);
    }
    
    @Bean
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(50);         // 50 threads!
        executor.setMaxPoolSize(100);         // Up to 100 threads!
        return executor;
    }
}

// This service will cause connection pool starvation
@Service
public class ProblematicBulkService {
    
    @Async
    @Transactional
    public CompletableFuture<Void> processBatch(List<Order> orders) {
        // 100 threads trying to get 10 connections = deadlock
        for (Order order : orders) {
            // Long-running transaction holds connection
            processOrder(order);
            
            // Calling another transactional method = needs another connection
            auditService.logOrderProcessing(order); // DEADLOCK RISK!
        }
        return CompletableFuture.completedFuture(null);
    }
}

@Configuration
public class OptimalDataSourceConfig {
    
    @Value("${app.database.max-connections:20}")
    private int maxConnections;
    
    @Bean
    public DataSource dataSource() {
        HikariConfig config = new HikariConfig();
        
        // Rule of thumb: connections >= active threads + buffer
        config.setMaximumPoolSize(maxConnections);
        config.setMinimumIdle(5);
        
        // Prevent connection hoarding
        config.setConnectionTimeout(5000);     // 5 seconds
        config.setIdleTimeout(300000);         // 5 minutes
        config.setMaxLifetime(1200000);        // 20 minutes
        
        // Detect leaked connections
        config.setLeakDetectionThreshold(60000); // 1 minute
        
        return new HikariDataSource(config);
    }
    
    @Bean
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // Keep thread pool smaller than connection pool
        executor.setCorePoolSize(maxConnections - 5);
        executor.setMaxPoolSize(maxConnections);
        executor.setQueueCapacity(100);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
}

// Connection-aware bulk processing
@Service
public class OptimalBulkService {
    
    @Transactional
    public void processBatchOptimized(List<Order> orders) {
        int batchSize = 50; // Tune based on connection pool size
        
        for (int i = 0; i < orders.size(); i += batchSize) {
            List<Order> batch = orders.subList(i, Math.min(i + batchSize, orders.size()));
            
            // Process batch in single transaction
            processBatchInternal(batch);
            
            // Flush and clear to prevent memory issues
            entityManager.flush();
            entityManager.clear();
        }
    }
    
    private void processBatchInternal(List<Order> batch) {
        // Bulk operations to reduce connection time
        orderRepository.saveAll(batch);
        
        // Batch audit logging
        List<AuditLog> auditLogs = batch.stream()
            .map(order -> new AuditLog("ORDER_PROCESSED", order.getId()))
            .collect(Collectors.toList());
        auditRepository.saveAll(auditLogs);
    }
}

// WRONG - Each record in its own transaction (SLOW!)
@Service
public class SlowETLService {
    
    public void importOrders(InputStream csvFile) {
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(csvFile))) {
            String line;
            while ((line = reader.readLine()) != null) {
                processOrderRecord(line); // Each call = new transaction!
            }
        }
    }
    
    @Transactional // New transaction per record = DISASTER
    private void processOrderRecord(String csvLine) {
        Order order = parseOrderFromCsv(csvLine);
        orderRepository.save(order);
        // Commit happens here - thousands of commits!
    }
}

// CORRECT - Batch processing with optimal transaction boundaries
@Service 
public class FastETLService {
    
    @Value("${etl.batch-size:1000}")
    private int batchSize;
    
    public void importOrdersOptimized(InputStream csvFile) {
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(csvFile))) {
            List<Order> batch = new ArrayList<>(batchSize);
            String line;
            
            while ((line = reader.readLine()) != null) {
                Order order = parseOrderFromCsv(line);
                batch.add(order);
                
                if (batch.size() >= batchSize) {
                    processBatch(batch);
                    batch.clear();
                }
            }
            
            // Process remaining records
            if (!batch.isEmpty()) {
                processBatch(batch);
            }
        }
    }
    
    @Transactional
    private void processBatch(List<Order> orders) {
        // Single transaction for entire batch
        orderRepository.saveAll(orders);
        
        // Bulk validation and error handling
        validateBatch(orders);
        
        // Bulk audit logging
        createAuditEntries(orders);
    }
}

// Connection pool configuration with proper sizing
func setupDatabase() *gorm.DB {
    dsn := "host=localhost user=postgres password=secret dbname=orders port=5432 sslmode=disable"
    
    db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
    if err != nil {
        log.Fatal("Failed to connect to database:", err)
    }
    
    sqlDB, err := db.DB()
    if err != nil {
        log.Fatal("Failed to get SQL DB:", err)
    }
    
    // Critical: Match pool size to application concurrency
    sqlDB.SetMaxOpenConns(20)        // Maximum connections
    sqlDB.SetMaxIdleConns(5)         // Idle connections
    sqlDB.SetConnMaxLifetime(time.Hour) // Connection lifetime
    sqlDB.SetConnMaxIdleTime(time.Minute * 30) // Idle timeout
    
    return db
}

// ETL with proper batching
type ETLService struct {
    db        *gorm.DB
    batchSize int
}

func (s *ETLService) ImportOrders(csvFile io.Reader) error {
    scanner := bufio.NewScanner(csvFile)
    batch := make([]*Order, 0, s.batchSize)
    
    for scanner.Scan() {
        order, err := s.parseOrderFromCSV(scanner.Text())
        if err != nil {
            log.Printf("Failed to parse order: %v", err)
            continue
        }
        
        batch = append(batch, order)
        
        if len(batch) >= s.batchSize {
            if err := s.processBatch(batch); err != nil {
                return fmt.Errorf("batch processing failed: %w", err)
            }
            batch = batch[:0] // Reset slice but keep capacity
        }
    }
    
    // Process remaining orders
    if len(batch) > 0 {
        return s.processBatch(batch)
    }
    
    return scanner.Err()
}

func (s *ETLService) processBatch(orders []*Order) error {
    return s.db.Transaction(func(tx *gorm.DB) error {
        // Batch insert for performance
        if err := tx.CreateInBatches(orders, len(orders)).Error; err != nil {
            return err
        }
        
        // Bulk audit logging in same transaction
        auditLogs := make([]*AuditLog, len(orders))
        for i, order := range orders {
            auditLogs[i] = &AuditLog{
                Action:    "ORDER_IMPORTED",
                OrderID:   order.ID,
                Timestamp: time.Now(),
            }
        }
        
        return tx.CreateInBatches(auditLogs, len(auditLogs)).Error
    })
}

@Service
public class DeadlockAwareAccountService {
    
    @Retryable(
        value = {CannotAcquireLockException.class, DeadlockLoserDataAccessException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 100, multiplier = 2, random = true)
    )
    @Transactional(isolation = Isolation.READ_COMMITTED)
    public void transferFunds(String fromAccountId, String toAccountId, BigDecimal amount) {
        
        // CRITICAL: Always acquire locks in consistent order to prevent deadlocks
        String firstId = fromAccountId.compareTo(toAccountId) < 0 ? fromAccountId : toAccountId;
        String secondId = fromAccountId.compareTo(toAccountId) < 0 ? toAccountId : fromAccountId;
        
        // Lock accounts in alphabetical order
        Account firstAccount = accountRepository.findByIdForUpdate(firstId);
        Account secondAccount = accountRepository.findByIdForUpdate(secondId);
        
        // Determine which is from/to after locking
        Account fromAccount = fromAccountId.equals(firstId) ? firstAccount : secondAccount;
        Account toAccount = fromAccountId.equals(firstId) ? secondAccount : firstAccount;
        
        if (fromAccount.getBalance().compareTo(amount) < 0) {
            throw new InsufficientFundsException();
        }
        
        fromAccount.setBalance(fromAccount.getBalance().subtract(amount));
        toAccount.setBalance(toAccount.getBalance().add(amount));
        
        accountRepository.save(fromAccount);
        accountRepository.save(toAccount);
    }
}

// Custom repository with explicit locking
@Repository
public interface AccountRepository extends JpaRepository<Account, String> {
    
    @Query("SELECT a FROM Account a WHERE a.id = :id")
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    Account findByIdForUpdate(@Param("id") String id);
    
    // Timeout-based locking to prevent indefinite waits
    @QueryHints({@QueryHint(name = "javax.persistence.lock.timeout", value = "5000")})
    @Query("SELECT a FROM Account a WHERE a.id = :id")
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    Account findByIdForUpdateWithTimeout(@Param("id") String id);
}

// Deadlock-aware service with timeout and retry
type AccountService struct {
    db           *gorm.DB
    lockTimeout  time.Duration
    retryAttempts int
}

func (s *AccountService) TransferFundsWithDeadlockHandling(
    ctx context.Context, 
    fromAccountID, toAccountID string, 
    amount int64,
) error {
    
    for attempt := 0; attempt < s.retryAttempts; attempt++ {
        err := s.attemptTransfer(ctx, fromAccountID, toAccountID, amount)
        
        if err == nil {
            return nil // Success
        }
        
        // Check if it's a deadlock or timeout error
        if s.isRetryableError(err) {
            // Exponential backoff with jitter
            backoff := time.Duration(attempt+1) * 100 * time.Millisecond
            jitter := time.Duration(rand.Intn(100)) * time.Millisecond
            
            select {
            case <-time.After(backoff + jitter):
                continue // Retry
            case <-ctx.Done():
                return ctx.Err()
            }
        }
        
        return err // Non-retryable error
    }
    
    return fmt.Errorf("transfer failed after %d attempts", s.retryAttempts)
}

func (s *AccountService) attemptTransfer(
    ctx context.Context, 
    fromAccountID, toAccountID string, 
    amount int64,
) error {
    
    // Set transaction timeout
    timeoutCtx, cancel := context.WithTimeout(ctx, s.lockTimeout)
    defer cancel()
    
    return s.db.WithContext(timeoutCtx).Transaction(func(tx *gorm.DB) error {
        // Lock in consistent order
        firstID, secondID := fromAccountID, toAccountID
        if fromAccountID > toAccountID {
            firstID, secondID = toAccountID, fromAccountID
        }
        
        var fromAccount, toAccount Account
        
        // Lock first account
        if err := tx.Set("gorm:query_option", "FOR UPDATE NOWAIT").
            Where("id = ?", firstID).First(&fromAccount).Error; err != nil {
            return err
        }
        
        // Lock second account
        if err := tx.Set("gorm:query_option", "FOR UPDATE NOWAIT").
            Where("id = ?", secondID).First(&toAccount).Error; err != nil {
            return err
        }
        
        // Ensure we have the right accounts
        if fromAccount.ID != fromAccountID {
            fromAccount, toAccount = toAccount, fromAccount
        }
        
        if fromAccount.Balance < amount {
            return fmt.Errorf("insufficient funds")
        }
        
        fromAccount.Balance -= amount
        toAccount.Balance += amount
        
        if err := tx.Save(&fromAccount).Error; err != nil {
            return err
        }
        
        return tx.Save(&toAccount).Error
    })
}

func (s *AccountService) isRetryableError(err error) bool {
    errStr := strings.ToLower(err.Error())
    
    // Common deadlock/timeout indicators
    retryablePatterns := []string{
        "deadlock detected",
        "lock timeout",
        "could not obtain lock",
        "serialization failure",
        "concurrent update",
    }
    
    for _, pattern := range retryablePatterns {
        if strings.Contains(errStr, pattern) {
            return true
        }
    }
    
    return false
}

// Command side - strict ACID transactions
@Service
public class OrderCommandService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private EventPublisher eventPublisher;
    
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public void createOrder(CreateOrderCommand command) {
        // Validate command
        validateOrderCommand(command);
        
        Order order = new Order(command);
        order = orderRepository.save(order);
        
        // Publish event for read model update
        eventPublisher.publish(new OrderCreatedEvent(order));
    }
    
    @Transactional(isolation = Isolation.READ_COMMITTED)
    public void updateOrderStatus(UpdateOrderStatusCommand command) {
        Order order = orderRepository.findById(command.getOrderId())
            .orElseThrow(() -> new OrderNotFoundException());
        
        order.updateStatus(command.getNewStatus());
        orderRepository.save(order);
        
        eventPublisher.publish(new OrderStatusChangedEvent(order));
    }
}

// Query side - optimized for reads, eventual consistency
@Service
public class OrderQueryService {
    
    @Autowired
    private OrderReadModelRepository readModelRepository;
    
    // Read-only transaction for consistency within the query
    @Transactional(readOnly = true)
    public OrderSummary getOrderSummary(String customerId) {
        List<OrderReadModel> orders = readModelRepository
            .findByCustomerId(customerId);
        
        return OrderSummary.builder()
            .totalOrders(orders.size())
            .totalAmount(orders.stream()
                .mapToLong(OrderReadModel::getTotalAmount)
                .sum())
            .recentOrders(orders.stream()
                .sorted((o1, o2) -> o2.getCreatedAt().compareTo(o1.getCreatedAt()))
                .limit(10)
                .collect(Collectors.toList()))
            .build();
    }
    
    // No transaction needed for simple lookups
    public List<OrderReadModel> searchOrders(OrderSearchCriteria criteria) {
        return readModelRepository.search(criteria);
    }
}

// Event handler updates read model asynchronously
@Component
public class OrderReadModelUpdater {
    
    @EventHandler
    @Async
    @Transactional
    public void handle(OrderCreatedEvent event) {
        OrderReadModel readModel = OrderReadModel.builder()
            .orderId(event.getOrderId())
            .customerId(event.getCustomerId())
            .totalAmount(event.getTotalAmount())
            .status(event.getStatus())
            .createdAt(event.getCreatedAt())
            .build();
        
        readModelRepository.save(readModel);
    }
    
    @EventHandler
    @Async
    @Transactional
    public void handle(OrderStatusChangedEvent event) {
        readModelRepository.updateStatus(
            event.getOrderId(), 
            event.getNewStatus(),
            event.getUpdatedAt()
        );
    }
}

@Configuration
public class MultiRegionDataSourceConfig {
    
    // Primary database in same region
    @Bean
    @Primary
    @ConfigurationProperties("spring.datasource.primary")
    public DataSource primaryDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    // Read replica in same availability zone
    @Bean
    @ConfigurationProperties("spring.datasource.read-replica")
    public DataSource readReplicaDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    // Cross-region backup for disaster recovery
    @Bean
    @ConfigurationProperties("spring.datasource.cross-region")
    public DataSource crossRegionDataSource() {
        return DataSourceBuilder.create().build();
    }
}

@Service
public class LocalityAwareOrderService {
    
    @Autowired
    @Qualifier("primaryDataSource")
    private DataSource writeDataSource;
    
    @Autowired
    @Qualifier("readReplicaDataSource") 
    private DataSource readDataSource;
    
    // Writes go to primary in same region
    @Transactional("primaryTransactionManager")
    public void createOrder(CreateOrderRequest request) {
        // Fast writes - same AZ latency ~1ms
        Order order = new Order(request);
        orderRepository.save(order);
    }
    
    // Reads from local replica
    @Transactional(value = "readReplicaTransactionManager", readOnly = true)
    public List<Order> getOrderHistory(String customerId) {
        // Even faster reads - same rack latency ~0.1ms
        return orderRepository.findByCustomerId(customerId);
    }
    
    // Critical path optimization
    @Transactional(timeout = 5) // Fail fast if network issues
    public void processUrgentOrder(UrgentOrderRequest request) {
        // Use connection pooling and keep-alive for predictable latency
        processOrder(request);
    }
}

type RegionalDatabaseConfig struct {
    primaryDB    *gorm.DB
    readReplica  *gorm.DB
    crossRegion  *gorm.DB
}

func NewRegionalDatabaseConfig(region string) *RegionalDatabaseConfig {
    // Select database endpoints based on current region
    primaryDSN := fmt.Sprintf("host=%s-primary.db user=app", region)
    readDSN := fmt.Sprintf("host=%s-read.db user=app", region)
    backupDSN := fmt.Sprintf("host=%s-backup.db user=app", getBackupRegion(region))
    
    return &RegionalDatabaseConfig{
        primaryDB:   connectWithLatencyOptimization(primaryDSN),
        readReplica: connectWithLatencyOptimization(readDSN),
        crossRegion: connectWithLatencyOptimization(backupDSN),
    }
}

func connectWithLatencyOptimization(dsn string) *gorm.DB {
    db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
    if err != nil {
        log.Fatal("Database connection failed:", err)
    }
    
    sqlDB, _ := db.DB()
    
    // Optimize for low-latency
    sqlDB.SetMaxOpenConns(20)
    sqlDB.SetMaxIdleConns(20)        // Keep connections alive
    sqlDB.SetConnMaxIdleTime(0)      // Never close idle connections
    sqlDB.SetConnMaxLifetime(time.Hour) // Rotate connections hourly
    
    return db
}

type OrderService struct {
    config *RegionalDatabaseConfig
}

func (s *OrderService) CreateOrder(ctx context.Context, request CreateOrderRequest) error {
    // Use primary database for writes
    return s.config.primaryDB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        order := &Order{
            CustomerID:  request.CustomerID,
            TotalAmount: request.TotalAmount,
            Status:      "PENDING",
        }
        
        return tx.Create(order).Error
    })
}

func (s *OrderService) GetOrderHistory(ctx context.Context, customerID string) ([]Order, error) {
    var orders []Order
    
    // Use read replica for queries - better performance
    err := s.config.readReplica.WithContext(ctx).
        Where("customer_id = ?", customerID).
        Order("created_at DESC").
        Find(&orders).Error
        
    return orders, err
}

// ACID Consistency: Data integrity constraints
@Entity
public class BankAccount {
    @Id
    private String accountId;
    
    @Column(nullable = false)
    @Min(0) // ACID Consistency: Balance cannot be negative
    private BigDecimal balance;
    
    @Version
    private Long version;
}

// CAP Consistency: Linearizability across distributed nodes
@Service
public class DistributedAccountService {
    
    // This requires coordination across all replicas
    @Transactional
    public void withdraw(String accountId, BigDecimal amount) {
        // All replicas must see this change at the same logical time
        // = CAP Consistency (Linearizability)
        
        BankAccount account = accountRepository.findById(accountId);
        if (account.getBalance().compareTo(amount) < 0) {
            throw new InsufficientFundsException(); // ACID Consistency violated
        }
        account.setBalance(account.getBalance().subtract(amount));
        accountRepository.save(account); // Maintains ACID invariants
    }
}

// DynamoDB allows you to choose consistency per operation
@Service
public class DynamoConsistencyService {
    
    @Autowired
    private DynamoDBClient dynamoClient;
    
    // Eventually consistent read - faster, cheaper, may be stale
    public Order getOrder(String orderId) {
        GetItemRequest request = GetItemRequest.builder()
            .tableName("Orders")
            .key(Map.of("orderId", AttributeValue.builder().s(orderId).build()))
            .consistentRead(false) // Eventually consistent
            .build();
            
        return dynamoClient.getItem(request);
    }
    
    // Strongly consistent read - slower, more expensive, always latest
    public Order getOrderStrong(String orderId) {
        GetItemRequest request = GetItemRequest.builder()
            .tableName("Orders")
            .key(Map.of("orderId", AttributeValue.builder().s(orderId).build()))
            .consistentRead(true) // Strongly consistent = linearizable
            .build();
            
        return dynamoClient.getItem(request);
    }
    
    // Quorum-based tunable consistency
    public void updateOrderWithQuorum(String orderId, String newStatus) {
        // W + R > N guarantees consistency
        // W=2, R=2, N=3 means writes must succeed on 2 nodes,
        // reads must check 2 nodes, guaranteeing overlap
        
        UpdateItemRequest request = UpdateItemRequest.builder()
            .tableName("Orders")
            .key(Map.of("orderId", AttributeValue.builder().s(orderId).build()))
            .updateExpression("SET #status = :status")
            .expressionAttributeNames(Map.of("#status", "status"))
            .expressionAttributeValues(Map.of(":status", AttributeValue.builder().s(newStatus).build()))
            .build();
            
        dynamoClient.updateItem(request);
    }
}

// Distributed lock implementation for financial operations
@Service
public class DynamoDistributedLockService {
    
    @Autowired
    private DynamoDBClient dynamoClient;
    
    private static final String LOCK_TABLE = "distributed_locks";
    private static final int LOCK_TIMEOUT_SECONDS = 30;
    
    public boolean acquireTransferLock(String fromAccount, String toAccount, String requestId) {
        // Always lock accounts in consistent order to prevent deadlocks
        List<String> sortedAccounts = Arrays.asList(fromAccount, toAccount)
            .stream()
            .sorted()
            .collect(Collectors.toList());
        
        String lockKey = String.join(":", sortedAccounts);
        long expiryTime = System.currentTimeMillis() + (LOCK_TIMEOUT_SECONDS * 1000);
        
        try {
            PutItemRequest request = PutItemRequest.builder()
                .tableName(LOCK_TABLE)
                .item(Map.of(
                    "lock_key", AttributeValue.builder().s(lockKey).build(),
                    "owner", AttributeValue.builder().s(requestId).build(),
                    "expires_at", AttributeValue.builder().n(String.valueOf(expiryTime)).build(),
                    "accounts", AttributeValue.builder().ss(sortedAccounts).build()
                ))
                .conditionExpression("attribute_not_exists(lock_key) OR expires_at < :current_time")
                .expressionAttributeValues(Map.of(
                    ":current_time", AttributeValue.builder().n(String.valueOf(System.currentTimeMillis())).build()
                ))
                .build();
                
            dynamoClient.putItem(request);
            return true;
            
        } catch (ConditionalCheckFailedException e) {
            return false; // Lock already held
        }
    }
    
    public void releaseLock(String fromAccount, String toAccount, String requestId) {
        List<String> sortedAccounts = Arrays.asList(fromAccount, toAccount)
            .stream()
            .sorted()
            .collect(Collectors.toList());
        
        String lockKey = String.join(":", sortedAccounts);
        
        DeleteItemRequest request = DeleteItemRequest.builder()
            .tableName(LOCK_TABLE)
            .key(Map.of("lock_key", AttributeValue.builder().s(lockKey).build()))
            .conditionExpression("owner = :request_id")
            .expressionAttributeValues(Map.of(
                ":request_id", AttributeValue.builder().s(requestId).build()
            ))
            .build();
            
        try {
            dynamoClient.deleteItem(request);
        } catch (ConditionalCheckFailedException e) {
            log.warn("Failed to release lock - may have been taken by another process: {}", lockKey);
        }
    }
}

// Usage in financial service
@Service
public class DynamoFinancialService {
    
    @Autowired
    private DynamoDistributedLockService lockService;
    
    public TransferResult transferFunds(String fromAccount, String toAccount, BigDecimal amount) {
        String requestId = UUID.randomUUID().toString();
        
        // Acquire distributed lock to prevent concurrent transfers
        if (!lockService.acquireTransferLock(fromAccount, toAccount, requestId)) {
            return TransferResult.rejected("Another transfer in progress for these accounts");
        }
        
        try {
            // Now safe to perform transfer with DynamoDB transactions
            Collection<TransactionWriteRequest> actions = new ArrayList<>();
            
            // Conditional update for from account
            actions.add(TransactionWriteRequest.builder()
                .update(Update.builder()
                    .tableName("accounts")
                    .key(Map.of("account_id", AttributeValue.builder().s(fromAccount).build()))
                    .updateExpression("SET balance = balance - :amount")
                    .conditionExpression("balance >= :amount") // Prevent negative balance
                    .expressionAttributeValues(Map.of(
                        ":amount", AttributeValue.builder().n(amount.toString()).build()
                    ))
                    .build())
                .build());
            
            // Conditional update for to account  
            actions.add(TransactionWriteRequest.builder()
                .update(Update.builder()
                    .tableName("accounts")
                    .key(Map.of("account_id", AttributeValue.builder().s(toAccount).build()))
                    .updateExpression("SET balance = balance + :amount")
                    .expressionAttributeValues(Map.of(
                        ":amount", AttributeValue.builder().n(amount.toString()).build()
                    ))
                    .build())
                .build());
            
            // Execute atomic transaction
            TransactWriteItemsRequest txRequest = TransactWriteItemsRequest.builder()
                .transactItems(actions)
                .build();
                
            dynamoClient.transactWriteItems(txRequest);
            return TransferResult.success();
            
        } catch (TransactionCanceledException e) {
            return TransferResult.rejected("Transfer failed - likely insufficient funds");
        } finally {
            lockService.releaseLock(fromAccount, toAccount, requestId);
        }
    }
}

// Real-world CAP tradeoffs in microservices
@Service
public class CAPAwareOrderService {
    
    // Partition tolerance is mandatory in distributed systems
    // So you choose: Consistency OR Availability
    
    // CP System: Choose Consistency over Availability
    @CircuitBreaker(fallbackMethod = "fallbackCreateOrder")
    public OrderResponse createOrderCP(CreateOrderRequest request) {
        try {
            // All replicas must acknowledge - may fail if partition occurs
            return orderService.createOrderWithQuorum(request);
        } catch (PartitionException e) {
            // System becomes unavailable during partition
            throw new ServiceUnavailableException("Cannot guarantee consistency during partition");
        }
    }
    
    // AP System: Choose Availability over Consistency  
    public OrderResponse createOrderAP(CreateOrderRequest request) {
        try {
            // Try strong consistency first
            return orderService.createOrderWithQuorum(request);
        } catch (PartitionException e) {
            // Fall back to available replicas - may create conflicts
            log.warn("Partition detected, falling back to eventual consistency");
            return orderService.createOrderEventual(request);
        }
    }
    
    // Fallback for CP system
    public OrderResponse fallbackCreateOrder(CreateOrderRequest request, Exception ex) {
        // Return cached response or friendly error
        return OrderResponse.builder()
            .status("DEFERRED")
            .message("Order will be processed when system recovers")
            .build();
    }
}

// CP System: Prioritize Consistency over Availability
@Service
public class FinancialConsistencyService {
    
    @CircuitBreaker(fallbackMethod = "rejectTransaction")
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public TransferResult transferFunds(String fromAccount, String toAccount, BigDecimal amount) {
        // Requires consensus across all replicas
        // May become unavailable during partitions
        
        if (!distributedLockService.acquireLock(fromAccount, toAccount)) {
            throw new ServiceUnavailableException("Cannot acquire distributed lock");
        }
        
        try {
            // This operation requires strong consistency across all nodes
            Account from = accountService.getAccountWithQuorum(fromAccount);
            Account to = accountService.getAccountWithQuorum(toAccount);
            
            if (from.getBalance().compareTo(amount) < 0) {
                throw new InsufficientFundsException();
            }
            
            // Both updates must succeed on majority of replicas
            accountService.updateWithQuorum(fromAccount, from.getBalance().subtract(amount));
            accountService.updateWithQuorum(toAccount, to.getBalance().add(amount));
            
            return TransferResult.success();
            
        } finally {
            distributedLockService.releaseLock(fromAccount, toAccount);
        }
    }
    
    public TransferResult rejectTransaction(String fromAccount, String toAccount, 
                                          BigDecimal amount, Exception ex) {
        // During partitions, reject rather than risk double spending
        return TransferResult.rejected("Cannot guarantee consistency during network partition");
    }
}

// DynamoDB with optimistic locking using version fields
@DynamoDBTable(tableName = "Orders")
public class DynamoOrder {
    
    @DynamoDBHashKey
    private String orderId;
    
    @DynamoDBAttribute
    private String customerId;
    
    @DynamoDBAttribute
    private Long totalAmount;
    
    @DynamoDBAttribute
    private String status;
    
    @DynamoDBVersionAttribute
    private Long version; // DynamoDB handles optimistic locking
    
    // getters, setters...
}

@Service
public class DynamoOrderService {
    
    @Autowired
    private DynamoDBMapper dynamoMapper;
    
    // Optimistic concurrency with retry
    @Retryable(
        value = {ConditionalCheckFailedException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 100, multiplier = 2)
    )
    public void updateOrderStatus(String orderId, String newStatus) {
        try {
            // Get current version
            DynamoOrder order = dynamoMapper.load(DynamoOrder.class, orderId);
            if (order == null) {
                throw new OrderNotFoundException();
            }
            
            // Update with version check
            order.setStatus(newStatus);
            dynamoMapper.save(order); // Fails if version changed
            
        } catch (ConditionalCheckFailedException e) {
            // Another process modified the record - retry
            throw e;
        }
    }
    
    // Multi-item transaction using DynamoDB transactions (limited)
    public void transferOrderItems(String fromOrderId, String toOrderId, List<String> itemIds) {
        Collection<TransactionWriteRequest> actions = new ArrayList<>();
        
        // Read both orders first
        DynamoOrder fromOrder = dynamoMapper.load(DynamoOrder.class, fromOrderId);
        DynamoOrder toOrder = dynamoMapper.load(DynamoOrder.class, toOrderId);
        
        // Prepare conditional updates
        fromOrder.removeItems(itemIds);
        toOrder.addItems(itemIds);
        
        actions.add(new TransactionWriteRequest()
            .withConditionCheck(new ConditionCheck()
                .withTableName("Orders")
                .withKey(Collections.singletonMap("orderId", new AttributeValue(fromOrderId)))
                .withConditionExpression("version = :version")
                .withExpressionAttributeValues(Collections.singletonMap(":version", 
                    new AttributeValue().withN(fromOrder.getVersion().toString())))))
        
        actions.add(new TransactionWriteRequest()
            .withUpdate(new Update()
                .withTableName("Orders")
                .withKey(Collections.singletonMap("orderId", new AttributeValue(fromOrderId)))
                .withUpdateExpression("SET #items = :items, version = version + :inc")
                .withConditionExpression("version = :currentVersion")
                .withExpressionAttributeNames(Collections.singletonMap("#items", "items"))
                .withExpressionAttributeValues(Map.of(
                    ":items", new AttributeValue().withSS(fromOrder.getItems()),
                    ":inc", new AttributeValue().withN("1"),
                    ":currentVersion", new AttributeValue().withN(fromOrder.getVersion().toString())
                ))));
        
        // Execute transaction
        dynamoClient.transactWriteItems(new TransactWriteItemsRequest()
            .withTransactItems(actions));
    }
}

// SAGA pattern for NoSQL databases without transactions
@Service
public class NoSQLOrderSaga {
    
    @Autowired
    private DynamoOrderService orderService;
    
    @Autowired
    private MongoInventoryService inventoryService;
    
    @Autowired
    private CassandraPaymentService paymentService;
    
    public void processOrderWithCompensation(CreateOrderRequest request) {
        CompensatingTransactionManager saga = new CompensatingTransactionManager();
        
        try {
            // Step 1: Create order in DynamoDB
            String orderId = saga.execute("CREATE_ORDER", 
                () -> orderService.createOrder(request),
                (orderId) -> orderService.deleteOrder(orderId)
            );
            
            // Step 2: Reserve inventory in MongoDB  
            String reservationId = saga.execute("RESERVE_INVENTORY",
                () -> inventoryService.reserveItems(request.getItems()),
                (reservationId) -> inventoryService.releaseReservation(reservationId)
            );
            
            // Step 3: Process payment in Cassandra
            String paymentId = saga.execute("PROCESS_PAYMENT",
                () -> paymentService.processPayment(request.getPaymentInfo()),
                (paymentId) -> paymentService.refundPayment(paymentId)
            );
            
            // All steps successful - confirm order
            orderService.confirmOrder(orderId, paymentId, reservationId);
            
        } catch (Exception e) {
            // Compensate all executed steps
            saga.compensateAll();
            throw new OrderProcessingException("Order processing failed", e);
        }
    }
}

// Generic compensating transaction manager
public class CompensatingTransactionManager {
    
    private final List<CompensatingAction> executedActions = new ArrayList<>();
    
    public <T> T execute(String stepName, Supplier<T> action, Consumer<T> compensation) {
        try {
            T result = action.get();
            executedActions.add(new CompensatingAction<>(stepName, result, compensation));
            return result;
        } catch (Exception e) {
            log.error("Step {} failed: {}", stepName, e.getMessage());
            throw e;
        }
    }
    
    public void compensateAll() {
        // Compensate in reverse order
        for (int i = executedActions.size() - 1; i >= 0; i--) {
            CompensatingAction action = executedActions.get(i);
            try {
                action.compensate();
                log.info("Compensated step: {}", action.getStepName());
            } catch (Exception e) {
                log.error("Compensation failed for step: {}", action.getStepName(), e);
                // In production, you'd send to dead letter queue
            }
        }
    }
    
    @AllArgsConstructor
    private static class CompensatingAction<T> {
        private final String stepName;
        private final T result;
        private final Consumer<T> compensationAction;
        
        public void compensate() {
            compensationAction.accept(result);
        }
        
        public String getStepName() {
            return stepName;
        }
    }
}

// Different strategies for different use cases
@Service
public class OrderService {
    
    // High-consistency financial operations
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public void processPayment(PaymentRequest request) {
        // Strict ACID guarantees
    }
    
    // Analytics operations can be eventually consistent
    @Transactional(readOnly = true, isolation = Isolation.READ_COMMITTED)
    public OrderAnalytics generateAnalytics(String customerId) {
        // Faster reads, acceptable if slightly stale
    }
    
    // Bulk operations need careful batching
    @Transactional
    public void processBulkOrders(List<CreateOrderRequest> requests) {
        int batchSize = 100;
        for (int i = 0; i < requests.size(); i += batchSize) {
            int end = Math.min(i + batchSize, requests.size());
            List<CreateOrderRequest> batch = requests.subList(i, end);
            
            for (CreateOrderRequest request : batch) {
                processOrder(request);
            }
            
            // Flush changes to avoid memory issues
            entityManager.flush();
            entityManager.clear();
        }
    }
}

@Component
public class TransactionMetrics {
    
    private final MeterRegistry meterRegistry;
    
    @EventListener
    public void handleTransactionCommit(TransactionCommitEvent event) {
        Timer.Sample sample = Timer.start(meterRegistry);
        sample.stop(Timer.builder("transaction.duration")
                .tag("status", "commit")
                .tag("name", event.getTransactionName())
                .register(meterRegistry));
    }
    
    @EventListener
    public void handleTransactionRollback(TransactionRollbackEvent event) {
        Counter.builder("transaction.rollback")
                .tag("reason", event.getRollbackReason())
                .register(meterRegistry)
                .increment();
    }
}

@TestConfiguration
static class TransactionTestConfig {
    
    @Bean
    @Primary
    public PlatformTransactionManager testTransactionManager() {
        // Use test-specific transaction manager that tracks state
        return new TestTransactionManager();
    }
}

@Test
public void testTransactionRollbackOnFailure() {
    // Given
    CreateOrderRequest request = createValidOrderRequest();
    
    // Mock payment service to fail
    when(paymentService.processPayment(any(), any()))
        .thenThrow(new PaymentException("Payment failed"));
    
    // When
    assertThatThrownBy(() -> orderService.processOrder(request))
        .isInstanceOf(PaymentException.class);
    
    // Then
    assertThat(orderRepository.count()).isEqualTo(0); // No order created
    assertThat(inventoryRepository.findBySku("TEST_SKU").getQuantity())
        .isEqualTo(100); // Inventory not decremented
}

// Simplified wait time calculation
func CalculateWaitTime(jobRequest, queuePosition, historicalStats, availableWorkers) time.Duration {
    // 1. Find position in priority-ordered queue
    queuePosition := findQueuePosition(jobRequest.Priority, jobRequest.SubmissionTime)
    
    // 2. Calculate jobs ahead in queue (70% of executing jobs assumed near completion)
    jobsAhead := queuePosition + int(float64(executingJobs) * 0.7)
    
    // 3. Estimate based on historical average execution time
    if historicalAverage > 0 && availableWorkers > 0 {
        estimatedWait := time.Duration(float64(jobsAhead) / float64(availableWorkers)) * 
                        time.Duration(historicalAverage) * time.Millisecond
    }
    
    // 4. Account for scheduled future execution
    if jobRequest.ScheduledAt.After(time.Now()) {
        scheduleDiff := time.Until(jobRequest.ScheduledAt)
        if scheduleDiff > estimatedWait {
            estimatedWait = scheduleDiff
        }
    }
    
    return estimatedWait
}

type JobStats struct {
    SucceededJobsAverage int64  // Average execution time
    ExecutingJobs       int32   // Currently running
    AntsCapacity        int     // Available worker capacity
    AntsAvailable       bool    // Worker availability status
}

sort.Slice(pendingJobs, func(i, j int) bool {
    if job1.Priority == job2.Priority {
        return job1.CreatedAt.Before(job2.CreatedAt)  // FCFS within priority
    }
    return job1.Priority > job2.Priority  // Higher priority first
})

# Job requests are processed in submission order
job_type: data-processing
description: FCFS example - processed in submission order
tasks:
  - task_type: process-data
    method: DOCKER
    container:
      image: python:3.9
    script:
      - python process_data.py

job_type: critical-analysis
priority: 10  # Higher priority job
description: Critical security analysis
tasks:
  - task_type: vulnerability-scan
    method: KUBERNETES
    container:
      image: security-scanner:latest

# Submitting jobs with different priorities
curl -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  --data '{"job_type": "critical-analysis", "job_priority": 10}' \
  $SERVER/api/jobs/requests

curl -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  --data '{"job_type": "routine-backup", "job_priority": 1}' \
  $SERVER/api/jobs/requests

# Short-running analysis jobs
job_type: quick-scan
tags: ["fast-worker", "analysis"]
tasks:
  - task_type: preflight-check
    method: DOCKER
    tags: ["cpu-optimized"]
    container:
      image: lightweight-scanner:latest

---
# Medium-duration static analysis
job_type: static-analysis
tags: ["medium-worker", "analysis"] 
tasks:
  - task_type: code-analysis
    method: KUBERNETES
    tags: ["memory-optimized"]
    container:
      image: static-analyzer:latest
      memory_limit: "4Gi"

---
# Long-running dynamic analysis
job_type: dynamic-analysis
tags: ["heavy-worker", "analysis"]
tasks:
  - task_type: device-testing
    method: KUBERNETES
    tags: ["gpu-enabled", "device-farm"]
    container:
      image: dynamic-analyzer:latest
      resources:
        cpu_limit: "8"
        memory_limit: "16Gi"

job_type: ml-training
description: Resource-aware ML model training
tasks:
  - task_type: train-model
    method: KUBERNETES
    tags: ["gpu-node", "ml-workload"]
    container:
      image: tensorflow/tensorflow:latest-gpu
      cpu_request: "4"
      cpu_limit: "8"
      memory_request: "8Gi"
      memory_limit: "16Gi"
      ephemeral_storage_request: "10Gi"
    node_selector:
      hardware: "gpu-enabled"
      instance-type: "ml-optimized"
    tolerations:
      - key: "gpu-workload"
        operator: "Equal"
        value: "true"
        effect: "NoSchedule"

job_type: geo-distributed-processing
description: Process data close to its source
tasks:
  - task_type: process-eu-data
    method: KUBERNETES
    tags: ["eu-region", "gdpr-compliant"]
    container:
      image: data-processor:latest
    affinity:
      node_affinity:
        required_during_scheduling_ignored_during_execution:
          node_selector_terms:
            - match_expressions:
                - key: "region"
                  operator: In
                  values: ["eu-west-1", "eu-central-1"]
                - key: "compliance"
                  operator: In
                  values: ["gdpr"]
    variables:
      DATA_REGION: "eu"
      COMPLIANCE_MODE: "strict"

Optimal_Delay = min(Max_Wait_Time, Expected_Benefit_Time)
Where:
- Max_Wait_Time = configured upper bound to prevent starvation
- Expected_Benefit_Time = estimated time until optimal conditions
- Locality_Benefit = (Remote_Cost - Local_Cost) / Transfer_Rate

job_type: nightly-etl
cron_trigger: "0 2 * * *"  # 2 AM daily
scheduled_at: "2024-12-25T02:00:00Z"  # One-time future execution

job_type: external-dependency-wait
tasks:
  - task_type: wait-for-api
    method: HTTP_GET
    url: https://api.service.com/status
    delay_between_retries: 30s
    retry: 20  # Maximum 10 minutes of polling
    on_exit_code:
      "200": proceed-with-processing    # Service ready
      "404": EXECUTING                  # Keep polling
      "503": EXECUTING                  # Service temporarily unavailable
      "FAILED": abort-job               # Permanent failure

job_type: gpu-intensive-training
tasks:
  - task_type: training
    method: KUBERNETES
    tags: ["gpu-v100", "high-memory"]
    timeout: 6h
    # Will delay until specific GPU resources become available

Queue_Capacity = (Allocated_Resources / Total_Cluster_Resources) × 100%
Effective_Capacity = min(Max_Capacity, Guaranteed_Capacity + Available_Borrowed_Capacity)
Resource_Utilization = Used_Resources / Effective_Capacity

# Organization-level capacity limits
tenant_limits:
  max_concurrent_jobs: 50
  max_cpu_hours_per_day: 200
  max_storage_gb: 500

# Job-level concurrency control  
job_type: batch-processing
max_concurrency: 3  # Limit concurrent instances

// Formicary's Fair Scheduling based on actual implementation
type FairScheduler struct {
    jobStatsRegistry *JobStatsRegistry
    serverConfig     *ServerConfig
}

func (fs *FairScheduler) CheckFairSchedulingConstraints(request *JobRequest) error {
    // Multi-level concurrency checking: Organization ? User ? Job-level
    
    // Level 1: Organization concurrency limits
    userExecuting, orgExecuting := fs.jobStatsRegistry.UserOrgExecuting(request)
    
    if orgExecuting >= fs.getMaxConcurrentOrgJobs(request.OrganizationID) {
        return fs.delayJobForConcurrencyExceeded(request, "organization", orgExecuting)
    }
    
    // Level 2: User concurrency limits  
    if userExecuting >= fs.getMaxConcurrentUserJobs(request.UserID) {
        return fs.delayJobForConcurrencyExceeded(request, "user", userExecuting)
    }
    
    // Level 3: Job-type concurrency limits
    executionCount := fs.jobStatsRegistry.GetExecutionCount(request.GetUserJobTypeKey())
    if executionCount >= request.GetMaxConcurrency() {
        return fs.delayJobForConcurrencyExceeded(request, "job-type", int(executionCount))
    }
    
    return nil
}

func (fs *FairScheduler) delayJobForConcurrencyExceeded(request *JobRequest, limitType string, currentCount int) error {
    // Intelligent delay calculation based on historical data
    avgCompletionTime := fs.jobStatsRegistry.GetAverageCompletionTime(request.JobType)
    
    // Dynamic wait factor: 25% of average completion time, bounded between 15-60 seconds
    waitFactor := min(max(avgCompletionTime/4, 15*time.Second), 60*time.Second)
    
    // Randomized delay to prevent thundering herd
    randomDelay := time.Duration(rand.Intn(int(waitFactor))) + waitFactor
    
    // Reschedule with delay
    request.ScheduledAt = time.Now().Add(randomDelay)
    
    // Logarithmic priority degradation (inspired by mobile security company approach)
    if request.Priority > 0 {
        // Priority degradation: log_e(original_priority), minimum of 1
        newPriority := max(1, int(math.Log(float64(request.Priority))))
        request.Priority = newPriority
        
        // Allow zero-priority jobs to bypass concurrency limits (emergency valve)
        if newPriority <= 0 {
            return nil // Allow execution despite limits
        }
    }
    
    // Update schedule attempts counter with exponential backoff
    request.ScheduleAttempts++
    
    return fmt.Errorf("%s concurrency limit exceeded: %d jobs running, rescheduling with %v delay", 
                     limitType, currentCount, randomDelay)
}

// Enhanced concurrency tracking from mobile security company experience
func (fs *FairScheduler) trackConcurrencyMetrics(request *JobRequest) {
    // Real-time metrics for monitoring fairness
    fs.metricsRegistry.Gauge("org_concurrent_jobs", map[string]string{
        "org_id": request.OrganizationID,
        "job_type": request.JobType,
    })
    
    fs.metricsRegistry.Gauge("user_concurrent_jobs", map[string]string{
        "user_id": request.UserID, 
        "job_type": request.JobType,
    })
}

Schedulability Test: ?(Ci/Ti) ? 1
Where Ci = execution time, Ti = period for periodic tasks

Priority(task) = 1 / (Deadline - Current_Time)
Schedulability_Test: ?(Execution_Time_i / Period_i) ? 1
Laxity = Deadline - Current_Time - Remaining_Execution_Time

# Short jobs queue
job_type: quick-validation
tags: ["short-queue"]
estimated_runtime: "5m"

# Long jobs queue  
job_type: full-analysis
tags: ["long-queue"] 
estimated_runtime: "2h"

Relative_Deadline = Period (for periodic tasks)
Absolute_Deadline = Arrival_Time + Relative_Deadline
Critical_Instant = Simultaneous release of all tasks

# Simulated EDF using priority and scheduled execution
job_type: time-critical-analysis
priority: {{.UrgencyScore}}  # Calculated based on deadline proximity
scheduled_at: "2024-12-31T23:59:59Z"
timeout: 2h
tasks:
  - task_type: urgent-processing
    method: KUBERNETES
    tags: ["priority-worker"]

job_type: speculative-execution
tasks:
  - task_type: main-task
    method: KUBERNETES
    timeout: 30m
    speculative_backup:
      enabled: true
      delay_threshold: "150%"  # Start backup if 50% slower than average
      resource_threshold: 0.3  # Only if 30%+ resources available

job_type: sensor-job
description: Wait for external conditions before proceeding
tasks:
  - task_type: wait-for-resource
    method: HTTP_GET
    url: https://api.example.com/resource/123
    delay_between_retries: 15s
    retry: 20  # Poll up to 20 times (5 minutes total)
    timeout: 15s
    on_exit_code:
      "200": process-resource    # Success - proceed
      "404": EXECUTING          # Resource not ready - poll again  
      "FAILED": FATAL           # Server error - abort job

  - task_type: process-resource
    method: DOCKER
    container:
      image: data-processor:latest
    script:
      - echo "Resource is now available, processing..."

Gang_Size = max(task_count, critical_path_parallelism)
Resource_Requirement = ?(individual_task_resources) for all gang members
Schedulability = available_resources >= Resource_Requirement

// Core gang scheduling logic from ResourceManager
func (rm *ManagerImpl) doReserveJobResources(
	requestID string,
	def *types.JobDefinition,
	dryRun bool) (reservations map[string]*common.AntReservation, err error) {
	
	reservations = make(map[string]*common.AntReservation)
	var alloc *common.AntReservation
	
	// Try to reserve resources for each task
	for _, task := range def.Tasks {
		alloc, err = rm.doReserve(requestID, task.TaskType, task.Method, task.Tags, dryRun)
		if err == nil {
			reservations[task.TaskType] = alloc
		} else {
			if !dryRun {
				// ALL-OR-NOTHING: Release all allocations and fail entire job
				_ = rm.ReleaseJobResources(requestID)
			}
			return nil, err
		}
	}
	return reservations, nil
}

// Check if all job resources are available without allocating
func (rm *ManagerImpl) CheckJobResources(job *types.JobDefinition) ([]*common.AntReservation, error) {
	if reservationsByTask, err := rm.doReserveJobResources("", job, true); err != nil {
		return nil, err // Gang scheduling not possible
	}
	// All tasks can be scheduled - proceed to actual reservation
}

// Atomically reserve resources for all tasks
func (rm *ManagerImpl) ReserveJobResources(requestID string, def *types.JobDefinition) (map[string]*common.AntReservation, error) {
	return rm.doReserveJobResources(requestID, def, false)
}

// Step 1: Check if gang scheduling is possible
if err = jobStateMachine.CheckAntResourcesAndConcurrencyForJob(); err != nil {
	// Gang scheduling failed - implement backoff strategy
	scheduleAttempts := request.ScheduleAttempts + 1
	scheduleSecs := math.Min(int(maxWait.Seconds()), scheduleAttempts*5)
	
	// Exponential backoff with priority degradation
	if scheduleAttempts >= 5 && scheduleAttempts%5 == 0 && request.JobPriority > 5 {
		request.JobPriority-- // Degrade priority every 5th attempt
	}
	
	request.ScheduledAt = request.ScheduledAt.Add(time.Duration(scheduleSecs) * time.Second)
	return fmt.Errorf("gang scheduling failed - will retry")
}

// Step 2: Perform actual gang scheduling
if err = jobStateMachine.ReserveJobResources(); err != nil {
	// Even after check, allocation failed - very rare race condition
	return fmt.Errorf("gang allocation failed after successful check")
}

job_type: distributed-ml-training
description: Gang scheduled ML training requiring coordinated execution
tasks:
  - task_type: parameter-server
    method: KUBERNETES
    tags: ["ml-cluster", "coordinator"]
    container:
      image: tensorflow:latest
      cpu_request: "4"
      memory_request: "8Gi"
    
  - task_type: worker-node-1
    method: KUBERNETES  
    tags: ["ml-cluster", "gpu-enabled"]
    container:
      image: tensorflow:gpu
      cpu_request: "8"
      memory_request: "16Gi"
      
  - task_type: worker-node-2
    method: KUBERNETES
    tags: ["ml-cluster", "gpu-enabled"]
    container:
      image: tensorflow:gpu
      cpu_request: "8" 
      memory_request: "16Gi"

  - task_type: aggregator
    method: KUBERNETES
    tags: ["ml-cluster"]
    dependencies: ["parameter-server", "worker-node-1", "worker-node-2"]

job_type: hybrid-ml-pipeline
priority: 8                    # Priority scheduling
max_concurrency: 3             # Capacity scheduling  
tags: ["gpu-cluster"]          # Matchmaking
cron_trigger: "0 */6 * * *"    # Delay scheduling
tasks:
  - task_type: data-preprocessing
    tags: ["cpu-optimized"]     # Multilevel queues
    method: KUBERNETES
    container:
      cpu_request: "2"          # Resource-aware
      memory_request: "4Gi"
  
  - task_type: model-training
    tags: ["gpu-optimized"]
    method: KUBERNETES
    container:
      image: tensorflow:gpu
    dependencies: ["data-preprocessing"]

job_type: parallel-video-encoding
description: Parallel processing with fork-join
tasks:
  - task_type: split-video
    method: DOCKER
    container:
      image: ffmpeg:latest
    script:
      - ffmpeg -i input.mp4 -f segment segment_%03d.mp4
    on_completed: fork-encode

  - task_type: fork-encode
    method: FORK_JOB
    fork_job_type: encode-segment
    variables:
      segment_count: 8
    on_completed: await-encoding

  - task_type: await-encoding  
    method: AWAIT_FORKED_JOB
    await_forked_tasks: ["fork-encode"]
    on_completed: merge-video

  - task_type: merge-video
    method: DOCKER
    container:
      image: ffmpeg:latest
    script:
      - ffmpeg -f concat -i segments.txt -c copy output.mp4

job_type: node-build
tasks:
  - task_type: install-deps
    method: DOCKER
    container:
      image: node:16
    cache:
      key: "node-deps-{{checksum 'package-lock.json'}}"
      paths:
        - node_modules
    script:
      - npm ci

job_type: deployment-pipeline
tasks:
  - task_type: build
    artifacts:
      paths: ["dist/"]
    
  - task_type: test
    dependencies: ["build"]  # Waits for build artifacts
    
  - task_type: deploy
    dependencies: ["test"]
    method: KUBERNETES

# Built-in metrics and alerts
metrics:
  - queue_depth_by_priority
  - average_wait_time_by_tag
  - resource_utilization_by_worker
  - job_completion_rate_by_tenant

alerts:
  - name: high_queue_depth
    condition: queue_depth > 1000
    action: scale_workers
    
  - name: poor_resource_utilization  
    condition: cpu_utilization < 30%
    action: consolidate_workers

error_handling:
  max_retries: 3
  dead_letter_queue: "failed-jobs"
  alert_on_dlq: true

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use tracing::{debug, info};

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum Value {
    Zero,
    One,
    Retreat, // Default value
}

impl std::fmt::Display for Value {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Value::Zero => write!(f, "ZERO"),
            Value::One => write!(f, "ONE"),
            Value::Retreat => write!(f, "RETREAT"),
        }
    }
}

#[derive(Clone, Debug)]
pub struct Configuration {
    pub source: usize,
    pub num_rounds: usize,
    pub num_processes: usize,
}

pub struct ByzantineEngine {
    config: Configuration,
    processes: Vec<Arc<Mutex<Process>>>,
    message_count: Arc<Mutex<usize>>,
}

pub struct Process {
    id: usize,
    config: Configuration,
    values: HashMap<String, Value>,
    is_faulty: bool,
}

impl Process {
    pub fn new(id: usize, config: Configuration) -> Self {
        let is_faulty = id == config.source || id == 2; // Configure faulty processes
        Process {
            id,
            config,
            values: HashMap::new(),
            is_faulty,
        }
    }

    pub fn receive_message(&mut self, path: String, value: Value) {
        debug!("Process {} received message: path={}, value={:?}", self.id, path, value);
        self.values.insert(path, value);
    }

    pub fn send_messages(&self, round: usize, processes: &[Arc<Mutex<Process>>], 
                        message_count: Arc<Mutex<usize>>) {
        if round == 0 && self.id == self.config.source {
            self.send_initial_messages(processes, message_count);
        } else if round > 0 {
            self.relay_messages(round, processes, message_count);
        }
    }

    fn send_initial_messages(&self, processes: &[Arc<Mutex<Process>>], 
                           message_count: Arc<Mutex<usize>>) {
        let base_value = Value::Zero;
        
        for (i, process) in processes.iter().enumerate() {
            if i != self.id {
                let value = if self.is_faulty {
                    // Faulty commander sends different values to different processes
                    if i % 2 == 0 { Value::Zero } else { Value::One }
                } else {
                    base_value.clone()
                };
                
                let value_for_log = value.clone(); // Clone for logging
                let mut proc = process.lock().unwrap();
                proc.receive_message(self.id.to_string(), value);
                *message_count.lock().unwrap() += 1;
                
                debug!("Commander {} sent {:?} to process {}", self.id, value_for_log, i);
            }
        }
    }

    fn relay_messages(&self, round: usize, processes: &[Arc<Mutex<Process>>], 
                     message_count: Arc<Mutex<usize>>) {
        let paths = self.get_paths_for_round(round - 1);
        
        for path in paths {
            if let Some(value) = self.values.get(&path) {
                let new_value = self.transform_value(value.clone());
                let new_path = format!("{}{}", path, self.id);
                
                for (i, process) in processes.iter().enumerate() {
                    if i != self.id && !self.path_contains_process(&new_path, i) {
                        let mut proc = process.lock().unwrap();
                        proc.receive_message(new_path.clone(), new_value.clone());
                        *message_count.lock().unwrap() += 1;
                        
                        debug!("Process {} relayed {:?} to process {} with path {}", 
                               self.id, new_value, i, new_path);
                    }
                }
            }
        }
    }

    fn transform_value(&self, value: Value) -> Value {
        if self.is_faulty && self.id == 2 {
            Value::One // Process 2 always sends One when faulty
        } else {
            value
        }
    }

    fn get_paths_for_round(&self, round: usize) -> Vec<String> {
        if round == 0 {
            vec![self.config.source.to_string()]
        } else {
            self.values.keys()
                .filter(|path| path.len() == round + 1)
                .cloned()
                .collect()
        }
    }

    fn path_contains_process(&self, path: &str, process_id: usize) -> bool {
        path.contains(&process_id.to_string())
    }

    pub fn decide(&self) -> Value {
        if self.id == self.config.source {
            // Source process uses its own value
            return if self.is_faulty { Value::One } else { Value::Zero };
        }

        self.majority_vote()
    }

    fn majority_vote(&self) -> Value {
        let mut counts = HashMap::new();
        counts.insert(Value::Zero, 0);
        counts.insert(Value::One, 0);
        counts.insert(Value::Retreat, 0);

        // Count values from the final round paths
        let final_paths: Vec<_> = self.values.keys()
            .filter(|path| path.len() == self.config.num_rounds + 1)
            .collect();

        if final_paths.is_empty() {
            // Count all available values if no final round paths
            for value in self.values.values() {
                *counts.entry(value.clone()).or_insert(0) += 1;
            }
        } else {
            for path in final_paths {
                if let Some(value) = self.values.get(path) {
                    *counts.entry(value.clone()).or_insert(0) += 1;
                }
            }
        }

        debug!("Process {} vote counts: {:?}", self.id, counts);

        // Find majority
        let total_votes: usize = counts.values().sum();
        if total_votes == 0 {
            return Value::Retreat;
        }

        let majority_threshold = total_votes / 2;
        
        for (value, count) in counts {
            if count > majority_threshold {
                return value;
            }
        }

        Value::Retreat // Default if no majority
    }

    pub fn is_faulty(&self) -> bool {
        self.is_faulty
    }

    pub fn is_source(&self) -> bool {
        self.id == self.config.source
    }
}

impl ByzantineEngine {
    pub fn new(source: usize, num_rounds: usize, num_processes: usize) -> Self {
        let config = Configuration { source, num_rounds, num_processes };
        let processes: Vec<Arc<Mutex<Process>>> = (0..num_processes)
            .map(|id| Arc::new(Mutex::new(Process::new(id, config.clone()))))
            .collect();

        ByzantineEngine {
            config,
            processes,
            message_count: Arc::new(Mutex::new(0)),
        }
    }

    pub fn run(&self) -> (Duration, usize) {
        info!("Starting Byzantine Generals algorithm with {} processes, {} rounds", 
              self.config.num_processes, self.config.num_rounds);
        
        let start = Instant::now();
        
        for round in 0..self.config.num_rounds {
            debug!("Starting round {}", round);
            
            let handles: Vec<_> = self.processes.iter().enumerate().map(|(_id, process)| {
                let process = Arc::clone(process);
                let processes = self.processes.clone();
                let message_count = Arc::clone(&self.message_count);
                
                thread::spawn(move || {
                    let proc = process.lock().unwrap();
                    proc.send_messages(round, &processes, message_count);
                })
            }).collect();

            for handle in handles {
                // Add timeout to prevent hanging
                if handle.join().is_err() {
                    eprintln!("Warning: Thread failed in round {}", round);
                }
            }
            
            debug!("Completed round {}", round);
            // Small delay to ensure message ordering
            thread::sleep(Duration::from_millis(10));
        }

        let duration = start.elapsed();
        let messages = *self.message_count.lock().unwrap();
        
        info!("Algorithm completed in {:.2}ms with {} messages", 
              duration.as_millis(), messages);
        
        self.print_results();
        
        (duration, messages)
    }

    fn print_results(&self) {
        println!("\nByzantine Generals Results:");
        println!("===========================");
        
        for (id, process) in self.processes.iter().enumerate() {
            let proc = process.lock().unwrap();
            if proc.is_source() {
                print!("Source ");
            }
            print!("Process {}", id);
            if proc.is_faulty() {
                println!(" is faulty");
            } else {
                println!(" decides on value {}", proc.decide());
            }
        }
        println!();
    }
}

pub fn benchmark_comprehensive(max_processes: usize) {
    let test_cases = generate_test_cases(max_processes);
    
    for (processes, rounds) in test_cases {
        if processes < 4 {
            continue; // Skip invalid cases
        }
        
        let source = processes / 3;
        let engine = ByzantineEngine::new(source, rounds, processes);
        
        let start_memory = get_memory_usage();
        let start = Instant::now();
        let (duration, messages) = engine.run();
        let _total_duration = start.elapsed();
        let end_memory = get_memory_usage();
        
        let memory_used = end_memory.saturating_sub(start_memory);
        
        println!("Rust,{},{},{},{:.2},{:.2}", 
                processes, rounds, messages, 
                duration.as_millis(), memory_used as f64 / 1024.0 / 1024.0);
    }
}

fn generate_test_cases(max_processes: usize) -> Vec<(usize, usize)> {
    let mut cases = Vec::new();
    
    for n in (4..=max_processes).step_by(3) {
        for m in 1..=3 {
            if 3 * m + 1 <= n {
                cases.push((n, m));
            }
        }
    }
    
    cases
}

fn get_memory_usage() -> usize {
    // Simplified memory usage - would need platform-specific code for accurate measurement
    std::process::id() as usize * 1024 // Placeholder
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_value_display() {
        assert_eq!(format!("{}", Value::Zero), "ZERO");
        assert_eq!(format!("{}", Value::One), "ONE");
        assert_eq!(format!("{}", Value::Retreat), "RETREAT");
    }

    #[test]
    fn test_process_creation() {
        let config = Configuration {
            source: 0,
            num_rounds: 2,
            num_processes: 4,
        };
        
        let process = Process::new(0, config.clone());
        assert!(process.is_source());
        assert!(process.is_faulty()); // Source is faulty in our test setup
        
        let process2 = Process::new(1, config);
        assert!(!process2.is_source());
        assert!(!process2.is_faulty());
    }

    #[test]
    fn test_engine_creation() {
        let engine = ByzantineEngine::new(0, 2, 4);
        assert_eq!(engine.config.source, 0);
        assert_eq!(engine.config.num_rounds, 2);
        assert_eq!(engine.config.num_processes, 4);
        assert_eq!(engine.processes.len(), 4);
    }

    #[test]
    fn test_minimum_byzantine_case() {
        let engine = ByzantineEngine::new(0, 1, 4);
        let (duration, messages) = engine.run();
        
        assert!(duration.as_nanos() > 0);
        assert!(messages > 0);
    }
}

defmodule ByzantineGenerals do
  @moduledoc """
  Byzantine Generals Problem implementation in Elixir
  Leverages the Actor model for natural distributed computing
  """

  require Logger

  defmodule Configuration do
    @moduledoc "Configuration for Byzantine Generals algorithm"
    defstruct [:source, :num_rounds, :num_processes]

    @type t :: %__MODULE__{
      source: non_neg_integer(),
      num_rounds: non_neg_integer(),
      num_processes: non_neg_integer()
    }
  end

  defmodule Process do
    @moduledoc "Individual process (general) in the Byzantine Generals algorithm"
    use GenServer
    require Logger

    defstruct [:id, :config, :values, :is_faulty, :message_count, :processes]

    @type value :: :zero | :one | :retreat
    @type path :: String.t()

    # Client API

    def start_link(%{id: id, config: config}) do
      GenServer.start_link(__MODULE__, %{id: id, config: config}, name: :"process_#{id}")
    end

    def receive_message(pid, path, value) do
      GenServer.call(pid, {:receive_message, path, value}, 10_000)
    end

    def send_messages(pid, round, processes) do
      GenServer.call(pid, {:send_messages, round, processes}, 10_000)
    end

    def decide(pid) do
      GenServer.call(pid, :decide, 5_000)
    end

    def is_faulty?(pid) do
      GenServer.call(pid, :is_faulty, 1_000)
    end

    def is_source?(pid) do
      GenServer.call(pid, :is_source, 1_000)
    end

    def get_message_count(pid) do
      GenServer.call(pid, :get_message_count, 1_000)
    end

    def get_values(pid) do
      GenServer.call(pid, :get_values, 1_000)
    end

    # Server callbacks

    @impl true
    def init(%{id: id, config: config}) do
      is_faulty = id == config.source || id == 2
      
      state = %__MODULE__{
        id: id,
        config: config,
        values: %{},
        is_faulty: is_faulty,
        message_count: 0,
        processes: []
      }
      
      Logger.debug("Process #{id} initialized, faulty: #{is_faulty}")
      {:ok, state}
    end

    @impl true
    def handle_call({:receive_message, path, value}, _from, state) do
      Logger.debug("Process #{state.id} received message: path=#{path}, value=#{value}")
      
      new_values = Map.put(state.values, path, value)
      new_count = state.message_count + 1
      
      {:reply, :ok, %{state | values: new_values, message_count: new_count}}
    end

    @impl true
    def handle_call({:send_messages, round, processes}, _from, state) do
      new_state = %{state | processes: processes}
      
      cond do
        round == 0 && state.id == state.config.source ->
          send_initial_messages(new_state)
        round > 0 ->
          relay_messages(new_state, round)
        true ->
          {:reply, :ok, new_state}
      end
    end

    @impl true
    def handle_call(:decide, _from, state) do
      decision = if state.id == state.config.source do
        # Source process uses its own value
        if state.is_faulty, do: :one, else: :zero
      else
        majority_vote(state)
      end
      
      {:reply, decision, state}
    end

    @impl true
    def handle_call(:is_faulty, _from, state) do
      {:reply, state.is_faulty, state}
    end

    @impl true
    def handle_call(:is_source, _from, state) do
      {:reply, state.id == state.config.source, state}
    end

    @impl true
    def handle_call(:get_message_count, _from, state) do
      {:reply, state.message_count, state}
    end

    @impl true
    def handle_call(:get_values, _from, state) do
      {:reply, state.values, state}
    end

    # Private functions

    defp send_initial_messages(state) do
      base_value = :zero
      
      Enum.each(state.processes, fn {id, pid} ->
        if id != state.id do
          value = if state.is_faulty do
            # Faulty commander sends different values
            if rem(id, 2) == 0, do: :zero, else: :one
          else
            base_value
          end
          
          receive_message(pid, Integer.to_string(state.id), value)
          Logger.debug("Commander #{state.id} sent #{value} to process #{id}")
        end
      end)
      
      {:reply, :ok, state}
    end

    defp relay_messages(state, round) do
      paths = get_paths_for_round(state, round - 1)
      
      Enum.each(paths, fn path ->
        case Map.get(state.values, path) do
          nil -> 
            :ok
          value ->
            new_value = transform_value(state, value)
            new_path = path <> Integer.to_string(state.id)
            
            Enum.each(state.processes, fn {id, pid} ->
              if id != state.id && !String.contains?(new_path, Integer.to_string(id)) do
                receive_message(pid, new_path, new_value)
                Logger.debug("Process #{state.id} relayed #{new_value} to #{id}, path: #{new_path}")
              end
            end)
        end
      end)
      
      {:reply, :ok, state}
    end

    defp transform_value(state, value) do
      if state.is_faulty && state.id == 2 do
        :one
      else
        value
      end
    end

    defp get_paths_for_round(state, round) do
      if round == 0 do
        [Integer.to_string(state.config.source)]
      else
        state.values
        |> Map.keys()
        |> Enum.filter(&(String.length(&1) == round + 1))
      end
    end

    defp majority_vote(state) do
      counts = Enum.reduce(state.values, %{zero: 0, one: 0, retreat: 0}, fn {_path, value}, acc ->
        Map.update!(acc, value, &(&1 + 1))
      end)
      
      Logger.debug("Process #{state.id} vote counts: #{inspect(counts)}")
      
      total_votes = Map.values(counts) |> Enum.sum()
      
      if total_votes == 0 do
        :retreat
      else
        majority_threshold = div(total_votes, 2)
        
        case Enum.find(counts, fn {_value, count} -> count > majority_threshold end) do
          {value, _count} -> value
          nil -> :retreat
        end
      end
    end
  end

  defmodule Engine do
    @moduledoc "Engine that orchestrates the Byzantine Generals algorithm"
    
    require Logger

    def run(source, num_rounds, num_processes, opts \\ []) do
      config = %Configuration{
        source: source,
        num_rounds: num_rounds,
        num_processes: num_processes
      }

      verbose = Keyword.get(opts, :verbose, true)
      
      if verbose do
        Logger.info("Starting Byzantine Generals: #{num_processes} processes, #{num_rounds} rounds, source: #{source}")
      end

      # Start processes
      processes = start_processes(config)
      
      start_time = :os.system_time(:millisecond)
      
      # Run algorithm rounds
      run_rounds(processes, num_rounds)
      
      end_time = :os.system_time(:millisecond)
      duration = end_time - start_time

      # Collect results
      {results, total_messages} = collect_results(processes, config, verbose)
      
      # Clean up
      cleanup_processes(processes)

      {duration, total_messages, results}
    end

    defp start_processes(config) do
      for id <- 0..(config.num_processes - 1) do
        {:ok, pid} = Process.start_link(%{id: id, config: config})
        {id, pid}
      end
    end

    defp run_rounds(processes, num_rounds, timeout \\ 30_000) do
      for round <- 0..(num_rounds - 1) do
        Logger.debug("Starting round #{round}")
        
        tasks = Enum.map(processes, fn {_id, pid} ->
          Task.async(fn -> 
            Process.send_messages(pid, round, processes)
          end)
        end)
        
        try do
          Task.await_many(tasks, timeout)
          # Small delay to ensure message ordering
          :timer.sleep(10)
        catch
          :exit, {:timeout, _} -> 
            Logger.error("Round #{round} timed out")
            throw(:timeout)
        end
      end
      :ok
    end

    defp collect_results(processes, _config, verbose) do
      total_messages = Enum.sum(Enum.map(processes, fn {_id, pid} ->
        Process.get_message_count(pid)
      end))

      results = Enum.map(processes, fn {id, pid} ->
        is_source = Process.is_source?(pid)
        is_faulty = Process.is_faulty?(pid)
        decision = if is_faulty, do: nil, else: Process.decide(pid)
        
        result = %{
          id: id,
          is_source: is_source,
          is_faulty: is_faulty,
          decision: decision
        }
        
        if verbose do
          print_process_result(result)
        end
        
        result
      end)

      {results, total_messages}
    end

    defp print_process_result(%{id: id, is_source: is_source, is_faulty: is_faulty, decision: decision}) do
      prefix = if is_source, do: "Source ", else: ""
      
      if is_faulty do
        IO.puts("#{prefix}Process #{id} is faulty")
      else
        IO.puts("#{prefix}Process #{id} decides on value #{decision}")
      end
    end

    defp cleanup_processes(processes) do
      Enum.each(processes, fn {_id, pid} -> 
        GenServer.stop(pid, :normal, 1000)
      end)
    end

    def benchmark(max_processes, opts \\ []) do
      verbose = Keyword.get(opts, :verbose, true)
      
      if verbose do
        IO.puts("Elixir Byzantine Generals Benchmark")
        IO.puts("===================================")
        IO.puts("Language,Processes,Rounds,Messages,Time(ms)")
      end
      
      test_cases = generate_test_cases(max_processes)
      
      results = Enum.map(test_cases, fn {processes, rounds} ->
        source = div(processes, 3)
        {time, messages, _results} = run(source, rounds, processes, verbose: false)
        
        result = %{
          language: "Elixir",
          processes: processes,
          rounds: rounds,
          messages: messages,
          time_ms: time
        }
        
        if verbose do
          IO.puts("Elixir,#{processes},#{rounds},#{messages},#{time}")
        end
        
        result
      end)
      
      results
    end

    defp generate_test_cases(max_processes) do
      for n <- 4..max_processes, rem(n - 1, 3) == 0 do
        for m <- 1..3, 3 * m + 1 <= n do
          {n, m}
        end
      end
      |> List.flatten()
    end
  end

  # Main module functions

  def run(source, num_rounds, num_processes, opts \\ []) do
    Engine.run(source, num_rounds, num_processes, opts)
  end

  def benchmark(max_processes \\ 20, opts \\ []) do
    Engine.benchmark(max_processes, opts)
  end

  def quick_test do
    IO.puts("Running quick test with 4 processes, 1 round...")
    {time, messages, results} = run(0, 1, 4)
    
    IO.puts("\nTest Results:")
    IO.puts("Time: #{time}ms")
    IO.puts("Messages: #{messages}")
    IO.puts("Processes reached consensus: #{length(results)}")
    IO.puts("? Test completed successfully")
    
    :ok
  end
end

defmodule ByzantineGenerals.Application do
  @moduledoc false
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      # Add supervised processes here if needed
    ]

    opts = [strategy: :one_for_one, name: ByzantineGenerals.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

-module(byzantine_generals).
-export([run/3, benchmark/1, quick_test/0, start/0, stop/0]).
-include_lib("kernel/include/logger.hrl").

-record(config, {source, num_rounds, num_processes}).
-record(process_state, {id, config, values, is_faulty, message_count, processes}).

%% Public API

%% Start the application
start() ->
    application:start(byzantine_generals).

%% Stop the application  
stop() ->
    application:stop(byzantine_generals).

%% Run the Byzantine Generals algorithm
run(Source, NumRounds, NumProcesses) ->
    ?LOG_INFO("Starting Byzantine Generals: ~p processes, ~p rounds, source: ~p", 
              [NumProcesses, NumRounds, Source]),
    
    Config = #config{source = Source, num_rounds = NumRounds, num_processes = NumProcesses},
    
    % Validate configuration
    case validate_config(Config) of
        ok -> 
            run_algorithm(Config);
        {error, Reason} ->
            ?LOG_ERROR("Invalid configuration: ~p", [Reason]),
            {error, Reason}
    end.

%% Run benchmark with different configurations
benchmark(MaxProcesses) ->
    ?LOG_INFO("Running Erlang Byzantine Generals Benchmark up to ~p processes", [MaxProcesses]),
    
    io:format("Erlang Byzantine Generals Benchmark~n"),
    io:format("===================================~n"),
    io:format("Language,Processes,Rounds,Messages,Time(ms)~n"),
    
    TestCases = generate_test_cases(MaxProcesses),
    
    Results = lists:map(fun({Processes, Rounds}) ->
        Source = Processes div 3,
        case run(Source, Rounds, Processes) of
            {ok, Time, Messages, _ProcessResults} ->
                io:format("Erlang,~p,~p,~p,~p~n", [Processes, Rounds, Messages, Time]),
                #{language => erlang, processes => Processes, rounds => Rounds, 
                  messages => Messages, time_ms => Time};
            {error, _Reason} ->
                #{error => true, processes => Processes, rounds => Rounds}
        end
    end, TestCases),
    
    Results.

%% Quick test function
quick_test() ->
    io:format("Running quick test with 4 processes, 1 round...~n"),
    case run(0, 1, 4) of
        {ok, Time, Messages, _Results} ->
            io:format("~nTest Results:~n"),
            io:format("Time: ~pms~n", [Time]),
            io:format("Messages: ~p~n", [Messages]),
            io:format("? Test completed successfully~n"),
            ok;
        {error, Reason} ->
            io:format("? Test failed: ~p~n", [Reason]),
            error
    end.

%% Internal functions

validate_config(#config{source = Source, num_rounds = NumRounds, num_processes = NumProcesses}) ->
    if
        NumProcesses < 4 ->
            {error, "Need at least 4 processes for Byzantine Generals Problem"};
        Source >= NumProcesses ->
            {error, "Source must be less than number of processes"};
        NumRounds < 1 ->
            {error, "Need at least 1 round"};
        true ->
            ok
    end.

run_algorithm(Config) ->
    % Start message counter
    CounterPid = spawn_link(fun() -> counter_loop(0) end),
    register(message_counter, CounterPid),
    
    StartTime = erlang:system_time(millisecond),
    
    try
        % Start processes
        ProcessPids = start_processes(Config),
        
        % Initialize processes with neighbor information
        initialize_processes(ProcessPids, Config),
        
        % Run algorithm rounds
        run_rounds(ProcessPids, Config),
        
        % Wait for completion
        timer:sleep(100),
        
        EndTime = erlang:system_time(millisecond),
        Duration = EndTime - StartTime,
        
        % Collect results
        {TotalMessages, ProcessResults} = collect_results(ProcessPids, Config),
        
        % Cleanup
        cleanup_processes(ProcessPids),
        unregister(message_counter),
        exit(CounterPid, normal),
        
        {ok, Duration, TotalMessages, ProcessResults}
        
    catch
        Class:Reason:Stacktrace ->
            ?LOG_ERROR("Algorithm failed: ~p:~p~n~p", [Class, Reason, Stacktrace]),
            % Cleanup on error
            catch unregister(message_counter),
            catch exit(CounterPid, kill),
            {error, {Class, Reason}}
    end.

start_processes(Config) ->
    NumProcesses = Config#config.num_processes,
    lists:map(fun(Id) -> 
        Pid = spawn_link(fun() -> process_loop(Id, Config) end),
        {Id, Pid}
    end, lists:seq(0, NumProcesses - 1)).

initialize_processes(ProcessPids, Config) ->
    lists:foreach(fun({_Id, Pid}) -> 
        Pid ! {init, ProcessPids, Config}
    end, ProcessPids).

run_rounds(ProcessPids, Config) ->
    NumRounds = Config#config.num_rounds,
    lists:foreach(fun(Round) ->
        ?LOG_DEBUG("Starting round ~p", [Round]),
        
        % Send messages for this round
        lists:foreach(fun({_Id, Pid}) ->
            Pid ! {send_messages, Round, self()}
        end, ProcessPids),
        
        % Wait for all processes to complete the round
        lists:foreach(fun({_Id, _Pid}) ->
            receive
                {round_complete, Round} -> ok
            after 5000 ->
                ?LOG_WARNING("Timeout waiting for round ~p completion", [Round])
            end
        end, ProcessPids),
        
        % Small delay between rounds
        timer:sleep(10)
    end, lists:seq(0, NumRounds - 1)).

collect_results(ProcessPids, Config) ->
    % Get total message count
    TotalMessages = get_message_count(),
    
    % Get process results
    ProcessResults = lists:map(fun({Id, Pid}) ->
        Pid ! {get_result, self()},
        receive
            {result, Id, Result} -> 
                print_process_result(Id, Result, Config#config.source),
                Result
        after 2000 ->
            ?LOG_WARNING("Timeout getting result from process ~p", [Id]),
            #{id => Id, error => timeout}
        end
    end, ProcessPids),
    
    {TotalMessages, ProcessResults}.

print_process_result(Id, Result, Source) ->
    Prefix = case Id of
        Source -> "Source ";
        _ -> ""
    end,
    
    case maps:get(is_faulty, Result, false) of
        true ->
            io:format("~sProcess ~p is faulty~n", [Prefix, Id]);
        false ->
            Decision = maps:get(decision, Result, retreat),
            io:format("~sProcess ~p decides on value ~p~n", [Prefix, Id, Decision])
    end.

cleanup_processes(ProcessPids) ->
    lists:foreach(fun({_Id, Pid}) -> 
        Pid ! stop,
        % Don't wait for exit - let them clean up
        ok
    end, ProcessPids).

generate_test_cases(MaxProcesses) ->
    lists:flatten([
        [{N, M} || M <- lists:seq(1, 3), 3 * M + 1 =< N]
        || N <- lists:seq(4, MaxProcesses, 3)
    ]).

%% Process implementation

process_loop(Id, Config) ->
    IsFaulty = (Id =:= Config#config.source) orelse (Id =:= 2),
    State = #process_state{
        id = Id, 
        config = Config, 
        values = #{}, 
        is_faulty = IsFaulty,
        message_count = 0,
        processes = []
    },
    ?LOG_DEBUG("Process ~p initialized, faulty: ~p", [Id, IsFaulty]),
    process_loop(State).

process_loop(State) ->
    receive
        {init, ProcessPids, Config} ->
            NewState = State#process_state{processes = ProcessPids, config = Config},
            process_loop(NewState);
            
        {receive_message, Path, Value} ->
            NewValues = maps:put(Path, Value, State#process_state.values),
            NewState = State#process_state{
                values = NewValues,
                message_count = State#process_state.message_count + 1
            },
            increment_message_count(),
            ?LOG_DEBUG("Process ~p received message: path=~s, value=~p", 
                      [State#process_state.id, Path, Value]),
            process_loop(NewState);
            
        {send_messages, Round, From} ->
            NewState = handle_send_messages(State, Round),
            From ! {round_complete, Round},
            process_loop(NewState);
            
        {get_result, From} ->
            Result = create_result(State),
            From ! {result, State#process_state.id, Result},
            process_loop(State);
            
        stop ->
            ?LOG_DEBUG("Process ~p stopping", [State#process_state.id]),
            ok;
            
        Other ->
            ?LOG_WARNING("Process ~p received unexpected message: ~p", 
                        [State#process_state.id, Other]),
            process_loop(State)
    end.

handle_send_messages(State, Round) ->
    Id = State#process_state.id,
    Config = State#process_state.config,
    
    if 
        Round =:= 0 andalso Id =:= Config#config.source ->
            send_initial_messages(State);
        Round > 0 ->
            relay_messages(State, Round);
        true ->
            State
    end.

send_initial_messages(State) ->
    BaseValue = zero,
    ProcessPids = State#process_state.processes,
    
    lists:foreach(fun({Id, Pid}) ->
        if Id =/= State#process_state.id ->
            Value = case State#process_state.is_faulty of
                true -> 
                    % Faulty commander sends different values
                    case Id rem 2 of
                        0 -> zero;
                        1 -> one
                    end;
                false -> 
                    BaseValue
            end,
            
            Pid ! {receive_message, integer_to_list(State#process_state.id), Value},
            ?LOG_DEBUG("Commander ~p sent ~p to process ~p", 
                      [State#process_state.id, Value, Id]);
        true -> 
            ok
        end
    end, ProcessPids),
    
    State.

relay_messages(State, Round) ->
    Paths = get_paths_for_round(State, Round - 1),
    ProcessPids = State#process_state.processes,
    
    lists:foreach(fun(Path) ->
        case maps:get(Path, State#process_state.values, undefined) of
            undefined -> 
                ok;
            Value ->
                NewValue = transform_value(State, Value),
                NewPath = Path ++ integer_to_list(State#process_state.id),
                
                lists:foreach(fun({Id, Pid}) ->
                    IdStr = integer_to_list(Id),
                    case Id =/= State#process_state.id of
                        true ->
                            case string:str(NewPath, IdStr) of
                                0 -> % IdStr not found in NewPath
                                    Pid ! {receive_message, NewPath, NewValue},
                                    ?LOG_DEBUG("Process ~p relayed ~p to ~p, path: ~s", 
                                              [State#process_state.id, NewValue, Id, NewPath]);
                                _ -> % IdStr found in NewPath, skip
                                    ok
                            end;
                        false -> % Same process, skip
                            ok
                    end
                end, ProcessPids)
        end
    end, Paths),
    
    State.

transform_value(State, Value) ->
    if State#process_state.is_faulty andalso State#process_state.id =:= 2 ->
        one;
    true ->
        Value
    end.

get_paths_for_round(State, Round) ->
    if Round =:= 0 ->
        [integer_to_list((State#process_state.config)#config.source)];
    true ->
        maps:fold(fun(Path, _Value, Acc) ->
            case length(Path) of
                Len when Len =:= Round + 1 -> [Path | Acc];
                _ -> Acc
            end
        end, [], State#process_state.values)
    end.

create_result(State) ->
    Decision = if State#process_state.id =:= (State#process_state.config)#config.source ->
        % Source process uses its own value
        case State#process_state.is_faulty of
            true -> one;
            false -> zero
        end;
    true ->
        majority_vote(State)
    end,
    
    #{
        id => State#process_state.id,
        is_source => State#process_state.id =:= (State#process_state.config)#config.source,
        is_faulty => State#process_state.is_faulty,
        decision => Decision,
        message_count => State#process_state.message_count
    }.

majority_vote(State) ->
    Values = maps:values(State#process_state.values),
    Counts = lists:foldl(fun(Value, Acc) ->
        maps:update_with(Value, fun(Count) -> Count + 1 end, 1, Acc)
    end, #{zero => 0, one => 0, retreat => 0}, Values),
    
    ?LOG_DEBUG("Process ~p vote counts: ~p", [State#process_state.id, Counts]),
    
    TotalVotes = maps:fold(fun(_Value, Count, Sum) -> Sum + Count end, 0, Counts),
    
    if TotalVotes =:= 0 ->
        retreat;
    true ->
        MajorityThreshold = TotalVotes div 2,
        case maps:fold(fun(Value, Count, Acc) ->
            if Count > MajorityThreshold -> Value;
            true -> Acc
            end
        end, retreat, Counts) of
            retreat -> retreat;
            Value -> Value
        end
    end.

%% Message counter implementation

counter_loop(Count) ->
    receive
        increment ->
            counter_loop(Count + 1);
        {get_count, From} ->
            From ! {count, Count},
            counter_loop(Count);
        reset ->
            counter_loop(0);
        stop ->
            ok;
        _ ->
            counter_loop(Count)
    end.

increment_message_count() ->
    case whereis(message_counter) of
        undefined -> ok;
        Pid -> Pid ! increment
    end.

get_message_count() ->
    case whereis(message_counter) of
        undefined -> 0;
        Pid ->
            Pid ! {get_count, self()},
            receive
                {count, Count} -> Count
            after 1000 -> 0
            end
    end.

// Example comprehensive benchmark
pub struct BenchmarkResults {
    pub language: String,
    pub num_processes: usize,
    pub num_rounds: usize,
    pub execution_time_ms: f64,
    pub messages_sent: usize,
    pub memory_peak_mb: f64,
    pub cpu_utilization: f64,
}

pub fn comprehensive_benchmark() {
    let test_cases = vec![
        (4, 1),   // Minimum viable case
        (7, 2),   // Small scale
        (10, 3),  // Medium scale
        (16, 5),  // Larger scale
    ];

    for (processes, rounds) in test_cases {
        // Rust benchmark
        let rust_result = benchmark_rust_detailed(processes, rounds);
        
        // Elixir benchmark (would call via Port)
        let elixir_result = benchmark_elixir_detailed(processes, rounds);
        
        // Erlang benchmark (would call via Port)
        let erlang_result = benchmark_erlang_detailed(processes, rounds);
        
        compare_results(vec![rust_result, elixir_result, erlang_result]);
    }
}

# Clone and run benchmarks
git clone https://github.com/bhatti/byz-sample
cd byz-sample
make benchmark