path: root/test/selfmount.py

#!/usr/bin/env python3
"""A session that opens, reads and saves its own 9P tree through a mount.

The editor runs under `9ns --mntgen`, so /mnt/9p/pardes/<name> inside its
namespace is its own tree, and is driven over its socket from outside. This
used to hang the editor outright: the realpath, the stat and the read went
out through the mount and came back as 9P requests only the editor's loop
could answer, while that loop was blocked making them. Now the connection's
task answers them while the editor waits (pardes.turn).

Needs 9ns (cloud9) on PATH or in ~/.local/bin, and a runtime directory; it
skips, loudly, without them.
"""
import fcntl
import os
from pathlib import Path
import pty
import shutil
import signal
import struct
import sys
import termios
import threading
import time

sys.path.insert(0, str(Path(__file__).resolve().parent))
from ninep import Client


def unix_mount(ns, binary, served, env, name, fixture):
    """A second editor under `9ns --unix <the first's socket>`, on the
    fixture (a command runs from a file's pane, not a terminal's): what its
    command pane sees in $PARDES_MOUNT."""
    second = name + 'u'
    sock = f'{os.environ["XDG_RUNTIME_DIR"]}/pardes-9p-{second}.sock'
    out = Path(f'/tmp/{second}.mount')
    env = dict(env, PARDES_DUMP=f'/tmp/{second}.dump.zon')
    pid, master = pty.fork()
    if pid == 0:
        os.execve(ns, ['9ns', '--unix', served, '--', binary, '--tty', '--9p=' + second, str(fixture)], env)
    fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack('HHHH', 30, 100, 0, 0))
    os.set_blocking(master, False)
    try:
        deadline = time.monotonic() + 30
        while time.monotonic() < deadline and not out.exists():
            try:
                os.read(master, 65536)
            except OSError:
                pass
            if os.path.exists(sock) and not out.with_suffix('.sent').exists():
                time.sleep(.3)
                with Client(sock) as client:
                    client.write('/ctl', b'Shell sh\n')
                    client.write('/exec', f'echo "${{PARDES_MOUNT-unset}}" > {out}.tmp; mv {out}.tmp {out}\n'.encode())
                out.with_suffix('.sent').touch()
            time.sleep(.1)
        return out.read_text().strip() if out.exists() else 'timeout'
    finally:
        os.kill(pid, signal.SIGTERM)
        time.sleep(.2)
        try:
            os.kill(pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        os.waitpid(pid, 0)
        for path in (out, out.with_suffix('.sent'), Path(f'/tmp/{second}.dump.zon')):
            path.unlink(missing_ok=True)


def run(binary):
    ns = shutil.which('9ns') or str(Path.home() / '.local/bin/9ns')
    runtime = os.environ.get('XDG_RUNTIME_DIR')
    if not os.path.exists(ns) or not runtime:
        print('selfmount: skipped (needs 9ns and XDG_RUNTIME_DIR)')
        return 0
    name = 'selfmount%d' % os.getpid()
    sock = f'{runtime}/pardes-9p-{name}.sock'
    fixture = Path(f'/tmp/{name}.txt')
    fixture.write_bytes(b'initial\n')
    env = {k: v for k, v in os.environ.items() if not k.startswith('PARDES_')}
    env.update(PARDES_NOTIME='1', TERM='xterm-256color', PARDES_DUMP=f'/tmp/{name}.dump.zon')
    for language in ['RS', 'C', 'GO', 'TS', 'PY']:
        env['PARDES_LSP_' + language] = ''
    pid, master = pty.fork()
    if pid == 0:
        os.execve(ns, ['9ns', '--mntgen', '--', binary, '--tty', '--9p=' + name, str(fixture)], env)
    fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack('HHHH', 30, 100, 0, 0))
    os.set_blocking(master, False)
    tail = bytearray()

    def drain():
        try:
            while True:
                chunk = os.read(master, 65536)
                if not chunk:
                    return
                tail.extend(chunk[-2000:])
        except OSError:
            pass

    failures = []

    def check(condition, what):
        print(('ok   ' if condition else 'FAIL ') + what)
        if not condition:
            failures.append(what)

    try:
        deadline = time.monotonic() + 15
        while not os.path.exists(sock):
            drain()
            if time.monotonic() > deadline:
                raise TimeoutError('socket never appeared: ' + tail.decode(errors='replace')[-400:])
            time.sleep(.02)
        time.sleep(.3)
        mount = f'/mnt/9p/pardes/{name}'
        with Client(sock) as client:
            index = client.read('/index').decode()
            rows = [row.rsplit(maxsplit=1)[0].split(maxsplit=3) for row in index.splitlines()]
            first = rows[0][0]
            # A Look at the tree's own index: realpath, stat and read all go
            # out through 9ns and back in, and the write returns when the
            # pane is open.
            started = time.monotonic()
            client.write('/look', (mount + '/index\n').encode())
            took = time.monotonic() - started
            check(took < 5, f'look at the mounted index returned ({took:.2f}s)')
            rows = [row.rsplit(maxsplit=1)[0].split(maxsplit=3) for row in client.read('/index').decode().splitlines()]
            opened = [row for row in rows if len(row) == 4 and row[3] == mount + '/index']
            check(len(opened) == 1, 'a pane opened on the mounted index')
            if opened:
                body = client.read(f'/pane/{opened[0][0]}/body').decode()
                check(body.startswith(index.splitlines()[0]), 'its body is the index as served')
            # A Look at pane 1's body through the mount, an edit of the copy,
            # and a Save: the save writes back through the mount into pane 1,
            # and the exec returns once it has.
            client.write('/look', f'{mount}/pane/{first}/body\n'.encode())
            rows = [row.rsplit(maxsplit=1)[0].split(maxsplit=3) for row in client.read('/index').decode().splitlines()]
            mirror = [row for row in rows if len(row) == 4 and row[3] == f'{mount}/pane/{first}/body']
            check(len(mirror) == 1, 'a pane opened on the mounted body')
            if mirror:
                pane = mirror[0][0]
                client.write(f'/pane/{pane}/addr', b'#0,#0')
                client.write(f'/pane/{pane}/data', b'HELLO ')
                client.write(f'/pane/{pane}/exec', b'Save\n')
                check(client.read(f'/pane/{first}/body') == b'HELLO initial\n', f'Save wrote back through the mount into pane {first}')
                text = ''.join(cell[0] for cell in client.screen()['cells'])
                check('saved ' + mount in text, 'the editor reported the save')
            # Through the kernel mount a write is cut at its msize: a
            # 1000-line burst and a 50 KB Edit block each still run whole,
            # line by line, and no piece of a line runs as a shell command.
            # sh: bash's and fish's prompt files are not reachable in 9ns's
            # namespace here, and a command pane needs no prompt marks.
            client.write('/ctl', b'Shell sh\n')
            # From the fixture's directory: a pane on the mount has one a
            # shell cannot enter.
            client.write('/focus', f'{first}\n'.encode())

            def run(line):
                """A command pane runs the line; answers its exit status."""
                seen = client.read('/log').decode().count('\nexit ')
                client.write('/exec', line)
                deadline = time.monotonic() + 20
                while time.monotonic() < deadline:
                    drain()
                    exits = [row for row in client.read('/log').decode().splitlines() if row.startswith('exit ')]
                    if len(exits) > seen:
                        return exits[-1].split()[2]
                    time.sleep(.1)
                return 'timeout'
            # A pane's command finds the session's own directory in
            # $PARDES_MOUNT (9ns --mntgen says NINE_MNTGEN=1).
            check(run(f'test "$PARDES_MOUNT" = {mount} && grep -q . "$PARDES_MOUNT/index"\n'.encode()) == '0',
                  '$PARDES_MOUNT is the session under the mntgen mount, and it answers')
            # Under `9ns --unix` the mount is another server, never this
            # session: a second editor started there leaves it unset.
            seen_mount = unix_mount(ns, binary, sock, env, name, fixture)
            check(seen_mount == 'unset', f'under 9ns --unix $PARDES_MOUNT is unset ({seen_mount!r})')
            # Followed from before the burst: a fresh open of the log reads
            # only what its 64 KiB ring still holds, which a burst this size
            # outgrows, and a follower the ring outruns reads `lost N`, so
            # every line is either heard once, in order, or a loss is said.
            heard = []

            def follow():
                with Client(sock) as follower:
                    fid = follower.open('/log', 2)
                    start = len(follower.read_fid(fid))
                    follower.rpc(118, struct.pack('<IQI', fid, 0, len(b'follow new\n')) + b'follow new\n')
                    ready.set()
                    deadline = time.monotonic() + 60
                    while time.monotonic() < deadline:
                        record = follower.read_fid(fid, start).decode()
                        heard.append(record)
                        if 'mnt-0999' in record:
                            return
            ready = threading.Event()
            reader = threading.Thread(target=follow, daemon=True)
            reader.start()
            check(ready.wait(10), 'a follower of the log is waiting')
            got = run(f"seq -f 'Msg mnt-%04g' 0 999 > {mount}/exec\n".encode())
            check(got == '0', f'a 1000-line burst written through the mount ({got!r})')
            reader.join(60)
            for r in heard:
                if r.startswith(('err ', 'exit ')):
                    print('     heard: ' + r.rstrip())
            said = [r.rstrip('\n').split(' ', 2)[2] for r in heard if r.startswith('msg ') and 'mnt-' in r]
            lost = [r for r in heard if r.startswith('lost ')]
            numbers = [int(s[4:]) for s in said]
            check(numbers == list(range(numbers[0] if numbers else 0, 1000)),
                  f'each line it heard ran once, whole, in order, through the last ({len(said)} heard)')
            check(len(said) == 1000 or (lost and heard.index(lost[0]) < heard.index(next(r for r in heard if 'mnt-' in r))),
                  f'no line was lost unsaid ({len(said)} heard, {lost!r})')
            check(not any('exit 127' in r for r in heard), 'no piece of a line ran as a command')
            scratch = int(client.read('/pane/new'))
            maker = "import sys; sys.stdout.write('Edit ,a\\n' + ''.join('line %05d of the block\\n' % i for i in range(2500)) + '.\\n')"
            check(run(f'python3 -c "{maker}" > {mount}/pane/{scratch}/ctl\n'.encode()) == '0', 'a 50 KB Edit block written through the mount')
            body = client.read(f'/pane/{scratch}/body').decode().splitlines()
            check(body == ['line %05d of the block' % i for i in range(2500)], f'it ran once, whole ({len(body)} lines)')
            # One 600 KiB line: 9ns sends it as 65512-byte Twrites, the last
            # one short, and none is taken for a whole line by its size.
            errs = client.read('/log').decode().count('\nerr ')
            long_edit = "import sys; sys.stdout.write('Edit ,c/' + 'q' * 600000 + '/\\n')"
            check(run(f'python3 -c "{long_edit}" > {mount}/pane/{scratch}/ctl\n'.encode()) == '0', 'a 600 KiB Edit line written through the mount')
            check(client.read(f'/pane/{scratch}/body') == b'q' * 600000, 'it ran once, whole')
            check(client.read('/log').decode().count('\nerr ') == errs, 'no piece of it was refused as a line of its own')
            # Over 1 MiB: refused once, its tail dropped through its newline.
            too_long = "import sys; sys.stdout.write('Msg ' + 'z' * 1150000 + '\\nMsg after-long\\n')"
            run(f'python3 -c "{too_long}" > {mount}/exec\n'.encode())
            log = client.read('/log').decode()
            check(log.count('over 1 MiB') == 1 and 'unknown control message' not in log and 'after-long' in log,
                  'a 1.1 MiB line is refused once, its tail never a line, the next line runs')
            # `pardes -` by hand through the mount: the directory, then a
            # program's output, escapes and all, on one open of /pager; the
            # +Pager reads back the plain text.
            got = run(f"{{ pwd; printf '\\033[31mred\\033[m\\n'; }} > {mount}/pager\n".encode())
            rows = [row.rsplit(maxsplit=1)[0].split(maxsplit=3) for row in client.read('/index').decode().splitlines()]
            paged = [row[0] for row in rows if len(row) == 4 and row[3].endswith('/+Pager')]
            check(got == '0' and len(paged) == 1, f'a +Pager made through the mount ({got!r}, {paged!r})')
            if paged:
                # The text goes in at the open's close, and a mount's close
                # (FUSE's release) comes after close(2) has returned.
                deadline = time.monotonic() + 5
                while (said := client.read(f'/pane/{paged[0]}/body')) != b'red\n' and time.monotonic() < deadline:
                    time.sleep(.05)
                check(said == b'red\n', f'its body is the plain text ({said!r})')
            # `>`'s truncate of exec changes nothing and waits on nothing,
            # and a last line with no newline runs once its open closes, at
            # the editor's next step, the close answered at once. Either
            # waiting on the editor hung: 9ns answers nothing else on the
            # mount while a setattr or a clunk is out, and the step may be
            # out reading a pane's file through it (the mounted index above).
            got = run(f"for i in $(seq 40); do printf 'Msg held-%d' $i > {mount}/exec || exit 9; done\n".encode())
            deadline = time.monotonic() + 5
            while 'held-40\n' not in (log := client.read('/log').decode()) and time.monotonic() < deadline:
                time.sleep(.05)
            check(got == '0' and all(f' held-{i}\n' in log for i in range(1, 41)),
                  f'40 closes holding a last line each through the mount, all answered, every line run ({got!r})')
            # An Edit whose filter writes the very ctl the Edit came by: 9ns
            # opens with FOPEN_PARALLEL_DIRECT_WRITES, so the filter's write
            # is not queued behind the one waiting for it. (Its own open may
            # not truncate, `1<>`, not `>`: Linux takes the file's lock
            # outright for an O_TRUNC open, and the waiting write holds it
            # shared; nothing a server answers can change that.)
            started = time.monotonic()
            got = run(f"printf 'Edit ,>echo Msg filter-hi 1<> {mount}/pane/{scratch}/ctl\\n' > {mount}/pane/{scratch}/ctl\n".encode())
            took = time.monotonic() - started
            check(got == '0' and took < 5 and 'filter-hi' in client.read('/log').decode(),
                  f'an Edit filter writing its own ctl through the mount answers promptly ({got!r}, {took:.1f}s)')
            # Another session dies under a burst of reads of it through the
            # registry mount: every read returns (an error is fine), none
            # hangs. 9ns invalidates the dead entry from a thread of its own,
            # never the one that must answer the reads holding its parent.
            other = name + 'k'
            other_sock = f'{runtime}/pardes-9p-{other}.sock'
            other_env = dict(env, PARDES_DUMP=f'/tmp/{other}.dump.zon')
            other_pid, other_master = pty.fork()
            if other_pid == 0:
                os.execve(binary, [binary, '--tty', '--9p=' + other], other_env)
            os.set_blocking(other_master, False)
            deadline = time.monotonic() + 15
            while not os.path.exists(other_sock) and time.monotonic() < deadline:
                try:
                    os.read(other_master, 65536)
                except OSError:
                    pass
                time.sleep(.05)
            codes = Path(f'/tmp/{name}.codes')
            codes.unlink(missing_ok=True)

            def kill_other():
                time.sleep(1.5)
                os.kill(other_pid, signal.SIGKILL)
            killer = threading.Thread(target=kill_other, daemon=True)
            killer.start()
            ran = run((f"for k in 1 2 3 4 5; do (end=$((SECONDS+4)); while [ $SECONDS -lt $end ]; do "
                       f"timeout 10 cat /mnt/9p/pardes/{other}/index > /dev/null 2>&1; echo $? >> {codes}; done) & done; wait\n").encode())
            killer.join(5)
            os.waitpid(other_pid, 0)
            got_codes = codes.read_text().split() if codes.exists() else []
            codes.unlink(missing_ok=True)
            for path in (other_sock, f'/tmp/{other}.dump.zon'):
                Path(path).unlink(missing_ok=True)
            check(ran == '0' and got_codes and '124' not in got_codes,
                  f'reads of a session killed under them all return ({ran!r}, {len(got_codes)} reads, {got_codes.count("124")} hung)')
            # Forty event reads held through the mount, a follower each, and
            # the mount still answers the rest: 9ns keeps workers past what
            # the editor holds (128), so `cat layout` is not queued behind
            # them for ever.
            held = run(f"for i in $(seq 40); do cat {mount}/pane/{first}/event > /dev/null & done; "
                       f"sleep 1; timeout 10 cat {mount}/layout > /dev/null; st=$?; kill $(jobs -p); exit $st\n".encode())
            check(held == '0', f'cat layout answers beside 40 held event reads ({held!r})')
            # fs.md's follow recipe, as written, with plain cat, its output
            # a pipe (where coreutils cat splices): it hears a record made
            # after it started.
            recipe = Path(f'/tmp/{name}.follow')
            ran = run(f"exec 3<>{mount}/log; echo follow >&3; "
                      f"(sleep 1; echo 'Msg recipe-heard' > {mount}/exec) & "
                      f"timeout 5 cat <&3 | cat > {recipe}; exit 0\n".encode())
            followed = recipe.read_text() if recipe.exists() else ''
            recipe.unlink(missing_ok=True)
            check(ran == '0' and 'recipe-heard' in followed, f'the log follow recipe with cat hears a new record ({followed[-120:]!r})')
            # A bash read loop on a pane's event: bash reads it a piece at a
            # time, and each record still arrives whole, in pieces.
            looped = Path(f'/tmp/{name}.events')
            ran = run((f"bash -c 'exec 3<{mount}/pane/{first}/event; "
                       f"(sleep 1; echo Msg ev-loop > {mount}/pane/{first}/exec) & "
                       f"read -r line <&3; echo \"$line\" > {looped}'\n").encode())
            got_line = looped.read_text() if looped.exists() else ''
            looped.unlink(missing_ok=True)
            check(ran == '0' and got_line.startswith('FX') and got_line.rstrip().endswith('Msg ev-loop'),
                  f'a bash read loop on event reads a whole record ({got_line!r})')
            # Past the session's 64 open records (write opens of exec take
            # them too) an open is refused EMFILE, "Too many open files".
            many = Path(f'/tmp/{name}.many')
            ran = run((f"timeout 10 bash -c 'for i in $(seq 80); do exec {{fd}}>>{mount}/exec || exit 9; done' 2> {many}; "
                       f"st=$?; echo rc=$st >> {many}; exit $st\n").encode())
            said_many = many.read_text() if many.exists() else ''
            many.unlink(missing_ok=True)
            check(ran == '9' and 'Too many open files' in said_many, f'an open past the 64 records is EMFILE ({ran!r}, {said_many[-80:]!r})')
            # An Edit whose command runs is held, not waited on: through the
            # mount, a status read beside a slow one answers at once, and a
            # command that reads this session's own tree runs.
            edited = int(client.read('/pane/new'))
            client.write(f'/pane/{edited}/body', b'pear\napple\n')
            took = Path(f'/tmp/{name}.took')
            ran = run((f"echo 'Edit , | sleep 3; sort' > {mount}/pane/{edited}/ctl & sleep .5; "
                       f"s=$(date +%s%N); timeout 5 cat {mount}/status > /dev/null; e=$(date +%s%N); "
                       f"echo $(( (e - s) / 1000000 )) > {took}; wait\n").encode())
            ms = int(took.read_text()) if took.exists() else -1
            took.unlink(missing_ok=True)
            check(ran == '0' and 0 <= ms < 1000, f'a status read beside a slow Edit answers at once ({ms} ms)')
            check(client.read(f'/pane/{edited}/body') == b'apple\npear\n', 'and the slow Edit applied')
            ran = run(f"echo 'Edit $ < cat {mount}/pane/{edited}/body' > {mount}/pane/{edited}/ctl\n".encode())
            check(ran == '0' and client.read(f'/pane/{edited}/body') == b'apple\npear\napple\npear\n',
                  f'an Edit whose command reads the session through its mount ({ran!r})')
            # A filter gets what a pane's command gets: the session's mount,
            # socket and pid.
            env_pane = int(client.read('/pane/new'))
            ran = run(f"echo 'Edit $ < printf \"%s|%s|%s\" \"$PARDES_MOUNT\" \"$PARDES_9P\" \"$PARDES_PID\"' > {mount}/pane/{env_pane}/ctl\n".encode())
            got_env = client.read(f'/pane/{env_pane}/body').decode().split('|')
            check(ran == '0' and len(got_env) == 3 and got_env[0] == mount and got_env[1] == sock and got_env[2].isdigit(),
                  f'an Edit filter has $PARDES_MOUNT, $PARDES_9P and $PARDES_PID ({got_env!r})')
            # An interrupted Edit write is flushed: its command is stopped and
            # nothing changes, and the next Edit runs.
            flushed = Path(f'/tmp/{name}.flushed')
            ran = run((f"timeout -s INT 1 sh -c \"echo 'Edit , | sleep 4; tr a-z A-Z' > {mount}/pane/{edited}/ctl\"; "
                       f"echo $? > {flushed}; sleep 1; echo 'Edit 1 | tr p P' > {mount}/pane/{edited}/ctl; echo $? >> {flushed}\n").encode())
            codes = flushed.read_text().split() if flushed.exists() else []
            flushed.unlink(missing_ok=True)
            time.sleep(4.5)
            check(codes == ['124', '0'] and client.read(f'/pane/{edited}/body') == b'aPPle\npear\napple\npear\n',
                  f'an interrupted Edit changes nothing, and the next runs ({codes!r}, {client.read(f"/pane/{edited}/body")!r})')
            client.write('/exec', b'Msg still here\n')
            # The terminal is read as it is waited on: a frame the editor
            # cannot write out (its pty full) is a frame /screen does not have.
            deadline = time.monotonic() + 8
            while 'still here' not in (text := ''.join(cell[0] for cell in client.screen()['cells'])) and time.monotonic() < deadline:
                drain()
                time.sleep(.1)
            check('still here' in text, 'the editor answers after all of that')
            # A filter that writes its own pane's ctl through the mount. Both
            # sides with `<>` run at once; a `>` or `>>` open on either side
            # waits in the kernel for the other write, and the command is
            # stopped at its 10 s limit: the Edit's write is answered then,
            # never held by a command it cannot kill.
            loop_pane = env_pane  # the screen has no room for another
            client.write(f'/pane/{loop_pane}/body', b'x\n', truncate=True)
            inner = Path(f'/tmp/{name}.inner')
            timed = Path(f'/tmp/{name}.timed')
            for outer_open, inner_open, quick in [('3<>', '1<>', True), ('>', '>', False), ('>>', '>', False),
                                                  ('3<>', '>>', False), ('>', '>>', False)]:
                # `clean`, which says nothing: a stopped filter's write still
                # lands once the Edit's is answered.
                inner.write_text(f'echo clean {inner_open} "$PARDES_MOUNT/pane/$winid/ctl"\n')
                edit = f'Edit 1 > sh {inner}'
                write = (f"exec 3<>{mount}/pane/{loop_pane}/ctl; echo '{edit}' >&3; rc=$?; exec 3>&-"
                         if outer_open == '3<>' else f"echo '{edit}' {outer_open} {mount}/pane/{loop_pane}/ctl; rc=$?")
                ran = run((f"s=$(date +%s%N); {write}; e=$(date +%s%N); "
                           f"echo $rc $(( (e - s) / 1000000 )) > {timed}\n").encode())
                rc_ms = timed.read_text().split() if timed.exists() else ['?', '-1']
                timed.unlink(missing_ok=True)
                ms = int(rc_ms[1])
                ok = ms >= 0 and (ms < 2000 if quick else ms < 13000)
                check(ran == '0' and ok, f'outer {outer_open}, filter {inner_open} on its own ctl: answered in {ms} ms (rc {rc_ms[0]})')
            inner.unlink(missing_ok=True)
            client.write('/exec', b'Msg after the stopped filters\n')
            check('msg ' in client.read('/log').decode().split('after the stopped filters')[0][-40:],
                  'and the editor answers after them')
    finally:
        os.kill(pid, signal.SIGTERM)
        time.sleep(.2)
        try:
            os.kill(pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        os.waitpid(pid, 0)
        for path in (fixture, Path(f'/tmp/{name}.dump.zon')):
            path.unlink(missing_ok=True)
    print('selfmount: ' + ('ok' if not failures else f'{len(failures)} failed'))
    return 0 if not failures else 1


if __name__ == '__main__':
    sys.exit(run(str(Path(sys.argv[1]).resolve())))