diff --git a/src/dm/dm_media.c b/src/dm/dm_media.c index 9ded229ac..780f67ed8 100644 --- a/src/dm/dm_media.c +++ b/src/dm/dm_media.c @@ -30,11 +30,13 @@ enum dm_file_work { DM_FILE_PREPARE, DM_FILE_RECEIVE, DM_FILE_CUSTODY, DM_FILE_DELETE }; struct dm_file_job; +struct dm_file_probe; struct dm_media_state { struct UTUN_INSTANCE* inst; char root[768]; char pending[768]; struct dm_file_job* jobs; + struct dm_file_probe* probes; void* timer; int closing; int cleanup_active; @@ -58,6 +60,17 @@ struct dm_file_job { unsigned held; }; +#define DM_SOURCE_MAX (DM_TARGET_MAX + 64) +struct dm_probe_source { uint64_t node, group, sent, rtt; int used; }; +struct dm_file_probe { + struct dm_file_probe* next; + uint8_t desc[DM_MEDIA_DESC_SIZE]; + uint64_t token, deadline, author; + char conv[64]; + unsigned count; + struct dm_probe_source sources[DM_SOURCE_MAX]; +}; + sqlite3_stmt* dm_media_prepare(struct UTUN_INSTANCE* inst, const char* sql) { sqlite3_stmt* st = NULL; int rc = sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL); @@ -741,7 +754,7 @@ static void dm_file_pull(struct dm_media_state* media, int custody, uint64_t gro } } -/* Общий авторизатор сервера transfer: A отдаёт B/storage, M — только B по custody. */ +/* Поставщиком может быть любая готовая реплика; UUID сам по себе права не даёт. */ static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_t id[16], uint64_t size, char* path, size_t cap) { struct dm_media_state* media = arg; @@ -749,8 +762,8 @@ static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_ if (media->closing) return -1; for (int custody = 0; custody <= 1; custody++) { sqlite3_stmt* st = dm_media_prepare(inst, custody ? - "SELECT author,recipient,plain_size,ready,expires FROM dm_custody WHERE id=?" : - "SELECT author,recipient,plain_size,ready,0 FROM dm_files WHERE id=? AND dir=1"); + "SELECT author,recipient,plain_size,ready,expires,body,NULL FROM dm_custody WHERE id=?" : + "SELECT author,recipient,plain_size,ready,0,body,conv_id FROM dm_files WHERE id=?"); if (!st) return -1; sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); int rc = sqlite3_step(st); @@ -759,10 +772,17 @@ static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_ uint64_t plain = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 2) : 0; int ready = rc == SQLITE_ROW ? sqlite3_column_int(st, 3) : 0; uint64_t expiry = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 4) : 0; + int allowed = 0; + struct dm_message_view view; + if (rc == SQLITE_ROW && !dm_message_parse(sqlite3_column_blob(st, 5), (size_t)sqlite3_column_bytes(st, 5), &view)) { + if (view.mode == DM_MODE_DEVICE) allowed = peer == author || peer == recipient; + else allowed = dm_party_member(inst, view.parties, peer) == 1 || dm_party_member(inst, view.parties + 1, peer) == 1; + } + if (!custody && rc == SQLITE_ROW && dm_conversation_access(inst, (const char*)sqlite3_column_text(st, 6), NULL, NULL)) ready = 0; sqlite3_finalize(st); if (!author || ready != 1 || size != dm_media_cipher_size(plain) || (custody && expiry <= ntp_time_get_seconds(inst))) continue; - if ((peer == recipient && dm_media_scope(inst, group, peer, NULL, author, recipient)) || + if ((allowed && dm_media_scope(inst, group, peer, NULL, inst->node_id, peer)) || (!custody && dm_media_scope(inst, group, peer, "storage", author, recipient))) return dm_file_path(media, custody, id, ".cipher", path, cap); } @@ -770,6 +790,109 @@ static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_ return -1; } +/* Не менять активную передачу из-за позднего ответа; оставшиеся ответы — резерв. */ +static struct dm_file_probe* dm_probe_find(struct dm_media_state* media, const uint8_t id[16]) { + for (struct dm_file_probe* p = media->probes; p; p = p->next) if (!memcmp(p->desc, id, 16)) return p; + return NULL; +} + +static void dm_probe_add(struct dm_media_state* media, struct dm_file_probe* p, uint64_t node, uint64_t preferred) { + if (node == media->inst->node_id) return; + for (unsigned i = 0; i < p->count; i++) if (p->sources[i].node == node) return; + uint64_t group; + if (!dm_route_group(media->inst, node, preferred, &group)) return; + if (p->count == DM_SOURCE_MAX) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM sources: candidate limit reached conv=%s", p->conv); return; } + struct dm_probe_source* source = p->sources + p->count++; + source->node = node; source->group = group; source->sent = get_time_tb(); + uint8_t request[64]; uint64_t plain, size; memcpy(&plain, p->desc + 16, 8); size = dm_media_cipher_size(plain); + memcpy(request, p->desc, 16); memcpy(request + 16, p->desc + 24, 32); + memcpy(request + 48, &p->token, 8); memcpy(request + 56, &size, 8); + if (dm_media_send(media->inst, group, node, DM_MEDIA_HAVE, request, sizeof(request))) source->used = 1; +} + +/* Одна ограниченная параллельная проба по устройствам обеих сторон и известным M. */ +static void dm_probe_start(struct dm_media_state* media, const uint8_t* body, size_t len, uint64_t preferred) { + struct dm_message_view view; + if (dm_message_parse(body, len, &view) || !view.media || dm_probe_find(media, view.media)) return; + struct dm_file_probe* p = u_calloc(1, sizeof(*p)); + if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM sources: allocation failed"); return; } + memcpy(p->desc, view.media, sizeof(p->desc)); p->author = view.author; + snprintf(p->conv, sizeof(p->conv), "%llu", (unsigned long long)view.conv); + struct dm_party parties[2]; struct dm_target targets[DM_TARGET_MAX]; size_t count; + if (dm_conversation_access(media->inst, p->conv, parties, NULL) || + dm_identity_targets(media->inst, parties, targets, DM_TARGET_MAX, &count) || RAND_bytes((uint8_t*)&p->token, 8) != 1) { + DEBUG_WARN(DEBUG_CATEGORY_DM, "DM sources: party/source snapshot unavailable conv=%s", p->conv); u_free(p); return; + } + p->deadline = get_time_tb() + 10000; p->next = media->probes; media->probes = p; + for (size_t i = 0; i < count; i++) dm_probe_add(media, p, targets[i].node, preferred); + sqlite3_stmt* st = dm_media_prepare(media->inst, "SELECT node,group_id FROM dm_file_sources WHERE id=? AND (expires=0 OR expires>?)"); + if (st) { + sqlite3_bind_blob(st, 1, p->desc, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, ntp_time_get_seconds(media->inst)); + uint64_t nodes[64], groups[64]; unsigned n = 0; int rc; + while ((rc = sqlite3_step(st)) == SQLITE_ROW && n < 64) { + nodes[n] = (uint64_t)sqlite3_column_int64(st, 0); groups[n++] = (uint64_t)sqlite3_column_int64(st, 1); + } + sqlite3_finalize(st); + if (rc != SQLITE_DONE) DEBUG_WARN(DEBUG_CATEGORY_DM, "DM sources: holder scan incomplete rc=%d count=%u", rc, n); + for (unsigned i = 0; i < n; i++) dm_probe_add(media, p, nodes[i], groups[i]); + } + DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM sources: parallel HAVE conv=%s candidates=%u window_ms=1000", p->conv, p->count); +} + +static void dm_probe_control(struct dm_media_state* media, uint64_t group, uint64_t peer, int reply, const uint8_t* request) { + uint64_t token, size; memcpy(&token, request + 48, 8); memcpy(&size, request + 56, 8); + if (reply) { + struct dm_file_probe* p = dm_probe_find(media, request); + if (!p || p->token != token || memcmp(request + 16, p->desc + 24, 32) || get_time_tb() > p->deadline) return; + for (unsigned i = 0; i < p->count; i++) { + struct dm_probe_source* s = p->sources + i; + if (s->node == peer && s->group == group && !s->used && !s->rtt) { + s->rtt = get_time_tb() - s->sent + 1; + DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM sources: HAVE peer=%llu rtt_us=%llu", (unsigned long long)peer, + (unsigned long long)(s->rtt * 100)); return; + } + } + return; + } + sqlite3_stmt* st = dm_media_prepare(media->inst, + "SELECT body FROM dm_files WHERE id=? UNION ALL SELECT body FROM dm_custody WHERE id=?"); + if (!st) return; + sqlite3_bind_blob(st, 1, request, 16, SQLITE_STATIC); sqlite3_bind_blob(st, 2, request, 16, SQLITE_STATIC); + int found = 0, rc; + while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + const uint8_t* desc = dm_message_media(sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0)); + if (desc && !memcmp(desc + 24, request + 16, 32)) { found = 1; break; } + } + sqlite3_finalize(st); + if (!found) return; + char path[1024]; struct stat sb; + if (!dm_file_lookup(media, group, peer, request, size, path, sizeof(path)) && !stat(path, &sb) && (uint64_t)sb.st_size == size) + dm_media_send(media->inst, group, peer, DM_MEDIA_HAVE_ACK, request, 64); +} + +/* После окна проб выбираем минимальный RTT. Ошибка transfer/проверки оставляет резервные источники. */ +static void dm_probe_tick(struct dm_media_state* media) { + struct dm_file_probe** link = &media->probes; + while (*link) { + struct dm_file_probe* p = *link; + if (get_time_tb() < p->deadline || dm_job_find(media, p->desc, 0)) { link = &p->next; continue; } + sqlite3_stmt* st = dm_media_prepare(media->inst, "SELECT f.ready,c.closed FROM dm_files f JOIN dm_conversations c USING(conv_id) WHERE f.id=?"); + if (!st) return; + sqlite3_bind_blob(st, 1, p->desc, 16, SQLITE_STATIC); int rc = sqlite3_step(st); + int pending = rc == SQLITE_ROW && sqlite3_column_int(st, 0) != 1 && !sqlite3_column_int(st, 1); + sqlite3_finalize(st); + unsigned best = p->count; + if (pending) for (unsigned i = 0; i < p->count; i++) if (!p->sources[i].used && p->sources[i].rtt && + (best == p->count || p->sources[i].rtt < p->sources[best].rtt)) best = i; + if (best == p->count) { *link = p->next; u_free(p); continue; } + struct dm_probe_source* s = p->sources + best; s->used = 1; + DEBUG_INFO(DEBUG_CATEGORY_DM, "DM sources: download conv=%s peer=%llu rtt_us=%llu", p->conv, + (unsigned long long)s->node, (unsigned long long)(s->rtt * 100)); + dm_file_pull(media, 0, s->group, s->node, p->desc, p->author); + link = &p->next; + } +} + /* Путь исходника копируется; caller может освободить строку после возврата. */ int dm_send_file(struct UTUN_INSTANCE* inst, const char* conv, const char* path, const struct attachment_info* info, int remove_source, uint64_t request_id) { @@ -1087,6 +1210,8 @@ static void dm_media_control(struct UTUN_INSTANCE* inst, uint64_t group, uint64_ uint8_t cmd, const uint8_t* p, size_t len) { if (!inst->dm_media || inst->dm_media->closing) return; if (cmd == DM_MEDIA_OFFER) dm_media_offer(inst, group, source, p, len); + else if ((cmd == DM_MEDIA_HAVE || cmd == DM_MEDIA_HAVE_ACK) && len == 64) + dm_probe_control(inst->dm_media, group, source, cmd == DM_MEDIA_HAVE_ACK, p); else if (cmd == DM_MEDIA_STORE) dm_custody_store(inst, group, source, p, len); else if (cmd == DM_MEDIA_DELETE && len == DM_MEDIA_RECEIPT_SIZE) dm_custody_delete(inst, group, source, p); else if (cmd == DM_MEDIA_RECEIPT && len == DM_MEDIA_RECEIPT_SIZE) { @@ -1228,6 +1353,7 @@ static void dm_media_tick(void* arg) { media->timer = NULL; uint64_t now = ntp_time_get_seconds(media->inst); dm_files_tick(media, now); + dm_probe_tick(media); dm_custody_tick(media, now); dm_cleanup_pump(media); media->timer = uasync_set_timeout(media->inst->ua, 10000, media, dm_media_tick, "dm_media"); @@ -1283,6 +1409,7 @@ void dm_media_quiesce(struct UTUN_INSTANCE* inst) { media->timer = NULL; etcp_router_unbind(inst, ETCP_RT_ID_DM_MEDIA); file_transfer_destroy(inst->file_transfer); + while (media->probes) { struct dm_file_probe* p = media->probes; media->probes = p->next; u_free(p); } DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: quiesced, waiting for workers"); }