diff --git a/src/stream/stream.c b/src/stream/stream.c index e7bddbbe..13647aeb 100644 --- a/src/stream/stream.c +++ b/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; владельцем остаётся сама подписка. */ diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index 417c385b..755f3453 100644 --- a/src/stream/stream_network.h +++ b/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 {