diff --git a/services/backup/java/com/android/server/backup/OperationStorage.java b/services/backup/java/com/android/server/backup/OperationStorage.java new file mode 100644 index 0000000000000..466f647fd0346 --- /dev/null +++ b/services/backup/java/com/android/server/backup/OperationStorage.java @@ -0,0 +1,156 @@ +/* + * Copyright (C) 2021 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.backup; + +import android.annotation.IntDef; + +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.util.Set; + +/** + * OperationStorage is an abstraction around a set of active operations. + * + * Operations are registered with a token that must first be obtained from + * {@link UserBackupManagerService#generateRandomIntegerToken()}. When + * registering, the caller may also associate a set of package names with + * the operation. + * + * TODO(b/208442527): have the token be generated within and returned by + * registerOperation, as it should be an internal detail. + * + * Operations have a type and a state. Although ints, the values that can + * be used are defined in {@link UserBackupManagerService}. If the type of + * an operation is OP_BACKUP, then it represents a task running backups. The + * task is provided when registering the operation because it provides a + * handle to cancel the backup. + */ +public interface OperationStorage { + + @Retention(RetentionPolicy.SOURCE) + @IntDef({ + OpState.PENDING, + OpState.ACKNOWLEDGED, + OpState.TIMEOUT + }) + public @interface OpState { + // The operation is in progress. + int PENDING = 0; + // The operation has been acknowledged. + int ACKNOWLEDGED = 1; + // The operation has timed out. + int TIMEOUT = -1; + } + + @Retention(RetentionPolicy.SOURCE) + @IntDef({ + OpType.BACKUP_WAIT, + OpType.RESTORE_WAIT, + OpType.BACKUP, + }) + public @interface OpType { + // Waiting for backup agent to respond during backup operation. + int BACKUP_WAIT = 0; + // Waiting for backup agent to respond during restore operation. + int RESTORE_WAIT = 1; + // An entire backup operation spanning multiple packages. + int BACKUP = 2; + } + + /** + * Record an ongoing operation of given type and in the given initial + * state. The associated task is used as a callback. + * + * @param token an operation token issued by + * {@link UserBackupManagerService#generateRandomIntegerToken()} + * @param initialState the state that the operation starts in + * @param task the {@link BackupRestoreTask} that is expected to + * remove the operation on completion, and which may + * be notified if the operation requires cancelling. + * @param type the type of the operation. + */ + void registerOperation(int token, @OpState int initialState, + BackupRestoreTask task, @OpType int type); + + /** + * See {@link #registerOperation()}. In addition this method accepts a set + * of package names which are associated with the operation. + * + * @param token See {@link #registerOperation()} + * @param initialState See {@link #registerOperation()} + * @param packageNames the package names to associate with the operation. + * @param task See {@link #registerOperation()} + * @param type See {@link #registerOperation()} + */ + void registerOperationForPackages(int token, @OpState int initialState, + Set packageNames, BackupRestoreTask task, @OpType int type); + + /** + * Remove the operation identified by token. This is called when the + * operation is no longer in progress and should be dropped. Any association + * with package names provided in {@link #registerOperation()} is dropped as + * well. + * + * @param token the operation token specified when registering the operation. + */ + void removeOperation(int token); + + /** + * Obtain the number of currently registered operations. + * + * @return the number of currently registered operations. + */ + int numOperations(); + + /** + * Determine if a backup operation is in progress or not. + * + * @return true if any operation is registered of type BACKUP and in + * state PENDING. + */ + boolean isBackupOperationInProgress(); + + /** + * Obtain a set of operation tokens for all pending operations that were + * registered with an association to the specified package name. + * + * @param packageName the name of the package used at registration time + * + * @return a set of operation tokens associated to package name. + */ + Set operationTokensForPackage(String packageName); + + /** + * Obtain a set of operation tokens for all pending operations that are + * of the specified operation type. + * + * @param type the type of the operation provided at registration time. + * + * @return a set of operation tokens for operations of that type. + */ + Set operationTokensForOpType(@OpType int type); + + /** + * Obtain a set of operation tokens for all pending operations that are + * currently in the specified operation state. + * + * @param state the state of the operation. + * + * @return a set of operation tokens for operations in that state. + */ + Set operationTokensForOpState(@OpState int state); +}; diff --git a/services/backup/java/com/android/server/backup/internal/LifecycleOperationStorage.java b/services/backup/java/com/android/server/backup/internal/LifecycleOperationStorage.java new file mode 100644 index 0000000000000..6908c60034b6e --- /dev/null +++ b/services/backup/java/com/android/server/backup/internal/LifecycleOperationStorage.java @@ -0,0 +1,356 @@ +/* + * Copyright (C) 2021 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.backup.internal; + +import static com.android.server.backup.BackupManagerService.DEBUG; +import static com.android.server.backup.BackupManagerService.MORE_DEBUG; + +import android.annotation.UserIdInt; +import android.util.Slog; +import android.util.SparseArray; + +import com.android.internal.annotations.GuardedBy; +import com.android.server.backup.BackupRestoreTask; +import com.android.server.backup.OperationStorage; + +import com.google.android.collect.Sets; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.IntConsumer; + +/** + * LifecycleOperationStorage is responsible for maintaining a set of currently + * active operations. Each operation has a type and state, and a callback that + * can receive events upon operation completion or cancellation. It may also + * be associated with one or more package names. + * + * An operation wraps a {@link BackupRestoreTask} within it. + * It's the responsibility of this task to remove the operation from this array. + * + * If type of operation is {@code OP_TYPE_WAIT}, it is waiting for an ACK or + * timeout. + * + * A BackupRestore task gets notified of AVK/timeout for the operation via + * {@link BackupRestoreTask#handleCancel()}, + * {@link BackupRestoreTask#operationComplete()} and {@code notifyAll} called + * on the {@code mCurrentOpLock}. + * + * {@link LifecycleOperationStorage#waitUntilOperationComplete(int)} is used in + * various places to 'wait' for notifyAll and detect change of pending state of + * an operation. So typically, an operation will be removed from this array by: + * - {@link BackupRestoreTask#handleCancel()} and + * - {@link BackupRestoreTask#operationComplete()} OR + * {@link BackupRestoreTask#waitUntilOperationComplete()}. + * Do not remove at both these places because {@code waitUntilOperationComplete} + * relies on the operation being present to determine its completion status. + * + * If type of operation is {@code OP_BACKUP}, it is a task running backups. It + * provides a handle to cancel backup tasks. + */ +public class LifecycleOperationStorage implements OperationStorage { + private static final String TAG = "LifecycleOperationStorage"; + + private final int mUserId; + + private final Object mOperationsLock = new Object(); + + // Bookkeeping of in-flight operations. The operation token is the index of + // the entry in the pending operations list. + @GuardedBy("mOperationsLock") + private final SparseArray mOperations = new SparseArray<>(); + + // Association from package name to one or more operations relating to that + // package. + @GuardedBy("mOperationsLock") + private final Map> mOpTokensByPackage = new HashMap<>(); + + public LifecycleOperationStorage(@UserIdInt int userId) { + this.mUserId = userId; + } + + /** See {@link OperationStorage#registerOperation()} */ + @Override + public void registerOperation(int token, @OpState int initialState, + BackupRestoreTask task, @OpType int type) { + registerOperationForPackages(token, initialState, Sets.newHashSet(), task, type); + } + + /** See {@link OperationStorage#registerOperationForPackages()} */ + @Override + public void registerOperationForPackages(int token, @OpState int initialState, + Set packageNames, BackupRestoreTask task, @OpType int type) { + synchronized (mOperationsLock) { + mOperations.put(token, new Operation(initialState, task, type)); + for (String packageName : packageNames) { + Set tokens = mOpTokensByPackage.get(packageName); + if (tokens == null) { + tokens = new HashSet(); + } + tokens.add(token); + mOpTokensByPackage.put(packageName, tokens); + } + } + } + + /** See {@link OperationStorage#removeOperation()} */ + @Override + public void removeOperation(int token) { + synchronized (mOperationsLock) { + mOperations.remove(token); + Set packagesWithTokens = mOpTokensByPackage.keySet(); + for (String packageName : packagesWithTokens) { + Set tokens = mOpTokensByPackage.get(packageName); + if (tokens == null) { + continue; + } + tokens.remove(token); + mOpTokensByPackage.put(packageName, tokens); + } + } + } + + /** See {@link OperationStorage#numOperations()}. */ + @Override + public int numOperations() { + synchronized (mOperationsLock) { + return mOperations.size(); + } + } + + /** See {@link OperationStorage#isBackupOperationInProgress()}. */ + @Override + public boolean isBackupOperationInProgress() { + synchronized (mOperationsLock) { + for (int i = 0; i < mOperations.size(); i++) { + Operation op = mOperations.valueAt(i); + if (op.type == OpType.BACKUP && op.state == OpState.PENDING) { + return true; + } + } + return false; + } + } + + /** See {@link OperationStorage#operationTokensForPackage()} */ + @Override + public Set operationTokensForPackage(String packageName) { + synchronized (mOperationsLock) { + final Set tokens = mOpTokensByPackage.get(packageName); + Set result = Sets.newHashSet(); + if (tokens != null) { + result.addAll(tokens); + } + return result; + } + } + + /** See {@link OperationStorage#operationTokensForOpType()} */ + @Override + public Set operationTokensForOpType(@OpType int type) { + Set tokens = Sets.newHashSet(); + synchronized (mOperationsLock) { + for (int i = 0; i < mOperations.size(); i++) { + final Operation op = mOperations.valueAt(i); + final int token = mOperations.keyAt(i); + if (op.type == type) { + tokens.add(token); + } + } + return tokens; + } + } + + /** See {@link OperationStorage#operationTokensForOpState()} */ + @Override + public Set operationTokensForOpState(@OpState int state) { + Set tokens = Sets.newHashSet(); + synchronized (mOperationsLock) { + for (int i = 0; i < mOperations.size(); i++) { + final Operation op = mOperations.valueAt(i); + final int token = mOperations.keyAt(i); + if (op.state == state) { + tokens.add(token); + } + } + return tokens; + } + } + + /** + * A blocking function that blocks the caller until the operation identified + * by {@code token} is complete - either via a message from the backup, + * agent or through cancellation. + * + * @param token the operation token specified when registering the operation + * @param callback a lambda which is invoked once only when the operation + * completes - ie. if this method is called twice for the + * same token, the lambda is not invoked the second time. + * @return true if the operation was ACKed prior to or during this call. + */ + public boolean waitUntilOperationComplete(int token, IntConsumer callback) { + if (MORE_DEBUG) { + Slog.i(TAG, "[UserID:" + mUserId + "] Blocking until operation complete for " + + Integer.toHexString(token)); + } + @OpState int finalState = OpState.PENDING; + Operation op = null; + synchronized (mOperationsLock) { + while (true) { + op = mOperations.get(token); + if (op == null) { + // mysterious disappearance: treat as success with no callback + break; + } else { + if (op.state == OpState.PENDING) { + try { + mOperationsLock.wait(); + } catch (InterruptedException e) { + Slog.w(TAG, "Waiting on mOperationsLock: ", e); + } + // When the wait is notified we loop around and recheck the current state + } else { + if (MORE_DEBUG) { + Slog.d(TAG, "[UserID:" + mUserId + + "] Unblocked waiting for operation token=" + + Integer.toHexString(token)); + } + // No longer pending; we're done + finalState = op.state; + break; + } + } + } + } + + removeOperation(token); + if (op != null) { + callback.accept(op.type); + } + if (MORE_DEBUG) { + Slog.v(TAG, "[UserID:" + mUserId + "] operation " + Integer.toHexString(token) + + " complete: finalState=" + finalState); + } + return finalState == OpState.ACKNOWLEDGED; + } + + /** + * Signals that an ongoing operation is complete: after a currently-active + * backup agent has notified us that it has completed the outstanding + * asynchronous backup/restore operation identified by the supplied + * {@code} token. + * + * @param token the operation token specified when registering the operation + * @param result a result code or error code for the completed operation + * @param callback a lambda that is invoked if the completion moves the + * operation from PENDING to ACKNOWLEDGED state. + */ + public void onOperationComplete(int token, long result, Consumer callback) { + if (MORE_DEBUG) { + Slog.v(TAG, "[UserID:" + mUserId + "] onOperationComplete: " + + Integer.toHexString(token) + " result=" + result); + } + Operation op = null; + synchronized (mOperationsLock) { + op = mOperations.get(token); + if (op != null) { + if (op.state == OpState.TIMEOUT) { + // The operation already timed out, and this is a late response. Tidy up + // and ignore it; we've already dealt with the timeout. + op = null; + mOperations.remove(token); + } else if (op.state == OpState.ACKNOWLEDGED) { + if (DEBUG) { + Slog.w(TAG, "[UserID:" + mUserId + "] Received duplicate ack for token=" + + Integer.toHexString(token)); + } + op = null; + mOperations.remove(token); + } else if (op.state == OpState.PENDING) { + // Can't delete op from mOperations. waitUntilOperationComplete can be + // called after we we receive this call. + op.state = OpState.ACKNOWLEDGED; + } + } + mOperationsLock.notifyAll(); + } + + // Invoke the operation's completion callback, if there is one. + if (op != null && op.callback != null) { + callback.accept(op.callback); + } + } + + /** + * Cancel the operation associated with {@code token}. Cancellation may be + * propagated to the operation's callback (a {@link BackupRestoreTask}) if + * the operation has one, and the cancellation is due to the operation + * timing out. + * + * @param token the operation token specified when registering the operation + * @param cancelAll this is passed on when propagating the cancellation + * @param operationTimedOutCallback a lambda that is invoked with the + * operation type where the operation is + * cancelled due to timeout, allowing the + * caller to do type-specific clean-ups. + */ + public void cancelOperation( + int token, boolean cancelAll, IntConsumer operationTimedOutCallback) { + // Notify any synchronous waiters + Operation op = null; + synchronized (mOperationsLock) { + op = mOperations.get(token); + if (MORE_DEBUG) { + if (op == null) { + Slog.w(TAG, "[UserID:" + mUserId + "] Cancel of token " + + Integer.toHexString(token) + " but no op found"); + } + } + int state = (op != null) ? op.state : OpState.TIMEOUT; + if (state == OpState.ACKNOWLEDGED) { + // The operation finished cleanly, so we have nothing more to do. + if (DEBUG) { + Slog.w(TAG, "[UserID:" + mUserId + "] Operation already got an ack." + + "Should have been removed from mCurrentOperations."); + } + op = null; + mOperations.delete(token); + } else if (state == OpState.PENDING) { + if (DEBUG) { + Slog.v(TAG, "[UserID:" + mUserId + "] Cancel: token=" + + Integer.toHexString(token)); + } + op.state = OpState.TIMEOUT; + // Can't delete op from mOperations here. waitUntilOperationComplete may be + // called after we receive cancel here. We need this op's state there. + operationTimedOutCallback.accept(op.type); + } + mOperationsLock.notifyAll(); + } + + // If there's a TimeoutHandler for this event, call it + if (op != null && op.callback != null) { + if (MORE_DEBUG) { + Slog.v(TAG, "[UserID:" + mUserId + " Invoking cancel on " + op.callback); + } + op.callback.handleCancel(cancelAll); + } + } +}; diff --git a/services/tests/servicestests/src/com/android/server/backup/internal/LifecycleOperationStorageTest.java b/services/tests/servicestests/src/com/android/server/backup/internal/LifecycleOperationStorageTest.java new file mode 100644 index 0000000000000..948acccba9478 --- /dev/null +++ b/services/tests/servicestests/src/com/android/server/backup/internal/LifecycleOperationStorageTest.java @@ -0,0 +1,194 @@ +/* + * Copyright (C) 2021 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.backup.internal; + +import static com.google.common.truth.Truth.assertThat; + +import static org.mockito.Mockito.verify; + +import android.platform.test.annotations.Presubmit; + +import androidx.test.runner.AndroidJUnit4; + +import com.android.server.backup.BackupRestoreTask; +import com.android.server.backup.OperationStorage.OpState; +import com.android.server.backup.OperationStorage.OpType; + +import com.google.android.collect.Sets; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import java.util.Set; + +@Presubmit +@RunWith(AndroidJUnit4.class) +public class LifecycleOperationStorageTest { + private static final int USER_ID = 0; + private static final int TOKEN_1 = 1; + private static final int TOKEN_2 = 2; + private static final int TOKEN_3 = 3; + private static final long RESULT = 123L; + + private static final String PKG_FOO = "com.android.foo"; + private static final String PKG_BAR = "com.android.bar"; + private static final String PKG_BAZ = "com.android.baz"; + private static final Set MULTIPLE_PKG = Sets.newHashSet(PKG_FOO); + private static final Set MULTIPLE_PKGS_1 = Sets.newHashSet(PKG_FOO, PKG_BAR); + private static final Set MULTIPLE_PKGS_2 = Sets.newHashSet(PKG_BAR, PKG_BAZ); + + @Mock private BackupRestoreTask mCallback; + private LifecycleOperationStorage mOpStorage; + + @Before + public void setUp() { + MockitoAnnotations.initMocks(/* testClass */ this); + mOpStorage = new LifecycleOperationStorage(USER_ID); + } + + @After + public void tearDown() {} + + @Test + public void testRegisterOperation_singleOperation() throws Exception { + mOpStorage.registerOperation(TOKEN_1, OpState.PENDING, mCallback, OpType.BACKUP_WAIT); + + Set tokens = mOpStorage.operationTokensForOpType(OpType.BACKUP_WAIT); + + assertThat(mOpStorage.numOperations()).isEqualTo(1); + assertThat(tokens).isEqualTo(only(TOKEN_1)); + } + + @Test + public void testRegisterOperation_multipleOperations() throws Exception { + mOpStorage.registerOperation(TOKEN_1, OpState.PENDING, mCallback, OpType.BACKUP_WAIT); + mOpStorage.registerOperation(TOKEN_2, OpState.ACKNOWLEDGED, mCallback, OpType.BACKUP_WAIT); + + Set typeWaitTokens = mOpStorage.operationTokensForOpType(OpType.BACKUP_WAIT); + Set statePendingTokens = mOpStorage.operationTokensForOpState(OpState.PENDING); + Set stateAcknowledgedTokens = + mOpStorage.operationTokensForOpState(OpState.ACKNOWLEDGED); + + assertThat(mOpStorage.numOperations()).isEqualTo(2); + assertThat(typeWaitTokens).isEqualTo(Sets.newHashSet(TOKEN_1, TOKEN_2)); + assertThat(statePendingTokens).isEqualTo(only(TOKEN_1)); + assertThat(stateAcknowledgedTokens).isEqualTo(only(TOKEN_2)); + } + + @Test + public void testRegisterOperationForPackages_singlePackage() throws Exception { + mOpStorage.registerOperationForPackages(TOKEN_1, OpState.PENDING, + MULTIPLE_PKG, mCallback, OpType.BACKUP_WAIT); + + Set tokens = mOpStorage.operationTokensForPackage(PKG_FOO); + + assertThat(mOpStorage.numOperations()).isEqualTo(1); + assertThat(tokens).isEqualTo(only(TOKEN_1)); + } + + @Test + public void testRegisterOperationForPackages_multiplePackage() throws Exception { + mOpStorage.registerOperationForPackages(TOKEN_1, OpState.PENDING, + MULTIPLE_PKGS_1, mCallback, OpType.BACKUP); + mOpStorage.registerOperationForPackages(TOKEN_2, OpState.PENDING, + MULTIPLE_PKGS_2, mCallback, OpType.BACKUP); + + Set tokensFoo = mOpStorage.operationTokensForPackage(PKG_FOO); + Set tokensBar = mOpStorage.operationTokensForPackage(PKG_BAR); + Set tokensBaz = mOpStorage.operationTokensForPackage(PKG_BAZ); + + assertThat(mOpStorage.numOperations()).isEqualTo(2); + assertThat(tokensFoo).isEqualTo(only(TOKEN_1)); + assertThat(tokensBar).isEqualTo(Sets.newHashSet(TOKEN_1, TOKEN_2)); + assertThat(tokensBaz).isEqualTo(only(TOKEN_2)); + } + + @Test + public void testRemoveOperation() throws Exception { + mOpStorage.registerOperation(TOKEN_2, OpState.PENDING, mCallback, OpType.BACKUP_WAIT); + + Set typeWaitTokens = mOpStorage.operationTokensForOpType(OpType.BACKUP_WAIT); + Set statePendingTokens = mOpStorage.operationTokensForOpState(OpState.PENDING); + + assertThat(mOpStorage.numOperations()).isEqualTo(1); + assertThat(typeWaitTokens).isEqualTo(only(TOKEN_2)); + assertThat(statePendingTokens).isEqualTo(only(TOKEN_2)); + + mOpStorage.removeOperation(TOKEN_2); + + typeWaitTokens = mOpStorage.operationTokensForOpType(OpType.BACKUP_WAIT); + statePendingTokens = mOpStorage.operationTokensForOpState(OpState.PENDING); + + assertThat(mOpStorage.numOperations()).isEqualTo(0); + assertThat(typeWaitTokens).isEmpty(); + assertThat(statePendingTokens).isEmpty(); + } + + @Test + public void testRemoveOperation_removesPackageMappings() throws Exception { + mOpStorage.registerOperationForPackages(TOKEN_1, OpState.PENDING, MULTIPLE_PKGS_1, + mCallback, OpType.BACKUP); + mOpStorage.registerOperationForPackages(TOKEN_2, OpState.PENDING, MULTIPLE_PKGS_2, + mCallback, OpType.BACKUP); + + mOpStorage.removeOperation(TOKEN_2); + + Set tokensFoo = mOpStorage.operationTokensForPackage(PKG_FOO); + Set tokensBar = mOpStorage.operationTokensForPackage(PKG_BAR); + Set tokensBaz = mOpStorage.operationTokensForPackage(PKG_BAZ); + + assertThat(mOpStorage.numOperations()).isEqualTo(1); + assertThat(tokensFoo).isEqualTo(only(TOKEN_1)); + assertThat(tokensBar).isEqualTo(only(TOKEN_1)); + assertThat(tokensBaz).isEmpty(); + } + + @Test + public void testIsBackupOperationInProgress() throws Exception { + mOpStorage.registerOperation(TOKEN_1, OpState.ACKNOWLEDGED, mCallback, OpType.RESTORE_WAIT); + assertThat(mOpStorage.isBackupOperationInProgress()).isFalse(); + + mOpStorage.registerOperation(TOKEN_2, OpState.TIMEOUT, mCallback, OpType.BACKUP_WAIT); + assertThat(mOpStorage.isBackupOperationInProgress()).isFalse(); + + mOpStorage.registerOperation(TOKEN_3, OpState.PENDING, mCallback, OpType.BACKUP); + assertThat(mOpStorage.isBackupOperationInProgress()).isTrue(); + } + + @Test + public void testOnOperationComplete_pendingAdvancesState_invokesCallback() throws Exception { + mOpStorage.registerOperation(TOKEN_1, OpState.PENDING, mCallback, OpType.BACKUP_WAIT); + + mOpStorage.onOperationComplete(TOKEN_1, RESULT, callback -> { + mCallback.operationComplete(RESULT); + }); + + assertThat(mOpStorage.operationTokensForOpType(OpType.BACKUP_WAIT)) + .isEqualTo(only(TOKEN_1)); + assertThat(mOpStorage.operationTokensForOpState(OpState.PENDING)).isEmpty(); + assertThat(mOpStorage.operationTokensForOpState(OpState.ACKNOWLEDGED)).isNotEmpty(); + verify(mCallback).operationComplete(RESULT); + } + + private Set only(Integer val) { + return Sets.newHashSet(val); + } +}