From 1d0459534de01c46d0c0254deefb3c5e3668c8b0 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 22:33:44 +0100 Subject: [PATCH] dm: track storage requests separately from shared immutable copies --- src/dm/dm_media.c | 351 +++++++++++++++++++++++++++++---------------- src/dm/dm_schema.c | 4 +- 2 files changed, 234 insertions(+), 121 deletions(-) diff --git a/src/dm/dm_media.c b/src/dm/dm_media.c index daa89da44..f18f46d85 100644 --- a/src/dm/dm_media.c +++ b/src/dm/dm_media.c @@ -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); } } } diff --git a/src/dm/dm_schema.c b/src/dm/dm_schema.c index ba4a2b64f..4c1640aa0 100644 --- a/src/dm/dm_schema.c +++ b/src/dm/dm_schema.c @@ -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");