Add a GetAllUnspentTxOuts API call to mobilecoind (#3663)

* Add a GetAllUnspentTxOuts API call to mobilecoind

Currently, there is no easy way to get all of the unspent tx outs
associated to a given monitor id. This is a very normal thing to
want to do if you are managing a wallet with many subaddresses,
because you have no way to subscribe to updates when payments come
in, so you would like to be able to poll for them efficiently.

The GetUnspentTxOutList API call exists, but it only allows you
to search for UTXOs one subaddress and token id at a time.

You get use the GetProcessedBlock API call, which allows you to
be more efficient, but it still creates a lot of extra work on
the caller side, because now you have to think about blocks and
keep track of a cursor as you advance through the chain.

There are other things about this API that are annoying --
the ProcessedTxOut tells you that you sent or received a TxOut
in a particular block, but not whether you still have it.
And many of the other calls, like GenerateTx, require you to
provide an UnspentTxOut for the InputList argument, but you can't
get an UnspentTxOut from a ProcessedTxOut, so you will have to
make another call to GetUnspentTxOutList if you want to do that.

---

By making a version of GetUnspentTxOutList that doesn't impose
any filtering, and just gives me all UTXOs that exist against
my monitor, it becomes much easier to determine if there was new
activity on my monitor.

I don't have to think about blocks anymore.

If I'm building an exchange, my loop can be like:

* Check if I got any new UTXOs
* If I got UTXOs on subaddress > 0, then that's a deposit
  * Report the deposit, using part of the UTXO as the inbound tx id.
  * Remote services can ignore it if it was a duplicate
* Sweep it to subaddress zero
  * Now I won't pick it up again and report it again if I get restarted
* Maybe make an optimization tx if subaddress 0 has many UTXOs.

This feels much simpler to me conceptually and much closer to how I
intuitively want an exchange integration to work, compared to the
GetProcessedBlock based approach.

---

Fortunately, it was extremely easy to implement this API and did
not require any changes to the mobilecoind database. This is because
of properties of how lmdb orders keys and how lmdb cursors work.

I was able to implement this in the time that I had to wait for
either of mobilecoind or full-service to finish downloading and
syncing the ledger, so that I can test other things.

If this PR is accepted, it will allow me to simplify a bunch of
other code that I had to write.

* fix clippies

mobilecoind/api/proto/mobilecoind_api.proto

@@ -21,6 +21,7 @@ service MobilecoindAPI {
     rpc GetMonitorList (google.protobuf.Empty) returns (GetMonitorListResponse) {}
     rpc GetMonitorStatus (GetMonitorStatusRequest) returns (GetMonitorStatusResponse) {}
     rpc GetUnspentTxOutList (GetUnspentTxOutListRequest) returns (GetUnspentTxOutListResponse) {}
+    rpc GetAllUnspentTxOut (GetAllUnspentTxOutRequest) returns (GetAllUnspentTxOutResponse) {}
 
     // Utilities
     rpc GenerateRootEntropy (google.protobuf.Empty) returns (GenerateRootEntropyResponse) {}
@@ -354,6 +355,14 @@ message GetUnspentTxOutListResponse {
     repeated UnspentTxOut output_list = 1;
 }
 
+// Get a list of all UnspentTxOuts for a given monitor, without any filtering
+message GetAllUnspentTxOutRequest {
+    bytes monitor_id = 1;
+}
+message GetAllUnspentTxOutResponse {
+    repeated UnspentTxOut output_list = 1;
+}
+
 //
 // Utilities
 //

mobilecoind/src/database.rs

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2022 The MobileCoin Foundation
+// Copyright (c) 2018-2023 The MobileCoin Foundation
 
 //! The mobilecoind database
 
@@ -219,6 +219,14 @@ impl Database {
         self.utxo_store.get_utxos(&db_txn, monitor_id, index)
     }
 
+    pub fn get_utxos_for_monitor(
+        &self,
+        monitor_id: &MonitorId,
+    ) -> Result<Vec<UnspentTxOut>, Error> {
+        let db_txn = self.env.begin_ro_txn()?;
+        self.utxo_store.get_utxos_for_monitor(&db_txn, monitor_id)
+    }
+
     pub fn update_attempted_spend(
         &self,
         utxo_ids: &[UtxoId],

mobilecoind/src/service.rs

@@ -343,7 +343,7 @@ impl<T: BlockchainConnection + UserTxConnection + 'static, FPR: FogPubkeyResolve
     ) -> Result<api::GetUnspentTxOutListResponse, RpcStatus> {
         // Get MonitorId from from the GRPC request.
         let monitor_id = MonitorId::try_from(&request.monitor_id)
-            .map_err(|err| rpc_internal_error("monitor_id.try_from.bytes", err, &self.logger))?;
+            .map_err(|err| rpc_invalid_arg_error("monitor_id.try_from.bytes", err, &self.logger))?;
 
         // Get UnspentTxOuts.
         let utxos = self
@@ -362,12 +362,37 @@ impl<T: BlockchainConnection + UserTxConnection + 'static, FPR: FogPubkeyResolve
         // Convert to protos.
         let proto_utxos: Vec<api::UnspentTxOut> = utxos.iter().map(|utxo| utxo.into()).collect();
 
-        // Returrn response.
+        // Return response.
         let mut response = api::GetUnspentTxOutListResponse::new();
         response.set_output_list(RepeatedField::from_vec(proto_utxos));
         Ok(response)
     }
 
+    fn get_all_unspent_tx_out_impl(
+        &mut self,
+        request: api::GetAllUnspentTxOutRequest,
+    ) -> Result<api::GetAllUnspentTxOutResponse, RpcStatus> {
+        // Get MonitorId from from the GRPC request.
+        let monitor_id = MonitorId::try_from(&request.monitor_id)
+            .map_err(|err| rpc_invalid_arg_error("monitor_id.try_from.bytes", err, &self.logger))?;
+
+        // Get UnspentTxOuts.
+        let utxos = self
+            .mobilecoind_db
+            .get_utxos_for_monitor(&monitor_id)
+            .map_err(|err| {
+                rpc_internal_error("mobilecoind_db.get_utxos_for_monitor", err, &self.logger)
+            })?;
+
+        // Convert to protos.
+        let proto_utxos: Vec<api::UnspentTxOut> = utxos.iter().map(|utxo| utxo.into()).collect();
+
+        // Return response.
+        let mut response = api::GetAllUnspentTxOutResponse::new();
+        response.set_output_list(RepeatedField::from_vec(proto_utxos));
+        Ok(response)
+    }
+
     fn generate_root_entropy_impl(
         &mut self,
         _request: api::Empty,
@@ -2277,6 +2302,7 @@ build_api! {
     get_monitor_list Empty GetMonitorListResponse get_monitor_list_impl,
     get_monitor_status GetMonitorStatusRequest GetMonitorStatusResponse get_monitor_status_impl,
     get_unspent_tx_out_list GetUnspentTxOutListRequest GetUnspentTxOutListResponse get_unspent_tx_out_list_impl,
+    get_all_unspent_tx_out GetAllUnspentTxOutRequest GetAllUnspentTxOutResponse get_all_unspent_tx_out_impl,
 
     // Utilities
     generate_root_entropy Empty GenerateRootEntropyResponse generate_root_entropy_impl,
@@ -2731,6 +2757,140 @@ mod test {
         );
     }
 
+    #[test_with_logger]
+    fn test_get_all_unspent_tx_out_impl(logger: Logger) {
+        let mut rng: StdRng = SeedableRng::from_seed([23u8; 32]);
+
+        let account_key = AccountKey::random(&mut rng);
+        let data = MonitorData::new(
+            account_key.clone(),
+            0,  // first_subaddress
+            20, // num_subaddresses
+            0,  // first_block
+            "", // name
+        )
+        .unwrap();
+
+        // 1 known recipient, 3 random recipients and no monitors.
+        let (mut ledger_db, mobilecoind_db, client, _server, _server_conn_manager) =
+            get_testing_environment(
+                BLOCK_VERSION,
+                3,
+                &[account_key.default_subaddress()],
+                &[],
+                logger.clone(),
+                &mut rng,
+            );
+
+        // Add a block with a non-MOB token ID.
+        add_block_to_ledger(
+            &mut ledger_db,
+            BLOCK_VERSION,
+            &vec![
+                AccountKey::random(&mut rng).default_subaddress(),
+                AccountKey::random(&mut rng).default_subaddress(),
+                AccountKey::random(&mut rng).default_subaddress(),
+                account_key.default_subaddress(),
+            ],
+            Amount::new(1000, 2.into()),
+            &[KeyImage::from(101)],
+            &mut rng,
+        )
+        .unwrap();
+
+        // Add a block with a non-MOB token ID, to an off subaddress
+        add_block_to_ledger(
+            &mut ledger_db,
+            BLOCK_VERSION,
+            &vec![
+                AccountKey::random(&mut rng).default_subaddress(),
+                AccountKey::random(&mut rng).default_subaddress(),
+                AccountKey::random(&mut rng).default_subaddress(),
+                account_key.subaddress(1),
+            ],
+            Amount::new(1000, 2.into()),
+            &[KeyImage::from(102)],
+            &mut rng,
+        )
+        .unwrap();
+
+        // Insert into database.
+        let id = mobilecoind_db.add_monitor(&data).unwrap();
+
+        // Allow the new monitor to process the ledger.
+        wait_for_monitors(&mobilecoind_db, &ledger_db, &logger);
+
+        // Query with the known id
+        let mut request = api::GetAllUnspentTxOutRequest::new();
+        request.set_monitor_id(id.to_vec());
+
+        let response = client
+            .get_all_unspent_tx_out(&request)
+            .expect("failed to get all unspent tx out");
+
+        let utxos: Vec<UnspentTxOut> = response
+            .output_list
+            .iter()
+            .map(|proto_utxo| {
+                UnspentTxOut::try_from(proto_utxo).expect("failed converting proto utxo")
+            })
+            .collect();
+
+        // Verify the data we got matches what we expected. This assumes knowledge about
+        // how the test ledger is constructed by the test utils.
+        let num_blocks = ledger_db.num_blocks().unwrap();
+        let account_tx_outs: Vec<TxOut> = (0..num_blocks)
+            .map(|idx| {
+                let block_contents = ledger_db.get_block_contents(idx).unwrap();
+                // We grab the 4th tx out in each block since the test ledger had 3 random
+                // recipients, followed by our known recipient.
+                // See the call to `get_testing_environment` at the beginning of the test.
+                block_contents.outputs[3].clone()
+            })
+            .collect();
+
+        let expected_utxos: Vec<UnspentTxOut> = account_tx_outs
+            .iter()
+            .enumerate()
+            .map(|(idx, tx_out)| {
+                let (amount, _) = tx_out
+                    .view_key_match(account_key.view_private_key())
+                    .unwrap();
+
+                // Get the expected subaddress index, based on block index. Everything is on 0
+                // except in the last block, where we used subaddrss 1.
+                let subaddress_index = if idx as u64 == num_blocks - 1 { 1 } else { 0 };
+
+                // Calculate the key image for this tx out.
+                let tx_public_key = RistrettoPublic::try_from(&tx_out.public_key).unwrap();
+                let onetime_private_key = recover_onetime_private_key(
+                    &tx_public_key,
+                    account_key.view_private_key(),
+                    &account_key.subaddress_spend_private(subaddress_index),
+                );
+                let key_image = KeyImage::from(&onetime_private_key);
+
+                // Craft the expected UnspentTxOut
+                UnspentTxOut {
+                    tx_out: tx_out.clone(),
+                    subaddress_index,
+                    key_image,
+                    value: amount.value,
+                    token_id: *amount.token_id,
+                    attempted_spend_height: 0,
+                    attempted_spend_tombstone: 0,
+                }
+            })
+            .collect();
+
+        // Compare - we should have one utxo in each block.
+        assert_eq!(utxos.len(), num_blocks as usize);
+        assert_eq!(
+            HashSet::from_iter(utxos.iter()),
+            HashSet::from_iter(expected_utxos.iter())
+        );
+    }
+
     #[test_with_logger]
     fn test_generate_root_entropy_impl(logger: Logger) {
         let mut rng: StdRng = SeedableRng::from_seed([23u8; 32]);

mobilecoind/src/utxo_store.rs

@@ -332,6 +332,19 @@ impl UtxoStore {
             .collect()
     }
 
+    /// Get all UnspentTxOuts for a given monitor
+    pub fn get_utxos_for_monitor(
+        &self,
+        db_txn: &impl Transaction,
+        monitor_id: &MonitorId,
+    ) -> Result<Vec<UnspentTxOut>, Error> {
+        let utxo_ids = self.get_utxo_ids_for_monitor_id(db_txn, monitor_id)?;
+        utxo_ids
+            .iter()
+            .map(|utxo_id| self.get_utxo_by_id(db_txn, utxo_id))
+            .collect()
+    }
+
     /// Get subaddress id by utxo id.
     pub fn get_subaddress_id_by_utxo_id(
         &self,
@@ -407,6 +420,49 @@ impl UtxoStore {
             .collect::<Result<Vec<_>, Error>>()
     }
 
+    /// Get all UtxoIds associated with a given monitor id
+    ///
+    /// This uses the fact that lmdb sorts keys lexicographically by bytes,
+    /// with low order bytes first, and the SubaddressId structure happens to
+    /// map the monitor id bytes first.
+    /// So if we open a cursor at subaddress (monitor_id, 0),
+    /// and walk until the subaddress changes, then we saw all records with that
+    /// monitor id.
+    fn get_utxo_ids_for_monitor_id(
+        &self,
+        db_txn: &impl Transaction,
+        monitor_id: &MonitorId,
+    ) -> Result<Vec<UtxoId>, Error> {
+        let mut cursor = db_txn.open_ro_cursor(self.subaddress_id_to_utxo_id)?;
+
+        let zero_subaddress_id = SubaddressId::new(monitor_id, 0);
+        let zero_subaddress_id_bytes = zero_subaddress_id.to_bytes();
+
+        let mut utxo_ids = Vec::<UtxoId>::default();
+        for iter in cursor.iter_dup_from(zero_subaddress_id.to_vec()) {
+            // The second iter is because, per docs, iter_dup_from returns an iterator over
+            // the duplicates
+            for result in iter {
+                match result {
+                    Ok((subaddress_id_bytes, utxo_id_bytes)) => {
+                        if subaddress_id_bytes[0..32] == zero_subaddress_id_bytes[0..32] {
+                            utxo_ids.push(UtxoId::try_from(utxo_id_bytes)?);
+                        } else {
+                            // We've moved on in the lexicographic ordering to a new monitor id,
+                            // so we're done here.
+                            break;
+                        }
+                    }
+                    Err(err) => {
+                        return Err(Error::from(err));
+                    }
+                }
+            }
+        }
+
+        Ok(utxo_ids)
+    }
+
     /// Get a single UnspentTxOut by its id.
     fn get_utxo_by_id(
         &self,