Execution labs: kiểm chứng ở ranh giới lỗi

CREATE UNIQUE INDEX one_business_effect
  ON business_effect (consumer_name, business_key);

CREATE OR REPLACE FUNCTION apply_charge(
  p_consumer text, p_event uuid, p_run text,
  p_key text, p_amount bigint, p_hash text
) RETURNS text LANGUAGE plpgsql AS $$
DECLARE
  v_claim uuid;
  v_effect bigint;
  v_hash text;
  v_amount bigint;
BEGIN
  INSERT INTO inbox(consumer_name, event_id, payload_hash)
  VALUES (p_consumer, p_event, p_hash)
  ON CONFLICT (consumer_name, event_id) DO NOTHING
  RETURNING event_id INTO v_claim;

  IF v_claim IS NULL THEN
    SELECT payload_hash INTO v_hash FROM inbox
      WHERE consumer_name=p_consumer AND event_id=p_event;
    IF v_hash IS DISTINCT FROM p_hash THEN
      RAISE EXCEPTION 'EVENT_ID_PAYLOAD_CONFLICT';
    END IF;
    RETURN 'DUPLICATE_EVENT';
  END IF;

  INSERT INTO business_effect
    (consumer_name,event_id,origin_run_id,business_key,amount,payload_hash)
  VALUES (p_consumer,p_event,p_run,p_key,p_amount,p_hash)
  ON CONFLICT (consumer_name,business_key) DO NOTHING
  RETURNING effect_id INTO v_effect;

  IF v_effect IS NULL THEN
    SELECT payload_hash, amount INTO v_hash, v_amount
      FROM business_effect
      WHERE consumer_name=p_consumer AND business_key=p_key;
    IF v_hash IS DISTINCT FROM p_hash OR v_amount IS DISTINCT FROM p_amount THEN
      RAISE EXCEPTION 'BUSINESS_KEY_PAYLOAD_CONFLICT';
    END IF;
    RETURN 'DUPLICATE_BUSINESS_OPERATION';
  END IF;
  RETURN 'APPLIED';
END $$;

// Integration sketch: use a separate transactional service/proxy.
// decodeAndValidate and audit are application code, not supplied APIs.
void onDelivery(Delivery d) {
    Event e = decodeAndValidate(d);
    audit("RECEIVED", e, d);
    service.applyInDatabase(e);      // Returns AFTER the proxy commits.
    audit("DB_COMMITTED", e, d);
    Failpoint.hit("AFTER_DB_COMMIT", e.id().toString());

    if (d.isRabbit()) {
        channel.basicAck(d.deliveryTag(), false); // Same receiving channel.
        audit("ACK_SENT", e, d);     // Does NOT mean broker-confirmed ack.
    } else {
        consumer.commitSync(Map.of(
            new TopicPartition(d.topic(), d.partition()),
            new OffsetAndMetadata(d.offset() + 1)));
        audit("OFFSET_COMMIT_CONFIRMED", e, d);
    }
    Failpoint.hit("AFTER_ACK_CALL", e.id().toString());
}

@Transactional
public void applyInDatabase(Event e) {
    // SELECT apply_charge(...); uses THIS transaction/connection.
    repository.applyCharge(e);
    Failpoint.hit("BEFORE_DB_COMMIT", e.id().toString());
}

#!/usr/bin/env python3
"""Finite, reviewed DLQ manifest replay for a LOCAL LAB.
Default is dry-run. Transport confirmation is NOT business success.
Requires pika==1.3.2 and confluent-kafka==2.10.0 only for --execute.
"""
import argparse
import base64
import hashlib
import json
import os
import sys
import signal
from contextlib import contextmanager
import time
from datetime import datetime, timezone
from pathlib import Path


@contextmanager
def deadline(seconds):
    # Main-thread POSIX watchdog; native Windows must use WSL for this lab.
    def expired(_signum, _frame):
        raise TimeoutError("Broker operation deadline exceeded")
    previous = signal.signal(signal.SIGALRM, expired)
    signal.setitimer(signal.ITIMER_REAL, seconds)
    try:
        yield
    finally:
        signal.setitimer(signal.ITIMER_REAL, 0)
        signal.signal(signal.SIGALRM, previous)

def utc(value):
    dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if dt.tzinfo is None:
        raise ValueError("Timestamp must contain a timezone")
    return dt.astimezone(timezone.utc)

def audit_write(f, record):
    record = {"at": datetime.now(timezone.utc).isoformat(), **record}
    f.write(json.dumps(record, ensure_ascii=False) + "\n")
    f.flush()
    os.fsync(f.fileno())

def parse_args():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--manifest", required=True)
    p.add_argument("--broker", choices=("rabbit", "kafka"), required=True)
    p.add_argument("--reason", required=True)
    p.add_argument("--schema", type=int, required=True)
    p.add_argument("--from-utc", required=True)
    p.add_argument("--to-utc", required=True)
    p.add_argument("--run-id", required=True)
    p.add_argument("--owner", required=True)
    p.add_argument("--approval")
    p.add_argument("--limit", type=int, default=100)
    p.add_argument("--rate", type=float, default=1.0)
    p.add_argument("--audit", required=True)
    p.add_argument("--execute", action="store_true")
    return p.parse_args()

def main():
    a = parse_args()
    if a.limit < 1 or not 0 < a.rate <= 1000:
        raise ValueError("Require limit >= 1 and 0 < rate <= 1000")
    if a.execute and not a.approval:
        raise ValueError("--execute requires --approval")
    start, end = utc(a.from_utc), utc(a.to_utc)
    if start >= end:
        raise ValueError("Require from-utc < to-utc; end is exclusive")
    manifest_bytes = Path(a.manifest).read_bytes()
    manifest_hash = hashlib.sha256(manifest_bytes).hexdigest()
    items = [json.loads(line) for line in manifest_bytes.splitlines() if line.strip()]
    candidates = []
    seen_dlq = set()
    for item in items:
        if item["dlq_id"] in seen_dlq:
            raise ValueError("Duplicate dlq_id in the frozen manifest")
        seen_dlq.add(item["dlq_id"])
        ts = utc(item["first_failed_at"])
        if not (item.get("replay_approved") is True
                and item["reason"] == a.reason
                and item["schema_version"] == a.schema
                and start <= ts < end):
            continue
        if not item.get("original_id") or not item.get("original_key"):
            raise ValueError("Approved replay requires original ID and key")
        if not isinstance(item["attempts"], int) or item["attempts"] < 0:
            raise ValueError("Invalid original attempt count")
        payload = base64.b64decode(item["payload_b64"], validate=True)
        if hashlib.sha256(payload).hexdigest() != item["payload_sha256"]:
            raise ValueError("Raw payload hash mismatch")
        candidates.append((item, payload))
    candidates.sort(key=lambda pair: (
        utc(pair[0]["first_failed_at"]), pair[0]["dlq_id"]))
    selected = candidates[:a.limit]

    # Exclusive create prevents accidental reuse/overwrite of an audit run.
    with open(a.audit, "x", encoding="utf-8") as log:
        audit_write(log, {
            "stage": "PLAN", "run_id": a.run_id, "owner": a.owner,
            "approval": a.approval, "execute": a.execute,
            "manifest_sha256": manifest_hash, "broker": a.broker,
            "reason": a.reason, "schema": a.schema,
            "from_utc": a.from_utc, "to_utc": a.to_utc,
            "rate_per_second": a.rate, "limit": a.limit,
            "selected_count": len(selected)})
        if not a.execute:
            for item, _ in selected:
                audit_write(log, {
                    "stage": "WOULD_PUBLISH", "dlq_id": item["dlq_id"],
                    "event_id": item["original_id"], "key": item["original_key"]})
            print(json.dumps({"mode": "dry-run", "selected": len(selected),
                              "manifest_sha256": manifest_hash}))
            return 0
        if not selected:
            return 0

        connection = producer = channel = None
        try:
            if a.broker == "rabbit":
                import pika
                params = pika.URLParameters(os.environ["RMQ_URL"])
                params.socket_timeout = 10
                params.blocked_connection_timeout = 10
                params.heartbeat = 30
                connection = pika.BlockingConnection(params)
                channel = connection.channel()
                channel.confirm_delivery()
            else:
                from confluent_kafka import Producer
                producer = Producer({
                    "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"],
                    "client.id": "replay-" + a.run_id,
                    "enable.idempotence": True,
                    "acks": "all",
                    "delivery.timeout.ms": 10000})
            next_due = time.monotonic()
            for item, payload in selected:
                delay = next_due - time.monotonic()
                if delay > 0:
                    if connection is not None:
                        connection.sleep(delay)  # Keep Rabbit heartbeats alive.
                    else:
                        time.sleep(delay)
                key = item["original_key"]
                metadata = {
                    "x-original-id": item["original_id"],
                    "x-origin-run-id": item["origin_run_id"],
                    "x-replay-run-id": a.run_id,
                    "x-dlq-id": item["dlq_id"],
                    "x-attempts": str(item["attempts"]),
                    "x-schema-version": str(item["schema_version"]),
                    "x-original-key": key}
                audit_write(log, {
                    "stage": "PUBLISH_PLANNED", "run_id": a.run_id,
                    "event_id": item["original_id"], "dlq_id": item["dlq_id"],
                    "payload_sha256": item["payload_sha256"]})
                broker_position = {}
                try:
                    if channel is not None:
                        import pika
                        with deadline(15):
                            channel.basic_publish(
                                exchange="lab.b.events", routing_key="work",
                                body=payload, mandatory=True,
                                properties=pika.BasicProperties(
                                    message_id=item["original_id"],
                                    content_type="application/json",
                                    delivery_mode=2, headers=metadata))
                        broker_position = {"exchange": "lab.b.events",
                                           "routing_key": "work"}
                    else:
                        outcomes = []
                        def delivered(error, message):
                            outcomes.append((error, message))
                        producer.produce(
                            "lab.b.events", key=key.encode(), value=payload,
                            headers=[(k, v.encode()) for k, v in metadata.items()],
                            on_delivery=delivered)
                        remaining = producer.flush(15)
                        if remaining or not outcomes or outcomes[0][0] is not None:
                            raise RuntimeError("Kafka delivery not confirmed")
                        m = outcomes[0][1]
                        broker_position = {"topic": m.topic(),
                                           "partition": m.partition(),
                                           "offset": m.offset()}
                    audit_write(log, {
                        "stage": "TRANSPORT_CONFIRMED", "run_id": a.run_id,
                        "event_id": item["original_id"],
                        "dlq_id": item["dlq_id"], **broker_position})
                except Exception as exc:
                    # Conservatively stop. Some errors are definitive, others
                    # mean the publish succeeded but its response was lost.
                    audit_write(log, {
                        "stage": "UNKNOWN_OR_FAILED_STOP", "run_id": a.run_id,
                        "event_id": item["original_id"], "dlq_id": item["dlq_id"],
                        "error_type": type(exc).__name__})
                    raise
                # Sequential confirms + spacing; no burst after a long stall.
                next_due = time.monotonic() + 1.0 / a.rate
        finally:
            if connection is not None and connection.is_open:
                try:
                    with deadline(5):
                        connection.close()
                except Exception as exc:
                    print("Cleanup warning: " + type(exc).__name__,
                          file=sys.stderr)
    return 0

if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as exc:
        print(type(exc).__name__ + ": " + str(exc), file=sys.stderr)
        sys.exit(2)

-- psql demo variables. Application workers use parameterized statements.
\set worker 'relay-1'
\set batch 10

BEGIN;
WITH picked AS (
  SELECT event_id FROM outbox
  WHERE sent_at IS NULL
    AND (lease_until IS NULL OR lease_until < clock_timestamp())
  ORDER BY created_at,event_id
  LIMIT :batch
  FOR UPDATE SKIP LOCKED
)
UPDATE outbox o
SET lease_owner=:'worker',
    lease_until=clock_timestamp() + interval '30 seconds',
    lease_epoch=o.lease_epoch+1,
    attempts=o.attempts+1
FROM picked p
WHERE o.event_id=p.event_id
RETURNING o.event_id,o.payload,o.lease_owner,o.lease_epoch,o.lease_until;
COMMIT;
-- Publish outside the DB transaction, not while holding row locks.

-- Values must come from THIS worker's returned claim, not a fresh SELECT.
\set worker 'relay-1'
\set event 'dddddddd-dddd-4ddd-8ddd-ddddddddddd1'
\set epoch 1

UPDATE outbox
SET sent_at=clock_timestamp(),lease_owner=NULL,lease_until=NULL
WHERE event_id=:'event'::uuid
  AND sent_at IS NULL
  AND lease_owner=:'worker'
  AND lease_epoch=:epoch
  AND lease_until > clock_timestamp()
RETURNING event_id,sent_at;
-- Exactly 1 row = successful mark; 0 = expired/lost/already marked claim.
-- A publish may still have succeeded when this UPDATE returns zero.

-- Optional heartbeat for a bounded, still-owned claim.
UPDATE outbox
SET lease_until=clock_timestamp() + interval '30 seconds'
WHERE event_id=:'event'::uuid AND sent_at IS NULL
  AND lease_owner=:'worker' AND lease_epoch=:epoch
  AND lease_until > clock_timestamp()
RETURNING lease_until;
-- If zero rows, STOP using the old claim. Do not silently adopt a new epoch.

-- Review a retention cutoff before running. Example ONLY; choose your own.
\set cutoff '2026-09-01T00:00:00Z'

-- Preview: only already-sent rows can be eligible.
SELECT count(*) AS eligible FROM outbox
WHERE sent_at < :'cutoff'::timestamptz
  AND lease_owner IS NULL;
SELECT count(*) AS pending_before FROM outbox WHERE sent_at IS NULL;

-- Bounded cleanup after archiving required evidence.
WITH eligible AS (
  SELECT event_id FROM outbox
  WHERE sent_at < :'cutoff'::timestamptz
    AND lease_owner IS NULL
  ORDER BY sent_at,event_id
  LIMIT 1000
  FOR UPDATE SKIP LOCKED
)
DELETE FROM outbox o USING eligible e
WHERE o.event_id=e.event_id
RETURNING o.event_id;

SELECT count(*) AS pending_after FROM outbox WHERE sent_at IS NULL;
-- pending_before == pending_after when relay/producer are paused for this test.
-- Inbox/unique business key retention is a separate policy; do not delete it here.