Change queue metrics

8 files changed, 57 insertions, 36 deletions

diff --git a/contrib/restricted/aws/aws-c-io/include/aws/io/io.h b/contrib/restricted/aws/aws-c-io/include/aws/io/io.h
index 9dc5039389..9c6e4614ed 100644
--- a/contrib/restricted/aws/aws-c-io/include/aws/io/io.h
+++ b/contrib/restricted/aws/aws-c-io/include/aws/io/io.h
@@ -241,6 +241,7 @@ enum aws_io_errors {
     AWS_ERROR_IO_PINNED_EVENT_LOOP_MISMATCH,
 
     AWS_ERROR_PKCS11_ENCODING_ERROR,
+    AWS_IO_TLS_ERROR_DEFAULT_TRUST_STORE_NOT_FOUND,
 
     AWS_IO_ERROR_END_RANGE = AWS_ERROR_ENUM_END_RANGE(AWS_C_IO_PACKAGE_ID),
     AWS_IO_INVALID_FILE_HANDLE = AWS_ERROR_INVALID_FILE_HANDLE,
diff --git a/contrib/restricted/aws/aws-c-io/include/aws/io/tls_channel_handler.h b/contrib/restricted/aws/aws-c-io/include/aws/io/tls_channel_handler.h
index 9e1c9d436c..1c19c59b58 100644
--- a/contrib/restricted/aws/aws-c-io/include/aws/io/tls_channel_handler.h
+++ b/contrib/restricted/aws/aws-c-io/include/aws/io/tls_channel_handler.h
@@ -24,11 +24,19 @@ enum aws_tls_versions {
 
 enum aws_tls_cipher_pref {
     AWS_IO_TLS_CIPHER_PREF_SYSTEM_DEFAULT = 0,
-    AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2019_06 = 1,
-    AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2019_11 = 2,
-    AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_02 = 3,
-    AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2020_02 = 4,
-    AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_07 = 5,
+
+    /* Deprecated */ AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2019_06 = 1,
+    /* Deprecated */ AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2019_11 = 2,
+    /* Deprecated */ AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_02 = 3,
+    /* Deprecated */ AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2020_02 = 4,
+    /* Deprecated */ AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_07 = 5,
+
+    /*
+     * This TLS cipher preference list contains post-quantum key exchange algorithms that have been submitted to NIST
+     * for potential future standardization. Support for this preference list, or PQ algorithms present in it, may be
+     * removed at any time in the future. PQ algorithms in this preference list will be used in hybrid mode, and always
+     * combined with a classical ECDHE key exchange.
+     */
     AWS_IO_TLS_CIPHER_PREF_PQ_TLSv1_0_2021_05 = 6,
 
     AWS_IO_TLS_CIPHER_PREF_END_RANGE = 0xFFFF
diff --git a/contrib/restricted/aws/aws-c-io/source/io.c b/contrib/restricted/aws/aws-c-io/source/io.c
index 55b9d0eb13..82145c996f 100644
--- a/contrib/restricted/aws/aws-c-io/source/io.c
+++ b/contrib/restricted/aws/aws-c-io/source/io.c
@@ -282,6 +282,11 @@ static struct aws_error_info s_errors[] = {
     AWS_DEFINE_ERROR_INFO_IO(
        AWS_ERROR_PKCS11_ENCODING_ERROR,
        "A PKCS#11 (Cryptoki) library function was unable to ASN.1 (DER) encode a data structure. See log for more details."),
+    AWS_DEFINE_ERROR_INFO_IO(
+       AWS_IO_TLS_ERROR_DEFAULT_TRUST_STORE_NOT_FOUND,
+        "Default TLS trust store not found on this system."
+        " Trusted CA certificates must be installed,"
+        " or \"override default trust store\" must be used while creating the TLS context."),
 };
 /* clang-format on */
 
diff --git a/contrib/restricted/aws/aws-c-io/source/s2n/s2n_tls_channel_handler.c b/contrib/restricted/aws/aws-c-io/source/s2n/s2n_tls_channel_handler.c
index e0e776361a..5c3561154d 100644
--- a/contrib/restricted/aws/aws-c-io/source/s2n/s2n_tls_channel_handler.c
+++ b/contrib/restricted/aws/aws-c-io/source/s2n/s2n_tls_channel_handler.c
@@ -189,11 +189,19 @@ void aws_tls_init_static_state(struct aws_allocator *alloc) {
 
     s_default_ca_dir = s_determine_default_pki_dir();
     s_default_ca_file = s_determine_default_pki_ca_file();
-    AWS_LOGF_DEBUG(
-        AWS_LS_IO_TLS,
-        "ctx: Based on OS, we detected the default PKI path as %s, and ca file as %s",
-        s_default_ca_dir,
-        s_default_ca_file);
+    if (s_default_ca_dir || s_default_ca_file) {
+        AWS_LOGF_DEBUG(
+            AWS_LS_IO_TLS,
+            "ctx: Based on OS, we detected the default PKI path as %s, and ca file as %s",
+            s_default_ca_dir,
+            s_default_ca_file);
+    } else {
+        AWS_LOGF_WARN(
+            AWS_LS_IO_TLS,
+            "Default TLS trust store not found on this system."
+            " TLS connections will fail unless trusted CA certificates are installed,"
+            " or \"override default trust store\" is used while creating the TLS context.");
+    }
 }
 
 void aws_tls_clean_up_static_state(void) {
@@ -210,11 +218,6 @@ bool aws_tls_is_cipher_pref_supported(enum aws_tls_cipher_pref cipher_pref) {
             return true;
             /* PQ Crypto no-ops on android for now */
 #ifndef ANDROID
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2019_06:
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2019_11:
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_02:
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2020_02:
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_07:
         case AWS_IO_TLS_CIPHER_PREF_PQ_TLSv1_0_2021_05:
             return true;
 #endif
@@ -1369,21 +1372,6 @@ static struct aws_tls_ctx *s_tls_ctx_new(
         case AWS_IO_TLS_CIPHER_PREF_SYSTEM_DEFAULT:
             /* No-Op, if the user configured a minimum_tls_version then a version-specific Cipher Preference was set */
             break;
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2019_06:
-            s2n_config_set_cipher_preferences(s2n_ctx->s2n_config, "KMS-PQ-TLS-1-0-2019-06");
-            break;
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2019_11:
-            s2n_config_set_cipher_preferences(s2n_ctx->s2n_config, "PQ-SIKE-TEST-TLS-1-0-2019-11");
-            break;
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_02:
-            s2n_config_set_cipher_preferences(s2n_ctx->s2n_config, "KMS-PQ-TLS-1-0-2020-02");
-            break;
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_SIKE_TLSv1_0_2020_02:
-            s2n_config_set_cipher_preferences(s2n_ctx->s2n_config, "PQ-SIKE-TEST-TLS-1-0-2020-02");
-            break;
-        case AWS_IO_TLS_CIPHER_PREF_KMS_PQ_TLSv1_0_2020_07:
-            s2n_config_set_cipher_preferences(s2n_ctx->s2n_config, "KMS-PQ-TLS-1-0-2020-07");
-            break;
         case AWS_IO_TLS_CIPHER_PREF_PQ_TLSv1_0_2021_05:
             s2n_config_set_cipher_preferences(s2n_ctx->s2n_config, "PQ-TLS-1-0-2021-05-26");
             break;
@@ -1507,7 +1495,7 @@ static struct aws_tls_ctx *s_tls_ctx_new(
                     goto cleanup_s2n_config;
                 }
             }
-        } else {
+        } else if (s_default_ca_file || s_default_ca_dir) {
             /* User wants to use the system's default trust store.
              *
              * Note that s2n's trust store always starts with libcrypto's default locations.
@@ -1522,6 +1510,14 @@ static struct aws_tls_ctx *s_tls_ctx_new(
                     AWS_LS_IO_TLS, "Failed to set ca_path: %s and ca_file %s\n", s_default_ca_dir, s_default_ca_file);
                 goto cleanup_s2n_config;
             }
+        } else {
+            /* Cannot find system's trust store */
+            aws_raise_error(AWS_IO_TLS_ERROR_DEFAULT_TRUST_STORE_NOT_FOUND);
+            AWS_LOGF_ERROR(
+                AWS_LS_IO_TLS,
+                "Default TLS trust store not found on this system."
+                " Install CA certificates, or \"override default trust store\".");
+            goto cleanup_s2n_config;
         }
 
         if (mode == S2N_SERVER && s2n_config_set_client_auth_type(s2n_ctx->s2n_config, S2N_CERT_AUTH_REQUIRED)) {
diff --git a/ydb/core/protos/counters_datashard.proto b/ydb/core/protos/counters_datashard.proto
index bcc87adcb1..8195eb0bfb 100644
--- a/ydb/core/protos/counters_datashard.proto
+++ b/ydb/core/protos/counters_datashard.proto
@@ -20,6 +20,7 @@ enum ESimpleCounters {
     COUNTER_MVCC_STATE_CHANGE_WAIT_TX_IN_FLY = 10                  [(CounterOpts) = {Name: "MvccStateChangeWaitTxInFly"}];
     COUNTER_MVCC_STATE_CHANGE_WAIT_IMMEDIATE_TX_IN_FLY = 11        [(CounterOpts) = {Name: "MvccStateChangeWaitImmediateTxInFly"}];
     COUNTER_MVCC_ENABLED = 12                                      [(CounterOpts) = {Name: "MvccEnabled"}];
+    COUNTER_CHANGE_QUEUE_SIZE  = 13                                [(CounterOpts) = {Name: "ChangeQueueSize"}];
 }
 
 enum ECumulativeCounters {
@@ -111,6 +112,8 @@ enum ECumulativeCounters {
     COUNTER_FULL_COMPACTION_DONE = 85                     [(CounterOpts) = {Name: "FullCompactionCount"}];
     COUNTER_TX_BACKGROUND_COMPACTION_FAILED_LOANED = 86   [(CounterOpts) = {Name: "TxCompactTableFailedLoaned"}];
     COUNTER_TX_COMPACT_BORROWED = 87                      [(CounterOpts) = {Name: "TxCompactBorrowed"}];
+    COUNTER_CHANGE_RECORDS_ENQUEUED = 88                  [(CounterOpts) = {Name: "ChangeRecordsEnqueued"}];
+    COUNTER_CHANGE_RECORDS_REMOVED = 89                   [(CounterOpts) = {Name: "ChangeRecordsRemoved"}];
 }
 
 enum EPercentileCounters {
diff --git a/ydb/core/tx/datashard/datashard.cpp b/ydb/core/tx/datashard/datashard.cpp
index f642a47d8c..5dc707fcf0 100644
--- a/ydb/core/tx/datashard/datashard.cpp
+++ b/ydb/core/tx/datashard/datashard.cpp
@@ -601,6 +601,9 @@ void TDataShard::RemoveChangeRecord(NIceDb::TNiceDb& db, ui64 order) {
     }
 
     ChangesQueue.erase(it);
+
+    IncCounter(COUNTER_CHANGE_RECORDS_REMOVED);
+    SetCounter(COUNTER_CHANGE_QUEUE_SIZE, ChangesQueue.size());
 }
 
 void TDataShard::EnqueueChangeRecords(TVector<NMiniKQL::IChangeCollector::TChange>&& records) {
@@ -627,6 +630,9 @@ void TDataShard::EnqueueChangeRecords(TVector<NMiniKQL::IChangeCollector::TChang
         }
     }
 
+    IncCounter(COUNTER_CHANGE_RECORDS_ENQUEUED, forward.size());
+    SetCounter(COUNTER_CHANGE_QUEUE_SIZE, ChangesQueue.size());
+
     Y_VERIFY(OutChangeSender);
     Send(OutChangeSender, new TEvChangeExchange::TEvEnqueueRecords(std::move(forward)));
 }
diff --git a/ydb/core/tx/datashard/datashard_change_receiving.cpp b/ydb/core/tx/datashard/datashard_change_receiving.cpp
index fcb74f4401..0cc14c5f17 100644
--- a/ydb/core/tx/datashard/datashard_change_receiving.cpp
+++ b/ydb/core/tx/datashard/datashard_change_receiving.cpp
@@ -266,6 +266,7 @@ class TDataShard::TTxApplyChangeRecords: public TTransactionBase<TDataShard> {
         }
 
         txc.DB.Update(tableInfo.LocalTid, rop, Key, Value, TRowVersion(record.GetStep(), record.GetTxId()));
+        tableInfo.Stats.UpdateTime = TAppData::TimeProvider->Now();
         AddRecordStatus(ctx, record.GetOrder(), NKikimrChangeExchange::TEvStatus::STATUS_OK);
 
         return true;
diff --git a/ydb/core/tx/datashard/datashard_impl.h b/ydb/core/tx/datashard/datashard_impl.h
index 5068c4c6f6..f974cdc429 100644
--- a/ydb/core/tx/datashard/datashard_impl.h
+++ b/ydb/core/tx/datashard/datashard_impl.h
@@ -2549,12 +2549,13 @@ protected:
             ev->Record.MutableTableStats()->SetTxRejectedByOverload(TabletCounters->Cumulative()[COUNTER_PREPARE_OVERLOADED].Get());
             ev->Record.MutableTableStats()->SetTxRejectedBySpace(TabletCounters->Cumulative()[COUNTER_PREPARE_OUT_OF_SPACE].Get());
             ev->Record.MutableTableStats()->SetTxCompleteLagMsec(TabletCounters->Simple()[COUNTER_TX_COMPLETE_LAG].Get());
-            ev->Record.MutableTableStats()->SetInFlightTxCount(TabletCounters->Simple()[COUNTER_TX_IN_FLY].Get() +
-                 TabletCounters->Simple()[COUNTER_IMMEDIATE_TX_IN_FLY].Get());
+            ev->Record.MutableTableStats()->SetInFlightTxCount(TabletCounters->Simple()[COUNTER_TX_IN_FLY].Get()
+                + TabletCounters->Simple()[COUNTER_IMMEDIATE_TX_IN_FLY].Get());
 
-            ev->Record.MutableTableStats()->SetRowUpdates(TabletCounters->Cumulative()[COUNTER_ENGINE_HOST_UPDATE_ROW].Get() +
-                                                          TabletCounters->Cumulative()[COUNTER_UPLOAD_ROWS].Get());
-            ev->Record.MutableTableStats()->SetRowDeletes(TabletCounters->Cumulative()[COUNTER_ENGINE_HOST_ERASE_ROW].Get());
+            ev->Record.MutableTableStats()->SetRowUpdates(TabletCounters->Cumulative()[COUNTER_ENGINE_HOST_UPDATE_ROW].Get()
+                + TabletCounters->Cumulative()[COUNTER_UPLOAD_ROWS].Get());
+            ev->Record.MutableTableStats()->SetRowDeletes(TabletCounters->Cumulative()[COUNTER_ENGINE_HOST_ERASE_ROW].Get()
+                + TabletCounters->Cumulative()[COUNTER_ERASE_ROWS].Get());
             ev->Record.MutableTableStats()->SetRowReads(TabletCounters->Cumulative()[COUNTER_ENGINE_HOST_SELECT_ROW].Get());
             ev->Record.MutableTableStats()->SetRangeReads(TabletCounters->Cumulative()[COUNTER_ENGINE_HOST_SELECT_RANGE].Get());
             ev->Record.MutableTableStats()->SetRangeReadRows(TabletCounters->Cumulative()[COUNTER_ENGINE_HOST_SELECT_RANGE_ROWS].Get());