[Private GATT] Refactor GattDatastore

Split into RawGattDatastore and GattDatastore, so we can add read blob /
reliable writes without making internal services more messy.

Test: unit
Bug: 255880936
Change-Id: Ie165fa45fca2a8e003da39f6515e3c41605fed43

12 files changed, 791 insertions, 125 deletions

diff --git a/system/rust/src/gatt/callbacks.rs b/system/rust/src/gatt/callbacks.rs
index d58a4d256d..3f342d1626 100644
--- a/system/rust/src/gatt/callbacks.rs
+++ b/system/rust/src/gatt/callbacks.rs
@@ -7,6 +7,7 @@ mod callback_transaction_manager;
 pub use callback_transaction_manager::{CallbackResponseError, CallbackTransactionManager};
 
 use async_trait::async_trait;
+use log::warn;
 
 use crate::packets::{AttAttributeDataChild, AttAttributeDataView, AttErrorCode};
 
@@ -28,7 +29,6 @@ pub trait GattCallbacks {
         handle: AttHandle,
         attr_type: AttributeBackingType,
         offset: u32,
-        is_long: bool,
     );
 
     /// Invoked when a client tries to write a characteristic/descriptor.
@@ -40,9 +40,7 @@ pub trait GattCallbacks {
         trans_id: TransactionId,
         handle: AttHandle,
         attr_type: AttributeBackingType,
-        offset: u32,
-        need_response: bool,
-        is_prepare: bool,
+        write_type: GattWriteType,
         value: AttAttributeDataView,
     );
 
@@ -53,11 +51,90 @@ pub trait GattCallbacks {
         conn_id: ConnectionId,
         result: Result<(), IndicationError>,
     );
+
+    /// Execute or cancel any prepared writes
+    fn on_execute(
+        &self,
+        conn_id: ConnectionId,
+        trans_id: TransactionId,
+        decision: TransactionDecision,
+    );
+}
+
+/// The various write types available (requests + commands)
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+pub enum GattWriteType {
+    /// Reliable, expects a response (WRITE_REQ or PREPARE_WRITE_REQ)
+    Request(GattWriteRequestType),
+    /// Unreliable, no response required (WRITE_CMD)
+    Command,
+}
+
+/// The types of write requests (that need responses)
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+pub enum GattWriteRequestType {
+    /// Atomic (WRITE_REQ)
+    Request,
+    /// Transactional, should not be committed yet (PREPARE_WRITE_REQ)
+    Prepare {
+        /// The byte offset at which to write
+        offset: u32,
+    },
+}
+
+/// Whether to commit or cancel a transaction
+#[derive(Clone, Copy, Debug)]
+pub enum TransactionDecision {
+    /// Commit all pending writes
+    Execute,
+    /// Discard all pending writes
+    Cancel,
 }
 
 /// This interface is an "async" version of the above, and is passed directly
 /// into the GattModule
 #[async_trait(?Send)]
+pub trait RawGattDatastore {
+    /// Read a characteristic from the specified connection at the given handle.
+    async fn read(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        offset: u32,
+        attr_type: AttributeBackingType,
+    ) -> Result<AttAttributeDataChild, AttErrorCode>;
+
+    /// Write data to a given characteristic on the specified connection.
+    async fn write(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        attr_type: AttributeBackingType,
+        write_type: GattWriteRequestType,
+        data: AttAttributeDataView<'_>,
+    ) -> Result<(), AttErrorCode>;
+
+    /// Write data to a given characteristic on the specified connection, without waiting
+    /// for a response from the upper layer.
+    fn write_no_response(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        attr_type: AttributeBackingType,
+        data: AttAttributeDataView<'_>,
+    );
+
+    /// Execute or cancel any prepared writes
+    async fn execute(
+        &self,
+        conn_id: ConnectionId,
+        decision: TransactionDecision,
+    ) -> Result<(), AttErrorCode>;
+}
+
+/// This interface simplifies the interface of RawGattDatastore by rejecting all unsupported
+/// operations, rather than requiring clients to do so.
+#[async_trait(?Send)]
 pub trait GattDatastore {
     /// Read a characteristic from the specified connection at the given handle.
     async fn read(
@@ -76,3 +153,275 @@ pub trait GattDatastore {
         data: AttAttributeDataView<'_>,
     ) -> Result<(), AttErrorCode>;
 }
+
+#[async_trait(?Send)]
+impl<T: GattDatastore + ?Sized> RawGattDatastore for T {
+    /// Read a characteristic from the specified connection at the given handle.
+    async fn read(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        offset: u32,
+        attr_type: AttributeBackingType,
+    ) -> Result<AttAttributeDataChild, AttErrorCode> {
+        if offset != 0 {
+            warn!("got read blob request for non-long attribute {handle:?}");
+            return Err(AttErrorCode::ATTRIBUTE_NOT_LONG);
+        }
+        self.read(conn_id, handle, attr_type).await
+    }
+
+    /// Write data to a given characteristic on the specified connection.
+    async fn write(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        attr_type: AttributeBackingType,
+        write_type: GattWriteRequestType,
+        data: AttAttributeDataView<'_>,
+    ) -> Result<(), AttErrorCode> {
+        match write_type {
+            GattWriteRequestType::Prepare { .. } => {
+                warn!("got prepare write attempt to characteristic {handle:?} not supporting write_without_response");
+                Err(AttErrorCode::WRITE_REQUEST_REJECTED)
+            }
+            GattWriteRequestType::Request => self.write(conn_id, handle, attr_type, data).await,
+        }
+    }
+
+    fn write_no_response(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        _: AttributeBackingType,
+        _: AttAttributeDataView<'_>,
+    ) {
+        // silently drop, since there's no way to return an error
+        warn!("got write command on {conn_id:?} to characteristic {handle:?} not supporting write_without_response");
+    }
+
+    /// Execute or cancel any prepared writes
+    async fn execute(&self, _: ConnectionId, _: TransactionDecision) -> Result<(), AttErrorCode> {
+        // we never do prepared writes, so who cares
+        return Ok(());
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use tokio::{sync::mpsc::error::TryRecvError, task::spawn_local};
+
+    use crate::{
+        gatt::mocks::mock_datastore::{MockDatastore, MockDatastoreEvents},
+        packets::OwnedAttAttributeDataView,
+        utils::{
+            packet::{build_att_data, build_view_or_crash},
+            task::block_on_locally,
+        },
+    };
+
+    use super::*;
+
+    const CONN_ID: ConnectionId = ConnectionId(1);
+    const HANDLE: AttHandle = AttHandle(1);
+    const DATA: [u8; 4] = [1, 2, 3, 4];
+
+    #[test]
+    fn test_regular_read_invoke() {
+        block_on_locally(async {
+            // arrange
+            let (datastore, mut rx) = MockDatastore::new();
+
+            // act: send read request
+            spawn_local(async move {
+                RawGattDatastore::read(
+                    &datastore,
+                    CONN_ID,
+                    HANDLE,
+                    0,
+                    AttributeBackingType::Characteristic,
+                )
+                .await
+            });
+            let resp = rx.recv().await.unwrap();
+
+            // assert: got read event
+            assert!(matches!(
+                resp,
+                MockDatastoreEvents::Read(CONN_ID, HANDLE, AttributeBackingType::Characteristic, _)
+            ));
+        });
+    }
+
+    #[test]
+    fn test_regular_read_response() {
+        block_on_locally(async {
+            // arrange
+            let (datastore, mut rx) = MockDatastore::new();
+
+            // act: send read request
+            let pending = spawn_local(async move {
+                RawGattDatastore::read(
+                    &datastore,
+                    CONN_ID,
+                    HANDLE,
+                    0,
+                    AttributeBackingType::Characteristic,
+                )
+                .await
+            });
+            let resp = rx.recv().await.unwrap();
+            let MockDatastoreEvents::Read(_, _, _, resp) = resp else {
+                unreachable!();
+            };
+            resp.send(Err(AttErrorCode::APPLICATION_ERROR)).unwrap();
+
+            // assert: got the supplied response
+            assert_eq!(pending.await.unwrap(), Err(AttErrorCode::APPLICATION_ERROR));
+        });
+    }
+
+    #[test]
+    fn test_rejected_read_blob() {
+        // arrange
+        let (datastore, mut rx) = MockDatastore::new();
+
+        // act: send read blob request
+        let resp = block_on_locally(RawGattDatastore::read(
+            &datastore,
+            CONN_ID,
+            HANDLE,
+            1,
+            AttributeBackingType::Characteristic,
+        ));
+
+        // assert: got the correct error code
+        assert_eq!(resp, Err(AttErrorCode::ATTRIBUTE_NOT_LONG));
+        // assert: no pending events
+        assert_eq!(rx.try_recv().unwrap_err(), TryRecvError::Empty);
+    }
+
+    fn make_data() -> OwnedAttAttributeDataView {
+        build_view_or_crash(build_att_data(AttAttributeDataChild::RawData(DATA.into())))
+    }
+
+    #[test]
+    fn test_write_request_invoke() {
+        block_on_locally(async {
+            // arrange
+            let (datastore, mut rx) = MockDatastore::new();
+
+            // act: send write request
+            spawn_local(async move {
+                RawGattDatastore::write(
+                    &datastore,
+                    CONN_ID,
+                    HANDLE,
+                    AttributeBackingType::Characteristic,
+                    GattWriteRequestType::Request,
+                    make_data().view(),
+                )
+                .await
+            });
+            let resp = rx.recv().await.unwrap();
+
+            // assert: got write event
+            assert!(matches!(
+                resp,
+                MockDatastoreEvents::Write(
+                    CONN_ID,
+                    HANDLE,
+                    AttributeBackingType::Characteristic,
+                    _,
+                    _
+                )
+            ));
+        });
+    }
+
+    #[test]
+    fn test_write_request_response() {
+        block_on_locally(async {
+            // arrange
+            let (datastore, mut rx) = MockDatastore::new();
+
+            // act: send write request
+            let pending = spawn_local(async move {
+                RawGattDatastore::write(
+                    &datastore,
+                    CONN_ID,
+                    HANDLE,
+                    AttributeBackingType::Characteristic,
+                    GattWriteRequestType::Request,
+                    make_data().view(),
+                )
+                .await
+            });
+            let resp = rx.recv().await.unwrap();
+            let MockDatastoreEvents::Write(_, _, _, _, resp) = resp else {
+                unreachable!();
+            };
+            resp.send(Err(AttErrorCode::APPLICATION_ERROR)).unwrap();
+
+            // assert: got the supplied response
+            assert_eq!(pending.await.unwrap(), Err(AttErrorCode::APPLICATION_ERROR));
+        });
+    }
+
+    #[test]
+    fn test_rejected_prepared_write() {
+        // arrange
+        let (datastore, mut rx) = MockDatastore::new();
+
+        // act: send prepare write request
+        let resp = block_on_locally(RawGattDatastore::write(
+            &datastore,
+            CONN_ID,
+            HANDLE,
+            AttributeBackingType::Characteristic,
+            GattWriteRequestType::Prepare { offset: 1 },
+            make_data().view(),
+        ));
+
+        // assert: got the correct error code
+        assert_eq!(resp, Err(AttErrorCode::WRITE_REQUEST_REJECTED));
+        // assert: no event sent up
+        assert_eq!(rx.try_recv().unwrap_err(), TryRecvError::Empty);
+    }
+
+    #[test]
+    fn test_dropped_write_command() {
+        // arrange
+        let (datastore, mut rx) = MockDatastore::new();
+
+        // act: send write command
+        RawGattDatastore::write_no_response(
+            &datastore,
+            CONN_ID,
+            HANDLE,
+            AttributeBackingType::Characteristic,
+            make_data().view(),
+        );
+
+        // assert: no event sent up
+        assert_eq!(rx.try_recv().unwrap_err(), TryRecvError::Empty);
+    }
+
+    #[test]
+    fn test_execute_noop() {
+        // arrange
+        let (datastore, mut rx) = MockDatastore::new();
+
+        // act: send execute request
+        let resp = block_on_locally(RawGattDatastore::execute(
+            &datastore,
+            CONN_ID,
+            TransactionDecision::Execute,
+        ));
+
+        // assert: succeeds trivially
+        assert!(resp.is_ok());
+        // assert: no event sent up
+        assert_eq!(rx.try_recv().unwrap_err(), TryRecvError::Empty);
+    }
+}
diff --git a/system/rust/src/gatt/callbacks/callback_transaction_manager.rs b/system/rust/src/gatt/callbacks/callback_transaction_manager.rs
index e5cd231417..153a1f2c20 100644
--- a/system/rust/src/gatt/callbacks/callback_transaction_manager.rs
+++ b/system/rust/src/gatt/callbacks/callback_transaction_manager.rs
@@ -1,7 +1,7 @@
 use std::{cell::RefCell, collections::HashMap, rc::Rc, time::Duration};
 
 use async_trait::async_trait;
-use log::{error, trace, warn};
+use log::{trace, warn};
 use tokio::{sync::oneshot, time::timeout};
 
 use crate::{
@@ -12,7 +12,10 @@ use crate::{
     packets::{AttAttributeDataChild, AttAttributeDataView, AttErrorCode},
 };
 
-use super::{AttributeBackingType, GattDatastore};
+use super::{
+    AttributeBackingType, GattWriteRequestType, GattWriteType, RawGattDatastore,
+    TransactionDecision,
+};
 
 struct PendingTransaction {
     response: oneshot::Sender<Result<AttAttributeDataChild, AttErrorCode>>,
@@ -25,33 +28,6 @@ struct PendingTransactionWatcher {
     rx: oneshot::Receiver<Result<AttAttributeDataChild, AttErrorCode>>,
 }
 
-enum PendingTransactionError {
-    SenderDropped,
-    Timeout,
-}
-
-impl PendingTransactionWatcher {
-    /// Wait for the transaction to resolve, or to hit the timeout. If the
-    /// timeout is reached, clean up state related to transaction watching.
-    async fn wait(
-        self,
-        manager: &CallbackTransactionManager,
-    ) -> Result<Result<AttAttributeDataChild, AttErrorCode>, PendingTransactionError> {
-        match timeout(TIMEOUT, self.rx).await {
-            Ok(Ok(result)) => Ok(result),
-            Ok(Err(_)) => Err(PendingTransactionError::SenderDropped),
-            Err(_) => {
-                manager
-                    .pending_transactions
-                    .borrow_mut()
-                    .pending_transactions
-                    .remove(&(self.conn_id, self.trans_id));
-                Err(PendingTransactionError::Timeout)
-            }
-        }
-    }
-}
-
 /// This struct converts the asynchronus read/write operations of GattDatastore
 /// into the callback-based interface expected by JNI
 pub struct CallbackTransactionManager {
@@ -113,41 +89,57 @@ impl CallbackTransactionManager {
 }
 
 impl PendingTransactionsState {
-    fn start_new_transaction(&mut self, conn_id: ConnectionId) -> PendingTransactionWatcher {
+    fn alloc_transaction_id(&mut self) -> TransactionId {
         let trans_id = TransactionId(self.next_transaction_id);
         self.next_transaction_id = self.next_transaction_id.wrapping_add(1);
+        trans_id
+    }
 
+    fn start_new_transaction(&mut self, conn_id: ConnectionId) -> PendingTransactionWatcher {
+        let trans_id = self.alloc_transaction_id();
         let (tx, rx) = oneshot::channel();
         self.pending_transactions.insert((conn_id, trans_id), PendingTransaction { response: tx });
         PendingTransactionWatcher { conn_id, trans_id, rx }
     }
 }
 
+impl PendingTransactionWatcher {
+    /// Wait for the transaction to resolve, or to hit the timeout. If the
+    /// timeout is reached, clean up state related to transaction watching.
+    async fn wait(
+        self,
+        manager: &CallbackTransactionManager,
+    ) -> Result<AttAttributeDataChild, AttErrorCode> {
+        if let Ok(Ok(result)) = timeout(TIMEOUT, self.rx).await {
+            result
+        } else {
+            manager
+                .pending_transactions
+                .borrow_mut()
+                .pending_transactions
+                .remove(&(self.conn_id, self.trans_id));
+            warn!("no response received from Java after timeout - returning UNLIKELY_ERROR");
+            Err(AttErrorCode::UNLIKELY_ERROR)
+        }
+    }
+}
+
 #[async_trait(?Send)]
-impl GattDatastore for CallbackTransactionManager {
+impl RawGattDatastore for CallbackTransactionManager {
     async fn read(
         &self,
         conn_id: ConnectionId,
         handle: AttHandle,
+        offset: u32,
         attr_type: AttributeBackingType,
     ) -> Result<AttAttributeDataChild, AttErrorCode> {
         let pending_transaction =
             self.pending_transactions.borrow_mut().start_new_transaction(conn_id);
         let trans_id = pending_transaction.trans_id;
 
-        self.callbacks.on_server_read(conn_id, trans_id, handle, attr_type, 0, false);
+        self.callbacks.on_server_read(conn_id, trans_id, handle, attr_type, offset);
 
-        match pending_transaction.wait(self).await {
-            Ok(value) => value,
-            Err(PendingTransactionError::SenderDropped) => {
-                warn!("sender side of {trans_id:?} dropped / timed out while handling request - most likely this response will not be sent over the air");
-                Err(AttErrorCode::UNLIKELY_ERROR)
-            }
-            Err(PendingTransactionError::Timeout) => {
-                warn!("no response received from Java after timeout - returning UNLIKELY_ERROR");
-                Err(AttErrorCode::UNLIKELY_ERROR)
-            }
-        }
+        pending_transaction.wait(self).await
     }
 
     async fn write(
@@ -155,25 +147,56 @@ impl GattDatastore for CallbackTransactionManager {
         conn_id: ConnectionId,
         handle: AttHandle,
         attr_type: AttributeBackingType,
+        write_type: GattWriteRequestType,
         data: AttAttributeDataView<'_>,
     ) -> Result<(), AttErrorCode> {
         let pending_transaction =
             self.pending_transactions.borrow_mut().start_new_transaction(conn_id);
         let trans_id = pending_transaction.trans_id;
 
-        self.callbacks.on_server_write(conn_id, trans_id, handle, attr_type, 0, true, false, data);
+        self.callbacks.on_server_write(
+            conn_id,
+            trans_id,
+            handle,
+            attr_type,
+            GattWriteType::Request(write_type),
+            data,
+        );
+
+        // the data passed back is irrelevant for write requests
+        pending_transaction.wait(self).await.map(|_| ())
+    }
 
-        match pending_transaction.wait(self).await {
-            Ok(value) => value.map(|_| ()), // the data passed back is irrelevant for write
-            // requests
-            Err(PendingTransactionError::SenderDropped) => {
-                error!("the CallbackTransactionManager dropped the sender TX without sending it");
-                Err(AttErrorCode::UNLIKELY_ERROR)
-            }
-            Err(PendingTransactionError::Timeout) => {
-                warn!("no response received from Java after timeout - returning UNLIKELY_ERROR");
-                Err(AttErrorCode::UNLIKELY_ERROR)
-            }
-        }
+    fn write_no_response(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        attr_type: AttributeBackingType,
+        data: AttAttributeDataView<'_>,
+    ) {
+        let trans_id = self.pending_transactions.borrow_mut().alloc_transaction_id();
+        self.callbacks.on_server_write(
+            conn_id,
+            trans_id,
+            handle,
+            attr_type,
+            GattWriteType::Command,
+            data,
+        );
+    }
+
+    async fn execute(
+        &self,
+        conn_id: ConnectionId,
+        decision: TransactionDecision,
+    ) -> Result<(), AttErrorCode> {
+        let pending_transaction =
+            self.pending_transactions.borrow_mut().start_new_transaction(conn_id);
+        let trans_id = pending_transaction.trans_id;
+
+        self.callbacks.on_execute(conn_id, trans_id, decision);
+
+        // the data passed back is irrelevant for execute requests
+        pending_transaction.wait(self).await.map(|_| ())
     }
 }
diff --git a/system/rust/src/gatt/ffi.rs b/system/rust/src/gatt/ffi.rs
index e9338ef267..1fb416cf97 100644
--- a/system/rust/src/gatt/ffi.rs
+++ b/system/rust/src/gatt/ffi.rs
@@ -22,6 +22,7 @@ use crate::{
 
 use super::{
     arbiter::{self, with_arbiter},
+    callbacks::{GattWriteRequestType, GattWriteType, TransactionDecision},
     channel::AttTransport,
     ids::{AdvertiserId, AttHandle, ConnectionId, ServerId, TransactionId, TransportIndex},
     server::{
@@ -97,6 +98,10 @@ mod inner {
             value: &[u8],
         );
 
+        /// This callback is invoked when executing / cancelling a write
+        #[cxx_name = "OnExecute"]
+        fn on_execute(self: &GattServerCallbacks, conn_id: u16, trans_id: u32, execute: bool);
+
         /// This callback is invoked when an indication has been sent and the
         /// peer device has confirmed it, or if some error occurred.
         #[cxx_name = "OnIndicationSentConfirmation"]
@@ -190,12 +195,15 @@ impl GattCallbacks for GattCallbacksImpl {
         handle: AttHandle,
         attr_type: AttributeBackingType,
         offset: u32,
-        is_long: bool,
     ) {
-        self.0
-            .as_ref()
-            .unwrap()
-            .on_server_read(conn_id.0, trans_id.0, handle.0, attr_type, offset, is_long);
+        self.0.as_ref().unwrap().on_server_read(
+            conn_id.0,
+            trans_id.0,
+            handle.0,
+            attr_type,
+            offset,
+            offset != 0,
+        );
     }
 
     fn on_server_write(
@@ -204,9 +212,7 @@ impl GattCallbacks for GattCallbacksImpl {
         trans_id: TransactionId,
         handle: AttHandle,
         attr_type: AttributeBackingType,
-        offset: u32,
-        need_response: bool,
-        is_prepare: bool,
+        write_type: GattWriteType,
         value: AttAttributeDataView,
     ) {
         self.0.as_ref().unwrap().on_server_write(
@@ -214,9 +220,12 @@ impl GattCallbacks for GattCallbacksImpl {
             trans_id.0,
             handle.0,
             attr_type,
-            offset,
-            need_response,
-            is_prepare,
+            match write_type {
+                GattWriteType::Request(GattWriteRequestType::Prepare { offset }) => offset,
+                _ => 0,
+            },
+            matches!(write_type, GattWriteType::Request { .. }),
+            matches!(write_type, GattWriteType::Request(GattWriteRequestType::Prepare { .. })),
             &value.get_raw_payload().collect::<Vec<_>>(),
         );
     }
@@ -234,6 +243,22 @@ impl GattCallbacks for GattCallbacksImpl {
             },
         )
     }
+
+    fn on_execute(
+        &self,
+        conn_id: ConnectionId,
+        trans_id: TransactionId,
+        decision: TransactionDecision,
+    ) {
+        self.0.as_ref().unwrap().on_execute(
+            conn_id.0,
+            trans_id.0,
+            match decision {
+                TransactionDecision::Execute => true,
+                TransactionDecision::Cancel => false,
+            },
+        )
+    }
 }
 
 /// Implementation of AttTransport wrapping the corresponding C++ method
diff --git a/system/rust/src/gatt/ffi/gatt_shim.cc b/system/rust/src/gatt/ffi/gatt_shim.cc
index 3957188844..128b591fec 100644
--- a/system/rust/src/gatt/ffi/gatt_shim.cc
+++ b/system/rust/src/gatt/ffi/gatt_shim.cc
@@ -121,5 +121,19 @@ void GattServerCallbacks::OnIndicationSentConfirmation(uint16_t conn_id,
                    base::Bind(callbacks.indication_sent_cb, conn_id, status));
 }
 
+void GattServerCallbacks::OnExecute(uint16_t conn_id, uint32_t trans_id,
+                                    bool execute) const {
+  auto addr = AddressOfConnection(conn_id);
+  if (!addr.has_value()) {
+    LOG_WARN("Dropping server execute write since connection %d not found",
+             conn_id);
+    return;
+  }
+
+  do_in_jni_thread(
+      FROM_HERE, base::Bind(callbacks.request_exec_write_cb, conn_id, trans_id,
+                            addr.value(), execute));
+}
+
 }  // namespace gatt
 }  // namespace bluetooth
diff --git a/system/rust/src/gatt/ffi/gatt_shim.h b/system/rust/src/gatt/ffi/gatt_shim.h
index 5c42ee2c02..6c7532fc01 100644
--- a/system/rust/src/gatt/ffi/gatt_shim.h
+++ b/system/rust/src/gatt/ffi/gatt_shim.h
@@ -50,6 +50,8 @@ class GattServerCallbacks {
 
   void OnIndicationSentConfirmation(uint16_t conn_id, int status) const;
 
+  void OnExecute(uint16_t conn_id, uint32_t trans_id, bool execute) const;
+
  private:
   const btgatt_server_callbacks_t& callbacks;
 };
diff --git a/system/rust/src/gatt/mocks.rs b/system/rust/src/gatt/mocks.rs
index cd7907ef23..53dbda316a 100644
--- a/system/rust/src/gatt/mocks.rs
+++ b/system/rust/src/gatt/mocks.rs
@@ -2,4 +2,5 @@
 pub mod mock_callbacks;
 pub mod mock_database_callbacks;
 pub mod mock_datastore;
+pub mod mock_raw_datastore;
 pub mod mock_transport;
diff --git a/system/rust/src/gatt/mocks/mock_callbacks.rs b/system/rust/src/gatt/mocks/mock_callbacks.rs
index 3c05ad2c9c..a5dbee253f 100644
--- a/system/rust/src/gatt/mocks/mock_callbacks.rs
+++ b/system/rust/src/gatt/mocks/mock_callbacks.rs
@@ -2,6 +2,7 @@
 
 use crate::{
     gatt::{
+        callbacks::{GattWriteType, TransactionDecision},
         ffi::AttributeBackingType,
         ids::{AttHandle, ConnectionId, TransactionId},
         server::IndicationError,
@@ -26,20 +27,20 @@ impl MockCallbacks {
 #[derive(Debug)]
 pub enum MockCallbackEvents {
     /// GattCallbacks#on_server_read_characteristic invoked
-    OnServerRead(ConnectionId, TransactionId, AttHandle, AttributeBackingType, u32, bool),
+    OnServerRead(ConnectionId, TransactionId, AttHandle, AttributeBackingType, u32),
     /// GattCallbacks#on_server_write_characteristic invoked
     OnServerWrite(
         ConnectionId,
         TransactionId,
         AttHandle,
         AttributeBackingType,
-        u32,
-        bool,
-        bool,
+        GattWriteType,
         OwnedAttAttributeDataView,
     ),
     /// GattCallbacks#on_indication_sent_confirmation invoked
     OnIndicationSentConfirmation(ConnectionId, Result<(), IndicationError>),
+    /// GattCallbacks#on_execute invoked
+    OnExecute(ConnectionId, TransactionId, TransactionDecision),
 }
 
 impl GattCallbacks for MockCallbacks {
@@ -50,12 +51,9 @@ impl GattCallbacks for MockCallbacks {
         handle: AttHandle,
         attr_type: AttributeBackingType,
         offset: u32,
-        is_long: bool,
     ) {
         self.0
-            .send(MockCallbackEvents::OnServerRead(
-                conn_id, trans_id, handle, attr_type, offset, is_long,
-            ))
+            .send(MockCallbackEvents::OnServerRead(conn_id, trans_id, handle, attr_type, offset))
             .unwrap();
     }
 
@@ -65,9 +63,7 @@ impl GattCallbacks for MockCallbacks {
         trans_id: TransactionId,
         handle: AttHandle,
         attr_type: AttributeBackingType,
-        offset: u32,
-        need_response: bool,
-        is_prepare: bool,
+        write_type: GattWriteType,
         value: AttAttributeDataView,
     ) {
         self.0
@@ -76,9 +72,7 @@ impl GattCallbacks for MockCallbacks {
                 trans_id,
                 handle,
                 attr_type,
-                offset,
-                need_response,
-                is_prepare,
+                write_type,
                 value.to_owned_packet(),
             ))
             .unwrap();
@@ -91,4 +85,13 @@ impl GattCallbacks for MockCallbacks {
     ) {
         self.0.send(MockCallbackEvents::OnIndicationSentConfirmation(conn_id, result)).unwrap();
     }
+
+    fn on_execute(
+        &self,
+        conn_id: ConnectionId,
+        trans_id: TransactionId,
+        decision: TransactionDecision,
+    ) {
+        self.0.send(MockCallbackEvents::OnExecute(conn_id, trans_id, decision)).unwrap()
+    }
 }
diff --git a/system/rust/src/gatt/mocks/mock_raw_datastore.rs b/system/rust/src/gatt/mocks/mock_raw_datastore.rs
new file mode 100644
index 0000000000..28aeac4131
--- /dev/null
+++ b/system/rust/src/gatt/mocks/mock_raw_datastore.rs
@@ -0,0 +1,125 @@
+//! Mocked implementation of GattDatastore for use in test
+
+use crate::{
+    gatt::{
+        callbacks::{GattWriteRequestType, RawGattDatastore, TransactionDecision},
+        ffi::AttributeBackingType,
+        ids::{AttHandle, ConnectionId},
+    },
+    packets::{
+        AttAttributeDataChild, AttAttributeDataView, AttErrorCode, OwnedAttAttributeDataView,
+        Packet,
+    },
+};
+use async_trait::async_trait;
+use log::info;
+use tokio::sync::{
+    mpsc::{self, unbounded_channel, UnboundedReceiver},
+    oneshot,
+};
+
+/// Routes calls to RawGattDatastore into a channel of MockRawDatastoreEvents
+pub struct MockRawDatastore(mpsc::UnboundedSender<MockRawDatastoreEvents>);
+
+impl MockRawDatastore {
+    /// Constructor. Returns self and the RX side of the associated channel.
+    pub fn new() -> (Self, UnboundedReceiver<MockRawDatastoreEvents>) {
+        let (tx, rx) = unbounded_channel();
+        (Self(tx), rx)
+    }
+}
+
+/// Events representing calls to GattDatastore
+#[derive(Debug)]
+pub enum MockRawDatastoreEvents {
+    /// A characteristic was read on a given handle. The oneshot is used to
+    /// return the value read.
+    Read(
+        ConnectionId,
+        AttHandle,
+        AttributeBackingType,
+        u32,
+        oneshot::Sender<Result<AttAttributeDataChild, AttErrorCode>>,
+    ),
+    /// A characteristic was written to on a given handle. The oneshot is used
+    /// to return whether the write succeeded.
+    Write(
+        ConnectionId,
+        AttHandle,
+        AttributeBackingType,
+        GattWriteRequestType,
+        OwnedAttAttributeDataView,
+        oneshot::Sender<Result<(), AttErrorCode>>,
+    ),
+    /// A characteristic was written to on a given handle, where the response was disregarded.
+    WriteNoResponse(ConnectionId, AttHandle, AttributeBackingType, OwnedAttAttributeDataView),
+    /// The prepared writes have been committed / aborted. The oneshot is used
+    /// to return whether this operation succeeded.
+    Execute(ConnectionId, TransactionDecision, oneshot::Sender<Result<(), AttErrorCode>>),
+}
+
+#[async_trait(?Send)]
+impl RawGattDatastore for MockRawDatastore {
+    async fn read(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        offset: u32,
+        attr_type: AttributeBackingType,
+    ) -> Result<AttAttributeDataChild, AttErrorCode> {
+        let (tx, rx) = oneshot::channel();
+        self.0.send(MockRawDatastoreEvents::Read(conn_id, handle, attr_type, offset, tx)).unwrap();
+        let resp = rx.await.unwrap();
+        info!("sending {resp:?} down from upper tester");
+        resp
+    }
+
+    async fn write(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        attr_type: AttributeBackingType,
+        write_type: GattWriteRequestType,
+        data: AttAttributeDataView<'_>,
+    ) -> Result<(), AttErrorCode> {
+        let (tx, rx) = oneshot::channel();
+        self.0
+            .send(MockRawDatastoreEvents::Write(
+                conn_id,
+                handle,
+                attr_type,
+                write_type,
+                data.to_owned_packet(),
+                tx,
+            ))
+            .unwrap();
+        rx.await.unwrap()
+    }
+
+    fn write_no_response(
+        &self,
+        conn_id: ConnectionId,
+        handle: AttHandle,
+        attr_type: AttributeBackingType,
+        data: AttAttributeDataView<'_>,
+    ) {
+        self.0
+            .send(MockRawDatastoreEvents::WriteNoResponse(
+                conn_id,
+                handle,
+                attr_type,
+                data.to_owned_packet(),
+            ))
+            .unwrap();
+    }
+
+    async fn execute(
+        &self,
+        conn_id: ConnectionId,
+        decision: TransactionDecision,
+    ) -> Result<(), AttErrorCode> {
+        let (tx, rx) = oneshot::channel();
+        self.0.send(MockRawDatastoreEvents::Execute(conn_id, decision, tx)).unwrap();
+        rx.await.unwrap()
+    }
+}
diff --git a/system/rust/src/gatt/server.rs b/system/rust/src/gatt/server.rs
index 53fb9dcfe6..886d9953a0 100644
--- a/system/rust/src/gatt/server.rs
+++ b/system/rust/src/gatt/server.rs
@@ -26,7 +26,7 @@ use self::{
     services::register_builtin_services,
 };
 
-use super::{callbacks::GattDatastore, channel::AttTransport, ids::AttHandle};
+use super::{callbacks::RawGattDatastore, channel::AttTransport, ids::AttHandle};
 use anyhow::{anyhow, bail, Result};
 use log::info;
 
@@ -87,7 +87,7 @@ impl GattModule {
         &mut self,
         server_id: ServerId,
         service: GattServiceWithHandle,
-        datastore: Rc<dyn GattDatastore>,
+        datastore: Rc<dyn RawGattDatastore>,
     ) -> Result<()> {
         self.databases
             .get(&server_id)
diff --git a/system/rust/src/gatt/server/gatt_database.rs b/system/rust/src/gatt/server/gatt_database.rs
index d27e7107d2..01d8372e44 100644
--- a/system/rust/src/gatt/server/gatt_database.rs
+++ b/system/rust/src/gatt/server/gatt_database.rs
@@ -14,7 +14,7 @@ use crate::{
         uuid::Uuid,
     },
     gatt::{
-        callbacks::GattDatastore,
+        callbacks::{GattWriteRequestType, RawGattDatastore},
         ffi::AttributeBackingType,
         ids::{AttHandle, ConnectionId},
     },
@@ -99,8 +99,8 @@ struct GattDatabaseSchema {
 #[derive(Clone)]
 enum AttAttributeBackingValue {
     Static(AttAttributeDataChild),
-    DynamicCharacteristic(Rc<dyn GattDatastore>),
-    DynamicDescriptor(Rc<dyn GattDatastore>),
+    DynamicCharacteristic(Rc<dyn RawGattDatastore>),
+    DynamicDescriptor(Rc<dyn RawGattDatastore>),
 }
 
 #[derive(Clone)]
@@ -166,7 +166,7 @@ impl GattDatabase {
     pub fn add_service_with_handles(
         &self,
         service: GattServiceWithHandle,
-        datastore: Rc<dyn GattDatastore>,
+        datastore: Rc<dyn RawGattDatastore>,
     ) -> Result<()> {
         let mut attributes = BTreeMap::new();
         let mut attribute_cnt = 0;
@@ -360,10 +360,24 @@ impl AttDatabase for AttDatabaseImpl {
         match value {
             AttAttributeBackingValue::Static(val) => return Ok(val),
             AttAttributeBackingValue::DynamicCharacteristic(datastore) => {
-                datastore.read(self.conn_id, handle, AttributeBackingType::Characteristic).await
+                datastore
+                    .read(
+                        self.conn_id,
+                        handle,
+                        /* offset */ 0,
+                        AttributeBackingType::Characteristic,
+                    )
+                    .await
             }
             AttAttributeBackingValue::DynamicDescriptor(datastore) => {
-                datastore.read(self.conn_id, handle, AttributeBackingType::Descriptor).await
+                datastore
+                    .read(
+                        self.conn_id,
+                        handle,
+                        /* offset */ 0,
+                        AttributeBackingType::Descriptor,
+                    )
+                    .await
             }
         }
     }
@@ -395,11 +409,25 @@ impl AttDatabase for AttDatabaseImpl {
             }
             AttAttributeBackingValue::DynamicCharacteristic(datastore) => {
                 datastore
-                    .write(self.conn_id, handle, AttributeBackingType::Characteristic, data)
+                    .write(
+                        self.conn_id,
+                        handle,
+                        AttributeBackingType::Characteristic,
+                        GattWriteRequestType::Request,
+                        data,
+                    )
                     .await
             }
             AttAttributeBackingValue::DynamicDescriptor(datastore) => {
-                datastore.write(self.conn_id, handle, AttributeBackingType::Descriptor, data).await
+                datastore
+                    .write(
+                        self.conn_id,
+                        handle,
+                        AttributeBackingType::Descriptor,
+                        GattWriteRequestType::Request,
+                        data,
+                    )
+                    .await
             }
         }
     }
diff --git a/system/rust/src/packets.pdl b/system/rust/src/packets.pdl
index 4df42117ee..48d410c809 100644
--- a/system/rust/src/packets.pdl
+++ b/system/rust/src/packets.pdl
@@ -63,9 +63,14 @@ enum AttErrorCode : 8 {
   INSUFFICIENT_AUTHENTICATION = 0x05,
   REQUEST_NOT_SUPPORTED = 0x06,
   ATTRIBUTE_NOT_FOUND = 0x0A,
+  ATTRIBUTE_NOT_LONG = 0x0B,
+  UNLIKELY_ERROR = 0x0E,
   UNSUPPORTED_GROUP_TYPE = 0x10,
   APPLICATION_ERROR = 0x80,
-  UNLIKELY_ERROR = 0x0E
+  WRITE_REQUEST_REJECTED = 0xFC,
+  CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR_IMPROPERLY_CONFIGURED = 0xFD,
+  PROCEDURE_ALREADY_IN_PROGRESS = 0xFE,
+  OUT_OF_RANGE = 0xFF,
 }
 
 struct AttHandle {
diff --git a/system/rust/tests/gatt_callbacks_test.rs b/system/rust/tests/gatt_callbacks_test.rs
index 8dd343d9c6..98b2838567 100644
--- a/system/rust/tests/gatt_callbacks_test.rs
+++ b/system/rust/tests/gatt_callbacks_test.rs
@@ -4,7 +4,10 @@ use std::{rc::Rc, time::Duration};
 
 use bluetooth_core::{
     gatt::{
-        callbacks::{CallbackResponseError, CallbackTransactionManager, GattDatastore},
+        callbacks::{
+            CallbackResponseError, CallbackTransactionManager, GattWriteRequestType, GattWriteType,
+            RawGattDatastore, TransactionDecision,
+        },
         ffi::AttributeBackingType,
         ids::{AttHandle, ConnectionId, ServerId, TransactionId, TransportIndex},
         mocks::mock_callbacks::{MockCallbackEvents, MockCallbacks},
@@ -23,6 +26,9 @@ const CONN_ID: ConnectionId = ConnectionId::new(TCB_IDX, SERVER_ID);
 const HANDLE_1: AttHandle = AttHandle(3);
 const BACKING_TYPE: AttributeBackingType = AttributeBackingType::Descriptor;
 
+const OFFSET: u32 = 12;
+const WRITE_REQUEST_TYPE: GattWriteRequestType = GattWriteRequestType::Prepare { offset: 7 };
+
 fn initialize_manager_with_connection(
 ) -> (Rc<CallbackTransactionManager>, UnboundedReceiver<MockCallbackEvents>) {
     let (callbacks, callbacks_rx) = MockCallbacks::new();
@@ -32,8 +38,9 @@ fn initialize_manager_with_connection(
 
 async fn pull_trans_id(events_rx: &mut UnboundedReceiver<MockCallbackEvents>) -> TransactionId {
     match events_rx.recv().await.unwrap() {
-        MockCallbackEvents::OnServerRead(_, trans_id, _, _, _, _) => trans_id,
-        MockCallbackEvents::OnServerWrite(_, trans_id, _, _, _, _, _, _) => trans_id,
+        MockCallbackEvents::OnServerRead(_, trans_id, _, _, _) => trans_id,
+        MockCallbackEvents::OnServerWrite(_, trans_id, _, _, _, _) => trans_id,
+        MockCallbackEvents::OnExecute(_, trans_id, _) => trans_id,
         _ => unimplemented!(),
     }
 }
@@ -45,11 +52,13 @@ fn test_read_characteristic_callback() {
         let (callback_manager, mut callbacks_rx) = initialize_manager_with_connection();
 
         // act: start read operation
-        spawn_local(async move { callback_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        spawn_local(
+            async move { callback_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await },
+        );
 
         // assert: verify the read callback is received
         let MockCallbackEvents::OnServerRead(
-            CONN_ID, _, HANDLE_1, BACKING_TYPE, 0, false,
+            CONN_ID, _, HANDLE_1, BACKING_TYPE, OFFSET,
         ) = callbacks_rx.recv().await.unwrap() else {
           unreachable!()
         };
@@ -65,8 +74,9 @@ fn test_read_characteristic_response() {
 
         // act: start read operation
         let cloned_manager = callback_manager.clone();
-        let pending_read =
-            spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        let pending_read = spawn_local(async move {
+            cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
         // provide a response
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
         callback_manager.send_response(CONN_ID, trans_id, data.clone()).unwrap();
@@ -86,16 +96,18 @@ fn test_sequential_reads() {
 
         // act: start read operation
         let cloned_manager = callback_manager.clone();
-        let pending_read_1 =
-            spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        let pending_read_1 = spawn_local(async move {
+            cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
         // respond to first
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
         callback_manager.send_response(CONN_ID, trans_id, data1.clone()).unwrap();
 
         // do a second read operation
         let cloned_manager = callback_manager.clone();
-        let pending_read_2 =
-            spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        let pending_read_2 = spawn_local(async move {
+            cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
         // respond to second
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
         callback_manager.send_response(CONN_ID, trans_id, data2.clone()).unwrap();
@@ -116,13 +128,15 @@ fn test_concurrent_reads() {
 
         // act: start read operation
         let cloned_manager = callback_manager.clone();
-        let pending_read_1 =
-            spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        let pending_read_1 = spawn_local(async move {
+            cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
 
         // do a second read operation
         let cloned_manager = callback_manager.clone();
-        let pending_read_2 =
-            spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        let pending_read_2 = spawn_local(async move {
+            cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
 
         // respond to first
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
@@ -146,9 +160,13 @@ fn test_distinct_transaction_ids() {
 
         // act: start two read operations concurrently
         let cloned_manager = callback_manager.clone();
-        spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        spawn_local(
+            async move { cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await },
+        );
         let cloned_manager = callback_manager.clone();
-        spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        spawn_local(
+            async move { cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await },
+        );
 
         // pull both trans_ids
         let trans_id_1 = pull_trans_id(&mut callbacks_rx).await;
@@ -168,7 +186,9 @@ fn test_invalid_trans_id() {
 
         // act: start a read operation
         let cloned_manager = callback_manager.clone();
-        spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        spawn_local(
+            async move { cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await },
+        );
         // respond with the correct conn_id but an invalid trans_id
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
         let invalid_trans_id = TransactionId(trans_id.0 + 1);
@@ -190,12 +210,14 @@ fn test_write_characteristic_callback() {
             build_view_or_crash(build_att_data(AttAttributeDataChild::RawData([1, 2].into())));
         let cloned_data = data.view().to_owned_packet();
         spawn_local(async move {
-            callback_manager.write(CONN_ID, HANDLE_1, BACKING_TYPE, cloned_data.view()).await
+            callback_manager
+                .write(CONN_ID, HANDLE_1, BACKING_TYPE, WRITE_REQUEST_TYPE, cloned_data.view())
+                .await
         });
 
         // assert: verify the write callback is received
         let MockCallbackEvents::OnServerWrite(
-            CONN_ID, _, HANDLE_1, BACKING_TYPE, 0, /* needs_response = */ true, false, recv_data
+            CONN_ID, _, HANDLE_1, BACKING_TYPE, GattWriteType::Request(WRITE_REQUEST_TYPE), recv_data
         ) = callbacks_rx.recv().await.unwrap() else {
           unreachable!()
         };
@@ -217,7 +239,9 @@ fn test_write_characteristic_response() {
             build_view_or_crash(build_att_data(AttAttributeDataChild::RawData([1, 2].into())));
         let cloned_manager = callback_manager.clone();
         let pending_write = spawn_local(async move {
-            cloned_manager.write(CONN_ID, HANDLE_1, BACKING_TYPE, data.view()).await
+            cloned_manager
+                .write(CONN_ID, HANDLE_1, BACKING_TYPE, GattWriteRequestType::Request, data.view())
+                .await
         });
         // provide a response with some error code
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
@@ -238,10 +262,9 @@ fn test_response_timeout() {
 
         // act: start operation
         let time_sent = Instant::now();
-        let pending_write =
-            spawn_local(
-                async move { callback_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await },
-            );
+        let pending_write = spawn_local(async move {
+            callback_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
 
         // assert: that we time-out after 15s
         assert_eq!(pending_write.await.unwrap(), Err(AttErrorCode::UNLIKELY_ERROR));
@@ -259,8 +282,9 @@ fn test_transaction_cleanup_after_timeout() {
 
         // act: start an operation
         let cloned_manager = callback_manager.clone();
-        let pending =
-            spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        let pending = spawn_local(async move {
+            cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
         // let it time out
         assert_eq!(pending.await.unwrap(), Err(AttErrorCode::UNLIKELY_ERROR));
@@ -281,8 +305,9 @@ fn test_listener_hang_up() {
 
         // act: start an operation
         let cloned_manager = callback_manager.clone();
-        let pending =
-            spawn_local(async move { cloned_manager.read(CONN_ID, HANDLE_1, BACKING_TYPE).await });
+        let pending = spawn_local(async move {
+            cloned_manager.read(CONN_ID, HANDLE_1, OFFSET, BACKING_TYPE).await
+        });
         let trans_id = pull_trans_id(&mut callbacks_rx).await;
         // cancel the listener, wait for it to stop
         pending.abort();
@@ -295,3 +320,69 @@ fn test_listener_hang_up() {
         assert_eq!(resp, Err(CallbackResponseError::ListenerHungUp(trans_id)));
     });
 }
+
+#[test]
+fn test_write_no_response_callback() {
+    start_test(async {
+        // arrange
+        let (callback_manager, mut callbacks_rx) = initialize_manager_with_connection();
+
+        // act: start write_no_response operation
+        let data =
+            build_view_or_crash(build_att_data(AttAttributeDataChild::RawData([1, 2].into())));
+        callback_manager.write_no_response(CONN_ID, HANDLE_1, BACKING_TYPE, data.view());
+
+        // assert: verify the write callback is received
+        let MockCallbackEvents::OnServerWrite(
+            CONN_ID, _, HANDLE_1, BACKING_TYPE, GattWriteType::Command, recv_data
+        ) = callbacks_rx.recv().await.unwrap() else {
+          unreachable!()
+        };
+        assert_eq!(
+            recv_data.view().get_raw_payload().collect::<Vec<_>>(),
+            data.view().get_raw_payload().collect::<Vec<_>>()
+        );
+    });
+}
+
+#[test]
+fn test_execute_characteristic_callback() {
+    start_test(async {
+        // arrange
+        let (callback_manager, mut callbacks_rx) = initialize_manager_with_connection();
+
+        // act: start execute operation
+        spawn_local(
+            async move { callback_manager.execute(CONN_ID, TransactionDecision::Cancel).await },
+        );
+
+        // assert: verify the execute callback is received
+        let MockCallbackEvents::OnExecute(
+            CONN_ID, _, TransactionDecision::Cancel
+        ) = callbacks_rx.recv().await.unwrap() else {
+          unreachable!()
+        };
+    });
+}
+
+#[test]
+fn test_execute_characteristic_response() {
+    start_test(async {
+        // arrange
+        let (callback_manager, mut callbacks_rx) = initialize_manager_with_connection();
+
+        // act: start execute operation
+        let cloned_manager = callback_manager.clone();
+        let pending_execute = spawn_local(async move {
+            cloned_manager.execute(CONN_ID, TransactionDecision::Cancel).await
+        });
+        // provide a response with some error code
+        let trans_id = pull_trans_id(&mut callbacks_rx).await;
+        callback_manager
+            .send_response(CONN_ID, trans_id, Err(AttErrorCode::WRITE_NOT_PERMITTED))
+            .unwrap();
+
+        // assert: that the error code was received
+        assert_eq!(pending_execute.await.unwrap(), Err(AttErrorCode::WRITE_NOT_PERMITTED));
+    });
+}