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

Sign stream batches before RS and authenticate reconstructed data

stream-multiview
evgeny 1 week ago
parent
7c8470f44a
commit
cee5b26c13
2 changed files with 181 additions and 100 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 164
      src/stream/stream_block.c
  2. 117
      src/stream/stream_buffer.c

164
src/stream/stream_block.c
Unescape View File

@ -19,7 +19,8 @@ struct stream_encoder {
/* Проверить ключ трансляции и параметры до расчёта размеров/выделения памяти. */
int stream_descriptor_validate(const struct stream_descriptor *d) {
if (!d || !d->group_id || !d->source_node_id || !d->source_epoch || !d->n || d->r >= d->n
|| d->block_size <= RS_HEADER_SIZE || d->block_size > RS_MAX_BLOCK_SIZE) {
|| d->block_size <= RS_HEADER_SIZE || d->block_size > RS_MAX_BLOCK_SIZE
|| (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) <= STREAM_BATCH_OVERHEAD) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "descriptor invalid: present=%d group=%llx source=%llx epoch=%llx n=%u r=%u size=%u",
d != NULL, d ? (unsigned long long)d->group_id : 0, d ? (unsigned long long)d->source_node_id : 0,
d ? (unsigned long long)d->source_epoch : 0, d ? d->n : 0, d ? d->r : 0, d ? d->block_size : 0);
@ -28,12 +29,13 @@ int stream_descriptor_validate(const struct stream_descriptor *d) {
return STREAM_OK;
}
size_t stream_block_size(const struct stream_descriptor *d) {
size_t stream_block_capacity(const struct stream_descriptor *d) {
return stream_descriptor_validate(d) == STREAM_OK ? d->block_size + STREAM_BLOCK_OVERHEAD : 0;
}
size_t stream_input_size(const struct stream_descriptor *d) {
return stream_descriptor_validate(d) == STREAM_OK ? (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) : 0;
return stream_descriptor_validate(d) == STREAM_OK
? (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD : 0;
}
/* Создать рабочую память RS и получить публичный ключ именно переданного Ed25519 seed. */
@ -64,8 +66,8 @@ int stream_encoder_create(struct stream_encoder **out, const struct stream_descr
stream_encoder_destroy(encoder);
return STREAM_ERR_FEC;
}
encoder->input = u_malloc((uint32_t)stream_input_size(d));
encoder->wire = u_malloc((uint32_t)(d->n * stream_block_size(d)));
encoder->input = u_malloc((uint32_t)rs_input_capacity(encoder->rs));
encoder->wire = u_malloc((uint32_t)(d->n * stream_block_capacity(d)));
if (!encoder->input || !encoder->wire) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder working memory allocation failed n=%u size=%u", d->n, d->block_size);
stream_encoder_destroy(encoder);
@ -74,7 +76,7 @@ int stream_encoder_create(struct stream_encoder **out, const struct stream_descr
*out = encoder;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder created group=%llx source=%llx epoch=%llx FEC=%u+%u block=%zu input=%zu",
(unsigned long long)d->group_id, (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch,
d->n - d->r, d->r, stream_block_size(d), stream_input_size(d));
d->n - d->r, d->r, stream_block_capacity(d), stream_input_size(d));
return STREAM_OK;
}
@ -84,7 +86,7 @@ void stream_encoder_destroy(struct stream_encoder *encoder) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder destroy source=%llx epoch=%llx",
(unsigned long long)encoder->descriptor.source_node_id, (unsigned long long)encoder->descriptor.source_epoch);
OPENSSL_cleanse(encoder->private_key, sizeof(encoder->private_key));
if (encoder->input) OPENSSL_cleanse(encoder->input, stream_input_size(&encoder->descriptor));
if (encoder->input) OPENSSL_cleanse(encoder->input, rs_input_capacity(encoder->rs));
u_free(encoder->input);
u_free(encoder->wire);
rs_free(encoder->rs);
@ -96,85 +98,137 @@ const uint8_t *stream_encoder_public_key(const struct stream_encoder *encoder) {
return encoder->public_key;
}
/* Сформировать канонический заголовок, RS-кодограммы и подпись каждого блока. */
/* Подписать метаданные и содержимое один раз, затем защитить всё сообщение кодированием RS. */
int stream_encoder_encode(struct stream_encoder *encoder, uint64_t seq, uint16_t timestamp_ms, uint8_t flags,
const uint8_t *input, size_t length, const uint8_t **blocks) {
if (!encoder || !input || !blocks || !length || seq == UINT64_MAX || (flags & ~STREAM_BLOCK_TIME_RESET)
const uint8_t *input, size_t length, const uint8_t **blocks, size_t *block_length) {
if (!encoder || !input || !blocks || !block_length || !length || seq == UINT64_MAX || (flags & ~STREAM_BLOCK_TIME_RESET)
|| length > stream_input_size(&encoder->descriptor)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode invalid arguments seq=%llu len=%zu flags=%u", (unsigned long long)seq, length, flags);
return STREAM_ERR_ARGUMENT;
}
const struct stream_descriptor *d = &encoder->descriptor;
size_t wire_size = stream_block_size(d);
size_t signed_size = STREAM_BATCH_HEADER + length, batch_size = signed_size + SC_SIGN_SIZE;
size_t rs_size = RS_HEADER_SIZE + (batch_size + d->n - d->r - 1) / (d->n - d->r);
size_t wire_size = STREAM_BLOCK_OVERHEAD + rs_size, stride = stream_block_capacity(d);
uint8_t *rs_blocks[RS_MAX_BLOCKS];
memcpy(encoder->input, input, length);
memset(encoder->input + length, 0, stream_input_size(d) - length);
uint8_t *batch = encoder->input;
memcpy(batch, STREAM_BATCH_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE);
stream_write_be(batch + STREAM_BLOCK_GROUP_OFF, d->group_id, 8);
stream_write_be(batch + STREAM_BLOCK_SOURCE_OFF, d->source_node_id, 8);
stream_write_be(batch + STREAM_BLOCK_EPOCH_OFF, d->source_epoch, 8);
batch[STREAM_BLOCK_N_OFF] = d->n; batch[STREAM_BLOCK_R_OFF] = d->r;
stream_write_be(batch + STREAM_BLOCK_SIZE_OFF, rs_size, 2);
stream_write_be(batch + STREAM_BLOCK_TIMESTAMP_OFF, timestamp_ms, 2);
stream_write_be(batch + STREAM_BLOCK_LENGTH_OFF, length, 4);
batch[STREAM_BLOCK_FLAGS_OFF] = flags;
stream_write_be(batch + STREAM_BATCH_SEQ_OFF, seq, 8);
memcpy(batch + STREAM_BATCH_HEADER, input, length);
if (sc_ed25519_sign(encoder->private_key, batch, signed_size, batch + signed_size) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode batch signature failed source=%llx seq=%llu length=%zu",
(unsigned long long)d->source_node_id, (unsigned long long)seq, length);
return STREAM_ERR_CRYPTO;
}
for (unsigned i = 0; i < d->n; i++) {
uint8_t *wire = encoder->wire + i * wire_size;
uint8_t *wire = encoder->wire + i * stride;
memcpy(wire, batch, STREAM_BLOCK_RS_OFF);
memcpy(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE);
stream_write_be(wire + STREAM_BLOCK_GROUP_OFF, d->group_id, 8);
stream_write_be(wire + STREAM_BLOCK_SOURCE_OFF, d->source_node_id, 8);
stream_write_be(wire + STREAM_BLOCK_EPOCH_OFF, d->source_epoch, 8);
wire[STREAM_BLOCK_N_OFF] = d->n;
wire[STREAM_BLOCK_R_OFF] = d->r;
stream_write_be(wire + STREAM_BLOCK_SIZE_OFF, d->block_size, 2);
stream_write_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, timestamp_ms, 2);
stream_write_be(wire + STREAM_BLOCK_LENGTH_OFF, length, 4);
wire[STREAM_BLOCK_FLAGS_OFF] = flags;
rs_blocks[i] = wire + STREAM_BLOCK_RS_OFF;
}
if (rs_encode(encoder->rs, seq, encoder->input, rs_blocks) != RS_OK) {
size_t encoded_size = 0;
if (rs_encode(encoder->rs, seq, batch, batch_size, rs_blocks, &encoded_size) != RS_OK || encoded_size != rs_size) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode RS failed seq=%llu", (unsigned long long)seq);
return STREAM_ERR_FEC;
}
for (unsigned i = 0; i < d->n; i++) {
uint8_t *wire = encoder->wire + i * wire_size;
if (sc_ed25519_sign(encoder->private_key, wire, wire_size - SC_SIGN_SIZE, wire + wire_size - SC_SIGN_SIZE) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode signature failed seq=%llu sub=%u", (unsigned long long)seq, i);
return STREAM_ERR_CRYPTO;
}
}
for (unsigned i = 0; i < d->n; i++) blocks[i] = encoder->wire + i * wire_size;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "encoded seq=%llu timestamp=%u len=%zu n=%u flags=%u",
(unsigned long long)seq, timestamp_ms, length, d->n, flags);
for (unsigned i = 0; i < d->n; i++) blocks[i] = encoder->wire + i * stride;
*block_length = wire_size;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "encoded signed batch seq=%llu timestamp=%u len=%zu n=%u flags=%u part=%zu max=%zu padding=%zu",
(unsigned long long)seq, timestamp_ms, length, d->n, flags, wire_size, stride,
(size_t)(d->n - d->r) * (rs_size - RS_HEADER_SIZE) - batch_size);
return STREAM_OK;
}
/* Проверить размеры, область подписи, дескриптор, метаданные и подпись до публикации view. */
int stream_block_verify(const struct stream_descriptor *d, const uint8_t public_key[32], const uint8_t *wire,
size_t length, struct stream_block_view *view) {
if (!public_key || !wire || !view || stream_descriptor_validate(d) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "verify invalid arguments length=%zu", length);
/* Только структурная проверка до хранения/пересылки; подлинность определяет восстановленная пачка. */
int stream_block_parse(const struct stream_descriptor *d, const uint8_t *wire, size_t length, struct stream_block_view *view) {
if (!wire || !view || stream_descriptor_validate(d) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "part parse invalid arguments length=%zu", length);
return STREAM_ERR_ARGUMENT;
}
if (length != stream_block_size(d) || memcmp(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE)
if (length <= STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE || length > stream_block_capacity(d)
|| memcmp(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE)
|| stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) != d->group_id
|| stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8) != d->source_node_id
|| stream_read_be(wire + STREAM_BLOCK_EPOCH_OFF, 8) != d->source_epoch
|| wire[STREAM_BLOCK_N_OFF] != d->n || wire[STREAM_BLOCK_R_OFF] != d->r
|| stream_read_be(wire + STREAM_BLOCK_SIZE_OFF, 2) != d->block_size) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify format/descriptor mismatch source=%llx epoch=%llx length=%zu expected=%zu",
(unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, length, stream_block_size(d));
|| stream_read_be(wire + STREAM_BLOCK_SIZE_OFF, 2) != length - STREAM_BLOCK_OVERHEAD) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "part format/descriptor mismatch source=%llx epoch=%llx length=%zu max=%zu",
(unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, length, stream_block_capacity(d));
return STREAM_ERR_FORMAT;
}
struct stream_block_view parsed = {
.seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8),
.data_length = (uint32_t)stream_read_be(wire + STREAM_BLOCK_LENGTH_OFF, 4),
.timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
.subblock = wire[STREAM_BLOCK_SUB_OFF], .flags = wire[STREAM_BLOCK_FLAGS_OFF], .rs = wire + STREAM_BLOCK_RS_OFF
.batch = {.seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8),
.data_length = (uint32_t)stream_read_be(wire + STREAM_BLOCK_LENGTH_OFF, 4),
.timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
.block_size = (uint16_t)(length - STREAM_BLOCK_OVERHEAD), .flags = wire[STREAM_BLOCK_FLAGS_OFF]},
.subblock = wire[STREAM_BLOCK_SUB_OFF], .rs = wire + STREAM_BLOCK_RS_OFF
};
if (parsed.seq == UINT64_MAX || parsed.subblock >= d->n || !parsed.data_length
|| parsed.data_length > stream_input_size(d) || (parsed.flags & ~STREAM_BLOCK_TIME_RESET)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify invalid metadata seq=%llu sub=%u length=%u flags=%u",
(unsigned long long)parsed.seq, parsed.subblock, parsed.data_length, parsed.flags);
const struct stream_batch_info *info = &parsed.batch;
if (info->seq == UINT64_MAX || parsed.subblock >= d->n || !info->data_length || info->data_length > stream_input_size(d)
|| (info->flags & ~STREAM_BLOCK_TIME_RESET)
|| info->block_size != RS_HEADER_SIZE + (STREAM_BATCH_OVERHEAD + info->data_length + d->n - d->r - 1) / (d->n - d->r)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "part invalid metadata seq=%llu sub=%u length=%u flags=%u size=%u",
(unsigned long long)info->seq, parsed.subblock, info->data_length, info->flags, info->block_size);
return STREAM_ERR_FORMAT;
}
*view = parsed;
return STREAM_OK;
}
/* Подпись проверяется после RS; внешние подсказки обязаны совпадать с подписанными полями пачки. */
int stream_batch_verify(const struct stream_descriptor *d, const uint8_t public_key[32],
const struct stream_batch_info *expected, const uint8_t *decoded, size_t decoded_length) {
if (!public_key || !expected || !decoded || stream_descriptor_validate(d) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "batch verify invalid arguments length=%zu", decoded_length);
return STREAM_ERR_ARGUMENT;
}
if (expected->seq == UINT64_MAX || !expected->data_length || expected->data_length > stream_input_size(d)
|| (expected->flags & ~STREAM_BLOCK_TIME_RESET) || expected->block_size > d->block_size
|| expected->block_size != RS_HEADER_SIZE + (STREAM_BATCH_OVERHEAD + expected->data_length + d->n - d->r - 1) / (d->n - d->r)
|| decoded_length != (size_t)(d->n - d->r) * (expected->block_size - RS_HEADER_SIZE)
|| decoded_length < STREAM_BATCH_OVERHEAD + expected->data_length) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch invalid size source=%llx seq=%llu decoded=%zu data=%u part=%u",
(unsigned long long)d->source_node_id, (unsigned long long)expected->seq,
decoded_length, expected->data_length, expected->block_size);
return STREAM_ERR_FORMAT;
}
if (sc_ed25519_verify(public_key, wire, length - SC_SIGN_SIZE, wire + length - SC_SIGN_SIZE) != SC_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify source signature failed source=%llx epoch=%llx seq=%llu sub=%u",
size_t signed_size = STREAM_BATCH_HEADER + expected->data_length;
if (sc_ed25519_verify(public_key, decoded, signed_size, decoded + signed_size) != SC_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch source signature rejected source=%llx epoch=%llx seq=%llu data=%u part=%u",
(unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch,
(unsigned long long)parsed.seq, parsed.subblock);
(unsigned long long)expected->seq, expected->data_length, expected->block_size);
return STREAM_ERR_SIGNATURE;
}
*view = parsed;
if (memcmp(decoded, STREAM_BATCH_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE)
|| stream_read_be(decoded + STREAM_BLOCK_GROUP_OFF, 8) != d->group_id
|| stream_read_be(decoded + STREAM_BLOCK_SOURCE_OFF, 8) != d->source_node_id
|| stream_read_be(decoded + STREAM_BLOCK_EPOCH_OFF, 8) != d->source_epoch
|| decoded[STREAM_BLOCK_N_OFF] != d->n || decoded[STREAM_BLOCK_R_OFF] != d->r
|| stream_read_be(decoded + STREAM_BLOCK_SIZE_OFF, 2) != expected->block_size
|| stream_read_be(decoded + STREAM_BLOCK_TIMESTAMP_OFF, 2) != expected->timestamp_ms
|| stream_read_be(decoded + STREAM_BLOCK_LENGTH_OFF, 4) != expected->data_length
|| decoded[STREAM_BLOCK_FLAGS_OFF] != expected->flags
|| stream_read_be(decoded + STREAM_BATCH_SEQ_OFF, 8) != expected->seq) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch signed metadata mismatch source=%llx epoch=%llx seq=%llu",
(unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, (unsigned long long)expected->seq);
return STREAM_ERR_FORMAT;
}
for (size_t i = signed_size + SC_SIGN_SIZE; i < decoded_length; i++) {
if (decoded[i]) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch nonzero padding source=%llx seq=%llu offset=%zu",
(unsigned long long)d->source_node_id, (unsigned long long)expected->seq, i);
return STREAM_ERR_FORMAT;
}
}
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "batch source signature verified source=%llx seq=%llu length=%u part=%u",
(unsigned long long)d->source_node_id, (unsigned long long)expected->seq, expected->data_length, expected->block_size);
return STREAM_OK;
}

117
src/stream/stream_buffer.c
Unescape View File

@ -23,8 +23,10 @@ struct stream_cached_batch {
uint32_t first_slot;
uint32_t data_length;
uint16_t timestamp_ms;
uint16_t block_size;
uint16_t count;
uint8_t flags;
int rejected;
};
struct stream_buffer {
@ -114,7 +116,7 @@ static void buffer_evict_oldest(struct stream_buffer *b) {
uint32_t before = b->slots[slot].sub_previous, after = b->slots[slot].sub_next;
if (before == STREAM_SLOT_NONE) b->sub_head[sub] = after; else b->slots[before].sub_next = after;
if (after == STREAM_SLOT_NONE) b->sub_tail[sub] = before; else b->slots[after].sub_previous = before;
if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, b->block_bytes, 0);
if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, STREAM_BLOCK_OVERHEAD + batch->block_size, 0);
u_free(b->slots[slot].path); b->slots[slot].path = NULL;
if (b->latest_valid[sub] && b->latest[sub] == batch->seq) b->latest_valid[sub] = 0;
b->slots[slot].next = b->free_slot;
@ -123,7 +125,7 @@ static void buffer_evict_oldest(struct stream_buffer *b) {
}
b->stats.expired_before = batch->seq + 1;
b->stats.blocks -= batch->count;
b->stats.stored_bytes -= batch->count * b->block_bytes;
b->stats.stored_bytes -= batch->count * (STREAM_BLOCK_OVERHEAD + batch->block_size);
b->stats.evicted_batches++;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer evicted seq=%llu blocks=%u retained=%u bytes=%zu low=%llu",
(unsigned long long)batch->seq, batch->count, b->stats.blocks, b->stats.stored_bytes,
@ -137,7 +139,7 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer create invalid arguments out=%p key=%p", (void *)out, (const void *)public_key);
return STREAM_ERR_ARGUMENT;
}
size_t block_bytes = stream_block_size(d);
size_t block_bytes = stream_block_capacity(d);
if (!bytes) bytes = STREAM_BUFFER_BYTES;
if (bytes > STREAM_BUFFER_BYTES || bytes < d->n * block_bytes) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer budget invalid bytes=%zu batch=%zu max=%u", bytes, d->n * block_bytes, STREAM_BUFFER_BYTES);
@ -156,7 +158,7 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip
b->slots = u_calloc(b->capacity, sizeof(*b->slots));
b->heap = u_calloc(b->capacity, sizeof(*b->heap));
b->hash = u_calloc(b->hash_size, sizeof(*b->hash));
b->decode_output = u_malloc((uint32_t)stream_input_size(d));
b->decode_output = u_malloc((uint32_t)(stream_input_size(d) + STREAM_BATCH_OVERHEAD));
if (!b->data || !b->slots || !b->heap || !b->hash || !b->decode_output) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer arrays allocation failed capacity=%u bytes=%zu", b->capacity, b->stats.capacity_bytes);
stream_buffer_destroy(b);
@ -194,7 +196,7 @@ void stream_buffer_destroy(struct stream_buffer *b) {
u_free(b);
}
/* Вставить только подписанный блок; OOM/конфликт проверяются до вытеснения ранее сохранённых данных. */
/* Сохранить часть после проверки формата; подпись источника проверяется при восстановлении пачки. */
int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t length) {
struct stream_path path = {.count = 1};
if (b) path.nodes[0] = b->descriptor.source_node_id;
@ -207,42 +209,40 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t
if (!path || !path->count || path->count > STREAM_PATH_MAX
|| stream_path_validate(path, b->descriptor.source_node_id, path->nodes[path->count - 1], 0) != STREAM_OK)
return STREAM_ERR_FORMAT;
struct stream_cached_batch *batch = NULL;
if (length == b->block_bytes) {
batch = buffer_batch(b, stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8));
uint8_t sub = wire[STREAM_BLOCK_SUB_OFF];
if (batch && sub < b->descriptor.n) {
uint32_t slot = buffer_slot(b, batch, sub);
if (slot != STREAM_SLOT_NONE && !memcmp(b->data + slot * b->block_bytes, wire, length)) {
b->stats.duplicates++;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer duplicate seq=%llu sub=%u", (unsigned long long)batch->seq, sub);
return STREAM_DUPLICATE;
}
}
}
struct stream_block_view view;
int result = stream_block_verify(&b->descriptor, b->public_key, wire, length, &view);
int result = stream_block_parse(&b->descriptor, wire, length, &view);
if (result != STREAM_OK) return result;
if (view.seq < b->stats.expired_before) {
const struct stream_batch_info *info = &view.batch;
struct stream_cached_batch *batch = buffer_batch(b, info->seq);
if (batch && batch->block_size == info->block_size) {
uint32_t slot = buffer_slot(b, batch, view.subblock);
if (slot != STREAM_SLOT_NONE && !memcmp(b->data + slot * b->block_bytes, wire, length)) {
b->stats.duplicates++;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer duplicate seq=%llu sub=%u", (unsigned long long)batch->seq, view.subblock);
return STREAM_DUPLICATE;
}
}
if (info->seq < b->stats.expired_before) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer expired incoming seq=%llu sub=%u low=%llu full=%d",
(unsigned long long)view.seq, view.subblock, (unsigned long long)b->stats.expired_before, b->stats.blocks == b->capacity);
(unsigned long long)info->seq, view.subblock, (unsigned long long)b->stats.expired_before, b->stats.blocks == b->capacity);
return STREAM_EXPIRED;
}
if (batch && (batch->timestamp_ms != view.timestamp_ms || batch->data_length != view.data_length || batch->flags != view.flags
if (batch && (batch->timestamp_ms != info->timestamp_ms || batch->data_length != info->data_length || batch->flags != info->flags
|| batch->block_size != info->block_size
|| buffer_slot(b, batch, view.subblock) != STREAM_SLOT_NONE)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer source conflict seq=%llu sub=%u timestamp=%u/%u length=%u/%u flags=%u/%u",
(unsigned long long)view.seq, view.subblock, batch->timestamp_ms, view.timestamp_ms,
batch->data_length, view.data_length, batch->flags, view.flags);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer part conflict seq=%llu sub=%u timestamp=%u/%u length=%u/%u flags=%u/%u size=%u/%u",
(unsigned long long)info->seq, view.subblock, batch->timestamp_ms, info->timestamp_ms,
batch->data_length, info->data_length, batch->flags, info->flags, batch->block_size, info->block_size);
return STREAM_ERR_CONFLICT;
}
if (b->stats.blocks == b->capacity && view.seq <= b->heap[0]->seq) {
if (b->stats.blocks == b->capacity && info->seq <= b->heap[0]->seq) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer full: incoming seq=%llu sub=%u cannot replace newer data oldest=%llu",
(unsigned long long)view.seq, view.subblock, (unsigned long long)b->heap[0]->seq);
(unsigned long long)info->seq, view.subblock, (unsigned long long)b->heap[0]->seq);
return STREAM_EXPIRED;
}
struct stream_cached_path *stored_path = path->count > 1 ? u_malloc(sizeof(*stored_path) + path->count * 8u) : NULL;
if (path->count > 1 && !stored_path) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path allocation failed seq=%llu", (unsigned long long)view.seq);
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path allocation failed seq=%llu", (unsigned long long)info->seq);
return STREAM_ERR_MEMORY;
}
if (stored_path) { stored_path->count = path->count; memcpy(stored_path->nodes, path->nodes, path->count * 8u); }
@ -250,14 +250,15 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t
if (new_batch) {
batch = u_calloc(1, sizeof(*batch));
if (!batch) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)view.seq);
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)info->seq);
u_free(stored_path); return STREAM_ERR_MEMORY;
}
batch->seq = view.seq;
batch->seq = info->seq;
batch->first_slot = STREAM_SLOT_NONE;
batch->timestamp_ms = view.timestamp_ms;
batch->data_length = view.data_length;
batch->flags = view.flags;
batch->timestamp_ms = info->timestamp_ms;
batch->data_length = info->data_length;
batch->block_size = info->block_size;
batch->flags = info->flags;
}
if (b->stats.blocks == b->capacity) buffer_evict_oldest(b);
if (new_batch) buffer_insert_batch(b, batch);
@ -265,7 +266,7 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t
b->free_slot = b->slots[slot].next;
b->slots[slot].subblock = view.subblock;
b->slots[slot].path = stored_path;
b->slots[slot].seq = view.seq; b->slots[slot].received_ms = get_time_tb() / 10;
b->slots[slot].seq = info->seq; b->slots[slot].received_ms = get_time_tb() / 10;
b->slots[slot].sub_next = STREAM_SLOT_NONE; b->slots[slot].sub_previous = b->sub_tail[view.subblock];
if (b->sub_tail[view.subblock] == STREAM_SLOT_NONE) b->sub_head[view.subblock] = slot;
else b->slots[b->sub_tail[view.subblock]].sub_next = slot;
@ -276,13 +277,13 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t
memcpy(b->data + slot * b->block_bytes, wire, length);
b->stats.blocks++;
b->stats.stored_bytes += length;
if (view.seq > b->stats.highest_seq) b->stats.highest_seq = view.seq;
if (!b->latest_valid[view.subblock] || view.seq > b->latest[view.subblock]) {
b->latest[view.subblock] = view.seq;
if (info->seq > b->stats.highest_seq) b->stats.highest_seq = info->seq;
if (!b->latest_valid[view.subblock] || info->seq > b->latest[view.subblock]) {
b->latest[view.subblock] = info->seq;
b->latest_valid[view.subblock] = 1;
}
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer stored seq=%llu sub=%u present=%u/%u bytes=%zu",
(unsigned long long)view.seq, view.subblock, batch->count, b->descriptor.n, b->stats.stored_bytes);
(unsigned long long)info->seq, view.subblock, batch->count, b->descriptor.n, b->stats.stored_bytes);
if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, length, 1);
return STREAM_OK;
}
@ -293,18 +294,20 @@ void stream_buffer_set_observer(struct stream_buffer *b, stream_buffer_event_fn
}
/* Список субпотока упорядочен по монотонному времени вставки, включая поступившие вне порядка индексы. */
int stream_buffer_pending(struct stream_buffer *b, uint64_t seq, uint8_t sub, uint64_t *count, uint64_t *oldest) {
if (!b || sub >= b->descriptor.n || (!count && !oldest)) {
int stream_buffer_pending(struct stream_buffer *b, uint64_t seq, uint8_t sub, uint64_t *count, uint64_t *bytes, uint64_t *oldest) {
if (!b || sub >= b->descriptor.n || (!count && !bytes && !oldest)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer pending invalid arguments sub=%u", sub); return STREAM_ERR_ARGUMENT;
}
uint64_t found = 0, first_ms = 0;
uint64_t found = 0, total_bytes = 0, first_ms = 0;
for (uint32_t slot = b->sub_head[sub]; slot != STREAM_SLOT_NONE; slot = b->slots[slot].sub_next) {
if (b->slots[slot].seq < seq) continue;
if (!found) first_ms = b->slots[slot].received_ms;
found++;
if (!count) break;
total_bytes += STREAM_BLOCK_OVERHEAD + stream_read_be(b->data + slot * b->block_bytes + STREAM_BLOCK_SIZE_OFF, 2);
if (!count && !bytes) break;
}
if (count) *count = found;
if (bytes) *bytes = total_bytes;
if (oldest) *oldest = first_ms;
return found ? STREAM_OK : STREAM_WAIT;
}
@ -363,7 +366,7 @@ int stream_buffer_read(struct stream_buffer *b, uint64_t seq, uint8_t subblock,
return STREAM_GAP;
}
*wire = b->data + slot * b->block_bytes;
*length = b->block_bytes;
*length = STREAM_BLOCK_OVERHEAD + batch->block_size;
*actual_seq = seq;
return STREAM_OK;
}
@ -407,6 +410,7 @@ int stream_buffer_decode(struct stream_buffer *b, uint64_t seq, uint8_t *output,
if (seq < b->stats.expired_before) return STREAM_EXPIRED;
struct stream_cached_batch *batch = buffer_batch(b, seq);
if (!batch) return !b->stats.blocks || seq > b->stats.highest_seq ? STREAM_WAIT : STREAM_GAP;
if (batch->rejected) return STREAM_REJECTED;
if (batch->count < b->descriptor.n - b->descriptor.r) return STREAM_WAIT;
if (capacity < batch->data_length) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode output too small seq=%llu capacity=%zu needed=%u",
@ -416,11 +420,34 @@ int stream_buffer_decode(struct stream_buffer *b, uint64_t seq, uint8_t *output,
const uint8_t *blocks[RS_MAX_BLOCKS] = {0};
for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE; slot = b->slots[slot].next)
blocks[b->slots[slot].subblock] = b->data + slot * b->block_bytes + STREAM_BLOCK_RS_OFF;
if (rs_decode(b->rs, blocks, b->decode_output) != RS_OK) {
size_t decoded_size = (size_t)(b->descriptor.n - b->descriptor.r) * (batch->block_size - RS_HEADER_SIZE);
if (rs_decode(b->rs, blocks, batch->block_size, b->decode_output, rs_input_capacity(b->rs)) != RS_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode RS failed seq=%llu present=%u", (unsigned long long)seq, batch->count);
return STREAM_ERR_FEC;
}
memcpy(output, b->decode_output, batch->data_length);
struct stream_batch_info expected = {.seq = batch->seq, .data_length = batch->data_length,
.timestamp_ms = batch->timestamp_ms, .block_size = batch->block_size, .flags = batch->flags};
int result = stream_batch_verify(&b->descriptor, b->public_key, &expected, b->decode_output, decoded_size);
if (result != STREAM_OK) {
if (result != STREAM_ERR_SIGNATURE && result != STREAM_ERR_FORMAT) return result;
batch->rejected = 1; b->stats.rejected_batches++;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer rejected batch group=%llx source=%llx epoch=%llx seq=%llu present=%u required=%u result=%d",
(unsigned long long)b->descriptor.group_id, (unsigned long long)b->descriptor.source_node_id,
(unsigned long long)b->descriptor.source_epoch, (unsigned long long)seq,
batch->count, b->descriptor.n - b->descriptor.r, result);
unsigned used = 0;
for (unsigned sub = 0; sub < b->descriptor.n && used < (unsigned)(b->descriptor.n - b->descriptor.r); sub++) {
if (!blocks[sub]) continue;
uint32_t slot = buffer_slot(b, batch, sub);
const struct stream_cached_path *path = b->slots[slot].path;
uint64_t supplier = path && path->count > 1 ? path->nodes[path->count - 2] : b->descriptor.source_node_id;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer rejected batch contribution seq=%llu sub=%u supplier=%llx part=%u",
(unsigned long long)seq, sub, (unsigned long long)supplier, batch->block_size);
used++;
}
return STREAM_REJECTED;
}
memcpy(output, b->decode_output + STREAM_BATCH_HEADER, batch->data_length);
*length = batch->data_length;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer decoded seq=%llu present=%u length=%u", (unsigned long long)seq, batch->count, batch->data_length);
return STREAM_OK;

Write Preview
Loading…
Cancel
Save