refactor(protocol): guard session release, clarify Data.owner contract

Add a NULL guard to protocol_release_memory_for_session so it no-ops like
the sibling session setters.  Correct the Data.owner doc comment, which
implied a non-zero protocol_charge always has an owner; document that
owner may be NULL for uncharged/ownerless Data, that any such charge
falls back to the bound session, and that a charged Data must not outlive
its owning session.  Note the lifetime contract on the release API too.

Extend tests/test_protocol.c to cover destroying a charged Data with no
session bound (the other half of the original bug) and to assert that
data_create/data_create_reserve start with owner == NULL and
protocol_charge == 0.

src/shared/data.h

@@ -12,9 +12,15 @@ typedef struct {
   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.  The charge must
-   * always be returned to this session, regardless of which session (if any) is
-   * bound to the destroying thread.  NULL for uncharged Data. */
+  /* 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;
 
@@ -24,7 +30,8 @@ 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.  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

@@ -41,6 +41,8 @@ static bool protocol_reserve_memory(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;

tests/test_protocol.c

@@ -413,8 +413,10 @@ static void test_protocol_accounting_release_does_not_underflow() {
 }
 
 /* A Data acquired on session A must return its connection-memory charge to A
-   even when a different session B is bound at destroy time: releasing against
-   the thread-local bound session would leak A's budget and drain B's. */
+   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];
@@ -431,23 +433,36 @@ static void test_receive_data_charge_follows_owning_session() {
   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_a = protocol_receive_data_limited(&session_a, 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_a);
+  EXPECT_NOT_NULL(data_a1);
+  EXPECT_NOT_NULL(data_a2);
   EXPECT_NOT_NULL(data_b);
-  EXPECT_TRUE(data_a->owner == &session_a);
+  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);
 
-  /* Destroy A's Data while the unrelated session B is the bound session. */
-  protocol_session_bind(&session_b);
-  data_destroy(data_a);
+  /* 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);
 
@@ -460,6 +475,24 @@ static void test_receive_data_charge_follows_owning_session() {
   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
@@ -623,4 +656,5 @@ void test_protocol() {
   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();
 }