Payment Reconciliation: Detecting Lost Orders, Over‑payments and Auto‑Repairing Them

Why payment reconciliation is needed

Even with idempotent requests, signed callbacks, at‑least‑once MQ delivery, local transaction persistence, and scheduled compensation queries, discrepancies still occur because payment callbacks can be delayed, duplicated, or lost; network and timing uncertainties exist between upstream channels and internal systems; high concurrency aims for eventual consistency rather than instantaneous consistency; and manual adjustments or abnormal releases can drift the ledger.

Therefore a payment reconciliation system is the final financial defense line in the payment chain.

Five core responsibilities of a production‑grade reconciliation system

Reliably fetch T+1 channel statements, parse, standardize and store them.

Efficiently compare channel statements with internal flows to discover lost orders, over‑payments, shortfalls, and amount mismatches.

Drive error handling via rules and a state machine, supporting automatic repair, manual takeover, and traceability.

Use transactional messages or an Outbox to guarantee consistency between repair results and downstream notifications.

Remain scalable, observable, and compensable under high concurrency, massive bills, and sharding.

Unified terminology for discrepancies

CHANNEL_ONLY : Channel has a successful record, internal system does not. Typical manifestation: user paid, order not successful. Risk: lost order, fulfillment failure, user complaint.

INTERNAL_ONLY : Internal system records success, channel has no successful record. Typical manifestation: system thinks money arrived, but channel shows none. Risk: over‑payment, fake success.

AMOUNT_MISMATCH : Both sides have records but amounts differ. Typical manifestation: differences in actual amount, discount, fee, split‑bill. Risk: settlement imbalance, financial discrepancy.

STATUS_MISMATCH : Same order number, different status. Typical manifestation: internal success, channel closed; or vice‑versa. Risk: status drift, compensation conflict.

DUPLICATE_RECORD : Same business order recorded multiple times. Typical manifestation: callback replay, duplicate consumption, manual duplicate entry. Risk: duplicate shipment, double billing.

Overall architecture

The system consists of the following components:

Bill Ingestor – downloads T+1 statements.

Parser SPI – adapts various channel formats (CSV, ZIP, Excel, fixed‑width).

Standardized Bill Store – unified schema for channel statements.

Payment Flow Store – projection of successful internal flows (via CDC/Outbox).

Reconcile Engine – shard‑based diff detection.

Diff Table ( recon_diff) – stores discovered discrepancies.

Repair Center – state‑machine driven error handling.

Order Service, Notification, Settlement – downstream consumers triggered via transactional messages or Outbox.

Key design points

Channel bills and internal flows are not compared directly in the business DB; a dedicated reconciliation store isolates the heavy read workload.

Parsing is plug‑in via SPI; each channel implements BillParser to produce ChannelBillDetail objects.

Difference detection and repair are decoupled – they have different pacing, retry policies, and permission boundaries.

Repair results and downstream notifications must be atomic (transactional MQ or Outbox).

The reconciliation system itself must be compensable – failed downloads, parsing interruptions, shard timeouts, or repair failures can be re‑run.

Core data model (four essential tables)

1. Channel bill detail

CREATE TABLE `channel_bill_detail` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `batch_no` VARCHAR(64) NOT NULL COMMENT '对账批次号',
  `recon_date` DATE NOT NULL COMMENT '账单日期',
  `channel` VARCHAR(32) NOT NULL COMMENT '渠道',
  `channel_order_no` VARCHAR(64) NOT NULL COMMENT '渠道订单号',
  `channel_trade_no` VARCHAR(64) DEFAULT NULL COMMENT '渠道流水号',
  `merchant_order_no` VARCHAR(64) DEFAULT NULL COMMENT '商户订单号',
  `trade_status` VARCHAR(32) NOT NULL COMMENT 'SUCCESS/CLOSED/REFUND',
  `amount` BIGINT NOT NULL COMMENT '订单金额，单位分',
  `fee_amount` BIGINT DEFAULT 0 COMMENT '手续费，单位分',
  `settle_amount` BIGINT DEFAULT 0 COMMENT '结算金额，单位分',
  `pay_time` DATETIME DEFAULT NULL,
  `settle_time` DATETIME DEFAULT NULL,
  `raw_line_no` INT NOT NULL COMMENT '原始行号',
  `shard_id` INT NOT NULL COMMENT '分片号',
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_batch_channel_trade` (`batch_no`, `channel`, `channel_order_no`),
  KEY `idx_recon_date_shard` (`recon_date`, `shard_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='渠道账单明细表';

2. Payment flow projection

CREATE TABLE `payment_flow_recon` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `payment_id` BIGINT NOT NULL COMMENT '支付流水主键',
  `recon_date` DATE NOT NULL,
  `channel` VARCHAR(32) NOT NULL,
  `merchant_order_no` VARCHAR(64) NOT NULL,
  `channel_order_no` VARCHAR(64) DEFAULT NULL,
  `pay_status` VARCHAR(32) NOT NULL COMMENT 'INIT/SUCCESS/CLOSED/REFUND',
  `amount` BIGINT NOT NULL COMMENT '支付金额，单位分',
  `pay_time` DATETIME DEFAULT NULL,
  `biz_type` VARCHAR(32) DEFAULT NULL,
  `tenant_id` BIGINT DEFAULT NULL,
  `shard_id` INT NOT NULL,
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_payment_id` (`payment_id`),
  KEY `idx_recon_date_shard` (`recon_date`, `shard_id`),
  KEY `idx_channel_order` (`channel`, `channel_order_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='对账专用支付流水表';

3. Difference table

CREATE TABLE `recon_diff` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `batch_no` VARCHAR(64) NOT NULL,
  `recon_date` DATE NOT NULL,
  `channel` VARCHAR(32) NOT NULL,
  `channel_order_no` VARCHAR(64) DEFAULT NULL,
  `merchant_order_no` VARCHAR(64) DEFAULT NULL,
  `diff_type` VARCHAR(32) NOT NULL COMMENT 'CHANNEL_ONLY/INTERNAL_ONLY/AMOUNT_MISMATCH/STATUS_MISMATCH',
  `channel_status` VARCHAR(32) DEFAULT NULL,
  `internal_status` VARCHAR(32) DEFAULT NULL,
  `channel_amount` BIGINT DEFAULT NULL,
  `internal_amount` BIGINT DEFAULT NULL,
  `diff_amount` BIGINT DEFAULT NULL,
  `risk_level` VARCHAR(16) NOT NULL DEFAULT 'P2',
  `status` VARCHAR(32) NOT NULL DEFAULT 'INIT',
  `repair_no` VARCHAR(64) DEFAULT NULL,
  `remark` VARCHAR(512) DEFAULT NULL,
  `version` INT NOT NULL DEFAULT 0,
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_batch_channel_order_type` (`batch_no`, `channel`, `channel_order_no`, `diff_type`),
  KEY `idx_status_risk` (`status`, `risk_level`, `create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='对账差异表';

4. Task table (for resumable batch execution)

CREATE TABLE `recon_task` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `batch_no` VARCHAR(64) NOT NULL,
  `recon_date` DATE NOT NULL,
  `channel` VARCHAR(32) NOT NULL,
  `task_type` VARCHAR(32) NOT NULL COMMENT 'DOWNLOAD/PARSE/RECON/REPAIR',
  `task_status` VARCHAR(32) NOT NULL COMMENT 'INIT/RUNNING/SUCCESS/FAILED',
  `shard_id` INT DEFAULT NULL,
  `retry_count` INT NOT NULL DEFAULT 0,
  `error_msg` VARCHAR(1024) DEFAULT NULL,
  `start_time` DATETIME DEFAULT NULL,
  `end_time` DATETIME DEFAULT NULL,
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_batch_task_shard` (`batch_no`, `task_type`, `channel`, `shard_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='对账任务执行表';

Bill ingestion – handling dirty data

Typical challenges include multiple file formats (CSV, ZIP, Excel, fixed‑width), varying encodings (UTF‑8, GBK), header/footer rows, empty or duplicate rows, inconsistent fields across products, delayed generation, and repeated or partial downloads.

The ingestion pipeline follows these steps:

Scheduled trigger.

Bill availability probing.

Download raw file.

Validate integrity (MD5/size/line count).

Decompress, decrypt, transcode.

Parse per‑channel.

Persist standardized model.

Record batch metadata.

Standard bill model (Java)

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ChannelBillDetail {
    private String batchNo;
    private LocalDate reconDate;
    private String channel;
    private String channelOrderNo;
    private String channelTradeNo;
    private String merchantOrderNo;
    private String tradeStatus;
    private Long amount;
    private Long feeAmount;
    private Long settleAmount;
    private LocalDateTime payTime;
    private LocalDateTime settleTime;
    private Integer rawLineNo;
    private Integer shardId;
}

Parser SPI

public interface BillParser {
    String channel();
    Flux<ChannelBillDetail> parse(InputStream inputStream, BillParseContext context);
}

WeChat CSV parser example

@Component
@Slf4j
public class WechatBillParser implements BillParser {
    private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    @Override
    public String channel() { return "WECHAT"; }
    @Override
    public Flux<ChannelBillDetail> parse(InputStream inputStream, BillParseContext context) {
        return Flux.using(() -> new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)),
            reader -> Flux.fromStream(reader.lines())
                .index()
                .filter(t -> t.getT1() > 0)
                .map(t -> parseLine(t.getT1().intValue(), t.getT2(), context))
                .filter(Objects::nonNull),
            r -> { try { r.close(); } catch (IOException e) { log.warn("close reader failed", e); } });
    }
    private ChannelBillDetail parseLine(int lineNo, String line, BillParseContext context) {
        if (line.isBlank() || line.startsWith("总计") || line.startsWith("#")) return null;
        String[] cols = line.split(",", -1);
        String channelOrderNo = trim(cols[0]);
        return ChannelBillDetail.builder()
            .batchNo(context.batchNo())
            .reconDate(context.reconDate())
            .channel("WECHAT")
            .channelOrderNo(channelOrderNo)
            .channelTradeNo(trim(cols[1]))
            .merchantOrderNo(trim(cols[2]))
            .tradeStatus(normalizeStatus(trim(cols[3])))
            .amount(toFen(trim(cols[4])))
            .feeAmount(toFen(trim(cols[5])))
            .settleAmount(toFen(trim(cols[6])))
            .payTime(LocalDateTime.parse(trim(cols[7]), TIME_FORMATTER))
            .settleTime(LocalDateTime.parse(trim(cols[8]), TIME_FORMATTER))
            .rawLineNo(lineNo)
            .shardId(ShardUtils.shardId(channelOrderNo))
            .build();
    }
    private String trim(String v) { return v == null ? null : v.replace("`", "").trim(); }
    private String normalizeStatus(String s) {
        return switch (s) {
            case "SUCCESS", "支付成功" -> "SUCCESS";
            case "REFUND", "已退款" -> "REFUND";
            case "CLOSED", "已关闭" -> "CLOSED";
            default -> "UNKNOWN";
        };
    }
    private Long toFen(String yuan) { return new BigDecimal(yuan).movePointRight(2).longValueExact(); }
}

Why internal payment flows need a reconciliation projection

Directly querying the core payment tables is impractical because:

Query dimensions differ – core tables are sharded by order_id, user_id, payment_id, while reconciliation needs channel_order_no, recon_date, channel.

Full‑table scans would compete with online transaction workloads.

Reconciliation requires a stable snapshot; a dedicated projection provides that.

Typical approach: the payment chain writes to the core DB, CDC/Outbox events project successful records into payment_flow_recon, and the reconciliation system reads only the projection.

Difference detection engine – production‑grade approach

Naïve JOIN on 10 million rows is slow and non‑parallelizable. The production solution uses three layers of optimization:

Hash‑partition both sides on the same channel_order_no (e.g., 64 shards).

Process each shard concurrently – each shard can be retried, resumed, or throttled independently.

Within a shard, batch‑load both sides into memory maps and compare, avoiding per‑row SQL joins.

Shard utility

public final class ShardUtils {
    private static final int SHARD_COUNT = 64;
    private ShardUtils() {}
    public static int shardId(String channelOrderNo) {
        return (channelOrderNo.hashCode() & Integer.MAX_VALUE) % SHARD_COUNT;
    }
    public static int shardCount() { return SHARD_COUNT; }
}

Reconcile engine (simplified)

@Service
@Slf4j
@RequiredArgsConstructor
public class ReconcileEngine {
    private final ChannelBillDetailMapper channelBillDetailMapper;
    private final PaymentFlowReconMapper paymentFlowReconMapper;
    private final ReconDiffMapper reconDiffMapper;
    private final ReconTaskMapper reconTaskMapper;
    private final ThreadPoolTaskExecutor reconExecutor;

    public void reconcileBatch(String batchNo, LocalDate reconDate, String channel) {
        IntStream.range(0, ShardUtils.shardCount())
            .forEach(shardId -> reconExecutor.execute(() -> safeRunShard(batchNo, reconDate, channel, shardId)));
    }

    private void safeRunShard(String batchNo, LocalDate reconDate, String channel, int shardId) {
        try {
            reconcileShard(batchNo, reconDate, channel, shardId);
        } catch (Exception e) {
            log.error("reconcile shard failed, batchNo={}, shardId={}", batchNo, shardId, e);
            reconTaskMapper.markFailed(batchNo, channel, "RECON", shardId, e.getMessage());
        }
    }

    @Transactional(rollbackFor = Exception.class)
    public void reconcileShard(String batchNo, LocalDate reconDate, String channel, int shardId) {
        if (!reconTaskMapper.tryStart(batchNo, channel, "RECON", shardId)) return;
        List<ChannelBillDetail> channelBills = channelBillDetailMapper.selectByBatchAndShard(batchNo, shardId);
        List<PaymentFlowRecon> internalFlows = paymentFlowReconMapper.selectByDateChannelAndShard(reconDate, channel, shardId);
        Map<String, ChannelBillDetail> channelMap = channelBills.stream()
            .collect(Collectors.toMap(ChannelBillDetail::getChannelOrderNo, Function.identity(), this::chooseLatestChannelRecord));
        Map<String, PaymentFlowRecon> internalMap = internalFlows.stream()
            .filter(f -> f.getChannelOrderNo() != null)
            .collect(Collectors.toMap(PaymentFlowRecon::getChannelOrderNo, Function.identity(), this::chooseLatestInternalRecord));
        List<ReconDiff> diffs = new ArrayList<>();
        for (Map.Entry<String, ChannelBillDetail> e : channelMap.entrySet()) {
            String orderNo = e.getKey();
            ChannelBillDetail bill = e.getValue();
            PaymentFlowRecon flow = internalMap.remove(orderNo);
            if (flow == null) { diffs.add(buildChannelOnlyDiff(batchNo, reconDate, bill)); continue; }
            if (!Objects.equals(bill.getTradeStatus(), flow.getPayStatus())) { diffs.add(buildStatusMismatch(batchNo, reconDate, bill, flow)); continue; }
            if (!Objects.equals(bill.getAmount(), flow.getAmount())) { diffs.add(buildAmountMismatch(batchNo, reconDate, bill, flow)); }
        }
        for (PaymentFlowRecon flow : internalMap.values()) {
            if ("SUCCESS".equals(flow.getPayStatus())) {
                diffs.add(buildInternalOnlyDiff(batchNo, reconDate, flow));
            }
        }
        if (!diffs.isEmpty()) reconDiffMapper.insertBatchIgnore(diffs);
        reconTaskMapper.markSuccess(batchNo, channel, "RECON", shardId);
    }

    private ChannelBillDetail chooseLatestChannelRecord(ChannelBillDetail a, ChannelBillDetail b) {
        return a.getSettleTime().isAfter(b.getSettleTime()) ? a : b;
    }
    private PaymentFlowRecon chooseLatestInternalRecord(PaymentFlowRecon a, PaymentFlowRecon b) {
        return a.getUpdateTime().isAfter(b.getUpdateTime()) ? a : b;
    }
    private ReconDiff buildChannelOnlyDiff(String batchNo, LocalDate reconDate, ChannelBillDetail bill) {
        return baseDiff(batchNo, reconDate, bill.getChannel(), bill.getChannelOrderNo(), bill.getMerchantOrderNo())
            .setDiffType("CHANNEL_ONLY")
            .setChannelStatus(bill.getTradeStatus())
            .setChannelAmount(bill.getAmount())
            .setInternalAmount(0L)
            .setDiffAmount(bill.getAmount())
            .setRiskLevel("P0");
    }
    private ReconDiff buildInternalOnlyDiff(String batchNo, LocalDate reconDate, PaymentFlowRecon flow) {
        return baseDiff(batchNo, reconDate, flow.getChannel(), flow.getChannelOrderNo(), flow.getMerchantOrderNo())
            .setDiffType("INTERNAL_ONLY")
            .setInternalStatus(flow.getPayStatus())
            .setChannelAmount(0L)
            .setInternalAmount(flow.getAmount())
            .setDiffAmount(flow.getAmount())
            .setRiskLevel("P0");
    }
    private ReconDiff buildStatusMismatch(String batchNo, LocalDate reconDate, ChannelBillDetail bill, PaymentFlowRecon flow) {
        return baseDiff(batchNo, reconDate, bill.getChannel(), bill.getChannelOrderNo(), bill.getMerchantOrderNo())
            .setDiffType("STATUS_MISMATCH")
            .setChannelStatus(bill.getTradeStatus())
            .setInternalStatus(flow.getPayStatus())
            .setChannelAmount(bill.getAmount())
            .setInternalAmount(flow.getAmount())
            .setDiffAmount(Math.abs(bill.getAmount() - flow.getAmount()))
            .setRiskLevel("P1");
    }
    private ReconDiff buildAmountMismatch(String batchNo, LocalDate reconDate, ChannelBillDetail bill, PaymentFlowRecon flow) {
        return baseDiff(batchNo, reconDate, bill.getChannel(), bill.getChannelOrderNo(), bill.getMerchantOrderNo())
            .setDiffType("AMOUNT_MISMATCH")
            .setChannelStatus(bill.getTradeStatus())
            .setInternalStatus(flow.getPayStatus())
            .setChannelAmount(bill.getAmount())
            .setInternalAmount(flow.getAmount())
            .setDiffAmount(bill.getAmount() - flow.getAmount())
            .setRiskLevel("P1");
    }
    private ReconDiff baseDiff(String batchNo, LocalDate reconDate, String channel, String channelOrderNo, String merchantOrderNo) {
        return new ReconDiff()
            .setBatchNo(batchNo)
            .setReconDate(reconDate)
            .setChannel(channel)
            .setChannelOrderNo(channelOrderNo)
            .setMerchantOrderNo(merchantOrderNo)
            .setStatus("INIT");
    }
}

Difference handling state machine

public enum DiffStatus {
    INIT,
    AUTO_REPAIRING,
    AUTO_REPAIR_SUCCESS,
    AUTO_REPAIR_FAILED,
    WAIT_MANUAL_REVIEW,
    MANUAL_REPAIRING,
    FIXED,
    IGNORED
}

Automatic repair workflow

Three‑step process for a CHANNEL_ONLY diff:

Fetch the diff from recon_diff where status='INIT'.

CAS update status to AUTO_REPAIRING (optimistic lock).

Dispatch to a handler based on diff_type.

Repair service entry

@Service
@Slf4j
@RequiredArgsConstructor
public class DiffRepairService {
    private final ReconDiffMapper reconDiffMapper;
    private final RepairDispatcher repairDispatcher;
    public void triggerAutoRepair(Long diffId) {
        ReconDiff diff = reconDiffMapper.selectById(diffId);
        if (diff == null || !"INIT".equals(diff.getStatus())) return;
        int updated = reconDiffMapper.casStatus(diffId, "INIT", "AUTO_REPAIRING");
        if (updated != 1) return;
        try {
            repairDispatcher.dispatch(diff);
        } catch (Exception e) {
            log.error("auto repair dispatch failed, diffId={}", diffId, e);
            reconDiffMapper.markAutoRepairFailed(diffId, e.getMessage());
        }
    }
}

Dispatcher and handler interface

public interface DiffRepairHandler {
    boolean supports(String diffType);
    void repair(ReconDiff diff);
}

@Component
@RequiredArgsConstructor
public class RepairDispatcher {
    private final List<DiffRepairHandler> handlers;
    public void dispatch(ReconDiff diff) {
        handlers.stream()
            .filter(h -> h.supports(diff.getDiffType()))
            .findFirst()
            .orElseThrow(() -> new IllegalStateException("No handler for diff type: " + diff.getDiffType()))
            .repair(diff);
    }
}

Channel‑only handler (auto‑repair lost order)

@Component
@Slf4j
@RequiredArgsConstructor
public class ChannelOnlyRepairHandler implements DiffRepairHandler {
    private final ChannelGateway channelGateway;
    private final PaymentRepairFacade paymentRepairFacade;
    private final ReconDiffMapper reconDiffMapper;
    @Override
    public boolean supports(String diffType) { return "CHANNEL_ONLY".equals(diffType); }
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void repair(ReconDiff diff) {
        ChannelTradeResult tradeResult = channelGateway.query(diff.getChannel(), diff.getChannelOrderNo());
        if (!tradeResult.isSuccess()) {
            reconDiffMapper.turnToManual(diff.getId(), "Channel query not success");
            return;
        }
        RepairCommand command = RepairCommand.builder()
            .repairNo("RP" + IdGenerator.nextId())
            .merchantOrderNo(diff.getMerchantOrderNo())
            .channelOrderNo(diff.getChannelOrderNo())
            .channel(diff.getChannel())
            .payAmount(diff.getChannelAmount())
            .payTime(tradeResult.getSuccessTime())
            .reason("RECON_AUTO_REPAIR")
            .build();
        paymentRepairFacade.repairSuccessPayment(command);
        reconDiffMapper.markFixed(diff.getId(), command.getRepairNo(), "channel only repaired");
    }
}

Idempotency guarantees

Bill import uniqueness: batch_no + channel_order_no unique key.

Diff generation uniqueness: batch_no + channel + channel_order_no + diff_type unique key.

Repair actions generate a unique repair_no; downstream consumers deduplicate by repair_no or event_key.

Manual UI actions use optimistic‑lock version field to avoid race conditions.

Engineering for scale

Two‑level slicing: first by channel + recon_date, then by shard_id (64‑128 parallel tasks).

Separate thread pools for bill download, reconciliation, and repair workers.

Batch IN queries (500‑2000 rows) to limit DB pressure.

Indexes focused on recon_date, channel, shard_id.

When tables grow beyond single‑table limits, apply sharding or move historical data to ClickHouse/Hive and use Spark/Flink for distributed diff.

Monitoring & alerting

Key metrics (exposed via Prometheus):

Bill download success rate.

Bill parsing latency.

Shard reconciliation duration.

Task failure & retry counts.

Daily total bills, internal successful flows, and total discrepancy amount.

Per‑type diff counts (P0‑P2).

Auto‑repair success rate and average repair time.

Manual‑review ratio and P0 open duration.

Counter diffCounter = Counter.builder("recon_diff_total")
    .tag("channel", channel)
    .tag("diffType", diffType)
    .register(meterRegistry);
Timer repairTimer = Timer.builder("recon_repair_duration")
    .tag("channel", channel)
    .register(meterRegistry);

Typical alerts:

Channel fails to obtain a bill for 30 minutes.

P0 lost‑order diff count exceeds threshold.

Auto‑repair failure rate spikes.

Tenant‑level diff surge.

Common pitfalls

Running full‑table joins on the core payment DB kills online performance.

Detecting diffs without a repair loop provides no business value.

Letting the reconciliation service directly modify order/payment tables blurs domain boundaries.

Missing optimistic‑lock (CAS) on diff status leads to race‑condition corruption.

Treating transactional MQ as a silver bullet solves DB‑MQ consistency only, not business idempotency.

Discarding raw channel files hampers audit and dispute resolution.

Roadmap from 0 to 1

Stage 1: Stable T+1 bill fetching, parsing, and basic diff (lost order, over‑payment, shortfall) with manual UI.

Stage 2: Implement automatic repair APIs, diff state machine, and transactional messaging/Outbox.

Stage 3: Add sharding, separate thread pools, scaling indexes, and comprehensive metrics.

Stage 4: Platformize – SPI for new channels, multi‑tenant isolation, rule‑engine configuration, analytics dashboards.

Why automatic repair must be paired with transactional messages or Outbox

Repair actions usually trigger downstream processes (order status change, fulfillment, granting of benefits, financial ledger updates, user notifications). If the database transaction succeeds but the MQ send fails, or vice‑versa, the system ends up in an inconsistent state.

Repair result persistence and repair event delivery must either both succeed or both fail.

RocketMQ transactional message example

@Service
@RequiredArgsConstructor
public class RepairMessageService {
    private final RocketMQTemplate rocketMQTemplate;
    public void sendRepairEvent(Long diffId) {
        rocketMQTemplate.sendMessageInTransaction(
            "recon-repair-tx-group",
            "repair-success-topic",
            MessageBuilder.withPayload(diffId).build(),
            diffId);
    }
}

@RocketMQTransactionListener(txProducerGroup = "recon-repair-tx-group")
@Slf4j
public class RepairTxListener implements RocketMQLocalTransactionListener {
    @Resource
    private RepairTxExecutor repairTxExecutor;
    @Resource
    private ReconDiffMapper reconDiffMapper;
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        Long diffId = (Long) arg;
        try {
            repairTxExecutor.execute(diffId);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("execute local tx failed, diffId={}", diffId, e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        Long diffId = (Long) msg.getPayload();
        ReconDiff diff = reconDiffMapper.selectById(diffId);
        if (diff == null) return RocketMQLocalTransactionState.ROLLBACK;
        if ("FIXED".equals(diff.getStatus())) return RocketMQLocalTransactionState.COMMIT;
        if ("AUTO_REPAIR_FAILED".equals(diff.getStatus()) || "WAIT_MANUAL_REVIEW".equals(diff.getStatus()))
            return RocketMQLocalTransactionState.ROLLBACK;
        return RocketMQLocalTransactionState.UNKNOWN;
    }
}

Kafka preferred Outbox pattern

CREATE TABLE `outbox_event` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `event_key` VARCHAR(64) NOT NULL,
  `aggregate_type` VARCHAR(64) NOT NULL,
  `aggregate_id` VARCHAR(64) NOT NULL,
  `event_type` VARCHAR(64) NOT NULL,
  `payload` JSON NOT NULL,
  `status` VARCHAR(16) NOT NULL DEFAULT 'PENDING',
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_event_key` (`event_key`),
  KEY `idx_status_create_time` (`status`, `create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Outbox事件表';

Within a local transaction, write the repair result, update the diff status, and insert an outbox_event. A CDC or background poller then publishes the event to Kafka, ensuring atomicity without relying on broker‑side transactions.

Engineering upgrades for massive scale

Task slicing: first by channel + recon_date, then by shard_id (64‑128 parallel tasks).

Thread‑pool isolation: separate pools for bill download, reconciliation, and repair workers.

Database pressure control: batch IN queries (500‑2000 rows), indexes on recon_date, channel, shard_id, and read from the projection store instead of the core payment DB.

Sharding evolution: single table → single‑table sharding → multi‑database sharding → historical data to ClickHouse/Hive for analytics.

When daily bill volume reaches billions, move raw files to object storage and use Spark/Flink for distributed diff, writing results back to the online diff center.

Consistency principles – where idempotency is mandatory

Bill import idempotency: batch_no + channel_order_no unique.

Diff generation idempotency: batch_no + channel + channel_order_no + diff_type unique.

Repair action idempotency: each repair generates a unique repair_no; repair APIs must accept repair_no for deduplication.

MQ consumer idempotency: downstream consumers (fulfillment, benefit granting, accounting) deduplicate by repair_no or event_key.

Manual handling idempotency: UI actions use version or CAS updates to avoid race conditions.

Case study – automatic recovery of a lost order

Scenario:

User paid at 23:58 on 2026‑08‑18; channel succeeded.

Payment callback was lost due to network jitter; internal order remained PAYING.

Channel generated a T+1 statement the next day.

Reconciliation discovered a CHANNEL_ONLY diff with risk level P0. The automatic repair flow:

Repair center scanned the diff.

Channel query confirmed success.

Payment domain's repairSuccessPayment recorded a successful payment.

Transactional message triggered downstream fulfillment.

Diff status updated to FIXED.

The user saw the order marked as paid and received the goods without manual intervention.

Manual audit UI essentials

Filter diffs by channel, date, tenant, business line.

View original channel bill line and internal flow details.

Inspect recent repair attempts, error reasons, and query results.

Execute manual actions: supplement order, refund, ignore, or create a work ticket.

All actions are logged with operator, timestamp, reason, and before/after status.

Full reconciliation lifecycle

Bill acquisition : scheduled pull of T+1 statements, retry with back‑off if missing, record batch metadata.

Bill standardization : decompress, transcode, parse, filter invalid rows, batch insert standardized model.

Flow projection preparation : CDC or outbox projects successful payments into payment_flow_recon, partitioned by recon_date, channel, shard_id.

Difference detection : shard‑parallel diff generation, produce diff types and aggregate daily statistics.

Error correction : automatic repair where possible, otherwise route to manual queue; all actions carry idempotency keys and audit logs.

Result persistence : daily reports, risk dashboards, metrics on diff rate, auto‑repair rate, manual intervention rate; feed high‑frequency diff types back to payment chain for preventive improvements.

Monitoring and alerting – essential metrics

Task layer: bill download success rate, parsing latency, shard reconciliation duration, task failure & retry counts.

Data layer: daily bill count & amount, internal successful flow count & amount, total diff count & amount, diff type distribution.

Repair layer: auto‑repair success rate, average repair time, manual hand‑off ratio, P0 diff open duration.

Business layer: lost‑order rate, over‑payment rate, amount‑mismatch rate, per‑channel anomaly spikes.

Common architectural mistakes

Running full‑table joins on the core payment DB.

Only detecting diffs without a repair loop.

Reconciliation service directly modifying order or payment tables.

Missing CAS on diff status leading to race conditions.

Relying on transactional MQ as a silver bullet without proper idempotency.

Discarding raw channel files, making audits impossible.

Conclusion – reconciliation as a core consistency system

Payment reconciliation is not a simple script; it is the financial consistency backbone of a payment system. It proves that every cent collected is correctly recorded, settled, and reflected to the user, turning potential revenue loss into an automated correction.