Improve context snapshots needed on ACLK connect (#21904)
* Add deferred context checkpoint handling in ACLK - Introduce deferred checkpoint logic to handle pending context loads and post-processing. - Log detailed reasons and timestamps for deferred checkpoints. - Update `aclk_sync_cfg_t` struct with additional fields for deferred checkpoint tracking. - Ensure proper handling of deferred counts and reasons upon context activation. * Refactor deferred context checkpoint logic in ACLK - Extract deferred checkpoint handling into a new `rrdcontext_checkpoint_defer()` function. - Introduce limits for maximum deferrals and cumulative deferral time. - Update `aclk_sync_cfg_t` to enhance tracking of deferred checkpoints with additional fields. - Improve logging to include detailed deferral reasons and time-related data. * Simplify deferred checkpoint handling by removing deferral limits and consolidating logic. Add support for replaying deferred checkpoints post context processing. * Add spinlock to protect pending context checkpoint handling in ACLK - Introduce `pending_ctx_spinlock` to ensure thread-safe access to pending checkpoint fields. - Lock critical sections in `rrdcontext_checkpoint_defer()` and checkpoint processing to prevent race conditions. - Initialize spinlock during ACLK sync configuration setup. * Improve deferred checkpoint handling, add timeout-based replay - Return success/failure state from `rrdcontext_checkpoint_save_pending()`. - Replay deferred checkpoints if post-processing queue is drained or on timeout. - Add `pending_ctx_saved_time_s` for tracking checkpoint save time. - Enhance spinlock safety and logging for deferred checkpoint replays and timeouts. * Add null check for `old_aclk_host_config` in ACLK cleanup logic * Pause incremental context streaming during deferred checkpoint replay, improve pending checkpoint replay logic - Ensure context streaming is paused until deferred checkpoints are replayed to avoid inconsistencies. - Add checks for pending context loads in checkpoint replay logic. - Allow timed-out deferred checkpoints to replay even if context loading is stuck. - Update logging to better reflect ongoing context activities during timeouts. * Add checkpoint generation tracking for improved deferred context handling - Introduce `pending_ctx_generation` to track checkpoint state changes. - Replace `pending_ctx_saved_time_s` with monotonic time for robustness in timeouts. - Add logic to validate stale checkpoints during execution and snapshot generation. - Enhance logging to include checkpoint generation details and invalidation notices. - Refactor and centralize checkpoint clearing logic with `rrdcontext_checkpoint_clear_pending_unsafe()`. * Switch to atomic operations for pending context generation handling - Replace non-atomic increments and loads with `__atomic_add_fetch` and `__atomic_load_n` to ensure thread-safe updates and accesses. - Maintain synchronization and prevent potential race conditions in checkpoint generation logic. * Improve thread safety and atomicity in ACLK pending context checkpoint handling - Adjust spinlock placement for correct synchronization during streaming pause. - Ensure consistent loading of `aclk_host_config` using atomic operations. - Add conditional logic to safely enable context streaming based on checkpoint state. * Reorder `aclk_host_config` atomic load for proper synchronization in context streaming logic * Downgrade checkpoint logging level to `DEBUG` for stale checkpoint skips in RRDCONTEXT.
Stelios Fragkakis committed
Apr 22, 2026 at 09:20 UTC
f58ecaef008a07879bac572775a63a49c097751d
5 files changed
+252
-26
src/database/contexts/rrdcontext-worker.c
+6
@@ -1072,6 +1072,9 @@ void rrdcontext_main(void *ptr) {
1072
dfe_start_reentrant(rrdhost_root_index, host) {
1073
if(unlikely(!service_running(SERVICE_CONTEXT))) break;
1074
1075
+ // Allow timed-out deferred checkpoints to replay even if context loading is stuck.
1076
+ rrdcontext_hub_pending_checkpoint_replay(host);
1077
+
1078
if(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))
1079
continue;
1080
@@ -1085,6 +1088,9 @@ void rrdcontext_main(void *ptr) {
1088
pp_queued_contexts_for_all_hosts += rrdcontext_queue_entries(&host->rrdctx.pp_queue);
1089
rrdcontext_post_process_queued_contexts(host);
1090
1091
+ // replay deferred checkpoint if post-processing drained the queue, or on timeout
1092
+ rrdcontext_hub_pending_checkpoint_replay(host);
1093
+
1094
hub_queued_contexts_for_all_hosts += rrdcontext_queue_entries(&host->rrdctx.hub_queue);
1095
rrdcontext_dispatch_queued_contexts_to_hub(host, now_ut);
1096
src/database/contexts/rrdcontext.c
+217
-26
@@ -1,6 +1,7 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
#include "rrdcontext-internal.h"
4
+#include "../sqlite/sqlite_aclk.h"
5
6
// ----------------------------------------------------------------------------
7
// visualizing flags
@@ -200,35 +201,81 @@ int rrdcontext_foreach_instance_with_rrdset_in_context(RRDHOST *host, const char
201
// ----------------------------------------------------------------------------
202
// ACLK interface
203
203
-void rrdcontext_hub_checkpoint_command(void *ptr) {
204
- struct ctxs_checkpoint *cmd = ptr;
204
+static void rrdcontext_checkpoint_clear_pending_unsafe(struct aclk_sync_cfg_t *aclk_host_config) {
205
+ freez(aclk_host_config->pending_ctx_claim_id);
206
+ freez(aclk_host_config->pending_ctx_node_id);
207
+ aclk_host_config->pending_ctx_claim_id = NULL;
208
+ aclk_host_config->pending_ctx_node_id = NULL;
209
+ aclk_host_config->pending_ctx_version_hash = 0;
210
+ aclk_host_config->pending_ctx_saved_monotonic_s = 0;
211
+ __atomic_store_n(&aclk_host_config->pending_ctx_checkpoint, false, __ATOMIC_RELEASE);
212
+}
213
206
- if(!claim_id_matches(cmd->claim_id)) {
207
- CLAIM_ID claim_id = claim_id_get();
208
- nd_log(NDLS_DAEMON, NDLP_WARNING,
209
- "RRDCONTEXT: received checkpoint command for claim_id '%s', node id '%s', "
210
- "but this is not our claim id. Ours '%s', received '%s'. Ignoring command.",
211
- cmd->claim_id, cmd->node_id,
212
- claim_id.str, cmd->claim_id);
214
+static uint64_t rrdcontext_checkpoint_invalidate_pending(RRDHOST *host) {
215
+ struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
216
+ if(!aclk_host_config)
217
+ return 0;
218
214
- return;
215
- }
219
+ spinlock_lock(&aclk_host_config->pending_ctx_spinlock);
220
+ uint64_t generation = __atomic_add_fetch(&aclk_host_config->pending_ctx_generation, 1, __ATOMIC_RELEASE);
221
+ rrdcontext_checkpoint_clear_pending_unsafe(aclk_host_config);
222
+ spinlock_unlock(&aclk_host_config->pending_ctx_spinlock);
223
217
- RRDHOST *host = rrdhost_find_by_node_id(cmd->node_id);
218
- if(!host) {
219
- nd_log(NDLS_DAEMON, NDLP_WARNING,
220
- "RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', "
221
- "but there is no node with such node id here. Ignoring command.",
222
- cmd->claim_id, cmd->node_id);
224
+ return generation;
225
+}
226
+
227
+static bool rrdcontext_checkpoint_generation_is_current(RRDHOST *host, const char *claim_id, const char *node_id, uint64_t generation) {
228
+ if(!generation)
229
+ return true;
230
+
231
+ struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
232
+ uint64_t current_generation = aclk_host_config ?
233
+ __atomic_load_n(&aclk_host_config->pending_ctx_generation, __ATOMIC_ACQUIRE) :
234
+ 0;
235
+
236
+ if(likely(aclk_host_config && current_generation == generation))
237
+ return true;
238
+
239
+ nd_log(NDLS_DAEMON, NDLP_DEBUG,
240
+ "RRDCONTEXT: skipping stale checkpoint for host '%s', claim id '%s', node id '%s' "
241
+ "(generation %"PRIu64", current %"PRIu64").",
242
+ rrdhost_hostname(host), claim_id, node_id, generation, current_generation);
243
+ return false;
244
+}
245
+
246
+// Save a pending checkpoint to be replayed when context processing completes.
247
+// Returns true if saved successfully, false if save failed (caller should execute immediately).
248
+static bool rrdcontext_checkpoint_save_pending(RRDHOST *host, struct ctxs_checkpoint *cmd) {
249
+ struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);
250
+ if(!aclk_host_config)
251
+ return false;
252
+
253
+ spinlock_lock(&aclk_host_config->pending_ctx_spinlock);
254
255
+ // Pause incremental context streaming until the deferred checkpoint is replayed.
256
+ rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
257
+
258
+ __atomic_add_fetch(&aclk_host_config->pending_ctx_generation, 1, __ATOMIC_RELEASE);
259
+ rrdcontext_checkpoint_clear_pending_unsafe(aclk_host_config);
260
+ aclk_host_config->pending_ctx_claim_id = strdupz(cmd->claim_id);
261
+ aclk_host_config->pending_ctx_node_id = strdupz(cmd->node_id);
262
+ aclk_host_config->pending_ctx_version_hash = cmd->version_hash;
263
+ aclk_host_config->pending_ctx_saved_monotonic_s = now_monotonic_sec();
264
+ __atomic_store_n(&aclk_host_config->pending_ctx_checkpoint, true, __ATOMIC_RELEASE);
265
+ spinlock_unlock(&aclk_host_config->pending_ctx_spinlock);
266
+ return true;
267
+}
268
+
269
+// Execute the checkpoint: compare version hash, send snapshot if needed, enable streaming.
270
+static void rrdcontext_checkpoint_execute(RRDHOST *host, const char *claim_id, const char *node_id, uint64_t version_hash, uint64_t generation) {
271
+ if(!rrdcontext_checkpoint_generation_is_current(host, claim_id, node_id, generation))
272
return;
225
- }
273
274
if(rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS)) {
275
nd_log(NDLS_DAEMON, NDLP_NOTICE,
229
- "RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', "
276
+ "RRDCONTEXT: checkpoint for claim id '%s', node id '%s', "
277
"while node '%s' has an active context streaming.",
231
- cmd->claim_id, cmd->node_id, rrdhost_hostname(host));
278
+ claim_id, node_id, rrdhost_hostname(host));
279
280
// disable it temporarily, so that our worker will not attempt to send messages in parallel
281
rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
@@ -236,16 +283,16 @@ void rrdcontext_hub_checkpoint_command(void *ptr) {
283
284
uint64_t our_version_hash = rrdcontext_version_hash(host);
285
239
- if(cmd->version_hash != our_version_hash) {
286
+ if(version_hash != our_version_hash) {
287
nd_log(NDLS_DAEMON, NDLP_NOTICE,
288
"RRDCONTEXT: received version hash %"PRIu64" for host '%s', does not match our version hash %"PRIu64". "
289
"Sending snapshot of all contexts.",
243
- cmd->version_hash, rrdhost_hostname(host), our_version_hash);
290
+ version_hash, rrdhost_hostname(host), our_version_hash);
291
292
// prepare the snapshot
293
char uuid_str[UUID_STR_LEN];
294
uuid_unparse_lower(host->node_id.uuid, uuid_str);
248
- contexts_snapshot_t bundle = contexts_snapshot_new(cmd->claim_id, uuid_str, our_version_hash);
295
+ contexts_snapshot_t bundle = contexts_snapshot_new(claim_id, uuid_str, our_version_hash);
296
297
// do a deep scan on every metric of the host to make sure all our data are updated
298
rrdcontext_recalculate_host_retention(host, RRD_FLAG_NONE, false);
@@ -253,6 +300,11 @@ void rrdcontext_hub_checkpoint_command(void *ptr) {
300
// calculate version hash and pack all the messages together in one go
301
our_version_hash = rrdcontext_version_hash_with_callback(host, rrdcontext_message_send_unsafe, true, bundle);
302
303
+ if(!rrdcontext_checkpoint_generation_is_current(host, claim_id, node_id, generation)) {
304
+ contexts_snapshot_delete(bundle);
305
+ return;
306
+ }
307
+
308
// update the version
309
contexts_snapshot_set_version(bundle, our_version_hash);
310
@@ -260,11 +312,32 @@ void rrdcontext_hub_checkpoint_command(void *ptr) {
312
aclk_send_contexts_snapshot(bundle);
313
}
314
315
+ if(!rrdcontext_checkpoint_generation_is_current(host, claim_id, node_id, generation))
316
+ return;
317
+
318
nd_log(NDLS_DAEMON, NDLP_DEBUG,
319
"RRDCONTEXT: host '%s' enabling streaming of contexts",
320
rrdhost_hostname(host));
321
267
- rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
322
+ struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
323
+ if(aclk_host_config) {
324
+ spinlock_lock(&aclk_host_config->pending_ctx_spinlock);
325
+
326
+ bool can_enable =
327
+ __atomic_load_n(&aclk_host_config->pending_ctx_generation, __ATOMIC_ACQUIRE) == generation &&
328
+ !__atomic_load_n(&aclk_host_config->pending_ctx_checkpoint, __ATOMIC_ACQUIRE);
329
+
330
+ if(can_enable)
331
+ rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
332
+
333
+ spinlock_unlock(&aclk_host_config->pending_ctx_spinlock);
334
+
335
+ if(!can_enable)
336
+ return;
337
+ }
338
+ else
339
+ rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
340
+
341
char node_str[UUID_STR_LEN];
342
uuid_unparse_lower(host->node_id.uuid, node_str);
343
nd_log(NDLS_ACCESS, NDLP_DEBUG,
@@ -272,6 +345,49 @@ void rrdcontext_hub_checkpoint_command(void *ptr) {
345
node_str, rrdhost_hostname(host));
346
}
347
348
+void rrdcontext_hub_checkpoint_command(void *ptr) {
349
+ struct ctxs_checkpoint *cmd = ptr;
350
+
351
+ if(!claim_id_matches(cmd->claim_id)) {
352
+ CLAIM_ID claim_id = claim_id_get();
353
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
354
+ "RRDCONTEXT: received checkpoint command for claim_id '%s', node id '%s', "
355
+ "but this is not our claim id. Ours '%s', received '%s'. Ignoring command.",
356
+ cmd->claim_id, cmd->node_id,
357
+ claim_id.str, cmd->claim_id);
358
+
359
+ return;
360
+ }
361
+
362
+ RRDHOST *host = rrdhost_find_by_node_id(cmd->node_id);
363
+ if(!host) {
364
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
365
+ "RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', "
366
+ "but there is no node with such node id here. Ignoring command.",
367
+ cmd->claim_id, cmd->node_id);
368
+
369
+ return;
370
+ }
371
+
372
+ if(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD) ||
373
+ rrdcontext_queue_entries(&host->rrdctx.pp_queue) > 0) {
374
+ nd_log(NDLS_DAEMON, NDLP_NOTICE,
375
+ "RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', "
376
+ "but host '%s' has pending context work. Saving checkpoint for replay after processing completes.",
377
+ cmd->claim_id, cmd->node_id, rrdhost_hostname(host));
378
+
379
+ if(rrdcontext_checkpoint_save_pending(host, cmd))
380
+ return;
381
+
382
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
383
+ "RRDCONTEXT: failed to save pending checkpoint for host '%s' (no aclk config). Executing immediately.",
384
+ rrdhost_hostname(host));
385
+ }
386
+
387
+ uint64_t generation = rrdcontext_checkpoint_invalidate_pending(host);
388
+ rrdcontext_checkpoint_execute(host, cmd->claim_id, cmd->node_id, cmd->version_hash, generation);
389
+}
390
+
391
void rrdcontext_hub_stop_streaming_command(void *ptr) {
392
struct stop_streaming_ctxs *cmd = ptr;
393
@@ -296,11 +412,19 @@ void rrdcontext_hub_stop_streaming_command(void *ptr) {
412
return;
413
}
414
415
+ bool had_pending_checkpoint = false;
416
+ struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
417
+ if(aclk_host_config)
418
+ had_pending_checkpoint = __atomic_load_n(&aclk_host_config->pending_ctx_checkpoint, __ATOMIC_ACQUIRE);
419
+
420
+ rrdcontext_checkpoint_invalidate_pending(host);
421
+
422
if(!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS)) {
423
nd_log(NDLS_DAEMON, NDLP_NOTICE,
424
"RRDCONTEXT: received stop streaming command for claim id '%s', node id '%s', "
302
- "but node '%s' does not have active context streaming. Ignoring command.",
303
- cmd->claim_id, cmd->node_id, rrdhost_hostname(host));
425
+ "but node '%s' does not have active context streaming%s.",
426
+ cmd->claim_id, cmd->node_id, rrdhost_hostname(host),
427
+ had_pending_checkpoint ? "; invalidated deferred checkpoint" : "");
428
429
return;
430
}
@@ -312,6 +436,73 @@ void rrdcontext_hub_stop_streaming_command(void *ptr) {
436
rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
437
}
438
439
+#define PENDING_CTX_CHECKPOINT_MAX_AGE_S 300
440
+
441
+void rrdcontext_hub_pending_checkpoint_replay(RRDHOST *host) {
442
+ struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);
443
+ if(!aclk_host_config || !__atomic_load_n(&aclk_host_config->pending_ctx_checkpoint, __ATOMIC_ACQUIRE))
444
+ return;
445
+
446
+ bool pending_context_load = rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD);
447
+ bool pp_queue_empty = rrdcontext_queue_entries(&host->rrdctx.pp_queue) <= 0;
448
+
449
+ spinlock_lock(&aclk_host_config->pending_ctx_spinlock);
450
+ if(!__atomic_load_n(&aclk_host_config->pending_ctx_checkpoint, __ATOMIC_RELAXED)) {
451
+ spinlock_unlock(&aclk_host_config->pending_ctx_spinlock);
452
+ return;
453
+ }
454
+
455
+ // replay when pp_queue is empty, or force replay on timeout
456
+ time_t now_s = now_monotonic_sec();
457
+ bool timed_out = (now_s - aclk_host_config->pending_ctx_saved_monotonic_s >= PENDING_CTX_CHECKPOINT_MAX_AGE_S);
458
+ if((pending_context_load || !pp_queue_empty) && !timed_out) {
459
+ spinlock_unlock(&aclk_host_config->pending_ctx_spinlock);
460
+ return;
461
+ }
462
+
463
+ char *claim_id = aclk_host_config->pending_ctx_claim_id;
464
+ char *node_id = aclk_host_config->pending_ctx_node_id;
465
+ uint64_t version_hash = aclk_host_config->pending_ctx_version_hash;
466
+ uint64_t generation = __atomic_load_n(&aclk_host_config->pending_ctx_generation, __ATOMIC_RELAXED);
467
+
468
+ aclk_host_config->pending_ctx_claim_id = NULL;
469
+ aclk_host_config->pending_ctx_node_id = NULL;
470
+ aclk_host_config->pending_ctx_version_hash = 0;
471
+ aclk_host_config->pending_ctx_saved_monotonic_s = 0;
472
+ __atomic_store_n(&aclk_host_config->pending_ctx_checkpoint, false, __ATOMIC_RELEASE);
473
+ spinlock_unlock(&aclk_host_config->pending_ctx_spinlock);
474
+
475
+ if(timed_out && (pending_context_load || !pp_queue_empty))
476
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
477
+ "RRDCONTEXT: pending checkpoint for host '%s' timed out after %d sec with context work still active. Forcing replay.",
478
+ rrdhost_hostname(host), PENDING_CTX_CHECKPOINT_MAX_AGE_S);
479
+
480
+ if(!claim_id || !node_id) {
481
+ freez(claim_id);
482
+ freez(node_id);
483
+ return;
484
+ }
485
+
486
+ // verify claim id is still valid
487
+ if(!claim_id_matches(claim_id)) {
488
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
489
+ "RRDCONTEXT: pending checkpoint for host '%s' has stale claim id '%s'. Discarding.",
490
+ rrdhost_hostname(host), claim_id);
491
+ freez(claim_id);
492
+ freez(node_id);
493
+ return;
494
+ }
495
+
496
+ nd_log(NDLS_DAEMON, NDLP_NOTICE,
497
+ "RRDCONTEXT: replaying deferred checkpoint for host '%s', claim id '%s', node id '%s'.",
498
+ rrdhost_hostname(host), claim_id, node_id);
499
+
500
+ rrdcontext_checkpoint_execute(host, claim_id, node_id, version_hash, generation);
501
+
502
+ freez(claim_id);
503
+ freez(node_id);
504
+}
505
+
506
ALWAYS_INLINE
507
bool rrdcontext_retention_match(RRDCONTEXT_ACQUIRED *rca, time_t after, time_t before) {
508
if(unlikely(!rca)) return false;
src/database/contexts/rrdcontext.h
+1
@@ -121,6 +121,7 @@ int rrdcontext_find_chart_uuid(RRDSET *st, nd_uuid_t *store_uuid);
121
122
void rrdcontext_hub_checkpoint_command(void *cmd);
123
void rrdcontext_hub_stop_streaming_command(void *cmd);
124
+void rrdcontext_hub_pending_checkpoint_replay(RRDHOST *host);
125
126
127
// ----------------------------------------------------------------------------
src/database/sqlite/sqlite_aclk.c
+17
@@ -992,6 +992,7 @@ void create_aclk_config(RRDHOST *host, nd_uuid_t *host_uuid __maybe_unused, nd_u
992
return;
993
994
struct aclk_sync_cfg_t *aclk_host_config = callocz(1, sizeof(struct aclk_sync_cfg_t));
995
+ spinlock_init(&aclk_host_config->pending_ctx_spinlock);
996
if (node_id && !uuid_is_null(*node_id))
997
uuid_unparse_lower(*node_id, aclk_host_config->node_id);
998
@@ -1221,6 +1222,22 @@ void destroy_aclk_config(RRDHOST *host)
1222
}
1223
1224
struct aclk_sync_cfg_t *old_aclk_host_config = __atomic_exchange_n(&host->aclk_host_config, NULL, __ATOMIC_RELAXED);
1225
+ if (!old_aclk_host_config)
1226
+ return;
1227
+
1228
+ // detach pending checkpoint strings under lock, to avoid racing with save/replay
1229
+ spinlock_lock(&old_aclk_host_config->pending_ctx_spinlock);
1230
+ char *pending_claim_id = old_aclk_host_config->pending_ctx_claim_id;
1231
+ char *pending_node_id = old_aclk_host_config->pending_ctx_node_id;
1232
+ old_aclk_host_config->pending_ctx_claim_id = NULL;
1233
+ old_aclk_host_config->pending_ctx_node_id = NULL;
1234
+ old_aclk_host_config->pending_ctx_version_hash = 0;
1235
+ old_aclk_host_config->pending_ctx_saved_monotonic_s = 0;
1236
+ __atomic_store_n(&old_aclk_host_config->pending_ctx_checkpoint, false, __ATOMIC_RELEASE);
1237
+ spinlock_unlock(&old_aclk_host_config->pending_ctx_spinlock);
1238
+
1239
+ freez(pending_claim_id);
1240
+ freez(pending_node_id);
1241
freez(old_aclk_host_config);
1242
}
1243
src/database/sqlite/sqlite_aclk.h
+11
@@ -38,6 +38,17 @@ typedef struct aclk_sync_cfg_t {
38
RRDHOST *host;
39
uv_timer_t timer;
40
bool timer_initialized;
41
+
42
+ // pending context checkpoint - saved when deferred during context load or post-processing
43
+ // protected by pending_ctx_spinlock (accessed from ACLK query thread and context worker thread)
44
+ SPINLOCK pending_ctx_spinlock;
45
+ uint64_t pending_ctx_generation;
46
+ bool pending_ctx_checkpoint;
47
+ char *pending_ctx_claim_id;
48
+ char *pending_ctx_node_id;
49
+ uint64_t pending_ctx_version_hash;
50
+ time_t pending_ctx_saved_monotonic_s;
51
+
52
int8_t send_snapshot;
53
bool stream_alerts;
54
int alert_count;