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

dm: support colocated endpoint and storage with independent file lifetimes

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 13:33:22 +01:00
1 parent c9b71ee1c9
commit 25d2ac981e
2 files changed
+38 -13
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_mailbox_media.c
+2 -1
View File
@@ -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);
src/dm/dm_media.c
+36 -12
View File
@@ -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);
}