301 lines
15 KiB
C
301 lines
15 KiB
C
/* Репликация: настоящие обработчики 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 (!packet_count && get_time_tb() < until) uasync_poll(ua, 100);
|
|
REQUIRE(packet_count);
|
|
}
|
|
|
|
/* Вручную запросить настоящую страницу без подмены реализации 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);
|
|
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;
|
|
}
|
|
|
|
int main(void) {
|
|
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); srand(17);
|
|
ua = uasync_create(); REQUIRE(ua); capture_blocks = 1;
|
|
test_group_boundary();
|
|
test_pages();
|
|
test_atomic_and_validation();
|
|
test_convergence_and_lifecycle();
|
|
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);
|
|
return 0;
|
|
}
|