jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
ac37710adcc233a25a21b37deafef915d00b44de
utun2/src/dm/dm_mailbox_media.c
T

498 lines
31 KiB
C
Raw Blame History

/* dm_mailbox_media — S хранит координатор отдельно от opaque custody на M. */
#include <stdio.h>
#include <string.h>
#include <openssl/sha.h>
#include <openssl/rand.h>
#include "../../lib/mem.h"
#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
#include "dm_media_priv.h"
#include "dm_media_receipt.h"
#include "dm_schema.h"
#include "dm_mailbox_media.h"
#include "../utun_instance.h"
#include "../ntp_time.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
struct dm_mb_media {
struct UTUN_INSTANCE* inst;
void* timer;
};
/* Регистрация идёт в транзакции PUT, до выдачи какого-либо задания M. */
int dm_mailbox_media_register(struct dm_mb_media* media, uint64_t group, uint64_t recipient,
const uint8_t* body, size_t len) {
const uint8_t* desc = dm_message_media(body, len);
if (!desc) return 0;
if (!media) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: registration without owner"); return -1; }
struct UTUN_INSTANCE* inst = media->inst;
uint64_t author, seq;
memcpy(&author, body + 24, 8); memcpy(&seq, body + 8, 8);
uint8_t hash[32]; SHA256(body, len, hash);
sqlite3_stmt* st = dm_media_prepare(inst, "INSERT INTO dm_media_jobs(id,author,recipient,seq,body,body_hash)"
" VALUES(?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING");
if (!st) return -1;
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)seq);
sqlite3_bind_blob(st, 5, body, (int)len, SQLITE_STATIC); sqlite3_bind_blob(st, 6, hash, 32, SQLITE_STATIC);
if (dm_media_step(inst, st)) return -1;
int inserted = sqlite3_changes(inst->topo_sqlite_db);
st = dm_media_prepare(inst, "SELECT body_hash FROM dm_media_jobs WHERE id=?");
if (!st) return -1;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
int exact = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), hash, 32);
sqlite3_finalize(st);
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 dm_media_targets_seed(inst, body, len, recipient);
}
/* Список S используется только для пересылки подписанного подтверждения B. */
void dm_mailbox_media_receipt(struct UTUN_INSTANCE* inst, const uint8_t* receipt) {
if (dm_media_verify_receipt(inst, receipt, NULL, 0)) return;
uint64_t author, recipient;
memcpy(&author, receipt + 24, 8); memcpy(&recipient, receipt + 32, 8);
uint64_t* groups = NULL;
int count = 0;
if (topo_node_sqlite_get_member_channels(inst->topo_sqlite_db, inst->node_id, &groups, &count)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: common group lookup failed"); return;
}
for (int i = 0; i < count; i++) {
char sql[192];
if (!topo_group_chat_allowed(inst, groups[i], inst->node_id)) continue;
snprintf(sql, sizeof(sql), "SELECT node_id FROM \"peers_%llu\" WHERE record_state=0 AND source=0 AND deleted=0 AND admitted=1",
(unsigned long long)groups[i]);
sqlite3_stmt* st = dm_media_prepare(inst, sql);
if (!st) continue;
uint64_t peers[64];
unsigned n = 0;
int rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW && n < 64) {
uint64_t node = (uint64_t)sqlite3_column_int64(st, 0);
if (node != author && node != recipient && dm_peer_role(inst, node, "supernode"))
peers[n++] = node;
}
if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: supernode scan failed rc=%d", rc);
sqlite3_finalize(st);
for (unsigned j = 0; j < n; j++) dm_media_send(inst, 0, peers[j], DM_MEDIA_RECEIPT, receipt, DM_MEDIA_RECEIPT_SIZE);
}
u_free(groups);
}
/* Подтверждение медиадоставки сохраняется до начала DELETE и никогда не отменяется STORED. */
static void dm_job_receipt(struct dm_mb_media* media, uint64_t group, uint64_t source, const uint8_t* r) {
struct UTUN_INSTANCE* inst = media->inst;
if (dm_media_receipt_kind(r) < 0) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM custody: invalid receipt domain"); return; }
uint64_t author, recipient;
memcpy(&author, r + 24, 8); memcpy(&recipient, r + 32, 8);
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 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 retry_at=0 WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) return;
if (!already) {
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);
}
}
/* Освобождение назначения и сброс текущего holder — один durable переход. */
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[] = { 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=?4 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 && i == 1 && !expired) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)(ntp_time_get_seconds(inst) + 60));
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: release SQL failed M=%llu rc=%d finalize=%d error=%s statement=%s",
(unsigned long long)holder, rc, final, sqlite3_errmsg(inst->topo_sqlite_db), sql[i]);
goto rollback;
}
changed[i] = sqlite3_changes(inst->topo_sqlite_db);
}
if (dm_media_exec(inst, "COMMIT")) goto rollback;
if (changed[0] || changed[1]) DEBUG_INFO(DEBUG_CATEGORY_DM,
"dm_mailbox_media: storage released M=%llu reason=%s released=%d reset=%d",
(unsigned long long)holder, expired ? "expired" : "rejected", changed[0], changed[1]);
return 0;
rollback:
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: storage release rolled back M=%llu reason=%s; retry retained",
(unsigned long long)holder, expired ? "expired" : "rejected");
dm_media_exec(inst, "ROLLBACK");
return -1;
}
/* Control от M принимается только для durable зарегистрированного назначения. */
void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t source,
uint8_t cmd, const uint8_t* p, size_t len) {
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; }
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,j.delete_on_open,j.expires 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;
int 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) rc = sqlite3_step(st);
uint64_t token[2], file[2]; memcpy(token, assignment, 16); memcpy(file, p, 16);
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM assignment: response assignment=%016llx%016llx file=%016llx%016llx M=%llu group=%llu cmd=%u rc=%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, (unsigned long long)group, cmd, rc);
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, 1);
int released = sqlite3_column_int(st, 2);
int policy = sqlite3_column_int(st, 3), accepted = holder == source && policy >= 0;
uint64_t saved_expiry = (uint64_t)sqlite3_column_int64(st, 4);
uint8_t receipt[DM_MEDIA_RECEIPT_SIZE];
int delivered = sqlite3_column_bytes(st, 0) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 0), sizeof(receipt));
rc = sqlite3_finalize(st);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: response lookup finalize failed rc=%d", rc); return; }
if ((cmd == DM_MEDIA_STORED || cmd == DM_MEDIA_RESERVED) && len == 25 && p[24] <= 1) {
uint64_t expiry;
memcpy(&expiry, p + 16, 8);
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;
}
if (accepted && (saved_expiry != expiry || policy != p[24])) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM assignment: changed reservation M=%llu expires=%llu/%llu policy=%d/%u",
(unsigned long long)source, (unsigned long long)saved_expiry, (unsigned long long)expiry, policy, p[24]); return;
}
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))"
" AND (delete_on_open=-1 OR stored<?3)");
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",
(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, 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);
sqlite3_finalize(st); return;
}
if (!dm_media_step(inst, st) && sqlite3_changes(inst->topo_sqlite_db))
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 && p[16] == 1 && !delivered) {
if (accepted) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM assignment: REJECT after confirmed reservation M=%llu",
(unsigned long long)source); return; }
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_mailbox_media: storage refused M=%llu reason=%u", (unsigned long long)source, p[16]);
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);
}
/* Сначала новый M, затем новая попытка после отказа по квоте. Истёкшие копии не возрождаем. */
static uint64_t dm_job_storage(struct dm_mb_media* media, const uint8_t id[16]) {
struct UTUN_INSTANCE* inst = media->inst;
uint64_t* groups = NULL;
int count = 0;
if (topo_node_sqlite_get_member_channels(inst->topo_sqlite_db, inst->node_id, &groups, &count)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM storage: membership lookup failed"); return 0;
}
uint64_t found = 0;
for (int retry = 0; retry < 2 && !found; retry++) for (int i = 0; i < count && !found; i++) {
if (!topo_group_chat_allowed(inst, groups[i], inst->node_id)) continue;
char sql[384];
snprintf(sql, sizeof(sql), "SELECT p.node_id,p.adm_tags FROM \"peers_%llu\" p"
" LEFT JOIN dm_media_holders h ON h.node_id=p.node_id AND h.id=?1"
" WHERE p.record_state=0 AND p.source=0 AND p.deleted=0 AND p.admitted=1 AND COALESCE(h.released,-1)=?2 ORDER BY p.node_id",
(unsigned long long)groups[i]);
sqlite3_stmt* st = dm_media_prepare(inst, sql);
if (!st) goto done;
int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 2, retry ? 3 : -1);
while (rc == SQLITE_OK && (rc = sqlite3_step(st)) == SQLITE_ROW) {
uint64_t node = (uint64_t)sqlite3_column_int64(st, 0), route = 0;
const char* tags = (const char*)sqlite3_column_text(st, 1);
char role[16]; rc = SQLITE_OK;
if (!tags || json_flat_get(tags, "storage", role, sizeof(role)) || strcmp(role, "yes")) continue;
if (!topo_group_chat_allowed(inst, groups[i], node)) continue;
if (node == inst->node_id || dm_route_group(inst, node, 0, &route)) { found = node; break; }
}
int final = sqlite3_finalize(st);
if ((rc != SQLITE_OK && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM storage: candidate scan failed rc=%d final=%d", rc, final); found = 0; goto done;
}
}
done:
u_free(groups);
return found;
}
/* Удаление неосвобождённых назначений повторяется до 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,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) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup scan bind failed rc=%d; body retained", rc);
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) {
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) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup scan failed rc=%d finalize=%d; body retained", rc, final); return;
}
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++) {
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)");
if (!st) return;
rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: cleanup body bind failed rc=%d; retry retained", rc);
sqlite3_finalize(st); return;
}
if (!dm_media_step(inst, st) && sqlite3_changes(inst->topo_sqlite_db))
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: all storage assignments released; message body removed");
}
}
/* Не получившие файл устройства узнают об удалённой копии; отказ в резерве не означает удаления. */
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;
struct UTUN_INSTANCE* inst = media->inst;
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 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);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); break; }
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != 16) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: job scan failed rc=%d", rc); sqlite3_finalize(st); break;
}
uint8_t id[16], body[DM_MEDIA_BODY_MAX], receipt[DM_MEDIA_RECEIPT_SIZE];
memcpy(id, sqlite3_column_blob(st, 0), 16);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 1), recipient = (uint64_t)sqlite3_column_int64(st, 2);
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);
int delivered = sqlite3_column_bytes(st, 7) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 7), sizeof(receipt));
sqlite3_finalize(st);
st = dm_media_prepare(inst, "UPDATE dm_media_jobs SET retry_at=? WHERE id=?");
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)"
",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) {
if (!holder) dm_job_cleanup(media, id, 1);
dm_media_send(inst, 0, author, DM_MEDIA_RECEIPT, receipt, sizeof(receipt));
continue;
}
if (holder && expires && expires <= now) {
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. */
}
if (!len || len > sizeof(body) || !dm_message_media(body, len)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: corrupt saved descriptor"); continue;
}
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 = 0;
if (!dm_route_group(inst, targets[t], 0, &gid)) continue;
uint8_t offer[24 + DM_MEDIA_BODY_MAX];
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) {
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,assignment_id) VALUES(?,?,?)"
" ON CONFLICT(id,node_id) DO UPDATE SET assignment_id=excluded.assignment_id,released=0 WHERE released=3");
if (!st) { dm_media_exec(inst, "ROLLBACK"); continue; }
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) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: bind failed rc=%d", rc);
sqlite3_finalize(st); dm_media_exec(inst, "ROLLBACK"); continue;
}
if (dm_media_step(inst, st)) { dm_media_exec(inst, "ROLLBACK"); continue; }
if (sqlite3_changes(inst->topo_sqlite_db) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM assignment: cannot replace unresolved assignment M=%llu", (unsigned long long)holder);
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; }
sqlite3_bind_int64(st, 1, (sqlite3_int64)holder); sqlite3_bind_blob(st, 2, id, 16, SQLITE_STATIC);
if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) { dm_media_exec(inst, "ROLLBACK"); continue; }
uint64_t token[2], file[2]; memcpy(token, assignment, 16); memcpy(file, id, 16);
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM assignment: committed assignment=%016llx%016llx file=%016llx%016llx M=%llu",
(unsigned long long)token[0], (unsigned long long)token[1], (unsigned long long)file[0],
(unsigned long long)file[1], (unsigned long long)holder);
}
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");
if (!media->timer) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: retry timer allocation failed");
}
struct dm_mb_media* dm_mailbox_media_create(struct UTUN_INSTANCE* inst) {
if (dm_media_schema_init(inst)) return NULL;
if (dm_media_exec(inst, "UPDATE dm_media_jobs SET retry_at=0")) return NULL;
struct dm_mb_media* media = u_calloc(1, sizeof(*media));
if (!media) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: owner allocation failed"); return NULL; }
media->inst = inst;
media->timer = uasync_set_timeout(inst->ua, 10000, media, dm_mailbox_media_tick, "dm_mailbox_media");
if (!media->timer) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_mailbox_media: timer allocation failed"); u_free(media); return NULL; }
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: initialized persistent coordinator");
return media;
}
void dm_mailbox_media_destroy(struct dm_mb_media* media) {
if (!media) return;
if (media->timer) uasync_cancel_timeout(media->inst->ua, media->timer);
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_mailbox_media: destroyed, pending custody/delete jobs retained");
u_free(media);
}
Reference in new issue View Git Blame Copy Permalink