diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 828f4ffd..74d7e6d1 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -270,7 +270,17 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v /* На пира ровно один stream NCD owner, общий для всех его выдач и групп. */ static struct stream_peer *peer_open(struct stream_network *net, uint64_t node) { struct stream_peer *p = peer_find(net, node); - if (p) return p; + if (p) { + if (!node_conn_direct_get_conn(p->ncd)) { + node_conn_direct_set_callback(p->ncd, NULL, NULL); node_conn_direct_close(p->ncd); p->ncd = NULL; + p->born = get_time_tb(); p->reverse_at = 0; + if (node_conn_direct_open(net->inst, node, ncd_changed, p, &p->ncd, NULL) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer NCD reopen failed peer=%llx", (unsigned long long)node); return NULL; + } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer NCD reopened peer=%llx", (unsigned long long)node); + } + return p; + } if (net->peer_count == STREAM_PEER_LIMIT) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream peer limit reached"); return NULL; } p = u_calloc(1, sizeof(*p)); if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer allocation failed"); return NULL; } @@ -885,8 +895,11 @@ static void network_tick(void *arg) { if (control_put(p, m, NULL, 0) == STREAM_OK) f->renewed = now; } } - if (!p->up && p->flows && now - p->born < STREAM_CONNECT_TB && (!p->reverse_at || now - p->reverse_at >= 10000)) { - struct stream_message m = {.descriptor = p->flows->descriptor, .command = STREAM_CMD_REVERSE_REQ}; + struct STREAM_RECEIVE *pending = NULL; + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) + if (f->outgoing && f->info.state < STREAM_RECEIVE_ACTIVE && now < f->deadline) { pending = f; break; } + if (!p->up && pending && (!p->reverse_at || now - p->reverse_at >= 10000)) { + struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ}; uint64_t saved = p->other_epoch; p->other_epoch = 0; if (route_control(p, m, NULL, 0) == STREAM_OK) p->reverse_at = now; p->other_epoch = saved;