241 lines
15 KiB
C
241 lines
15 KiB
C
/* Проверяем настоящую схему media_delivery: restart, сброс старого кеша и сохранность media_files. */
|
|
#include "media_delivery_fixture.h"
|
|
#include "../src/routing_layer/topo_group.h"
|
|
|
|
#ifdef __linux__
|
|
enum sql_fault { SQL_NONE, SQL_PREPARE, SQL_BIND, SQL_STEP_FIRST, SQL_STEP_AFTER, SQL_FINALIZE };
|
|
static enum sql_fault sql_fault;
|
|
static sqlite3_stmt* fault_statement;
|
|
static unsigned fault_steps, fault_injected;
|
|
int __real_sqlite3_prepare_v2(sqlite3*, const char*, int, sqlite3_stmt**, const char**);
|
|
int __real_sqlite3_bind_int64(sqlite3_stmt*, int, sqlite3_int64);
|
|
int __real_sqlite3_step(sqlite3_stmt*);
|
|
int __real_sqlite3_finalize(sqlite3_stmt*);
|
|
|
|
/* Ошибки внедряются только в внешний обход; вложенные SQL анонса остаются настоящими. */
|
|
int __wrap_sqlite3_prepare_v2(sqlite3* db, const char* command, int size, sqlite3_stmt** st, const char** tail) {
|
|
static const char prefix[] = "SELECT id,media_id,block_id,chunk,chat_id FROM media_files";
|
|
int target = sql_fault && !strncmp(command, prefix, sizeof(prefix) - 1);
|
|
if (target && sql_fault == SQL_PREPARE) { *st = NULL; fault_injected++; return SQLITE_NOMEM; }
|
|
int rc = __real_sqlite3_prepare_v2(db, command, size, st, tail);
|
|
if (target && rc == SQLITE_OK) fault_statement = *st;
|
|
return rc;
|
|
}
|
|
int __wrap_sqlite3_bind_int64(sqlite3_stmt* st, int column, sqlite3_int64 value) {
|
|
if (st == fault_statement && sql_fault == SQL_BIND) { fault_injected++; return SQLITE_NOMEM; }
|
|
return __real_sqlite3_bind_int64(st, column, value);
|
|
}
|
|
int __wrap_sqlite3_step(sqlite3_stmt* st) {
|
|
if (st == fault_statement) {
|
|
unsigned step = fault_steps++;
|
|
if (sql_fault == SQL_STEP_FIRST || (sql_fault == SQL_STEP_AFTER && step == 1)) {
|
|
fault_injected++; return SQLITE_IOERR;
|
|
}
|
|
}
|
|
return __real_sqlite3_step(st);
|
|
}
|
|
int __wrap_sqlite3_finalize(sqlite3_stmt* st) {
|
|
int target = st && st == fault_statement;
|
|
if (target) fault_statement = NULL;
|
|
int rc = __real_sqlite3_finalize(st);
|
|
if (target && sql_fault == SQL_FINALIZE) { fault_injected++; return SQLITE_IOERR; }
|
|
return rc;
|
|
}
|
|
#endif
|
|
|
|
static void sql(sqlite3* db, const char* command) {
|
|
REQUIRE(sqlite3_exec(db, command, NULL, NULL, NULL) == 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 && sqlite3_step(st) == SQLITE_ROW);
|
|
sqlite3_int64 value = sqlite3_column_int64(st, 0); REQUIRE(sqlite3_finalize(st) == SQLITE_OK); return value;
|
|
}
|
|
|
|
#define ANN_MEDIA "x'0102030405060708090a0b0c0d0e0f10'"
|
|
#define ANN_BLOCK "x'2122232425262728292a2b2c2d2e2f30'"
|
|
#define ANN_BLOCK_SECOND "x'2122232425262728292a2b2c2d2e2f31'"
|
|
|
|
/* Данные нарочно добавляются в обход media_index writers: проверяем повреждённую, но допустимую схемой БД. */
|
|
static void insert_announcement(sqlite3* db, const char* media, const char* block, const char* chunk, const char* channel) {
|
|
char command[1024];
|
|
int n = snprintf(command, sizeof(command), "INSERT INTO media_files"
|
|
"(media_id,block_id,content_hash,chat_id,location,node_id,node_sign,timestamp,file_size,chunk_size,chunk,offset)"
|
|
" VALUES(%s,%s,zeroblob(32),%s,'announcement.bin',256,zeroblob(64),1,1,1,%s,0)",
|
|
media ? media : ANN_MEDIA, block ? block : ANN_BLOCK, channel ? channel : "'111'", chunk ? chunk : "0");
|
|
REQUIRE(n > 0 && (size_t)n < sizeof(command)); sql(db, command);
|
|
}
|
|
|
|
/* Только удалённый CLIENT имеет роль supernode: refresh не открывает CM и не маскирует чтение плохих строк. */
|
|
static void add_announcement_group(struct UTUN_INSTANCE* inst, uint64_t id) {
|
|
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, "announcement_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 = id; g->group_type = TOPO_GROUP_TYPE_CHAT;
|
|
snprintf(g->channel_id, sizeof(g->channel_id), "%llu", (unsigned long long)id);
|
|
queue_data_put_with_index(inst->topo_groups->group_list, e);
|
|
char command[512];
|
|
snprintf(command, sizeof(command), "CREATE TABLE peers_%s(node_id INTEGER PRIMARY KEY,adm_tags TEXT);"
|
|
"INSERT INTO peers_%s VALUES(256,'{}'),(512,'{\"supernode\":\"yes\"}')", g->channel_id, g->channel_id);
|
|
sql(inst->topo_sqlite_db, command);
|
|
}
|
|
|
|
static void check_announcement(const struct packet* packet, uint64_t group, uint32_t chunk, uint8_t last_block) {
|
|
REQUIRE(packet->from == SERVER && packet->to == CLIENT && packet->group == group && packet->len == MEDIA_HAVE_BLOCK_SIZE);
|
|
const struct media_pkt_have_block* hb = (const struct media_pkt_have_block*)packet->data;
|
|
REQUIRE(hb->subcmd == MEDIA_SUBCMD_HAVE_BLOCK && hb->group_id == group && hb->chunk == chunk && hb->timestamp == 1);
|
|
for (unsigned i = 0; i < 16; i++) {
|
|
REQUIRE(hb->media_id[i] == i + 1);
|
|
REQUIRE(hb->block_id[i] == (i == 15 ? last_block : i + 0x21));
|
|
}
|
|
}
|
|
|
|
/* Проверяем не только отсутствие crash: каждый HAVE_BLOCK обязан точно соответствовать валидной исходной строке. */
|
|
static void test_announcements(void) {
|
|
struct UTUN_INSTANCE* a = instance(SERVER);
|
|
add_announcement_group(a, GROUP); add_announcement_group(a, 1); add_announcement_group(a, UINT64_MAX);
|
|
const struct {
|
|
const char *name, *media, *block, *chunk, *channel;
|
|
uint64_t expected_group;
|
|
uint32_t expected_chunk;
|
|
} cases[] = {
|
|
{"valid", NULL, NULL, NULL, NULL, GROUP, 0},
|
|
{"media0", "zeroblob(0)", NULL}, {"media1", "zeroblob(1)", NULL},
|
|
{"media15", "zeroblob(15)", NULL}, {"media17", "zeroblob(17)", NULL},
|
|
{"media_text", "'1234567890123456'", NULL}, {"media_int", "1", NULL}, {"media_float", "1.5", NULL},
|
|
{"block0", NULL, "zeroblob(0)"}, {"block1", NULL, "zeroblob(1)"},
|
|
{"block15", NULL, "zeroblob(15)"}, {"block17", NULL, "zeroblob(17)"},
|
|
{"block_text", NULL, "'1234567890123456'"}, {"block_int", NULL, "1"}, {"block_float", NULL, "1.5"},
|
|
{"chunk_negative", NULL, NULL, "-1"}, {"chunk_overflow", NULL, NULL, "4294967296"},
|
|
{"chunk_text", NULL, NULL, "'invalid'"}, {"chunk_float", NULL, NULL, "1.5"}, {"chunk_blob", NULL, NULL, "x'00'"},
|
|
{"chunk_max", NULL, NULL, "4294967295", NULL, GROUP, UINT32_MAX},
|
|
{"channel_empty", NULL, NULL, NULL, "''"}, {"channel_zero", NULL, NULL, NULL, "'0'"},
|
|
{"channel_leading_zero", NULL, NULL, NULL, "'0111'"}, {"channel_suffix", NULL, NULL, NULL, "'111junk'"},
|
|
{"channel_plus", NULL, NULL, NULL, "'+111'"}, {"channel_negative", NULL, NULL, NULL, "'-111'"},
|
|
{"channel_left_space", NULL, NULL, NULL, "' 111'"}, {"channel_right_space", NULL, NULL, NULL, "'111 '"},
|
|
{"channel_blob", NULL, NULL, NULL, "x'313131'"}, {"channel_float", NULL, NULL, NULL, "1.5"},
|
|
{"channel_nul", NULL, NULL, NULL, "CAST(x'313131006a756e6b' AS TEXT)"},
|
|
{"channel_overflow", NULL, NULL, NULL, "'18446744073709551616'"},
|
|
{"channel_too_long", NULL, NULL, NULL, "'184467440737095516150'"},
|
|
{"channel_one", NULL, NULL, NULL, "'1'", 1, 0},
|
|
{"channel_max", NULL, NULL, NULL, "'18446744073709551615'", UINT64_MAX, 0},
|
|
};
|
|
unsigned valid = 0;
|
|
debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_WARN);
|
|
for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
|
|
sql(a->md.db, "DELETE FROM media_files");
|
|
insert_announcement(a->md.db, cases[i].media, cases[i].block, cases[i].chunk, cases[i].channel);
|
|
reset(); capture_blocks = 1;
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "announcement case=%s expected_packets=%d", cases[i].name, cases[i].expected_group != 0);
|
|
media_delivery_announce_local_blocks(a);
|
|
REQUIRE(packet_count == (cases[i].expected_group ? 1 : 0));
|
|
if (cases[i].expected_group) { check_announcement(&packets[0], cases[i].expected_group, cases[i].expected_chunk, 0x30); valid++; }
|
|
REQUIRE(sqlite3_next_stmt(a->md.db, NULL) == NULL);
|
|
}
|
|
sql(a->md.db, "DELETE FROM media_files");
|
|
insert_announcement(a->md.db, NULL, NULL, NULL, NULL);
|
|
insert_announcement(a->md.db, "x'01'", "x'ff'", "1", NULL);
|
|
insert_announcement(a->md.db, NULL, ANN_BLOCK_SECOND, "4294967295", "'18446744073709551615'");
|
|
reset(); capture_blocks = 1;
|
|
media_delivery_announce_local_blocks(a); REQUIRE(packet_count == 2 && packets[0].group != packets[1].group);
|
|
for (int i = 0; i < packet_count; i++) {
|
|
const struct media_pkt_have_block* hb = (const struct media_pkt_have_block*)packets[i].data;
|
|
check_announcement(&packets[i], hb->chunk ? UINT64_MAX : GROUP, hb->chunk ? UINT32_MAX : 0, hb->chunk ? 0x31 : 0x30);
|
|
}
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "announcement validation cases=%zu valid=%u mixed=2/3 PASS", sizeof(cases) / sizeof(cases[0]), valid);
|
|
|
|
#ifdef __linux__
|
|
sql(a->md.db, "DELETE FROM media_files");
|
|
insert_announcement(a->md.db, NULL, NULL, NULL, NULL);
|
|
insert_announcement(a->md.db, NULL, ANN_BLOCK_SECOND, "1", NULL);
|
|
for (enum sql_fault fault = SQL_PREPARE; fault <= SQL_FINALIZE; fault++) {
|
|
const char* logfile = "test_media_announcement_fault.log";
|
|
reset(); capture_blocks = 1; sql_fault = fault; fault_steps = fault_injected = 0;
|
|
debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_INFO);
|
|
debug_enable_file_output(logfile, 1);
|
|
media_delivery_announce_local_blocks(a);
|
|
debug_disable_file_output();
|
|
REQUIRE(fault_injected == 1 && fault_statement == NULL);
|
|
int expected = fault == SQL_FINALIZE ? 2 : fault == SQL_STEP_AFTER ? 1 : 0;
|
|
REQUIRE(packet_count == expected && sqlite3_next_stmt(a->md.db, NULL) == NULL);
|
|
FILE* file = fopen(logfile, "rb"); REQUIRE(file);
|
|
char log[8192]; size_t n = fread(log, 1, sizeof(log) - 1, file); log[n] = 0;
|
|
REQUIRE(!ferror(file) && fgetc(file) == EOF && !fclose(file) && !remove(logfile));
|
|
REQUIRE(strstr(log, "[ERROR]") && !strstr(log, "local block scan complete"));
|
|
REQUIRE(strstr(log, fault == SQL_PREPARE ? "announce_local_blocks prepare failed rc=" : "SQL phase=enumerate local blocks"));
|
|
if (fault != SQL_PREPARE) REQUIRE(strstr(log, "local block scan incomplete"));
|
|
sql_fault = SQL_NONE; reset(); capture_blocks = 1;
|
|
media_delivery_announce_local_blocks(a); REQUIRE(packet_count == 2 && sqlite3_next_stmt(a->md.db, NULL) == NULL);
|
|
unsigned seen = 0;
|
|
for (int i = 0; i < packet_count; i++) {
|
|
uint32_t chunk = ((const struct media_pkt_have_block*)packets[i].data)->chunk;
|
|
REQUIRE(chunk < 2 && !(seen & (1u << chunk))); seen |= 1u << chunk;
|
|
check_announcement(&packets[i], GROUP, chunk, chunk ? 0x31 : 0x30);
|
|
}
|
|
REQUIRE(seen == 3);
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "announcement SQL fault=%d injected=1 packets_before_error=%d recovery=2 PASS", fault, expected);
|
|
}
|
|
#endif
|
|
struct TOPO_GROUPS* groups = a->topo_groups;
|
|
destroy(a);
|
|
while (groups->group_list->head) {
|
|
struct ll_entry* e = groups->group_list->head;
|
|
queue_remove_data(groups->group_list, e); queue_entry_free(e);
|
|
}
|
|
queue_free(groups->group_list); u_free(groups);
|
|
debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_ERROR);
|
|
}
|
|
|
|
int main(void) {
|
|
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
|
|
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
|
|
ua = uasync_create(); REQUIRE(ua);
|
|
struct UTUN_INSTANCE* a = instance(SERVER);
|
|
uint8_t epoch[16]; memcpy(epoch, a->md.source_epoch, 16);
|
|
sql(a->md.db, "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status)"
|
|
" VALUES(zeroblob(16),111,256,0,1,1),(zeroblob(16),222,256,0,1,1);"
|
|
"INSERT INTO super_sync VALUES(111,512,zeroblob(16),4294967297),(222,512,zeroblob(16),8)");
|
|
REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM block_availability") == 2);
|
|
REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM super_sync") == 2);
|
|
media_delivery_destroy(a);
|
|
REQUIRE(!media_delivery_init(a) && !memcmp(epoch, a->md.source_epoch, 16));
|
|
REQUIRE(scalar(a->md.db, "SELECT last_recv_id FROM super_sync WHERE group_id=111") == 4294967297LL);
|
|
REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM block_availability") == 2);
|
|
|
|
/* Настоящая регистрация остаётся целой при пересоздании вспомогательного каталога. */
|
|
REQUIRE(!media_index_set_base(a->md.db, "/tmp"));
|
|
uint8_t media[16] = {1}, block[16] = {2}, hash[32] = {0};
|
|
REQUIRE(!media_index_register_downloaded(a->md.db, media, block, hash, "111", "media-repl-preserved.bin",
|
|
SERVER, 1, 1, 0, 0));
|
|
REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM media_files") == 1);
|
|
media_delivery_destroy(a);
|
|
sql(a->topo_sqlite_db, "DROP TABLE block_availability");
|
|
REQUIRE(!media_delivery_init(a) && memcmp(epoch, a->md.source_epoch, 16));
|
|
REQUIRE(!scalar(a->md.db, "SELECT COUNT(*) FROM super_sync"));
|
|
REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM media_files") == 1);
|
|
memcpy(epoch, a->md.source_epoch, 16);
|
|
media_delivery_announce_media(a, GROUP, media, block, 1);
|
|
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'");
|
|
media_delivery_announce_media(a, GROUP, media, block, 1);
|
|
REQUIRE(scalar(a->md.db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'") == high);
|
|
|
|
/* Старая схема и её курсоры сбрасываются одной транзакцией, epoch меняется. */
|
|
media_delivery_destroy(a);
|
|
sql(a->topo_sqlite_db, "DROP TABLE media_replication_meta; DROP TABLE block_availability; DROP TABLE super_sync;"
|
|
"CREATE TABLE block_availability(id INTEGER PRIMARY KEY,block_uuid BLOB);"
|
|
"CREATE TABLE super_sync(peer_node_id INTEGER PRIMARY KEY,last_recv_id INTEGER);"
|
|
"INSERT INTO super_sync VALUES(512,99)");
|
|
REQUIRE(!media_delivery_init(a) && memcmp(epoch, a->md.source_epoch, 16));
|
|
REQUIRE(!scalar(a->md.db, "SELECT COUNT(*) FROM super_sync"));
|
|
REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM media_files") == 1);
|
|
destroy(a); test_announcements(); uasync_destroy(ua, 0);
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "media catalog SQL restart/reset/preservation: PASS");
|
|
return 0;
|
|
}
|