Handle connection closures from proxy peers in async

Makefile

@@ -59,6 +59,7 @@ bin/res/%.txt: res/%.*
 bin/sesimos: bin/server.o bin/logger.o bin/cache_handler.o bin/async.o bin/workers.o \
              bin/worker/request_handler.o bin/worker/tcp_acceptor.o \
              bin/worker/fastcgi_handler.o bin/worker/local_handler.o bin/worker/proxy_handler.o \
+             bin/worker/proxy_peer_handler.o \
              bin/worker/ws_frame_handler.o bin/worker/chunk_handler.o bin/worker/fastcgi_frame_handler.o \
              bin/lib/http_static.o bin/res/default.o bin/res/proxy.o bin/res/style.o \
              bin/res/icon_error.o bin/res/icon_info.o bin/res/icon_success.o bin/res/icon_warning.o \
@@ -90,6 +91,8 @@ bin/worker/local_handler.o: src/worker/func.h
 
 bin/worker/proxy_handler.o: src/worker/func.h
 
+bin/worker/proxy_peer_handler.o: src/worker/func.h
+
 bin/worker/ws_frame_handler.o: src/worker/func.h
 
 bin/worker/fastcgi_frame_handler.o: src/worker/func.h

src/async.c

@@ -108,7 +108,7 @@ static int async_exec(evt_listen_t *evt, async_evt_t r_events) {
     int ret, e = errno;
     if (r_events & evt->events) {
         // specified event(s) occurred
-        if (evt->socket && !sock_has_pending(evt->socket, 0)) {
+        if (!(evt->flags & ASYNC_IGNORE_PENDING) && evt->socket && !sock_has_pending(evt->socket, 0)) {
             evt->err_cb(evt->arg);
             ret = 0;
         } else {

src/async.h

@@ -12,6 +12,7 @@
 #include "lib/sock.h"
 
 #define ASYNC_KEEP 1
+#define ASYNC_IGNORE_PENDING 2
 
 #define ASYNC_IN   0x01
 #define ASYNC_PRI  0x02

src/lib/proxy.c

@@ -137,16 +137,20 @@ proxy_ctx_t *proxy_get_by_conf(host_config_t *conf) {
     return NULL;
 }
 
-void proxy_unlock_ctx(proxy_ctx_t *ctx) {
+int proxy_unlock_ctx(proxy_ctx_t *ctx) {
     int n = (int) ((ctx - proxies) / MAX_PROXY_CNX_PER_HOST);
-    if (ctx->close) {
-        proxy_close(ctx);
-        ctx->close = 0;
-    }
+    if (ctx->close) proxy_close(ctx);
+
     debug("Released proxy connection slot %i/%i", (ctx - proxies) % MAX_PROXY_CNX_PER_HOST, MAX_PROXY_CNX_PER_HOST);
     ctx->in_use = 0;
     ctx->client = NULL;
     sem_post(&available[n]);
+    if (!ctx->close) {
+        return 1;
+    } else {
+        ctx->close = 0;
+        return 0;
+    }
 }
 
 int proxy_request_header(http_req *req, sock *sock) {

src/lib/proxy.h

@@ -35,7 +35,7 @@ void proxy_close_all(void);
 
 proxy_ctx_t *proxy_get_by_conf(host_config_t *conf);
 
-void proxy_unlock_ctx(proxy_ctx_t *ctx);
+int proxy_unlock_ctx(proxy_ctx_t *ctx);
 
 int proxy_request_header(http_req *req, sock *sock);
 

src/worker/func.h

@@ -71,6 +71,8 @@ void fastcgi_frame_handler_func(fastcgi_ctx_t *ctx);
 
 void proxy_handler_func(client_ctx_t *ctx);
 
+void proxy_peer_handler_func(proxy_ctx_t *ctx);
+
 void ws_frame_handler_func(ws_ctx_t *ctx);
 
 void chunk_handler_func(chunk_ctx_t *ctx);

src/worker/proxy_handler.c

@@ -27,16 +27,21 @@ void proxy_handler_func(client_ctx_t *ctx) {
     respond(ctx);
 
     if (ret == 1) {
-        proxy_unlock_ctx(ctx->proxy);
+        // error status code
+        if (proxy_unlock_ctx(ctx->proxy) == 1)
+            proxy_peer_handle(ctx->proxy);
     } else if (ctx->use_proxy == 0) {
+        // proxy not used
         proxy_close(ctx->proxy);
         proxy_unlock_ctx(ctx->proxy);
     } else if (ctx->use_proxy == 1) {
+        // proxy is used
         if (proxy_handler_2(ctx) == 1) {
             // chunked
             return;
         }
-        proxy_unlock_ctx(ctx->proxy);
+        if (proxy_unlock_ctx(ctx->proxy) == 1)
+            proxy_peer_handle(ctx->proxy);
     } else if (ctx->use_proxy == 2) {
         // WebSocket
         ws_handle_connection(ctx);
@@ -112,7 +117,8 @@ static int proxy_handler_1(client_ctx_t *ctx) {
 }
 
 static void proxy_chunk_next_cb(chunk_ctx_t *ctx) {
-    proxy_unlock_ctx(ctx->client->proxy);
+    if (proxy_unlock_ctx(ctx->client->proxy) == 1)
+        proxy_peer_handle(ctx->client->proxy);
 
     ctx->client->proxy = NULL;
     request_complete(ctx->client);

src/worker/proxy_peer_handler.c

@@ -0,0 +1,17 @@
+/**
+ * sesimos - secure, simple, modern web server
+ * @brief Proxy peer handler
+ * @file src/worker/proxy_peer_handler.c
+ * @author Lorenz Stechauner
+ * @date 2023-07-07
+ */
+
+#include "func.h"
+#include "../logger.h"
+#include "../lib/utils.h"
+
+void proxy_peer_handler_func(proxy_ctx_t *ctx) {
+    if (!ctx->initialized || ctx->in_use) return;
+    logger_set_prefix("[%s%*s%s]", BLD_STR, ADDRSTRLEN, ctx->host, CLR_STR);
+    proxy_close(ctx);
+}

src/workers.c

@@ -13,7 +13,7 @@
 #include "async.h"
 
 static mpmc_t tcp_acceptor_ctx, request_handler_ctx, local_handler_ctx, fastcgi_handler_ctx, proxy_handler_ctx,
-              ws_frame_handler_ctx, chunk_handler_ctx, fastcgi_frame_handler_ctx;
+              proxy_peer_handler_ctx, ws_frame_handler_ctx, chunk_handler_ctx, fastcgi_frame_handler_ctx;
 
 int workers_init(void) {
     mpmc_init(&tcp_acceptor_ctx,          8, 64, (void (*)(void *)) tcp_acceptor_func,          "tcp");
@@ -21,6 +21,7 @@ int workers_init(void) {
     mpmc_init(&local_handler_ctx,         8, 64, (void (*)(void *)) local_handler_func,         "local");
     mpmc_init(&fastcgi_handler_ctx,       8, 64, (void (*)(void *)) fastcgi_handler_func,       "fcgi");
     mpmc_init(&proxy_handler_ctx,         8, 64, (void (*)(void *)) proxy_handler_func,         "proxy");
+    mpmc_init(&proxy_peer_handler_ctx,    1,  8, (void (*)(void *)) proxy_peer_handler_func,    "prxy_p");
     mpmc_init(&ws_frame_handler_ctx,      8, 64, (void (*)(void *)) ws_frame_handler_func,      "ws");
     mpmc_init(&chunk_handler_ctx,         8, 64, (void (*)(void *)) chunk_handler_func,         "chunk");
     mpmc_init(&fastcgi_frame_handler_ctx, 8, 64, (void (*)(void *)) fastcgi_frame_handler_func, "fcgi_f");
@@ -32,6 +33,7 @@ void workers_stop(void) {
     mpmc_stop(&local_handler_ctx);
     mpmc_stop(&fastcgi_handler_ctx);
     mpmc_stop(&proxy_handler_ctx);
+    mpmc_stop(&proxy_peer_handler_ctx);
     mpmc_stop(&request_handler_ctx);
     mpmc_stop(&ws_frame_handler_ctx);
     mpmc_stop(&chunk_handler_ctx);
@@ -43,6 +45,7 @@ void workers_destroy(void) {
     mpmc_destroy(&local_handler_ctx);
     mpmc_destroy(&fastcgi_handler_ctx);
     mpmc_destroy(&proxy_handler_ctx);
+    mpmc_destroy(&proxy_peer_handler_ctx);
     mpmc_destroy(&request_handler_ctx);
     mpmc_destroy(&ws_frame_handler_ctx);
     mpmc_destroy(&chunk_handler_ctx);
@@ -92,6 +95,17 @@ int proxy_handle(client_ctx_t *ctx) {
     return mpmc_queue(&proxy_handler_ctx, ctx);
 }
 
+static int proxy_peer_handle_cb(proxy_ctx_t *ctx) {
+    return mpmc_queue(&proxy_peer_handler_ctx, ctx);
+}
+
+int proxy_peer_handle(proxy_ctx_t *ctx) {
+    return async(&ctx->proxy, ASYNC_WAIT_READ, ASYNC_IGNORE_PENDING, ctx,
+                 (void (*)(void *)) proxy_peer_handle_cb,
+                 (void (*)(void *)) proxy_peer_handle_cb,
+                 (void (*)(void *)) proxy_peer_handle_cb);
+}
+
 static int ws_handle_frame_cb(ws_ctx_t *ctx) {
     return mpmc_queue(&ws_frame_handler_ctx, ctx);
 }

src/workers.h

@@ -29,6 +29,8 @@ int fastcgi_handle_frame(fastcgi_ctx_t *ctx);
 
 int proxy_handle(client_ctx_t *ctx);
 
+int proxy_peer_handle(proxy_ctx_t *ctx);
+
 int ws_handle_frame(ws_ctx_t *ctx);
 
 int handle_chunk(chunk_ctx_t *ctx);