From 4503bdfa7ee6ce0374fb58651f5a1beb7c8b77cc Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Mon, 28 Sep 2026 13:16:39 +0300
Subject: [PATCH] diagnostics: explain and rate-limit rejected ETCP session
 data

---
 src/transport_layer/etcp.h         |  1 +
 src/transport_layer/etcp_session.c | 12 +++++++++++-
 tests/test_etcp_session.c          | 22 ++++++++++++++++++++++
 3 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/src/transport_layer/etcp.h b/src/transport_layer/etcp.h
index 53b8c0dd..2b1dbc32 100644
--- a/src/transport_layer/etcp.h
+++ b/src/transport_layer/etcp.h
@@ -204,6 +204,7 @@ struct ETCP_CONN {
     uint64_t session_candidate, session_challenge, session_started; // эпоха-кандидат, challenge, время начала (0.1 ms)
     void* session_timer;            // таймер согласования транспортной сессии
     uint8_t session_required;       // только подтверждение challenge может завершить локальный reset
+    uint64_t session_data_errors;   // отвергнутые заголовки DATA; ограничивает повторяющиеся сообщения об ошибке
     uint8_t tx_state;                // 0 - n/a, 1 - data_wait (queues empty), 2 - link_wait (link busy)
     uint8_t links_up;                // 0 - канал не готов для передачи, 1 - канал готов для передачи (хотя бы один линк не down)
     uint8_t reset_done;              // 0 - рукопожатие не завершено (реинит разрешён), 1 - соединение стабильно (реинит заблокирован)
diff --git a/src/transport_layer/etcp_session.c b/src/transport_layer/etcp_session.c
index b2268b0e..cd51e223 100644
--- a/src/transport_layer/etcp_session.c
+++ b/src/transport_layer/etcp_session.c
@@ -112,7 +112,17 @@ int etcp_session_receive(struct ETCP_CONN* c, const uint8_t* data, size_t len) {
 
 int etcp_session_accept_data(struct ETCP_CONN* c, const uint8_t* data, size_t len) {
     if (len <= ETCP_SESSION_HEADER_SIZE || data[0] != ETCP_SESSION_DATA || data[ETCP_SESSION_HEADER_SIZE] > 1) {
-        DEBUG_WARN(DEBUG_CATEGORY_ETCP,"[%s] invalid session data size=%zu",c->log_name,len); return 0;
+        c->session_data_errors++;
+        if (c->session_data_errors <= 3 || c->session_data_errors % 1024 == 0) {
+            int code = len ? data[0] : -1;
+            int section = len > ETCP_SESSION_HEADER_SIZE ? data[ETCP_SESSION_HEADER_SIZE] : -1;
+            const char* reason = len <= ETCP_SESSION_HEADER_SIZE ? "short header" :
+                (code != ETCP_SESSION_DATA ? "missing session envelope" : "invalid stream section");
+            DEBUG_WARN(DEBUG_CATEGORY_ETCP,
+                "[%s] invalid session data: %s size=%zu code=%d expected=%u section=%d count=%llu",
+                c->log_name, reason, len, code, ETCP_SESSION_DATA, section, (unsigned long long)c->session_data_errors);
+        }
+        return 0;
     }
     uint64_t peer = read_id(data+1), target = read_id(data+9);
     if (peer != c->peer_reset_id || target != c->reset_id || c->reinit_pending || !c->initialized) {
diff --git a/tests/test_etcp_session.c b/tests/test_etcp_session.c
index 73ee22f9..04dd7b9a 100644
--- a/tests/test_etcp_session.c
+++ b/tests/test_etcp_session.c
@@ -110,6 +110,27 @@ static struct frame stream_packet(int data) {
     CHECK(etcp_encrypt_send(p)>0); memory_pool_free(ep[0].inst.pkt_pool,p);
     CHECK(tail == head+1); struct frame f=frames[head++ % 4096]; return f;
 }
+static void data_header_validation(void) {
+    setup(0);
+    uint8_t data[ETCP_SESSION_HEADER_SIZE + ETCP_ACK_BASE_SIZE] = {0};
+    struct ETCP_CONN* c = ep[1].c;
+    etcp_session_encode(data, ETCP_SESSION_DATA, c->peer_reset_id, c->reset_id);
+    data[ETCP_SESSION_HEADER_SIZE] = ETCP_SECTION_ACK;
+    CHECK(etcp_session_accept_data(c, data, sizeof(data)));
+    CHECK(!etcp_session_accept_data(c, data, 0));
+    CHECK(!etcp_session_accept_data(c, data, ETCP_SESSION_HEADER_SIZE));
+    data[0] = ETCP_SECTION_ACK;
+    CHECK(!etcp_session_accept_data(c, data, sizeof(data)));
+    data[0] = ETCP_SESSION_DATA;
+    data[ETCP_SESSION_HEADER_SIZE] = ETCP_SESSION_DATA;
+    for (unsigned i = 3; i < 1024; i++) CHECK(!etcp_session_accept_data(c, data, sizeof(data)));
+    CHECK(c->session_data_errors == 1024);
+    data[ETCP_SESSION_HEADER_SIZE] = ETCP_SECTION_ACK;
+    CHECK(etcp_session_accept_data(c, data, sizeof(data)));
+    CHECK(c->session_data_errors == 1024 && c->initialized && !c->reinit_pending);
+    teardown();
+    puts("PASS session DATA header validation and rejection counter");
+}
 static void timestamp_boundaries(void) {
     setup(0);
     struct ETCP_LINK* link = &ep[0].links[0];
@@ -161,6 +182,7 @@ int main(void) {
     const char* log=getenv("SESSION_TEST_LOG");
     if (log) { debug_enable_file_output(log,1); debug_set_category_level(DEBUG_CATEGORY_ETCP,DEBUG_LEVEL_DEBUG); }
     size_t baseline=u_get_allocated_count();
+    data_header_validation(); CHECK(u_get_allocated_count() == baseline);
     timestamp_boundaries(); CHECK(u_get_allocated_count() == baseline);
     receive_window_case(0); CHECK(u_get_allocated_count() == baseline);
     receive_window_case(1); CHECK(u_get_allocated_count() == baseline);