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

dm: bind custody replies to persisted assignment identities

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 22:37:35 +01:00
1 parent 1d0459534d
commit 13d838988e
2 files changed
+104 -35
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_mailbox_media.c
+94 -30
View File
@@ -2,6 +2,7 @@
#include <stdio.h>
#include <string.h>
#include <openssl/sha.h>
#include <openssl/rand.h>
#include "../../lib/mem.h"
#include "../../lib/debug_config.h"
@@ -111,18 +112,24 @@ static void dm_job_receipt(struct dm_mb_media* media, uint64_t group, uint64_t s
}
/* Освобождение назначения и сброс текущего holder — один durable переход. */
static int dm_job_release(struct dm_mb_media* media, const uint8_t id[16], uint64_t holder, int expired) {
static int dm_job_release(struct dm_mb_media* media, const uint8_t id[16], uint64_t holder,
const uint8_t assignment[16], int expired) {
struct UTUN_INSTANCE* inst = media->inst;
if (dm_media_exec(inst, "BEGIN IMMEDIATE")) return -1;
const char* sql[] = { "UPDATE dm_media_holders SET released=1 WHERE id=? AND node_id=? AND released=0", expired ?
"UPDATE dm_media_jobs SET holder=0,stored=0,expired=1,retry_at=0 WHERE id=? AND holder=? AND receipt IS NULL" :
"UPDATE dm_media_jobs SET holder=0,stored=0,retry_at=0 WHERE id=? AND holder=? AND receipt IS NULL" };
const char* sql[] = { expired ?
"UPDATE dm_media_holders SET released=2 WHERE id=? AND node_id=? AND assignment_id=? AND released=0" :
"UPDATE dm_media_holders SET released=3 WHERE id=? AND node_id=? AND assignment_id=? AND released=0", expired ?
"UPDATE dm_media_jobs SET holder=0,stored=0,expired=1,retry_at=0 WHERE id=?1 AND holder=?2 AND receipt IS NULL"
" AND EXISTS(SELECT 1 FROM dm_media_holders h WHERE h.id=?1 AND h.node_id=?2 AND h.assignment_id=?3)" :
"UPDATE dm_media_jobs SET holder=0,stored=0,expires=0,delete_on_open=-1,retry_at=0 WHERE id=?1 AND holder=?2 AND receipt IS NULL"
" AND EXISTS(SELECT 1 FROM dm_media_holders h WHERE h.id=?1 AND h.node_id=?2 AND h.assignment_id=?3)" };
int changed[2];
for (int i = 0; i < 2; i++) {
sqlite3_stmt* st = dm_media_prepare(inst, sql[i]);
if (!st) goto rollback;
int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)holder);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, assignment, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) {
@@ -150,23 +157,36 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
if (!media) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: control after stop"); return; }
struct UTUN_INSTANCE* inst = media->inst;
if (cmd == DM_MEDIA_RECEIPT && len == DM_MEDIA_RECEIPT_SIZE) { dm_job_receipt(media, group, source, p); return; }
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT j.author,j.recipient,j.receipt,j.holder FROM dm_media_jobs j"
" JOIN dm_media_holders h ON h.id=j.id WHERE j.id=? AND h.node_id=?");
if (!p || !(((cmd == DM_MEDIA_STORED || cmd == DM_MEDIA_RESERVED) && len == 41) ||
((cmd == DM_MEDIA_DELETE_ACK || cmd == DM_MEDIA_EXPIRED) && len == 32) || (cmd == DM_MEDIA_REJECT && len == 33))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM assignment: invalid response cmd=%u bytes=%zu", cmd, len); return;
}
const uint8_t* assignment = p; p += 16; len -= 16;
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT j.receipt,j.holder,h.released FROM dm_media_jobs j"
" JOIN dm_media_holders h ON h.id=j.id WHERE j.id=? AND h.node_id=? AND h.assignment_id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)source);
sqlite3_bind_blob(st, 3, assignment, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); return; }
if (rc == SQLITE_DONE) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM assignment: unknown/stale response M=%llu cmd=%u", (unsigned long long)source, cmd);
sqlite3_finalize(st); return;
}
if (rc != SQLITE_ROW) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: location lookup failed rc=%d", rc); sqlite3_finalize(st); return; }
uint64_t holder = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t holder = (uint64_t)sqlite3_column_int64(st, 1);
int released = sqlite3_column_int(st, 2);
uint8_t receipt[DM_MEDIA_RECEIPT_SIZE];
int delivered = sqlite3_column_bytes(st, 2) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 2), sizeof(receipt));
int delivered = sqlite3_column_bytes(st, 0) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 0), sizeof(receipt));
sqlite3_finalize(st);
if ((cmd == DM_MEDIA_STORED || cmd == DM_MEDIA_RESERVED) && len == 25 && p[24] <= 1) {
uint64_t expiry;
memcpy(&expiry, p + 16, 8);
if (delivered) { dm_media_send(inst, group, source, DM_MEDIA_DELETE, receipt, sizeof(receipt)); return; }
if (holder != source || expiry <= ntp_time_get_seconds(inst)) {
if (delivered) {
uint8_t request[16 + DM_MEDIA_RECEIPT_SIZE]; memcpy(request, assignment, 16); memcpy(request + 16, receipt, sizeof(receipt));
dm_media_send(inst, group, source, DM_MEDIA_DELETE, request, sizeof(request)); return;
}
if (released || holder != source || expiry <= ntp_time_get_seconds(inst)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: stale/unauthorized STORED source=%llu expiry=%llu",
(unsigned long long)source, (unsigned long long)expiry); return;
}
@@ -180,11 +200,12 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: custody committed M=%llu expires=%llu",
(unsigned long long)source, (unsigned long long)expiry);
} else if ((cmd == DM_MEDIA_DELETE_ACK || cmd == DM_MEDIA_EXPIRED) && len == 16) {
if (cmd == DM_MEDIA_EXPIRED && !delivered) { dm_job_release(media, p, source, 1); return; }
st = dm_media_prepare(inst, "UPDATE dm_media_holders SET released=1 WHERE id=? AND node_id=? AND released=0");
if (cmd == DM_MEDIA_EXPIRED && !delivered) { dm_job_release(media, p, source, assignment, 1); return; }
st = dm_media_prepare(inst, "UPDATE dm_media_holders SET released=1 WHERE id=? AND node_id=? AND assignment_id=? AND released=0");
if (!st) return;
rc = sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)source);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, assignment, 16, SQLITE_STATIC);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: ACK bind failed M=%llu cmd=%u rc=%d",
(unsigned long long)source, cmd, rc);
@@ -194,7 +215,7 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: assignment released M=%llu cmd=%u", (unsigned long long)source, cmd);
} else if (cmd == DM_MEDIA_REJECT && len == 17 && !delivered) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: storage refused M=%llu reason=%u", (unsigned long long)source, p[16]);
dm_job_release(media, p, source, 0);
if (!released) dm_job_release(media, p, source, assignment, 0);
} else DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: unexpected control cmd=%u bytes=%zu", cmd, len);
}
@@ -236,7 +257,7 @@ static uint64_t dm_job_storage(struct dm_mb_media* media, const uint8_t id[16])
/* Удаление неосвобождённых назначений повторяется до DELETE_ACK; истёкшие завершаются по TTL. */
static void dm_job_cleanup(struct dm_mb_media* media, const uint8_t id[16], int terminal) {
struct UTUN_INSTANCE* inst = media->inst;
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT node_id FROM dm_media_holders WHERE id=? AND released=0 LIMIT 16");
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT node_id,assignment_id FROM dm_media_holders WHERE id=? AND released=0 LIMIT 16");
if (!st) return;
int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc != SQLITE_OK) {
@@ -244,9 +265,11 @@ static void dm_job_cleanup(struct dm_mb_media* media, const uint8_t id[16], int
sqlite3_finalize(st); return;
}
uint64_t peers[16];
uint8_t assignments[16][16];
unsigned count = 0;
while ((rc = sqlite3_step(st)) == SQLITE_ROW && count < 16) {
peers[count++] = (uint64_t)sqlite3_column_int64(st, 0);
if (sqlite3_column_bytes(st, 1) != 16) { rc = SQLITE_CORRUPT; break; }
peers[count] = (uint64_t)sqlite3_column_int64(st, 0); memcpy(assignments[count++], sqlite3_column_blob(st, 1), 16);
}
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
@@ -263,8 +286,10 @@ static void dm_job_cleanup(struct dm_mb_media* media, const uint8_t id[16], int
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: corrupt/overflowing receipt set rc=%d", rc); return; }
for (unsigned i = 0; i < count; i++) for (unsigned j = 0; j < fact_count; j++)
dm_media_send(inst, 0, peers[i], DM_MEDIA_DELETE, facts[j], DM_MEDIA_RECEIPT_SIZE);
for (unsigned i = 0; i < count; i++) for (unsigned j = 0; j < fact_count; j++) {
uint8_t request[16 + DM_MEDIA_RECEIPT_SIZE]; memcpy(request, assignments[i], 16); memcpy(request + 16, facts[j], sizeof(facts[j]));
dm_media_send(inst, 0, peers[i], DM_MEDIA_DELETE, request, sizeof(request));
}
if (!count && terminal) {
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET body=NULL WHERE id=? AND receipt IS NOT NULL AND body IS NOT NULL"
" AND NOT EXISTS(SELECT 1 FROM dm_media_holders h WHERE h.id=dm_media_jobs.id AND h.released=0)");
@@ -279,6 +304,36 @@ static void dm_job_cleanup(struct dm_mb_media* media, const uint8_t id[16], int
}
}
/* Не получившие файл устройства узнают об удалённой копии; отказ в резерве не означает удаления. */
static void dm_job_source_notices(struct UTUN_INSTANCE* inst, const uint8_t id[16]) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT h.node_id,j.body_hash,t.node FROM dm_media_holders h"
" JOIN dm_media_jobs j ON j.id=h.id JOIN dm_file_targets t ON t.id=h.id"
" WHERE h.id=? AND h.released IN (1,2) AND h.node_id<>j.author AND t.received IS NULL");
if (!st) return;
struct source_notice { uint64_t peer; uint8_t body[56]; } *notices = NULL;
size_t count = 0, capacity = 0;
int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
if (sqlite3_column_bytes(st, 1) != 32) { rc = SQLITE_CORRUPT; break; }
if (count == capacity) {
size_t next = capacity ? capacity * 2 : 8;
struct source_notice* grown = u_realloc(notices, next * sizeof(*notices));
if (!grown) { rc = SQLITE_NOMEM; break; }
notices = grown; capacity = next;
}
uint64_t holder = (uint64_t)sqlite3_column_int64(st, 0);
notices[count].peer = (uint64_t)sqlite3_column_int64(st, 2);
memcpy(notices[count].body, id, 16); memcpy(notices[count].body + 16, sqlite3_column_blob(st, 1), 32);
memcpy(notices[count++].body + 48, &holder, 8);
}
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM sources: terminal assignment snapshot failed rc=%d final=%d", rc, final);
} else for (size_t i = 0; i < count; i++)
dm_media_send(inst, 0, notices[i].peer, DM_MEDIA_SOURCE_GONE, notices[i].body, sizeof(notices[i].body));
u_free(notices);
}
/* Snapshot statement закрывается до control send, включая локальное совмещение S/M. */
static void dm_mailbox_media_tick(void* arg) {
struct dm_mb_media* media = arg;
@@ -286,8 +341,9 @@ static void dm_mailbox_media_tick(void* arg) {
media->timer = NULL;
uint64_t now = ntp_time_get_seconds(inst);
for (int i = 0; i < 8; i++) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,author,recipient,body,holder,stored,expires,receipt,expired,delete_on_open"
" FROM dm_media_jobs WHERE retry_at<=? ORDER BY retry_at,seq LIMIT 1");
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT j.id,j.author,j.recipient,j.body,j.holder,j.stored,j.expires,j.receipt,"
"j.expired,j.delete_on_open,h.assignment_id,h.released FROM dm_media_jobs j LEFT JOIN dm_media_holders h"
" ON h.id=j.id AND h.node_id=j.holder WHERE j.retry_at<=? ORDER BY j.retry_at,j.seq LIMIT 1");
if (!st) break;
sqlite3_bind_int64(st, 1, (sqlite3_int64)now);
int rc = sqlite3_step(st);
@@ -301,6 +357,12 @@ static void dm_mailbox_media_tick(void* arg) {
size_t len = (size_t)sqlite3_column_bytes(st, 3);
if (len <= sizeof(body) && len) memcpy(body, sqlite3_column_blob(st, 3), len);
uint64_t holder = (uint64_t)sqlite3_column_int64(st, 4);
uint8_t assignment[16];
if (holder && sqlite3_column_bytes(st, 10) != 16) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: missing/corrupt ID M=%llu", (unsigned long long)holder); sqlite3_finalize(st); break;
}
if (holder) memcpy(assignment, sqlite3_column_blob(st, 10), 16);
int released = sqlite3_column_int(st, 11);
int stored = sqlite3_column_int(st, 5), expired = sqlite3_column_int(st, 8);
int policy = sqlite3_column_int(st, 9);
uint64_t expires = (uint64_t)sqlite3_column_int64(st, 6);
@@ -311,6 +373,7 @@ static void dm_mailbox_media_tick(void* arg) {
if (!st) break;
sqlite3_bind_int64(st, 1, (sqlite3_int64)(now + 5)); sqlite3_bind_blob(st, 2, id, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) break;
dm_job_source_notices(inst, id);
if (!delivered && policy >= 0 && dm_media_retention_done(inst, id, policy) == 1) {
st = dm_media_prepare(inst, policy ?
"UPDATE dm_media_jobs SET receipt=(SELECT taken FROM dm_file_targets WHERE id=? AND side=1 AND taken IS NOT NULL LIMIT 1)"
@@ -329,7 +392,7 @@ static void dm_mailbox_media_tick(void* arg) {
continue;
}
if (holder && expires && expires <= now) {
if (!dm_job_release(media, id, holder, 1))
if (!dm_job_release(media, id, holder, assignment, 1))
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: custody deadline reconciled M=%llu expires=%llu now=%llu",
(unsigned long long)holder, (unsigned long long)expires, (unsigned long long)now);
continue; /* Новый снимок после commit; при ошибке задача повторится через retry_at. */
@@ -348,22 +411,25 @@ static void dm_mailbox_media_tick(void* arg) {
uint64_t gid = 0;
if (!dm_route_group(inst, targets[t], 0, &gid)) continue;
uint8_t offer[24 + DM_MEDIA_BODY_MAX];
uint64_t location = stored && expires > now ? holder : author;
uint64_t expiry = stored ? expires : 0;
uint64_t location = stored && !released && expires > now ? holder : author;
uint64_t expiry = stored && !released ? expires : 0;
memcpy(offer, targets + t, 8); memcpy(offer + 8, &location, 8); memcpy(offer + 16, &expiry, 8);
memcpy(offer + 24, body, len);
dm_media_send(inst, gid, targets[t], DM_MEDIA_OFFER, offer, len + 24);
if (expired || (stored && expires <= now)) dm_media_send(inst, gid, targets[t], DM_MEDIA_EXPIRED, id, 16);
}
if (!expired) {
if (!holder) {
if (!dm_peer_role(inst, inst->node_id, "supernode")) continue;
holder = dm_job_storage(media, id);
if (!holder) continue;
if (RAND_bytes(assignment, sizeof(assignment)) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: random ID generation failed"); continue;
}
if (dm_media_exec(inst, "BEGIN IMMEDIATE")) continue;
st = dm_media_prepare(inst, "INSERT INTO dm_media_holders(id,node_id) VALUES(?,?)");
st = dm_media_prepare(inst, "INSERT INTO dm_media_holders(id,node_id,assignment_id) VALUES(?,?,?)");
if (!st) { dm_media_exec(inst, "ROLLBACK"); continue; }
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)holder);
sqlite3_bind_blob(st, 3, assignment, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) { dm_media_exec(inst, "ROLLBACK"); continue; }
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET holder=?,stored=0 WHERE id=? AND receipt IS NULL");
if (!st) { dm_media_exec(inst, "ROLLBACK"); continue; }
@@ -371,10 +437,8 @@ static void dm_mailbox_media_tick(void* arg) {
if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) { dm_media_exec(inst, "ROLLBACK"); continue; }
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: assigned storage M=%llu", (unsigned long long)holder);
}
if (!stored) {
uint8_t put[8 + DM_MEDIA_BODY_MAX]; memcpy(put, &recipient, 8); memcpy(put + 8, body, len);
dm_media_send(inst, 0, holder, DM_MEDIA_STORE, put, len + 8);
}
uint8_t put[24 + DM_MEDIA_BODY_MAX]; memcpy(put, assignment, 16); memcpy(put + 16, &recipient, 8); memcpy(put + 24, body, len);
dm_media_send(inst, 0, holder, DM_MEDIA_STORE, put, len + 24);
}
}
media->timer = uasync_set_timeout(inst->ua, 10000, media, dm_mailbox_media_tick, "dm_mailbox_media");
src/dm/dm_media.c
+10 -5
View File
@@ -1231,8 +1231,8 @@ static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[1
}
if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) goto rollback;
dm_custody_notify(inst, id);
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: custody copy removed reason=%s super=%llu",
delivered ? "delivered" : "expired", (unsigned long long)super);
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: custody copy removed reason=%s; all coordinators notified",
delivered ? "delivered" : "expired");
return;
rollback:
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: custody retirement rolled back; retry retained, no ACK");
@@ -1320,6 +1320,8 @@ static void dm_media_offer(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: unauthorized OFFER source=%llu holder=%llu",
(unsigned long long)source, (unsigned long long)holder); return;
}
/* На узле автора оригинал и custody — разные копии; срок custody не ограничивает поиск оригинала. */
if (holder == author) expiry = 0;
uint8_t receipt[DM_RECEIPT_SIZE];
if (dm_accept_message(inst, p, len, receipt)) return;
dm_mailbox_receipt(inst, receipt);
@@ -1362,16 +1364,20 @@ static void dm_source_gone(struct UTUN_INSTANCE* inst, uint64_t source, const ui
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM source: unauthorized removal source=%llu holder=%llu",
(unsigned long long)source, (unsigned long long)holder); return;
}
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body FROM dm_files WHERE id=?");
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body,author FROM dm_files WHERE id=?");
if (!st) return;
int rc = sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
uint8_t hash[32];
if (rc == SQLITE_ROW) SHA256(sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0), hash);
int original = rc == SQLITE_ROW && (uint64_t)sqlite3_column_int64(st, 1) == holder;
int exact = rc == SQLITE_ROW && !memcmp(hash, p + 16, 32), final = sqlite3_finalize(st);
if (!exact || final != SQLITE_OK) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM source: removal for unknown/conflicting file rc=%d final=%d", rc, final); return;
}
if (original) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM source: custody ended on author=%llu; original remains eligible", (unsigned long long)holder); return;
}
if (dm_media_exec(inst, "BEGIN IMMEDIATE")) return;
st = dm_media_prepare(inst, "INSERT INTO dm_file_sources(id,node,expires) VALUES(?,?,-1)"
" ON CONFLICT(id,node) DO UPDATE SET expires=-1");
@@ -1386,8 +1392,7 @@ static void dm_source_gone(struct UTUN_INSTANCE* inst, uint64_t source, const ui
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)holder);
if (rc != SQLITE_OK) { sqlite3_finalize(st); goto rollback; }
if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) goto rollback;
struct dm_file_probe* probe = dm_probe_find(inst->dm_media, p);
if (probe) for (unsigned i = 0; i < probe->count; i++) if (probe->sources[i].node == holder) probe->sources[i].used = 1;
/* HAVE от устройства стороны может относиться к его личной реплике; статус custody её не отменяет. */
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM source: retired holder=%llu; other sources and local file retained", (unsigned long long)holder);
return;
rollback: