diff --git a/.aspect/axl.axl b/.aspect/axl.axl index f5eef9cdc..c60c34bf4 100644 --- a/.aspect/axl.axl +++ b/.aspect/axl.axl @@ -17,6 +17,7 @@ load("@aspect//private/lib/bazel_results.axl", "bb_clientd_root", "compute_repro load("@aspect//private/lib/check_dispatch.axl", "SURFACE_PR_COMMENT", "SURFACE_STATUS_CHECK", "TEMPLATE_SCOPE_KEYS", "resolve_templates", "snippet_budget_for") load("@aspect//private/lib/ci.axl", "detect_build_url") load("@aspect//private/lib/environment.axl", "color_enabled", "detect_ci", "parse_git_url_name", "sanitize_filename") +load("@aspect//private/lib/execlog.axl", "execlog") load( "@aspect//private/lib/gazelle_results.axl", "dedupe_subtrees", @@ -4139,6 +4140,7 @@ def impl(ctx: TaskContext) -> int: tc = test_result_cell_text(tc) tc = test_results_subset_carries_progress(tc) tc = test_denied_terminal_post_warning(tc) + tc = test_execlog_lib(ctx, tc, temp_dir) print(tc, "tests passed") return 0 @@ -4239,3 +4241,88 @@ axl = task( # the test run sees at most one fire per id. traits = tips.TRAITS, ) + +def test_execlog_lib(ctx: TaskContext, tc: int, temp_dir: str) -> int: + """Covers the three pieces of `private/lib/execlog.axl` that are pure + functions of their inputs: input-set flattening, and the two differs that + turn a pair of resolved spawns into the lines a reader sees. + + The end-to-end path (reading a real compact log) is exercised by + `aspect execlog diff` in CI, not here โ€” it needs a build to produce a log.""" + + # A set whose members are files, keyed by entry id, resolves to sorted + # `path@digest` strings. + paths = {1: "a.txt", 2: "b.txt", 3: "c.txt"} + digests = {1: "aaa", 2: "bbb", 3: "ccc"} + sets = {10: ([1, 2], []), 11: ([3], [10])} + + flat = execlog.testonly_flatten(sets, paths, digests, 10, 100) + tc = test_case(tc, flat == ["a.txt@aaa", "b.txt@bbb"], "flatten resolves a flat input set") + + # A nested set contributes its own members and everything below it, once. + flat = execlog.testonly_flatten(sets, paths, digests, 11, 100) + tc = test_case( + tc, + flat == ["a.txt@aaa", "b.txt@bbb", "c.txt@ccc"], + "flatten walks transitive sets", + ) + + # Input set id 0 is Bazel's "no inputs", not a set to look up. + tc = test_case(tc, execlog.testonly_flatten(sets, paths, digests, 0, 100) == [], "flatten treats id 0 as empty") + + # A cycle cannot hang the walk. Bazel does not emit one, but a truncated + # log can leave a dangling id, and a wedged CLI is a worse failure than a + # wrong answer. + cyclic = {20: ([1], [21]), 21: ([2], [20])} + flat = execlog.testonly_flatten(cyclic, paths, digests, 20, 100) + tc = test_case(tc, flat == ["a.txt@aaa", "b.txt@bbb"], "flatten terminates on a cyclic set") + + # An id with no entry (a log that references something it never recorded) + # is skipped rather than failing the whole diff. + flat = execlog.testonly_flatten({30: ([1, 999], [])}, paths, digests, 30, 100) + tc = test_case(tc, flat == ["a.txt@aaa"], "flatten skips an unknown id") + + # The maps differ reports changed, added and removed separately, because + # the three have different causes: a rebuilt input, a new dep, a dropped one. + reason = execlog.testonly_diff_maps( + "inputs", + {"same.txt": "h1", "moved.txt": "h2", "gone.txt": "h3"}, + {"same.txt": "h1", "moved.txt": "CHANGED", "new.txt": "h4"}, + ) + tc = test_case(tc, reason.kind == "inputs", "diff_maps keeps its kind") + tc = test_case(tc, reason.summary == "1 changed, 1 added, 1 removed", "diff_maps counts each bucket") + tc = test_case( + tc, + reason.items == ["changed moved.txt", "added new.txt", "removed gone.txt"], + "diff_maps orders changed, then added, then removed", + ) + + # Identical maps produce no items, which is what suppresses the section. + reason = execlog.testonly_diff_maps("env", {"A": "1"}, {"A": "1"}) + tc = test_case(tc, reason.items == [], "diff_maps is empty when nothing moved") + + # Args are positional, so the differ reports the index: an inserted flag + # shifts everything after it, and saying so is more useful than a set diff. + reason = execlog.testonly_diff_lists(["cc", "-c", "a.c"], ["cc", "-O2", "-c", "a.c"]) + tc = test_case(tc, len(reason.items) == 3, "diff_lists reports every shifted position") + tc = test_case(tc, reason.items[0] == "[1] -c -> -O2", "diff_lists names the index that changed") + tc = test_case( + tc, + reason.items[2] == "[3] -> a.c", + "diff_lists marks a position the shorter side does not have", + ) + + # The hash is order-sensitive and separator-safe: two lists that concatenate + # to the same string must not collide, or a diff misses a real change. + tc = test_case( + tc, + execlog.testonly_hash(["ab", "c"]) != execlog.testonly_hash(["a", "bc"]), + "hash does not collide across element boundaries", + ) + tc = test_case( + tc, + execlog.testonly_hash(["a", "b"]) == execlog.testonly_hash(["a", "b"]), + "hash is stable", + ) + + return tc diff --git a/crates/aspect-cli/src/builtins/aspect/MODULE.aspect b/crates/aspect-cli/src/builtins/aspect/MODULE.aspect index 43870780f..76daeb069 100644 --- a/crates/aspect-cli/src/builtins/aspect/MODULE.aspect +++ b/crates/aspect-cli/src/builtins/aspect/MODULE.aspect @@ -11,6 +11,7 @@ use_task("test.axl", "test") use_task("axl_add.axl", "add") use_task("delivery.axl", "delivery") use_task("cache_diff.axl", "diff") +use_task("execlog.axl", "diff", "list_actions") use_task("lint.axl", "lint") use_task("format.axl", "format") use_task("gazelle.axl", "gazelle") diff --git a/crates/aspect-cli/src/builtins/aspect/README.md b/crates/aspect-cli/src/builtins/aspect/README.md index c819a7761..dd1bccfa9 100644 --- a/crates/aspect-cli/src/builtins/aspect/README.md +++ b/crates/aspect-cli/src/builtins/aspect/README.md @@ -14,6 +14,7 @@ Aspect-CLI ships with six built-in tasks that drive Bazel for the most common CI | [format](#format) | `bazel run` of a `format_multirun` | `aspect format [--scope=changed\|all]` | `format_results` | | [gazelle](#gazelle) | `bazel run` of a `gazelle()` / `aspect_gazelle()` target | `aspect gazelle [--check]` | `gazelle_results` | | [delivery](#delivery) | Multi-phase delivery flow | `aspect delivery //pkg/foo:release //pkg/bar:release` | `delivery_results` | +| [execlog](#execlog) | Reads a compact execution log | `aspect execlog diff before.binpb.zst after.binpb.zst` | none (no Bazel run) | Every task: @@ -177,6 +178,93 @@ Renderer: `delivery_results`. The body shows counts-by-outcome, per-outcome tabl --- + +--- + +## execlog + +[`execlog.axl`](execlog.axl) ยท offline analysis of `--execution_log_compact_file` artifacts. +The only built-in tasks that run no Bazel command: they read logs a previous build produced. + +### `aspect execlog diff ` + +Two builds that should have been identical were not. This says which actions moved and why. + +``` +$ aspect execlog diff /tmp/before.binpb.zst /tmp/after.binpb.zst +2 action(s) changed and would not have been a cache hit: + +Genrule //gen:banner + output: bazel-out/darwin_arm64-fastbuild/bin/gen/banner.txt + inputs: 1 changed, 0 added, 0 removed + changed gen/greeting.txt + +Genrule //gen:report + output: bazel-out/darwin_arm64-fastbuild/bin/gen/report.txt + inputs: 1 changed, 0 added, 0 removed + changed bazel-out/darwin_arm64-fastbuild/bin/gen/banner.txt +``` + +That is the propagation chain you want: a source file changed, so the action reading it changed, +so the action reading *its* output changed. An action whose key did not move is not listed. + +Five dimensions of the action key are compared โ€” `args`, `env`, `inputs`, `tools`, `platform` โ€” +and every one that moved is reported, not just the first, because an action with two problems +sends the reader round the loop twice otherwise. `env` is the one that finds non-hermeticity: + +``` +Genrule //gen:stable + env: 1 changed, 0 added, 0 removed + changed DEMO_TOKEN +``` + +Exits 0 by default: a changed action is a finding, not an error. `--fail-on-change` makes it fail +a CI step asserting two builds are identical. + +### `aspect execlog list ` + +One line per action: mnemonic, label, runner, cache hit. `--output=json` adds each action's +computed key, which is what `diff` compares. `--mnemonic=` and `--cached=` filter. + +### How it fits in memory + +A 20 MB compact log expands to gigabytes once every input set is flattened, and a diff needs two. +So [`lib/execlog.axl`](private/lib/execlog.axl) never holds a flattened log: + +1. **Index** each log once, keeping three id-keyed maps and one `Fingerprint` per spawn. A + fingerprint is five hashes and a label, so it scales with the number of actions rather than + the number of files. +2. **Detail** only the spawns whose fingerprints differ, by walking the logs again. On a healthy + build that is a handful of actions out of tens of thousands, and it is skipped entirely when + nothing differs. + +Two passes over a compressed file are much cheaper than one flattened log in memory. +`--summary-only` stops after the first pass when even that is too much. + +Ported from the Go `execlog-diff` tool, which streams and diffs concurrently; this is the same +analysis with a bounded working set instead of a pipeline. + +### Reading a log from your own task + +`bazel.execution_log.read(path = ...)` returns an iterator of `ExecLogEntry` decoded from a +compact log on disk, on a background thread, so a log larger than memory can be walked as long as +the loop body does not keep every entry. + +```python +entries = bazel.execution_log.read(path = "before.binpb.zst") +spawns = 0 +for entry in entries: + if type(entry.type) == "spawn": + spawns += 1 +if entries.error() != None: + ctx.std.process.exit(1, "before.binpb.zst: " + entries.error()) +``` + +A missing file, or one that is not a zstd frame, fails immediately with a traceback. A log that is +truncated or corrupt part-way through ends the iteration early and sets `error()`, so check it when +a partial read would give a wrong answer. See `spawn_from_path` for the one truncation case zstd +cannot report. + ## Cross-cutting features | Feature | File | Activation | What it does | diff --git a/crates/aspect-cli/src/builtins/aspect/execlog.axl b/crates/aspect-cli/src/builtins/aspect/execlog.axl new file mode 100644 index 000000000..76c1a0497 --- /dev/null +++ b/crates/aspect-cli/src/builtins/aspect/execlog.axl @@ -0,0 +1,219 @@ +"""`aspect execlog diff` and `aspect execlog list`. + +Answers the question a cache-miss investigation actually starts with: two builds +that should have been identical were not, and which action moved first? + +`diff` compares two compact execution logs and, for every action present in +both whose action key changed, names the dimension that changed and the exact +arguments, environment variables, or input files behind it. +""" + +load("./private/lib/execlog.axl", "execlog") + +_MAX_ITEMS = 12 + +def _require_file(ctx: TaskContext, path: str, what: str) -> None: + if not ctx.std.fs.exists(path): + ctx.std.process.exit(1, "{} not found: {}".format(what, path)) + +def _render(ctx: TaskContext, result, verbose: bool) -> None: + out = ctx.std.io.stdout + + for warning in result.truncated: + print("WARNING: execution log ended early, the comparison is partial. " + warning) + + if not result.changed: + out.write("No action in both logs changed its action key.\n") + else: + out.write("{} action(s) changed and would not have been a cache hit:\n\n".format(len(result.changed))) + + for change in result.changed: + out.write("{} {}\n".format(change.mnemonic, change.label)) + out.write(" output: {}\n".format(change.key)) + for reason in change.reasons: + out.write(" {}: {}\n".format(reason.kind, reason.summary)) + shown = reason.items if verbose else reason.items[:_MAX_ITEMS] + for item in shown: + out.write(" {}\n".format(item)) + hidden = len(reason.items) - len(shown) + if hidden > 0: + out.write(" ... and {} more (pass --verbose)\n".format(hidden)) + out.write("\n") + + if result.only_before or result.only_after: + out.write("{} action(s) ran only in the first log, {} only in the second.\n".format( + len(result.only_before), + len(result.only_after), + )) + if verbose: + for key in result.only_before: + out.write(" only in first: {}\n".format(key)) + for key in result.only_after: + out.write(" only in second: {}\n".format(key)) + +def _diff_impl(ctx: TaskContext) -> int | TaskConclusion: + logs = ctx.args.logs + if len(logs) != 2: + return TaskConclusion( + exit_code = 1, + message = "Pass exactly two compact execution logs, e.g. " + + "`aspect execlog diff before.binpb.zst after.binpb.zst`.", + ) + + _require_file(ctx, logs[0], "first execution log") + _require_file(ctx, logs[1], "second execution log") + + ctx.task.phase("read", description = "Reading both execution logs", emoji = "๐Ÿ“–") + result = execlog.diff(ctx, logs[0], logs[1], not ctx.args.summary_only) + + ctx.task.phase("report", description = "Explaining the differences", emoji = "๐Ÿ”") + _render(ctx, result, ctx.args.verbose) + + scanned = "{} and {} entries, {} and {} actions, {} in both".format( + result.before_entries, + result.after_entries, + result.before_spawns, + result.after_spawns, + result.matched, + ) + + if result.truncated: + return TaskConclusion(exit_code = 1, text = "Partial logs, " + scanned) + + if not result.changed: + return TaskConclusion(exit_code = 0, text = "No action changed, " + scanned) + + return TaskConclusion( + exit_code = 1 if ctx.args.fail_on_change else 0, + text = "{} action(s) changed, {}".format(len(result.changed), scanned), + flagged = not ctx.args.fail_on_change, + ) + +def _list_impl(ctx: TaskContext) -> int | TaskConclusion: + logs = ctx.args.logs + if len(logs) != 1: + return TaskConclusion( + exit_code = 1, + message = "Pass one compact execution log, e.g. `aspect execlog list build.binpb.zst`.", + ) + + _require_file(ctx, logs[0], "execution log") + + index = execlog.index(ctx, logs[0]) + out = ctx.std.io.stdout + + rows = [] + for key in sorted(index.spawns): + fp = index.spawns[key] + if ctx.args.mnemonic and fp.mnemonic != ctx.args.mnemonic: + continue + if ctx.args.cached == "only" and not fp.cache_hit: + continue + if ctx.args.cached == "exclude" and fp.cache_hit: + continue + rows.append(fp) + + if ctx.args.output == "json": + out.write(json.encode([{ + "label": fp.label, + "mnemonic": fp.mnemonic, + "runner": fp.runner, + "cache_hit": fp.cache_hit, + "output": fp.key, + "action_key": fp.overall, + } for fp in rows]) + "\n") + else: + for fp in rows: + out.write("{} {} {}{}\n".format( + fp.mnemonic, + fp.label, + fp.runner, + " (cached)" if fp.cache_hit else "", + )) + + if index.truncated != None: + print("WARNING: execution log ended early, the listing is partial. " + index.truncated) + return TaskConclusion(exit_code = 1, text = "Partial log, {} action(s)".format(len(rows))) + + return TaskConclusion( + exit_code = 0, + text = "{} action(s) of {} entries".format(len(rows), index.entries), + ) + +diff = task( + group = ["execlog"], + summary = "Compare two compact execution logs and explain every action that changed.", + description = ( + "Takes two `--execution_log_compact_file` artifacts from builds that should " + + "have been identical, and reports every action present in both whose action " + + "key moved, so it would have missed the cache the second time.\n\n" + + "For each one it names which part of the key changed โ€” arguments, environment, " + + "inputs, tool inputs, or the execution platform โ€” and for inputs it lists the " + + "files whose digests differ and the ones added or removed.\n\n" + + "Produce the logs with:\n" + + " bazel build --execution_log_compact_file=before.binpb.zst //...\n" + + " bazel build --execution_log_compact_file=after.binpb.zst //...\n\n" + + "or from an AXL task with `execution_log = [bazel.execution_log.compact_file(" + + "path = ...)]`.\n\n" + + "Exits 0 by default even when actions changed, because a changed action is a " + + "finding rather than an error. Pass --fail-on-change to make it fail a CI step " + + "that asserts two builds are identical." + ), + implementation = _diff_impl, + args = { + "verbose": args.boolean( + default = False, + short = "v", + description = "List every differing item rather than the first {}.".format(_MAX_ITEMS), + ), + "summary_only": args.boolean( + default = False, + description = "Report which dimension changed without the second pass that " + + "resolves the individual files. Faster on very large logs.", + ), + "fail_on_change": args.boolean( + default = False, + description = "Exit non-zero when any action changed.", + ), + "logs": args.positional( + minimum = 0, + maximum = 2, + default = [], + description = "The two compact execution logs to compare.", + ), + }, +) + +list_actions = task( + group = ["execlog"], + kind = "list", + summary = "List the actions recorded in a compact execution log.", + description = ( + "One line per action: mnemonic, label, runner, and whether it was a cache hit. " + + "Use --output=json for a machine-readable listing that includes each action's " + + "computed key, which is what `aspect execlog diff` compares." + ), + implementation = _list_impl, + args = { + "mnemonic": args.string( + default = "", + description = "Only list actions with this mnemonic, e.g. CppCompile.", + ), + "cached": args.string( + default = "all", + values = ["all", "only", "exclude"], + description = "Filter by cache hit.", + ), + "output": args.string( + default = "lines", + values = ["lines", "json"], + description = "Output format.", + ), + "logs": args.positional( + minimum = 0, + maximum = 1, + default = [], + description = "The compact execution log to list.", + ), + }, +) diff --git a/crates/aspect-cli/src/builtins/aspect/private/lib/execlog.axl b/crates/aspect-cli/src/builtins/aspect/private/lib/execlog.axl new file mode 100644 index 000000000..1b1be0116 --- /dev/null +++ b/crates/aspect-cli/src/builtins/aspect/private/lib/execlog.axl @@ -0,0 +1,416 @@ +"""Read compact execution logs and explain why an action would not be a cache hit. + +A compact execution log is a stream of entries that reference each other by id: +a `spawn` names an `input_set`, which names `file` ids and nested set ids, and a +file carries the digest that actually feeds Bazel's action key. Nothing in the +log states "this action changed"; you get there by rebuilding each spawn's +identity from those references and comparing it across two logs. + +The shape of the work is set by how big these get. A 20 MB compact log expands +to gigabytes once every input set is flattened, and a diff needs two of them, so +nothing here holds a flattened input list for a whole log. Instead: + + Pass 1 (`index`) Per log, walk once and keep three small maps keyed by entry + id, plus one `Fingerprint` per spawn. A fingerprint is five + hashes and a label, so its size tracks the number of actions + rather than the number of files. + Pass 2 (`detail`) Walk the same logs again, resolving full input lists only for + the spawns whose fingerprints differ. On a healthy build that + is a handful of actions out of tens of thousands. + +Two passes over a compressed file are far cheaper than holding one flattened +log in memory, and the second pass is skipped entirely when nothing differs. + +Ported from the Go `execlog-diff` tool. That one streams and diffs concurrently; +this is the same analysis with a bounded working set instead of a pipeline. +""" + +load("@std//hash.axl", "sha256") + +_INVOCATION = "invocation" +_FILE = "file" +_DIRECTORY = "directory" +_UNRESOLVED_SYMLINK = "unresolved_symlink" +_INPUT_SET = "input_set" +_RUNFILES_TREE = "runfiles_tree" +_SPAWN = "spawn" +_SYMLINK_ACTION = "symlink_action" + +Fingerprint = record( + key = field(str), + label = field(str), + mnemonic = field(str), + runner = field(str), + cache_hit = field(bool), + args_hash = field(str), + env_hash = field(str), + inputs_hash = field(str), + tools_hash = field(str), + platform_hash = field(str), + overall = field(str), +) + +Detail = record( + args = field(list), + env = field(dict), + inputs = field(dict), + tools = field(dict), +) + +Index = record( + spawns = field(dict), + entries = field(int), + truncated = field(str | None, default = None), +) + +def _hash(parts: list) -> str: + h = sha256() + for part in parts: + h.update(str(part)) + + # Length-prefixing would be tidier, but a NUL cannot appear in an + # argv entry or an env name, so it is already unambiguous here. + h.update("\x00") + return h.hexdigest() + +def _digest_of(entry) -> str: + d = getattr(entry, "digest", None) + return d.hash if d != None else "" + +def _flatten(sets: dict, paths: dict, digests: dict, root_id: int, budget: int) -> list: + """Resolve an input set id to a sorted list of `path@digest` strings. + + Iterative rather than recursive: input sets nest arbitrarily deep in a large + repo, and a blown evaluation stack part-way through a log is a worse failure + than a slow one. `budget` bounds the walk at the number of entries in the + log, which is more ids than any one set can reach. + """ + if root_id == 0: + return [] + + seen = {} + stack = [root_id] + out = [] + + for _ in range(budget): + if not stack: + break + current = stack.pop() + if current in seen: + continue + seen[current] = True + + if current in sets: + direct, transitive = sets[current] + for child in direct: + stack.append(child) + for child in transitive: + stack.append(child) + continue + + path = paths.get(current) + if path != None: + out.append(path + "@" + digests.get(current, "")) + + return sorted(out) + +def _index(ctx: TaskContext, log_path: str) -> Index: + """Walk a log once and fingerprint every spawn in it.""" + paths = {} + digests = {} + sets = {} + spawns = {} + count = 0 + + entries = bazel.execution_log.read(path = log_path) + for entry in entries: + count += 1 + payload = entry.type + kind = type(payload) + + if kind == _FILE: + paths[entry.id] = payload.path + digests[entry.id] = _digest_of(payload) + elif kind == _DIRECTORY or kind == _UNRESOLVED_SYMLINK: + paths[entry.id] = payload.path + elif kind == _INPUT_SET: + sets[entry.id] = (payload.input_ids, payload.transitive_set_ids) + elif kind == _RUNFILES_TREE: + # A spawn names the tree, not the set inside it, so the tree has to + # resolve as if it were a set or every runfile drops out of the key. + paths[entry.id] = payload.path + sets[entry.id] = ([], [payload.input_set_id]) + elif kind == _SPAWN: + fp = _fingerprint(payload, sets, paths, digests, count) + + # Two spawns can share a key when an action is retried. The last one + # is the one whose outputs the build kept. + spawns[fp.key] = fp + + return Index( + spawns = spawns, + entries = count, + truncated = entries.error(), + ) + +def _spawn_key(spawn, paths: dict) -> str: + """The identity of an action across two builds. + + Bazel gives no stable action id, so the primary output path stands in: it is + unique within a build and identical between two builds of the same target, + which is exactly the property a diff key needs. Outputs are ids into the + entry table, except when Bazel could not resolve one and recorded the raw + path instead. + """ + for output in spawn.outputs: + value = output.type + if type(value) == "string": + return value + resolved = paths.get(value) + if resolved != None: + return resolved + + # A spawn with no usable output is rare (a failed action that produced + # nothing). Falling back to the label keeps it in the diff instead of + # silently dropping it, at the cost of colliding with its siblings. + return spawn.target_label + "!" + spawn.mnemonic + +def _fingerprint(spawn, sets: dict, paths: dict, digests: dict, budget: int) -> Fingerprint: + args_hash = _hash(spawn.args) + env_hash = _hash(sorted([v.name + "=" + v.value for v in spawn.env_vars])) + inputs_hash = _hash(_flatten(sets, paths, digests, spawn.input_set_id, budget)) + tools_hash = _hash(_flatten(sets, paths, digests, spawn.tool_set_id, budget)) + platform_hash = _hash(_platform_properties(spawn)) + + return Fingerprint( + key = _spawn_key(spawn, paths), + label = spawn.target_label, + mnemonic = spawn.mnemonic, + runner = spawn.runner, + cache_hit = spawn.cache_hit, + args_hash = args_hash, + env_hash = env_hash, + inputs_hash = inputs_hash, + tools_hash = tools_hash, + platform_hash = platform_hash, + overall = _hash([args_hash, env_hash, inputs_hash, tools_hash, platform_hash]), + ) + +def _platform_properties(spawn) -> list: + platform = spawn.platform + if platform == None: + return [] + return sorted([p.name + "=" + p.value for p in platform.properties]) + +def _detail(ctx: TaskContext, log_path: str, wanted: dict) -> dict: + """Second pass: resolve full inputs for the spawns named in `wanted`.""" + paths = {} + digests = {} + sets = {} + found = {} + count = 0 + + entries = bazel.execution_log.read(path = log_path) + for entry in entries: + count += 1 + payload = entry.type + kind = type(payload) + + if kind == _FILE: + paths[entry.id] = payload.path + digests[entry.id] = _digest_of(payload) + elif kind == _DIRECTORY or kind == _UNRESOLVED_SYMLINK: + paths[entry.id] = payload.path + elif kind == _INPUT_SET: + sets[entry.id] = (payload.input_ids, payload.transitive_set_ids) + elif kind == _RUNFILES_TREE: + paths[entry.id] = payload.path + sets[entry.id] = ([], [payload.input_set_id]) + elif kind == _SPAWN: + key = _spawn_key(payload, paths) + if key not in wanted: + continue + found[key] = Detail( + args = list(payload.args), + env = {v.name: v.value for v in payload.env_vars}, + inputs = _as_map(_flatten(sets, paths, digests, payload.input_set_id, count)), + tools = _as_map(_flatten(sets, paths, digests, payload.tool_set_id, count)), + ) + + return found + +def _as_map(flattened: list) -> dict: + out = {} + for item in flattened: + path, _, digest = item.rpartition("@") + out[path] = digest + return out + +Change = record( + key = field(str), + label = field(str), + mnemonic = field(str), + reasons = field(list), +) + +Reason = record( + kind = field(str), + summary = field(str), + items = field(list, default = []), +) + +def _compare(before: Fingerprint, after: Fingerprint) -> list: + """Name every dimension of the action key that moved. + + All of them, not the first: an action whose inputs *and* environment changed + has two problems, and reporting only the first sends the reader round the + loop twice. + """ + reasons = [] + if before.args_hash != after.args_hash: + reasons.append("args") + if before.env_hash != after.env_hash: + reasons.append("env") + if before.inputs_hash != after.inputs_hash: + reasons.append("inputs") + if before.tools_hash != after.tools_hash: + reasons.append("tools") + if before.platform_hash != after.platform_hash: + reasons.append("platform") + return reasons + +def _diff_lists(before: list, after: list) -> Reason: + items = [] + limit = max(len(before), len(after)) + for i in range(limit): + b = before[i] if i < len(before) else "" + a = after[i] if i < len(after) else "" + if b != a: + items.append("[{}] {} -> {}".format(i, b, a)) + return Reason( + kind = "args", + summary = "{} argument(s) differ".format(len(items)), + items = items, + ) + +def _diff_maps(kind: str, before: dict, after: dict) -> Reason: + changed = [] + added = [] + removed = [] + + for key in sorted(after): + if key not in before: + added.append(key) + elif before[key] != after[key]: + changed.append(key) + for key in sorted(before): + if key not in after: + removed.append(key) + + items = [] + for key in changed: + items.append("changed " + key) + for key in added: + items.append("added " + key) + for key in removed: + items.append("removed " + key) + + return Reason( + kind = kind, + summary = "{} changed, {} added, {} removed".format(len(changed), len(added), len(removed)), + items = items, + ) + +def _explain(before: Detail, after: Detail, reasons: list) -> list: + out = [] + for reason in reasons: + if reason == "args": + out.append(_diff_lists(before.args, after.args)) + elif reason == "env": + out.append(_diff_maps("env", before.env, after.env)) + elif reason == "inputs": + out.append(_diff_maps("inputs", before.inputs, after.inputs)) + elif reason == "tools": + out.append(_diff_maps("tools", before.tools, after.tools)) + else: + out.append(Reason(kind = reason, summary = "differs")) + return out + +DiffResult = record( + changed = field(list), + only_before = field(list), + only_after = field(list), + matched = field(int), + before_entries = field(int), + after_entries = field(int), + before_spawns = field(int), + after_spawns = field(int), + truncated = field(list, default = []), +) + +def _diff(ctx: TaskContext, before_path: str, after_path: str, explain: bool) -> DiffResult: + before = _index(ctx, before_path) + after = _index(ctx, after_path) + + truncated = [] + if before.truncated != None: + truncated.append(before_path + ": " + before.truncated) + if after.truncated != None: + truncated.append(after_path + ": " + after.truncated) + + differing = {} + matched = 0 + for key in before.spawns: + if key not in after.spawns: + continue + matched += 1 + reasons = _compare(before.spawns[key], after.spawns[key]) + if reasons: + differing[key] = reasons + + before_detail = {} + after_detail = {} + if explain and differing: + before_detail = _detail(ctx, before_path, differing) + after_detail = _detail(ctx, after_path, differing) + + changed = [] + for key in sorted(differing): + fp = after.spawns[key] + reasons = differing[key] + explained = [] + if key in before_detail and key in after_detail: + explained = _explain(before_detail[key], after_detail[key], reasons) + else: + explained = [Reason(kind = r, summary = "differs") for r in reasons] + changed.append(Change( + key = key, + label = fp.label, + mnemonic = fp.mnemonic, + reasons = explained, + )) + + return DiffResult( + changed = changed, + only_before = sorted([k for k in before.spawns if k not in after.spawns]), + only_after = sorted([k for k in after.spawns if k not in before.spawns]), + matched = matched, + before_entries = before.entries, + after_entries = after.entries, + before_spawns = len(before.spawns), + after_spawns = len(after.spawns), + truncated = truncated, + ) + +execlog = struct( + index = _index, + diff = _diff, + Change = Change, + DiffResult = DiffResult, + Fingerprint = Fingerprint, + Reason = Reason, + testonly_flatten = _flatten, + testonly_diff_maps = _diff_maps, + testonly_diff_lists = _diff_lists, + testonly_hash = _hash, +) diff --git a/crates/axl-runtime/src/engine/bazel/iter/execlog.rs b/crates/axl-runtime/src/engine/bazel/iter/execlog.rs index a1359b728..de930e37f 100644 --- a/crates/axl-runtime/src/engine/bazel/iter/execlog.rs +++ b/crates/axl-runtime/src/engine/bazel/iter/execlog.rs @@ -1,4 +1,6 @@ use std::cell::RefCell; +use std::sync::Arc; +use std::sync::Mutex; use allocative::Allocative; use fibre::RecvError; @@ -23,17 +25,34 @@ use axl_proto::tools::protos::ExecLogEntry; use derive_more::Display; use fibre::spmc::Receiver; +/// Set by a decoder thread when it stops early. `None` means the stream ended +/// because it ran out of entries, which is the only clean way for it to end. +pub type DecodeFailure = Arc>>; + #[derive(ProvidesStaticType, Display, Trace, NoSerialize, Allocative, Debug)] #[display("")] pub struct ExecutionLogIterator { #[allocative(skip)] recv: RefCell>, + /// Only populated when the entries come from a file read that owns its own + /// decoder thread. The live build path reports its errors through the build. + #[allocative(skip)] + failure: Option, } impl ExecutionLogIterator { pub fn new(recv: Receiver) -> Self { Self { recv: RefCell::new(recv), + failure: None, + } + } + + /// An iterator whose producer reports decode failures through `failure`. + pub fn with_failure_slot(recv: Receiver, failure: DecodeFailure) -> Self { + Self { + recv: RefCell::new(recv), + failure: Some(failure), } } } @@ -67,6 +86,35 @@ pub(crate) fn execlog_methods(registry: &mut MethodsBuilder) { .into_anyhow_result()?; Ok(this.recv.borrow().is_closed()) } + + /// Why the stream stopped early, or `None` if it ran to the end of the log. + /// + /// Iteration ends silently when the decoder gives up, so a caller that must not + /// act on a partial log checks this once the loop is over. Always `None` for a + /// stream attached to a running build, where a decode failure fails the build. + /// + /// ```python + /// entries = bazel.execution_log.read(path = "a.binpb.zst") + /// for entry in entries: + /// ... + /// if entries.error() != None: + /// ctx.std.process.exit(1, "a.binpb.zst: " + entries.error()) + /// ``` + fn error<'v>(this: values::Value<'v>) -> anyhow::Result> { + let this = this + .downcast_ref_err::() + .into_anyhow_result()?; + let Some(failure) = this.failure.as_ref() else { + return Ok(NoneOr::None); + }; + let failure = failure + .lock() + .map_err(|e| anyhow::anyhow!("execution log decoder state was poisoned: {e}"))?; + Ok(match failure.as_ref() { + Some(msg) => NoneOr::Other(msg.clone()), + None => NoneOr::None, + }) + } } #[starlark_value(type = "ExecutionLogIterator")] diff --git a/crates/axl-runtime/src/engine/bazel/iter/mod.rs b/crates/axl-runtime/src/engine/bazel/iter/mod.rs index b52e5144d..41ba817fd 100644 --- a/crates/axl-runtime/src/engine/bazel/iter/mod.rs +++ b/crates/axl-runtime/src/engine/bazel/iter/mod.rs @@ -1,4 +1,4 @@ -mod execlog; +pub mod execlog; mod workspace_event; pub use execlog::ExecutionLogIterator; diff --git a/crates/axl-runtime/src/engine/bazel/mod.rs b/crates/axl-runtime/src/engine/bazel/mod.rs index b2c01b49c..58f173768 100644 --- a/crates/axl-runtime/src/engine/bazel/mod.rs +++ b/crates/axl-runtime/src/engine/bazel/mod.rs @@ -1351,6 +1351,40 @@ fn register_execlog_sinks(globals: &mut GlobalsBuilder) { ) -> anyhow::Result { Ok(sink::execlog::ExecLogSink::CompactFile { path }) } + + /// Read a compact execution log that is already on disk. + /// + /// `path` is a finished `--execution_log_compact_file` artifact, the same thing + /// `bazel.execution_log.compact_file(path = ...)` writes. Returns an iterator of + /// `ExecLogEntry`, decoded on a background thread, so a log far larger than memory + /// can be walked as long as the loop body does not keep every entry. + /// + /// A missing file, or one that is not zstd, fails here. A log that is truncated or + /// corrupt part-way through ends the iteration early and sets `error()`; check it + /// after the loop when a partial read would give a wrong answer. + /// + /// ```python + /// entries = bazel.execution_log.read(path = "before.binpb.zst") + /// spawns = 0 + /// for entry in entries: + /// if entry.type != None and hasattr(entry.type, "mnemonic"): + /// spawns += 1 + /// if entries.error() != None: + /// ctx.std.process.exit(1, "before.binpb.zst: " + entries.error()) + /// ``` + fn read( + #[starlark(require = named)] path: String, + ) -> anyhow::Result { + let failure: iter::execlog::DecodeFailure = Default::default(); + let stream = stream::ExecLogStream::spawn_from_path(path.clone().into(), failure.clone()) + .map_err(|e| anyhow::anyhow!("reading execution log '{path}': {e}"))?; + let recv = stream.receiver(); + // The decoder thread owns the file and ends itself at EOF or on error, so the + // stream handle has nothing left to do once the receiver is cloned out of it. + // Dropping it here detaches the thread; failures come back through `failure`. + drop(stream); + Ok(iter::ExecutionLogIterator::with_failure_slot(recv, failure)) + } } #[starlark_module] diff --git a/crates/axl-runtime/src/engine/bazel/stream/execlog.rs b/crates/axl-runtime/src/engine/bazel/stream/execlog.rs index b68a4129d..3d6b147a1 100644 --- a/crates/axl-runtime/src/engine/bazel/stream/execlog.rs +++ b/crates/axl-runtime/src/engine/bazel/stream/execlog.rs @@ -10,6 +10,7 @@ use std::path::PathBuf; use std::thread::JoinHandle; use std::{env, thread}; +use super::super::iter::execlog::DecodeFailure; use super::super::sink::retry::SinkOutcome; use super::util::{MultiTeeReader, read_varint}; use thiserror::Error; @@ -49,6 +50,9 @@ impl Read for RetryRead { } } +/// Leading bytes of every zstd frame, little-endian 0xFD2FB528. +const ZSTD_MAGIC: [u8; 4] = [0x28, 0xB5, 0x2F, 0xFD]; + #[derive(Debug)] pub struct ExecLogStream { handle: JoinHandle>, @@ -272,6 +276,107 @@ impl ExecLogStream { )) } + /// Decode a compact execution log that already exists on disk. + /// + /// Unlike [`spawn`](Self::spawn) and [`spawn_with_file`](Self::spawn_with_file), which + /// follow a file Bazel is still writing, this reads a finished + /// `--execution_log_compact_file` artifact: a plain zstd frame of + /// varint-length-prefixed `ExecLogEntry` messages. The end of the file is the end of + /// the stream, so EOF terminates rather than `BrokenPipe`. + /// + /// Sends block. A file this is pointed at was produced by a build that already + /// finished, so there is no build to slow down, and silently dropping entries the way + /// the live path does would make a diff wrong rather than slow. + /// + /// ## Truncation is usually, not always, detected + /// + /// A log cut off part-way through sets `failure` and ends the stream early, so a + /// caller that checks it does not compare a prefix and call it a clean result. The + /// detection comes from zstd, which reports `incomplete frame` when it runs out of + /// input mid-block. When the cut lands exactly on a block boundary, the decoder sees + /// a clean end of input and the entries before it are a structurally valid stream, so + /// the read looks successful. Measured against a real 134 KB log, cuts at 40% and + /// beyond were reported; cuts at 10% and 25% were not. + /// + /// Closing that last gap means checking zstd's end-of-frame marker, which the `zstd` + /// crate's `Decoder` does not expose; it needs `zstd_safe::DCtx` directly. Worth doing + /// if a partial log is ever mistaken for a real answer in practice. + pub fn spawn_from_path(path: PathBuf, failure: DecodeFailure) -> io::Result { + // Fail here, on the caller's thread, for the mistakes worth a traceback: a + // path that does not exist, or a file that is not a zstd frame at all. + // `Decoder::new` reads lazily, so it accepts a text file and only complains + // once the consumer starts iterating; checking the magic number is what makes + // "you pointed this at your BEP file" an error rather than an empty result. + // A log that is truncated or corrupt part-way through still surfaces through + // `failure`, because that cannot be known without reading it all. + let mut magic = [0u8; 4]; + File::open(&path)?.read_exact(&mut magic).map_err(|e| { + io::Error::new( + e.kind(), + format!("{path:?} is too short to be a zstd frame"), + ) + })?; + if magic != ZSTD_MAGIC { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "{path:?} is not a zstd frame, so it is not a compact execution log. \ + Pass the file written by --execution_log_compact_file." + ), + )); + } + + let (mut sender, recv) = bounded::(1000); + let report = failure.clone(); + let handle = thread::spawn(move || { + let mut buf: Vec = Vec::with_capacity(1024 * 5); + // 10 is the maximum size of a varint so start with that size. + buf.resize(10, 0); + + let mut out_raw = Decoder::new(File::open(&path)?)?; + + let mut read = || -> Result { + let (size, _) = match read_varint(&mut out_raw) { + Ok(it) => it, + // A clean end of file between entries: the log is fully read. + Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => return Ok(false), + Err(err) => return Err(err.into()), + }; + if size > buf.len() { + buf.resize(size, 0); + } + out_raw.read_exact(&mut buf[0..size])?; + sender.send(ExecLogEntry::decode(&buf[0..size])?)?; + Ok(true) + }; + + loop { + match read() { + Ok(true) => continue, + Ok(false) => { + sender.close()?; + return Ok(()); + } + Err(err) => { + // Record before closing: a consumer blocked in `recv` wakes on + // the close and may read `error()` immediately afterwards. + if let Ok(mut slot) = report.lock() { + *slot = Some(err.to_string()); + } + let _ = sender.close(); + return Err(err); + } + } + } + }); + + Ok(Self { + handle, + recv: Some(recv), + file_sink_handles: vec![], + }) + } + pub fn receiver(&self) -> Receiver { self.recv .as_ref() @@ -311,3 +416,165 @@ impl ExecLogStream { reader_result } } + +#[cfg(test)] +mod tests { + use super::*; + use axl_proto::tools::protos::exec_log_entry; + + /// Write `entries` as a compact execution log: one zstd frame over + /// varint-length-prefixed `ExecLogEntry` messages, which is the format + /// `--execution_log_compact_file` produces. + fn write_log(path: &std::path::Path, entries: &[ExecLogEntry]) { + let file = File::create(path).unwrap(); + let mut encoder = zstd::Encoder::new(file, 0).unwrap(); + for entry in entries { + encoder + .write_all(&entry.encode_length_delimited_to_vec()) + .unwrap(); + } + encoder.finish().unwrap().flush().unwrap(); + } + + fn file_entry(id: u32, path: &str) -> ExecLogEntry { + ExecLogEntry { + id, + r#type: Some(exec_log_entry::Type::File(exec_log_entry::File { + path: path.to_string(), + digest: None, + })), + } + } + + fn drain(path: PathBuf) -> (Vec, Option) { + let failure: DecodeFailure = Default::default(); + let stream = ExecLogStream::spawn_from_path(path, failure.clone()).unwrap(); + let recv = stream.receiver(); + // Drop the stream's own receiver clone before consuming, exactly as the + // Starlark `read()` does. In fibre's SPMC every clone is an independent + // subscriber the sender must not lap, so an unconsumed one deadlocks the + // producer on any log longer than the channel bound. + drop(stream); + let mut out = vec![]; + while let Ok(entry) = recv.recv() { + out.push(entry); + } + // The reader thread records the failure before closing the channel, so + // by the time recv() reports disconnection the slot is already set. + let err = failure.lock().unwrap().clone(); + (out, err) + } + + #[test] + fn reads_every_entry_of_a_complete_log() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("log.binpb.zst"); + let entries: Vec<_> = (1..=250) + .map(|i| file_entry(i, &format!("f{i}.txt"))) + .collect(); + write_log(&path, &entries); + + let (got, err) = drain(path); + assert_eq!(err, None, "a complete log should not report a failure"); + assert_eq!(got.len(), 250); + assert_eq!(got[0].id, 1); + assert_eq!(got[249].id, 250); + } + + #[test] + fn rejects_a_file_that_is_not_zstd() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("not-a-log.zst"); + std::fs::write(&path, b"this is plainly not a zstd frame").unwrap(); + + let failure: DecodeFailure = Default::default(); + let err = ExecLogStream::spawn_from_path(path, failure).unwrap_err(); + assert!( + err.to_string().contains("is not a zstd frame"), + "expected the magic-number check to name the problem, got: {err}" + ); + } + + #[test] + fn rejects_a_file_too_short_to_have_a_header() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("tiny.zst"); + std::fs::write(&path, b"ab").unwrap(); + + let failure: DecodeFailure = Default::default(); + let err = ExecLogStream::spawn_from_path(path, failure).unwrap_err(); + assert!( + err.to_string().contains("too short"), + "expected a length complaint, got: {err}" + ); + } + + #[test] + fn a_missing_file_fails_on_the_calling_thread() { + let failure: DecodeFailure = Default::default(); + let err = + ExecLogStream::spawn_from_path(PathBuf::from("/definitely/not/here.zst"), failure) + .unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::NotFound); + } + + #[test] + fn a_truncated_log_yields_a_prefix_rather_than_every_entry() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("log.binpb.zst"); + let entries: Vec<_> = (1..=200) + .map(|i| file_entry(i, &format!("f{i}.txt"))) + .collect(); + write_log(&path, &entries); + + // Lop off the tail, keeping the header so the magic check still passes. + let bytes = std::fs::read(&path).unwrap(); + let truncated = dir.path().join("truncated.binpb.zst"); + std::fs::write(&truncated, &bytes[..bytes.len() / 2]).unwrap(); + + let (got, _) = drain(truncated); + assert!( + got.len() < 200, + "expected fewer than every entry back from a truncated log, got {}", + got.len() + ); + } + + #[test] + fn a_corrupt_frame_reports_why_it_stopped() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("log.binpb.zst"); + let entries: Vec<_> = (1..=2000) + .map(|i| { + file_entry( + i, + &format!("some/deep/path/to/a/source/file/number/{i}.txt"), + ) + }) + .collect(); + write_log(&path, &entries); + + // Flip bytes in the middle of the compressed payload. Unlike truncation, + // which can land on a block boundary and look like a clean end, this is + // always a frame zstd refuses, so it pins the reporting path itself. + let mut bytes = std::fs::read(&path).unwrap(); + let middle = bytes.len() / 2; + for b in &mut bytes[middle..middle + 32] { + *b ^= 0xFF; + } + let corrupt = dir.path().join("corrupt.binpb.zst"); + std::fs::write(&corrupt, &bytes).unwrap(); + + let (got, err) = drain(corrupt); + assert!( + err.is_some(), + "a corrupt frame must set the failure slot, or a diff silently compares \ + a prefix; got {} entries and no error", + got.len() + ); + assert!( + got.len() < 2000, + "expected the stream to stop short, got every entry back" + ); + } +}