OCI Queue Service: Building Decoupled Microservices with Terraform

Synchronous API calls between microservices create tight coupling. Service A calls Service B and waits. If Service B is slow, Service A is slow. If Service B is down, Service A fails. If traffic spikes, both services need to scale simultaneously or requests start timing out. This is a fragile architecture for anything beyond simple workloads.

Message queues decouple producers from consumers. Service A puts a message in a queue and moves on. Service B reads from the queue at its own pace. If Service B is temporarily unavailable, messages accumulate in the queue and get processed when it comes back. If traffic spikes, the queue absorbs the burst and consumers process it without the producer being affected.

OCI Queue is Oracle’s managed message queue service. It is HTTP-based, requires no client libraries beyond a standard HTTP client, supports at-least-once delivery, and integrates natively with OCI IAM, Monitoring, and Functions. This post covers deploying OCI Queue with Terraform, building producers and consumers in Python, configuring dead letter queues, and handling visibility timeouts correctly.

Architecture

			
Order Service (producer)
        |
   OCI Queue (orders-processing-queue)
        |
   -------------------------
   |                       |
Notification           Inventory
Consumer               Consumer
        |
Dead Letter Queue
(failed messages)

		

Step 1: IAM Policy

			
resource "oci_identity_dynamic_group" "queue_consumers_dg" {
  compartment_id = var.tenancy_ocid
  name           = "queue-consumer-instances"
  description    = "Instances and Functions that consume from OCI Queue"
  matching_rule  = "Any {instance.compartment.id = '${var.compartment_id}', resource.type = 'fnfunc'}"
}
resource "oci_identity_policy" "queue_policy" {
  compartment_id = var.compartment_id
  name           = "oci-queue-policy"
  description    = "Permissions for OCI Queue producers and consumers"
  statements = [
    "Allow group ${var.producer_group_name} to use queues in compartment id ${var.compartment_id}",
    "Allow dynamic-group queue-consumer-instances to use queues in compartment id ${var.compartment_id}",
    "Allow dynamic-group queue-consumer-instances to read queues in compartment id ${var.compartment_id}"
  ]
}

		

Step 2: Provision Queue and Dead Letter Queue

			
resource "oci_queue_queue" "orders_dlq" {
  compartment_id         = var.compartment_id
  display_name           = "orders-processing-dlq"
  visibility_in_seconds  = 30
  timeout_in_seconds     = 30
  retention_in_seconds   = 604800
  max_in_flight_messages = 100
  defined_tags = {
    "Operations.Environment" = "production"
    "Operations.Purpose"     = "dead-letter-queue"
    "Operations.ManagedBy"   = "terraform"
  }
}
resource "oci_queue_queue" "orders_queue" {
  compartment_id                   = var.compartment_id
  display_name                     = "orders-processing-queue"
  visibility_in_seconds            = 60
  timeout_in_seconds               = 30
  retention_in_seconds             = 86400
  max_in_flight_messages           = 500
  dead_letter_queue_delivery_count = 3
  custom_encryption_key_id         = var.vault_key_id
  defined_tags = {
    "Operations.Environment" = "production"
    "Operations.Application" = "orders"
    "Operations.ManagedBy"   = "terraform"
  }
}
output "queue_endpoint" {
  value = oci_queue_queue.orders_queue.messages_endpoint
}
output "queue_id" {
  value = oci_queue_queue.orders_queue.id
}

		

The visibility_in_seconds = 60 setting is critical. When a consumer reads a message, OCI Queue hides it from other consumers for 60 seconds. If the consumer processes and deletes it within that window, it is gone. If the consumer crashes, the message becomes visible again after 60 seconds. The dead_letter_queue_delivery_count = 3 means after three failed visibility timeout expirations, OCI Queue moves the message to the DLQ automatically.

Step 3: Producer Implementation

			
import oci
import json
import logging
import base64
from datetime import datetime, timezone
from typing import List
logger = logging.getLogger(__name__)
class OrderQueueProducer:
    def __init__(self, queue_id: str, messages_endpoint: str):
        self.queue_id          = queue_id
        self.messages_endpoint = messages_endpoint
        config = oci.config.from_file()
        self.client = oci.queue.QueueClient(
            config,
            service_endpoint=messages_endpoint
        )
    def publish_order_event(self, order_id: str, event_type: str, payload: dict) -> str:
        message_body = json.dumps({
            "event_type":   event_type,
            "order_id":     order_id,
            "payload":      payload,
            "published_at": datetime.now(timezone.utc).isoformat(),
            "version":      "1.0"
        })
        put_details = oci.queue.models.PutMessagesDetails(
            messages=[
                oci.queue.models.PutMessagesDetailsEntry(
                    content=base64.b64encode(message_body.encode()).decode()
                )
            ]
        )
        response = self.client.put_messages(
            queue_id=self.queue_id,
            put_messages_details=put_details
        )
        message_id = response.data.messages[0].id
        logger.info(f"Published {event_type} for order {order_id}, message_id: {message_id}")
        return message_id
    def publish_batch(self, events: List[dict]) -> List[str]:
        if len(events) > 20:
            raise ValueError("OCI Queue supports maximum 20 messages per batch")
        entries = [
            oci.queue.models.PutMessagesDetailsEntry(
                content=base64.b64encode(json.dumps(event).encode()).decode()
            )
            for event in events
        ]
        response = self.client.put_messages(
            queue_id=self.queue_id,
            put_messages_details=oci.queue.models.PutMessagesDetails(messages=entries)
        )
        return [msg.id for msg in response.data.messages]

		

Step 4: Consumer with Visibility Timeout Renewal

For long-running processing tasks, the consumer must renew the visibility timeout before it expires, otherwise OCI Queue makes the message visible again and another consumer picks it up, causing duplicate processing.

			
import oci
import json
import base64
import logging
import time
import threading
from typing import Callable
logger = logging.getLogger(__name__)
class OrderQueueConsumer:
    def __init__(self, queue_id: str, messages_endpoint: str, visibility_in_seconds: int = 60):
        self.queue_id              = queue_id
        self.messages_endpoint     = messages_endpoint
        self.visibility_in_seconds = visibility_in_seconds
        self.running               = False
        config = oci.config.from_file()
        self.client = oci.queue.QueueClient(config, service_endpoint=messages_endpoint)
    def _renew_visibility(self, receipt: str, stop_event: threading.Event):
        renewal_interval = self.visibility_in_seconds * 0.7
        while not stop_event.wait(timeout=renewal_interval):
            try:
                self.client.update_message(
                    queue_id=self.queue_id,
                    message_receipt=receipt,
                    update_message_details=oci.queue.models.UpdateMessageDetails(
                        visibility_in_seconds=self.visibility_in_seconds
                    )
                )
                logger.debug(f"Renewed visibility for receipt {receipt[:20]}...")
            except Exception as e:
                logger.warning(f"Failed to renew visibility: {e}")
                break
    def process_message(self, message, handler: Callable) -> bool:
        content = json.loads(base64.b64decode(message.content).decode())
        receipt = message.receipt
        stop_renewal = threading.Event()
        renewal_thread = threading.Thread(
            target=self._renew_visibility,
            args=(receipt, stop_renewal),
            daemon=True
        )
        renewal_thread.start()
        try:
            handler(content)
            self.client.delete_message(queue_id=self.queue_id, message_receipt=receipt)
            logger.info(f"Processed and deleted: {content.get('event_type')}")
            return True
        except Exception as e:
            logger.error(f"Processing failed: {e}. Message will become visible again.")
            return False
        finally:
            stop_renewal.set()
            renewal_thread.join(timeout=2)
    def consume(self, handler: Callable, max_messages: int = 10):
        self.running = True
        while self.running:
            try:
                response = self.client.get_messages(
                    queue_id=self.queue_id,
                    limit=max_messages,
                    visibility_in_seconds=self.visibility_in_seconds,
                    timeout_in_seconds=20
                )
                for message in response.data.messages:
                    self.process_message(message, handler)
            except oci.exceptions.ServiceError as e:
                logger.error(f"Queue service error: {e}")
                time.sleep(5)
            except KeyboardInterrupt:
                self.running = False

		

Step 5: Dead Letter Queue Monitoring

			
resource "oci_monitoring_alarm" "dlq_messages_alarm" {
  compartment_id        = var.compartment_id
  display_name          = "orders-dlq-has-messages"
  is_enabled            = true
  metric_compartment_id = var.compartment_id
  namespace             = "oci_queue"
  query                 = "VisibleMessages[5m]{queueId = '${oci_queue_queue.orders_dlq.id}'}.max() > 0"
  severity              = "CRITICAL"
  pending_duration      = "PT5M"
  destinations          = [var.ops_notification_topic_id]
  body                  = "Messages have arrived in the orders DLQ. These represent failed processing attempts requiring investigation."
}
resource "oci_monitoring_alarm" "queue_depth_alarm" {
  compartment_id        = var.compartment_id
  display_name          = "orders-queue-depth-high"
  is_enabled            = true
  metric_compartment_id = var.compartment_id
  namespace             = "oci_queue"
  query                 = "VisibleMessages[5m]{queueId = '${oci_queue_queue.orders_queue.id}'}.max() > 1000"
  severity              = "WARNING"
  pending_duration      = "PT10M"
  destinations          = [var.ops_notification_topic_id]
  body                  = "Orders queue depth exceeded 1000 messages. Consumers may be falling behind."
}

		

Step 6: Deploying Consumers on OKE

			
apiVersion: apps/v1
kind: Deployment
metadata:
  name: orders-queue-consumer
  namespace: orders
spec:
  replicas: 3
  selector:
    matchLabels:
      app: orders-queue-consumer
  template:
    metadata:
      labels:
        app: orders-queue-consumer
    spec:
      serviceAccountName: queue-consumer-sa
      containers:
        - name: consumer
          image: <region>.ocir.io/<namespace>/orders-consumer:latest
          env:
            - name: QUEUE_ID
              valueFrom:
                secretKeyRef:
                  name: queue-config
                  key: queue-id
            - name: QUEUE_ENDPOINT
              valueFrom:
                secretKeyRef:
                  name: queue-config
                  key: messages-endpoint
          resources:
            requests:
              cpu: "250m"
              memory: "256Mi"
            limits:
              cpu: "500m"
              memory: "512Mi"

		

Key Operational Considerations

Idempotency is mandatory. OCI Queue guarantees at-least-once delivery, which means the same message can be delivered more than once. Your handler must be idempotent: processing the same order event twice must produce the same result as processing it once. Use the order_id and event_type combination as an idempotency key and check a processed-events store before executing any state-changing operation.

Message ordering is not guaranteed. If your use case requires strict ordering of events for the same order, include a sequence number in the message payload and handle out-of-order delivery in the consumer logic.

The visibility timeout must exceed your maximum processing time. If processing occasionally takes 45 seconds, set the timeout to at least 90 seconds. A message that expires its visibility timeout before deletion gets reprocessed, triggering your idempotency logic and wasting consumer capacity.

Long polling reduces empty receive calls. Setting timeout_in_seconds=20 in the get_messages call tells OCI Queue to wait up to 20 seconds for a message before returning an empty response. Without long polling, a quiet queue generates a constant stream of empty API calls.

Regards,
Osama

#OCI #OracleCloud #Microservices #MessageQueue #Terraform #CloudNative #InfrastructureAsCode #IaC #OracleCloudInfrastructure #PlatformEngineering #CloudArchitecture #DevOps #DistributedSystems #BackendEngineering #TechBlog #AsyncMessaging #CloudEngineering #Oracle #OCIQueue #EventDriven