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

dm: enforce monotonic storage decisions and update fault scenarios

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 22:43:20 +01:00
1 parent 515da0304c
commit 27b0166324
2 files changed
+60 -34
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_mailbox_media.c
+7 -3
View File
@@ -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 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]);
@@ -213,7 +215,9 @@ void dm_mailbox_media_recv(struct dm_mb_media* media, uint64_t group, uint64_t s
}
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 && !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);
tests/test_dm_media.c
+53 -31
View File
@@ -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++;
}