Define stream pressure negotiation and timestamp metric snapshots

src/stream/stream_network.h

@@ -7,6 +7,7 @@
 
 #include "stream.h"
 #include "stream_wire.h"
+#include "stream_load.h"
 
 struct stream_network;
 struct STREAM_RECEIVE;
@@ -16,6 +17,9 @@ struct stream_receive_info {
     uint64_t handle, peer, start, next, blocks, gaps;
     enum stream_receive_state state;
     int reason;
+    struct stream_metric_snapshot metrics, supplier_metrics;
+    uint32_t rtt_ms;
+    int metrics_valid, supplier_metrics_valid, rtt_valid, close_requested;
 };
 /* out заранее NULL; seq=UINT64_MAX: первая выдача с конца, остальные ждут её проверенный общий старт.
  * peer должен быть онлайн-кандидатом группы. Повтор peer+stream+sub не создаёт вторую выдачу. */
@@ -25,10 +29,22 @@ void stream_receive_close(struct STREAM_RECEIVE *receive);
  * Буфер/подписка/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);
+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;
+    uint32_t transport_queued_packets;
+    double cpu_percent;
+    uint64_t callback_late_ms;
+    unsigned required_count, released_count;
+    int global_pressure, request_active;
+};
+int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t peer, struct stream_peer_load_info *info);
 
 int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out);
 void stream_network_destroy(struct stream_network *network);
 void stream_network_changed(struct stream_network *network);
 void stream_network_cancel(struct stream_network *network, const struct stream_descriptor *descriptor);
 void stream_network_membership_changed(struct stream_network *network, uint64_t node_id, const char *channel);
+void stream_network_cache_event(struct stream_network *network, const struct stream_descriptor *descriptor,
+                                const uint8_t *wire, size_t length, int added);
 #endif

src/stream/stream_wire.c

@@ -7,7 +7,7 @@
 /* Заголовок одинаков у raw и router payload; cmd raw сохраняется внутри routed-кодограммы. */
 int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *m) {
     if (!wire || !m || length < STREAM_WIRE_HEADER || wire[0] != ETCP_ID_STREAM || wire[1] != STREAM_WIRE_VERSION
-        || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_REVERSE_REQ) {
+        || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_OVERLOAD_SELECT) {
         DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire header invalid bytes=%zu", length); return STREAM_ERR_FORMAT;
     }
     struct stream_message result = {.command = wire[2], .subblock = wire[3], .payload = wire + STREAM_WIRE_HEADER,
@@ -60,6 +60,25 @@ size_t stream_path_write(uint8_t *wire, const struct stream_path *path) {
     return 1 + path->count * 8;
 }
 
+/* Неизвестное измерение явно отличается от нулевого lag/rate; числа не сужаются до старых uint16 RTT. */
+void stream_metric_write(uint8_t *wire, const struct stream_metric_snapshot *s, int valid) {
+    wire[0] = valid ? 1 | (s->source_rate_valid ? 2 : 0) | (s->receive_rate_valid ? 4 : 0) | (s->stopped ? 8 : 0) : 0;
+    stream_write_be(wire + 1, s->lag_ms, 8); stream_write_be(wire + 9, s->source_rate_bytes_sec, 8);
+    stream_write_be(wire + 17, s->receive_rate_bytes_sec, 8); stream_write_be(wire + 25, s->blocks, 8);
+    stream_write_be(wire + 33, s->late_blocks, 8); stream_write_be(wire + 41, s->stops, 8);
+}
+
+int stream_metric_read(const uint8_t *wire, struct stream_metric_snapshot *s, int *valid) {
+    if (wire[0] & ~15u || (!(wire[0] & 1) && wire[0])) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream metric flags invalid=%u", wire[0]); return STREAM_ERR_FORMAT;
+    }
+    *s = (struct stream_metric_snapshot){.lag_ms = stream_read_be(wire + 1, 8), .source_rate_bytes_sec = stream_read_be(wire + 9, 8),
+        .receive_rate_bytes_sec = stream_read_be(wire + 17, 8), .blocks = stream_read_be(wire + 25, 8),
+        .late_blocks = stream_read_be(wire + 33, 8), .stops = stream_read_be(wire + 41, 8),
+        .source_rate_valid = !!(wire[0] & 2), .receive_rate_valid = !!(wire[0] & 4), .stopped = !!(wire[0] & 8)};
+    *valid = !!(wire[0] & 1); return STREAM_OK;
+}
+
 /* END имеет собственный domain и подписывает ключ запуска + последний индекс; UINT64_MAX означает пустой поток. */
 int stream_end_sign(const struct stream_descriptor *d, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]) {
     if (!d || !key || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream END sign invalid arguments"); return STREAM_ERR_ARGUMENT; }

src/stream/stream_wire.h

@@ -4,22 +4,28 @@
 #define STREAM_WIRE_H
 
 #include "stream_proto.h"
+#include "stream_metrics.h"
 
-#define STREAM_WIRE_VERSION 1
+#define STREAM_WIRE_VERSION 2
 #define STREAM_WIRE_HEADER 52
 #define STREAM_PATH_MAX 32
 #define STREAM_END_SIZE 106
-#define STREAM_INFO_SIZE 29 /* actual:8, first:8, last:8, n:1, r:1, block_size:2, path_count:1 */
+#define STREAM_METRIC_WIRE_SIZE 49
+#define STREAM_INFO_METRIC_OFF 36
+#define STREAM_INFO_PATH_OFF (STREAM_INFO_METRIC_OFF + STREAM_METRIC_WIRE_SIZE)
+#define STREAM_INFO_SIZE (STREAM_INFO_PATH_OFF + 1) /* probe:8, границы/FEC:28, метрики:49, count:1 */
+#define STREAM_OVERLOAD_MAX 64
+#define STREAM_OVERLOAD_HEADER 18
 
 enum stream_command {
     STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT,
     STREAM_CMD_REJECT, STREAM_CMD_RENEW, STREAM_CMD_UNSUBSCRIBE, STREAM_CMD_CLOSED,
-    STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ
+    STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ, STREAM_CMD_OVERLOAD, STREAM_CMD_OVERLOAD_SELECT
 };
 enum stream_reject {
     STREAM_REASON_STOPPED = 1, STREAM_REASON_POSITION, STREAM_REASON_EXPIRED, STREAM_REASON_STALE,
     STREAM_REASON_CONFLICT, STREAM_REASON_LIMIT, STREAM_REASON_LOOP, STREAM_REASON_TRANSPORT,
-    STREAM_REASON_LEASE, STREAM_REASON_MEMBER, STREAM_REASON_PROTOCOL
+    STREAM_REASON_LEASE, STREAM_REASON_MEMBER, STREAM_REASON_PROTOCOL, STREAM_REASON_OVERLOAD
 };
 struct stream_path { uint64_t nodes[STREAM_PATH_MAX]; uint8_t count; };
 struct stream_message {
@@ -37,6 +43,8 @@ void stream_wire_header(uint8_t *wire, const struct stream_message *message);
 int stream_path_validate(const struct stream_path *path, uint64_t source, uint64_t sender, uint64_t forbidden);
 int stream_path_parse(const uint8_t *wire, size_t length, struct stream_path *path, size_t *used);
 size_t stream_path_write(uint8_t *wire, const struct stream_path *path);
+void stream_metric_write(uint8_t *wire, const struct stream_metric_snapshot *snapshot, int valid);
+int stream_metric_read(const uint8_t *wire, struct stream_metric_snapshot *snapshot, int *valid);
 int stream_end_sign(const struct stream_descriptor *descriptor, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]);
 int stream_end_verify(const struct stream_descriptor *descriptor, const uint8_t key[32], const uint8_t *wire, size_t length,
                       uint64_t *last);