Merge changes from topic "relm-service-retry" into main

* changes:
  Add reliable message retry support to the Context Hub Service
  Make ContextHubTestModeManager a dedicated class
  Add reliable message retry fields to ContextHubServiceTransaction

5 files changed, 422 insertions, 148 deletions

diff --git a/services/core/java/com/android/server/location/contexthub/ContextHubService.java b/services/core/java/com/android/server/location/contexthub/ContextHubService.java
index 7a722bc914f7..381b66735e9a 100644
--- a/services/core/java/com/android/server/location/contexthub/ContextHubService.java
+++ b/services/core/java/com/android/server/location/contexthub/ContextHubService.java
@@ -81,7 +81,6 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.PriorityQueue;
-import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -162,8 +161,8 @@ public class ContextHubService extends IContextHubService.Stub {
             new PriorityQueue<>(
                     Comparator.comparingLong(ReliableMessageRecord::getTimestamp));
 
-    // The test mode manager that manages behaviors during test mode.
-    private final TestModeManager mTestModeManager = new TestModeManager();
+    // The test mode manager that manages behaviors during test mode
+    private final ContextHubTestModeManager mTestModeManager = new ContextHubTestModeManager();
 
     // The period of the recurring time
     private static final int PERIOD_METRIC_QUERY_DAYS = 1;
@@ -229,9 +228,11 @@ public class ContextHubService extends IContextHubService.Stub {
             if (Flags.reliableMessageImplementation()
                     && Flags.reliableMessageTestModeBehavior()
                     && mIsTestModeEnabled.get()
-                    && mTestModeManager.handleNanoappMessage(mContextHubId, hostEndpointId,
-                            message, nanoappPermissions, messagePermissions)) {
-                // The TestModeManager handled the nanoapp message, so return here.
+                    && mTestModeManager.handleNanoappMessage(() -> {
+                        handleClientMessageCallback(mContextHubId, hostEndpointId, message,
+                                nanoappPermissions, messagePermissions);
+                    }, message)) {
+                // The ContextHubTestModeManager handled the nanoapp message, so return here.
                 return;
             }
 
@@ -261,8 +262,6 @@ public class ContextHubService extends IContextHubService.Stub {
      * Records a reliable message from a nanoapp for duplicate detection.
      */
     private static class ReliableMessageRecord {
-        public static final int TIMEOUT_NS = 1000000000;
-
         public int mContextHubId;
         public long mTimestamp;
         public int mMessageSequenceNumber;
@@ -297,56 +296,8 @@ public class ContextHubService extends IContextHubService.Stub {
         }
 
         public boolean isExpired() {
-            return mTimestamp + TIMEOUT_NS < SystemClock.elapsedRealtimeNanos();
-        }
-    }
-
-    /**
-     * A class to manage behaviors during test mode. This is used for testing.
-     */
-    private class TestModeManager {
-        /**
-         * Probability (in percent) of duplicating a message.
-         */
-        private static final int MESSAGE_DUPLICATION_PROBABILITY_PERCENT = 50;
-
-        /**
-         * The number of total messages to send when the duplicate event happens.
-         */
-        private static final int NUM_MESSAGES_TO_DUPLICATE = 3;
-
-        /**
-         * A probability percent for a certain event.
-         */
-        private static final int MAX_PROBABILITY_PERCENT = 100;
-
-        private final Random mRandom = new Random();
-
-        /**
-         * @return whether the message was handled
-         * @see ContextHubServiceCallback#handleNanoappMessage
-         */
-        public boolean handleNanoappMessage(int contextHubId,
-                short hostEndpointId, NanoAppMessage message,
-                List<String> nanoappPermissions, List<String> messagePermissions) {
-            if (!message.isReliable()) {
-                return false;
-            }
-
-            if (Flags.reliableMessageDuplicateDetectionService()
-                    && mRandom.nextInt(MAX_PROBABILITY_PERCENT)
-                    < MESSAGE_DUPLICATION_PROBABILITY_PERCENT) {
-                Log.i(TAG, "[TEST MODE] Duplicating message ("
-                        + NUM_MESSAGES_TO_DUPLICATE
-                        + " sends) with message sequence number: "
-                        + message.getMessageSequenceNumber());
-                for (int i = 0; i < NUM_MESSAGES_TO_DUPLICATE; ++i) {
-                    handleClientMessageCallback(contextHubId, hostEndpointId,
-                            message, nanoappPermissions, messagePermissions);
-                }
-                return true;
-            }
-            return false;
+            return mTimestamp + ContextHubTransactionManager.RELIABLE_MESSAGE_TIMEOUT.toNanos()
+                    < SystemClock.elapsedRealtimeNanos();
         }
     }
 
diff --git a/services/core/java/com/android/server/location/contexthub/ContextHubServiceTransaction.java b/services/core/java/com/android/server/location/contexthub/ContextHubServiceTransaction.java
index 6da7a6500d54..2ec9bdb75349 100644
--- a/services/core/java/com/android/server/location/contexthub/ContextHubServiceTransaction.java
+++ b/services/core/java/com/android/server/location/contexthub/ContextHubServiceTransaction.java
@@ -27,53 +27,65 @@ import java.util.concurrent.TimeUnit;
  *
  * @hide
  */
-/* package */ abstract class ContextHubServiceTransaction {
+abstract class ContextHubServiceTransaction {
     private final int mTransactionId;
+
     @ContextHubTransaction.Type
     private final int mTransactionType;
 
-    /** The ID of the nanoapp this transaction is targeted for, null if not applicable. */
     private final Long mNanoAppId;
 
-    /**
-     * The host package associated with this transaction.
-     */
     private final String mPackage;
 
-    /**
-     * The message sequence number associated with this transaction, null if not applicable.
-     */
     private final Integer mMessageSequenceNumber;
 
-    /**
-     * true if the transaction has already completed, false otherwise
-     */
+    private long mNextRetryTime;
+
+    private long mTimeoutTime;
+
+    /** The number of times the transaction has been started (start function called). */
+    private int mNumCompletedStartCalls;
+
+    private final short mHostEndpointId;
+
     private boolean mIsComplete = false;
 
-    /* package */ ContextHubServiceTransaction(int id, int type, String packageName) {
+    ContextHubServiceTransaction(int id, int type, String packageName) {
         mTransactionId = id;
         mTransactionType = type;
         mNanoAppId = null;
         mPackage = packageName;
         mMessageSequenceNumber = null;
+        mNextRetryTime = Long.MAX_VALUE;
+        mTimeoutTime = Long.MAX_VALUE;
+        mNumCompletedStartCalls = 0;
+        mHostEndpointId = Short.MAX_VALUE;
     }
 
-    /* package */ ContextHubServiceTransaction(int id, int type, long nanoAppId,
+    ContextHubServiceTransaction(int id, int type, long nanoAppId,
             String packageName) {
         mTransactionId = id;
         mTransactionType = type;
         mNanoAppId = nanoAppId;
         mPackage = packageName;
         mMessageSequenceNumber = null;
+        mNextRetryTime = Long.MAX_VALUE;
+        mTimeoutTime = Long.MAX_VALUE;
+        mNumCompletedStartCalls = 0;
+        mHostEndpointId = Short.MAX_VALUE;
     }
 
-    /* package */ ContextHubServiceTransaction(int id, int type, String packageName,
-            int messageSequenceNumber) {
+    ContextHubServiceTransaction(int id, int type, String packageName,
+            int messageSequenceNumber, short hostEndpointId) {
         mTransactionId = id;
         mTransactionType = type;
         mNanoAppId = null;
         mPackage = packageName;
         mMessageSequenceNumber = messageSequenceNumber;
+        mNextRetryTime = Long.MAX_VALUE;
+        mTimeoutTime = Long.MAX_VALUE;
+        mNumCompletedStartCalls = 0;
+        mHostEndpointId = hostEndpointId;
     }
 
     /**
@@ -95,7 +107,7 @@ import java.util.concurrent.TimeUnit;
      *
      * @param result the result of the transaction
      */
-    /* package */ void onTransactionComplete(@ContextHubTransaction.Result int result) {
+    void onTransactionComplete(@ContextHubTransaction.Result int result) {
     }
 
     /**
@@ -106,44 +118,51 @@ import java.util.concurrent.TimeUnit;
      * @param result           the result of the query
      * @param nanoAppStateList the list of nanoapps given by the query response
      */
-    /* package */ void onQueryResponse(
+    void onQueryResponse(
             @ContextHubTransaction.Result int result, List<NanoAppState> nanoAppStateList) {
     }
 
-    /**
-     * @return the ID of this transaction
-     */
-    /* package */ int getTransactionId() {
+    int getTransactionId() {
         return mTransactionId;
     }
 
-    /**
-     * @return the type of this transaction
-     * @see ContextHubTransaction.Type
-     */
     @ContextHubTransaction.Type
-    /* package */ int getTransactionType() {
+    int getTransactionType() {
         return mTransactionType;
     }
 
-    /**
-     * @return the message sequence number of this transaction
-     */
     Integer getMessageSequenceNumber() {
         return mMessageSequenceNumber;
     }
 
+    long getNextRetryTime() {
+        return mNextRetryTime;
+    }
+
+    long getTimeoutTime() {
+        return mTimeoutTime;
+    }
+
+    int getNumCompletedStartCalls() {
+        return mNumCompletedStartCalls;
+    }
+
+    short getHostEndpointId() {
+        return mHostEndpointId;
+    }
+
     /**
      * Gets the timeout period as defined in IContexthub.hal
      *
      * @return the timeout of this transaction in the specified time unit
      */
-    /* package */ long getTimeout(TimeUnit unit) {
+    long getTimeout(TimeUnit unit) {
         switch (mTransactionType) {
             case ContextHubTransaction.TYPE_LOAD_NANOAPP:
                 return unit.convert(30L, TimeUnit.SECONDS);
             case ContextHubTransaction.TYPE_RELIABLE_MESSAGE:
-                return unit.convert(1000L, TimeUnit.MILLISECONDS);
+                return unit.convert(ContextHubTransactionManager.RELIABLE_MESSAGE_TIMEOUT.toNanos(),
+                        TimeUnit.NANOSECONDS);
             case ContextHubTransaction.TYPE_UNLOAD_NANOAPP:
             case ContextHubTransaction.TYPE_ENABLE_NANOAPP:
             case ContextHubTransaction.TYPE_DISABLE_NANOAPP:
@@ -159,14 +178,23 @@ import java.util.concurrent.TimeUnit;
      *
      * Should only be called as a result of a response from a Context Hub callback
      */
-    /* package */ void setComplete() {
+    void setComplete() {
         mIsComplete = true;
     }
 
-    /**
-     * @return true if the transaction has already completed, false otherwise
-     */
-    /* package */ boolean isComplete() {
+    void setNextRetryTime(long nextRetryTime) {
+        mNextRetryTime = nextRetryTime;
+    }
+
+    void setTimeoutTime(long timeoutTime) {
+        mTimeoutTime = timeoutTime;
+    }
+
+    void setNumCompletedStartCalls(int numCompletedStartCalls) {
+        mNumCompletedStartCalls = numCompletedStartCalls;
+    }
+
+    boolean isComplete() {
         return mIsComplete;
     }
 
@@ -187,7 +215,18 @@ import java.util.concurrent.TimeUnit;
             out.append(", messageSequenceNumber = ");
             out.append(mMessageSequenceNumber);
         }
+        if (mTransactionType == ContextHubTransaction.TYPE_RELIABLE_MESSAGE) {
+            out.append(", nextRetryTime = ");
+            out.append(mNextRetryTime);
+            out.append(", timeoutTime = ");
+            out.append(mTimeoutTime);
+            out.append(", numCompletedStartCalls = ");
+            out.append(mNumCompletedStartCalls);
+            out.append(", hostEndpointId = ");
+            out.append(mHostEndpointId);
+        }
         out.append(")");
+
         return out.toString();
     }
 }
diff --git a/services/core/java/com/android/server/location/contexthub/ContextHubTestModeManager.java b/services/core/java/com/android/server/location/contexthub/ContextHubTestModeManager.java
new file mode 100644
index 000000000000..e50324eb7c83
--- /dev/null
+++ b/services/core/java/com/android/server/location/contexthub/ContextHubTestModeManager.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2024 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.android.server.location.contexthub;
+
+import android.chre.flags.Flags;
+import android.hardware.location.NanoAppMessage;
+import android.util.Log;
+
+import java.util.Random;
+
+/**
+ * A class to manage behaviors during test mode. This is used for testing.
+ * @hide
+ */
+public class ContextHubTestModeManager {
+    private static final String TAG = "ContextHubTestModeManager";
+
+    /** Probability (in percent) of duplicating a message. */
+    private static final int MESSAGE_DROP_PROBABILITY_PERCENT = 20;
+
+    /** Probability (in percent) of duplicating a message. */
+    private static final int MESSAGE_DUPLICATION_PROBABILITY_PERCENT = 20;
+
+    /** The number of total messages to send when the duplicate event happens. */
+    private static final int NUM_MESSAGES_TO_DUPLICATE = 3;
+
+    /** A probability percent for a certain event. */
+    private static final int MAX_PROBABILITY_PERCENT = 100;
+
+    private final Random mRandom = new Random();
+
+    /**
+     * @return whether the message was handled
+     * @see ContextHubServiceCallback#handleNanoappMessage
+     */
+    public boolean handleNanoappMessage(Runnable handleMessage, NanoAppMessage message) {
+        if (Flags.reliableMessageDuplicateDetectionService()
+                && message.isReliable()
+                && mRandom.nextInt(MAX_PROBABILITY_PERCENT)
+                        < MESSAGE_DUPLICATION_PROBABILITY_PERCENT) {
+            Log.i(TAG, "[TEST MODE] Duplicating message ("
+                    + NUM_MESSAGES_TO_DUPLICATE
+                    + " sends) with message sequence number: "
+                    + message.getMessageSequenceNumber());
+            for (int i = 0; i < NUM_MESSAGES_TO_DUPLICATE; ++i) {
+                handleMessage.run();
+            }
+            return true;
+        }
+        return false;
+    }
+
+    /**
+     * @return whether the message was handled
+     * @see IContextHubWrapper#sendMessageToContextHub
+     */
+    public boolean sendMessageToContextHub(NanoAppMessage message) {
+        if (Flags.reliableMessageRetrySupportService()
+                && message.isReliable()
+                && mRandom.nextInt(MAX_PROBABILITY_PERCENT)
+                        < MESSAGE_DROP_PROBABILITY_PERCENT) {
+            Log.i(TAG, "[TEST MODE] Dropping message with message sequence number: "
+                    + message.getMessageSequenceNumber());
+            return true;
+        }
+        return false;
+    }
+}
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 ec94e2be2c59..3051379d7b35 100644
--- a/services/core/java/com/android/server/location/contexthub/ContextHubTransactionManager.java
+++ b/services/core/java/com/android/server/location/contexthub/ContextHubTransactionManager.java
@@ -16,19 +16,26 @@
 
 package com.android.server.location.contexthub;
 
+import android.chre.flags.Flags;
 import android.hardware.location.ContextHubTransaction;
 import android.hardware.location.IContextHubTransactionCallback;
 import android.hardware.location.NanoAppBinary;
 import android.hardware.location.NanoAppMessage;
 import android.hardware.location.NanoAppState;
 import android.os.RemoteException;
+import android.os.SystemClock;
 import android.util.Log;
 
+import java.time.Duration;
 import java.util.ArrayDeque;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -47,34 +54,30 @@ import java.util.concurrent.atomic.AtomicInteger;
 /* package */ class ContextHubTransactionManager {
     private static final String TAG = "ContextHubTransactionManager";
 
-    /*
-     * Maximum number of transaction requests that can be pending at a time
-     */
+    public static final Duration RELIABLE_MESSAGE_TIMEOUT = Duration.ofSeconds(1);
+
     private static final int MAX_PENDING_REQUESTS = 10000;
 
-    /*
-     * The proxy to talk to the Context Hub
-     */
+    private static final int RELIABLE_MESSAGE_MAX_NUM_RETRY = 3;
+
+    private static final Duration RELIABLE_MESSAGE_RETRY_WAIT_TIME = Duration.ofMillis(250);
+
+    private static final Duration RELIABLE_MESSAGE_MIN_WAIT_TIME = Duration.ofNanos(1000);
+
     private final IContextHubWrapper mContextHubProxy;
 
-    /*
-     * The manager for all clients for the service.
-     */
     private final ContextHubClientManager mClientManager;
 
-    /*
-     * The nanoapp state manager for the service
-     */
     private final NanoAppStateManager mNanoAppStateManager;
 
-    /*
-     * A queue containing the current transactions
-     */
     private final ArrayDeque<ContextHubServiceTransaction> mTransactionQueue = new ArrayDeque<>();
 
-    /*
-     * The next available transaction ID
-     */
+    private final Map<Integer, ContextHubServiceTransaction> mReliableMessageTransactionMap =
+            new HashMap<>();
+
+    /** A set of host endpoint IDs that have an active pending transaction. */
+    private final Set<Short> mReliableMessageHostEndpointIdActiveSet = new HashSet<>();
+
     private final AtomicInteger mNextAvailableId = new AtomicInteger();
 
     /**
@@ -86,10 +89,12 @@ import java.util.concurrent.atomic.AtomicInteger;
             new AtomicInteger(new Random().nextInt(Integer.MAX_VALUE / 2));
 
     /*
-     * An executor and the future object for scheduling timeout timers
+     * An executor and the future object for scheduling timeout timers and
+     * for scheduling the processing of reliable message transactions.
      */
-    private final ScheduledThreadPoolExecutor mTimeoutExecutor = new ScheduledThreadPoolExecutor(1);
+    private final ScheduledThreadPoolExecutor mExecutor = new ScheduledThreadPoolExecutor(1);
     private ScheduledFuture<?> mTimeoutFuture = null;
+    private ScheduledFuture<?> mReliableMessageTransactionFuture = null;
 
     /*
      * The list of previous transaction records.
@@ -333,7 +338,7 @@ import java.util.concurrent.atomic.AtomicInteger;
             IContextHubTransactionCallback transactionCallback, String packageName) {
         return new ContextHubServiceTransaction(mNextAvailableId.getAndIncrement(),
                 ContextHubTransaction.TYPE_RELIABLE_MESSAGE, packageName,
-                mNextAvailableMessageSequenceNumber.getAndIncrement()) {
+                mNextAvailableMessageSequenceNumber.getAndIncrement(), hostEndpointId) {
             @Override
             /* package */ int onTransact() {
                 try {
@@ -416,16 +421,23 @@ import java.util.concurrent.atomic.AtomicInteger;
             return;
         }
 
-        if (mTransactionQueue.size() == MAX_PENDING_REQUESTS) {
+        if (mTransactionQueue.size() >= MAX_PENDING_REQUESTS
+                || mReliableMessageTransactionMap.size() >= MAX_PENDING_REQUESTS) {
             throw new IllegalStateException("Transaction queue is full (capacity = "
                     + MAX_PENDING_REQUESTS + ")");
         }
 
-        mTransactionQueue.add(transaction);
         mTransactionRecordDeque.add(new TransactionRecord(transaction.toString()));
-
-        if (mTransactionQueue.size() == 1) {
-            startNextTransaction();
+        if (Flags.reliableMessageRetrySupportService()
+                && transaction.getTransactionType()
+                        == ContextHubTransaction.TYPE_RELIABLE_MESSAGE) {
+            mReliableMessageTransactionMap.put(transaction.getMessageSequenceNumber(), transaction);
+            mExecutor.execute(() -> processMessageTransactions());
+        } else {
+            mTransactionQueue.add(transaction);
+            if (mTransactionQueue.size() == 1) {
+                startNextTransaction();
+            }
         }
     }
 
@@ -455,26 +467,42 @@ import java.util.concurrent.atomic.AtomicInteger;
 
     /* package */
     synchronized void onMessageDeliveryResponse(int messageSequenceNumber, boolean success) {
-        ContextHubServiceTransaction transaction = mTransactionQueue.peek();
-        if (transaction == null) {
-            Log.w(TAG, "Received unexpected transaction response (no transaction pending)");
+        if (!Flags.reliableMessageRetrySupportService()) {
+            ContextHubServiceTransaction transaction = mTransactionQueue.peek();
+            if (transaction == null) {
+                Log.w(TAG, "Received unexpected transaction response (no transaction pending)");
+                return;
+            }
+
+            Integer transactionMessageSequenceNumber = transaction.getMessageSequenceNumber();
+            if (transaction.getTransactionType() != ContextHubTransaction.TYPE_RELIABLE_MESSAGE
+                    || transactionMessageSequenceNumber == null
+                    || transactionMessageSequenceNumber != messageSequenceNumber) {
+                Log.w(TAG, "Received unexpected message transaction response (expected message "
+                        + "sequence number = "
+                        + transaction.getMessageSequenceNumber()
+                        + ", received messageSequenceNumber = " + messageSequenceNumber + ")");
+                return;
+            }
+
+            transaction.onTransactionComplete(success ? ContextHubTransaction.RESULT_SUCCESS :
+                            ContextHubTransaction.RESULT_FAILED_AT_HUB);
+            removeTransactionAndStartNext();
             return;
         }
 
-        Integer transactionMessageSequenceNumber = transaction.getMessageSequenceNumber();
-        if (transaction.getTransactionType() != ContextHubTransaction.TYPE_RELIABLE_MESSAGE
-                || transactionMessageSequenceNumber == null
-                || transactionMessageSequenceNumber != messageSequenceNumber) {
-            Log.w(TAG, "Received unexpected message transaction response (expected message "
-                    + "sequence number = "
-                    + transaction.getMessageSequenceNumber()
-                    + ", received messageSequenceNumber = " + messageSequenceNumber + ")");
+        ContextHubServiceTransaction transaction =
+                mReliableMessageTransactionMap.get(messageSequenceNumber);
+        if (transaction == null) {
+            Log.w(TAG, "Could not find reliable message transaction with message sequence number"
+                    + messageSequenceNumber);
             return;
         }
 
-        transaction.onTransactionComplete(success ? ContextHubTransaction.RESULT_SUCCESS :
-                        ContextHubTransaction.RESULT_FAILED_AT_HUB);
-        removeTransactionAndStartNext();
+        completeMessageTransaction(transaction,
+                success ? ContextHubTransaction.RESULT_SUCCESS
+                        : ContextHubTransaction.RESULT_FAILED_AT_HUB);
+        mExecutor.execute(() -> processMessageTransactions());
     }
 
     /**
@@ -503,6 +531,15 @@ import java.util.concurrent.atomic.AtomicInteger;
      */
     /* package */
     synchronized void onHubReset() {
+        if (Flags.reliableMessageRetrySupportService()) {
+            Iterator<Map.Entry<Integer, ContextHubServiceTransaction>> iter =
+                    mReliableMessageTransactionMap.entrySet().iterator();
+            while (iter.hasNext()) {
+                completeMessageTransaction(iter.next().getValue(),
+                        ContextHubTransaction.RESULT_FAILED_AT_HUB, iter);
+            }
+        }
+
         ContextHubServiceTransaction transaction = mTransactionQueue.peek();
         if (transaction == null) {
             return;
@@ -566,7 +603,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 
                 long timeoutMs = transaction.getTimeout(TimeUnit.MILLISECONDS);
                 try {
-                    mTimeoutFuture = mTimeoutExecutor.schedule(
+                    mTimeoutFuture = mExecutor.schedule(
                             onTimeoutFunc, timeoutMs, TimeUnit.MILLISECONDS);
                 } catch (Exception e) {
                     Log.e(TAG, "Error when schedule a timer", e);
@@ -579,6 +616,136 @@ import java.util.concurrent.atomic.AtomicInteger;
         }
     }
 
+    /**
+     * Processes message transactions, starting and completing them as needed.
+     * This function is called when adding a message transaction or when a timer
+     * expires for an existing message transaction's retry or timeout. The
+     * internal processing loop will iterate at most twice as if one iteration
+     * completes a transaction, the next iteration can only start new transactions.
+     * If the first iteration does not complete any transaction, the loop will
+     * only iterate once.
+     */
+    private synchronized void processMessageTransactions() {
+        if (!Flags.reliableMessageRetrySupportService()) {
+            return;
+        }
+
+        if (mReliableMessageTransactionFuture != null) {
+            mReliableMessageTransactionFuture.cancel(/* mayInterruptIfRunning= */ false);
+            mReliableMessageTransactionFuture = null;
+        }
+
+        long now = SystemClock.elapsedRealtimeNanos();
+        long nextExecutionTime = Long.MAX_VALUE;
+        boolean continueProcessing;
+        do {
+            continueProcessing = false;
+            Iterator<Map.Entry<Integer, ContextHubServiceTransaction>> iter =
+                    mReliableMessageTransactionMap.entrySet().iterator();
+            while (iter.hasNext()) {
+                ContextHubServiceTransaction transaction = iter.next().getValue();
+                short hostEndpointId = transaction.getHostEndpointId();
+                int numCompletedStartCalls = transaction.getNumCompletedStartCalls();
+                if (numCompletedStartCalls == 0
+                        && mReliableMessageHostEndpointIdActiveSet.contains(hostEndpointId)) {
+                    continue;
+                }
+
+                long nextRetryTime = transaction.getNextRetryTime();
+                long timeoutTime = transaction.getTimeoutTime();
+                boolean transactionTimedOut = timeoutTime <= now;
+                boolean transactionHitMaxRetries = nextRetryTime <= now
+                        && numCompletedStartCalls > RELIABLE_MESSAGE_MAX_NUM_RETRY;
+                if (transactionTimedOut || transactionHitMaxRetries) {
+                    completeMessageTransaction(transaction,
+                            ContextHubTransaction.RESULT_FAILED_TIMEOUT, iter);
+                    continueProcessing = true;
+                } else {
+                    if (nextRetryTime <= now || numCompletedStartCalls <= 0) {
+                        startMessageTransaction(transaction, now);
+                    }
+
+                    nextExecutionTime = Math.min(nextExecutionTime,
+                            transaction.getNextRetryTime());
+                    nextExecutionTime = Math.min(nextExecutionTime,
+                            transaction.getTimeoutTime());
+                }
+            }
+        } while (continueProcessing);
+
+        if (nextExecutionTime < Long.MAX_VALUE) {
+            mReliableMessageTransactionFuture = mExecutor.schedule(
+                    () -> processMessageTransactions(),
+                    Math.max(nextExecutionTime - SystemClock.elapsedRealtimeNanos(),
+                            RELIABLE_MESSAGE_MIN_WAIT_TIME.toNanos()),
+                    TimeUnit.NANOSECONDS);
+        }
+    }
+
+    /**
+     * Completes a message transaction and removes it from the reliable message map.
+     *
+     * @param transaction The transaction to complete.
+     * @param result The result code.
+     */
+    private void completeMessageTransaction(ContextHubServiceTransaction transaction,
+            @ContextHubTransaction.Result int result) {
+        completeMessageTransaction(transaction, result, /* iter= */ null);
+    }
+
+    /**
+     * Completes a message transaction and removes it from the reliable message map using iter.
+     *
+     * @param transaction The transaction to complete.
+     * @param result The result code.
+     * @param iter The iterator for the reliable message map - used to remove the message directly.
+     */
+    private void completeMessageTransaction(ContextHubServiceTransaction transaction,
+            @ContextHubTransaction.Result int result,
+            Iterator<Map.Entry<Integer, ContextHubServiceTransaction>> iter) {
+        transaction.onTransactionComplete(result);
+
+        if (iter == null) {
+            mReliableMessageTransactionMap.remove(transaction.getMessageSequenceNumber());
+        } else {
+            iter.remove();
+        }
+        mReliableMessageHostEndpointIdActiveSet.remove(transaction.getHostEndpointId());
+
+        Log.d(TAG, "Successfully completed reliable message transaction with "
+                + "message sequence number: " + transaction.getMessageSequenceNumber()
+                + " and result: " + result);
+    }
+
+    /**
+     * Starts a message transaction.
+     *
+     * @param transaction The transaction to start.
+     * @param now The now time.
+     */
+    private void startMessageTransaction(ContextHubServiceTransaction transaction, long now) {
+        int numCompletedStartCalls = transaction.getNumCompletedStartCalls();
+        @ContextHubTransaction.Result int result = transaction.onTransact();
+        if (result == ContextHubTransaction.RESULT_SUCCESS) {
+            Log.d(TAG, "Successfully "
+                    + (numCompletedStartCalls == 0 ? "started" : "retried")
+                    + " reliable message transaction with message sequence number: "
+                    + transaction.getMessageSequenceNumber());
+        } else {
+            Log.w(TAG, "Could not start reliable message transaction with "
+                    + "message sequence number: "
+                    + transaction.getMessageSequenceNumber()
+                    + ", result: " + result);
+        }
+
+        transaction.setNextRetryTime(now + RELIABLE_MESSAGE_RETRY_WAIT_TIME.toNanos());
+        if (transaction.getTimeoutTime() == Long.MAX_VALUE) { // first time starting transaction
+            transaction.setTimeoutTime(now + RELIABLE_MESSAGE_TIMEOUT.toNanos());
+        }
+        transaction.setNumCompletedStartCalls(numCompletedStartCalls + 1);
+        mReliableMessageHostEndpointIdActiveSet.add(transaction.getHostEndpointId());
+    }
+
     private int toStatsTransactionResult(@ContextHubTransaction.Result int result) {
         switch (result) {
             case ContextHubTransaction.RESULT_SUCCESS:
@@ -605,19 +772,34 @@ import java.util.concurrent.atomic.AtomicInteger;
 
     @Override
     public String toString() {
-        StringBuilder sb = new StringBuilder(100);
-        ContextHubServiceTransaction[] arr;
+        StringBuilder sb = new StringBuilder();
+        int i = 0;
         synchronized (this) {
-            arr = mTransactionQueue.toArray(new ContextHubServiceTransaction[0]);
-        }
-        for (int i = 0; i < arr.length; i++) {
-            sb.append(i + ": " + arr[i] + "\n");
-        }
+            for (ContextHubServiceTransaction transaction: mTransactionQueue) {
+                sb.append(i);
+                sb.append(": ");
+                sb.append(transaction.toString());
+                sb.append("\n");
+                ++i;
+            }
 
-        sb.append("Transaction History:\n");
-        Iterator<TransactionRecord> iterator = mTransactionRecordDeque.descendingIterator();
-        while (iterator.hasNext()) {
-            sb.append(iterator.next() + "\n");
+            if (Flags.reliableMessageRetrySupportService()) {
+                for (ContextHubServiceTransaction transaction:
+                        mReliableMessageTransactionMap.values()) {
+                    sb.append(i);
+                    sb.append(": ");
+                    sb.append(transaction.toString());
+                    sb.append("\n");
+                    ++i;
+                }
+            }
+
+            sb.append("Transaction History:\n");
+            Iterator<TransactionRecord> iterator = mTransactionRecordDeque.descendingIterator();
+            while (iterator.hasNext()) {
+                sb.append(iterator.next());
+                sb.append("\n");
+            }
         }
         return sb.toString();
     }
diff --git a/services/core/java/com/android/server/location/contexthub/IContextHubWrapper.java b/services/core/java/com/android/server/location/contexthub/IContextHubWrapper.java
index 552809bc7453..4fc3d8715a88 100644
--- a/services/core/java/com/android/server/location/contexthub/IContextHubWrapper.java
+++ b/services/core/java/com/android/server/location/contexthub/IContextHubWrapper.java
@@ -52,6 +52,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * @hide
@@ -432,10 +433,16 @@ public abstract class IContextHubWrapper {
 
         // Use this thread in case where the execution requires to be on a service thread.
         // For instance, AppOpsManager.noteOp requires the UPDATE_APP_OPS_STATS permission.
-        private HandlerThread mHandlerThread =
+        private final HandlerThread mHandlerThread =
                 new HandlerThread("Context Hub AIDL callback", Process.THREAD_PRIORITY_BACKGROUND);
         private Handler mHandler;
 
+        // True if test mode is enabled for the Context Hub
+        private final AtomicBoolean mIsTestModeEnabled = new AtomicBoolean(false);
+
+        // The test mode manager that manages behaviors during test mode
+        private final ContextHubTestModeManager mTestModeManager = new ContextHubTestModeManager();
+
         private class ContextHubAidlCallback extends
                 android.hardware.contexthub.IContextHubCallback.Stub {
             private final int mContextHubId;
@@ -549,6 +556,8 @@ public abstract class IContextHubWrapper {
             } else {
                 Log.e(TAG, "mHandleServiceRestartCallback is not set");
             }
+
+            mIsTestModeEnabled.set(false);
         }
 
         public Pair<List<ContextHubInfo>, List<String>> getHubs() throws RemoteException {
@@ -659,7 +668,17 @@ public abstract class IContextHubWrapper {
             try {
                 var msg = ContextHubServiceUtil.createAidlContextHubMessage(
                         hostEndpointId, message);
-                hub.sendMessageToHub(contextHubId, msg);
+
+                // Only process the message normally if not using test mode manager or if
+                // the test mode manager call returned false as this indicates it did not
+                // process the message.
+                boolean useTestModeManager = Flags.reliableMessageImplementation()
+                        && Flags.reliableMessageTestModeBehavior()
+                        && mIsTestModeEnabled.get();
+                if (!useTestModeManager || !mTestModeManager.sendMessageToContextHub(message)) {
+                    hub.sendMessageToHub(contextHubId, msg);
+                }
+
                 return ContextHubTransaction.RESULT_SUCCESS;
             } catch (RemoteException | ServiceSpecificException e) {
                 return ContextHubTransaction.RESULT_FAILED_UNKNOWN;
@@ -828,6 +847,7 @@ public abstract class IContextHubWrapper {
                 return false;
             }
 
+            mIsTestModeEnabled.set(enable);
             try {
                 hub.setTestMode(enable);
                 return true;