#!/usr/bin/env python3
"""U87 公开入口：多租户配额、代际租约与 checkpoint/业务提交，真实 MySQL + Spark。

前置：本机 MySQL（DBngin，独有库 study_u87_<runid>）、JDK 21、Spark 4.0.1
发行包（复用 U86 缓存，首次自动下载并校验 SHA-512）、venv（pymysql）。

场景：
  1. 配额裁决：两进程并发扣减同一租户令牌，超限拒绝且不扣减；他租户不受影响。
  2. 代际租约：gen1 持有者被 SIGKILL → 过期 → gen2 接管 → 旧代际写入被拒。
  3. checkpoint vs 业务提交：Spark Structured Streaming 重放，幂等写。
  4. 组合：配额超限时任务被拒且无业务行；恢复配额后成功。

结束时停进程、DROP 数据库、清理孤儿库（runtime 目录保留作为证据日志）；清理不完整使退出码非 0。
只有 exit 0 才把证据复制回仓库 evidence/。
"""
import glob
import hashlib
import json
import os
import pathlib
import platform
import shutil
import signal
import subprocess
import sys
import tarfile
import time
import traceback
import urllib.request
from datetime import datetime

LAB = pathlib.Path(__file__).resolve().parent
CACHE = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u87'
SPARK_CACHE = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u86'
SPARK_DIR = SPARK_CACHE / 'spark-4.0.1-bin-hadoop3'
SPARK_TGZ = SPARK_CACHE / 'downloads' / 'spark-4.0.1-bin-hadoop3.tgz'
SPARK_URL = 'https://archive.apache.org/dist/spark/spark-4.0.1/spark-4.0.1-bin-hadoop3.tgz'
SPARK_SHA512 = ('9198602c6b931b46686f32a25793b3bb58b522cd98a5b6a94d2484bae32e3e7b520d60'
                'f4bffe72ba29ff5c9ecd862443841ee47dde0f2f9e1bf52539f7baef41')
VENV = SPARK_CACHE / 'venv'
EVIDENCE = LAB / 'evidence' / 'u87-2026-10-28.json'

run_id = datetime.now().strftime('%H%M%S') + hashlib.sha1(str(time.time_ns()).encode()).hexdigest()[:4]
DB = f'study_u87_{run_id.lower()}'
RUNTIME = CACHE / 'runtime' / f'run-{run_id}'
MYSQL = {'bin': None, 'host': os.environ.get('MYSQL_HOST', '127.0.0.1'),
         'port': os.environ.get('MYSQL_PORT', '3306'),
         'user': os.environ.get('MYSQL_USER', 'root'),
         'password': os.environ.get('MYSQL_PASSWORD', '')}
JAVA_HOME = subprocess.check_output(['/usr/libexec/java_home', '-v', '21'], text=True).strip()
SPARK_SUBMIT = None
procs = []
observations = []
scenarios = []
errors = []
started_all = datetime.now().astimezone().isoformat()


class Blocked(SystemExit):
    pass


def say(line):
    print(line, flush=True)
    observations.append(line)


def phase(name):
    print(f'\n===== {name} =====', flush=True)


def record(name, observed, started):
    scenarios.append({'name': name, 'startedAt': started,
                      'finishedAt': datetime.now().astimezone().isoformat(),
                      'observed': observed})


def die(message):
    raise Blocked(f'U87 BLOCKED/FAILED: {message}')


def assert_true(cond, message):
    if not cond:
        raise AssertionError(message)


def sha256_file(path):
    digest = hashlib.sha256()
    with open(path, 'rb') as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b''):
            digest.update(chunk)
    return digest.hexdigest()


def sha512_file(path):
    digest = hashlib.sha512()
    with open(path, 'rb') as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b''):
            digest.update(chunk)
    return digest.hexdigest()


def mysql_bin():
    for candidate in filter(None, [os.environ.get('MYSQL_BIN'), shutil.which('mysql'),
                                   *sorted(glob.glob('/Users/Shared/DBngin/mysql/*/bin/mysql')),
                                   '/opt/homebrew/bin/mysql', '/usr/local/bin/mysql']):
        if pathlib.Path(candidate).is_file():
            return candidate
    die('未找到 mysql 客户端：设置 MYSQL_BIN')


def mysql(sql, db=None, timeout=180):
    args = [MYSQL['bin'], '-h', MYSQL['host'], '-P', MYSQL['port'], '-u', MYSQL['user'],
            '--batch', '--default-character-set=utf8mb4']
    if MYSQL['password']:
        args.insert(1, f"-p{MYSQL['password']}")
    if db:
        args.append(db)
    args += ['-e', sql]
    done = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
    if done.returncode != 0:
        raise RuntimeError(f'mysql failed: {sql[:120]} → {done.stderr.strip()[:300]}')
    return done.stdout


def sql_rows(sql):
    out = mysql(sql, db=DB).strip().splitlines()
    return out[1:] if out else []


def py(*args, **kwargs):
    return subprocess.run([str(VENV / 'bin' / 'python'), *args], **kwargs)


def ensure_spark():
    if SPARK_DIR.is_dir() and (SPARK_DIR / 'bin' / 'spark-submit').is_file():
        return 'cached'
    SPARK_TGZ.parent.mkdir(parents=True, exist_ok=True)
    if not SPARK_TGZ.is_file() or sha512_file(SPARK_TGZ) != SPARK_SHA512:
        tmp = SPARK_TGZ.with_suffix('.part')
        for attempt in range(5):
            try:
                urllib.request.urlretrieve(SPARK_URL, tmp)
                break
            except Exception as exc:  # noqa: BLE001
                say(f'U87 download retry {attempt + 1}: {exc}')
                time.sleep(5)
        if sha512_file(tmp) != SPARK_SHA512:
            die('Spark 发行包 SHA-512 校验失败')
        tmp.rename(SPARK_TGZ)
    with tarfile.open(SPARK_TGZ) as tar:
        tar.extractall(SPARK_CACHE)
    return 'extracted'


def ensure_venv():
    if not (VENV / 'bin' / 'python').is_file():
        subprocess.run([sys.executable, '-m', 'venv', str(VENV)], check=True)
    done = py('-c', 'import pymysql, pyspark; print("ok")', capture_output=True, text=True)
    if done.returncode != 0:
        py('-m', 'pip', 'install', '--quiet', 'pymysql==1.1.2', 'pyspark==4.0.1', check=True)


def spark_env():
    env = dict(os.environ, JAVA_HOME=JAVA_HOME)
    env['SPARK_LOCAL_DIRS'] = str(RUNTIME / 'spark-local')
    site = py('-c', 'import site; print(site.getsitepackages()[0])',
              capture_output=True, text=True, check=True).stdout.strip()
    env['PYTHONPATH'] = site + os.pathsep + env.get('PYTHONPATH', '')
    env['PYSPARK_PYTHON'] = str(VENV / 'bin' / 'python')
    env['PYSPARK_DRIVER_PYTHON'] = str(VENV / 'bin' / 'python')
    return env


def kill_group(proc, sig, wait):
    if proc is None or proc.poll() is not None:
        return
    try:
        os.killpg(proc.pid, sig)
    except ProcessLookupError:
        pass
    try:
        proc.wait(timeout=wait)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        proc.wait(timeout=10)


def spawn(name, code):
    script = RUNTIME / 'workers' / f'{name}.py'
    script.parent.mkdir(parents=True, exist_ok=True)
    script.write_text(code, encoding='utf-8')
    log = open(RUNTIME / f'{name}.log', 'w')
    proc = subprocess.Popen([str(VENV / 'bin' / 'python'), str(script)],
                            stdout=log, stderr=subprocess.STDOUT,
                            env=dict(os.environ, PYTHONPATH=str(VENV / 'lib' / 'python3.9' / 'site-packages')),
                            start_new_session=True)
    procs.append(proc)
    return proc, script, log


def wait_row(sql, predicate, description, timeout=60, interval=0.5):
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        rows = sql_rows(sql)
        last = rows
        if predicate(rows):
            return rows
        time.sleep(interval)
    die(f'超时等待 {description}（{timeout}s），最后: {last}')


COMMON = f'''
import json, os, signal, sys, time
import pymysql

DB = {DB!r}
CONN = dict(host={MYSQL['host']!r}, port={MYSQL['port']}, user={MYSQL['user']!r},
            password={MYSQL['password']!r}, database=DB, charset='utf8mb4',
            autocommit=False)


def conn():
    return pymysql.connect(**CONN)


def emit(line):
    print('U87OBS ' + line, flush=True)
'''


def scenario_quota():
    phase('场景1：租户配额令牌桶，并发扣减与超限拒绝')
    started = datetime.now().astimezone().isoformat()
    mysql('INSERT INTO tenant_quota (tenant, tokens, updated_at) '
          f"VALUES ('tenant-a', 5, NOW()), ('tenant-b', 5, NOW())", db=DB)
    worker = COMMON + '''
from concurrent.futures import ThreadPoolExecutor

def consume(tenant, tag):
    connection = conn()
    try:
        with connection.cursor() as cur:
            cur.execute('UPDATE tenant_quota SET tokens = tokens - 1, updated_at = NOW() '
                        'WHERE tenant = %s AND tokens > 0', (tenant,))
            affected = cur.rowcount
        connection.commit()
    finally:
        connection.close()
    return tag, affected, tenant

tasks = [(tenant, f'{tenant}-{i}') for tenant in ('tenant-a', 'tenant-b') for i in range(3)]
results = []
with ThreadPoolExecutor(max_workers=6) as pool:
    for tag, affected, tenant in pool.map(lambda t: consume(*t), tasks):
        results.append((tag, affected))
a_ok = sum(1 for tag, affected in results if tag.startswith('tenant-a') and affected == 1)
b_ok = sum(1 for tag, affected in results if tag.startswith('tenant-b') and affected == 1)
emit('quotaRequests tenantA=' + str(a_ok) + ' tenantB=' + str(b_ok)
     + ' total=' + str(len(results)))
'''
    proc, script, log = spawn('quota', worker)
    proc.wait(timeout=120)
    log.close()
    text = (RUNTIME / 'quota.log').read_text()
    line = next((l for l in text.splitlines() if l.startswith('U87OBS quotaRequests')), None)
    assert_true(line, '应有 quotaRequests 观察行')
    rows = sql_rows("SELECT tenant, tokens FROM tenant_quota ORDER BY tenant")
    tokens = {row.split('\t')[0]: int(row.split('\t')[1]) for row in rows}
    observed = [line, f'U87 quotaRemaining tenantA={tokens.get("tenant-a")} '
                      f'tenantB={tokens.get("tenant-b")} requestedEach=3']
    # 两个租户各请求 3 次，初始 5 → 各自剩余 2（不多扣，也不会扣成负数）
    assert_true(tokens.get('tenant-a') == 2 and tokens.get('tenant-b') == 2,
                '两租户都应剩余 2 个令牌（共 3 次成功）')
    record('quota-token-bucket', observed, started)


def scenario_lease_generation():
    phase('场景2：代际租约，SIGKILL→过期→接管→旧代际被拒')
    started = datetime.now().astimezone().isoformat()
    mysql("INSERT INTO lease_state (name, owner, generation, expires_at) "
          "VALUES ('scheduler', NULL, 0, NOW())", db=DB)
    holder = COMMON + '''
import signal

connection = conn()
with connection.cursor() as cur:
    cur.execute("UPDATE lease_state SET owner = 'worker-a', generation = generation + 1, "
                "expires_at = DATE_ADD(NOW(), INTERVAL 3 SECOND) "
                "WHERE name = 'scheduler' AND (owner IS NULL OR expires_at < NOW())")
    acquired = cur.rowcount
    connection.commit()
    cur.execute("SELECT generation FROM lease_state WHERE name = 'scheduler'")
    generation = cur.fetchone()[0]
connection.close()
emit('leaseAcquire holder=worker-a affected=' + str(acquired) + ' generation=' + str(generation))
if acquired != 1:
    sys.exit(2)
time.sleep(300)
'''
    proc_a, script_a, log_a = spawn('lease-a', holder)
    wait_row("SELECT generation, IFNULL(owner,'-') FROM lease_state WHERE name='scheduler'",
             lambda rows: rows and rows[0].split('\t')[1] == 'worker-a', 'worker-a 取得租约', 30)
    gen_a = int(sql_rows("SELECT generation FROM lease_state WHERE name='scheduler'")[0])
    kill_group(proc_a, signal.SIGKILL, 10)
    log_a.close()
    observed = [f'U87 leaseAcquire holder=worker-a generation={gen_a}',
                'U87 leaseHolderKilled signal=SIGKILL']

    # 等租约过期（3 秒）后由**同名** worker-a 重启并在新代际接管——
    # owner 无法区分新旧，只有 generation 能区分；再加一条只按 owner 的计数作为对照
    time.sleep(4)
    taker = COMMON + f'''
worker_a_generation = {gen_a}
connection = conn()
with connection.cursor() as cur:
    cur.execute("UPDATE lease_state SET owner = 'worker-a', generation = generation + 1, "
                "expires_at = DATE_ADD(NOW(), INTERVAL 60 SECOND) "
                "WHERE name = 'scheduler' AND expires_at < NOW()")
    took = cur.rowcount
    connection.commit()
    cur.execute("SELECT generation, owner FROM lease_state WHERE name = 'scheduler'")
    generation, owner = cur.fetchone()
connection.close()
emit('leaseTakeover owner=' + owner + ' affected=' + str(took) + ' generation=' + str(generation))

# 对照：只按 owner 匹配的行数（同名接管后仍为 1——owner 无法区分新旧）
connection = conn()
with connection.cursor() as cur:
    cur.execute("SELECT COUNT(*) FROM lease_state WHERE name = 'scheduler' AND owner = 'worker-a'")
    owner_matches = cur.fetchone()[0]
connection.close()
emit('leaseOwnerOnlyMatches owner=worker-a count=' + str(owner_matches))

# 旧代际的迟到写入必须被拒：同名 + 旧 generation 的条件更新应影响 0 行
connection = conn()
with connection.cursor() as cur:
    cur.execute("UPDATE lease_state SET expires_at = DATE_ADD(NOW(), INTERVAL 60 SECOND) "
                "WHERE name = 'scheduler' AND owner = 'worker-a' AND generation = %s",
                (worker_a_generation,))
    stale = cur.rowcount
    connection.commit()
connection.close()
emit('leaseStaleWrite owner=worker-a generation=' + str(worker_a_generation)
     + ' affected=' + str(stale))
'''
    proc_b, script_b, log_b = spawn('lease-b', taker)
    proc_b.wait(timeout=120)
    log_b.close()
    text = (RUNTIME / 'lease-b.log').read_text()
    take_line = next((l for l in text.splitlines() if l.startswith('U87OBS leaseTakeover')), None)
    owner_line = next((l for l in text.splitlines() if l.startswith('U87OBS leaseOwnerOnlyMatches')), None)
    stale_line = next((l for l in text.splitlines() if l.startswith('U87OBS leaseStaleWrite')), None)
    assert_true(take_line and 'affected=1' in take_line, '同名重启应成功接管')
    assert_true(owner_line and 'count=1' in owner_line,
                '只按 owner 匹配应命中 1 行（owner 无法区分新旧）')
    assert_true(stale_line and 'affected=0' in stale_line,
                '同名旧代际的写入必须影响 0 行（generation 是唯一区分）')
    rows = sql_rows("SELECT owner, generation FROM lease_state WHERE name='scheduler'")
    owner, generation = rows[0].split('\t')
    observed += [take_line, owner_line, stale_line,
                 f'U87 leaseFinal owner={owner} generation={generation}']
    assert_true(owner == 'worker-a' and int(generation) == gen_a + 1,
                '最终持有者应为同名重启的 worker-a 且代际 +1')
    record('lease-generation', observed, started)


def scenario_checkpoint():
    phase('场景3：checkpoint 与业务提交：重放与幂等（Spark Structured Streaming）')
    started = datetime.now().astimezone().isoformat()
    input_dir = RUNTIME / 'stream-input'
    checkpoint_dir = RUNTIME / 'stream-checkpoint'
    ready_dir = RUNTIME / 'stream-ready'
    for path in (input_dir, checkpoint_dir, ready_dir):
        path.mkdir(parents=True, exist_ok=True)
    mysql('CREATE TABLE business_orders (tenant VARCHAR(32) NOT NULL, '
          'order_id VARCHAR(32) NOT NULL, amount INT NOT NULL, PRIMARY KEY (tenant, order_id))',
          db=DB)
    (input_dir / 'batch1.json').write_text('\n'.join(
        '{"tenant": "tenant-a", "order_id": "o-%02d", "amount": %d}' % (i, 10 + i)
        for i in range(5)) + '\n')
    task = RUNTIME / 'stream_task.py'
    task.write_text(f'''
import os, sys
import pymysql
from pyspark.sql import SparkSession

DB = {DB!r}
CONN = dict(host={MYSQL['host']!r}, port={MYSQL['port']}, user={MYSQL['user']!r},
            password={MYSQL['password']!r}, database=DB, charset='utf8mb4', autocommit=False)
INPUT = {str(input_dir)!r}
CHECKPOINT = {str(checkpoint_dir)!r}
READY = {str(ready_dir)!r}
FAIL_ONCE = os.environ.get('FAIL_ONCE') == '1'

spark = (SparkSession.builder.appName('study-u87-stream').master('local[2]')
         .config('spark.sql.shuffle.partitions', '2').getOrCreate())


def write_batch(batch_df, epoch_id):
    rows = batch_df.collect()
    print('U87BATCH epoch=%d rows=%d' % (epoch_id, len(rows)), flush=True)
    if FAIL_ONCE and epoch_id == 1:
        print('U87BATCH business failure injected at epoch %d' % epoch_id, flush=True)
        raise RuntimeError('injected business write failure')
    connection = pymysql.connect(**CONN)
    try:
        with connection.cursor() as cur:
            for row in rows:
                cur.execute('INSERT IGNORE INTO business_orders (tenant, order_id, amount) '
                            'VALUES (%s, %s, %s)', (row.tenant, row.order_id, row.amount))
        connection.commit()
    finally:
        connection.close()


stream = (spark.readStream.schema('tenant STRING, order_id STRING, amount INT')
          .json(INPUT))
query = (stream.writeStream
         .foreachBatch(write_batch)
         .option('checkpointLocation', CHECKPOINT)
         .trigger(availableNow=True)
         .start())
query.awaitTermination()
spark.stop()
''', encoding='utf-8')

    def run_stream(tag, fail_once):
        env = spark_env()
        env['FAIL_ONCE'] = '1' if fail_once else '0'
        log = open(RUNTIME / f'stream-{tag}.log', 'w')
        proc = subprocess.Popen([str(SPARK_SUBMIT), str(task)], stdout=log,
                                stderr=subprocess.STDOUT, env=env, start_new_session=True)
        procs.append(proc)
        proc.wait(timeout=420)
        log.close()
        return proc.returncode

    code1 = run_stream('first', fail_once=False)
    rows1 = sql_rows('SELECT COUNT(*) FROM business_orders')
    observed = [f'U87 checkpointFirstRun exit={code1} businessRows={rows1[0]}']
    assert_true(code1 == 0 and rows1[0] == '5', '第一批应写入 5 行业务')

    (input_dir / 'batch2.json').write_text('\n'.join(
        '{"tenant": "tenant-a", "order_id": "o-%02d", "amount": %d}' % (i, 20 + i)
        for i in range(5, 8)) + '\n')
    code2 = run_stream('fail', fail_once=True)
    rows2 = sql_rows('SELECT COUNT(*) FROM business_orders')
    observed += [f'U87 checkpointFailedBusiness exit={code2} businessRows={rows2[0]}',
                 'U87 checkpointConclusion 引擎可先失败；checkpoint 是否推进要看下一次运行实际读取的批次']

    assert_true(code2 != 0, '第二批业务写失败时 spark-submit 应以非零退出')
    assert_true(rows2[0] == '5', '失败批不应写入业务行')
    code3 = run_stream('replay', fail_once=False)
    rows3 = sql_rows('SELECT COUNT(*) FROM business_orders')
    replay_log = (RUNTIME / 'stream-replay.log').read_text()
    observed += [f'U87 checkpointReplay exit={code3} businessRows={rows3[0]}',
                 f'U87 checkpointReplayBatches epoch1Reprocessed='
                 f'{"U87BATCH epoch=1" in replay_log}']
    assert_true(code3 == 0, '重放运行应成功')
    assert_true(int(rows3[0]) == 8, '重放后业务行应为 8（5+3）')
    assert_true('U87BATCH epoch=1' in replay_log, '重放应再次处理失败批次 epoch=1')

    # 真正触发重复投递：删掉 commit log 中 batch 1 的提交记录（模拟"业务已写、提交记录丢失"），
    # 重启后引擎会重放同一批；唯一键 INSERT IGNORE 应使业务行数保持不变。
    commits_dir = checkpoint_dir / 'commits'
    batch1_commit = commits_dir / '1'
    assert_true(batch1_commit.is_file(), 'checkpoint 应有 batch 1 的提交记录')
    batch1_commit.unlink()
    code4 = run_stream('replay2', fail_once=False)
    rows5 = sql_rows('SELECT COUNT(*) FROM business_orders')
    replay2_log = (RUNTIME / 'stream-replay2.log').read_text()
    tamper_guard = 'Multiple streaming queries are concurrently using' in replay2_log
    observed.append(f'U87 checkpointReplayAgain exit={code4} businessRows={rows5[0]} '
                    f'epoch1Reprocessed={"U87BATCH epoch=1" in replay2_log} '
                    f'tamperGuardFired={tamper_guard}')
    # 删除提交记录是一种"篡改 checkpoint"：Spark 4 会重放 batch 1（真实重复投递），
    # 随后在提交时检测到日志不一致并以 STREAM_FAILED 拒绝继续。业务侧幂等使行数保持 8。
    assert_true('U87BATCH epoch=1' in replay2_log,
                '删除提交记录后引擎应真正重放 batch 1（重复投递）')
    assert_true(int(rows5[0]) == 8, '重复投递同一批后业务行仍应为 8（唯一键幂等）')
    assert_true(code4 != 0 and tamper_guard,
                '篡改 commit log 后 Spark 4 应以 STREAM_FAILED 拒绝继续（保护而非幂等失败）')
    record('checkpoint-vs-business', observed, started)


def scenario_quota_blocks_then_recovers():
    phase('场景4：配额超限拒绝任务且无业务行；恢复配额后成功')
    started = datetime.now().astimezone().isoformat()
    mysql("INSERT IGNORE INTO tenant_quota (tenant, tokens, updated_at) "
          f"VALUES ('tenant-c', 0, NOW())", db=DB)
    mysql('CREATE TABLE IF NOT EXISTS tenant_jobs (job_id VARCHAR(64) PRIMARY KEY, '
          'tenant VARCHAR(32) NOT NULL, status VARCHAR(16) NOT NULL)', db=DB)
    mysql('CREATE TABLE IF NOT EXISTS quota_business (tenant VARCHAR(32) NOT NULL, '
          'job_id VARCHAR(64) NOT NULL, PRIMARY KEY (tenant, job_id))', db=DB)
    worker = COMMON + '''
job = 'job-c-1'
connection = conn()
with connection.cursor() as cur:
    cur.execute('UPDATE tenant_quota SET tokens = tokens - 1 WHERE tenant = %s AND tokens > 0',
                ('tenant-c',))
    allowed = cur.rowcount
    if allowed:
        cur.execute('INSERT IGNORE INTO quota_business (tenant, job_id) VALUES (%s, %s)',
                    ('tenant-c', job))
    cur.execute('INSERT INTO tenant_jobs (job_id, tenant, status) VALUES (%s, %s, %s)',
                (job, 'tenant-c', 'ACCEPTED' if allowed else 'REJECTED'))
connection.commit()
connection.close()
emit('quotaGate allowed=' + str(allowed))
'''
    proc, script, log = spawn('quota-gate', worker)
    proc.wait(timeout=60)
    log.close()
    status1 = sql_rows("SELECT status FROM tenant_jobs WHERE job_id='job-c-1'")[0]
    tokens_block = sql_rows("SELECT tokens FROM tenant_quota WHERE tenant='tenant-c'")[0]
    assert_true(status1 == 'REJECTED', '配额为 0 时应拒绝任务')
    assert_true(tokens_block == '0', '拒绝时令牌不应被扣减')
    mysql("UPDATE tenant_quota SET tokens = 3, updated_at = NOW() WHERE tenant='tenant-c'", db=DB)
    worker2 = worker.replace("job = 'job-c-1'", "job = 'job-c-2'")
    proc2, script2, log2 = spawn('quota-gate-2', worker2)
    proc2.wait(timeout=60)
    log2.close()
    status2 = sql_rows("SELECT status FROM tenant_jobs WHERE job_id='job-c-2'")[0]
    tokens = sql_rows("SELECT tokens FROM tenant_quota WHERE tenant='tenant-c'")[0]
    business_blocked = sql_rows("SELECT COUNT(*) FROM quota_business WHERE job_id='job-c-1'")[0]
    business_ok = sql_rows("SELECT COUNT(*) FROM quota_business WHERE job_id='job-c-2'")[0]
    observed = [f'U87 quotaGate blockedStatus={status1} tokensAtBlock={tokens_block} '
                f'businessRowsBlocked={business_blocked}',
                f'U87 quotaGate recoveredStatus={status2} tokensAfter={tokens} '
                f'businessRowsAccepted={business_ok}']
    assert_true(status2 == 'ACCEPTED', '恢复配额后任务应被接受')
    assert_true(int(tokens) == 2, '恢复后应扣减一个令牌')
    assert_true(business_blocked == '0', '被拒绝的任务不应产生业务行')
    assert_true(business_ok == '1', '被接受的任务应产生 1 行业务')
    record('quota-block-and-recover', observed, started)


def cleanup():
    phase('清理')
    for proc in procs:
        kill_group(proc, signal.SIGTERM, 5)
    for proc in procs:
        if proc.poll() is None:
            kill_group(proc, signal.SIGKILL, 10)
    survivors = [p.pid for p in procs if p.poll() is None]
    dropped = False
    drop_err = None
    try:
        mysql(f'DROP DATABASE IF EXISTS `{DB}`')
        dropped = True
    except Exception as exc:  # noqa: BLE001
        drop_err = str(exc)
    orphan_rows = [row for row in
                   mysql("SHOW DATABASES LIKE 'study\\\\_u87\\\\_%'").strip().splitlines()[1:]
                   if row]
    orphans_dropped = []
    for row in orphan_rows:
        name = row.strip()
        try:
            mysql(f'DROP DATABASE IF EXISTS `{name}`')
            orphans_dropped.append(name)
        except Exception as exc:  # noqa: BLE001
            errors.append(f'孤儿库清理失败 {name}: {exc}')
    remaining = [r for r in mysql("SHOW DATABASES LIKE 'study\\\\_u87\\\\_%'").strip().splitlines()[1:] if r]
    result = {'processesSurviving': survivors, 'databaseDropped': dropped,
              'dropError': drop_err, 'orphanDatabasesDropped': orphans_dropped,
              'labDatabasesRemaining': remaining, 'runtimeDir': str(RUNTIME)}
    if survivors or not dropped or remaining:
        errors.append(f'清理不完整: {result}')
    return result


def run():
    global SPARK_SUBMIT
    exit_code = 0
    failure = None
    environment = {}
    cleanup_result = None
    try:
        MYSQL['bin'] = mysql_bin()
        RUNTIME.mkdir(parents=True, exist_ok=True)
        mysql_version = mysql('SELECT VERSION()').splitlines()[1]
        mysql(f'CREATE DATABASE IF NOT EXISTS `{DB}` DEFAULT CHARACTER SET utf8mb4')
        mysql(f'CREATE TABLE {DB}.tenant_quota (tenant VARCHAR(32) PRIMARY KEY, '
              'tokens INT NOT NULL, updated_at DATETIME NOT NULL)')
        mysql(f'CREATE TABLE {DB}.lease_state (name VARCHAR(32) PRIMARY KEY, '
              'owner VARCHAR(32) NULL, generation INT NOT NULL, expires_at DATETIME NOT NULL)')
        spark_state = ensure_spark()
        ensure_venv()
        SPARK_SUBMIT = str(SPARK_DIR / 'bin' / 'spark-submit')
        environment = {
            'os': platform.platform(), 'arch': platform.machine(),
            'jdk': subprocess.run([f'{JAVA_HOME}/bin/java', '-version'],
                                  capture_output=True, text=True).stderr.splitlines()[0],
            'mysql': mysql_version,
            'spark': next((line.strip() for line in
                           (subprocess.run([SPARK_SUBMIT, '--version'], capture_output=True,
                                           text=True, env=dict(os.environ, JAVA_HOME=JAVA_HOME))
                            .stdout + subprocess.run([SPARK_SUBMIT, '--version'],
                                                     capture_output=True, text=True,
                                                     env=dict(os.environ, JAVA_HOME=JAVA_HOME))
                            .stderr).splitlines() if 'version' in line.lower()), None)
                     if SPARK_SUBMIT else None,
            'sparkTarSha512Prefix': SPARK_SHA512[:16], 'sparkState': spark_state,
            'pymysql': next((item.name.split('-')[1].replace('.dist', '') for item in
                             (VENV / 'lib' / 'python3.9' / 'site-packages').glob('[Pp]y[Mm][Yy][Ss][Qq][Ll]-*.dist-info')),
                            'unknown'),
            'externalServices': f'MySQL {MYSQL["host"]}:{MYSQL["port"]} 独有库 {DB}；'
                                'Spark local[2] 本机 Structured Streaming；无集群',
        }
        say(f'U87 preflight db={DB} mysql={mysql_version}')

        scenario_quota()
        scenario_lease_generation()
        scenario_checkpoint()
        scenario_quota_blocks_then_recovers()
        say('U87 scenarios completed')
    except SystemExit as exc:
        exit_code = 1
        failure = str(exc)
        errors.append(failure)
        print(failure, flush=True)
    except Exception as exc:  # noqa: BLE001
        exit_code = 1
        failure = f'{type(exc).__name__}: {exc}'
        errors.append(failure)
        traceback.print_exc()
    finally:
        try:
            cleanup_result = cleanup()
        except Exception as cleanup_exc:  # noqa: BLE001
            errors.append(f'清理异常: {cleanup_exc}')
            cleanup_result = {'error': str(cleanup_exc)}
            exit_code = exit_code or 1
        if cleanup_result and (cleanup_result.get('processesSurviving') or
                               not cleanup_result.get('databaseDropped') or
                               cleanup_result.get('labDatabasesRemaining')):
            exit_code = exit_code or 1

    evidence = {
        'ticket': 'U87 #124',
        'generatedAt': datetime.now().astimezone().isoformat(),
        'startedAt': started_all, 'runId': run_id,
        'input': 'python3 labs/data-tenant-quota/run.py',
        'environment': environment, 'scenarios': scenarios, 'observed': observations,
        'exitCode': exit_code, 'failure': failure, 'errors': errors,
        'rawLogs': str(RUNTIME), 'cleanup': cleanup_result,
        'runnerSha256': sha256_file(pathlib.Path(__file__).resolve()),
        'limits': [
            '租户配额在 MySQL 行锁上实现（UPDATE ... WHERE tokens > 0），不是独立配额服务；'
            '并发为 6 个线程，未做高并发压测。',
            '代际租约用 expires_at 列 + 条件更新实现；SIGKILL 是真实注入，未覆盖时钟回拨与多机时钟偏差。',
            'checkpoint 场景用 Spark Structured Streaming 的 availableNow 触发，业务写走 foreachBatch + pymysql；'
            '不是 Flink checkpoint，也不覆盖 exactly-once 端到端语义（业务侧靠唯一键 INSERT IGNORE 保证幂等）。',
            'Spark 为 local[2]；资源超限的"集群级"配额（YARN/K8s 队列、Spark dynamic allocation）未运行。',
            'MySQL 为本机 DBngin 8.4.x 独有库，结束 DROP；所有数据为合成标签。',
        ],
    }
    local = RUNTIME / 'evidence.json'
    local.write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + '\n')
    if exit_code == 0:
        EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(local, EVIDENCE)
        print(f'\nU87 exit=0 evidence={EVIDENCE}', flush=True)
    else:
        print(f'\nU87 exit={exit_code} evidence(local)={local}', flush=True)
    return exit_code


if __name__ == '__main__':
    sys.exit(run())