// vtest flaky: true // vtest retry: 3 module main import net import net.html import net.http import net.mbedtls import sync.pool import time const concurrent_large_response_requests = 6 const concurrent_large_response_workers = 3 const concurrent_large_response_links = 2_000 const concurrent_large_response_chunk_size = 8_192 const concurrent_large_response_test_cert = '-----BEGIN CERTIFICATE-----\nMIIEOTCCAyECFG64Q2g46jZb3kRbDOJWX/BwjSp6MA0GCSqGSIb3DQEBCwUAMEUx\nCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRl\ncm5ldCBXaWRnaXRzIFB0eSBMdGQwIBcNMjMwODAyMTcyOTQyWhgPMjA1MDEyMTcx\nNzI5NDJaMGsxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDYWxpZm9ybmlhMRQwEgYD\nVQQHDAtMb3MgQW5nZWxlczEdMBsGA1UECgwUQ2F0YWx5c3QgRGV2ZWxvcG1lbnQx\nEjAQBgNVBAMMCWxvY2FsaG9zdDCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoC\nggIBALqAI4fqUi+QBVWcsXglouLdOML5+w0+1hSR1KdO0Q5XPdQAs/yYWJ+KUkDw\nG++rfy9DUPq7FNRBVurXQkcAtn6gXdllGUSjwUiDo/N4mMOyS/2sufBuaeww7jVi\nrppH+zwP1tUnjRd6khl6bi1Ian9VSzr3Iy9CkXIg1GU4CPXkOydLeoQfepXxWoK1\nOUNwT3VKC/stAfY3j/NIIeiJYkyuRGFCkxn/BUjN+AsXiTugRcYKEFHdIPkOuCXp\nYbhf+lLsczpxCs3rdZG9b/N6mEDCzXTmeHkmsjdPTf+1k5DZZvKzVBBrgdxCgBb7\n5RwjF5v9WmnIc33wWgfJC6FaUzj9NYxYUbPHD+jTz0rJB/jj4u/xJlM/e5NRmXdW\n70pOMKXtWjRSolLOFIPKLY1qs3KMTAZxKKWPDDF7WlMJxMRt7nnnks5yw43Nog4C\njDLk1ZgETnPpLgo3jbmJdIv+OHKTJrBlVvDq7VTyixCoS5G8KoOmyQJhaXG6NwE2\niVhH5JIKgzgCfetfDsnjxqJ/qtrFXPa8FF2TsomD0NK/GZmIcs+9OeVB75Jn5uhF\nfLHScpiTbuu5w3P/LI/MqihLRB6RRNnRzPH8fIg5bYC9b770ta/8GcFRuYE8t+UR\nGtqXJoIKixbDlqV54kal8FQzYzhETf9+NM6Kb/lKEfG/pslvAgMBAAEwDQYJKoZI\nhvcNAQELBQADggEBALI3uNiNO0QE1brA3QYFK+d9ZroB72NrJ0UNkzYHDg2Fc6xg\n4aVVfaxY08+TmKc0JlMOW+pUxeCW/+UBSngdQiR9EE9xm0k0XIrAsy9RXxRvEtPu\nM1VI2h7ayp1Y2BrnQinevTSgtqLRyS1VbOFRl1FiyVvinw2I0KsDdAMNevAPXcOa\nQ8pUgUq6f56DkhocQaj+hxD/uV8HryNxuoSXnPhvfTN3z4YRGzsaWevJ9EYJliOM\n+XugcqfFJ+W7/QCEcAHCL+Bw6OydG5NFORr3p57PXjjcL/uKmxPBrWg2Bz6uT4uR\nMhj0zttiFHLAt9jGfyk6W57UNUja1e1ggftJJhs=\n-----END CERTIFICATE-----\n' const concurrent_large_response_test_key = '-----BEGIN RSA PRIVATE KEY-----\nMIIJKQIBAAKCAgEAuoAjh+pSL5AFVZyxeCWi4t04wvn7DT7WFJHUp07RDlc91ACz\n/JhYn4pSQPAb76t/L0NQ+rsU1EFW6tdCRwC2fqBd2WUZRKPBSIOj83iYw7JL/ay5\n8G5p7DDuNWKumkf7PA/W1SeNF3qSGXpuLUhqf1VLOvcjL0KRciDUZTgI9eQ7J0t6\nhB96lfFagrU5Q3BPdUoL+y0B9jeP80gh6IliTK5EYUKTGf8FSM34CxeJO6BFxgoQ\nUd0g+Q64JelhuF/6UuxzOnEKzet1kb1v83qYQMLNdOZ4eSayN09N/7WTkNlm8rNU\nEGuB3EKAFvvlHCMXm/1aachzffBaB8kLoVpTOP01jFhRs8cP6NPPSskH+OPi7/Em\nUz97k1GZd1bvSk4wpe1aNFKiUs4Ug8otjWqzcoxMBnEopY8MMXtaUwnExG3ueeeS\nznLDjc2iDgKMMuTVmAROc+kuCjeNuYl0i/44cpMmsGVW8OrtVPKLEKhLkbwqg6bJ\nAmFpcbo3ATaJWEfkkgqDOAJ9618OyePGon+q2sVc9rwUXZOyiYPQ0r8ZmYhyz705\n5UHvkmfm6EV8sdJymJNu67nDc/8sj8yqKEtEHpFE2dHM8fx8iDltgL1vvvS1r/wZ\nwVG5gTy35REa2pcmggqLFsOWpXniRqXwVDNjOERN/340zopv+UoR8b+myW8CAwEA\nAQKCAgEAkcoffF0JOBMOiHlAJhrNtSiX+ZruzNDlCxlgshUjyWEbfQG7sWbqSHUZ\njZflTrqyZqDpyca7Jp2ZM2Vocxa0klIMayfj08trCaOWY3pPeROE4d3HUJMPjEpH\nvEXTFdnVJIOBPgl3+vWfBfm17QIh9j4X3BVbVNNl3WCaiDGAl699Kl+Pe38cFeCh\nD3JZPEWsZ5SlvwjU8sNGbThjAWN8C1NjMuCXG4hGej5Ae3M/nPPR91jgnw4Me4Ut\nIL3K3RVyGqaqAPJjLsu0kWQUArJAGMfvUkXjwVklkaUV5SHtJBs+pdTXjyprTmJR\nvSXWWON5zkAEEJNY7QcZaeKYi96PFLUFI+ciEdnXn74CfSKhgZCBo+OyFZjDWW5R\nNmgAbZTN2RW0z+V54Lg36JfJrmiGs8TN06KwNjFo+iOJCdQnoUSIhTlmMfVbXPah\ntRfQvwqtfqVS9W/jkiGq9yDDqyXx093R/QTM/XqDlWJ2iOJFppOJefGFCWF6Fwll\nVT9povTAGQmXFiAxwFZxWtbFa0i8fP5QG80X6l/gRklSd6ZXAVvcLkaFGqxunDAe\nrYC2jBwHWRpVmbxw880SWRzlAsJXc7M8PQnBTlyX1mFZNnwAJgqplz0BQHQhQh4V\nqNfisUm9smtda+Hr9GBBUxs09ulery3I0lQjsArVxPqPVgUbFPECggEBANqLA5fH\n2LupOBoFH/fK5jixyGdSB8eJvU+XuS8RBBexnzTQApmDHiU7Axa/cKvxAfUgwBpU\n6OIsL6Lq6wowVInBgo7GraACwspGMIP8Z7+A8qDgSWIcpXP21Ny2RW+nukdH8ZnV\nTFtiFxLYU9GRfzSUcqvE0miKfMGP/S9Cqbew00K6CQ2xurLTR2AchfUQZJJIg7eF\nRBoftthXLQ+s1JoiLJX2gqCliFy32RMAUP+pKvKVJmVQh8bxEkoEzTV2eY7eTxsH\nJDH5hD66EZ5bW/nVAMruJ3iKjy3WvjDbnddNAz9IFKrd1RMP9dgSEKuSv/HhqwPe\n1q9Wm6LWZo8BlYcCggEBANp3M14QMcMxRlZE0TiSopi1CaE8OG0C9apToS1dol2s\n4lCsWHVPIC516LMPGU0bmCdtwJey1mgXQEKVxCWHkVhhoCKT/tN53o5qkptrhrXL\npbqmRfoMXI7LwJU+Vqi5fwSPGrSR/IzHwCUL7pHTbYN7wT5rr2rcC84XYSX31TFm\nNfMnbDuUk33ycAo07Vqts5A5FN+xViEUMFSDmfA2XmOAV77awz0l/3n3qOg9lQYe\nU4Av2nT19lGELirLInkB1ndLirWAcLaCBXKOLW4bzpNm9Bt8aiziVzcUzlJlLa+1\nnb/7//xzKi0eM/BhyJfhsmOz5B8AQ6Ca/keDk8M7JtkCggEARl8DDinE6VCpBv/l\ndlX4YgMlQ9fPN3pr4ig58iTpi3Ofj1L3s1TcLSLecMG+Vy9o8PTVxuTWhJWz1SMO\nAh7j6ePM1Yq2N9MLxDRrxOROyASOnCz8lEIjKL8vdc6fdz+sJO3OpzleuAJS6beM\n7euK6XRvpE3hbtZBK9bgsQonOkYPEOp0pds4AgM0dYdZvzrDF7OP7lVUQ5E4wFr5\n4JVHdEZS0wsoru/+g9STaqHscxaXBLvwPCl9Pxs7R2haZ7+5jr6Y/FwFVK5C3ivu\nJm7GpCDpe27KeO8tAZancXYWUlCzHfpo5Ug/Jz85a5UNlyHO+uUuuzVTLeyWew3M\nwnnBGwKCAQEAqGTBP3wUH3TX1p9s9cJxemvxZEra44woeIXF8wX9pV8hgzWVabb4\nA1f3ai31Pq5KdfnvPf8nrUxex/RRIOyCaDG4EW8qOS/zEKutHgef6nly4ZBQ2BC3\nN4pug5ttiNiSw5za5NyyYoGF5ghweA8UlwjJR6gRqri6kL0MsQt7VXyHkUmN787y\ncV5yZiut2PuTMVQOdu5miVDagAqAmdwOnXvMJtzRKU0kw4rWs0zklbbCfkhkh0sf\n9m2AeJPjmoqEGags3wKF3ugR8t8MvZbJgG0XNCiOXtKIj3iGIJTExm+jjNxd0OWk\nWOqy9lMpH4lky91ZtVuqxR0za0RMnWv24QKCAQBe8l0w9AYVNGDLv1jyPcbsncty\nNYI81yqe2mL+TC00sMCeil7C7WCP7kRklY01rH5q5gJ9Q1UV+bOj2fQdXDmQ5Bgo\n41jseh44gkbuXAeWcSDrDkJCrfvlNqFobTmUb8cdb9aQlHYfOJ31367LJspiw2SY\nmCbnLQ5sMnyBiMkcn0GfBV6IAkZVN73DPa8a1m/0Qrrv1GmBJFVbuZd9d/hAWpHa\nekhXPq0Sta+RNDfBR3aI5lAmVA17qRGiubQYJ+Ldq0aRJ40fGE51ctoSU/5RMcmh\n6+Qro+jSC94L46xMFp+1J5atgB1p/jVzTT/Ws7SLyotYUSL8zU7tcLiycQXs\n-----END RSA PRIVATE KEY-----\n' struct ConcurrentLargeResponseContext { expected_body string expected_link_count int } struct ConcurrentLargeResponseResult { status_code int body_matches bool link_count int err string } fn test_https_large_bodies_can_be_processed_concurrently() { // Watchdog thread to prevent the test from hanging indefinitely on CI. spawn fn () { time.sleep(120 * time.second) eprintln('mbedtls concurrent large response test: timeout reached, exiting.') exit(1) }() mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0')! port := port_listener.addr()!.port()! port_listener.close()! mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{ cert: concurrent_large_response_test_cert cert_key: concurrent_large_response_test_key validate: false in_memory_verification: true })! server := spawn serve_large_html_responses(mut listener, concurrent_large_response_body(), concurrent_large_response_requests) mut urls := []string{cap: concurrent_large_response_requests} for i in 0 .. concurrent_large_response_requests { urls << 'https://127.0.0.1:${port}/?req=${i}' } expected_body := concurrent_large_response_body() mut pp := pool.new_pool_processor(callback: concurrent_large_response_worker) pp.set_max_jobs(concurrent_large_response_workers) pp.set_shared_context(&ConcurrentLargeResponseContext{ expected_body: expected_body expected_link_count: concurrent_large_response_links }) pp.work_on_items(urls) server.wait() for result in pp.get_results_ref[ConcurrentLargeResponseResult]() { assert result.err == '' assert result.status_code == 200 assert result.body_matches assert result.link_count == concurrent_large_response_links } } fn test_https_request_timeout_closes_the_connection() { mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0')! port := port_listener.addr()!.port()! port_listener.close()! mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{ cert: concurrent_large_response_test_cert cert_key: concurrent_large_response_test_key validate: false in_memory_verification: true })! server := spawn serve_incomplete_https_response(mut listener) mut req := http.new_request(.get, 'https://127.0.0.1:${port}', '') req.read_timeout = 250 * time.millisecond req.validate = false req.do() or { assert server.wait() return } panic('expected the HTTPS request to time out') } fn test_https_chunked_body_is_returned_by_http_get_text() { mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0')! port := port_listener.addr()!.port()! port_listener.close()! mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{ cert: concurrent_large_response_test_cert cert_key: concurrent_large_response_test_key validate: false in_memory_verification: true })! expected_body := concurrent_large_response_body() server := spawn serve_chunked_html_response(mut listener, expected_body) body := http.get_text('https://127.0.0.1:${port}/chunked') server.wait() assert body == expected_body } fn concurrent_large_response_body() string { return '' + '
payload
'.repeat(concurrent_large_response_links) + '' } fn serve_incomplete_https_response(mut listener mbedtls.SSLListener) bool { defer { listener.shutdown() or {} } mut conn := listener.accept() or { panic(err) } defer { conn.shutdown() or {} } mut request_buf := []u8{len: 2048} _ = conn.read(mut request_buf) or { return false } conn.write_string('HTTP/1.1 200 OK\r\nContent-Length: 2\r\n') or { return false } conn.set_read_timeout(time.second) mut drain_buf := []u8{len: 128} _ = conn.read(mut drain_buf) or { return err.code() != net.err_timed_out_code } return false } fn serve_large_html_responses(mut listener mbedtls.SSLListener, body string, request_count int) { defer { listener.shutdown() or {} } mut handlers := []thread{cap: request_count} for _ in 0 .. request_count { mut conn := listener.accept() or { panic(err) } handlers << spawn write_large_html_response(mut conn, body) } handlers.wait() } fn serve_chunked_html_response(mut listener mbedtls.SSLListener, body string) { defer { listener.shutdown() or {} } mut conn := listener.accept() or { panic(err) } write_chunked_html_response(mut conn, body) } fn write_large_html_response(mut conn mbedtls.SSLConn, body string) { defer { conn.shutdown() or {} } mut request_buf := []u8{len: 2048} _ = conn.read(mut request_buf) or { return } header := 'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: ${body.len}\r\nConnection: close\r\n\r\n' conn.write_string(header) or { return } mut start := 0 for start < body.len { end := if start + concurrent_large_response_chunk_size > body.len { body.len } else { start + concurrent_large_response_chunk_size } conn.write_string(body[start..end]) or { return } start = end time.sleep(1 * time.millisecond) } } fn write_chunked_html_response(mut conn mbedtls.SSLConn, body string) { defer { conn.shutdown() or {} } mut request_buf := []u8{len: 2048} _ = conn.read(mut request_buf) or { return } header := 'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n' conn.write_string(header) or { return } mut start := 0 for start < body.len { end := if start + concurrent_large_response_chunk_size > body.len { body.len } else { start + concurrent_large_response_chunk_size } chunk := body[start..end] conn.write_string('${chunk.len:x}\r\n') or { return } conn.write_string(chunk) or { return } conn.write_string('\r\n') or { return } start = end time.sleep(1 * time.millisecond) } conn.write_string('0\r\n\r\n') or { return } } fn concurrent_large_response_worker(mut pp pool.PoolProcessor, idx int, _wid int) &ConcurrentLargeResponseResult { url := pp.get_item[string](idx) ctx := unsafe { &ConcurrentLargeResponseContext(pp.get_shared_context()) } resp := http.fetch( method: .get url: url validate: false ) or { return &ConcurrentLargeResponseResult{ err: '${err}' } } doc := html.parse(resp.body) return &ConcurrentLargeResponseResult{ status_code: resp.status_code body_matches: resp.body == ctx.expected_body link_count: doc.get_tags(name: 'a').len } }