@cryptotaxi247 / netdata-1 / commits / c6438952c

Improve stale replication (#21357)

* Implement stuck replication detection and response handling * Fix checks and commit response for not found chart * Initialize replication stuck detection counter and refine replication completion conditions * Refine streaming response handling to prevent data loss during buffer overflow

Stelios Fragkakis committed Nov 30, 2025 at 20:54 UTC c6438952c2b4b393bf1f5c0a7e31c09478e8807a

4 files changed +171 -1

src/database/rrdset-index-id.c

@@ -103,6 +103,9 @@ static void rrdset_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v

103

104 rw_spinlock_init(&st->alerts.spinlock);

105

106 + // Initialize replication stuck detection counter

107 + st->replication_empty_response_count = 0;

108 +

109 // initialize the db tiers

110 {

111 for(size_t tier = 0; tier < nd_profile.storage_tiers; tier++) {

src/database/rrdset.h

@@ -228,6 +228,9 @@ struct rrdset {

228 } replay;

229 #endif // NETDATA_LOG_REPLICATION_REQUESTS

230

231 + // Replication stuck detection - outside debug flag for production safety

232 + uint8_t replication_empty_response_count; // track consecutive empty responses

233 +

234 SPINLOCK destroy_lock;

235 };

236

src/plugins.d/pluginsd_replication.c

+115

@@ -359,6 +359,9 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, si

359 ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARSER *parser) {

360 if (num_words < 7) { // accepts 7, but the 7th is optional

361 nd_log(NDLS_DAEMON, NDLP_ERR, "REPLAY: malformed " PLUGINSD_KEYWORD_REPLAY_END " command");

362 + RRDSET *st = pluginsd_get_scope_chart(parser);

363 + if(st)

364 + st->replication_empty_response_count = 0;

365 return PARSER_RC_ERROR;

366 }

367

@@ -402,6 +405,10 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS

405

406 parser->user.data_collections_count++;

407

408 + // Reset empty response counter when we receive actual data

409 + if(parser->user.replay.rset_enabled && st)

410 + st->replication_empty_response_count = 0;

411 +

412 if(parser->user.replay.rset_enabled && st->rrdhost->receiver) {

413 time_t now = now_realtime_sec();

414 time_t started = st->rrdhost->receiver->replication.first_time_s;

@@ -433,6 +440,9 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS

440 st->replay.log_next_data_collection = true;

441 #endif

442

443 + if(start_streaming)

444 + st->replication_empty_response_count = 0;

445 +

446 if (start_streaming) {

447 #ifdef REPLICATION_TRACKING

448 st->stream.rcv.who = REPLAY_WHO_FINISHED;

@@ -465,6 +475,111 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS

475 return PARSER_RC_OK;

476 }

477

478 + // ========================================================================

479 + // SAFETY NET: Detect stuck replication loops

480 + // ========================================================================

481 + //

482 + // We received start_streaming=false, which means we need to send another

483 + // replication request. However, we need to detect if we're stuck in an

484 + // infinite retry loop where no progress is being made.

485 + //

486 + // This can happen when:

487 + // 1. Parent already has newer data than child

488 + // 2. Child keeps splitting responses due to buffer constraints

489 + // 3. Network issues causing repeated empty/failed responses

490 +

491 + // Check parent's current retention to detect if we're already caught up

492 + time_t local_first_entry = 0, local_last_entry = 0;

493 + rrdset_get_retention_of_tier_for_collected_chart(

494 + st, &local_first_entry, &local_last_entry, now_realtime_sec(), 0);

495 +

496 + // Detect suspicious pattern: parent requested data but is already caught up

497 + // This indicates we're in a loop where child keeps splitting responses

498 + // even though parent doesn't need more data.

499 + bool parent_already_caught_up = (local_last_entry >= last_entry_child);

500 + bool requested_non_empty_range = (first_entry_requested != 0 || last_entry_requested != 0);

501 + bool is_suspicious_response = (requested_non_empty_range && parent_already_caught_up);

502 +

503 + bool should_check_for_stuck_replication = false;

504 +

505 + // Track consecutive suspicious responses - applies to all builds

506 + if(is_suspicious_response) {

507 + st->replication_empty_response_count++;

508 + // After 3 consecutive suspicious responses, we need to investigate

509 + if(st->replication_empty_response_count >= 3) {

510 + should_check_for_stuck_replication = true;

511 + }

512 + } else {

513 + // Reset counter if this was a legitimate response (parent still catching up)

514 + st->replication_empty_response_count = 0;

515 + }

516 +

517 + if (should_check_for_stuck_replication) {

518 + // We already have local_first_entry and local_last_entry from above

519 +

520 + // Check multiple conditions to ensure we're truly stuck:

521 + //

522 + // Condition 1: Parent has data that covers or exceeds child's retention

523 + // (We already checked this in parent_already_caught_up, but verify again)

524 + bool parent_has_equal_or_newer_data = (local_last_entry >= last_entry_child);

525 +

526 + // Calculate the gap for logging purposes

527 + time_t gap_to_child = (last_entry_child > local_last_entry) ?

528 + (last_entry_child - local_last_entry) : 0;

529 +

530 + // Condition 2: Parent's data is reasonably recent

531 + time_t wall_clock = now_realtime_sec();

532 + bool parent_data_is_recent = (local_last_entry > 0 &&

533 + (wall_clock - local_last_entry) < 300);

534 +

535 + // Only finish replication if parent has equal or newer data than child

536 + // Do NOT terminate if there's any gap, as that would cause data loss

537 + if (parent_has_equal_or_newer_data) {

538 +

539 + // Log with appropriate level based on confidence

540 + ND_LOG_FIELD_PRIORITY level = (parent_has_equal_or_newer_data && parent_data_is_recent) ?

541 + NDLP_INFO : NDLP_WARNING;

542 +

543 + nd_log(NDLS_DAEMON, level,

544 + "PLUGINSD REPLAY: 'host:%s/chart:%s' detected stuck replication loop. "

545 + "Parent last entry: %llu, Child last entry: %llu, Gap: %llu seconds, "

546 + "Empty responses: %u. Forcing replication to finish.",

547 + rrdhost_hostname(host), rrdset_id(st),

548 + (unsigned long long)local_last_entry,

549 + (unsigned long long)last_entry_child,

550 + (unsigned long long)gap_to_child,

551 + (unsigned int)st->replication_empty_response_count

552 + );

553 +

554 + st->replication_empty_response_count = 0;

555 +

556 + // IMPORTANT: Mark as finished and decrement counter NOW, before sending final request.

557 + // This prevents infinite loops even if child continues to respond with start_streaming=false.

558 + // The next REPLAY_END will see FINISHED flag and handle accordingly.

559 + RRDSET_FLAGS old = rrdset_flag_set_and_clear(

560 + st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED,

561 + RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS | RRDSET_FLAG_SYNC_CLOCK);

562 +

563 + if(!(old & RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED)) {

564 + if(rrdhost_receiver_replicating_charts_minus_one(st->rrdhost) == 0)

565 + pulse_host_status(host, PULSE_HOST_STATUS_RCV_RUNNING, 0);

566 + }

567 +

568 + pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END);

569 + host->stream.rcv.status.replication.percent = 100.0;

570 + worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, host->stream.rcv.status.replication.percent);

571 +

572 + // Send one final request to notify child. If child responds with start_streaming=true,

573 + // it will start streaming. If it responds with start_streaming=false, the next

574 + // REPLAY_END will see the FINISHED flag and log a warning but not loop forever.

575 + bool ok = replicate_chart_request(send_to_plugin, parser, host, st,

576 + first_entry_child, last_entry_child, child_world_time,

577 + 0, 0); // prev_wanted = 0,0 to trigger empty request path

578 +

579 + return ok ? PARSER_RC_OK : PARSER_RC_ERROR;

580 + }

581 + }

582 +

583 #ifdef REPLICATION_TRACKING

584 st->stream.rcv.who = REPLAY_WHO_ME;

585 #endif

src/streaming/stream-replication-sender.c

+50 -1

@@ -442,6 +442,16 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s

442

443 if(buffer_strlen(wb) > max_msg_size && last_end_time_in_buffer) {

444 q->query.before = last_end_time_in_buffer;

445 +

446 + // CRITICAL: When splitting a response due to buffer overflow, we MUST set

447 + // enable_streaming=false to prevent data loss. If we send start_streaming=true

448 + // with partial data, the parent will think replication is complete when we've

449 + // actually truncated the requested interval, causing permanent data loss.

450 + //

451 + // The parent-side stuck detection will handle infinite loops differently by:

452 + // 1. Detecting when no progress is made after multiple rounds

453 + // 2. Explicitly marking replication as FINISHED before sending final request

454 + // 3. Handling the child's response appropriately whether it says true or false

455 q->query.enable_streaming = false;

456

457 internal_error(

@@ -1198,8 +1208,47 @@ static bool replication_execute_request(struct replication_request *rq, bool wor

1208 if(!rq->st) {

1209 __atomic_add_fetch(&replication_globals.atomic.error_not_found, 1, __ATOMIC_RELAXED);

1210 nd_log(NDLS_DAEMON, NDLP_ERR,

1201 - "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' not found",

1211 + "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' not found, sending empty response to unblock parent",

1212 rrdhost_hostname(rq->sender->host), string2str(rq->chart_id));

1213 +

1214 + // CRITICAL: Parent is waiting for a response! We MUST send REPLAY_END even if chart not found

1215 + // Otherwise parent will wait forever with chart stuck in replicating state.

1216 + // Send empty response with start_streaming=true to finish replication for this non-existent chart.

1217 + BUFFER *wb = sender_thread_buffer(rq->sender, REPLICATION_THREAD_BUFFER_INITIAL_SIZE);

1218 +

1219 + bool with_slots = (rq->sender->capabilities & STREAM_CAP_SLOTS) ? true : false;

1220 + NUMBER_ENCODING integer_encoding = (rq->sender->capabilities & STREAM_CAP_IEEE754) ?

1221 + NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL;

1222 +

1223 + buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN, sizeof(PLUGINSD_KEYWORD_REPLAY_BEGIN) - 1);

1224 + if(with_slots) {

1225 + buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);

1226 + buffer_print_uint64_encoded(wb, integer_encoding, 0); // slot 0 for unknown chart

1227 + }

1228 + buffer_fast_strcat(wb, " '", 2);

1229 + buffer_fast_strcat(wb, string2str(rq->chart_id), string_strlen(rq->chart_id));

1230 + buffer_fast_strcat(wb, "'\n", 2);

1231 +

1232 + // Send REPLAY_END with empty data and start_streaming=true to unblock parent

1233 + buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_END " ", sizeof(PLUGINSD_KEYWORD_REPLAY_END) - 1 + 1);

1234 + buffer_print_int64_encoded(wb, integer_encoding, 0); // update_every

1235 + buffer_fast_strcat(wb, " ", 1);

1236 + buffer_print_uint64_encoded(wb, integer_encoding, 0); // db_first_entry

1237 + buffer_fast_strcat(wb, " ", 1);

1238 + buffer_print_uint64_encoded(wb, integer_encoding, 0); // db_last_entry

1239 + buffer_fast_strcat(wb, " true ", 7); // start_streaming=true (force finish)

1240 + buffer_print_uint64_encoded(wb, integer_encoding, 0); // after

1241 + buffer_fast_strcat(wb, " ", 1);

1242 + buffer_print_uint64_encoded(wb, integer_encoding, 0); // before

1243 + buffer_fast_strcat(wb, " ", 1);

1244 + buffer_print_uint64_encoded(wb, integer_encoding, now_realtime_sec()); // wall_clock_time

1245 + buffer_fast_strcat(wb, "\n", 1);

1246 +

1247 + sender_commit(rq->sender, wb, STREAM_TRAFFIC_TYPE_REPLICATION);

1248 + __atomic_add_fetch(&rq->sender->host->stream.snd.status.replication.counter_out, 1, __ATOMIC_RELAXED);

1249 + replication_replied_add();

1250 +

1251 + ret = true; // Consider this a successful response

1252 goto cleanup;

1253 }

1254 }