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

dm: track storage requests separately from shared immutable copies

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 22:33:44 +01:00
1 parent 75ab57798e
commit 1d0459534d
2 files changed
+234 -121
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_media.c
+231 -120
View File
@@ -580,17 +580,77 @@ done:
static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[16]);
/* Политика/expiry фиксируются при резервировании; STORED требует готового ciphertext. */
static void dm_custody_status(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t id[16]) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT expires,delete_on_open,ready FROM dm_custody WHERE id=?");
/* Ответ относится к одному обращению S; свойства физической копии общие для всех обращений. */
static void dm_custody_status(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer,
const uint8_t assignment[16], const uint8_t id[16]) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT r.rejected,c.ready,c.expires,c.delete_on_open,d.id IS NOT NULL,d.receipt"
" FROM dm_custody_requests r LEFT JOIN dm_custody c ON c.id=r.id LEFT JOIN dm_custody_done d ON d.id=r.id"
" WHERE r.supernode=? AND r.assignment_id=? AND r.id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); int rc = sqlite3_step(st);
if (rc != SQLITE_ROW) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: status unavailable rc=%d", rc); sqlite3_finalize(st); return; }
uint64_t expires = (uint64_t)sqlite3_column_int64(st, 0);
uint8_t status[25]; memcpy(status, id, 16); memcpy(status + 16, &expires, 8); status[24] = (uint8_t)sqlite3_column_int(st, 1);
int ready = sqlite3_column_int(st, 2); sqlite3_finalize(st);
if (ready >= 0 && expires > ntp_time_get_seconds(inst))
int rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)peer);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, assignment, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_ROW) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: request status unavailable S=%llu rc=%d",
(unsigned long long)peer, rc); sqlite3_finalize(st); return; }
int rejected = sqlite3_column_int(st, 0), ready = sqlite3_column_type(st, 1) == SQLITE_NULL ? -1 : sqlite3_column_int(st, 1);
uint64_t expires = (uint64_t)sqlite3_column_int64(st, 2);
uint8_t policy = (uint8_t)sqlite3_column_int(st, 3);
int retired = sqlite3_column_int(st, 4), delivered = sqlite3_column_bytes(st, 5) == DM_MEDIA_RECEIPT_SIZE;
rc = sqlite3_finalize(st);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: status finalize failed rc=%d", rc); return; }
uint8_t status[41]; memcpy(status, assignment, 16); memcpy(status + 16, id, 16);
if (rejected) {
status[32] = (uint8_t)rejected; dm_media_send(inst, group, peer, DM_MEDIA_REJECT, status, 33);
} else if (retired) {
/* Все факты нужны S при политике «получили все»; один последний receipt недостаточен. */
if (delivered) {
st = dm_media_prepare(inst, "SELECT received FROM dm_file_targets WHERE id=?1 AND received IS NOT NULL"
" UNION ALL SELECT taken FROM dm_file_targets WHERE id=?1 AND taken IS NOT NULL");
if (!st) return;
rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
uint8_t facts[2 * DM_TARGET_MAX][DM_MEDIA_RECEIPT_SIZE]; unsigned count = 0;
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW && count < 2 * DM_TARGET_MAX) {
if (sqlite3_column_bytes(st, 0) != DM_MEDIA_RECEIPT_SIZE) { rc = SQLITE_CORRUPT; break; }
memcpy(facts[count++], sqlite3_column_blob(st, 0), DM_MEDIA_RECEIPT_SIZE);
}
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: retired facts scan failed rc=%d final=%d", rc, final); return;
}
for (unsigned i = 0; i < count; i++) dm_media_send(inst, group, peer, DM_MEDIA_RECEIPT, facts[i], sizeof(facts[i]));
}
dm_media_send(inst, group, peer, delivered ? DM_MEDIA_DELETE_ACK : DM_MEDIA_EXPIRED, status, 32);
} else if (ready >= 0 && expires > ntp_time_get_seconds(inst)) {
memcpy(status + 32, &expires, 8); status[40] = policy;
dm_media_send(inst, group, peer, ready ? DM_MEDIA_STORED : DM_MEDIA_RESERVED, status, sizeof(status));
}
}
/* Snapshot закрыт до send: локальный S=M может синхронно завершить копию из callback. */
static void dm_custody_notify(struct UTUN_INSTANCE* inst, const uint8_t id[16]) {
struct custody_target { uint64_t peer; uint8_t assignment[16]; } *targets = NULL;
size_t count = 0, capacity = 0;
sqlite3_stmt* st = dm_media_prepare(inst,
"SELECT supernode,assignment_id FROM dm_custody_requests WHERE id=? AND rejected=0");
if (!st) return;
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) != 16) { rc = SQLITE_CORRUPT; break; }
if (count == capacity) {
size_t next = capacity ? capacity * 2 : 8;
struct custody_target* grown = u_realloc(targets, next * sizeof(*targets));
if (!grown) { rc = SQLITE_NOMEM; break; }
targets = grown; capacity = next;
}
targets[count].peer = (uint64_t)sqlite3_column_int64(st, 0);
memcpy(targets[count++].assignment, sqlite3_column_blob(st, 1), 16);
}
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DM,
"DM custody: subscriber snapshot failed rc=%d final=%d count=%zu; S retries retained", rc, final, count);
else for (size_t i = 0; i < count; i++) dm_custody_status(inst, 0, targets[i].peer, targets[i].assignment, id);
u_free(targets);
}
/* Только uasync публикует проверенный файл и изменяет durable состояние. */
@@ -626,7 +686,7 @@ static void dm_file_done(void* arg, int error) {
dm_job_free(job);
return;
}
const char* sql = custody ? "SELECT body,supernode,ready,expires FROM dm_custody WHERE id=?" :
const char* sql = custody ? "SELECT body,author,ready,expires FROM dm_custody WHERE id=?" :
"SELECT body,author,ready,0 FROM dm_files WHERE id=?";
sqlite3_stmt* st = dm_media_prepare(inst, sql);
if (!st) goto fail;
@@ -671,7 +731,7 @@ static void dm_file_done(void* arg, int error) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: recovered expired file conv=%s generation=%llu bytes=%llu",
job->conv, (unsigned long long)job->generation, (unsigned long long)job->size);
if (custody) {
dm_custody_status(inst, 0, peer, job->id);
dm_custody_notify(inst, job->id);
} else {
if (acknowledge) {
dm_media_send(inst, 0, peer, DM_MEDIA_RECEIPT, receipt, sizeof(receipt));
@@ -1020,99 +1080,115 @@ void dm_media_action_trampoline(void* arg) {
/* STORE сначала резервирует место и фиксирует expiry; повтор не продлевает TTL. */
static void dm_custody_store(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source, const uint8_t* p, size_t len) {
if (len < 8) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: short STORE"); return; }
if (len < 24) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: short STORE"); return; }
const uint8_t* assignment = p;
static const uint8_t zero[16];
if (!memcmp(assignment, zero, 16)) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: zero assignment ID"); return; }
p += 16; len -= 16;
uint64_t recipient, author, size;
memcpy(&recipient, p, 8); p += 8; len -= 8;
const uint8_t* desc = dm_message_media(p, len);
if (!desc || dm_verify_message(inst, recipient, p, len)) return;
memcpy(&author, p + 24, 8); memcpy(&size, desc + 16, 8);
uint8_t hash[32]; SHA256(p, len, hash);
sqlite3_stmt* st = dm_media_prepare(inst,
"SELECT id,body_hash FROM dm_custody_requests WHERE supernode=? AND assignment_id=?");
if (!st) return;
int rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)source);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, assignment, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int exists = rc == SQLITE_ROW;
int exact = exists && sqlite3_column_bytes(st, 0) == 16 && !memcmp(sqlite3_column_blob(st, 0), desc, 16) &&
sqlite3_column_bytes(st, 1) == 32 && !memcmp(sqlite3_column_blob(st, 1), hash, 32);
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: assignment lookup failed rc=%d final=%d", rc, final); return;
}
if (exists) {
if (!exact) DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: assignment conflicts with file/body S=%llu", (unsigned long long)source);
else dm_custody_status(inst, group, source, assignment, desc);
return;
}
if (!dm_peer_role(inst, source, "supernode") || !dm_peer_role(inst, inst->node_id, "storage")) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: unauthorized STORE source=%llu group=%llu",
(unsigned long long)source, (unsigned long long)group);
return;
}
uint8_t hash[32]; SHA256(p, len, hash);
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body_hash,receipt FROM dm_custody_done WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
int exact = sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), hash, 32);
uint8_t receipt[DM_MEDIA_RECEIPT_SIZE];
int delivered = sqlite3_column_bytes(st, 1) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 1), sizeof(receipt));
sqlite3_finalize(st);
if (!exact) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: STORE conflicts with retired UUID"); return; }
dm_media_send(inst, group, source, delivered ? DM_MEDIA_RECEIPT : DM_MEDIA_EXPIRED,
delivered ? receipt : desc, delivered ? sizeof(receipt) : 16);
return;
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: tombstone lookup failed rc=%d", rc); return; }
st = dm_media_prepare(inst, "SELECT body,ready,expires FROM dm_custody WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
int exact = sqlite3_column_bytes(st, 0) == (int)len && !memcmp(sqlite3_column_blob(st, 0), p, len);
int ready = sqlite3_column_int(st, 1);
uint64_t expiry = (uint64_t)sqlite3_column_int64(st, 2);
sqlite3_finalize(st);
if (!exact) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: conflicting custody body"); return; }
if (ready >= 0 && expiry > ntp_time_get_seconds(inst)) dm_custody_status(inst, group, source, desc);
return;
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody lookup failed rc=%d", rc); return; }
st = dm_media_prepare(inst, "SELECT COALESCE(sum(cipher_size),0) FROM dm_custody");
if (!st) return;
rc = sqlite3_step(st);
if (rc != SQLITE_ROW) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody quota read failed rc=%d error=%s; STORE retained for retry",
rc, sqlite3_errmsg(inst->topo_sqlite_db));
sqlite3_finalize(st);
return;
}
uint64_t used = (uint64_t)sqlite3_column_int64(st, 0);
rc = sqlite3_finalize(st);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody quota finalize failed rc=%d error=%s; STORE retained for retry",
rc, sqlite3_errmsg(inst->topo_sqlite_db));
return;
}
uint64_t quota = (uint64_t)chat_setting_get_int(inst, "dm_media_storage_mb", 1024) * 1024 * 1024;
uint64_t cipher_size = dm_media_cipher_size(size);
if (used > quota || cipher_size > quota - used) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: custody quota refused used=%llu requested=%llu quota=%llu",
(unsigned long long)used, (unsigned long long)cipher_size, (unsigned long long)quota);
uint8_t reject[17]; memcpy(reject, desc, 16); reject[16] = 1;
dm_media_send(inst, group, source, DM_MEDIA_REJECT, reject, sizeof(reject));
return;
}
uint64_t expires = ntp_time_get_seconds(inst) + (uint64_t)chat_setting_get_int(inst, "dm_media_ttl_sec", 604800);
int policy = chat_setting_get_int(inst, "dm_media_delete_on_open", 1);
if (dm_media_exec(inst, "BEGIN IMMEDIATE")) return;
st = dm_media_prepare(inst, "INSERT INTO dm_custody(id,author,recipient,plain_size,cipher_size,body,body_hash,"
"supernode,expires,delete_on_open) VALUES(?,?,?,?,?,?,?,?,?,?)");
if (!st) { dm_media_exec(inst, "ROLLBACK"); return; }
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
sqlite3_bind_int64(st, 3, (sqlite3_int64)recipient); sqlite3_bind_int64(st, 4, (sqlite3_int64)size);
sqlite3_bind_int64(st, 5, (sqlite3_int64)cipher_size); sqlite3_bind_blob(st, 6, p, (int)len, SQLITE_STATIC);
sqlite3_bind_blob(st, 7, hash, 32, SQLITE_STATIC);
sqlite3_bind_int64(st, 8, (sqlite3_int64)source); sqlite3_bind_int64(st, 9, (sqlite3_int64)expires);
sqlite3_bind_int(st, 10, policy);
if (dm_media_step(inst, st) || dm_media_targets_seed(inst, p, len, recipient) || dm_media_exec(inst, "COMMIT")) {
dm_media_exec(inst, "ROLLBACK"); return;
st = dm_media_prepare(inst, "SELECT body_hash FROM dm_custody WHERE id=?1 UNION ALL SELECT body_hash FROM dm_custody_done WHERE id=?1");
if (!st) goto rollback;
rc = sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
exists = rc == SQLITE_ROW;
exact = !exists || (sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), hash, 32));
if (rc == SQLITE_ROW) rc = sqlite3_step(st);
final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK || !exact) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: copy/tombstone conflict or lookup failure rc=%d final=%d exact=%d", rc, final, exact);
goto rollback;
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: custody reserved author=%llu recipient=%llu bytes=%llu expires=%llu delete_on_open=%d",
(unsigned long long)author, (unsigned long long)recipient, (unsigned long long)cipher_size, (unsigned long long)expires, policy);
dm_custody_status(inst, group, source, desc);
int rejected = 0;
uint64_t cipher_size = dm_media_cipher_size(size);
if (!exists) {
st = dm_media_prepare(inst, "SELECT COALESCE(sum(cipher_size),0) FROM dm_custody");
if (!st) goto rollback;
rc = sqlite3_step(st);
uint64_t used = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 0) : 0;
final = sqlite3_finalize(st);
if (rc != SQLITE_ROW || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: quota read failed rc=%d final=%d; no REJECT", rc, final); goto rollback;
}
uint64_t quota = (uint64_t)chat_setting_get_int(inst, "dm_media_storage_mb", 1024) * 1024 * 1024;
rejected = used > quota || cipher_size > quota - used;
if (rejected) DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: quota refused used=%llu requested=%llu quota=%llu",
(unsigned long long)used, (unsigned long long)cipher_size, (unsigned long long)quota);
}
if (!exists && !rejected) {
uint64_t expires = ntp_time_get_seconds(inst) + (uint64_t)chat_setting_get_int(inst, "dm_media_ttl_sec", 604800);
int policy = chat_setting_get_int(inst, "dm_media_delete_on_open", 1);
st = dm_media_prepare(inst, "INSERT INTO dm_custody(id,author,recipient,plain_size,cipher_size,body,body_hash,"
"expires,delete_on_open) VALUES(?,?,?,?,?,?,?,?,?)");
if (!st) goto rollback;
rc = sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)recipient);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)size);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 5, (sqlite3_int64)cipher_size);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 6, p, (int)len, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 7, hash, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 8, (sqlite3_int64)expires);
if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 9, policy);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: reservation bind failed rc=%d", rc); sqlite3_finalize(st); goto rollback;
}
if (dm_media_step(inst, st) || dm_media_targets_seed(inst, p, len, recipient)) goto rollback;
}
st = dm_media_prepare(inst, "INSERT INTO dm_custody_requests(supernode,assignment_id,id,body_hash,rejected) VALUES(?,?,?,?,?)");
if (!st) goto rollback;
rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)source);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, assignment, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, desc, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 4, hash, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 5, rejected);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: assignment bind failed rc=%d", rc); sqlite3_finalize(st); goto rollback;
}
if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) goto rollback;
uint64_t token[2], file[2]; memcpy(token, assignment, 16); memcpy(file, desc, 16);
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM custody: assignment=%016llx%016llx file=%016llx%016llx S=%llu existing_copy=%d rejected=%d",
(unsigned long long)token[0], (unsigned long long)token[1], (unsigned long long)file[0], (unsigned long long)file[1],
(unsigned long long)source, exists, rejected);
dm_custody_status(inst, group, source, assignment, desc);
return;
rollback:
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: STORE rolled back S=%llu; request can retry, no response", (unsigned long long)source);
dm_media_exec(inst, "ROLLBACK");
}
/* Удаление подтверждается только после durable удаления файла и копии из БД. */
static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[16]) {
struct UTUN_INSTANCE* inst = media->inst;
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body_hash,receipt,supernode,delete_on_open FROM dm_custody WHERE id=? AND ready=-1");
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body_hash,receipt,delete_on_open FROM dm_custody WHERE id=? AND ready=-1");
if (!st) return;
int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc != SQLITE_OK) {
@@ -1128,8 +1204,7 @@ static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[1
memcpy(hash, sqlite3_column_blob(st, 0), sizeof(hash));
int delivered = sqlite3_column_bytes(st, 1) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 1), sizeof(receipt));
uint64_t super = (uint64_t)sqlite3_column_int64(st, 2);
int policy = sqlite3_column_int(st, 3);
int policy = sqlite3_column_int(st, 2);
int final = sqlite3_finalize(st);
if (final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete state finalize failed rc=%d; retry retained", final); return;
@@ -1155,13 +1230,12 @@ static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[1
sqlite3_finalize(st); goto rollback;
}
if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) goto rollback;
dm_media_send(inst, 0, super, delivered ? DM_MEDIA_DELETE_ACK : DM_MEDIA_EXPIRED, id, 16);
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);
return;
rollback:
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: custody retirement rolled back super=%llu; retry retained, no ACK",
(unsigned long long)super);
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: custody retirement rolled back; retry retained, no ACK");
dm_media_exec(inst, "ROLLBACK");
}
@@ -1200,7 +1274,8 @@ static int dm_custody_complete(struct UTUN_INSTANCE* inst, const uint8_t id[16],
}
/* DELETE переносит факт; удаление определяет сохранённая политика M. */
static void dm_custody_delete(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source, const uint8_t* r) {
static void dm_custody_delete(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source,
const uint8_t assignment[16], const uint8_t* r) {
if (dm_media_receipt_kind(r) < 0) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: unknown receipt domain"); return; }
/* Для существующей копии точная подпись B разрешает удаление даже после отзыва роли M/S. */
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body,delete_on_open FROM dm_custody WHERE id=?");
@@ -1223,7 +1298,10 @@ static void dm_custody_delete(struct UTUN_INSTANCE* inst, uint64_t group, uint64
rc = sqlite3_step(st);
int exact = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), r + 56, 32);
sqlite3_finalize(st);
if (exact) dm_media_send(inst, group, source, DM_MEDIA_DELETE_ACK, r + 40, 16);
if (exact) {
uint8_t ack[32]; memcpy(ack, assignment, 16); memcpy(ack + 16, r + 40, 16);
dm_media_send(inst, group, source, DM_MEDIA_DELETE_ACK, ack, sizeof(ack));
}
else if (rc == SQLITE_DONE) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM custody: fact before STORE; S retains it for retry");
else DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: delete conflicts with retired artifact rc=%d", rc);
}
@@ -1260,6 +1338,16 @@ static void dm_media_offer(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)holder);
sqlite3_bind_int64(st, 3, (sqlite3_int64)expiry);
if (dm_media_step(inst, st)) return;
st = dm_media_prepare(inst, "SELECT expires FROM dm_file_sources WHERE id=? AND node=?");
if (!st) return;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)holder);
int rc = sqlite3_step(st);
int current = rc == SQLITE_ROW && sqlite3_column_int64(st, 0) >= 0 && (uint64_t)sqlite3_column_int64(st, 0) == expiry;
int final = sqlite3_finalize(st);
if (!current || final != SQLITE_OK) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM source: stale/conflicting OFFER holder=%llu expiry=%llu rc=%d final=%d",
(unsigned long long)holder, (unsigned long long)expiry, rc, final); return;
}
st = dm_media_prepare(inst, "UPDATE dm_files SET holder=?,expires=?,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);
@@ -1267,6 +1355,46 @@ static void dm_media_offer(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t
dm_media_step(inst, st);
}
/* Исчезновение одной копии не отменяет файл, другого поставщика или готовую локальную историю. */
static void dm_source_gone(struct UTUN_INSTANCE* inst, uint64_t source, const uint8_t* p) {
uint64_t holder; memcpy(&holder, p + 48, 8);
if (!holder || (source != holder && !dm_peer_role(inst, source, "supernode"))) {
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=?");
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 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 (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");
if (!st) goto rollback;
rc = sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC);
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)) goto rollback;
st = dm_media_prepare(inst, "UPDATE dm_files SET holder=author,expires=0,retry_at=0 WHERE id=? AND holder=? AND ready<=0");
if (!st) goto rollback;
rc = sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC);
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;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM source: retired holder=%llu; other sources and local file retained", (unsigned long long)holder);
return;
rollback:
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM source: removal rolled back holder=%llu rc=%d", (unsigned long long)holder, rc);
dm_media_exec(inst, "ROLLBACK");
}
/* Строгие границы control, независимые подписи DM и квитанций. */
static void dm_media_control(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source,
uint8_t cmd, const uint8_t* p, size_t len) {
@@ -1275,30 +1403,16 @@ static void dm_media_control(struct UTUN_INSTANCE* inst, uint64_t group, uint64_
else if ((cmd == DM_MEDIA_HAVE || cmd == DM_MEDIA_HAVE_ACK) && len == 64)
dm_probe_control(inst->dm_media, group, source, cmd == DM_MEDIA_HAVE_ACK, p);
else if (cmd == DM_MEDIA_STORE) dm_custody_store(inst, group, source, p, len);
else if (cmd == DM_MEDIA_DELETE && len == DM_MEDIA_RECEIPT_SIZE) dm_custody_delete(inst, group, source, p);
else if (cmd == DM_MEDIA_DELETE && len == 16 + DM_MEDIA_RECEIPT_SIZE) dm_custody_delete(inst, group, source, p, p + 16);
else if (cmd == DM_MEDIA_SOURCE_GONE && len == 56) dm_source_gone(inst, source, p);
else if (cmd == DM_MEDIA_RECEIPT && len == DM_MEDIA_RECEIPT_SIZE) {
uint64_t author; memcpy(&author, p + 24, 8);
if (author == inst->node_id) {
dm_media_accept_receipt(inst, p);
} else dm_mailbox_media_recv(dm_mailbox_get_media(inst), group, source, cmd, p, len);
} else if (((cmd == DM_MEDIA_STORED || cmd == DM_MEDIA_RESERVED) && len == 25) || (cmd == DM_MEDIA_DELETE_ACK && len == 16) ||
(cmd == DM_MEDIA_EXPIRED && len == 16) || (cmd == DM_MEDIA_REJECT && len == 17)) {
} else if (((cmd == DM_MEDIA_STORED || cmd == DM_MEDIA_RESERVED) && len == 41) || (cmd == DM_MEDIA_DELETE_ACK && len == 32) ||
(cmd == DM_MEDIA_EXPIRED && len == 32) || (cmd == DM_MEDIA_REJECT && len == 33)) {
dm_mailbox_media_recv(dm_mailbox_get_media(inst), group, source, cmd, p, len);
if (cmd == DM_MEDIA_EXPIRED) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT author,recipient,conv_id FROM dm_files WHERE id=? AND dir=0 AND ready<=0");
if (!st) return;
sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint64_t author = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 0) : 0;
uint64_t recipient = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 1) : 0;
uint64_t conv = rc == SQLITE_ROW ? strtoull((const char*)sqlite3_column_text(st, 2), NULL, 10) : 0;
sqlite3_finalize(st);
if (author && dm_peer_role(inst, source, "supernode")) {
st = dm_media_prepare(inst, "UPDATE dm_files SET ready=-1,expires=0 WHERE id=? AND ready<=0");
if (st) { sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC); dm_media_step(inst, st); }
dm_media_notify(inst, conv);
}
}
} else DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: invalid control cmd=%u bytes=%zu", cmd, len);
}
@@ -1404,7 +1518,7 @@ 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,expires,ready,delete_on_open,"
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,body,author,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;
@@ -1418,11 +1532,10 @@ static void dm_custody_tick(struct dm_media_state* media, uint64_t now) {
memcpy(id, sqlite3_column_blob(st, 0), 16);
size_t len = (size_t)sqlite3_column_bytes(st, 1);
if (len <= sizeof(body)) memcpy(body, sqlite3_column_blob(st, 1), len);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 2), super = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t expires = (uint64_t)sqlite3_column_int64(st, 4);
int ready = sqlite3_column_int(st, 5);
int policy = sqlite3_column_int(st, 6);
int local = sqlite3_column_int(st, 7);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 2), expires = (uint64_t)sqlite3_column_int64(st, 3);
int ready = sqlite3_column_int(st, 4);
int policy = sqlite3_column_int(st, 5);
int local = sqlite3_column_int(st, 6);
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=?");
@@ -1437,8 +1550,6 @@ static void dm_custody_tick(struct dm_media_state* media, uint64_t now) {
uint64_t gid = 0;
if (desc && local) dm_file_pull(media, 1, 0, inst->node_id, desc, author);
else if (desc && dm_route_group(inst, author, 0, &gid)) dm_file_pull(media, 1, gid, author, desc, author);
} else {
dm_custody_status(inst, 0, super, id);
}
}
}
src/dm/dm_schema.c
+3 -1
View File
@@ -52,7 +52,7 @@ int dm_media_schema_init(struct UTUN_INSTANCE* inst) {
"expires INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(id,node));"
"CREATE TABLE IF NOT EXISTS dm_custody(id BLOB PRIMARY KEY,author INTEGER NOT NULL,recipient INTEGER NOT NULL,"
"plain_size INTEGER NOT NULL,cipher_size INTEGER NOT NULL,body BLOB NOT NULL,body_hash BLOB NOT NULL,"
"supernode INTEGER NOT NULL,expires INTEGER NOT NULL,ready INTEGER NOT NULL DEFAULT 0,"
"expires INTEGER NOT NULL,ready INTEGER NOT NULL DEFAULT 0,"
"receipt BLOB,retry_at INTEGER NOT NULL DEFAULT 0);"
"CREATE TABLE IF NOT EXISTS dm_custody_done(id BLOB PRIMARY KEY,body_hash BLOB NOT NULL,receipt BLOB);"
"CREATE TABLE IF NOT EXISTS dm_custody_requests(supernode INTEGER NOT NULL,assignment_id BLOB NOT NULL,"
@@ -68,6 +68,8 @@ int dm_media_schema_init(struct UTUN_INSTANCE* inst) {
if (!found && (schema_exec(db, "ALTER TABLE dm_media_holders ADD COLUMN assignment_id BLOB") ||
schema_exec(db, "UPDATE dm_media_holders SET assignment_id=randomblob(16)"))) goto fail;
if (schema_exec(db, "CREATE UNIQUE INDEX IF NOT EXISTS dm_media_assignment_id ON dm_media_holders(assignment_id)")) goto fail;
found = schema_has_column(db, "dm_custody", "supernode");
if (found < 0 || (found && schema_exec(db, "ALTER TABLE dm_custody DROP COLUMN supernode"))) goto fail;
const char* tables[] = {"dm_custody", "dm_custody_done", "dm_media_jobs"};
for (unsigned i = 0; i < 3; i++) {
found = schema_has_column(db, tables[i], "delete_on_open");