Concurrency Patterns in Magento

Race Condition Example

// BAD: Race condition
class InventoryService
{
    public function decrementStock(ProductId $productId, int $quantity): void
    {
        $stock = $this->inventoryRepository->get($productId);
        $newStock = $stock->getQuantity() - $quantity;  // Read
        $stock->setQuantity($newStock);
        $this->inventoryRepository->save($stock);        // Write
    }
}
// Two concurrent requests read same stock value
// Both decrement and save, losing one decrement

Preventing Race Conditions

// GOOD: Atomic operation
class InventoryService
{
    public function decrementStock(ProductId $productId, int $quantity): void
    {
        $connection = $this->resource->getConnection();
        $connection->query(
            "UPDATE cataloginventory_stock_item 
             SET qty = qty - ? 
             WHERE product_id = ? AND qty >= ?",
            [$quantity, $productId->getValue(), $quantity]
        );
    }
}

Pessimistic Locking

// Database-level locking
class OrderService
{
    public function reserveStock(Order $order): void
    {
        $connection = $this->resource->getConnection();
        $connection->beginTransaction();
        
        try {
            // Lock row for update
            $connection->query(
                "SELECT * FROM cataloginventory_stock_item 
                 WHERE product_id = ? FOR UPDATE",
                [$order->getProductId()]
            );
            
            // Safe to modify
            $this->decrementStock($order->getProductId(), $order->getQuantity());
            
            $connection->commit();
        } catch (Exception $e) {
            $connection->rollBack();
            throw $e;
        }
    }
}

Magento Product Save Race Condition

// BAD: Race condition on product save
class ProductUpdater
{
    public function updatePrice(ProductId $id, Money $price): void
    {
        $product = $this->productRepository->get($id);
        $product->setPrice($price->getAmount());
        $this->productRepository->save($product);
    }
}

// GOOD: Use resource model for atomic update
public function updatePrice(ProductId $id, Money $price): void
{
    $this->productResource->saveAttribute(
        $this->productFactory->create()->setId($id->getValue()),
        ['price' => $price->getAmount()]
    );
}

Key Takeaway

Race conditions occur when concurrent operations read and write shared state. Use atomic database operations or pessimistic locking to prevent them.

Optimistic Locking Concept

1. Read data with version number
2. Modify data
3. Update WHERE version = original_version
4. If affected rows = 0, someone else modified - retry

Implementation

// Optimistic locking implementation
class ProductUpdater
{
    public function updateProduct(ProductId $id, array $data): void
    {
        $maxRetries = 3;
        $attempt = 0;
        
        while ($attempt < $maxRetries) {
            // Read with version
            $product = $this->productRepository->get($id);
            $originalVersion = $product->getData('lock_version');
            
            // Modify
            foreach ($data as $key => $value) {
                $product->setData($key, $value);
            }
            $product->setData('lock_version', $originalVersion + 1);
            
            // Update with version check
            $connection = $this->resource->getConnection();
            $affected = $connection->update(
                'catalog_product_entity',
                $product->getData(),
                ['entity_id = ? AND lock_version = ?', $id->getValue(), $originalVersion]
            );
            
            if ($affected > 0) {
                return;  // Success
            }
            
            $attempt++;
        }
        
        throw new OptimisticLockException('Failed to update after retries');
    }
}

Magento EAV Optimistic Locking

// EAV optimistic locking
public function saveWithLock(ProductInterface $product): void
{
    $connection = $this->resource->getConnection();
    $tableName = $this->resource->getTable('catalog_product_entity');
    
    $currentVersion = $product->getData('lock_version');
    
    $connection->update(
        $tableName,
        [
            'lock_version' => $currentVersion + 1,
            // other fields...
        ],
        ['entity_id = ? AND lock_version = ?', $product->getId(), $currentVersion]
    );
}

Version Conflict Handling

// Handle version conflicts
class ConflictResolver
{
    public function resolve(\Closure $operation): mixed
    {
        $maxRetries = 3;
        $attempt = 0;
        
        while ($attempt < $maxRetries) {
            try {
                return $operation();
            } catch (OptimisticLockException $e) {
                $attempt++;
                
                if ($attempt >= $maxRetries) {
                    throw new ConflictException('Unable to resolve conflict');
                }
                
                // Notify user or merge changes
                $this->notifyConflict($e);
            }
        }
    }
}

Key Takeaway

Optimistic locking uses version numbers to detect conflicts. Update with version check and retry on conflict. Suitable for low-contention scenarios.

Message Queue Pattern

// Producer: Add task to queue
class InventoryUpdater
{
    public function updateStock(ProductId $productId, int $quantity): void
    {
        $message = $this->messageFactory->create(
            'inventory.update',
            ['product_id' => $productId->getValue(), 'quantity' => $quantity]
        );
        $this->publisher->publish($message);
    }
}

// Consumer: Process one task at a time
class InventoryUpdateHandler
{
    public function handle(Message $message): void
    {
        $data = $message->getBody();
        
        // Sequential processing - no race condition
        $this->inventoryService->decrementStock(
            new ProductId($data['product_id']),
            $data['quantity']
        );
    }
}

RabbitMQ Configuration

<!-- Queue configuration -->
<config>
    <queue>
        <queue name="inventory.update">
            <consumer>inventory_updater</consumer>
            <handler>Vendor\Inventory\Model\Queue\Handler::process</handler>
        </queue>
    </queue>
</config>

Async Order Processing

// Process order asynchronously to avoid concurrency issues
class AsyncOrderProcessor
{
    public function process(Order $order): void
    {
        // Immediate: Validate and save order
        $this->orderRepository->save($order);
        
        // Async: Heavy processing via queue
        $this->publisher->publish(
            'order.process',
            ['order_id' => $order->getId()]
        );
    }
}

// Sequential consumer processing
class OrderProcessorHandler
{
    public function handle(Message $message): void
    {
        $order = $this->orderRepository->get($message->getOrderId());
        
        // Process sequentially - no concurrency issues
        $this->processPayment($order);
        $this->updateInventory($order);
        $this->sendConfirmation($order);
    }
}

Key Takeaway

Queue-based concurrency serializes concurrent operations. Tasks processed one at a time, eliminating race conditions. Use for high-contention scenarios.

Magento Lock Implementation

// Use Magento's Lock service
class InventoryService
{
    private LockManagerInterface $lockManager;
    
    public function decrementStock(ProductId $productId, int $quantity): void
    {
        $lockKey = 'inventory_' . $productId->getValue();
        
        if ($this->lockManager->lock($lockKey, 5)) {
            try {
                $this->doDecrement($productId, $quantity);
            } finally {
                $this->lockManager->unlock($lockKey);
            }
        } else {
            throw new LockException('Could not acquire inventory lock');
        }
    }
}

Cart Concurrency

// Handle concurrent cart updates
class CartService
{
    public function addItem(CartId $cartId, ProductId $productId, int $quantity): void
    {
        $cart = $this->cartRepository->get($cartId);
        
        // Check stock atomically
        $stock = $this->inventoryService->checkStock($productId, $quantity);
        if (!$stock) {
            throw new InsufficientStockException();
        }
        
        // Add item with version check
        $cart->addItem($productId, $quantity);
        $this->cartRepository->save($cart);
    }
}

Checkout Concurrency

// Prevent double checkout
class CheckoutService
{
    public function placeOrder(Cart $cart): Order
    {
        $lockKey = 'checkout_' . $cart->getId();
        
        if (!$this->lockManager->lock($lockKey, 30)) {
            throw new ConcurrentCheckoutException('Checkout already in progress');
        }
        
        try {
            // Validate cart
            $this->validateCart($cart);
            
            // Process payment
            $payment = $this->processPayment($cart);
            
            // Create order
            $order = $this->createOrder($cart, $payment);
            
            return $order;
        } finally {
            $this->lockManager->unlock($lockKey);
        }
    }
}

Inventory Reservation

// Reserve inventory during checkout
class InventoryReservation
{
    public function reserve(Cart $cart): void
    {
        foreach ($cart->getItems() as $item) {
            $this->inventoryService->reserve(
                $item->getProductId(),
                $item->getQuantity(),
                $cart->getId()  // reservation ID
            );
        }
    }
    
    public function release(CartId $cartId): void
    {
        $this->inventoryService->releaseReservation($cartId);
    }
}

Key Takeaway

Use Magento's Lock service for distributed locking. Implement inventory reservation during checkout. Handle concurrent cart and checkout operations safely.