diff --git a/src/dm/dm_mailbox_media.c b/src/dm/dm_mailbox_media.c index c00f65f7f..50c542342 100644 --- a/src/dm/dm_mailbox_media.c +++ b/src/dm/dm_mailbox_media.c @@ -70,7 +70,7 @@ void dm_mailbox_media_receipt(struct UTUN_INSTANCE* inst, const uint8_t* receipt int rc; while ((rc = sqlite3_step(st)) == SQLITE_ROW && n < 64) { uint64_t node = (uint64_t)sqlite3_column_int64(st, 0); - if (node != author && node != recipient && dm_media_scope(inst, groups[i], node, "supernode", author, recipient)) + if (node != author && node != recipient && dm_media_scope(inst, groups[i], node, "supernode", author, inst->node_id)) peers[n++] = node; } if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: supernode scan failed rc=%d", rc); @@ -85,12 +85,9 @@ void dm_mailbox_media_receipt(struct UTUN_INSTANCE* inst, const uint8_t* receipt /* Подтверждение медиадоставки сохраняется до начала DELETE и никогда не отменяется STORED. */ static void dm_job_receipt(struct dm_mb_media* media, uint64_t group, uint64_t source, const uint8_t* r) { struct UTUN_INSTANCE* inst = media->inst; - if (dm_media_verify_receipt(inst, r, NULL, 0)) return; + if (dm_media_receipt_kind(r) < 0) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: invalid receipt domain"); return; } uint64_t author, recipient; memcpy(&author, r + 24, 8); memcpy(&recipient, r + 32, 8); - if (!dm_media_scope(inst, group, inst->node_id, "supernode", author, recipient)) { - DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: receipt outside S scope source=%llu", (unsigned long long)source); return; - } sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body,receipt FROM dm_media_jobs WHERE id=?"); if (!st) return; sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC); diff --git a/src/dm/dm_media.c b/src/dm/dm_media.c index 780f67ed8..7a574ceee 100644 --- a/src/dm/dm_media.c +++ b/src/dm/dm_media.c @@ -1198,7 +1198,14 @@ static void dm_media_offer(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t } return; } - sqlite3_stmt* st = dm_media_prepare(inst, "UPDATE dm_files SET holder=?,expires=?,group_id=?,retry_at=0 WHERE id=? AND ready<=0"); + sqlite3_stmt* st = dm_media_prepare(inst, + "INSERT INTO dm_file_sources(id,node,group_id,expires) VALUES(?,?,?,?) ON CONFLICT(id,node) DO UPDATE SET" + " group_id=excluded.group_id WHERE expires=excluded.expires"); + if (!st) return; + sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)holder); + sqlite3_bind_int64(st, 3, (sqlite3_int64)group); sqlite3_bind_int64(st, 4, (sqlite3_int64)expiry); + if (dm_media_step(inst, st)) return; + st = dm_media_prepare(inst, "UPDATE dm_files SET holder=?,expires=?,group_id=?,retry_at=0 WHERE id=? AND ready<=0"); if (!st) return; sqlite3_bind_int64(st, 1, (sqlite3_int64)holder); sqlite3_bind_int64(st, 2, (sqlite3_int64)expiry); sqlite3_bind_int64(st, 3, (sqlite3_int64)group); sqlite3_bind_blob(st, 4, desc, 16, SQLITE_STATIC); @@ -1252,11 +1259,34 @@ done: if (e) { queue_dgram_free(e); queue_entry_free(e); } } +/* Автор повторяет все полученные факты к S, остальные устройства — только свои. */ +static void dm_file_forward_facts(struct dm_media_state* media, const uint8_t id[16], uint64_t author, uint64_t group) { + struct UTUN_INSTANCE* inst = media->inst; + sqlite3_stmt* st = dm_media_prepare(inst, + "SELECT received FROM dm_file_targets WHERE id=?1 AND received IS NOT NULL AND (?2=0 OR node=?2)" + " UNION ALL SELECT taken FROM dm_file_targets WHERE id=?1 AND taken IS NOT NULL AND (?2=0 OR node=?2)"); + if (!st) return; + sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + sqlite3_bind_int64(st, 2, author == inst->node_id ? 0 : (sqlite3_int64)inst->node_id); + uint8_t facts[2 * DM_TARGET_MAX][DM_MEDIA_RECEIPT_SIZE]; unsigned n = 0; int rc; + while ((rc = sqlite3_step(st)) == SQLITE_ROW && n < 2 * DM_TARGET_MAX) { + if (sqlite3_column_bytes(st, 0) != DM_MEDIA_RECEIPT_SIZE) { rc = SQLITE_CORRUPT; break; } + memcpy(facts[n++], sqlite3_column_blob(st, 0), DM_MEDIA_RECEIPT_SIZE); + } + sqlite3_finalize(st); + if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM facts: scan failed rc=%d", rc); return; } + uint64_t gid; int route = author != inst->node_id && dm_route_group(inst, author, group, &gid); + for (unsigned i = 0; i < n; i++) { + if (route) dm_media_send(inst, gid, author, DM_MEDIA_RECEIPT, facts[i], DM_MEDIA_RECEIPT_SIZE); + dm_mailbox_media_receipt(inst, facts[i]); + } +} + /* Текст и файл имеют независимые durable retries; наличие text ACK не останавливает медиа. */ static void dm_files_tick(struct dm_media_state* media, uint64_t now) { struct UTUN_INSTANCE* inst = media->inst; for (int i = 0; i < 8; i++) { - sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,dir,author,recipient,group_id,body,ready,holder,expires,receipt" + sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,dir,author,recipient,group_id,body,ready,manual,first_attempt,plain_size" " FROM dm_files WHERE retry_at<=? AND conv_id IN" " (SELECT conv_id FROM dm_conversations WHERE closed=0) ORDER BY retry_at,seq LIMIT 1"); if (!st) return; @@ -1266,44 +1296,47 @@ static void dm_files_tick(struct dm_media_state* media, uint64_t now) { if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != 16) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: local files scan failed rc=%d", rc); sqlite3_finalize(st); return; } - uint8_t id[16], body[DM_MEDIA_BODY_MAX], receipt[DM_MEDIA_RECEIPT_SIZE]; + uint8_t id[16], body[DM_MEDIA_BODY_MAX]; memcpy(id, sqlite3_column_blob(st, 0), 16); int dir = sqlite3_column_int(st, 1), ready = sqlite3_column_int(st, 6); - uint64_t author = (uint64_t)sqlite3_column_int64(st, 2), recipient = (uint64_t)sqlite3_column_int64(st, 3); - uint64_t group = (uint64_t)sqlite3_column_int64(st, 4), holder = (uint64_t)sqlite3_column_int64(st, 7); - uint64_t expiry = (uint64_t)sqlite3_column_int64(st, 8); + uint64_t author = (uint64_t)sqlite3_column_int64(st, 2), group = (uint64_t)sqlite3_column_int64(st, 4); + int manual = sqlite3_column_int(st, 7); + uint64_t first = (uint64_t)sqlite3_column_int64(st, 8), size = (uint64_t)sqlite3_column_int64(st, 9); size_t len = (size_t)sqlite3_column_bytes(st, 5); - int delivered = sqlite3_column_bytes(st, 9) == sizeof(receipt); if (len <= sizeof(body)) memcpy(body, sqlite3_column_blob(st, 5), len); - if (delivered) memcpy(receipt, sqlite3_column_blob(st, 9), sizeof(receipt)); sqlite3_finalize(st); - st = dm_media_prepare(inst, "UPDATE dm_files SET retry_at=? WHERE id=?"); + st = dm_media_prepare(inst, "UPDATE dm_files SET retry_at=?1,first_attempt=CASE WHEN first_attempt=0 THEN ?3 ELSE first_attempt END WHERE id=?2"); if (!st) return; - sqlite3_bind_int64(st, 1, (sqlite3_int64)(now + (dir && delivered ? 3600 : 5))); + sqlite3_bind_int64(st, 1, (sqlite3_int64)(now + 5)); sqlite3_bind_int64(st, 3, (sqlite3_int64)now); sqlite3_bind_blob(st, 2, id, 16, SQLITE_STATIC); if (dm_media_step(inst, st)) return; const uint8_t* desc = len <= sizeof(body) ? dm_message_media(body, len) : NULL; if (!desc) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: corrupt saved file descriptor"); continue; } - if (dir && !delivered) { - uint64_t gid = group; - if (dm_route_group(inst, recipient, group, &gid)) { + if (dir) { + st = dm_media_prepare(inst, "SELECT node FROM dm_file_targets WHERE id=? AND received IS NULL ORDER BY side DESC,node"); + if (!st) return; + sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + uint64_t nodes[DM_TARGET_MAX]; unsigned count = 0; + while ((rc = sqlite3_step(st)) == SQLITE_ROW && count < DM_TARGET_MAX) nodes[count++] = (uint64_t)sqlite3_column_int64(st, 0); + sqlite3_finalize(st); + if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: target scan failed rc=%d", rc); continue; } + for (unsigned j = 0; j < count; j++) { + uint64_t gid; + if (!dm_route_group(inst, nodes[j], group, &gid)) continue; uint8_t offer[24 + DM_MEDIA_BODY_MAX]; - memcpy(offer, &recipient, 8); memcpy(offer + 8, &author, 8); memset(offer + 16, 0, 8); + memcpy(offer, nodes + j, 8); memcpy(offer + 8, &author, 8); memset(offer + 16, 0, 8); memcpy(offer + 24, body, len); - dm_media_send(inst, gid, recipient, DM_MEDIA_OFFER, offer, len + 24); - } else dm_mailbox_put(inst, recipient, author, body, len); - } else if (!dir && ready == 1 && delivered) { - uint64_t gid = group; - if (dm_route_group(inst, author, group, &gid)) dm_media_send(inst, gid, author, DM_MEDIA_RECEIPT, receipt, sizeof(receipt)); - dm_mailbox_media_receipt(inst, receipt); - } else if (!dir && ready <= 0 && !dm_job_find(media, id, 0)) { - uint64_t peer = author, gid = group; - if (!dm_route_group(inst, author, group, &gid)) { - peer = holder; - if (holder == author || (expiry && expiry <= now) || !dm_route_group(inst, holder, group, &gid)) continue; + dm_media_send(inst, gid, nodes[j], DM_MEDIA_OFFER, offer, len + 24); } - dm_file_pull(media, 0, gid, peer, desc, author); + if (count && first && now >= first + 5) dm_mailbox_store_file(inst, nodes[0], body, len); + } else if (ready <= 0 && !dm_job_find(media, id, 0)) { + uint64_t limit = inst->config->global.chatserver_storage_unit_size; + if (manual || (chat_setting_get_int(inst, "storage_autoload", 1) && limit && size <= limit)) + dm_probe_start(media, body, len, group); + else DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM files: waiting for manual download bytes=%llu autoload=%d limit=%llu", + (unsigned long long)size, chat_setting_get_int(inst, "storage_autoload", 1), (unsigned long long)limit); } + dm_file_forward_facts(media, id, author, group); } }