Expire stale stream control without retaining orphan issuances

src/stream/stream_network.c

@@ -148,8 +148,13 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin
                    (unsigned long long)p->node_id, p->control->count, m.command, length);
         p->load.rejected++; p->network->load.rejected++; return STREAM_ERR_STATE;
     }
-    struct ll_entry *e = ll_alloc_lldgram((uint16_t)(STREAM_WIRE_HEADER + length));
-    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control allocation failed command=%u", m.command); return STREAM_ERR_MEMORY; }
+    struct ll_entry *e = queue_entry_new(sizeof(uint64_t));
+    if (e) e->dgram = u_malloc((uint32_t)(STREAM_WIRE_HEADER + length));
+    if (!e || !e->dgram) {
+        if (e) queue_entry_free(e);
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control allocation failed command=%u", m.command); return STREAM_ERR_MEMORY;
+    }
+    uint64_t queued_ms = get_time_tb() / 10; memcpy(e->data, &queued_ms, sizeof(queued_ms));
     m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
     stream_wire_header(e->dgram, &m);
     if (length) memcpy(e->dgram + STREAM_WIRE_HEADER, payload, length);
@@ -760,6 +765,24 @@ static void transport_changed(struct ETCP_CONN *conn, int status, void *arg) {
     if (p) peer_reset(p, STREAM_REASON_TRANSPORT);
 }
 
+/* FIFO ограничена и по времени. Отмена неподтверждённого SUBSCRIBE/ACCEPT не допускает DATA без живого согласования. */
+static void expire_controls(struct stream_peer *p, uint64_t now) {
+    while (p->control->head) {
+        struct ll_entry *e = p->control->head; uint64_t queued_ms; memcpy(&queued_ms, e->data, sizeof(queued_ms));
+        if (now - queued_ms <= STREAM_LOAD_MAX_AGE_MS) break;
+        e = queue_data_get(p->control); struct stream_message m;
+        if (stream_wire_parse(e->dgram, e->len, &m) == STREAM_OK) {
+            int outgoing = m.command == STREAM_CMD_INFO_REQ || m.command == STREAM_CMD_SUBSCRIBE;
+            struct STREAM_RECEIVE *f = flow_find(p, &m, outgoing);
+            if (f && (outgoing || m.command == STREAM_CMD_ACCEPT)) flow_close(f, STREAM_REASON_OVERLOAD, 0);
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control expired peer=%llx command=%u handle=%llu age=%llu",
+                       (unsigned long long)p->node_id, m.command, (unsigned long long)m.handle, (unsigned long long)(now - queued_ms));
+        }
+        p->load.rejected++; p->network->load.rejected++;
+        queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(p->control);
+    }
+}
+
 /* Один таймер обслуживает сроки/продления и ограниченные reverse-попытки; DATA не опрашивается таймером. */
 static void network_tick(void *arg) {
     struct stream_network *net = arg; net->timer = NULL;
@@ -771,6 +794,7 @@ static void network_tick(void *arg) {
     while (p) {
         struct stream_peer *next = p->next;
         p->busy++;
+        expire_controls(p, now_ms);
         for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
             if (f->info.state == STREAM_RECEIVE_CLOSED) continue;
             f->info.metrics_valid = stream_metrics_poll(&f->metrics, now_ms, &f->info.metrics) == STREAM_OK;