When is flush needed?

[petrzjunior]

I am trying to use this library in my project and found an interesting sequence of events:

Server init -> client init -> handshake started -> client write -> client read -> handshake finished

In this sequence, client write is buffered until the handshake is finished. But then, it is not actually sent and so a response never comes back. I managed to fix this by either waiting on handshake .handshake().await or by flushing the client after write before read .flush().await.

Since this crate has only little documentation, I wanted to clarify if this is a bug or a feature. Should I call flush whenever I expect a response to come back?

[mmastrac]

This is definitely a bug -- the handshake is supposed to wake up readers and writers on completion.

Are you using a multi-threaded runtime?

[petrzjunior]

No, I am using single-threaded tokio::test. I tried flavor = "multi_thread" as well with different number of threads, but the result is still the same.

I double checked my code and the usecase is to test an external proxy program. The testcase works by sending data to a proxy input and trying to read from the proxy output. When the data is written before the handshakes completes, they are never flushed into to proxy and the read on the other side of the proxy hangs indefinitely. The fix is to flush immediately after the write.

The code is similar to this:

#[tokio::test]
async fn should_forward_messages() {
	let client = TcpStream::connect(SocketAddrV4::new(ip, port))
		.timeout(Duration::from_secs(1))
		.await
		.expect("timeout")
		.expect("connect failed");
	let client_tls = TlsStream::new_client_side(
		iso_client_tcp,
		ClientConnection::new(tls_config, ServerName::try_from("localhost").unwrap()).unwrap(),
		None,
	);

	client_tls.write(message).await.expect("failed writing");
	// this fixes the hang:
	// client_tls.flush().await.expect("failed flushing");
	
	read_other_side().await.timeout(Duration::from_secs(1)); // hangs
}

I cannot share the exact code here, but I am happy to update it and debug it, if you guide me.

[mmastrac]

@petrzjunior Does PR #35 fix the issue? There was definitely a race here, but I'm wondering if there's a different failure mode.

If #35 doesn't fix it and you can capture this in a test without proprietary code, I can definitely take a closer look.

[mmastrac]

I think I understand what might be happening here for you, and I fixed an unrelated bug that happened to exist 😄

rustls-tokio-stream only guarantees that it will make progress while you are actively polling it. If you write before the handshake completes, the stream will buffer the data until the next active polling operation after the handshake completes. Unfortunately, this means that a buffered write-before-handshake will fail to flush if you don't poll something.

In this case, polling for flush is probably the right way to fix it. It should also work if you poll for reads on the client after the write.

In the case where a TlsStream is split and there is a task always reading and a task always writing, it's guaranteed that all written data will be flushed at some point.

[petrzjunior]

If I understand it correctly, the problem is, that I am awaiting a Future unrelated to the rustls-tokio-stream and therefore it never wakes?

By the way, here is the trace log:

2025-01-27T22:03:03.704593Z DEBUG : Connecting to 127.0.0.1:19319
2025-01-27T22:03:03.704897Z  INFO : Connected to 127.0.0.1:19319
2025-01-27T22:03:03.704919Z  INFO : Sending logon request
2025-01-27T22:03:03.705129Z TRACE send_message: ::server: Sent: HELLO
2025-01-27T22:03:03.705208Z  INFO : Sending echo request
2025-01-27T22:03:03.705357Z TRACE send_message: ::server: Sent: TEST
2025-01-27T22:03:03.705999Z DEBUG : Running cron job
2025-01-27T22:03:03.708010Z DEBUG : New client accepted
[1738015383708] write buf
[1738015383708] w=Ok(250)
[1738015383708] r(4096)=Ok(250)
2025-01-27T22:03:03.708453Z TRACE rustls::server::hs: we got a clienthello ClientHelloPayload { client_version: TLSv1_2, random: 918961dc634efccf8d5c27414ba251d93c260aea278e461a44d5b6efaefb7d7f, session_id: b7154c8406d8218332a30efe2ce4ef77ea38d17872d4466fdd923900d0c95dc9, cipher_suites: [TLS13_AES_256_GCM_SHA384, TLS13_AES_128_GCM_SHA256, TLS_EMPTY_RENEGOTIATION_INFO_SCSV], compression_methods: [Null], extensions: [PresharedKeyModes([PSK_DHE_KE]), SignatureAlgorithms([ECDSA_NISTP384_SHA384, ECDSA_NISTP256_SHA256, ECDSA_NISTP521_SHA512, ED25519, RSA_PSS_SHA512, ]), KeyShare([KeyShareEntry { group: X25519, payload: 71c185e04330c6de15e6152d1d7a8c0b0811308a116c0eff29235d999e35ec37 }]), SupportedVersions([TLSv1_3]), Protocols([ProtocolName(6832), ProtocolName(687474702f312e31)]), NamedGroups([X25519, secp256r1, secp384r1]), ExtendedMasterSecretRequest, EcPointFormats([Uncompressed]), CertificateStatusRequest(Ocsp(OcspCertificateStatusRequest { responder_ids: [], extensions:  })), ServerName([ServerName { typ: HostName, payload: HostName(DnsName("localhost")) }])] }    
2025-01-27T22:03:03.708505Z TRACE rustls::server::server_conn: Accepted::client_hello(): ClientHello {
    server_name: Some(
        DnsName(
            "localhost",
        ),
    ),
    signature_schemes: [
        ECDSA_NISTP384_SHA384,
        ECDSA_NISTP256_SHA256,
        ECDSA_NISTP521_SHA512,
        ED25519,
        RSA_PSS_SHA512,
        RSA_PSS_SHA384,
        RSA_PSS_SHA256,
        RSA_PKCS1_SHA512,
        RSA_PKCS1_SHA384,
        RSA_PKCS1_SHA256,
    ],
    alpn: Some(
        [
            ProtocolName(
                6832,
            ),
            ProtocolName(
                687474702f312e31,
            ),
        ],
    ),
    server_cert_types: None,
    client_cert_types: None,
    cipher_suites: [
        TLS13_AES_256_GCM_SHA384,
        TLS13_AES_128_GCM_SHA256,
        TLS13_CHACHA20_POLY1305_SHA256,
        TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
        TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
        TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
        TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
        TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
        TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
        TLS_EMPTY_RENEGOTIATION_INFO_SCSV,
    ],
    certificate_authorities: None,
}    
2025-01-27T22:03:03.711308Z TRACE rustls::server::hs: Resolving server certificate: ClientHello {
    server_name: Some(
        DnsName(
            "localhost",
        ),
    ),
    signature_schemes: [
        ECDSA_NISTP384_SHA384,
        ECDSA_NISTP256_SHA256,
        ECDSA_NISTP521_SHA512,
        ED25519,
        RSA_PSS_SHA512,
        RSA_PSS_SHA384,
        RSA_PSS_SHA256,
        RSA_PKCS1_SHA512,
        RSA_PKCS1_SHA384,
        RSA_PKCS1_SHA256,
    ],
    alpn: Some(
        [
            ProtocolName(
                6832,
            ),
            ProtocolName(
                687474702f312e31,
            ),
        ],
    ),
    server_cert_types: None,
    client_cert_types: None,
    cipher_suites: [
        TLS13_AES_256_GCM_SHA384,
        TLS13_AES_128_GCM_SHA256,
        TLS13_CHACHA20_POLY1305_SHA256,
        TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
        TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
        TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
        TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
        TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
        TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
        TLS_EMPTY_RENEGOTIATION_INFO_SCSV,
    ],
    certificate_authorities: None,
}    
2025-01-27T22:03:03.711337Z DEBUG rustls::server::hs: decided upon suite TLS13_AES_256_GCM_SHA384    
2025-01-27T22:03:03.711436Z TRACE rustls::server::tls13::client_hello: sending server hello Message { version: TLSv1_2, payload: Handshake { parsed: HandshakeMessagePayload { typ: ServerHello, payload: ServerHello(ServerHelloPayload { extensions: [KeyShare(KeyShareEntry { group: X25519, payload: 0052b0c9d73c8881d6433bf49a6280e80fa51edd67bda4608a45a499a97fb04d }), SupportedVersions(TLSv1_3)], legacy_version: TLSv1_2, random: b21a8c866deea25d714a82dc4316a5c67e22698af062142e097f50089e6e5f4b, session_id: b7154c8406d8218332a30efe2ce4ef77ea38d17872d4466fdd923900d0c95dc9, cipher_suite: TLS13_AES_256_GCM_SHA384, compression_method: Null }) }, encoded: 020000760303b21a8c866deea25d714a82dc4316a5c67e22698af062142e097f50089e6e5f4b20b7154c8406d8218332a30efe2ce4ef77ea38d17872d4466fdd923900d0c95dc9130200002e00330024001d00200052b0c9d73c8881d6433bf49a6280e80fa51edd67bda4608a45a499a97fb04d002b00020304 } }    
2025-01-27T22:03:03.711527Z DEBUG rustls::server::hs: Chosen ALPN protocol [104, 50]    
2025-01-27T22:03:03.711537Z TRACE rustls::server::tls13::client_hello: sending encrypted extensions HandshakeMessagePayload { typ: EncryptedExtensions, payload: EncryptedExtensions([Protocols([ProtocolName(6832)]), ServerNameAck]) }    
2025-01-27T22:03:03.711549Z TRACE rustls::server::tls13::client_hello: Sending CertificateRequest HandshakeMessagePayload { typ: CertificateRequest, payload: CertificateRequestTls13(CertificateRequestPayloadTls13 { context: , extensions: [SignatureAlgorithms([ECDSA_NISTP384_SHA384, ]), AuthorityNames([DistinguishedName(xxx)])] }) }    
2025-01-27T22:03:03.711570Z TRACE rustls::server::tls13::client_hello: sending certificate HandshakeMessagePayload { typ: Certificate, payload: CertificateTls13(CertificatePayloadTls13 { context: , entries: [CertificateEntry { cert: CertificateDer(0xxxx), exts: [] }] }) }    
2025-01-27T22:03:03.716127Z TRACE rustls::server::tls13::client_hello: sending certificate-verify HandshakeMessagePayload { typ: CertificateVerify, payload: CertificateVerify(DigitallySignedStruct { scheme: RSA_PSS_SHA512, sig: xxx }) }    
2025-01-27T22:03:03.716255Z TRACE rustls::server::tls13::client_hello: sending finished HandshakeMessagePayload { typ: Finished, payload: Finished(e69c683d42ff26364c219c105fca4865295c0491efcedfb0cec6ccc34cd8d04d9f733227a8d70e063c93a1c3c9a07151) }    
[1738015383716] w=Ok(127)
[1738015383716] w=Ok(6)
[1738015383716] w=Ok(2211)
[1738015383716] r(4096)=Err(Kind(WouldBlock))
[1738015383716] r(4096)=Ok(2344)
[1738015383722] w=Ok(6)
[1738015383722] w=Ok(2049)
[1738015383722] r(4096)=Ok(6)
2025-01-27T22:03:03.722159Z TRACE rustls::conn: Dropping CCS    
[1738015383722] r(4096)=Err(Kind(WouldBlock))
[1738015383762] r(4096)=Ok(2049)
2025-01-27T22:03:03.763577Z TRACE rustls::server::tls13: client CertificateVerify OK    
2025-01-27T22:03:03.763762Z TRACE rustls::server::tls13: sending new ticket HandshakeMessagePayload { typ: NewSessionTicket, payload: NewSessionTicketTls13(NewSessionTicketPayloadTls13 { lifetime: 86400, age_add: 1162361866, nonce: 2a3f8c25ffe9cbb8fb0eac958e05fa65dce27b54632480fd522bd8bad9d94a54, ticket: a0bb5678ec282091508005d2704939699534e84e22da7b479d155989b21867ef, exts: [] }) } (stateless: false)    
2025-01-27T22:03:03.763847Z TRACE rustls::server::tls13: sending new ticket HandshakeMessagePayload { typ: NewSessionTicket, payload: NewSessionTicketTls13(NewSessionTicketPayloadTls13 { lifetime: 86400, age_add: 2003928252, nonce: 60a65a5a3c2c531366f0142d85572533a39f44b8fb40c66196467bdec4a26577, ticket: bf07ce354a4a7ed2b87d44d5b341bf0330c62e2fff0fa9d11db6cd916aaa646a, exts: [] }) } (stateless: false)    
[1738015383763] w=Ok(184)
[1738015383764] finalize handshake
[1738015383764] join=Ok(Ok(HandshakeResult(PollEvented { io: Some(TcpStream { addr: 127.0.0.1:22339, peer: 127.0.0.1:37844, fd: 12 }) }, Server(ServerConnection))))
[1738015383764] hs buf=0
[1738015383764] r(4096)=Err(Kind(WouldBlock))
[1738015384709] dropping TlsStream { Handshaking }

[petrzjunior]

#35 does not fix the issue.

[mmastrac]

If I understand it correctly, the problem is, that I am awaiting a Future unrelated to the rustls-tokio-stream and therefore it never wakes?

Correct. It might be possible to rework the end-of-handshake logic to fully flush the write buffer, but that would add some complications to the write paths and likely require adding another post-handshake state (ie: post-handshake flushing).

It should also be possible to add a configuration option to disallow writes before the handshake completes which would also fix it, but that would likely require some refactoring of the TlsStream constructors.

[petrzjunior]

I understand. What do you propose to fix this particular use case on my side? Is it sufficient to add a manual flush after the write?

[mmastrac]

I'd recommend that you await the handshake after connection and avoid any pre-handshake buffering, but it would also be sufficient to flush after the writes. Flushing adds a little bit of overhead -- that code path could probably be optimized a bit.

[petrzjunior]

I added a handshake await and it works for me just fine. Feel free to close this issue.