fix: enforce max alloc across protocol workers

src/client/client_send.c

@@ -456,6 +456,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
 
 static int scan_directory_multithreaded(void* pipeline_context) {
   PipelineContextSender* context = (PipelineContextSender*)pipeline_context;
+  protocol_session_bind(&context->allocation_session);
   ScannerOptions options = scanner_options_from_config(context->config, 4);
   ParallelScanner* scanner =
       parallel_scanner_create_with_options(context->config->send_directory, &options);
@@ -464,6 +465,7 @@ static int scan_directory_multithreaded(void* pipeline_context) {
   if (scanner == NULL) {
     log_message(LOG_LEVEL_ERROR, "Failed to create parallel scanner");
     pipeline_cancel(context);
+    protocol_session_unbind();
     return thrd_error;
   }
   while ((current_chunk = parallel_scanner_next(scanner)) != NULL) {
@@ -475,6 +477,7 @@ static int scan_directory_multithreaded(void* pipeline_context) {
         pipeline_cancel(context);
         chunk_destroy(current_chunk);
         parallel_scanner_destroy(scanner);
+        protocol_session_unbind();
         return thrd_error;
       }
     }
@@ -485,6 +488,7 @@ static int scan_directory_multithreaded(void* pipeline_context) {
       chunk_destroy(current_chunk);
       pipeline_cancel(context);
       parallel_scanner_destroy(scanner);
+      protocol_session_unbind();
       return thrd_error;
     }
   }
@@ -496,6 +500,7 @@ static int scan_directory_multithreaded(void* pipeline_context) {
     cnd_broadcast(&context->condition_not_full_scanner);
     mtx_unlock(&context->mutex_scanner);
     pipeline_cancel(context);
+    protocol_session_unbind();
     return thrd_error;
   }
   mtx_lock(&context->mutex_scanner);
@@ -504,11 +509,13 @@ static int scan_directory_multithreaded(void* pipeline_context) {
   mtx_unlock(&context->mutex_scanner);
 
   parallel_scanner_destroy(scanner);
+  protocol_session_unbind();
   return thrd_success;
 }
 
 static int load_files_multithreaded(void* pipeline_context) {
   PipelineContextSender* context = (PipelineContextSender*)pipeline_context;
+  protocol_session_bind(&context->allocation_session);
   while (true) {
     Chunk* chunk = queue_dequeue_multithreaded(
         context->queue_scanner, &context->mutex_scanner, &context->condition_not_empty_scanner,
@@ -518,6 +525,7 @@ static int load_files_multithreaded(void* pipeline_context) {
       context->loader_done = true;
       cnd_signal(&context->condition_not_empty_loader);
       mtx_unlock(&context->mutex_loader);
+      protocol_session_unbind();
       return thrd_success;
     }
     if (!context->config->use_sendfile) {
@@ -529,6 +537,7 @@ static int load_files_multithreaded(void* pipeline_context) {
           log_message(LOG_LEVEL_ERROR, "Failed to load file data");
           chunk_destroy(chunk);
           pipeline_cancel(context);
+          protocol_session_unbind();
           return thrd_error;
         }
       }
@@ -539,6 +548,7 @@ static int load_files_multithreaded(void* pipeline_context) {
                                             &context->cancelled)) {
       chunk_destroy(chunk);
       pipeline_cancel(context);
+      protocol_session_unbind();
       return thrd_error;
     }
   }

src/server/server.c

@@ -294,6 +294,7 @@ void handler(int file_descriptor) {
       protocol_session_unbind();
       return;
     }
+    protocol_session_set_max_alloc(&context->session, config->max_alloc);
     context->session.total_allocated_bytes = session.total_allocated_bytes;
     thrd_t receiver, writer;
     bool receiver_created = thrd_create(&receiver, receive_thread, context) == thrd_success;

src/shared/config.c

@@ -262,6 +262,8 @@ static bool receive_core_fields(int fd, Config* c) {
   int value;
   if (!receive_n_data(fd, &c->max_alloc, sizeof(c->max_alloc)) || c->max_alloc == 0)
     return false;
+  if (c->max_alloc > MAX_SERVER_ALLOC)
+    c->max_alloc = MAX_SERVER_ALLOC;
   protocol_session_set_max_alloc(NULL, c->max_alloc);
   c->send_directory = receive_str(fd);
   c->receive_root_directory = receive_str(fd);

src/shared/config.h

@@ -129,7 +129,7 @@ typedef struct Config {
   char* compress_choice;
 } Config;
 
-#define PROTOCOL_VERSION "2.2.0"
+#define PROTOCOL_VERSION "2.3.0"
 #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
 
 Config* config_create(void);

src/shared/multiprocessing.c

@@ -29,6 +29,8 @@ PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* que
   context->progress_bytes = 0;
   context->sender_done = false;
   atomic_init(&context->cancelled, false);
+  protocol_session_init(&context->allocation_session, -1, -1);
+  protocol_session_set_max_alloc(&context->allocation_session, config->max_alloc);
   int init = 0;
   if (mtx_init(&context->mutex_scanner, mtx_plain) != thrd_success)
     goto fail;

src/shared/multiprocessing.h

@@ -28,6 +28,7 @@ typedef struct {
   unsigned long long progress_bytes;
   bool sender_done;
   atomic_bool cancelled;
+  ProtocolSession allocation_session;
 } PipelineContextSender;
 
 typedef struct PipelineContextReceiver {

src/shared/protocol.h

@@ -19,6 +19,8 @@
 /* Aggregate bytes retained by one received deletion manifest. */
 #define MAX_MANIFEST_BYTES (16ULL * 1024 * 1024)
 #define DEFAULT_MAX_ALLOC (1ULL * 1024 * 1024 * 1024)
+/* Server policy ceiling for a client-provided allocation limit. */
+#define MAX_SERVER_ALLOC (256ULL * 1024 * 1024)
 
 typedef struct ssl_st SSL;