Require monotonic consumer cursor across stream receive restart

tests/test_stream_service.c

@@ -325,16 +325,22 @@ static void test_network(void) {
     for (struct etcp_status_cbk_entry *cb = nodes[1]->conn_status_cbks; cb; cb = cb->next)
         if (cb->arg == stream_network_get(nodes[1])) { cb->fn(stable, ETCP_CONN_STATUS_REINIT, cb->arg); notified = 1; break; }
     CHECK(notified && stable->links_up);
-    /* Новый bootstrap может получить уже сохранённый блок: DUPLICATE обязан установить общий старт. */
+    /* Старый DUPLICATE устанавливает общий старт, но не повторяет уже выданную приложению пачку. */
     CHECK(stream_receive_restart(subscribers[1]) == STREAM_OK);
-    network_data[1] = 0; network_next[1] = 1; first_request[0] = 0;
+    first_request[0] = 0;
     for (unsigned i = 0; i < 8; i++) {
         struct stream_receive_info info; CHECK(stream_receive_get_info(up[i], &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED);
         stream_receive_close(up[i]); up[i] = NULL;
         CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, i, UINT64_MAX, &up[i]) == STREAM_OK);
     }
-    run_until(&network_data[1], 1);
     struct stream_receive_info current_generation; CHECK(stream_receive_get_info(up[0], &current_generation) == STREAM_OK);
+    deadline = get_time_tb() + 100000;
+    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(up[0], &current_generation) == STREAM_OK); }
+    while (!current_generation.blocks && get_time_tb() < deadline);
+    test_clock_run_for(ua, 1000);
+    struct stream_local restarted; CHECK(stream_local_get(nodes[1], &origin.descriptor, &restarted) == STREAM_OK);
+    CHECK(current_generation.blocks == 1 && restarted.anchored && restarted.start == 2 && restarted.next == 2);
+    CHECK(network_data[1] == 1 && network_next[1] == 2);
     CHECK(current_generation.local_epoch != previous_generation.local_epoch && current_generation.peer_epoch == previous_generation.peer_epoch);
     CHECK(stable == instance_find_conn(nodes[1], nodes[3]->node_id) && stable->links_up);
     /* Node2 видит подписку relay через обычный Merkle-обмен, а не копирование из его SQLite. */
@@ -727,7 +733,7 @@ static void test_automatic(void) {
     CHECK(automatic_data[2] == 400); stream_close(subscribers[2]); stream_close(publisher);
     subscribers[2] = publisher = NULL;
     test_clock_run_for(ua, 1000);
-    /* END первого субпотока не выключает выбор недостающих: для 3+1 нужны ещё данные второй выдачи. */
+    /* END первого субпотока не выключает выбор недостающих: для 2+1 нужны ещё данные второй выдачи. */
     parameters.n = 3;
     CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
     CHECK(stream_get_info(publisher, &origin) == STREAM_OK);