数据摄取错误处理方案

以下为“日志源A”数据摄取失败的标准化错误处理流程，适用于流式或批式摄取场景（如 Filebeat/Fluent Bit → Kafka/Kinesis → 流处理/Flink/Spark → 数据湖/数仓）。

一、目标与范围
- 目标：在不影响整体吞吐的前提下，稳定识别、隔离与恢复摄取失败，保证可追踪、可重放、数据一致性与最小数据丢失。
- 范围：源端采集、传输链路、解析/转换、目标写入、状态管理（偏移/检查点）、告警与回放。

二、参考架构（抽象）
- 源端代理：Agent/Collector（如 Fluent Bit）将日志推送至消息队列。
- 消息队列：Kafka/Kinesis，承接背压与重放。
- 摄取作业：Flink/Spark/自研消费者，完成解析、校验、清洗、入湖/入仓。
- 存储：对象存储（S3/OSS/HDFS）或数据仓库（Iceberg/Hudi/Delta、ClickHouse、BigQuery 等）。
- 附属组件：Schema Registry、DLQ（死信队列）、监控告警、审计与重放工具。

三、错误类型与判定
- 源端采集失败：网络不可达、鉴权失败、证书过期、源接口限流/429、Agent 崩溃或本地缓冲写满、时间漂移。
- 传输链路失败：Broker 不可用、分区 Leader 选举中、连接超时、吞吐不足导致高滞后。
- 解析/模式失败：编码不符（UTF-8/GBK）、压缩格式不匹配、消息超限、必填字段缺失、Schema 变更不兼容。
- 转换/质量失败：业务规则校验失败、类型转换异常、PII 脱敏缺失。
- 目标写入失败：对象存储/仓库不可用、权限不足、表/分区不存在、Schema 不兼容、事务提交失败。
- 状态/检查点失败：偏移提交失败、Flink Checkpoint 失败、幂等键冲突。

判定原则：
- 瞬时错误（网络抖动、限流、短时不可用）→ 可重试。
- 永久错误（鉴权错误、格式不可解析、Schema 永久不兼容）→ 直接送入 DLQ/隔离区。
- 模糊错误初判为瞬时，但在阈值内连续失败升级为永久错误策略。

四、处理流程（文字版流程图）
1) 采集器/消费者接收消息 → 基础校验（大小、编码、必填字段、Schema ID）。
2) 校验失败 → 标记为永久错误 → 写入 DLQ（携带源偏移、分区、错误码、首次出现时间、尝试次数、原始载荷指针）。
3) 校验通过 → 执行解析/转换。
4) 转换失败：
   - 可修复（缺维、外部维表短暂不可用）→ 有界重试（指数退避+抖动）→ 失败后入 DLQ。
   - 不可修复（字段类型根本不兼容）→ 直接入 DLQ。
5) 目标写入：
   - 使用幂等/事务写入；失败则按瞬时/永久分类重试。
   - 成功后提交偏移/检查点；失败则不提交偏移，避免丢数。
6) 对连续失败的源分区/租户开启断路器，降速或暂停消费，防雪崩。
7) 观测系统实时汇总成功率、延迟、滞后、DLQ 增长，触发告警与自动化修复。
8) 故障解除后，基于偏移/检查点或 DLQ 进行选择性回放，确保整体一致性。

五、关键策略与配置
1) 重试与退避
- 指数退避：base_delay × 2^attempt，叠加抖动（±20%）。
- 最大尝试次数：瞬时错误 3–7 次；分层（解析<写入<外部依赖）。
- 断路器：窗口内失败率>阈值（如 50%/5 分钟）则熔断，冷却期后半开探测。
- 限速与背压：基于消费者滞后、sink QPS、失败率动态调节并发/批量。

2) DLQ/隔离区
- 形态：DLQ Topic（Kafka）或隔离存储桶（原始文件+错误元数据）。
- 记录项：source_id、topic/partition/offset、event_time、ingest_time、schema_id、error_code、error_detail_hash、retriable、attempt_count、payload_ptr（或原文）、PII 标识。
- 保留策略：与合规匹配（如 7–30 天）；加密、访问控制。
- 重放工具：支持按时间/错误类型/租户范围回放到主 Topic 或旁路修复管道。

3) 幂等与事务
- 事件键：event_id（源唯一键或 hash），在 sink 端做去重/幂等写（MERGE/UPSERT）。
- Kafka 侧：启用 idempotent producer、必要时使用事务性写入（producer tx + sink 事务/两阶段提交），或在对象存储采用临时区→清单文件→原子提交。
- Flink：启用 Exactly-Once（Checkpoint + TwoPhaseCommit Sink）；Spark 结构化流使用幂等输出与 checkpoint 目录。

4) Schema 与数据质量
- Schema Registry 强制兼容策略（如 backward/forward）；不兼容变更触发阻断与告警。
- DQ 规则：必填/枚举/范围/时间戳新鲜度；失败送 DQ-DLQ 并标注规则 ID。
- 部分容忍：可配置字段降级（如缺少非关键字段时默认值+标记）。

5) 观测与告警
- 指标：
  - ingest_success_rate、error_rate_by_type、DLQ_rate、DLQ_backlog
  - consumer_lag、end-to-end_latency、throughput、checkpoint_duration/failure
  - sink_write_failures、schema_incompat_count
- SLO 示例：5 分钟窗口成功率≥99%，端到端延迟P95≤2 分钟，DLQ 比例≤0.5%。
- 告警：零摄取 X 分钟、失败率激增、DLQ 突增、滞后>阈值、连续 checkpoint 失败、Schema 不兼容事件。

6) 自动化修复
- 鉴权失败：自动刷新 Token/轮转密钥，失败升级为人工。
- 限流/429：读取 Retry-After，动态降速。
- Schema 漂移：自动注册兼容字段（仅允许向后兼容），不兼容变更阻断并创建变更请求。
- 目标不可用：切换到暂存区（staging）+延迟提交。
- 时钟漂移：触发 NTP 校正；时间窗口内矫正 event_time。

7) 安全与合规
- DLQ 和原始日志加密、细粒度访问控制、PII 脱敏/标记。
- 审计表记录每次丢弃/隔离/重放决策与操作者。

六、运行手册（应急步骤）
- 步骤 1：确认范围
  - 查看 Dashboard：error_rate、DLQ_backlog、consumer_lag、sink 可用性。
  - 定位失败集中在采集、解析、写入还是检查点。
- 步骤 2：分类
  - 瞬时 vs 永久：依据错误码/重试历史/断路器状态。
- 步骤 3：处置
  - 瞬时：增加退避、降低并发、确认下游可用性；必要时启用暂存区。
  - 永久：保持主链路运行，将问题记录入 DLQ；修复后从 DLQ 或偏移回放。
- 步骤 4：修复
  - 鉴权：更新凭据；源端连通性检查；证书/CA 更新。
  - Schema：更新注册与转换逻辑；回放前在影子环境验证。
  - 目标写入：恢复写入权限/配额/表结构；验证幂等键。
- 步骤 5：回放与验证
  - 从 DLQ/偏移区间回放；监控重复写入与延迟；对账审计（投入/产出条数、哈希抽样）。
- 步骤 6：复盘
  - 根因分析（RCA）、SLO 影响、是否需要阈值/退避/分区/限速策略调整。

七、最小实现示例（Kafka 消费者，带重试与 DLQ，伪代码）
- 假设：
  - 主 Topic：logs.sourceA
  - DLQ Topic：dlq.sourceA
  - 永久错误：Schema 不兼容/必填缺失/无法解析
  - 瞬时错误：Broker/Sink 超时、限流

示例（Python 伪代码）：
- 省略依赖导入与连接细节，仅示意控制流程。

def handle_message(msg):
    payload = msg.value()
    meta = {"topic": msg.topic(), "partition": msg.partition(), "offset": msg.offset()}
    try:
        record = parse_and_validate(payload)  # 编码/Schema/必填校验
    except PermanentError as e:
        send_to_dlq(payload, meta, error_code="PARSE_PERM", detail=str(e))
        return "DLQ"
    except TransientError as e:
        raise e

    attempt = 0
    while True:
        try:
            upsert_to_sink(record)  # 幂等写，使用 event_id 去重
            return "OK"
        except TransientError as e:
            attempt += 1
            if attempt > MAX_RETRIES:
                send_to_dlq(payload, meta, error_code="SINK_RETRY_EXCEEDED", detail=str(e), attempts=attempt)
                return "DLQ"
            sleep(backoff_with_jitter(attempt))
        except PermanentError as e:
            send_to_dlq(payload, meta, error_code="SINK_PERM", detail=str(e))
            return "DLQ"

for msg in consumer:
    result = handle_message(msg)
    if result in ("OK", "DLQ"):
        consumer.commit(msg)  # 仅在成功写入或已入 DLQ 后提交偏移，避免丢数

- 关键点：
  - 不在“写入未成功”时提交偏移。
  - 区分永久/瞬时错误，DLQ 记全量上下文。
  - 幂等写入保障回放安全。
  - backoff_with_jitter 实现指数退避+抖动。

八、元数据与审计
- 错误审计表（示例字段）：event_id、source_id、topic/partition/offset、error_type、error_code、attempt_count、first_seen_ts、last_seen_ts、operator/auto_action、replay_job_id、final_status。
- 数据血缘：记录从源到目标的 run_id、schema_id、变更版本，便于追溯与合规。

九、验收标准
- 失败被正确分类并隔离，主链路无阻塞。
- 可观测：关键 SLI 实时可见，告警不漏报。
- 可恢复：DLQ 回放闭环可验证，重复写可控（幂等）。
- 合规安全：敏感数据在 DLQ 亦受控与可审计。

以上流程可直接落地到现有 Kafka/Flink/Airflow 等栈；对其他队列或云原生服务（Kinesis、Pub/Sub、BigQuery、S3 等）仅需替换对应的客户端与幂等/事务机制实现。

Error handling process for ingestion failures from Convergence Layer B

Scope
- Applies to both streaming (e.g., Kafka topics or event bus from Layer B) and batch ingestion (e.g., files, API exports).
- Target systems include the landing/bronze zone in the lakehouse (e.g., object storage + transactional tables such as Delta/Iceberg) and downstream CDC or warehouse loads.

1) Failure detection
- Health checks before ingestion:
  - Source availability: endpoint reachable, topic exists, file manifest present, authentication/authorization valid.
  - Contract verification: schema compatibility check against registry; required fields present; partitioning keys available.
  - Data readiness: watermark/timestamp not older than threshold; batch completeness via manifest or count expectations.
- Runtime detection:
  - Transport errors: timeouts, TLS/SSL failures, DNS errors.
  - Schema/format errors: parse failures, incompatible schema evolution.
  - Content/data quality errors: nulls in required columns, invalid enums, out-of-range values, referential integrity breaks.
  - Compute/runtime errors: OOM, executor failures, container/node issues.
  - Sink/write errors: permission denied, quota exceeded, transactional commit failures.

2) Classification and routing
- Classify each failure to one of:
  - Upstream unavailable or rate-limited.
  - Transient transport/infrastructure.
  - Contract violation (schema/format).
  - Data quality violation.
  - Duplicate/late events or replay artifacts.
  - Sink/write path failure.
- Route records or batches accordingly:
  - Transient: retry pipeline; no quarantine.
  - Contract/data quality: quarantine at DLQ/quarantine zone.
  - Sink failure: halt and rollback; no partial commits.

3) Handling logic (core flow)
- Pre-ingestion guard:
  - If upstream unavailable or SLAs not met, pause ingestion, trigger alert, and backoff retries.
- Ingestion execution:
  - Use idempotent writes and transactional sinks (Delta/Iceberg transactional commit; or framework two-phase commit).
  - For streaming, commit offsets/checkpoints only after successful sink commit.
- On error:
  - Transient transport/infrastructure:
    - Apply exponential backoff with jitter; maximum retry window; circuit breaker after threshold to protect downstream.
    - Maintain backpressure (reduce poll rate or pause consumer).
  - Schema/format violations:
    - Route offending records (or entire batch) to DLQ/quarantine with full context.
    - Stop automated schema evolution unless explicitly allowed; require approval and change management.
  - Data quality violations:
    - If policy is “reject on failure,” send bad records to quarantine and continue for valid records only if row-level quarantine is approved; otherwise fail the batch atomically.
    - If “quarantine-and-continue” is enabled, ensure downstream consumers can tolerate partial loads and completeness flags.
  - Sink/write failures:
    - Abort transaction and rollback; do not commit offsets or checkpoints. Retry after remediation.
- Post-error actions:
  - Persist a failure event to an ingestion_status/audit table with correlation_id, source_dataset, run_id, batch_id, counts, error_class, and next_action.
  - Escalate via alerts per severity policy (e.g., P2 for repeated transient, P1 for contract breaches).

4) Streaming ingestion specifics (Kafka/Flink/Spark Structured Streaming)
- Idempotency and exactly-once:
  - Use framework support:
    - Flink: enable checkpointing and two-phase commit sinks; offsets advance only on successful sink commit.
    - Spark Structured Streaming: use checkpointLocation; in foreachBatch, write to a transactional table (Delta/Iceberg) then commit; checkpoint update happens after successful write.
  - Avoid committing Kafka offsets before sink write.
- DLQ handling:
  - For parse/schema failures: produce to a DLQ topic or write to a quarantine table/object path with metadata.
  - Include fields: original payload, topic/partition/offset, schema_id, parse_error, ingestion_time, run_id.
- Backpressure:
  - Dynamically reduce max in-flight requests; pause topic partitions if sustained errors; resume upon recovery.

Example (Spark Structured Streaming, DLQ and transactional sink):
- Pseudocode:
  - val stream = spark.readStream.format("kafka").option("subscribe", "layerB.topic").load()
  - val parsed = parseAndValidate(stream) // separate valid and invalid
  - invalid.writeStream.format("delta").option("checkpointLocation", ".../chk/dlq").start(".../quarantine/layerB/topic")
  - valid.writeStream.format("delta").option("checkpointLocation", ".../chk/sink").option("mergeSchema", "false").start(".../bronze/layerB/topic")

5) Batch ingestion specifics (files/API exports)
- Readiness:
  - Use manifest files or control tables to assert batch completeness before load.
  - Verify checksum and size; reject corrupted files.
- Atomicity:
  - Load into a temporary staging path/table; validate; atomically swap/commit to bronze upon success.
- Quarantine:
  - For row-level failures, write bad records to quarantine; for file-level failures, move entire file to quarantine path and mark batch failed.
- Retry:
  - Exponential backoff, bounded retries; if consistent failure, open incident and stop downstream dependencies.

Example (Airflow DAG pattern):
- Use retries + failure callbacks + quarantine move:
  - Ingest task with retries=5, retry_delay=5m, on_failure_callback=move_to_quarantine_and_emit_alert
  - Validation task: schema + dq checks; if fail, short-circuit downstream, emit audit entry.
  - Commit task: transactional upsert to bronze; only runs if validation passed.
  - Sensor tasks for source readiness and manifest presence.

6) Quarantine/DLQ structure
- Storage:
  - Object path: quarantine/layerB/{dataset}/{yyyy}/{MM}/{dd}/{batch_id}/
  - Table: dlq_layerB(dataset STRING, batch_id STRING, run_id STRING, error_class STRING, error_detail STRING, payload VARIANT/BINARY, source_pointer STRUCT, schema_id STRING, event_time TIMESTAMP, ingestion_time TIMESTAMP)
- Access:
  - Restricted to data engineering and quality teams; immutable writes.
- Retention:
  - Configurable (e.g., 30–90 days), with purge policy and GDPR considerations.
- Indexing:
  - Partition by dataset and date; add bloom/secondary indexes if needed for large DLQs.

7) Replay and recovery
- Preconditions:
  - Root cause fixed (e.g., schema updated, upstream corrected, infrastructure restored).
  - Approved change ticket if contract evolved.
- Replay process:
  - Select quarantine records for affected dataset and window.
  - Re-run validation with updated rules.
  - Write to bronze using idempotent keys (natural keys or event_id) and deduplicate.
  - Update audit table: mark records as recovered with recovery_run_id and outcome.
- Batch re-ingestion:
  - Reprocess failed batches after fix; ensure transactional overwrite/merge to avoid partial duplicates.

8) Observability and alerting
- Metrics:
  - Ingestion success/failure counts per dataset and per run.
  - DLQ rate and size; schema violation counts.
  - Consumer lag (streaming); time-to-land (batch).
  - Sink commit latency; rollback counts.
- Logs:
  - Structured logs with correlation_id, run_id, source_dataset, partition/offset or file path, error_class, stacktrace_hash.
- Alerts:
  - Severity by error class:
    - P1: Contract violations, repeated sink failures, data loss risk.
    - P2: Persistent transient failures beyond threshold.
    - P3: Minor data quality deviations within tolerance.
- Runbooks:
  - Linked to alerts; include steps for triage, rollback, replay, and contacting Layer B owners.

9) Consistency and data contracts
- Schema registry integration:
  - Enforce compatibility policy (e.g., backward or full).
  - Reject incompatible schemas and open change request.
- Idempotency:
  - Use event_id or primary keys; deduplicate on write (Merge into Delta/Iceberg).
- Atomicity:
  - All-or-nothing per batch unless explicitly allowed row-level quarantine with downstream tolerance flags (e.g., completeness_percentage).

10) Access control and escalation
- Role-based permissions:
  - Ingestion services: read from Layer B, write to bronze/quarantine; limited delete.
  - Ops team: replay/quarantine purge via controlled jobs.
- Escalation:
  - Notify Layer B owner when contract breaches or sustained data quality issues occur.
  - Document RCA with impacted datasets, time window, and remediation steps.

Minimal code examples

Airflow task with retry and failure callback:
- PythonOperator(
    task_id="ingest_layerB_datasetX",
    python_callable=ingest_fn,
    retries=5,
    retry_delay=timedelta(minutes=5),
    on_failure_callback=lambda ctx: quarantine_and_alert(ctx)
  )

Spark batch ingestion with badRecordsPath:
- spark.read.format("json").option("badRecordsPath", "s3://quarantine/layerB/datasetX/bad/").load(src) \
  .transform(validate_fn) \
  .write.format("delta").mode("append").save("s3://bronze/layerB/datasetX/")

Flink sink with two-phase commit (conceptual):
- env.enableCheckpointing(60000, EXACTLY_ONCE)
- stream.transform(validate).addSink(new TwoPhaseCommitSinkToDelta(...))

Outcome
- The process ensures failures from Convergence Layer B are detected early, isolated safely, retried appropriately, and remediated through controlled replay, while maintaining data integrity (no partial commits, idempotent writes) and providing full observability, auditability, and compliance with data contracts.

目的
- ゲートウェイソースCからのデータ取り込み（ingestion）が失敗した際に、サービス継続性・データ整合性・再処理容易性を確保するための標準的なエラー処理フローを定義する。

前提
- 取り込みは push（Cが送信）または pull（我々が取得）いずれにも対応。
- ストリーミング（例: Kafka/gRPC/WebSocket）とバッチ（例: REST/ファイル）を想定。
- 生データは Raw/landing ゾーンに、整形後は Staging/Curated に保存。オフセット/チェックポイントで進捗管理。
- スキーマレジストリとDLQ（Dead Letter Queue）/隔離ストレージを用意。

失敗検知
- 同期呼び出し: 呼び出し結果（HTTPステータス、gRPCコード）、タイムアウト、TLS/接続エラーを検知。
- 非同期/ストリーミング: コンシューマラグ、再平衡失敗、ブローカー接続断、レコードデコード失敗を検知。
- リアルタイムのメトリクスとログ（失敗率、リトライ回数、遅延、エラー分類）を記録し、相関ID・ソースID（C）・オフセット/チェックポイント・タイムスタンプを付与。

失敗の分類
- 交通層/ネットワーク: DNS/TCP/TLS/タイムアウト/接続リセット。
- プロトコル層: HTTP 5xx、gRPC UNAVAILABLE 等の一時障害、HTTP 4xx（400/401/403/404/429）等のクライアント側。
- 認証・認可: トークン期限切れ、鍵/証明書不一致。
- レート制限: 429、Retry-After 指定あり/なし。
- データ/スキーマ: 形式不正、必須項目欠落、型不一致、スキーマ互換性違反、配列長超過、圧縮/暗号化不一致。
- データ品質: 重複、順序乱れ、外れ値、参照整合性違反、PIIポリシー違反。
- リソース/バックプレッシャ: メモリ/ディスク不足、ファイルハンドル枯渇、スレッドプール飽和、ストレージ書き込み失敗。
- 下流書き込み: 一意制約違反、トランザクション失敗、コミット不可。

自動対応フロー
- 一時障害（ネットワーク、5xx、UNAVAILABLE、408/ETIMEDOUT）
  - 指数バックオフ＋ジッタで自動リトライ。例: base=500ms, factor=2, max=5–7回。
  - サーキットブレーカを適用。一定連続失敗で open、クールダウン後に half-open。
  - ストリーミングはコンシューマのレートを下げ、ポーズ/リジュームでバックプレッシャ制御。
- レート制限（429）
  - Retry-After/RateLimit-Reset を優先。なければバックオフにフォールバック。
  - 優先度キューで他ソースを先行処理。Cはスロットリング。
- 認証・認可（401/403、証明書エラー）
  - トークン自動更新/キーローテーションを実施。失敗時は即座に取り込み停止、アラート発報。
  - 永続化された失敗イベントに資格情報メタデータを添付（秘密情報は記録しない）。
- クライアントエラー（400/404、スキーマ/ペイロード不正）
  - リトライせずにDLQ/隔離ストレージへ格納。原本ペイロード＋エラー理由＋スキーマバージョンを付与。
  - スキーマレジストリの互換性ルール（後方互換など）に基づきコンシューマを一時停止し、スキーマ更新フローへ。
- データ品質
  - 重複はイベントID/一意キーでデデュープ（ストレージ側UPSERT/幂等化キーを使用）。
  - 順序保証が必要な場合はパーティションキー単位で順序バッファリング。違反時は隔離。
- リソース不足
  - 自動スケール、ワーカー並列度の動的調整、ローカル耐久バッファ（ディスク）へのスプール。
  - ストレージ書き込み失敗はトランザクション/バルク単位で再試行。コミット成功前にオフセットは進めない。

永続化・監査
- 失敗イベントストア（例: Elasticsearch/OLAPテーブル）に以下を記録:
  - correlation_id、source_id=C、イベントキー/オフセット、試行回数、エラー分類（transport/protocol/auth/schema/dq/resource/write）、ステータスコード/例外、発生時刻。
- メトリクス/トレース:
  - ingestion_failure_rate、retry_count、open_circuit_count、dlq_enqueue_count、consumer_lag、time_to_recovery。

通知・アラート
- しきい値超過（例: 5分間で失敗率 > 2%、DLQ流入 > N件、サーキット open）で Pager/SNS/Slack 通知。
- 認証・スキーマ互換性違反は即時クリティカル。

再処理（リカバリ）フロー
- 原因解消後（例: 資格情報修正、スキーマ更新、レート制限緩和）:
  - DLQ/隔離データをバッチで再処理。処理速度を制限し、下流に負荷をかけない。
  - スキーマは固定バージョンで再検証。変換ロジックのバージョンをイベントに付与。
  - ストリーミングは last committed offset から再開。コミットは「書き込み成功」後にのみ実施。
  - 完全性検証: ソースCのシーケンス番号/件数と突合。ギャップがあれば再取得要求。

整合性・冪等性
- at-least-once を基本とし、下流は冪等なUPSERT/合成キーで重複防止。
- exactly-once が必要な場合は、トランザクションログ＋オフセットコミットの二相制御、またはブローカーのトランザクショナルAPIを使用。
- バッチ取り込みはウォーターマーク/チェックポイントで再開点を管理。部分成功時は部分コミット禁止、ロールバックか粒度を下げた再試行。

セキュリティ・コンプライアンス
- 隔離データは暗号化保存、PIIはマスキング。アクセス制御・監査ログを必須。
- 保持期間（TTL）と削除ポリシーを定義。再処理完了後に隔離データを消去。

実装要点（抜粋）
- リトライポリシ
  - 5xx/ネットワーク/UNAVAILABLE/408 はリトライ対象。
  - 4xx（400/404）やスキーマ不一致は非リトライでDLQ。
  - 429 は Retry-After 準拠の遅延。
- サーキットブレーカ
  - 連続失敗数/失敗率で open、一定時間後 half-open、単発成功で close。
- バックプレッシャ
  - コンシューマのポーズ/リジューム、最大同時処理数の動的制御、優先度キュー。
- スキーマ管理
  - 互換性チェック（後方互換）に失敗したら取り込み停止、スキーマ進化の承認プロセスへ。

擬似コード（リトライ/隔離の骨子）
- function ingestFromGatewayC(event):
  - for attempt in 1..max_attempts:
    - res = call_gateway_c(event)
    - if res.success:
      - write_to_storage(event)
      - commit_offset(event)
      - return OK
    - if is_permanent_error(res):  // 4xx(400/404), schema violation, payload invalid
      - send_to_DLQ(event, res)
      - record_failure(event, res, attempt)
      - return DLQ
    - delay = compute_backoff(attempt, jitter, parse_retry_after(res))
    - sleep(delay)
    - if circuit_breaker_open():
      - record_failure(event, res, attempt)
      - alert("circuit_open")
      - return RETRY_LATER
  - open_circuit()
  - alert("ingestion_failed_threshold")
  - return FAIL

運用フロー（人手介入）
- 重大障害（認証/スキーマ/永続的なペイロード不正）:
  - インシデント起票、原因切り分け、修正のデプロイ。
  - ソースCとの合意済みスキーマ・レート・認証情報を再確認。
- 復旧後の後処理:
  - DLQ再処理の結果確認、欠損・重複のレポート化。
  - ランブック更新、しきい値・バックオフの再調整。

このフローにより、ソースCからの取り込み失敗時でも、サービス劣化を最小化しつつ、データ完全性・再処理可能性・監査可能性を維持する。