jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Define media block attempts and shared byte-source ownership

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 19:08:14 +02:00
1 parent 68129c0151
commit f15d53947e
3 files changed
+57 -34
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.h
+29 -23
View File
@@ -13,6 +13,7 @@ extern "C" {
#include <stdint.h>
#include "../lib/ll_queue.h"
#include "../lib/sqlite3.h"
#include "../lib/file_io.h"
struct UTUN_INSTANCE;
@@ -90,30 +91,29 @@ struct media_served_node {
#define MD_MAX_RELAY_DOWNSTREAM 5
struct relay_downstream {
uint64_t node_id;
uint64_t group_id;
uint64_t sent_offset; // сколько байт уже отправлено этому пиру
struct queue_waiter_handle waiter;
};
struct relay_block_ctx {
struct ll_entry ll; // индекс по block_id (16 байт)
/* Источник байтов блока: producer загрузки и каждая выдача удерживают отдельную ссылку.
* Индекс заимствует источник; последний владелец удаляет собственный temp.
* available — только непрерывный, успешно закрытый префикс; verified разрешает отправить DONE. */
struct media_block_source {
uint8_t block_id[16];
uint8_t media_id[16];
uint64_t group_id;
uint32_t chunk;
uint64_t file_offset; // всего байт получено от апстрима (размер .chunk_N)
char chunk_file[2048]; // путь к .chunk_N файлу для relay-чтения
struct relay_downstream downstream[MD_MAX_RELAY_DOWNSTREAM];
int downstream_count;
struct UTUN_INSTANCE* inst;
uint32_t size;
uint32_t available;
uint64_t base; // Начало блока в индексированном файле; для temp — 0.
unsigned refs;
uint8_t indexed, owned, verified, failed;
struct ll_entry* entry;
struct media_delivery_ctx* md;
struct file_io_temp storage; // Путь temp либо индексированного файла; открытых FILE не хранит.
};
struct stream_ctx;
struct relay_fwd_ctx;
struct media_destination;
struct media_delivery_ctx {
struct stream_ctx* streams;
struct relay_fwd_ctx* forwards;
struct media_destination* destinations; // Активные писатели нормализованных путей, входящие и исходящие.
uint64_t self_node_id;
int initialized;
uint8_t is_supernode; // из adm_tags в peers_*
@@ -128,7 +128,7 @@ struct media_delivery_ctx {
struct ll_queue* served_nodes; // media_served_node — только суперузел
struct ll_queue* super_peers; // media_super_peer — только суперузел
struct ll_queue* downloads; // media_download — активные загрузки
struct ll_queue* relay_blocks; // relay_block_ctx — relay в процессе загрузки
struct ll_queue* block_sources; // media_block_source, заимствованный индекс по block_id.
struct ll_queue* file_loads; // md_file_load — per-file download count (source node)
struct UTUN_INSTANCE* inst;
};
@@ -176,13 +176,19 @@ void media_delivery_announce_local_blocks(struct UTUN_INSTANCE* inst);
* загрузки с group_id, их relay-контексты и super_peers этой группы. */
void media_delivery_cancel_channel(struct UTUN_INSTANCE* inst, uint64_t group_id);
// ── Внутренние операции загрузчика: relay ──
struct relay_block_ctx* md_relay_find(struct media_delivery_ctx* md, const uint8_t* block_id);
/* Зарезервировать путь в общей базе перед любой подготовкой файла; NULL — занят/ошибка.
* Владелец освобождает handle после завершения всех записей, включая ошибку/отмену. Только uasync. */
struct media_destination* media_delivery_reserve(struct UTUN_INSTANCE* inst, const char* path);
void media_delivery_release(struct UTUN_INSTANCE* inst, struct media_destination* destination);
struct relay_block_ctx* md_relay_add(struct media_delivery_ctx* md,
const uint8_t* block_id, const uint8_t* media_id, uint32_t chunk);
void md_relay_remove(struct media_delivery_ctx* md, const uint8_t* block_id);
/* Создать уникальный temp блока с одной producer-ссылкой; NULL — ошибка.
* publish будит выдачи после записи/проверки. abort отсоединяет индекс и завершает выдачи с ошибкой.
* release снимает ссылку; проверенный temp живёт до завершения последней выдачи. Только uasync. */
struct media_block_source* md_block_create(struct media_delivery_ctx* md, uint64_t group_id,
const uint8_t media_id[16], const uint8_t block_id[16], uint32_t chunk, uint32_t size, const char* destination);
void md_block_publish(struct media_block_source* source, uint32_t available, int verified);
void md_block_abort(struct media_block_source* source);
void md_block_release(struct media_block_source* source);
// ── Внутренний учёт нагрузки на файл ──
struct md_file_load* md_file_load_find(struct media_delivery_ctx* md, const uint8_t* media_id);
src/media_delivery/media_delivery_proto.h
+17 -5
View File
@@ -43,6 +43,14 @@ enum {
#define MEDIA_QUERY_OK 0
#define MEDIA_QUERY_DB_ERROR 1
/* Каждая попытка BLOCK_REQ получает новый UUID; все ответы/CANCEL повторяют его.
* Группа и отправитель берутся только из router. Байты идут непрерывно от offset=0.
* DONE подтверждает конец передачи; подпись берётся из метаданных сообщения, а не из ответа держателя. */
#define MEDIA_BLOCK_DATA_MAX 1024
#define MEDIA_BLOCK_SIZE_MAX (25U * 1024U * 1024U)
#define MEDIA_BLOCK_OK 0
#define MEDIA_BLOCK_UNAVAILABLE 1
/* ── packed-структуры пакетов ── */
#pragma pack(push, 1)
@@ -125,11 +133,10 @@ struct media_pkt_super_ack {
struct media_pkt_block_req {
uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_REQ
uint64_t group_id; // CHAT-группа для маршрутизации и прямого коннекта
uint8_t media_id[16];
uint8_t block_id[16];
uint8_t transfer_id[16]; // UUID попытки; одинаков только при дублировании запроса.
uint32_t chunk;
uint64_t offset; // смещение в блоке (обычно 0)
};
#define MEDIA_BLOCK_REQ_SIZE (sizeof(struct media_pkt_block_req))
@@ -138,21 +145,23 @@ struct media_pkt_block_chunk {
uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_CHUNK
uint8_t media_id[16];
uint8_t block_id[16];
uint8_t transfer_id[16];
uint32_t chunk;
uint32_t offset; // смещение в блоке
uint16_t data_len; // 1..1024
// далее: data[data_len]
};
#define MEDIA_BLOCK_CHUNK_HDR_SIZE (sizeof(struct media_pkt_block_chunk)) // 1+16+16+4+4+2=43
#define MEDIA_BLOCK_CHUNK_HDR_SIZE (sizeof(struct media_pkt_block_chunk)) // 59
struct media_pkt_block_done {
uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_DONE
uint8_t media_id[16];
uint8_t block_id[16];
uint8_t transfer_id[16];
uint32_t chunk;
uint32_t total_size; // полный размер блока
uint8_t block_sig[64]; // Ed25519 подпись блока
uint8_t status; // MEDIA_BLOCK_OK / MEDIA_BLOCK_UNAVAILABLE; ошибка не подтверждает байты.
};
#define MEDIA_BLOCK_DONE_SIZE (sizeof(struct media_pkt_block_done))
@@ -161,6 +170,7 @@ struct media_pkt_cancel {
uint8_t subcmd; // MEDIA_SUBCMD_CANCEL
uint8_t media_id[16];
uint8_t block_id[16];
uint8_t transfer_id[16];
uint32_t chunk;
};
@@ -174,6 +184,7 @@ struct media_pkt_block_overloaded {
uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_OVERLOADED
uint8_t media_id[16];
uint8_t block_id[16];
uint8_t transfer_id[16];
uint32_t chunk;
uint32_t retry_after_ms; // рекомендованная задержка перед повтором (ms), 0 = default 2000
};
@@ -182,12 +193,13 @@ struct media_pkt_block_relay_full {
uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_RELAY_FULL
uint8_t media_id[16];
uint8_t block_id[16];
uint8_t transfer_id[16];
uint32_t chunk;
uint16_t num_relay_nodes; // 1..5
// далее: relay_node_ids[num_relay_nodes * 8]
};
#define MEDIA_RELAY_FULL_HDR_SIZE (sizeof(struct media_pkt_block_relay_full)) // 1+16+16+4+2=39
#define MEDIA_RELAY_FULL_HDR_SIZE (sizeof(struct media_pkt_block_relay_full)) // 55
#pragma pack(pop)
src/media_delivery/media_download.h
+11 -6
View File
@@ -22,6 +22,8 @@ extern "C" {
struct UTUN_INSTANCE;
struct media_index_result;
struct CONN_MGR_HANDLE;
struct media_block_source;
struct media_destination;
/* максимальное число узлов-держателей одного блока (кандидатов на фейловер) */
#define MD_MAX_HOLDERS_PER_BLOCK MEDIA_QUERY_MAX_HOLDERS
@@ -44,11 +46,13 @@ enum md_block_state {
/* ── состояние одного блока ── */
struct media_download_block {
uint8_t block_id[16];
uint8_t transfer_id[16]; /* Текущая попытка, меняется до отправки BLOCK_REQ. */
struct media_block_source* source; // Собственная producer-ссылка на файл этой попытки.
uint64_t holders[MD_MAX_HOLDERS_PER_BLOCK]; /* узлы, у кого есть блок */
int num_holders;
int holder_idx; /* текущий держатель (-1 = нет) */
int attempts; /* 0..max (фейловеры/провалы блока) */
uint32_t bytes_received; /* max(offset+len) полученных байт */
uint32_t bytes_received; /* Непрерывный успешно записанный и закрытый префикс. */
uint32_t expected_size; /* реальный размер блока (последний короче) */
uint8_t state; /* enum md_block_state */
uint64_t last_progress_tb; /* get_time_tb() последнего чанка/прогресса */
@@ -71,6 +75,8 @@ struct media_download {
struct ll_entry ll;
struct ll_entry* entry; /* Владеющая запись; после удаления из индекса живёт до последнего refs. */
unsigned refs; /* Активная загрузка и обработчики, вызывающие пользовательские callbacks. */
struct media_destination* destination; // Исключительное право записи dest_path до terminal.
uint8_t author_pubkey[32]; /* Точный ключ автора из БД, скопирован до начала загрузки. */
uint8_t media_id[16];
uint64_t group_id;
char dest_path[1024];
@@ -131,14 +137,13 @@ int media_download_cancel(struct UTUN_INSTANCE* inst,
void media_download_handle_query_resp(struct UTUN_INSTANCE* inst,
uint64_t group_id, uint64_t from_node, const uint8_t* data, size_t len);
void media_download_handle_chunk(struct UTUN_INSTANCE* inst,
const uint8_t* data, size_t len);
uint64_t group_id, uint64_t from_node, const uint8_t* data, size_t len);
void media_download_handle_done(struct UTUN_INSTANCE* inst,
const uint8_t* data, size_t len);
uint64_t group_id, uint64_t from_node, const uint8_t* data, size_t len);
void media_download_handle_overloaded(struct UTUN_INSTANCE* inst,
const uint8_t* data, size_t len);
uint64_t group_id, uint64_t from_node, const uint8_t* data, size_t len);
void media_download_handle_relay_full(struct UTUN_INSTANCE* inst,
const uint8_t* media_id, const uint8_t* block_id,
uint32_t chunk, const uint64_t* node_ids, int num_nodes);
uint64_t group_id, uint64_t from_node, const uint8_t* data, size_t len);
/* ── conn_mgr callback (non-static for test visibility) ── */
struct CONN_MGR_HANDLE;