Keep active block and worker temporaries during media cache cleanup

src/chat/chat_msg.c

@@ -1642,12 +1642,13 @@ static void orphan_walk_cb(void* arg, const char* full_path, int64_t size) {
     char* found = bsearch(&rel, ow->rels, (size_t)ow->rel_count, sizeof(char*), cache_str_cmp);
     if (found) return;
 
-    if (remove(full_path) == 0) {
+    int removed = file_io_remove_orphan(full_path);
+    if (removed > 0) {
         ow->removed++;
         ow->removed_bytes += (uint64_t)(size > 0 ? size : 0);
         DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cache orphan removed: %s (%lld bytes)",
                    CC_ID, rel, (long long)size);
-    } else {
+    } else if (removed < 0) {
         DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cache orphan remove failed: %s", CC_ID, full_path);
     }
 }

tests/test_encrypted_attachment.c

@@ -78,7 +78,10 @@ void __wrap_media_download_handle_chunk(struct UTUN_INSTANCE* inst, uint64_t gro
     int length = snprintf(orphan, sizeof(orphan), "%s.unowned", temporary);
     if (length < 0 || length >= (int)sizeof(orphan)) { failure(t, "orphan fixture path"); return; }
     FILE* file = fopen(orphan, "wb");
-    if (!file || fputs("stale", file) < 0 || fclose(file)) { failure(t, "orphan fixture creation"); return; }
+    if (!file) { failure(t, "orphan fixture creation"); return; }
+    int error = fputs("stale", file) < 0;
+    if (fclose(file)) error = 1;
+    if (error) { failure(t, "orphan fixture write/close"); return; }
     uint32_t prefix = block->bytes_received;
     t->forced_cleanup++;
     struct chat_core_ctx* cc = CC(inst);

tests/test_media_primitives.c

@@ -24,6 +24,7 @@ static enum fault fault;
 static int injected, read_failed, rng_after = -1, allocation_after = -1;
 static FILE *input_file, *output_file;
 static pthread_barrier_t *post_barrier;
+static pthread_barrier_t *temp_barrier;
 extern FILE *__real_fopen(const char*, const char*);
 extern FILE *__real_fdopen(int, const char*);
 extern size_t __real_fread(void*, size_t, size_t, FILE*);
@@ -54,6 +55,8 @@ FILE *__wrap_fdopen(int fd, const char *mode) {
     if (inject(FDOPEN)) return NULL;
     FILE *file = __real_fdopen(fd, mode);
     if (!strcmp(mode, "wb+")) output_file = file;
+    pthread_barrier_t *barrier = __atomic_load_n(&temp_barrier, __ATOMIC_ACQUIRE);
+    if (barrier) { pthread_barrier_wait(barrier); pthread_barrier_wait(barrier); }
     return file;
 }
 size_t __wrap_fread(void *data, size_t size, size_t count, FILE *file) {
@@ -162,6 +165,39 @@ static void copy_tests(void) {
 }
 
 /* Ошибки чтения/close и неверные аргументы не оставляют правдоподобный hash/signature. */
+static void temporary_tests(void) {
+    struct file_io_temp outputs[3] = {0};
+    char paths[3][FILE_IO_PATH_MAX], stale[FILE_IO_PATH_MAX];
+    for (unsigned i = 0; i < 3; i++) {
+        assert(!file_io_temp_open(&outputs[i], destination));
+        snprintf(paths[i], sizeof(paths[i]), "%s", outputs[i].temporary);
+        assert(file_io_remove_orphan(paths[i]) == 0);
+        assert(fputs("owned", outputs[i].file) >= 0 && !file_io_temp_finish(&outputs[i]));
+        assert(file_io_remove_orphan(paths[i]) == 0); check_text(paths[i], "owned");
+    }
+    snprintf(stale, sizeof(stale), "%s.tmp.unowned", destination); write_text(stale, "stale");
+    assert(file_io_remove_orphan(stale) == 1 && file_io_remove_orphan(stale) == 0);
+    file_io_temp_discard(&outputs[1]); file_io_temp_discard(&outputs[1]);
+    assert(ma_file_size(paths[1]) == -1);
+    write_text(paths[1], "unowned"); assert(file_io_remove_orphan(paths[1]) == 1);
+    assert(!file_io_temp_publish(&outputs[0], source, destination)); check_text(destination, "owned");
+    write_text(paths[0], "unowned"); assert(file_io_remove_orphan(paths[0]) == 1);
+    assert(file_io_remove_orphan(paths[2]) == 0);
+    file_io_temp_discard(&outputs[2]);
+#ifdef __linux__
+    struct file_io_temp failed = {0};
+    fault = FDOPEN; injected = 0;
+    assert(file_io_temp_open(&failed, destination) == -1 && injected); fault = NONE;
+    assert(failed.temporary[0] && file_io_remove_orphan(failed.temporary) == 0);
+    file_io_temp_discard(&failed);
+    assert(!file_io_temp_open(&failed, destination) && !file_io_temp_finish(&failed));
+    fault = PUBLISH; injected = 0;
+    assert(file_io_temp_publish(&failed, source, destination) == -1 && injected); fault = NONE;
+    assert(file_io_remove_orphan(failed.temporary) == 0); file_io_temp_discard(&failed);
+#endif
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS active temp ownership: open/finished, multiple generations, publish/discard and failures");
+}
+
 static void crypto_tests(void) {
     uint8_t output[64], zero[64] = {0};
     memset(output, 0xa5, sizeof(output));
@@ -252,6 +288,38 @@ static void registration_tests(struct UASYNC *ua) {
 }
 
 #ifdef __linux__
+struct temp_job { struct file_io_temp output; pthread_barrier_t barrier; int publish; };
+static void* temp_worker(void* arg) {
+    struct temp_job* job = arg;
+    assert(!file_io_temp_open(&job->output, destination));
+    assert(fputs("worker", job->output.file) >= 0 && !file_io_temp_finish(&job->output));
+    pthread_barrier_wait(&job->barrier); pthread_barrier_wait(&job->barrier);
+    if (job->publish) assert(!file_io_temp_publish(&job->output, source, destination));
+    else file_io_temp_discard(&job->output);
+    return NULL;
+}
+static void temporary_thread_tests(void) {
+    for (int publish = 0; publish < 2; publish++) {
+        struct temp_job job = {.publish = publish}; pthread_t thread;
+        assert(!pthread_barrier_init(&job.barrier, NULL, 2));
+        __atomic_store_n(&temp_barrier, &job.barrier, __ATOMIC_RELEASE);
+        assert(!pthread_create(&thread, NULL, temp_worker, &job));
+        pthread_barrier_wait(&job.barrier); /* fdopen ещё не вернул управление temp_open. */
+        char path[FILE_IO_PATH_MAX]; snprintf(path, sizeof(path), "%s", job.output.temporary);
+        assert(ma_file_size(path) == 0 && file_io_remove_orphan(path) == 0);
+        pthread_barrier_wait(&job.barrier);
+        pthread_barrier_wait(&job.barrier); /* Worker закрыл FILE, но сохраняет владение. */
+        assert(file_io_remove_orphan(path) == 0); check_text(path, "worker");
+        __atomic_store_n(&temp_barrier, NULL, __ATOMIC_RELEASE);
+        pthread_barrier_wait(&job.barrier);
+        assert(!pthread_join(thread, NULL) && !pthread_barrier_destroy(&job.barrier));
+        assert(ma_file_size(path) == -1 && file_io_remove_orphan(path) == 0);
+        write_text(path, "unowned"); assert(file_io_remove_orphan(path) == 1);
+        if (publish) check_text(destination, "worker");
+    }
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS worker temp protected during open and after finish; publish/discard release ownership");
+}
+
 struct job {
     struct media_async *owner; struct UASYNC *ua;
     unsigned work, done; int error, restart, cancel_actions;
@@ -316,11 +384,11 @@ int main(void) {
 #endif
     assert(!utun_mkdir(root, 0700)); snprintf(source, sizeof(source), "%s/source", root);
     snprintf(destination, sizeof(destination), "%s/destination", root);
-    copy_tests(); crypto_tests(); size_and_walk_tests();
+    copy_tests(); temporary_tests(); crypto_tests(); size_and_walk_tests();
     struct UASYNC *ua = uasync_create(); assert(ua);
     registration_tests(ua);
 #ifdef __linux__
-    engine_tests(ua);
+    temporary_thread_tests(); engine_tests(ua);
 #endif
     uasync_destroy(ua, 0); assert(!ma_dir_walk(root, remove_file, NULL) && !rmdir(root));
     assert(u_get_allocated_count() == before);