Standalone WebKitGTK evaluate_javascript leak test (multi-variant)

webkit_eval_leak_test_v2.py

"""
WebKitGTK evaluate_javascript leak test -- v2.

Fixes from v1:
- Properly calls evaluate_javascript_finish() via a GAsyncReadyCallback
  (v1 passed None, which left GTask results unconsumed)
- Supports --static-js flag to test with identical JS each eval
  (v1 embedded a unique ID in each eval, polluting JSC bytecode cache)

Tests four combinations to isolate the cause:
  1. Unique JS + no callback  (original v1 behavior -- known flawed)
  2. Unique JS + callback     (tests bytecode cache pollution alone)
  3. Static JS + no callback  (tests GTask accumulation alone)
  4. Static JS + callback     (both fixes -- should show no leak)

Usage:
    python3 webkit_eval_leak_test_v2.py [--rate 94] [--duration 120] [--payload-size 16]
           [--static-js] [--no-callback] [--all-variants]
"""

import argparse
import json
import math
import os
import signal
import sys
import time

import gi
gi.require_version("Gtk", "3.0")
gi.require_version("WebKit2", "4.1")
from gi.repository import GLib, Gtk, WebKit2, Gio


def find_webkit_child_pid(parent_pid):
    """Find the WebKitWebProcess child of our process."""
    try:
        for entry in os.listdir("/proc"):
            if not entry.isdigit():
                continue
            try:
                with open(f"/proc/{entry}/status") as f:
                    status = f.read()
                if f"PPid:\t{parent_pid}" in status and "WebKitWebProces" in status:
                    return int(entry)
            except (FileNotFoundError, PermissionError):
                continue
    except Exception:
        pass
    return None


def get_rss_kb(pid):
    """Read VmRSS from /proc/<pid>/status."""
    try:
        with open(f"/proc/{pid}/status") as f:
            for line in f:
                if line.startswith("VmRSS:"):
                    return int(line.split()[1])
    except (FileNotFoundError, PermissionError):
        pass
    return None


def get_private_dirty_kb(pid):
    """Read Private_Dirty from /proc/<pid>/smaps_rollup."""
    try:
        with open(f"/proc/{pid}/smaps_rollup") as f:
            for line in f:
                if line.startswith("Private_Dirty:"):
                    return int(line.split()[1])
    except (FileNotFoundError, PermissionError):
        pass
    return None


def on_js_finished(webview, result):
    """GAsyncReadyCallback that calls evaluate_javascript_finish to consume the GTask result."""
    try:
        webview.run_javascript_finish(result)
    except Exception:
        pass


def run_single_test(rate, duration, payload_size, static_js, use_callback):
    """Run a single test variant and return measurements."""
    interval_ms = max(1, int(1000 / rate))
    total_evals = rate * duration

    sample_payload = [round(math.sin(i * 0.4) * 0.5 + 0.5, 6) for i in range(payload_size)]
    payload_json = json.dumps(sample_payload)

    if static_js:
        # Same JS every time -- no bytecode cache pollution
        js_code = f"""
        (function() {{
            var event = {{
                event: 'mic-level',
                id: 0,
                payload: {payload_json}
            }};
            var processed = event.payload.map(function(v, i) {{ return v * 0.7 + 0.3; }});
        }})();
        """
    else:
        # Template with unique ID each time (Tauri's behavior)
        js_template = f"""
        (function() {{
            var event = {{
                event: 'mic-level',
                id: {{ID}},
                payload: {payload_json}
            }};
            var processed = event.payload.map(function(v, i) {{ return v * 0.7 + 0.3; }});
        }})();
        """

    variant_name = f"{'static' if static_js else 'unique'} JS, {'with' if use_callback else 'no'} callback"
    print(f"\n{'='*60}")
    print(f"  Variant: {variant_name}")
    print(f"  Rate: {rate}/sec, Duration: {duration}s, Total: {total_evals}")
    print(f"{'='*60}")

    window = Gtk.Window(title=f"Leak Test - {variant_name}")
    window.set_default_size(200, 100)
    window.connect("destroy", Gtk.main_quit)

    webview = WebKit2.WebView()
    window.add(webview)
    webview.load_html("<html><body><p>Testing...</p></body></html>", None)

    parent_pid = os.getpid()
    eval_count = [0]
    webkit_pid = [None]
    baseline_rss = [None]
    baseline_dirty = [None]
    start_time = [None]
    measurements = []

    def take_measurement(label):
        pid = webkit_pid[0]
        if pid is None:
            return
        rss = get_rss_kb(pid)
        dirty = get_private_dirty_kb(pid)
        if rss is not None:
            elapsed = time.time() - start_time[0]
            measurements.append((elapsed, eval_count[0], rss, dirty or 0))
            rss_mb = rss / 1024
            dirty_mb = (dirty or 0) / 1024
            delta_rss = (rss - baseline_rss[0]) / 1024 if baseline_rss[0] else 0
            delta_dirty = ((dirty - baseline_dirty[0]) / 1024) if baseline_dirty[0] and dirty else 0
            print(f"  [{label}] t={elapsed:.0f}s evals={eval_count[0]:,} "
                  f"RSS={rss_mb:.1f}MB(+{delta_rss:.1f}) "
                  f"PrivDirty={dirty_mb:.1f}MB(+{delta_dirty:.1f})")

    def do_eval():
        if static_js:
            js = js_code
        else:
            js = js_template.replace("{ID}", str(eval_count[0]))

        if use_callback:
            webview.run_javascript(js, None, on_js_finished)
        else:
            webview.run_javascript(js, None, None)

        eval_count[0] += 1

        if webkit_pid[0] is None:
            webkit_pid[0] = find_webkit_child_pid(parent_pid)
            if webkit_pid[0]:
                baseline_rss[0] = get_rss_kb(webkit_pid[0])
                baseline_dirty[0] = get_private_dirty_kb(webkit_pid[0])
                start_time[0] = time.time()
                print(f"  WebKitWebProcess PID: {webkit_pid[0]}")
                print(f"  Baseline RSS: {(baseline_rss[0] or 0) / 1024:.1f} MB")

        if eval_count[0] % (rate * 10) == 0:
            take_measurement("running")

        if eval_count[0] >= total_evals:
            take_measurement("final")
            print(f"\n  Waiting 10 seconds idle...")
            GLib.timeout_add(10000, post_idle_check)
            return False

        return True

    def post_idle_check():
        take_measurement("10s idle")

        if measurements and baseline_rss[0]:
            first_rss = baseline_rss[0]
            last_rss = measurements[-1][2]
            total_delta_mb = (last_rss - first_rss) / 1024

            print(f"\n  --- Result: {variant_name} ---")
            print(f"  Total evaluations: {eval_count[0]:,}")
            print(f"  RSS growth after idle: {total_delta_mb:.1f} MB")
            if total_delta_mb > 5:
                print(f"  LEAK DETECTED")
            else:
                print(f"  No significant leak")

        Gtk.main_quit()
        return False

    def on_load_changed(wv, event):
        if event == WebKit2.LoadEvent.FINISHED:
            GLib.timeout_add(1000, start_test)

    def start_test():
        GLib.timeout_add(interval_ms, do_eval)
        return False

    webview.connect("load-changed", on_load_changed)
    signal.signal(signal.SIGINT, lambda *_: Gtk.main_quit())

    window.show_all()
    Gtk.main()

    return measurements


def main():
    parser = argparse.ArgumentParser(description="WebKitGTK evaluate_javascript leak test v2")
    parser.add_argument("--rate", type=int, default=94)
    parser.add_argument("--duration", type=int, default=120)
    parser.add_argument("--payload-size", type=int, default=16)
    parser.add_argument("--static-js", action="store_true", help="Use identical JS each eval (no unique IDs)")
    parser.add_argument("--no-callback", action="store_true", help="Pass None for callback (don't call _finish)")
    parser.add_argument("--all-variants", action="store_true", help="Run all 4 combinations sequentially")
    args = parser.parse_args()

    print("WebKitGTK evaluate_javascript leak test v2")
    print(f"  Rate: {args.rate}/sec, Duration: {args.duration}s, Payload: {args.payload_size} floats")

    if args.all_variants:
        print("\nRunning all 4 variants sequentially...")
        for static_js, use_callback in [
            (False, False),  # original v1 behavior
            (False, True),   # unique JS + callback
            (True, False),   # static JS + no callback
            (True, True),    # both fixes
        ]:
            run_single_test(
                rate=args.rate,
                duration=args.duration,
                payload_size=args.payload_size,
                static_js=static_js,
                use_callback=use_callback,
            )
    else:
        run_single_test(
            rate=args.rate,
            duration=args.duration,
            payload_size=args.payload_size,
            static_js=args.static_js,
            use_callback=not args.no_callback,
        )


if __name__ == "__main__":
    main()