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

Test DM release rollback and expiry recovery after restart

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 17:54:45 +01:00
1 parent cb08faae04
commit 5788d2557b
2 files changed
+181 -3
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/Makefile.am
+1 -1
View File
@@ -894,7 +894,7 @@ test_dm_media_LDFLAGS = $(test_clock_ldflags)
if OS_LINUX
test_dm_media_CFLAGS += -DDM_QUOTA_FAULT_INJECTION
test_dm_media_LDFLAGS += -Wl,--wrap=sqlite3_prepare_v2 -Wl,--wrap=sqlite3_step \
-Wl,--wrap=sqlite3_finalize -Wl,--wrap=etcp_router_conn_send
-Wl,--wrap=sqlite3_bind_blob -Wl,--wrap=sqlite3_finalize -Wl,--wrap=sqlite3_exec -Wl,--wrap=etcp_router_conn_send
endif
test_dm_worker_SOURCES = test_dm_worker.c
tests/test_dm_media.c
+180 -2
View File
@@ -56,6 +56,27 @@ static enum quota_fault quota_fault;
static unsigned quota_hits, quota_attempts, quota_rejects, quota_stored;
static int quota_sequence;
enum release_fault { RELEASE_NONE, RELEASE_HOLDER_PREPARE, RELEASE_HOLDER_BIND, RELEASE_HOLDER_STEP, RELEASE_HOLDER_FINALIZE,
RELEASE_JOB_PREPARE, RELEASE_JOB_BIND, RELEASE_JOB_STEP, RELEASE_JOB_FINALIZE, RELEASE_COMMIT };
static sqlite3* release_db;
static sqlite3* expiry_storage_db;
static enum release_fault release_fault;
static unsigned release_hits, release_checks, expiry_blocks, expiry_notices;
static int expiry_block, expiry_restarted;
/* Выделить два 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";
return release_db && db == release_db && sql && !strncmp(sql, prefix, strlen(prefix));
}
static int release_inject(enum release_fault kind) {
if (release_fault != kind) return 0;
release_fault = RELEASE_NONE; release_hits++;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM release test: one-shot SQL fault=%d hit=%u", kind, release_hits);
return 1;
}
static int quota_query(sqlite3* db, const char* sql) {
return quota_db && db == quota_db && sql && !strcmp(sql, "SELECT COALESCE(sum(cipher_size),0) FROM dm_custody");
}
@@ -71,6 +92,8 @@ static int quota_inject(enum quota_fault kind) {
extern int __real_sqlite3_prepare_v2(sqlite3*, const char*, int, sqlite3_stmt**, const char**);
int __wrap_sqlite3_prepare_v2(sqlite3* db, const char* sql, int len, sqlite3_stmt** st, const char** tail) {
if ((release_query(db, sql, 0) && release_inject(RELEASE_HOLDER_PREPARE)) ||
(release_query(db, sql, 1) && release_inject(RELEASE_JOB_PREPARE))) { *st = NULL; return SQLITE_IOERR; }
if (quota_query(db, sql)) {
quota_attempts++;
if (quota_sequence) quota_fault = quota_attempts <= 3 ? (enum quota_fault)quota_attempts : QUOTA_NONE;
@@ -79,21 +102,57 @@ int __wrap_sqlite3_prepare_v2(sqlite3* db, const char* sql, int len, sqlite3_stm
return __real_sqlite3_prepare_v2(db, sql, len, st, tail);
}
extern int __real_sqlite3_step(sqlite3_stmt*);
extern int __real_sqlite3_bind_blob(sqlite3_stmt*, int, const void*, int, void (*)(void*));
int __wrap_sqlite3_bind_blob(sqlite3_stmt* st, int index, const void* data, int bytes, void (*destroy)(void*)) {
sqlite3* db = sqlite3_db_handle(st);
const char* sql = sqlite3_sql(st);
if ((release_query(db, sql, 0) && release_inject(RELEASE_HOLDER_BIND)) ||
(release_query(db, sql, 1) && release_inject(RELEASE_JOB_BIND))) return SQLITE_NOMEM;
return __real_sqlite3_bind_blob(st, index, data, bytes, destroy);
}
int __wrap_sqlite3_step(sqlite3_stmt* st) {
sqlite3* db = sqlite3_db_handle(st);
const char* sql = sqlite3_sql(st);
if (release_query(db, sql, 0) && release_inject(RELEASE_HOLDER_STEP)) return SQLITE_IOERR;
if (release_query(db, sql, 1)) {
if (release_inject(RELEASE_JOB_STEP)) return SQLITE_IOERR;
if (expiry_block) {
expiry_blocks++;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM expiry test: block job reset attempt=%u", expiry_blocks);
return SQLITE_IOERR;
}
}
if (quota_query(sqlite3_db_handle(st), sqlite3_sql(st))) {
if (quota_inject(QUOTA_STEP)) return SQLITE_IOERR;
if (quota_inject(QUOTA_EMPTY)) return SQLITE_DONE;
}
return __real_sqlite3_step(st);
}
extern int __real_sqlite3_exec(sqlite3*, const char*, int (*)(void*, int, char**, char**), void*, char**);
int __wrap_sqlite3_exec(sqlite3* db, const char* sql, int (*callback)(void*, int, char**, char**), void* arg, char** error) {
if (release_db && db == release_db && !strcmp(sql, "COMMIT") && release_inject(RELEASE_COMMIT)) {
if (error) *error = NULL;
return SQLITE_IOERR;
}
return __real_sqlite3_exec(db, sql, callback, arg, error);
}
extern int __real_sqlite3_finalize(sqlite3_stmt*);
int __wrap_sqlite3_finalize(sqlite3_stmt* st) {
int fail = st && quota_query(sqlite3_db_handle(st), sqlite3_sql(st)) && quota_inject(QUOTA_FINALIZE);
if (st && ((release_query(sqlite3_db_handle(st), sqlite3_sql(st), 0) && release_inject(RELEASE_HOLDER_FINALIZE)) ||
(release_query(sqlite3_db_handle(st), sqlite3_sql(st), 1) && release_inject(RELEASE_JOB_FINALIZE)))) fail = 1;
int rc = __real_sqlite3_finalize(st);
return fail ? SQLITE_IOERR : rc;
}
extern int __real_etcp_router_conn_send(struct ETCP_ROUTER_CONN*, const uint8_t*, size_t, int);
int __wrap_etcp_router_conn_send(struct ETCP_ROUTER_CONN* conn, const uint8_t* data, size_t len, int mode) {
if (expiry_storage_db && conn->inst->topo_sqlite_db == expiry_storage_db &&
conn->svc_id == ETCP_RT_ID_DM_MEDIA && len && data[0] == DM_MEDIA_EXPIRED) {
expiry_notices++;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM expiry test: drop EXPIRED notice=%u; S must reconcile its deadline", expiry_notices);
return 0;
}
if (quota_db && conn->inst->topo_sqlite_db == quota_db && conn->svc_id == ETCP_RT_ID_DM_MEDIA && len) {
if (data[0] == DM_MEDIA_REJECT) quota_rejects++;
if (data[0] == DM_MEDIA_STORED) quota_stored++;
@@ -235,6 +294,83 @@ static void cancelled(void* arg, int error) {
}
#ifdef DM_QUOTA_FAULT_INJECTION
/* Снимок точного назначения S; тело/квитанция не должны меняться переходом освобождения. */
static int release_state(struct media_test* t, sqlite3_int64 state[6]) {
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"
" 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);
}
sqlite3_finalize(st);
if (rc != SQLITE_ROW) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM release test: state read failed rc=%d", rc);
return rc == SQLITE_ROW ? 0 : -1;
}
/* После синхронного component-контроля вернуть живую фикстуру до следующего uasync callback. */
static int release_restore(struct media_test* t, const sqlite3_int64 state[6]) {
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);
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_finalize(st);
if (rc != SQLITE_DONE) goto fail;
rc = sqlite3_prepare_v2(db, "UPDATE dm_media_holders SET released=? WHERE id=? AND node_id=?", -1, &st, NULL);
if (rc == SQLITE_OK) {
sqlite3_bind_int64(st, 1, state[5]); sqlite3_bind_blob(st, 2, t->id, 16, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)t->ids[M]); rc = sqlite3_step(st);
}
sqlite3_finalize(st);
if (rc == SQLITE_DONE) return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM release test: fixture restore failed rc=%d error=%s", rc, sqlite3_errmsg(db));
return -1;
}
/* Реальный зарегистрированный 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[24]; memcpy(packet, t->id, 16); packet[16] = 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];
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);
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);
if (release_state(t, completed) || completed[0] || completed[1] || completed[2] != (cmd == DM_MEDIA_EXPIRED) ||
completed[3] != before[3] || completed[4] || completed[5] != 1 ||
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], DM_MEDIA_STORED, packet, sizeof(packet));
packet[16] = 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",
cmd, kind, release_checks);
}
}
release_db = NULL;
return 0;
}
/* При ошибке чтения свободное M не выдаёт REJECT/STORED и не меняет durable состояние. */
static int quota_failures(struct media_test* t, const uint8_t* request, size_t len) {
struct UTUN_INSTANCE* m = t->nodes[M];
@@ -379,6 +515,9 @@ static void tick(void* arg) {
quota_sequence = 0;
#endif
if (snapshot(t, 2)) { failure(t, "offline body snapshot"); return; }
#ifdef DM_QUOTA_FAULT_INJECTION
if (release_failures(t)) { failure(t, "atomic storage release"); return; }
#endif
utun_instance_destroy(a); t->nodes[A] = NULL;
utun_instance_destroy(m); t->nodes[M] = NULL;
utun_instance_destroy(s); t->nodes[S] = NULL;
@@ -398,6 +537,7 @@ static void tick(void* arg) {
uint8_t stored[24]; memcpy(stored, t->id, 16);
uint64_t expiry = ntp_time_get_seconds(s) + 100; memcpy(stored + 16, &expiry, 8);
control(t, S, t->ids[M], DM_MEDIA_STORED, stored, sizeof(stored));
control(t, S, t->ids[M], DM_MEDIA_EXPIRED, t->id, 16);
uint8_t put[8 + sizeof(t->body)]; memcpy(put, &t->ids[B], 8); memcpy(put + 8, t->body, t->body_len);
control(t, M, t->ids[S], DM_MEDIA_STORE, put, 8 + t->body_len);
if (value(m, "SELECT count(*) FROM dm_custody") != 0 ||
@@ -410,6 +550,10 @@ static void tick(void* arg) {
if (live(a, t->ids[B]) && value(a, "SELECT count(*) FROM dm_files WHERE receipt IS NOT NULL") == 2) {
utun_instance_destroy(b); t->nodes[B] = NULL;
if (chat_setting_set(m, "dm_media_ttl_sec", "1")) { failure(t, "TTL setting"); return; }
#ifdef DM_QUOTA_FAULT_INJECTION
release_db = s->topo_sqlite_db; expiry_storage_db = m->topo_sqlite_db; expiry_block = 1;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM expiry test: lost EXPIRED and SQL failures enabled until M retires custody");
#endif
t->phase++;
}
break;
@@ -421,11 +565,44 @@ static void tick(void* arg) {
break;
case 8:
if (value(m, "SELECT count(*) FROM dm_custody_done WHERE receipt IS NULL") == 1) {
#ifdef DM_QUOTA_FAULT_INJECTION
if (!expiry_restarted) {
if (!expiry_notices || !expiry_blocks) break;
sqlite3_int64 state[6];
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;
}
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM expiry test: M retired; SQL rollback intact blocks=%u lost EXPIRED=%u; restart S",
expiry_blocks, expiry_notices);
expiry_block = 0; release_db = NULL;
utun_instance_destroy(s); t->nodes[S] = NULL;
if (start_node(t, S)) { failure(t, "expiry coordinator restart"); return; }
release_db = t->nodes[S]->topo_sqlite_db; expiry_restarted = 1;
break;
}
#endif
if (value(m, "SELECT count(*) FROM dm_custody") != 0 || value(s, "SELECT count(*) FROM dm_media_jobs WHERE expired=1") != 1) break;
if (snapshot(t, 3)) { failure(t, "expired body snapshot"); return; }
uint8_t stored[24]; memcpy(stored, t->id, 16);
uint64_t expiry = ntp_time_get_seconds(s) + 100; memcpy(stored + 16, &expiry, 8);
control(t, S, t->ids[M], DM_MEDIA_STORED, stored, sizeof(stored));
uint8_t put[8 + sizeof(t->body)]; memcpy(put, &t->ids[B], 8); memcpy(put + 8, t->body, t->body_len);
control(t, M, t->ids[S], DM_MEDIA_STORE, put, 8 + t->body_len);
if (value(m, "SELECT count(*) FROM dm_custody") != 0) { failure(t, "late STORE extended TTL"); return; }
if (value(m, "SELECT count(*) FROM dm_custody") != 0 ||
value(s, "SELECT count(*) FROM dm_media_jobs WHERE receipt IS NULL AND (stored<>0 OR holder<>0 OR expired<>1)") ||
value(s, "SELECT count(*) FROM dm_media_jobs WHERE receipt IS NOT NULL") != 1) {
failure(t, "late STORE/STORED extended TTL or forged delivery"); return;
}
#ifdef DM_QUOTA_FAULT_INJECTION
sqlite3_int64 state[6];
if (release_checks != 18 || !expiry_restarted || !expiry_notices || !expiry_blocks ||
release_state(t, state) || state[5] != 1 || !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",
release_checks, expiry_blocks, expiry_notices);
#endif
t->phase++;
uasync_stop(t->ua); return;
}
@@ -439,6 +616,7 @@ static void tick(void* arg) {
int main(void) {
struct media_test t = {0};
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
size_t allocations = u_get_allocated_count();
test_clock_init(1000); test_clock_sync_wall_time();
if (getenv("UTUN_TEST_DEBUG")) {
@@ -479,7 +657,7 @@ int main(void) {
t.timer = uasync_set_timeout(t.ua, 100, &t, tick, "dm_media_test");
test_clock_mainloop(t.ua);
#ifdef DM_QUOTA_FAULT_INJECTION
quota_db = NULL;
quota_db = NULL; release_db = NULL; expiry_storage_db = NULL;
#endif
for (int i = 0; i < NODES; i++) if (t.nodes[i]) utun_instance_destroy(t.nodes[i]);
uasync_destroy(t.ua, 0);