Add a tool that checks the translated shaders of a shader cache offline, with the game's own load checks and spirv-val.

tools/validate_shader_cache.py reads every module and its reflection from the pack and the loose files, says how many the game would refuse and translate again, runs spirv-val on the others and groups what it refuses by message. It writes nothing to the cache. tests/validate_shader_cache_test.py checks it on a made-up cache.

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>

docs/building.md

@@ -174,6 +174,12 @@ of them.
    `make_release.py`). A list belongs to the record layout of the runtime that wrote it; another runtime says so and
    builds nothing from it. `python tests\pipeline_list_test.py` checks the tool on made-up folders.
 
+   What a game folder's shader cache holds can be checked without the game:
+   `python tools\validate_shader_cache.py <game folder>\shader-cache` reads every translated shader from the pack and
+   the loose files, applies the checks the game makes when it loads one (those that fail are translated again at
+   their next use) and runs `spirv-val` on the rest; `--every N` samples, `--out FILE` lists the failed ones. It writes
+   nothing to the cache. `python tests\validate_shader_cache_test.py` checks the tool on a made-up cache.
+
 7. Run. The game keeps running when its window loses the focus (`assume_focus = 1` in `nx2navi.ini`, the default; the
    Game tab switches it). With 0 it pauses, as it does on the Switch when Home is pressed.
 

tests/validate_shader_cache_test.py

@@ -0,0 +1,157 @@
+"""tools/validate_shader_cache.py on a fake shader cache: the entries the game would refuse (a cut or zero-filled
+module, another magic number, a reflection that is not one JSON object naming its stage, a missing half of a pair) are
+named, packed and loose entries are both read with the loose one standing for its packed copy, --every samples, and,
+where spirv-val is installed, a whole module that is not valid SPIR-V is refused by it. No game files.
+
+Run: python tests/validate_shader_cache_test.py
+"""
+import hashlib
+import struct
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / 'tools'))
+import pack_shader_cache as psc  # noqa: E402
+import validate_shader_cache as vsc  # noqa: E402
+
+RES = b'{"stage":"cs","textures":[]}'
+
+
+def name(seed):
+    return hashlib.sha256(seed.encode()).hexdigest()
+
+
+def insn(op, *operands):
+    return struct.pack('<%dI' % (1 + len(operands)), (1 + len(operands)) << 16 | op, *operands)
+
+
+def text(s):
+    b = s.encode() + b'\0'
+    b += b'\0' * (-len(b) % 4)
+    return struct.unpack('<%dI' % (len(b) // 4), b)
+
+
+def module(memory_model=True):
+    """A compute shader that does nothing, SPIR-V 1.3: ids 1 main, 2 void, 3 the function type, 4 the label."""
+    body = insn(17, 1)                                   # OpCapability Shader
+    if memory_model:
+        body += insn(14, 0, 1)                           # OpMemoryModel Logical GLSL450
+    body += insn(15, 5, 1, *text('main'))                # OpEntryPoint GLCompute %1 "main"
+    body += insn(16, 1, 17, 1, 1, 1)                     # OpExecutionMode %1 LocalSize 1 1 1
+    body += insn(19, 2) + insn(33, 3, 2)                 # OpTypeVoid, OpTypeFunction
+    body += insn(54, 2, 1, 0, 3) + insn(248, 4) + insn(253) + insn(56)   # OpFunction, OpLabel, OpReturn, OpFunctionEnd
+    return struct.pack('<5I', vsc.SPIRV_MAGIC, 0x00010300, 0, 5, 0) + body
+
+
+def cut_module():
+    """A module whose last instruction says three words and has one: what a write cut short leaves."""
+    return module() + struct.pack('<I', 3 << 16 | 0)
+
+
+def put(cache, seed, spv=None, res=None):
+    if spv is not None:
+        (cache / (name(seed) + '.spv')).write_bytes(spv)
+    if res is not None:
+        (cache / (name(seed) + '.res.json')).write_bytes(res)
+    return name(seed)
+
+
+class ValidateTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.cache = Path(self.tmp.name) / 'shader-cache'
+        self.cache.mkdir()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def check(self, **kw):
+        checked, failures = vsc.validate(self.cache, say=lambda s: None, **kw)
+        return checked, {k: (o, p) for k, o, p in failures}
+
+    def test_game_checks(self):
+        good = module()
+        ok = put(self.cache, 'ok', good, RES)
+        cut = put(self.cache, 'cut', good[:-6], RES)                       # not a whole number of words
+        short = put(self.cache, 'short', cut_module(), RES)                 # the last instruction runs past the end
+        zero = put(self.cache, 'zero', good[:40] + b'\0' * (len(good) - 40), RES)
+        magic = put(self.cache, 'magic', b'\x01' + good[1:], RES)
+        nul = put(self.cache, 'nul', good, b'\0' * len(RES))
+        stage = put(self.cache, 'stage', good, b'{"stage":"xx"}')
+        array = put(self.cache, 'array', good, b'[1]')
+        no_res = put(self.cache, 'no-res', good, None)
+        no_spv = put(self.cache, 'no-spv', None, RES)
+        checked, bad = self.check()
+        self.assertEqual(checked, 10)
+        self.assertNotIn(ok, bad)
+        self.assertIn('not a whole SPIR-V module', bad[cut][1])
+        self.assertIn('past the end', bad[short][1])
+        self.assertIn('zero words', bad[zero][1])
+        self.assertIn('magic', bad[magic][1])
+        self.assertIn('NUL', bad[nul][1])
+        self.assertIn('no stage', bad[stage][1])
+        self.assertIn('not one JSON object', bad[array][1])
+        self.assertEqual(bad[no_res][1], 'no .res.json')
+        self.assertEqual(bad[no_spv][1], 'no .spv')
+        self.assertEqual(len(bad), 9)
+        self.assertTrue(all(o == 'loose' for o, _ in bad.values()))
+
+    def test_stages_and_spaces(self):
+        for st in ('vs', 'ps', 'cs', 'ts', 'ms'):
+            self.assertEqual(vsc.reflection_problem(b' \n{ "stage" : "%s" }\n' % st.encode()), '')
+        self.assertNotEqual(vsc.reflection_problem(b'{"stage":"gs"}'), '')
+        self.assertNotEqual(vsc.reflection_problem(b'{}'), '')
+        self.assertEqual(vsc.spirv_problem(module()), '')
+
+    def test_pack_and_loose(self):
+        good = module()
+        packed_ok = put(self.cache, 'packed-ok', good, RES)
+        packed_bad = put(self.cache, 'packed-bad', cut_module(), RES)
+        fixed = put(self.cache, 'fixed', cut_module(), RES)
+        psc.pack(self.cache)
+        self.assertFalse(list(self.cache.glob('*.spv')))
+        put(self.cache, 'fixed', good, None)             # a newer translation beside the pack's bad copy
+        loose_bad = put(self.cache, 'loose-bad', good, b'{}')
+        checked, bad = self.check()
+        self.assertEqual(checked, 4)
+        self.assertEqual(set(bad), {packed_bad, loose_bad})
+        self.assertEqual(bad[packed_bad][0], 'pack')
+        self.assertEqual(bad[loose_bad][0], 'loose')
+        self.assertNotIn(fixed, bad)
+        self.assertNotIn(packed_ok, bad)
+
+    def test_every_and_out(self):
+        for i in range(10):
+            put(self.cache, 'n%d' % i, cut_module() if i % 2 else module(), RES)
+        checked, bad = self.check()
+        self.assertEqual((checked, len(bad)), (10, 5))
+        checked, _ = self.check(every=3)
+        self.assertEqual(checked, 4)
+        out = Path(self.tmp.name) / 'failed.txt'
+        self.check(out=out)
+        lines = out.read_text(encoding='utf-8').splitlines()
+        self.assertEqual(len(lines), 5)
+        self.assertTrue(all(ln.split()[0] in bad and ln.split()[1] == 'loose' for ln in lines))
+
+    def test_groups(self):
+        a = vsc.group_of("spirv-val: error: line 12: ID '34[%34]' has not been defined")
+        b = vsc.group_of("spirv-val: error: line 907: ID '5[%main]' has not been defined")
+        self.assertEqual(a, b)
+
+    @unittest.skipUnless(vsc.find_spirv_val(), 'spirv-val is not installed')
+    def test_spirv_val(self):
+        ok = put(self.cache, 'ok', module(), RES)
+        invalid = put(self.cache, 'invalid', module(memory_model=False), RES)   # whole, and not valid SPIR-V
+        checked, bad = self.check(val=vsc.find_spirv_val())
+        self.assertEqual(checked, 2)
+        self.assertNotIn(ok, bad)
+        self.assertTrue(bad[invalid][1].startswith('spirv-val: '), bad[invalid][1])
+        _, without = self.check()                        # the game's checks alone take both
+        self.assertEqual(without, {})
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/validate_shader_cache.py

@@ -0,0 +1,200 @@
+"""Check the translated shaders an install's shader cache holds, offline: every <key>.spv with its <key>.res.json, from
+<cache>/shader-cache.pack and from the loose files (a loose file stands for its packed copy, as the game reads them).
+
+The game checks little of a cached shader when it loads one (runtime/gfx_shader.c disk_load_from: the reflection is one
+JSON object that names its stage, the SPIR-V has its magic and whole instructions) and hands the module to the driver
+as it is. A module the driver then compiles wrongly, or crashes on, shows as a missing draw or a crash in play, far
+from its cause. This tool puts the question to the whole cache at once:
+
+  1. the game's own checks, so the count says how many entries the game would translate again;
+  2. spirv-val on each module that passes them (--target-env vulkan1.3, as tools/shader_census.py validates fresh
+     translations), unless --no-val is given or spirv-val is not found.
+
+The modules are the translator's output; the pipeline code rewrites a copy of each before the driver sees it
+(runtime/vk_pipeline.c make_module), and that copy is not what is checked here. tools/shader_census.py --translate
+validates translations made now from the game files; this validates what a folder already holds, which is also what
+earlier translators wrote and what an interrupted write left.
+
+    python tools\\validate_shader_cache.py <install folder>\\shader-cache [--every N] [--jobs N] [--no-val] [--out FILE]
+
+--every N checks every N-th shader (by key order), for a quick look at a large cache; --out writes one line per failed
+shader (key, origin, what failed). Exit code 0 when every checked entry passed, 1 otherwise, 2 for a folder that is not
+a shader cache. The game may be running: nothing is written to the cache.
+"""
+import argparse
+import os
+import re
+import shutil
+import struct
+import subprocess
+import sys
+import time
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import pack_shader_cache as psc  # noqa: E402
+
+SPV, RES = psc.KINDS['.spv'], psc.KINDS['.res.json']
+SPIRV_MAGIC = 0x07230203
+STAGES = (b'"vs"', b'"ps"', b'"cs"', b'"ts"', b'"ms"')   # gfx_shader.c reflection_valid
+TARGET_ENV = 'vulkan1.3'
+
+
+def find_spirv_val():
+    """The path of spirv-val, or None: MSYS2 UCRT64's on Windows, else the one on PATH."""
+    msys = Path(r'C:\msys64\ucrt64\bin\spirv-val.exe')
+    if os.name == 'nt' and msys.is_file():
+        return str(msys)
+    return shutil.which('spirv-val')
+
+
+def spirv_problem(data):
+    """What gfx_shader.c disk_load_from refuses in a module ('' when it takes it): size, magic, and every instruction
+    after the five header words having a word count that ends inside the module (spirv_complete)."""
+    if data is None:
+        return 'no .spv'
+    if len(data) < 20 or len(data) % 4:
+        return 'the .spv is %d bytes: not a whole SPIR-V module' % len(data)
+    words = len(data) // 4
+    if struct.unpack_from('<I', data, 0)[0] != SPIRV_MAGIC:
+        return 'the .spv has another magic number'
+    i = 5
+    while i < words:
+        n = struct.unpack_from('<I', data, 4 * i)[0] >> 16
+        if not n:
+            return 'an instruction of zero words at word %d (a zero-filled stretch)' % i
+        i += n
+    return '' if i == words else 'the last instruction runs past the end of the module'
+
+
+def reflection_problem(data):
+    """What gfx_shader.c reflection_valid refuses in a .res.json ('' when it takes it)."""
+    if data is None:
+        return 'no .res.json'
+    if b'\0' in data:
+        return 'the .res.json holds NUL bytes (an interrupted write)'
+    body = data.strip()
+    if len(body) < 2 or body[:1] != b'{' or body[-1:] != b'}':
+        return 'the .res.json is not one JSON object'
+    at = body.find(b'"stage"')
+    quote = body.find(b'"', at + 7) if at >= 0 else -1
+    if quote < 0 or not body.startswith(STAGES, quote):
+        return 'the .res.json names no stage the game knows'
+    return ''
+
+
+class Cache:
+    """The .spv and .res.json entries of a shader cache folder: loose files, then the pack."""
+
+    def __init__(self, folder):
+        self.folder = Path(folder)
+        self.loose = {k: p for k, p in psc.loose_files(self.folder).items() if k[0] in (SPV, RES)}
+        self.pack_path = self.folder / psc.PACK
+        self.packed = {k: v for k, v in psc.read_pack(self.pack_path).items() if k[0] in (SPV, RES)}
+
+    def keys(self):
+        """Every key with a .spv or a .res.json, sorted."""
+        return sorted({name for _, name in self.loose} | {name for _, name in self.packed})
+
+    def read(self, kind, name, pack_file):
+        """(data or None, 'loose' | 'pack' | '') of one entry."""
+        path = self.loose.get((kind, name))
+        if path is not None:
+            return psc.read_loose(path), 'loose'
+        at = self.packed.get((kind, name))
+        if at is None or pack_file is None:
+            return None, ''
+        return os.pread(pack_file, at[1], at[0]) if hasattr(os, 'pread') else pread(pack_file, at[1], at[0]), 'pack'
+
+
+def pread(fd, size, offset):
+    """os.pread where the platform has none (Windows): callers hold one descriptor per thread."""
+    os.lseek(fd, offset, os.SEEK_SET)
+    return os.read(fd, size)
+
+
+def check_one(cache, name, val, pack_file):
+    """(key hex, origin, problem): problem '' when the entry passed every check asked for."""
+    spv, origin = cache.read(SPV, name, pack_file)
+    res, res_origin = cache.read(RES, name, pack_file)
+    origin = origin or res_origin
+    problem = reflection_problem(res) or spirv_problem(spv)
+    if not problem and val:
+        r = subprocess.run([val, '--target-env', TARGET_ENV, '-'], input=spv, capture_output=True)
+        if r.returncode:
+            text = (r.stderr or r.stdout).decode('utf-8', 'replace').strip().splitlines()
+            problem = 'spirv-val: ' + (text[0] if text else 'exit code %d' % r.returncode)
+    return name.hex(), origin, problem
+
+
+def group_of(problem):
+    """A problem with its numbers and ids taken out, so the same fault in many shaders counts as one group."""
+    return re.sub(r"'\d+\[%[^\]]*\]'|\b\d+\b", 'N', re.sub(r'^spirv-val: error: line \d+: ', 'spirv-val: ', problem))
+
+
+def validate(folder, every=1, jobs=0, val=None, out=None, say=print):
+    """Checks the cache at folder; returns (checked, failures) with failures a list of (key, origin, problem)."""
+    cache = Cache(folder)
+    names = cache.keys()[::max(1, every)]
+    jobs = jobs or min(32, (os.cpu_count() or 4))
+    failures, done, t0, last = [], 0, time.monotonic(), time.monotonic()
+
+    def work(chunk):
+        # one descriptor of the pack per task: positioned reads without a shared file position
+        fd = os.open(cache.pack_path, os.O_RDONLY | getattr(os, 'O_BINARY', 0)) if cache.packed else None
+        try:
+            return [check_one(cache, n, val, fd) for n in chunk]
+        finally:
+            if fd is not None:
+                os.close(fd)
+
+    chunks = [names[i:i + 64] for i in range(0, len(names), 64)]
+    with ThreadPoolExecutor(max_workers=jobs) as ex:
+        for results in ex.map(work, chunks):
+            done += len(results)
+            failures += [r for r in results if r[2]]
+            if time.monotonic() - last >= 10:
+                last = time.monotonic()
+                say(f'  {done} of {len(names)} shaders checked, {len(failures)} failed, {last - t0:.0f} s')
+    if out:
+        Path(out).write_text(''.join(f'{k} {o or "-"} {p}\n' for k, o, p in failures), encoding='utf-8')
+    return len(names), failures
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.split('\n\n')[0])
+    ap.add_argument('cache', help='the shader cache folder (<install folder>/shader-cache)')
+    ap.add_argument('--every', type=int, default=1, help='check every N-th shader only')
+    ap.add_argument('--jobs', type=int, default=0, help='threads (default: the CPUs, at most 32)')
+    ap.add_argument('--no-val', action='store_true', help="the game's own checks only, without spirv-val")
+    ap.add_argument('--out', help='write one line per failed shader to this file')
+    args = ap.parse_args()
+    folder = Path(args.cache)
+    if not folder.is_dir():
+        print(f'{folder}: not a folder', file=sys.stderr)
+        return 2
+    val = None if args.no_val else find_spirv_val()
+    if not args.no_val and not val:
+        print("spirv-val not found (MSYS2: pacman -S mingw-w64-ucrt-x86_64-spirv-tools): the game's own checks only")
+    t0 = time.monotonic()
+    checked, failures = validate(folder, args.every, args.jobs, val, args.out)
+    if not checked:
+        print(f'{folder}: no translated shaders (no {psc.PACK} and no loose .spv files)', file=sys.stderr)
+        return 2
+    refused = sum(1 for f in failures if not f[2].startswith('spirv-val: '))
+    print(f'{checked} shaders checked in {time.monotonic() - t0:.0f} s: {checked - len(failures)} passed, {refused} '
+          f"the game would translate again (its own checks), {len(failures) - refused} refused by spirv-val"
+          + ('' if val else ' (not run)'))
+    groups = Counter(group_of(f[2]) for f in failures)
+    for text, n in groups.most_common(20):
+        first = next(f for f in failures if group_of(f[2]) == text)
+        print(f'  {n:6d}  {text}   (first: {first[0][:16]}, {first[1] or "no file"})')
+    if len(groups) > 20:
+        print(f'  ... and {len(groups) - 20} more kinds; --out FILE lists every failed shader')
+    return 1 if failures else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())