Use observed provider quality and keep empty-source queue accounting bounded

src/stream/stream_network.c

@@ -628,6 +628,7 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
                 stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL);
                 load_account(f, count, count * stream_block_size(&f->descriptor), 1, 0);
                 f->pending_highest = stream_read_be(payload + 24, 8);
+                if (f->pending_highest == UINT64_MAX) f->pending_highest = 0;
                 const uint8_t *wire; size_t length; uint64_t seq;
                 if (stream_buffer_read(local->buffer, UINT64_MAX, f->subblock, &wire, &length, &seq) == STREAM_OK)
                     stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),

src/stream/stream_optimizer.c

@@ -192,11 +192,17 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig
             double score = problems(&info->supplier_metrics, info->supplier_metrics_valid);
             double old_score = problems(&previous->supplier_metrics, previous->supplier_metrics_valid);
             uint64_t rtt = total_rtt(info), old_rtt = total_rtt(previous);
-            if (score < old_score || (score == old_score && (rtt < old_rtt
-                || (rtt == old_rtt && (shared_path(o, info, sub) < shared_path(o, previous, sub)
-                    || (shared_path(o, info, sub) == shared_path(o, previous, sub)
-                        && (info->path.count < previous->path.count || (info->path.count == previous->path.count && s->node < best->node))))))))
-                best = s;
+            int better = score < old_score;
+            if (score == old_score) {
+                better = rtt < old_rtt;
+                if (rtt == old_rtt) {
+                    unsigned shared = shared_path(o, info, sub), old_shared = shared_path(o, previous, sub);
+                    better = shared < old_shared;
+                    if (shared == old_shared) better = info->path.count < previous->path.count
+                        || (info->path.count == previous->path.count && s->node < best->node);
+                }
+            }
+            if (better) best = s;
         }
     }
     return best;
@@ -277,7 +283,15 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
         if (pick->current) {
             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++; }
+            else {
+                if (info.blocks) 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;
+                    supplier->offers[sub].info.supplier_metrics_valid = 1;
+                    supplier->offers[sub].info.last_available = info.next - 1;
+                }
+            }
         }
         if (!pick->replacement) continue;
         struct stream_receive_info replacement; stream_receive_get_info(pick->replacement, &replacement);