/* Репликация: настоящие обработчики media/SQLite, управляемый CM/транспорт и ошибки. */
#include "media_delivery_fixture.h"
#include "../src/routing_layer/topo_group.h"
#include "../src/routing_layer/conn_mgr.h"

static int cm_opened, cm_closed;
struct fake_handle {
    struct UTUN_INSTANCE* inst;
    uint64_t node, group;
    conn_mgr_cb_t cb;
    void* arg;
    void* soon;
};

static void cm_up(void* arg) {
    struct fake_handle* h = arg; h->soon = NULL;
    h->cb((struct CONN_MGR_HANDLE*)h, h->node, h->group, CONN_EVENT_UP, h->arg);
}

int __wrap_conn_mgr_open(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t node,
                         conn_mgr_cb_t cb, void* arg, struct CONN_MGR_HANDLE** out) {
    struct fake_handle* h = u_calloc(1, sizeof(*h)); REQUIRE(h);
    h->inst = inst; h->group = group; h->node = node; h->cb = cb; h->arg = arg;
    *out = (struct CONN_MGR_HANDLE*)h; cm_opened++;
    h->soon = uasync_call_soon(ua, h, cm_up); REQUIRE(h->soon); return 0;
}

void __wrap_conn_mgr_close(struct CONN_MGR_HANDLE* handle) {
    struct fake_handle* h = (struct fake_handle*)handle; REQUIRE(h);
    if (h->soon) uasync_call_soon_cancel(ua, h->soon);
    cm_closed++; u_free(h);
}

static void sql(sqlite3* db, const char* command) {
    int rc = sqlite3_exec(db, command, NULL, NULL, NULL);
    if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "test SQL=%s rc=%d error=%s", command, rc, sqlite3_errmsg(db));
    REQUIRE(rc == SQLITE_OK);
}

static sqlite3_int64 scalar(sqlite3* db, const char* command) {
    sqlite3_stmt* st = NULL; REQUIRE(sqlite3_prepare_v2(db, command, -1, &st, NULL) == SQLITE_OK);
    REQUIRE(sqlite3_step(st) == SQLITE_ROW);
    sqlite3_int64 value = sqlite3_column_int64(st, 0);
    REQUIRE(sqlite3_finalize(st) == SQLITE_OK); return value;
}

static uint64_t read_le(const void* ptr, unsigned n) {
    const uint8_t* p = ptr; uint64_t v = 0;
    for (unsigned i = 0; i < n; i++) v |= (uint64_t)p[i] << (i * 8);
    return v;
}

static void write_le(void* ptr, uint64_t v, unsigned n) {
    uint8_t* p = ptr;
    for (unsigned i = 0; i < n; i++) p[i] = (uint8_t)(v >> (i * 8));
}

/* Только транспорт управляемый: каталог, роли, router-очереди и обработчики — настоящие.
 * Таблица peers отражает результат уже проверенного member_sync; криптография членства проверяется его тестами. */
static void add_group(struct UTUN_INSTANCE* inst, uint64_t group) {
    if (!inst->topo_groups) {
        inst->topo_groups = u_calloc(1, sizeof(*inst->topo_groups)); REQUIRE(inst->topo_groups);
        inst->topo_groups->instance = inst;
        inst->topo_groups->group_list = queue_new(ua, 16, 0, 8, "test_groups"); REQUIRE(inst->topo_groups->group_list);
    }
    struct ll_entry* e = queue_entry_new(sizeof(struct TOPO_GROUP) - sizeof(struct ll_entry)); REQUIRE(e);
    struct TOPO_GROUP* g = (struct TOPO_GROUP*)e;
    g->instance = inst; g->group_id = group; g->group_type = TOPO_GROUP_TYPE_CHAT;
    snprintf(g->channel_id, sizeof(g->channel_id), "%llu", (unsigned long long)group);
    queue_data_put_with_index(inst->topo_groups->group_list, e);
    char command[512];
    snprintf(command, sizeof(command), "CREATE TABLE peers_%llu(node_id INTEGER PRIMARY KEY,adm_tags TEXT);"
        "INSERT INTO peers_%llu VALUES(256,'{\"supernode\":\"yes\"}'),(512,'{\"supernode\":\"yes\"}'),(768,'{\"supernode\":\"yes\"}')",
        (unsigned long long)group, (unsigned long long)group);
    sql(inst->md.db, command);
}

static struct UTUN_INSTANCE* node(uint64_t id) {
    struct UTUN_INSTANCE* inst = instance(id);
    add_group(inst, GROUP); return inst;
}

static void free_node(struct UTUN_INSTANCE* inst) {
    struct TOPO_GROUPS* groups = inst->topo_groups;
    destroy(inst);
    if (groups) { queue_free(groups->group_list); u_free(groups); }
}

static struct media_super_peer* peer(struct UTUN_INSTANCE* inst, uint64_t remote, uint64_t group) {
    uint64_t key[2] = {remote, group};
    struct ll_entry* e = inst->md.super_peers ? queue_find_data_by_index(inst->md.super_peers, key) : NULL;
    return e ? (struct media_super_peer*)e->data : NULL;
}

static void poll_ready(void) { for (int i = 0; i < 8; i++) uasync_poll(ua, 0); }

static void expire(struct media_super_peer* p) {
    REQUIRE(p); p->deadline_tb = 0;
    uint64_t until = get_time_tb() + 10000;
    while (get_time_tb() < until) {
        uasync_poll(ua, 100);
        for (int i = 0; i < packet_count; i++) if (packets[i].data[0] == MEDIA_SUBCMD_SYNC_REQ
            && packets[i].from == p->md->self_node_id && packets[i].to == p->peer_node_id && packets[i].group == p->group_id) return;
    }
    REQUIRE(0);
}

/* Вручную запросить настоящую страницу без подмены реализации SELECT/сериализации. */
static struct packet page(struct UTUN_INSTANCE* source, uint64_t receiver, uint64_t group, uint64_t after,
                           const uint8_t epoch[16], uint8_t token) {
    struct packet p = {.from = receiver, .to = source->node_id, .group = group, .len = MEDIA_SYNC_REQ_SIZE};
    struct media_pkt_sync_req* q = (struct media_pkt_sync_req*)p.data;
    q->subcmd = MEDIA_SUBCMD_SYNC_REQ; memset(q->request_id, token, 16);
    if (epoch) memcpy(q->source_epoch, epoch, 16);
    write_le(&q->after_id, after, 8);
    receive(source, &p); poll_ready();
    struct packet response = {0};
    while (packet_count) {
        struct packet out = take_packet();
        if (out.data[0] == MEDIA_SUBCMD_SYNC_RESP && out.to == receiver && out.group == group
            && ((struct media_pkt_sync_resp*)out.data)->request_id[0] == token) response = out;
    }
    REQUIRE(response.len); return response;
}

static void synchronize(struct UTUN_INSTANCE** nodes, int count, int noisy) {
    int delivered = 0, idle = 0;
    while (idle < 8) {
        uasync_poll(ua, 0);
        if (!packet_count) { idle++; continue; }
        idle = 0; REQUIRE(delivered++ < 10000);
        int at = noisy ? rand() % packet_count : 0;
        struct packet p = packets[at];
        memmove(packets + at, packets + at + 1, (size_t)(--packet_count - at) * sizeof(*packets));
        for (int i = 0; i < count; i++) if (nodes[i]->node_id == p.to) {
            receive(nodes[i], &p);
            if (noisy && delivered % 7 == 0) receive(nodes[i], &p);
        }
    }
}

static void begin_pull(struct UTUN_INSTANCE* a, uint64_t remote, uint64_t group) {
    struct packet p = {.from = remote, .to = a->node_id, .group = group, .len = 1, .data = {MEDIA_SUBCMD_CHANGED}};
    receive(a, &p); poll_ready();
    struct media_super_peer* sp = peer(a, remote, group); REQUIRE(sp);
    if (!sp->request_pending) expire(sp);
}

static struct packet request_from(struct UTUN_INSTANCE* a, uint64_t remote) {
    struct packet q = {0};
    poll_ready();
    while (packet_count) {
        struct packet p = take_packet();
        if (p.from == a->node_id && p.to == remote && p.data[0] == MEDIA_SUBCMD_SYNC_REQ) q = p;
    }
    REQUIRE(q.len); return q;
}

static void test_group_boundary(void) {
    struct UTUN_INSTANCE* a = node(SERVER);
    struct packet p = {.from = CLIENT, .to = SERVER, .group = GROUP, .len = MEDIA_HAVE_BLOCK_SIZE};
    struct media_pkt_have_block* hb = (struct media_pkt_have_block*)p.data;
    hb->subcmd = MEDIA_SUBCMD_HAVE_BLOCK; hb->group_id = GROUP + 1; hb->block_id[0] = 1; hb->timestamp = 1;
    receive(a, &p);
    REQUIRE(!scalar(a->md.db, "SELECT COUNT(*) FROM block_availability"));
    hb->group_id = GROUP; receive(a, &p);
    REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM block_availability") == 1);
    sqlite3_int64 high = scalar(a->md.db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'");
    receive(a, &p);
    REQUIRE(scalar(a->md.db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'") == high);
    hb->timestamp = 2; receive(a, &p);
    REQUIRE(scalar(a->md.db, "SELECT id FROM block_availability") > high);
    hb->subcmd = MEDIA_SUBCMD_BLOCK_PROCESSING; hb->timestamp = 3; receive(a, &p);
    REQUIRE(scalar(a->md.db, "SELECT status FROM block_availability") == 1);
    free_node(a); reset(); capture_blocks = 1;
}

static void test_pages(void) {
    const int counts[] = {0, 1, 32, 33, 65, 10000};
    for (unsigned t = 0; t < sizeof(counts) / sizeof(*counts); t++) {
        struct UTUN_INSTANCE* a = node(SERVER);
        struct media_index_result r = result(counts[t] ? counts[t] : 1);
        if (counts[t]) seed(a, &r, counts[t], 1, 0);
        /* Реальный 64-bit AUTOINCREMENT и пропуски после DELETE/чужой группы. */
        sql(a->md.db, "UPDATE block_availability SET id=id+4294967296;"
            "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(zeroblob(16),222,1001,0,1,1);"
            "DELETE FROM block_availability WHERE group_id=222;");
        uint64_t cursor = 0; int seen = 0, pages = 0;
        for (;;) {
            struct packet p = page(a, CLIENT, GROUP, cursor, a->md.source_epoch, 1);
            const struct media_pkt_sync_resp* h = (const struct media_pkt_sync_resp*)p.data;
            REQUIRE(h->status == MEDIA_SYNC_OK && read_le(&h->from_id, 8) == cursor);
            unsigned n = (unsigned)read_le(&h->num_entries, 2);
            REQUIRE(n <= 32 && p.len == sizeof(*h) + n * sizeof(struct media_pkt_sync_entry));
            const struct media_pkt_sync_entry* rows = (const struct media_pkt_sync_entry*)(p.data + sizeof(*h));
            for (unsigned i = 0; i < n; i++) {
                REQUIRE(read_le(&rows[i].id, 8) > cursor);
                REQUIRE(read_le(&rows[i].node_id, 8) == 1000);
                cursor = read_le(&rows[i].id, 8);
            }
            seen += n; pages++;
            REQUIRE(read_le(&h->next_id, 8) >= cursor);
            cursor = read_le(&h->next_id, 8);
            if (!h->more) break;
        }
        REQUIRE(seen == counts[t] && pages == (counts[t] ? (counts[t] + 31) / 32 : 1));
        REQUIRE(cursor == (uint64_t)scalar(a->md.db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'"));
        media_index_result_free(&r); free_node(a); reset(); capture_blocks = 1;
    }
}

static void test_atomic_and_validation(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);
    const struct media_pkt_sync_req* query = (const struct media_pkt_sync_req*)q.data;
    struct packet response = page(b, SERVER, GROUP, read_le(&query->after_id, 8), query->source_epoch, query->request_id[0]);
    memcpy(((struct media_pkt_sync_resp*)response.data)->request_id, query->request_id, 16);
    struct media_super_peer* sp = peer(a, CLIENT, GROUP); REQUIRE(sp && sp->request_pending);
    uint64_t deadline = sp->deadline_tb;
    struct packet bad = response;
    for (size_t len = 1; len < response.len; len++) { bad.len = len; receive(a, &bad); }
    bad = response; bad.group++; receive(a, &bad);
    bad = response; bad.from = 768; receive(a, &bad);
    bad = response; ((struct media_pkt_sync_resp*)bad.data)->request_id[1] ^= 1; receive(a, &bad);
    bad = response; ((struct media_pkt_sync_resp*)bad.data)->more = 1; receive(a, &bad);
    bad = response;
    struct media_pkt_sync_entry* rows = (struct media_pkt_sync_entry*)(bad.data + MEDIA_SYNC_RESP_HDR_SIZE);
    rows[1].status = 3; receive(a, &bad);
    for (int fault = 0; fault < 8; fault++) {
        bad = response;
        struct media_pkt_sync_resp* h = (struct media_pkt_sync_resp*)bad.data;
        rows = (struct media_pkt_sync_entry*)(bad.data + sizeof(*h));
        if (fault == 0) bad.len++;
        if (fault == 1) write_le(&h->num_entries, 33, 2);
        if (fault == 2) h->status = 3;
        if (fault == 3) write_le(&h->next_id, UINT64_MAX, 8);
        if (fault == 4) rows[1].id = rows[0].id;
        if (fault == 5) memset(&rows[0].node_id, 0, 8);
        if (fault == 6) write_le(&rows[1].timestamp, UINT64_MAX, 8);
        if (fault == 7) memcpy(rows[1].block_id, rows[0].block_id, 16);
        receive(a, &bad);
    }
    REQUIRE(sp->request_pending && sp->deadline_tb == deadline && !scalar(a->md.db, "SELECT COUNT(*) FROM super_sync"));
    REQUIRE(!scalar(a->md.db, "SELECT COUNT(*) FROM block_availability"));
    const char* failures[] = {
        "CREATE TRIGGER fail BEFORE INSERT ON block_availability WHEN NEW.chunk=1 BEGIN SELECT RAISE(ABORT,'row fault'); END",
        "CREATE TRIGGER fail BEFORE INSERT ON super_sync BEGIN SELECT RAISE(ABORT,'cursor fault'); END"
    };
    for (unsigned i = 0; i < 2; i++) {
        sql(a->md.db, failures[i]); receive(a, &response);
        REQUIRE(sp->request_pending && !scalar(a->md.db, "SELECT COUNT(*) FROM block_availability")
            && !scalar(a->md.db, "SELECT COUNT(*) FROM super_sync"));
        sql(a->md.db, "DROP TRIGGER fail");
    }
    sql(a->md.db, "BEGIN"); receive(a, &response); sql(a->md.db, "ROLLBACK");
    REQUIRE(sp->request_pending && !scalar(a->md.db, "SELECT COUNT(*) FROM super_sync"));
    receive(a, &response);
    REQUIRE(!sp->request_pending && scalar(a->md.db, "SELECT COUNT(*) FROM block_availability") == 2);
    sqlite3_int64 high = scalar(a->md.db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'");
    deadline = sp->deadline_tb; receive(a, &response);
    REQUIRE(deadline == sp->deadline_tb && scalar(a->md.db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'") == high);
    free_node(a); free_node(b); media_index_result_free(&r); reset(); capture_blocks = 1;
}

static void test_convergence_and_lifecycle(void) {
    struct UTUN_INSTANCE* nodes[3] = {node(SERVER), node(CLIENT), node(768)};
    for (int n = 0; n < 3; n++) {
        struct media_index_result r = result(65);
        for (int i = 0; i < 65; i++) r.block_ids[i * 16 + 15] = (uint8_t)n;
        seed(nodes[n], &r, 65, 1, 0); media_index_result_free(&r);
        media_delivery_refresh_roles(nodes[n]);
    }
    synchronize(nodes, 3, 1);
    for (int i = 0; i < 3; i++) {
        REQUIRE(scalar(nodes[i]->md.db, "SELECT COUNT(*) FROM block_availability") == 195);
        REQUIRE(scalar(nodes[i]->md.db, "SELECT COUNT(*) FROM super_sync") == 2);
    }
    sqlite3_int64 highs[3];
    for (int i = 0; i < 3; i++) {
        highs[i] = scalar(nodes[i]->md.db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'");
        for (struct ll_entry* e = nodes[i]->md.super_peers->head; e; e = e->next)
            ((struct media_super_peer*)e->data)->deadline_tb = 0;
    }
    /* Настоящий watchdog запускает периодическую сверку, а не тестовая копия алгоритма. */
    uint64_t until = get_time_tb() + 3000;
    while (get_time_tb() < until) uasync_poll(ua, 100);
    synchronize(nodes, 3, 1);
    for (int i = 0; i < 3; i++) REQUIRE(highs[i] == scalar(nodes[i]->md.db,
        "SELECT seq FROM sqlite_sequence WHERE name='block_availability'"));
    add_group(nodes[0], 222); media_delivery_refresh_roles(nodes[0]); poll_ready();
    REQUIRE(peer(nodes[0], CLIENT, GROUP) && peer(nodes[0], CLIENT, 222));
    sql(nodes[0]->md.db, "UPDATE peers_111 SET adm_tags=NULL WHERE node_id=256");
    media_delivery_refresh_roles(nodes[0]);
    REQUIRE(!peer(nodes[0], CLIENT, GROUP) && peer(nodes[0], CLIENT, 222));
    REQUIRE(!scalar(nodes[0]->md.db, "SELECT COUNT(*) FROM super_sync WHERE group_id=111"));
    REQUIRE(!scalar(nodes[0]->md.db, "SELECT COUNT(*) FROM block_availability WHERE group_id=111 AND node_id!=256"));
    sql(nodes[0]->md.db, "UPDATE peers_111 SET adm_tags='{\"supernode\":\"yes\"}' WHERE node_id=256");
    media_delivery_refresh_roles(nodes[0]);
    synchronize(nodes, 3, 0);
    REQUIRE(scalar(nodes[0]->md.db, "SELECT COUNT(*) FROM block_availability WHERE group_id=111") == 195);
    for (int i = 0; i < 3; i++) free_node(nodes[i]);
    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; }

static int deny_rows(void* arg, int action, const char* table, const char* column, const char* db, const char* trigger) {
    (void)arg; (void)column; (void)db; (void)trigger;
    return action == SQLITE_READ && table && !strcmp(table, "block_availability") ? SQLITE_DENY : SQLITE_OK;
}

/* Обновление после последнего snapshot: обе перестановки CHANGED/RESP и потеря уведомления. */
static void test_hints_and_read_error(void) {
    for (int order = 0; order < 3; order++) {
        struct UTUN_INSTANCE* a = node(SERVER), *b = node(CLIENT);
        begin_pull(a, CLIENT, GROUP);
        struct packet q = request_from(a, CLIENT);
        struct media_super_peer* sp = peer(a, CLIENT, GROUP);
        uint64_t deadline = sp->deadline_tb;
        REQUIRE(sqlite3_set_authorizer(b->md.db, deny_rows, NULL) == SQLITE_OK);
        struct packet error = respond_to(b, &q);
        REQUIRE(((struct media_pkt_sync_resp*)error.data)->status == MEDIA_SYNC_DB_ERROR && error.len == MEDIA_SYNC_RESP_HDR_SIZE);
        receive(a, &error); REQUIRE(sp->request_pending && sp->deadline_tb == deadline);
        REQUIRE(sqlite3_set_authorizer(b->md.db, NULL, NULL) == SQLITE_OK);
        struct packet empty = respond_to(b, &q);
        struct media_index_result r = result(1); seed(b, &r, 1, 1, 0);
        struct packet hint = {.from = CLIENT, .to = SERVER, .group = GROUP, .len = 1, .data = {MEDIA_SUBCMD_CHANGED}};
        if (order == 0) receive(a, &hint);
        receive(a, &empty);
        if (order == 1) receive(a, &hint);
        if (order != 0) { packet_count = 0; expire(sp); } /* order=2: потерянный hint, истечение idle-сверки. */
        q = request_from(a, CLIENT);
        struct packet filled = respond_to(b, &q); receive(a, &filled);
        REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM block_availability") == 1 && !sp->request_pending);
        free_node(a); free_node(b); media_index_result_free(&r); reset(); capture_blocks = 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);
    debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
    ua = uasync_create(); REQUIRE(ua); capture_blocks = 1;
    test_group_boundary();
    test_pages();
    test_atomic_and_validation();
    test_convergence_and_lifecycle();
    test_retry_and_epoch();
    test_hints_and_read_error();
    test_backpressure();
    REQUIRE(cm_opened == cm_closed);
    uasync_destroy(ua, 0);
    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "media replication regressions: PASS CM=%d/%d", cm_opened, cm_closed);
    return 0;
}