Audio hangs after Windows App disconnect: Weston semaphore wait and PulseAudio blocked socket read #1508

#!/usr/bin/env python3
"""A bounded, non-recovering PulseAudio playback probe for WSLg disconnects."""
import argparse
import array
import datetime as dt
import json
import math
import os
from pathlib import Path
import select
import shutil
import signal
import subprocess
import sys
import threading
import time


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--duration', type=float, default=0,
                        help='Seconds to run; 0 runs until Ctrl-C (default).')
    parser.add_argument('--log-dir', type=Path, help='New directory for this run.')
    parser.add_argument('--device', help='Optional PulseAudio sink name.')
    parser.add_argument('--mark', type=Path, help='Add a timestamped marker to an existing run directory, then exit.')
    parser.add_argument('--message', default='User marker')
    args = parser.parse_args()
    if args.mark:
        with (args.mark / 'markers.jsonl').open('a') as f:
            f.write(json.dumps({'utc': dt.datetime.now(dt.timezone.utc).isoformat(),
                                'message': args.message}) + '\n')
        return
    if args.duration < 0:
        parser.error('--duration must be nonnegative')
    for tool in ('pacat', 'pactl'):
        if not shutil.which(tool):
            parser.error(f'{tool} is required (Ubuntu package: pulseaudio-utils)')
    stamp = dt.datetime.now(dt.timezone.utc).strftime('%Y%m%dT%H%M%S.%fZ')
    folder = args.log_dir or Path.home() / 'wsl-audio-logs' / stamp
    folder.mkdir(parents=True, exist_ok=False)
    events = (folder / 'events.jsonl').open('a', buffering=1)
    lock = threading.Lock()
    started = time.monotonic()
    stop = threading.Event()

    def log(event, **fields):
        row = {'utc': dt.datetime.now(dt.timezone.utc).isoformat(),
               'elapsed_s': round(time.monotonic() - started, 3), 'event': event, **fields}
        with lock:
            events.write(json.dumps(row) + '\n')
        if event in ('start', 'health', 'heartbeat', 'player_exit', 'stop', 'error'):
            print(f"{row['utc']} {event}: {fields}", flush=True)

    def snapshot(reason):
        # Read only tails, including rotated newfile logs, so dumps stay bounded.
        root = Path('/mnt/wslg')
        try:
            pulse = sorted(root.glob('pulseaudio.log*'), key=lambda p: p.stat().st_mtime)[-3:]
            for path in pulse + [root / 'weston.log', root / 'stderr.log']:
                try:
                    with path.open('rb') as f:
                        f.seek(0, 2)
                        f.seek(max(0, f.tell() - 12000))
                        content = f.read(12000).decode(errors='replace')
                    log('wslg_log_tail', reason=reason, path=str(path), text=content)
                except OSError as e:
                    log('snapshot_error', path=str(path), error=str(e))
        except OSError as e:
            log('snapshot_error', error=str(e))

    def probe(*cmd):
        before = time.monotonic()
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=2)
            return {'status': 'ok' if result.returncode == 0 else 'error',
                    'code': result.returncode, 'seconds': round(time.monotonic()-before, 3),
                    'stdout': result.stdout[-24000:], 'stderr': result.stderr[-4000:]}
        except subprocess.TimeoutExpired:
            # subprocess.run kills and reaps the probe on timeout.
            return {'status': 'timeout', 'seconds': round(time.monotonic()-before, 3)}
        except OSError as e:
            return {'status': 'error', 'error': str(e)}

    def health_loop():
        previous = None
        count = 0
        while not stop.is_set():
            result = probe('pactl', 'info')
            log('health', **result)
            changed = result['status'] != previous
            if changed:
                snapshot('health_' + result['status'])
            if result['status'] == 'ok' and (changed or count % 6 == 0):
                for kind in ('sinks', 'sink-inputs'):
                    if stop.is_set():
                        break
                    log('pulse_details', kind=kind, **probe('pactl', 'list', kind))
            previous = result['status']
            count += 1
            stop.wait(5)

    def signal_stop(signum, frame):
        stop.set()

    signal.signal(signal.SIGINT, signal_stop)
    signal.signal(signal.SIGTERM, signal_stop)
    print(f'Logs: {folder}\nQuiet 440 Hz tone: half a second every two seconds. Ctrl-C stops.', flush=True)
    log('start', pulse_server=os.environ.get('PULSE_SERVER'), device=args.device,
        uname=list(os.uname()), note='Writes measure bytes accepted by pacat, not audible output. No automatic recovery.')
    # Smooth the edges of a quiet (-30 dBFS peak) tone. Maintain a continuous
    # stream, including silence, to test device lifetime across disconnection.
    rate = 48000
    pcm = array.array('h')
    for i in range(rate * 2):
        t = i / rate
        envelope = max(0, min(1, t / .01, (.5 - t) / .01))
        sample = int(1000 * envelope * math.sin(2 * math.pi * 440 * t))
        pcm.extend((sample, sample))
    if sys.byteorder != 'little':
        pcm.byteswap()
    wave = pcm.tobytes()
    command = ['pacat', '--playback', '--raw', '--verbose', '--format=s16le',
               '--rate=48000', '--channels=2', '--latency-msec=100',
               '--client-name=WSLg logging audio player', '--stream-name=Disconnect probe']
    if args.device:
        command += ['--device', args.device]
    # pacat's verbose timing output uses carriage returns; retain it verbatim.
    stderr = (folder / 'pacat.log').open('wb')
    player = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=stderr)
    os.set_blocking(player.stdin.fileno(), False)
    thread = threading.Thread(target=health_loop, daemon=True)
    thread.start()
    total = 0
    last_write = time.monotonic()
    next_heartbeat = last_write
    exited = False
    try:
        while not stop.is_set():
            now = time.monotonic()
            if args.duration and now - started >= args.duration:
                break
            if now >= next_heartbeat:
                log('heartbeat', bytes_accepted=total, audio_seconds_accepted=round(total/(rate*4), 3),
                    seconds_since_write=round(now-last_write, 3), player_returncode=player.poll())
                next_heartbeat = now + 5
            if player.poll() is not None or exited:
                if not exited:
                    log('player_exit', code=player.returncode)
                    snapshot('player_exit')
                    exited = True
                stop.wait(.2)
                continue  # Keep observing the service; never restart the player.
            _, writable, _ = select.select([], [player.stdin], [], .2)
            if writable:
                offset = total % len(wave)
                try:
                    n = os.write(player.stdin.fileno(), wave[offset:offset + 4096])
                    total += n
                    last_write = time.monotonic()
                except BlockingIOError:
                    pass
                except BrokenPipeError:
                    log('player_exit', code=player.poll(), reason='broken pipe')
                    snapshot('broken_pipe')
                    exited = True
    finally:
        stop.set()
        if player.poll() is None:
            player.terminate()
            try:
                player.wait(timeout=2)
            except subprocess.TimeoutExpired:
                player.kill()
                player.wait()
        player.stdin.close()
        stderr.close()
        thread.join(timeout=8)
        snapshot('stop')
        log('stop', player_returncode=player.returncode, bytes_accepted=total)
        events.close()
    return 0


if __name__ == '__main__':
    raise SystemExit(main())