Bound stale offer storage, detect initial DATA stalls and report verified FEC protection

src/stream/stream.c

@@ -372,7 +372,8 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
     struct stream_session *s = session_find(service, d->group_id, d->source_node_id);
     int created = s == NULL;
     if (s && (publisher || s->descriptor.source_epoch != d->source_epoch || (!s->source && !!s->optimizer != automatic))) {
-        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source already owned with another publication/epoch"); return STREAM_ERR_STATE;
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream instance conflicts with publication/epoch/selection mode automatic=%d existing_auto=%d",
+                   automatic, s->optimizer != NULL); return STREAM_ERR_STATE;
     }
     struct STREAM_HANDLE *h = u_calloc(1, sizeof(*h));
     if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream handle allocation failed"); return STREAM_ERR_MEMORY; }
@@ -702,10 +703,10 @@ int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate
         if (node->node_id == inst->node_id || !stream_member_online(inst, group->group_id, node->node_id)) continue;
         struct TOPO_NODE *identity = topo_node_registry_find(inst->topo_groups, node->node_id);
         int source = node->node_id == s->descriptor.source_node_id;
-        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidate peer=%llx type=%u source=%d online=1",
+        DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate peer=%llx type=%u source=%d online=1",
                     (unsigned long long)node->node_id, identity ? identity->client_type : UINT8_MAX, source);
         if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source) {
-            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidate excluded peer=%llx reason=mobile_relay",
+            DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate excluded peer=%llx reason=mobile_relay",
                         (unsigned long long)node->node_id); continue;
         }
         struct stream_member_state state;
@@ -716,7 +717,7 @@ int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate
     }
     if (!found) { u_free(candidates); candidates = NULL; }
     *out = candidates; *count = found;
-    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidates source=%llx eligible=%zu BGP=%zu",
+    DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidates source=%llx eligible=%zu BGP=%zu",
                 (unsigned long long)s->descriptor.source_node_id, found, capacity);
     return STREAM_OK;
 }

src/stream/stream_network.c

@@ -872,9 +872,11 @@ static void network_tick(void *arg) {
         for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
             if (f->info.state == STREAM_RECEIVE_CLOSED) continue;
             f->info.metrics_valid = stream_metrics_poll(&f->metrics, now_ms, &f->info.metrics) == STREAM_OK;
-            if ((!f->outgoing || (f->info.state < STREAM_RECEIVE_ACTIVE && !(f->info.probe_only && f->info.info_ready)))
+            if ((!f->outgoing || (f->info.state < STREAM_RECEIVE_ACTIVE && !(f->info.probe_only && f->info.info_ready))
+                || (f->outgoing && !f->info.probe_only && f->info.state == STREAM_RECEIVE_ACTIVE && !f->info.blocks
+                    && f->info.last_available != UINT64_MAX))
                 && now >= f->deadline)
-                flow_close(f, f->info.state < STREAM_RECEIVE_ACTIVE ? STREAM_REASON_TRANSPORT : STREAM_REASON_LEASE, 1);
+                flow_close(f, f->outgoing ? STREAM_REASON_TRANSPORT : STREAM_REASON_LEASE, 1);
             else if (f->outgoing && f->info.state >= STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) {
                 struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle,
                     .command = STREAM_CMD_RENEW, .subblock = f->subblock};

src/stream/stream_optimizer.c

@@ -284,7 +284,8 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
             stream_receive_get_info(pick->current, &info);
             if (info.state == STREAM_RECEIVE_CLOSED || !supplier_find(o, info.peer)) receive_failed(o, &pick->current, sub, now);
             else {
-                if (info.blocks) received++; busy++;
+                if (info.blocks && info.metrics_valid && !info.metrics.stopped) received++;
+                busy++;
                 struct stream_supplier *supplier = supplier_find(o, info.peer);
                 if (supplier->offers && info.metrics_valid) {
                     supplier->offers[sub].info.supplier_metrics = info.metrics;
@@ -364,9 +365,23 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
         unsigned used = usage(o, s->node, o->descriptor.n);
         if (used) o->info.providers++;
         if (used > o->descriptor.r) o->info.protected_by_fec = 0;
+        if (used || s->probe || !s->offers) continue;
+        uint64_t keep_until = 0;
+        for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
+            struct stream_offer *offer = &s->offers[sub];
+            uint64_t until = offer->info.info_ms + STREAM_OFFER_TTL_MS;
+            if (until < offer->retry_ms) until = offer->retry_ms;
+            if (until > keep_until) keep_until = until;
+        }
+        if (now >= keep_until) { u_free(s->offers); s->offers = NULL; }
     }
     for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
-        if (o->selected[sub].current) o->info.subflows++;
+        if (o->selected[sub].current) {
+            struct stream_receive_info info; stream_receive_get_info(o->selected[sub].current, &info);
+            o->info.subflows++;
+            if (info.state != STREAM_RECEIVE_ACTIVE || !info.blocks || !info.metrics_valid || info.metrics.stopped || info.close_requested)
+                o->info.protected_by_fec = 0;
+        }
         if (o->selected[sub].replacement) o->info.overlaps++;
     }
     if (o->info.subflows != o->descriptor.n) o->info.protected_by_fec = 0;