@cryptotaxi247 / netdata-1 / commits / e82c14760

Enhance error handling for journal v2 migration (#21514)

* Improve metric deletion handling in migration and cache processing - Add checks to safely handle deleted metrics during page migration and cache operations. - Return NULL for invalid metric duplications in `mrg_metric_dup`. - Ensure proper resource cleanup and state marking when encountering deleted metrics. Enhance error handling for journal v2 migration: skip metrics with NULL UUID * Additional checks * Add null checks for `STORAGE_COLLECT_HANDLE` and reorganize `pdc` initialization logic - Safely return when `STORAGE_COLLECT_HANDLE` is null in key storage engine methods. - Reorder and streamline `pdc` initialization to avoid redundancy and improve clarity. * Add `pdc` cleanup in error path to prevent potential resource leaks * Add internal checks and cleanup logic to metric duplication error path * Validate UUID early in journal v2 migration and reduce redundancy in error paths - Add early UUID validity checks during metric migration to avoid unnecessary operations. - Adjust inflight query counters in `pagecache` on query completion for accurate tracking.

Stelios Fragkakis committed Jan 8, 2026 at 19:18 UTC e82c14760a5e6b4cc9cf1e12897576c3c414da5a

6 files changed +55 -6

src/database/engine/cache.c

+16 -1

@@ -2482,6 +2482,21 @@ void pgc_open_cache_to_journal_v2(

2482 }

2483

2484 METRIC *metric = mrg_metric_dup(main_mrg, (METRIC *)page->metric_id);

2485 + if(!metric) {

2486 + // metric has been deleted, skip this page

2487 + page_transition_unlock(cache, page);

2488 + page_release(cache, page, false);

2489 + continue;

2490 + }

2491 +

2492 + // Check UUID validity early, before any JudyL modifications

2493 + nd_uuid_t *uuid = mrg_metric_uuid(main_mrg, metric);

2494 + if (unlikely(!uuid)) {

2495 + mrg_metric_release(main_mrg, metric);

2496 + page_transition_unlock(cache, page);

2497 + page_release(cache, page, false);

2498 + continue;

2499 + }

2500

2501 page_flag_set(page, PGC_PAGE_IS_BEING_MIGRATED_TO_V2);

2502

@@ -2524,7 +2539,7 @@ void pgc_open_cache_to_journal_v2(

2539 if(!*PValue) {

2540 mi = aral_mallocz(ar_mi);

2541 mi->metric = metric;

2527 - mi->uuid = mrg_metric_uuid(main_mrg, metric);

2542 + mi->uuid = uuid;

2543 mi->first_time_s = page->start_time_s;

2544 mi->last_time_s = page->end_time_s;

2545 mi->number_of_pages = 1;

src/database/engine/mrg.c

+3 -1

@@ -124,7 +124,9 @@ bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) {

124

125 ALWAYS_INLINE

126 METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {

127 - metric_acquire(mrg, metric);

127 + if(!metric_acquire(mrg, metric))

128 + return NULL;

129 +

130 return metric;

131 }

132

src/database/engine/pagecache.c

+15 -3

@@ -863,14 +863,26 @@ ALWAYS_INLINE_HOT void pg_cache_preload(struct rrdeng_query_handle *handle) {

863 __atomic_add_fetch(&handle->ctx->atomic.inflight_queries, 1, __ATOMIC_RELAXED);

864 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.currently_running_queries, 1, __ATOMIC_RELAXED);

865 handle->pdc = pdc_get();

866 + handle->pdc->ctx = handle->ctx;

867 + handle->pdc->refcount = 1;

868 + spinlock_init(&handle->pdc->refcount_spinlock);

869 handle->pdc->metric = mrg_metric_dup(main_mrg, handle->metric);

870 + if(!handle->pdc->metric) {

871 + // metric has been deleted, mark completions and return

872 + completion_init(&handle->pdc->prep_completion);

873 + completion_init(&handle->pdc->page_completion);

874 + completion_mark_complete(&handle->pdc->prep_completion);

875 + completion_mark_complete(&handle->pdc->page_completion);

876 + pdc_release_and_destroy_if_unreferenced(handle->pdc, true, true);

877 + handle->pdc = NULL;

878 + __atomic_sub_fetch(&handle->ctx->atomic.inflight_queries, 1, __ATOMIC_RELAXED);

879 + __atomic_sub_fetch(&rrdeng_cache_efficiency_stats.currently_running_queries, 1, __ATOMIC_RELAXED);

880 + return;

881 + }

882 handle->pdc->start_time_s = handle->start_time_s;

883 handle->pdc->end_time_s = handle->end_time_s;

884 handle->pdc->priority = handle->priority;

885 handle->pdc->optimal_end_time_s = handle->end_time_s;

871 - handle->pdc->ctx = handle->ctx;

872 - handle->pdc->refcount = 1;

873 - spinlock_init(&handle->pdc->refcount_spinlock);

886 completion_init(&handle->pdc->prep_completion);

887 completion_init(&handle->pdc->page_completion);

888

src/database/engine/pdc.c

+3 -1

@@ -388,7 +388,9 @@ static bool epdl_check_if_pages_are_already_in_cache(struct rrdengine_instance *

388 // PDC logic

389

390 static ALWAYS_INLINE void pdc_destroy(PDC *pdc) {

391 - mrg_metric_release(main_mrg, pdc->metric);

391 + if(pdc->metric)

392 + mrg_metric_release(main_mrg, pdc->metric);

393 +

394 completion_destroy(&pdc->prep_completion);

395 completion_destroy(&pdc->page_completion);

396

src/database/engine/rrdengineapi.c

@@ -283,6 +283,15 @@ STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *smh, uin

283 #endif

284

285 metric = mrg_metric_dup(main_mrg, metric);

286 + if(!metric) {

287 +#ifdef NETDATA_INTERNAL_CHECKS

288 + if(is_1st_metric_writer)

289 + mrg_metric_clear_writer(main_mrg, (METRIC *)smh);

290 + else

291 + __atomic_sub_fetch(&ctx->atomic.collectors_running_duplicate, 1, __ATOMIC_RELAXED);

292 +#endif

293 + return NULL;

294 + }

295

296 struct rrdeng_collect_handle *handle;

297

src/database/storage-engine.h

@@ -168,6 +168,9 @@ static void storage_engine_store_metric(

168 STORAGE_COLLECT_HANDLE *sch, usec_t point_in_time_ut,

169 NETDATA_DOUBLE n, NETDATA_DOUBLE min_value, NETDATA_DOUBLE max_value,

170 uint16_t count, uint16_t anomaly_count, SN_FLAGS flags) {

171 + if(unlikely(!sch))

172 + return;

173 +

174 internal_fatal(!is_valid_backend(sch->seb), "STORAGE: invalid backend");

175

176 #ifdef ENABLE_DBENGINE

@@ -288,6 +291,9 @@ int rrddim_collect_finalize(STORAGE_COLLECT_HANDLE *sch);

291 // returns 1 if it's safe to delete the dimension

292

293 static inline int storage_engine_store_finalize(STORAGE_COLLECT_HANDLE *sch) {

294 + if(unlikely(!sch))

295 + return 1; // safe to delete if no handle

296 +

297 internal_fatal(!is_valid_backend(sch->seb), "STORAGE: invalid backend");

298

299 #ifdef ENABLE_DBENGINE

@@ -304,6 +310,9 @@ void rrdeng_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *sch

310 void rrddim_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *sch, int update_every);

311

312 static inline void storage_engine_store_change_collection_frequency(STORAGE_COLLECT_HANDLE *sch, int update_every) {

313 + if(unlikely(!sch))

314 + return;

315 +

316 internal_fatal(!is_valid_backend(sch->seb), "STORAGE: invalid backend");

317

318 #ifdef ENABLE_DBENGINE