diff --git a/.claude/skills/sglang-runtime-context/SKILL.md b/.claude/skills/sglang-runtime-context/SKILL.md index bb5f5f96b..730175ede 100644 --- a/.claude/skills/sglang-runtime-context/SKILL.md +++ b/.claude/skills/sglang-runtime-context/SKILL.md @@ -27,10 +27,13 @@ resolved configuration lives in the namespace bags.** - Every publishing process entry calls `publish(server_args, role=...)` (`run_scheduler_process`, the Ray `SchedulerActor`, the DP controller, tokenizer, - encoder, weight-cache daemon, launcher, ...). The one deliberate exception is the - detokenizer: its processes never publish and read only the raw config handed to - their constructors — code that can run detokenizer-side must not use the - namespace accessors. `publish` snapshots the resolved field + detokenizer, encoder, weight-cache daemon, ...); the roles are enumerated once, + as the keys of `ROLE_NAMESPACE_SETS` — there is no `launcher` role, the launch + path publishes as `tokenizer`. The remaining non-publisher is + `run_multi_detokenizer_router_process`: it *is* handed a `ServerArgs`, and uses + it only for `configure_logger(server_args)` today, so it has nothing to publish + for — a bag read added under that entry needs a `publish` at the entry first. + `publish` snapshots the resolved field values into the config bags; the accessors (`get_exec()` etc.) fail closed before it runs. `role` records which process type published, and keys per-role namespace enforcement: `SGLANG_ROLE_NAMESPACES=record` audits which namespaces each role's diff --git a/python/sglang/srt/disaggregation/encode_server.py b/python/sglang/srt/disaggregation/encode_server.py index 032485c82..5e1ff18e2 100644 --- a/python/sglang/srt/disaggregation/encode_server.py +++ b/python/sglang/srt/disaggregation/encode_server.py @@ -309,13 +309,15 @@ class MMEncoder: ``base_gpu_id + rank`` — the DP launcher's per-worker placement. It is this instance's value, not a config change, so it travels as an argument.""" + # The DP and TP encoder workers are spawned, so this constructor is + # the first publish in those processes. + publish(server_args, role="encoder") logger.info(f"init MMEncoder {rank}/{server_args.tp_size}") self.server_args = server_args configure_media_url_security( server_args.allowed_media_domains, server_args.media_url_max_file_size_mb, ) - publish(server_args, role="encoder") self.rank = rank # DP rank for metric labels; overridden by run_dp_worker in DP mode. # 0 in the single-instance (non-DP) path. @@ -3952,6 +3954,9 @@ def _unregister_encoder_url_from_bootstrap(server_args: ServerArgs): def launch_server(server_args: ServerArgs): configure_logger(server_args, prefix=" encode_server") + # Publish before the launch path reads configuration; the encoder built + # below re-projects the same object. + publish(server_args, role="encoder") if server_args.dp_size > 1: _launch_server_dp(server_args) return diff --git a/python/sglang/srt/entrypoints/engine.py b/python/sglang/srt/entrypoints/engine.py index 1b8ec6889..3e98f7208 100644 --- a/python/sglang/srt/entrypoints/engine.py +++ b/python/sglang/srt/entrypoints/engine.py @@ -99,6 +99,7 @@ from sglang.srt.observability.trace import process_tracing_init, trace_set_threa from sglang.srt.parser.template_detection import resolve_auto_parsers from sglang.srt.parser.template_manager import TemplateManager from sglang.srt.plugins import load_plugins +from sglang.srt.runtime_context import publish from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.utils import ( MultiprocessingSerializer, @@ -1109,6 +1110,11 @@ class Engine(EngineScoreMixin, EngineBase): ): resolve_auto_parsers(server_args) + # Resolution is complete here; this process goes on to host the + # tokenizer manager or the multi-tokenizer router, whose own publish + # re-projects the same object. + publish(server_args, role="tokenizer") + # Launch daemons (daemon mode only). The handles travel back to the # Engine that spawned them; shutdown() reaps from there. weight_cache_daemon_procs: List = [] diff --git a/python/sglang/srt/managers/detokenizer_manager.py b/python/sglang/srt/managers/detokenizer_manager.py index 19a7f4938..ad92bb817 100644 --- a/python/sglang/srt/managers/detokenizer_manager.py +++ b/python/sglang/srt/managers/detokenizer_manager.py @@ -39,6 +39,7 @@ from sglang.srt.managers.io_struct import ( ) from sglang.srt.managers.multi_tokenizer_mixin import MultiHttpWorkerDetokenizerMixin from sglang.srt.observability.cpu_monitor import start_cpu_monitor_thread +from sglang.srt.runtime_context import publish from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.utils import configure_logger, freeze_gc, kill_itself_when_parent_died from sglang.srt.utils.hf_transformers_utils import get_tokenizer @@ -520,6 +521,7 @@ def run_detokenizer_process( kill_itself_when_parent_died() setproctitle.setproctitle("sglang::detokenizer") configure_logger(server_args) + publish(server_args, role="detokenizer") parent_process = psutil.Process().parent() manager = None diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index 1d949342c..3f1e7e661 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -5016,6 +5016,8 @@ def run_scheduler_process( ): # Load plugins so hooks can override Scheduler and its dependencies. load_plugins() + # Publish before anything in this process reads configuration. + publish(server_args, role="scheduler") dp_rank = configure_scheduler_process( server_args, gpu_id, @@ -5029,9 +5031,6 @@ def run_scheduler_process( display_dp_rank=display_dp_rank, display_moe_ep_rank=display_moe_ep_rank, ) - # Scheduler.__init__ reads the config namespaces before the model - # worker's own publish. - publish(server_args, role="scheduler") parent_process = psutil.Process().parent() # Set up tracing diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 3ab16fa57..daca62449 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -393,6 +393,9 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin): ): # Parse args self.server_args = server_args + # In a tokenizer-worker process this is the process's first publish; + # the in-process path re-projects the object the launcher published. + set_global_server_args_for_tokenizer(server_args) self.startup_time: Optional[Dict[str, Any]] = None self._config_updates: List[Tuple[str, Dict[str, Any]]] = [] self.elastic_worker_count = server_args.dp_size @@ -407,7 +410,6 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin): self.skip_tokenizer_init = server_args.skip_tokenizer_init self.preferred_sampling_params = server_args.preferred_sampling_params self.crash_dump_folder = server_args.crash_dump_folder - set_global_server_args_for_tokenizer(server_args) # Init model config self.init_model_config() diff --git a/python/sglang/srt/ray/scheduler_actor.py b/python/sglang/srt/ray/scheduler_actor.py index 70a0255ac..d5323a423 100644 --- a/python/sglang/srt/ray/scheduler_actor.py +++ b/python/sglang/srt/ray/scheduler_actor.py @@ -75,6 +75,10 @@ class SchedulerActor: actual_gpu_id = gpu_id logger.info(f"[TP{tp_rank}] Using passed gpu_id: {gpu_id}") + # This actor takes the place of run_scheduler_process, which is where + # a forked scheduler publishes. + publish(server_args, role="scheduler") + # Configure worker (logging, process title, etc.) dp_rank = configure_scheduler_process( server_args, @@ -99,10 +103,6 @@ class SchedulerActor: f"[TP{tp_rank}] Bound to NUMA node {numa_node} for GPU {actual_gpu_id}" ) - # This actor constructs Scheduler directly (no run_scheduler_process), - # which reads the config namespaces before the model worker's publish. - publish(server_args, role="scheduler") - # Create scheduler (loads model into GPU, initializes NCCL) self.scheduler = Scheduler( server_args=server_args, diff --git a/python/sglang/srt/runtime_context.py b/python/sglang/srt/runtime_context.py index 01a809ce9..8e02aa62e 100644 --- a/python/sglang/srt/runtime_context.py +++ b/python/sglang/srt/runtime_context.py @@ -1155,6 +1155,7 @@ ROLE_NAMESPACE_SETS: dict[str, frozenset[str] | None] = { # a wrong set fails a request rather than a test. "tokenizer": None, # Deployment shapes not exercised locally; audit before restricting. + "detokenizer": None, "encoder": None, "expert_backup": None, "weight_cache_daemon": None, @@ -1259,9 +1260,8 @@ def _dump_recorded_namespace_reads() -> None: def publish(server_args, *, role: str, hf_config: Any = None) -> RuntimeContext: """Install process-wide config for this OS process. - Records the process ``role`` (``tokenizer`` / ``scheduler`` / - ``dp_controller`` / ``encoder`` / ``expert_backup`` / - ``weight_cache_daemon`` / ``test``) and + Records the process ``role`` — one of the ``ROLE_NAMESPACE_SETS`` keys, + which is the one place the roles are enumerated — and projects the config bags. Draft workers skip publish (they must not clobber the target). ``role`` is provenance, and — when ``SGLANG_ROLE_NAMESPACES`` is ``enforce`` — the key into ``ROLE_NAMESPACE_SETS`` for fail-closed diff --git a/test/registered/unit/test_publish_precedes_bag_reads.py b/test/registered/unit/test_publish_precedes_bag_reads.py new file mode 100644 index 000000000..aed5093cd --- /dev/null +++ b/test/registered/unit/test_publish_precedes_bag_reads.py @@ -0,0 +1,471 @@ +"""A process entry publishes before it reads a config namespace. + +The functions checked here are found by walking the package for `publish` +calls, not by listing them: a hand-kept list can name a function that no longer +exists and still pass, which is how the Ray actor entry went unchecked. + +Every such function starts a process (or is the first thing a spawned worker +runs), so the runtime context it inherits is empty. A bag read placed above the +publish raises `config namespace ... not published` -- in a spawned worker, +which no unit test starts, so the failure only shows up as a server that never +comes up. + +A process entry reaches its bag reads through what it calls -- `Scheduler(...)`, +`configure_scheduler_process(...)`, `self.init_tokenizer_and_processor()` -- not +by naming an accessor itself, so a scan of the entry's own body sees nothing to +order and passes whatever the code does. The read line is therefore taken over +what the entry calls: a call resolved inside the module, through a parameter's +default (`detokenizer_manager_class=DetokenizerManager`), or one hop out through +that module's import table, followed the same way at every depth. Following the +import table only out of the entry's own body would stop one call short of the +expert-backup read, which is reached as `ExpertBackupManager(...)` -> +`backup_weights_from_disk` -> imported loader code -> `get_model()`. + +Reaching no read is not a pass. The walk is a static one, and every call it +cannot resolve -- a callable handed in as a parameter, an attribute off +something other than `self` -- turns into "reaches no accessor", which is also +what a defect looks like. So every publisher that reaches none is named in +`_UNREAD_ENTRIES` with the reason, and that map is asserted against the whole +set the walk finds: a publisher that stops reaching a read, or a new one that +never reached any, fails here instead of becoming an entry with nothing to +check. Restricting the comparison to `_KNOWN_ENTRIES` would exempt exactly the +newly discovered entry the derivation exists to catch. + +What this cannot pin: a publish moving across code that reads only the handed +`server_args` instance. Such code names no accessor, so there is no read for the +walk to order it against. +""" + +import ast +import pathlib +import unittest + +import sglang +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(est_time=40, suite="base-a-test-cpu") + +_PACKAGE_ROOT = pathlib.Path(sglang.__file__).resolve().parent + +_ACCESSORS = frozenset( + { + "get_exec", + "get_memory", + "get_schedule", + "get_model", + "get_spec", + "get_serving", + "get_observability", + "get_disagg", + "get_lora", + "get_mm", + "get_device", + "get_parallel", + } +) + +# The process entries that must be found by the walk. A derivation that stops +# matching -- an import rewritten, a publish moved behind a helper -- would +# otherwise leave this test green over an empty set. +_KNOWN_ENTRIES = frozenset( + { + ("srt/managers/scheduler.py", "run_scheduler_process"), + ("srt/managers/detokenizer_manager.py", "run_detokenizer_process"), + ( + "srt/managers/data_parallel_controller.py", + "run_data_parallel_controller_process", + ), + ("srt/ray/scheduler_actor.py", "__init__"), + ("srt/disaggregation/encode_server.py", "__init__"), + ("srt/disaggregation/encode_server.py", "launch_server"), + ("srt/managers/tokenizer_manager.py", "__init__"), + ("srt/entrypoints/engine.py", "_launch_subprocesses"), + ( + "srt/elastic_ep/expert_backup_manager.py", + "run_expert_backup_manager_process", + ), + ("srt/weight_cache/daemon.py", "load"), + } +) + +# Every publisher the walk finds whose callees reach no bag accessor at this +# revision, and why. Asserted exactly against what the walk finds -- not +# intersected with `_KNOWN_ENTRIES`, which would drop a newly discovered entry +# and check no ordering for the one case the derivation exists to catch. +# "Reaches none" is also what the walk answers when it cannot resolve a call, so +# every one of them is named. An entry leaves this map in the commit that gives +# it a bag read. +_UNREAD_ENTRIES: dict = { + # Not process entries: these publish to set up a context for themselves. + ("kernels/aot/tests/test_fused_qk_norm_rope.py", "test_fused_qk_norm_rope"): ( + "a kernel test publishing its own context" + ), + ("multimodal_gen/test/unit/test_disagg_trace.py", "_srt_trace_server_args"): ( + "a trace fixture publishing its own context" + ), + ("srt/entrypoints/engine.py", "_launch_subprocesses"): ( + "its subprocess targets and its tokenizer-manager factory arrive as " + "parameters, so the walk resolves none of them" + ), + ("srt/managers/detokenizer_manager.py", "run_detokenizer_process"): ( + "DetokenizerManager reads the handed instance at this revision" + ), + ("srt/managers/tokenizer_manager.py", "__init__"): ( + "the constructor and the init_* helpers it calls read the handed " + "instance at this revision" + ), +} + +# `publish` itself and its named wrappers live here; a call inside them is the +# definition, not a process entry. +_PUBLISH_HOMES = frozenset({"srt/runtime_context.py", "srt/server_args.py"}) + +_CONFIG_MODULES = frozenset({"sglang.srt.runtime_context", "sglang.srt.server_args"}) + + +def _module_path(dotted: str): + """The package-relative file a `sglang.` import names, if it is one.""" + if not dotted.startswith("sglang."): + return None + parts = dotted.split(".")[1:] + for candidate in ("/".join(parts) + ".py", "/".join(parts) + "/__init__.py"): + if (_PACKAGE_ROOT / candidate).exists(): + return candidate + return None + + +_CALLS = {} + + +def _calls(fn): + """(callee key, line) per call in this body. + + `self.f()` / `cls.f()` are keyed to the owning class; a bare name is a + module-level def, an imported name, or a class -- for a class the call runs + its `__init__`. `x.f()` for any other `x` yields two keys: the bare `x`, + which resolves when `x` is a class (`PortArgs.init_new()`), and a + module-qualified one that keeps `f`, which resolves when `x` is a module + the file imported. Keeping only the bare name loses `f` entirely, so a + helper called as `foo.initialize()` contributes nothing to the walk. + Anything deeper (`a.b.c()`, a callable off an attribute) is not resolved. + """ + if id(fn) in _CALLS: + return _CALLS[id(fn)] + out = [] + for node in ast.walk(fn): + if not isinstance(node, ast.Call): + continue + func = node.func + if isinstance(func, ast.Name): + out.append((("name", func.id), node.lineno)) + elif isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name): + if func.value.id in ("self", "cls"): + out.append((("self", func.attr), node.lineno)) + else: + out.append((("name", func.value.id), node.lineno)) + out.append(((f"module:{func.value.id}", func.attr), node.lineno)) + _CALLS[id(fn)] = out + return out + + +class _Module: + """One parsed module: what it calls the config API, and what it defines. + + The publisher and accessor names are resolved from the imports rather than + matched by name: a model's ``index_topk_share.publish()`` and a platform's + ``get_device()`` are unrelated methods that a name-only match reports as + config calls. + """ + + def __init__(self, rel: str, tree): + self.rel, self.tree = rel, tree + self.publishers, self.accessors, self.qualified = set(), set(), set() + self.imported = {} + # Names bound to another sglang module rather than to a symbol in one: + # `import sglang.srt.foo as foo` / `from sglang.srt import foo`. Without + # these, `foo.initialize()` reaches nothing. + self.modules = {} + self.functions = {} + self.classes = {} + self.owner = {} + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + if node.module in _CONFIG_MODULES: + for alias in node.names: + local = alias.asname or alias.name + if alias.name == "publish" or alias.name.startswith( + "set_global_server_args" + ): + self.publishers.add(local) + elif alias.name in _ACCESSORS: + self.accessors.add(local) + target = _module_path(node.module) + if target is not None and target != rel: + for alias in node.names: + self.imported[alias.asname or alias.name] = (target, alias.name) + for alias in node.names: + # `from sglang.srt import runtime_context` binds the module + # itself; `runtime_context.get_serving()` is the same read. + dotted = f"{node.module}.{alias.name}" + if dotted in _CONFIG_MODULES: + self.qualified.add(alias.asname or alias.name) + bound = _module_path(dotted) + if bound is not None and bound != rel: + self.modules[alias.asname or alias.name] = bound + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.name in _CONFIG_MODULES: + self.qualified.add(alias.asname or alias.name.split(".")[0]) + # Only `import a.b.c as name` binds a usable name; without + # the alias the call reads `a.b.c.f()`, which is deeper than + # this walk resolves. + bound = _module_path(alias.name) if alias.asname else None + if bound is not None and bound != rel: + self.modules[alias.asname] = bound + elif isinstance(node, ast.ClassDef): + methods = self.classes.setdefault(node.name, {}) + for stmt in node.body: + if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)): + methods[stmt.name] = stmt + self.owner[id(stmt)] = node.name + for node in ast.walk(tree): + if ( + isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and id(node) not in self.owner + ): + self.functions.setdefault(node.name, node) + self._direct = {} + + def is_publish(self, node) -> bool: + if not isinstance(node, ast.Call): + return False + if isinstance(node.func, ast.Name) and node.func.id in self.publishers: + return True + return ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id in self.qualified + ) + + def is_read(self, node) -> bool: + if not isinstance(node, ast.Call): + return False + if isinstance(node.func, ast.Name) and node.func.id in self.accessors: + return True + return ( + isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id in self.qualified + and node.func.attr in _ACCESSORS + ) + + def resolve(self, key, cls): + """The def in this module a callee key names, if it names one. + + A module-qualified key names nothing here: its def lives in the module + the alias is bound to, which `_targets` follows. + """ + kind, name = key + if kind == "self": + return self.classes.get(cls, {}).get(name) + if kind.startswith("module:"): + return None + if name in self.functions: + return self.functions[name] + return self.classes.get(name, {}).get("__init__") + + def direct_read(self, fn) -> bool: + """Whether this body names an accessor itself.""" + if id(fn) not in self._direct: + self._direct[id(fn)] = any(self.is_read(n) for n in ast.walk(fn)) + return self._direct[id(fn)] + + +_MODULES = {} + + +def _module(rel: str): + # A module this walk cannot parse would resolve to "reaches no config", + # which is the answer that hides a defect. utf-8-sig because a file in the + # package carries a BOM; anything still unparsable fails the test. + if rel not in _MODULES: + _MODULES[rel] = _Module( + rel, ast.parse((_PACKAGE_ROOT / rel).read_text(encoding="utf-8-sig")) + ) + return _MODULES[rel] + + +def _defaulted_parameters(fn): + """`{parameter: default name}` for the parameters this def gives a plain + name as default. `run_detokenizer_process` reaches `DetokenizerManager` + only this way -- the body calls the parameter, and the class it is really + handed is written once, as that parameter's default.""" + if id(fn) in _DEFAULTS: + return _DEFAULTS[id(fn)] + arguments = fn.args + positional = arguments.posonlyargs + arguments.args + pairs = list( + zip(positional[len(positional) - len(arguments.defaults) :], arguments.defaults) + ) + pairs += zip(arguments.kwonlyargs, arguments.kw_defaults) + _DEFAULTS[id(fn)] = { + parameter.arg: default.id + for parameter, default in pairs + if isinstance(default, ast.Name) + } + return _DEFAULTS[id(fn)] + + +_DEFAULTS = {} + + +def _targets(mod, key, cls, fn=None): + """The (module, def, owning class) a callee key names, here and one hop out. + + A bare name that is one of `fn`'s parameters resolves through that + parameter's default as well, which is how a factory handed in as an + argument is followed to the class the entry actually constructs. + + A module-qualified key (`foo.initialize()`) resolves in the module `foo` is + bound to, so a helper reached that way joins the walk instead of dropping + out of it. + """ + keys = [key] + if fn is not None and key[0] == "name": + default = _defaulted_parameters(fn).get(key[1]) + if default is not None: + keys.append(("name", default)) + for key in keys: + target = mod.resolve(key, cls) + if target is not None: + yield mod, target, mod.owner.get(id(target), cls) + if key[0].startswith("module:"): + home = mod.modules.get(key[0][len("module:") :]) + if home is None: + continue + other = _module(home) + target = other.resolve(("name", key[1]), None) + if target is not None: + yield other, target, other.owner.get(id(target)) + continue + hop = mod.imported.get(key[1]) if key[0] == "name" else None + if hop is None: + continue + other = _module(hop[0]) + target = other.resolve(("name", hop[1]), None) + if target is not None: + yield other, target, other.owner.get(id(target)) + + +# The callee names from a def down to the read it reaches, for the defs a +# witness has been found for. Only "reaches a read" is carried between +# questions: a witness path stays one, while "reaches none" can be the answer a +# recursive call gets when it re-enters a def the walk is still inside, which +# is that call's answer and not the def's own. +_WITNESS = {} + + +def _witness(mod, fn, cls, seen): + """The callees from this def down to a bag read, ending in the file that + reads, or None for a def that reaches no read. + + `seen` belongs to one question. Skipping a def already on this walk keeps + the answer for the def the question was asked about -- that def opened the + skipped one, so a read below it still comes back up the path that opened + it -- and bounds the walk on a call cycle. + """ + key = (mod.rel, id(fn)) + if key in _WITNESS: + return _WITNESS[key] + if key in seen: + return None + seen.add(key) + if mod.direct_read(fn): + _WITNESS[key] = [mod.rel] + return _WITNESS[key] + for call, _ in _calls(fn): + for target in _targets(mod, call, cls, fn): + below = _witness(*target, seen) + if below is not None: + _WITNESS[key] = [call[1]] + below + return _WITNESS[key] + return None + + +def _publishing_functions(): + """(relative path, function node, module) per publisher.""" + for path in sorted(_PACKAGE_ROOT.rglob("*.py")): + rel = path.relative_to(_PACKAGE_ROOT).as_posix() + if rel in _PUBLISH_HOMES: + continue + mod = _module(rel) + if mod is None or not mod.publishers: + continue + for fn in ast.walk(mod.tree): + if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if any(mod.is_publish(n) for n in ast.walk(fn)): + yield rel, fn, mod + + +def _first_read(fn, mod): + """(line, what read it) of the earliest config read this entry reaches.""" + cls = mod.owner.get(id(fn)) + marks = [(n.lineno, "a bag accessor") for n in ast.walk(fn) if mod.is_read(n)] + for call, line in _calls(fn): + for target in _targets(mod, call, cls, fn): + if target[1] is fn: + continue + below = _witness(*target, set()) + if below is not None: + chain = [call[1]] + below + marks.append( + (line, " -> ".join(chain[:-1]) + f", which reads in {chain[-1]}") + ) + break + return min(marks) if marks else None + + +class TestPublishPrecedesBagReads(CustomTestCase): + def test_every_publishing_entry_publishes_first(self): + offenders = [] + found = set() + unread = set() + for rel, fn, mod in _publishing_functions(): + found.add((rel, fn.name)) + # ast.walk yields breadth-first, so the first match is not the + # earliest line; take the minimum. + publish_line = min(n.lineno for n in ast.walk(fn) if mod.is_publish(n)) + read = _first_read(fn, mod) + if read is None: + unread.add((rel, fn.name)) + elif read[0] < publish_line: + offenders.append( + f"{rel}:{fn.name} reaches a config namespace at line " + f"{read[0]} through {read[1]}, before its publish at " + f"{publish_line}" + ) + self.assertEqual( + sorted(_KNOWN_ENTRIES - found), + [], + "the walk stopped finding known process entries; the derivation " + "is broken, not the tree", + ) + self.assertEqual( + sorted(unread), + sorted(_UNREAD_ENTRIES), + "a publisher reaching no bag accessor checks nothing; either the " + "walk stopped resolving a call, or a publisher appeared or moved " + "its reads and _UNREAD_ENTRIES has to say so", + ) + self.assertEqual( + offenders, + [], + "a spawned worker starts with an empty context:\n " + + "\n ".join(offenders), + ) + + +if __name__ == "__main__": + unittest.main()