Default stream redundancy to zero on Android and desktop

2026-10-21 12:42:23 +02:00
1 parent 34db81c5cf
commit 5870efdf04
@@ -8,7 +8,7 @@

 Принятые решения:
 - Транспорт — существующий ETCP с быстрыми переповторами, мультилинком, шифрованием и управлением нагрузкой.
- Параметры FEC задаёт источник; 6+2 — значение по умолчанию, а не ограничение протокола.
+- Параметры FEC задаёт источник; 8+0 — значение по умолчанию, а не ограничение протокола.
 - Источник подписывает всю пачку один раз до RS; подпись восстанавливается вместе с содержимым и проверяется после decode.
 - В источник добавляется циклический timestamp uint16_t в миллисекундах; он сохраняется при всей ретрансляции.
  Задержка оценивается относительно предыдущих данных по дельтам timestamp и времени прихода, включая полную остановку.
@@ -68,7 +68,7 @@ INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получа

 Источник формирует пачки и кодирует их lib/reed_solomon:
 - n — полное число субпотоков, r — число проверочных, k = n-r — число исходных.
- По умолчанию n=8, r=2, k=6. Любые k кодограмм одной пачки восстанавливают её содержимое.
+- По умолчанию n=8, r=0, k=8. Любые k кодограмм одной пачки восстанавливают её содержимое.
 - subblock от 0 до n-1 определяет субпоток; первые k блоков систематические, остальные проверочные.
 - batch_seq — общий 64-битный индекс пачки; блок субпотока X определяется парой (batch_seq, X).
  Позиция N в запросе — batch_seq, а не байтовое смещение или отдельный счётчик субпотока.
@@ -78,7 +78,7 @@ INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получа
 Дескриптор содержит ключ трансляции, n, r и максимальный размер RS-кодограммы block_size.
 Параметры задаёт только источник и не меняет в пределах source_epoch; изменение создаёт новый запуск.
 В настройках desktop/Android раздел «Трансляция» сохраняет общее число субблоков и число избыточных RS.
-В [chatserver] это stream_subblocks (1..255, по умолчанию 8) и stream_rs_redundant (0..n-1, по умолчанию 2).
+В [chatserver] это stream_subblocks (1..255, по умолчанию 8) и stream_rs_redundant (0..n-1, по умолчанию 0).
 Медиапубликатор читает параметры при запуске и один раз выделяет рабочий буфер по ёмкости пачки,
 но не больше предела медиаблока 65536 байт. Изменение настроек действует только со следующего запуска.
 Дескриптор подтверждается подписанной опцией источника; поля FEC также входят в подписанную пачку.
@@ -87,9 +87,9 @@ INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получа

 Максимальный размер части субпотока по умолчанию — 1300 байт: внешние метаданные 47 и RS до 1253 байт.
 Внутри RS находятся подписанные метаданные 55 байт, содержимое и одна 64-байтная подпись.
-block_size включает 9-байтный заголовок RS seq/subblock. Максимальная длина содержимого при 6+2 — 7345 байт.
+block_size включает 9-байтный заголовок RS seq/subblock. Максимальная длина содержимого при 8+0 — 9833 байт.
 При длине содержимого L фактический RS-size = 9 + ceil((55 + L + 64)/k), полный размер части = 47 + RS-size.
-Например, L=1000 даёт восемь частей по 243 байта; L=7345 даёт восемь частей по 1300 байт.
+Например, при 8+0 L=1000 даёт восемь частей по 196 байт; L=9833 даёт восемь частей по 1300 байт.
 Один контекст RS принимает пачки разной длины без пересоздания матрицы; все части одной пачки равны по размеру.
 Handle получателя, поколение stream-сессии, изменяемый путь и служебные заголовки ETCP идут сверх этого бюджета.
 Не обещаем соответствия одной кодограммы одному UDP-пакету: необходимую фрагментацию выполняет ETCP normalizer.
@@ -743,7 +743,7 @@ Restart в той же source_epoch сохраняет consumer cursor: bootstra
 смешивание отвергается до изменения данных. stream_publish_finish не освобождает буфер источника.

 Медиаблок uAV2 находится целиком внутри входа RS, поэтому header восстанавливается вместе с данными.
-При штатных 6+2 полный медиаблок <=7345 байт. Раскладка big-endian, без packed-структур:
+При штатных 8+0 полный медиаблок <=9833 байт. Раскладка big-endian, без packed-структур:

 fixed header (32 байта):
  magic="uAV2":4, video_codec:1, audio_codec:1, video_count:1, audio_count:1,
@@ -36,7 +36,7 @@ struct chat_setting_def {

 static const struct chat_setting_def g_setting_defs[] = {
    {"stream_subblocks",           CHAT_SETTING_INT,  8,   1, 255},
-    {"stream_rs_redundant",        CHAT_SETTING_INT,  2,   0, 254},
+    {"stream_rs_redundant",        CHAT_SETTING_INT,  0,   0, 254},
    {"storage_autoload",            CHAT_SETTING_BOOL, 1,   0, 1},
    {"opus_codec_preset",           CHAT_SETTING_INT,  1,   0, 4},
    {"radio_opus_mode",             CHAT_SETTING_INT,  0,   0, 1},
@@ -32,7 +32,7 @@
 #define STREAM_BLOCK_DEFAULT_SIZE 1300
 #define STREAM_RS_DEFAULT_SIZE (STREAM_BLOCK_DEFAULT_SIZE - STREAM_BLOCK_OVERHEAD)
 #define STREAM_DEFAULT_N 8
-#define STREAM_DEFAULT_R 2
+#define STREAM_DEFAULT_R 0
 #define STREAM_DEFAULT_INPUT_CAPACITY \
    ((STREAM_DEFAULT_N - STREAM_DEFAULT_R) * (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD)
 #define STREAM_BLOCK_MAX_SIZE (RS_MAX_BLOCK_SIZE + STREAM_BLOCK_OVERHEAD)
@@ -69,7 +69,7 @@ int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
    size_t before = u_get_allocated_count(); test_clock_init(100); ua = uasync_create(); CHECK(ua);
    struct UTUN_INSTANCE *inst = utun_instance_create_from_str(ua,
-        "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstream_subblocks=8\nstream_rs_redundant=0\nstorage_autoload=0\nmember_files_enabled=0\n");
+        "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n");
    CHECK(inst && !utun_core_start(inst) && !chat_service_start(inst));
    chat_core_create_channel_auto(inst, "Stream \"test\" 😀");
    sqlite3_stmt *statement = NULL;
@@ -20,6 +20,8 @@ static const struct stream_descriptor descriptor = {
    .group_id = UINT64_C(0x0123456789abcdef), .source_node_id = UINT64_C(0xfedcba9876543210),
    .source_epoch = UINT64_C(0x0102030405060708), .block_size = STREAM_RS_DEFAULT_SIZE, .n = 8, .r = 2
 };
+/* Стенд потерь использует 6+2 независимо от настроек публикации по умолчанию. */
+#define TEST_INPUT_CAPACITY (6 * (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD)
 static uint8_t private_key[32];
 static unsigned cache_added, cache_removed;
 static void cache_observer(void *arg, const uint8_t *wire, size_t length, int added) {
@@ -75,8 +77,8 @@ static void pattern(uint8_t *data, size_t length) {
 static void test_signed_blocks(void) {
    struct stream_encoder *encoder = NULL;
    CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
-    CHECK(stream_block_capacity(&descriptor) == 1300 && stream_input_size(&descriptor) == STREAM_DEFAULT_INPUT_CAPACITY);
-    uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input));
+    CHECK(stream_block_capacity(&descriptor) == 1300 && stream_input_size(&descriptor) == TEST_INPUT_CAPACITY);
+    uint8_t input[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input));
    const uint8_t *blocks[RS_MAX_BLOCKS];
    uint64_t seq = UINT64_C(0x1122334455667788);
    size_t size = 0;
@@ -97,7 +99,7 @@ static void test_signed_blocks(void) {
    }
    struct rs_codec *rs = NULL;
    CHECK(rs_init(&rs, descriptor.n, descriptor.r, descriptor.block_size) == RS_OK);
-    uint8_t decoded[STREAM_DEFAULT_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD];
+    uint8_t decoded[TEST_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD];
    CHECK(rs_decode(rs, received, descriptor.block_size, decoded, sizeof(decoded)) == RS_OK);
    CHECK(!memcmp(decoded, STREAM_BATCH_DOMAIN, 12));
    CHECK(!memcmp(decoded + STREAM_BATCH_HEADER, input, sizeof(input)));
@@ -184,7 +186,7 @@ static void test_wire(void) {
 static void test_fec_and_padding(void) {
    struct stream_encoder *encoder = NULL;
    CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
-    uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input));
+    uint8_t input[TEST_INPUT_CAPACITY], output[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input));
    const uint8_t *blocks[RS_MAX_BLOCKS];
    size_t size = 0;
    CHECK(stream_encoder_encode(encoder, 0, 15, 0, input, sizeof(input), blocks, &size) == STREAM_OK && size == 1300);
@@ -241,7 +243,7 @@ static void test_variable_batches(void) {
    struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL;
    CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
    CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 64 * 1300) == STREAM_OK);
-    uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input));
+    uint8_t input[TEST_INPUT_CAPACITY], output[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input));
    const size_t lengths[] = {1, 7, sizeof(input), 17, 1000, 6, 2};
    size_t sizes[sizeof(lengths) / sizeof(lengths[0])], stored = 0;
    const uint8_t *blocks[RS_MAX_BLOCKS];
@@ -313,7 +315,7 @@ static void test_rejected_batches(void) {
    CHECK(stream_encoder_encode(encoder, 20, 42, 0, input, 17, blocks, &size) == STREAM_OK);
    const uint8_t *received[RS_MAX_BLOCKS] = {0};
    for (unsigned sub = 0; sub < 8; sub++) received[sub] = blocks[sub] + STREAM_BLOCK_RS_OFF;
-    uint8_t decoded[STREAM_DEFAULT_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD];
+    uint8_t decoded[TEST_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD];
    size_t rs_size = size - STREAM_BLOCK_OVERHEAD, decoded_size = 6 * (rs_size - RS_HEADER_SIZE);
    CHECK(rs_decode(rs, received, rs_size, decoded, sizeof(decoded)) == RS_OK);
    struct stream_block_view view; CHECK(stream_block_parse(&descriptor, blocks[0], size, &view) == STREAM_OK);
@@ -463,7 +465,7 @@ static void test_source_timer(void) {
    struct UASYNC *ua = uasync_create(); CHECK(ua != NULL);
    struct stream_source *source = NULL; struct source_events events = {0};
    CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK);
-    uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input));
+    uint8_t input[TEST_INPUT_CAPACITY], output[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input));
    CHECK(stream_source_input_capacity(source) == sizeof(input));
    size_t accepted = 999;
    CHECK(stream_source_write(source, NULL, 0, &accepted) == STREAM_OK && accepted == 0);
@@ -27,6 +27,7 @@
 #define GROUP1 UINT64_C(710001)
 #define GROUP2 UINT64_C(710002)
 #define SINGLE_INPUT_CAPACITY (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE - STREAM_BATCH_OVERHEAD)
+#define PRESSURE_INPUT_CAPACITY (6 * (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD)
 #define CHECK(expr) do { if (!(expr)) { \
    DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_service failed line=%d: %s", __LINE__, #expr); exit(1); \
 } } while (0)
@@ -46,7 +47,7 @@ static etcp_recv_fn saved_raw[4], saved_route;
 static int drop_renew, gap_batch;
 static unsigned pressure_data, overload_seen, selects_seen;
 static int ignore_select;
-static uint8_t pressure_payload[STREAM_DEFAULT_INPUT_CAPACITY], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER];
+static uint8_t pressure_payload[PRESSURE_INPUT_CAPACITY], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER];
 static size_t stale_select_length;
 static etcp_recv_fn pressure_handlers[4];
 static int simulated_cpu;
@@ -501,7 +502,8 @@ static void test_network(void) {
        CHECK(member_sync_start(nodes[i], nodes[3]->node_id, "710002", NULL, NULL) == 0);
    }
    struct STREAM_HANDLE *publisher = NULL, *subscribers[3] = {0};
-    CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK);
+    struct stream_parameters fec_parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE};
+    CHECK(stream_publish_open(nodes[3], GROUP2, &fec_parameters, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    /* Подписанный END отдельно отвергает подмену подписи, индекса и группы. */
    uint8_t end[STREAM_END_SIZE]; uint64_t last;
@@ -675,7 +677,8 @@ static void test_network(void) {
 static void test_pressure(void) {
    memset(pressure_payload, 42, sizeof(pressure_payload));
    struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL;
-    CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK);
+    struct stream_parameters parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE};
+    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    wait_state(1, 3, GROUP2, origin.descriptor.source_epoch, 0);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &subscriber) == STREAM_OK);
@@ -1560,7 +1563,7 @@ static void lifecycle_event(struct UTUN_INSTANCE *inst, const struct stream_even
    else CHECK(event->type == STREAM_EVENT_CLOSED);
 }

-/* Первый поставщик имеет лишь 3 из 6 нужных частей; пауза источника не оправдывает зависший bootstrap. */
+/* Первый поставщик имеет лишь 3 части пачки; пауза источника не оправдывает зависший bootstrap. */
 static void test_partial_bootstrap_and_end(void) {
    struct STREAM_HANDLE *publisher = NULL, *relay = NULL, *viewer = NULL;
    CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
@@ -1755,7 +1758,8 @@ int main(int argc, char **argv) {
    test_profile_atomicity();
    stream_set_event_callback(nodes[0], on_stream, NULL); stream_set_event_callback(nodes[1], on_stream, NULL);
    struct STREAM_HANDLE *publisher = NULL, *publisher2 = NULL, *other_group = NULL, *mobile = NULL, *desktop = NULL;
-    CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
+    struct stream_parameters fec_parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE};
+    CHECK(stream_publish_open(nodes[0], GROUP1, &fec_parameters, &publisher) == STREAM_OK);
    CHECK(stream_publish_open(nodes[0], GROUP2, NULL, &publisher2) == STREAM_OK);
    struct stream_info info; CHECK(stream_get_info(publisher, &info) == STREAM_OK);
    uint64_t epoch = info.descriptor.source_epoch;
@@ -233,7 +233,7 @@ class ConfigProvider(private val context: Context) {
            key == "chat.storage_maxsize_gb" -> context.dataStore.data.first()[chatStorageMaxsizeGb] ?: 1
            key == "chat.opus_codec_preset" -> context.dataStore.data.first()[chatOpusCodecPreset] ?: 1
            key == "chat.stream_subblocks" -> context.dataStore.data.first()[streamSubblocks] ?: 8
-            key == "chat.stream_rs_redundant" -> context.dataStore.data.first()[streamRsRedundant] ?: 2
+            key == "chat.stream_rs_redundant" -> context.dataStore.data.first()[streamRsRedundant] ?: 0
            key == "chat.call_jitter_max_reserve_ms" -> context.dataStore.data.first()[callJitterMaxReserveMs] ?: 1000
            key == "chat.compressor_enabled" -> context.dataStore.data.first()[chatCompressorEnabled] ?: 1
            key == "chat.compressor_max_gain_db" -> context.dataStore.data.first()[chatCompressorMaxGainDb] ?: 15
@@ -245,7 +245,7 @@ SettingsDialog::SettingsDialog(const QString& configPath, DbManager* db, UtunNod
    QSettings streamConfig(m_configPath, QSettings::IniFormat);
    m_streamSubblocks->setValue(streamConfig.value("chatserver/stream_subblocks", 8).toInt());
    m_streamRedundant->setRange(0, m_streamSubblocks->value() - 1);
-    m_streamRedundant->setValue(streamConfig.value("chatserver/stream_rs_redundant", 2).toInt());
+    m_streamRedundant->setValue(streamConfig.value("chatserver/stream_rs_redundant", 0).toInt());
    connect(m_streamSubblocks, QOverload<int>::of(&QSpinBox::valueChanged), this, [this](int n) {
        m_streamRedundant->setMaximum(n - 1);
    });