streaming compression, query planner and replication fixes (#14023)

* streaming compression, query planner and replication fixes

* remove journal v2 stats from global statistics

* disable sql for checking past sql UUIDs

* single threaded replication

* final replication thread using dictionaries and JudyL for sorting the pending requests

* do not timeout the sending socket when there are pending replication requests

* streaming receiver using read() instead of fread()

* remove FILE * from streaming - now using posix read() and write()

* increase timeouts to 10 minutes

* apply sender timeout only when there are metrics that are supposed to be streamed

* error handling in replication

* remove retries on socket read timeout; better error messages

* take into account inbound traffic too to detect that a connection is stale

* remove race conditions from replication thread

* make sure deleted entries are marked as executed, so that even if deletion fails, they will not be executed

* 2 minutes timeout to retry streaming to a parent that already has this node

* remove unecessary condition check

* fix compilation warnings

* include judy in replication

* wrappers to handle retries for SSL_read and SSL_write

* compressed bytes read monitoring

* recursive locks on replication to make it faster during flush or cleanup

* replication completion chart at the receiver side

* simplified recursive mutex

* simplified recursive mutex again

1 files changed, 221 insertions, 207 deletions

diff --git a/streaming/receiver.c b/streaming/receiver.c
index 40673f05b4..32b510e7ce 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -3,7 +3,12 @@
 #include "rrdpush.h"
 #include "parser/parser.h"
 
+// IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly
 #define WORKER_RECEIVER_JOB_BYTES_READ (WORKER_PARSER_FIRST_JOB - 1)
+#define WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED (WORKER_PARSER_FIRST_JOB - 2)
+
+// this has to be the same at parser.h
+#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 3)
 
 #if WORKER_PARSER_FIRST_JOB < 1
 #error The define WORKER_PARSER_FIRST_JOB needs to be at least 1
@@ -110,185 +115,182 @@ PARSER_RC streaming_claimed_id(char **words, size_t num_words, void *user)
     return PARSER_RC_OK;
 }
 
+static int read_stream(struct receiver_state *r, char* buffer, size_t size) {
+    if(unlikely(!size)) {
+        internal_error(true, "%s() asked to read zero bytes", __FUNCTION__);
+        return 0;
+    }
 
-#ifndef ENABLE_COMPRESSION
-/* The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing.
- */
-static int receiver_read(struct receiver_state *r, FILE *fp) {
 #ifdef ENABLE_HTTPS
-    if (r->ssl.conn && !r->ssl.flags) {
-        ERR_clear_error();
-        int desired = sizeof(r->read_buffer) - r->read_len - 1;
-        int ret = SSL_read(r->ssl.conn, r->read_buffer + r->read_len, desired);
-        if (ret > 0 ) {
-            r->read_len += ret;
-            worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret);
-            return 0;
-        }
-        // Don't treat SSL_ERROR_WANT_READ or SSL_ERROR_WANT_WRITE differently on blocking socket
-        u_long err;
-        char buf[256];
-        while ((err = ERR_get_error()) != 0) {
-            ERR_error_string_n(err, buf, sizeof(buf));
-            error("STREAM %s [receive from %s] ssl error: %s", r->hostname, r->client_ip, buf);
-        }
-        return 1;
-    }
+    if (r->ssl.conn && r->ssl.flags == NETDATA_SSL_HANDSHAKE_COMPLETE)
+        return (int)netdata_ssl_read(r->ssl.conn, buffer, size);
 #endif
-    if (!fgets(r->read_buffer, sizeof(r->read_buffer), fp))
-        return 1;
-    r->read_len = strlen(r->read_buffer);
-    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, r->read_len);
-    return 0;
-}
-#else
-/*
- * The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing.
- * if SSL encryption is on, then use SSL API for reading stream data.
- * Use line oriented fgets() in buffer from receiver_state is provided.
- * In other cases use fread to read binary data from socket.
- * Return zero on success and the number of bytes were read using pointer in the last argument.
- */
-static int read_stream(struct receiver_state *r, FILE *fp, char* buffer, size_t size, int* ret) {
-    if (!ret)
-        return 1;
-    *ret = 0;
-#ifdef ENABLE_HTTPS
-    if (r->ssl.conn && !r->ssl.flags) {
-        ERR_clear_error();
-        if (buffer != r->read_buffer + r->read_len) {
-            *ret = SSL_read(r->ssl.conn, buffer, size);
-            if (*ret > 0 ) 
-                return 0;
-        } else {
-            // we need to receive data with LF to parse compression header
-            size_t ofs = 0;
-            int res = 0;
-            errno = 0;
-            while (ofs < size) {
-                do {
-                    res = SSL_read(r->ssl.conn, buffer + ofs, 1);
-                    // When either SSL_ERROR_SYSCALL (OpenSSL < 3.0) or SSL_ERROR_SSL(OpenSSL > 3.0) happens,
-                    // the connection was lost https://www.openssl.org/docs/man3.0/man3/SSL_get_error.html,
-                    // without the test we will have an infinite loop https://github.com/netdata/netdata/issues/13092
-                    int local_ssl_err = SSL_get_error(r->ssl.conn, res);
-                    if (local_ssl_err == SSL_ERROR_SYSCALL || local_ssl_err == SSL_ERROR_SSL) {
-                        error("The SSL connection has error SSL_ERROR_SYSCALL(%d) and system is registering errno = %d",
-                              local_ssl_err, errno);
-                        return 1;
-                    }
-                } while (res == 0);
-
-                if (res < 0)
-                    break;
-                if (buffer[ofs] == '\n')
-                    break;
-                ofs += res;
-            }
-            if (res > 0) {
-                ofs += res;
-                *ret = ofs;
-                buffer[ofs] = 0;
-                return 0;
-            }
-        }
-        // Don't treat SSL_ERROR_WANT_READ or SSL_ERROR_WANT_WRITE differently on blocking socket
-        u_long err;
-        char buf[256];
-        while ((err = ERR_get_error()) != 0) {
-            ERR_error_string_n(err, buf, sizeof(buf));
-            error("STREAM %s [receive from %s] ssl error: %s", r->hostname, r->client_ip, buf);
-        }
-        return 1;
+
+    ssize_t bytes_read = read(r->fd, buffer, size);
+    if(bytes_read == 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS)) {
+        error("STREAM: %s(): timeout while waiting for data on socket!", __FUNCTION__);
+        bytes_read = -3;
     }
-#endif
-    if (buffer != r->read_buffer + r->read_len) {
-        // read to external buffer
-        *ret = fread(buffer, 1, size, fp);
-        if (!*ret)
-            return 1;
-    } else {
-        if (!fgets(r->read_buffer, sizeof(r->read_buffer), fp))
-            return 1;
-        *ret = strlen(r->read_buffer);
+    else if (bytes_read == 0) {
+        error("STREAM: %s(): EOF while reading data from socket!", __FUNCTION__);
+        bytes_read = -1;
     }
-    return 0;
+    else if (bytes_read < 0) {
+        error("STREAM: %s() failed to read from socket!", __FUNCTION__);
+        bytes_read = -2;
+    }
+
+//    do {
+//        bytes_read = (int) fread(buffer, 1, size, fp);
+//        if (unlikely(bytes_read <= 0)) {
+//            if(feof(fp)) {
+//                internal_error(true, "%s(): fread() failed with EOF", __FUNCTION__);
+//                bytes_read = -2;
+//            }
+//            else if(ferror(fp)) {
+//                internal_error(true, "%s(): fread() failed with ERROR", __FUNCTION__);
+//                bytes_read = -3;
+//            }
+//            else bytes_read = 0;
+//        }
+//        else
+//            worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, bytes_read);
+//    } while(bytes_read == 0);
+
+    return (int)bytes_read;
 }
 
-/*
- * Get the next line of data for parsing.
- * Return data from the decompressor buffer if available.
- * Otherwise read next line from the socket and check for compression header.
- * Return the line was read If no compression header was found.
- * Otherwise read the entire block of compressed data, decompress it
- * and return it in receiver_state buffer.
- * Return zero on success.
- */
-static int receiver_read(struct receiver_state *r, FILE *fp) {
-    // check any decompressed data  present
-    if (r->decompressor && r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) {
-        size_t available = sizeof(r->read_buffer) - r->read_len;
+static bool receiver_read_uncompressed(struct receiver_state *r) {
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(r->read_buffer[r->read_len] != '\0')
+        fatal("%s(): read_buffer does not start with zero", __FUNCTION__ );
+#endif
+
+    int bytes_read = read_stream(r, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1);
+    if(unlikely(bytes_read <= 0))
+        return false;
+
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)bytes_read);
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_read);
+
+    r->read_len += bytes_read;
+    r->read_buffer[r->read_len] = '\0';
+
+    return true;
+}
+
+#ifdef ENABLE_COMPRESSION
+static bool receiver_read_compressed(struct receiver_state *r) {
+
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(r->read_buffer[r->read_len] != '\0')
+        fatal("%s: read_buffer does not start with zero #2", __FUNCTION__ );
+#endif
+
+    // first use any available uncompressed data
+    if (r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) {
+        size_t available = sizeof(r->read_buffer) - r->read_len - 1;
         if (available) {
             size_t len = r->decompressor->get(r->decompressor, r->read_buffer + r->read_len, available);
             if (!len) {
-                internal_error(true, "decompressor returned zero length");
-                return 1;
+                internal_error(true, "decompressor returned zero length #1");
+                return false;
             }
 
-            r->read_len += len;
+            r->read_len += (int)len;
+            r->read_buffer[r->read_len] = '\0';
         }
-        return 0;
+        else
+            internal_error(true, "The line to read is too big! Already have %d bytes in read_buffer.", r->read_len);
+
+        return true;
     }
 
-    int ret = 0;
-    if (read_stream(r, fp, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1, &ret)) {
-        internal_error(true, "read_stream() failed (1).");
-        return 1;
+    // no decompressed data available
+    // read the compression signature of the next block
+
+    if(unlikely(r->read_len + r->decompressor->signature_size > sizeof(r->read_buffer) - 1)) {
+        internal_error(true, "The last incomplete line does not leave enough room for the next compression header! Already have %d bytes in read_buffer.", r->read_len);
+        return false;
     }
 
-    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret);
+    // read the compression signature from the stream
+    // we have to do a loop here, because read_stream() may return less than the data we need
+    int bytes_read = 0;
+    do {
+        int ret = read_stream(r, r->read_buffer + r->read_len + bytes_read, r->decompressor->signature_size - bytes_read);
+        if (unlikely(ret <= 0))
+            return false;
 
-    if (!is_compressed_data(r->read_buffer, ret)) {
-        r->read_len += ret;
-        return 0;
+        bytes_read += ret;
+    } while(unlikely(bytes_read < (int)r->decompressor->signature_size));
+
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)bytes_read);
+
+    if(unlikely(bytes_read != (int)r->decompressor->signature_size))
+        fatal("read %d bytes, but expected compression signature of size %zu", bytes_read, r->decompressor->signature_size);
+
+    size_t compressed_message_size = r->decompressor->start(r->decompressor, r->read_buffer + r->read_len, bytes_read);
+    if (unlikely(!compressed_message_size)) {
+        internal_error(true, "multiplexed uncompressed data in compressed stream!");
+        r->read_len += bytes_read;
+        r->read_buffer[r->read_len] = '\0';
+        return true;
+    }
+
+    if(unlikely(compressed_message_size > COMPRESSION_MAX_MSG_SIZE)) {
+        error("received a compressed message of %zu bytes, which is bigger than the max compressed message size supported of %zu. Ignoring message.",
+              compressed_message_size, (size_t)COMPRESSION_MAX_MSG_SIZE);
+        return false;
     }
 
-    if (unlikely(!r->decompressor)) 
-        r->decompressor = create_decompressor();
-    
-    size_t bytes_to_read = r->decompressor->start(r->decompressor, r->read_buffer, ret);
+    // delete compression header from our read buffer
+    r->read_buffer[r->read_len] = '\0';
 
-    // Read the entire block of compressed data because
-    // we're unable to decompress incomplete block
-    char compressed[bytes_to_read];
+    // Read the entire compressed block of compressed data
+    char compressed[compressed_message_size];
+    size_t compressed_bytes_read = 0;
     do {
-        if (read_stream(r, fp, compressed, bytes_to_read, &ret)) {
-            internal_error(true, "read_stream() failed (2).");
-            return 1;
+        size_t start = compressed_bytes_read;
+        size_t remaining = compressed_message_size - start;
+
+        int last_read_bytes = read_stream(r, &compressed[start], remaining);
+        if (unlikely(last_read_bytes <= 0)) {
+            internal_error(true, "read_stream() failed #2, with code %d", last_read_bytes);
+            return false;
         }
 
-        worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret);
+        compressed_bytes_read += last_read_bytes;
 
-        // Send input data to decompressor
-        if (ret)
-            r->decompressor->put(r->decompressor, compressed, ret);
+    } while(unlikely(compressed_message_size > compressed_bytes_read));
 
-        bytes_to_read -= ret;
-    } while (bytes_to_read > 0);
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)compressed_bytes_read);
 
-    // Decompress
-    size_t bytes_to_parse = r->decompressor->decompress(r->decompressor);
+    // decompress the compressed block
+    size_t bytes_to_parse = r->decompressor->decompress(r->decompressor, compressed, compressed_bytes_read);
     if (!bytes_to_parse) {
         internal_error(true, "no bytes to parse.");
-        return 1;
+        return false;
     }
 
-    // Fill read buffer with decompressed data
-    r->read_len = r->decompressor->get(r->decompressor, r->read_buffer, sizeof(r->read_buffer));
-    return 0;
-}
+    worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_to_parse);
 
-#endif
+    // fill read buffer with decompressed data
+    size_t len = (int)r->decompressor->get(r->decompressor, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1);
+    if (!len) {
+        internal_error(true, "decompressor returned zero length #2");
+        return false;
+    }
+    r->read_len += (int)len;
+    r->read_buffer[r->read_len] = '\0';
+
+    return true;
+}
+#else // !ENABLE_COMPRESSION
+static bool receiver_read_compressed(struct receiver_state *r) {
+    return receiver_read_uncompressed(r);
+}
+#endif // ENABLE_COMPRESSION
 
 /* Produce a full line if one exists, statefully return where we start next time.
  * When we hit the end of the buffer with a partial line move it to the beginning for the next fill.
@@ -302,7 +304,10 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b
     char *de = &buffer[buffer_length - 2];
 
     if(ss >= se) {
+        *ds = '\0';
+        *pos = 0;
         r->read_len = 0;
+        r->read_buffer[r->read_len] = '\0';
         return NULL;
     }
 
@@ -333,6 +338,9 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b
     // move everything to the beginning
     memmove(r->read_buffer, &r->read_buffer[start], r->read_len - start);
     r->read_len -= (int)start;
+    r->read_buffer[r->read_len] = '\0';
+    *ds = '\0';
+    *pos = 0;
     return NULL;
 }
 
@@ -342,7 +350,7 @@ static void streaming_parser_thread_cleanup(void *ptr) {
     parser_destroy(parser);
 }
 
-static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp_in, FILE *fp_out, void *ssl) {
+static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) {
     size_t result;
 
     PARSER_USER_OBJECT user = {
@@ -353,7 +361,7 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, F
         .trust_durations = 1
     };
 
-    PARSER *parser = parser_init(rpt->host, &user, fp_in, fp_out, PARSER_INPUT_SPLIT, ssl);
+    PARSER *parser = parser_init(rpt->host, &user, NULL, NULL, fd, PARSER_INPUT_SPLIT, ssl);
 
     rrd_collector_started();
 
@@ -365,36 +373,56 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, F
 
     user.parser = parser;
 
+    bool compressed_connection = false;
 #ifdef ENABLE_COMPRESSION
-    if (rpt->decompressor)
-        rpt->decompressor->reset(rpt->decompressor);
+    if(stream_has_capability(rpt, STREAM_CAP_COMPRESSION)) {
+        compressed_connection = true;
+
+        if (!rpt->decompressor)
+            rpt->decompressor = create_decompressor();
+        else
+            rpt->decompressor->reset(rpt->decompressor);
+    }
 #endif
 
-    char buffer[PLUGINSD_LINE_MAX + 2];
-    do {
-        if(receiver_read(rpt, fp_in)) break;
+    rpt->read_buffer[0] = '\0';
+    rpt->read_len = 0;
 
-        size_t pos = 0;
-        while(receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &pos)) {
-            if(unlikely(netdata_exit)) {
-                internal_error(true, "exiting...");
-                goto done;
-            }
-            if(unlikely(rpt->shutdown)) {
-                internal_error(true, "parser shutdown...");
-                goto done;
-            }
-            if (unlikely(parser_action(parser,  buffer))) {
-                internal_error(true, "parser_action() failed...");
-                goto done;
-            }
+    size_t read_buffer_start = 0;
+    char buffer[PLUGINSD_LINE_MAX + 2] = "";
+    while(!netdata_exit) {
+        if(!receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &read_buffer_start)) {
+            bool have_new_data;
+            if(compressed_connection)
+                have_new_data = receiver_read_compressed(rpt);
+            else
+                have_new_data = receiver_read_uncompressed(rpt);
+
+            if(!have_new_data)
+                break;
+
+            rpt->last_msg_t = now_realtime_sec();
+            continue;
         }
 
-        rpt->last_msg_t = now_realtime_sec();
+        if(unlikely(netdata_exit)) {
+            internal_error(true, "exiting...");
+            goto done;
+        }
+        if(unlikely(rpt->shutdown)) {
+            internal_error(true, "parser shutdown...");
+            goto done;
+        }
+
+        if (unlikely(parser_action(parser,  buffer))) {
+            internal_error(true, "parser_action() failed on keyword '%s'.", buffer);
+            break;
+        }
     }
-    while(!netdata_exit);
 
 done:
+    internal_error(true, "Streaming receiver thread stopping...");
+
     result = user.count;
 
     // free parser with the pop function
@@ -644,42 +672,11 @@ static int rrdpush_receive(struct receiver_state *rpt)
         error("STREAM %s [receive from [%s]:%s]: cannot remove the non-blocking flag from socket %d", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd);
 
     struct timeval timeout;
-    timeout.tv_sec = 120;
+    timeout.tv_sec = 600;
     timeout.tv_usec = 0;
     if (unlikely(setsockopt(rpt->fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout) != 0))
         error("STREAM %s [receive from [%s]:%s]: cannot set timeout for socket %d", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd);
 
-    // convert the socket to a FILE *
-    // It seems that the same FILE * cannot be used for both reading and writing.
-    // (reads and writes seem to interfere with each other, with undefined results).
-
-    int fd_in = rpt->fd;
-    int fd_out = fcntl(rpt->fd, F_DUPFD_CLOEXEC, 0);
-    if(fd_out == -1) {
-        error("STREAM %s [receive from [%s]:%s]: failed to duplicate FD %d.", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd);
-        log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rrdhost_hostname(rpt->host), "FAILED - SOCKET ERROR");
-        close(fd_in);
-        return 0;
-    }
-
-    FILE *fp_out = fdopen(fd_out, "w");
-    if(!fp_out) {
-        error("STREAM %s [receive from [%s]:%s]: failed to get a FILE pointer for fd_out %d.", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd);
-        log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rrdhost_hostname(rpt->host), "FAILED - SOCKET ERROR");
-        close(fd_in);
-        close(fd_out);
-        return 0;
-    }
-
-    FILE *fp_in  = fdopen(fd_in, "r");
-    if(!fp_in) {
-        error("STREAM %s [receive from [%s]:%s]: failed to get a FILE pointer for fd_in %d.", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd);
-        log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rrdhost_hostname(rpt->host), "FAILED - SOCKET ERROR");
-        close(fd_in);
-        fclose(fp_out);
-        return 0;
-    }
-
     rrdhost_wrlock(rpt->host);
 /* if(rpt->host->connected_senders > 0) {
         rrdhost_unlock(rpt->host);
@@ -724,12 +721,19 @@ static int rrdpush_receive(struct receiver_state *rpt)
 
     rrdhost_set_is_parent_label(++localhost->senders_count);
 
-    rrdcontext_host_child_connected(rpt->host);
+    if(stream_has_capability(rpt->host->receiver, STREAM_CAP_REPLICATION)) {
+        RRDSET *st;
+        rrdset_foreach_read(st, rpt->host) {
+            rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS | RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
+        }
+        rrdset_foreach_done(st);
+    }
 
+    rrdcontext_host_child_connected(rpt->host);
 
     rrdhost_flag_clear(rpt->host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED);
 
-    size_t count = streaming_parser(rpt, &cd, fp_in, fp_out,
+    size_t count = streaming_parser(rpt, &cd, rpt->fd,
 #ifdef ENABLE_HTTPS
                                     (rpt->ssl.conn) ? &rpt->ssl : NULL
 #else
@@ -746,6 +750,15 @@ static int rrdpush_receive(struct receiver_state *rpt)
     error("STREAM %s [receive from [%s]:%s]: disconnected (completed %zu updates).",
           rpt->hostname, rpt->client_ip, rpt->client_port, count);
 
+    if(stream_has_capability(rpt->host->receiver, STREAM_CAP_REPLICATION)) {
+        RRDSET *st;
+        rrdset_foreach_read(st, rpt->host) {
+            rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
+            rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
+        }
+        rrdset_foreach_done(st);
+    }
+
     rrdcontext_host_child_disconnected(rpt->host);
 
 #ifdef ENABLE_ACLK
@@ -779,8 +792,7 @@ static int rrdpush_receive(struct receiver_state *rpt)
     }
 
     // cleanup
-    fclose(fp_in);
-    fclose(fp_out);
+    close(rpt->fd);
     return (int)count;
 }
 
@@ -791,7 +803,9 @@ void *rrdpush_receiver_thread(void *ptr) {
     info("STREAM %s [%s]:%s: receive thread created (task id %d)", rpt->hostname, rpt->client_ip, rpt->client_port, gettid());
 
     worker_register("STREAMRCV");
-    worker_register_job_custom_metric(WORKER_RECEIVER_JOB_BYTES_READ, "received bytes", "bytes/s", WORKER_METRIC_INCREMENTAL);
+    worker_register_job_custom_metric(WORKER_RECEIVER_JOB_BYTES_READ, "received bytes", "bytes/s", WORKER_METRIC_INCREMENT);
+    worker_register_job_custom_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, "uncompressed bytes", "bytes/s", WORKER_METRIC_INCREMENT);
+    worker_register_job_custom_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, "replication completion", "%", WORKER_METRIC_ABSOLUTE);
     rrdpush_receive(rpt);
     worker_unregister();