AiDOPWarehouse/scripts/s8/sched-exec-regression.sh

#!/usr/bin/env bash
# S8-REGRESSION-FIXTURE-1 调度执行专用回归脚本（dev/test only, aidopdev）。
#
# 覆盖 4 类 case：
#   A. runtime schema sanity      — watch_rule 调度运行时列 + rule_detection_state 表/索引存在
#   B. scheduler execution sanity — demo rule 10 (DEMO_ORDER_DELIVERY_TIMEOUT) next_run_at=NOW
#                                  → 等待 Job tick → last_run_at/next_run_at 推进 / lock 三件套释放
#   C. antiflap trigger=3         — TEMP_SCHED_TIMEOUT_ANTIFLAP 三次 run-once → hit_count 1/2/3
#                                  → 第三次 CREATED；TEMP rule disabled、TEMP exception soft delete
#   D. detection log edge         — CREATED（C 已覆盖）/ REFRESHED（第四次 run-once）
#                                  / RECOVERED（高风险，SKIP，由专项脚本覆盖）
#
# 退出码：FAIL=0 → 0；FAIL>0 → 1；致命错误 → 2。

set -uo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=./s8-regression-common.sh
source "${SCRIPT_DIR}/s8-regression-common.sh"

auth_load

DEMO_TIMEOUT_RULE='DEMO_ORDER_DELIVERY_TIMEOUT'
DEMO_TIMEOUT_RULE_ID=10
TEMP_RULE='TEMP_SCHED_TIMEOUT_ANTIFLAP'
TEMP_DEDUP_OBJ='TEMP-ANTIFLAP-001'

baseline_before=$(read_baseline)
DEMO_SNAPSHOT=$(snapshot_demo_rule_state)
echo "==== sched-exec-regression  baseline_before=${baseline_before} demo_snapshot=${DEMO_SNAPSHOT} ===="

# ---------------------------------------------------------------------------
# A. runtime schema sanity
# ---------------------------------------------------------------------------
echo "---- A. runtime schema sanity ----"
required_cols=(next_run_at last_run_at last_status last_error last_duration_ms last_run_id \
               lock_token locked_by lock_until running_started_at consecutive_failure_count \
               paused_until pause_reason trigger_count_required recover_count_required)
for col in "${required_cols[@]}"; do
  cnt=$(mysql_run "SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='${DB_NAME}' AND TABLE_NAME='ado_s8_watch_rule' AND COLUMN_NAME='${col}';")
  if [[ "${cnt}" == "1" ]]; then
    record_pass "watch_rule.${col} exists"
  else
    record_fail "watch_rule.${col} missing"
  fi
done

state_table=$(mysql_run "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='${DB_NAME}' AND TABLE_NAME='ado_s8_rule_detection_state';")
if [[ "${state_table}" == "1" ]]; then
  record_pass "ado_s8_rule_detection_state table exists"
else
  record_fail "ado_s8_rule_detection_state table missing"
fi

state_idx=$(mysql_run "SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA='${DB_NAME}' AND TABLE_NAME='ado_s8_rule_detection_state' AND INDEX_NAME='uk_s8_rule_detection_state_dedup';")
if [[ "${state_idx}" -ge 1 ]]; then
  record_pass "uk_s8_rule_detection_state_dedup exists"
else
  record_fail "uk_s8_rule_detection_state_dedup missing"
fi

# ---------------------------------------------------------------------------
# B. scheduler execution sanity（demo rule 10）
#    策略：将 next_run_at 置为 NOW；轮询 last_run_at 是否推进，最多等 90s（Job 间隔 60s）。
#    注意：仅刷新 last_run_at/next_run_at/lock 等运行态字段，不改 enabled/paused/trigger/recover/params/expression。
# ---------------------------------------------------------------------------
echo "---- B. scheduler execution sanity ----"
if ! require_demo_rule "${DEMO_TIMEOUT_RULE}"; then
  record_skip "B: ${DEMO_TIMEOUT_RULE} not found or disabled — cannot validate scheduler execution"
else
  pre_last_run_at=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} last_run_at)
  echo "B pre-state: last_run_at=${pre_last_run_at}"
  mysql_run_strict "UPDATE ado_s8_watch_rule SET next_run_at=NOW(), lock_token=NULL, locked_by=NULL, lock_until=NULL, running_started_at=NULL WHERE id=${DEMO_TIMEOUT_RULE_ID};" >/dev/null
  echo "B nudged: next_run_at=NOW(); waiting Job tick (poll up to 90s)…"

  picked_up=0
  for ((i=1; i<=18; i++)); do
    sleep 5
    cur_last_run_at=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} last_run_at)
    if [[ "${cur_last_run_at}" != "${pre_last_run_at}" && "${cur_last_run_at}" != "NULL" ]]; then
      picked_up=1
      break
    fi
  done

  if (( picked_up == 1 )); then
    record_pass "B: rule ${DEMO_TIMEOUT_RULE_ID} picked up by Job (last_run_at advanced ${pre_last_run_at} -> ${cur_last_run_at})"
    last_status=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} last_status)
    last_run_id=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} last_run_id)
    last_duration_ms=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} last_duration_ms)
    next_run_at=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} next_run_at)
    lock_token=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} lock_token)
    locked_by=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} locked_by)
    lock_until=$(get_rule_field ${DEMO_TIMEOUT_RULE_ID} lock_until)
    echo "B post-state: last_status=${last_status} last_run_id=${last_run_id} last_duration_ms=${last_duration_ms} next_run_at=${next_run_at} lock_token=${lock_token} locked_by=${locked_by} lock_until=${lock_until}"

    if [[ "${last_status}" == "SUCCESS" ]]; then
      record_pass "B: last_status=SUCCESS"
    else
      record_fail "B: last_status expected SUCCESS, got ${last_status}"
    fi
    [[ "${last_run_id}" != "NULL" && -n "${last_run_id}" ]] && record_pass "B: last_run_id populated (${last_run_id})" || record_fail "B: last_run_id missing"
    [[ "${last_duration_ms}" != "NULL" && "${last_duration_ms}" != "0" ]] && record_pass "B: last_duration_ms=${last_duration_ms}" || record_fail "B: last_duration_ms missing/zero"
    [[ "${next_run_at}" != "NULL" ]] && record_pass "B: next_run_at advanced (${next_run_at})" || record_fail "B: next_run_at NULL"
    if [[ "${lock_token}" == "NULL" && "${locked_by}" == "NULL" && "${lock_until}" == "NULL" ]]; then
      record_pass "B: lock 三件套已释放"
    else
      record_fail "B: lock not released (lock_token=${lock_token} locked_by=${locked_by} lock_until=${lock_until})"
    fi
  else
    record_skip "B: Job did not pick up rule within 90s (poll timeout) — possibly Job paused or background load; not failing"
  fi
fi

# ---------------------------------------------------------------------------
# C. antiflap trigger=3 (TEMP fixture)
# ---------------------------------------------------------------------------
echo "---- C. antiflap trigger=3 (TEMP fixture) ----"

TEMP_PARAMS='{"dueAtField":"due_at","statusField":"status","completedStates":["CLOSED","DONE","COMPLETED"],"objectCodeField":"related_object_code","objectIdField":"source_object_id","graceMinutes":0,"exceptionTypeCode":"DELIVERY_DELAY"}'
TEMP_EXPR="SELECT '${TEMP_DEDUP_OBJ}' AS related_object_code, '${TEMP_DEDUP_OBJ}' AS source_object_id, DATE_SUB(NOW(), INTERVAL 1 HOUR) AS due_at, 'PENDING' AS status"

# 幂等 upsert，并清空 detection_state 残留 hit_count 以保证从 0 起。
existing=$(mysql_run "SELECT COUNT(*) FROM ado_s8_watch_rule WHERE rule_code='${TEMP_RULE}' AND tenant_id=${TENANT_ID} AND factory_id=${FACTORY_ID};")
if [[ "${existing}" == "0" ]]; then
  mysql_run_strict "INSERT INTO ado_s8_watch_rule
    (tenant_id, factory_id, rule_code, scene_code, data_source_id, watch_object_type, expression, severity,
     poll_interval_seconds, enabled, created_at, rule_type, source_object_type, params_json,
     consecutive_failure_count, trigger_count_required, recover_count_required, next_run_at)
    VALUES (${TENANT_ID}, ${FACTORY_ID}, '${TEMP_RULE}', 'S7', 1, 'ORDER',
     \"${TEMP_EXPR}\", 'HIGH', 60, 1, NOW(), 'TIMEOUT', 'ORDER', '${TEMP_PARAMS}',
     0, 3, 1, NULL);" >/dev/null
  echo "C: inserted TEMP rule ${TEMP_RULE}"
else
  mysql_run_strict "UPDATE ado_s8_watch_rule SET enabled=1, expression=\"${TEMP_EXPR}\", params_json='${TEMP_PARAMS}', trigger_count_required=3, recover_count_required=1, paused_until=NULL, pause_reason=NULL, lock_token=NULL, locked_by=NULL, lock_until=NULL, consecutive_failure_count=0, last_status=NULL, last_error=NULL, updated_at=NOW() WHERE rule_code='${TEMP_RULE}' AND tenant_id=${TENANT_ID} AND factory_id=${FACTORY_ID};" >/dev/null
  echo "C: re-armed TEMP rule ${TEMP_RULE}"
fi

TEMP_RULE_ID=$(get_rule_id_by_code "${TEMP_RULE}")
echo "C: TEMP_RULE_ID=${TEMP_RULE_ID}"

# 清理可能残留的 detection_state（防止此前测试遗留 hit_count > 0）
mysql_run_strict "DELETE FROM ado_s8_rule_detection_state WHERE rule_code='${TEMP_RULE}' AND tenant_id=${TENANT_ID} AND factory_id=${FACTORY_ID};" >/dev/null
# 清理可能残留的 TEMP exception（同 dedup_key 旧记录），避免被识别为已存在。Soft delete only.
mysql_run_strict "UPDATE ado_s8_exception SET is_deleted=1, updated_at=NOW() WHERE source_rule_code='${TEMP_RULE}' AND is_deleted=0;" >/dev/null

marker_c=$(mysql_run "SELECT NOW();")
sleep 1

# tick 1 / 2 / 3
for tick in 1 2 3; do
  resp=$(run_once_endpoint)
  if [[ -z "${resp}" ]]; then
    record_fail "C: tick${tick} run-once returned empty response"
    break
  fi
  hit_count=$(mysql_run "SELECT IFNULL(consecutive_hit_count,0) FROM ado_s8_rule_detection_state WHERE rule_code='${TEMP_RULE}' AND dedup_key='T${TENANT_ID}:F${FACTORY_ID}:R${TEMP_RULE}:ORDER:${TEMP_DEDUP_OBJ}';")
  active_excs=$(mysql_run "SELECT COUNT(*) FROM ado_s8_exception WHERE source_rule_code='${TEMP_RULE}' AND is_deleted=0 AND status<>'CLOSED';")
  echo "C tick${tick}: hit_count=${hit_count} active_excs=${active_excs}"
  case "${tick}" in
    1)
      if [[ "${hit_count}" == "1" && "${active_excs}" == "0" ]]; then
        record_pass "C tick1 pending: hit_count=1, no exception"
      else
        record_fail "C tick1 expected hit_count=1/active=0, got hit_count=${hit_count}/active=${active_excs}"
      fi
      ;;
    2)
      if [[ "${hit_count}" == "2" && "${active_excs}" == "0" ]]; then
        record_pass "C tick2 pending: hit_count=2, no exception"
      else
        record_fail "C tick2 expected hit_count=2/active=0, got hit_count=${hit_count}/active=${active_excs}"
      fi
      ;;
    3)
      if [[ "${hit_count}" -ge 3 && "${active_excs}" == "1" ]]; then
        record_pass "C tick3 fired: hit_count=${hit_count}, exception created"
      else
        record_fail "C tick3 expected hit_count>=3/active=1, got hit_count=${hit_count}/active=${active_excs}"
      fi
      ;;
  esac
  sleep 1
done

# CREATED detection_log
created_cnt=$(mysql_run "SELECT COUNT(*) FROM ado_s8_detection_log WHERE rule_code='${TEMP_RULE}' AND detect_result='CREATED' AND detected_at >= '${marker_c}';")
[[ "${created_cnt}" -ge 1 ]] && record_pass "D: CREATED detection_log present (${created_cnt})" || record_fail "D: CREATED detection_log missing"

# state.active_exception_id 写入校验
state_active_excid=$(mysql_run "SELECT IFNULL(active_exception_id,0) FROM ado_s8_rule_detection_state WHERE rule_code='${TEMP_RULE}' AND dedup_key='T${TENANT_ID}:F${FACTORY_ID}:R${TEMP_RULE}:ORDER:${TEMP_DEDUP_OBJ}';")
[[ "${state_active_excid}" != "0" ]] && record_pass "C: state.active_exception_id=${state_active_excid}" || record_fail "C: state.active_exception_id not set"

# ---------------------------------------------------------------------------
# D. detection log edge — REFRESHED（CREATED 已在 C 覆盖；RECOVERED 风险高，SKIP）
# ---------------------------------------------------------------------------
echo "---- D. detection log edge ----"
sleep 1
marker_d=$(mysql_run "SELECT NOW();")
sleep 1
run_once_endpoint >/dev/null
refreshed_cnt=$(mysql_run "SELECT COUNT(*) FROM ado_s8_detection_log WHERE rule_code='${TEMP_RULE}' AND detect_result='REFRESHED' AND detected_at >= '${marker_d}';")
if [[ "${refreshed_cnt}" -ge 1 ]]; then
  record_pass "D: REFRESHED detection_log present (${refreshed_cnt})"
else
  # 可能是命中后 refresh 路径走 last_detected_at 更新但不写 REFRESHED log；放宽到 last_detected_at 推进。
  ts=$(mysql_run "SELECT IFNULL(MAX(last_detected_at),'NULL') FROM ado_s8_exception WHERE source_rule_code='${TEMP_RULE}' AND is_deleted=0;")
  if [[ "${ts}" > "${marker_d}" ]]; then
    record_pass "D: last_detected_at advanced past marker_d (${ts} > ${marker_d})"
  else
    record_fail "D: REFRESHED log absent and last_detected_at did not advance (ts=${ts}, marker=${marker_d})"
  fi
fi

record_skip "D: RECOVERED edge — TEMP fixture 翻转风险高（需要业务态切换），由专项脚本覆盖"

# ---------------------------------------------------------------------------
# 收尾：TEMP fixture 关闭 + TEMP exception soft delete
# ---------------------------------------------------------------------------
echo "---- TEMP fixture cleanup ----"
mysql_run_strict "UPDATE ado_s8_watch_rule SET enabled=0, paused_until=NULL, pause_reason=NULL, lock_token=NULL, locked_by=NULL, lock_until=NULL, updated_at=NOW() WHERE rule_code='${TEMP_RULE}' AND tenant_id=${TENANT_ID} AND factory_id=${FACTORY_ID};" >/dev/null
TEMP_EXC_IDS=$(mysql_run "SELECT GROUP_CONCAT(id) FROM ado_s8_exception WHERE source_rule_code='${TEMP_RULE}' AND is_deleted=0;")
mysql_run_strict "UPDATE ado_s8_exception SET is_deleted=1, updated_at=NOW() WHERE source_rule_code='${TEMP_RULE}' AND is_deleted=0;" >/dev/null
cleanup_temp_sched_approval_ghost_tasks
echo "TEMP cleanup: rule_id=${TEMP_RULE_ID} disabled; soft-deleted exception ids=[${TEMP_EXC_IDS:-none}]"

# 确认 TEMP rule 不再 enabled，且默认列表（is_deleted=0）不可见。
final_temp_enabled=$(get_rule_field "${TEMP_RULE_ID}" enabled)
final_temp_visible=$(mysql_run "SELECT COUNT(*) FROM ado_s8_exception WHERE source_rule_code='${TEMP_RULE}' AND is_deleted=0;")
[[ "${final_temp_enabled}" == "0" ]] && record_pass "cleanup: TEMP rule enabled=0" || record_fail "cleanup: TEMP rule still enabled=${final_temp_enabled}"
[[ "${final_temp_visible}" == "0" ]] && record_pass "cleanup: TEMP exception 不出现在默认列表" || record_fail "cleanup: TEMP exception 仍在默认列表 (${final_temp_visible})"

# ---------------------------------------------------------------------------
# demo rule 10/11/12 漂移检测（不依赖 enabled=1 假设；driver arm 后 child 看到 1，
# 单跑看到 0，两种场景都只检查"未漂移"）
# ---------------------------------------------------------------------------
echo "---- demo rule 10/11/12 漂移检测 ----"
assert_demo_rule_state_unchanged "${DEMO_SNAPSHOT}"

# ---------------------------------------------------------------------------
# baseline 守恒
# ---------------------------------------------------------------------------
echo "---- baseline assert ----"
assert_baseline_unchanged "${baseline_before}"

# ---------------------------------------------------------------------------
# summary
# ---------------------------------------------------------------------------
echo
print_summary
echo "TEMP_RULE=${TEMP_RULE} TEMP_RULE_ID=${TEMP_RULE_ID} TEMP_EXC_IDS=[${TEMP_EXC_IDS:-none}]"
exit_by_summary