DLQ สำหรับ AI Agent: กักข้อความที่ล้มเหลวซ้ำ แล้วส่งกลับอย่างมีขอบเขต

Reliability · หนึ่งทักษะ: กักข้อความที่ล้มเหลวซ้ำ และตัดสินใจส่งกลับด้วยขอบเขตที่ตรวจได้ ตัวอย่างและ Labs เป็น simulation; โค้ดและ tests ยัง NOT RUN

• คำตอบสั้น
• ประตูที่ต้องผ่านก่อน redrive
• ตัวอย่างตั้งแต่รับจนจบ
• อ่านโครงสร้างโค้ดจำลอง
• แบบฝึกหัดและ rubric
• คำถามที่พบบ่อย
• สำรวจแผนภาพและทดลอง lifecycle

คำตอบสั้น

เมื่อข้อความล้มเหลวครบสามครั้งในรอบจำลอง ให้หยุดส่งกลับคิวหลักและกักระเบียนเดิมไว้ตรวจสอบ กักกัน หรือ quarantine หมายถึงการแยกข้อความออกจากเส้นทางงานปกติเพื่อไม่ให้มันใช้ทรัพยากรและบังงานอื่น ส่วน poison message คือข้อความที่ล้มเหลวซ้ำเพราะเนื้อหา เงื่อนไข หรือความเข้ากันได้ ไม่ใช่แค่ความผิดพลาดชั่วคราว

ก่อน redrive ให้ตรวจสาเหตุ แก้ consumer ให้รองรับ schema ที่ข้อความใช้โดยไม่แก้ body ตรวจว่าไม่มีผลข้างเคียงภายนอกที่ยังไม่รู้ และผ่าน gate อนุมัติที่กำหนด การ redrive คือการนำข้อความที่กักไว้มาลองใหม่หลังมี remediation ที่ชัดเจน ตัวอย่างนี้จำกัดหนึ่ง redrive ต่อระเบียนและใช้ receive budget สามครั้งต่อรอบ

นี่คือ mock ในหน่วยความจำเพื่อสอนการตัดสินใจ ไม่เชื่อม broker ไม่บันทึกถาวร ไม่สร้างผลข้างเคียงทางธุรกิจ และไม่รับประกัน exactly-once หรือความสำเร็จของบริการจริง

เป้าหมายการเรียนรู้และขอบเขต

เมื่อจบบท ผู้อ่านควรออกแบบวงจรหนึ่งวงที่มีขีดจำกัดการรับ แยกสถานะ RETRY, QUARANTINED, HELD และ REDRIVEN ออกจากกัน ตรวจหลักฐานก่อนเปิดทางให้ลองใหม่ และอธิบายได้ว่าทำไมการหยุดอาจเป็นผลลัพธ์ที่ถูกต้องเมื่อข้อมูลยังไม่ครบ

ทักษะเดียวของบทนี้คือการออกแบบ quarantine และ controlled redrive แบบมีขอบเขต โดยใช้ poison message ที่เกิดจาก schema mismatch เป็นตัวอย่าง เรียนรู้การกำหนด receive budget, รักษาอัตลักษณ์ของงาน, ตรวจ body hash, รักษาอายุเดิม และปิดประตูเมื่อตรวจพบเงื่อนไขเสี่ยง

บทนี้ไม่ได้สอนการทำ queue ให้ทนทาน การจัดเก็บ receipt หรือการประสานผลข้าม transaction บท receipt ก่อน ACK กล่าวถึงการยืนยันงานและ conflict ส่วน Transactional Outbox กล่าวถึงการส่งผลออกจาก transaction ให้สอดคล้องกัน บทนี้รับช่วงปัญหาคนละระยะ: เมื่อข้อความถูกแยกแล้ว จะตัดสินใจ HOLD หรือ redrive อย่างไร

เหตุใด retry อย่างเดียวไม่พอ

การลองใหม่เหมาะกับความผิดพลาดชั่วคราว เช่น บริการปลายทางไม่ตอบชั่วครู่หรือการเชื่อมต่อหลุด แต่หากข้อมูลขาดฟิลด์หรือ consumer ไม่รู้จัก schema ใหม่ การเรียกซ้ำด้วยอินพุตเดิมจะได้ความล้มเหลวเดิม การตั้ง retry ไม่จำกัดจึงแปลงข้อบกพร่องเดียวให้เป็นการใช้ทรัพยากรต่อเนื่อง งานที่ถูกต้องในคิวเดียวกันอาจต้องรอ และ log อาจเต็มไปด้วยเหตุการณ์ซ้ำจนมองไม่เห็นต้นเหตุ

สมมติระบบรับคำขอตรวจวิดีโอที่ operation_id คือ video-review-17 ข้อความมี schema_version 2 แต่ consumer รุ่นที่รันอยู่รองรับ schema 1 เท่านั้น handler ปฏิเสธข้อความเพราะยังอ่านโครงสร้างใหม่ไม่ได้ ในตัวอย่างเรานับได้สาม receive ภายในหนึ่งรอบ จากนั้นย้ายระเบียนเดิมไป quarantine แทนการลองต่อเอง

DLQ หรือ dead-letter queue เป็นคิวแยกที่รับข้อความเมื่อเงื่อนไขการส่งของคิวหลักครบกำหนด เช่น จำนวน receive สูงสุด การแยกช่วยหยุดข้อความที่ทำให้เกิดปัญหาในเส้นทางปกติ แต่ DLQ ไม่วิเคราะห์สาเหตุ ไม่แก้ข้อมูล และไม่รับรองว่าผลข้างเคียงไม่เกิดขึ้นแล้ว การเห็นข้อความใน DLQ จึงเป็นจุดเริ่มตรวจ ไม่ใช่สัญญาณให้ replay ทันที

จำนวนสามในบทเป็น synthetic receive budget สำหรับตัวอย่าง ไม่ใช่การตั้งค่า maxReceiveCount ที่นำไปใช้กับ Amazon SQS ได้ทันที Broker จริงมี semantics ของการรับ การมองเห็นข้อความ การยืนยัน และ retention เฉพาะของตนเอง

คำศัพท์และสถานะของวงจร

READY หมายถึงระเบียนพร้อมเข้าสู่รอบรับ RETRY คือการลองภายในงบของรอบปัจจุบัน QUARANTINED คือครบงบแล้วและถูกแยกไว้ตรวจ HELD คือหยุดเพราะ gate ใด gate หนึ่งไม่ผ่าน REDRIVEN คือผ่าน gate และอนุญาตให้เริ่มรอบใหม่ และ SUCCEEDED ใน mock คือ handler fixture ยอมรับ schema ได้ ไม่ใช่หลักฐานว่างานธุรกิจจริงเสร็จสมบูรณ์

การแยกสถานะทำให้ผู้ปฏิบัติงานเห็นว่าระบบกำลังทำอะไรและเหตุใดจึงหยุด RETRY บอกว่ายังมีงบในรอบ ไม่ควรใช้แทนสถานะ “รอคนตรวจ” QUARANTINED ไม่ควรเริ่มวงจรใหม่เอง HELD คือการปิดประตูแบบชัดเจน และ REDRIVEN บอกว่าได้รับอนุญาตให้เริ่มใหม่แล้วแต่ยังไม่ยืนยันผลลัพธ์ทางธุรกิจ

ระบบจริงควรบันทึกการเปลี่ยนสถานะและผู้ตัดสินใจอย่างตรวจสอบย้อนหลังได้ แต่ mock นี้มีเพียง field ใน RAM ไม่มี durable event log, audit identity หรือการกู้คืนหลังโปรเซสจบ

ประตูที่ต้องผ่านก่อน redrive

การ redrive ต้องตรวจเงื่อนไขทั้งหมดก่อนเปลี่ยนสถานะ ระเบียนต้องอยู่ใน QUARANTINED ผู้อนุมัติใน fixture ต้องเป็นจริง remediation ต้องระบุว่าปรับแก้แล้วและ handler ต้องรองรับ schema ของ body ผลข้างเคียงภายนอกต้องไม่อยู่ในสถานะ unknown อายุข้อความต้องน้อยกว่า 12 synthetic ticks body hash ต้องตรงกับค่าตอนสร้าง และจำนวน redrive ต้องยังไม่เกินหนึ่งครั้ง

หากขาดข้อใดข้อหนึ่งให้ HOLD พร้อมเหตุผลที่เฉพาะเจาะจง เช่น approval_required, unfixed, unknown_effect, expired, body_changed หรือ replay_limit การตอบเหตุผลช่วยให้ผู้ดูแลรู้ว่าต้องหาอะไรต่อ แทนการลองใหม่เพื่อดูว่าจะโชคดีหรือไม่ ต้องระวังว่าการรายงานเหตุผลไม่เปิดเผยข้อมูลอ่อนไหวใน log ด้วย

ค่า boolean approved ในตัวอย่างเป็นเพียง fixture สำหรับสาธิต gate ไม่ใช่ระบบยืนยันตัวตนหรือสิทธิ์จริง ในระบบใช้งานต้องผูกการอนุมัติกับ principal ที่ตรวจสอบได้ กำหนดขอบเขตสิทธิ์ จำกัดชนิดงาน เก็บเวลาและเหตุผล และป้องกันไม่ให้ worker ทั่วไปปลอมค่าดังกล่าว

อัตลักษณ์ของธุรกิจกับตัวนับการส่ง

การส่งซ้ำเป็นความพยายามใหม่ในการประมวลผล operation ทางธุรกิจเดิม จึงรักษา operation_id ให้คงที่ตลอดวงจร ตัวอย่างคำนวณ SHA-256 จาก JSON body ที่ serialize ด้วยการเรียงชื่อ key และใช้ตัวคั่นคงที่ วิธีนี้ทำให้ตรวจได้ว่า body ที่กำลังจะ replay ตรงกับ body ที่กักไว้หรือไม่ trace ID ใช้เชื่อมเหตุการณ์ในตัวอย่าง แต่อย่าใช้แทน business identity

ตัวนับ receive ต่อรอบ จำนวน receive รวม replay counter และ transport message ID เป็นข้อมูลเกี่ยวกับความพยายามหรือการส่ง ไม่ควรนำมาแทน operation_id หากเริ่มรอบใหม่ ตัวนับรอบปัจจุบันกลับเป็นศูนย์ แต่ total_receives ยังคงบอกประวัติทั้งหมด การแยกตัวนับช่วยป้องกันการเข้าใจผิดว่าการเริ่มรอบใหม่ทำให้ประวัติเดิมหายไป

การ remediation ในโจทย์คือปรับ consumer fixture ให้รองรับ schema 2 โดยไม่แก้ body หากแก้ body กลางทาง อาจเปลี่ยนความหมายของคำขอและทำให้หลักฐานไม่ตรงกับ payload ที่ผู้ส่งสร้างไว้ หากพบข้อมูลต้นทางผิดจริงให้ HOLD และทำกระบวนการแก้ที่ตรวจสอบได้ ไม่แก้ข้อความอย่างเงียบ ๆ เพื่อให้ handler ผ่าน

hash ไม่ได้ยืนยันผู้ส่ง ไม่ตรวจความถูกต้องเชิงธุรกิจ และไม่สร้างตัวตนถาวร สถานะทั้งหมดอยู่ใน RAM หายเมื่อโปรเซสจบ ไม่มี persistence, transaction หรือการประสานงานข้าม worker จึงห้ามอ้างว่า mock ป้องกัน duplicate ในระบบจริง

ตัวอย่างตั้งแต่รับจนจบ

อินพุตที่ใช้คือ operation_id video-review-17, body ที่มี schema_version 2 และ video_id V-17, trace_id trace-demo-17 consumer เริ่มต้นรองรับ schema 1 receive cycle เรียก handler ได้สูงสุดสามครั้ง เนื่องจาก schema 2 ยังไม่เข้ากัน ทุกครั้งจึงไม่สำเร็จ รอบจบที่ QUARANTINED ด้วย receives_this_cycle เท่ากับ 3 และ total_receives เท่ากับ 3

ผู้ตรวจทบทวนเหตุการณ์และยืนยันใน fixture ว่าไม่มีผลข้างเคียงภายนอกค้างอยู่ จากนั้นปรับ supported schemas ของ consumer เป็น {1,2} โดยไม่เปลี่ยนข้อความ body คง operation ID และ hash เดิม ตั้ง approval และ remediation fixtures เป็นจริง แล้วเรียก redrive หนึ่งครั้ง สถานะเปลี่ยนเป็น REDRIVEN ตัวนับรอบเริ่มใหม่แต่ total_receives ยังเป็นสาม อายุยังนับจากเวลา creation เดิม

เมื่อ consume หลัง redrive handler fixture รองรับ schema 2 จึงได้ SUCCEEDED และ total_receives เป็นสี่ นี่คือผลที่โค้ดกำหนดตามเงื่อนไข ไม่ใช่ output ที่เราสังเกตจากการรัน เพราะไม่รันตัวอย่างหรือ tests

หาก approval เป็นเท็จจะได้ HELD:approval_required หาก consumer ยังไม่รองรับหรือ remediation ไม่พร้อมจะได้ HELD:unfixed ถ้ามี unknown effect จะได้ HELD:unknown_effect ถ้าเวลาถึง 12 ticks จะได้ HELD:expired ถ้า body เปลี่ยนจะได้ HELD:body_changed และเมื่อใช้ replay ไปแล้วจะได้ HELD:replay_limit ไม่มีการเพิ่มงบอัตโนมัติ

อ่านโครงสร้างโค้ดจำลอง

Record เก็บ operation ID, body, trace ID, creation tick, state, counters, unknown-effect flag และ original hash Demo เก็บเวลาจำลอง schema ที่ handler รองรับ record เดียว และรายการ quarantine การใช้ record เดียวทำให้บทเรียนอ่านง่าย แต่ไม่ได้จำลองการเลือกข้อความหลายรายการหรือการทำงานพร้อมกัน

receive_cycle ยอมเริ่มจาก READY เท่านั้น รอบหลัง redrive ต้องผ่าน consume เท่านั้น รีเซ็ตตัวนับรอบปัจจุบัน แต่ไม่ล้าง total จากนั้นวนไม่เกินสามครั้ง เพิ่มทั้งตัวนับรอบและจำนวนรวม หาก schema ตรงก็จบที่ SUCCEEDED หากไม่ตรงหลังครบงบ จึงเปลี่ยน record เดิมเป็น QUARANTINED และเก็บไว้ในรายการกัก

redrive รับ approval fixture, remediation fixture และ optional body สำหรับแสดงการตรวจ hash โดยไม่ได้แก้ body ภายใน ตรวจสถานะก่อน จากนั้นตรวจ approval, remediation และ compatibility, unknown effect, expiry, hash และ replay budget ตามลำดับ หากผิดข้อใดเปลี่ยนเป็น HELD และคืนสาเหตุ หากผ่านจึงเพิ่ม redrive counter ตั้ง receives_this_cycle เป็นศูนย์ และเปลี่ยนเป็น REDRIVEN

consume ยอมทำงานเฉพาะ REDRIVEN และตรวจ ID, body hash, อายุ, unknown effect และ schema ซ้ำก่อนเพิ่มตัวนับหรือทำเครื่องหมาย SUCCEEDED หากไม่ผ่านจะคืน HELD พร้อมเหตุผล receive_cycle ไม่รับสถานะ REDRIVEN จึงใช้ข้ามด่านนี้ไม่ได้ โค้ดนี้ไม่ได้เรียก handler ภายนอกจริง Synthetic clock เป็นเลขที่ผู้เรียกปรับเอง ไม่มีเวลาเป็นวินาทีหรืออัตราการไหล

"""In-memory teaching mock. No broker, IO, or external effects. NOT RUN."""
from dataclasses import dataclass, field
from hashlib import sha256
import json

MAX_RECEIVES, MAX_REDRIVES, TTL_TICKS = 3, 1, 12

def body_hash(body):
    raw = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return sha256(raw.encode("utf-8")).hexdigest()

@dataclass
class Record:
    operation_id: str
    body: dict
    trace_id: str
    created_at: int
    state: str = "READY"
    receives_this_cycle: int = 0
    total_receives: int = 0
    redrives: int = 0
    unknown_effect: bool = False
    original_operation_id: str = field(init=False)
    original_hash: str = field(init=False)

    def __post_init__(self):
        self.original_operation_id = self.operation_id
        self.original_hash = body_hash(self.body)

class Demo:
    def __init__(self):
        self.now = 0
        self.supported_schemas = {1}
        self.record = Record("video-review-17",
            {"schema_version": 2, "video_id": "V-17"},
            "trace-demo-17", self.now)
        self.quarantine = []

    def receive_cycle(self):
        r = self.record
        if r.state != "READY":
            return r.state
        if r.operation_id != r.original_operation_id or body_hash(r.body) != r.original_hash:
            r.state = "HELD"
            return r.state
        r.receives_this_cycle = 0
        for _ in range(MAX_RECEIVES):
            r.receives_this_cycle += 1
            r.total_receives += 1
            if r.body["schema_version"] in self.supported_schemas:
                r.state = "SUCCEEDED"
                return r.state
            r.state = "RETRY"
        r.state = "QUARANTINED"
        self.quarantine.append(r)
        return r.state

    def redrive(self, *, approved, remediation_fixed, body=None):
        r = self.record
        candidate = r.body if body is None else body
        reason = None
        if r.state != "QUARANTINED": reason = "not_quarantined"
        elif not approved: reason = "approval_required"
        elif not remediation_fixed or candidate.get("schema_version") not in self.supported_schemas:
            reason = "unfixed"
        elif r.unknown_effect: reason = "unknown_effect"
        elif self.now - r.created_at >= TTL_TICKS: reason = "expired"
        elif r.operation_id != r.original_operation_id: reason = "operation_id_changed"
        elif body_hash(r.body) != r.original_hash or body_hash(candidate) != r.original_hash:
            reason = "body_changed"
        elif r.redrives >= MAX_REDRIVES: reason = "replay_limit"
        if reason:
            r.state = "HELD"
            return "HELD:" + reason
        r.redrives += 1
        r.receives_this_cycle = 0
        r.state = "REDRIVEN"
        return r.state

    def _hold(self, reason):
        self.record.state = "HELD"
        return "HELD:" + reason

    def consume(self):
        r = self.record
        if r.state != "REDRIVEN": return r.state
        if r.unknown_effect:
            return self._hold("unknown_effect")
        if self.now - r.created_at >= TTL_TICKS:
            return self._hold("expired")
        if r.operation_id != r.original_operation_id:
            return self._hold("operation_id_changed")
        if body_hash(r.body) != r.original_hash:
            return self._hold("body_changed")
        if r.body["schema_version"] not in self.supported_schemas:
            return self._hold("unfixed")
        r.receives_this_cycle += 1
        r.total_receives += 1
        r.state = ("SUCCEEDED" if
                   r.body["schema_version"] in self.supported_schemas else "HELD")
        return r.state

if __name__ == "__main__":
    d = Demo()
    print(d.receive_cycle(), d.record.receives_this_cycle, d.record.total_receives)
    d.supported_schemas = {1, 2}  # consumer remediation; body unchanged
    print(d.redrive(approved=True, remediation_fixed=True))
    print(d.consume(), d.record.operation_id, d.record.original_hash)

Runtime, dependency, สิทธิ์ และคำสั่ง

กำหนดให้ใช้ Python 3.12 ขึ้นไปและ standard library เท่านั้น ได้แก่ dataclasses, hashlib, json และ unittest ไม่ต้องติดตั้ง dependency เพิ่ม ไม่ต้องตั้ง environment variable ตัวอย่างไม่เรียก network/database/broker/API/LLM หรือเขียนไฟล์ข้อมูล ผู้เรียนต้องมีสิทธิ์อ่าน source และ standard library ในเครื่องเมื่อตัดสินใจทดลองเอง จึงไม่มีค่าบริการ API ตัวอย่างไม่ต้องใช้สิทธิ์เครือข่าย เขียนไฟล์ หรือเข้าถึงข้อมูลผู้ใช้

หากผู้เรียนต้องการทดลอง ให้ตรวจ source และทำสำเนาใน sandbox ส่วนตัวก่อน คำสั่งที่แนะนำคือ python -B mock_dlq.py เพื่ออ่าน transition และ python -B -m unittest -v test_mock_dlq.py เพื่อเรียกกรณีทดสอบ นี่เป็นคำแนะนำ ไม่ใช่การกระทำที่เกิดขึ้นในงานนี้ ห้ามชี้การทดลองไปยังคิวหรือระบบจริง

Input: operation_id video-review-17, body schema_version 2 และ video_id V-17, trace_id trace-demo-17, consumer เริ่มรองรับ {1}, receive limit 3 ต่อ cycle, replay limit 1, TTL 12 synthetic ticks การแก้ไขรองรับ schema 2 ใน consumer โดยคง body เดิม

Expected output โดยอนุมาน: receive cycle คืน QUARANTINED พร้อมตัวนับ 3/3; หลังเปลี่ยน fixture และผ่าน gate redrive คืน REDRIVEN; consume คืน SUCCEEDED พร้อม operation ID และ body hash เดิม ค่าเหล่านี้เป็นผลตามตรรกะที่เขียน ไม่ใช่ผลที่สังเกตจาก runtime

เมื่อเลือกทดลองเอง ให้บันทึก mock_dlq.py และ test_mock_dlq.py จากโค้ดด้านบนไว้ในโฟลเดอร์สำเนาเดียวกัน เปิด terminal ในโฟลเดอร์นั้นและใช้ Python 3.12 ขึ้นไป ตัวเลือก -B ปิดการเขียน bytecode cache การอ่าน source/standard library ยังจำเป็น

กรณีทดสอบและสถานะ

ชุด unittest ที่เขียนไว้ครอบคลุมการกักเมื่อครบสาม receive, ความสำเร็จหลัง remediation ที่รองรับ schema 2, การคง operation ID และ hash, total count ที่ไม่หายเมื่อเริ่มรอบใหม่, การปฏิเสธ approval และ remediation ที่ขาด, handler ที่ไม่รองรับ, unknown effect, expiry, body ที่เปลี่ยน, replay budget ที่หมด และการพยายาม redrive record ที่ยังไม่ถูกกัก

แต่ละกรณี gate ที่ไม่ผ่านควรจบ HELD พร้อมเหตุผล ไม่เพิ่ม replay counter และไม่เรียก consumer เพิ่ม นี่เป็นชุดกรณีทดสอบที่ผู้เขียนกำหนด ยังไม่มีการรันโค้ดหรือชุดทดสอบ สถานะจึงเป็น NOT RUN ไม่ใช่ PASS

การตรวจ UI ของ interactive และ 3D Lab เป็น UNKNOWN จนกว่าจะมีผู้ใช้เปิดและทดลองจริง การมีโค้ดตัวอย่างหรือเขียน test case ไม่ได้ทำให้การทดสอบ runtime, integration หรือ UI เกิดขึ้นเอง

มีกรณีที่เขียนเพิ่มสำหรับการแก้ ID/body/schema หรือผลที่ไม่ทราบหลัง redrive, เวลา 11 เทียบ 12 ticks และการพยายามเรียก receive_cycle ข้าม final gate ทุกกรณียังเป็น NOT RUN

"""Authored tests only. NOT RUN."""
import unittest
from mock_dlq import Demo, body_hash

class QuarantineRedriveTests(unittest.TestCase):
    def quarantined(self):
        d = Demo()
        self.assertEqual(d.receive_cycle(), "QUARANTINED")
        return d

    def test_three_receives_quarantine(self):
        d = Demo()
        self.assertEqual(d.receive_cycle(), "QUARANTINED")
        self.assertEqual((d.record.receives_this_cycle, d.record.total_receives), (3, 3))

    def test_compatible_remediation_preserves_identity_and_history(self):
        d = self.quarantined()
        original = (d.record.operation_id, d.record.original_hash)
        d.supported_schemas = {1, 2}
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "REDRIVEN")
        self.assertEqual((d.record.operation_id, d.record.original_hash), original)
        self.assertEqual(d.record.total_receives, 3)
        self.assertEqual(d.consume(), "SUCCEEDED")
        self.assertEqual(d.record.total_receives, 4)

    def test_denies_approval_or_remediation(self):
        for kw, expected in [
            ({"approved": False, "remediation_fixed": True}, "HELD:approval_required"),
            ({"approved": True, "remediation_fixed": False}, "HELD:unfixed")]:
            d = self.quarantined()
            d.supported_schemas = {1, 2}
            self.assertEqual(d.redrive(**kw), expected)

    def test_denies_incompatible_handler(self):
        d = self.quarantined()
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "HELD:unfixed")

    def test_denies_unknown_effect_and_expiry(self):
        d = self.quarantined()
        d.supported_schemas = {1, 2}
        d.record.unknown_effect = True
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "HELD:unknown_effect")
        d = self.quarantined()
        d.supported_schemas = {1, 2}
        d.now = 12
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "HELD:expired")

    def test_expiry_boundary_before_and_at_ttl(self):
        before = self.quarantined()
        before.supported_schemas = {1, 2}
        before.now = 11
        self.assertEqual(before.redrive(approved=True, remediation_fixed=True), "REDRIVEN")
        at = self.quarantined()
        at.supported_schemas = {1, 2}
        at.now = 12
        self.assertEqual(at.redrive(approved=True, remediation_fixed=True), "HELD:expired")

    def test_denies_changed_body(self):
        d = self.quarantined()
        d.supported_schemas = {1, 2}
        changed = {**d.record.body, "video_id": "OTHER"}
        self.assertNotEqual(body_hash(changed), d.record.original_hash)
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True, body=changed),
                         "HELD:body_changed")
        self.assertEqual(d.record.body["video_id"], "V-17")

    def test_denies_replay_limit(self):
        d = self.quarantined()
        d.supported_schemas = {1, 2}
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "REDRIVEN")
        self.assertEqual(d.consume(), "SUCCEEDED")
        d.record.state = "QUARANTINED"  # fixture reaches exhausted-budget gate
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "HELD:replay_limit")

    def test_old_candidate_cannot_mask_mutated_record_body(self):
        d = self.quarantined()
        d.supported_schemas = {1, 2}
        original = dict(d.record.body)
        d.record.body["video_id"] = "MUTATED"
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True, body=original),
                         "HELD:body_changed")

    def test_consume_rechecks_mutations_after_redrive(self):
        for mutation in ("body", "operation_id", "schema", "unknown", "expiry"):
            d = self.quarantined()
            d.supported_schemas = {1, 2}
            self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "REDRIVEN")
            if mutation == "body":
                d.record.body["video_id"] = "MUTATED"
            elif mutation == "operation_id":
                d.record.operation_id = "other-operation"
            elif mutation == "schema":
                d.supported_schemas = {1}
            elif mutation == "unknown":
                d.record.unknown_effect = True
            else:
                d.now = 12
            reason = {"body": "body_changed", "operation_id": "operation_id_changed",
                      "schema": "unfixed", "unknown": "unknown_effect", "expiry": "expired"}[mutation]
            self.assertEqual(d.consume(), "HELD:" + reason, mutation)
            self.assertEqual(d.record.total_receives, 3, mutation)

    def test_redrive_rejects_changed_operation_id(self):
        d = self.quarantined()
        d.supported_schemas = {1, 2}
        d.record.operation_id = "other-operation"
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True),
                         "HELD:operation_id_changed")

    def test_receive_cycle_cannot_skip_redrive_final_gate(self):
        d = self.quarantined()
        d.supported_schemas = {1, 2}
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "REDRIVEN")
        d.now = 12
        self.assertEqual(d.receive_cycle(), "REDRIVEN")
        self.assertEqual(d.record.total_receives, 3)
        self.assertEqual(d.consume(), "HELD:expired")

    def test_non_quarantined_cannot_redrive(self):
        d = Demo()
        self.assertEqual(d.redrive(approved=True, remediation_fixed=True), "HELD:not_quarantined")

if __name__ == "__main__":
    unittest.main()

เหตุใด HOLD จึงเป็นผลที่ถูกต้อง

หากคำขอไปบริการภายนอก timeout หลังส่ง request แต่ก่อนรับ response ระบบอาจไม่รู้ว่าผลข้างเคียงเกิดแล้วหรือยัง การส่งซ้ำจากการเดาว่าไม่เกิดอาจสร้างรายการซ้ำ การหยุดเป็น HELD:unknown_effect เปิดให้ตรวจด้วย operation ID หรือหลักฐานปลายทางก่อน อย่าทำให้ความไม่รู้กลายเป็นคำยืนยันว่าล้มเหลว เพียงเพื่อให้ระบบเดินต่อ

HOLD ยังเหมาะเมื่อข้อความหมดอายุเพราะผลอาจไม่ควรเกิดแล้ว เมื่อ body hash เปลี่ยนเพราะไม่ใช่ข้อมูลเดิม เมื่อ remediation ยังไม่เข้ากัน เมื่อ budget หมด หรือไม่มีหลักฐานการอนุมัติ การเพิ่มจำนวนครั้งเพื่อให้ระบบ “ผ่าน” ไม่แก้ต้นเหตุและอาจขยายผลกระทบ

บันทึกเหตุผลและหลักฐานที่ต้องใช้ต่อ เช่นผลตรวจปลายทาง consumer version ที่แก้แล้ว body hash operation ID และเวลา หากยังไม่รู้ว่าผลข้างเคียงเกิดหรือไม่ ให้คง HOLD ระหว่าง reconcile การหยุดที่อธิบายได้เป็นส่วนหนึ่งของ recovery ที่ปลอดภัย

สิ่งที่ต้องตรวจเมื่อเลือก broker จริง

เอกสาร Amazon SQS อธิบาย DLQ และ maxReceiveCount รวมถึง retention และ FIFO caveat ซึ่งเป็นพฤติกรรมเฉพาะ SQS เอกสาร redrive อธิบายการควบคุมอัตราย้ายและข้อจำกัดของขั้นย้าย เช่น ไม่ใช้ redrive เพื่อกรองหรือแก้เนื้อหาข้อความ และมี permission ที่เกี่ยวข้อง ต้องอ่านคู่มือของบริการที่จะใช้จริงก่อนตั้งค่า ไม่ยกค่าจำลองจากบทไปใช้โดยตรง

เอกสาร RabbitMQ อธิบาย dead-letter ว่าเป็นการ republish ซึ่งอาจล้มเหลวและมีความเสี่ยงจาก cycle รายละเอียดขึ้นกับ exchange, queue และ policy ที่ตั้งไว้ ส่วน SQS StartMessageMoveTask เริ่ม asynchronous move task การที่เริ่ม task สำเร็จไม่ได้บอกว่าการดำเนินการทางธุรกิจสำเร็จแล้ว งานนี้ไม่เรียก API จริงและไม่วัด throughput หรืออัตรา clock

ระเบียนจริงที่ตรวจสอบได้ควรตอบว่า operation คืออะไร ข้อความมาจากไหน hash เดิมคืออะไร handler รุ่นใดล้มเหลว รับกี่ครั้งและเมื่อใด เหตุใดจึงจัดเป็น poison ผลข้างเคียงมีสถานะใด ใครอนุมัติ ใช้ remediation รุ่นไหน และ replay budget เหลือเท่าใด ข้อมูลนี้เป็นแนวทางออกแบบ ไม่ได้ถูกเก็บถาวรโดย mock

สิทธิ์ ความเสี่ยง และต้นทุน

mock ไม่ใช้ permission ภายนอกและไม่มีค่าใช้จ่ายบริการเพราะทำงานใน RAM ต้นทุนจริงขึ้นกับปริมาณข้อความ จำนวน receive ระยะ retention ปริมาณ log อัตราการย้าย และค่าใช้จ่ายของระบบปลายทาง บทนี้ไม่มีข้อมูล workload หรือราคาสำหรับประเมินจึงไม่ใส่ตัวเลข และการจำกัดหนึ่ง record ต่อการเรียกเป็นลักษณะสาธิต ไม่ใช่ throughput target

ในระบบจริงควรแยกสิทธิ์ดู DLQ ออกจากสิทธิ์ redrive จำกัดขอบเขตคิว ตรวจตัวตนผู้อนุมัติ บันทึกเหตุผลและเวลาตัดสินใจ ปกป้อง payload ที่อาจมีข้อมูลส่วนบุคคล และตั้ง retention ตามนโยบายองค์กร อย่าให้ worker ทุกตัวมีสิทธิ์ลบหรือส่งซ้ำโดยไม่มี gate

body hash ไม่ยืนยันผู้ส่ง ไม่รับรองความถูกต้องเชิงธุรกิจ boolean approval ไม่ใช่ access control mock ไม่มี persistence, broker ACK, atomic writes, multi-worker coordination, restart recovery, audit identity, exactly-once guarantee หรือผลกระทบ production ห้ามเรียกว่า production-ready

Copyable task brief หกส่วน

1. Goal: หยุด retry ไม่จำกัดและกัก schema-mismatch poison message
2. Tools and data: Python standard library, in-memory fixture; stable operation_id, schema-2 body, trace ID, receive 3, replay 1, TTL 12 synthetic ticks
3. Scope: remediation ปรับ consumer compatibility; ไม่มี broker, network, file/database writes, API, secrets หรือผลข้างเคียงจริง
4. Deliverable: state machine, ID/hash snapshot, counters, quarantine, gated redrive, HOLD reasons และ authored tests
5. Acceptance: quarantine เมื่อครบสาม; คง ID/hash/age/total; redrive เดียวเมื่อทุก gate ผ่าน; ตรวจ integrity/expiry/effect/schema ซ้ำก่อน consume
6. Stop: HOLD เมื่อ unknown effect, expiry, body/ID change, incompatible schema, missing approval/remediation หรือ budget หมด

เกณฑ์ยอมรับและเงื่อนไขหยุด

• ครบสาม receive ในหนึ่งรอบแล้วสถานะเป็น QUARANTINED และไม่มี retry เพิ่มโดยอัตโนมัติ
• operation ID, body hash, creation age และ total receive คงเดิมผ่าน quarantine และ redrive
• ผ่าน gate ทุกข้อจึง redrive ได้ และไม่เกินหนึ่งครั้งต่อ record
• ทุกเหตุปฏิเสธคืน HELD พร้อมเหตุผลที่ชัด และไม่เรียก consumer เพิ่ม
• อธิบายชัดว่า mock ไม่ durable, ไม่ atomic, ไม่เชื่อม broker และไม่รับประกัน exactly-once

หยุด เมื่อ hash เปลี่ยน, ผลข้างเคียงยังไม่ทราบ, หมดอายุ, replay budget หมด หรือไม่มีหลักฐานว่า remediation รองรับข้อความ อย่าเพิ่ม budget เพื่อให้ผ่านและอย่าแก้ body เพื่อหลบ schema mismatch

แบบฝึกหัดและ rubric

ปรับกรณีเป็น schema 3 แต่ consumer หลัง remediation รองรับ schema 2 เท่านั้น คาดการณ์ state และเหตุผล จากนั้นเปรียบเทียบ clock หนึ่ง tick ก่อนหมดอายุกับเวลาที่ถึง TTL พอดี พร้อมอธิบายว่าเหตุใดอายุต้องไม่เริ่มใหม่เมื่อ redrive สุดท้ายเพิ่ม unknown external effect แล้วเขียนบันทึก HOLD ที่ระบุหลักฐานจากระบบปลายทางซึ่งต้องตรวจต่อ

ตรวจด้วย acceptance ที่สังเกตได้: schema 3 ที่ handler ไม่รองรับต้องถูก HOLD; tick 11 อนุญาต redrive เมื่อ gate อื่นผ่าน แต่ tick 12 ต้องหยุด; unknown effect ต้องไม่ consume; การเปลี่ยน ID หรือ body หลัง redrive ต้องเป็น HELD ก่อนนับ receive เพิ่ม แบบฝึกหัดจำกัดใน mock ห้ามชี้ไปยังคิวจริง

คำถามที่พบบ่อย

DLQ แก้ poison message เองหรือไม่?
ไม่ DLQ แยกข้อความตามนโยบาย ผู้ปฏิบัติงานยังต้องตรวจสาเหตุและเลือก HOLD หรือ redrive

ทำไมไม่แก้ body ให้ handler อ่านได้?
เพราะอาจเปลี่ยนเจตนาคำขอ ให้แก้ consumer และคง hash เดิม

redrive แปลว่าสำเร็จไหม?
ไม่ หมายถึงผ่าน gate และเริ่มลองใหม่ ผลธุรกิจต้องตรวจแยก

เมื่อใดควร HOLD?
เมื่อ approval หรือ remediation ไม่ครบ ผลภายนอกไม่ทราบ หมดอายุ hash เปลี่ยน หรือ budget หมด

เลขสามเป็นค่า SQS ไหม?
ไม่ เป็นงบสังเคราะห์ semantics จริงขึ้นกับ broker และ configuration

mock ยืนยัน exactly-once หรือ durability ไหม?
ไม่ สถานะอยู่ใน RAM ไม่มี broker หรือ atomic storage

approval boolean คือ authorization จริงไหม?
ไม่ เป็น fixture เท่านั้น

โค้ดและ tests รันหรือยัง?
ยัง NOT RUN; expected output อนุมานจาก logic และ UI QA เป็น UNKNOWN

แหล่งอ้างอิงและสถานะ

• Amazon SQS dead-letter queues — isolation, maxReceiveCount, retention และ FIFO caveat
• Configuring dead-letter queue redrive — controlled redrive และข้อจำกัดเฉพาะ SQS
• RabbitMQ Dead Letter Exchanges — republish, failure และ cycle เฉพาะ RabbitMQ
• StartMessageMoveTask API — การเริ่ม asynchronous move task

สรุปเอกสารจำกัดอยู่ใน claims ที่ research brief ระบุ ส่วน mock เป็น engineering analysis เพื่อการศึกษา ไม่ใช่ benchmark หรือ implementation ของ vendor Code execution: NOT RUN. Authored tests: NOT RUN. Owner UI QA: UNKNOWN.