From 27b0166324cb55d06c221e01cf5054b4bdff8e35 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 22:43:20 +0100 Subject: [PATCH] dm: enforce monotonic storage decisions and update fault scenarios --- src/dm/dm_mailbox_media.c | 10 +++-- tests/test_dm_media.c | 84 ++++++++++++++++++++++++--------------- 2 files changed, 60 insertions(+), 34 deletions(-) diff --git a/src/dm/dm_mailbox_media.c b/src/dm/dm_mailbox_media.c index df3e49811..5761fde8c 100644 --- a/src/dm/dm_mailbox_media.c +++ b/src/dm/dm_mailbox_media.c @@ -162,7 +162,7 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s 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 FROM dm_media_jobs j" + sqlite3_stmt* st = dm_media_prepare(inst, "SELECT j.receipt,j.holder,h.released,j.delete_on_open 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; sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)source); @@ -175,6 +175,7 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s 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 accepted = holder == source && sqlite3_column_int(st, 3) >= 0; 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)); @@ -191,7 +192,8 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s (unsigned long long)source, (unsigned long long)expiry); 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))"); + " 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 storedtopo_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 && !delivered) { + } 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); diff --git a/tests/test_dm_media.c b/tests/test_dm_media.c index 2848f6f44..718772c5e 100644 --- a/tests/test_dm_media.c +++ b/tests/test_dm_media.c @@ -89,7 +89,7 @@ static int retire_inject(enum retire_fault kind) { /* Выделить два UPDATE перехода, не затрагивая регистрацию/приём сообщений. */ static int release_query(sqlite3* db, const char* sql, int job) { - const char* prefix = job ? "UPDATE dm_media_jobs SET holder=0,stored=0" : "UPDATE dm_media_holders SET released=1"; + const char* prefix = job ? "UPDATE dm_media_jobs SET holder=0,stored=0" : "UPDATE dm_media_holders SET released="; return release_db && db == release_db && sql && !strncmp(sql, prefix, strlen(prefix)); } @@ -361,17 +361,31 @@ static void cancelled(void* arg, int error) { *calls += error == -2 ? 1 : 100; } +static int assignment_id(struct media_test* t, uint8_t out[16]) { + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(t->nodes[S]->topo_sqlite_db, + "SELECT assignment_id FROM dm_media_holders WHERE id=? AND node_id=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, t->id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)t->ids[M]); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + int found = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 16; + if (found) memcpy(out, sqlite3_column_blob(st, 0), 16); + int final = sqlite3_finalize(st); + if (!found || final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM test: assignment unavailable rc=%d final=%d", rc, final); + return found && final == SQLITE_OK ? 0 : -1; +} + #ifdef DM_QUOTA_FAULT_INJECTION /* Снимок точного назначения S; тело/квитанция не должны меняться переходом освобождения. */ -static int release_state(struct media_test* t, sqlite3_int64 state[6]) { +static int release_state(struct media_test* t, sqlite3_int64 state[7]) { sqlite3_stmt* st = NULL; sqlite3* db = t->nodes[S]->topo_sqlite_db; - int rc = sqlite3_prepare_v2(db, "SELECT j.holder,j.stored,j.expired,j.expires,j.retry_at,h.released" + int rc = sqlite3_prepare_v2(db, "SELECT j.holder,j.stored,j.expired,j.expires,j.retry_at,h.released,j.delete_on_open" " FROM dm_media_jobs j JOIN dm_media_holders h ON h.id=j.id WHERE j.id=? AND h.node_id=?", -1, &st, NULL); if (rc == SQLITE_OK) { sqlite3_bind_blob(st, 1, t->id, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)t->ids[M]); rc = sqlite3_step(st); - if (rc == SQLITE_ROW) for (int i = 0; i < 6; i++) state[i] = sqlite3_column_int64(st, i); + if (rc == SQLITE_ROW) for (int i = 0; i < 7; i++) state[i] = sqlite3_column_int64(st, i); } sqlite3_finalize(st); if (rc != SQLITE_ROW) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM release test: state read failed rc=%d", rc); @@ -379,14 +393,14 @@ static int release_state(struct media_test* t, sqlite3_int64 state[6]) { } /* После синхронного component-контроля вернуть живую фикстуру до следующего uasync callback. */ -static int release_restore(struct media_test* t, const sqlite3_int64 state[6]) { +static int release_restore(struct media_test* t, const sqlite3_int64 state[7]) { sqlite3_stmt* st = NULL; sqlite3* db = t->nodes[S]->topo_sqlite_db; int rc = sqlite3_prepare_v2(db, - "UPDATE dm_media_jobs SET holder=?,stored=?,expired=?,expires=?,retry_at=? WHERE id=?", -1, &st, NULL); + "UPDATE dm_media_jobs SET holder=?,stored=?,expired=?,expires=?,retry_at=?,delete_on_open=? WHERE id=?", -1, &st, NULL); if (rc == SQLITE_OK) { for (int i = 0; i < 5; i++) sqlite3_bind_int64(st, i + 1, state[i]); - sqlite3_bind_blob(st, 6, t->id, 16, SQLITE_STATIC); rc = sqlite3_step(st); + sqlite3_bind_int64(st, 6, state[6]); sqlite3_bind_blob(st, 7, t->id, 16, SQLITE_STATIC); rc = sqlite3_step(st); } sqlite3_finalize(st); if (rc != SQLITE_DONE) goto fail; @@ -404,31 +418,35 @@ fail: /* Реальный зарегистрированный job: отказ каждого UPDATE/COMMIT откатывает оба состояния, повтор завершает переход. */ static int release_failures(struct media_test* t) { - sqlite3_int64 before[6], after[6], completed[6]; - if (release_state(t, before)) return -1; - uint8_t packet[25] = {0}; memcpy(packet, t->id, 16); packet[16] = 1; + sqlite3_int64 before[7], after[7], completed[7], original[7]; + if (release_state(t, original)) return -1; + uint8_t packet[41] = {0}; if (assignment_id(t, packet)) return -1; + memcpy(packet + 16, t->id, 16); packet[32] = 1; release_db = t->nodes[S]->topo_sqlite_db; const uint8_t commands[] = { DM_MEDIA_EXPIRED, DM_MEDIA_REJECT }; for (size_t c = 0; c < sizeof(commands); c++) { uint8_t cmd = commands[c]; + memcpy(before, original, sizeof(before)); + if (cmd == DM_MEDIA_REJECT) { before[1] = before[3] = 0; before[6] = -1; } + if (release_restore(t, before)) return -1; for (enum release_fault kind = RELEASE_HOLDER_PREPARE; kind <= RELEASE_COMMIT; kind++) { release_hits = 0; release_fault = kind; - control(t, S, t->ids[M], cmd, packet, cmd == DM_MEDIA_EXPIRED ? 16 : 17); + control(t, S, t->ids[M], cmd, packet, cmd == DM_MEDIA_EXPIRED ? 32 : 33); release_fault = RELEASE_NONE; if (t->failed || release_hits != 1 || release_state(t, after) || memcmp(before, after, sizeof(before)) || !sqlite3_get_autocommit(release_db)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM release test: partial transition cmd=%u fault=%d hits=%u", cmd, kind, release_hits); return -1; } - control(t, S, t->ids[M], cmd, packet, cmd == DM_MEDIA_EXPIRED ? 16 : 17); + control(t, S, t->ids[M], cmd, packet, cmd == DM_MEDIA_EXPIRED ? 32 : 33); if (release_state(t, completed) || completed[0] || completed[1] || completed[2] != (cmd == DM_MEDIA_EXPIRED) || - completed[3] != before[3] || completed[4] || completed[5] != 1 || + completed[3] != before[3] || completed[4] || completed[5] != (cmd == DM_MEDIA_EXPIRED ? 2 : 3) || value(t->nodes[S], "SELECT count(*) FROM dm_media_jobs WHERE receipt IS NOT NULL") || value(t->nodes[S], "SELECT count(*) FROM dm_media_jobs WHERE body IS NULL")) return -1; - control(t, S, t->ids[M], cmd, packet, cmd == DM_MEDIA_EXPIRED ? 16 : 17); - uint64_t expiry = (uint64_t)before[3]; memcpy(packet + 16, &expiry, 8); + control(t, S, t->ids[M], cmd, packet, cmd == DM_MEDIA_EXPIRED ? 32 : 33); + uint64_t expiry = (uint64_t)original[3]; memcpy(packet + 32, &expiry, 8); control(t, S, t->ids[M], DM_MEDIA_STORED, packet, sizeof(packet)); - packet[16] = 1; + packet[32] = 1; if (release_state(t, after) || memcmp(completed, after, sizeof(after)) || release_restore(t, before)) return -1; release_checks++; DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM release test: cmd=%u fault=%d rollback/retry/duplicate/late STORED PASS check=%u", @@ -436,7 +454,7 @@ static int release_failures(struct media_test* t) { } } release_db = NULL; - return 0; + return release_restore(t, original); } /* При ошибке чтения свободное M не выдаёт REJECT/STORED и не меняет durable состояние. */ @@ -485,24 +503,27 @@ static int protocol_checks(struct media_test* t) { if (dm_list_messages_json(a, t->conv, 200, 0, json, sizeof(json), &len) || !strstr(json, "\"media_status\":\"delivered\"") || !strstr(json, "\"media_path\":") || !strstr(json, "\"content_type\":\"audio/opus\"") || !strstr(json, "\"duration_ms\":3000")) return -1; - uint8_t request[8 + sizeof(t->body)]; - memcpy(request, &t->ids[B], 8); memcpy(request + 8, t->body, t->body_len); - uint8_t* metadata = (uint8_t*)dm_message_media(request + 8, t->body_len); + uint8_t request[24 + sizeof(t->body)] = {5}; + memcpy(request + 16, &t->ids[B], 8); memcpy(request + 24, t->body, t->body_len); + uint8_t* metadata = (uint8_t*)dm_message_media(request + 24, t->body_len); metadata[0] ^= 0x80; uint64_t size = 2 * 1024 * 1024; memcpy(metadata + 16, &size, 8); - if (sc_ed25519_sign(a->my_ed25519_privkey, request + 8, t->body_len - DM_MSG_SIG_SIZE, - request + 8 + t->body_len - DM_MSG_SIG_SIZE) || - dm_verify_message(m, t->ids[B], request + 8, t->body_len) || chat_setting_set(m, "dm_media_storage_mb", "1")) return -1; + if (sc_ed25519_sign(a->my_ed25519_privkey, request + 24, t->body_len - DM_MSG_SIG_SIZE, + request + 24 + t->body_len - DM_MSG_SIG_SIZE) || + dm_verify_message(m, t->ids[B], request + 24, t->body_len) || chat_setting_set(m, "dm_media_storage_mb", "1")) return -1; #ifdef DM_QUOTA_FAULT_INJECTION quota_db = m->topo_sqlite_db; quota_rejects = quota_stored = 0; #endif - control(t, M, t->ids[S], DM_MEDIA_STORE, request, 8 + t->body_len); + control(t, M, t->ids[S], DM_MEDIA_STORE, request, 24 + t->body_len); if (value(m, "SELECT count(*) FROM dm_custody") != 0 || chat_setting_set(m, "dm_media_storage_mb", "1024")) return -1; #ifdef DM_QUOTA_FAULT_INJECTION if (quota_rejects != 1 || quota_stored) return -1; + control(t, M, t->ids[S], DM_MEDIA_STORE, request, 24 + t->body_len); + if (quota_rejects != 2 || quota_stored || value(m, "SELECT count(*) FROM dm_custody")) return -1; quota_db = NULL; - if (quota_failures(t, request, 8 + t->body_len)) return -1; + request[0]++; /* Новая попытка после окончательного отказа имеет другой ID. */ + if (quota_failures(t, request, 24 + t->body_len)) return -1; #endif char path[768]; snprintf(path, sizeof(path), "%s/cancelled.download", t->dir); int calls = 0; @@ -583,7 +604,7 @@ static void tick(void* arg) { break; case 4: #ifdef DM_QUOTA_FAULT_INJECTION - if (quota_rejects || value(s, "SELECT count(*) FROM dm_media_holders WHERE released=1")) { + if (quota_rejects || value(s, "SELECT count(*) FROM dm_media_holders WHERE released<>0")) { failure(t, "transient quota read failure permanently rejected M"); return; } #endif @@ -698,7 +719,7 @@ static void tick(void* arg) { #ifdef DM_QUOTA_FAULT_INJECTION if (!expiry_restarted) { if (!expiry_notices || !expiry_blocks) break; - sqlite3_int64 state[6]; + sqlite3_int64 state[7]; if (snapshot(t, 3) || release_state(t, state) || state[0] != (sqlite3_int64)t->ids[M] || state[1] != 1 || state[2] || state[5] || !sqlite3_get_autocommit(release_db)) { failure(t, "timer release failure left partial state"); return; @@ -726,9 +747,9 @@ static void tick(void* arg) { failure(t, "late STORE/STORED extended TTL or forged delivery"); return; } #ifdef DM_QUOTA_FAULT_INJECTION - sqlite3_int64 state[6]; + sqlite3_int64 state[7]; if (release_checks != 18 || !expiry_restarted || !expiry_notices || !expiry_blocks || - release_state(t, state) || state[5] != 1 || !sqlite3_get_autocommit(release_db)) { + release_state(t, state) || state[5] != 2 || !sqlite3_get_autocommit(release_db)) { failure(t, "expiry did not recover after SQL failure and lost control"); return; } DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM expiry test: release checks=%u blocked=%u lost EXPIRED=%u restart=1 recovered PASS", @@ -740,8 +761,9 @@ static void tick(void* arg) { } break; case 9: - if (value(b, "SELECT count(*) FROM dm_files WHERE dir=0 AND seq=3 AND ready=-1 AND receipt IS NULL") == 1) { - DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM recovery test: B received EXPIRED while A offline; returning original author"); + if (value(b, "SELECT count(*) FROM dm_files WHERE dir=0 AND seq=3 AND ready=0 AND receipt IS NULL") == 1 && + value(b, "SELECT count(*) FROM dm_file_sources WHERE expires=-1") == 1) { + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM recovery test: custody source gone while A offline; returning original author"); if (start_node(t, A)) { failure(t, "source return after EXPIRED"); return; } t->phase++; }