src/client/client_send.c

@@ -1157,6 +1157,7 @@ static int send_append(const Client* client, File* file, Config* config,
     tail_view.data = (char*)file->data->data + off;
     tail_view.size = tail_len;
     tail_view.protocol_charge = 0;
+    tail_view.owner = NULL;
     ok = send_data(fd, &tail_view);
   }
   return ok ? 0 : -1;

src/shared/data.c

@@ -23,6 +23,7 @@ Data* data_create_reserve(size_t size) {
   d->data = NULL;
   d->size = size;
   d->protocol_charge = 0;
+  d->owner = NULL;
   return d;
 }
 
@@ -36,14 +37,19 @@ Data* data_create(void* data, size_t data_size) {
   new_data->data = data;
   new_data->size = data_size;
   new_data->protocol_charge = 0;
+  new_data->owner = NULL;
   return new_data;
 }
 
 void data_destroy(Data* data) {
   if (data == NULL)
     return;
-  if (data->protocol_charge != 0)
+  if (data->protocol_charge != 0) {
+    if (data->owner != NULL)
+      protocol_release_memory_for_session(data->owner, data->protocol_charge);
+    else
       protocol_release_memory(data->protocol_charge);
+  }
   free(data->data);
   free(data);
 }

src/shared/data.h

@@ -3,11 +3,25 @@
 
 #include <stdlib.h>
 
+/* Forward declaration for the connection budget a received Data is charged
+ * against; defined in protocol.h (which includes this header). */
+typedef struct ProtocolSession ProtocolSession;
+
 typedef struct {
   void* data;
   size_t size;
   /* Non-zero only for a buffer charged to the protocol connection budget. */
   size_t protocol_charge;
+  /* Session whose budget `protocol_charge` was reserved from.  When non-NULL,
+   * the charge is returned to this session directly, regardless of which
+   * session (if any) is bound to the destroying thread.  owner is not
+   * guaranteed to be set whenever protocol_charge is non-zero: it is NULL for
+   * uncharged Data and for Data that has no recorded owner, in which case any
+   * charge falls back to the session bound at destroy time.
+   *
+   * Lifetime contract: a Data with a non-NULL owner must not outlive that
+   * ProtocolSession -- data_destroy dereferences owner to return the charge. */
+  ProtocolSession* owner;
 } Data;
 
 Data* data_create_empty(size_t data_size);
@@ -15,5 +29,9 @@ Data* data_create_reserve(size_t size);
 Data* data_create(void* data, size_t data_size);
 void data_destroy(Data* data);
 void protocol_release_memory(size_t charge);
+/* Release `charge` against `session` directly instead of the thread-local bound
+ * session.  Used by data_destroy to honor Data.owner; `session` must outlive
+ * the Data whose charge is being returned.  A NULL session is a no-op. */
+void protocol_release_memory_for_session(ProtocolSession* session, size_t charge);
 
 #endif

src/shared/protocol.c

@@ -40,7 +40,9 @@ static bool protocol_reserve_memory(ProtocolSession* session, size_t charge) {
   }
 }
 
-static void protocol_release_memory_for_session(ProtocolSession* session, size_t charge) {
+void protocol_release_memory_for_session(ProtocolSession* session, size_t charge) {
+  if (!session)
+    return;
   unsigned long long allocated = atomic_load(&session->total_allocated_bytes);
   while (true) {
     unsigned long long remaining = (unsigned long long)charge >= allocated ? 0 : allocated - charge;
@@ -573,6 +575,7 @@ Data* protocol_receive_data_limited(ProtocolSession* session, unsigned long long
     return NULL;
   }
   result->protocol_charge = allocation_size;
+  result->owner = session;
   return result;
 }
 

tests/test_protocol.c

@@ -412,6 +412,87 @@ static void test_protocol_accounting_release_does_not_underflow() {
   protocol_session_unbind();
 }
 
+/* A Data acquired on session A must return its connection-memory charge to A
+   regardless of what (if anything) is bound at destroy time.  The original bug
+   had two halves: destroying A's Data while a different session is bound leaks
+   A and drains the bound session, and destroying it with nothing bound leaks A
+   and drains the legacy fallback session. */
+static void test_receive_data_charge_follows_owning_session() {
+  int pipe_a[2];
+  int pipe_b[2];
+  EXPECT_EQ_INT(pipe(pipe_a), 0);
+  EXPECT_EQ_INT(pipe(pipe_b), 0);
+
+  ProtocolSession session_a;
+  ProtocolSession session_b;
+  protocol_session_init(&session_a, pipe_a[0], pipe_a[1]);
+  protocol_session_init(&session_b, pipe_b[0], pipe_b[1]);
+  protocol_session_set_max_alloc(&session_a, 64);
+  protocol_session_set_max_alloc(&session_b, 64);
+
+  unsigned long long size = 8;
+  EXPECT_EQ_INT((int)write(pipe_a[1], &size, sizeof(size)), (int)sizeof(size));
+  EXPECT_EQ_INT((int)write(pipe_a[1], "12345678", 8), 8);
+  EXPECT_EQ_INT((int)write(pipe_a[1], &size, sizeof(size)), (int)sizeof(size));
+  EXPECT_EQ_INT((int)write(pipe_a[1], "ABCDEFGH", 8), 8);
+  EXPECT_EQ_INT((int)write(pipe_b[1], &size, sizeof(size)), (int)sizeof(size));
+  EXPECT_EQ_INT((int)write(pipe_b[1], "abcdefgh", 8), 8);
+
+  Data* data_a1 = protocol_receive_data_limited(&session_a, 8);
+  Data* data_a2 = protocol_receive_data_limited(&session_a, 8);
+  Data* data_b = protocol_receive_data_limited(&session_b, 8);
+  EXPECT_NOT_NULL(data_a1);
+  EXPECT_NOT_NULL(data_a2);
+  EXPECT_NOT_NULL(data_b);
+  EXPECT_TRUE(data_a1->owner == &session_a);
+  EXPECT_TRUE(data_a2->owner == &session_a);
+  EXPECT_TRUE(data_b->owner == &session_b);
+  EXPECT_EQ_INT((int)atomic_load(&session_a.total_allocated_bytes), 16);
+  EXPECT_EQ_INT((int)atomic_load(&session_b.total_allocated_bytes), 8);
+
+  /* Half 1: destroy A's Data while the unrelated session B is bound.  The
+     charge must go to A, not to the bound B. */
+  protocol_session_bind(&session_b);
+  data_destroy(data_a1);
+  protocol_session_unbind();
+
+  EXPECT_EQ_INT((int)atomic_load(&session_a.total_allocated_bytes), 8);
+  EXPECT_EQ_INT((int)atomic_load(&session_b.total_allocated_bytes), 8);
+
+  /* Half 2: destroy A's remaining Data with NO session bound.  The charge must
+     still go to A, not to the legacy fallback session. */
+  protocol_session_unbind();
+  data_destroy(data_a2);
+  EXPECT_EQ_INT((int)atomic_load(&session_a.total_allocated_bytes), 0);
+  EXPECT_EQ_INT((int)atomic_load(&session_b.total_allocated_bytes), 8);
+
+  data_destroy(data_b);
+  EXPECT_EQ_INT((int)atomic_load(&session_b.total_allocated_bytes), 0);
+
+  close(pipe_a[0]);
+  close(pipe_a[1]);
+  close(pipe_b[0]);
+  close(pipe_b[1]);
+}
+
+/* Freshest Data holds no connection charge; only a bounded receive binds an
+   owner and a charge, so creation helpers must start uncharged and unowned. */
+static void test_data_create_starts_uncharged_and_unowned() {
+  void* buf = malloc(8);
+  EXPECT_NOT_NULL(buf);
+  Data* created = data_create(buf, 8);
+  EXPECT_NOT_NULL(created);
+  EXPECT_TRUE(created->owner == NULL);
+  EXPECT_EQ_INT((int)created->protocol_charge, 0);
+  data_destroy(created);
+
+  Data* reserved = data_create_reserve(64);
+  EXPECT_NOT_NULL(reserved);
+  EXPECT_TRUE(reserved->owner == NULL);
+  EXPECT_EQ_INT((int)reserved->protocol_charge, 0);
+  data_destroy(reserved);
+}
+
 static void test_protocol_session_io_timeout() {
   /* Default is the built-in 60 s window; the setter stores exactly what it is
    * given (<= 0 means "fall back to the default") so callers can propagate
@@ -574,4 +655,6 @@ void test_protocol() {
   test_protocol_accounting_reservation_is_atomic();
   test_protocol_string_accounting_is_transient();
   test_protocol_accounting_release_does_not_underflow();
+  test_receive_data_charge_follows_owning_session();
+  test_data_create_starts_uncharged_and_unowned();
 }