diff --git a/tests/Makefile.am b/tests/Makefile.am index fd98ee28..d4b7391a 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -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 diff --git a/tests/test_dm_media.c b/tests/test_dm_media.c index 80993b14..14c57dd4 100644 --- a/tests/test_dm_media.c +++ b/tests/test_dm_media.c @@ -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);