Account variable stream parts and continue after rejected batches

src/stream/stream.c

@@ -306,6 +306,12 @@ static void session_deliver(void *arg) {
             }
             event.type = STREAM_EVENT_DATA; event.first_seq = event.last_seq = s->next_seq++;
             event.data = s->output; event.length = length; emit = 1; again = 1;
+        } else if (result == STREAM_REJECTED) {
+            event.type = STREAM_EVENT_GAP; event.first_seq = event.last_seq = s->next_seq++;
+            emit = again = 1;
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream rejected batch GAP source=%llx epoch=%llx seq=%llu; receiving continues",
+                       (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch,
+                       (unsigned long long)event.first_seq);
         } else if (result == STREAM_EXPIRED) {
             event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = stats.expired_before - 1;
             s->next_seq = stats.expired_before; emit = 1; again = 1;

src/stream/stream_network.c

@@ -131,7 +131,7 @@ static uint64_t flow_age(struct STREAM_RECEIVE *f, uint64_t now) {
     if (f->outgoing || !f->load.queued.packets) return 0;
     struct stream_local local; uint64_t oldest = 0;
     if (stream_local_get(f->peer->network->inst, &f->descriptor, &local) != STREAM_OK) return 0;
-    if (stream_buffer_pending(local.buffer, f->info.next, f->subblock, NULL, &oldest) != STREAM_OK) return 0;
+    if (stream_buffer_pending(local.buffer, f->info.next, f->subblock, NULL, NULL, &oldest) != STREAM_OK) return 0;
     if (oldest < f->admitted_ms) oldest = f->admitted_ms;
     return now > oldest ? now - oldest : 0;
 }
@@ -736,9 +736,9 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
                 f->info.handle = m.handle; f->requested = requested; f->info.start = f->info.next = stream_read_be(payload + 8, 8);
                 if (f->info.start == UINT64_MAX) { f->info.start = f->info.next = 0; stream_write_be(payload + 8, 0, 8); }
                 f->info.state = STREAM_RECEIVE_ACTIVE;
-                uint64_t count = 0;
-                stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL);
-                load_account(f, count, count * stream_block_size(&f->descriptor), 1, 0);
+                uint64_t count = 0, bytes = 0;
+                stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, &bytes, NULL);
+                load_account(f, count, bytes, 1, 0);
                 f->pending_highest = stream_read_be(payload + 24, 8);
                 if (f->pending_highest == UINT64_MAX) f->pending_highest = 0;
                 const uint8_t *wire; size_t length; uint64_t seq;
@@ -886,9 +886,14 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
         }
     } else if (m.command == STREAM_CMD_DATA && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
         struct stream_path path; size_t used;
-        if (stream_path_parse(m.payload, m.length, &path, &used) != STREAM_OK || m.length - used != stream_block_size(&f->descriptor)
+        if (stream_path_parse(m.payload, m.length, &path, &used) != STREAM_OK
             || stream_path_validate(&path, f->descriptor.source_node_id, sender, net->inst->node_id) != STREAM_OK) {
             flow_close(f, STREAM_REASON_LOOP, 1);
+        } else if (m.length - used <= STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE
+                   || m.length - used > stream_block_capacity(&f->descriptor)) {
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA part size invalid peer=%llx bytes=%zu max=%zu",
+                       (unsigned long long)sender, m.length - used, stream_block_capacity(&f->descriptor));
+            flow_close(f, STREAM_REASON_PROTOCOL, 1);
         } else {
             const uint8_t *block = m.payload + used; uint64_t seq = stream_read_be(block + STREAM_BLOCK_SEQ_OFF, 8);
             if (block[STREAM_BLOCK_SUB_OFF] != f->subblock || seq != f->info.next
@@ -899,7 +904,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
                 int result = stream_ingest_path(net->inst, block, m.length - used, &path);
                 if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) {
                     stream_metrics_observe(&f->metrics, seq, (uint16_t)stream_read_be(block + STREAM_BLOCK_TIMESTAMP_OFF, 2),
-                                           block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)stream_block_size(&f->descriptor), get_time_tb() / 10);
+                                           block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)(m.length - used), get_time_tb() / 10);
                     if (f->info.path.count != path.count || memcmp(f->info.path.nodes, path.nodes, path.count * sizeof(uint64_t)))
                         f->info.path_rtt_valid = 0;
                     f->info.next = seq + 1; f->info.blocks++; f->info.path = path; f->deadline = get_time_tb() + STREAM_LEASE_TB;

src/stream/stream_wire.h

@@ -10,7 +10,7 @@
 extern "C" {
 #endif
 
-#define STREAM_WIRE_VERSION 4
+#define STREAM_WIRE_VERSION 5
 #define STREAM_WIRE_HEADER 52
 #define STREAM_PATH_MAX 32
 #define STREAM_END_SIZE 106