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

Reject premature stream completion and failed ACCEPT enqueue

master
evgeny 3 days ago
parent
0b202bb3e7
commit
ee36c578e2
1 changed files with 19 additions and 7 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. 26
      src/stream/stream_network.c

26
src/stream/stream_network.c
Unescape View File

@ -477,12 +477,16 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
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);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u index=%llu reason=%d range=%llu..%llu watermark=%llu",
(unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)requested,
reason, (unsigned long long)first, (unsigned long long)last, (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[28] * 8);
m.command = STREAM_CMD_ACCEPT;
if (control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8) != STREAM_OK) {
flow_close(f, STREAM_REASON_LIMIT, 0); return;
}
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);
}
@ -531,8 +535,10 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
}
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;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session peer=%llx command=%u handle=%llu sender=%llx/%llx receiver=%llx/%llx",
(unsigned long long)sender, m.command, (unsigned long long)m.handle,
(unsigned long long)m.sender_epoch, (unsigned long long)(p ? p->other_epoch : 0),
(unsigned long long)m.receiver_epoch, (unsigned long long)(p ? p->own_epoch : 0)); return;
}
p->busy++;
struct STREAM_RECEIVE *f = flow_find(p, &m, m.command != STREAM_CMD_SUBSCRIBE && m.command != STREAM_CMD_RENEW
@ -578,7 +584,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
|| stream_read_be(m.payload + 8, 8) <= f->info.next) flow_close(f, STREAM_REASON_PROTOCOL, 1);
else { f->info.next = stream_read_be(m.payload + 8, 8); f->info.gaps++; }
} else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
if (stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED;
uint64_t last = m.length == STREAM_END_SIZE ? stream_read_be(m.payload + 34, 8) : UINT64_MAX;
if (m.length == STREAM_END_SIZE && last != UINT64_MAX && f->info.next == last + 1
&& stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED;
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;
@ -749,7 +757,11 @@ int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8
if (f->outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d)) bootstrap = 0;
struct STREAM_RECEIVE *f = flow_new(p, &d, sub, 1);
if (!f) return STREAM_ERR_MEMORY;
f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; *out = f; peer_schedule(p); return STREAM_OK;
f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; *out = f; peer_schedule(p);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive opened peer=%llx group=%llx source=%llx epoch=%llx sub=%u index=%llu bootstrap=%d",
(unsigned long long)node, (unsigned long long)d.group_id, (unsigned long long)d.source_node_id,
(unsigned long long)d.source_epoch, sub, (unsigned long long)seq, bootstrap);
return STREAM_OK;
}
void stream_receive_close(struct STREAM_RECEIVE *f) {

Write Preview
Loading…
Cancel
Save