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

Coordinate custody using device facts and the storage node policy

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 12:58:57 +01:00
1 parent 1dcd2f1ce6
commit b700763062
1 file changed
+60 -22
Download Patch File Download Diff File Expand all files Collapse all files
src/dm/dm_mailbox_media.c
+60 -22
View File
@@ -7,6 +7,7 @@
#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
#include "dm_media_priv.h"
#include "dm_media_receipt.h"
#include "dm_mailbox_media.h"
#include "../utun_instance.h"
#include "../ntp_time.h"
@@ -46,7 +47,7 @@ int dm_mailbox_media_register(struct dm_mb_media* media, uint64_t group, uint64_
if (!exact) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: conflicting media UUID/body"); return -1; }
if (inserted) DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: registered author=%llu recipient=%llu seq=%llu group=%llu",
(unsigned long long)author, (unsigned long long)recipient, (unsigned long long)seq, (unsigned long long)group);
return 0;
return dm_media_targets_seed(inst, body, len, recipient);
}
/* Список S используется только для пересылки подписанного подтверждения B. */
@@ -90,23 +91,24 @@ static void dm_job_receipt(struct dm_mb_media* media, uint64_t group, uint64_t s
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_hash,receipt FROM dm_media_jobs WHERE id=?");
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);
int rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); return; } /* Нет выданных копий на этом S. */
int exact = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), r + 56, 32);
int already = rc == SQLITE_ROW && sqlite3_column_bytes(st, 1) == DM_MEDIA_RECEIPT_SIZE;
if (already && sqlite3_column_type(st, 0) == SQLITE_NULL) { sqlite3_finalize(st); return; }
int exact = rc == SQLITE_ROW && !dm_media_receipt_save(inst, r, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0));
sqlite3_finalize(st);
if (!exact) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: receipt body hash mismatch rc=%d", rc); return; }
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET receipt=?,retry_at=0 WHERE id=? AND receipt IS NULL");
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET retry_at=0 WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, r, DM_MEDIA_RECEIPT_SIZE, SQLITE_STATIC); sqlite3_bind_blob(st, 2, r + 40, 16, SQLITE_STATIC);
sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) return;
if (!already) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: recipient confirmed file; delete jobs activated source=%llu",
(unsigned long long)source);
dm_media_send(inst, group, author, DM_MEDIA_RECEIPT, r, DM_MEDIA_RECEIPT_SIZE);
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "dm_mailbox_media: device fact stored source=%llu kind=%d",
(unsigned long long)source, dm_media_receipt_kind(r));
if (source != author) dm_media_send(inst, group, author, DM_MEDIA_RECEIPT, r, DM_MEDIA_RECEIPT_SIZE);
}
}
@@ -164,7 +166,7 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
int delivered = sqlite3_column_bytes(st, 2) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 2), sizeof(receipt));
sqlite3_finalize(st);
if (cmd == DM_MEDIA_STORED && len == 24) {
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; }
@@ -173,9 +175,11 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: stale/unauthorized STORED source=%llu expiry=%llu",
(unsigned long long)source, (unsigned long long)expiry); return;
}
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET stored=1,expires=?,retry_at=0 WHERE id=? AND receipt IS NULL AND stored=0");
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET stored=MAX(stored,?3),expires=?1,delete_on_open=?4,retry_at=0"
" WHERE id=?2 AND receipt IS NULL AND (delete_on_open=-1 OR (delete_on_open=?4 AND expires=?1))");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)expiry); sqlite3_bind_blob(st, 2, p, 16, SQLITE_STATIC);
sqlite3_bind_int(st, 3, cmd == DM_MEDIA_STORED); sqlite3_bind_int(st, 4, p[24]);
if (dm_media_step(inst, st)) return;
if (sqlite3_changes(inst->topo_sqlite_db))
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: custody committed M=%llu expires=%llu",
@@ -223,7 +227,7 @@ static uint64_t dm_job_storage(struct dm_mb_media* media, uint64_t group, const
}
/* Удаление неосвобождённых назначений повторяется до DELETE_ACK; истёкшие завершаются по TTL. */
static void dm_job_cleanup(struct dm_mb_media* media, const uint8_t id[16], const uint8_t* receipt) {
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,group_id FROM dm_media_holders WHERE id=? AND released=0 LIMIT 16");
if (!st) return;
@@ -241,8 +245,20 @@ static void dm_job_cleanup(struct dm_mb_media* media, const uint8_t id[16], cons
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup scan failed rc=%d finalize=%d; body retained", rc, final); return;
}
for (unsigned i = 0; i < count; i++) dm_media_send(inst, groups[i], peers[i], DM_MEDIA_DELETE, receipt, DM_MEDIA_RECEIPT_SIZE);
if (!count) {
st = dm_media_prepare(inst, "SELECT received FROM dm_file_targets WHERE id=? AND received IS NOT NULL"
" UNION ALL SELECT taken FROM dm_file_targets WHERE id=? AND taken IS NOT NULL");
if (!st) return;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_blob(st, 2, id, 16, SQLITE_STATIC);
uint8_t facts[2 * DM_TARGET_MAX][DM_MEDIA_RECEIPT_SIZE]; unsigned fact_count = 0;
while ((rc = sqlite3_step(st)) == SQLITE_ROW && fact_count < 2 * DM_TARGET_MAX) {
if (sqlite3_column_bytes(st, 0) != DM_MEDIA_RECEIPT_SIZE) { rc = SQLITE_CORRUPT; break; }
memcpy(facts[fact_count++], sqlite3_column_blob(st, 0), DM_MEDIA_RECEIPT_SIZE);
}
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, groups[i], peers[i], DM_MEDIA_DELETE, facts[j], DM_MEDIA_RECEIPT_SIZE);
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)");
if (!st) return;
@@ -263,7 +279,7 @@ 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,group_id,holder,stored,expires,receipt,expired"
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,author,recipient,body,group_id,holder,stored,expires,receipt,expired,delete_on_open"
" FROM dm_media_jobs WHERE retry_at<=? ORDER BY retry_at,seq LIMIT 1");
if (!st) break;
sqlite3_bind_int64(st, 1, (sqlite3_int64)now);
@@ -279,6 +295,7 @@ static void dm_mailbox_media_tick(void* arg) {
if (len <= sizeof(body) && len) memcpy(body, sqlite3_column_blob(st, 3), len);
uint64_t group = (uint64_t)sqlite3_column_int64(st, 4), holder = (uint64_t)sqlite3_column_int64(st, 5);
int stored = sqlite3_column_int(st, 6), expired = sqlite3_column_int(st, 9);
int policy = sqlite3_column_int(st, 10);
uint64_t expires = (uint64_t)sqlite3_column_int64(st, 7);
int delivered = sqlite3_column_bytes(st, 8) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 8), sizeof(receipt));
@@ -287,13 +304,25 @@ 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;
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)"
",retry_at=0 WHERE id=?" :
"UPDATE dm_media_jobs SET receipt=(SELECT received FROM dm_file_targets WHERE id=? AND side=1 AND received IS NOT NULL LIMIT 1)"
",retry_at=0 WHERE id=?");
if (!st) break;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_blob(st, 2, id, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) break;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM custody: retention complete policy=%d", policy); continue;
}
if (holder) dm_job_cleanup(media, id, delivered);
if (delivered) {
dm_job_cleanup(media, id, receipt);
if (!holder) dm_job_cleanup(media, id, 1);
uint64_t gid = group;
if (dm_route_group(inst, author, group, &gid)) dm_media_send(inst, gid, author, DM_MEDIA_RECEIPT, receipt, sizeof(receipt));
continue;
}
if (holder && stored && expires <= now) {
if (holder && expires && expires <= now) {
if (!dm_job_release(media, id, holder, 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);
@@ -303,16 +332,25 @@ static void dm_mailbox_media_tick(void* arg) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: corrupt saved descriptor"); continue;
}
if (!dm_media_scope(inst, group, inst->node_id, "supernode", author, recipient)) continue;
uint64_t gid = group;
if (dm_route_group(inst, recipient, group, &gid)) {
st = dm_media_prepare(inst, "SELECT node FROM dm_file_targets WHERE id=? AND received IS NULL");
if (!st) break;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
uint64_t targets[DM_TARGET_MAX]; unsigned n = 0;
while ((rc = sqlite3_step(st)) == SQLITE_ROW && n < DM_TARGET_MAX) targets[n++] = (uint64_t)sqlite3_column_int64(st, 0);
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM custody: targets scan failed rc=%d", rc); continue; }
for (unsigned t = 0; t < n; t++) {
uint64_t gid = group;
if (!dm_route_group(inst, targets[t], group, &gid)) continue;
uint8_t offer[24 + DM_MEDIA_BODY_MAX];
uint64_t location = stored && expires > now ? holder : author;
uint64_t expiry = stored ? expires : 0;
memcpy(offer, &recipient, 8); memcpy(offer + 8, &location, 8); memcpy(offer + 16, &expiry, 8);
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, recipient, DM_MEDIA_OFFER, offer, len + 24);
if (expired || (stored && expires <= now)) dm_media_send(inst, gid, recipient, DM_MEDIA_EXPIRED, id, 16);
} else {
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) {
holder = dm_job_storage(media, group, id, author, recipient);
if (!holder) continue;