Remember optimistically sent data until we have gotten a CONNECTED

Since we can retry failed streams under some circumstances, we need
to be ready to send data queued on them.

4 files changed, 68 insertions, 7 deletions

diff --git a/src/or/connection.c b/src/or/connection.c
index 8b9fb126d..1ccd2b660 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -444,7 +444,12 @@ _connection_free(connection_t *conn)
     tor_free(edge_conn->chosen_exit_name);
     if (edge_conn->socks_request)
       socks_request_free(edge_conn->socks_request);
-
+    if (edge_conn->pending_optimistic_data) {
+      generic_buffer_free(edge_conn->pending_optimistic_data);
+    }
+    if (edge_conn->sending_optimistic_data) {
+      generic_buffer_free(edge_conn->sending_optimistic_data);
+    }
     rend_data_free(edge_conn->rend_data);
   }
   if (conn->type == CONN_TYPE_CONTROL) {
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
index 20f8bc916..79f0db037 100644
--- a/src/or/connection_edge.c
+++ b/src/or/connection_edge.c
@@ -738,6 +738,12 @@ connection_ap_detach_retriable(edge_connection_t *conn, origin_circuit_t *circ,
 {
   control_event_stream_status(conn, STREAM_EVENT_FAILED_RETRIABLE, reason);
   conn->_base.timestamp_lastread = time(NULL);
+
+  if (conn->pending_optimistic_data) {
+    generic_buffer_set_to_copy(&conn->sending_optimistic_data,
+                               conn->pending_optimistic_data);
+  }
+
   if (!get_options()->LeaveStreamsUnattached || conn->use_begindir) {
     /* If we're attaching streams ourself, or if this connection is
      * a tunneled directory connection, then just attach it. */
@@ -2429,7 +2435,8 @@ connection_ap_handshake_send_begin(edge_connection_t *ap_conn)
   control_event_stream_status(ap_conn, STREAM_EVENT_SENT_CONNECT, 0);
 
   /* If there's queued-up data, send it now */
-  if (connection_get_inbuf_len(TO_CONN(ap_conn)) &&
+  if ((connection_get_inbuf_len(TO_CONN(ap_conn)) ||
+       ap_conn->sending_optimistic_data) &&
       connection_ap_supports_optimistic_data(ap_conn)) {
     log_info(LD_APP, "Sending up to %ld bytes of queued-up data",
              connection_get_inbuf_len(TO_CONN(ap_conn)));
diff --git a/src/or/or.h b/src/or/or.h
index c78426733..7097b5f0f 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -1237,11 +1237,20 @@ typedef struct edge_connection_t {
    * NATd connection */
   unsigned int is_transparent_ap:1;
 
-  /** Set if this connection's target exit node allows optimistic data.
-   * (That is, data sent on this stream before the exit has sent a
-   * CONNECTED cell.)*/
+  /** For AP connections only: Set if this connection's target exit node
+   * allows optimistic data.  (That is, data sent on this stream before
+   * the exit has sent a CONNECTED cell.)*/
   unsigned int exit_allows_optimistic_data : 1;
 
+  /** For AP connections only: buffer for data that we have sent
+   * optimistically, which we might need to re-send if we have to
+   * retry this connection. */
+  generic_buffer_t *pending_optimistic_data;
+  /* For AP connections only: buffer for data that we previously sent
+  * optimistically which we are currently re-sending as we retry this
+  * connection. */
+  generic_buffer_t *sending_optimistic_data;
+
   /** If this is a DNSPort connection, this field holds the pending DNS
    * request that we're going to try to answer.  */
   struct evdns_server_request *dns_server_request;
diff --git a/src/or/relay.c b/src/or/relay.c
index df6d0a8a5..3fa31a3f3 100644
--- a/src/or/relay.c
+++ b/src/or/relay.c
@@ -944,6 +944,12 @@ connection_edge_process_relay_cell_not_open(
           break;
       }
     }
+    /* This is definitely a success, so forget about any pending data we
+     * had sent. */
+    if (conn->pending_optimistic_data) {
+      generic_buffer_free(conn->pending_optimistic_data);
+      conn->pending_optimistic_data = NULL;
+    }
 
     /* handle anything that might have queued */
     if (connection_edge_package_raw_inbuf(conn, 1, NULL) < 0) {
@@ -1343,6 +1349,10 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
   char payload[CELL_PAYLOAD_SIZE];
   circuit_t *circ;
   unsigned domain = conn->cpath_layer ? LD_APP : LD_EXIT;
+  int sending_from_optimistic = 0;
+  const int sending_optimistically =
+    conn->_base.type == CONN_TYPE_AP &&
+    conn->_base.state != AP_CONN_STATE_OPEN;
 
   tor_assert(conn);
 
@@ -1375,7 +1385,18 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
     return 0;
   }
 
-  amount_to_process = connection_get_inbuf_len(TO_CONN(conn));
+  sending_from_optimistic = conn->sending_optimistic_data != NULL;
+
+  if (PREDICT_UNLIKELY(sending_from_optimistic)) {
+    amount_to_process = generic_buffer_len(conn->sending_optimistic_data);
+    if (PREDICT_UNLIKELY(!amount_to_process)) {
+      log_warn(LD_BUG, "sending_optimistic_data was non-NULL but empty");
+      amount_to_process = connection_get_inbuf_len(TO_CONN(conn));
+      sending_from_optimistic = 0;
+    }
+  } else {
+    amount_to_process = connection_get_inbuf_len(TO_CONN(conn));
+  }
 
   if (!amount_to_process)
     return 0;
@@ -1391,11 +1412,30 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
   stats_n_data_bytes_packaged += length;
   stats_n_data_cells_packaged += 1;
 
-  connection_fetch_from_buf(payload, length, TO_CONN(conn));
+  if (PREDICT_UNLIKELY(sending_from_optimistic)) {
+    /* XXX023 We could be more efficient here by sometimes packing
+     * previously-sent optimistic data in the same cell with data
+     * from the inbuf. */
+    generic_buffer_get(conn->sending_optimistic_data, payload, length);
+    if (!generic_buffer_len(conn->sending_optimistic_data)) {
+        generic_buffer_free(conn->sending_optimistic_data);
+        conn->sending_optimistic_data = NULL;
+    }
+  } else {
+    connection_fetch_from_buf(payload, length, TO_CONN(conn));
+  }
 
   log_debug(domain,"(%d) Packaging %d bytes (%d waiting).", conn->_base.s,
             (int)length, (int)connection_get_inbuf_len(TO_CONN(conn)));
 
+  if (sending_optimistically && !sending_from_optimistic) {
+    /* This is new optimistic data; remember it in case we need to detach and
+       retry */
+    if (!conn->pending_optimistic_data)
+      conn->pending_optimistic_data = generic_buffer_new();
+    generic_buffer_add(conn->pending_optimistic_data, payload, length);
+  }
+
   if (connection_edge_send_command(conn, RELAY_COMMAND_DATA,
                                    payload, length) < 0 )
     /* circuit got marked for close, don't continue, don't need to mark conn */