etcp: фикс double-free буферов паковщика и повторных free после неудачного queue_data_put

- pn_send_to_etcp/pn_unpacker_cb: занулять pn->data/pn->recvpart ДО queue_data_put
  (ре-энтрантный close/reinit в синхронной цепочке освобождал буфер повторно)
- etcp_output_try_assembly: гард state==2/reinit_pending + убрать re-put освобождённого entry
- etcp_conn_input: гард ре-энтрантного reinit/close в цикле секций, локальный etcp вместо pkt->link
- убраны повторные free после queue_data_put (он сам освобождает entry+dgram) в 6 местах
- диагностика: obj=%p в DOUBLE FREE, логи free dgram в pn_deinit/ack_recv

src/call/call.c

@@ -158,9 +158,11 @@ static struct call_session* call_session_new(struct call_ctx* ctx, uint64_t call
     s->ctx = ctx;
     s->tx_q = queue_new(ctx->inst->ua, 0, 0, 0, "call_txq");
     if (!s->tx_q) { queue_entry_free(qe); return NULL; }
+    // При провале queue_data_put_with_index сам освобождает qe (вместе с s), поэтому
+    // сохраняем tx_q заранее: queue_entry_free(qe) повторно дал бы double-free.
+    struct ll_queue* tx_q = s->tx_q;
     if (queue_data_put_with_index(ctx->sessions, qe) != 0) {
-        queue_free(s->tx_q);
-        queue_entry_free(qe);
+        queue_free(tx_q);
         return NULL;
     }
     return s;

src/chat/merkle_sync.c

@@ -342,7 +342,8 @@ static int ms_outq_enqueue(struct ms_session* s, const uint8_t* payload, size_t
     e->dgram = buf; e->len = (uint16_t)(1 + plen);
 
     int was_empty = (queue_entry_count(s->out_q) == 0);
-    if (queue_data_put(s->out_q, e) != 0) { u_free(buf); queue_entry_free(e); return -1; }
+    // При провале queue_data_put сам освобождает entry+dgram, повторный free дал бы double-free.
+    if (queue_data_put(s->out_q, e) != 0) return -1;
     if (was_empty && conn->send_input_q)
         queue_waiter_wait(conn->send_input_q, &s->waiter, ms_outq_drain_cb, s);
     return 0;

src/transport_layer/dummynet.c

@@ -304,10 +304,10 @@ static void dummynet_delay_callback(void* user_arg) {
 
     /* Добавляем в очередь */
     uint32_t id = (uint32_t)(uintptr_t)entry;
+    // При провале queue_data_put сам освобождает entry, повторный free дал бы double-free.
     if (queue_data_put(dir->queue, entry) != 0) {
         dir->stats.dropped++;
         DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Dir %d: queue_data_put failed", dir_idx);
-        queue_entry_free(entry);
         return;
     }
 

src/transport_layer/etcp.c

@@ -786,10 +786,9 @@ int etcp_int_send(struct ETCP_CONN* etcp, const void* data, uint16_t len) {
     DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] created PACKET %p with data %p (len=%zu)", etcp->log_name, pkt, packet_data, len);
 
     // Add to input queue - input_queue_cb will process it
+    // При провале queue_data_put сам освобождает entry+dgram, повторный free дал бы double-free.
     if (queue_data_put(etcp->input_queue, (struct ll_entry*)pkt) != 0) {
         DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to add to input queue", etcp->log_name);
-        queue_dgram_free(&pkt->ll);
-        queue_entry_free(&pkt->ll);
         return -1;
     }
 
@@ -923,10 +922,9 @@ static void input_queue_cb(struct ll_queue* q, void* arg) {
     int len=p->ll.len;// сохраним len
 
     // Add to send queue
+    // При провале 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);
-        queue_dgram_free(&p->ll);
-        queue_entry_free(&p->ll);
         DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] EXIT (queue put failed)", etcp->log_name);
         return;
     }
@@ -1397,16 +1395,18 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) {
         DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] moving packet id=%u to output_queue (qlen=%d)", etcp->log_name,
                    next_expected_id, etcp->output_queue->count);
         uint16_t pkt_len = rx_pkt->ll.len;
-        if (queue_data_put(etcp->output_queue, (struct ll_entry*)rx_pkt) == 0) {
-            delivered_bytes += pkt_len;
-            delivered_count++;
-        } else {
+        // На провале queue_data_put сам освобождает rx_pkt — re-put дал бы double-free.
+        if (queue_data_put(etcp->output_queue, (struct ll_entry*)rx_pkt) != 0) {
             DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to add packet id=%u to output_queue", etcp->log_name,
                        next_expected_id);
-            // Put it back in recv_q if we can't add to output_queue
-            queue_data_put_with_index(etcp->recv_q, (struct ll_entry*)rx_pkt);
             break;
         }
+        // pn_unpacker_cb (вызывается из put синхронно) мог ре-энтрантно сделать reinit/close:
+        // при reinit last_delivered_id уже сброшен в 0 и recv_q очищен, при close state==2.
+        if (etcp->state == 2 || etcp->reinit_pending) break;
+
+        delivered_bytes += pkt_len;
+        delivered_count++;
 
         // Update state for next iteration
         etcp->last_delivered_id = next_expected_id;
@@ -1506,6 +1506,7 @@ void etcp_ack_recv(struct ETCP_CONN* etcp, uint32_t seq, uint16_t ts, uint16_t d
     DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TX removed packet seq=%u from wait_ack, unacked_bytes now %u total acked=%u", etcp->log_name, seq, etcp->unacked_bytes, etcp->ack_packets_count);
 
     if (acked_pkt->ll.dgram) {
+        DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TX-ACK free dgram=%p seq=%u", etcp->log_name, acked_pkt->ll.dgram, seq);
         memory_pool_free(etcp->instance->data_pool, acked_pkt->ll.dgram);
     }
     memory_pool_free(etcp->inflight_pool, acked_pkt);
@@ -1542,6 +1543,8 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
     DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RX pkt dlen=%d", etcp->log_name, len);
 
     while (len >= 1) {
+        // Ре-энтрантный reinit/close (из pn_unpacker_cb/etcp_int_recv) — дальше разбирать пакет нельзя.
+        if (etcp->state == 2 || etcp->reinit_pending) break;
         uint8_t type = data[0];
 
         // Process sections as per protocol.txt
@@ -1812,7 +1815,8 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
 
     }
 
-    if (memory_pool_is_freed(pkt->link->etcp->instance->pkt_pool, pkt)) {
+    // Используем локальный etcp (валиден до phase 2 close): pkt->link мог быть освобождён ре-энтрантным close.
+    if (memory_pool_is_freed(etcp->instance->pkt_pool, pkt)) {
         DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pkt=%p ALREADY FREED in pkt_pool — HALTING", (void*)pkt);
         volatile int _halt = 1; while (_halt) {}
     }

src/transport_layer/pkt_normalizer.c

@@ -116,6 +116,7 @@ void pn_deinit(struct PKTNORM* pn) {
     }
 
     if (pn->data) {
+        DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "pn_deinit: free pn->data=%p ptr=%d", pn->data, pn->data_ptr);
         memory_pool_free(pn->etcp->instance->data_pool, pn->data);
     }
     if (pn->recvpart) {
@@ -244,14 +245,17 @@ static void pn_send_to_etcp(struct PKTNORM* pn) {
     frag->ll.len        = pn->data_ptr;
     frag->ll.dgram_pool = pn->etcp->instance->data_pool;
     frag->ll.memlen     = pn->etcp->instance->data_pool->object_size;
-    
-    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "pn->etcp: size=%d memlen=%d frag_size=%d", frag->ll.len, frag->ll.memlen, pn->frag_size);
-    if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP)) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP, "NORM->ETCP", pn->data, frag->ll.len);
 
-    queue_data_put(pn->etcp->input_queue, (struct ll_entry*)frag);
-    // Сбросить структуру (dgram передан во фрагмент, не освобождаем)
+    // Зануляем pn->data ДО queue_data_put: последняя синхронно гонит input_queue_cb → send_q → wait_ack,
+    // в которой может ре-энтрантно сработать close → pn_deinit → free(pn->data).
+    // Иначе pn_deinit освободит буфер, уже переданный во INFLIGHT-запись (double free).
     pn->data = NULL;
     pn->data_ptr = 0;
+
+    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "pn->etcp: size=%d memlen=%d frag_size=%d", frag->ll.len, frag->ll.memlen, pn->frag_size);
+    if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP)) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP, "NORM->ETCP", frag->ll.dgram, frag->ll.len);
+
+    queue_data_put(pn->etcp->input_queue, (struct ll_entry*)frag);
 }
 
 // Internal: Renew sndpart buffer
@@ -404,10 +408,13 @@ static void pn_unpacker_cb(struct ll_queue* q, void* arg) {
                 uint32_t recv_len = pn->recvpart->len;
                 DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "unpacked dgram (size=%d)", recv_len);
                 if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP)) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DUMP, "NORM->", pn->recvpart->dgram, recv_len);
-                queue_data_put(pn->output, pn->recvpart);
+                // Зануляем recvpart ДО queue_data_put: etcp_int_recv может ре-энтрантно вызвать
+                // reinit → pn_unpacker_reset_state → free(pn->recvpart), что даст double-free.
+                struct ll_entry* out = pn->recvpart;
+                pn->recvpart = NULL;
+                queue_data_put(pn->output, out);
                 pn->out_total_pkts++;
                 pn->out_total_bytes += recv_len;
-                pn->recvpart = NULL;
             }
         }
 

src/tun_if.c

@@ -72,10 +72,9 @@ static void tun_read_callback(int fd, void* user_arg)
     tun->packets_read++;
 
     // Add to output queue (TUN → routing)
+    // При провале queue_data_put сам освобождает entry+dgram, повторный free дал бы double-free.
     if (queue_data_put(tun->output_queue, pkt) != 0) {
         DEBUG_ERROR(DEBUG_CATEGORY_TUN, "Failed to add packet to output queue");
-        u_free(packet_data);
-        queue_entry_free(pkt);
         tun->read_errors++;
         return;
     }