feat(inputlayer-py): subscribe(), watch() and on() with typed Change events - #245

    Finished `release` profile [optimized] target(s) in 0.16s
Resolved 67 packages in 1ms
Checked 63 packages in 0.76ms
============================= test session starts ==============================
platform linux -- Python 3.10.22, pytest-9.1.1, pluggy-1.6.0 -- ~/.no-mistakes/worktrees/9eea230c3aae/01M43H9FB9Q10HGP9AX4J9GDNH/packages/inputlayer-py/.venv/bin/python
cachedir: .pytest_cache
rootdir: ~/.no-mistakes/worktrees/9eea230c3aae/01M43H9FB9Q10HGP9AX4J9GDNH/packages/inputlayer-py
configfile: pyproject.toml
plugins: langsmith-0.12.1, asyncio-1.3.0, anyio-4.13.0
asyncio: mode=auto, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collecting ... collected 58 items

tests/docs_queries_live.py::test_basic_queries_all_rows PASSED           [  1%]
tests/docs_queries_live.py::test_basic_queries_with_a_filter PASSED      [  3%]
tests/docs_queries_live.py::test_selecting_specific_columns PASSED       [  5%]
tests/docs_queries_live.py::test_a_projection_is_a_set PASSED            [  6%]
tests/docs_queries_live.py::test_joins PASSED                            [  8%]
tests/docs_queries_live.py::test_self_joins PASSED                       [ 10%]
tests/docs_queries_live.py::test_computed_columns PASSED                 [ 12%]
tests/docs_queries_live.py::test_ordering_and_pagination PASSED          [ 13%]
tests/docs_queries_live.py::test_sort_and_limit_run_in_the_engine PASSED [ 15%]
tests/docs_queries_live.py::test_aggregations_group_by_department_with_stats PASSED [ 17%]
tests/docs_queries_live.py::test_agg_two_measures_same_column PASSED     [ 18%]
tests/docs_queries_live.py::test_aggregates_leave_no_session_rule PASSED [ 20%]
tests/docs_queries_live.py::test_count_without_a_column PASSED           [ 22%]
tests/docs_queries_live.py::test_aggregations_top_k_per_group PASSED     [ 24%]
tests/docs_queries_live.py::test_aggregate_ordered_and_limited PASSED    [ 25%]
tests/docs_queries_live.py::test_working_with_results PASSED             [ 27%]
tests/docs_queries_live.py::test_query_plans PASSED                      [ 29%]
tests/docs_queries_live.py::test_debug_takes_the_arguments_of_query PASSED [ 31%]
tests/docs_queries_live.py::test_debug_shows_the_plan_of_an_aggregate_query PASSED [ 32%]
tests/docs_queries_live.py::test_why_returns_the_selected_columns_with_a_proof_per_row PASSED [ 34%]
tests/docs_queries_live.py::test_why_explains_an_aggregate_query PASSED  [ 36%]
tests/docs_queries_live.py::test_why_returns_a_column_of_the_second_joined_relation PASSED [ 37%]
tests/docs_queries_live.py::test_why_returns_a_computed_column PASSED    [ 39%]
tests/docs_queries_live.py::test_an_offset_without_a_limit_pages_in_the_engine PASSED [ 41%]
tests/docs_queries_live.py::test_why_orders_and_paginates_like_query PASSED [ 43%]
tests/docs_queries_live.py::test_why_returns_the_distinct_rows_of_a_projection_like_query PASSED [ 44%]
tests/docs_queries_live.py::test_paging_a_projection_ordered_by_an_unselected_column_is_refused PASSED [ 46%]
tests/docs_queries_live.py::test_why_orders_and_limits_an_aggregate_query_like_query PASSED [ 48%]
tests/docs_queries_live.py::test_raw_iql PASSED                          [ 50%]
tests/docs_queries_live.py::test_derived_relations_querying_a_recursive_rule PASSED [ 51%]
tests/docs_queries_live.py::test_sessions_session_facts_mix_with_persistent_data PASSED [ 53%]
tests/docs_queries_live.py::test_session_insert_keeps_every_fact PASSED  [ 55%]
tests/docs_queries_live.py::test_distance_functions PASSED               [ 56%]
tests/docs_queries_live.py::test_temporal_functions PASSED               [ 58%]
tests/docs_queries_live.py::test_or_conditions_merge_their_branches PASSED [ 60%]
tests/compile_rules_live.py::test_every_fixture_command_runs_and_rejected_spellings_fail PASSED [ 62%]
tests/compile_rules_live.py::test_rule_definition_reads_the_clauses PASSED [ 63%]
tests/compile_rules_live.py::test_session_rules_list_and_drop_by_clause PASSED [ 65%]
tests/compile_rules_live.py::test_datetime_columns_define_insert_and_read_back PASSED [ 67%]
tests/compile_rules_live.py::test_in_and_not_in_queries PASSED           [ 68%]
tests/compile_rules_live.py::test_in_in_a_rule_and_a_conditional_delete PASSED [ 70%]
tests/compile_rules_live.py::test_ground_negation_binds_a_constant_for_the_request_only PASSED [ 72%]
tests/compile_rules_live.py::test_negation_sharing_no_variable_is_refused_before_sending PASSED [ 74%]
tests/compile_rules_live.py::test_view_with_ground_negation PASSED       [ 75%]
tests/compile_rules_live.py::test_session_view_with_ground_negation PASSED [ 77%]
tests/compile_rules_live.py::test_negated_comparison PASSED              [ 79%]
tests/compile_rules_live.py::test_literals_round_trip_through_the_engine PASSED [ 81%]
tests/compile_rules_live.py::test_unrepresentable_values_are_refused_before_sending PASSED [ 82%]
tests/subscriptions_live.py::test_history_matches_the_oracle_generator PASSED [ 84%]
tests/subscriptions_live.py::test_snapshot_plus_deltas_equals_a_fresh_query_over_random_histories PASSED [ 86%]
tests/subscriptions_live.py::test_projection_two_supporting_tuples PASSED [ 87%]
tests/subscriptions_live.py::test_streamed_snapshot_and_streamed_deltas_are_assembled PASSED [ 89%]
tests/subscriptions_live.py::test_a_seq_gap_and_a_reset_end_in_unverified_then_an_exact_resync PASSED [ 91%]
tests/subscriptions_live.py::test_a_slow_consumer_gets_unverified_then_a_resync_once_it_has_read PASSED [ 93%]
tests/subscriptions_live.py::test_reconnect_gives_unverified_at_once_then_only_what_changed PASSED [ 94%]
tests/subscriptions_live.py::test_an_acl_revoke_ends_the_subscription_with_access_denied PASSED [ 96%]
tests/subscriptions_live.py::test_refused_before_or_at_the_engine PASSED [ 98%]
tests/subscriptions_live.py::test_watch_yields_the_whole_result_and_on_calls_back PASSED [100%]

============================= 58 passed in 22.23s ==============================

== S1 sync client: blocking iterator yields snapshot then delta
event 1: snapshot inserted= [E(a=1, b=1), E(a=2, b=2)] rev= 7 seq= 0 verified= True
event 2: delta inserted= [E(a=3, b=3)] retracted= [] seq= 1 verified= True
S1 PASS

== S2 auto_reconnect=False, server dies, connection gives up; new subscribe() fails fast
kg connection is il connection: False | kg conn state: open
kg connection state: closed closed_for_good: True
raised ConnectionLost after 0.001s: Not connected: the connection is closed
S2 PASS

== S3 first open while the server is down: transient failures retried, snapshot once it is back
kg connection state before subscribe: reconnecting
after 2s with server down, first event pending: True | errored: False
server restarted; first event after 0.34s: snapshot rows= [(1, 1), (2, 2), (3, 3)] verified= True
S3 PASS

"""Live drive of the P1-F fix-round behaviours plus the sync iterator against a
disposable inputlayer-server (private port, temp data dir)."""

import asyncio
import os
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import urllib.request

from inputlayer import ConnectionLost, InputLayer, InputLayerSync, Relation

BIN = sys.argv[1]
PORT = 8198
URL = f"ws://127.0.0.1:{PORT}/ws"
DATA = tempfile.mkdtemp(prefix="il-p1f-drive-")
server = None


class E(Relation):
    a: int
    b: int


def start():
    global server
    env = dict(os.environ, INPUTLAYER_ADMIN_PASSWORD="admin")
    server = subprocess.Popen(
        [BIN, "--host", "127.0.0.1", "--data-dir", DATA, "--port", str(PORT)],
        env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    for _ in range(80):
        try:
            urllib.request.urlopen(f"http://127.0.0.1:{PORT}/health", timeout=1)
            return
        except Exception:
            time.sleep(0.25)
    raise RuntimeError("server did not start")


def stop():
    server.terminate()
    server.wait()


def log(*a):
    print(*a, flush=True)


async def seed():
    il = InputLayer(URL, username="admin", password="admin")
    await il.connect()
    kg = il.knowledge_graph("p1f_drive")
    await kg.define(E)
    await kg.insert([E(a=1, b=1), E(a=2, b=2)])
    await il.close()


def scenario_sync_iterator():
    log("\n== S1 sync client: blocking iterator yields snapshot then delta")
    with InputLayerSync(URL, username="admin", password="admin") as il:
        kg = il.knowledge_graph("p1f_drive")
        sub = kg.subscribe(E)
        it = iter(sub)
        first = next(it)
        log("event 1:", first.kind, "inserted=", first.inserted, "rev=", first.revision,
            "seq=", first.seq, "verified=", first.verified)
        assert first.kind == "snapshot" and sorted(r.a for r in first.inserted) == [1, 2]
        with InputLayerSync(URL, username="admin", password="admin") as w:
            w.knowledge_graph("p1f_drive").insert(E(a=3, b=3))
        second = next(it)
        log("event 2:", second.kind, "inserted=", second.inserted, "retracted=", second.retracted,
            "seq=", second.seq, "verified=", second.verified)
        assert second.kind == "delta" and second.inserted == [E(a=3, b=3)] and second.seq == 1
        sub.close()
    log("S1 PASS")


async def scenario_given_up():
    log("\n== S2 auto_reconnect=False, server dies, connection gives up; new subscribe() fails fast")
    il = InputLayer(URL, username="admin", password="admin", auto_reconnect=False)
    await il.connect()
    kg = il.knowledge_graph("p1f_drive")
    await kg.query(E)  # open the graph's own connection
    log("kg connection is il connection:", kg._conn is il._conn, "| kg conn state:", kg._conn.state)
    stop()
    for _ in range(100):
        if kg._conn.state == "closed":
            break
        await asyncio.sleep(0.05)
    log("kg connection state:", kg._conn.state, "closed_for_good:", kg._conn.closed_for_good)
    t0 = time.monotonic()
    try:
        await asyncio.wait_for(kg.subscribe(E).__anext__(), 10)
        log("S2 FAIL: got an event")
        raise SystemExit(1)
    except ConnectionLost as e:
        log(f"raised {type(e).__name__} after {time.monotonic() - t0:.3f}s: {e}")
    except asyncio.TimeoutError:
        log("S2 FAIL: hung (no event, no error within 10s)")
        raise SystemExit(1)
    await il.close()
    log("S2 PASS")


async def scenario_first_open_retry():
    log("\n== S3 first open while the server is down: transient failures retried, snapshot once it is back")
    start()
    il = InputLayer(URL, username="admin", password="admin", reconnect_delay=0.1,
                    max_reconnect_attempts=1000)
    await il.connect()
    kg = il.knowledge_graph("p1f_drive")
    await kg.query(E)  # make sure the graph is selected
    stop()
    for _ in range(100):
        if kg._conn.state != "open":
            break
        await asyncio.sleep(0.05)
    log("kg connection state before subscribe:", kg._conn.state)
    sub = kg.subscribe(E)
    task = asyncio.ensure_future(sub.__anext__())
    await asyncio.sleep(2.0)
    log("after 2s with server down, first event pending:", not task.done(),
        "| errored:", task.done() and task.exception() is not None)
    assert not task.done(), task.exception() if task.done() else None
    t = time.monotonic()
    threading.Thread(target=start).start()
    ev = await asyncio.wait_for(task, 60)
    log(f"server restarted; first event after {time.monotonic() - t:.2f}s:",
        ev.kind, "rows=", sorted((r.a, r.b) for r in ev.inserted), "verified=", ev.verified)
    assert ev.kind == "snapshot" and ev.verified
    await sub.close()
    await il.close()
    log("S3 PASS")


try:
    start()
    asyncio.run(seed())
    scenario_sync_iterator()
    asyncio.run(scenario_given_up())
    asyncio.run(scenario_first_open_retry())
finally:
    if server and server.poll() is None:
        stop()
    shutil.rmtree(DATA, ignore_errors=True)

S1 snapshot inserted=[E(1,1),E(2,2)] seq=0 verified=True -> delta inserted=[E(3,3)] seq=1 PASS
S2 kg conn closed_for_good=True; raised ConnectionLost after 0.001s PASS
S3 conn reconnecting; first event still pending after 2s with no error; server restarted -> snapshot after 0.34s verified=True PASS