############################################################################
# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
############################################################################
"""Known Slurm issues that may make any test fail.

Classifiers used by conftest.py to label a failure with its Issue/Ticket, and to
xfail it on versions older than the fix.
"""
import logging
import os
import re

import atf

# Teardown failures of the module, that a crash (or a gcore of a hung daemon)
# typically causes. Like any other failure, they are explained by the known
# issues of the coredumps found, see _related_coredumps().
JOBS_NOT_CANCELLED = "Not all jobs were successfully cancelled"
DAEMONS_NOT_STOPPED = "Not all Slurm daemons were successfully stopped"


# The failure of a coredump found by the module teardown, followed by its path
COREDUMP_DETECTED = "Coredump detected: "


def analyze_coredump(coredump):
    """
    Return the evidence of a coredump to classify it, as {"path", "time",
    "bin_path", "slurm_prefix", "bt_file", "bt", "error"}, where path is its
    new path and time is when it was dumped (None if unknown).

    The coredump is moved into the logs dir, and gdb is run on it to get its
    binary and to save its full backtrace in bt_file. If its binary cannot be
    found, bin_path is None and error explains why.

    The analysis is kept in atf.properties["coredumps"] for its original and
    its new path, so gdb only runs once for each coredump.
    """
    coredumps = atf.properties["coredumps"]
    if coredump in coredumps:
        return coredumps[coredump]

    orig_coredump = coredump

    # Ensure that the core is copied into the logs dir
    if atf.properties["slurm-logs-dir"] not in coredump:
        coredump_new = (
            f"{atf.properties['slurm-logs-dir']}/{os.path.basename(coredump)}"
        )
        logging.debug(f"Moving coredump from {coredump} to {coredump_new}")
        atf.run_command(f"mv {coredump} {coredump_new}", user="root", quiet=True)
        coredump = coredump_new

    # mv keeps the mtime, that is when the core was dumped
    try:
        core_time = os.path.getmtime(coredump)
    except OSError as e:
        logging.warning(f"Unable to get the time of {coredump}: {e}")
        core_time = None

    # Get the binary that generated the coredump using gdb's
    # "info auxv": AT_EXECFN is the kernel-recorded full path of the
    # executable. Avoids 'file' command issues with cores having "too
    # many program headers".
    # debuginfod off: It may be too slow
    # Use atf.resolve_core_binary() to avoid issue with upgraded
    # components
    gdb_res = atf.run_command(
        f"gdb"
        f" -batch"
        f' -iex "set debuginfod enabled off"'
        f" -c {coredump}"
        f' -ex "info auxv"'
        f' -ex "quit"',
        user="root",
        quiet=True,
        timeout=300,
    )
    if gdb_res["exit_code"] != 0:
        logging.warning(
            f"Error getting the binary from the coredump: {gdb_res['exit_code']}"
        )
    gdb_out = gdb_res["stdout"] + gdb_res["stderr"]
    bin_match = re.search(r'AT_EXECFN.*?"([^"\n]+)"', gdb_out) or re.search(
        r"Core was generated by [`']([^ '`\n]+)", gdb_out
    )

    execfn_path = bin_match.group(1) if bin_match else None
    bin_path = atf.resolve_core_binary(coredump, execfn_path) if execfn_path else None

    if not bin_path:
        if not execfn_path:
            logging.warning(f"Unable to get the binary of {coredump}: {gdb_out}")
            error = "unable to get binary from gdb"
        else:
            error = f"unable to resolve binary by build-id (AT_EXECFN={execfn_path})"
        analysis = {
            "path": coredump,
            "time": core_time,
            "bin_path": None,
            "slurm_prefix": None,
            "bt_file": None,
            "bt": "",
            "error": error,
        }
    else:
        # Generate the .bt file if binary was found
        bt_file = f"{coredump}.bt"
        logging.debug(
            f"Saving the full backtrace of {coredump} from {bin_path} into {bt_file}"
        )
        # debuginfod off: It may be too slow
        results = atf.run_command(
            f"gdb"
            f" -batch"
            f' -iex "set debuginfod enabled off"'
            f' -ex "info files"'
            f' -ex "set pagination off"'
            f' -ex "set confirm off"'
            f' -ex "set print pretty on"'
            f' -ex "set max-value-size unlimited"'
            f' -ex "set print array-indexes on"'
            f' -ex "set print array off"'
            f' -ex "thread apply all bt full"'
            f' -ex "quit"'
            f" {bin_path} {coredump}",
            user="root",
            quiet=True,
            timeout=300,
        )
        if results["exit_code"] != 0:
            logging.warning(f"Error generating the bt file: {results['exit_code']}")
        with open(bt_file, "w") as f:
            f.write(results["stdout"])
            f.write(results["stderr"])

        analysis = {
            "path": coredump,
            "time": core_time,
            "bin_path": bin_path,
            "slurm_prefix": os.path.dirname(os.path.dirname(bin_path)),
            "bt_file": bt_file,
            "bt": results["stdout"] + results["stderr"],
            "error": None,
        }

    coredumps[orig_coredump] = coredumps[coredump] = analysis
    return analysis


def _related_coredumps(failure, report):
    """
    Return the analysis of the coredumps that may explain a failure of the
    report.

    A coredump is a failure on its own, reported by the module teardown, but a
    crash may also make a test fail in any other way: a timeout, a wrong output,
    a client that is killed by a signal while it was expected to fail, the
    jobs that cannot be cancelled in the teardown... So any failure may be
    explained by the coredumps that happened before it:
    - The coredumps of the daemons dumped since the module started, as they run
      for the whole module.
    - The coredumps of the clients dumped since the phase of the report started
      (setup, call or teardown), as they only run for the test that called
      them.
    A coredump whose time is unknown may also explain any failure.

    A COREDUMP_DETECTED failure is only explained by its own coredump.

    Every coredump is analyzed once in the module (see analyze_coredump()), if
    not before then by the module teardown, so considering all of them for any
    failure costs no extra gdb runs.
    """
    if failure.startswith(COREDUMP_DETECTED):
        return [analyze_coredump(failure[len(COREDUMP_DETECTED) :])]

    related = []
    for coredump in atf.get_coredumps():
        analysis = analyze_coredump(coredump)
        if not analysis["bin_path"]:
            continue
        if os.path.basename(os.path.dirname(analysis["bin_path"])) == "sbin":
            since = atf.properties.get("module-start-time", 0)
        else:
            since = report.start
        if analysis["time"] is None or analysis["time"] >= since:
            related.append(analysis)
    return related


def classify_failure(failure, report):
    """
    Return (xfail, reason) of the known issue that explains a failure, or None
    if it's not known.
    xfail is True if the issue is fixed in a newer version than the one tested,
    otherwise the failure is kept but reported with the known reason.

    The failure is a single line with the minimal info to classify it: the
    failure of a test, or one of the failures of the module teardown, like a
    teardown failure or a coredump found (COREDUMP_DETECTED followed by its
    path).

    Each known issue checks the failure and the evidence it needs: the
    backtraces of the coredumps related to the failure (see
    _related_coredumps()), the Slurm logs, and the report. The whole Slurm
    logs are checked, not only the part of the failing test, so the fingerprint
    of an earlier test of the same module may also match. Matching only the
    logs, without the failure or a coredump, should be very exceptional: only
    for issues that may randomly make any test fail.

    The report is the pytest report of the phase that failed (the test setup
    or call, or the module teardown). Its caplog (test_log) is the log
    captured for that phase, so it includes the output of the commands run
    by the test, like the errors printed by a client.

    Known issues are sorted by Ticket/Issue id.
    """
    if not atf.properties["auto-config"]:
        return None

    # Gather evidences. Only the issues with a fingerprint in the failure, or
    # in a coredump, can explain a coredump (is_coredump).
    is_coredump = failure.startswith(COREDUMP_DETECTED)
    test_log = report.caplog
    coredumps = [
        (coredump["bin_path"], coredump["bt"], coredump["slurm_prefix"])
        for coredump in _related_coredumps(failure, report)
        if coredump["bin_path"]
    ]
    ctld_log = atf.get_log("slurmctld")

    reason = "Ticket Unknown: Fixed issue in slurmrestd in 25.05: SIGABRT in con_set_polling(): has_in || has_out"
    component = "sbin/slurmrestd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/conmgr/con.c" in bt
            and "src/conmgr/watch.c" in bt
            and "con_set_polling" in bt
            and "_handle_connection" in bt
            and "has_in || has_out" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 5):
                return False, reason
            else:
                return True, reason

    reason = "Ticket Unknown: Fixed issue in slurmdbd in 25.05: SIGABRT in _connection_fini_callback(): pthread_mutex_lock(): Invalid argument"
    component = "sbin/slurmdbd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/common/log.c" in bt
            and "src/slurmdbd/rpc_mgr.c" in bt
            and "_connection_fini_callback" in bt
            and "_service_connection" in bt
            and "fatal_abort" in bt
            and "pthread_mutex_lock" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 5):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 22310: Known issue when shutting down slurmdbd: SIGSEGV in _service_connection()"
    component = "sbin/slurmdbd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGSEGV" in bt
            and "src/common/persist_conn.c" in bt
            and "_service_connection" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (26, 5):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 22326: Known issue in slurmdbd: SIGABRT in  _list_find_first_lock(): Assertion (l != NULL)"
    component = "sbin/slurmdbd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/common/list.c" in bt
            and "src/slurmdbd/proc_req.c" in bt
            and "_list_find_first_lock" in bt
            and "_process_service_connection" in bt
            and "l != NULL" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 5):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 22370: Known issue in slurmd with TLS: SIGSEGV in _pack_buf_msg() relaying a response to slurmstepd. Fixed in 25.11+"
    component = "sbin/slurmd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGSEGV" in bt
            and "src/common/stepd_proxy.c" in bt
            and "in _pack_buf_msg" in bt
            and "in _slurmd_pack_msg_to_stepd" in bt
            and "in _on_data_local_socket" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 11):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 24122: Known issue in slurmctld: SIGABRT in conn_g_recv(): Assertion (plugin_inited == PLUGIN_INITED) failed"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/interfaces/conn.c" in bt
            and "src/common/slurm_protocol_socket.c" in bt
            and "src/common/forward.c" in bt
            and "conn_g_recv" in bt
            and "_fwd_tree_thread" in bt
            and "plugin_inited == PLUGIN_INITED" in bt
        ):
            # Fixed in 25.11.5 and 25.5.8
            if (
                (atf.get_version(component, slurm_prefix=slurm_prefix) < (25, 5))
                or (
                    (25, 5) < atf.get_version(component, slurm_prefix=slurm_prefix)
                    and atf.get_version(component, slurm_prefix=slurm_prefix)
                    < (25, 5, 8)
                )
                or (
                    (25, 11) < atf.get_version(component, slurm_prefix=slurm_prefix)
                    and atf.get_version(component, slurm_prefix=slurm_prefix)
                    < (25, 11, 5)
                )
            ):
                return True, reason
            else:
                return False, reason

    reason = "Ticket 24562: Known issue when shutting down slurmdbd: SIGABORT in acct_storage_g_close_connection(): Assertion (plugin_inited != PLUGIN_NOT_INITED). Fixed in 25.11.7+."
    component = "sbin/slurmdbd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/interfaces/accounting_storage.c" in bt
            and "acct_storage_g_close_connection" in bt
            and "plugin_inited != PLUGIN_NOT_INITED" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 11, 7):
                return False, reason
            else:
                return True, reason
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/interfaces/accounting_storage.c" in bt
            and "clusteracct_storage_g_node_up" in bt
            and "plugin_inited != PLUGIN_NOT_INITED" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 11, 7):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 24708: slurmstepd notified srun before REQUEST_STEP_COMPLETE, so a job could complete without its last step's exit code. Fixed in 26.05+"
    component = "sbin/slurmstepd"
    if not is_coredump and "jobcomp DerivedExitCode" in failure:
        if atf.get_version(component) >= (26, 5):
            return False, reason
        else:
            return True, reason

    reason = "Ticket 24853: Issue in slurmdbd in 25.05: SIGABRT in slurm_persist_conn_recv_thread_init: service_conn"
    component = "sbin/slurmdbd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/common/persist_conn.c" in bt
            and "src/slurmdbd/rpc_mgr.c" in bt
            and "slurm_persist_conn_recv_thread_init" in bt
            and "service_conn" in bt
            and "__xassert_failed" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix)[:2] != (25, 5):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 24905: Known issue with slurmstepd with stepmgr"
    component = "sbin/slurmstepd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGSEGV" in bt
            and "src/slurmd/slurmstepd/mgr.c" in bt
            and "_spawn_job_container" in bt
            and "for (uint32_t i = 0; i < step->node_tasks; i++)" in bt
        ):
            # "slurmstepd -V" won't work, but slurmd should have the same version
            if atf.get_version("sbin/slurmd", slurm_prefix=slurm_prefix) >= (25, 11, 6):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 24907: Known issue with slurmstepd and jobacct_gather"
    component = "sbin/slurmstepd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/interfaces/jobacct_gather.c" in bt
            and "jobacctinfo_aggregate" in bt
            and "step_partial_comp" in bt
            and "dest=0x0" in bt
        ):
            # "slurmstepd -V" won't work, but slurmd should have the same version
            if atf.get_version("sbin/slurmd", slurm_prefix=slurm_prefix) > (25, 11, 6):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 24952: Known issue with slurmctld auth_g_destroy(): Assertion (g_context_num > 0)"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/interfaces/auth.c" in bt
            and "auth_g_destroy" in bt
            and "slurm_receive_msgs" in bt
            and "g_context_num > 0" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 5):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 24991: Known issue with slurmstepd and _wait_for_job_running"
    component = "sbin/slurmstepd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGSEGV" in bt
            and "src/slurmd/slurmstepd/req.c" in bt
            and "_wait_for_job_running" in bt
            and "___pthread_mutex_lock" in bt
        ):
            # TODO: Add version when t24991 is fixed
            return False, reason

    reason = "Ticket 25013: Known issue with slurmctld: Assertion (con_flag(con, FLAG_READ_EOF) || con_flag(con, FLAG_IS_LISTEN)). Fixed in 25.05.8+"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/conmgr/con.c" in bt
            and "in close_con" in bt
            and "con_flag(con, FLAG_READ_EOF) || con_flag(con, FLAG_IS_LISTEN)" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 5, 8):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 25070: Known issue with slurmd: fatal: _forward_thread: pthread_mutex_lock()/unlock(): Invalid argument. Fixed in 25.11.6+"
    component = "sbin/slurmd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/common/log.c" in bt
            and "in fatal_abort" in bt
            and "in _forward_thread" in bt
            and (
                "%s: pthread_mutex_lock(): %m" in bt
                or "%s: pthread_mutex_unlock(): %m" in bt
            )
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 11, 6):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 25095: Known issue with slurmctld: SIGABRT: slurmdb_destroy_assoc_usage fixed in 25.11.6"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/common/slurmdb_defs.c" in bt
            and "in slurm_xfree" in bt
            and "in slurmdb_destroy_assoc_usage" in bt
            and "in slurmdb_free_assoc_rec_members" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 11, 6):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 25135: Known issue with slurmctld with s2n: SIGSEGV: _on_s2n_error. Fixed in 26.05.2+"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGSEGV" in bt
            and "src/plugins/tls/s2n/tls_s2n.c" in bt
            and "src/interfaces/conn.c" in bt
            and "in _on_s2n_error" in bt
            and "in _negotiate" in bt
            and "in tls_p_create_conn" in bt
            and "s2n_connection_get_delay" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (26, 5, 2):
                return False, reason
            else:
                return True, reason

    reason = "Ticket 25141: Known issue with slurmctld on reconfigure: SIGABRT in jobacct_storage_g_*_complete(): Assertion (plugin_inited != PLUGIN_NOT_INITED). Fixed in 25.11+"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/interfaces/accounting_storage.c" in bt
            and re.search(r"jobacct_storage_g_(step|job)_complete", bt)
            and "plugin_inited != PLUGIN_NOT_INITED" in bt
            and "Reconfigure signal (SIGHUP) received" in ctld_log
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (25, 11):
                return False, reason
            else:
                return True, reason
    if not is_coredump and re.search(
        r"jobacct_storage_g_(step|job)_complete\(\): Assertion \(plugin_inited != PLUGIN_NOT_INITED\) failed\.",
        ctld_log,
    ):
        if atf.get_version(component) >= (25, 11):
            return False, reason
        else:
            return True, reason

    reason = "Ticket 25406: slurmctld stalled on shutdown or reconfigure due to a lost poll interrupt (stuck in conmgr_quiesce()). Fixed in 26.05.3+"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal" not in bt
            and (
                (
                    "in conmgr_quiesce" in bt
                    and "in _slurmctld_background" in bt
                    and "in pollctl_poll" in bt
                )
                # The gcore may miss the stuck stack, but the log shows the stall
                # (incl. the idle slurmscriptd gcore'd together with slurmctld,
                # but not the slurmscriptd stuck in conmgr_fini of Issue 51358)
                or (
                    "in conmgr_fini" not in bt
                    and atf.get_log_delays(
                        "slurmctld",
                        r"Terminate signal|Reconfigure signal|slurmctld terminating",
                        r"Saving all slurm state",
                        min_delay=30,
                    )
                )
            )
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (26, 5, 3):
                return False, reason
            else:
                return True, reason
    if (
        "reconfigure" in failure and "Socket timed out" in failure
    ) or "Slurmctld is still running" in failure:
        stalls = atf.get_log_delays(
            "slurmctld",
            r"Terminate signal|Reconfigure signal|slurmctld terminating",
            r"Saving all slurm state",
            min_delay=30,
        )
        if stalls:
            logging.warning(
                f"{reason}:\n"
                + "\n".join(f"{start}\n{end}" for start, end, _ in stalls)
            )
            if atf.get_version(component) >= (26, 5, 3):
                return False, reason
            else:
                return True, reason

    reason = "Issue 50192: slurmrestd - SIGABRT in _foreach_add_path() on repeated -d data_parser. Fixed in 26.05.5+"
    component = "sbin/slurmrestd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/slurmrestd/openapi.c" in bt
            and "_foreach_add_path" in bt
            and "!data_key_get(spec->paths, path->path)" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (26, 5, 5):
                return False, reason
            else:
                return True, reason

    reason = "Issue 50963: salloc got the allocation but never ran its command"
    if "salloc not responding" in failure:
        allocated = set(
            re.findall(r"_slurm_rpc_allocate_resources JobId=(\d+)", ctld_log)
        )
        released = set(
            re.findall(r"_slurm_rpc_complete_job_allocation: JobId=(\d+)", ctld_log)
        )
        slurmd_logs = atf.get_log("slurmd")
        stuck = {
            job_id
            for job_id in allocated - released
            if not re.search(rf"\[{job_id}\.\d+\]", slurmd_logs)
        }
        if stuck:
            logging.warning(f"{reason}: JobId={','.join(sorted(stuck))}")
            return False, reason

    reason = "Issue 50974: Known issue about slurmd stuck waiting for a completing job during shutdown. Seems fixed in 26.05+"
    component = "sbin/slurmd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "pause_for_job_completion" in bt
            and "_rpc_terminate_job" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (26, 5):
                return False, reason
            else:
                return True, reason

    reason = "Issue 50979: OpenSSL atexit teardown race at shutdown (concurrent with OPENSSL_cleanup). Fixed in 26.11+."
    signals = ["SIGABRT", "SIGSEGV"]
    components = [
        "sbin/slurmd",
        "sbin/slurmctld",
        "bin/srun",
        "bin/salloc",
        "bin/sbatch",
        "bin/scancel",
        "bin/scontrol",
        "bin/sbcast",
    ]
    frames_in_bt = ["OPENSSL_cleanup", "__nptl_deallocate_tsd"]
    # The main thread may be caught inside OPENSSL_cleanup() (atexit), or
    # already past it running DSO destructors (_dl_fini & co.).
    teardown_frames = (
        "exit",
        "__GI_exit",
        "__run_exit_handlers",
        "__funcs_on_exit",
        "_dl_fini",
        "_dl_call_fini",
        "__cxa_finalize",
        "__do_global_dtors_aux",
    )
    for bin_path, bt, slurm_prefix in coredumps:
        component_match = next((c for c in components if bin_path.endswith(c)), None)
        in_teardown = any(f"in {frame} (" in bt for frame in teardown_frames)
        # libcrypto used after OPENSSL_cleanup() freed its global locks, e.g. a
        # TLS handshake on srun's listener while srun exits (the main thread may
        # not be unwindable by then).
        freed_crypto_lock = "rwlock=0x0" in bt and any(
            f"in CRYPTO_THREAD_{lock}_lock (" in bt for lock in ("read", "write")
        )
        if (
            component_match
            and any(f"Program terminated with signal {s}" in bt for s in signals)
            and "libcrypto.so" in bt
            and (
                (
                    in_teardown
                    and (
                        any(f in bt for f in frames_in_bt) or "in s2n_negotiate (" in bt
                    )
                )
                or freed_crypto_lock
            )
        ):
            if atf.get_version(component_match, slurm_prefix=slurm_prefix) >= (26, 11):
                return False, reason
            else:
                return True, reason

    reason = 'Issue 50980 (#51138 form): sattach exits 0 with "no tasks running" against a step that just started'
    if (
        "sattach" in failure
        and "no tasks running" in failure
        and "'exit_code': 0" in failure
    ):
        slurmd_logs = atf.get_log("slurmd")
        for step in re.findall(r"sattach (\d+\.\d+)", failure):
            if re.search(
                rf"_rpc_launch_tasks: starting for JobId=\S+ .*StepId={re.escape(step)}\b",
                slurmd_logs,
            ):
                logging.warning(f"{reason}: StepId={step}")
                # TODO: Add version when i50980 is fixed
                return False, reason

    reason = "Issue 51031 (!4273): Node events in the same second may be lost. Fixed in 26.11."
    component = "sbin/slurmdbd"
    if not is_coredump and re.search(
        r"sacctmgr\b.*\b(show|list)\s+events?\b", test_log, re.IGNORECASE
    ):
        inserted = set()
        lost = []
        for node, time in re.findall(
            r"DBD_NODE_STATE_(?:DOWN|TRES_DRAIN): NODE:(\S+) .* TIME:(\d+)",
            atf.get_log("slurmdbd"),
        ):
            if (node, time) in inserted:
                lost.append(f"{node}@{time}")
            inserted.add((node, time))
        if lost:
            logging.warning(f"{reason}: {','.join(lost)}")
            if atf.get_version(component) >= (26, 11):
                return False, reason
            else:
                return True, reason

    reason = "Issue 51060: slurmctld - SIGABRT: _kill_job_step(): Assertion (job_ptr->job_id == job_step_kill_msg->step_id.job_id) failed"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "src/slurmctld/job_mgr.c" in bt
            and "_kill_job_step" in bt
            and "job_ptr->job_id == job_step_kill_msg->step_id.job_id" in bt
        ):
            # TODO: Add version when i51060 is fixed
            return False, reason

    reason = "Issue 51111: srun/salloc - SIGABRT on shutdown: tls_create(): Assertion (con->input_fd >= 0) failed"
    for bin_path, bt, slurm_prefix in coredumps:
        components = ["bin/srun", "bin/salloc"]
        if (
            any(bin_path.endswith(c) for c in components)
            and "Program terminated with signal SIGABRT" in bt
            and "src/conmgr/tls.c" in bt
            and "in tls_create" in bt
            and "con->input_fd >= 0" in bt
        ):
            # TODO: Add version when i51111 is fixed
            return False, reason

    reason = "Issue 51119: srun - SIGSEGV in slurm_free_node_alias_addrs on powered down cloud node with stepmgr. Fixed in 26.05."
    component = "bin/srun"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGSEGV" in bt
            and "src/common/xmalloc.c" in bt
            and "src/common/slurm_protocol_defs.c" in bt
            and "src/api/allocate.c" in bt
            and "in slurm_xfree" in bt
            and "in slurm_free_node_alias_addrs_members" in bt
            and "in slurm_free_node_alias_addrs" in bt
            and "in slurm_job_step_create" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (26, 5):
                return False, reason
            else:
                return True, reason

    reason = "Issue 51306: slurmctld stalled sending an SRUN_* message to srun with an untrusted cert. Fixed in 26.11."
    component = "sbin/slurmctld"
    if (
        not is_coredump
        and atf.get_config_parameter(
            "TLSType", default="tls/none", live=False, quiet=True
        ).casefold()
        != "tls/none"
    ):
        stalls = atf.get_log_delays(
            "slurmctld",
            r"Spawning RPC agent for msg_type SRUN_",
            r"S2N_ERR_CERT_UNTRUSTED",
            min_delay=5,
        )
        if stalls:
            logging.warning(
                f"{reason}:\n"
                + "\n".join(f"{start}\n{end}" for start, end, _ in stalls)
            )
            if atf.get_version(component) >= (26, 11):
                return False, reason
            else:
                # convert to xfail
                return True, reason

    reason = "Issue 51358: slurmscriptd stuck in conmgr_fini() waiting for a watch() thread that already returned. Fixed in 26.11+"
    component = "sbin/slurmctld"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal" not in bt
            and "slurmctld: slurmscriptd" in bt
            and "in wait_for_watch" in bt
            and "in conmgr_fini" in bt
            and "in slurmscriptd_run_slurmscriptd" in bt
        ):
            if atf.get_version(component, slurm_prefix=slurm_prefix) >= (26, 11):
                return False, reason
            else:
                return True, reason

    reason = "Issue 51359: slurmd - SIGABRT on shutdown: slurm_unpack_msg_and_forward(): Assertion (fd >= 0) failed"
    component = "sbin/slurmd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "in slurm_unpack_msg_and_forward" in bt
            and "in _try_parse_rpc" in bt
            and "fd >= 0" in bt
        ):
            # TODO: Add version when i51359 is fixed
            return False, reason

    reason = "Issue 51360: sattach never exits after the attached step ends"
    if "sattach" in failure and (
        "not responding" in failure or re.search(r"\b110\b", failure)
    ):
        slurmd_logs = atf.get_log("slurmd")
        for step in re.findall(r"_handle_attach for StepId=(\S+)", slurmd_logs):
            if f"[{step}] done with step" in slurmd_logs:
                logging.warning(f"{reason}: StepId={step}")
                return False, reason

    reason = "Issue 51369: slurmd - double s2n_config_free() in _cleanup_tls_conn() racing certificate renewal"
    component = "sbin/slurmd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and re.search(r"Program terminated with signal SIG(SEGV|ABRT)", bt)
            and "in s2n_config_free" in bt
            and "in _cleanup_tls_conn" in bt
        ):
            # TODO: Add version when i51369 is fixed
            return False, reason

    reason = "Issue 51370: a one-way message to a client (e.g. SRUN_JOB_COMPLETE, SRUN_STEP_SIGNAL, SRUN_STEPS_DRAINED) was lost with s2n"
    # Only the client's own receive error: S2N_ERR_CLOSED alone is not enough,
    # and s2n_negotiate() errors are spurious handshake noise (Issue 50973)
    client_error = r"(srun|salloc|sattach|scrun|swait): error: tls_p_recv: s2n_recv\(\) failed S2N_ERR_CLOSED"
    if not is_coredump:
        found = re.search(client_error, failure) or (
            "swait" in failure
            and "swait: error: tls_p_recv: s2n_recv() failed S2N_ERR_CLOSED" in test_log
        )
        job = re.search(r"Job \((\d+)\) did not reach the \w+ state", failure)
        if not found and job:
            # Without swait's error in the test log (e.g. in the batch output),
            # the job didn't end although an swait was subscribed to its stepmgr
            # and all the steps launched in it already ended.
            job_id = job.group(1)
            slurmd_logs = atf.get_log("slurmd")
            steps = set(
                re.findall(
                    rf"_rpc_launch_tasks: starting for JobId={job_id} .*StepId=({job_id}\.\d+)",
                    slurmd_logs,
                )
            )
            found = (
                f"[{job_id}.extern stepmgr] debug:  Handling REQUEST_STEPS_DRAINED_SUBSCRIBE"
                in slurmd_logs
                and steps
                and all(f"[{step}] done with step" in slurmd_logs for step in steps)
            )
            if found:
                logging.warning(f"{reason}: JobId={job_id}")
        if found:
            # TODO: Add version when i51370 is fixed
            return False, reason

    reason = "Issue 51380: slurmd - SIGABRT on shutdown: on_extract(): Assertion (!get_buf_offset(con->in)) failed"
    component = "sbin/slurmd"
    for bin_path, bt, slurm_prefix in coredumps:
        if (
            component in bin_path
            and "Program terminated with signal SIGABRT" in bt
            and "in __xassert_failed" in bt
            and "!get_buf_offset(con->in)" in bt
            and "in on_extract" in bt
        ):
            # TODO: Add version when i51380 is fixed
            return False, reason

    # A coredump that is not a known issue is always a failure
    if is_coredump:
        coredump = analyze_coredump(failure[len(COREDUMP_DETECTED) :])
        if not coredump["bin_path"]:
            return False, f"Unknown coredump, {coredump['error']}"
        return False, f"Unknown coredump, see {coredump['bt_file']}"

    return None


def classify_failures(failures, report):
    """
    Return (xfail, reason) for the failures of a test report, or None if none
    of them is a known issue.

    Each failure is a string with the minimal info to classify it: the failure
    of a test, or one of the failures of the module teardown, like a teardown
    failure or a coredump found (COREDUMP_DETECTED followed by its path).
    The report is the pytest report of the phase that failed (see
    classify_failure()).

    The report is an xfail if all its failures are known issues to xfail.
    Otherwise it's a failure, and the reason has one line for each failure that
    is not an xfail, with its known reason if any.
    """

    verdicts = [(failure, classify_failure(failure, report)) for failure in failures]

    # The coredumps that explain another failure with the same known issue are
    # already reported, so the module teardown doesn't report them again.
    for failure, verdict in verdicts:
        if not verdict or failure.startswith(COREDUMP_DETECTED):
            continue
        for analysis in _related_coredumps(failure, report):
            coredump_failure = f"{COREDUMP_DETECTED}{analysis['path']}"
            if classify_failure(coredump_failure, report) == verdict:
                analysis["reported"] = True

    if not any(verdict for _, verdict in verdicts):
        return None

    if all(verdict and verdict[0] for _, verdict in verdicts):
        return True, "; ".join(dict.fromkeys(verdict[1] for _, verdict in verdicts))

    lines = []
    for failure, verdict in verdicts:
        if not verdict:
            lines.append(failure)
        elif not verdict[0]:
            lines.append(f"{verdict[1]}: {failure}")
    return False, "\n".join(lines)