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

Make media send queue policy explicit and document announcement errors

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 21:14:42 +02:00
1 parent f2283eff6b
commit 0bcdb9930a
2 files changed
+16 -20
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.c
+15 -19
View File
@@ -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,
src/media_delivery/media_delivery_proto.h
+1 -1
View File
@@ -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 {