Idempotent Consumer

บริบท

ทุก relay ในโมดูลนี้ — ไม่ว่าจะ tail transaction log หรือ poll outbox — ส่งแบบ at least once relay สามารถ publish ข้อความแล้ว crash ก่อนบันทึกว่าทำไปแล้ว พอ restart จึง publish ซ้ำอีกรอบ broker เองก็ส่งซ้ำเมื่อ consumer ไม่ ack ทันเวลา ผลลัพธ์จึงหลีกเลี่ยงไม่ได้: consumer ของคุณบางครั้งจะได้รับข้อความเดียวกันมากกว่าหนึ่งครั้ง

ปัญหา

handler จำนวนมากไม่ปลอดภัยที่จะรันสองครั้ง การเครดิตเงินเข้าบัญชี การลดสต็อก การชาร์จบัตร หรือการส่งของ ล้วนเป็นการดำเนินงานที่ duplicate ก่อให้เกิดความเสียหายจริง — ชาร์จซ้ำ สต็อกติดลบ พัสดุสองชิ้น คุณไม่สามารถป้องกันการส่งซ้ำได้ (นั่นเป็นกลไกความปลอดภัยของ broker) และคุณไม่สามารถเรียกร้องการส่งแบบ exactly-once จากระบบที่ไม่ได้ให้สิ่งนั้น

ดังนั้นแรงต่าง ๆ คือ: ข้อความสามารถและจะถูกส่งซ้ำ บาง side effect ต้องไม่เกิดขึ้นสองครั้ง และคุณไม่สามารถผลักปัญหา deduplication กลับไปให้ broker

วิธีแก้

ทำให้ consumer เป็น idempotent — การประมวลผลข้อความสองครั้งมีผลเหมือนกับการประมวลผลครั้งเดียว มีสองแนวทาง และ handler ที่ดีที่สุดรวมทั้งสองเข้าด้วยกัน

แนวทางทั่วไปคือการ ติดตาม message id ที่ประมวลผลแล้ว ทุกข้อความพก id ที่ไม่ซ้ำกัน (กำหนดตอนที่สร้างแถว outbox) consumer เก็บตาราง processed_messages ไว้ พอข้อความมาถึง ก็พยายามบันทึก id แล้วนำ side effect ไปใช้ ใน local transaction เดียวกัน ถ้า id มีอยู่แล้ว ข้อความนั้นเป็น duplicate และถูก ack โดยไม่นำ effect ไปใช้ซ้ำ การบันทึก id และทำงานใน transaction เดียวคือสิ่งที่ทำให้สิ่งนี้รัดกุม — การ crash จะ commit ทั้งคู่หรือไม่เกิดเลย

แนวทางตามธรรมชาติคือการ ออกแบบ side effect ให้เป็น idempotent โดยเนื้อแท้: upsert ที่ key ด้วย business id, conditional update ที่ป้องกันด้วย status หรือ SET balance = X แทน balance = balance + Y เมื่อการดำเนินงานเป็น idempotent โดยธรรมชาติ คุณอาจไม่ต้องการตาราง dedup เลย

sequenceDiagram
  participant B as Message Broker
  participant C as Consumer
  participant DB as Consumer DB
  B->>C: deliver msg 42
  C->>DB: BEGIN, INSERT id 42, apply effect, COMMIT
  C-->>B: ack
  B->>C: redeliver msg 42 (duplicate)
  C->>DB: INSERT id 42 - already exists
  Note over C,DB: duplicate detected, effect skipped
  C-->>B: ack (no double effect)

consumer บันทึกแต่ละ message id พร้อม effect ที่เกิดขึ้น ข้อความซ้ำที่ถูกส่งมาใหม่จะถูกตรวจจับและข้ามไป

ตัวอย่าง

handler insert message id และนำ side effect ไปใช้ใน transaction เดียว การละเมิด unique-constraint บน id หมายถึง “ประมวลผลแล้ว” ดังนั้น duplicate จึงถูกกลืนและ ack

// Returns true if newly processed, false if it was a duplicate.
async function handleMessage(pool: Pool, msg: Message): Promise<boolean> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    const res = await client.query(
      'INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT DO NOTHING',
      [msg.id],
    );
    if (res.rowCount === 0) {
      await client.query('ROLLBACK'); // duplicate: id already recorded
      return false;
    }

    // Side effect runs in the SAME transaction as recording the id.
    await client.query(
      'UPDATE accounts SET balance = balance + $1 WHERE id = $2',
      [msg.amount, msg.accountId],
    );

    await client.query('COMMIT');
    return true;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

def handle_message(conn, msg) -> bool:
    """Return True if newly processed, False if it was a duplicate."""
    with conn:  # one transaction: record id + apply effect
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO processed_messages (id) VALUES (%s) "
                "ON CONFLICT DO NOTHING",
                (msg.id,),
            )
            if cur.rowcount == 0:
                return False  # duplicate: id already recorded

            # Side effect runs in the same transaction as recording the id.
            cur.execute(
                "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                (msg.amount, msg.account_id),
            )
    return True

// Returns true if newly processed, false if it was a duplicate.
func HandleMessage(ctx context.Context, db *sql.DB, msg Message) (bool, error) {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return false, err
    }
    defer tx.Rollback()

    res, err := tx.ExecContext(ctx,
        `INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT DO NOTHING`,
        msg.ID)
    if err != nil {
        return false, err
    }
    if n, _ := res.RowsAffected(); n == 0 {
        return false, nil // duplicate: id already recorded
    }

    // Side effect runs in the same transaction as recording the id.
    if _, err := tx.ExecContext(ctx,
        `UPDATE accounts SET balance = balance + $1 WHERE id = $2`,
        msg.Amount, msg.AccountID); err != nil {
        return false, err
    }

    return true, tx.Commit()
}

use sqlx::{Pool, Postgres};

/// Returns Ok(true) if newly processed, Ok(false) if it was a duplicate.
async fn handle_message(pool: &Pool<Postgres>, msg: &Message) -> Result<bool, sqlx::Error> {
    let mut tx = pool.begin().await?; // record id + apply effect in one tx

    let inserted = sqlx::query(
        "INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT DO NOTHING",
    )
    .bind(msg.id)
    .execute(&mut *tx)
    .await?;

    if inserted.rows_affected() == 0 {
        return Ok(false); // duplicate: id already recorded
    }

    // Side effect runs in the same transaction as recording the id.
    sqlx::query("UPDATE accounts SET balance = balance + $1 WHERE id = $2")
        .bind(msg.amount)
        .bind(msg.account_id)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok(true)
}

ผลลัพธ์ที่ตามมา

สิ่งที่คุณได้:

ความปลอดภัยจากการส่งซ้ำ duplicate ถูกตรวจจับและข้าม ดังนั้นการส่งแบบ at-least-once จึงกลายเป็น การประมวลผล แบบ exactly-once อย่างมีประสิทธิภาพ — การรับประกันที่สำคัญจริง ๆ
สัญญาที่สะอาดกับส่วนที่เหลือของโมดูล เพราะ consumer เป็น idempotent ฝั่ง outbox และ relay จึงออกแบบให้เรียบง่ายและส่งซ้ำได้โดยไม่ต้องกลัว

สิ่งที่คุณต้องแลก:

state ที่ต้องเก็บและตัดทิ้ง ตาราง processed_messages โตขึ้น คุณต้องตัด id เก่าทิ้ง (เช่น ตามช่วงเวลา) และกำหนดขนาดให้เหมาะกับ throughput ของคุณ
effect ต้องใช้ transaction ร่วมกัน การบันทึก id และการนำ side effect ไปใช้ต้องเป็น atomic ถ้า side effect แตะระบบภายนอกที่ไม่สามารถเข้าร่วม transaction ได้ คุณก็กลับไปสู่ dual-write ขนาดเล็กกว่า — ในกรณีนั้นควรเลือกการดำเนินงานที่เป็น idempotent โดยธรรมชาติ
การเรียงลำดับยังคงเป็นปัญหาของคุณ idempotency หยุดการประมวลผลซ้ำ แต่ไม่รับประกันว่าข้อความจะมาถึงตามลำดับ ถ้าลำดับสำคัญ ให้จัดการแยกต่างหาก

เนื้อหาที่เกี่ยวข้อง

Transactional Outbox — กำหนด message id ที่ไม่ซ้ำกันซึ่งแพตเทิร์นนี้ใช้ทำ dedup
Transaction Log Tailing — relay ที่ส่งแบบ at-least-once ซึ่ง pattern นี้เข้ามาดูดซับให้
Polling Publisher — อีก relay แบบ at-least-once ที่แพตเทิร์นนี้ป้องกัน

ข้อดี	ข้อแลกเปลี่ยน
message ที่ deliver ซ้ำไม่ทำให้ผลลัพธ์เปลี่ยน	ต้องมี idempotency store — storage เพิ่ม
ระบบทนต่อ at-least-once delivery guarantee	lookup idempotency key เพิ่ม latency เล็กน้อย
retry อย่างปลอดภัยโดยไม่กลัว duplicate effect	idempotency key ต้อง expire อย่างระมัดระวัง
ง่ายต่อการ reason เกี่ยวกับ correctness	ต้องออกแบบ idempotency key ให้ unique และ stable

ข้อผิดพลาดที่พบบ่อย

Idempotency Key ที่ไม่ Stable — key เปลี่ยนทุก retry อาการ:

ใช้ timestamp หรือ random UUID เป็น idempotency key
retry ทุกครั้งถือว่าเป็น operation ใหม่ — duplicate effect เกิดได้เสมอ
ใช้ message ID จาก broker หรือ business key แทน

Idempotency Key ที่ไม่ Expire — เก็บ key ตลอดไป อาการ:

idempotency store ใหญ่ขึ้นไม่จำกัด
query ช้าเมื่อ store ใหญ่มาก
กำหนด TTL ที่เหมาะสมตาม message retention ของ broker

💡 ตัวอย่างจากของจริง

Stripe:

idempotency key บน payment API เป็น best practice ที่ document ชัดเจน

client ส่ง Idempotency-Key header ใน POST request

Stripe เก็บ result 24 ชั่วโมง — retry ด้วย key เดิมได้รับ result เดิม

AWS SQS:

at-least-once delivery โดย design — consumer ต้องเป็น idempotent เอง

FIFO queue มี message deduplication built-in (5 นาที window)

🧩 มุ่งสู่การประมวลผลแบบ idempotent ไม่ใช่การส่งแบบ exactly-once

การส่งแบบ exactly-once ข้าม network แทบเป็นไปไม่ได้ การประมวลผล แบบ exactly-once ทำได้และเป็นสิ่งที่คุณต้องการจริง ๆ บันทึก message id และนำ effect ไปใช้ใน transaction เดียว หรือทำให้ effect เป็น idempotent โดยธรรมชาติ แล้ว duplicate ก็หยุดมีความสำคัญ วินัยฝั่ง consumer นี้คือสิ่งที่ทำให้ทุกแพตเทิร์นอื่นในโมดูลนี้ยังคงเรียบง่าย