From a011c57237a66528da632eddd64aa02add5fcc27 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 21:10:44 +0200 Subject: [PATCH] Cover replication retries, source reset and router backpressure --- src/media_delivery/media_delivery.h | 3 +- tests/media_delivery_fixture.h | 1 + tests/test_media_replication.c | 108 ++++++++++++++++++++++++++++ 3 files changed, 110 insertions(+), 2 deletions(-) diff --git a/src/media_delivery/media_delivery.h b/src/media_delivery/media_delivery.h index 26237849..0874ed82 100644 --- a/src/media_delivery/media_delivery.h +++ b/src/media_delivery/media_delivery.h @@ -28,7 +28,6 @@ struct UTUN_INSTANCE; #define MEDIA_REPL_WATCHDOG_TB 2000 // контроль дедлайна/очереди/CM каждые 200ms #define MEDIA_QUERY_TIMEOUT_TB 20000 // таймаут QUERY 2s #define MEDIA_HAVE_BLOCK_TIMEOUT_TB 20000 // таймаут HAVE_BLOCK_ACK 2s -#define MEDIA_RECONNECT_COOLDOWN_TB 360000000 // 1 час (0.1ms) #define MEDIA_DL_STALL_TIMEOUT_TB 20000 // таймаут простоя загрузки 2s (0.1ms) #define MEDIA_DL_MAX_ATTEMPTS 3 // попыток на блок до ошибки @@ -121,7 +120,7 @@ struct media_delivery_ctx { // ── Жизненный цикл и роль узла ── -// Создать таблицы/контекст и подключить router и подписки; 0/-1. Общая БД и группы должны быть готовы. +// Создать таблицы/контекст и подключить router и подписки; 0/-1. Общие БД, media_index и chat_core уже готовы. int media_delivery_init(struct UTUN_INSTANCE* inst); // Отменить загрузки/выдачи/репликацию и снять подписки; workers исходящих писателей уже завершены, группы/БД ещё живы. diff --git a/tests/media_delivery_fixture.h b/tests/media_delivery_fixture.h index 2cd0964c..480917a3 100644 --- a/tests/media_delivery_fixture.h +++ b/tests/media_delivery_fixture.h @@ -49,6 +49,7 @@ int __wrap_etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t struct ll_entry* entry, int force, int mode) { (void)force; (void)mode; int cmd = entry->dgram[1], rc = cmd == fail_send ? -1 : 0; + if (cmd == MEDIA_SUBCMD_SYNC_REQ || cmd == MEDIA_SUBCMD_SYNC_RESP || cmd == MEDIA_SUBCMD_CHANGED) REQUIRE(!force); if (cmd == MEDIA_SUBCMD_BLOCK_REQ) request_count++; if (cmd == MEDIA_SUBCMD_QUERY) query_count++; if (!rc && (cmd == MEDIA_SUBCMD_QUERY || cmd == MEDIA_SUBCMD_QUERY_RESP || capture_blocks)) { diff --git a/tests/test_media_replication.c b/tests/test_media_replication.c index 007f13f7..71576db2 100644 --- a/tests/test_media_replication.c +++ b/tests/test_media_replication.c @@ -286,6 +286,112 @@ static void test_convergence_and_lifecycle(void) { reset(); capture_blocks = 1; } +static struct packet respond_to(struct UTUN_INSTANCE* source, const struct packet* q) { + receive(source, q); poll_ready(); + struct packet answer = {0}; + while (packet_count) { + struct packet p = take_packet(); + if (p.from == source->node_id && p.to == q->from && p.group == q->group && p.data[0] == MEDIA_SUBCMD_SYNC_RESP + && !memcmp(((struct media_pkt_sync_resp*)p.data)->request_id, + ((const struct media_pkt_sync_req*)q->data)->request_id, 16)) answer = p; + } + REQUIRE(answer.len); return answer; +} + +static int reject_commit(void* arg) { (void)arg; return 1; } + +/* Потери запроса/ответа, backoff с тем же UUID, ошибка commit и пересоздание БД источника. */ +static void test_retry_and_epoch(void) { + struct UTUN_INSTANCE* a = node(SERVER), *b = node(CLIENT); + struct media_index_result r = result(2); seed(b, &r, 2, 1, 0); + begin_pull(a, CLIENT, GROUP); + struct packet q = request_from(a, CLIENT); + struct media_super_peer* sp = peer(a, CLIENT, GROUP); + expire(sp); + struct packet retry = request_from(a, CLIENT); + REQUIRE(q.len == retry.len && !memcmp(q.data, retry.data, q.len) && sp->timeout_tb > MEDIA_REPL_TIMEOUT_TB); + struct packet old = respond_to(b, &retry); + /* Ответ потерян, повторный запрос не меняет UUID и допускает новый снимок источника. */ + sql(b->md.db, "UPDATE block_availability SET id=id+100,timestamp=timestamp+1;" + "UPDATE sqlite_sequence SET seq=102 WHERE name='block_availability'"); + expire(sp); retry = request_from(a, CLIENT); + REQUIRE(!memcmp(q.data, retry.data, q.len)); + struct packet newer = respond_to(b, &retry); + sqlite3_commit_hook(a->md.db, reject_commit, NULL); + receive(a, &newer); + REQUIRE(sp->request_pending && !scalar(a->md.db, "SELECT COUNT(*) FROM block_availability") + && !scalar(a->md.db, "SELECT COUNT(*) FROM super_sync") && sqlite3_get_autocommit(a->md.db)); + sqlite3_commit_hook(a->md.db, NULL, NULL); + receive(a, &newer); receive(a, &old); + REQUIRE(!sp->request_pending && scalar(a->md.db, "SELECT last_recv_id FROM super_sync WHERE peer_node_id=512") == 102); + + /* DOWN/UP отменяет RPC; старый ответ не завершает запрос нового подключения. */ + struct fake_handle* h = (struct fake_handle*)sp->connection; REQUIRE(h); + h->cb(sp->connection, CLIENT, GROUP, CONN_EVENT_DOWN, h->arg); + h->cb(sp->connection, CLIENT, GROUP, CONN_EVENT_UP, h->arg); + struct packet current = request_from(a, CLIENT); + REQUIRE(memcmp(((struct media_pkt_sync_req*)q.data)->request_id, + ((struct media_pkt_sync_req*)current.data)->request_id, 16)); + receive(a, &old); REQUIRE(sp->request_pending); + uint8_t epoch[16]; memcpy(epoch, b->md.source_epoch, 16); + media_delivery_destroy(b); sql(b->topo_sqlite_db, "DROP TABLE block_availability"); + REQUIRE(!media_delivery_init(b) && memcmp(epoch, b->md.source_epoch, 16)); + r.block_ids[0] = 0xEE; seed(b, &r, 1, 1, 0); + struct packet fresh = respond_to(b, ¤t); + REQUIRE(read_le(&((struct media_pkt_sync_resp*)fresh.data)->from_id, 8) == 0); + receive(a, &fresh); REQUIRE(!sp->request_pending); + REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM block_availability WHERE substr(block_uuid,1,1)=x'ee'") == 1); + sqlite3_stmt* st = NULL; + REQUIRE(sqlite3_prepare_v2(a->md.db, "SELECT source_epoch FROM super_sync WHERE peer_node_id=512", -1, &st, NULL) == SQLITE_OK); + REQUIRE(sqlite3_step(st) == SQLITE_ROW && !memcmp(sqlite3_column_blob(st, 0), b->md.source_epoch, 16)); + REQUIRE(sqlite3_finalize(st) == SQLITE_OK); + /* Источник восстановлен из более старого снимка с прежней эпохой: отдельный reset только RX-курсора. */ + sql(a->md.db, "UPDATE super_sync SET last_recv_id=999 WHERE peer_node_id=512"); + packet_count = 0; expire(sp); current = request_from(a, CLIENT); + fresh = respond_to(b, ¤t); + REQUIRE(((struct media_pkt_sync_resp*)fresh.data)->status == MEDIA_SYNC_CURSOR_INVALID); + receive(a, &fresh); + REQUIRE(!sp->request_pending && !scalar(a->md.db, "SELECT COUNT(*) FROM super_sync WHERE peer_node_id=512")); + REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM block_availability") == 3); + free_node(a); free_node(b); media_index_result_free(&r); reset(); capture_blocks = 1; +} + +/* Реальные router send_q/waiter: заполнение, синхронное пробуждение и reset очереди. */ +static void test_backpressure(void) { + for (int restart = 0; restart < 2; restart++) { + struct UTUN_INSTANCE* a = node(SERVER); + struct ETCP_ROUTER_CONN* channel = etcp_router_conn_get(a, GROUP, CLIENT, ETCP_RT_ID_MEDIA_DELIVERY); REQUIRE(channel); + queue_set_threshold(channel->send_q, 0, 0); + struct ll_entry* occupied = queue_entry_new(0); REQUIRE(occupied && !queue_data_put(channel->send_q, occupied)); + begin_pull(a, CLIENT, GROUP); + struct media_super_peer* sp = peer(a, CLIENT, GROUP); REQUIRE(sp && sp->request_pending && !packet_count); + for (int i = 1; i <= 100; i++) { + struct packet p = {.from = CLIENT, .to = SERVER, .group = GROUP, .len = MEDIA_SYNC_REQ_SIZE}; + struct media_pkt_sync_req* req = (struct media_pkt_sync_req*)p.data; + req->subcmd = MEDIA_SUBCMD_SYNC_REQ; req->request_id[0] = (uint8_t)i; + receive(a, &p); poll_ready(); + } + REQUIRE(!packet_count && sp->response_pending && sp->response.request_id[0] == 100); + if (restart) { + etcp_router_conn_restart(a, GROUP, CLIENT, ETCP_RT_ID_MEDIA_DELIVERY); + uint64_t until = get_time_tb() + 10000; + while (!packet_count && get_time_tb() < until) uasync_poll(ua, 100); + } else { + occupied = queue_data_get(channel->send_q); REQUIRE(occupied); queue_entry_free(occupied); poll_ready(); + } + int responses = 0, requests = 0; + while (packet_count) { + struct packet p = take_packet(); + if (p.data[0] == MEDIA_SUBCMD_SYNC_REQ) requests++; + if (p.data[0] == MEDIA_SUBCMD_SYNC_RESP) { + REQUIRE(((struct media_pkt_sync_resp*)p.data)->request_id[0] == 100); responses++; + } + } + REQUIRE(responses == 1 && requests == 1); + free_node(a); reset(); capture_blocks = 1; + } +} + int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); srand(17); ua = uasync_create(); REQUIRE(ua); capture_blocks = 1; @@ -293,6 +399,8 @@ int main(void) { test_pages(); test_atomic_and_validation(); test_convergence_and_lifecycle(); + test_retry_and_epoch(); + test_backpressure(); REQUIRE(cm_opened == cm_closed); uasync_destroy(ua, 0); DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "media replication regressions: PASS CM=%d/%d", cm_opened, cm_closed);