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

Commit exact DM custody tombstones before acknowledging deletion

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 19:45:51 +01:00
1 parent 3f195c058b
commit 497792da73
3 files changed
+112 -11
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_media.c
+28 -8
View File
@@ -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");
}
src/dm/dm_media_priv.h
+1 -1
View File
@@ -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);
tests/test_dm_media.c
+83 -2
View File
@@ -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);