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

Exercise real media replication pages, rollback and catalog lifecycle

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 21:00:48 +02:00
1 parent d3bc77922b
commit 00502bf4d4
3 files changed
+332 -475
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/Makefile.am
+2 -1
View File
@@ -881,9 +881,10 @@ if OS_LINUX
test_media_index_regressions_LDFLAGS = -Wl,--wrap=ma_sha256_file
endif
test_media_delivery_sql_SOURCES = test_media_delivery_sql.c
test_media_delivery_sql_SOURCES = test_media_delivery_sql.c media_delivery_fixture.h
test_media_delivery_sql_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/media_delivery -I$(top_srcdir)/lib
test_media_delivery_sql_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
test_media_delivery_sql_LDFLAGS = -Wl,--wrap=etcp_route_send -Wl,--wrap=RAND_bytes
test_media_delivery_download_SOURCES = test_media_delivery_download.c
test_media_delivery_download_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/media_delivery -I$(top_srcdir)/src/media_async -I$(top_srcdir)/lib -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/chat
tests/test_media_delivery_sql.c
+49 -463
View File
@@ -1,473 +1,59 @@
// test_media_delivery_sql.c — SQL-таблицы + протокольные структуры media_delivery
//
// Покрытие:
// 1. Создание таблиц block_availability + super_sync + индексов
// 2. INSERT OR REPLACE (upsert) — дубликат не создаёт вторую строку
// 3. SELECT с фильтром по block_uuid, сортировка timestamp DESC
// 4. SELECT с id > N, сортировка ASC, LIMIT
// 5. super_sync: INSERT OR REPLACE + SELECT (дефолт 0)
// 6. DELETE WHERE node_id=?
// 7. DELETE WHERE node_id!=? (чистка чужих при выключении суперузла)
// 8. Размеры packed-структур
// 9. Проверка полей packed-структур (нет padding-сюрпризов)
// 10. Построение/разбор MEDIA_QUERY
// 11. Построение/разбор MEDIA_HAVE_BLOCK
// 12. Построение/разбор MEDIA_SUPER_HELLO
/* Проверяем настоящую схему media_delivery: restart, сброс старого кеша и сохранность media_files. */
#include "media_delivery_fixture.h"
#include "../src/media_delivery/media_delivery_proto.h"
#include "../lib/debug_config.h"
#include <sqlite3.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
static int g_passed = 0, g_failed = 0, g_total = 0;
#define TEST(name) do { g_total++; printf(" %-50s", name); } while(0)
#define OK() do { g_passed++; printf("OK\n"); } while(0)
#define FAIL(fmt, ...) do { g_failed++; printf("FAIL: " fmt "\n", ##__VA_ARGS__); } while(0)
static void make_uuid(uint8_t buf[16]) {
for (int i = 0; i < 16; i++) buf[i] = (uint8_t)(rand() & 0xFF);
static void sql(sqlite3* db, const char* command) {
REQUIRE(sqlite3_exec(db, command, NULL, NULL, NULL) == SQLITE_OK);
}
static sqlite3* make_db(void) {
sqlite3* db = NULL;
sqlite3_open(":memory:", &db);
sqlite3_exec(db, "PRAGMA journal_mode=memory", NULL, NULL, NULL);
return db;
}
static int create_tables(sqlite3* db) {
const char* sql_ba =
"CREATE TABLE IF NOT EXISTS block_availability ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" block_uuid BLOB NOT NULL,"
" group_id INTEGER NOT NULL,"
" node_id INTEGER NOT NULL,"
" chunk INTEGER NOT NULL,"
" timestamp INTEGER NOT NULL,"
" status INTEGER NOT NULL DEFAULT 1)"; // 0=processing, 1=completed
int rc = sqlite3_exec(db, sql_ba, NULL, NULL, NULL);
if (rc != SQLITE_OK) return -1;
sqlite3_exec(db, "CREATE UNIQUE INDEX IF NOT EXISTS idx_ba_uuid_node ON block_availability(block_uuid, node_id)", NULL, NULL, NULL);
sqlite3_exec(db, "CREATE INDEX IF NOT EXISTS idx_ba_id ON block_availability(id)", NULL, NULL, NULL);
sqlite3_exec(db, "CREATE INDEX IF NOT EXISTS idx_ba_node_id ON block_availability(node_id)", NULL, NULL, NULL);
const char* sql_ss =
"CREATE TABLE IF NOT EXISTS super_sync ("
" peer_node_id INTEGER PRIMARY KEY,"
" last_recv_id INTEGER NOT NULL DEFAULT 0)";
rc = sqlite3_exec(db, sql_ss, NULL, NULL, NULL);
return (rc == SQLITE_OK) ? 0 : -1;
}
static int table_exists(sqlite3* db, const char* name) {
sqlite3_stmt* s = NULL;
int ok = 0;
if (sqlite3_prepare_v2(db, "SELECT name FROM sqlite_master WHERE type='table' AND name=?", -1, &s, NULL) == SQLITE_OK) {
sqlite3_bind_text(s, 1, name, -1, SQLITE_STATIC);
ok = (sqlite3_step(s) == SQLITE_ROW);
sqlite3_finalize(s);
}
return ok;
}
static int index_exists(sqlite3* db, const char* name) {
sqlite3_stmt* s = NULL;
int ok = 0;
if (sqlite3_prepare_v2(db, "SELECT name FROM sqlite_master WHERE type='index' AND name=?", -1, &s, NULL) == SQLITE_OK) {
sqlite3_bind_text(s, 1, name, -1, SQLITE_STATIC);
ok = (sqlite3_step(s) == SQLITE_ROW);
sqlite3_finalize(s);
}
return ok;
}
static int count_rows(sqlite3* db, const char* table) {
char sql[128]; snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM %s", table);
sqlite3_stmt* s = NULL;
int n = -1;
if (sqlite3_prepare_v2(db, sql, -1, &s, NULL) == SQLITE_OK) {
if (sqlite3_step(s) == SQLITE_ROW) n = sqlite3_column_int(s, 0);
sqlite3_finalize(s);
}
return n;
}
/* ── test cases ── */
static void test_tables(void) {
TEST("create tables exist"); {
sqlite3* db = make_db();
if (create_tables(db) != 0) { FAIL("create_tables failed"); sqlite3_close(db); return; }
int ba = table_exists(db, "block_availability");
int ss = table_exists(db, "super_sync");
if (ba && ss) OK(); else FAIL("ba=%d ss=%d", ba, ss);
sqlite3_close(db);
}
TEST("create tables idempotent"); {
sqlite3* db = make_db();
if (create_tables(db) != 0) { FAIL("first create"); sqlite3_close(db); return; }
if (create_tables(db) != 0) { FAIL("second create"); sqlite3_close(db); return; }
OK();
sqlite3_close(db);
}
TEST("indices exist"); {
sqlite3* db = make_db();
create_tables(db);
int i1 = index_exists(db, "idx_ba_uuid_node");
int i2 = index_exists(db, "idx_ba_id");
int i3 = index_exists(db, "idx_ba_node_id");
if (i1 && i2 && i3) OK(); else FAIL("i1=%d i2=%d i3=%d", i1, i2, i3);
sqlite3_close(db);
}
}
static void test_ba_insert(void) {
TEST("ba_insert basic"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; make_uuid(uuid);
uint64_t gid = 0x100, nid = 0x200;
const char* sql = "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp) VALUES(?,?,?,?,?)";
for (int i = 0; i < 3; i++) {
sqlite3_stmt* s = NULL;
sqlite3_prepare_v2(db, sql, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_bind_int64(s, 2, (sqlite3_int64)gid);
sqlite3_bind_int64(s, 3, (sqlite3_int64)(nid + (uint64_t)i));
sqlite3_bind_int(s, 4, i);
sqlite3_bind_int64(s, 5, (sqlite3_int64)time(NULL));
sqlite3_step(s); sqlite3_finalize(s);
}
int n = count_rows(db, "block_availability");
if (n == 3) OK(); else FAIL("expected 3 rows got %d", n);
sqlite3_close(db);
}
TEST("ba_insert upsert (unique uuid+node)"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; make_uuid(uuid);
const char* sql = "INSERT OR REPLACE INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp) VALUES(?,?,?,?,?)";
for (int pass = 0; pass < 2; pass++) {
sqlite3_stmt* s = NULL;
sqlite3_prepare_v2(db, sql, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_bind_int64(s, 2, (sqlite3_int64)1);
sqlite3_bind_int64(s, 3, (sqlite3_int64)0x300);
sqlite3_bind_int(s, 4, 0);
sqlite3_bind_int64(s, 5, (sqlite3_int64)(pass + 100));
sqlite3_step(s); sqlite3_finalize(s);
}
int n = count_rows(db, "block_availability");
if (n == 1) OK(); else FAIL("expected 1 row (upsert) got %d", n);
sqlite3_close(db);
}
}
static void test_ba_find(void) {
TEST("ba_find nodes by uuid"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; make_uuid(uuid);
const char* ins = "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp) VALUES(?,?,?,?,?)";
for (int i = 0; i < 5; i++) {
sqlite3_stmt* s = NULL; sqlite3_prepare_v2(db, ins, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_bind_int64(s, 2, (sqlite3_int64)1);
sqlite3_bind_int64(s, 3, (sqlite3_int64)(0x10 + i));
sqlite3_bind_int(s, 4, i);
sqlite3_bind_int64(s, 5, (sqlite3_int64)(2000 - i));
sqlite3_step(s); sqlite3_finalize(s);
}
const char* q = "SELECT node_id FROM block_availability WHERE block_uuid=? ORDER BY timestamp DESC LIMIT 3";
sqlite3_stmt* s = NULL;
sqlite3_prepare_v2(db, q, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
int count = 0, first_node = -1;
while (sqlite3_step(s) == SQLITE_ROW) { if (count == 0) first_node = sqlite3_column_int(s, 0); count++; }
sqlite3_finalize(s);
if (count == 3 && first_node == 0x10) OK(); else FAIL("count=%d first_node=0x%x", count, first_node);
sqlite3_close(db);
}
}
static void test_ba_get_since(void) {
TEST("ba_get_since id > N"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; make_uuid(uuid);
const char* ins = "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp) VALUES(?,?,?,?,?)";
for (int i = 0; i < 6; i++) {
sqlite3_stmt* s = NULL; sqlite3_prepare_v2(db, ins, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_bind_int64(s, 2, (sqlite3_int64)1);
sqlite3_bind_int64(s, 3, (sqlite3_int64)(0x20 + i));
sqlite3_bind_int(s, 4, i);
sqlite3_bind_int64(s, 5, (sqlite3_int64)1000);
sqlite3_step(s); sqlite3_finalize(s);
}
const char* q = "SELECT id FROM block_availability WHERE id > ? ORDER BY id ASC LIMIT 4";
sqlite3_stmt* s = NULL;
sqlite3_prepare_v2(db, q, -1, &s, NULL);
sqlite3_bind_int64(s, 1, 1);
int count = 0;
while (sqlite3_step(s) == SQLITE_ROW) count++;
sqlite3_finalize(s);
if (count == 4) OK(); else FAIL("expected 4 rows got %d", count);
sqlite3_close(db);
}
}
/* ── status column tests ── */
static void test_ba_status(void) {
/* processing→completed: DELETE old + INSERT new, ids don't reuse */
TEST("ba_complete: new id > processing id"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; for (int i = 0; i < 16; i++) uuid[i] = (uint8_t)(0xAA + i);
sqlite3_stmt* s = NULL;
/* 1. INSERT processing (status=0) */
sqlite3_prepare_v2(db, "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,1,0xAA,0,100,0)", -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_step(s); sqlite3_finalize(s);
/* 2. get processing id */
int64_t pid = -1;
sqlite3_prepare_v2(db, "SELECT id FROM block_availability WHERE node_id=0xAA AND status=0", -1, &s, NULL);
if (sqlite3_step(s) == SQLITE_ROW) pid = sqlite3_column_int64(s, 0);
sqlite3_finalize(s);
/* 3. DELETE processing + INSERT completed (new id) */
sqlite3_exec(db, "DELETE FROM block_availability WHERE block_uuid=x'AAABACADAEAFB0B1B2B3B4B5B6B7B8B9' AND node_id=0xAA AND status=0", NULL, NULL, NULL);
sqlite3_prepare_v2(db, "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,1,0xAA,0,200,1)", -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_step(s); sqlite3_finalize(s);
/* 4. get completed id */
int64_t cid = -1;
sqlite3_prepare_v2(db, "SELECT id,status FROM block_availability WHERE node_id=0xAA", -1, &s, NULL);
int st = -1;
if (sqlite3_step(s) == SQLITE_ROW) { cid = sqlite3_column_int64(s, 0); st = sqlite3_column_int(s, 1); }
sqlite3_finalize(s);
int n = count_rows(db, "block_availability");
if (n == 1 && st == 1 && cid > pid) OK(); else FAIL("rows=%d status=%d pid=%lld cid=%lld (expected 1 row, status=1, cid>pid)", n, st, (long long)pid, (long long)cid);
sqlite3_close(db);
}
/* completed already exists → no-op, no duplicate */
TEST("ba_complete: no-op when completed exists"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; for (int i = 0; i < 16; i++) uuid[i] = (uint8_t)(0xBB + i);
sqlite3_stmt* s = NULL;
/* 1. INSERT completed */
sqlite3_prepare_v2(db, "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,1,0xBB,0,300,1)", -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_step(s); sqlite3_finalize(s);
int64_t id1 = -1;
sqlite3_prepare_v2(db, "SELECT id FROM block_availability WHERE node_id=0xBB", -1, &s, NULL);
if (sqlite3_step(s) == SQLITE_ROW) id1 = sqlite3_column_int64(s, 0);
sqlite3_finalize(s);
/* 2. DELETE processing (none) + check completed exists + skip INSERT */
sqlite3_exec(db, "DELETE FROM block_availability WHERE block_uuid=x'BBBCBDBEBFC0C1C2C3C4C5C6C7C8C9' AND node_id=0xBB AND status=0", NULL, NULL, NULL);
int exists = 0;
sqlite3_prepare_v2(db, "SELECT 1 FROM block_availability WHERE block_uuid=? AND node_id=0xBB AND status=1", -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
exists = (sqlite3_step(s) == SQLITE_ROW);
sqlite3_finalize(s);
/* 3. Don't insert if exists */
int64_t id2 = -1;
sqlite3_prepare_v2(db, "SELECT id FROM block_availability WHERE node_id=0xBB", -1, &s, NULL);
if (sqlite3_step(s) == SQLITE_ROW) id2 = sqlite3_column_int64(s, 0);
sqlite3_finalize(s);
int n = count_rows(db, "block_availability");
if (n == 1 && exists && id2 == id1) OK(); else FAIL("rows=%d exists=%d id1=%lld id2=%lld (expected 1 row, same id)", n, exists, (long long)id1, (long long)id2);
sqlite3_close(db);
}
/* processing after completed INSERT → CONSTRAINT violation */
TEST("ba_processing after completed: constraint violation"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; for (int i = 0; i < 16; i++) uuid[i] = (uint8_t)(0xCC + i);
sqlite3_stmt* s = NULL;
/* 1. INSERT completed */
sqlite3_prepare_v2(db, "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,1,0xCC,0,400,1)", -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
sqlite3_step(s); sqlite3_finalize(s);
/* 2. try INSERT processing (same uuid+node) → must fail with unique constraint */
int rc = SQLITE_OK;
sqlite3_prepare_v2(db, "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,1,0xCC,0,500,0)", -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC);
rc = sqlite3_step(s);
sqlite3_finalize(s);
/* 3. completed is still there, unchanged */
int n = count_rows(db, "block_availability");
int64_t ts_check = -1;
sqlite3_prepare_v2(db, "SELECT timestamp FROM block_availability WHERE node_id=0xCC", -1, &s, NULL);
if (sqlite3_step(s) == SQLITE_ROW) ts_check = sqlite3_column_int64(s, 0);
sqlite3_finalize(s);
if (n == 1 && rc == SQLITE_CONSTRAINT && ts_check == 400) OK();
else FAIL("rows=%d rc=%d ts=%lld (expected 1 row, CONSTRAINT, ts=400)", n, rc, (long long)ts_check);
sqlite3_close(db);
}
}
static void test_ss_ops(void) {
TEST("super_sync insert+select"); {
sqlite3* db = make_db(); create_tables(db);
sqlite3_exec(db, "INSERT OR REPLACE INTO super_sync(peer_node_id,last_recv_id) VALUES(0xAA, 42)", NULL, NULL, NULL);
sqlite3_stmt* s = NULL;
sqlite3_prepare_v2(db, "SELECT last_recv_id FROM super_sync WHERE peer_node_id=?", -1, &s, NULL);
sqlite3_bind_int64(s, 1, 0xAA);
int v = -1;
if (sqlite3_step(s) == SQLITE_ROW) v = sqlite3_column_int(s, 0);
sqlite3_finalize(s);
if (v == 42) OK(); else FAIL("expected 42 got %d", v);
sqlite3_close(db);
}
TEST("super_sync default 0"); {
sqlite3* db = make_db(); create_tables(db);
sqlite3_stmt* s = NULL;
sqlite3_prepare_v2(db, "SELECT last_recv_id FROM super_sync WHERE peer_node_id=?", -1, &s, NULL);
sqlite3_bind_int64(s, 1, 0xCC);
int v = -1;
if (sqlite3_step(s) == SQLITE_ROW) v = sqlite3_column_int(s, 0);
else v = 0; /* no row = default */
sqlite3_finalize(s);
if (v == 0) OK(); else FAIL("expected 0 got %d", v);
sqlite3_close(db);
}
}
static void test_ba_delete(void) {
TEST("ba_delete by node_id"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; make_uuid(uuid);
const char* ins = "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp) VALUES(?,?,?,?,?)";
sqlite3_stmt* s = NULL;
sqlite3_prepare_v2(db, ins, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC); sqlite3_bind_int64(s, 2, 1);
sqlite3_bind_int64(s, 3, 0x500); sqlite3_bind_int(s, 4, 0);
sqlite3_bind_int64(s, 5, 100); sqlite3_step(s); sqlite3_finalize(s);
sqlite3_prepare_v2(db, ins, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC); sqlite3_bind_int64(s, 2, 1);
sqlite3_bind_int64(s, 3, 0x600); sqlite3_bind_int(s, 4, 0);
sqlite3_bind_int64(s, 5, 100); sqlite3_step(s); sqlite3_finalize(s);
if (count_rows(db, "block_availability") != 2) { FAIL("setup failed"); sqlite3_close(db); return; }
sqlite3_exec(db, "DELETE FROM block_availability WHERE node_id=0x500", NULL, NULL, NULL);
int n = count_rows(db, "block_availability");
if (n == 1) OK(); else FAIL("expected 1 got %d", n);
sqlite3_close(db);
}
TEST("ba_delete foreign (node_id != self)"); {
sqlite3* db = make_db(); create_tables(db);
uint8_t uuid[16]; make_uuid(uuid);
const char* ins = "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp) VALUES(?,?,?,?,?)";
sqlite3_stmt* s = NULL;
for (int i = 0; i < 3; i++) {
sqlite3_prepare_v2(db, ins, -1, &s, NULL);
sqlite3_bind_blob(s, 1, uuid, 16, SQLITE_STATIC); sqlite3_bind_int64(s, 2, 1);
sqlite3_bind_int64(s, 3, (sqlite3_int64)(0x700 + i)); sqlite3_bind_int(s, 4, i);
sqlite3_bind_int64(s, 5, 100); sqlite3_step(s); sqlite3_finalize(s);
}
char buf[128]; snprintf(buf, sizeof(buf), "DELETE FROM block_availability WHERE node_id!=0x700");
sqlite3_exec(db, buf, NULL, NULL, NULL);
int n = count_rows(db, "block_availability");
if (n == 1) OK(); else FAIL("expected 1 (self) got %d", n);
sqlite3_close(db);
}
}
static void test_proto_sizes(void) {
TEST("packed struct sizes"); {
int ok = 1;
if (sizeof(struct media_pkt_query) != 35) { FAIL("query: %zu != 35", sizeof(struct media_pkt_query)); ok = 0; }
if (sizeof(struct media_pkt_have_block) != 117) { FAIL("have_block: %zu != 117", sizeof(struct media_pkt_have_block)); ok = 0; }
if (sizeof(struct media_pkt_super_hello) != 17) { FAIL("super_hello: %zu != 17", sizeof(struct media_pkt_super_hello)); ok = 0; }
if (sizeof(struct media_pkt_super_ack) != 5) { FAIL("super_ack: %zu != 5", sizeof(struct media_pkt_super_ack)); ok = 0; }
if (sizeof(struct media_pkt_block_req) != 53) { FAIL("block_req: %zu != 53", sizeof(struct media_pkt_block_req)); ok = 0; }
if (sizeof(struct media_pkt_cancel) != 53) { FAIL("cancel: %zu != 53", sizeof(struct media_pkt_cancel)); ok = 0; }
if (sizeof(struct media_pkt_block_done) != 58) { FAIL("block_done: %zu != 58", sizeof(struct media_pkt_block_done)); ok = 0; }
if (sizeof(struct media_pkt_block_chunk) != 59) { FAIL("block_chunk: %zu != 59", sizeof(struct media_pkt_block_chunk)); ok = 0; }
if (sizeof(struct media_pkt_block_overloaded) != 57) { FAIL("overloaded: %zu != 57", sizeof(struct media_pkt_block_overloaded)); ok = 0; }
if (sizeof(struct media_pkt_block_relay_full) != 55) { FAIL("relay_full: %zu != 55", sizeof(struct media_pkt_block_relay_full)); ok = 0; }
if (sizeof(struct media_pkt_query_resp_entry) != 49) { FAIL("query_resp_entry: %zu != 49", sizeof(struct media_pkt_query_resp_entry)); ok = 0; }
if (sizeof(struct media_pkt_query_resp) != 36) { FAIL("query_resp: %zu != 36", sizeof(struct media_pkt_query_resp)); ok = 0; }
if (ok) OK();
/* continue running even if sizes mismatch */
}
}
static void test_proto_build(void) {
TEST("build MEDIA_QUERY"); {
uint8_t mid[16]; memset(mid, 0xAB, 16);
uint8_t bids[32]; memset(bids, 0xCD, 32);
uint8_t pkt[256];
struct media_pkt_query* q = (struct media_pkt_query*)pkt;
q->subcmd = MEDIA_SUBCMD_QUERY;
memset(q->request_id, 0xEF, 16);
memcpy(q->media_id, mid, 16);
q->num_blocks = 2;
memcpy(pkt + MEDIA_QUERY_HDR_SIZE, bids, 32);
if (q->subcmd == MEDIA_SUBCMD_QUERY && q->request_id[0] == 0xEF &&
q->num_blocks == 2 && memcmp(q->media_id, mid, 16) == 0 &&
memcmp(pkt + MEDIA_QUERY_HDR_SIZE, bids, 32) == 0) OK();
else FAIL();
}
TEST("build MEDIA_HAVE_BLOCK"); {
struct media_pkt_have_block hb;
memset(&hb, 0, sizeof(hb));
hb.subcmd = MEDIA_SUBCMD_HAVE_BLOCK;
hb.group_id = 0xAA;
hb.chunk = 5;
hb.timestamp = 999;
if (hb.subcmd == MEDIA_SUBCMD_HAVE_BLOCK && hb.group_id == 0xAA &&
hb.chunk == 5 && hb.timestamp == 999) OK();
else FAIL();
}
TEST("build MEDIA_SUPER_HELLO"); {
struct media_pkt_super_hello sh;
sh.subcmd = MEDIA_SUBCMD_SUPER_HELLO;
sh.last_recv_id = 12345;
sh.group_id = 0x11223344;
if (sh.subcmd == MEDIA_SUBCMD_SUPER_HELLO && sh.last_recv_id == 12345 && sh.group_id == 0x11223344) OK();
else FAIL();
}
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;
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
srand((unsigned)time(NULL));
printf("=== test_media_delivery_sql ===\n");
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);
test_tables();
test_ba_insert();
test_ba_find();
test_ba_get_since();
test_ba_status();
test_ss_ops();
test_ba_delete();
test_proto_sizes();
test_proto_build();
/* Настоящая регистрация остаётся целой при пересоздании вспомогательного каталога. */
REQUIRE(!media_index_set_base(a->md.db, "/tmp"));
uint8_t media[16] = {1}, block[16] = {2}, sig[64] = {0};
REQUIRE(!media_index_register_downloaded(a->md.db, media, block, "111", "media-repl-preserved.bin",
SERVER, sig, 1, 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);
printf("\n%d/%d passed, %d failed\n", g_passed, g_total, g_failed);
return g_failed > 0 ? 1 : 0;
/* Старая схема и её курсоры сбрасываются одной транзакцией, 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); uasync_destroy(ua, 0);
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "media catalog SQL restart/reset/preservation: PASS");
return 0;
}
tests/test_media_replication.c
+281 -11
View File
@@ -1,30 +1,300 @@
/* Репликация: настоящие обработчики media/SQLite, управляемый транспорт и ошибки. */
/* Репликация: настоящие обработчики media/SQLite, управляемый CM/транспорт и ошибки. */
#include "media_delivery_fixture.h"
#include "../src/routing_layer/topo_group.h"
#include "../src/routing_layer/conn_mgr.h"
static sqlite3_int64 scalar(sqlite3* db, const char* sql) {
sqlite3_stmt* st = NULL; REQUIRE(sqlite3_prepare_v2(db, sql, -1, &st, NULL) == SQLITE_OK);
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;
}
/* Группа сообщения задаётся router; другое значение внутри HAVE не даёт записать чужую группу. */
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 = instance(SERVER);
a->md.is_supernode = 1;
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 WHERE group_id=112") == 0);
destroy(a); reset();
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);
ua = uasync_create(); REQUIRE(ua);
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");
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "media replication regressions: PASS CM=%d/%d", cm_opened, cm_closed);
return 0;
}