diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index a36578d..d78f7e3 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -893,73 +893,141 @@ fn receive( mod udp_tests { use super::*; + const REBIND_ATTEMPTS: usize = 16; + + // Closing an exclusive socket releases its ephemeral port before the next + // bind. Another thread/process can claim it in that gap. Retry the entire + // fixture on EADDRINUSE only; successful iterations must run every assertion. + fn retry_udp_rebind(mut iteration: impl FnMut() -> Result<()>) -> Result<()> { + for attempt in 1..=REBIND_ATTEMPTS { + match iteration() { + Ok(()) => return Ok(()), + Err(error) => { + eprintln!("UDP rebind attempt {attempt}/{REBIND_ATTEMPTS}: {error:?}"); + if error.os != Some(libc::EADDRINUSE) || attempt == REBIND_ATTEMPTS { + return Err(error); + } + } + } + } + unreachable!("the final attempt always returns") + } + + #[test] + fn udp_rebind_retry_limit_and_errno_are_strict() { + let busy = Error::from(std::io::Error::from_raw_os_error(libc::EADDRINUSE)); + let mut attempts = 0; + retry_udp_rebind(|| { + attempts += 1; + if attempts < REBIND_ATTEMPTS { + Err(busy) + } else { + Ok(()) + } + }) + .expect("the final permitted attempt can succeed"); + assert_eq!(attempts, REBIND_ATTEMPTS); + + attempts = 0; + let error = retry_udp_rebind(|| { + attempts += 1; + Err(busy) + }) + .expect_err("exhaustion must fail, never skip the subject"); + assert_eq!(attempts, REBIND_ATTEMPTS); + assert_eq!(error.os, Some(libc::EADDRINUSE)); + + // Even the same portable kind cannot authorize a retry without the + // exact OS code. Other failures must not get hidden by a later success. + for fatal in [ + Error::from(std::io::Error::from_raw_os_error(libc::EADDRNOTAVAIL)), + Error::new(busy.kind), + ] { + attempts = 0; + let error = retry_udp_rebind(|| { + attempts += 1; + Err(fatal) + }) + .expect_err("only EADDRINUSE can retry"); + assert_eq!(attempts, 1); + assert_eq!(error.os, fatal.os); + assert_eq!(error.kind, fatal.kind); + } + } + #[test] fn default_udp_bind_does_not_enable_address_sharing() { let mut checked = 0; for addr in ["127.0.0.1:0", "[::1]:0"] { - let mut backend = - Unix::new(&Config::default(), BufferPool::new(2, 64)).expect("backend"); - let first = Handle { - owner: 1, - key: 1 << 32, - }; - let second = Handle { - owner: 1, - key: (1 << 32) | 1, - }; - backend - .open( - first, - Open::Udp { - addr: addr.parse().expect("address"), - opts: UdpOpts::default(), - }, - ) - .expect("first bind"); - let fd = backend - .get(first) - .expect("resource") - .transport - .fd - .as_raw_fd(); - let mut reuse = -1i32; - let mut len = std::mem::size_of_val(&reuse) as libc::socklen_t; - assert_eq!( - // SAFETY: live owned socket and correctly sized integer output. - unsafe { - libc::getsockopt( - fd, - libc::SOL_SOCKET, - libc::SO_REUSEADDR, - (&mut reuse as *mut i32).cast(), - &mut len, + retry_udp_rebind(|| { + let mut backend = + Unix::new(&Config::default(), BufferPool::new(2, 64)).expect("backend"); + let first = Handle { + owner: 1, + key: 1 << 32, + }; + let second = Handle { + owner: 1, + key: (1 << 32) | 1, + }; + backend + .open( + first, + Open::Udp { + addr: addr.parse().expect("address"), + opts: UdpOpts::default(), + }, ) - }, - 0 - ); - assert_eq!(reuse, 0, "default UDP must reserve its endpoint: {addr}"); - let addr = backend.local_addr(first).expect("bound address"); - let error = backend - .open( - second, - Open::Udp { - addr, - opts: UdpOpts::default(), - }, - ) - .expect_err("a live default endpoint cannot be shared"); - assert_eq!(error.os, Some(libc::EADDRINUSE)); - backend.release(first); - backend - .open( - second, - Open::Udp { - addr, - opts: UdpOpts::default(), + .expect("first bind"); + let fd = backend + .get(first) + .expect("resource") + .transport + .fd + .as_raw_fd(); + let mut reuse = -1i32; + let mut len = std::mem::size_of_val(&reuse) as libc::socklen_t; + assert_eq!( + // SAFETY: live owned socket and correctly sized integer output. + unsafe { + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_REUSEADDR, + (&mut reuse as *mut i32).cast(), + &mut len, + ) }, - ) - .expect("released endpoint can be rebound"); - assert_eq!(backend.local_addr(second).expect("rebound address"), addr); + 0 + ); + assert_eq!(reuse, 0, "default UDP must reserve its endpoint: {addr}"); + let addr = backend.local_addr(first).expect("bound address"); + let error = backend + .open( + second, + Open::Udp { + addr, + opts: UdpOpts::default(), + }, + ) + .expect_err("a live default endpoint cannot be shared"); + assert_eq!(error.os, Some(libc::EADDRINUSE)); + backend.release(first); + backend + .open( + second, + Open::Udp { + addr, + opts: UdpOpts::default(), + }, + ) + .inspect_err(|error| { + eprintln!("rebind released endpoint {addr}: {error:?}"); + })?; + assert_eq!(backend.local_addr(second).expect("rebound address"), addr); + Ok(()) + }) + .expect("UDP rebind must succeed within 16 complete attempts"); checked += 1; } assert_eq!(checked, 2, "IPv4 and IPv6 binding policies ran"); @@ -993,149 +1061,157 @@ mod udp_tests { fn cancelled_udp_with_cached_events_survives_exact_fd_and_port_reuse() { let mut received = 0; for addr in ["127.0.0.1:0", "[::1]:0"] { - let mut backend = - Unix::new(&Config::default(), BufferPool::new(2, 64)).expect("backend"); - let old = Handle { - owner: 1, - key: 1 << 32, - }; - let next = Handle { - owner: 1, - key: 2 << 32, - }; - let old_op = OpId { - owner: 1, - key: 1 << 32, - }; - let next_op = OpId { - owner: 1, - key: 2 << 32, - }; - backend - .open( - old, - Open::Udp { - addr: addr.parse().expect("address"), - opts: UdpOpts::default(), - }, - ) - .expect("old socket"); - let endpoint = backend.local_addr(old).expect("old endpoint"); - let peer = std::net::UdpSocket::bind(addr).expect("peer"); - let from = peer.local_addr().expect("peer address"); - let mut events = Vec::with_capacity(1); - backend - .submit(Request { - op: old_op, - handle: old, - operation: Operation::RecvFrom(ReadBuf::Pooled), - }) - .expect("old receive"); - backend - .poll(Some(Duration::ZERO), &mut events) - .expect("arm old receive"); - assert!(events.is_empty()); - assert_eq!( - peer.send_to(&[], endpoint).expect("old zero-byte packet"), - 0 - ); - // Collect a real old-generation readiness event without consuming - // its datagram; the next poll must discard this cached event batch. - backend.polled.clear(); - backend - .poller - .wait(Some(Duration::from_secs(1)), &mut backend.polled) - .expect("old readiness"); - assert!(backend.polled.iter().any(|e| e.key == old.key() && e.read)); - backend.cancel(old_op).expect("cancel old receive"); - backend - .poll(Some(Duration::ZERO), &mut events) - .expect("acknowledgement"); - assert_eq!(events.len(), 1); - assert_eq!(events[0].op, old_op); - assert!(events[0].terminal); - assert!(matches!(events[0].result, Ok(Outcome::Cancelled))); - events.clear(); - let transport = backend.detach(old).expect("detach quiescent socket"); - let fd = transport.fd.as_raw_fd(); - let replacement = socket::create(endpoint, true).expect("replacement socket"); - assert_ne!(replacement.as_raw_fd(), fd); - // Keep the destination descriptor owned throughout: dup2 atomically - // closes the old socket (discarding its queued datagram) and reuses - // exactly its fd, without racing other tests for an unowned fd slot. - // SAFETY: distinct live owned fds; transport retains ownership of the - // destination and replacement owns the source reference. - assert_eq!(unsafe { libc::dup2(replacement.as_raw_fd(), fd) }, fd); - drop(replacement); - socket::configure(fd).expect("restore descriptor flags after dup2"); - let address = Addr::new(endpoint); - // SAFETY: fd now owns a fresh UDP socket; address has initialized storage. - assert_eq!(unsafe { libc::bind(fd, address.ptr(), address.len) }, 0); - backend - .attach(next, transport) - .expect("attach next generation"); - assert_eq!( + retry_udp_rebind(|| { + let mut backend = + Unix::new(&Config::default(), BufferPool::new(2, 64)).expect("backend"); + let old = Handle { + owner: 1, + key: 1 << 32, + }; + let next = Handle { + owner: 1, + key: 2 << 32, + }; + let old_op = OpId { + owner: 1, + key: 1 << 32, + }; + let next_op = OpId { + owner: 1, + key: 2 << 32, + }; backend - .get(next) - .expect("new resource") - .transport - .fd - .as_raw_fd(), - fd - ); - assert_eq!(backend.local_addr(next).expect("reused port"), endpoint); - assert_eq!(old.index(), next.index()); - assert_ne!(old.key(), next.key()); - backend - .submit(Request { - op: next_op, - handle: next, - operation: Operation::RecvFrom(ReadBuf::Pooled), - }) - .expect("next receive"); - assert_eq!(old_op.index(), next_op.index()); - assert_eq!( - backend.cancel(old_op).expect_err("stale cancellation").kind, - ErrorKind::NotFound - ); - backend.release(old); // A stale handle must not release the new fd. - backend - .poll(Some(Duration::ZERO), &mut events) - .expect("arm replacement receive"); - assert!(events.is_empty(), "old event batch crossed generations"); - // A stale event is not a completion, and must not cause a no-spin - // violation after the new socket reports EAGAIN. - let at = Instant::now() + Duration::from_millis(2); - let info = backend - .poll(Some(Duration::from_millis(2)), &mut events) - .expect("new empty socket waits"); - assert!(events.is_empty(), "old datagram crossed socket lifetime"); - assert_eq!(info.waits, 1); - assert!(Instant::now() >= at); - assert_eq!(peer.send_to(b"new", endpoint).expect("new packet"), 3); - backend - .poll(Some(Duration::from_secs(1)), &mut events) - .expect("new delivery"); - assert_eq!(events.len(), 1); - let event = events.pop().expect("one completion"); - assert_eq!(event.op, next_op); - assert!(event.terminal); - match event.result { - Ok(Outcome::RecvFrom { - n, - from: actual, - lease: Some(lease), - }) => { - assert_eq!(actual, from); - assert_eq!(n, 3); - assert_eq!(lease.as_slice(), b"new"); + .open( + old, + Open::Udp { + addr: addr.parse().expect("address"), + opts: UdpOpts::default(), + }, + ) + .expect("old socket"); + let endpoint = backend.local_addr(old).expect("old endpoint"); + let peer = std::net::UdpSocket::bind(addr).expect("peer"); + let from = peer.local_addr().expect("peer address"); + let mut events = Vec::with_capacity(1); + backend + .submit(Request { + op: old_op, + handle: old, + operation: Operation::RecvFrom(ReadBuf::Pooled), + }) + .expect("old receive"); + backend + .poll(Some(Duration::ZERO), &mut events) + .expect("arm old receive"); + assert!(events.is_empty()); + assert_eq!( + peer.send_to(&[], endpoint).expect("old zero-byte packet"), + 0 + ); + // Collect a real old-generation readiness event without consuming + // its datagram; the next poll must discard this cached event batch. + backend.polled.clear(); + backend + .poller + .wait(Some(Duration::from_secs(1)), &mut backend.polled) + .expect("old readiness"); + assert!(backend.polled.iter().any(|e| e.key == old.key() && e.read)); + backend.cancel(old_op).expect("cancel old receive"); + backend + .poll(Some(Duration::ZERO), &mut events) + .expect("acknowledgement"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].op, old_op); + assert!(events[0].terminal); + assert!(matches!(events[0].result, Ok(Outcome::Cancelled))); + events.clear(); + let transport = backend.detach(old).expect("detach quiescent socket"); + let fd = transport.fd.as_raw_fd(); + let replacement = socket::create(endpoint, true).expect("replacement socket"); + assert_ne!(replacement.as_raw_fd(), fd); + // Keep the destination descriptor owned throughout: dup2 atomically + // closes the old socket (discarding its queued datagram) and reuses + // exactly its fd, without racing other tests for an unowned fd slot. + // SAFETY: distinct live owned fds; transport retains ownership of the + // destination and replacement owns the source reference. + assert_eq!(unsafe { libc::dup2(replacement.as_raw_fd(), fd) }, fd); + drop(replacement); + socket::configure(fd).expect("restore descriptor flags after dup2"); + let address = Addr::new(endpoint); + // SAFETY: fd now owns a fresh UDP socket; address has initialized storage. + if unsafe { libc::bind(fd, address.ptr(), address.len) } < 0 { + let error = last_error(); + eprintln!("bind reused fd {fd} at {endpoint}: {error:?}"); + return Err(error); } - other => panic!("unexpected reused-socket result: {other:?}"), - } - backend - .poll(Some(Duration::ZERO), &mut events) - .expect("no duplicates"); - assert!(events.is_empty()); + backend + .attach(next, transport) + .expect("attach next generation"); + assert_eq!( + backend + .get(next) + .expect("new resource") + .transport + .fd + .as_raw_fd(), + fd + ); + assert_eq!(backend.local_addr(next).expect("reused port"), endpoint); + assert_eq!(old.index(), next.index()); + assert_ne!(old.key(), next.key()); + backend + .submit(Request { + op: next_op, + handle: next, + operation: Operation::RecvFrom(ReadBuf::Pooled), + }) + .expect("next receive"); + assert_eq!(old_op.index(), next_op.index()); + assert_eq!( + backend.cancel(old_op).expect_err("stale cancellation").kind, + ErrorKind::NotFound + ); + backend.release(old); // A stale handle must not release the new fd. + backend + .poll(Some(Duration::ZERO), &mut events) + .expect("arm replacement receive"); + assert!(events.is_empty(), "old event batch crossed generations"); + // A stale event is not a completion, and must not cause a no-spin + // violation after the new socket reports EAGAIN. + let at = Instant::now() + Duration::from_millis(2); + let info = backend + .poll(Some(Duration::from_millis(2)), &mut events) + .expect("new empty socket waits"); + assert!(events.is_empty(), "old datagram crossed socket lifetime"); + assert_eq!(info.waits, 1); + assert!(Instant::now() >= at); + assert_eq!(peer.send_to(b"new", endpoint).expect("new packet"), 3); + backend + .poll(Some(Duration::from_secs(1)), &mut events) + .expect("new delivery"); + assert_eq!(events.len(), 1); + let event = events.pop().expect("one completion"); + assert_eq!(event.op, next_op); + assert!(event.terminal); + match event.result { + Ok(Outcome::RecvFrom { + n, + from: actual, + lease: Some(lease), + }) => { + assert_eq!(actual, from); + assert_eq!(n, 3); + assert_eq!(lease.as_slice(), b"new"); + } + other => panic!("unexpected reused-socket result: {other:?}"), + } + backend + .poll(Some(Duration::ZERO), &mut events) + .expect("no duplicates"); + assert!(events.is_empty()); + Ok(()) + }) + .expect("UDP rebind must succeed within 16 complete attempts"); received += 1; } assert_eq!( diff --git a/docs/lanes/udp-fix.md b/docs/lanes/udp-fix.md new file mode 100644 index 0000000..08dc5ea --- /dev/null +++ b/docs/lanes/udp-fix.md @@ -0,0 +1,108 @@ +# udp-fix + +Base: `c506775`. Implementation and verification complete. Default-parallel +workspace has one unrelated process/signal failure; documented serial execution passes. +Read DESIGN.md, CONTRIBUTING.md, integration and relevant core lane reports. +No applicable AGENTS.md. The integrator owns commits (`.git` is read-only). + +## Implemented / cause + +- Confirmed the proposed release/rebind race in a controlled run: four threads + repeatedly bound exclusive IPv4/IPv6 ephemeral UDP ports; a temporary 10 ms + scheduling gap after `dup2` exposed `EADDRINUSE`, macOS errno **48**, on iteration + 18 at `127.0.0.1:56207`, fd 3. The diagnostic-only baseline without the gap + passed 2,000 runs. This proves the mechanism, not the unrecorded original errno. +- Both the exact fd/port reuse test and the default-bind policy test now retry + the **whole address-family iteration**, at most 16 attempts, only after the + released-port bind returns exactly `EADDRINUSE`. Every other error and exhausted + retries fail with errno. Successful IPv4 and IPv6 iterations must still execute + every original assertion, including cached readiness, cancelled operations, + discarded old datagrams, actual waiting, new bytes/source and no duplicates. +- The module's one raw `libc::bind` captures errno immediately and reports fd and + endpoint before propagating the error. Policy tests prove last-attempt success, + exact exhaustion and immediate rejection of other/missing OS codes. +- `scripts/stress-udp-reuse.py` runs positive-count UDP suites with four active + ephemeral-port churn threads, bounded retained sockets and per-run logs. + The same 10 ms controlled gap with the fix passed **2,000/2,000** suites and + recovered **78 actual bind conflicts** (50 IPv4, 28 IPv6), all errno 48. + Final-source stress without the gap also passed **2,000/2,000**, recovering + two real errno-48 conflicts. No successful family iteration was skipped. + +## Verification + +Logs and saved reproducer binaries/sources: `.tools/udp-fix/`. Controlled builds +inserted `std::thread::sleep(Duration::from_millis(10));` immediately after +`drop(replacement)`, saved the test binary, then restored the source. No pause is +in the final source. The stress command uses 2,000 iterations by default. +An initial final-stress run reused the saved-gap build because restoring source +also restored its older mtime. Binary SHA-256 comparison caught this. The source +mtime was refreshed, Cargo visibly recompiled, the new binary hash differs, and +the 20 core runs and final stress were rerun. That cached run is not final-source +verification; its logs remain in `cached-delayed-stress/`. + +| Command | Result | +| --- | --- | +| `cargo test --locked -p turnloop --lib --no-run --message-format=json` (diagnostics-only, old/fixed 10 ms gap builds) | PASS | +| `python3 scripts/stress-udp-reuse.py --binary .tools/udp-fix/baseline-tests --logs .tools/udp-fix/baseline-stress` | PASS, 2,000 suites, 1,919,705 competing binds | +| `python3 scripts/stress-udp-reuse.py --binary .tools/udp-fix/delayed-tests --logs .tools/udp-fix/delayed-stress` | Expected FAIL at iteration 18, errno 48; 85,513 competing binds | +| `python3 scripts/stress-udp-reuse.py --binary .tools/udp-fix/fixed-delayed-tests --logs .tools/udp-fix/fixed-delayed-stress --require-contention` | PASS, 2,000 complete suites, 78 real conflicts recovered, 5,354,966 competing binds | +| `python3 scripts/stress-udp-reuse.py --binary /usr/bin/true --iterations 1 --logs .tools/udp-fix/empty-control` | Expected FAIL: exit-zero process with no executed tests rejected | +| `cargo fmt --all`; `cargo fmt --check`; `git diff --check` | PASS | +| `cargo test --locked -p turnloop backend::unix::udp_tests:: -- --nocapture` | PASS, all four tests actually ran | +| `cargo clippy --workspace --all-targets -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS native | +| `cargo clippy --locked --workspace --all-targets --all-features -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS native | +| `cargo clippy --locked -p turnloop --all-targets --target x86_64-unknown-linux-gnu -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS | +| `cargo clippy --locked -p turnloop --all-targets --all-features --target x86_64-unknown-linux-gnu -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS | +| `cargo clippy --locked -p turnloop --all-targets --all-features --target aarch64-unknown-linux-gnu -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS | +| `cargo clippy --locked -p turnloop --all-targets --all-features --target x86_64-pc-windows-msvc -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS | +| `cargo clippy --locked -p turnloop --all-targets --all-features --target wasm32-wasip2 -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS | +| `cargo +nightly-2026-09-07 clippy --locked -p turnloop --all-targets --all-features --target wasm32-wasip3 -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS, repository's p3 toolchain | +| `cargo clippy --locked -p turnloop --all-targets --all-features --target wasm32-unknown-unknown -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS | +| `cargo +stable check --locked --workspace --all-targets --all-features` | PASS | +| `cargo test -p turnloop` ×20 via `python3 .tools/udp-fix/repeat-core.py` | PASS, 12 tests each; original exact-reuse subject required in every run; default parallel threads, `RUST_TEST_THREADS` unset | +| `bash scripts/ci/no-tokio.sh` | PASS, eight targets plus union, default/all features | +| `python3 scripts/ci/soak.py` | PASS, 251 versions; inherited rustls security exception unchanged | +| `python3 -m py_compile scripts/stress-udp-reuse.py` | PASS | +| `cargo test --locked -p turnloop --lib --no-run` after restoring source mtime | PASS, actual final recompilation and distinct binary hash | +| `cargo test --workspace` | FAIL, unrelated `registered_processes_and_signals_do_not_spin` at `native_surface.rs:440` (`zero <= 1`); all 12 core tests and 10 core allocation tests passed | +| `cargo test --workspace -- --test-threads=1` | PASS, 237 tests including the failing parallel subject and allocation gates; 12 ignored service tests are UNRUN | +| `python3 scripts/stress-udp-reuse.py --logs .tools/udp-fix/final-stress` after fresh rebuild | PASS, 2,000 complete suites, two real conflicts recovered, 1,868,347 competing binds | + +The workspace failure is in an unchanged contract binary; changed Rust is entirely +inside `#[cfg(all(test, not(loom)))] mod udp_tests`. Parallel signal stress can +interrupt another test's kevent wait; kqueue counts EINTR as an empty wait. +This is consistent with the failure, not an errno-instrumented diagnosis. +CONTRIBUTING.md:45–46 already requires serial contracts for signal/process and +allocator isolation. The serial run is supplemental; it does not erase the +requested default-parallel failure. The no-spin test/gate remains unchanged. + +## Same-pattern audit / deviations + +- Fixed `default_udp_bind_does_not_enable_address_sharing`, the other UDP + close-then-exact-rebind test. The explicit-sharing test keeps both sockets live. +- `turnloop-contract::refused_connect_once` releases a TCP listener before + connecting, so an unrelated listener could steal its port. This does not rebind + the port; retries would hide an incorrect successful connection. Left unchanged + for the existing portable refusal-fixture follow-up, including WASI. +- TCP fixture gaps, listed without changing their lifecycle policies: + `scripts/test-servers.py:145` (`sql_port`, used by PostgreSQL/MySQL/SMTP), `:317` + (`redis_port`, explicit candidate and bus-port probes), `:525` (MongoDB socket + reservations), and `scripts/ci/browser_driver.py:55` (WebDriver). They release a + reservation before a child binds. A fix needs verified child bind errors and + bounded restart/cleanup, rather than retrying arbitrary startup failures. +- Related tests in `scripts/ci/test_servers.py:220`, `:238`, `:287`, `:302` assume + an unused/just-closed TCP port remains closed while probing refusal/cleanup. + They do not rebind; retrying would obscure their cleanup/refusal subjects. + Other audited Rust TCP/UDP tests, native/web HTTP fixtures and the proxy keep + their listeners alive or let the serving process bind port zero directly. +- No production changes, dependencies, sharing options, wait paths, allocation + thresholds, CI/soak policy or DESIGN.md edits. No new runtime operation requires + extending an allocation gate; existing allocation tests remain mandatory. + +## Open questions / next steps + +- No UDP implementation questions remain. Keep the unrelated default-parallel + process/signal no-spin failure visible; investigate its isolation separately. + No-spin assertions and the required CI commands were not modified. +- Linux/Windows runtime: UNRUN (no hosts); cross-checks are compilation only. + Integrator should run native tests on those hosts and commit the working tree. diff --git a/scripts/stress-udp-reuse.py b/scripts/stress-udp-reuse.py new file mode 100644 index 0000000..114d65b --- /dev/null +++ b/scripts/stress-udp-reuse.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""Run the Unix UDP regressions while threads churn exclusive ephemeral ports.""" + +import argparse +from collections import deque +from concurrent.futures import ThreadPoolExecutor +import json +from pathlib import Path +import re +import socket +import subprocess +import threading + + +def churn(family, host, stop, ready): + held = deque() + count = 0 + try: + while not stop.is_set(): + sock = socket.socket(family, socket.SOCK_DGRAM) + try: + sock.bind((host, 0)) + except BaseException: + sock.close() + raise + held.append(sock) + count += 1 + if len(held) > 128: + held.popleft().close() + ready.set() + finally: + for sock in held: + sock.close() + return count + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--iterations", type=int, default=2000) + parser.add_argument("--binary", type=Path, help="Use a previously built test binary") + parser.add_argument("--logs", type=Path, default=Path(".tools/udp-fix/stress")) + parser.add_argument("--require-contention", action="store_true", + help="Fail unless a real released-port bind failed during a passing suite") + args = parser.parse_args() + if args.iterations < 1: + parser.error("--iterations must be positive") + args.logs.mkdir(parents=True, exist_ok=True) + binary = args.binary + if binary is None: + command = ["cargo", "test", "--locked", "-p", "turnloop", "--lib", + "--no-run", "--message-format=json"] + print("Building:", " ".join(command), flush=True) + build = subprocess.run(command, text=True, stdout=subprocess.PIPE, check=True) + artifacts = [json.loads(line) for line in build.stdout.splitlines()] + binaries = [a["executable"] for a in artifacts + if a.get("reason") == "compiler-artifact" + and a["target"]["name"] == "turnloop" and a.get("executable")] + if len(binaries) != 1: + raise RuntimeError(f"expected one core test executable: {binaries}") + binary = Path(binaries[0]) + command = [str(binary.resolve()), "backend::unix::udp_tests::", "--nocapture"] + stop = threading.Event() + workers = [(socket.AF_INET, "127.0.0.1"), (socket.AF_INET6, "::1")] * 2 + status = 0 + counts = [] + recovered = 0 + with ThreadPoolExecutor(max_workers=len(workers)) as pool: + ready = [threading.Event() for _ in workers] + futures = [pool.submit(churn, family, host, stop, signal) + for (family, host), signal in zip(workers, ready)] + try: + for signal in ready: + if not signal.wait(10): + raise RuntimeError("ephemeral-port worker did not bind") + for iteration in range(1, args.iterations + 1): + for future in futures: + if future.done(): + future.result() + raise RuntimeError("port churn stopped before the tests") + run = subprocess.run(command, text=True, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, timeout=30) + (args.logs / f"{iteration:05}.log").write_text(run.stdout) + passed = re.search(r"test result: ok\. ([1-9]\d*) passed; 0 failed; 0 ignored;", + run.stdout) + subjects = ["default_udp_bind_does_not_enable_address_sharing", + "udp_port_sharing_requires_explicit_opt_in", + "cancelled_udp_with_cached_events_survives_exact_fd_and_port_reuse"] + ran = all(f"test backend::unix::udp_tests::{name} ... ok" in run.stdout + for name in subjects) + if run.returncode or not passed or not ran: + print(run.stdout, flush=True) + print(f"FAIL iteration {iteration}: exit={run.returncode}", flush=True) + status = 1 + break + recovered += len(re.findall(r"(?:bind reused fd|rebind released endpoint) .*Error \{", + run.stdout)) + if iteration % 100 == 0 or iteration == args.iterations: + print(f"PASS {iteration}/{args.iterations} UDP suites", flush=True) + finally: + stop.set() + counts = [future.result() for future in futures] + if not all(count > 0 for count in counts): + raise RuntimeError(f"churn did not execute on every thread: {counts}") + print(f"Exclusive ephemeral binds per worker (IPv4, IPv6, IPv4, IPv6): {counts}", flush=True) + print(f"Real rebind failures recovered in passing suites: {recovered}", flush=True) + if args.require_contention and not recovered: + print("FAIL: no actual released-port contention observed", flush=True) + status = 1 + return status + + +if __name__ == "__main__": + raise SystemExit(main())