myxce/scripts/check-no-fork.py

#!/usr/bin/env python3
"""Enforce the no-fork rule (.kiro/steering/tech.md, "Process creation").

XCE daemons are multi-threaded (libxce delaymgr/netio/vtaskmgr threads, xcecd's
thread pool); fork() in a multi-threaded process copies only the calling thread,
so any lock another thread held (malloc, logger, stdio) stays locked in the
child. Component code therefore never calls fork-family functions; children are
started with posix_spawn(), or by systemd over D-Bus.

Scanned: C sources of the components the rule covers (COVERED below), minus
test code. Excluded by decision (2026-10-10): tests, the harness (xce_test),
the autoscaler (not finalized; provider calls use posix_spawn since bug-1969)
and the Go frontend.

A call is allowed only in a process proven single-threaded, marked on the call
line or the line above with:  /* xce-fork-ok: <why it is single-threaded> */

Usage: check-no-fork.py [repo-root] [file ...]   (exit 1 on a violation)
"""
import os
import re
import sys

COVERED = ["src/controller", "src/scheduler", "src/lib", "src/worker",
           "src/esd"]
EXCLUDED_DIRS = {"testsuite", "test", "tests"}
# Fork-family calls. uv_spawn() is libuv's fork()+exec on Linux.
BANNED = re.compile(
    r"(?<![\w.>])(fork|vfork|forkpty|daemon|system|popen|uv_spawn)\s*\(")
MARK = re.compile(r"xce-fork-ok:\s*\S")


def strip_code(text):
    """Blank comments and string/char literals, keeping line structure, so
    names inside them never match. Returns the stripped text."""
    out, i, n = [], 0, len(text)
    while i < n:
        c = text[i]
        if text.startswith("/*", i):
            j = text.find("*/", i + 2)
            j = n if j < 0 else j + 2
            out.append(re.sub(r"[^\n]", " ", text[i:j]))
            i = j
        elif text.startswith("//", i):
            j = text.find("\n", i)
            j = n if j < 0 else j
            out.append(" " * (j - i))
            i = j
        elif c in "\"'":
            j = i + 1
            while j < n and text[j] != c and text[j] != "\n":
                j += 2 if text[j] == "\\" else 1
            j = min(j + 1, n)
            out.append(c + " " * (j - i - 2) + c if j - i >= 2 else c)
            i = j
        else:
            out.append(c)
            i += 1
    return "".join(out)


def disabled_lines(lines):
    """Line indexes inside `#if 0` blocks (nesting aware)."""
    dead, depth = set(), 0
    for k, line in enumerate(lines):
        s = line.strip()
        if depth:
            if re.match(r"#\s*if", s):
                depth += 1
            elif re.match(r"#\s*endif", s):
                depth -= 1
            elif depth == 1 and re.match(r"#\s*(else|elif)", s):
                depth = 0
            dead.add(k)
        elif re.match(r"#\s*if\s+0\b", s):
            depth = 1
            dead.add(k)
    return dead


def check_file(path, rel):
    raw = open(path, encoding="utf-8", errors="replace").read()
    raw_lines = raw.split("\n")
    code_lines = strip_code(raw).split("\n")
    dead = disabled_lines(raw_lines)
    bad = []
    for k, line in enumerate(code_lines):
        if k in dead:
            continue
        for m in BANNED.finditer(line):
            prev = raw_lines[k - 1] if k else ""
            if MARK.search(raw_lines[k]) or MARK.search(prev):
                continue
            bad.append(f"{rel}:{k + 1}: {m.group(1)}() "
                       f"{raw_lines[k].strip()}")
    return bad


def covered_files(root):
    for top in COVERED:
        for d, dirs, files in os.walk(os.path.join(root, top)):
            dirs[:] = [x for x in dirs if x not in EXCLUDED_DIRS
                       and not x.startswith(".")]
            for f in files:
                if f.endswith((".c", ".h")) and not f.startswith("test"):
                    yield os.path.join(d, f)


def main(argv):
    root = os.path.abspath(argv[1]) if len(argv) > 1 else os.getcwd()
    if len(argv) > 2:
        files = [os.path.join(root, f) for f in argv[2:]]
        covered = set(map(os.path.abspath, covered_files(root)))
        files = [f for f in files if os.path.abspath(f) in covered]
    else:
        files = list(covered_files(root))
    bad = []
    for f in sorted(files):
        bad += check_file(f, os.path.relpath(f, root))
    if bad:
        print("no-fork rule violated (.kiro/steering/tech.md, "
              "\"Process creation\"):")
        print("\n".join("  " + b for b in bad))
        print("Use posix_spawn() (or a systemd unit over D-Bus). If the process "
              "is proven single-threaded, mark the call with "
              "/* xce-fork-ok: <why> */.")
        return 1
    print(f"no-fork rule: {len(files)} file(s) OK")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))