Files
utun2/tests/test_chat_join_faults.c
T

306 lines
19 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/* Реальный chat_sync/NCD/SQLite с управляемым транспортом: отказы, корреляция и teardown. */
#include "utun_instance.h"
#include "chat/chat_core.h"
#include "chat/chat_sync.h"
#include "chat/chat_join.h"
#include "chat/db_sync.h"
#include "routing_layer/topo_group.h"
#include "chat/chat_event.h"
#include "chat/chat_join_port.h"
#include "chat/invite_link.h"
#include "transport_layer/etcp.h"
#include "transport_layer/etcp_api.h"
#include "transport_layer/etcp_connections.h"
#include "transport_layer/node_conn_direct.h"
#include "routing_layer/topo_node.h"
#include "../lib/mem.h"
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "join fault line=%d: %s", __LINE__, #x); exit(1); } } while (0)
static struct UTUN_INSTANCE* inst;
static struct UASYNC* ua;
static struct ETCP_CONN* conn;
static char invitation[INVITE_LINK_MAX];
static uint8_t request_wire[512], response_wire[8192], member_wire[8192], ready_wire[32];
static size_t response_size, member_size, ready_size;
static struct UTUN_INSTANCE* server;
static struct ETCP_CONN* server_conn;
static uint64_t channel;
static size_t request_size;
static unsigned allocations, fail_allocation, events[512];
static int results[512], fail_send, reenter, reenter_result;
static const char* fail_timer;
static int fail_runtime, fail_ncd, nested_up;
static unsigned info_sends;
static pthread_mutex_t race_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t race_condition = PTHREAD_COND_INITIALIZER;
static int race_gate, race_waiting, race_release, race_result;
static uint64_t peer;
void* __real_u_calloc_impl(uint32_t, uint32_t, const char*);
void* __real_u_malloc_impl(uint32_t, const char*);
static int allocation_failed(const char* site) {
if (fail_runtime == 1 && strstr(site, "db_sync.c")) { fail_runtime = 0; return 1; }
if (!strstr(site, "chat_sync.c") && !strstr(site, "chat_join_port.c")) return 0;
allocations++;
return fail_allocation && allocations == fail_allocation;
}
void* __wrap_u_calloc_impl(uint32_t count, uint32_t size, const char* site) {
if (allocation_failed(site)) return NULL;
void* data = __real_u_calloc_impl(count, size, site);
if (race_gate && strstr(site, "chat_join_port.c")) {
pthread_mutex_lock(&race_lock); race_waiting = 1; pthread_cond_signal(&race_condition);
while (!race_release) pthread_cond_wait(&race_condition, &race_lock);
pthread_mutex_unlock(&race_lock);
}
return data;
}
void* __wrap_u_malloc_impl(uint32_t size, const char* site) {
if (allocation_failed(site)) return NULL;
return __real_u_malloc_impl(size, site);
}
void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
void* __wrap_uasync_set_timeout(struct UASYNC* loop, int delay, void* arg, timeout_callback_t fn, const char* name) {
if (fail_timer && !strcmp(name, fail_timer)) return NULL;
return __real_uasync_set_timeout(loop, delay, arg, fn, name);
}
struct TOPO_GROUP* __real_topo_groups_create_group(struct TOPO_GROUPS*, uint64_t, uint8_t, const char*);
struct TOPO_GROUP* __wrap_topo_groups_create_group(struct TOPO_GROUPS* groups, uint64_t id, uint8_t type, const char* name) {
if (fail_runtime == 2) { fail_runtime = 0; return NULL; }
return __real_topo_groups_create_group(groups, id, type, name);
}
void __real_topo_group_add_node_cbk(struct TOPO_GROUP*, topo_node_event_fn, void*);
void __wrap_topo_group_add_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg) {
if (fail_runtime == 3) { fail_runtime = 0; return; }
__real_topo_group_add_node_cbk(group, fn, arg);
}
int __real_node_conn_direct_open_node(struct UTUN_INSTANCE*, uint64_t, ncd_callback, void*, struct NODE_CONN_DIRECT**, struct TOPO_NODE*, struct ETCP_SOCKET*);
int __wrap_node_conn_direct_open_node(struct UTUN_INSTANCE* owner, uint64_t id, ncd_callback cb, void* arg,
struct NODE_CONN_DIRECT** handle, struct TOPO_NODE* node, struct ETCP_SOCKET* socket) {
if (fail_ncd) { *handle = NULL; return NCD_ERR; }
if (nested_up && owner == inst) { nested_up = 0; etcp_fire_conn_status(conn, ETCP_CONN_STATUS_UP); }
return __real_node_conn_direct_open_node(owner, id, cb, arg, handle, node, socket);
}
int __real_etcp_send(struct ETCP_CONN*, struct ll_entry*);
int __wrap_etcp_send(struct ETCP_CONN* c, struct ll_entry* entry) {
if (entry->len >= 18 && entry->dgram[0] == ETCP_ID_CHAT_SYNC) {
if (entry->dgram[9] == CS_MSG_JOIN_INFO_REQ) {
if (c->instance == inst) info_sends++;
CHECK(entry->len <= sizeof(request_wire)); request_size = entry->len;
memcpy(request_wire, entry->dgram, request_size);
if (fail_send) return -1;
}
if (entry->dgram[9] == CS_MSG_JOIN_INFO_RESP) { response_size = entry->len; CHECK(response_size <= sizeof(response_wire)); memcpy(response_wire, entry->dgram, response_size); }
if (entry->dgram[9] == CS_MSG_JOIN_REQUEST) { member_size = entry->len; CHECK(member_size <= sizeof(member_wire)); memcpy(member_wire, entry->dgram, member_size); }
if (entry->dgram[9] == CS_MSG_JOIN_READY) { ready_size = entry->len; CHECK(ready_size <= sizeof(ready_wire)); memcpy(ready_wire, entry->dgram, ready_size); }
queue_dgram_free(entry); queue_entry_free(entry); return 0;
}
return __real_etcp_send(c, entry);
}
static void event(struct UTUN_INSTANCE* owner, int type, const uint8_t* data, int len) {
if (type != CHAT_EVT_CONNECT_RESULT) return;
CHECK(owner == inst && len == 28);
uint64_t request; int32_t result;
memcpy(&request, data + 20, 8); memcpy(&result, data + 8, 4);
CHECK(request < 512); events[request]++; results[request] = result;
if (reenter && request == 30) {
reenter = 0; reenter_result = chat_sync_join_link(inst, 31, invitation, strlen(invitation));
}
}
static void deliver_error(const uint8_t token[8]) {
struct ll_entry* entry = ll_alloc_lldgram(18); CHECK(entry);
entry->len = 18; entry->dgram[0] = ETCP_ID_CHAT_SYNC;
uint64_t group = htobe64(channel); memcpy(entry->dgram + 1, &group, 8);
entry->dgram[9] = CS_MSG_ERROR; memcpy(entry->dgram + 10, token, 8);
inst->api_bindings.callbacks[ETCP_ID_CHAT_SYNC](conn, entry);
}
static void deliver(struct UTUN_INSTANCE* target, struct ETCP_CONN* connection, const uint8_t* bytes, size_t size) {
struct ll_entry* entry = ll_alloc_lldgram(size); CHECK(entry); entry->len = size; memcpy(entry->dgram, bytes, size);
target->api_bindings.callbacks[ETCP_ID_CHAT_SYNC](connection, entry);
}
static void description(void) {
response_size = 0; deliver(server, server_conn, request_wire, request_size); CHECK(response_size > 200);
}
static int channels(void) {
sqlite3_stmt* st = NULL; CHECK(sqlite3_prepare_v2(inst->topo_sqlite_db, "SELECT count(*) FROM channels", -1, &st, NULL) == SQLITE_OK);
CHECK(sqlite3_step(st) == SQLITE_ROW); int value = sqlite3_column_int(st, 0); sqlite3_finalize(st); return value;
}
static void* producer(void* arg) {
struct chat_join_port* port = arg;
CHECK(!chat_join_port_submit(port, 60, invitation, strlen(invitation)));
return NULL;
}
static void* stopped_producer(void* arg) {
race_result = chat_join_port_submit(arg, 71, invitation, strlen(invitation)); return NULL;
}
static void stop_in_batch(void* arg) { chat_service_stop(arg); }
int main(void) {
size_t before = u_get_allocated_count();
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
ua = uasync_create(); CHECK(ua);
inst = utun_instance_create_from_str(ua, "[global]\nauto_sockets=no\n[ntp]\nenabled=no\n"
"[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n");
CHECK(inst && !utun_core_start(inst) && !chat_service_start(inst));
chat_event_set_handler(inst, event);
server = utun_instance_create_from_str(ua, "[global]\nauto_sockets=no\n[ntp]\nenabled=no\n"
"[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n");
CHECK(server && !utun_core_start(server) && !chat_service_start(server));
chat_core_create_channel_auto(server, "fault channel");
CHECK(server->topo_groups->group_list->head);
channel = ((struct TOPO_GROUP*)server->topo_groups->group_list->head)->group_id;
CHECK(!chat_join_store_local_key(server, channel, 123));
struct InviteData invite = { .channelId = channel, .join_key = 123, .addrCount = 4 };
memcpy(invite.pubkey, server->my_keys.public_key, 32); peer = server->node_id;
for (unsigned i = 0; i < invite.addrCount; i++) {
invite.addrs[i].family = i & 1 ? 6 : 4; invite.addrs[i].port = 12345;
invite.addrs[i].proto = 1; invite.addrs[i].address[0] = 127; invite.addrs[i].address[3] = 1;
}
invite.reality.has = 1; memcpy(invite.reality.server_pubkey, server->my_keys.public_key, 32);
snprintf(invite.reality.server_name, sizeof(invite.reality.server_name), "example.org"); invite.addrs[3].proto = 6;
CHECK(invite_link_encode(&invite, NULL, invitation, sizeof(invitation)) > 0);
conn = etcp_connection_create(inst, "join fault transport"); CHECK(conn);
conn->peer_node_id = peer; CHECK(!etcp_conn_index_peer(conn));
CHECK(sc_init_ctx(&conn->crypto_ctx, &inst->my_keys) == SC_OK);
CHECK(sc_set_peer_public_key(&conn->crypto_ctx, invite.pubkey, SC_PEER_PUBKEY_BIN) == SC_OK);
conn->links_up = 1; etcp_conn_ready(conn); conn->initialized = 1;
struct NODE_CONN_DIRECT* other_owner = NULL;
CHECK(node_conn_direct_open(inst, peer, NULL, NULL, &other_owner, NULL) != NCD_ERR);
uasync_drain_immediate(ua);
server_conn = etcp_connection_create(server, "join server fixture"); CHECK(server_conn);
server_conn->peer_node_id = inst->node_id; CHECK(!etcp_conn_index_peer(server_conn));
CHECK(sc_init_ctx(&server_conn->crypto_ctx, &server->my_keys) == SC_OK);
CHECK(sc_set_peer_public_key(&server_conn->crypto_ctx, inst->my_keys.public_key, SC_PEER_PUBKEY_BIN) == SC_OK);
server_conn->links_up = 1; etcp_conn_ready(server_conn); server_conn->initialized = 1;
CHECK(!channels());
CHECK(chat_sync_join_link(inst, 1, "utun://", 7) < 0 && !channels() && !events[1]);
CHECK(!chat_sync_join_link(inst, 10, invitation, strlen(invitation)) && request_size > 18);
uint8_t old[8]; memcpy(old, request_wire + 10, 8);
CHECK(chat_sync_join_link(inst, 11, invitation, strlen(invitation)) == CHAT_CONNECT_BUSY && !events[11]);
chat_sync_cancel_join(inst, 999); CHECK(!events[10]);
chat_sync_cancel_join(inst, 10); CHECK(events[10] == 1 && results[10] == CHAT_CONNECT_CANCELLED);
CHECK(!chat_sync_join_link(inst, 12, invitation, strlen(invitation)));
uint8_t current[8]; memcpy(current, request_wire + 10, 8); CHECK(memcmp(old, current, 8));
deliver_error(old); CHECK(!events[12]);
deliver_error(current); CHECK(events[12] == 1 && results[12] == CHAT_CONNECT_ERROR);
deliver_error(current); CHECK(events[12] == 1);
CHECK(node_conn_direct_get_conn(other_owner) == conn && !conn->close_requested);
CHECK(!channels());
reenter = 1; CHECK(!chat_sync_join_link(inst, 30, invitation, strlen(invitation)));
chat_sync_cancel_join(inst, 30); CHECK(!reenter_result && events[30] == 1);
chat_sync_cancel_join(inst, 31); CHECK(events[31] == 1);
/* Сначала измерить все реальные точки выделения этого пути, затем отказать в каждой. */
allocations = 0; CHECK(!chat_sync_join_link(inst, 40, invitation, strlen(invitation)));
unsigned total = allocations; CHECK(total >= 5); chat_sync_cancel_join(inst, 40);
for (unsigned i = 1; i <= total; i++) {
allocations = 0; fail_allocation = i;
int rc = chat_sync_join_link(inst, 41, invitation, strlen(invitation));
fail_allocation = 0;
CHECK(rc < 0 && !events[41] && !channels());
CHECK(node_conn_direct_get_conn(other_owner) == conn && !conn->close_requested);
CHECK(!chat_sync_join_link(inst, 42, invitation, strlen(invitation))); chat_sync_cancel_join(inst, 42);
}
for (unsigned i = 0; i < 2; i++) {
fail_timer = i ? "cs_info_req" : "join_transport_timeout";
CHECK(chat_sync_join_link(inst, 43, invitation, strlen(invitation)) < 0 && !events[43]); fail_timer = NULL;
CHECK(!chat_sync_join_link(inst, 44, invitation, strlen(invitation))); chat_sync_cancel_join(inst, 44);
}
fail_send = 1; CHECK(chat_sync_join_link(inst, 45, invitation, strlen(invitation)) < 0 && !events[45]); fail_send = 0;
CHECK(!chat_sync_join_link(inst, 46, invitation, strlen(invitation))); chat_sync_cancel_join(inst, 46);
fail_ncd = 1; CHECK(chat_sync_join_link(inst, 47, invitation, strlen(invitation)) < 0 && !events[47]); fail_ncd = 0;
CHECK(node_conn_direct_get_conn(other_owner) == conn && !conn->close_requested);
unsigned previous_sends = info_sends; nested_up = 1;
CHECK(!chat_sync_join_link(inst, 48, invitation, strlen(invitation)) && !nested_up && info_sends == previous_sends + 1);
chat_sync_cancel_join(inst, 48); CHECK(events[48] == 1);
/* Сервер обслуживает чужой JOIN, пока сам ждёт завершения локального JOIN. */
struct InviteData reverse = invite;
memcpy(reverse.pubkey, inst->my_keys.public_key, 32);
char reverse_link[INVITE_LINK_MAX]; CHECK(invite_link_encode(&reverse, NULL, reverse_link, sizeof(reverse_link)) > 0);
CHECK(!chat_sync_join_link(server, 90, reverse_link, strlen(reverse_link)));
CHECK(!sqlite3_exec(inst->topo_sqlite_db, "CREATE TEMP TRIGGER reject_join BEFORE INSERT ON channels "
"BEGIN SELECT RAISE(ABORT,'join fault'); END", NULL, NULL, NULL));
CHECK(!chat_sync_join_link(inst, 80, invitation, strlen(invitation))); description();
deliver(inst, conn, response_wire, response_size);
CHECK(events[80] == 1 && results[80] == CHAT_CONNECT_ERROR && !channels() && sqlite3_get_autocommit(inst->topo_sqlite_db));
CHECK(!sqlite3_exec(inst->topo_sqlite_db, "DROP TRIGGER reject_join", NULL, NULL, NULL));
CHECK(!chat_sync_join_link(inst, 81, invitation, strlen(invitation))); description();
size_t signature_offset = 19 + response_wire[18] + 8 + 32 + 32 + 64 + 8 + 32 + 32 + 64 + 8;
CHECK(signature_offset + 64 < response_size); response_wire[signature_offset] ^= 1;
deliver(inst, conn, response_wire, response_size);
CHECK(events[81] == 1 && results[81] == CHAT_CONNECT_ERROR && !channels() && sqlite3_get_autocommit(inst->topo_sqlite_db));
CHECK(!topo_groups_find(inst->topo_groups, channel));
for (unsigned failure = 1; failure <= 3; failure++) {
uint64_t request = 85 + failure;
CHECK(!chat_sync_join_link(inst, request, invitation, strlen(invitation))); description();
fail_runtime = (int)failure; deliver(inst, conn, response_wire, response_size); CHECK(!fail_runtime);
CHECK(events[request] == 1 && results[request] == CHAT_CONNECT_ERROR && channels() == 1);
CHECK(!topo_groups_find(inst->topo_groups, channel) && !db_sync_instance_find(inst, channel));
}
CHECK(!chat_sync_join_link(inst, 82, invitation, strlen(invitation))); description();
fail_timer = "cs_join"; deliver(inst, conn, response_wire, response_size); fail_timer = NULL;
CHECK(events[82] == 1 && results[82] == CHAT_CONNECT_ERROR && channels() == 1);
CHECK(!chat_sync_join_link(inst, 83, invitation, strlen(invitation))); description();
member_size = 0; deliver(inst, conn, response_wire, response_size); CHECK(member_size > 200 && !events[83]);
ready_size = 0; deliver(server, server_conn, member_wire, member_size); CHECK(ready_size == 26);
deliver(inst, conn, ready_wire, ready_size); CHECK(events[83] == 1 && !results[83]);
deliver(inst, conn, ready_wire, ready_size); CHECK(events[83] == 1);
uint8_t previous_ready[32]; memcpy(previous_ready, ready_wire, ready_size);
for (unsigned failure = 0; failure < 2; failure++) {
uint64_t request = 84 + failure;
CHECK(!chat_sync_join_link(inst, request, invitation, strlen(invitation))); description();
deliver(inst, conn, response_wire, response_size); CHECK(!events[request]);
deliver(inst, conn, previous_ready, 26); CHECK(!events[request]);
deliver(server, server_conn, member_wire, member_size); CHECK(ready_size == 26);
if (failure) { memset(ready_wire + 18, 0, 8); deliver(inst, conn, ready_wire, 26); }
else deliver(inst, conn, ready_wire, 25);
CHECK(events[request] == 1 && results[request] == CHAT_CONNECT_ERROR);
}
chat_sync_cancel_join(server, 90);
struct chat_join_port* port = chat_sync_join_port(inst); CHECK(port);
allocations = 0; fail_allocation = 1;
CHECK(chat_join_port_submit(port, 50, invitation, strlen(invitation)) < 0 && !events[50]); fail_allocation = 0;
pthread_t thread; CHECK(!pthread_create(&thread, NULL, producer, port)); CHECK(!pthread_join(thread, NULL));
CHECK(chat_join_port_submit(port, 60, invitation, strlen(invitation)) == CHAT_CONNECT_BUSY);
chat_service_stop(inst); CHECK(events[60] == 1 && results[60] == CHAT_CONNECT_CANCELLED);
CHECK(chat_join_port_submit(port, 61, invitation, strlen(invitation)) == CHAT_CONNECT_CANCELLED);
CHECK(!chat_service_start(inst)); CHECK(chat_join_port_submit(port, 62, invitation, strlen(invitation)) == CHAT_CONNECT_CANCELLED);
chat_join_port_release(port);
port = chat_sync_join_port(inst); CHECK(port);
CHECK(!chat_join_port_submit(port, 65, invitation, strlen(invitation))); uasync_poll(ua, 0);
CHECK(!events[65] && chat_join_port_submit(port, 65, invitation, strlen(invitation)) == CHAT_CONNECT_BUSY);
CHECK(!chat_join_port_submit(port, 66, invitation, strlen(invitation))); uasync_poll(ua, 0);
CHECK(events[66] == 1 && results[66] == CHAT_CONNECT_BUSY && !events[65]);
chat_sync_cancel_join(inst, 65); CHECK(events[65] == 1 && results[65] == CHAT_CONNECT_CANCELLED);
/* Детерминированная гонка: stop между копированием аргументов producer и публикацией post. */
race_gate = 1; CHECK(!pthread_create(&thread, NULL, stopped_producer, port));
pthread_mutex_lock(&race_lock);
while (!race_waiting) pthread_cond_wait(&race_condition, &race_lock);
pthread_mutex_unlock(&race_lock);
chat_service_stop(inst);
pthread_mutex_lock(&race_lock); race_release = 1; pthread_cond_signal(&race_condition); pthread_mutex_unlock(&race_lock);
CHECK(!pthread_join(thread, NULL)); race_gate = 0;
CHECK(race_result == CHAT_CONNECT_CANCELLED && !events[71]); chat_join_port_release(port);
CHECK(!chat_service_start(inst));
/* close после извлечения списка post: отмена не находит task, пустой callback обязан освободить его. */
port = chat_sync_join_port(inst); CHECK(port);
struct posted_task* stop = u_calloc(1, sizeof(*stop)); CHECK(stop);
stop->callback = stop_in_batch; stop->arg = inst; uasync_post_reserved(ua, stop);
CHECK(!chat_join_port_submit(port, 70, invitation, strlen(invitation)));
uasync_poll(ua, 0); CHECK(events[70] == 1 && results[70] == CHAT_CONNECT_CANCELLED);
chat_join_port_release(port);
node_conn_direct_close(other_owner);
utun_instance_destroy(server); utun_instance_destroy(inst); uasync_poll(ua, 0); uasync_destroy(ua, 0);
CHECK(u_get_allocated_count() == before);
printf("join faults: %u allocation failures, deadlines, send, stale replies, reentry, ownership, thread and restart PASS\n", total);
return 0;
}