diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 74d7e6d1..476195b6 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -155,7 +155,7 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control allocation failed command=%u", m.command); return STREAM_ERR_MEMORY; } uint64_t queued_ms = get_time_tb() / 10; memcpy(e->data, &queued_ms, sizeof(queued_ms)); - m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch; + m.sender_epoch = p->own_epoch; m.receiver_epoch = m.command == STREAM_CMD_INFO_REQ ? 0 : p->other_epoch; stream_wire_header(e->dgram, &m); if (length) memcpy(e->dgram + STREAM_WIRE_HEADER, payload, length); e->len = STREAM_WIRE_HEADER + length; @@ -235,17 +235,26 @@ static void peer_sweep(struct stream_peer *p) { } /* Invalidation сохраняет транспорт чужих владельцев; старые DATA больше не имеют живого handle. */ -static void peer_reset(struct stream_peer *p, int reason) { +static void peer_reset(struct stream_peer *p, int reason, int rotate, struct STREAM_RECEIVE *keep) { peer_cancel_wait(p); struct ll_entry *e; while ((e = queue_data_get(p->control))) { queue_dgram_free(e); queue_entry_free(e); } - for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) flow_close(f, reason, 0); + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) { + if (f == keep) continue; + if (!rotate && f->outgoing && !f->info.handle && f->info.state < STREAM_RECEIVE_REQUESTED) { + f->info_sent = f->info_ready = f->info.info_ready = 0; f->probe = f->probe_sent_ms = 0; + f->info.local_epoch = f->info.peer_epoch = f->info.info_ms = 0; + f->info.supplier_metrics_valid = f->info.rtt_valid = f->info.path_rtt_valid = 0; + f->info.state = STREAM_RECEIVE_CONNECTING; + } else flow_close(f, reason, 0); + } p->other_epoch = p->highwater = 0; p->next_handle = 1; p->last_sent = NULL; - p->own_epoch = new_epoch(); peer_sweep(p); + if (rotate) p->own_epoch = new_epoch(); + peer_sweep(p); memset(&p->pressure, 0, sizeof(p->pressure)); memset(&p->requested, 0, sizeof(p->requested)); p->pressure_id = 0; p->cooling = 0; stream_load_init(&p->load, get_time_tb() / 10); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx reason=%d", - (unsigned long long)p->node_id, (unsigned long long)p->own_epoch, reason); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx rotate=%d reason=%d", + (unsigned long long)p->node_id, (unsigned long long)p->own_epoch, rotate, reason); } /* NCD подтверждает UP собственного владельца, даже если ETCP уже использовался BGP/другим стримом. */ @@ -262,7 +271,7 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v .subblock = f->subblock, .command = STREAM_CMD_CLOSED}; route_control(p, m, &reason, 1); } - peer_reset(p, STREAM_REASON_TRANSPORT); + peer_reset(p, STREAM_REASON_TRANSPORT, 1, NULL); } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id); DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream NCD event peer=%llx event=%d up=%d", (unsigned long long)p->node_id, event, p->up); } @@ -591,7 +600,7 @@ static void peer_pump(void *arg) { if (etcp_send(conn, e) != 0) { queue_dgram_free(e); queue_entry_free(e); DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control send failed peer=%llx", (unsigned long long)p->node_id); - peer_reset(p, STREAM_REASON_TRANSPORT); + peer_reset(p, STREAM_REASON_TRANSPORT, 1, NULL); } else sent = 1; queue_resume_callback(p->control); } else { @@ -712,8 +721,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t if (!p) p = peer_open(net, sender); if (!p || (m.receiver_epoch && m.receiver_epoch != p->own_epoch)) return; if (p->other_epoch && p->other_epoch != m.sender_epoch) { - peer_reset(p, STREAM_REASON_TRANSPORT); - /* Remote INFO начинает новое поколение; его следующий ответ подтвердит свежий local epoch. */ + peer_reset(p, STREAM_REASON_TRANSPORT, 0, NULL); + /* Изменившийся sender уже отличает пару поколений; own не вращаем в ответ, чтобы не получить ping-pong. */ } p->other_epoch = m.sender_epoch; uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0}; @@ -729,7 +738,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t return; } if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch - || (p->other_epoch && m.sender_epoch != p->other_epoch)) { + || (p->other_epoch && m.sender_epoch != p->other_epoch && m.command != STREAM_CMD_INFO_RESP)) { DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session peer=%llx command=%u handle=%llu sender=%llx/%llx receiver=%llx/%llx", (unsigned long long)sender, m.command, (unsigned long long)m.handle, (unsigned long long)m.sender_epoch, (unsigned long long)(p ? p->other_epoch : 0), @@ -744,7 +753,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale INFO probe peer=%llx", (unsigned long long)sender); } else if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &f->info, &start) == STREAM_OK && stream_metric_read(m.payload + STREAM_INFO_METRIC_OFF, &f->info.supplier_metrics, &f->info.supplier_metrics_valid) == STREAM_OK) { - p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1; peer_schedule(p); + if (p->other_epoch && p->other_epoch != m.sender_epoch) peer_reset(p, STREAM_REASON_TRANSPORT, 0, f); + p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1; + f->info.local_epoch = p->own_epoch; f->info.peer_epoch = p->other_epoch; peer_schedule(p); uint64_t rtt = get_time_tb() / 10 - f->probe_sent_ms; f->info.rtt_valid = f->probe_sent_ms != 0; f->info.rtt_ms = rtt > UINT32_MAX ? UINT32_MAX : (uint32_t)rtt; for (struct STREAM_RECEIVE *active = p->flows; active; active = active->next) { @@ -847,7 +858,7 @@ static void transport_changed(struct ETCP_CONN *conn, int status, void *arg) { struct stream_network *net = arg; if (status != ETCP_CONN_STATUS_REINIT || net->stopping) return; struct stream_peer *p = peer_find(net, conn->peer_node_id); - if (p) peer_reset(p, STREAM_REASON_TRANSPORT); + if (p) peer_reset(p, STREAM_REASON_TRANSPORT, 1, NULL); } /* FIFO ограничена и по времени. Отмена неподтверждённого SUBSCRIBE/ACCEPT не допускает DATA без живого согласования. */ diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index 3f16ef17..417c385b 100644 --- a/src/stream/stream_network.h +++ b/src/stream/stream_network.h @@ -19,7 +19,7 @@ struct STREAM_RECEIVE; enum stream_receive_state { STREAM_RECEIVE_CONNECTING, STREAM_RECEIVE_INFO, STREAM_RECEIVE_REQUESTED, STREAM_RECEIVE_ACTIVE, STREAM_RECEIVE_ENDED, STREAM_RECEIVE_CLOSED }; struct stream_receive_info { - uint64_t handle, peer, start, next, blocks, gaps; + uint64_t handle, peer, start, next, blocks, gaps, local_epoch, peer_epoch; enum stream_receive_state state; int reason; struct stream_metric_snapshot metrics, supplier_metrics;