From ee36c578e264dc51d5e617fc2d216c90ee0c05b2 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Sun, 11 Oct 2026 01:16:05 +0200
Subject: [PATCH] Reject premature stream completion and failed ACCEPT enqueue

---
 src/stream/stream_network.c | 26 +++++++++++++++++++-------
 1 file changed, 19 insertions(+), 7 deletions(-)

diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c
index e962033d..414e5b66 100644
--- a/src/stream/stream_network.c
+++ b/src/stream/stream_network.c
@@ -477,12 +477,16 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
         if (m.subblock < local->descriptor.n) stream_buffer_range(local->buffer, m.subblock, &first, &last);
         stream_write_be(rejection + 1, first, 8); stream_write_be(rejection + 9, last, 8);
         m.command = STREAM_CMD_REJECT; control_put(p, m, rejection, sizeof(rejection));
-        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u reason=%d watermark=%llu",
-                   (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, reason, (unsigned long long)p->highwater);
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u index=%llu reason=%d range=%llu..%llu watermark=%llu",
+                   (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)requested,
+                   reason, (unsigned long long)first, (unsigned long long)last, (unsigned long long)p->highwater);
         return;
     }
     f->deadline = get_time_tb() + STREAM_LEASE_TB;
-    m.command = STREAM_CMD_ACCEPT; control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8);
+    m.command = STREAM_CMD_ACCEPT;
+    if (control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8) != STREAM_OK) {
+        flow_close(f, STREAM_REASON_LIMIT, 0); return;
+    }
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream ACCEPT queued peer=%llx handle=%llu sub=%u start=%llu lease=6000ms",
                (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)f->info.start);
 }
@@ -531,8 +535,10 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
     }
     if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch
         || (p->other_epoch && m.sender_epoch != p->other_epoch)) {
-        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session packet peer=%llx command=%u handle=%llu",
-                    (unsigned long long)sender, m.command, (unsigned long long)m.handle); return;
+        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session peer=%llx command=%u handle=%llu sender=%llx/%llx receiver=%llx/%llx",
+                    (unsigned long long)sender, m.command, (unsigned long long)m.handle,
+                    (unsigned long long)m.sender_epoch, (unsigned long long)(p ? p->other_epoch : 0),
+                    (unsigned long long)m.receiver_epoch, (unsigned long long)(p ? p->own_epoch : 0)); return;
     }
     p->busy++;
     struct STREAM_RECEIVE *f = flow_find(p, &m, m.command != STREAM_CMD_SUBSCRIBE && m.command != STREAM_CMD_RENEW
@@ -578,7 +584,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
             || stream_read_be(m.payload + 8, 8) <= f->info.next) flow_close(f, STREAM_REASON_PROTOCOL, 1);
         else { f->info.next = stream_read_be(m.payload + 8, 8); f->info.gaps++; }
     } else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
-        if (stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED;
+        uint64_t last = m.length == STREAM_END_SIZE ? stream_read_be(m.payload + 34, 8) : UINT64_MAX;
+        if (m.length == STREAM_END_SIZE && last != UINT64_MAX && f->info.next == last + 1
+            && stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED;
         else flow_close(f, STREAM_REASON_PROTOCOL, 1);
     } else if (m.command == STREAM_CMD_RENEW && f && !m.length) {
         f->deadline = get_time_tb() + STREAM_LEASE_TB;
@@ -749,7 +757,11 @@ int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8
             if (f->outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d)) bootstrap = 0;
     struct STREAM_RECEIVE *f = flow_new(p, &d, sub, 1);
     if (!f) return STREAM_ERR_MEMORY;
-    f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; *out = f; peer_schedule(p); return STREAM_OK;
+    f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; *out = f; peer_schedule(p);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive opened peer=%llx group=%llx source=%llx epoch=%llx sub=%u index=%llu bootstrap=%d",
+               (unsigned long long)node, (unsigned long long)d.group_id, (unsigned long long)d.source_node_id,
+               (unsigned long long)d.source_epoch, sub, (unsigned long long)seq, bootstrap);
+    return STREAM_OK;
 }
 
 void stream_receive_close(struct STREAM_RECEIVE *f) {