diff --git a/src/member_files/member_files.c b/src/member_files/member_files.c index 116b058d..8d8e211c 100644 --- a/src/member_files/member_files.c +++ b/src/member_files/member_files.c @@ -73,7 +73,7 @@ static void mf_local_deliver(void* arg) { int member_files_member(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t node, int storage) { struct TOPO_GROUP* g = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, group) : NULL; if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT || !node || !inst->topo_sqlite_db) { - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: ACL missing CHAT group=%llu node=%llu role=%d", + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: ACL missing CHAT group=%llu node=%llu role=%d", (unsigned long long)group, (unsigned long long)node, storage); return 0; } char sql[256]; @@ -90,19 +90,22 @@ int member_files_member(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t nod int storage_role = tags && json_flat_get(tags, "storage", role, sizeof(role)) == 0 && !strcmp(role, "yes"); int allowed = rc == SQLITE_ROW && !sqlite3_column_int(st, 1) && !sqlite3_column_int(st, 3) && (!storage || (storage_role && sqlite3_column_bytes(st, 2) == 64)); - if (!allowed) DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, - "member_files: ACL denied group=%llu node=%llu require_storage=%d row=%d storage=%d source=%d signature=%d deleted=%d", - (unsigned long long)group, (unsigned long long)node, storage, rc == SQLITE_ROW, + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, + "member_files: ACL allowed=%d group=%llu node=%llu require_storage=%d row=%d storage=%d source=%d signature=%d deleted=%d", + allowed, (unsigned long long)group, (unsigned long long)node, storage, rc == SQLITE_ROW, storage_role, sqlite3_column_int(st, 1), sqlite3_column_bytes(st, 2), sqlite3_column_int(st, 3)); if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: membership step failed rc=%d", rc); sqlite3_finalize(st); return allowed; } static int mf_server_allowed(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* q) { - return chat_setting_get_int(f->inst, "member_files_enabled", 1) && - member_files_member(f->inst, group, f->inst->node_id, 1) && member_files_member(f->inst, group, peer, 0) && - (!q->owner || member_files_member(f->inst, group, q->owner, 0)) && - (q->command < MF_BEGIN_WRITE || (q->owner && q->owner == peer)); + int enabled = chat_setting_get_int(f->inst, "member_files_enabled", 1); + int allowed = enabled && member_files_member(f->inst, group, f->inst->node_id, 1) && + member_files_member(f->inst, group, peer, 0) && (!q->owner || member_files_member(f->inst, group, q->owner, 0)) && + (q->command < MF_BEGIN_WRITE || (q->owner && q->owner == peer)); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: server access id=%llu cmd=%s enabled=%d allowed=%d owner_is_peer=%d", + (unsigned long long)q->request_id, mf_command_text(q->command), enabled, allowed, q->owner == peer); + return allowed; } /* Router подтверждает доставку отдельно; здесь отправка всегда соблюдает порог очереди. */ @@ -114,20 +117,29 @@ static int mf_send(struct member_files* f, uint64_t group, uint64_t peer, const local->files = f; local->group = group; local->message = *m; local->task = uasync_call_soon(f->inst->ua, local, mf_local_deliver); if (!local->task) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: local delivery scheduling failed"); u_free(local); return -1; } - local->next = f->locals; f->locals = local; f->local_count++; return 0; + local->next = f->locals; f->locals = local; f->local_count++; + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: local queued id=%llu cmd=%s kind=%u count=%u", + (unsigned long long)m->request_id, mf_command_text(m->command), m->kind, f->local_count); + return 0; } struct ETCP_ROUTER_CONN* rc = etcp_router_conn_get(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES); if (!rc) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: router allocation failed"); return -1; } queue_set_threshold(rc->send_q, 0, 0); etcp_router_set_max_inflight(rc, 16); - if (!etcp_router_send_q_has_room(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES)) return 1; + if (!etcp_router_send_q_has_room(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES)) { + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: tx waiting id=%llu cmd=%s send_q=%u inflight=%u sync=%u no_route=%u", + (unsigned long long)m->request_id, mf_command_text(m->command), rc->send_q->count, rc->inflight_q->count, + rc->peer_sync_done, rc->no_route); + return 1; + } uint8_t wire[MF_WIRE_MAX]; int n = mf_encode(m, wire, sizeof(wire)); if (n < 0) return -1; + uint32_t seq = rc->tx_seq; int result = etcp_router_conn_send(rc, wire, (size_t)n, ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED); - if (!result) DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, - "member_files: tx id=%llu kind=%u cmd=%u group=%llx peer=%llx owner=%llx status=%u bytes=%d", - (unsigned long long)m->request_id, m->kind, m->command, (unsigned long long)group, - (unsigned long long)peer, (unsigned long long)m->owner, m->status, n); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, + "member_files: tx id=%llu kind=%u cmd=%s group=%llx peer=%llx owner=%llx status=%s bytes=%d seq=%u accepted=%d", + (unsigned long long)m->request_id, m->kind, mf_command_text(m->command), (unsigned long long)group, + (unsigned long long)peer, (unsigned long long)m->owner, mf_status_text(m->status), n, seq, result == 0); if (result) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: send failed group=%llu peer=%llu cmd=%u rc=%d", (unsigned long long)group, (unsigned long long)peer, m->command, result); return result ? -1 : 0; @@ -145,8 +157,9 @@ static void mf_pending_finish(struct mf_pending* p, const struct mf_message* res struct mf_message result; if (!response) { result = p->request; result.kind = MF_RESPONSE; result.status = (uint16_t)status; result.data_len = 0; response = &result; } mf_response_fn cb = p->callback; void* arg = p->arg; - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: request done id=%llu cmd=%u status=%s pending=%u", - (unsigned long long)p->request.request_id, p->request.command, mf_status_text(response->status), f->pending_count); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: request done id=%llu cmd=%s status=%s pending=%u elapsed_ms=%llu", + (unsigned long long)p->request.request_id, mf_command_text(p->request.command), mf_status_text(response->status), + f->pending_count, (unsigned long long)((get_time_tb() - p->started) / 10)); u_free(p); cb(arg, response); } @@ -170,7 +183,7 @@ static void mf_client_ready(struct ll_queue* q, void* arg) { (void)q; mf_pending static void mf_connected(struct CONN_MGR_HANDLE* h, uint64_t peer, uint64_t group, enum conn_mgr_event event, void* arg) { (void)h; (void)peer; (void)group; struct mf_pending* p = arg; - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: CM event=%d request=%llu", event, (unsigned long long)p->request.request_id); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: CM event=%d request=%llu", event, (unsigned long long)p->request.request_id); if (event == CONN_EVENT_UP) mf_pending_wake(p); } @@ -199,6 +212,11 @@ struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t gro DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: duplicate local request id"); u_free(p); return NULL; } p->next = f->pending; f->pending = p; f->pending_count++; + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, + "member_files: request id=%llu cmd=%s group=%llx server=%llx owner=%llx path=%s offset=%llu size=%llu pending=%u", + (unsigned long long)p->request.request_id, mf_command_text(p->request.command), (unsigned long long)group, + (unsigned long long)server, (unsigned long long)p->request.owner, p->request.path, + (unsigned long long)p->request.offset, (unsigned long long)p->request.size, f->pending_count); if (server != inst->node_id && conn_mgr_open(inst, group, server, mf_connected, p, &p->cm) != 0) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: CM launch failed server=%llu", (unsigned long long)server); if (p->pump) uasync_call_soon_cancel(inst->ua, p->pump); @@ -240,8 +258,8 @@ static void mf_server_ready(struct ll_queue* q, void* arg) { static void mf_server_work(void* arg) { struct mf_job* j = arg; - if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: disk start id=%llu cmd=%u group=%llx peer=%llx path=%s", - (unsigned long long)j->request.request_id, j->request.command, (unsigned long long)j->group, + if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk start id=%llu cmd=%s group=%llx peer=%llx path=%s", + (unsigned long long)j->request.request_id, mf_command_text(j->request.command), (unsigned long long)j->group, (unsigned long long)j->peer, j->request.path); if (j->request.command) mf_storage_execute(j->files->storage, j->group, j->peer, &j->request, &j->response); else mf_storage_sweep(j->files->storage, get_time_tb()); @@ -252,8 +270,8 @@ static void mf_server_done(void* arg, int error) { struct member_files* f = j->files; f->worker = NULL; if (f->closing || !j->request.command) { mf_job_free(j); if (!f->closing) mf_server_pump(f); return; } if (error) { j->response = j->request; j->response.kind = MF_RESPONSE; j->response.status = MF_BUSY; j->response.data_len = 0; } - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: disk done id=%llu cmd=%u status=%u error=%d jobs=%u", - (unsigned long long)j->request.request_id, j->request.command, j->response.status, error, f->job_count); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk done id=%llu cmd=%s status=%s error=%d jobs=%u", + (unsigned long long)j->request.request_id, mf_command_text(j->request.command), mf_status_text(j->response.status), error, f->job_count); j->state = 2; mf_server_pump(f); } @@ -312,7 +330,11 @@ static void mf_server_pump(struct member_files* f) { } j = next; } - if (f->worker) return; + if (f->worker) { + DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: server waiting for disk id=%llu cmd=%s jobs=%u", + (unsigned long long)f->worker->request.request_id, mf_command_text(f->worker->request.command), f->job_count); + return; + } for (j = f->jobs; j; j = j->next) if (!j->state) break; if (!j && f->sweep_due && f->job_count < MF_MAX_JOBS) { j = u_calloc(1, sizeof(*j)); @@ -325,8 +347,8 @@ static void mf_server_pump(struct member_files* f) { j->response.data_len = 0; j->state = 2; mf_server_pump(f); return; } j->state = 1; f->worker = j; - if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: disk queued id=%llu cmd=%u jobs=%u", - (unsigned long long)j->request.request_id, j->request.command, f->job_count); + if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk queued id=%llu cmd=%s jobs=%u", + (unsigned long long)j->request.request_id, mf_command_text(j->request.command), f->job_count); mf_worker_submit(f->disk, mf_server_work, j, mf_server_done, j); } @@ -359,8 +381,8 @@ static void mf_tick(void* arg) { /* Общая обработка проверенного сетевого сообщения или команды внутри экземпляра. */ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* message) { struct mf_message m = *message; - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: rx id=%llu kind=%u cmd=%u group=%llx peer=%llx owner=%llx jobs=%u worker=%p", - (unsigned long long)m.request_id, m.kind, m.command, (unsigned long long)group, + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: rx id=%llu kind=%u cmd=%s group=%llx peer=%llx owner=%llx jobs=%u worker=%p", + (unsigned long long)m.request_id, m.kind, mf_command_text(m.command), (unsigned long long)group, (unsigned long long)peer, (unsigned long long)m.owner, f->job_count, (void*)f->worker); if (m.kind == MF_RESPONSE) { for (struct mf_pending* p = f->pending; p; p = p->next) { @@ -394,7 +416,11 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c static void mf_recv(struct ETCP_CONN* conn, struct ll_entry* e) { struct member_files* f = conn ? conn->instance->member_files : NULL; - if (!f || f->closing || !e || !e->dgram || e->len <= ROUTER_SVC_PAYLOAD_OFF) goto out; + if (!f || f->closing || !e || !e->dgram || e->len <= ROUTER_SVC_PAYLOAD_OFF) { + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: recv discarded context=%p closing=%d bytes=%u", + (void*)f, f ? f->closing : 0, e ? e->len : 0); + goto out; + } uint64_t group, peer; memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&peer, e->dgram + ROUTER_SVC_SRC_OFF, 8); if ((e->dgram[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) != @@ -482,6 +508,8 @@ static void mf_gui_done(void* arg, const struct mf_message* response) { struct mf_gui_request* q = arg; uint8_t event[16 + MF_WIRE_MAX]; mf_put64(event, q->group); mf_put64(event + 8, q->server); int n = mf_encode(response, event + 16, MF_WIRE_MAX); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: GUI result id=%llu cmd=%s status=%s bytes=%d", + (unsigned long long)response->request_id, mf_command_text(response->command), mf_status_text(response->status), n); if (n > 0) chat_event_post(q->inst, CHAT_EVT_MEMBER_FILES_RESPONSE, event, n + 16); u_free(q); } diff --git a/src/member_files/protocol.c b/src/member_files/protocol.c index 9e3e262a..4725b1ea 100644 --- a/src/member_files/protocol.c +++ b/src/member_files/protocol.c @@ -107,3 +107,9 @@ const char* mf_status_text(int status) { "Busy", "File changed", "Storage full", "I/O error", "Invalid packet", "Hash mismatch", "Cancelled", "Timeout" }; return status >= 0 && status <= MF_TIMEOUT ? texts[status] : "Unknown error"; } + +const char* mf_command_text(int command) { + static const char* commands[] = { "SWEEP", "LIST", "STAT", "READ", "BEGIN_WRITE", "WRITE", "COMMIT", "ABORT", + "MKDIR", "RENAME", "REMOVE" }; + return command >= 0 && command <= MF_REMOVE ? commands[command] : "UNKNOWN"; +} diff --git a/src/member_files/protocol.h b/src/member_files/protocol.h index 02caa2ea..6c0073ec 100644 --- a/src/member_files/protocol.h +++ b/src/member_files/protocol.h @@ -50,6 +50,7 @@ int mf_path_valid(const char* path, int root_allowed); int mf_encode(const struct mf_message* message, uint8_t* wire, size_t capacity); int mf_decode(const uint8_t* wire, size_t length, struct mf_message* message); const char* mf_status_text(int status); +const char* mf_command_text(int command); #ifdef __cplusplus } #endif