Exercise real manifest recovery and bounded backfill overflow

tests/test_chat_attachment.c

@@ -726,12 +726,19 @@ static struct local_snapshot local_read(struct fixture* f) {
 
 static void local_equal(struct fixture* f, const struct local_snapshot* expected) {
     struct local_snapshot seen = local_read(f);
-    if (memcmp(&seen, expected, sizeof(seen))) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA,
+    int matches = seen.exists == expected->exists && seen.author == expected->author && seen.key_size == expected->key_size &&
+        seen.info_size == expected->info_size && seen.ready == expected->ready && seen.downloaded == expected->downloaded &&
+        seen.wanted == expected->wanted && seen.decrypting == expected->decrypting && seen.error == expected->error &&
+        !memcmp(seen.id, expected->id, sizeof(seen.id)) && !memcmp(seen.message, expected->message, sizeof(seen.message)) &&
+        !memcmp(seen.hash, expected->hash, sizeof(seen.hash)) && !memcmp(seen.key, expected->key, sizeof(seen.key)) &&
+        !memcmp(seen.info, expected->info, sizeof(seen.info)) && !strcmp(seen.type, expected->type) &&
+        !strcmp(seen.cipher, expected->cipher) && !strcmp(seen.plain, expected->plain);
+    if (!matches) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA,
         "local state test: exists/key/info/ready/downloaded/wanted/decrypting/error=%d/%d/%d/%d/%d/%d/%d/%d expected=%d/%d/%d/%d/%d/%d/%d/%d",
         seen.exists, seen.key_size, seen.info_size, seen.ready, seen.downloaded, seen.wanted, seen.decrypting, seen.error,
         expected->exists, expected->key_size, expected->info_size, expected->ready, expected->downloaded,
         expected->wanted, expected->decrypting, expected->error);
-    assert(!memcmp(&seen, expected, sizeof(seen)));
+    assert(matches);
     sqlite3_stmt* st = NULL;
     assert(sqlite3_prepare_v2(f->reader, "SELECT key,ready FROM attachment_keys WHERE channel='900'", -1, &st, NULL) == SQLITE_OK);
     assert(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && sqlite3_column_int(st, 1) == 0);
@@ -828,7 +835,8 @@ static void test_observation(void) {
     sql(f.inst->topo_sqlite_db, "BEGIN;UPDATE attachment_keys SET ready=8 WHERE channel='900'");
     assert(observe(&f, &owner, manifest) == -1 && !sqlite3_get_autocommit(f.inst->topo_sqlite_db)); local_equal(&f, &owner);
     sql(f.inst->topo_sqlite_db, "ROLLBACK");
-    fault = BEGIN; assert(!chat_attachment_observe(f.inst, f.channel, owner.message, owner.author, "{\"ct\":\"text\",\"d\":\"x\"}", 21));
+    const char* ordinary = "{\"ct\":\"text\",\"d\":\"x\"}";
+    fault = BEGIN; assert(!chat_attachment_observe(f.inst, f.channel, owner.message, owner.author, ordinary, strlen(ordinary)));
     assert(fault == BEGIN); fault = NONE;
     fault_prefix = fault_transaction = NULL; u_free(manifest); cleanup(&f);
     DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS observation:%u SQL faults/retries, new/duplicate records, key/metadata preserved, veto/ABORT/IGNORE/collision/foreign transaction", cases);
@@ -918,6 +926,29 @@ static void test_index_recovery(void) {
     DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS index recovery: failed observation still notifies committed message; detached backfill retries another failure; startup restores cached record without network");
 }
 
+static void test_backfill_capacity(void) {
+    struct fixture f; setup(&f); block_index = 1; unsigned channels = CHAT_MEDIA_BACKFILL_CHANNELS + 1;
+    for (unsigned i = 0; i < channels; i++) {
+        if (i) {
+            char name[64], query[256]; snprintf(name, sizeof(name), "backfill capacity %u", i);
+            chat_core_create_channel_auto(f.inst, name);
+            snprintf(query, sizeof(query), "SELECT channel_id FROM channels WHERE name='%s'", name); sqlite3_stmt* st = NULL;
+            assert(sqlite3_prepare_v2(f.reader, query, -1, &st, NULL) == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW);
+            snprintf(f.channel, sizeof(f.channel), "%s", sqlite3_column_text(st, 0)); assert(sqlite3_finalize(st) == SQLITE_OK);
+        }
+        source(&f); publish(&f); erase_index(&f); f.rows--;
+        chat_media_autodownload_backfill_channel(f.inst, f.channel);
+    }
+    assert(CC(f.inst)->media_channel_backfill_pending == CHAT_MEDIA_BACKFILL_CHANNELS);
+    assert(CC(f.inst)->media_channel_backfill_timer && CC(f.inst)->media_backfill_timer);
+    unsigned received = f.received_events, changed = f.attachment_events;
+    block_index = 0; drive_backfill(&f, 0); drive_backfill(&f, 1);
+    assert(rows(&f) == f.rows + channels && !CC(f.inst)->media_channel_backfill_pending);
+    assert(!CC(f.inst)->media_backfill_timer && !CC(f.inst)->media_channel_backfill_timer);
+    assert(f.received_events == received && f.attachment_events == changed + channels); cleanup(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS backfill capacity:17 channels, bounded16 queue, full-scan fallback restores all without duplicate history/events");
+}
+
 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();
@@ -929,6 +960,7 @@ int main(void) {
     test_channel_delete_rollback();
     test_attachment_binding(); test_attachment_identity();
     test_observation(); test_local_flags(); test_index_recovery();
+    test_backfill_capacity();
     assert(u_get_allocated_count() == baseline);
     DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment: independent committed key/metadata before worker, allocations restored");
     return 0;

tests/test_encrypted_attachment.c

@@ -43,6 +43,7 @@ struct media_test {
     unsigned grant_bind_faults, grant_rollbacks, result_acks; /* Linux: отказ записи GRANT и настоящий повтор до расшифровки. */
     unsigned decision_failures, decision_events; /* Linux: два отказа автора, restart и одно событие после commit. */
     unsigned request_rejections, request_restarts, request_sends, request_acks, request_events;
+    unsigned observation_faults, observation_events, observation_recovered;
     int request_external; /* Linux: транзакция фикстуры вокруг первого настоящего REQUEST от B. */
     int grant_state, grant_outbox; /* Снимок получателя до единственного отказа bind. */
     int failed;
@@ -68,6 +69,25 @@ static timeout_callback_t backfill_cb;
 static etcp_recv_fn attachment_receive;
 static void request_restart(void* arg);
 
+/* История уже committed даже при отказе локального индекса; сообщение не должно пропасть из событий UI. */
+static void observation_event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int bytes) {
+    (void)data; (void)bytes;
+    struct media_test* t = active_test;
+    if (type != CHAT_EVT_MSG_RECEIVED || !t || inst != t->nodes[B]) return;
+    sqlite3* reader = NULL; sqlite3_stmt* st = NULL;
+    int rc = sqlite3_open_v2(sqlite3_db_filename(inst->topo_sqlite_db, "main"), &reader, SQLITE_OPEN_READONLY, NULL);
+    if (rc == SQLITE_OK) rc = sqlite3_prepare_v2(reader, "SELECT (SELECT count(*) FROM msg_" CHANNEL
+        " WHERE json_extract(data,'$.ct')='" ENCRYPTED_ATTACHMENT_TYPE "'),(SELECT count(*) FROM attachment_keys)", -1, &st, NULL);
+    int saved = rc == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW && sqlite3_column_int(st, 0) == 1 && sqlite3_column_int(st, 1) == 0;
+    if (sqlite3_finalize(st) != SQLITE_OK) saved = 0;
+    if (sqlite3_close(reader) != SQLITE_OK) saved = 0;
+    if (!saved || t->observation_faults != 1 || !sqlite3_get_autocommit(inst->topo_sqlite_db) ||
+        CC(inst)->media_channel_backfill_pending != 1 || !CC(inst)->media_channel_backfill_timer)
+        failure(t, "committed message notification lost or index retry missing after SQL failure");
+    t->observation_events++;
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS observation failure: reader history=1 index=0 before UI event, bounded backfill queued");
+}
+
 static int author_request_count(struct media_test* t) {
     char query[256];
     snprintf(query, sizeof(query), "SELECT count(*) FROM attachment_key_requests WHERE channel='" CHANNEL
@@ -182,6 +202,13 @@ static void verify_grant_rollback(void* arg) {
 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*)) {
     struct media_test* t = active_test;
+    const char* observation = "INSERT INTO attachment_keys(channel,crypto_id,message_id,author";
+    if (t && !t->observation_faults && t->nodes[B] && index == 5 && sqlite3_db_handle(st) == t->nodes[B]->topo_sqlite_db &&
+        !strncmp(sqlite3_sql(st), observation, strlen(observation))) {
+        t->observation_faults++;
+        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "encrypted attachment fixture: one observation hash bind NOMEM bytes=%d", bytes);
+        return SQLITE_NOMEM;
+    }
     if (t && !t->grant_bind_faults && t->nodes[B] && index == 1 && sqlite3_db_handle(st) == t->nodes[B]->topo_sqlite_db &&
         strstr(sqlite3_sql(st), "UPDATE attachment_keys SET key=?,error=0 WHERE channel=? AND message_id=? AND author=?")) {
         t->grant_bind_faults++;
@@ -199,7 +226,25 @@ extern int __real_etcp_router_conn_send(struct ETCP_ROUTER_CONN*, const uint8_t*
 int __wrap_etcp_router_conn_send(struct ETCP_ROUTER_CONN* conn, const uint8_t* data, size_t length, int mode) {
     struct media_test* t = active_test;
     if (t && conn->svc_id == ETCP_RT_ID_ATTACHMENT_KEYS && length == 85 && !memcmp(data, "AK01", 4)) {
-        if (conn->inst == t->nodes[B] && conn->remote_node_id == t->ids[A] && data[4] == 1) t->request_sends++;
+        if (conn->inst == t->nodes[B] && conn->remote_node_id == t->ids[A] && data[4] == 1) {
+            t->request_sends++;
+            if (!t->observation_recovered) {
+                sqlite3* reader = NULL; sqlite3_stmt* st = NULL;
+                int rc = sqlite3_open_v2(sqlite3_db_filename(conn->inst->topo_sqlite_db, "main"), &reader, SQLITE_OPEN_READONLY, NULL);
+                if (rc == SQLITE_OK) rc = sqlite3_prepare_v2(reader, "SELECT crypto_id,cipher_hash FROM attachment_keys"
+                    " WHERE channel='" CHANNEL "' AND message_id=? AND key IS NULL AND downloaded=1 AND wanted=1", -1, &st, NULL);
+                if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, data + 5, 32, SQLITE_STATIC);
+                int saved = rc == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 16 &&
+                    sqlite3_column_bytes(st, 1) == 32 && !memcmp(sqlite3_column_blob(st, 0), data + 37, 16) &&
+                    !memcmp(sqlite3_column_blob(st, 1), data + 53, 32);
+                if (sqlite3_finalize(st) != SQLITE_OK) saved = 0;
+                if (sqlite3_close(reader) != SQLITE_OK) saved = 0;
+                if (!saved || t->observation_faults != 1 || t->observation_events != 1 || !sqlite3_get_autocommit(conn->inst->topo_sqlite_db))
+                    failure(t, "REQUEST without recovered committed manifest index after SQL failure");
+                t->observation_recovered++;
+                DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS observation recovery: independent reader identity/hash/downloaded/wanted committed before real REQUEST");
+            }
+        }
         if (conn->inst == t->nodes[A] && conn->remote_node_id == t->ids[B] && data[4] == 2) {
             sqlite3* reader = NULL; sqlite3_stmt* st = NULL;
             int rc = sqlite3_open_v2(sqlite3_db_filename(conn->inst->topo_sqlite_db, "main"), &reader, SQLITE_OPEN_READONLY, NULL);
@@ -331,7 +376,11 @@ static int start_node(struct media_test* t, int index) {
         DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "encrypted attachment fixture: node=%d port=%u retained for restart", index, t->ports[index]);
     }
     if (t->nodes[index] && index != S) test_etcp_set_peer(t->nodes[index], "to_s", t->nodes[S], "s1");
-    return t->nodes[index] ? start_with_recovery(t->nodes[index], db) : -1;
+    int rc = t->nodes[index] ? start_with_recovery(t->nodes[index], db) : -1;
+#ifdef __linux__
+    if (!rc && index == B) chat_event_set_handler(t->nodes[index], observation_event);
+#endif
+    return rc;
 }
 
 #ifdef __linux__
@@ -623,6 +672,8 @@ int main(void) {
         failure(&t, "author decision failures, restart and committed event were not exercised");
     if (!t.failed && (t.request_rejections != 1 || t.request_restarts != 1 || t.request_sends < 2 || !t.request_acks ||
         t.request_events || t.request_external)) failure(&t, "REQUEST rejection, restart, real retry and committed ACK were not exercised");
+    if (!t.failed && (t.observation_faults != 1 || t.observation_events != 1 || t.observation_recovered != 1))
+        failure(&t, "observation SQL failure, committed message notification and real index recovery were not exercised");
 #endif
     for (int i = 0; i < NODES; i++) if (t.nodes[i]) utun_instance_destroy(t.nodes[i]);
     uasync_destroy(t.ua, 0);