Make stream restart terminate explicitly if session scheduling fails

src/transport_layer/etcp.c

@@ -605,6 +605,7 @@ void etcp_conn_ready(struct ETCP_CONN* conn) {
         etcp_cbk_fire(conn, ETCP_CBK_EVENT_REINIT);
         etcp_fire_conn_status(conn, ETCP_CONN_STATUS_REINIT);
     }
+    if (conn->normalizer && !conn->close_requested) queue_resume_callback(conn->normalizer->input);
     etcp_conn_wake_sender(conn);
 }
 
@@ -848,7 +849,7 @@ static void input_queue_cb(struct ll_queue* q, void* arg) {
         DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot allocate INFLIGHT_PACKET (pool=%p etcp=%p)", etcp->log_name, etcp->inflight_pool, etcp);
         queue_dgram_free(&in_pkt->ll);
         queue_entry_free((struct ll_entry*)in_pkt);
-        queue_resume_callback(q);
+        etcp_conn_fatal_reinit(etcp, "TX inflight OOM");
         return;
     }
 
@@ -871,7 +872,7 @@ static void input_queue_cb(struct ll_queue* q, void* arg) {
     // При провале queue_data_put_with_index сам освобождает entry+dgram, повторный free дал бы double-free.
     if (queue_data_put_with_index(etcp->input_send_q, &p->ll) != 0) {
         DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to add packet seq=%u to input_send_q", etcp->log_name, p->seq);
-        DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] EXIT (queue put failed)", etcp->log_name);
+        etcp_conn_fatal_reinit(etcp, "TX inflight queue failure");
         return;
     }
     etcp->unacked_bytes += len;
@@ -1132,11 +1133,12 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) {
 //                etcp->log_name, etcp, etcp->last_delivered_id, queue_entry_count(etcp->recv_q));
 
     uint32_t next_expected_id = etcp->last_delivered_id + 1;
+    uint32_t reset = etcp->reset_count;
     int delivered_count = 0;
     uint32_t delivered_bytes = 0;
 
     // Look for contiguous packets starting from next_expected_id
-    while (1) {
+    while (!etcp->close_requested && !etcp->reinit_pending && etcp->reset_count == reset) {
         struct ETCP_FRAGMENT* rx_pkt = (struct ETCP_FRAGMENT*)queue_find_data_by_index(etcp->recv_q, &next_expected_id);
         if (!rx_pkt) {
             // No more contiguous packets found
@@ -1165,7 +1167,7 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) {
         }
         // pn_unpacker_cb (вызывается из put синхронно) мог ре-энтрантно сделать reinit/close:
         // при reinit last_delivered_id уже сброшен в 0 и recv_q очищен, при close state==2.
-        if (etcp->state == 2 || etcp->reinit_pending) break;
+        if (etcp->close_requested || etcp->state == 2 || etcp->reinit_pending || etcp->reset_count != reset) break;
 
         delivered_bytes += pkt_len;
         delivered_count++;
@@ -1230,7 +1232,7 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
 
     while (len >= 1) {
         // Ре-энтрантный reinit/close (из pn_unpacker_cb/etcp_int_recv) — дальше разбирать пакет нельзя.
-        if (etcp->state == 2 || etcp->reinit_pending) break;
+        if (etcp->close_requested || etcp->state == 2 || etcp->reinit_pending) break;
         uint8_t type = data[0];
 
         // Process sections as per protocol.txt

src/transport_layer/etcp_session.c

@@ -10,6 +10,7 @@
 
 static uint64_t read_id(const uint8_t* p) { uint64_t v; memcpy(&v,p,8); return be64toh(v); }
 static void write_id(uint8_t* p, uint64_t v) { v = htobe64(v); memcpy(p,&v,8); }
+static void session_schedule(struct ETCP_CONN* c, int delay_tb);
 void etcp_session_encode(uint8_t* data, uint8_t code, uint64_t sender, uint64_t target) {
     data[0] = code; write_id(data+1,sender); write_id(data+9,target);
 }
@@ -47,20 +48,25 @@ static void session_tick(void* arg) {
     }
     if (c->session_required) session_send(c,ETCP_SESSION_HELLO,c->peer_reset_id,0);
     if (c->session_candidate) session_send(c,ETCP_SESSION_CHALLENGE,c->session_candidate,c->session_challenge);
-    c->session_timer = uasync_set_timeout(c->instance->ua,SESSION_RETRY_TB,c,session_tick,"etcp_session");
+    session_schedule(c, SESSION_RETRY_TB);
 }
 
-static void session_schedule(struct ETCP_CONN* c) {
-    if (c->session_timer) return;
-    c->session_started = get_time_tb();
-    c->session_timer = uasync_set_timeout(c->instance->ua,1,c,session_tick,"etcp_session");
+static void session_schedule(struct ETCP_CONN* c, int delay_tb) {
+    if (c->session_timer || c->close_requested) return;
+    if (delay_tb == 1) c->session_started = get_time_tb();
+    c->session_timer = uasync_set_timeout(c->instance->ua, delay_tb, c, session_tick, "etcp_session");
+    if (!c->session_timer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] session timer OOM delay_tb=%d required=%u; closing",
+            c->log_name, delay_tb, c->session_required);
+        c->callbacks_running++; etcp_connection_close(c); c->callbacks_running--;
+    }
 }
 
 void etcp_session_start(struct ETCP_CONN* c) {
     etcp_session_cancel(c); c->session_required = 1;
     DEBUG_INFO(DEBUG_CATEGORY_ETCP,"[%s] session negotiation started local=%016llx peer=%016llx",c->log_name,
         (unsigned long long)c->reset_id,(unsigned long long)c->peer_reset_id);
-    session_schedule(c);
+    session_schedule(c, 1);
 }
 
 void etcp_session_observe(struct ETCP_CONN* c, uint64_t peer_epoch) {
@@ -74,7 +80,7 @@ void etcp_session_observe(struct ETCP_CONN* c, uint64_t peer_epoch) {
             }
         } while (!c->session_challenge);
     }
-    session_schedule(c);
+    session_schedule(c, 1);
 }
 
 int etcp_session_receive(struct ETCP_CONN* c, const uint8_t* data, size_t len) {