Merge "Cancel pending reliable message transactions on HAL restart" into main

3 files changed, 296 insertions, 124 deletions

diff --git a/services/core/java/com/android/server/location/contexthub/ContextHubEndpointBroker.java b/services/core/java/com/android/server/location/contexthub/ContextHubEndpointBroker.java
index 940bcb4c6ba1..f40d0dd18213 100644
--- a/services/core/java/com/android/server/location/contexthub/ContextHubEndpointBroker.java
+++ b/services/core/java/com/android/server/location/contexthub/ContextHubEndpointBroker.java
@@ -43,7 +43,10 @@ import android.util.SparseArray;
 import com.android.internal.annotations.GuardedBy;
 
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
 
 /**
  * A class that represents a broker for the endpoint registered by the client app. This class
@@ -111,6 +114,11 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
 
         private final boolean mRemoteInitiated;
 
+        /**
+         * The set of seq # for pending reliable messages started by this endpoint for this session.
+         */
+        private final Set<Integer> mPendingSequenceNumbers = new HashSet<>();
+
         SessionInfo(HubEndpointInfo remoteEndpointInfo, boolean remoteInitiated) {
             mRemoteEndpointInfo = remoteEndpointInfo;
             mRemoteInitiated = remoteInitiated;
@@ -131,6 +139,24 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
         public boolean isActive() {
             return mSessionState == SessionState.ACTIVE;
         }
+
+        public boolean isReliableMessagePending(int sequenceNumber) {
+            return mPendingSequenceNumbers.contains(sequenceNumber);
+        }
+
+        public void setReliableMessagePending(int sequenceNumber) {
+            mPendingSequenceNumbers.add(sequenceNumber);
+        }
+
+        public void setReliableMessageCompleted(int sequenceNumber) {
+            mPendingSequenceNumbers.remove(sequenceNumber);
+        }
+
+        public void forEachPendingReliableMessage(Consumer<Integer> consumer) {
+            for (int sequenceNumber : mPendingSequenceNumbers) {
+                consumer.accept(sequenceNumber);
+            }
+        }
     }
 
     /** A map between a session ID which maps to its current state. */
@@ -208,10 +234,7 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
             try {
                 mSessionInfoMap.put(sessionId, new SessionInfo(destination, false));
                 mHubInterface.openEndpointSession(
-                        sessionId,
-                        halEndpointInfo.id,
-                        mHalEndpointInfo.id,
-                        serviceDescriptor);
+                        sessionId, halEndpointInfo.id, mHalEndpointInfo.id, serviceDescriptor);
             } catch (RemoteException | IllegalArgumentException | UnsupportedOperationException e) {
                 Log.e(TAG, "Exception while calling HAL openEndpointSession", e);
                 cleanupSessionResources(sessionId);
@@ -286,34 +309,42 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
     public void sendMessage(
             int sessionId, HubMessage message, IContextHubTransactionCallback callback) {
         super.sendMessage_enforcePermission();
-        Message halMessage = ContextHubServiceUtil.createHalMessage(message);
-        if (!isSessionActive(sessionId)) {
-            throw new SecurityException(
-                    "sendMessage called on inactive session (id= " + sessionId + ")");
-        }
-
-        if (callback == null) {
-            try {
-                mHubInterface.sendMessageToEndpoint(sessionId, halMessage);
-            } catch (RemoteException e) {
-                Log.w(TAG, "Exception while sending message on session " + sessionId, e);
+        synchronized (mOpenSessionLock) {
+            SessionInfo info = mSessionInfoMap.get(sessionId);
+            if (info == null) {
+                throw new IllegalArgumentException(
+                        "sendMessage for invalid session id=" + sessionId);
             }
-        } else {
-            ContextHubServiceTransaction transaction =
-                    mTransactionManager.createSessionMessageTransaction(
-                            mHubInterface, sessionId, halMessage, mPackageName, callback);
-            try {
-                mTransactionManager.addTransaction(transaction);
-            } catch (IllegalStateException e) {
-                Log.e(
-                        TAG,
-                        "Unable to add a transaction in sendMessageToEndpoint "
-                                + "(session ID = "
-                                + sessionId
-                                + ")",
-                        e);
-                transaction.onTransactionComplete(
-                        ContextHubTransaction.RESULT_FAILED_SERVICE_INTERNAL_FAILURE);
+            if (!info.isActive()) {
+                throw new SecurityException(
+                        "sendMessage called on inactive session (id= " + sessionId + ")");
+            }
+
+            Message halMessage = ContextHubServiceUtil.createHalMessage(message);
+            if (callback == null) {
+                try {
+                    mHubInterface.sendMessageToEndpoint(sessionId, halMessage);
+                } catch (RemoteException e) {
+                    Log.w(TAG, "Exception while sending message on session " + sessionId, e);
+                }
+            } else {
+                ContextHubServiceTransaction transaction =
+                        mTransactionManager.createSessionMessageTransaction(
+                                mHubInterface, sessionId, halMessage, mPackageName, callback);
+                try {
+                    mTransactionManager.addTransaction(transaction);
+                    info.setReliableMessagePending(transaction.getMessageSequenceNumber());
+                } catch (IllegalStateException e) {
+                    Log.e(
+                            TAG,
+                            "Unable to add a transaction in sendMessageToEndpoint "
+                                    + "(session ID = "
+                                    + sessionId
+                                    + ")",
+                            e);
+                    transaction.onTransactionComplete(
+                            ContextHubTransaction.RESULT_FAILED_SERVICE_INTERNAL_FAILURE);
+                }
             }
         }
     }
@@ -393,7 +424,9 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
                 int id = mSessionInfoMap.keyAt(i);
                 int count = i + 1;
                 sb.append(
-                        "  " + count + ". id="
+                        "  "
+                                + count
+                                + ". id="
                                 + id
                                 + ", remote:"
                                 + mSessionInfoMap.get(id).getRemoteEndpointInfo());
@@ -461,13 +494,24 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
     /* package */ void onMessageReceived(int sessionId, HubMessage message) {
         byte code = onMessageReceivedInternal(sessionId, message);
         if (code != ErrorCode.OK && message.isResponseRequired()) {
-            sendMessageDeliveryStatus(
-                    sessionId, message.getMessageSequenceNumber(), code);
+            sendMessageDeliveryStatus(sessionId, message.getMessageSequenceNumber(), code);
         }
     }
 
     /* package */ void onMessageDeliveryStatusReceived(
             int sessionId, int sequenceNumber, byte errorCode) {
+        synchronized (mOpenSessionLock) {
+            SessionInfo info = mSessionInfoMap.get(sessionId);
+            if (info == null || !info.isActive()) {
+                Log.w(TAG, "Received delivery status for invalid session: id=" + sessionId);
+                return;
+            }
+            if (!info.isReliableMessagePending(sequenceNumber)) {
+                Log.w(TAG, "Received delivery status for unknown seq: " + sequenceNumber);
+                return;
+            }
+            info.setReliableMessageCompleted(sequenceNumber);
+        }
         mTransactionManager.onMessageDeliveryResponse(sequenceNumber, errorCode == ErrorCode.OK);
     }
 
@@ -492,7 +536,6 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
                 onCloseEndpointSession(id, Reason.HUB_RESET);
             }
         }
-        // TODO(b/390029594): Cancel any ongoing reliable communication transactions
     }
 
     private Optional<Byte> onEndpointSessionOpenRequestInternal(
@@ -515,9 +558,11 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
             mSessionInfoMap.put(sessionId, new SessionInfo(initiator, true));
         }
 
-        boolean success = invokeCallback(
-                (consumer) ->
-                        consumer.onSessionOpenRequest(sessionId, initiator, serviceDescriptor));
+        boolean success =
+                invokeCallback(
+                        (consumer) ->
+                                consumer.onSessionOpenRequest(
+                                        sessionId, initiator, serviceDescriptor));
         return success ? Optional.empty() : Optional.of(Reason.UNSPECIFIED);
     }
 
@@ -590,8 +635,15 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
     private boolean cleanupSessionResources(int sessionId) {
         synchronized (mOpenSessionLock) {
             SessionInfo info = mSessionInfoMap.get(sessionId);
-            if (info != null && !info.isRemoteInitiated()) {
-                mEndpointManager.returnSessionId(sessionId);
+            if (info != null) {
+                if (!info.isRemoteInitiated()) {
+                    mEndpointManager.returnSessionId(sessionId);
+                }
+                info.forEachPendingReliableMessage(
+                        (sequenceNumber) -> {
+                            mTransactionManager.onMessageDeliveryResponse(
+                                    sequenceNumber, /* success= */ false);
+                        });
                 mSessionInfoMap.remove(sessionId);
             }
             return info != null;
@@ -646,10 +698,7 @@ public class ContextHubEndpointBroker extends IContextHubEndpoint.Stub
                         try {
                             mWakeLock.release();
                         } catch (RuntimeException e) {
-                            Log.e(
-                                    TAG,
-                                    "Releasing the wakelock for all acquisitions fails - ",
-                                    e);
+                            Log.e(TAG, "Releasing the wakelock for all acquisitions fails - ", e);
                             break;
                         }
                     }
diff --git a/services/core/java/com/android/server/location/contexthub/ContextHubTransactionManager.java b/services/core/java/com/android/server/location/contexthub/ContextHubTransactionManager.java
index a430a82fc13b..6a1db0223db5 100644
--- a/services/core/java/com/android/server/location/contexthub/ContextHubTransactionManager.java
+++ b/services/core/java/com/android/server/location/contexthub/ContextHubTransactionManager.java
@@ -29,6 +29,7 @@ import android.os.SystemClock;
 import android.util.Log;
 
 import com.android.internal.annotations.GuardedBy;
+import com.android.internal.annotations.VisibleForTesting;
 
 import java.time.Duration;
 import java.util.ArrayDeque;
@@ -165,52 +166,61 @@ import java.util.concurrent.atomic.AtomicInteger;
     /**
      * Creates a transaction for loading a nanoapp.
      *
-     * @param contextHubId       the ID of the hub to load the nanoapp to
-     * @param nanoAppBinary      the binary of the nanoapp to load
+     * @param contextHubId the ID of the hub to load the nanoapp to
+     * @param nanoAppBinary the binary of the nanoapp to load
      * @param onCompleteCallback the client on complete callback
      * @return the generated transaction
      */
     /* package */ ContextHubServiceTransaction createLoadTransaction(
-            int contextHubId, NanoAppBinary nanoAppBinary,
-            IContextHubTransactionCallback onCompleteCallback, String packageName) {
+            int contextHubId,
+            NanoAppBinary nanoAppBinary,
+            IContextHubTransactionCallback onCompleteCallback,
+            String packageName) {
         return new ContextHubServiceTransaction(
-                mNextAvailableId.getAndIncrement(), ContextHubTransaction.TYPE_LOAD_NANOAPP,
-                nanoAppBinary.getNanoAppId(), packageName) {
+                mNextAvailableId.getAndIncrement(),
+                ContextHubTransaction.TYPE_LOAD_NANOAPP,
+                nanoAppBinary.getNanoAppId(),
+                packageName) {
             @Override
-                /* package */ int onTransact() {
+            /* package */ int onTransact() {
                 try {
                     return mContextHubProxy.loadNanoapp(
                             contextHubId, nanoAppBinary, this.getTransactionId());
                 } catch (RemoteException e) {
-                    Log.e(TAG, "RemoteException while trying to load nanoapp with ID 0x" +
-                            Long.toHexString(nanoAppBinary.getNanoAppId()), e);
+                    Log.e(
+                            TAG,
+                            "RemoteException while trying to load nanoapp with ID 0x"
+                                    + Long.toHexString(nanoAppBinary.getNanoAppId()),
+                            e);
                     return ContextHubTransaction.RESULT_FAILED_UNKNOWN;
                 }
             }
 
             @Override
-                /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
+            /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
                 ContextHubStatsLog.write(
                         ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED,
                         nanoAppBinary.getNanoAppId(),
                         nanoAppBinary.getNanoAppVersion(),
                         ContextHubStatsLog
-                            .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_TYPE__TYPE_LOAD,
+                                .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_TYPE__TYPE_LOAD,
                         toStatsTransactionResult(result));
 
-                ContextHubEventLogger.getInstance().logNanoappLoad(
-                        contextHubId,
-                        nanoAppBinary.getNanoAppId(),
-                        nanoAppBinary.getNanoAppVersion(),
-                        nanoAppBinary.getBinary().length,
-                        result == ContextHubTransaction.RESULT_SUCCESS);
+                ContextHubEventLogger.getInstance()
+                        .logNanoappLoad(
+                                contextHubId,
+                                nanoAppBinary.getNanoAppId(),
+                                nanoAppBinary.getNanoAppVersion(),
+                                nanoAppBinary.getBinary().length,
+                                result == ContextHubTransaction.RESULT_SUCCESS);
 
                 if (result == ContextHubTransaction.RESULT_SUCCESS) {
                     // NOTE: The legacy JNI code used to do a query right after a load success
                     // to synchronize the service cache. Instead store the binary that was
                     // requested to load to update the cache later without doing a query.
                     mNanoAppStateManager.addNanoAppInstance(
-                            contextHubId, nanoAppBinary.getNanoAppId(),
+                            contextHubId,
+                            nanoAppBinary.getNanoAppId(),
                             nanoAppBinary.getNanoAppVersion());
                 }
                 try {
@@ -228,42 +238,51 @@ import java.util.concurrent.atomic.AtomicInteger;
     /**
      * Creates a transaction for unloading a nanoapp.
      *
-     * @param contextHubId       the ID of the hub to unload the nanoapp from
-     * @param nanoAppId          the ID of the nanoapp to unload
+     * @param contextHubId the ID of the hub to unload the nanoapp from
+     * @param nanoAppId the ID of the nanoapp to unload
      * @param onCompleteCallback the client on complete callback
      * @return the generated transaction
      */
     /* package */ ContextHubServiceTransaction createUnloadTransaction(
-            int contextHubId, long nanoAppId, IContextHubTransactionCallback onCompleteCallback,
+            int contextHubId,
+            long nanoAppId,
+            IContextHubTransactionCallback onCompleteCallback,
             String packageName) {
         return new ContextHubServiceTransaction(
-                mNextAvailableId.getAndIncrement(), ContextHubTransaction.TYPE_UNLOAD_NANOAPP,
-                nanoAppId, packageName) {
+                mNextAvailableId.getAndIncrement(),
+                ContextHubTransaction.TYPE_UNLOAD_NANOAPP,
+                nanoAppId,
+                packageName) {
             @Override
-                /* package */ int onTransact() {
+            /* package */ int onTransact() {
                 try {
                     return mContextHubProxy.unloadNanoapp(
                             contextHubId, nanoAppId, this.getTransactionId());
                 } catch (RemoteException e) {
-                    Log.e(TAG, "RemoteException while trying to unload nanoapp with ID 0x" +
-                            Long.toHexString(nanoAppId), e);
+                    Log.e(
+                            TAG,
+                            "RemoteException while trying to unload nanoapp with ID 0x"
+                                    + Long.toHexString(nanoAppId),
+                            e);
                     return ContextHubTransaction.RESULT_FAILED_UNKNOWN;
                 }
             }
 
             @Override
-                /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
+            /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
                 ContextHubStatsLog.write(
-                        ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED, nanoAppId,
+                        ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED,
+                        nanoAppId,
                         0 /* nanoappVersion */,
                         ContextHubStatsLog
-                            .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_TYPE__TYPE_UNLOAD,
+                                .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_TYPE__TYPE_UNLOAD,
                         toStatsTransactionResult(result));
 
-                ContextHubEventLogger.getInstance().logNanoappUnload(
-                        contextHubId,
-                        nanoAppId,
-                        result == ContextHubTransaction.RESULT_SUCCESS);
+                ContextHubEventLogger.getInstance()
+                        .logNanoappUnload(
+                                contextHubId,
+                                nanoAppId,
+                                result == ContextHubTransaction.RESULT_SUCCESS);
 
                 if (result == ContextHubTransaction.RESULT_SUCCESS) {
                     mNanoAppStateManager.removeNanoAppInstance(contextHubId, nanoAppId);
@@ -283,31 +302,37 @@ import java.util.concurrent.atomic.AtomicInteger;
     /**
      * Creates a transaction for enabling a nanoapp.
      *
-     * @param contextHubId       the ID of the hub to enable the nanoapp on
-     * @param nanoAppId          the ID of the nanoapp to enable
+     * @param contextHubId the ID of the hub to enable the nanoapp on
+     * @param nanoAppId the ID of the nanoapp to enable
      * @param onCompleteCallback the client on complete callback
      * @return the generated transaction
      */
     /* package */ ContextHubServiceTransaction createEnableTransaction(
-            int contextHubId, long nanoAppId, IContextHubTransactionCallback onCompleteCallback,
+            int contextHubId,
+            long nanoAppId,
+            IContextHubTransactionCallback onCompleteCallback,
             String packageName) {
         return new ContextHubServiceTransaction(
-                mNextAvailableId.getAndIncrement(), ContextHubTransaction.TYPE_ENABLE_NANOAPP,
+                mNextAvailableId.getAndIncrement(),
+                ContextHubTransaction.TYPE_ENABLE_NANOAPP,
                 packageName) {
             @Override
-                /* package */ int onTransact() {
+            /* package */ int onTransact() {
                 try {
                     return mContextHubProxy.enableNanoapp(
                             contextHubId, nanoAppId, this.getTransactionId());
                 } catch (RemoteException e) {
-                    Log.e(TAG, "RemoteException while trying to enable nanoapp with ID 0x" +
-                            Long.toHexString(nanoAppId), e);
+                    Log.e(
+                            TAG,
+                            "RemoteException while trying to enable nanoapp with ID 0x"
+                                    + Long.toHexString(nanoAppId),
+                            e);
                     return ContextHubTransaction.RESULT_FAILED_UNKNOWN;
                 }
             }
 
             @Override
-                /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
+            /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
                 try {
                     onCompleteCallback.onTransactionComplete(result);
                 } catch (RemoteException e) {
@@ -320,31 +345,37 @@ import java.util.concurrent.atomic.AtomicInteger;
     /**
      * Creates a transaction for disabling a nanoapp.
      *
-     * @param contextHubId       the ID of the hub to disable the nanoapp on
-     * @param nanoAppId          the ID of the nanoapp to disable
+     * @param contextHubId the ID of the hub to disable the nanoapp on
+     * @param nanoAppId the ID of the nanoapp to disable
      * @param onCompleteCallback the client on complete callback
      * @return the generated transaction
      */
     /* package */ ContextHubServiceTransaction createDisableTransaction(
-            int contextHubId, long nanoAppId, IContextHubTransactionCallback onCompleteCallback,
+            int contextHubId,
+            long nanoAppId,
+            IContextHubTransactionCallback onCompleteCallback,
             String packageName) {
         return new ContextHubServiceTransaction(
-                mNextAvailableId.getAndIncrement(), ContextHubTransaction.TYPE_DISABLE_NANOAPP,
+                mNextAvailableId.getAndIncrement(),
+                ContextHubTransaction.TYPE_DISABLE_NANOAPP,
                 packageName) {
             @Override
-                /* package */ int onTransact() {
+            /* package */ int onTransact() {
                 try {
                     return mContextHubProxy.disableNanoapp(
                             contextHubId, nanoAppId, this.getTransactionId());
                 } catch (RemoteException e) {
-                    Log.e(TAG, "RemoteException while trying to disable nanoapp with ID 0x" +
-                            Long.toHexString(nanoAppId), e);
+                    Log.e(
+                            TAG,
+                            "RemoteException while trying to disable nanoapp with ID 0x"
+                                    + Long.toHexString(nanoAppId),
+                            e);
                     return ContextHubTransaction.RESULT_FAILED_UNKNOWN;
                 }
             }
 
             @Override
-                /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
+            /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
                 try {
                     onCompleteCallback.onTransactionComplete(result);
                 } catch (RemoteException e) {
@@ -447,18 +478,20 @@ import java.util.concurrent.atomic.AtomicInteger;
     /**
      * Creates a transaction for querying for a list of nanoapps.
      *
-     * @param contextHubId       the ID of the hub to query
+     * @param contextHubId the ID of the hub to query
      * @param onCompleteCallback the client on complete callback
      * @return the generated transaction
      */
     /* package */ ContextHubServiceTransaction createQueryTransaction(
-            int contextHubId, IContextHubTransactionCallback onCompleteCallback,
+            int contextHubId,
+            IContextHubTransactionCallback onCompleteCallback,
             String packageName) {
         return new ContextHubServiceTransaction(
-                mNextAvailableId.getAndIncrement(), ContextHubTransaction.TYPE_QUERY_NANOAPPS,
+                mNextAvailableId.getAndIncrement(),
+                ContextHubTransaction.TYPE_QUERY_NANOAPPS,
                 packageName) {
             @Override
-                /* package */ int onTransact() {
+            /* package */ int onTransact() {
                 try {
                     return mContextHubProxy.queryNanoapps(contextHubId);
                 } catch (RemoteException e) {
@@ -468,12 +501,12 @@ import java.util.concurrent.atomic.AtomicInteger;
             }
 
             @Override
-                /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
+            /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
                 onQueryResponse(result, Collections.emptyList());
             }
 
             @Override
-                /* package */ void onQueryResponse(
+            /* package */ void onQueryResponse(
                     @ContextHubTransaction.Result int result, List<NanoAppState> nanoAppStateList) {
                 try {
                     onCompleteCallback.onQueryResponse(result, nanoAppStateList);
@@ -539,6 +572,14 @@ import java.util.concurrent.atomic.AtomicInteger;
         }
     }
 
+    @VisibleForTesting
+    /* package */
+    int numReliableMessageTransactionPending() {
+        synchronized (mReliableMessageLock) {
+            return mReliableMessageTransactionMap.size();
+        }
+    }
+
     /**
      * Handles a transaction response from a Context Hub.
      *
@@ -585,18 +626,21 @@ import java.util.concurrent.atomic.AtomicInteger;
     void onMessageDeliveryResponse(int messageSequenceNumber, boolean success) {
         if (!Flags.reliableMessageRetrySupportService()) {
             TransactionAcceptConditions conditions =
-                    transaction -> transaction.getTransactionType()
-                            == ContextHubTransaction.TYPE_RELIABLE_MESSAGE
-                    && transaction.getMessageSequenceNumber()
-                            == messageSequenceNumber;
+                    transaction ->
+                            transaction.getTransactionType()
+                                            == ContextHubTransaction.TYPE_RELIABLE_MESSAGE
+                                    && transaction.getMessageSequenceNumber()
+                                            == messageSequenceNumber;
             ContextHubServiceTransaction transaction = getTransactionAndHandleNext(conditions);
             if (transaction == null) {
-                Log.w(TAG, "Received unexpected message delivery response (expected"
-                        + " message sequence number = "
-                        + messageSequenceNumber
-                        + ", received messageSequenceNumber = "
-                        + messageSequenceNumber
-                        + ")");
+                Log.w(
+                        TAG,
+                        "Received unexpected message delivery response (expected"
+                                + " message sequence number = "
+                                + messageSequenceNumber
+                                + ", received messageSequenceNumber = "
+                                + messageSequenceNumber
+                                + ")");
                 return;
             }
 
@@ -640,8 +684,10 @@ import java.util.concurrent.atomic.AtomicInteger;
      */
     /* package */
     void onQueryResponse(List<NanoAppState> nanoAppStateList) {
-        TransactionAcceptConditions conditions = transaction ->
-                transaction.getTransactionType() == ContextHubTransaction.TYPE_QUERY_NANOAPPS;
+        TransactionAcceptConditions conditions =
+                transaction ->
+                        transaction.getTransactionType()
+                                == ContextHubTransaction.TYPE_QUERY_NANOAPPS;
         ContextHubServiceTransaction transaction = getTransactionAndHandleNext(conditions);
         if (transaction == null) {
             Log.w(TAG, "Received unexpected query response");
@@ -968,24 +1014,33 @@ import java.util.concurrent.atomic.AtomicInteger;
     private int toStatsTransactionResult(@ContextHubTransaction.Result int result) {
         switch (result) {
             case ContextHubTransaction.RESULT_SUCCESS:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_SUCCESS;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_SUCCESS;
             case ContextHubTransaction.RESULT_FAILED_BAD_PARAMS:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_BAD_PARAMS;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_BAD_PARAMS;
             case ContextHubTransaction.RESULT_FAILED_UNINITIALIZED:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_UNINITIALIZED;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_UNINITIALIZED;
             case ContextHubTransaction.RESULT_FAILED_BUSY:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_BUSY;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_BUSY;
             case ContextHubTransaction.RESULT_FAILED_AT_HUB:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_AT_HUB;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_AT_HUB;
             case ContextHubTransaction.RESULT_FAILED_TIMEOUT:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_TIMEOUT;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_TIMEOUT;
             case ContextHubTransaction.RESULT_FAILED_SERVICE_INTERNAL_FAILURE:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_SERVICE_INTERNAL_FAILURE;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_SERVICE_INTERNAL_FAILURE;
             case ContextHubTransaction.RESULT_FAILED_HAL_UNAVAILABLE:
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_HAL_UNAVAILABLE;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_HAL_UNAVAILABLE;
             case ContextHubTransaction.RESULT_FAILED_UNKNOWN:
             default: /* fall through */
-                return ContextHubStatsLog.CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_UNKNOWN;
+                return ContextHubStatsLog
+                        .CHRE_CODE_DOWNLOAD_TRANSACTED__TRANSACTION_RESULT__TRANSACTION_RESULT_FAILED_UNKNOWN;
         }
     }
 
diff --git a/services/tests/servicestests/src/com/android/server/location/contexthub/ContextHubEndpointTest.java b/services/tests/servicestests/src/com/android/server/location/contexthub/ContextHubEndpointTest.java
index a4e77c00d647..1de864cb4eb0 100644
--- a/services/tests/servicestests/src/com/android/server/location/contexthub/ContextHubEndpointTest.java
+++ b/services/tests/servicestests/src/com/android/server/location/contexthub/ContextHubEndpointTest.java
@@ -17,9 +17,9 @@
 package com.android.server.location.contexthub;
 
 import static com.google.common.truth.Truth.assertThat;
-
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -33,15 +33,21 @@ import android.hardware.contexthub.HubMessage;
 import android.hardware.contexthub.IContextHubEndpoint;
 import android.hardware.contexthub.IContextHubEndpointCallback;
 import android.hardware.contexthub.IEndpointCommunication;
+import android.hardware.contexthub.Message;
 import android.hardware.contexthub.MessageDeliveryStatus;
 import android.hardware.contexthub.Reason;
+import android.hardware.location.IContextHubTransactionCallback;
+import android.hardware.location.NanoAppState;
 import android.os.Binder;
 import android.os.RemoteException;
 import android.platform.test.annotations.Presubmit;
-
+import android.util.Log;
 import androidx.test.ext.junit.runners.AndroidJUnit4;
 import androidx.test.platform.app.InstrumentationRegistry;
 
+import java.util.Collections;
+import java.util.List;
+
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -51,11 +57,11 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
 
-import java.util.Collections;
-
 @RunWith(AndroidJUnit4.class)
 @Presubmit
 public class ContextHubEndpointTest {
+    private static final String TAG = "ContextHubEndpointTest";
+
     private static final int SESSION_ID_RANGE = ContextHubEndpointManager.SERVICE_SESSION_RANGE;
     private static final int MIN_SESSION_ID = 0;
     private static final int MAX_SESSION_ID = MIN_SESSION_ID + SESSION_ID_RANGE - 1;
@@ -206,6 +212,68 @@ public class ContextHubEndpointTest {
         assertThat(mEndpointManager.getNumAvailableSessions()).isEqualTo(SESSION_ID_RANGE);
     }
 
+    @Test
+    public void testMessageTransaction() throws RemoteException {
+        IContextHubEndpoint endpoint = registerExampleEndpoint();
+        testMessageTransactionInternal(endpoint, /* deliverMessageStatus= */ true);
+
+        unregisterExampleEndpoint(endpoint);
+    }
+
+    @Test
+    public void testMessageTransactionCleanupOnUnregistration() throws RemoteException {
+        IContextHubEndpoint endpoint = registerExampleEndpoint();
+        testMessageTransactionInternal(endpoint, /* deliverMessageStatus= */ false);
+
+        unregisterExampleEndpoint(endpoint);
+        assertThat(mTransactionManager.numReliableMessageTransactionPending()).isEqualTo(0);
+    }
+
+    /** A helper method to create a session and validates reliable message sending. */
+    private void testMessageTransactionInternal(
+            IContextHubEndpoint endpoint, boolean deliverMessageStatus) throws RemoteException {
+        HubEndpointInfo targetInfo =
+                new HubEndpointInfo(
+                        TARGET_ENDPOINT_NAME,
+                        TARGET_ENDPOINT_ID,
+                        ENDPOINT_PACKAGE_NAME,
+                        Collections.emptyList());
+        int sessionId = endpoint.openSession(targetInfo, /* serviceDescriptor= */ null);
+        mEndpointManager.onEndpointSessionOpenComplete(sessionId);
+
+        final int messageType = 1234;
+        HubMessage message =
+                new HubMessage.Builder(messageType, new byte[] {1, 2, 3, 4, 5})
+                        .setResponseRequired(true)
+                        .build();
+        IContextHubTransactionCallback callback =
+                new IContextHubTransactionCallback.Stub() {
+                    @Override
+                    public void onQueryResponse(int result, List<NanoAppState> nanoappList) {
+                        Log.i(TAG, "Received onQueryResponse callback, result=" + result);
+                    }
+
+                    @Override
+                    public void onTransactionComplete(int result) {
+                        Log.i(TAG, "Received onTransactionComplete callback, result=" + result);
+                    }
+                };
+        endpoint.sendMessage(sessionId, message, callback);
+        ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
+        verify(mMockEndpointCommunications, timeout(1000))
+                .sendMessageToEndpoint(eq(sessionId), messageCaptor.capture());
+        Message halMessage = messageCaptor.getValue();
+        assertThat(halMessage.type).isEqualTo(message.getMessageType());
+        assertThat(halMessage.content).isEqualTo(message.getMessageBody());
+        assertThat(mTransactionManager.numReliableMessageTransactionPending()).isEqualTo(1);
+
+        if (deliverMessageStatus) {
+            mEndpointManager.onMessageDeliveryStatusReceived(
+                    sessionId, halMessage.sequenceNumber, ErrorCode.OK);
+            assertThat(mTransactionManager.numReliableMessageTransactionPending()).isEqualTo(0);
+        }
+    }
+
     private IContextHubEndpoint registerExampleEndpoint() throws RemoteException {
         HubEndpointInfo info =
                 new HubEndpointInfo(