From b70076306215f8e94a98838bb363ef0fbca20cc2 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 12:58:57 +0100 Subject: [PATCH] Coordinate custody using device facts and the storage node policy --- src/dm/dm_mailbox_media.c | 82 ++++++++++++++++++++++++++++----------- 1 file changed, 60 insertions(+), 22 deletions(-) diff --git a/src/dm/dm_mailbox_media.c b/src/dm/dm_mailbox_media.c index ca4fdfe74..c00f65f7f 100644 --- a/src/dm/dm_mailbox_media.c +++ b/src/dm/dm_mailbox_media.c @@ -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;