fix(pairing): destroy port backing before refunding capacity

Co-authored-by: Medulla <medulla@tinyhumans.ai>

Author: senamakel

crates/tinysecurity-crypto/src/pairing.rs

@@ -267,19 +267,29 @@ pub struct PortBuffer {
     inner: Arc<Inner>,
     account: usize,
     binding: [u8; 32],
-    bytes: StagingBytes,
+    bytes: Option<StagingBytes>,
     slot: usize,
     charge: usize,
+    #[cfg(test)]
+    after_drop_unlock: Option<Box<dyn FnOnce() + Send>>,
 }
 impl Drop for PortBuffer {
     fn drop(&mut self) {
         if let Ok(mut state) = self.inner.state.lock() {
-            state.pending_bytes = state.pending_bytes.saturating_sub(self.charge);
             if let Some(pending) = state.accounts[self.account].pending.as_mut()
                 && pending.binding == self.binding
             {
                 pending.buffers[self.slot] = None;
             }
+            // No backing Arc escapes these two private owners. Destroy the final
+            // Arc, its zeroizing Vec and allocation before admission can observe
+            // a refund; implicit field Drop would happen after unlocking State.
+            drop(self.bytes.take());
+            state.pending_bytes = state.pending_bytes.saturating_sub(self.charge);
+        }
+        #[cfg(test)]
+        if let Some(observe) = self.after_drop_unlock.take() {
+            observe();
         }
     }
 }
@@ -291,7 +301,12 @@ impl PortBuffer {
     pub fn with_bytes(&mut self, f: impl FnOnce(&mut [u8]) -> Result<()>) -> Result<()> {
         let mut state = self.inner.lock()?;
         self.inner.check(&mut state, self.account, &self.binding)?;
-        let mut bytes = self.bytes.lock().map_err(|_| Error::NativeDenied)?;
+        let mut bytes = self
+            .bytes
+            .as_ref()
+            .ok_or(Error::NativeDenied)?
+            .lock()
+            .map_err(|_| Error::NativeDenied)?;
         f(&mut bytes)
     }
 }
@@ -346,7 +361,12 @@ impl StagedDelivery {
             return Err(Error::NativeDenied);
         }
         self.with_staged_secret(|secret| {
-            let mut out = buffer.bytes.lock().map_err(|_| Error::NativeDenied)?;
+            let mut out = buffer
+                .bytes
+                .as_ref()
+                .ok_or(Error::NativeDenied)?
+                .lock()
+                .map_err(|_| Error::NativeDenied)?;
             if out.len() < secret.len() {
                 return Err(Error::NativeCapacity);
             }
@@ -629,9 +649,11 @@ impl Inner {
             inner: Arc::clone(self),
             account,
             binding: *binding,
-            bytes,
+            bytes: Some(bytes),
             slot,
             charge,
+            #[cfg(test)]
+            after_drop_unlock: None,
         })
     }
     fn lock(&self) -> Result<MutexGuard<'_, State>> {

crates/tinysecurity-crypto/src/pairing_tests.rs

@@ -1127,7 +1127,16 @@ async fn full_backing_charge_survives_revoke_and_releases_only_with_owned_buffer
             })
             .is_err()
     );
-    assert!(buffer.bytes.lock().checked().iter().all(|b| *b == 0));
+    assert!(
+        buffer
+            .bytes
+            .as_ref()
+            .checked()
+            .lock()
+            .checked()
+            .iter()
+            .all(|b| *b == 0)
+    );
     task.abort();
     let _ = task.await;
     assert_eq!(owner.inner.state.lock().checked().pending_bytes, charge);
@@ -1398,3 +1407,93 @@ async fn unpublished_retained_token_reservation_releases_once_on_cancel() {
     assert_eq!(state.reserved_tokens, 0);
     assert_eq!(state.tokens, 0);
 }
+
+#[tokio::test]
+async fn refunded_port_capacity_is_reusable_only_after_backing_destruction() {
+    let (_, clock, _, verifier, entropy) = setup(true, vec![], Limits::default());
+    let mut second = registration(true, vec![]);
+    second.account = "second".into();
+    let owner = Arc::new(
+        PairingOwner::construct(
+            vec![registration(true, vec![]), second],
+            1,
+            Limits {
+                pending_bytes: 2 * OP_BYTES + 8192 + 256,
+                ..Limits::default()
+            },
+            300_000,
+            clock,
+            Arc::new(AggregateConsumer),
+            verifier.clone(),
+            entropy,
+        )
+        .checked(),
+    );
+    let first = reference(&owner).await;
+    let other = TrustedAccountRequest {
+        account: "second",
+        ..account_request()
+    };
+    let second = match owner.begin(&other).await.checked() {
+        BeginOutcome::Active(reference) => reference,
+        BeginOutcome::Inactive => unreachable!("second challenge must be active"),
+    };
+    verifier.pause.store(true, Ordering::SeqCst);
+    let mut first_work = Box::pin(owner.verify(attempt(&first, "000000")));
+    let mut second_attempt = attempt(&second, "000000");
+    second_attempt.account = other;
+    let mut second_work = Box::pin(owner.verify(second_attempt));
+    for work in [&mut first_work, &mut second_work] {
+        let polled =
+            std::future::poll_fn(|cx| std::task::Poll::Ready(work.as_mut().poll(cx))).await;
+        assert!(polled.is_pending());
+    }
+    let deliveries: Vec<_> = (0..2)
+        .map(|account| StagedDelivery {
+            inner: owner.inner.clone(),
+            account,
+            binding: owner.inner.state.lock().checked().accounts[account]
+                .pending
+                .as_ref()
+                .checked()
+                .binding,
+        })
+        .collect();
+    let mut first_buffer = deliveries[0].allocate_buffer(8192).checked();
+    first_buffer
+        .with_bytes(|bytes| {
+            bytes.fill(7);
+            Ok(())
+        })
+        .checked();
+    assert!(matches!(
+        deliveries[1].allocate_buffer(8192),
+        Err(Error::NativeCapacity)
+    ));
+    // Weak observation adds no strong backing owner. This callback is the exact
+    // preemption boundary after unlocking State and before implicit field Drop;
+    // it runs outside the lock, with both accounts' original work still pending.
+    let first_backing = Arc::downgrade(first_buffer.bytes.as_ref().checked());
+    let next_delivery = StagedDelivery {
+        inner: owner.inner.clone(),
+        account: 1,
+        binding: deliveries[1].binding,
+    };
+    let observed = Arc::new(AtomicBool::new(false));
+    let observed_drop = observed.clone();
+    first_buffer.after_drop_unlock = Some(Box::new(move || {
+        let next_buffer = next_delivery.allocate_buffer(8192).checked();
+        assert_eq!(
+            first_backing.strong_count(),
+            0,
+            "refunded capacity admitted another backing before the old backing was destroyed"
+        );
+        observed_drop.store(true, Ordering::SeqCst);
+        drop(next_buffer);
+    }));
+    drop(first_buffer);
+    assert!(observed.load(Ordering::SeqCst));
+    drop(first_work);
+    drop(second_work);
+    assert_eq!(owner.inner.state.lock().checked().pending_bytes, 0);
+}