feat(ws): stream oversized subscription snapshots and deltas with bounded, complete delivery - #222

"""Live driver for W05 against a running inputlayer-server.

Usage: drive_ws.py <port> <scenario>
Each scenario prints a transcript of frame types/sizes and asserts the contract.
"""
import asyncio, json, sys, time

import websockets

PORT = int(sys.argv[1])
SCEN = sys.argv[2]
PW = "w05-live-password"
KG = sys.argv[3] if len(sys.argv) > 3 else "default"
URL = f"ws://127.0.0.1:{PORT}/ws?kg={KG}"
MAX_FRAME = 16 * 1024 * 1024


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


class C:
    def __init__(self, name):
        self.name = name

    async def open(self, **kw):
        self.ws = await websockets.connect(URL, max_size=None, **kw)
        await self.send({"type": "login", "username": "admin", "password": PW})
        n, r = await self.recv()
        assert r["type"] == "authenticated", r
        log(f"[{self.name}] authenticated protocol_version={r.get('protocol_version')}")
        return self

    async def send(self, v):
        await self.ws.send(json.dumps(v))

    async def recv(self, timeout=60):
        while True:
            raw = await asyncio.wait_for(self.ws.recv(), timeout)
            v = json.loads(raw)
            t = v.get("type", "")
            if t.endswith("_update") or t.endswith("_change"):
                continue
            return len(raw), v

    async def execute(self, program, quiet=False):
        await self.send({"type": "execute", "program": program})
        n, first = await self.recv()
        frames = [(n, first)]
        if first["type"] == "result_start":
            while True:
                n, f = await self.recv()
                frames.append((n, f))
                if f["type"] != "result_chunk":
                    break
        if not quiet:
            log(f"[{self.name}] {program[:70]!r} -> " + ", ".join(f"{f['type']}({n}B)" for n, f in frames[:4]) + (f" ... {len(frames)} frames" if len(frames) > 4 else ""))
        return frames


def summarize(frames):
    out = []
    for n, f in frames:
        d = {k: v for k, v in f.items() if k not in ("inserted", "retracted", "rows", "columns", "row_provenance")}
        for k in ("inserted", "retracted", "rows"):
            if k in f:
                d[k + "_len"] = len(f[k])
        d["bytes"] = n
        out.append(d)
    return out


async def read_delta(c, sub):
    """Read one logical delta (single frame or stream) for sub; returns (kind, frames)."""
    n, f = await c.recv()
    frames = [(n, f)]
    if f["type"] == "subscription_delta_start":
        while True:
            n, g = await c.recv()
            frames.append((n, g))
            if g["type"] == "subscription_delta_end":
                break
    return f["type"], frames


def check_stream(frames, expect_ins):
    start = frames[0][1]
    chunks = [f for _, f in frames[1:-1]]
    end = frames[-1][1]
    assert start["type"] == "subscription_delta_start"
    assert all(f["type"] == "subscription_delta_chunk" for f in chunks)
    assert [f["chunk_index"] for f in chunks] == list(range(len(chunks)))
    assert all(f["seq"] == start["seq"] for _, f in frames)
    assert max(n for n, _ in frames) <= MAX_FRAME
    ins = sum(len(f["inserted"]) for f in chunks)
    assert end["chunk_count"] == len(chunks) and end["inserted_count"] == ins == expect_ins, (end, ins)
    log(f"  stream ok: seq={start['seq']} chunks={len(chunks)} inserted={ins} largest_frame={max(n for n,_ in frames)}B total={sum(n for n,_ in frames)}B")


async def setup_big(c, n, pad, rel="big"):
    facts = ", ".join(f"({i})" for i in range(n))
    await c.execute(f"+n[{facts}]", quiet=True)
    await c.execute(f'+pad("{"x" * pad}")', quiet=True)
    await c.execute("+switch(0)", quiet=True)
    await c.execute(f"+{rel}(X, P) <- n(X), switch(1), pad(P)", quiet=True)


async def large_delta():
    w = await C("writer").open()
    a = await C("agent").open()
    await setup_big(w, 4000, 5000)
    r = await a.execute(".subscribe big ?big(X, P)")
    assert r[0][1]["type"] == "result"
    t0 = time.monotonic()
    await w.execute("+switch(1)")
    kind, frames = await read_delta(a, "big")
    log(f"  first frames: {json.dumps(summarize(frames[:2] + frames[-1:]))}")
    check_stream(frames, 4000)
    log(f"  ~{sum(n for n,_ in frames)/2**20:.1f} MiB delta (> 16 MiB frame limit) delivered in {time.monotonic()-t0:.2f}s")
    await w.execute("+n(4000)")
    kind, frames = await read_delta(a, "big")
    log(f"  follow-up: {json.dumps(summarize(frames))}")
    assert kind == "subscription_delta" and frames[0][1]["seq"] == 2
    log("PASS large_delta")


async def large_snapshot():
    w = await C("writer").open()
    await setup_big(w, 3000, 1000)
    await w.execute("+switch(1)")
    a = await C("agent").open()
    frames = await a.execute(".subscribe snap ?big(X, P)")
    s = summarize(frames)
    log(f"  start: {json.dumps(s[0])}")
    log(f"  end:   {json.dumps(s[-1])}")
    assert s[0]["type"] == "result_start" and s[0]["subscribed"]["subscription"] == "snap"
    assert s[-1]["type"] == "result_end" and s[-1]["row_count"] == 3000
    assert sum(x.get("rows_len", 0) for x in s[1:-1]) == 3000
    assert [x["chunk_index"] for x in s[1:-1]] == list(range(len(s) - 2))
    await w.execute("-n(0)")
    kind, f = await read_delta(a, "snap")
    log(f"  delta after snapshot: {json.dumps(summarize(f))}")
    assert kind == "subscription_delta" and f[0][1]["seq"] == 1
    # Python SDK assembles the same streamed reply and keeps `subscribed`.
    sys.path.insert(0, "packages/inputlayer-py/src")
    from inputlayer.connection import Connection
    conn = Connection(URL.split("?")[0], username="admin", password=PW, initial_kg=KG, auto_reconnect=False)
    await conn.connect()
    try:
        res = await conn.execute(".subscribe sdk ?big(X, P)")
        log(f"  python SDK: rows={len(res.rows)} row_count={res.row_count} subscribed={res.subscribed}")
        assert len(res.rows) == 2999 and res.subscribed is not None
    finally:
        await conn.close()
    log("PASS large_snapshot")


async def oversized_row():
    w = await C("writer").open()
    a = await C("agent").open()
    # Two 9 MiB strings joined in one row: an 18 MiB row, over the 16 MiB frame
    # limit, while each write stays under the inbound limit.
    half = 9 * 1024 * 1024
    await w.execute("+n(0)", quiet=True)
    await w.execute(f'+pa("{"x" * half}")', quiet=True)
    await w.execute(f'+pb("{"y" * half}")', quiet=True)
    await w.execute("+switch(0)", quiet=True)
    await w.execute("+big(X, P, Q) <- n(X), switch(1), pa(P), pb(Q)", quiet=True)
    r = await a.execute(".subscribe big ?big(X, P, Q)")
    assert r[0][1]["type"] == "result"
    await w.execute("+switch(1)")
    n, reset = await a.recv()
    log(f"  push: {json.dumps(reset)}")
    assert reset["type"] == "subscription_reset" and reset["subscription"] == "big"
    await w.execute("+n(1)")
    try:
        n, f = await a.recv(timeout=0.5)
        raise AssertionError(f"unexpected push after reset: {f['type']}")
    except asyncio.TimeoutError:
        log("  no further pushes after reset (0.5s)")
    r = await a.execute(".unsubscribe big")
    log(f"  .unsubscribe after reset: {json.dumps(summarize(r))[:300]}")
    assert r[0][1]["type"] == "error"
    r = await a.execute(".subscribe big ?big(X, P, Q)")
    log(f"  re-subscribe (undeliverable snapshot): {json.dumps(summarize(r))[:400]}")
    assert len(r) == 1 and r[0][1]["type"] == "error"
    await w.execute("+n(2)")
    try:
        n, f = await a.recv(timeout=0.5)
        raise AssertionError(f"unexpected push: {f['type']}")
    except asyncio.TimeoutError:
        log("  nothing registered, no pushes")
    log("PASS oversized_row")


class RawStalled:
    """A WebSocket client over a raw socket with a small receive buffer that
    simply stops calling recv(): a client that stops reading."""

    def __init__(self):
        import socket, base64, os
        self.s = socket.socket()
        self.s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 64 * 1024)
        self.s.connect(("127.0.0.1", PORT))
        key = base64.b64encode(os.urandom(16)).decode()
        self.s.sendall((f"GET /ws?kg={KG} HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
                        f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())
        self.buf = b""
        while b"\r\n\r\n" not in self.buf:
            self.buf += self.s.recv(4096)
        self.buf = self.buf.split(b"\r\n\r\n", 1)[1]

    def send(self, v):
        import os, struct
        data = json.dumps(v).encode()
        mask = os.urandom(4)
        n = len(data)
        hdr = bytes([0x81]) + (bytes([0x80 | n]) if n < 126 else bytes([0x80 | 126]) + struct.pack(">H", n))
        self.s.sendall(hdr + mask + bytes(b ^ mask[i % 4] for i, b in enumerate(data)))

    def recv_frame(self):
        import struct
        def need(k):
            while len(self.buf) < k:
                d = self.s.recv(65536)
                if not d:
                    raise EOFError
                self.buf += d
        need(2)
        op, n = self.buf[0] & 0x0F, self.buf[1] & 0x7F
        off = 2
        if n == 126:
            need(4); n = struct.unpack(">H", self.buf[2:4])[0]; off = 4
        elif n == 127:
            need(10); n = struct.unpack(">Q", self.buf[2:10])[0]; off = 10
        need(off + n)
        payload, self.buf = self.buf[off:off + n], self.buf[off + n:]
        return op, payload

    def recv_json(self):
        while True:
            op, p = self.recv_frame()
            if op == 1:
                v = json.loads(p)
                if not v["type"].endswith("_change") and not v["type"].endswith("_update"):
                    return v


async def stalled_reader():
    w = await C("writer").open()
    await setup_big(w, 8000, 5000)
    stalled = RawStalled()
    stalled.send({"type": "login", "username": "admin", "password": PW})
    log(f"[stalled] {stalled.recv_json()['type']} (raw socket, SO_RCVBUF=64KiB)")
    stalled.send({"type": "execute", "program": ".subscribe big ?big(X, P)"})
    log(f"[stalled] .subscribe big -> {stalled.recv_json()['type']}; now stops reading")
    reader = await C("reader").open()
    await reader.execute(".subscribe big ?big(X, P)")
    await reader.execute(".subscribe small ?switch(X)")
    t0 = time.monotonic()
    await w.execute("+switch(1)")
    # stalled never reads again
    got = set()
    while got != {"big", "small"}:
        kind, frames = await read_delta(reader, None)
        sub = frames[0][1]["subscription"]
        if kind == "subscription_delta_start":
            check_stream(frames, 8000)
        log(f"  reader got {kind} for {sub} at +{time.monotonic()-t0:.2f}s")
        got.add(sub)
    # stalled: after its socket buffers fill, server must close it within ws_send_timeout_ms
    # We probe from the server side: a small write must still be delivered to reader quickly.
    t1 = time.monotonic()
    await w.execute("+switch(2)")
    kind, frames = await read_delta(reader, None)
    log(f"  reader small delta latency while stalled peer pending: {(time.monotonic()-t1)*1000:.1f} ms ({kind}, {frames[0][1]['subscription']})")
    # Now drain the stalled socket: the server must have closed it, so EOF
    # arrives well before the whole ~40 MiB delta.
    await asyncio.sleep(1)
    total, kinds = 0, {}
    stalled.s.settimeout(10)
    try:
        while True:
            op, p = stalled.recv_frame()
            total += len(p)
            if op == 1:
                k = json.loads(p)["type"]
                kinds[k] = kinds.get(k, 0) + 1
            elif op == 8:
                kinds["close"] = kinds.get("close", 0) + 1
    except (EOFError, ConnectionResetError) as e:
        log(f"  stalled client: server closed the connection ({type(e).__name__}) after {total/2**20:.1f} MiB that were already in socket buffers; frames seen: {kinds}")
    assert total < 40 * 2**20, "stalled client got everything; not disconnected"
    assert kinds.get("subscription_delta_end", 0) == 0, "partial stream must not end as complete"
    log("PASS stalled_reader")


async def disconnect_mid_stream():
    w = await C("writer").open()
    await setup_big(w, 4000, 5000)
    leaving = await C("leaving").open()
    staying = await C("staying").open()
    await leaving.execute(".subscribe big ?big(X, P)")
    await staying.execute(".subscribe big ?big(X, P)")
    await w.execute("+switch(1)")
    n, start = await leaving.recv()
    assert start["type"] == "subscription_delta_start"
    n, chunk = await leaving.recv()
    log(f"  leaving got {start['type']} and {chunk['type']} #{chunk.get('chunk_index')}, then drops the socket")
    leaving.ws.transport.abort() if hasattr(leaving.ws, "transport") else None
    await leaving.ws.close()
    kind, frames = await read_delta(staying, "big")
    check_stream(frames, 4000)
    # server still serves new connections + subscribers
    await w.execute("+n(4000)")
    kind, frames = await read_delta(staying, "big")
    assert kind == "subscription_delta"
    r = await (await C("fresh").open()).execute("?n(4000)")
    log(f"  server healthy after mid-stream disconnect: {r[0][1]['type']} rows={len(r[0][1]['rows'])}")
    log("PASS disconnect_mid_stream")


async def small_latency():
    w = await C("writer").open()
    a = await C("agent").open()
    await w.execute("+m[" + ", ".join(f"({i})" for i in range(1)) + "]", quiet=True)
    await w.execute("+on(0)", quiet=True)
    await w.execute("+wide(X) <- m(X), on(1)", quiet=True)
    await w.execute("+k[" + ", ".join(f"({i})" for i in range(300)) + "]", quiet=True)
    await w.execute("+many(X) <- k(X), on(1)", quiet=True)
    await a.execute(".subscribe one ?wide(X)")
    await a.execute(".subscribe many ?many(X)")
    lat = []
    for i in range(1, 41):
        t = time.monotonic()
        await w.execute(f"+on({i % 2})" if i % 2 else f"-on(1)", quiet=True)
        seen = set()
        while len(seen) < 2:
            kind, frames = await read_delta(a, None)
            assert kind == "subscription_delta", kind
            assert len(frames) == 1
            seen.add(frames[0][1]["subscription"])
        lat.append((time.monotonic() - t) * 1000)
    lat.sort()
    log(f"  40 commits, each pushing a 1-row and a 300-row delta as single frames; write->both-deltas latency ms: p50={lat[20]:.2f} p90={lat[36]:.2f} max={lat[-1]:.2f}")
    log("PASS small_latency")


async def result_cap():
    w = await C("writer").open()
    a = await C("agent").open()
    await w.execute("+c[" + ", ".join(f"({i})" for i in range(50)) + "]", quiet=True)
    await w.execute("+gate(0)", quiet=True)
    await w.execute("+capped(X) <- c(X), gate(1)", quiet=True)
    r = await a.execute(".subscribe cap ?capped(X)")
    assert r[0][1]["type"] == "result"
    await w.execute("+gate(1)")
    n, f = await a.recv()
    log(f"  push when result exceeds max_result_rows: {json.dumps(summarize([(n, f)]))[:400]}")
    assert f["type"] in ("subscription_error", "subscription_reset"), f
    r = await a.execute(".subscribe cap2 ?capped(X)")
    log(f"  .subscribe over the cap: {json.dumps(summarize(r))[:400]}")
    assert len(r) == 1 and r[0][1]["type"] == "error"
    await w.execute("-c[" + ", ".join(f"({i})" for i in range(5, 50)) + "]")
    kind, fr = await read_delta(a, "cap")
    assert kind == "subscription_delta" and len(fr[0][1]["inserted"]) == 5
    log(f"  after shrinking under the cap: {json.dumps(summarize(fr))[:300]}")
    log("PASS result_cap")


asyncio.run(globals()[SCEN]())

[writer] authenticated protocol_version=3
[agent] authenticated protocol_version=3
[agent] '.subscribe big ?big(X, P)' -> result(188B)
[writer] '+switch(1)' -> result(197B)
  first frames: [{"type": "subscription_delta_start", "subscription": "big", "generation": 1, "knowledge_graph": "s_delta", "seq": 1, "revision": 16, "bytes": 141}, {"type": "subscription_delta_chunk", "subscription": "big", "generation": 1, "seq": 1, "chunk_index": 0, "inserted_len": 209, "retracted_len": 0, "bytes": 1047184}, {"type": "subscription_delta_end", "subscription": "big", "generation": 1, "seq": 1, "chunk_count": 20, "inserted_count": 4000, "retracted_count": 0, "bytes": 136}]
  stream ok: seq=1 chunks=20 inserted=4000 largest_frame=1047192B total=20041637B
  ~19.1 MiB delta (> 16 MiB frame limit) delivered in 3.48s
[writer] '+n(4000)' -> result(192B)
  follow-up: [{"type": "subscription_delta", "subscription": "big", "generation": 1, "knowledge_graph": "s_delta", "seq": 2, "revision": 17, "inserted_len": 1, "retracted_len": 0, "bytes": 5173}]
PASS large_delta

[writer] authenticated protocol_version=3
[writer] '+switch(1)' -> result(197B)
[agent] authenticated protocol_version=3
[agent] '.subscribe snap ?big(X, P)' -> result_start(182B), result_chunk(504987B), result_chunk(504995B), result_chunk(504990B) ... 8 frames
  start: {"type": "result_start", "total_count": 3000, "truncated": false, "execution_time_ms": 240, "errors": [], "subscribed": {"subscription": "snap", "generation": 1, "revision": 22}, "bytes": 182}
  end:   {"type": "result_end", "row_count": 3000, "chunk_count": 6, "bytes": 54}
[writer] '-n(0)' -> result(189B)
  delta after snapshot: [{"type": "subscription_delta", "subscription": "snap", "generation": 1, "knowledge_graph": "s_snap", "seq": 1, "revision": 23, "inserted_len": 0, "retracted_len": 1, "bytes": 1170}]
  python SDK: rows=2999 row_count=2999 subscribed=Subscribed(subscription='sdk', generation=1, revision=23)
PASS large_snapshot

[writer] authenticated protocol_version=3
[agent] authenticated protocol_version=3
[agent] '.subscribe big ?big(X, P, Q)' -> result(187B)
[writer] '+switch(1)' -> result(197B)
  push: {"type": "subscription_reset", "subscription": "big", "generation": 1, "message": "Delta 1 has a row of 18874377 bytes, over the 16777216-byte message limit. The subscription was removed; subscribe again."}
[writer] '+n(1)' -> result(192B)
  no further pushes after reset (0.5s)
[agent] '.unsubscribe big' -> error(70B)
  .unsubscribe after reset: [{"type": "error", "message": "No subscription 'big' on this connection.", "bytes": 70}]
[agent] '.subscribe big ?big(X, P, Q)' -> error(123B)
  re-subscribe (undeliverable snapshot): [{"type": "error", "message": "The result has a row of 18874377 bytes, over the 16777216-byte message limit", "code": "internal", "bytes": 123}]
[writer] '+n(2)' -> result(192B)
  nothing registered, no pushes
PASS oversized_row

[writer] authenticated protocol_version=3
[stalled] authenticated (raw socket, SO_RCVBUF=64KiB)
[stalled] .subscribe big -> result; now stops reading
[reader] authenticated protocol_version=3
[reader] '.subscribe big ?big(X, P)' -> result(188B)
[reader] '.subscribe small ?switch(X)' -> result(195B)
[writer] '+switch(1)' -> result(197B)
  reader got subscription_delta for small at +0.01s
  stream ok: seq=1 chunks=39 inserted=8000 largest_frame=1047194B total=40083994B
  reader got subscription_delta_start for big at +4.26s
[writer] '+switch(2)' -> result(197B)
  reader small delta latency while stalled peer pending: 6.9 ms (subscription_delta, small)
  stalled client: server closed the connection (EOFError) after 2.0 MiB that were already in socket buffers; frames seen: {'persistent_update': 1, 'subscription_delta_start': 1, 'subscription_delta_chunk': 2}
PASS stalled_reader
2026-10-03T22:27:13.278797Z  WARN tokio-rt-worker ThreadId(30) inputlayer::protocol::rest::handlers::ws::outbound: ws_send_timeout timeout=Some(300ms)

[writer] authenticated protocol_version=3
[leaving] authenticated protocol_version=3
[staying] authenticated protocol_version=3
[leaving] '.subscribe big ?big(X, P)' -> result(188B)
[staying] '.subscribe big ?big(X, P)' -> result(188B)
[writer] '+switch(1)' -> result(197B)
  leaving got subscription_delta_start and subscription_delta_chunk #0, then drops the socket
  stream ok: seq=1 chunks=20 inserted=4000 largest_frame=1047192B total=20041636B
[writer] '+n(4000)' -> result(192B)
[fresh] authenticated protocol_version=3
[fresh] '?n(4000)' -> result(300B)
  server healthy after mid-stream disconnect: result rows=1
PASS disconnect_mid_stream

[writer] authenticated protocol_version=3
[agent] authenticated protocol_version=3
[agent] '.subscribe one ?wide(X)' -> result(187B)
[agent] '.subscribe many ?many(X)' -> result(188B)
  40 commits, each pushing a 1-row and a 300-row delta as single frames; write->both-deltas latency ms: p50=13.70 p90=17.63 max=19.91
PASS small_latency

[writer] authenticated protocol_version=3
[agent] authenticated protocol_version=3
[agent] '.subscribe cap ?capped(X)' -> result(186B)
[writer] '+gate(1)' -> result(195B)
  push when result exceeds max_result_rows: [{"type": "subscription_error", "subscription": "cap", "generation": 1, "message": "Subscription result exceeds storage.performance.max_result_rows (10); no complete result to deliver. Narrow the query.", "bytes": 196}]
[agent] '.subscribe cap2 ?capped(X)' -> error(147B)
  .subscribe over the cap: [{"type": "error", "message": "Subscription result exceeds storage.performance.max_result_rows (10); no complete result to deliver. Narrow the query.", "bytes": 147}]
[writer] '-c[(5), (6), (7), (8), (9), (10), (11), (12), (13), (14), (15), (16), ' -> result(192B)
  after shrinking under the cap: [{"type": "subscription_delta", "subscription": "cap", "generation": 1, "knowledge_graph": "s_cap", "seq": 1, "revision": 11, "inserted_len": 5, "retracted_len": 0, "bytes": 177}]
PASS result_cap

2026-10-03T22:25:40.625075Z  WARN tokio-rt-worker ThreadId(36) inputlayer::protocol::rest::handlers::ws::stream: ws_row_too_large row=0 bytes=18874377 max=16777216
2026-10-03T22:25:40.625318Z  WARN tokio-rt-worker ThreadId(31) inputlayer::protocol::rest::handlers::ws::push: ws_subscription_reset subscription=big generation=1 reason=Delta 1 has a row of 18874377 bytes, over the 16777216-byte message limit
2026-10-03T22:25:43.937362Z  WARN tokio-rt-worker ThreadId(04) inputlayer::protocol::rest::handlers::ws::stream: ws_row_too_large row=0 bytes=18874377 max=16777216
2026-10-03T22:27:13.278797Z  WARN tokio-rt-worker ThreadId(30) inputlayer::protocol::rest::handlers::ws::outbound: ws_send_timeout timeout=Some(300ms)
2026-10-03T22:27:37.976024Z  WARN tokio-rt-worker ThreadId(33) inputlayer::protocol::rest::handlers::ws::outbound: ws_send_timeout timeout=Some(300ms)

    Finished `test` profile [unoptimized + debuginfo] target(s) in 25.86s
     Running tests/ws_delivery_tests.rs (target/debug/deps/ws_delivery_tests-1f3bb6cc525e3cf9)

running 5 tests
test a_large_snapshot_streams_as_the_subscribe_reply ... ok
test a_disconnect_mid_stream_releases_only_that_connection ... ok
test a_delta_over_the_frame_limit_streams_whole_and_numbering_continues ... ok
test a_client_that_stops_reading_is_disconnected_and_isolated ... ok
test a_row_no_frame_can_carry_resets_the_subscription ... ok

test result: ok. 5 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 7.44s

    Finished `test` profile [unoptimized + debuginfo] target(s) in 30.56s
     Running unittests src/lib.rs (target/debug/deps/inputlayer-a741300da475ef27)

running 18 tests
test protocol::rest::handlers::ws::outbound::tests::unbound_connection_is_not_fenced ... ok
test protocol::rest::handlers::ws::push::tests::integers_are_estimated_exactly ... ok
test protocol::rest::handlers::ws::push::tests::the_estimate_covers_the_serialized_rows ... ok
test protocol::rest::handlers::ws::push::tests::a_small_delta_of_many_rows_is_encoded_without_the_blocking_pool ... ok
test protocol::rest::handlers::ws::outbound::tests::revoked_credential_stops_data_frames_but_not_control_frames ... ok
test protocol::rest::handlers::ws::stream::tests::a_small_delta_is_one_frame_serialized_as_before ... ok
test protocol::rest::handlers::ws::stream::tests::a_small_result_is_one_frame_serialized_as_before ... ok
test protocol::rest::handlers::ws::stream::tests::errors_and_resets_pass_through_whole ... ok
test protocol::rest::handlers::ws::push::tests::a_wide_small_integer_delta_that_fits_one_frame_stays_inline ... ok
test protocol::rest::handlers::ws::push::tests::a_wide_delta_behind_a_short_first_row_goes_to_the_blocking_pool ... ok
test protocol::rest::handlers::ws::outbound::tests::a_stalled_reader_fails_the_connection_after_the_send_timeout ... ok
test protocol::rest::handlers::ws::push::tests::a_delta_too_large_for_one_frame_is_estimated_over_the_budget ... ok
test protocol::rest::handlers::ws::outbound::tests::without_a_send_timeout_a_stalled_reader_is_waited_for ... ok
test protocol::rest::handlers::ws::stream::tests::a_large_delta_streams_as_one_logical_delta ... ok
test protocol::rest::handlers::ws::stream::tests::a_large_result_streams_as_one_bounded_ordered_result ... ok
test protocol::rest::handlers::ws::outbound::tests::oversized_frame_becomes_an_error_for_the_same_request ... ok
test protocol::rest::handlers::ws::stream::tests::a_delta_with_a_row_no_frame_can_carry_is_undeliverable ... ok
test protocol::rest::handlers::ws::stream::tests::a_result_with_a_row_no_frame_can_carry_is_undeliverable ... ok

test result: ok. 18 passed; 0 failed; 0 ignored; 0 measured; 2928 filtered out; finished in 1.13s

   Compiling thiserror-impl v1.0.69
   Compiling thiserror v1.0.69
test streamed::tests::nothing_applies_before_the_end ... ok
test streamed::tests::a_complete_stream_is_one_delta ... ok
test streamed::tests::missing_duplicate_or_foreign_chunks_break_the_stream ... ok
test result: ok. 9 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
test server::tests::push_wire_shape ... ok
test server::tests::notices_and_pushes_never_carry_an_id ... ok
test server::tests::streamed_delta_wire_shape ... ok
test server::tests::streamed_subscribe_reply_names_the_subscription ... ok
test result: ok. 19 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

tests/test_connection.py::TestConnectionStreaming::test_stream_assembly PASSED [  7%]
tests/test_connection.py::TestConnectionStreaming::test_streamed_subscribe_keeps_its_subscription PASSED [ 15%]
tests/test_connection.py::TestConnectionStreaming::test_incomplete_stream_is_not_a_result[chunks0-end0] PASSED [ 23%]
tests/test_connection.py::TestConnectionStreaming::test_incomplete_stream_is_not_a_result[chunks1-end1] PASSED [ 30%]
tests/test_connection.py::TestConnectionStreaming::test_incomplete_stream_is_not_a_result[chunks2-end2] PASSED [ 38%]
tests/test_connection.py::TestConnectionNotifications::test_notices_and_subscription_pushes_are_not_replies PASSED [ 46%]
tests/test_protocol.py::TestDeserializeStreaming::test_result_start PASSED [ 53%]
tests/test_protocol.py::TestDeserializeStreaming::test_result_chunk PASSED [ 61%]
tests/test_protocol.py::TestDeserializeStreaming::test_result_end PASSED [ 69%]
tests/test_protocol.py::TestRequestIds::test_subscribe_reply_names_its_generation PASSED [ 76%]
tests/test_protocol.py::TestDeserializePushes::test_subscription_frames PASSED [ 84%]
tests/test_protocol.py::TestDeserializePushes::test_streamed_delta_and_reset_frames PASSED [ 92%]
tests/test_protocol.py::TestDeserializePushes::test_streamed_subscribe_reply_names_the_subscription PASSED [100%]
====================== 13 passed, 45 deselected in 0.05s =======================

 ✓ tests/protocol.test.ts > deserializeMessage > deserializes streaming messages 0ms
 ✓ tests/protocol.test.ts > request ids and pushes > classifies notices and subscription frames as pushes 0ms
 ✓ tests/statement-errors.test.ts > error frames > notices and subscription pushes do not answer a call 3ms
 ✓ tests/statement-errors.test.ts > chunked results > errors on result_start survive assembly 5ms
 ✓ tests/statement-errors.test.ts > chunked results > a successful stream keeps its timing 2ms
 ✓ tests/statement-errors.test.ts > chunked results > an error frame mid-stream rejects typed 2ms
 ✓ tests/statement-errors.test.ts > chunked results > a streamed subscribe keeps its subscription 2ms
 ✓ tests/statement-errors.test.ts > chunked results > a missing chunk is not a result 2ms
 ✓ tests/statement-errors.test.ts > chunked results > a repeated chunk is not a result 2ms
 ✓ tests/statement-errors.test.ts > chunked results > rows that do not add up is not a result 2ms
      Tests  32 passed (32)
   Duration  856ms (transform 304ms, setup 0ms, collect 414ms, tests 98ms, environment 0ms, prepare 245ms)