Skip to content

tls: yield downstream read retries to the event loop [Backport to 5.0] #12141

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edsiper merged 10 commits into from
Jul 30, 2026
Merged

tls: yield downstream read retries to the event loop [Backport to 5.0] #12141

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
52dc27c
tls: yield downstream reads to event loop
edsiper Jul 22, 2026
dc675fa
in_forward: retain retryable TLS connections
edsiper Jul 22, 2026
86a2f07
in_http: retain retryable TLS connections
edsiper Jul 22, 2026
9ef9b18
in_mqtt: retain retryable TLS connections
edsiper Jul 22, 2026
854f656
in_syslog: retain retryable TLS connections
edsiper Jul 22, 2026
f30e3a3
in_tcp: retain retryable TLS connections
edsiper Jul 22, 2026
a4e27cb
in_unix_socket: retain retryable TLS connections
edsiper Jul 22, 2026
1e2a2a4
http_server: retain retryable TLS connections
edsiper Jul 22, 2026
ff9bd14
tls: preserve downstream retry event dispatch
edsiper Jul 30, 2026
b166ceb
tests: internal: preserve TLS retry event dispatch
edsiper Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions include/fluent-bit/flb_io.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,15 @@
/* Other features */
#define FLB_IO_IPV6 32 /* network I/O uses IPv6 */

/* Retryable network I/O results */
#define FLB_IO_WANT_READ -0x7e4
#define FLB_IO_WANT_WRITE -0x7e6

static inline int flb_io_net_is_retry(ssize_t result)
{
return result == FLB_IO_WANT_READ || result == FLB_IO_WANT_WRITE;
}

struct flb_connection;

int flb_io_net_accept(struct flb_connection *connection,
Expand Down
5 changes: 3 additions & 2 deletions include/fluent-bit/tls/flb_tls.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,15 +25,16 @@
#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_coro.h>
#include <fluent-bit/flb_io.h>
#include <stddef.h>

#define FLB_TLS_ALPN_MAX_LENGTH 16

#define FLB_TLS_CLIENT "Fluent Bit"

/* TLS backend return status on read/write */
#define FLB_TLS_WANT_READ -0x7e4
#define FLB_TLS_WANT_WRITE -0x7e6
#define FLB_TLS_WANT_READ FLB_IO_WANT_READ
#define FLB_TLS_WANT_WRITE FLB_IO_WANT_WRITE

/* Cert Flags */
#define FLB_TLS_CA_ROOT 1
Expand Down
8 changes: 8 additions & 0 deletions plugins/in_forward/fw_conn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,10 @@ static int fw_conn_event_internal(struct flb_connection *connection)
flb_plg_trace(ctx->ins, "handshake status = %d", conn->handshake_status);

ret = fw_prot_secure_forward_handshake(ctx->ins, conn);
if (flb_io_net_is_retry(ret)) {
return 0;
}

if (ret == -1) {
flb_plg_trace(ctx->ins, "fd=%i closed connection", event->fd);
fw_conn_del(conn);
Expand Down Expand Up @@ -108,6 +112,10 @@ static int fw_conn_event_internal(struct flb_connection *connection)
(void *) &conn->buf[conn->buf_len],
available);

if (flb_io_net_is_retry(bytes)) {
return 0;
}

if (bytes > 0) {
read_bytes = (size_t) bytes;
flb_plg_trace(ctx->ins, "read()=%zd pre_len=%zu now_len=%zu",
Expand Down
43 changes: 31 additions & 12 deletions plugins/in_forward/fw_prot.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,45 +140,54 @@ static inline void print_msgpack_error_code(struct flb_input_instance *in,

/* Read a secure forward msgpack message for handshake */
static int secure_forward_read(struct flb_input_instance *in,
struct flb_connection *connection,
char *buf, size_t size, size_t *out_len)
struct fw_conn *conn, size_t size,
size_t *out_len)
{
int ret;
size_t off;
size_t avail;
size_t buf_off = 0;
msgpack_unpacked result;

msgpack_unpacked_init(&result);
while (1) {
avail = size - buf_off;
if (avail < 1) {
if (conn->buf_len >= size) {
goto error;
}

avail = size - conn->buf_len;

/* Read the message */
ret = flb_io_net_read(connection, buf + buf_off, size - buf_off);
ret = flb_io_net_read(conn->connection,
conn->buf + conn->buf_len, avail);
Comment on lines +160 to +161

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Bound secure-forward reads to the allocated buffer

When secure-forward is enabled and buffer_chunk_size is configured below 1023, conn->buf is only that configured size, but check_ping() calls this helper with size == 1023, so flb_io_net_read() can write past the end of conn->buf before msgpack validation runs. Please cap avail by conn->buf_size - conn->buf_len or grow the connection buffer before reading the PING.

Useful? React with 👍 / 👎.

if (flb_io_net_is_retry(ret)) {
msgpack_unpacked_destroy(&result);
return ret;
}

if (ret <= 0) {
flb_plg_debug(in, "read %d byte(s)", ret);
goto error;
}
buf_off += ret;
conn->buf_len += ret;

/* Validate */
off = 0;
ret = msgpack_unpack_next(&result, buf, buf_off, &off);
ret = msgpack_unpack_next(&result, conn->buf, conn->buf_len, &off);
switch (ret) {
case MSGPACK_UNPACK_SUCCESS:
msgpack_unpacked_destroy(&result);
*out_len = buf_off;
*out_len = conn->buf_len;
return 0;
case MSGPACK_UNPACK_CONTINUE:
break;
default:
print_msgpack_error_code(in, ret, "handshake");
goto error;
};
}

error:
conn->buf_len = 0;
msgpack_unpacked_destroy(&result);
return -1;
}
Expand Down Expand Up @@ -506,7 +515,6 @@ static int check_ping(struct flb_input_instance *ins,
flb_sds_t *out_shared_key_salt)
{
int ret;
char buf[1024];
size_t out_len;
size_t off;
msgpack_unpacked result;
Expand All @@ -531,7 +539,12 @@ static int check_ping(struct flb_input_instance *ins,
}

/* Wait for client PING */
ret = secure_forward_read(ins, conn->connection, buf, sizeof(buf) - 1, &out_len);
ret = secure_forward_read(ins, conn, 1023, &out_len);
if (flb_io_net_is_retry(ret)) {
flb_free(serverside);
return ret;
}

if (ret == -1) {
flb_free(serverside);
flb_plg_error(ins, "handshake error expecting PING");
Expand All @@ -541,7 +554,8 @@ static int check_ping(struct flb_input_instance *ins,
/* Unpack message and validate */
off = 0;
msgpack_unpacked_init(&result);
ret = msgpack_unpack_next(&result, buf, out_len, &off);
ret = msgpack_unpack_next(&result, conn->buf, out_len, &off);
conn->buf_len = 0;
if (ret != MSGPACK_UNPACK_SUCCESS) {
flb_free(serverside);
print_msgpack_error_code(ins, ret, "PING");
Expand Down Expand Up @@ -1206,6 +1220,11 @@ int fw_prot_secure_forward_handshake(struct flb_input_instance *ins,
reason = flb_sds_create_size(32);
flb_plg_debug(ins, "protocol: checking PING");
ping_ret = check_ping(ins, conn, &shared_key_salt);
if (flb_io_net_is_retry(ping_ret)) {
flb_sds_destroy(reason);
return ping_ret;
}

if (ping_ret == -1) {
flb_plg_error(ins, "handshake error checking PING");

Expand Down
4 changes: 4 additions & 0 deletions plugins/in_http/http_conn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,10 @@ static int http_conn_event(void *data)
(void *) &conn->buf_data[conn->buf_len],
available);

if (flb_io_net_is_retry(bytes)) {
return 0;
}

if (bytes <= 0) {
flb_plg_trace(ctx->ins, "fd=%i closed connection", event->fd);
http_conn_del(conn);
Expand Down
4 changes: 4 additions & 0 deletions plugins/in_mqtt/mqtt_conn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,10 @@ int mqtt_conn_event(void *data)
(void *) &conn->buf[conn->buf_len],
available);

if (flb_io_net_is_retry(bytes)) {
return 0;
}

if (bytes > 0) {
conn->buf_len += bytes;
flb_plg_trace(ctx->ins, "[fd=%i] read()=%i bytes",
Expand Down
4 changes: 4 additions & 0 deletions plugins/in_syslog/syslog_conn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,10 @@ int syslog_stream_conn_event(void *data)
(void *) &conn->buf_data[conn->buf_len],
available);

if (flb_io_net_is_retry(bytes)) {
return 0;
}

if (bytes > 0) {
flb_plg_trace(ctx->ins, "read()=%i pre_len=%zu now_len=%zu",
bytes, conn->buf_len, conn->buf_len + bytes);
Expand Down
5 changes: 5 additions & 0 deletions plugins/in_tcp/tcp_conn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -452,6 +452,11 @@ int tcp_conn_event(void *data)
(void *) &conn->buf_data[conn->buf_len],
available);

if (flb_io_net_is_retry(bytes)) {
conn->busy = FLB_FALSE;
return 0;
}

if (bytes <= 0) {
flb_plg_trace(ctx->ins, "fd=%i closed connection", event->fd);
conn->busy = FLB_FALSE;
Expand Down
4 changes: 4 additions & 0 deletions plugins/in_unix_socket/unix_socket_conn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,6 +269,10 @@ int unix_socket_conn_event(void *data)
(void *) &conn->buf_data[conn->buf_len],
available);

if (flb_io_net_is_retry(bytes)) {
return 0;
}

if (bytes <= 0) {
if (!ctx->dgram_mode_flag) {
flb_plg_trace(ctx->ins, "fd=%i closed connection", event->fd);
Expand Down
4 changes: 4 additions & 0 deletions src/http_server/flb_http_server.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,10 @@ static int flb_http_server_session_read(struct flb_http_server_session *session)
(void *) session->read_buffer,
session->read_buffer_size);

if (flb_io_net_is_retry(result)) {
return 0;
}

if (result <= 0) {
return -1;
}
Expand Down
33 changes: 31 additions & 2 deletions src/tls/flb_tls.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,10 +162,10 @@ static inline int io_tls_event_switch(struct flb_tls_session *session,
event = &session->connection->event;
event_loop = session->connection->evl;

if ((event->mask & mask) == 0) {
if ((event->mask & (MK_EVENT_READ | MK_EVENT_WRITE)) != mask) {
ret = mk_event_add(event_loop,
event->fd,
FLB_ENGINE_EV_THREAD,
event->type,
mask, event);

event->priority = FLB_ENGINE_PRIORITY_CONNECT;
Expand All @@ -175,6 +175,8 @@ static inline int io_tls_event_switch(struct flb_tls_session *session,

return -1;
}

event->mask = mask;
}

return 0;
Expand Down Expand Up @@ -345,12 +347,19 @@ int flb_tls_set_client_thumbprints(struct flb_tls *tls, const char *thumbprints)

int flb_tls_net_read(struct flb_tls_session *session, void *buf, size_t len)
{
int event_driven;
time_t timeout_timestamp;
time_t current_timestamp;
struct flb_tls *tls;
int ret;

tls = session->tls;
event_driven = FLB_FALSE;

if (session->connection->type == FLB_DOWNSTREAM_CONNECTION &&
MK_EVENT_IS_REGISTERED((&session->connection->event))) {
event_driven = FLB_TRUE;
}

if (session->connection->net->io_timeout > 0) {
timeout_timestamp = time(NULL) + session->connection->net->io_timeout;
Expand All @@ -365,6 +374,14 @@ int flb_tls_net_read(struct flb_tls_session *session, void *buf, size_t len)
current_timestamp = time(NULL);

if (ret == FLB_TLS_WANT_READ) {
if (event_driven) {
if (io_tls_event_switch(session, MK_EVENT_READ) == -1) {
return -1;
}

return FLB_TLS_WANT_READ;
}

if (timeout_timestamp > 0 &&
timeout_timestamp <= current_timestamp) {
return ret;
Expand All @@ -373,6 +390,14 @@ int flb_tls_net_read(struct flb_tls_session *session, void *buf, size_t len)
goto retry_read;
}
else if (ret == FLB_TLS_WANT_WRITE) {
if (event_driven) {
if (io_tls_event_switch(session, MK_EVENT_READ | MK_EVENT_WRITE) == -1) {

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve custom event dispatch for TLS write retries

When a downstream TLS read returns FLB_TLS_WANT_WRITE, this new path calls io_tls_event_switch(), whose mk_event_add(..., FLB_ENGINE_EV_THREAD, ...) re-registers the already-registered input/http-server connection as a thread event instead of the FLB_ENGINE_EV_CUSTOM handler it was using. In the normal downstream input registrations there is no coroutine waiting for that follow-up writable notification, so the engine consumes the event without invoking the connection handler and the TLS connection can remain stuck in the retry state; preserve the existing event type/handler when switching the mask.

Useful? React with 👍 / 👎.

return -1;
}

return FLB_TLS_WANT_WRITE;
}

goto retry_read;
}
else if (ret < 0) {
Expand All @@ -382,6 +407,10 @@ int flb_tls_net_read(struct flb_tls_session *session, void *buf, size_t len)
return -1;
}

if (event_driven && io_tls_event_switch(session, MK_EVENT_READ) == -1) {
return -1;
}

return ret;
}

Expand Down
Loading
Loading