Fix CONNECT_ACK state bugs in client/server (#60)

## Client bug
Bug would manifest as a `AWS_ERROR_EVENT_STREAM_RPC_PROTOCOL_ERROR` if an `APPLICATION_MESSAGE` was sent immediately upon receiving a `CONNECT_ACK`

Turns out the client was setting `state=CONNECT_PROCESSED` in two places:
1) When writing the CONNECT message
2) When the CONNECT message's write-completion callback fired.

Fix is to ONLY set `state=CONNECT_PROCESSED` in place 1.

Bug would occur if the CONNECT_ACK was received before the CONNECT's write-complete callback fired. The state would go from CONNECT_ACK_PROCESSED back to CONNECT_PROCESSED.

This bug began manifesting when aws-c-io changed (PR# awslabs/aws-c-io#353) so that the write-completion callback never fired synchronously. This made it possible to receive a CONNECT_ACK before the CONNECT's write-completion callback fired. The write-completion callback always worked this way on windows though, so it is still righteous to fix the bug in aws-c-event-stream, rather than make further changes to aws-c-io.

## Server bug

Server was only setting `state=CONNECT_ACK_PROCESSED` in one place, but it was setting it from the write-completion callback. Seems like a similar thing could happen:
1) server sends CONNECT_ACK
2) client receives CONNECT_ACK
3) client send APPLICATION_MESSAGE
4) server receives APPLICATION_MESSAGE. This causes error because state is not yet `CONNECT_ACK_PROCESSED`
5) CONNECT_ACK write-completion fires, we set `STATE=CONNECT_ACK_PROCESSED`, but too late connection is already shutting down in error

Fix is to set `state=CONNECT_ACK_PROCESSED` when writing the CONNECT_ACK message, not when the write completes.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: graebm

source/event_stream_rpc_client.c

@@ -412,11 +412,10 @@ static void s_on_protocol_message_written_fn(
         (void *)message);
 
     if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) {
-        AWS_LOGF_DEBUG(
+        AWS_LOGF_TRACE(
             AWS_LS_EVENT_STREAM_RPC_CLIENT,
-            "id=%p: connect message flushed to the wire, waiting on connect ack.",
+            "id=%p: connect message flushed to the wire.",
             (void *)message_args->connection);
-        aws_atomic_store_int(&message_args->connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED);
     }
 
     if (message_args->end_stream) {
@@ -535,7 +534,10 @@ static int s_send_protocol_message(
     }
 
     if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT) {
-        AWS_LOGF_DEBUG(AWS_LS_EVENT_STREAM_RPC_CLIENT, "id=%p: sending connect message", (void *)connection);
+        AWS_LOGF_DEBUG(
+            AWS_LS_EVENT_STREAM_RPC_CLIENT,
+            "id=%p: sending connect message, waiting on connect ack",
+            (void *)connection);
         aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_PROCESSED);
     }
 

source/event_stream_rpc_server.c

@@ -484,12 +484,10 @@ static void s_on_protocol_message_written_fn(
         aws_error_debug_str(error_code));
 
     if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) {
-        AWS_LOGF_INFO(
+        AWS_LOGF_TRACE(
             AWS_LS_EVENT_STREAM_RPC_SERVER,
-            "id=%p: connect ack message sent, the connect handshake is completed",
+            "id=%p: connect ack message flushed to wire",
             (void *)message_args->connection);
-        aws_atomic_store_int(
-            &message_args->connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED);
     }
 
     if (message_args->end_stream) {
@@ -580,13 +578,20 @@ static int s_send_protocol_message(
         }
     }
 
-    if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK &&
-        !(message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_CONNECTION_ACCEPTED)) {
-        AWS_LOGF_DEBUG(
+    if (message_args->message_type == AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_CONNECT_ACK) {
+        AWS_LOGF_INFO(
             AWS_LS_EVENT_STREAM_RPC_SERVER,
-            "id=%p: connection ack was rejected closing connection",
+            "id=%p: sending connect ack message, the connect handshake is completed",
             (void *)connection);
-        args->terminate_connection = true;
+        aws_atomic_store_int(&connection->handshake_state, CONNECTION_HANDSHAKE_STATE_CONNECT_ACK_PROCESSED);
+
+        if (!(message_args->message_flags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_CONNECTION_ACCEPTED)) {
+            AWS_LOGF_DEBUG(
+                AWS_LS_EVENT_STREAM_RPC_SERVER,
+                "id=%p: connection ack was rejected closing connection",
+                (void *)connection);
+            args->terminate_connection = true;
+        }
     }
 
     args->flush_fn = flush_fn;

tests/event_stream_rpc_client_connection_test.c

@@ -1066,6 +1066,7 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
     test_data->user_data = &client_test_data;
     test_data->server_continuation_user_data = &client_test_data;
 
+    /* client sends CONNECT */
     struct aws_byte_buf connect_payload = aws_byte_buf_from_c_str("{ \"message\": \" connect message \" }");
     struct aws_event_stream_rpc_message_args connect_args = {
         .headers_count = 0,
@@ -1078,6 +1079,8 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
         test_data->client_connection, &connect_args, s_rpc_client_message_flush, &client_test_data));
 
     aws_mutex_lock(&client_test_data.sync_lock);
+
+    /* ...wait until sent and received... */
     aws_condition_variable_wait_pred(
         &client_test_data.sync_cvar,
         &client_test_data.sync_lock,
@@ -1092,6 +1095,7 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
         client_test_data.received_payload.len);
     aws_byte_buf_clean_up(&client_test_data.received_payload);
 
+    /* server sends CONNECT_ACK */
     client_test_data.received_message_type = 0;
     client_test_data.message_received = false;
     client_test_data.message_sent = false;
@@ -1102,6 +1106,7 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
     ASSERT_SUCCESS(aws_event_stream_rpc_server_connection_send_protocol_message(
         test_data->server_connection, &connect_args, s_rpc_client_message_flush, &client_test_data));
 
+    /* ...wait until sent and received... */
     aws_condition_variable_wait_pred(
         &client_test_data.sync_cvar,
         &client_test_data.sync_lock,
@@ -1116,6 +1121,7 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
         client_test_data.received_payload.len);
     aws_byte_buf_clean_up(&client_test_data.received_payload);
 
+    /* client sends message creating new stream */
     client_test_data.received_message_type = 0;
     client_test_data.message_received = false;
     client_test_data.message_sent = false;
@@ -1143,6 +1149,7 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
     ASSERT_SUCCESS(aws_event_stream_rpc_client_continuation_activate(
         client_token, operation_name, &operation_args, s_rpc_client_message_flush, &client_test_data));
 
+    /* ...wait until sent and received... */
     aws_condition_variable_wait_pred(
         &client_test_data.sync_cvar,
         &client_test_data.sync_lock,
@@ -1164,6 +1171,7 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
         client_test_data.last_seen_operation_name.len);
     aws_byte_buf_clean_up(&client_test_data.last_seen_operation_name);
 
+    /* server sends response with TERMINATE_STREAM flag set */
     client_test_data.received_message_type = 0;
     client_test_data.message_received = false;
     client_test_data.message_sent = false;
@@ -1174,6 +1182,14 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
     ASSERT_SUCCESS(aws_event_stream_rpc_server_continuation_send_message(
         client_test_data.server_token, &operation_args, s_rpc_client_message_flush, &client_test_data));
 
+    /* ...wait until sent and received... */
+    aws_condition_variable_wait_pred(
+        &client_test_data.sync_cvar,
+        &client_test_data.sync_lock,
+        s_rpc_client_message_transmission_completed_pred,
+        &client_test_data);
+
+    /* ...wait until client stream closed... */
     aws_condition_variable_wait_pred(
         &client_test_data.sync_cvar,
         &client_test_data.sync_lock,
@@ -1190,6 +1206,7 @@ static int s_test_event_stream_rpc_client_connection_continuation_send_message_o
 
     aws_byte_buf_clean_up(&client_test_data.received_payload);
 
+    /* should not be allowed to send further stream messages */
     ASSERT_ERROR(
         AWS_ERROR_EVENT_STREAM_RPC_STREAM_CLOSED,
         aws_event_stream_rpc_client_continuation_send_message(