char/avril

git clone https://git.t4t.associates/char/avril

Charlotte Somtests: remove flaky eventsource testa7f08a12026-10-20

mod common;

use std::io::{self, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

struct Request {
    path: String,
    headers: String,
}

fn request(stream: &mut TcpStream) -> io::Result<Request> {
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let mut buffer = Vec::new();
    let mut byte = [0; 1];
    loop {
        match stream.read(&mut byte) {
            Ok(0) | Err(_) => break,
            Ok(_) => {
                buffer.push(byte[0]);
                if buffer.ends_with(b"\r\n\r\n") {
                    break;
                }
            }
        }
    }
    let request = String::from_utf8_lossy(&buffer);
    // Header names are matched case-insensitively; values are not.
    let lines = request
        .lines()
        .map(|line| match line.split_once(':') {
            Some((name, value)) => format!("{}:{value}", name.to_ascii_lowercase()),
            None => line.to_owned(),
        })
        .collect::<Vec<_>>();
    let first = lines.first().map(String::as_str).unwrap_or_default();
    let mut words = first.split_whitespace();
    match words.next() {
        None => return Err(io::ErrorKind::UnexpectedEof.into()),
        method => assert_eq!(method, Some("GET")),
    }
    Ok(Request {
        path: words.next().unwrap_or_default().to_owned(),
        headers: lines[1..].join("\n"),
    })
}

fn respond(stream: &mut TcpStream, status: &str, headers: &str, body: &[u8]) -> io::Result<()> {
    let mut response = format!("HTTP/1.1 {status}\r\n{headers}\r\n").into_bytes();
    response.extend_from_slice(body);
    stream.write_all(&response)
}

/// Accept `count` connections; each handler receives the request and may
/// stream the body in stages before returning, which ends the connection.
fn server(
    count: usize,
    handler: impl Fn(Request, &mut TcpStream) -> io::Result<()> + Send + 'static,
) -> (String, JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let address = listener.local_addr().unwrap().to_string();
    listener.set_nonblocking(true).unwrap();
    let thread = std::thread::spawn(move || {
        let deadline = Instant::now() + Duration::from_secs(5);
        let mut remaining = count;
        while remaining != 0 {
            let (mut socket, _) = loop {
                match listener.accept() {
                    Ok(connection) => break connection,
                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                        assert!(Instant::now() < deadline, "Missing EventSource request");
                        std::thread::sleep(Duration::from_millis(5));
                    }
                    Err(error) => panic!("{error}"),
                }
            };
            let Ok(request) = request(&mut socket) else {
                continue;
            };
            remaining -= 1;
            if handler(request, &mut socket).is_err() {
                continue;
            }
        }
    });
    (format!("http://{address}"), thread)
}

fn event_stream_headers() -> &'static str {
    "Content-Type: text/event-stream\r\nConnection: close\r\n"
}

/// Stream `chunks` with pauses so chunk boundaries land where the test
/// intends; coalesced reads parse identically either way.
fn stream_chunks(stream: &mut TcpStream, chunks: &[&[u8]]) -> io::Result<()> {
    respond(stream, "200 OK", event_stream_headers(), b"")?;
    for chunk in chunks {
        stream.write_all(chunk)?;
        stream.flush()?;
        std::thread::sleep(Duration::from_millis(30));
    }
    Ok(())
}

fn header(headers: &str, name: &str) -> Option<String> {
    headers.lines().find_map(|line| {
        let (field, value) = line.split_once(':')?;
        (field.trim() == name).then(|| value.trim().to_owned())
    })
}

#[test]
fn user_agent_defaults_follow_navigator_on_connect_and_reconnect() {
    let (url, thread) = server(3, |request, stream| {
        let ua = header(&request.headers, "user-agent").expect("Missing User-Agent");
        respond(
            stream,
            "200 OK",
            event_stream_headers(),
            format!("retry: 0\ndata: {ua}\n\n").as_bytes(),
        )
    });
    let source = r#"
        const original = navigator.userAgent;
        await new Promise((resolve, reject) => {
            const source = new EventSource('URL/default');
            source.onmessage = event => {
                source.close();
                if (event.data !== original) reject(Error('Initial UA'));
                else resolve();
            };
            source.onerror = () => {source.close(); reject(Error('Initial connection failed'));};
        });
        navigator.userAgent = 'custom/1.0';
        await new Promise((resolve, reject) => {
            const source = new EventSource('URL/custom');
            let count = 0;
            source.onmessage = event => {
                const expected = count++ === 0 ? 'custom/1.0' : 'reconnected/2.0';
                if (event.data !== expected) {
                    source.close();
                    reject(Error('Expected UA ' + expected + ', got ' + event.data));
                } else if (count === 1) {
                    navigator.userAgent = 'reconnected/2.0';
                } else {
                    source.close();
                    resolve();
                }
            };
        });
        console.log('EventSource UA');
    "#
    .replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"EventSource UA\n");
    thread.join().unwrap();
}

#[test]
fn surface_brands_coercion_and_clone_rejection() {
    let out = common::run(
        r#"
        const original = 'HTTP://Example.com/plain%20path?query=1#frag';
        const source = new EventSource(original);
        if (source.url !== 'http://example.com/plain%20path?query=1#frag') throw Error('URL serialization');
        if (source.readyState !== 0 || source.withCredentials !== false) throw Error('Initial attributes');
        const credentialed = new EventSource('http://127.0.0.1:1/cred', {withCredentials: true});
        if (credentialed.withCredentials !== true) throw Error('withCredentials option');
        const falsy = new EventSource('http://127.0.0.1:1/falsy', {withCredentials: 0});
        if (falsy.withCredentials !== false) throw Error('Falsy withCredentials');
        const nulled = new EventSource('http://127.0.0.1:1/n', null);
        if (nulled.withCredentials !== false || nulled.readyState !== 0) throw Error('Null options');
        for (const constant of ['CONNECTING', 'OPEN', 'CLOSED']) {
            const descriptor = Object.getOwnPropertyDescriptor(EventSource.prototype, constant);
            if (!descriptor.enumerable || descriptor.writable || descriptor.get) throw Error('Prototype constant ' + constant);
            const staticDescriptor = Object.getOwnPropertyDescriptor(EventSource, constant);
            if (!staticDescriptor.enumerable || staticDescriptor.writable) throw Error('Static constant ' + constant);
            if (source[constant] !== staticDescriptor.value) throw Error('Constant value ' + constant);
        }
        if (Object.prototype.toString.call(source) !== '[object EventSource]') throw Error('toStringTag');
        class Sub extends EventSource {}
        const sub = new Sub('http://127.0.0.1:1/sub');
        if (!(sub instanceof Sub) || !(sub instanceof EventSource) || !(sub instanceof EventTarget)) throw Error('Subclass');
        if (sub.readyState !== 0 || sub.withCredentials !== false) throw Error('Subclass attributes');
        if (source.onopen !== null || source.onmessage !== null || source.onerror !== null) throw Error('Default handlers');
        const illegal = (operation) => {
            try {
                operation();
                throw Error('Accepted illegal invocation');
            } catch (error) {
                if (!(error instanceof TypeError)) throw error;
            }
        };
        illegal(() => Object.getOwnPropertyDescriptor(EventSource.prototype, 'readyState').get.call({}));
        illegal(() => EventSource.prototype.close.call({}));
        illegal(() => Object.getOwnPropertyDescriptor(EventSource.prototype, 'url').get.call(new EventTarget()));
        illegal(() => Object.getOwnPropertyDescriptor(EventSource.prototype, 'withCredentials').get.call({}));
        for (const bad of [() => new EventSource(), () => new EventSource(Symbol()), () => new EventSource('http://127.0.0.1:1/x', Symbol())]) {
            try {
                bad();
                throw Error('Accepted invalid constructor argument');
            } catch (error) {
                if (!(error instanceof TypeError)) throw error;
            }
        }
        const syntax = (operation) => {
            try {
                operation();
                throw Error('Accepted invalid URL');
            } catch (error) {
                if (!(error instanceof DOMException) || error.name !== 'SyntaxError') throw error;
            }
        };
        syntax(() => new EventSource('http://['));
        const relative = new EventSource('./events');
        if (relative.url !== new URL('./events', import.meta.url).href) throw Error('Relative URL');
        relative.close();
        const marker = {};
        try {
            new EventSource({toString() { throw marker; }});
            throw Error('Swallowed coercion failure');
        } catch (error) {
            if (error !== marker) throw error;
        }
        for (const clone of [
            () => structuredClone(source),
            () => new MessageChannel().port1.postMessage(source),
        ]) {
            try {
                clone();
                throw Error('Accepted a clone');
            } catch (error) {
                if (!(error instanceof DOMException) || error.name !== 'DataCloneError') throw error;
            }
        }
        credentialed.close();
        falsy.close();
        nulled.close();
        sub.close();
        source.close();
        if (source.readyState !== 2) throw Error('Closed ready state');
        source.close();
        if (source.readyState !== 2) throw Error('Second close');
        console.log('surface ok');
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"surface ok\n");
}

#[test]
fn streaming_line_endings_chunk_boundaries_and_fields() {
    let (url, thread) = server(1, |_, stream| {
        stream_chunks(
            stream,
            &[
                // CR terminated comment split from its line terminator.
                b": comment\r",
                b"\ndata: first\n",
                b"data: second\r\n",
                b"data:third\n",
                b"data:  spaced\n",
                b"event: custom\n",
                b"\n",
                // An incomplete event that the stream end must discard.
                b"data: dropped",
            ],
        )
    });
    let out = common::run(
        &format!(
            r#"
        const log = [];
        const es = new EventSource('{url}');
        es.addEventListener('custom', (event) => {{
            log.push(`custom:${{JSON.stringify(event.data)}}:${{event.lastEventId}}`);
        }});
        es.addEventListener('message', () => log.push('unexpected message'));
        es.addEventListener('error', () => {{
            if (es.readyState !== 0) throw Error('Reconnecting ready state');
            es.close();
            console.log(log.join(' | '));
        }});
    "#
        )
        .replace("{url}", &url),
        &[],
    );
    assert_eq!(
        out.stdout,
        "custom:\"first\\nsecond\\nthird\\n spaced\":\n".as_bytes()
    );
    thread.join().unwrap();
}

#[test]
fn byte_order_marks_utf8_splits_and_field_edge_cases() {
    let (url, thread) = server(3, |request, stream| {
        match request.path.as_str() {
            "/bom" => stream_chunks(
                stream,
                &[
                    // The BOM must be stripped exactly once even when split.
                    b"\xef",
                    b"\xbb\xbfdata: bom\n\n",
                    // A second BOM becomes U+FEFF content.
                    b"data: \xef\xbb\xbfkept\n\n",
                    // A UTF-8 sequence split across chunk boundaries.
                    b"id: h\xc3",
                    b"\xa9llo\r\ndata: caf\xc3",
                    b"\xa9\n\n",
                ],
            ),
            "/ids" => stream_chunks(
                stream,
                &[
                    b"id: a\n\ndata: one\n\n",
                    // NUL rejects the id; the buffer is unchanged.
                    b"id: b\x00\ndata: two\n\n",
                    // A blank line commits the id buffer without data.
                    b"id: c\n\n",
                    // An empty id resets it for the next commit.
                    b"id:\n\ndata: three\n\n",
                ],
            ),
            _ => stream_chunks(
                stream,
                &[
                    // Field names are case-sensitive and unknown fields are ignored.
                    b"DATA: upper\neVENT: mixed\nrandom: value\n",
                    b"event: one\nevent: two\ndata: replaced\n\n",
                    // The type buffer resets after a dispatch.
                    b"data: back to message\n\n",
                    // Empty data values still dispatch.
                    b"data:\n\n",
                    b"data: \n\n",
                    // retry only accepts plain ASCII digits.
                    b"retry: +5\nretry: 1 2\nretry: 0x10\nretry: \nretry: -1\n",
                    b"data: after retry\n\n",
                ],
            ),
        }
    });
    let out = common::run(
        &format!(
            r#"
        const log = [];
        const events = (source, types) => {{
            for (const type of types) {{
                if (type === 'error') continue;
                source.addEventListener(type, (event) => {{
                    log.push(`${{type}}:${{JSON.stringify(event.data)}}:${{JSON.stringify(event.lastEventId)}}`);
                }});
            }}
        }};
        function next(path) {{
            const source = new EventSource('{url}/' + path);
            events(source, path === 'cases' ? ['message', 'two', 'error'] : ['message', 'error']);
            source.addEventListener('error', () => {{
                source.close();
                if (path === 'bom') next('ids');
                else if (path === 'ids') next('cases');
                else console.log(log.join(' | '));
            }});
        }}
        next('bom');
    "#
        )
        .replace("{url}", &url),
        &[],
    );
    let expected = "message:\"bom\":\"\" | message:\"\u{feff}kept\":\"\" | message:\"café\":\"héllo\" \
         | message:\"one\":\"a\" | message:\"two\":\"a\" | message:\"three\":\"\" \
         | two:\"replaced\":\"\" | message:\"back to message\":\"\" | message:\"\":\"\" \
         | message:\"\":\"\" | message:\"after retry\":\"\"\n";
    assert_eq!(out.stdout, expected.as_bytes());
    thread.join().unwrap();
}

#[test]
fn eof_discards_incomplete_event_without_dispatching() {
    let (url, thread) = server(1, |_, stream| {
        respond(
            stream,
            "200 OK",
            event_stream_headers(),
            b"id: lost\ndata: incomplete\nevent: never",
        )
    });
    let out = common::run(
        &format!(
            r#"
        const log = [];
        const es = new EventSource('{url}');
        es.addEventListener('message', () => log.push('message'));
        es.addEventListener('open', () => log.push('open'));
        es.addEventListener('error', () => {{
            if (es.readyState !== 0) throw Error('Ready state after EOF');
            log.push('error');
            es.close();
            console.log(log.join(' | '));
        }});
    "#
        )
        .replace("{url}", &url),
        &[],
    );
    assert_eq!(out.stdout, b"open | error\n");
    thread.join().unwrap();
}

#[test]
fn events_run_in_tasks_and_close_suppresses_queued_messages() {
    let (url, thread) = server(1, |_, stream| {
        respond(
            stream,
            "200 OK",
            event_stream_headers(),
            b"data: one\n\ndata: two\n\ndata: three\n\n",
        )
    });
    let out = common::run(
        &format!(
            r#"
        const log = [];
        queueMicrotask(() => log.push('microtask'));
        Promise.resolve().then(() => log.push('promise'));
        let closed = false;
        const es = new EventSource('{url}');
        es.onmessage = (event) => {{
            log.push(`message:${{event.data}}`);
            if (!closed) {{
                closed = true;
                es.close();
                if (es.readyState !== 2) throw Error('Ready state inside handler');
            }}
        }};
        setTimeout(() => console.log(log.join(' | ')), 300);
    "#
        )
        .replace("{url}", &url),
        &[],
    );
    assert_eq!(out.stdout, b"microtask | promise | message:one\n");
    thread.join().unwrap();
}

#[test]
fn reconnect_honors_retry_and_sends_last_event_id() {
    let (times_sender, times) = mpsc::channel();
    let (received_sender, received) = mpsc::channel();
    let (url, thread) = {
        let times = times_sender.clone();
        let received = received_sender.clone();
        server(2, move |request, stream| {
            times.send(Instant::now()).unwrap();
            let id = header(&request.headers, "last-event-id");
            received.send(id.clone()).unwrap();
            if id.is_none() {
                stream_chunks(
                    stream,
                    &[b"retry: 100\r\nid: \xf0\x9f\x92\xa9\r\ndata: hello\r\n\r\n"],
                )
            } else {
                stream_chunks(stream, &[b"data: again\r\n\r\n"])
            }
        })
    };
    let out = common::run(
        &format!(
            r#"
        const log = [];
        const es = new EventSource('{url}');
        es.addEventListener('message', (event) => {{
            if (event.origin !== '{url}') throw Error('Origin: ' + event.origin);
            log.push(`message:${{event.data}}:${{event.lastEventId}}`);
            if (event.data === 'again') es.close();
        }});
        es.addEventListener('error', () => {{
            if (es.readyState !== 0) throw Error('Ready state while waiting to reconnect');
            log.push('error');
        }});
        es.addEventListener('open', () => {{
            log.push(`open:${{es.readyState}}`);
        }});
        setTimeout(() => console.log(log.join(' | ')), 1500);
    "#
        )
        .replace("{url}", &url),
        &[],
    );
    assert_eq!(
        out.stdout,
        "open:1 | message:hello:\u{1f4a9} | error | open:1 | message:again:\u{1f4a9}\n".as_bytes(),
    );
    thread.join().unwrap();
    assert!(
        received.recv().unwrap().is_none(),
        "the first request must not carry Last-Event-ID"
    );
    assert_eq!(
        received.recv().unwrap().unwrap(),
        "\u{1f4a9}",
        "the reconnect must send the raw UTF-8 Last-Event-ID"
    );
    let first = times.recv().unwrap();
    let second = times.recv().unwrap();
    let elapsed = second.duration_since(first);
    assert!(
        elapsed >= Duration::from_millis(100),
        "reconnected after {elapsed:?}"
    );
    assert!(
        elapsed < Duration::from_millis(2500),
        "reconnection took {elapsed:?}"
    );
}

#[test]
fn non_200_bad_mime_and_204_terminate_without_retry() {
    let (url, thread) = server(3, |request, stream| match request.path.as_str() {
        "/status" => respond(stream, "404 Not Found", "Connection: close\r\n", b""),
        "/mime" => respond(
            stream,
            "200 OK",
            "Content-Type: text/plain\r\nConnection: close\r\n",
            b"data: rejected\n\n",
        ),
        _ => respond(stream, "204 No Content", "Connection: close\r\n", b""),
    });
    let out = common::run(
        &format!(
            r#"
        const log = [];
        for (const path of ['status', 'mime', 'empty']) {{
            await new Promise((resolve) => {{
                const es = new EventSource('{url}/' + path);
                es.addEventListener('open', () => log.push(`${{path}}:open`));
                es.addEventListener('message', () => log.push(`${{path}}:message`));
                es.addEventListener('error', (event) => {{
                    if (es.readyState !== 2) throw Error('Ready state after terminal failure');
                    if (!(event instanceof Event) || event instanceof ErrorEvent) throw Error('Error event interface');
                    log.push(`${{path}}:error`);
                    resolve();
                }});
            }});
        }}
        console.log(log.join(' | '));
    "#
        )
        .replace("{url}", &url),
        &[],
    );
    assert_eq!(out.stdout, b"status:error | mime:error | empty:error\n");
    thread.join().unwrap();
}

#[test]
fn redirects_keep_the_original_url_and_use_the_final_origin() {
    let (target_url, target_thread) = server(1, |_, stream| {
        respond(
            stream,
            "200 OK",
            event_stream_headers(),
            b"data: redirected\n\n",
        )
    });
    let (redirect_url, redirect_thread) = {
        let target_url = target_url.clone();
        server(1, move |_, stream| {
            respond(
                stream,
                "302 Found",
                &format!("Location: {target_url}\r\nConnection: close\r\n"),
                b"",
            )
        })
    };
    let out = common::run(
        &format!(
            r#"
        const es = new EventSource('{redirect_url}');
        if (es.url !== '{redirect_url}/') throw Error('Original URL changed');
        es.addEventListener('message', (event) => {{
            if (event.origin !== '{target_url}') throw Error('Final origin: ' + event.origin);
            if (es.url !== '{redirect_url}/') throw Error('URL after redirect');
            if (event.data !== 'redirected') throw Error('Redirected data');
            es.close();
            console.log('redirect ok');
        }});
        es.addEventListener('error', (event) => {{
            throw Error('Unexpected error: ' + event.message);
        }});
    "#
        )
        .replace("{redirect_url}", &redirect_url)
        .replace("{target_url}", &target_url),
        &[],
    );
    assert_eq!(out.stdout, b"redirect ok\n");
    redirect_thread.join().unwrap();
    target_thread.join().unwrap();
}

#[test]
fn net_permission_failures_are_asynchronous() {
    let out = common::run(
        r#"
        import { pledge } from 'avril:std';
        pledge({ net: false });
        const log = [];
        const es = new EventSource('http://127.0.0.1:1/');
        if (es.readyState !== 0) throw Error('Synchronous failure');
        log.push('connecting');
        es.addEventListener('error', (event) => {
            if (es.readyState !== 0) throw Error('Ready state after denied connection');
            if (event instanceof ErrorEvent) throw Error('Wrong error event');
            log.push('denied');
            es.close();
            console.log(log.join(' | '));
        });
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"connecting | denied\n");
}

#[test]
fn call_time_primordials_survive_guest_monkeypatching() {
    let (url, thread) = server(1, |_, stream| {
        stream_chunks(
            stream,
            &[b"event: hello\ndata: bonjour\nid: id\xf0\x9f\x92\xa9x\n\n"],
        )
    });
    let out = common::run(
        &format!(
            r#"
        const poison = () => {{ throw Error('Guest replacement invoked'); }};
        Math.floor = poison;
        Math.min = poison;
        Number.parseInt = poison;
        Promise.prototype.then = poison;
        Object.defineProperty(Promise.prototype, 'constructor', {{ get: poison }});
        Object.defineProperty(String.prototype, 'indexOf', {{ value: poison }});
        Object.defineProperty(String.prototype, 'charCodeAt', {{ value: poison }});
        Object.defineProperty(String.prototype, 'slice', {{ value: poison }});
        Object.defineProperty(String.prototype, 'toLowerCase', {{ value: poison }});
        RegExp.prototype.exec = poison;
        Object.defineProperty(URL.prototype, 'href', {{ get: poison }});
        Object.defineProperty(URL.prototype, 'origin', {{ get: poison }});
        Object.defineProperty(URL.prototype, 'protocol', {{ get: poison }});
        Object.defineProperty(Object.getPrototypeOf(Uint8Array.prototype), 'length', {{ get: poison }});
        TextDecoder.prototype.decode = poison;
        TextEncoder.prototype.encode = poison;
        const addEventListener = EventTarget.prototype.addEventListener;
        EventTarget.prototype.addEventListener = poison;
        const es = new EventSource('{url}');
        addEventListener.call(es, 'message', () => {{ throw Error('message event should not fire'); }});
        addEventListener.call(es, 'hello', (event) => {{
            if (event.data !== 'bonjour') throw Error('Custom event');
            es.close();
            console.log('primordials safe');
        }});
    "#
        )
        .replace("{url}", &url),
        &[],
    );
    assert_eq!(out.stdout, b"primordials safe\n");
    thread.join().unwrap();
}

#[test]
fn workers_expose_event_source() {
    let dir = tempfile::tempdir().unwrap();
    let worker = dir.path().join("worker.js");
    std::fs::write(
        &worker,
        r#"
        if (typeof EventSource !== 'function') throw Error('Missing EventSource');
        const es = new EventSource('http://127.0.0.1:1/worker');
        if (!(es instanceof EventTarget) || es.readyState !== 0) throw Error('Worker EventSource');
        es.addEventListener('error', (event) => {
            if (es.readyState !== 0) throw Error('Ready state after denied connection');
            es.close();
            postMessage('worker eventsource ok');
            close();
        });
    "#,
    )
    .unwrap();
    let main = r#"
        if (typeof EventSource !== 'function') throw Error('Missing EventSource in main');
        const worker = new Worker(new URL('./worker.js', import.meta.url), {type: 'module', name: 'es'});
        const answer = new Promise((resolve, reject) => {worker.onmessage = e => resolve(e.data); worker.onerror = reject;});
        console.log(await answer);
    "#;
    std::fs::write(dir.path().join("main.js"), main).unwrap();
    let output = std::process::Command::new(env!("CARGO_BIN_EXE_avril"))
        .arg("run")
        .arg(dir.path().join("main.js"))
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .output()
        .unwrap();
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(output.stdout, b"worker eventsource ok\n");
}

#[test]
fn reconnects_preserve_the_resume_cursor_until_an_explicit_reset() {
    let count = std::sync::atomic::AtomicUsize::new(0);
    let (url, thread) = server(4, move |request, stream| {
        let attempt = count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        assert_eq!(
            header(&request.headers, "cache-control").as_deref(),
            Some("no-cache")
        );
        assert_eq!(
            header(&request.headers, "pragma").as_deref(),
            Some("no-cache")
        );
        assert_eq!(
            header(&request.headers, "last-event-id").as_deref(),
            if attempt == 1 || attempt == 2 {
                Some("cursor")
            } else {
                None
            }
        );
        let body: &[u8] = match attempt {
            0 => b"retry: 5\nid: cursor\ndata: first\n\n",
            1 => b"data: second\n\n",
            2 => b"id:\ndata: reset\n\n",
            _ => b"data: final\n\n",
        };
        respond(stream, "200 OK", event_stream_headers(), body)
    });
    let output = common::run(
        &format!(
            r#"
        const es = new EventSource('{url}'), seen = [];
        const completed = new Promise(resolve => es.onmessage = event => {{
            seen.push([event.data, event.lastEventId]);
            if (event.data === 'final') {{ es.close(); resolve(); }}
        }});
        await completed;
        console.log(JSON.stringify(seen));
    "#
        ),
        &[],
    );
    assert_eq!(
        output.stdout,
        b"[[\"first\",\"cursor\"],[\"second\",\"cursor\"],[\"reset\",\"\"],[\"final\",\"\"]]\n"
    );
    thread.join().unwrap();
}

#[test]
fn long_fragmented_lines_stream_without_rescanning_the_prefix() {
    const LENGTH: usize = 16 * 1024 * 1024;
    let (url, thread) = server(1, |_, stream| {
        respond(stream, "200 OK", event_stream_headers(), b"data: ")?;
        for _ in 0..LENGTH / 1024 {
            stream.write_all(&[b'x'; 1024])?;
        }
        stream.write_all(b"\n\n")
    });
    let output = common::run(
        &format!(
            r#"
        const es = new EventSource('{url}');
        await new Promise(resolve => es.onmessage = event => {{
            if (event.data.length !== {LENGTH} || event.data[0] !== 'x' || event.data.at(-1) !== 'x') throw Error('Long line');
            es.close(); resolve();
        }});
        console.log('long line');
    "#
        ),
        &[],
    );
    assert_eq!(output.stdout, b"long line\n");
    thread.join().unwrap();
}

#[test]
fn control_character_ids_do_not_prevent_reconnection() {
    let count = std::sync::atomic::AtomicUsize::new(0);
    let (url, thread) = server(5, move |request, stream| {
        let attempt = count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        assert_eq!(
            header(&request.headers, "last-event-id").as_deref(),
            if attempt == 3 {
                Some("valid\tid")
            } else {
                None
            }
        );
        let body: &[u8] = match attempt {
            0 => b"retry: 5\nid: a\x01b\ndata: first\n\n",
            1 => b"data: persisted\n\n",
            2 => b"id: valid\tid\ndata: valid\n\n",
            3 => b"id: a\x7fb\ndata: second\n\n",
            _ => b"data: final\n\n",
        };
        respond(stream, "200 OK", event_stream_headers(), body)
    });
    let output = common::run(
        &format!(
            r#"
        const es = new EventSource('{url}'), ids = [];
        await new Promise(resolve => es.onmessage = event => {{
            ids.push(event.lastEventId);
            if (event.data === 'final') {{es.close(); resolve();}}
        }});
        if (JSON.stringify(ids) !== JSON.stringify(['a\x01b','a\x01b','valid\tid','a\x7fb','a\x7fb'])) throw Error('Cursor changed');
        console.log('controls');
    "#
        ),
        &[],
    );
    assert_eq!(output.stdout, b"controls\n");
    thread.join().unwrap();
}

#[test]
fn reflected_construction_and_live_handlers_survive_replaced_public_bases() {
    let output = common::run(
        r#"
        const Ctor = EventSource, Parent = EventTarget;
        const add = Parent.prototype.addEventListener, dispatch = Parent.prototype.dispatchEvent;
        const poison = () => {throw Error('Guest replacement invoked');};
        Object.setPrototypeOf(Ctor, function() {throw Error('Dynamic super invoked');});
        Parent.prototype.addEventListener = Parent.prototype.removeEventListener = poison;
        Event.prototype.preventDefault = poison;
        globalThis.EventTarget = poison;
        class Sub extends Ctor {}
        const order = [], options = {
            get withCredentials() {order.push('options'); return true;}
        };
        const source = new Sub({toString() {order.push('url'); return 'data:text/plain,events';}}, options);
        source.close();
        if (order.join('|') !== 'url|options' || !source.withCredentials) throw Error('Conversion order');
        if (!(source instanceof Sub) || !(source instanceof Ctor) || !(source instanceof Parent)) {
            throw Error('Reflected inheritance');
        }
        const log = [];
        source.onopen = function(event) {
            if (this !== source || event.target !== source) throw Error('Handler receiver');
            log.push('first');
            source.onopen = function() {log.push('second');};
            return false;
        };
        add.call(source, 'open', () => log.push('observer'));
        const canceled = new Event('open', {cancelable: true});
        if (dispatch.call(source, canceled) || !canceled.defaultPrevented) throw Error('Handler cancellation');
        if (!dispatch.call(source, new Event('open', {cancelable: true}))) throw Error('Replacement handler');
        source.onopen = {handleEvent: poison};
        if (source.onopen !== null) throw Error('Non-callable handler');
        dispatch.call(source, new Event('open'));
        source.onopen = () => log.push('third');
        dispatch.call(source, new Event('open'));
        if (log.join('|') !== 'first|observer|second|observer|observer|observer|third') {
            throw Error('Handler replacement or removal changed ordering');
        }
        if (Reflect.ownKeys(source).length !== 0) throw Error('Unexpected instance fields');
        for (const name of ['CONNECTING', 'OPEN', 'CLOSED']) {
            for (const owner of [Ctor, Ctor.prototype]) {
                const d = Object.getOwnPropertyDescriptor(owner, name);
                if (d.writable || d.configurable || !d.enumerable) throw Error('Constant descriptor');
            }
        }
        const handler = Object.getOwnPropertyDescriptor(Ctor.prototype, 'onopen');
        if (Ctor.length !== 1 || fetch.length !== 1 || Ctor.prototype.close.length !== 0 ||
            handler.get.length !== 0 || handler.set.length !== 1) throw Error('Callable shape');
        console.log('reflected construction and handlers');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"reflected construction and handlers\n");
}