Verify independent stream groups share one direct ETCP session

tests/test_stream_service.c

@@ -34,6 +34,7 @@ static unsigned data_events[4], end_events[4], closed_events[4];
 static int close_readers_in_callback;
 static unsigned network_data[4], network_end[4], accepts, reverse_requests, dropped_renew;
 static unsigned network_gaps[4];
+static unsigned parallel_data;
 static uint64_t network_next[4] = {0, 1, 1, 0};
 static uint8_t first_request[STREAM_WIRE_HEADER + 8];
 static etcp_recv_fn saved_raw[4], saved_route;
@@ -59,7 +60,7 @@ static void network_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
                 stream_write_be(entry->dgram + STREAM_WIRE_HEADER + 8, 3, 8);
             }
         }
-        if (node == 3 && command == STREAM_CMD_RENEW && drop_renew) {
+        if (node == 3 && command == STREAM_CMD_RENEW && drop_renew && stream_read_be(entry->dgram + 4, 8) == GROUP2) {
             dropped_renew++; queue_dgram_free(entry); queue_entry_free(entry); return;
         }
     }
@@ -74,7 +75,15 @@ static void network_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
 
 static void on_network(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
     (void)arg; unsigned node = inst == nodes[1] ? 1 : 2;
-    CHECK(event->descriptor.group_id == GROUP2 && event->descriptor.source_node_id == nodes[3]->node_id);
+    CHECK(event->descriptor.source_node_id == nodes[3]->node_id);
+    if (event->descriptor.group_id == GROUP1) {
+        if (event->type == STREAM_EVENT_DATA) {
+            CHECK(node == 1 && event->first_seq == parallel_data && event->length == sizeof(expected));
+            CHECK(!memcmp(event->data, expected, sizeof(expected))); parallel_data++;
+        } else CHECK(event->type == STREAM_EVENT_CLOSED);
+        return;
+    }
+    CHECK(event->descriptor.group_id == GROUP2);
     if (event->type == STREAM_EVENT_DATA) {
         CHECK(event->length == sizeof(expected) && !memcmp(event->data, expected, sizeof(expected)));
         CHECK(event->first_seq == network_next[node]++); network_data[node]++;
@@ -260,6 +269,21 @@ static void test_network(void) {
     memcpy(duplicate->dgram, first_request, sizeof(first_request)); duplicate->len = sizeof(first_request); CHECK(etcp_send(conn, duplicate) == 0);
     run_until(&accepts, before + 1);
     struct stream_receive_info info; CHECK(stream_receive_get_info(up[0], &info) == STREAM_OK && info.blocks == 1 && info.next == 2);
+    /* Вторая группа на том же peer/NCD/ETCP использует иной FEC и сохраняется после закрытия выдач GROUP2. */
+    struct STREAM_HANDLE *parallel_publisher = NULL, *parallel_subscriber = NULL;
+    struct STREAM_RECEIVE *parallel = NULL; struct stream_parameters parameters = {.n = 1, .r = 0, .block_size = STREAM_RS_DEFAULT_SIZE};
+    CHECK(stream_publish_open(nodes[3], GROUP1, &parameters, &parallel_publisher) == STREAM_OK);
+    struct stream_info parallel_info; CHECK(stream_get_info(parallel_publisher, &parallel_info) == STREAM_OK);
+    wait_state(1, 3, GROUP1, parallel_info.descriptor.source_epoch, 0);
+    CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[3]->node_id, &parallel_subscriber) == STREAM_OK);
+    size_t accepted = 0;
+    CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
+    deadline = get_time_tb() + 100000;
+    do { test_clock_poll(ua, 100); CHECK(stream_get_info(parallel_publisher, &parallel_info) == STREAM_OK); }
+    while (!parallel_info.buffer.blocks && get_time_tb() < deadline);
+    CHECK(parallel_info.buffer.blocks == 1);
+    CHECK(stream_receive_open(parallel_subscriber, nodes[3]->node_id, 0, UINT64_MAX, &parallel) == STREAM_OK);
+    run_until(&parallel_data, 1);
     /* Одна и та же routed reverse-попытка не меняет сессию/действующие handles. */
     struct stream_message reverse; CHECK(stream_wire_parse(first_request, sizeof(first_request), &reverse) == STREAM_OK);
     reverse.command = STREAM_CMD_REVERSE_REQ; reverse.receiver_epoch = reverse.handle = 0;
@@ -281,6 +305,9 @@ static void test_network(void) {
     drop_renew = 1; test_clock_run_for(ua, 71000); drop_renew = 0;
     CHECK(dropped_renew && stream_receive_get_info(up[0], &info) == STREAM_OK
           && info.state == STREAM_RECEIVE_CLOSED && info.reason == STREAM_REASON_LEASE);
+    CHECK(stream_receive_get_info(parallel, &info) == STREAM_OK && info.state == STREAM_RECEIVE_ACTIVE);
+    CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
+    run_until(&parallel_data, 2);
     CHECK(stream_handle_active(reader1) && stream_handle_active(reader2));
     /* REINIT не обязан разрушать ETCP_CONN, но старые stream handles становятся недействительны. */
     conn = instance_find_conn(nodes[2], nodes[1]->node_id); CHECK(conn != NULL);
@@ -291,6 +318,7 @@ static void test_network(void) {
     CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED
           && info.reason == STREAM_REASON_TRANSPORT);
     for (unsigned i = 0; i < 8; i++) { stream_receive_close(up[i]); stream_receive_close(down[i]); }
+    stream_receive_close(parallel); stream_close(parallel_subscriber); stream_close(parallel_publisher);
     for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, saved_raw[i]) == 0);
     CHECK(etcp_router_bind(nodes[3], ETCP_RT_ID_STREAM, saved_route) == 0);
     for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);