Prefer FEC parity cuts and refresh receive progress on inspection

src/stream/stream.c

@@ -57,8 +57,8 @@ static void cache_event(void *arg, const uint8_t *wire, size_t length, int added
     if (s->closing || s->invalidated) return;
     uint8_t sub = wire[STREAM_BLOCK_SUB_OFF]; uint64_t seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
     if (added) {
-        stream_metrics_observe(&s->metrics[sub], seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
-                               wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
+        if (!s->metrics[sub].ended) stream_metrics_observe(&s->metrics[sub], seq,
+            (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
         if (s->end_valid && seq == s->last_seq) stream_metrics_end(&s->metrics[sub]);
     }
     stream_network_cache_event(s->service->network, &s->descriptor, wire, length, added);

src/stream/stream_network.c

@@ -922,10 +922,10 @@ int stream_receive_restart(struct STREAM_HANDLE *subscription) {
     return stream_local_restart(inst, &d);
 }
 
-int stream_receive_get_info(const struct STREAM_RECEIVE *f, struct stream_receive_info *info) {
+int stream_receive_get_info(struct STREAM_RECEIVE *f, struct stream_receive_info *info) {
     if (!f || !info) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream receive info invalid arguments"); return STREAM_ERR_ARGUMENT; }
     *info = f->info;
-    /* Poll таймером сохраняет историю остановок; getter не меняет состояние const handle. */
+    info->metrics_valid = stream_metrics_poll(&f->metrics, get_time_tb() / 10, &info->metrics) == STREAM_OK;
     if (f->peer && f->peer->requested.active) for (unsigned i = 0; i < f->peer->requested.count; i++)
         if (f->peer->requested.cuts[i].selected && f->peer->requested.cuts[i].handle == f->info.handle) info->close_requested = 1;
     return STREAM_OK;
@@ -1021,7 +1021,8 @@ static int removal_cost(struct stream_peer *p, struct STREAM_RECEIVE *candidate)
             if (!selected && !seen[f->subblock]) { seen[f->subblock] = 1; distinct++; }
         }
     }
-    return distinct >= (unsigned)(candidate->descriptor.n - candidate->descriptor.r) ? 0 : 1;
+    int parity = candidate->subblock >= candidate->descriptor.n - candidate->descriptor.r;
+    return (distinct >= (unsigned)(candidate->descriptor.n - candidate->descriptor.r) ? 0 : 4) + (parity ? 0 : 1);
 }
 
 /* Выбор лишь предлагает source закрытия. Клиент не отсоединяет handles до CLOSED: поздний SELECT не режет лишние после fallback. */
@@ -1031,7 +1032,7 @@ static void pressure_select(struct stream_peer *p, uint64_t now) {
     uint64_t reduction = 0; unsigned selected = 0;
     for (unsigned i = 0; i < r->count; i++) if (r->cuts[i].selected) { reduction += r->cuts[i].rate; selected++; }
     while (reduction < r->required_rate) {
-        int best = -1, best_cost = 2;
+        int best = -1, best_cost = 8;
         for (unsigned i = 0; i < r->count; i++) {
             struct STREAM_RECEIVE *f = flow_by_handle(p, r->cuts[i].handle, 1);
             if (r->cuts[i].selected || !f) continue;

src/stream/stream_network.h

@@ -28,7 +28,7 @@ void stream_receive_close(struct STREAM_RECEIVE *receive);
 /* Отменить текущие upstream данного локального экземпляра и начать новую попытку с конца.
  * Буфер/подписка/downstream сохраняются; старые receive handles становятся CLOSED и всё равно требуют close. */
 int stream_receive_restart(struct STREAM_HANDLE *subscription);
-int stream_receive_get_info(const struct STREAM_RECEIVE *receive, struct stream_receive_info *info);
+int stream_receive_get_info(struct STREAM_RECEIVE *receive, struct stream_receive_info *info);
 struct stream_peer_load_info {
     struct stream_load_snapshot load;
     uint64_t request_id, required_rate, released_rate, deadline_ms, transport_queued_bytes, inflight_bytes;