From 497792da739aaebde83943ce8fb2311d5e341663 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 19:45:51 +0100 Subject: [PATCH] Commit exact DM custody tombstones before acknowledging deletion --- src/dm/dm_media.c | 36 ++++++++++++++---- src/dm/dm_media_priv.h | 2 +- tests/test_dm_media.c | 85 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 112 insertions(+), 11 deletions(-) diff --git a/src/dm/dm_media.c b/src/dm/dm_media.c index 44918862..b4138b0b 100644 --- a/src/dm/dm_media.c +++ b/src/dm/dm_media.c @@ -68,8 +68,10 @@ int dm_media_step(struct UTUN_INSTANCE* inst, sqlite3_stmt* st) { int rc = sqlite3_step(st); if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: SQL step rc=%d error=%s statement=%s", rc, sqlite3_errmsg(inst->topo_sqlite_db), sqlite3_sql(st)); - sqlite3_finalize(st); - return rc == SQLITE_DONE ? 0 : -1; + int final = sqlite3_finalize(st); + if (final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: SQL finalize rc=%d step=%d error=%s", + final, rc, sqlite3_errmsg(inst->topo_sqlite_db)); + return rc == SQLITE_DONE && final == SQLITE_OK ? 0 : -1; } int dm_media_exec(struct UTUN_INSTANCE* inst, const char* sql) { @@ -879,8 +881,12 @@ static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[1 struct UTUN_INSTANCE* inst = media->inst; sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body_hash,receipt,supernode,group_id FROM dm_custody WHERE id=? AND ready=-1"); if (!st) return; - sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); - int rc = sqlite3_step(st); + int rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + if (rc != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete state bind failed rc=%d; retry retained", rc); + sqlite3_finalize(st); return; + } + rc = sqlite3_step(st); uint8_t hash[32], receipt[DM_MEDIA_RECEIPT_SIZE]; if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != sizeof(hash)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete state missing/corrupt rc=%d", rc); @@ -890,23 +896,37 @@ static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[1 int delivered = sqlite3_column_bytes(st, 1) == sizeof(receipt); if (delivered) memcpy(receipt, sqlite3_column_blob(st, 1), sizeof(receipt)); uint64_t super = (uint64_t)sqlite3_column_int64(st, 2), group = (uint64_t)sqlite3_column_int64(st, 3); - sqlite3_finalize(st); + int final = sqlite3_finalize(st); + if (final != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete state finalize failed rc=%d; retry retained", final); return; + } if (dm_media_exec(inst, "BEGIN IMMEDIATE")) return; st = dm_media_prepare(inst, "INSERT INTO dm_custody_done(id,body_hash,receipt) VALUES(?,?,?)" " ON CONFLICT(id) DO UPDATE SET receipt=COALESCE(excluded.receipt,receipt)"); if (!st) goto rollback; - sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_blob(st, 2, hash, 32, SQLITE_STATIC); - if (delivered) sqlite3_bind_blob(st, 3, receipt, sizeof(receipt), SQLITE_STATIC); else sqlite3_bind_null(st, 3); + rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, hash, 32, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = delivered ? sqlite3_bind_blob(st, 3, receipt, sizeof(receipt), SQLITE_STATIC) : sqlite3_bind_null(st, 3); + if (rc != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: tombstone bind failed rc=%d; custody retained", rc); + sqlite3_finalize(st); goto rollback; + } if (dm_media_step(inst, st)) goto rollback; st = dm_media_prepare(inst, "DELETE FROM dm_custody WHERE id=? AND ready=-1"); if (!st) goto rollback; - sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + rc = sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); + if (rc != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody delete bind failed rc=%d; tombstone rolled back", rc); + sqlite3_finalize(st); goto rollback; + } if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) goto rollback; dm_media_send(inst, group, super, delivered ? DM_MEDIA_DELETE_ACK : DM_MEDIA_EXPIRED, id, 16); DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: custody copy removed reason=%s super=%llu", delivered ? "delivered" : "expired", (unsigned long long)super); return; rollback: + DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: custody retirement rolled back super=%llu; retry retained, no ACK", + (unsigned long long)super); dm_media_exec(inst, "ROLLBACK"); } diff --git a/src/dm/dm_media_priv.h b/src/dm/dm_media_priv.h index fd629194..a17a08a4 100644 --- a/src/dm/dm_media_priv.h +++ b/src/dm/dm_media_priv.h @@ -12,7 +12,7 @@ * inst/db/sql обязательны; вызывающий связывает параметры и освобождает statement. */ sqlite3_stmt* dm_media_prepare(struct UTUN_INSTANCE* inst, const char* sql); /* Шаг записи, затем finalize (включая ошибку): statement передаётся во владение helper. - * 0 только при SQLITE_DONE, иначе -1 с логом; результат finalize отдельно не проверяется. */ + * 0 только при SQLITE_DONE и успешном finalize, иначе -1 с логом; внутри транзакции rollback делает вызывающий. */ int dm_media_step(struct UTUN_INSTANCE* inst, sqlite3_stmt* st); /* Выполнить SQL без результата: 0/-1 с логом. Владение транзакцией определяется SQL вызывающего. */ int dm_media_exec(struct UTUN_INSTANCE* inst, const char* sql); diff --git a/tests/test_dm_media.c b/tests/test_dm_media.c index 8dc0f7ce..1ac1d7d1 100644 --- a/tests/test_dm_media.c +++ b/tests/test_dm_media.c @@ -66,6 +66,27 @@ static enum release_fault release_fault; static unsigned release_hits, release_checks, expiry_blocks, expiry_notices; static int expiry_block, expiry_restarted; +enum retire_fault { RETIRE_NONE, RETIRE_INSERT_PREPARE, RETIRE_ID_BIND, RETIRE_HASH_BIND, RETIRE_RECEIPT_BIND, + RETIRE_INSERT_STEP, RETIRE_INSERT_FINALIZE, RETIRE_DELETE_PREPARE, RETIRE_DELETE_BIND, + RETIRE_DELETE_STEP, RETIRE_DELETE_FINALIZE, RETIRE_COMMIT }; +static sqlite3* retire_db; +static enum retire_fault retire_fault; +static unsigned retire_hits, retire_checks, retire_acks, retire_restarted; +static int retire_drop_ack; + +static int retire_query(sqlite3* db, const char* sql, int deletion) { + const char* prefix = deletion ? "DELETE FROM dm_custody WHERE id=?" : "INSERT INTO dm_custody_done"; + return retire_db && db == retire_db && sql && !strncmp(sql, prefix, strlen(prefix)); +} + +/* Держим отказ до проверки таймером теста: повторный DELETE не должен успеть скрыть неверный commit. */ +static int retire_inject(enum retire_fault kind) { + if (retire_fault != kind) return 0; + retire_hits++; + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM retirement test: SQL fault=%d hit=%u", kind, retire_hits); + return 1; +} + /* Выделить два 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"; @@ -94,6 +115,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 ((retire_query(db, sql, 0) && retire_inject(RETIRE_INSERT_PREPARE)) || + (retire_query(db, sql, 1) && retire_inject(RETIRE_DELETE_PREPARE))) { *st = NULL; return SQLITE_IOERR; } 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)) { @@ -108,6 +131,9 @@ extern int __real_sqlite3_bind_blob(sqlite3_stmt*, int, const void*, int, 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 ((retire_query(db, sql, 0) && index >= 1 && index <= 3 && + retire_inject((enum retire_fault)(RETIRE_ID_BIND + index - 1))) || + (retire_query(db, sql, 1) && retire_inject(RETIRE_DELETE_BIND))) return SQLITE_NOMEM; 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); @@ -116,6 +142,8 @@ int __wrap_sqlite3_bind_blob(sqlite3_stmt* st, int index, const void* data, int int __wrap_sqlite3_step(sqlite3_stmt* st) { sqlite3* db = sqlite3_db_handle(st); const char* sql = sqlite3_sql(st); + if ((retire_query(db, sql, 0) && retire_inject(RETIRE_INSERT_STEP)) || + (retire_query(db, sql, 1) && retire_inject(RETIRE_DELETE_STEP))) return SQLITE_IOERR; 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; @@ -133,6 +161,10 @@ int __wrap_sqlite3_step(sqlite3_stmt* 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 (retire_db && db == retire_db && !strcmp(sql, "COMMIT") && retire_inject(RETIRE_COMMIT)) { + if (error) *error = NULL; + return SQLITE_IOERR; + } if (release_db && db == release_db && !strcmp(sql, "COMMIT") && release_inject(RELEASE_COMMIT)) { if (error) *error = NULL; return SQLITE_IOERR; @@ -142,6 +174,8 @@ int __wrap_sqlite3_exec(sqlite3* db, const char* sql, int (*callback)(void*, int 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 && ((retire_query(sqlite3_db_handle(st), sqlite3_sql(st), 0) && retire_inject(RETIRE_INSERT_FINALIZE)) || + (retire_query(sqlite3_db_handle(st), sqlite3_sql(st), 1) && retire_inject(RETIRE_DELETE_FINALIZE)))) fail = 1; 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); @@ -149,6 +183,14 @@ int __wrap_sqlite3_finalize(sqlite3_stmt* st) { } 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 (retire_db && conn->inst->topo_sqlite_db == retire_db && conn->svc_id == ETCP_RT_ID_DM_MEDIA && + len && data[0] == DM_MEDIA_DELETE_ACK) { + retire_acks++; + if (retire_drop_ack) { + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM retirement test: drop DELETE_ACK count=%u", retire_acks); + return 0; + } + } 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++; @@ -547,10 +589,49 @@ static void tick(void* arg) { utun_instance_destroy(m); t->nodes[M] = NULL; utun_instance_destroy(s); t->nodes[S] = NULL; if (start_node(t, S) || start_node(t, M) || start_node(t, B)) { failure(t, "custody restart"); return; } +#ifdef DM_QUOTA_FAULT_INJECTION + retire_db = t->nodes[M]->topo_sqlite_db; + retire_fault = RETIRE_INSERT_PREPARE; retire_drop_ack = 1; +#endif t->phase++; } break; case 5: +#ifdef DM_QUOTA_FAULT_INJECTION + if (!retire_restarted) { + if (!retire_hits) break; + if (retire_acks || !file_correct(t) || !sqlite3_get_autocommit(retire_db) || + value(m, "SELECT count(*) FROM dm_custody WHERE ready=-1 AND receipt IS NOT NULL") != 1 || + value(m, "SELECT count(*) FROM dm_custody_done") || + value(s, "SELECT count(*) FROM dm_media_jobs WHERE receipt IS NOT NULL AND body IS NOT NULL") != 1 || + value(s, "SELECT count(*) FROM dm_media_holders WHERE released=0") != 1) { + failure(t, "failed retirement emitted ACK or lost durable custody/tombstone state"); return; + } + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM retirement test: fault=%d rollback intact; no ACK PASS", retire_fault); + retire_checks++; retire_hits = 0; + if (retire_fault < RETIRE_COMMIT) { retire_fault++; break; } + retire_fault = RETIRE_NONE; retire_db = NULL; + utun_instance_destroy(m); t->nodes[M] = NULL; + utun_instance_destroy(s); t->nodes[S] = NULL; + if (start_node(t, S) || start_node(t, M)) { failure(t, "retirement retry restart"); return; } + retire_db = t->nodes[M]->topo_sqlite_db; retire_restarted = 1; + break; + } + if (retire_restarted == 1) { + if (!retire_acks || value(m, "SELECT count(*) FROM dm_custody_done WHERE receipt IS NOT NULL") != 1) break; + if (retire_checks != RETIRE_COMMIT || value(m, "SELECT count(*) FROM dm_custody") || + value(s, "SELECT count(*) FROM dm_media_jobs WHERE receipt IS NOT NULL AND body IS NOT NULL") != 1 || + value(s, "SELECT count(*) FROM dm_media_holders WHERE released=0") != 1) { + failure(t, "lost DELETE_ACK released S before retry"); return; + } + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "DM retirement test: 11 SQL faults, restart M, lost ACK; restarting S for replay"); + retire_drop_ack = 0; + utun_instance_destroy(s); t->nodes[S] = NULL; + if (start_node(t, S)) { failure(t, "lost DELETE_ACK coordinator restart"); return; } + retire_restarted = 2; + break; + } +#endif if (file_correct(t) && value(m, "SELECT count(*) FROM dm_custody") == 0 && value(s, "SELECT count(*) FROM dm_media_jobs WHERE receipt IS NOT NULL AND body IS NULL") == 1) { char json[8192]; size_t length; @@ -678,7 +759,7 @@ static void tick(void* arg) { break; } if (t->failed) return; - if (t->ticks > 9000) { failure(t, "timeout"); return; } + if (t->ticks > 18000) { failure(t, "timeout"); return; } t->timer = uasync_set_timeout(t->ua, 100, t, tick, "dm_media_test"); } @@ -726,7 +807,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; release_db = NULL; expiry_storage_db = NULL; + quota_db = NULL; release_db = NULL; expiry_storage_db = NULL; retire_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);