Reproduce automatic stream stall after early END with incomplete FEC batch

tests/test_stream_service.c

@@ -50,6 +50,7 @@ static unsigned global_data[4];
 static unsigned automatic_data[4], automatic_gaps[4], automatic_end[4], automatic_probes[4], automatic_accepts;
 static uint64_t automatic_next[4];
 static int automatic_drop_relay;
+static int automatic_drop_tail;
 static etcp_recv_fn automatic_handlers[4];
 static uint8_t automatic_request[STREAM_WIRE_HEADER + 1];
 err_t __real_uasync_get_load(struct UASYNC *async, struct uasync_load *load);
@@ -79,6 +80,9 @@ static void automatic_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
         if (automatic_drop_relay && node == 2 && conn->peer_node_id == nodes[1]->node_id && command == STREAM_CMD_DATA) {
             queue_dgram_free(entry); queue_entry_free(entry); return;
         }
+        if (automatic_drop_tail && node == 2 && entry->dgram[3] && (command == STREAM_CMD_DATA || command == STREAM_CMD_END)) {
+            queue_dgram_free(entry); queue_entry_free(entry); return;
+        }
     }
     automatic_handlers[node](conn, entry);
 }
@@ -87,7 +91,7 @@ static void on_automatic(struct UTUN_INSTANCE *inst, const struct stream_event *
     (void)arg; unsigned node = inst == nodes[1] ? 1 : 2;
     CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
     if (event->type == STREAM_EVENT_DATA) {
-        CHECK(event->length == 1180 && !memcmp(event->data, pressure_payload, 1180));
+        CHECK(event->length == stream_input_size(&event->descriptor) && !memcmp(event->data, pressure_payload, event->length));
         CHECK(event->first_seq == automatic_next[node]++); automatic_data[node]++;
     } else if (event->type == STREAM_EVENT_GAP) {
         CHECK(event->first_seq == automatic_next[node] && event->last_seq >= event->first_seq);
@@ -722,9 +726,35 @@ static void test_automatic(void) {
     CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&automatic_end[2], 2);
     CHECK(automatic_data[2] == 400); stream_close(subscribers[2]); stream_close(publisher);
     test_clock_run_for(ua, 1000);
+    /* END первого субпотока не выключает выбор недостающих: для 3+1 нужны ещё данные второй выдачи. */
+    parameters.n = 3;
+    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
+    CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
+    for (unsigned i = 1; i < 3; i++) {
+        wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
+        CHECK(stream_subscribe_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK);
+        CHECK(stream_selection_configure(subscribers[i], &settings) == STREAM_OK);
+        automatic_next[i] = 0;
+    }
+    automatic_drop_tail = 1;
+    test_clock_run_for(ua, 10000);
+    size_t input_size = stream_input_size(&origin.descriptor);
+    memset(pressure_payload, 0xa9, input_size);
+    CHECK(stream_publish_write(publisher, pressure_payload, input_size, &accepted) == STREAM_OK && accepted == input_size);
+    CHECK(stream_publish_finish(publisher) == STREAM_OK);
+    struct stream_local terminal;
+    deadline = get_time_tb() + 100000;
+    do { test_clock_poll(ua, 100); CHECK(stream_local_get(nodes[2], &origin.descriptor, &terminal) == STREAM_OK); }
+    while (!terminal.finished && get_time_tb() < deadline);
+    CHECK(terminal.finished && terminal.anchored && !terminal.next && automatic_data[2] == 400 && automatic_end[2] == 2);
+    automatic_drop_tail = 0;
+    run_until(&automatic_data[2], 401); run_until(&automatic_end[2], 3);
+    CHECK(automatic_next[2] == 1 && !automatic_gaps[2]);
+    for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
+    test_clock_run_for(ua, 1000);
     for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_handlers[i]) == 0);
     stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
-    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] automatic streams: INFO-only empty source, shared owners, 30s cycle, FEC spread, overlap, stall and CLOSED failover");
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] automatic streams: INFO-only, shared owners, 30s cycle, FEC spread, overlap, stall, CLOSED and early END drain");
 }
 
 /* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */