Target completion transaction faults without affecting history jobs

src/chat/chat_attachment.c

@@ -173,7 +173,7 @@ static int ak_store_file_result(struct ak_file* job) {
         return -1;
     }
     const char* stage = "begin completion";
-    int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL);
+    int rc = sqlite3_exec(db, "BEGIN IMMEDIATE /* attachment completion */", NULL, NULL, NULL);
     if (rc != SQLITE_OK) goto fail;
     stage = "presence";
     int present = ak_present(job);
@@ -211,7 +211,7 @@ static int ak_store_file_result(struct ak_file* job) {
             stage = "row count"; rc = SQLITE_CONSTRAINT; goto fail;
         }
     }
-    stage = "commit completion"; rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
+    stage = "commit completion"; rc = sqlite3_exec(db, "COMMIT /* attachment completion */", NULL, NULL, NULL);
     if (rc != SQLITE_OK) goto fail;
     DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "attachment keys: completion committed channel=%s encrypt=%d present=%d success=%d",
         job->channel, job->encrypt, present, success);

tests/test_chat_attachment.c

@@ -35,8 +35,8 @@ struct fixture {
     sqlite3* reader;
     char directory[256], source[512], channel[32];
     uint8_t id[16], key[32];
-    unsigned rows, events, submissions, decrypt_submissions, key_events;
-    int success, inspect_submit, decrypt_mode, allow_worker, result_error;
+    unsigned rows, events, submissions, decrypt_submissions, key_events, deleted_events;
+    int success, inspect_submit, decrypt_mode, allow_worker, result_error, delete_mode;
 };
 static struct fixture* active;
 
@@ -63,14 +63,16 @@ void* __wrap_uasync_set_timeout(struct UASYNC* ua, int delay, void* arg, timeout
 
 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 (*cb)(void*, int, char**, char**), void* arg, char** error) {
-    if (!strcmp(sql, "BEGIN IMMEDIATE") && take(BEGIN)) return SQLITE_BUSY;
-    if (!strcmp(sql, "COMMIT") && take(COMMIT)) return SQLITE_BUSY;
-    if (!strcmp(sql, "ROLLBACK") && fail_rollback) {
+    int completion_scope = fault_prefix && strcmp(fault_prefix, "UPDATE attachment_keys SET decrypting=1 WHERE");
+    int targeted = !completion_scope || strstr(sql, "/* attachment completion */");
+    if (targeted && !strncmp(sql, "BEGIN IMMEDIATE", strlen("BEGIN IMMEDIATE")) && take(BEGIN)) return SQLITE_BUSY;
+    if (targeted && !strncmp(sql, "COMMIT", strlen("COMMIT")) && take(COMMIT)) return SQLITE_BUSY;
+    if (!strcmp(sql, "ROLLBACK") && fail_rollback && (!completion_scope || completion_transaction)) {
         fail_rollback = 0; rollback_fault++;
         DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment test: injected rollback failure"); return SQLITE_IOERR;
     }
     int rc = __real_sqlite3_exec(db, sql, cb, arg, error);
-    if (!strcmp(sql, "COMMIT") || !strcmp(sql, "ROLLBACK")) completion_transaction = 0;
+    if (!strncmp(sql, "COMMIT", strlen("COMMIT")) || !strcmp(sql, "ROLLBACK")) completion_transaction = 0;
     return rc;
 }
 
@@ -184,6 +186,7 @@ void __wrap_media_async_submit(struct media_async* ma, struct UASYNC* ua, ma_wor
 }
 
 static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int length) {
+    if (type == CHAT_EVT_CHANNEL_DELETED && active && active->delete_mode) { active->deleted_events++; return; }
     if (type == CHAT_EVT_ATTACHMENT_KEYS_CHANGED && active && active->decrypt_mode) {
         assert(inst == active->inst && sqlite3_get_autocommit(inst->topo_sqlite_db));
         decryption_state(active, !active->result_error, 0, active->result_error); active->key_events++; return;
@@ -514,6 +517,19 @@ static void test_completion_limit(void) {
     DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS completion limit:16 retained results, seventeenth source preserved, retry publishes all exactly once");
 }
 
+static void test_channel_delete_rollback(void) {
+    struct fixture f; setup(&f); source(&f); publish(&f); f.delete_mode = 1;
+    char path[512]; plaintext_path(&f, path);
+    const char* trigger = "CREATE TEMP TRIGGER reject_channel BEFORE DELETE ON channels BEGIN SELECT RAISE(ABORT,'channel'); END";
+    sql(f.inst->topo_sqlite_db, trigger); chat_core_delete_channel(f.inst, f.channel);
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "delete test: failed metadata delete reader_keys=%u expected=%u plaintext_bytes=%lld events=%u",
+        rows(&f), f.rows, (long long)ma_file_size(path), f.deleted_events);
+    assert(!f.deleted_events && rows(&f) == f.rows && ma_file_size(path) == sizeof(content)); committed(&f, 1);
+    sql(f.inst->topo_sqlite_db, "DROP TRIGGER reject_channel"); chat_core_delete_channel(f.inst, f.channel);
+    assert(f.deleted_events == 1 && rows(&f) == f.rows - 1 && ma_file_size(path) < 0); cleanup(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS channel delete: metadata failure retains committed key/plaintext; retry removes channel and derived records together");
+}
+
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_INFO);
     utun_instance_set_tun_init_enabled(0); size_t baseline = u_get_allocated_count();
@@ -522,6 +538,7 @@ int main(void) {
     for (unsigned i = BEGIN; i < FAULT_COUNT; i++) assert(hits[i] == (i == COMMIT ? 2 : 1));
     test_decryption_claim(); test_decryption_completion(); test_completion_controls(); test_completion_stop();
     test_worker_error_completion(); test_completion_limit();
+    test_channel_delete_rollback();
     assert(u_get_allocated_count() == baseline);
     DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment: independent committed key/metadata before worker, allocations restored");
     return 0;