Task Deduplication & Idempotency

Tại Sao Quan Trọng

Cloud Tasks guarantee at-least-once delivery, không exactly-once. Nếu bạn không:

Hiểu deduplication window — bạn sẽ tạo duplicate tasks không cần
Implement idempotency — bạn sẽ xử lý payment 2 lần, data corrupt

At-Least-Once Semantics

Why Not Exactly-Once?

Network ideal: Request → Handler → Response always reaches back
Network reality: Many failure points

Case 1: Handler processes successfully, response lost before Cloud Tasks receives
  Cloud Tasks: "No response, assume failed"
  Retry: Task executed again (duplicate work)

Case 2: Handler slow, Cloud Tasks timeout before handler finishes
  Handler: Finishes processing after timeout
  Cloud Tasks: "Timeout, retry"
  Retry: Task executed again

Exactly-once would require:
  - Global coordinator (performance hit)
  - Distributed consensus (complexity)
  - Not worth cost vs benefit

Solution: At-least-once + idempotent handlers.

Task ID & Deduplication Window

Task ID Semantics

Task ID là duy nhất trong queue scope:

projects/PROJECT/locations/REGION/queues/QUEUE_NAME/tasks/TASK_ID
                                                          ↑ unique here

Deduplication Window: 24 Hours After Deletion

Timeline:
[10:00] Task created with ID="payment-12345"
[10:05] Task completed, deleted

Dedup window: 10:05 → next day 10:05

During window:
[10:30] Try CreateTask with ID="payment-12345" again
  → Error: ALREADY_EXISTS

After window (next day 10:06):
[10:06 next day] Try CreateTask with ID="payment-12345"
  → Success: New task created (window expired)

Why 24 Hour Window?

Use case: Idempotent creation with retries

Client tries to create payment task:

Attempt 1 (10:00):
  CreateTask(task_id="payment-12345")
  → Success, but response lost

Attempt 2 (10:01):
  CreateTask(task_id="payment-12345")
  → Error: ALREADY_EXISTS
  → Client knows: "first attempt must have succeeded"

Attempt 3 (next day 10:10):
  CreateTask(task_id="payment-12345")
  → Success: new task (old window expired)
  → Risk: might duplicate if handler didn't clean up
  → But 24h > typical SLA (task completes within day)

Idempotent Handler Pattern

Handler Must Be Idempotent

// WRONG: Not idempotent
func PaymentHandler(w http.ResponseWriter, r *http.Request) {
    var req struct {
        PaymentID string `json:"payment_id"`
        Amount    int    `json:"amount"`
    }
    json.NewDecoder(r.Body).Decode(&req)
    
    // Every execution inserts
    db.Exec("INSERT INTO payments VALUES (?, ?)", req.PaymentID, req.Amount)
    
    w.WriteHeader(http.StatusOK)
}

// If called twice:
// Execution 1: INSERT payment-12345, $100 → success
// Execution 2: INSERT payment-12345, $100 → duplicate key error!

Pattern 1: Upsert (Database Primary Key)

// CORRECT: Idempotent via upsert
func PaymentHandler(w http.ResponseWriter, r *http.Request) {
    var req struct {
        PaymentID string `json:"payment_id"`
        Amount    int    `json:"amount"`
    }
    json.NewDecoder(r.Body).Decode(&req)
    
    // Upsert: insert or update if exists
    db.Exec(
        "INSERT INTO payments (payment_id, amount) VALUES (?, ?) ON DUPLICATE KEY UPDATE amount = ?",
        req.PaymentID, req.Amount, req.Amount,
    )
    
    w.WriteHeader(http.StatusOK)
}

// If called twice:
// Execution 1: INSERT payment-12345, $100 → success (row created)
// Execution 2: INSERT payment-12345, $100 → update (no change, but not error)
// Result: exactly once, idempotent

Pattern 2: Check Then Act

// CORRECT: Check existence first
func EmailHandler(w http.ResponseWriter, r *http.Request) {
    var req struct {
        NewsletterID string `json:"newsletter_id"`
        UserEmail    string `json:"user_email"`
    }
    json.NewDecoder(r.Body).Decode(&req)
    
    // Trace ID: combine fields into unique key
    traceID := fmt.Sprintf("%s:%s", req.NewsletterID, req.UserEmail)
    
    // Check if already sent
    existing := db.QueryRow(
        "SELECT sent_at FROM newsletter_sends WHERE trace_id = ?",
        traceID,
    ).Scan(&sent_at)
    
    if existing == nil {
        // Already sent, return success (idempotent)
        w.WriteHeader(http.StatusOK)
        return
    }
    
    // First time: send email
    err := sendEmail(req.UserEmail, ...)
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
    
    // Record that we sent
    db.Exec(
        "INSERT INTO newsletter_sends (trace_id, sent_at) VALUES (?, ?)",
        traceID, time.Now(),
    )
    
    w.WriteHeader(http.StatusOK)
}

// If called twice:
// Execution 1: Check fails (no row), send email, insert
// Execution 2: Check succeeds (row exists), return 200 OK (no re-send)
// Result: exactly once, idempotent

Pattern 3: Trace ID Injection

// Client side: set fixed trace ID (unique per logical operation)
client.CreateTask(ctx, &TaskRequest{
    Parent: "projects/X/locations/Y/queues/emails",
    Task: &Task{
        Name: fmt.Sprintf("...queues/emails/tasks/newsletter-2024-01-{%s}", user_email),
        HttpRequest: &HttpRequest{
            Url: "https://email-service.internal/send",
            Headers: map[string]string{
                "X-Idempotency-Key": fmt.Sprintf("newsletter-2024-01-%s", user_email),
            },
            Body: []byte(...),
        },
    },
})

// Handler side: extract and use trace ID
func EmailHandler(w http.ResponseWriter, r *http.Request) {
    traceID := r.Header.Get("X-Idempotency-Key")
    
    // Same as Pattern 2: check if already processed
    if isAlreadyProcessed(traceID) {
        w.WriteHeader(http.StatusOK)
        return
    }
    
    // Process
    err := process(traceID)
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
    
    w.WriteHeader(http.StatusOK)
}

Deduplication Boundary Cases

Case 1: Task Within Dedup Window

[10:00] CreateTask(id="payment-12345") → success
[10:01] CreateTask(id="payment-12345") → ALREADY_EXISTS error
[11:00] CreateTask(id="payment-12345") → ALREADY_EXISTS error (still within 24h)

Next day:
[10:01] CreateTask(id="payment-12345") → success (24h boundary passed)

Case 2: Deleted Task, New Task with Same ID

[10:00] CreateTask(id="payment-12345")
[10:05] Task completed, automatically deleted

[11:00] Try CreateTask(id="payment-12345") again
  → ALREADY_EXISTS (dedup window still active)

[10:06 next day] CreateTask(id="payment-12345")
  → Success (window expired)

Case 3: What If Handler Never Completes?

[10:00] CreateTask(id="payment-12345")
[10:05] Handler receives task, starts processing
[10:10] Handler still processing (slow operation)
[10:20] Cloud Tasks timeout (10 min default for App Engine)
  → Retry scheduled

[10:21] Try CreateTask(id="payment-12345") again
  → ALREADY_EXISTS (task still in queue, dedup window active)

Problem: Old task still running, new attempt blocked by dedup
Solution: Use long handler timeout OR ensure handler completes in time

Task ID Strategy

Anti-Pattern: Random IDs

// WRONG
taskID := generateRandomID() // UUID or random string
client.CreateTask(ctx, &TaskRequest{
    Parent: "...queues/my-queue",
    Task: &Task{
        Name: fmt.Sprintf("...queues/my-queue/tasks/%s", taskID),
    },
})

Problem:
- Random ID doesn't reflect logical operation
- Dedup doesn't help (each retry gets new ID)
- If task fails and you retry, you create duplicate task
- Dedup window useless

Pattern: Deterministic IDs

// CORRECT: ID derived from logical operation
paymentID := "payment-12345"
userID := "user-67890"
traceID := fmt.Sprintf("%s-%s-%d", paymentID, userID, time.Now().Unix())

client.CreateTask(ctx, &TaskRequest{
    Parent: "...queues/payment",
    Task: &Task{
        Name: fmt.Sprintf("...queues/payment/tasks/%s", traceID),
    },
})

// On client retry (no response received):
taskID := fmt.Sprintf("%s-%s-%d", paymentID, userID, sametime.Unix())
// Same ID → dedup kicks in → no duplicate task created

Pattern: Resource ID + Version

// Order processing
orderID := "order-99999"
version := 1

taskID := fmt.Sprintf("order-%s-v%d", orderID, version)

client.CreateTask(ctx, &TaskRequest{
    Parent: "...queues/orders",
    Task: &Task{
        Name: fmt.Sprintf("...queues/orders/tasks/%s", taskID),
    },
})

// If need to reprocess order with version 2:
taskID := fmt.Sprintf("order-%s-v%d", orderID, 2)
// Different ID → new task (not blocked by dedup)

Idempotency Storage

Where to Store Idempotency Key?

Option 1: Database (same as business logic)
  Pros: ACID guarantees, consistent
  Cons: Extra DB call, latency
  
Option 2: Cache (Redis)
  Pros: Fast, low latency
  Cons: TTL-based (might expire), cache miss = reprocess
  
Option 3: Business Data itself
  Pros: No extra storage, consistent
  Cons: Must design schema carefully

Example: Payment Service

// Database schema
CREATE TABLE payments (
    payment_id VARCHAR PRIMARY KEY,
    amount INT,
    status ENUM('PENDING', 'COMPLETED', 'FAILED'),
    created_at TIMESTAMP,
    updated_at TIMESTAMP,
);

// Handler: check status first
func PaymentHandler(w http.ResponseWriter, r *http.Request) {
    var req struct {
        PaymentID string `json:"payment_id"`
        Amount    int    `json:"amount"`
    }
    json.NewDecoder(r.Body).Decode(&req)
    
    // Check status
    var status string
    err := db.QueryRow("SELECT status FROM payments WHERE payment_id = ?", req.PaymentID).Scan(&status)
    
    if err == nil && status == "COMPLETED" {
        // Already processed, idempotent return
        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, `{"status": "already_completed"}`)
        return
    }
    
    // Process new payment
    result := processPayment(req.PaymentID, req.Amount)
    
    if result.Success {
        db.Exec("UPDATE payments SET status = 'COMPLETED' WHERE payment_id = ?", req.PaymentID)
        w.WriteHeader(http.StatusOK)
    } else {
        w.WriteHeader(http.StatusInternalServerError)
    }
}

Task Deduplication & Idempotency ​

Tại Sao Quan Trọng ​

At-Least-Once Semantics ​

Why Not Exactly-Once? ​

Task ID & Deduplication Window ​

Task ID Semantics ​

Deduplication Window: 24 Hours After Deletion ​

Why 24 Hour Window? ​

Idempotent Handler Pattern ​

Handler Must Be Idempotent ​

Pattern 1: Upsert (Database Primary Key) ​

Pattern 2: Check Then Act ​

Pattern 3: Trace ID Injection ​

Deduplication Boundary Cases ​

Case 1: Task Within Dedup Window ​

Case 2: Deleted Task, New Task with Same ID ​

Case 3: What If Handler Never Completes? ​

Task ID Strategy ​

Anti-Pattern: Random IDs ​

Pattern: Deterministic IDs ​

Pattern: Resource ID + Version ​

Idempotency Storage ​

Where to Store Idempotency Key? ​

Example: Payment Service ​

References ​

Task Deduplication & Idempotency

Tại Sao Quan Trọng

At-Least-Once Semantics

Why Not Exactly-Once?

Task ID & Deduplication Window

Task ID Semantics

Deduplication Window: 24 Hours After Deletion

Why 24 Hour Window?

Idempotent Handler Pattern

Handler Must Be Idempotent

Pattern 1: Upsert (Database Primary Key)

Pattern 2: Check Then Act

Pattern 3: Trace ID Injection

Deduplication Boundary Cases

Case 1: Task Within Dedup Window

Case 2: Deleted Task, New Task with Same ID

Case 3: What If Handler Never Completes?

Task ID Strategy

Anti-Pattern: Random IDs

Pattern: Deterministic IDs

Pattern: Resource ID + Version

Idempotency Storage

Where to Store Idempotency Key?

Example: Payment Service

References