backport of fixes from balance-parents (#19012)

* fix replication crash on exit

* fix crash on exit due to incomplete begin-set-end v2

* fix incorrect check of function in rrdhost_status()

src/streaming/receiver.c

@@ -384,6 +384,10 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
         buffer->buffer[0] = '\0';
     }
 
+    // cleanup the sender buffer, because we may end-up reusing an incomplete buffer
+    sender_thread_buffer_free();
+    parser->user.v2.stream_buffer.wb = NULL;
+
     // make sure send_to_plugin() will not write any data to the socket
     spinlock_lock(&parser->writer.spinlock);
     parser->fd_output = -1;

src/streaming/replication.c

@@ -612,6 +612,7 @@ static struct replication_query *replication_response_prepare(
 }
 
 void replication_response_cancel_and_finalize(struct replication_query *q) {
+    if(!q) return;
     replication_query_finalize(NULL, q, false);
 }
 

src/streaming/rrdhost-status.c

@@ -224,7 +224,7 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {
 
     if(host == localhost)
         s->ingest.type = RRDHOST_INGEST_TYPE_LOCALHOST;
-    else if(has_receiver || rrdhost_flag_set(host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED))
+    else if(has_receiver || !rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED))
         s->ingest.type = RRDHOST_INGEST_TYPE_CHILD;
     else if(rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST))
         s->ingest.type = RRDHOST_INGEST_TYPE_VIRTUAL;