Chapter 3 · Distributed Coordination and Containerized Deployment Advanced

Chapter goals:

• Master the correct implementation of Redis distributed locks to avoid duplicate consumption
• Understand the importance of idempotency design in Agent tasks
• Learn to use Docker multi-stage builds to optimize Python image size
• Be able to orchestrate multi-container Agent services using docker-compose
• Lay the foundation for production environment deployment with containerization

---

3.1 Distributed Locks: Preventing Duplicate Consumption

In Chapter 2, we learned about Celery Worker consuming tasks from Redis queue. But what happens when you have multiple Workers consuming from the same queue?

         ┌─────────────┐
         │  Redis Queue │
         └──────┬──────┘
                │
       ┌────────┼────────┐
       ▼        ▼        ▼
   Worker 1  Worker 2  Worker 3

Without coordination, multiple Workers might:

1. Fetch the same task (duplicate execution)
2. Overwrite each other's results (data corruption)
3. Compete for shared resources (race conditions)

3.1.1 Redis Distributed Lock Implementation

Redis provides SET key value NX PX milliseconds for atomic lock acquisition:

import redis
import uuid
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def acquire_lock(resource: str, ttl: int = 10) -> str | None:
    """
    Acquire distributed lock using SET NX PX
    Returns lock_id on success, None on failure
    """
    lock_id = str(uuid.uuid4())
    # NX = only set if not exists, PX = expire in milliseconds
    acquired = r.set(f"lock:{resource}", lock_id, nx=True, ex=ttl)
    if acquired:
        return lock_id
    return None

def release_lock(resource: str, lock_id: str) -> bool:
    """
    Release lock using Lua script (atomic check + delete)
    """
    script = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    else
        return 0
    end
    """
    return r.eval(script, 1, f"lock:{resource}", lock_id) == 1

3.1.2 Safe Task Processing Pattern

@app.task(bind=True)
def process_task(self, task_id: str, data: dict):
    lock_id = acquire_lock(f"task:{task_id}", ttl=30)
    if not lock_id:
        # Another Worker is processing this task, requeue
        raise self.retry(countdown=5)
    
    try:
        # Process task...
        result = do_work(data)
        return result
    finally:
        release_lock(f"task:{task_id}", lock_id)

---

3.2 Idempotency Design

Idempotency means: calling the same operation multiple times produces the same result. This is critical for:

• Retry scenarios (network timeout → retry → duplicate execution)
• Message queue re-delivery (Worker crashes before ack → message redelivered)
• Distributed system consistency

3.2.1 Idempotency Keys

Use a unique key to track processed tasks:

from redis import Redis

r = Redis()

def process_with_idempotency(task_id: str, payload: dict):
    # Check if already processed
    if r.exists(f"processed:{task_id}"):
        return r.get(f"result:{task_id}")
    
    # Process task
    result = do_expensive_work(payload)
    
    # Mark as processed (with TTL for cleanup)
    pipe = r.pipeline()
    pipe.set(f"processed:{task_id}", "1", ex=3600)
    pipe.set(f"result:{task_id}", json.dumps(result), ex=3600)
    pipe.execute()
    
    return result

3.2.2 Database-Level Idempotency

For database operations, use unique constraints:

-- Create table with unique constraint
CREATE TABLE agent_tasks (
    task_id VARCHAR(36) PRIMARY KEY,
    status VARCHAR(20),
    result TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- INSERT ... ON DUPLICATE KEY UPDATE (MySQL)
INSERT INTO agent_tasks (task_id, status, result)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE status = VALUES(status);

---

3.3 Docker Multi-Stage Builds

Python Docker images can be huge (1GB+). Multi-stage builds optimize size:

# Stage 1: Build stage
FROM python:3.11-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: Runtime stage
FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /install /usr/local
COPY . .

# Non-root user for security
RUN useradd -m agentuser
USER agentuser

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

3.3.1 Image Size Comparison

Stage	Image Size
Full Python image	~900MB
Multi-stage build	~150MB
Alpine-based	~80MB

---

3.4 Docker Compose for Agent Services

# docker-compose.yml
version: "3.9"

services:
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  api:
    build: .
    ports: ["8000:8000"]
    environment:
      - REDIS_URL=redis://redis:6379/0
      - DATABASE_URL=postgresql://user:pass@db:5432/agent
    depends_on:
      redis:
        condition: service_healthy
      db:
        condition: service_healthy
    restart: unless-stopped

  worker:
    build: .
    command: celery -A tasks worker --loglevel=info --concurrency=4
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped

  db:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=agent
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 10s
      timeout: 5s
      retries: 5

volumes:
  redis_data:
  postgres_data:

3.4.1 Starting Services

# Build and start all services
docker compose up -d --build

# View logs
docker compose logs -f api worker

# Stop services
docker compose down

# Stop and remove volumes (clean slate)
docker compose down -v

---

3.5 Production Deployment Checklist

3.5.1 Resource Limits

services:
  worker:
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G
        reservations:
          cpus: '1'
          memory: 2G

3.5.2 Health Checks

healthcheck:
  test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
  interval: 30s
  timeout: 10s
  retries: 3
  start_period: 40s

3.5.3 Log Rotation

services:
  api:
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

---

Chapter Summary

1. Distributed Locks: Use SET NX PX for atomic lock acquisition, Lua scripts for safe release;
2. Idempotency: Essential for retry scenarios, use unique keys or database constraints;
3. Multi-Stage Builds: Reduce image size from 900MB to ~150MB;
4. Docker Compose: Orchestrate multi-service Agent deployments;
5. Production Considerations: Resource limits, health checks, log rotation.

📌 Containerization is the foundation for cloud deployment. Next chapter covers AWS ECS/Fargate deployment.

---

🛠️ Hands-on Practice

1. Implement a Redis-based distributed lock class with automatic lease renewal (watchdog pattern).

2. Design an idempotent task processor using database unique constraints, test with concurrent retries.

3. Create a docker-compose.yml for a complete Agent service (FastAPI + Celery Worker + Redis + PostgreSQL) with health checks and resource limits.

---

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What is the purpose of SET key value NX PX in Redis?

2. Why is idempotency critical in distributed systems?

3. What is the main benefit of Docker multi-stage builds?

4. Which command starts Docker Compose services in detached mode?