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

Honor local auto-download settings and persist storage demand independently

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 13:03:25 +01:00
1 parent bfd14abdbb
commit 89b435e72c
2 files changed
+61 -31
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_mailbox_media.c
+2 -5
View File
@@ -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);
src/dm/dm_media.c
+59 -26
View File
@@ -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);
}
}