From 25d2ac981e41c6db7749b185a880086b68cfcde8 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 13:33:22 +0100 Subject: [PATCH] dm: support colocated endpoint and storage with independent file lifetimes --- src/dm/dm_mailbox_media.c | 3 ++- src/dm/dm_media.c | 48 +++++++++++++++++++++++++++++---------- 2 files changed, 38 insertions(+), 13 deletions(-) diff --git a/src/dm/dm_mailbox_media.c b/src/dm/dm_mailbox_media.c index 70d974ea6..e7eeaad69 100644 --- a/src/dm/dm_mailbox_media.c +++ b/src/dm/dm_mailbox_media.c @@ -216,7 +216,8 @@ static uint64_t dm_job_storage(struct dm_mb_media* media, uint64_t group, const uint64_t node = (uint64_t)sqlite3_column_int64(st, 0); const char* tags = (const char*)sqlite3_column_text(st, 1); char role[16]; - if (node == a || node == b || !tags || json_flat_get(tags, "storage", role, sizeof(role)) || strcmp(role, "yes")) continue; + if (!tags || json_flat_get(tags, "storage", role, sizeof(role)) || strcmp(role, "yes")) continue; + if (!dm_media_scope(inst, group, node, "storage", a, b)) continue; if (node == inst->node_id || etcp_router_get_path(inst, group, node, ETCP_RT_ID_DM_MEDIA).next_hop_node_id) { found = node; break; } } if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: storage scan failed rc=%d", rc); diff --git a/src/dm/dm_media.c b/src/dm/dm_media.c index 156bd89e5..e33bea037 100644 --- a/src/dm/dm_media.c +++ b/src/dm/dm_media.c @@ -293,7 +293,11 @@ void dm_media_forget_committed(struct UTUN_INSTANCE* inst, const char* conv) { sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC); int rc; while ((rc = sqlite3_step(st)) == SQLITE_ROW) { - if (sqlite3_column_bytes(st, 0) == 16) file_transfer_cancel_object(inst->file_transfer, sqlite3_column_blob(st, 0)); + if (sqlite3_column_bytes(st, 0) == 16) { + char path[1024]; + if (!dm_file_path(inst->dm_media, 0, sqlite3_column_blob(st, 0), ".cipher", path, sizeof(path))) + file_transfer_cancel_path(inst->file_transfer, path); + } else DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: damaged deleted media ID conv=%s", conv); } if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: cancellation scan failed conv=%s rc=%d", conv, rc); @@ -512,7 +516,7 @@ static void dm_job_free(struct dm_file_job* job) { } /* Копирование своего source фиксирует один снимок до шифрования. */ -static int dm_file_copy(const char* source, const char* dest) { +static int dm_file_copy(const char* source, const char* dest, uint64_t limit) { FILE* in = fopen(source, "rb"); FILE* out = in ? fopen(dest, "wb") : NULL; int error = -1; @@ -522,7 +526,7 @@ static int dm_file_copy(const char* source, const char* dest) { size_t n; while ((n = fread(buf, 1, sizeof(buf), in))) { size += n; - if (size > DM_MEDIA_MAX_SIZE || fwrite(buf, 1, n, out) != n) { + if (size > limit || fwrite(buf, 1, n, out) != n) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: copy size/write failed bytes=%llu", (unsigned long long)size); goto done; } @@ -544,9 +548,11 @@ static void dm_file_work(void* arg) { return; } if (job->work == DM_FILE_PREPARE) { - int copied = dm_file_copy(job->input, job->plain); + int copied = dm_file_copy(job->input, job->plain, DM_MEDIA_MAX_SIZE); if (job->remove_source && dm_file_remove(job->input)) return; if (copied) return; + } else if (job->input[0] && dm_file_copy(job->input, job->cipher, dm_media_cipher_size(job->size))) { + return; } FILE* in = fopen(job->work == DM_FILE_PREPARE ? job->plain : job->cipher, "rb"); FILE* out = job->work == DM_FILE_CUSTODY ? NULL : fopen(job->temp, "wb"); @@ -661,6 +667,13 @@ static void dm_file_done(void* arg, int error) { custody, ready, (unsigned long long)expires); goto fail; } + if (!custody && job->input[0]) { + st = dm_media_prepare(inst, "SELECT 1 FROM dm_custody WHERE id=? AND ready=1 AND expires>?"); + if (!st) goto fail; + sqlite3_bind_blob(st, 1, job->id, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, ntp_time_get_seconds(inst)); + int active = sqlite3_step(st) == SQLITE_ROW; sqlite3_finalize(st); + if (!active) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM local download: custody expired during copy"); goto fail; } + } uint8_t receipt[DM_MEDIA_RECEIPT_SIZE]; if (!custody && dm_media_receipt_make(inst, body, len, 0, receipt)) goto fail; if (dm_media_exec(inst, "BEGIN IMMEDIATE")) goto fail; @@ -745,6 +758,12 @@ static void dm_file_pull(struct dm_media_state* media, int custody, uint64_t gro if (dm_job_hold(job)) { dm_job_free(job); return; } } job->next = media->jobs; media->jobs = job; + if (peer == media->inst->node_id) { + if (dm_file_path(media, !custody, job->id, ".cipher", job->input, sizeof(job->input))) { dm_file_done(job, -1); return; } + DEBUG_INFO(DEBUG_CATEGORY_DM, "DM local copy: custody=%d bytes=%llu", custody, (unsigned long long)job->size); + media_async_submit(media->inst->media_async, media->inst->ua, dm_file_work, job, dm_file_done, job); + return; + } job->transfer = file_transfer_pull(media->inst->file_transfer, group, peer, job->id, dm_media_cipher_size(job->size), job->cipher, dm_file_received, job); if (!job->transfer) { dm_file_done(job, -1); return; } @@ -797,10 +816,9 @@ static struct dm_file_probe* dm_probe_find(struct dm_media_state* media, const u } 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; + uint64_t group = preferred; + if (node != media->inst->node_id && !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(); @@ -824,7 +842,7 @@ static void dm_probe_start(struct dm_media_state* media, const uint8_t* body, si 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); + for (size_t i = 0; i < count; i++) if (targets[i].node != media->inst->node_id) 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)); @@ -1162,6 +1180,11 @@ rollback: /* Удаление больших файлов и fsync каталога выполняются worker, а не uasync. */ static void dm_custody_remove(struct dm_media_state* media, const uint8_t id[16]) { + char cipher[1024], download[1024]; + if (dm_file_path(media, 1, id, ".cipher", cipher, sizeof(cipher)) || + dm_file_path(media, 1, id, ".download", download, sizeof(download))) return; + file_transfer_cancel_path(media->inst->file_transfer, cipher); + file_transfer_cancel_path(media->inst->file_transfer, download); if (dm_job_find(media, id, 1)) return; struct dm_file_job* job = u_calloc(1, sizeof(*job)); if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete worker allocation failed"); return; } @@ -1202,7 +1225,6 @@ static void dm_custody_delete(struct UTUN_INSTANCE* inst, uint64_t group, uint64 int valid = !dm_media_receipt_save(inst, r, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0)); sqlite3_finalize(st); if (!valid || dm_custody_complete(inst, r + 40, policy) != 1) return; - file_transfer_cancel_object(inst->file_transfer, r + 40); dm_custody_remove(inst->dm_media, r + 40); return; } @@ -1392,7 +1414,8 @@ static void dm_files_tick(struct dm_media_state* media, uint64_t now) { static void dm_custody_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,body,author,supernode,group_id,expires,ready,delete_on_open FROM dm_custody" + sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,body,author,supernode,group_id,expires,ready,delete_on_open," + "EXISTS(SELECT 1 FROM dm_files f WHERE f.id=dm_custody.id AND f.ready=1) FROM dm_custody" " WHERE retry_at<=? ORDER BY retry_at LIMIT 1"); if (!st) return; sqlite3_bind_int64(st, 1, (sqlite3_int64)now); @@ -1409,6 +1432,7 @@ static void dm_custody_tick(struct dm_media_state* media, uint64_t now) { uint64_t group = (uint64_t)sqlite3_column_int64(st, 4), expires = (uint64_t)sqlite3_column_int64(st, 5); int ready = sqlite3_column_int(st, 6); int policy = sqlite3_column_int(st, 7); + int local = sqlite3_column_int(st, 8); sqlite3_finalize(st); if (ready >= 0 && dm_custody_complete(inst, id, policy) == 1) ready = -1; st = dm_media_prepare(inst, "UPDATE dm_custody SET retry_at=?,ready=CASE WHEN expires<=? THEN -1 ELSE ready END WHERE id=?"); @@ -1417,12 +1441,12 @@ static void dm_custody_tick(struct dm_media_state* media, uint64_t now) { sqlite3_bind_blob(st, 3, id, 16, SQLITE_STATIC); if (dm_media_step(inst, st)) return; if (expires <= now || ready < 0) { - file_transfer_cancel_object(inst->file_transfer, id); dm_custody_remove(media, id); } else if (!ready) { const uint8_t* desc = len <= sizeof(body) ? dm_message_media(body, len) : NULL; uint64_t gid = group; - if (desc && dm_route_group(inst, author, group, &gid)) dm_file_pull(media, 1, gid, author, desc, author); + if (desc && local) dm_file_pull(media, 1, group, inst->node_id, desc, author); + else if (desc && dm_route_group(inst, author, group, &gid)) dm_file_pull(media, 1, gid, author, desc, author); } else { dm_custody_status(inst, group, super, id); }