MySQL upgrade paths: 8.0 to 8.4, 9.7 and 26.x

checker_session.py

"""The pre-upgrade checks the manual asks for before replacing the server binaries.

Two tools, one 8.4.11 instance seeded with three objects for the checks to judge:

1. A table named CUBE with a column named QUALIFY. Both words are reserved in the 8.4 and the 26.7 keyword lists
   (Reference Manual, 11.3 "Keywords and Reserved Words"), so on either version they work as names only when
   quoted. The 26.7.0 release notes list them, with EXTERNAL and TABLESAMPLE, among the words that were not
   properly defined as reserved keywords.
2. An account that still authenticates with mysql_native_password, re-enabled on 8.4 only through the
   --mysql-native-password=ON start option (the plugin is disabled by default in 8.4 and removed in 9.0.0,
   MySQL 8.4 Reference Manual, 8.4.1.1).
3. A plain sales.orders table, so mysqlcheck has a clean table to show.

Then:

- mysqlcheck --check-upgrade (MySQL 8.4 Reference Manual, "4.5.3 mysqlcheck"): runs CHECK TABLE ... FOR UPGRADE on
  every table and prints a one-line status per table.
- MySQL Shell 26.7.0's util.checkForServerUpgrade() with outputFormat = "JSON" (MySQL Shell 26.7, "12.2 Upgrade
  Checker Utility"), first with targetVersion = "26.7.0", then with "9.7.0", the first hop of the supported path
  from 8.4. Each report prints the serverVersion, the targetVersion, the error/warning/notice totals and only the
  checks that found something; the full JSON report is what a deployment script would read.

The driver masks ports and the run folder, and the Shell runs with --no-wizard inside a config home pinned to the
run folder (MYSQLSH_USER_CONFIG_HOME), so every run prints the same output and writes nothing outside its folder.
"""

import json
import re
import subprocess
import sys

from mysql_session import Server, tarball

MYSQLSH = tarball("mysql-shell-26.7.0") / "bin" / "mysqlsh"


SEED = """
-- Two words reserved in both the 8.4 and the 26.7 keyword lists, used as quoted names:
-- CUBE for a table and QUALIFY for a column.
CREATE DATABASE sales;
CREATE TABLE sales.`CUBE` (
  id INT PRIMARY KEY,
  `QUALIFY` VARCHAR(40)
);
INSERT INTO sales.`CUBE` VALUES (1, 'north'), (2, 'south');

-- A regular shop table so mysqlcheck's output has at least one clean 'OK' row.
CREATE TABLE sales.orders (id INT PRIMARY KEY, total DECIMAL(10,2));

-- An account on mysql_native_password: the plugin is OFF by default on 8.4, so this needs
-- --mysql-native-password=ON on the server start.
CREATE USER 'legacy_app'@'127.0.0.1' IDENTIFIED WITH mysql_native_password BY 'oldpass';
GRANT SELECT ON sales.* TO 'legacy_app'@'127.0.0.1';
"""


def run_mysqlsh_check(server, target):
    """Runs util.checkForServerUpgrade(null, {targetVersion, outputFormat: 'JSON'}) through mysqlsh and returns the
    parsed JSON. The explicit null for the connectionData argument tells the function to use the current session,
    which stops MySQL Shell from parsing the options dict as a connection spec. The config home is pinned to the
    run folder, so mysqlsh writes nothing outside it."""
    cfg = server.base / "mysqlsh-home"
    cfg.mkdir(exist_ok=True)
    uri = f"mysql://root:{server.password}@127.0.0.1:{server.port}?ssl-mode=DISABLED"
    env = dict(server.env)
    env["MYSQLSH_USER_CONFIG_HOME"] = str(cfg)
    js = (
        'println(JSON.stringify(util.checkForServerUpgrade('
        'null, {targetVersion: "' + target + '", outputFormat: "JSON"})))'
    )
    done = subprocess.run(
        [str(MYSQLSH), "--no-wizard", "--log-level=1", "--mysql", "--uri", uri, "--js", "-e", js],
        env=env, cwd=server.base, capture_output=True, text=True,
    )
    if done.returncode != 0:
        print(server.clean(done.stdout))
        print(server.clean(done.stderr))
        sys.exit(f"mysqlsh ended with status {done.returncode}")
    # Shell prints the JSON on standard output; standard error carries a password-in-URI warning we ignore.
    text = done.stdout
    for match in re.finditer(r"\{.*\}", text, re.DOTALL):
        try:
            return json.loads(match.group(0))
        except json.JSONDecodeError:
            continue
    sys.exit("could not find the Shell JSON report on standard output")


def summarize(report, target):
    """Prints the command a deployment script would run and the short summary of its JSON report: the versions, the
    error/warning/notice totals and one line per check that found something, with its problems counted by level."""
    print("$ mysqlsh --no-wizard --uri root@127.0.0.1:<port> -- util check-for-server-upgrade"
          f" --target-version={target} --output-format=JSON")
    # Clean the serverVersion string (it starts with the real version plus a build tag that stays stable here).
    print(f"serverVersion: {report['serverVersion'].split(' - ')[0]}")
    print(f"targetVersion: {report['targetVersion']}")
    print(f"errorCount: {report['errorCount']}  warningCount: {report['warningCount']}  "
          f"noticeCount: {report['noticeCount']}")
    print("checks that found something:")
    for check in report.get("checksPerformed", []):
        problems = check.get("detectedProblems") or []
        if not problems:
            continue
        levels = {}
        for p in problems:
            levels[p["level"]] = levels.get(p["level"], 0) + 1
        counts = ", ".join(f"{k.lower()}s={v}" for k, v in sorted(levels.items()))
        print(f"  {check['id']}: {counts}")
    manual = report.get("manualChecks") or []
    if manual:
        print(f"manual checks the utility reports: {len(manual)}")


with Server(options=["--mysql-native-password=ON"]) as my:
    print("# 1. Start MySQL 8.4.11 with --mysql-native-password=ON (the plugin is OFF by default on 8.4).")
    print("#    Seed one reserved-word table, one plain table and one native-password account.")
    my.up()
    my._call(["mysql", f"--defaults-file={my.client_cnf}"], stdin=SEED)
    my.mask(re.escape(my.password), "<password>")
    my.sql(
        "SELECT COUNT(*) AS reserved_tables FROM information_schema.tables "
        "WHERE table_schema='sales' AND table_name='CUBE'; "
        "SELECT user, host, plugin FROM mysql.user WHERE user='legacy_app'",
        note="what the seed left behind on the 8.4.11 server",
    )
    print("# 2. mysqlcheck --check-upgrade: a one-line status per table against the current server")
    print("#    (CHECK TABLE ... FOR UPGRADE). Clean tables read 'OK'; a table that will not survive the upgrade")
    print("#    reads another status.")
    my.shell("mysqlcheck", "--defaults-file=client.cnf", "--check-upgrade", "sales",
             shown=["mysqlcheck", "--defaults-file=client.cnf", "--check-upgrade", "sales"])

    print("# 3. MySQL Shell 26.7.0's util.checkForServerUpgrade() targeting 26.7.0, two hops away through 9.7 LTS.")
    print("#    The JSON report is long, so the run prints its totals and the checks that found something.")
    summarize(run_mysqlsh_check(my, "26.7.0"), "26.7.0")

    print()
    print("# 4. The same checker targeting 9.7.0, the first hop of the supported path from 8.4.")
    summarize(run_mysqlsh_check(my, "9.7.0"), "9.7.0")

# 1. Start MySQL 8.4.11 with --mysql-native-password=ON (the plugin is OFF by default on 8.4).
#    Seed one reserved-word table, one plain table and one native-password account.
-- what the seed left behind on the 8.4.11 server
$ mysql -t -e "SELECT COUNT(*) AS reserved_tables FROM information_schema.tables WHERE table_schema='sales' AND table_name='CUBE'; SELECT user, host, plugin FROM mysql.user WHERE user='legacy_app'"
+-----------------+
| reserved_tables |
+-----------------+
|               1 |
+-----------------+
+------------+-----------+-----------------------+
| user       | host      | plugin                |
+------------+-----------+-----------------------+
| legacy_app | 127.0.0.1 | mysql_native_password |
+------------+-----------+-----------------------+

# 2. mysqlcheck --check-upgrade: a one-line status per table against the current server
#    (CHECK TABLE ... FOR UPGRADE). Clean tables read 'OK'; a table that will not survive the upgrade
#    reads another status.
$ mysqlcheck --defaults-file=client.cnf --check-upgrade sales
sales.cube                                         OK
sales.orders                                       OK

# 3. MySQL Shell 26.7.0's util.checkForServerUpgrade() targeting 26.7.0, two hops away through 9.7 LTS.
#    The JSON report is long, so the run prints its totals and the checks that found something.
$ mysqlsh --no-wizard --uri root@127.0.0.1:<port> -- util check-for-server-upgrade --target-version=26.7.0 --output-format=JSON
serverVersion: 8.4.11
targetVersion: 26.7.0
errorCount: 1  warningCount: 21  noticeCount: 0
checks that found something:
  sysVars: warnings=21
  authMethodUsage: errors=1

# 4. The same checker targeting 9.7.0, the first hop of the supported path from 8.4.
$ mysqlsh --no-wizard --uri root@127.0.0.1:<port> -- util check-for-server-upgrade --target-version=9.7.0 --output-format=JSON
serverVersion: 8.4.11
targetVersion: 9.7.0
errorCount: 1  warningCount: 21  noticeCount: 0
checks that found something:
  sysVars: warnings=21
  authMethodUsage: errors=1

mysql_session.py

"""Throwaway MySQL 8.4.11 servers for the examples of this lesson.

Every session creates a server of its own: `mysqld --initialize-insecure` makes a new data directory inside a new
folder in the examples folder, the server starts on it, the session runs `mysql` client scripts and MySQL's programs
against it and prints what they printed, and then the server is stopped and the folder deleted, also when something
fails. The server is the pinned MySQL 8.4.11 from the mise installations next to the Python that runs this file;
Server(basedir=...) runs the programs of another unpacked MySQL tarball instead, and tarball(name) finds one.

Safety: the server listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system
reports as unused; the X Protocol listener is off (--mysqlx=OFF); its Unix socket, temporary files, error log and
import/export folder all stay inside the run folder. Right after the first start the session gives root a password made
up for the run (never printed) and connects only with a client option file that holds it, so no password ever appears
on a command line. Every program starts with --no-defaults or --defaults-file, so no option file of the computer
(/etc/my.cnf, ~/.my.cnf) can change a result, and HOME points at the run folder, so ~/.mylogin.cnf is not read either.
A printed mysqld command shows the options that matter to the lesson; the session also passes, on the command line,
the folders it keeps inside the run folder (--basedir, --tmpdir, --log-error, --pid-file, --secure-file-priv) and
the port, which is why performance_schema.variables_info reports those variables as COMMAND_LINE.
The sessions read only this folder and the pinned MySQL installations, and they connect to nothing but 127.0.0.1.

The output is exactly what the programs printed, with these substitutions so that a record never shows the computer it
ran on and is the same on every run: the run folder is shown as ".", the folder of MySQL's programs is left out of
their paths and MySQL's own folder is shown as <basedir>, the computer's name is shown as "lab-host" and the operating-system user's name as "learner", the time a
statement took (the client's "(0.01 sec)") is left out, the time stamp SET PERSIST writes into mysqld-auto.cnf is
shown as <when it was set>, error-log lines lose their timestamp and thread number, and a session can name more values
that change between runs with my.mask(pattern, replacement), which then prints in their place. A client table whose cells
change is drawn again so that its borders still line up.
The sessions never print the password, and the run's port prints as <port>. A session stops with status 1 when a command that should have worked
fails, or when a script reports a different number of errors than the session expects, so a run that went wrong is
never recorded as if it had worked.
"""

import os
import pwd
import re
import secrets
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import textwrap
import time
from pathlib import Path

MYSQL_VERSION = "8.4.11"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
PINNED = INSTALLS / "mysql" / MYSQL_VERSION
OS_USER = pwd.getpwuid(os.geteuid()).pw_name
HOST = socket.gethostname().split(".")[0]
# 2026-10-17T16:50:01.123456Z 0 [ERROR] ... -> [ERROR] ...
LOG_PREFIX = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d+Z \d+ ", re.M)


def tarball(name):
    """The folder of an unpacked official tarball, such as mysql-9.7.2, for Server(basedir=...). It is the folder
    <name> in a folder called toolchains, looked for in every folder above the one the session runs in and above
    this file, directly or inside one of the folders of a tmp folder there. The search stops below the home folder."""
    home = Path(pwd.getpwuid(os.geteuid()).pw_dir).resolve()
    seen = set()
    for start in (Path.cwd().resolve(), Path(__file__).resolve().parent):
        for parent in (start, *start.parents):
            if parent in seen or parent == home or parent == Path(parent.anchor):
                break
            seen.add(parent)
            candidates = [parent / "toolchains" / name]
            if (parent / "tmp").is_dir():
                candidates += sorted((parent / "tmp").glob(f"*/toolchains/{name}"))
            for found in candidates:
                if (found / "bin").is_dir():
                    return found
    sys.exit(f"no toolchains folder above this one holds {name}")


def free_port():
    """A TCP port on 127.0.0.1 that nothing listens on now."""
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


class Server:
    """One throwaway server: initialise, start; stopped and deleted when the with block ends.

    options: extra mysqld options for every start (for example "--sql-mode=" or "--character-set-server=latin1").
    basedir: the folder of an unpacked MySQL tarball to run instead of the pinned 8.4.11.
    masks: (pattern, replacement) pairs, as for mask().
    """

    def __init__(self, options=(), redo=None, basedir=None, masks=(), label=None, port_label=None):
        self.options = [str(o) for o in options]
        self.redo = redo
        self.basedir = Path(basedir).resolve() if basedir else PINNED
        self.bin = self.basedir / "bin"
        self.label = label                       # a stable name for this server in the record ("source", "replica")
        self.port_label = port_label             # the text that shows in place of this server's port
        self.uuid = None
        self.masks = []
        for pattern, replacement in masks:
            self.mask(pattern, replacement)
        self.proc = None

    def mask(self, pattern, replacement):
        """From now on every printed line shows text matching the regular expression as the replacement: for a value
        that changes from run to run (an uptime, a time stamp), so that the recorded output stays the same."""
        self.masks.append((re.compile(pattern), replacement))

    # ---- the folder ------------------------------------------------------------------------------------------------
    def __enter__(self):
        if not (self.bin / "mysqld").exists():
            sys.exit(f"no MySQL server program in {self.basedir}" if self.basedir != PINNED
                     else f"MySQL {MYSQL_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="my", dir=".")).resolve()
        for sub in ("tmp", "files"):
            (self.base / sub).mkdir(mode=0o700)
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        self.client_cnf = self.base / "client.cnf"
        self._write_client_cnf(password=None)
        self.env = {key: value for key, value in os.environ.items()
                    if not key.startswith("MYSQL") and key not in ("TZ", "HOME", "LANG", "LC_ALL")}
        self.env.update({"HOME": str(self.base), "TZ": "UTC", "LANG": "en_US.UTF-8",
                         "MYSQL_TEST_LOGIN_FILE": str(self.base / "no-login-paths.cnf")})
        for pattern in ("*.sql", "*.cnf", "*.csv", "*.txt"):
            for f in Path(".").glob(pattern):
                shutil.copy(f, self.base / f.name)
        return self

    def __exit__(self, *exc):
        if self.proc and self.proc.poll() is None:
            self.stop(quiet=True)
        shutil.rmtree(self.base, ignore_errors=True)

    def _write_client_cnf(self, password):
        lines = ["[client]", "user=root", "host=127.0.0.1", f"port={self.port}", "protocol=TCP"]
        if password:
            lines.append(f"password={password}")
        self.client_cnf.write_text("\n".join(lines) + "\n")
        self.client_cnf.chmod(0o600)

    def server_options(self):
        b = self.base
        return ["--no-defaults", f"--basedir={self.basedir}", f"--datadir={b / 'data'}", f"--tmpdir={b / 'tmp'}",
                f"--log-error={b / 'mysqld.err'}", f"--pid-file={b / 'mysqld.pid'}",
                f"--secure-file-priv={b / 'files'}",
                *([f"--innodb-redo-log-capacity={self.redo}"] if self.redo else [])]

    # ---- the server ------------------------------------------------------------------------------------------------
    def initialize(self, show=False):
        """mysqld --initialize-insecure: a new data directory whose root account has no password yet."""
        args = ["mysqld", *self.server_options(), "--initialize-insecure"]
        code, out = self._call(args)
        if show:
            _show(["mysqld", "--no-defaults", "--initialize-insecure", "--datadir=./data"], out)
        if code != 0:
            print(self.log())
            sys.exit(f"mysqld --initialize-insecure ended with status {code}")

    def start(self, extra=(), defaults_file=None, expect_fail=False, show=False):
        """Starts mysqld on 127.0.0.1 and waits until it answers. With expect_fail, prints the error-log lines of a
        start that must fail, and returns them."""
        log = self.base / "mysqld.err"
        before = log.read_text() if log.exists() else ""
        opts = self.server_options()
        if defaults_file:
            opts[0] = f"--defaults-file={self.base / defaults_file}"
        args = [str(self.bin / "mysqld"), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
                "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *[str(e) for e in extra]]
        if show:
            shown = ["mysqld", opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
                     "--port=$PORT", "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *extra]
            print("$ " + " ".join(_arg(a) if a != "--port=$PORT" else a for a in shown) + " &")
            print()
        with open(self.base / "mysqld.out", "a") as out:
            self.proc = subprocess.Popen(args, stdout=out, stderr=subprocess.STDOUT, cwd=self.base, env=self.env)
        for _ in range(600):
            if self.proc.poll() is not None:
                break
            if self._ping():
                break
            time.sleep(0.1)
        if self.proc.poll() is not None:
            new = (log.read_text() if log.exists() else "")[len(before):]
            lines = [l for l in self.clean(LOG_PREFIX.sub("", new)).splitlines() if "[ERROR]" in l]
            if expect_fail:
                print("\n".join(lines))
                print(f"(mysqld ended with status {self.proc.returncode})")
                print()
                self.proc = None
                return lines
            print("\n".join(lines))
            sys.exit(f"mysqld ended with status {self.proc.returncode} while starting")
        if expect_fail:
            sys.exit("mysqld started, but the session expected it to fail")
        if not self._ping():
            sys.exit("mysqld did not answer within 60 seconds")
        return []

    def _ping(self):
        code, _ = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "--connect-timeout=2", "ping"])
        return code == 0

    def up(self, extra=(), defaults_file=None, show=False):
        """Initialise, start (with start()'s options), and give root the run's password, quietly."""
        self.initialize()
        self.start(extra=extra, defaults_file=defaults_file, show=show)
        return self.login()

    def login(self):
        """Gives root the run's password (never printed) on a server started from a new data directory, and makes
        every later client of the session use it. If this server has a label, its random server_uuid is masked to a
        stable value so GTID sets read the same on every run. A server started with --super-read-only=ON (a replica)
        blocks even root's ALTER USER, so super_read_only is turned off for the one password statement and restored.
        The statement runs with sql_log_bin off, for two reasons: the binary log a lesson decodes then holds only the
        lesson's own statements, and the replica keeps its own root password (each server of a run has a different one,
        so a replicated ALTER USER would lock the session out of the replica)."""
        sro = self.query("SELECT @@global.super_read_only") == "1"
        if sro:
            self.query("SET GLOBAL super_read_only=OFF")
        self.silent("SET SESSION sql_log_bin = 0;\n"
                    f"ALTER USER 'root'@'localhost' IDENTIFIED BY '{self.password}'")
        self._write_client_cnf(self.password)
        if sro:
            self.query("SET GLOBAL super_read_only=ON")
        if self.label:
            self.uuid = self.query("SELECT @@server_uuid")
            self.mask(re.escape(self.uuid), f"<{self.label}-uuid>")
        return self

    def stop(self, quiet=False):
        """A clean shutdown (mysqladmin shutdown), then waits until the process has ended."""
        code, out = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "shutdown"])
        if code != 0 and not quiet:
            sys.exit(f"mysqladmin shutdown failed: {out}")
        try:
            self.proc.wait(timeout=120)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
            self.proc.wait(timeout=60)
        self.proc = None

    def restart(self, extra=(), defaults_file=None, show=False):
        self.stop()
        self.start(extra=extra, defaults_file=defaults_file, show=show)

    # ---- running things --------------------------------------------------------------------------------------------
    def _call(self, args, stdin=None, merge=True, clean=True):
        """Runs one of MySQL's programs in the run folder: (status, cleaned output). merge=False keeps only what it
        wrote to standard output (its warnings on standard error are dropped). clean=False returns the output exactly
        as the program wrote it, for a value the session feeds back into SQL (a GTID set, a file name): a masked value
        is text for the record only and MySQL would reject it."""
        args = [str(a) for a in args]
        done = subprocess.run([str(self.bin / args[0]), *args[1:]], stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT if merge else subprocess.DEVNULL,
                              text=True, cwd=self.base, env=self.env, input=stdin)
        return done.returncode, (self.clean(done.stdout) if clean else done.stdout)

    def output(self, *args, connect=False, status=0):
        """Runs one of MySQL's programs without printing anything and returns its standard output, for example the
        SQL that mysql_tzinfo_to_sql writes, which the session saves with write_file(..., show=False) and runs next."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run, merge=False)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """The substitutions of the docstring. A table of the client whose cells change is drawn again with the
        new column widths, so its borders still line up."""
        lines, out, i = text.split("\n"), [], 0
        while i < len(lines):
            if lines[i].startswith("+-"):
                j = i
                while j < len(lines) and lines[j][:1] in ("+", "|"):
                    j += 1
                block = lines[i:j]
                done = [self._clean(l) for l in block]
                out.extend(done if done == block else _redraw(block, self._clean))
                i = j
            else:
                out.append(self._clean(lines[i]))
                i += 1
        return "\n".join(out)

    def _clean(self, text):
        text = text.replace(str(self.bin) + "/", "").replace(str(self.basedir), "<basedir>")
        text = text.replace(str(self.base) + "/", "./").replace(str(self.base), ".")
        text = re.sub(r" \(\d+\.\d{2,3} sec\)", "", text)  # 8.4 prints 2 decimals, 9.7 and 26.7 print 3
        text = re.sub(r'"Timestamp": \d+', '"Timestamp": <when it was set>', text)
        text = re.sub(rf"\b{re.escape(HOST)}\b", "lab-host", text)
        text = re.sub(rf"\b{re.escape(OS_USER)}\b", "learner", text)
        text = re.sub(rf"(?<![\d.]){self.port}(?![\d.])", self.port_label or "<port>", text)
        for pattern, replacement in self.masks:
            text = pattern.sub(replacement, text)
        return text

    def log(self, pattern=None):
        """The error log without timestamps and thread numbers, optionally only the lines matching a pattern."""
        text = self.clean(LOG_PREFIX.sub("", (self.base / "mysqld.err").read_text()))
        if pattern:
            text = "\n".join(l for l in text.splitlines() if re.search(pattern, l))
        return text

    def show_log(self, pattern, note=None):
        """Prints the error-log lines that match a pattern, as grep would."""
        print(f"$ grep -E {_arg(pattern)} mysqld.err" + (f"   # {note}" if note else ""))
        print(self.log(pattern))
        print()

    def query(self, sql, db=None):
        """One value from one query, unprinted and UNMASKED (the session uses it to wait or to decide, and often
        feeds it back into SQL: a masked value would be a syntax error). Use gtid_executed() or clean() for a value
        that goes into the record."""
        args = ["mysql", f"--defaults-file={self.client_cnf}", "-N", "-B", *([db] if db else [])]
        code, out = self._call(args, stdin=sql, clean=False)
        if code != 0:
            sys.exit(f"the query failed: {self.clean(out)}")
        return out.strip()

    def silent(self, sql):
        """Runs a statement without printing it (used only for the run's password)."""
        self.query(sql)

    def mysql(self, script, errors=0, status=None, db=None, flags=("-t", "-vvv", "--skip-comments"), note=None, force=None):
        """Runs a client script from standard input and prints the command and its output. `errors` is how many
        ERROR lines the script must print; with errors > 0 the client runs on past an error (--force) and then ends
        with status 0, without --force it stops at the first error with status 1. The default flags print each
        statement (-vvv) as a table (-t) without the script's comments (--skip-comments)."""
        force = errors > 0 if force is None else force
        shown = ["mysql", *flags, *(["--force"] if force else []), *([db] if db else [])]
        args = ["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]]
        code, out = self._call(args, stdin=(self.base / script).read_text())
        if note:
            print(f"-- {note}")
        print("$ " + " ".join(_arg(a) for a in shown) + f" < {script}")
        if out.strip():
            print(out.rstrip("\n"))
        print()
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        expected_status = (1 if errors and not force else 0) if status is None else status
        if found != errors or code != expected_status:
            sys.exit(f"{script}: {found} errors and status {code}; the session expects {errors} and {expected_status}")
        return out

    def shell(self, *args, status=0, shown=None, connect=False):
        """Runs one of MySQL's programs and prints the command and its output. connect=True gives a client program
        (mysqladmin, mysqlshow, mysql) the run's client option file, which the printed command leaves out."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run)
        _show(shown or args, out)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def sql(self, statement, flags=("-t",), errors=0, db=None, note=None):
        """Runs one statement with mysql -e and prints the command and its output."""
        shown = ["mysql", *flags, *([db] if db else []), "-e", statement]
        code, out = self._call(["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]])
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        if found != errors or (code != 0) != (errors > 0):
            sys.exit(f"{statement}: {found} errors and status {code}; the session expects {errors}")
        return out

    def write_file(self, path, text, show=True):
        """Writes a text file into the run folder (an option file, a script) and prints it, as cat would."""
        (self.base / path).write_text(textwrap.dedent(text).lstrip("\n"))
        if show:
            self.show_file(path)

    def show_file(self, path):
        print(f"$ cat {path}")
        print(self.clean((self.base / path).read_text()).rstrip("\n"))
        print()

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the run (sorted, folders with a slash), as `ls -p` would list them."""
        path = self.base / folder
        names = sorted(n + ("/" if (path / n).is_dir() else "") for n in os.listdir(path) if n.startswith(prefix))
        print(f"$ ls -p {folder}" + (f" | grep '^{prefix}'" if prefix else ""))
        print(textwrap.fill("  ".join(names), width=78, break_on_hyphens=False))
        print()
        return names


    # ---- replication (two or three servers on 127.0.0.1) -----------------------------------------------------------
    def gtid_executed(self, raw=False):
        """This server's @@global.gtid_executed as one line (no newlines). The value is masked for the record (the
        server_uuid reads as <label-uuid>); raw=True gives the real set, which is what MySQL accepts in a statement."""
        value = self.query("SELECT @@global.gtid_executed").replace("\n", "")
        return value if raw else self.clean(value)

    def track(self, other):
        """Teach this server's record to mask another server's port and server_uuid, so a statement that names the
        other server (CHANGE REPLICATION SOURCE TO, SHOW REPLICA STATUS) reads the same on every run."""
        if other.port:
            self.mask(rf"(?<![\d.]){other.port}(?![\d.])", other.port_label or "<port>")
        if other.uuid:
            self.mask(re.escape(other.uuid), f"<{other.label}-uuid>")
        self.mask(re.escape(str(other.base)), ".")

    def change_source(self, source, user="repl", password="replpass", auto_position=True, ssl=False,
                      verify_ca=False, channel=None, delay=None, extra=(), show=True, run=True):
        """Builds and (run=True) runs CHANGE REPLICATION SOURCE TO on this replica, pointing at `source`. With
        verify_ca the source's auto-generated ca.pem is copied into the replica's data directory and the replica
        verifies the server certificate's issuer chains to it (OpenSSL's VERIFY_CA mode).
        A relative SOURCE_SSL_CA is resolved relative to the replica's data directory, so the file is copied there,
        not into the run folder.
        VERIFY_IDENTITY (SOURCE_SSL_VERIFY_SERVER_CERT=1, which also checks the server certificate's SAN against the
        host) is not wired up here: MySQL's --initialize-insecure auto-generates self-signed certificates whose
        subject is "MySQL_Server_<version>_Auto_Generated_Server_Certificate" and whose SAN does not include
        127.0.0.1, so VERIFY_IDENTITY would fail with "certificate verify failed". A production setup issues server
        certificates whose SAN includes the host and then enables SOURCE_SSL_VERIFY_SERVER_CERT=1."""
        self.track(source)
        opts = ["SOURCE_HOST='127.0.0.1'", f"SOURCE_PORT={source.port}", f"SOURCE_USER='{user}'",
                f"SOURCE_PASSWORD='{password}'"]
        if auto_position:
            opts.append("SOURCE_AUTO_POSITION=1")
        if ssl:
            opts.append("SOURCE_SSL=1")
        if verify_ca:
            shutil.copy(source.base / "data" / "ca.pem", self.base / "data" / "source-ca.pem")
            opts.append("SOURCE_SSL_CA='source-ca.pem'")
        if delay is not None:
            opts.append(f"SOURCE_DELAY={delay}")
        opts += [str(e) for e in extra]
        stmt = "CHANGE REPLICATION SOURCE TO\n  " + ",\n  ".join(opts)
        if channel:
            stmt += f"\n  FOR CHANNEL '{channel}'"
        stmt += ";"
        self.write_file("_change_source.sql", stmt + "\n", show=False)
        if run:
            if show:
                self.mysql("_change_source.sql")
            else:
                self.query(stmt)
        return stmt

    def start_replica(self, user=None, password=None, channel=None, show=True):
        """START REPLICA, optionally with USER/PASSWORD given at start time (not stored in the metadata repository)."""
        parts = ["START REPLICA"]
        if user:
            parts.append(f"USER='{user}' PASSWORD='{password}'")
        if channel:
            parts.append(f"FOR CHANNEL '{channel}'")
        stmt = " ".join(parts) + ";"
        if show:
            self.write_file("_start_replica.sql", stmt + "\n", show=False)
            self.mysql("_start_replica.sql")
        else:
            self.query(stmt)

    def stop_replica(self, channel=None, show=False):
        stmt = "STOP REPLICA" + (f" FOR CHANNEL '{channel}'" if channel else "")
        if show:
            self.sql(stmt)
        else:
            self.query(stmt)

    def wait_replica_running(self, channel=None, timeout=30):
        """Waits until both replication threads report ON in performance_schema (deterministic, no fixed sleep)."""
        where = f" WHERE CHANNEL_NAME='{channel}'" if channel else ""
        for _ in range(timeout * 10):
            io = self.query(f"SELECT SERVICE_STATE FROM performance_schema.replication_connection_status{where}")
            applier = self.query(f"SELECT SERVICE_STATE FROM performance_schema.replication_applier_status{where}")
            if io == "ON" and applier == "ON":
                return
            time.sleep(0.1)
        print(self.sql("SHOW REPLICA STATUS", flags=("-E",)))
        sys.exit("the replica did not reach the running state in time")

    def wait_caught_up(self, source, timeout=30):
        """Blocks until this replica has executed every transaction the source had committed when called."""
        gset = source.gtid_executed(raw=True)
        if not gset:
            return
        result = self.query(f"SELECT WAIT_FOR_EXECUTED_GTID_SET('{gset}', {timeout})")
        if result != "0":
            sys.exit(f"the replica did not catch up (WAIT_FOR_EXECUTED_GTID_SET returned {result})")

    def replica_status(self, fields, channel=None, note=None, label="SHOW REPLICA STATUS"):
        """Prints only the named SHOW REPLICA STATUS fields, in the order given, vertically (phone-friendly)."""
        stmt = "SHOW REPLICA STATUS" + (f" FOR CHANNEL '{channel}'" if channel else "")
        code, out = self._call(["mysql", f"--defaults-file={self.client_cnf}", "-E", "-e", stmt])
        if code != 0:
            sys.exit(f"SHOW REPLICA STATUS failed: {out}")
        values = {}
        for line in out.splitlines():
            m = re.match(r"\s*([A-Za-z_]+):\s?(.*)$", line)
            if m:
                values[m.group(1)] = m.group(2)
        if note:
            print(f"-- {note}")
        print(f"mysql> {label}\\G")
        for field in fields:
            if field in values:
                print(f"  {field}: {values[field]}")
        print()
        return values


def _redraw(block, clean):
    """Draws a client table again after its cells were cleaned. The columns come from the first border line; the row
    after it is the header (left-aligned); a column whose data cells were padded on the left (numbers) stays
    right-aligned (an empty cell says nothing about its column)."""
    cuts = [k for k, ch in enumerate(block[0]) if ch == "+"]
    rows = [None if line.startswith("+") else [line[x + 1:y] for x, y in zip(cuts, cuts[1:])] for line in block]
    header = next(k for k, r in enumerate(rows) if r is not None)
    ncol = len(cuts) - 1
    right = [any(r is not None and k != header and r[c].strip() and r[c].startswith("  ") for k, r in enumerate(rows))
             for c in range(ncol)]
    cells = [None if r is None else [clean(x.strip()) for x in r] for r in rows]
    widths = [max(len(r[c]) for r in cells if r is not None) for c in range(ncol)]
    drawn = []
    for k, r in enumerate(cells):
        if r is None:
            drawn.append("+" + "+".join("-" * (w + 2) for w in widths) + "+")
        else:
            drawn.append("|" + "|".join(" " + (t.rjust(w) if right[c] and k != header else t.ljust(w)) + " "
                                        for c, (t, w) in enumerate(zip(r, widths))) + "|")
    return drawn


def _arg(a):
    """A command-line argument as a reader would type it: quotes only where the shell needs them."""
    a = str(a)
    if re.fullmatch(r"[\w@%+=:,./-]+", a):
        return a
    if not re.search(r"[\"$`\\!]", a):
        return f'"{a}"'
    return shlex.quote(a)


def _show(args, out):
    print("$ " + " ".join(_arg(a) for a in args))
    if out.strip():
        print(out.rstrip("\n"))
    print()

upgrade_chain_session.py

"""The in-place upgrade chain on one data directory: 8.4.11 -> 9.7.2 -> 26.7.0, and the refusal from 26.7.0 against
an 8.4.11 data directory. One server, one data directory, three MySQL versions swapped under it.

Rules the output demonstrates (MySQL 26.7 Reference Manual, 1.3 "MySQL Releases: Innovation and LTS", and 9.7
Reference Manual, 3.2 "Upgrade Paths"):

- An LTS release can upgrade within its own line or into the immediately following compatibility lineage, and only
  9.7.x LTS may step directly into the first calendar-versioned lineage, the one 26.7.0 belongs to.
- A direct 8.4.11 to 26.7.0 start therefore fails with MY-014060: the source and target are not in the same
  compatibility lineage, and 8.4 is not the immediately preceding LTS for 26.7 (9.7 is).
- Every stop before a swap is a slow shutdown (innodb_fast_shutdown = 0): InnoDB finishes a full purge and a change
  buffer merge before it stops, which the 9.7 manual's entry for innodb_fast_shutdown (17.14) advises before an
  upgrade or downgrade between major releases.

The mysql_session driver masks ports, basedirs and the run folder and strips log timestamps, so every run prints
the same output.
"""

from pathlib import Path

from mysql_session import Server, tarball


def switch(my, basedir):
    """Point this server at another unpacked MySQL tarball, so the next my.start() runs that version's mysqld on the
    same data directory."""
    my.basedir = Path(basedir).resolve()
    my.bin = my.basedir / "bin"


with Server(redo="8M") as my:
    print("# 1. Initialise and start MySQL 8.4.11. Write a database the chain will carry across every upgrade.")
    my.up()
    my.sql(
        "CREATE DATABASE shop; "
        "CREATE TABLE shop.orders (id INT PRIMARY KEY, total DECIMAL(10,2)); "
        "INSERT INTO shop.orders VALUES (1, 499.00), (2, 1250.50); "
        "SELECT VERSION()"
    )

    print("# 2. Slow-shutdown the 8.4.11 server, then start 26.7.0 against the same data directory.")
    print("#    26.7.0 refuses: 8.4 is not the immediately preceding LTS for 26.7, and the two lineages differ.")
    my.query("SET GLOBAL innodb_fast_shutdown = 0")
    my.stop()
    switch(my, tarball("mysql-26.7.0"))
    my.start(expect_fail=True)

    print("# 3. Switch to 9.7.2 and start on the same directory. 9.7 is the immediately following LTS, so the server")
    print("#    upgrades the data dictionary (80300 -> 90200) and the server metadata (80411 -> 90702), then runs.")
    switch(my, tarball("mysql-9.7.2"))
    my.start()
    my.sql("SELECT VERSION(); SELECT COUNT(*) AS orders FROM shop.orders")
    my.show_log("MY-01(1090|3413|3381)", note="data-dictionary and server upgrade lines")

    print("# 4. Slow-shutdown 9.7.2 and start 26.7.0 against the same directory.")
    print("#    9.7 LTS is the one bridge into the first calendar lineage, so 26.7.0 upgrades and starts.")
    my.query("SET GLOBAL innodb_fast_shutdown = 0")
    my.stop()
    switch(my, tarball("mysql-26.7.0"))
    my.start()
    my.sql("SELECT VERSION(); SELECT COUNT(*) AS orders FROM shop.orders")
    my.show_log("MY-013381", note="server upgrade from 9.7.2 (90702) to 26.7.0 (260700)")

# 1. Initialise and start MySQL 8.4.11. Write a database the chain will carry across every upgrade.
$ mysql -t -e "CREATE DATABASE shop; CREATE TABLE shop.orders (id INT PRIMARY KEY, total DECIMAL(10,2)); INSERT INTO shop.orders VALUES (1, 499.00), (2, 1250.50); SELECT VERSION()"
+-----------+
| VERSION() |
+-----------+
| 8.4.11    |
+-----------+

# 2. Slow-shutdown the 8.4.11 server, then start 26.7.0 against the same data directory.
#    26.7.0 refuses: 8.4 is not the immediately preceding LTS for 26.7, and the two lineages differ.
[ERROR] [MY-014060] [Server] Invalid MySQL server upgrade: Cannot upgrade from 80411 to 260700. The source and target must be in the same compatibility lineage, unless the source is the immediately preceding LTS for the target.
[ERROR] [MY-010020] [Server] Data Dictionary initialization failed.
[ERROR] [MY-010119] [Server] Aborting
(mysqld ended with status 1)

# 3. Switch to 9.7.2 and start on the same directory. 9.7 is the immediately following LTS, so the server
#    upgrades the data dictionary (80300 -> 90200) and the server metadata (80411 -> 90702), then runs.
$ mysql -t -e "SELECT VERSION(); SELECT COUNT(*) AS orders FROM shop.orders"
+-----------+
| VERSION() |
+-----------+
| 9.7.2     |
+-----------+
+--------+
| orders |
+--------+
|      2 |
+--------+

$ grep -E "MY-01(1090|3413|3381)" mysqld.err   # data-dictionary and server upgrade lines
[System] [MY-011090] [Server] Data dictionary upgrading from version '80300' to '90200'.
[System] [MY-013413] [Server] Data dictionary upgrade from version '80300' to '90200' completed.
[System] [MY-013381] [Server] Server upgrade from '80411' to '90702' started.
[System] [MY-013381] [Server] Server upgrade from '80411' to '90702' completed.

# 4. Slow-shutdown 9.7.2 and start 26.7.0 against the same directory.
#    9.7 LTS is the one bridge into the first calendar lineage, so 26.7.0 upgrades and starts.
$ mysql -t -e "SELECT VERSION(); SELECT COUNT(*) AS orders FROM shop.orders"
+-----------+
| VERSION() |
+-----------+
| 26.7.0    |
+-----------+
+--------+
| orders |
+--------+
|      2 |
+--------+

$ grep -E MY-013381 mysqld.err   # server upgrade from 9.7.2 (90702) to 26.7.0 (260700)
[System] [MY-013381] [Server] Server upgrade from '80411' to '90702' started.
[System] [MY-013381] [Server] Server upgrade from '80411' to '90702' completed.
[System] [MY-013381] [Server] Server upgrade from '90702' to '260700' started.
[System] [MY-013381] [Server] Server upgrade from '90702' to '260700' completed.

mysql_session.py

"""Throwaway MySQL 8.4.11 servers for the examples of this lesson.

Every session creates a server of its own: `mysqld --initialize-insecure` makes a new data directory inside a new
folder in the examples folder, the server starts on it, the session runs `mysql` client scripts and MySQL's programs
against it and prints what they printed, and then the server is stopped and the folder deleted, also when something
fails. The server is the pinned MySQL 8.4.11 from the mise installations next to the Python that runs this file;
Server(basedir=...) runs the programs of another unpacked MySQL tarball instead, and tarball(name) finds one.

Safety: the server listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system
reports as unused; the X Protocol listener is off (--mysqlx=OFF); its Unix socket, temporary files, error log and
import/export folder all stay inside the run folder. Right after the first start the session gives root a password made
up for the run (never printed) and connects only with a client option file that holds it, so no password ever appears
on a command line. Every program starts with --no-defaults or --defaults-file, so no option file of the computer
(/etc/my.cnf, ~/.my.cnf) can change a result, and HOME points at the run folder, so ~/.mylogin.cnf is not read either.
A printed mysqld command shows the options that matter to the lesson; the session also passes, on the command line,
the folders it keeps inside the run folder (--basedir, --tmpdir, --log-error, --pid-file, --secure-file-priv) and
the port, which is why performance_schema.variables_info reports those variables as COMMAND_LINE.
The sessions read only this folder and the pinned MySQL installations, and they connect to nothing but 127.0.0.1.

The output is exactly what the programs printed, with these substitutions so that a record never shows the computer it
ran on and is the same on every run: the run folder is shown as ".", the folder of MySQL's programs is left out of
their paths and MySQL's own folder is shown as <basedir>, the computer's name is shown as "lab-host" and the operating-system user's name as "learner", the time a
statement took (the client's "(0.01 sec)") is left out, the time stamp SET PERSIST writes into mysqld-auto.cnf is
shown as <when it was set>, error-log lines lose their timestamp and thread number, and a session can name more values
that change between runs with my.mask(pattern, replacement), which then prints in their place. A client table whose cells
change is drawn again so that its borders still line up.
The sessions never print the password, and the run's port prints as <port>. A session stops with status 1 when a command that should have worked
fails, or when a script reports a different number of errors than the session expects, so a run that went wrong is
never recorded as if it had worked.
"""

import os
import pwd
import re
import secrets
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import textwrap
import time
from pathlib import Path

MYSQL_VERSION = "8.4.11"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
PINNED = INSTALLS / "mysql" / MYSQL_VERSION
OS_USER = pwd.getpwuid(os.geteuid()).pw_name
HOST = socket.gethostname().split(".")[0]
# 2026-10-17T16:50:01.123456Z 0 [ERROR] ... -> [ERROR] ...
LOG_PREFIX = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d+Z \d+ ", re.M)


def tarball(name):
    """The folder of an unpacked official tarball, such as mysql-9.7.2, for Server(basedir=...). It is the folder
    <name> in a folder called toolchains, looked for in every folder above the one the session runs in and above
    this file, directly or inside one of the folders of a tmp folder there. The search stops below the home folder."""
    home = Path(pwd.getpwuid(os.geteuid()).pw_dir).resolve()
    seen = set()
    for start in (Path.cwd().resolve(), Path(__file__).resolve().parent):
        for parent in (start, *start.parents):
            if parent in seen or parent == home or parent == Path(parent.anchor):
                break
            seen.add(parent)
            candidates = [parent / "toolchains" / name]
            if (parent / "tmp").is_dir():
                candidates += sorted((parent / "tmp").glob(f"*/toolchains/{name}"))
            for found in candidates:
                if (found / "bin").is_dir():
                    return found
    sys.exit(f"no toolchains folder above this one holds {name}")


def free_port():
    """A TCP port on 127.0.0.1 that nothing listens on now."""
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


class Server:
    """One throwaway server: initialise, start; stopped and deleted when the with block ends.

    options: extra mysqld options for every start (for example "--sql-mode=" or "--character-set-server=latin1").
    basedir: the folder of an unpacked MySQL tarball to run instead of the pinned 8.4.11.
    masks: (pattern, replacement) pairs, as for mask().
    """

    def __init__(self, options=(), redo=None, basedir=None, masks=(), label=None, port_label=None):
        self.options = [str(o) for o in options]
        self.redo = redo
        self.basedir = Path(basedir).resolve() if basedir else PINNED
        self.bin = self.basedir / "bin"
        self.label = label                       # a stable name for this server in the record ("source", "replica")
        self.port_label = port_label             # the text that shows in place of this server's port
        self.uuid = None
        self.masks = []
        for pattern, replacement in masks:
            self.mask(pattern, replacement)
        self.proc = None

    def mask(self, pattern, replacement):
        """From now on every printed line shows text matching the regular expression as the replacement: for a value
        that changes from run to run (an uptime, a time stamp), so that the recorded output stays the same."""
        self.masks.append((re.compile(pattern), replacement))

    # ---- the folder ------------------------------------------------------------------------------------------------
    def __enter__(self):
        if not (self.bin / "mysqld").exists():
            sys.exit(f"no MySQL server program in {self.basedir}" if self.basedir != PINNED
                     else f"MySQL {MYSQL_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="my", dir=".")).resolve()
        for sub in ("tmp", "files"):
            (self.base / sub).mkdir(mode=0o700)
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        self.client_cnf = self.base / "client.cnf"
        self._write_client_cnf(password=None)
        self.env = {key: value for key, value in os.environ.items()
                    if not key.startswith("MYSQL") and key not in ("TZ", "HOME", "LANG", "LC_ALL")}
        self.env.update({"HOME": str(self.base), "TZ": "UTC", "LANG": "en_US.UTF-8",
                         "MYSQL_TEST_LOGIN_FILE": str(self.base / "no-login-paths.cnf")})
        for pattern in ("*.sql", "*.cnf", "*.csv", "*.txt"):
            for f in Path(".").glob(pattern):
                shutil.copy(f, self.base / f.name)
        return self

    def __exit__(self, *exc):
        if self.proc and self.proc.poll() is None:
            self.stop(quiet=True)
        shutil.rmtree(self.base, ignore_errors=True)

    def _write_client_cnf(self, password):
        lines = ["[client]", "user=root", "host=127.0.0.1", f"port={self.port}", "protocol=TCP"]
        if password:
            lines.append(f"password={password}")
        self.client_cnf.write_text("\n".join(lines) + "\n")
        self.client_cnf.chmod(0o600)

    def server_options(self):
        b = self.base
        return ["--no-defaults", f"--basedir={self.basedir}", f"--datadir={b / 'data'}", f"--tmpdir={b / 'tmp'}",
                f"--log-error={b / 'mysqld.err'}", f"--pid-file={b / 'mysqld.pid'}",
                f"--secure-file-priv={b / 'files'}",
                *([f"--innodb-redo-log-capacity={self.redo}"] if self.redo else [])]

    # ---- the server ------------------------------------------------------------------------------------------------
    def initialize(self, show=False):
        """mysqld --initialize-insecure: a new data directory whose root account has no password yet."""
        args = ["mysqld", *self.server_options(), "--initialize-insecure"]
        code, out = self._call(args)
        if show:
            _show(["mysqld", "--no-defaults", "--initialize-insecure", "--datadir=./data"], out)
        if code != 0:
            print(self.log())
            sys.exit(f"mysqld --initialize-insecure ended with status {code}")

    def start(self, extra=(), defaults_file=None, expect_fail=False, show=False):
        """Starts mysqld on 127.0.0.1 and waits until it answers. With expect_fail, prints the error-log lines of a
        start that must fail, and returns them."""
        log = self.base / "mysqld.err"
        before = log.read_text() if log.exists() else ""
        opts = self.server_options()
        if defaults_file:
            opts[0] = f"--defaults-file={self.base / defaults_file}"
        args = [str(self.bin / "mysqld"), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
                "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *[str(e) for e in extra]]
        if show:
            shown = ["mysqld", opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
                     "--port=$PORT", "--socket=mysql.sock", "--mysqlx=OFF", *self.options, *extra]
            print("$ " + " ".join(_arg(a) if a != "--port=$PORT" else a for a in shown) + " &")
            print()
        with open(self.base / "mysqld.out", "a") as out:
            self.proc = subprocess.Popen(args, stdout=out, stderr=subprocess.STDOUT, cwd=self.base, env=self.env)
        for _ in range(600):
            if self.proc.poll() is not None:
                break
            if self._ping():
                break
            time.sleep(0.1)
        if self.proc.poll() is not None:
            new = (log.read_text() if log.exists() else "")[len(before):]
            lines = [l for l in self.clean(LOG_PREFIX.sub("", new)).splitlines() if "[ERROR]" in l]
            if expect_fail:
                print("\n".join(lines))
                print(f"(mysqld ended with status {self.proc.returncode})")
                print()
                self.proc = None
                return lines
            print("\n".join(lines))
            sys.exit(f"mysqld ended with status {self.proc.returncode} while starting")
        if expect_fail:
            sys.exit("mysqld started, but the session expected it to fail")
        if not self._ping():
            sys.exit("mysqld did not answer within 60 seconds")
        return []

    def _ping(self):
        code, _ = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "--connect-timeout=2", "ping"])
        return code == 0

    def up(self, extra=(), defaults_file=None, show=False):
        """Initialise, start (with start()'s options), and give root the run's password, quietly."""
        self.initialize()
        self.start(extra=extra, defaults_file=defaults_file, show=show)
        return self.login()

    def login(self):
        """Gives root the run's password (never printed) on a server started from a new data directory, and makes
        every later client of the session use it. If this server has a label, its random server_uuid is masked to a
        stable value so GTID sets read the same on every run. A server started with --super-read-only=ON (a replica)
        blocks even root's ALTER USER, so super_read_only is turned off for the one password statement and restored.
        The statement runs with sql_log_bin off, for two reasons: the binary log a lesson decodes then holds only the
        lesson's own statements, and the replica keeps its own root password (each server of a run has a different one,
        so a replicated ALTER USER would lock the session out of the replica)."""
        sro = self.query("SELECT @@global.super_read_only") == "1"
        if sro:
            self.query("SET GLOBAL super_read_only=OFF")
        self.silent("SET SESSION sql_log_bin = 0;\n"
                    f"ALTER USER 'root'@'localhost' IDENTIFIED BY '{self.password}'")
        self._write_client_cnf(self.password)
        if sro:
            self.query("SET GLOBAL super_read_only=ON")
        if self.label:
            self.uuid = self.query("SELECT @@server_uuid")
            self.mask(re.escape(self.uuid), f"<{self.label}-uuid>")
        return self

    def stop(self, quiet=False):
        """A clean shutdown (mysqladmin shutdown), then waits until the process has ended."""
        code, out = self._call(["mysqladmin", f"--defaults-file={self.client_cnf}", "shutdown"])
        if code != 0 and not quiet:
            sys.exit(f"mysqladmin shutdown failed: {out}")
        try:
            self.proc.wait(timeout=120)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
            self.proc.wait(timeout=60)
        self.proc = None

    def restart(self, extra=(), defaults_file=None, show=False):
        self.stop()
        self.start(extra=extra, defaults_file=defaults_file, show=show)

    # ---- running things --------------------------------------------------------------------------------------------
    def _call(self, args, stdin=None, merge=True, clean=True):
        """Runs one of MySQL's programs in the run folder: (status, cleaned output). merge=False keeps only what it
        wrote to standard output (its warnings on standard error are dropped). clean=False returns the output exactly
        as the program wrote it, for a value the session feeds back into SQL (a GTID set, a file name): a masked value
        is text for the record only and MySQL would reject it."""
        args = [str(a) for a in args]
        done = subprocess.run([str(self.bin / args[0]), *args[1:]], stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT if merge else subprocess.DEVNULL,
                              text=True, cwd=self.base, env=self.env, input=stdin)
        return done.returncode, (self.clean(done.stdout) if clean else done.stdout)

    def output(self, *args, connect=False, status=0):
        """Runs one of MySQL's programs without printing anything and returns its standard output, for example the
        SQL that mysql_tzinfo_to_sql writes, which the session saves with write_file(..., show=False) and runs next."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run, merge=False)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """The substitutions of the docstring. A table of the client whose cells change is drawn again with the
        new column widths, so its borders still line up."""
        lines, out, i = text.split("\n"), [], 0
        while i < len(lines):
            if lines[i].startswith("+-"):
                j = i
                while j < len(lines) and lines[j][:1] in ("+", "|"):
                    j += 1
                block = lines[i:j]
                done = [self._clean(l) for l in block]
                out.extend(done if done == block else _redraw(block, self._clean))
                i = j
            else:
                out.append(self._clean(lines[i]))
                i += 1
        return "\n".join(out)

    def _clean(self, text):
        text = text.replace(str(self.bin) + "/", "").replace(str(self.basedir), "<basedir>")
        text = text.replace(str(self.base) + "/", "./").replace(str(self.base), ".")
        text = re.sub(r" \(\d+\.\d{2,3} sec\)", "", text)  # 8.4 prints 2 decimals, 9.7 and 26.7 print 3
        text = re.sub(r'"Timestamp": \d+', '"Timestamp": <when it was set>', text)
        text = re.sub(rf"\b{re.escape(HOST)}\b", "lab-host", text)
        text = re.sub(rf"\b{re.escape(OS_USER)}\b", "learner", text)
        text = re.sub(rf"(?<![\d.]){self.port}(?![\d.])", self.port_label or "<port>", text)
        for pattern, replacement in self.masks:
            text = pattern.sub(replacement, text)
        return text

    def log(self, pattern=None):
        """The error log without timestamps and thread numbers, optionally only the lines matching a pattern."""
        text = self.clean(LOG_PREFIX.sub("", (self.base / "mysqld.err").read_text()))
        if pattern:
            text = "\n".join(l for l in text.splitlines() if re.search(pattern, l))
        return text

    def show_log(self, pattern, note=None):
        """Prints the error-log lines that match a pattern, as grep would."""
        print(f"$ grep -E {_arg(pattern)} mysqld.err" + (f"   # {note}" if note else ""))
        print(self.log(pattern))
        print()

    def query(self, sql, db=None):
        """One value from one query, unprinted and UNMASKED (the session uses it to wait or to decide, and often
        feeds it back into SQL: a masked value would be a syntax error). Use gtid_executed() or clean() for a value
        that goes into the record."""
        args = ["mysql", f"--defaults-file={self.client_cnf}", "-N", "-B", *([db] if db else [])]
        code, out = self._call(args, stdin=sql, clean=False)
        if code != 0:
            sys.exit(f"the query failed: {self.clean(out)}")
        return out.strip()

    def silent(self, sql):
        """Runs a statement without printing it (used only for the run's password)."""
        self.query(sql)

    def mysql(self, script, errors=0, status=None, db=None, flags=("-t", "-vvv", "--skip-comments"), note=None, force=None):
        """Runs a client script from standard input and prints the command and its output. `errors` is how many
        ERROR lines the script must print; with errors > 0 the client runs on past an error (--force) and then ends
        with status 0, without --force it stops at the first error with status 1. The default flags print each
        statement (-vvv) as a table (-t) without the script's comments (--skip-comments)."""
        force = errors > 0 if force is None else force
        shown = ["mysql", *flags, *(["--force"] if force else []), *([db] if db else [])]
        args = ["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]]
        code, out = self._call(args, stdin=(self.base / script).read_text())
        if note:
            print(f"-- {note}")
        print("$ " + " ".join(_arg(a) for a in shown) + f" < {script}")
        if out.strip():
            print(out.rstrip("\n"))
        print()
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        expected_status = (1 if errors and not force else 0) if status is None else status
        if found != errors or code != expected_status:
            sys.exit(f"{script}: {found} errors and status {code}; the session expects {errors} and {expected_status}")
        return out

    def shell(self, *args, status=0, shown=None, connect=False):
        """Runs one of MySQL's programs and prints the command and its output. connect=True gives a client program
        (mysqladmin, mysqlshow, mysql) the run's client option file, which the printed command leaves out."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run)
        _show(shown or args, out)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def sql(self, statement, flags=("-t",), errors=0, db=None, note=None):
        """Runs one statement with mysql -e and prints the command and its output."""
        shown = ["mysql", *flags, *([db] if db else []), "-e", statement]
        code, out = self._call(["mysql", f"--defaults-file={self.client_cnf}", *shown[1:]])
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        if found != errors or (code != 0) != (errors > 0):
            sys.exit(f"{statement}: {found} errors and status {code}; the session expects {errors}")
        return out

    def write_file(self, path, text, show=True):
        """Writes a text file into the run folder (an option file, a script) and prints it, as cat would."""
        (self.base / path).write_text(textwrap.dedent(text).lstrip("\n"))
        if show:
            self.show_file(path)

    def show_file(self, path):
        print(f"$ cat {path}")
        print(self.clean((self.base / path).read_text()).rstrip("\n"))
        print()

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the run (sorted, folders with a slash), as `ls -p` would list them."""
        path = self.base / folder
        names = sorted(n + ("/" if (path / n).is_dir() else "") for n in os.listdir(path) if n.startswith(prefix))
        print(f"$ ls -p {folder}" + (f" | grep '^{prefix}'" if prefix else ""))
        print(textwrap.fill("  ".join(names), width=78, break_on_hyphens=False))
        print()
        return names


    # ---- replication (two or three servers on 127.0.0.1) -----------------------------------------------------------
    def gtid_executed(self, raw=False):
        """This server's @@global.gtid_executed as one line (no newlines). The value is masked for the record (the
        server_uuid reads as <label-uuid>); raw=True gives the real set, which is what MySQL accepts in a statement."""
        value = self.query("SELECT @@global.gtid_executed").replace("\n", "")
        return value if raw else self.clean(value)

    def track(self, other):
        """Teach this server's record to mask another server's port and server_uuid, so a statement that names the
        other server (CHANGE REPLICATION SOURCE TO, SHOW REPLICA STATUS) reads the same on every run."""
        if other.port:
            self.mask(rf"(?<![\d.]){other.port}(?![\d.])", other.port_label or "<port>")
        if other.uuid:
            self.mask(re.escape(other.uuid), f"<{other.label}-uuid>")
        self.mask(re.escape(str(other.base)), ".")

    def change_source(self, source, user="repl", password="replpass", auto_position=True, ssl=False,
                      verify_ca=False, channel=None, delay=None, extra=(), show=True, run=True):
        """Builds and (run=True) runs CHANGE REPLICATION SOURCE TO on this replica, pointing at `source`. With
        verify_ca the source's auto-generated ca.pem is copied into the replica's data directory and the replica
        verifies the server certificate's issuer chains to it (OpenSSL's VERIFY_CA mode).
        A relative SOURCE_SSL_CA is resolved relative to the replica's data directory, so the file is copied there,
        not into the run folder.
        VERIFY_IDENTITY (SOURCE_SSL_VERIFY_SERVER_CERT=1, which also checks the server certificate's SAN against the
        host) is not wired up here: MySQL's --initialize-insecure auto-generates self-signed certificates whose
        subject is "MySQL_Server_<version>_Auto_Generated_Server_Certificate" and whose SAN does not include
        127.0.0.1, so VERIFY_IDENTITY would fail with "certificate verify failed". A production setup issues server
        certificates whose SAN includes the host and then enables SOURCE_SSL_VERIFY_SERVER_CERT=1."""
        self.track(source)
        opts = ["SOURCE_HOST='127.0.0.1'", f"SOURCE_PORT={source.port}", f"SOURCE_USER='{user}'",
                f"SOURCE_PASSWORD='{password}'"]
        if auto_position:
            opts.append("SOURCE_AUTO_POSITION=1")
        if ssl:
            opts.append("SOURCE_SSL=1")
        if verify_ca:
            shutil.copy(source.base / "data" / "ca.pem", self.base / "data" / "source-ca.pem")
            opts.append("SOURCE_SSL_CA='source-ca.pem'")
        if delay is not None:
            opts.append(f"SOURCE_DELAY={delay}")
        opts += [str(e) for e in extra]
        stmt = "CHANGE REPLICATION SOURCE TO\n  " + ",\n  ".join(opts)
        if channel:
            stmt += f"\n  FOR CHANNEL '{channel}'"
        stmt += ";"
        self.write_file("_change_source.sql", stmt + "\n", show=False)
        if run:
            if show:
                self.mysql("_change_source.sql")
            else:
                self.query(stmt)
        return stmt

    def start_replica(self, user=None, password=None, channel=None, show=True):
        """START REPLICA, optionally with USER/PASSWORD given at start time (not stored in the metadata repository)."""
        parts = ["START REPLICA"]
        if user:
            parts.append(f"USER='{user}' PASSWORD='{password}'")
        if channel:
            parts.append(f"FOR CHANNEL '{channel}'")
        stmt = " ".join(parts) + ";"
        if show:
            self.write_file("_start_replica.sql", stmt + "\n", show=False)
            self.mysql("_start_replica.sql")
        else:
            self.query(stmt)

    def stop_replica(self, channel=None, show=False):
        stmt = "STOP REPLICA" + (f" FOR CHANNEL '{channel}'" if channel else "")
        if show:
            self.sql(stmt)
        else:
            self.query(stmt)

    def wait_replica_running(self, channel=None, timeout=30):
        """Waits until both replication threads report ON in performance_schema (deterministic, no fixed sleep)."""
        where = f" WHERE CHANNEL_NAME='{channel}'" if channel else ""
        for _ in range(timeout * 10):
            io = self.query(f"SELECT SERVICE_STATE FROM performance_schema.replication_connection_status{where}")
            applier = self.query(f"SELECT SERVICE_STATE FROM performance_schema.replication_applier_status{where}")
            if io == "ON" and applier == "ON":
                return
            time.sleep(0.1)
        print(self.sql("SHOW REPLICA STATUS", flags=("-E",)))
        sys.exit("the replica did not reach the running state in time")

    def wait_caught_up(self, source, timeout=30):
        """Blocks until this replica has executed every transaction the source had committed when called."""
        gset = source.gtid_executed(raw=True)
        if not gset:
            return
        result = self.query(f"SELECT WAIT_FOR_EXECUTED_GTID_SET('{gset}', {timeout})")
        if result != "0":
            sys.exit(f"the replica did not catch up (WAIT_FOR_EXECUTED_GTID_SET returned {result})")

    def replica_status(self, fields, channel=None, note=None, label="SHOW REPLICA STATUS"):
        """Prints only the named SHOW REPLICA STATUS fields, in the order given, vertically (phone-friendly)."""
        stmt = "SHOW REPLICA STATUS" + (f" FOR CHANNEL '{channel}'" if channel else "")
        code, out = self._call(["mysql", f"--defaults-file={self.client_cnf}", "-E", "-e", stmt])
        if code != 0:
            sys.exit(f"SHOW REPLICA STATUS failed: {out}")
        values = {}
        for line in out.splitlines():
            m = re.match(r"\s*([A-Za-z_]+):\s?(.*)$", line)
            if m:
                values[m.group(1)] = m.group(2)
        if note:
            print(f"-- {note}")
        print(f"mysql> {label}\\G")
        for field in fields:
            if field in values:
                print(f"  {field}: {values[field]}")
        print()
        return values


def _redraw(block, clean):
    """Draws a client table again after its cells were cleaned. The columns come from the first border line; the row
    after it is the header (left-aligned); a column whose data cells were padded on the left (numbers) stays
    right-aligned (an empty cell says nothing about its column)."""
    cuts = [k for k, ch in enumerate(block[0]) if ch == "+"]
    rows = [None if line.startswith("+") else [line[x + 1:y] for x, y in zip(cuts, cuts[1:])] for line in block]
    header = next(k for k, r in enumerate(rows) if r is not None)
    ncol = len(cuts) - 1
    right = [any(r is not None and k != header and r[c].strip() and r[c].startswith("  ") for k, r in enumerate(rows))
             for c in range(ncol)]
    cells = [None if r is None else [clean(x.strip()) for x in r] for r in rows]
    widths = [max(len(r[c]) for r in cells if r is not None) for c in range(ncol)]
    drawn = []
    for k, r in enumerate(cells):
        if r is None:
            drawn.append("+" + "+".join("-" * (w + 2) for w in widths) + "+")
        else:
            drawn.append("|" + "|".join(" " + (t.rjust(w) if right[c] and k != header else t.ljust(w)) + " "
                                        for c, (t, w) in enumerate(zip(r, widths))) + "|")
    return drawn


def _arg(a):
    """A command-line argument as a reader would type it: quotes only where the shell needs them."""
    a = str(a)
    if re.fullmatch(r"[\w@%+=:,./-]+", a):
        return a
    if not re.search(r"[\"$`\\!]", a):
        return f'"{a}"'
    return shlex.quote(a)


def _show(args, out):
    print("$ " + " ".join(_arg(a) for a in args))
    if out.strip():
        print(out.rstrip("\n"))
    print()