+end_stream_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+end:
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(struct lttcomm_relayd_generic_reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"rotate session stream\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+
+end_no_reply:
+ return ret;
+}
+
+/*
+ * relay_mkdir: Create a folder on the disk.
+ */
+static int relay_mkdir(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_mkdir path_info_header;
+ struct lttcomm_relayd_generic_reply reply;
+ char *path = NULL;
+ size_t header_len;
+ ssize_t send_ret;
+ struct lttng_buffer_view path_view;
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to create a directory before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ /*
+ * This client is not supposed to use this command since
+ * it predates its introduction.
+ */
+ ERR("relay_mkdir command is unsupported before LTTng 2.11");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ header_len = sizeof(path_info_header);
+ if (payload->size < header_len) {
+ ERR("Unexpected payload size in \"relay_mkdir\": expected >= %zu bytes, got %zu bytes",
+ header_len, payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+
+ memcpy(&path_info_header, payload->data, header_len);
+
+ path_info_header.length = be32toh(path_info_header.length);
+
+ if (payload->size < header_len + path_info_header.length) {
+ ERR("Unexpected payload size in \"relay_mkdir\" including path: expected >= %zu bytes, got %zu bytes",
+ header_len + path_info_header.length, payload->size);
+ ret = -1;
+ goto end_no_session;
+ }
+
+ /* Ensure that it fits in local path length. */
+ if (path_info_header.length >= LTTNG_PATH_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("Path name argument of mkdir command (%" PRIu32 " bytes) exceeds the maximal length allowed (%d bytes)",
+ path_info_header.length, LTTNG_PATH_MAX);
+ goto end;
+ }
+
+ path_view = lttng_buffer_view_from_view(payload, header_len,
+ path_info_header.length);
+
+ path = create_output_path(path_view.data);
+ if (!path) {
+ ERR("Failed to create output path");
+ ret = -1;
+ goto end;
+ }
+
+ DBG("MKDIR command has path \"%s\", changed to \"%s\"", path_view.data, path);
+ ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, -1, -1);
+ if (ret < 0) {
+ ERR("relay creating output directory");
+ goto end;
+ }
+
+ ret = 0;
+
+end:
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"mkdir\" command reply (ret = %zd)", send_ret);
+ ret = -1;
+ }
+
+end_no_session:
+ free(path);
+ return ret;
+}
+
+static int validate_rotate_rename_path_length(const char *path_type,
+ uint32_t path_length)
+{
+ int ret = 0;
+
+ if (path_length > LTTNG_PATH_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("rotate rename \"%s\" path name length (%" PRIu32 " bytes) exceeds the allowed size of %i bytes",
+ path_type, path_length, LTTNG_PATH_MAX);
+ } else if (path_length == 0) {
+ ret = -EINVAL;
+ ERR("rotate rename \"%s\" path name has an illegal length of 0", path_type);
+ }
+ return ret;
+}
+
+/*
+ * relay_rotate_rename: rename the trace folder after a rotation is
+ * completed. We are not closing any fd here, just moving the folder, so it
+ * works even if data is still in-flight.
+ */
+static int relay_rotate_rename(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret;
+ ssize_t send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_generic_reply reply;
+ struct lttcomm_relayd_rotate_rename header;
+ size_t header_len;
+ size_t received_paths_size;
+ char *complete_old_path = NULL, *complete_new_path = NULL;
+ struct lttng_buffer_view old_path_view;
+ struct lttng_buffer_view new_path_view;
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to rename a trace folder before version check");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ ERR("relay_rotate_rename command is unsupported before LTTng 2.11");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ header_len = sizeof(header);
+ if (payload->size < header_len) {
+ ERR("Unexpected payload size in \"relay_rotate_rename\": expected >= %zu bytes, got %zu bytes",
+ header_len, payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ memcpy(&header, payload->data, header_len);
+
+ header.old_path_length = be32toh(header.old_path_length);
+ header.new_path_length = be32toh(header.new_path_length);
+ received_paths_size = header.old_path_length + header.new_path_length;
+
+ if (payload->size < header_len + received_paths_size) {
+ ERR("Unexpected payload size in \"relay_rotate_rename\" including paths: expected >= %zu bytes, got %zu bytes",
+ header_len, payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ /* Ensure the paths don't exceed their allowed size. */
+ ret = validate_rotate_rename_path_length("old", header.old_path_length);
+ if (ret) {
+ goto end;
+ }
+ ret = validate_rotate_rename_path_length("new", header.new_path_length);
+ if (ret) {
+ goto end;
+ }
+
+ old_path_view = lttng_buffer_view_from_view(payload, header_len,
+ header.old_path_length);
+ new_path_view = lttng_buffer_view_from_view(payload,
+ header_len + header.old_path_length,
+ header.new_path_length);
+
+ /* Validate that both paths received are NULL terminated. */
+ if (old_path_view.data[old_path_view.size - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"old\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+ if (new_path_view.data[new_path_view.size - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"new\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+
+ DBG("ROTATE_RENAME command has argument old path = \"%s\", new_path = \"%s\"",
+ old_path_view.data, new_path_view.data);
+ complete_old_path = create_output_path(old_path_view.data);
+ if (!complete_old_path) {
+ ERR("Failed to build old output path in rotate_rename command");
+ ret = -1;
+ goto end;
+ }
+
+ complete_new_path = create_output_path(new_path_view.data);
+ if (!complete_new_path) {
+ ERR("Failed to build new output path in rotate_rename command");
+ ret = -1;
+ goto end;
+ }
+ DBG("Expanded ROTATE_RENAME arguments to old path = \"%s\", new_path = \"%s\"",
+ complete_old_path, complete_new_path);
+
+ ret = utils_mkdir_recursive(complete_new_path, S_IRWXU | S_IRWXG,
+ -1, -1);
+ if (ret < 0) {
+ ERR("Failed to mkdir() rotate_rename's \"new\" output directory at \"%s\"",
+ complete_new_path);
+ goto end;
+ }
+
+ /*
+ * If a domain has not yet created its channel, the domain-specific
+ * folder might not exist, but this is not an error.
+ */
+ ret = rename(complete_old_path, complete_new_path);
+ if (ret < 0 && errno != ENOENT) {
+ PERROR("Renaming chunk in rotate_rename command from \"%s\" to \"%s\"",
+ complete_old_path, complete_new_path);
+ goto end;
+ }
+ ret = 0;
+
+end:
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(reply), 0);
+ if (send_ret < sizeof(reply)) {
+ ERR("Failed to send \"rotate rename\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+
+end_no_reply:
+ free(complete_old_path);
+ free(complete_new_path);
+ return ret;
+}
+
+/*
+ * Check if all the streams in the session have completed the last rotation.
+ * The chunk_id value is used to distinguish the cases where a stream was
+ * closed on the consumerd before the rotation started but it still active on
+ * the relayd, and the case where a stream appeared on the consumerd/relayd
+ * after the last rotation started (in that case, it is already writing in the
+ * new chunk folder).
+ */
+static
+int relay_rotate_pending(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_rotate_pending msg;
+ struct lttcomm_relayd_rotate_pending_reply reply;
+ struct lttng_ht_iter iter;
+ struct relay_stream *stream;
+ int ret = 0;
+ ssize_t send_ret;
+ uint64_t chunk_id;
+ bool rotate_pending = false;
+
+ DBG("Rotate pending command received");
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ ERR("Unsupported feature before 2.11");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (payload->size < sizeof(msg)) {
+ ERR("Unexpected payload size in \"relay_rotate_pending\": expected >= %zu bytes, got %zu bytes",
+ sizeof(msg), payload->size);
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ memcpy(&msg, payload->data, sizeof(msg));
+
+ chunk_id = be64toh(msg.chunk_id);
+
+ DBG("Evaluating rotate pending for session \"%s\" and chunk id %" PRIu64,
+ session->session_name, chunk_id);
+
+ /*
+ * Iterate over all the streams in the session and check if they are
+ * still waiting for data to perform their rotation.
+ */
+ rcu_read_lock();
+ cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
+ node.node) {
+ if (!stream_get(stream)) {
+ continue;
+ }
+ if (stream->trace->session != session) {
+ stream_put(stream);
+ continue;
+ }
+ pthread_mutex_lock(&stream->lock);
+ if (stream->rotate_at_seq_num != -1ULL) {
+ /* We have not yet performed the rotation. */
+ rotate_pending = true;
+ DBG("Stream %" PRIu64 " is still rotating",
+ stream->stream_handle);
+ } else if (stream->current_chunk_id.value <= chunk_id) {
+ /*
+ * Stream closed on the consumer but still active on the
+ * relay.
+ */
+ rotate_pending = true;
+ DBG("Stream %" PRIu64 " did not exist on the consumer "
+ "when the last rotation started, but is"
+ "still waiting for data before getting"
+ "closed",
+ stream->stream_handle);
+ }
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+ if (rotate_pending) {
+ goto send_reply;
+ }
+ }
+
+send_reply:
+ rcu_read_unlock();
+ memset(&reply, 0, sizeof(reply));
+ reply.generic.ret_code = htobe32((uint32_t) LTTNG_OK);
+ reply.is_pending = (uint8_t) !!rotate_pending;
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(reply), 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"rotate pending\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+
+end_no_reply:
+ return ret;
+}
+
+#define DBG_CMD(cmd_name, conn) \
+ DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd);
+
+static int relay_process_control_command(struct relay_connection *conn,
+ const struct lttcomm_relayd_hdr *header,
+ const struct lttng_buffer_view *payload)
+{
+ int ret = 0;
+
+ switch (header->cmd) {
+ case RELAYD_CREATE_SESSION:
+ DBG_CMD("RELAYD_CREATE_SESSION", conn);
+ ret = relay_create_session(header, conn, payload);
+ break;
+ case RELAYD_ADD_STREAM:
+ DBG_CMD("RELAYD_ADD_STREAM", conn);
+ ret = relay_add_stream(header, conn, payload);
+ break;
+ case RELAYD_START_DATA:
+ DBG_CMD("RELAYD_START_DATA", conn);
+ ret = relay_start(header, conn, payload);
+ break;
+ case RELAYD_SEND_METADATA:
+ DBG_CMD("RELAYD_SEND_METADATA", conn);
+ ret = relay_recv_metadata(header, conn, payload);
+ break;
+ case RELAYD_VERSION:
+ DBG_CMD("RELAYD_VERSION", conn);
+ ret = relay_send_version(header, conn, payload);
+ break;
+ case RELAYD_CLOSE_STREAM:
+ DBG_CMD("RELAYD_CLOSE_STREAM", conn);
+ ret = relay_close_stream(header, conn, payload);
+ break;
+ case RELAYD_DATA_PENDING:
+ DBG_CMD("RELAYD_DATA_PENDING", conn);
+ ret = relay_data_pending(header, conn, payload);
+ break;
+ case RELAYD_QUIESCENT_CONTROL:
+ DBG_CMD("RELAYD_QUIESCENT_CONTROL", conn);
+ ret = relay_quiescent_control(header, conn, payload);
+ break;
+ case RELAYD_BEGIN_DATA_PENDING:
+ DBG_CMD("RELAYD_BEGIN_DATA_PENDING", conn);
+ ret = relay_begin_data_pending(header, conn, payload);
+ break;
+ case RELAYD_END_DATA_PENDING:
+ DBG_CMD("RELAYD_END_DATA_PENDING", conn);
+ ret = relay_end_data_pending(header, conn, payload);
+ break;
+ case RELAYD_SEND_INDEX:
+ DBG_CMD("RELAYD_SEND_INDEX", conn);
+ ret = relay_recv_index(header, conn, payload);
+ break;
+ case RELAYD_STREAMS_SENT:
+ DBG_CMD("RELAYD_STREAMS_SENT", conn);
+ ret = relay_streams_sent(header, conn, payload);
+ break;
+ case RELAYD_RESET_METADATA:
+ DBG_CMD("RELAYD_RESET_METADATA", conn);
+ ret = relay_reset_metadata(header, conn, payload);
+ break;
+ case RELAYD_ROTATE_STREAM:
+ DBG_CMD("RELAYD_ROTATE_STREAM", conn);
+ ret = relay_rotate_session_stream(header, conn, payload);
+ break;
+ case RELAYD_ROTATE_RENAME:
+ DBG_CMD("RELAYD_ROTATE_RENAME", conn);
+ ret = relay_rotate_rename(header, conn, payload);
+ break;
+ case RELAYD_ROTATE_PENDING:
+ DBG_CMD("RELAYD_ROTATE_PENDING", conn);
+ ret = relay_rotate_pending(header, conn, payload);
+ break;
+ case RELAYD_MKDIR:
+ DBG_CMD("RELAYD_MKDIR", conn);
+ ret = relay_mkdir(header, conn, payload);
+ break;
+ case RELAYD_UPDATE_SYNC_INFO:
+ default:
+ ERR("Received unknown command (%u)", header->cmd);
+ relay_unknown_command(conn);
+ ret = -1;
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+static enum relay_connection_status relay_process_control_receive_payload(
+ struct relay_connection *conn)
+{
+ int ret = 0;
+ enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
+ struct lttng_dynamic_buffer *reception_buffer =
+ &conn->protocol.ctrl.reception_buffer;
+ struct ctrl_connection_state_receive_payload *state =
+ &conn->protocol.ctrl.state.receive_payload;
+ struct lttng_buffer_view payload_view;
+
+ if (state->left_to_receive == 0) {
+ /* Short-circuit for payload-less commands. */
+ goto reception_complete;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock,
+ reception_buffer->data + state->received,
+ state->left_to_receive, MSG_DONTWAIT);
+ if (ret < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK) {
+ PERROR("Unable to receive command payload on sock %d",
+ conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ }
+ goto end;
+ } else if (ret == 0) {
+ DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_CLOSED;
+ goto end;
+ }
+
+ assert(ret > 0);
+ assert(ret <= state->left_to_receive);
+
+ state->left_to_receive -= ret;
+ state->received += ret;
+
+ if (state->left_to_receive > 0) {
+ /*
+ * Can't transition to the protocol's next state, wait to
+ * receive the rest of the header.
+ */
+ DBG3("Partial reception of control connection protocol payload (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
+ state->received, state->left_to_receive,
+ conn->sock->fd);
+ goto end;
+ }
+
+reception_complete:
+ DBG("Done receiving control command payload: fd = %i, payload size = %" PRIu64 " bytes",
+ conn->sock->fd, state->received);
+ /*
+ * The payload required to process the command has been received.
+ * A view to the reception buffer is forwarded to the various
+ * commands and the state of the control is reset on success.
+ *
+ * Commands are responsible for sending their reply to the peer.
+ */
+ payload_view = lttng_buffer_view_from_dynamic_buffer(reception_buffer,
+ 0, -1);
+ ret = relay_process_control_command(conn,
+ &state->header, &payload_view);
+ if (ret < 0) {
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end;
+ }
+
+ ret = connection_reset_protocol_state(conn);
+ if (ret) {
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ }
+end:
+ return status;
+}
+
+static enum relay_connection_status relay_process_control_receive_header(
+ struct relay_connection *conn)
+{
+ int ret = 0;
+ enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
+ struct lttcomm_relayd_hdr header;
+ struct lttng_dynamic_buffer *reception_buffer =
+ &conn->protocol.ctrl.reception_buffer;
+ struct ctrl_connection_state_receive_header *state =
+ &conn->protocol.ctrl.state.receive_header;
+
+ assert(state->left_to_receive != 0);
+
+ ret = conn->sock->ops->recvmsg(conn->sock,
+ reception_buffer->data + state->received,
+ state->left_to_receive, MSG_DONTWAIT);
+ if (ret < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK) {
+ PERROR("Unable to receive control command header on sock %d",
+ conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ }
+ goto end;
+ } else if (ret == 0) {
+ DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_CLOSED;
+ goto end;
+ }
+
+ assert(ret > 0);
+ assert(ret <= state->left_to_receive);
+
+ state->left_to_receive -= ret;
+ state->received += ret;
+
+ if (state->left_to_receive > 0) {
+ /*
+ * Can't transition to the protocol's next state, wait to
+ * receive the rest of the header.
+ */
+ DBG3("Partial reception of control connection protocol header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
+ state->received, state->left_to_receive,
+ conn->sock->fd);
+ goto end;
+ }
+
+ /* Transition to next state: receiving the command's payload. */
+ conn->protocol.ctrl.state_id =
+ CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD;
+ memcpy(&header, reception_buffer->data, sizeof(header));
+ header.circuit_id = be64toh(header.circuit_id);
+ header.data_size = be64toh(header.data_size);
+ header.cmd = be32toh(header.cmd);
+ header.cmd_version = be32toh(header.cmd_version);
+ memcpy(&conn->protocol.ctrl.state.receive_payload.header,
+ &header, sizeof(header));
+
+ DBG("Done receiving control command header: fd = %i, cmd = %" PRIu32 ", cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes",
+ conn->sock->fd, header.cmd, header.cmd_version,
+ header.data_size);
+
+ if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) {
+ ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.",
+ header.data_size);
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end;
+ }
+
+ conn->protocol.ctrl.state.receive_payload.left_to_receive =
+ header.data_size;
+ conn->protocol.ctrl.state.receive_payload.received = 0;
+ ret = lttng_dynamic_buffer_set_size(reception_buffer,
+ header.data_size);
+ if (ret) {
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end;
+ }
+
+ if (header.data_size == 0) {
+ /*
+ * Manually invoke the next state as the poll loop
+ * will not wake-up to allow us to proceed further.
+ */
+ status = relay_process_control_receive_payload(conn);
+ }
+end:
+ return status;
+}
+
+/*
+ * Process the commands received on the control socket
+ */
+static enum relay_connection_status relay_process_control(
+ struct relay_connection *conn)
+{
+ enum relay_connection_status status;
+
+ switch (conn->protocol.ctrl.state_id) {
+ case CTRL_CONNECTION_STATE_RECEIVE_HEADER:
+ status = relay_process_control_receive_header(conn);
+ break;
+ case CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD:
+ status = relay_process_control_receive_payload(conn);
+ break;
+ default:
+ ERR("Unknown control connection protocol state encountered.");
+ abort();
+ }
+
+ return status;
+}
+
+/*
+ * Handle index for a data stream.
+ *
+ * Called with the stream lock held.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
+ bool rotate_index, bool *flushed, uint64_t total_size)
+{
+ int ret = 0;
+ uint64_t data_offset;
+ struct relay_index *index;
+
+ /* Get data offset because we are about to update the index. */
+ data_offset = htobe64(stream->tracefile_size_current);
+
+ DBG("handle_index_data: stream %" PRIu64 " net_seq_num %" PRIu64 " data offset %" PRIu64,
+ stream->stream_handle, net_seq_num, stream->tracefile_size_current);
+
+ /*
+ * Lookup for an existing index for that stream id/sequence
+ * number. If it exists, the control thread has already received the
+ * data for it, thus we need to write it to disk.
+ */
+ index = relay_index_get_by_id_or_create(stream, net_seq_num);
+ if (!index) {
+ ret = -1;
+ goto end;
+ }
+
+ if (rotate_index || !stream->index_file) {
+ const char *stream_path;