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

Cover replication retries, source reset and router backpressure

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 21:10:44 +02:00
1 parent 9e2a1e5208
commit a011c57237
3 files changed
+110 -2
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.h
+1 -2
View File
@@ -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 исходящих писателей уже завершены, группы/БД ещё живы.
tests/media_delivery_fixture.h
+1
View File
@@ -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)) {
tests/test_media_replication.c
+108
View File
@@ -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, &current);
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, &current);
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);