diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index b673995a..4e705ac3 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -32,9 +32,8 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry); static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg); static void md_on_props_changed(uint64_t node_id, const char* adm_tags, const char* channel_id, void* arg); static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id, uint64_t group_id); -static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, const uint8_t* data, size_t len); -static int md_send_packet(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, - const uint8_t* data, size_t len, int force); +static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, + const uint8_t* data, size_t len, int force); static void md_repl_pump(void* arg); static void md_repl_watchdog(void* arg); static void md_repl_notify(struct media_delivery_ctx* md, uint64_t group_id); @@ -506,7 +505,7 @@ static void md_repl_pump(void* arg) { if (kind == 0) len = md_repl_page(peer, packet); else if (kind == 1) { data = (const uint8_t*)&peer->request; len = sizeof(peer->request); } else packet[0] = MEDIA_SUBCMD_CHANGED; - if (md_send_packet(md->inst, peer->group_id, peer->peer_node_id, data, len, 0)) return; + if (md_send(md->inst, peer->group_id, peer->peer_node_id, data, len, 0)) return; if (kind == 0) peer->response_pending = 0; else if (kind == 1) peer->request_send = 0; else peer->changed = 0; @@ -550,14 +549,9 @@ static void md_repl_notify(struct media_delivery_ctx* md, uint64_t group_id) { /* ── send via etcp_router helper ── */ -static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, - const uint8_t* data, size_t len) { - return md_send_packet(inst, group_id, dst_node_id, data, len, 1); -} - /* Репликация использует force=0; остальные команды сохраняют существующую политику очереди. */ -static int md_send_packet(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, - const uint8_t* data, size_t len, int force) { +static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, + const uint8_t* data, size_t len, int force) { if (!inst || !data || len == 0) return -1; struct TOPO_GROUP* grp = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; struct ETCP_CONN* conn = grp ? topo_group_find_conn_for_node(grp, dst_node_id) : NULL; @@ -633,7 +627,7 @@ static void md_handle_query(struct media_delivery_ctx* md, uint64_t group_id, ui } } int rc = md_send(md->inst, group_id, from_node, response, - MEDIA_QUERY_RESP_HDR_SIZE + (size_t)r->num_blocks * sizeof(*results)); + MEDIA_QUERY_RESP_HDR_SIZE + (size_t)r->num_blocks * sizeof(*results), 1); DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: QUERY peer=%016llx group=%016llx blocks=%u status=%u send=%d", MD_ID, (unsigned long long)from_node, (unsigned long long)group_id, q->num_blocks, r->status, rc); } @@ -660,7 +654,7 @@ static void md_handle_announcement(struct media_delivery_ctx* md, uint64_t group struct media_pkt_have_block_ack ack = {.subcmd = MEDIA_SUBCMD_HAVE_BLOCK_ACK, .status = changed < 0 ? 2 : changed ? 0 : 1}; memcpy(ack.media_id, hb->media_id, 16); memcpy(ack.block_id, hb->block_id, 16); - md_send(md->inst, group_id, from_node, (const uint8_t*)&ack, sizeof(ack)); + md_send(md->inst, group_id, from_node, (const uint8_t*)&ack, sizeof(ack), 1); } DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: announcement group=%llx holder=%llx status=%d changed=%d", MD_ID, (unsigned long long)group_id, (unsigned long long)from_node, status, changed); @@ -797,6 +791,8 @@ static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64 return; } peer->response_pending = 0; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: replication path unavailable group=%llx peer=%llx event=%d", MD_ID, + (unsigned long long)group_id, (unsigned long long)node_id, event); etcp_router_cancel_send_ready(md->inst, group_id, node_id, ETCP_RT_ID_MEDIA_DELIVERY, &peer->waiter); if (peer->pump) { uasync_call_soon_cancel(md->inst->ua, peer->pump); peer->pump = NULL; } if (event == CONN_EVENT_TIMEOUT) { @@ -903,7 +899,7 @@ static void stream_pump(void* arg) { .total_size = source->size, .status = source->failed ? MEDIA_BLOCK_UNAVAILABLE : MEDIA_BLOCK_OK}; memcpy(done.media_id, source->media_id, 16); memcpy(done.block_id, source->block_id, 16); memcpy(done.transfer_id, sc->transfer_id, 16); - if (!md_send(md->inst, sc->group_id, sc->dst_node_id, (const uint8_t*)&done, sizeof(done))) + if (!md_send(md->inst, sc->group_id, sc->dst_node_id, (const uint8_t*)&done, sizeof(done), 1)) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: TX DONE peer=%016llx chunk=%u bytes=%u status=%u", MD_ID, (unsigned long long)sc->dst_node_id, source->chunk, sc->offset, done.status); stream_free(sc); return; @@ -929,7 +925,7 @@ static void stream_pump(void* arg) { source->failed = 1; md_block_detach(source); md_block_wake(source, sc); continue; } - if (md_send(md->inst, sc->group_id, sc->dst_node_id, packet, sizeof(*ch) + count)) { + if (md_send(md->inst, sc->group_id, sc->dst_node_id, packet, sizeof(*ch) + count, 1)) { stream_free(sc); return; } sc->offset += count; sc->last_progress_tb = get_time_tb(); @@ -947,13 +943,13 @@ static void md_block_reply(struct media_delivery_ctx* md, uint64_t group, uint64 .chunk = request->chunk, .retry_after_ms = 2000}; memcpy(reply.media_id, request->media_id, 16); memcpy(reply.block_id, request->block_id, 16); memcpy(reply.transfer_id, request->transfer_id, 16); - md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply)); + md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply), 1); } else { struct media_pkt_block_done reply = {.subcmd = MEDIA_SUBCMD_BLOCK_DONE, .chunk = request->chunk, .status = MEDIA_BLOCK_UNAVAILABLE}; memcpy(reply.media_id, request->media_id, 16); memcpy(reply.block_id, request->block_id, 16); memcpy(reply.transfer_id, request->transfer_id, 16); - md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply)); + md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply), 1); } } @@ -996,7 +992,7 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t group_id full->subcmd = MEDIA_SUBCMD_BLOCK_RELAY_FULL; full->chunk = req->chunk; full->num_relay_nodes = (uint16_t)count; memcpy(full->media_id, req->media_id, 16); memcpy(full->block_id, req->block_id, 16); memcpy(full->transfer_id, req->transfer_id, 16); memcpy(packet + sizeof(*full), nodes, count * sizeof(uint64_t)); - md_send(md->inst, group_id, from_node, packet, sizeof(*full) + count * sizeof(uint64_t)); return; + md_send(md->inst, group_id, from_node, packet, sizeof(*full) + count * sizeof(uint64_t), 1); return; } md_block_reply(md, group_id, from_node, req, 1); return; } @@ -1581,7 +1577,7 @@ void media_delivery_send_have_block(struct UTUN_INSTANCE* inst, uint64_t group_i DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: Ed25519 sign failed for HAVE_BLOCK", MD_ID); return; } - md_send(inst, group_id, dst_node_id, (const uint8_t*)&hb, sizeof(hb)); + md_send(inst, group_id, dst_node_id, (const uint8_t*)&hb, sizeof(hb), 1); } void media_delivery_announce_media(struct UTUN_INSTANCE* inst, uint64_t group_id, diff --git a/src/media_delivery/media_delivery_proto.h b/src/media_delivery/media_delivery_proto.h index 53fce8d6..cf1cd1f4 100644 --- a/src/media_delivery/media_delivery_proto.h +++ b/src/media_delivery/media_delivery_proto.h @@ -126,7 +126,7 @@ struct media_pkt_have_block_ack { uint8_t subcmd; // MEDIA_SUBCMD_HAVE_BLOCK_ACK uint8_t media_id[16]; uint8_t block_id[16]; - uint8_t status; // 0=OK, 1=dup/already + uint8_t status; // 0=сохранено изменение, 1=повтор/старое объявление, 2=ошибка БД. }; struct media_pkt_sync_req {