Change data capture for EloqDoc: log scan, schema-op timestamps, cluster gate, multi-node fixes

.github/workflows/tests.yml

@@ -213,11 +213,15 @@ jobs:
             exit 1
           fi
           # test_rocksdb_cloud_sst_reader exercises the rocksdb-cloud storage
-          # provider end-to-end against RustFS. The DBCloud::Open-based cases
-          # currently crash inside librocksdb-cloud (ABI/vtable mismatch) and are
-          # tracked separately, so they are excluded here.
+          # provider end-to-end against RustFS; test_cdc_scan_cloud_state,
+          # test_cdc_retention_floor_cloud_state and
+          # test_cdc_ddl_history_cloud_state the CDC log scan, the CDC
+          # retention floor and the CDC DDL history on the cloud log state.
+          # The DBCloud::Open-based cases currently crash inside
+          # librocksdb-cloud (ABI/vtable mismatch) and are tracked separately,
+          # so they are excluded here.
           cd eloq_log_service/bld_cloud
-          ./log_server_rocksdb_cloud_tests "test_rocksdb_cloud_sst_reader" \
+          ./log_server_rocksdb_cloud_tests "test_rocksdb_cloud_sst_reader,test_cdc_scan_cloud_state,test_cdc_retention_floor_cloud_state,test_cdc_ddl_history_cloud_state" \
             --region us-east-1 \
             --aws_access_key_id "$AWS_ACCESS_KEY_ID" \
             --aws_secret_key "$AWS_SECRET_ACCESS_KEY"

build_eloq_log_service.cmake

@@ -281,13 +281,18 @@ ADD_LIBRARY(logservice
     ${LOG_SOURCE_DIR}/src/log_state_rocksdb_impl.cpp
     ${LOG_SOURCE_DIR}/src/log_state_rocksdb_cloud_impl.cpp
     ${LOG_SOURCE_DIR}/src/log_state_memory_impl.cpp
+    ${LOG_SOURCE_DIR}/src/cdc_long_poll.cpp
+    ${LOG_SOURCE_DIR}/src/cdc_scan.cpp
     ${LOG_SOURCE_DIR}/src/fault_inject.cpp
     ${CMAKE_CURRENT_LIST_DIR}/third_party/ini/INIReader.cpp
     ${CMAKE_CURRENT_LIST_DIR}/third_party/ini/ini.c
     ${TX_LOG_PROTOS_SOURCE_DIR}/log.pb.cc
     )
 
 target_include_directories(logservice PUBLIC ${LOG_INCLUDE_DIR})
+# The log groups' metrics use eloq_metrics' header-only Meter against the
+# registry the process passes in, so the library itself is not linked.
+target_include_directories(logservice PUBLIC ${ELOQ_METRICS_INCLUDE_DIR})
 if (WITH_LOG_STATE MATCHES "ROCKSDB|ROCKSDB_CLOUD_S3|ROCKSDB_CLOUD_GCS")
     target_include_directories(logservice PUBLIC ${ROCKSDB_INCLUDE_PATH})
 endif()

build_tx_service.cmake

@@ -200,6 +200,8 @@ SET(ELOQ_SOURCES
     ${TX_SERVICE_SOURCE_DIR}/src/fault/fault_inject.cpp
     ${TX_SERVICE_SOURCE_DIR}/src/dead_lock_check.cpp
     ${TX_SERVICE_SOURCE_DIR}/src/tx_index_operation.cpp
+    ${TX_SERVICE_SOURCE_DIR}/src/tx_cluster_gate_operation.cpp
+    ${TX_SERVICE_SOURCE_DIR}/src/tx_upsert_table_operation.cpp
     ${TX_SERVICE_SOURCE_DIR}/src/sk_generator.cpp
     ${TX_SERVICE_SOURCE_DIR}/src/data_sync_task.cpp
     ${TX_SERVICE_SOURCE_DIR}/src/store/snapshot_manager.cpp

core/src/data_substrate.cpp

@@ -20,6 +20,7 @@
  *
  */
 
+#include <bthread/bthread.h>
 #include <gflags/gflags.h>
 #include <sys/sysinfo.h>
 
@@ -33,6 +34,9 @@
 // clang-format on
 #include "data_substrate.h"
 #include "log_server.h"
+#if (WITH_LOG_SERVICE)
+#include "log_service_metrics.h"
+#endif
 #include "sequences/sequences.h"
 #include "store/data_store_handler.h"
 #include "util.h"
@@ -87,6 +91,35 @@ const auto NUM_VCPU = std::thread::hardware_concurrency();
 // Static mutex for protecting initialization state
 static std::mutex instance_mutex_;
 
+namespace
+{
+#ifdef ELOQ_MODULE_ENABLED
+/**
+ * @brief Starts brpc's bthread workers if none runs yet, by running one
+ * bthread to completion.
+ *
+ * With ELOQ_MODULE_ENABLED, EloqStore's shards run on the bthread workers as
+ * a brpc module, so a request to EloqStore completes only once workers exist.
+ * Attaching a data shard in cloud mode is such a request, made while the
+ * storage handler starts. A serving node has started the workers by then, its
+ * brpc servers and clients having run bthreads; a bootstrap may have started
+ * none, and would wait on the attach forever.
+ * @return true if the workers run.
+ */
+bool EnsureBthreadWorkers()
+{
+    bthread_t tid;
+    if (bthread_start_background(
+            &tid, nullptr, [](void *) -> void * { return nullptr; }, nullptr) !=
+        0)
+    {
+        return false;
+    }
+    return bthread_join(tid, nullptr) == 0;
+}
+#endif
+}  // namespace
+
 DataSubstrate &DataSubstrate::Instance()
 {
     static DataSubstrate instance_;
@@ -180,6 +213,25 @@ bool DataSubstrate::Start()
         LOG(ERROR) << "Failed to initialize metrics";
         return false;
     }
+#if (WITH_LOG_SERVICE)
+    // The log service started in phase 1, before the registry existed.
+    if (metrics::enable_metrics && metrics::enable_log_service_metrics &&
+        instance.log_server_ != nullptr)
+    {
+        instance.log_server_->EnableMetrics(instance.metrics_registry_.get(),
+                                            instance.tx_service_common_labels_);
+    }
+#endif
+
+#ifdef ELOQ_MODULE_ENABLED
+    // The storage handler's first EloqStore requests need running bthread
+    // workers (EnsureBthreadWorkers).
+    if (!EnsureBthreadWorkers())
+    {
+        LOG(ERROR) << "Failed to start the bthread workers";
+        return false;
+    }
+#endif
 
     // Phase 3: Initialize storage handler
     if (instance.GetCoreConfig().enable_data_store &&
@@ -817,6 +869,18 @@ bool DataSubstrate::LoadCoreAndNetworkConfig(const INIReader &config_reader)
             : config_reader.GetBoolean(
                   "local", "enable_mvcc", FLAGS_enable_mvcc);
 
+    // The shard cleaner's release threshold is a tx service gflag that the
+    // cleaner reads directly; the config file sets it unless the command line
+    // already did.
+    const char *field_release = "shard_clean_release_kb";
+    if (CheckCommandLineFlagIsDefault(field_release) &&
+        config_reader.HasValue("local", field_release))
+    {
+        GFLAGS_NAMESPACE::SetCommandLineOption(
+            field_release,
+            config_reader.Get("local", field_release, "").c_str());
+    }
+
     // Load network configuration
     if (!LoadNetworkConfig(core_config_.bootstrap,
                            config_reader,

core/src/log_init.cpp

@@ -115,6 +115,13 @@ DEFINE_string(notify_checkpointer_threshold_size,
               "When the replay log size reaches this threshold the txlog "
               "server sends a checkpoint request to tx_service.");
 
+DEFINE_uint64(txlog_cdc_retention_seconds,
+              0,
+              "How far behind the checkpoint the log state keeps the data log "
+              "for change data capture: no purge removes a record within this "
+              "window. Set it above the worst-case CDC consumer lag. 0 keeps "
+              "no window, so every checkpointed record may be purged.");
+
 bool DataSubstrate::InitializeLogService(const INIReader &config_reader)
 {
     std::string txlog_service =
@@ -284,6 +291,14 @@ bool DataSubstrate::InitializeLogService(const INIReader &config_reader)
                           "notify_checkpointer_threshold_size",
                           FLAGS_notify_checkpointer_threshold_size));
 
+            [[maybe_unused]] uint64_t txlog_cdc_retention_seconds =
+                !CheckCommandLineFlagIsDefault("txlog_cdc_retention_seconds")
+                    ? FLAGS_txlog_cdc_retention_seconds
+                    : config_reader.GetInteger(
+                          "local",
+                          "txlog_cdc_retention_seconds",
+                          FLAGS_txlog_cdc_retention_seconds);
+
 #if defined(LOG_STATE_TYPE_RKDB_ALL)
             std::string txlog_rocksdb_storage_path =
                 !CheckCommandLineFlagIsDefault("txlog_rocksdb_storage_path")
@@ -534,7 +549,8 @@ bool DataSubstrate::InitializeLogService(const INIReader &config_reader)
                     logserver_snapshot_interval,
                     enable_txlog_request_checkpoint,
                     check_replay_log_size_interval_sec,
-                    notify_checkpointer_threshold_size_val);
+                    notify_checkpointer_threshold_size_val,
+                    txlog_cdc_retention_seconds);
             }
 #else
             size_t txlog_rocksdb_sst_files_size_limit_val =
@@ -585,7 +601,8 @@ bool DataSubstrate::InitializeLogService(const INIReader &config_reader)
                     logserver_snapshot_interval,
                     enable_txlog_request_checkpoint,
                     check_replay_log_size_interval_sec,
-                    notify_checkpointer_threshold_size_val);
+                    notify_checkpointer_threshold_size_val,
+                    txlog_cdc_retention_seconds);
             }
 #endif
 #endif

core/src/tx_service_init.cpp

@@ -7,7 +7,18 @@
 #include "eloq_log_wrapper.h"
 #include "sequences/sequences.h"
 #include "tx_service.h"
-DEFINE_int32(checkpointer_interval, 10, "Checkpointer interval in seconds");
+DEFINE_int32(checkpointer_interval,
+             10,
+             "Checkpointer interval in seconds; values below 5 are raised to "
+             "5, negative values are rejected");
+
+/**
+ * @brief The shortest checkpointer interval, in seconds.
+ *
+ * A shorter configured interval is raised to this. A zero interval would run
+ * checkpoint rounds back to back.
+ */
+static constexpr int64_t kMinCheckpointerIntervalSec = 5;
 DEFINE_int32(checkpointer_min_ckpt_request_interval,
              5,
              "Minimum checkpoint request interval in seconds, to avoid too "
@@ -42,6 +53,11 @@ DEFINE_string(hm_bin_path,
 DEFINE_uint32(deadlock_check_interval_seconds,
               10,
               "Deadlock check interval in seconds");
+DEFINE_int32(ddl_kv_retry_interval_seconds,
+             30,
+             "Pause in seconds before a schema op runs a failed data store "
+             "step again after its point of no return; 0 is raised to 1, "
+             "negative values are rejected");
 DEFINE_bool(realtime_sampling,
             true,
             "Whether enable realtime sampling. If disable it, user may need "
@@ -65,12 +81,28 @@ DEFINE_uint64(dirty_memory_size_threshold_mb,
 
 bool DataSubstrate::InitializeTxService(const INIReader &config_reader)
 {
-    uint64_t checkpointer_interval =
+    // Read signed: a negative value stored as unsigned becomes an interval of
+    // about 136 years, which silently stops the periodic checkpoints.
+    int64_t checkpointer_interval =
         !CheckCommandLineFlagIsDefault("checkpointer_interval")
             ? FLAGS_checkpointer_interval
             : config_reader.GetInteger("local",
                                        "checkpointer_interval",
                                        FLAGS_checkpointer_interval);
+    if (checkpointer_interval < 0)
+    {
+        LOG(ERROR) << "Invalid `checkpointer_interval` "
+                   << checkpointer_interval << ": it must not be negative";
+        return false;
+    }
+    if (checkpointer_interval < kMinCheckpointerIntervalSec)
+    {
+        LOG(WARNING) << "`checkpointer_interval` " << checkpointer_interval
+                     << " is below the minimum of "
+                     << kMinCheckpointerIntervalSec << " seconds; using "
+                     << kMinCheckpointerIntervalSec;
+        checkpointer_interval = kMinCheckpointerIntervalSec;
+    }
 
     uint64_t checkpointer_delay_seconds =
         !CheckCommandLineFlagIsDefault("checkpointer_delay_seconds")
@@ -144,6 +176,31 @@ bool DataSubstrate::InitializeTxService(const INIReader &config_reader)
                                        FLAGS_deadlock_check_interval_seconds);
     txservice::DeadLockCheck::SetTimeInterval(deadlock_check_interval_seconds);
 
+    // Read signed, as checkpointer_interval is: a negative value stored as
+    // unsigned would stall a retrying schema op for decades.
+    int64_t ddl_kv_retry_interval_seconds =
+        !CheckCommandLineFlagIsDefault("ddl_kv_retry_interval_seconds")
+            ? FLAGS_ddl_kv_retry_interval_seconds
+            : config_reader.GetInteger("local",
+                                       "ddl_kv_retry_interval_seconds",
+                                       FLAGS_ddl_kv_retry_interval_seconds);
+    if (ddl_kv_retry_interval_seconds < 0)
+    {
+        LOG(ERROR) << "Invalid `ddl_kv_retry_interval_seconds` "
+                   << ddl_kv_retry_interval_seconds
+                   << ": it must not be negative";
+        return false;
+    }
+    if (ddl_kv_retry_interval_seconds == 0)
+    {
+        // Retries need a pause between them.
+        LOG(WARNING) << "`ddl_kv_retry_interval_seconds` 0 is below the "
+                        "minimum of 1 second; using 1";
+        ddl_kv_retry_interval_seconds = 1;
+    }
+    txservice::SchemaOp::SetKvRetryInterval(
+        std::chrono::seconds(ddl_kv_retry_interval_seconds));
+
     bool realtime_sampling =
         !CheckCommandLineFlagIsDefault("realtime_sampling")
             ? FLAGS_realtime_sampling

docs/01-architecture-overview.md

@@ -18,7 +18,7 @@ The proprietary host manager participates through RPC-visible topology and leade
 
 ## Lifecycle and data flow
 
-`DataSubstrate::Init` loads process and cluster configuration. API layers then call `RegisterEngine` with their catalog factory and prebuilt tables. `Start` initializes log, metrics and storage dependencies before constructing and starting the transaction service. Shutdown reverses ownership so transaction work cannot outlive its persistence dependencies.
+`DataSubstrate::Init` loads process and cluster configuration. API layers then call `RegisterEngine` with their catalog factory and prebuilt tables. `Start` initializes log, metrics and storage dependencies before constructing and starting the transaction service; an in-process log server starts before the metrics registry exists and is handed it afterwards (`LogServer::EnableMetrics`). Shutdown reverses ownership so transaction work cannot outlive its persistence dependencies.
 
 A foreground request follows this path: