diff --git a/src/media_delivery/media_delivery.h b/src/media_delivery/media_delivery.h index 232e1832..0af1edcb 100644 --- a/src/media_delivery/media_delivery.h +++ b/src/media_delivery/media_delivery.h @@ -13,6 +13,7 @@ extern "C" { #include #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); diff --git a/src/media_delivery/media_delivery_proto.h b/src/media_delivery/media_delivery_proto.h index b930a4b1..03d95cef 100644 --- a/src/media_delivery/media_delivery_proto.h +++ b/src/media_delivery/media_delivery_proto.h @@ -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) diff --git a/src/media_delivery/media_download.h b/src/media_delivery/media_download.h index 42791a94..c313658b 100644 --- a/src/media_delivery/media_download.h +++ b/src/media_delivery/media_download.h @@ -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;