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

Report stream ring bounds and handle expired and final cursors

master
evgeny 4 days ago
parent
c2179b3959
commit
d83e85b8ca
3 changed files with 32 additions and 20 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. 3
      src/stream/stream.c
  2. 47
      src/stream/stream_network.c
  3. 2
      src/stream/stream_wire.h

3
src/stream/stream.c
Unescape View File

@ -513,7 +513,8 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l
if (!s) { DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "unsubscribed stream ignored group=%llx source=%llx",
(unsigned long long)group, (unsigned long long)source); return STREAM_WAIT; }
if (s->source) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "incoming DATA to own publisher rejected"); return STREAM_ERR_STATE; }
if (!path || path->count >= STREAM_PATH_MAX || stream_path_validate(path, source, path->nodes[path->count - 1], inst->node_id) != STREAM_OK)
if (!path || !path->count || path->count >= STREAM_PATH_MAX
|| stream_path_validate(path, source, path->nodes[path->count - 1], inst->node_id) != STREAM_OK)
return STREAM_ERR_FORMAT;
if (s->end_valid && (s->last_seq == UINT64_MAX || stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8) > s->last_seq)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA beyond signed END source=%llx", (unsigned long long)source);

47
src/stream/stream_network.c
Unescape View File

@ -267,25 +267,27 @@ static int authorized(struct stream_network *net, uint64_t peer, const struct st
/* INFO и ACCEPT описывают реальный последний/начальный блок с его историей, а не BGP-метрику. */
static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, uint64_t peer, uint64_t self, uint8_t *payload) {
const uint8_t *wire; size_t length; uint64_t actual, first, last; struct stream_path path;
if (stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual) != STREAM_OK
int result = stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual);
if (result == STREAM_EXPIRED) return STREAM_REASON_EXPIRED;
if (result != STREAM_OK
|| stream_buffer_range(local->buffer, sub, &first, &last) != STREAM_OK
|| stream_buffer_path(local->buffer, actual, sub, &path) != STREAM_OK) return STREAM_REASON_POSITION;
if (stream_path_validate(&path, local->descriptor.source_node_id, self, peer) != STREAM_OK || path.count == STREAM_PATH_MAX)
return STREAM_REASON_LOOP;
stream_write_be(payload, actual, 8); stream_write_be(payload + 8, last, 8);
payload[16] = local->descriptor.n; payload[17] = local->descriptor.r;
stream_write_be(payload + 18, local->descriptor.block_size, 2);
/* byte 20 — число узлов, следует непосредственно за параметрами. */
stream_path_write(payload + 20, &path); return 0;
stream_write_be(payload, actual, 8); stream_write_be(payload + 8, first, 8); stream_write_be(payload + 16, last, 8);
payload[24] = local->descriptor.n; payload[25] = local->descriptor.r;
stream_write_be(payload + 26, local->descriptor.block_size, 2);
stream_path_write(payload + 28, &path); return 0;
}
static int parse_info(const struct stream_message *m, const struct stream_descriptor *d, uint64_t peer, uint64_t self, uint64_t *start) {
struct stream_path path; size_t used;
if (m->length < STREAM_INFO_SIZE || m->payload[16] != d->n || m->payload[17] != d->r
|| stream_read_be(m->payload + 18, 2) != d->block_size
|| stream_path_parse(m->payload + 20, m->length - 20, &path, &used) != STREAM_OK || m->length != 20 + used
if (m->length < STREAM_INFO_SIZE || m->payload[24] != d->n || m->payload[25] != d->r
|| stream_read_be(m->payload + 26, 2) != d->block_size
|| stream_path_parse(m->payload + 28, m->length - 28, &path, &used) != STREAM_OK || m->length != 28 + used
|| path.count == STREAM_PATH_MAX || stream_path_validate(&path, d->source_node_id, peer, self) != STREAM_OK
|| stream_read_be(m->payload, 8) == UINT64_MAX || stream_read_be(m->payload, 8) > stream_read_be(m->payload + 8, 8)) {
|| stream_read_be(m->payload, 8) == UINT64_MAX || stream_read_be(m->payload, 8) > stream_read_be(m->payload + 16, 8)
|| stream_read_be(m->payload + 8, 8) > stream_read_be(m->payload + 16, 8)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO/ACCEPT invalid command=%u bytes=%zu", m->command, m->length);
return STREAM_ERR_FORMAT;
}
@ -316,7 +318,8 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 1); return 0; }
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock};
const uint8_t *wire; size_t length; uint64_t actual;
int result = stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual);
int result = f->info.next == UINT64_MAX ? STREAM_WAIT
: stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual);
if (result == STREAM_OK) {
struct stream_path path;
if (stream_buffer_path(local.buffer, actual, f->subblock, &path) != STREAM_OK
@ -332,7 +335,11 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
}
if (result == STREAM_EXPIRED || result == STREAM_GAP) {
uint64_t next;
if (stream_buffer_next_seq(local.buffer, f->info.next, f->subblock, &next) == STREAM_OK && next > f->info.next) {
int available = stream_buffer_next_seq(local.buffer, f->info.next, f->subblock, &next) == STREAM_OK;
if (!available && local.finished && local.last != UINT64_MAX && f->info.next <= local.last) {
next = local.last + 1; available = 1;
}
if (available && next > f->info.next) {
uint8_t payload[16]; stream_write_be(payload, f->info.next, 8); stream_write_be(payload + 8, next, 8);
m.command = STREAM_CMD_GAP;
if (raw_send(p, m, payload, sizeof(payload)) == STREAM_OK) {
@ -442,19 +449,22 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
} else if (f && !reason) {
reason = info_payload(local, m.subblock, f->info.start, p->node_id, p->network->inst->node_id, payload);
/* Идемпотентный ACCEPT возвращает первоначальный индекс даже если тот уже вытеснен. */
if (reason == STREAM_REASON_POSITION) {
if (reason == STREAM_REASON_POSITION || reason == STREAM_REASON_EXPIRED) {
reason = info_payload(local, m.subblock, UINT64_MAX, p->node_id, p->network->inst->node_id, payload);
if (!reason) stream_write_be(payload, f->info.start, 8);
}
}
if (reason) {
uint8_t value = (uint8_t)reason; m.command = STREAM_CMD_REJECT; control_put(p, m, &value, 1);
uint8_t rejection[17] = {(uint8_t)reason}; uint64_t first = UINT64_MAX, last = UINT64_MAX;
if (m.subblock < local->descriptor.n) stream_buffer_range(local->buffer, m.subblock, &first, &last);
stream_write_be(rejection + 1, first, 8); stream_write_be(rejection + 9, last, 8);
m.command = STREAM_CMD_REJECT; control_put(p, m, rejection, sizeof(rejection));
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u reason=%d watermark=%llu",
(unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, reason, (unsigned long long)p->highwater);
return;
}
f->deadline = get_time_tb() + STREAM_LEASE_TB;
m.command = STREAM_CMD_ACCEPT; control_put(p, m, payload, STREAM_INFO_SIZE + payload[20] * 8);
m.command = STREAM_CMD_ACCEPT; control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream ACCEPT queued peer=%llx handle=%llu sub=%u start=%llu lease=6000ms",
(unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)f->info.start);
}
@ -490,10 +500,10 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
int reason = info_payload(&local, m.subblock, UINT64_MAX, sender, net->inst->node_id, payload);
m.command = reason ? STREAM_CMD_REJECT : STREAM_CMD_INFO_RESP;
if (reason) { uint8_t value = (uint8_t)reason; control_put(p, m, &value, 1); }
else control_put(p, m, payload, STREAM_INFO_SIZE + payload[20] * 8);
else control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8);
return;
}
if (!p || !p->up || m.receiver_epoch != p->own_epoch
if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch
|| (p->other_epoch && m.sender_epoch != p->other_epoch)) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session packet peer=%llx command=%u handle=%llu",
(unsigned long long)sender, m.command, (unsigned long long)m.handle); return;
@ -546,7 +556,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
else flow_close(f, STREAM_REASON_PROTOCOL, 1);
} else if (m.command == STREAM_CMD_RENEW && f && !m.length) f->deadline = get_time_tb() + STREAM_LEASE_TB;
else if (m.command == STREAM_CMD_UNSUBSCRIBE && f && m.length == 1) flow_close(f, m.payload[0], 1);
else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && m.length == 1) flow_close(f, m.payload[0], 0);
else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && (m.length == 1 || m.length == 17))
flow_close(f, m.payload[0], 0);
else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream unknown/inactive issuance peer=%llx command=%u handle=%llu sub=%u bytes=%zu",
(unsigned long long)sender, m.command, (unsigned long long)m.handle, m.subblock, m.length);
p->busy--; peer_sweep(p);

2
src/stream/stream_wire.h
Unescape View File

@ -9,7 +9,7 @@
#define STREAM_WIRE_HEADER 52
#define STREAM_PATH_MAX 32
#define STREAM_END_SIZE 106
#define STREAM_INFO_SIZE 21
#define STREAM_INFO_SIZE 29 /* actual:8, first:8, last:8, n:1, r:1, block_size:2, path_count:1 */
enum stream_command {
STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT,

Write Preview
Loading…
Cancel
Save