Preserve delivered cursor when stream restarts from a stale supplier

src/stream/stream.c

@@ -604,8 +604,12 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l
     int result = stream_buffer_put_path(s->buffer, wire, length, &retained);
     if (result == STREAM_OK || (result == STREAM_DUPLICATE && !s->anchored)) {
         if (!s->anchored) {
-            s->anchored = 1; s->start_seq = s->next_seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
+            uint64_t received = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
+            uint64_t start = s->restarting && received < s->restart_from ? s->restart_from : received;
+            s->anchored = 1; s->start_seq = s->next_seq = start;
             if (s->restarting && s->next_seq > s->restart_from) { s->restart_gap = 1; s->restart_last = s->next_seq - 1; }
+            if (s->restarting) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream restart anchored source=%llx received=%llu floor=%llu start=%llu",
+                (unsigned long long)source, (unsigned long long)received, (unsigned long long)s->restart_from, (unsigned long long)start);
             s->restarting = 0;
         }
         session_schedule(s);
@@ -684,11 +688,15 @@ int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descrip
     }
     struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
     if (s->delivery) { uasync_call_soon_cancel(inst->ua, s->delivery); s->delivery = NULL; }
-    if (s->anchored) { s->restarting = 1; s->restart_from = s->next_seq; }
+    if (s->anchored) {
+        if (!s->restart_gap) s->restart_from = s->next_seq;
+        s->restarting = 1;
+    }
+    s->restart_gap = 0;
     s->anchored = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0;
-    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u",
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u floor=%llu known=%d",
                (unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
-               (unsigned long long)key->source_epoch, local.buffer != NULL); return STREAM_OK;
+               (unsigned long long)key->source_epoch, local.buffer != NULL, (unsigned long long)s->restart_from, s->restarting); return STREAM_OK;
 }
 
 /* Заимствованный локальный handle нужен только на время tick; владельцем остаётся сама подписка. */

src/stream/stream_network.h

@@ -41,7 +41,8 @@ int stream_probe_next(struct STREAM_RECEIVE *probe, uint8_t subblock);
 int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t peer, uint8_t subblock, uint64_t seq, struct STREAM_RECEIVE **out);
 void stream_receive_close(struct STREAM_RECEIVE *receive);
 /* Отменить текущие upstream данного локального экземпляра и начать новую попытку с конца.
- * Буфер/подписка/downstream сохраняются; старые receive handles становятся CLOSED и всё равно требуют close. */
+ * Буфер/подписка/downstream и нижняя граница consumer cursor сохраняются: уже выданные DATA не повторяются.
+ * Старые receive handles становятся CLOSED и всё равно требуют close. */
 int stream_receive_restart(struct STREAM_HANDLE *subscription);
 int stream_receive_get_info(struct STREAM_RECEIVE *receive, struct stream_receive_info *info);
 struct stream_peer_load_info {