Async Operations — Bulk API, Indexation, and Patterns

Synchronous vs Asynchronous

Synchronous: Client -> API -> Process -> Response (blocks)
Asynchronous: Client -> API -> Queue -> Response (immediate)

Bulk API Endpoint

POST /rest/async/bulk
Content-Type: application/json

[
  {
    "httpMethod": "POST",
    "requestUri": "/rest/V1/products",
    "body": {"product": {"sku": "SKU1", "name": "Product 1"}}
  },
  {
    "httpMethod": "POST",
    "requestUri": "/rest/V1/products",
    "body": {"product": {"sku": "SKU2", "name": "Product 2"}}
  }
]

Response

{
  "bulk_id": "3a1b2c3d-4e5f-6789-abcd-ef0123456789",
  "status_link": "/rest/V1/bulk/3a1b2c3d-4e5f-6789-abcd-ef0123456789/status"
}

Check Bulk Status

GET /rest/V1/bulk/{bulk_id}/status

{
  "bulk_id": "3a1b2c3d-4e5f-6789-abcd-ef0123456789",
  "status": "complete",
  "result": {
    "success_count": 2,
    "error_count": 0
  }
}

Enable Async Indexing

// app/etc/env.php
return [
    'indexer' => [
        'async_indexing' => true
    ]
];

How Async Indexing Works

1. Product save triggers index update message
2. Message published to queue
3. Consumer processes batch reindex
4. Index updated in background

// Automatic with async_indexing enabled
$this->product->save(); // Triggers async index update
// Response returned immediately

Manual Async Reindex

# Queue reindex operations
php bin/magento indexer:set-mode realtime catalog_product_flat

# Process queued reindex
php bin/magento queue:consumers:start index.update products --max-messages=1000

Benefits

• Faster page loads (no index wait)
• Better API response times
• Reduced database locks
• Improved scalability

Fire and Forget

namespace Vendor\Module\Service;

use Magento\Framework\MessageQueue\PublisherInterface;

class OrderNotifier
{
    public function __construct(
        private PublisherInterface $publisher
    ) {}

    public function notify(int $orderId): void
    {
        $this->publisher->publish(
            'vendor.order.notify',
            json_encode(['order_id' => $orderId])
        );
        // Return immediately, process in background
    }
}

Request-Reply Pattern

// Publish request
$requestId = $this->uuid->generate();
$this->publisher->publish(
    'vendor.export.request',
    json_encode(['request_id' => $requestId, 'data' => $data])
);

// Wait for reply (poll)
while (!$this->isComplete($requestId)) {
    sleep(1);
}
return $this->getResult($requestId);

Saga Pattern (Multi-Step)

// Step 1: Validate
$this->publisher->publish('order.validate', $data);
// Consumer triggers Step 2 on success
// Step 2: Process payment
$this->publisher->publish('order.payment', $data);
// Consumer triggers Step 3 on success
// Step 3: Ship
$this->publisher->publish('order.ship', $data);

Throttling

class ThrottledPublisher
{
    private int $count = 0;
    private int $limit = 100;

    public function publish(string $topic, string $body): void
    {
        $this->publisher->publish($topic, $body);
        $this->count++;
        
        if ($this->count >= $this->limit) {
            sleep(1); // Throttle
            $this->count = 0;
        }
    }
}

Async vs Sync Decision Matrix

Batch Size Optimization

// Too small: overhead per message
$batchSize = 1;

// Too large: memory issues
$batchSize = 100000;

// Optimal: balance throughput and memory
$batchSize = 500;

Performance Metrics

-- Queue depth monitoring
SELECT queue_name, COUNT(*) as pending
FROM queue_message
WHERE status = 'new'
GROUP BY queue_name
ORDER BY pending DESC;

-- Processing rate
SELECT
    DATE(created_at) as day,
    COUNT(*) as messages_processed
FROM queue_message
WHERE status = 'complete'
GROUP BY DATE(created_at);

Monitoring Dashboard

-- Consumer performance
SELECT
    consumer_name,
    COUNT(*) as processed,
    AVG(TIMESTAMPDIFF(SECOND, started_at, finished_at)) as avg_time
FROM queue_consumer
WHERE finished_at > DATE_SUB(NOW(), INTERVAL 24 HOUR)
GROUP BY consumer_name;