std/src/net/split_test.nv - main - nv-lang/nova

std/src/net/split_test.nv
221 строка10 KB
Evgeniy Golovin
style(std): D452 — migrate to canonical match-arm/statement separators
10 авг 2026, 03:57
10 авг 2026, 03:57685589a
// SPDX-License-Identifier: MIT OR Apache-2.0
// Plan 182 Ф.1 (migrated from nova_tests/plan91_16: tcp_split_echo_slow.nv,
// tcp_split_mock.nv) — TcpStream.into_split() into TcpReadHalf/TcpWriteHalf.
// Read and write run on DIFFERENT halves backed by independent C-side park
// slots (read_scope/read_slot vs write_scope/write_slot) — a single shared
// op_slot would corrupt under the cross-fiber read/write traffic this test
// drives. Halves only expose the raw byte methods (@read/@write) — no
// read_text/write_str convenience — so tests decode through a caller buffer.

module std.net

import std.time.duration

fn split_test_one_byte(v int) -> []u8 {
    mut out []u8 = []u8.new()
    out.push(v as u8)
    out
}

// Plan 249 Ф.0-а regression: `TcpWriteHalf consume @close()` must send a
// real TCP FIN (half-close) to the peer, WHILE the read half stays
// genuinely open/usable — this is the exact shape a bidirectional relay
// (`pipe_bidirectional`) needs when one direction sees EOF.
//
// Before the fix, closing the write half alone only decremented net.c's
// internal split-refcount and returned WITHOUT calling `uv_shutdown` (see
// `TcpWriteHalf consume @close()`'s doc comment above, this module) — the
// peer's second read blocked forever. Guarded by a manual `CancelToken` +
// `supervised(cancel:)` watchdog, NOT `supervised(timeout:)`: a separate,
// distinct finding from this same plan window is that `timeout:` does not
// reach a `.read()` nested inside a sibling `spawn` body (confirmed hung
// past 30s in an isolated probe; manual `CancelToken.cancel()` from a
// third watchdog spawn DOES reach it, in ~3s) — using `timeout:` here would
// make a REGRESSION of this very fix hang the whole `nova test` run instead
// of failing it.
test "TCP split: TcpWriteHalf.close() alone delivers FIN, read half stays usable (#Ф.0-а)" {
    with Net = real_net() {
        ro ch_port = Channel[int].new(1)
        ro ch_result_a = Channel[int].new(1)
        ro ch_result_b = Channel[int].new(1)
        ro tok = CancelToken.new()
        supervised(cancel: tok) {
            // Watchdog: if this regresses, B's second read (and then A's
            // echo read) would block forever — cancel at 2.5s so the test
            // FAILS observably instead of hanging the suite.
            spawn {
                2500.to_millis().sleep()
                tok.cancel()
            }
            spawn {
                consume lst = TcpListener.bind(SocketAddr.loopback(0))!!
                ch_port.tx.send(lst.local_port())
                ro accepted = lst.accept()
                lst.close()
                match accepted {
                    Ok(consume conn) => {
                        consume (ar, aw) = conn.into_split()
                        ro _ = aw.write(split_test_one_byte(0x2A))
                        aw.close()          // write-half close ONLY — pin under test
                        mut echo_buf []u8 = []u8.new()
                        echo_buf.resize(1, 0 as u8)
                        ro echoed = ar.read(echo_buf)   // ar stays genuinely open/parked here
                        ar.close()
                        match echoed {
                            Ok(n) => ch_result_a.tx.send(1000 + n * 10 + (echo_buf[0] as int))
                            Err(_) => ch_result_a.tx.send(-1001)
                        }
                    }
                    Err(_) => ch_result_a.tx.send(-1002)
                }
            }
            spawn {
                ro port = match ch_port.rx.recv() { Some(p) => p, None => 0 }
                match TcpStream.connect(SocketAddr.loopback(port)) {
                    Ok(consume conn) => {
                        consume (br, bw) = conn.into_split()
                        mut buf1 []u8 = []u8.new()
                        buf1.resize(1, 0 as u8)
                        ro n1 = match br.read(buf1) { Ok(k) => k, Err(_) => -1 }
                        mut buf2 []u8 = []u8.new()
                        buf2.resize(1, 0 as u8)
                        ro n2 = match br.read(buf2) { Ok(k) => k, Err(_) => -2 }
                        ro _ = bw.write(split_test_one_byte(0x2A))
                        br.close()
                        bw.close()
                        ch_result_b.tx.send(n1 * 100 + n2)
                    }
                    Err(_) => ch_result_b.tx.send(-999)
                }
            }
        }
        ro r1 = match ch_result_a.rx.recv() { Some(v) => v, None => -998 }
        ro r2 = match ch_result_b.rx.recv() { Some(v) => v, None => -998 }
        // Healthy: r2 == 100 (n1=1 byte, n2=0 EOF); r1 == 1052 (A's
        // ar.read got the 1-byte echo back). Broken: watchdog cancels ->
        // r1 == -1001, r2 == 98.
        assert(r1 == 1052)
        assert(r2 == 100)
    }
}

test "TCP split: server + client exchange via read/write halves _slow" {
    with Net = real_net() {
        // D415 (E_CONCURRENT_MUT_CAPTURE): listener живёт ЦЕЛИКОМ в server-
        // файбере, порт публикуется каналом — mut-захвата нет; заодно это
        // канонический M:N-паттерн (bind внутри файбера его I/O, см. udp_test.nv).
        ro ch = Channel[int].new(1)   // server port → client

        supervised {
            // Client fiber: connect, split, write request, read reply.
            spawn {
                ro port = match ch.rx.recv() { Some(x) => x, None => 0 }
                consume conn = match TcpStream.connect(SocketAddr.loopback(port)) {
                    Ok(consume s) => s
                    Err(_) => panic("connect failed")
                }
                mut (cli_rd, cli_wr) = conn.into_split()

                ro wr = cli_wr.write("client-hello")
                assert(wr.is_ok())
                ro nbytes = match wr { Ok(n) => n, Err(_) => panic("client write") }
                assert(nbytes == 12)

                mut reply_buf []u8 = []u8.new()
                reply_buf.resize(64, 0 as u8)
                ro rd = cli_rd.read(reply_buf)
                assert(rd.is_ok())
                ro reply_n = match rd { Ok(n) => n, Err(_) => panic("client read") }
                assert(unsafe { reply_buf[0..reply_n].to_str_unchecked() } == "server-greeting")

                cli_rd.close()
                cli_wr.close()
            }

            // Server fiber: bind + accept в одном файбере, listener не покидает его.
            spawn {
                consume lst = match TcpListener.bind(SocketAddr.loopback(0)) {
                    Ok(consume l) => l
                    Err(_) => panic("bind failed")
                }
                ch.tx.send(lst.local_port())
                // Explicit close before `panic` on this (never-taken outside
                // a real failure) error arm — kept for clarity/symmetry with
                // byte_surface_test.nv's UdpSocket sites even though Plan
                // 217.1 (D432) gave `TcpListener` its own `@cleanup`, so a
                // panic here would now auto-close `lst` regardless.
                consume srv = match lst.accept() {
                    Ok(consume s) => s
                    Err(_) => { lst.close(); panic("accept failed") }
                }
                mut (srv_rd, srv_wr) = srv.into_split()

                mut req_buf []u8 = []u8.new()
                req_buf.resize(64, 0 as u8)
                ro incoming = srv_rd.read(req_buf)
                assert(incoming.is_ok())
                ro req_n = match incoming { Ok(n) => n, Err(_) => { lst.close(); panic("srv read") } }
                assert(unsafe { req_buf[0..req_n].to_str_unchecked() } == "client-hello")

                ro reply_wr = srv_wr.write("server-greeting")
                assert(reply_wr.is_ok())

                srv_rd.close()
                srv_wr.close()
                lst.close()
            }
        }
    }
}

test "split mock stream — read half + write half" {
    with Net = mock_net() {
        supervised {
            spawn {
                consume s = match TcpStream.connect(SocketAddr.loopback(0)) {
                    Ok(consume c) => c
                    Err(_) => panic("connect")
                }
                mut (rdh, wrh) = s.into_split()

                ro w1 = wrh.write("hello")
                assert(w1.is_ok())
                ro nbytes = match w1 { Ok(n) => n, Err(_) => -1 }
                assert(nbytes == 5)

                mut buf []u8 = []u8.new()
                buf.resize(64, 0 as u8)
                ro got = rdh.read(buf)
                assert(got.is_ok())
                ro n = match got { Ok(n) => n, Err(_) => 0 }
                assert(unsafe { buf[0..n].to_str_unchecked() } == "mock-reply")

                rdh.close()
                wrh.close()
            }
        }
    }
}

// №498 fix (Plan 258 Ф.1, registry docs/plans/221.1-bug-sweep.md): `into_split()`
// on a stream with a LIVE `share()`-copy must panic (sole ownership required —
// see `into_split()`'s own doc comment, tcp.nv) WITHOUT mutating `rc` on the way
// there. Before the fix, `@rc.fetch_sub(1) != 1` decremented the shared counter
// UNCONDITIONALLY, ahead of the check that can fail — the copy that survives the
// panicking fiber (`s2` below) would have been left owning a permanently-wrong
// counter (a real close from `s2` afterwards would either double-release or
// never trigger the underlying extern-close). This path had NO coverage before
// this fixture. The fix (tcp.nv `@into_split()`) makes the check-and-mutate ONE
// atomic `compare_exchange(1, 0)`: on failure the CAS is a no-op by construction
// (hardware/runtime CAS contract, exercised generically by
// std/src/runtime/sync_test.nv — not re-proven here), so `s2.rc` is unaffected
// by the panicking attempt; this fixture's job is only to pin that the
// precondition violation actually panics (previously unverified).
test "into_split panics on a live share()-copy (rc != 1, #498)" panics "sole ownership" {
    with Net = mock_net() {
        mut s = must(TcpStream.connect(SocketAddr.loopback(0)))
        mut s2 = s.share()
        assert(s2.rc.load() == 2)
        ro _pair = s.into_split()   // must panic: rc == 2, not 1 — s2 is a live copy
    }
}