feat(leaser): migrate leases in batches

contracts/leaser/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "leaser"
-version = "0.3.0"
+version = "0.4.0"
 edition.workspace = true
 authors.workspace = true
 license.workspace = true
@@ -33,9 +33,10 @@ thiserror = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 
 [dev-dependencies]
-lease = { workspace = true, features = ["testing"]}
+lease = { workspace = true, features = ["testing"] }
 finance = { workspace = true, features = ["testing"] }
 currency = { workspace = true, features = ["testing"] }
+platform = { workspace = true, features = ["testing"] }
 schema = { workspace = true }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]

contracts/leaser/src/contract.rs

@@ -66,10 +66,18 @@ pub fn execute(
             currency,
             max_ltd,
         ),
-        ExecuteMsg::MigrateLeases { new_code_id } => {
-            SingleUserAccess::check_owner_access(deps.storage, &info.sender)
-                .and_then(move |()| leaser::try_migrate_leases(deps.storage, new_code_id.u64()))
-        }
+        ExecuteMsg::MigrateLeases {
+            new_code_id,
+            max_leases,
+        } => SingleUserAccess::check_owner_access(deps.storage, &info.sender).and_then(move |()| {
+            leaser::try_migrate_leases(deps.storage, new_code_id.u64(), max_leases)
+        }),
+        ExecuteMsg::MigrateLeasesCont {
+            key: start_past,
+            max_leases,
+        } => SingleUserAccess::check_owner_access(deps.storage, &info.sender).and_then(move |()| {
+            leaser::try_migrate_leases_cont(deps.storage, start_past, max_leases)
+        }),
     }
     .map(response::response_only_messages)
 }

contracts/leaser/src/error.rs

@@ -1,3 +1,5 @@
+use std::num::TryFromIntError;
+
 use thiserror::Error;
 
 use sdk::cosmwasm_std::StdError;
@@ -7,6 +9,9 @@ pub enum ContractError {
     #[error("[Leaser] [Std] {0}")]
     Std(#[from] StdError),
 
+    #[error("[Leaser] integer conversion {0}")]
+    Conversion(#[from] TryFromIntError),
+
     #[error("[Leaser] {0}")]
     Finance(#[from] finance::error::Error),
 

contracts/leaser/src/leaser.rs

@@ -5,15 +5,15 @@ use finance::{currency::SymbolOwned, liability::Liability, percent::Percent};
 use lease::api::{ConnectionParams, DownpaymentCoin, InterestPaymentSpec};
 use lpp::{msg::ExecuteMsg, stub::LppRef};
 use oracle::stub::OracleRef;
-use platform::batch::Batch;
+use platform::batch::{Batch, Emit, Emitter};
 use platform::message::Response as MessageResponse;
 use sdk::cosmwasm_std::{Addr, Deps, StdResult, Storage};
 
 use crate::{
     cmd::Quote,
     error::ContractError,
     migrate::{self},
-    msg::{ConfigResponse, QuoteResponse},
+    msg::{ConfigResponse, NbInstances, QuoteResponse},
     result::ContractResult,
     state::{config::Config, leases::Leases},
 };
@@ -92,24 +92,58 @@ pub(super) fn try_configure(
 pub(super) fn try_migrate_leases(
     storage: &mut dyn Storage,
     new_code_id: u64,
+    max_leases: NbInstances,
 ) -> ContractResult<MessageResponse> {
     Config::update_lease_code(storage, new_code_id)?;
 
-    migrate::migrate_leases(Leases::iter(storage), new_code_id)
-        .and_then(|batch| update_lpp(storage, new_code_id, batch))
+    let leases = Leases::iter(storage, None);
+    migrate::migrate_leases(leases, new_code_id, max_leases)
+        .and_then(|result| result.try_add_msgs(|msgs| update_lpp_impl(storage, new_code_id, msgs)))
+        .map(|result| {
+            MessageResponse::messages_with_events(result.msgs, emit_status(result.last_instance))
+        })
 }
 
-pub(super) fn update_lpp(
+pub(super) fn try_migrate_leases_cont(
     storage: &mut dyn Storage,
+    start_past: Addr,
+    max_leases: NbInstances,
+) -> ContractResult<MessageResponse> {
+    let lease_code_id = Config::load(storage)?.lease_code_id;
+
+    let leases = Leases::iter(storage, Some(start_past));
+    migrate::migrate_leases(leases, lease_code_id, max_leases).map(|result| {
+        MessageResponse::messages_with_events(result.msgs, emit_status(result.last_instance))
+    })
+}
+
+pub(super) fn update_lpp(
+    storage: &dyn Storage,
     new_code_id: u64,
     mut batch: Batch,
-) -> ContractResult<MessageResponse> {
+) -> ContractResult<Batch> {
+    update_lpp_impl(storage, new_code_id, &mut batch).map(|()| batch)
+}
+
+fn update_lpp_impl(
+    storage: &dyn Storage,
+    new_code_id: u64,
+    batch: &mut Batch,
+) -> ContractResult<()> {
     let lpp = Config::load(storage)?.lpp_addr;
     let lpp_update_code = ExecuteMsg::NewLeaseCode {
         lease_code_id: new_code_id.into(),
     };
     batch
         .schedule_execute_wasm_no_reply::<_, Nls>(&lpp, lpp_update_code, None)
-        .map(|()| batch.into())
         .map_err(Into::into)
 }
+
+fn emit_status(last_instance: Option<Addr>) -> Emitter {
+    let emitter = Emitter::of_type("migrate-leases");
+    if let Some(last) = last_instance {
+        emitter.emit("contunuation-key", last)
+    } else {
+        emitter.emit("status", "done")
+    }
+}

contracts/leaser/src/migrate.rs

@@ -2,66 +2,97 @@ use lease::api::MigrateMsg;
 use platform::batch::Batch;
 use sdk::cosmwasm_std::Addr;
 
-use crate::{error::ContractError, result::ContractResult};
+use crate::{error::ContractError, msg::NbInstances, result::ContractResult};
 
-pub fn migrate_leases<I>(leases: I, lease_code_id: u64) -> ContractResult<Batch>
+#[derive(Default)]
+#[cfg_attr(test, derive(Debug, Eq, PartialEq))]
+pub struct MigrationResult {
+    pub msgs: Batch,
+    /// None if the number of processed instances is less than the `max_leases`
+    pub last_instance: Option<Addr>,
+}
+
+pub fn migrate_leases<I>(
+    leases: I,
+    lease_code_id: u64,
+    max_leases: NbInstances,
+) -> ContractResult<MigrationResult>
 where
     I: Iterator<Item = ContractResult<Addr>>,
 {
-    let no_msgs = MigrateBatch::new(lease_code_id);
-    let migrated_msgs = leases.fold(no_msgs, MigrateBatch::migrate_lease);
+    let no_msgs = MigrateBatch::new(lease_code_id, max_leases);
+    let migrated_msgs = leases
+        .take(max_leases.try_into()?)
+        .fold(no_msgs, MigrateBatch::migrate_lease);
     migrated_msgs.try_into()
 }
 
+impl MigrationResult {
+    pub fn try_add_msgs<F>(mut self, add_fn: F) -> ContractResult<Self>
+    where
+        F: FnOnce(&mut Batch) -> ContractResult<()>,
+    {
+        add_fn(&mut self.msgs).map(|()| self)
+    }
+}
 struct MigrateBatch {
     new_code_id: u64,
-    may_batch: ContractResult<Batch>,
+    leases_left: NbInstances,
+    may_result: ContractResult<MigrationResult>,
 }
 impl MigrateBatch {
-    fn new(new_code_id: u64) -> Self {
+    fn new(new_code_id: u64, max_leases: NbInstances) -> Self {
         Self {
             new_code_id,
-            may_batch: Ok(Batch::default()),
+            leases_left: max_leases,
+            may_result: Ok(Default::default()),
         }
     }
 
     fn migrate_lease(mut self, lease_contract: ContractResult<Addr>) -> Self {
-        let op = |mut batch: Batch| {
+        let op = |result: MigrationResult| {
             lease_contract.and_then(|lease| {
-                batch
-                    .schedule_migrate_wasm_no_reply(&lease, MigrateMsg {}, self.new_code_id)
-                    .map(|_| batch)
-                    .map_err(Into::into)
+                self.leases_left -= 1;
+                result
+                    .try_add_msgs(|msgs| {
+                        msgs.schedule_migrate_wasm_no_reply(&lease, MigrateMsg {}, self.new_code_id)
+                            .map_err(Into::into)
+                    })
+                    .map(|mut res| {
+                        if self.leases_left == NbInstances::MIN {
+                            res.last_instance = Some(lease);
+                        }
+                        res
+                    })
             })
         };
 
-        self.may_batch = self.may_batch.and_then(op);
+        self.may_result = self.may_result.and_then(op);
         self
     }
 }
 
-impl TryFrom<MigrateBatch> for Batch {
+impl TryFrom<MigrateBatch> for MigrationResult {
     type Error = ContractError;
     fn try_from(this: MigrateBatch) -> Result<Self, Self::Error> {
-        this.may_batch
+        this.may_result
     }
 }
 
 #[cfg(test)]
 mod test {
     use lease::api::MigrateMsg;
-    use platform::batch::Batch;
     use sdk::cosmwasm_std::Addr;
 
-    use crate::ContractError;
+    use crate::{migrate::MigrationResult, ContractError};
 
     #[test]
     fn no_leases() {
         let new_code = 242;
         let no_leases = vec![];
         assert_eq!(
-            Ok(Batch::default()),
-            super::migrate_leases(no_leases.into_iter().map(Ok), new_code)
+            Ok(MigrationResult::default()),
+            super::migrate_leases(no_leases.into_iter().map(Ok), 2, new_code)
         );
     }
 
@@ -77,7 +108,7 @@ mod test {
         ];
         assert_eq!(
             Err(ContractError::ParseError { err: err.into() }),
-            super::migrate_leases(no_leases.into_iter(), new_code)
+            super::migrate_leases(no_leases.into_iter(), 5, new_code)
         );
     }
 
@@ -88,15 +119,66 @@ mod test {
         let addr2 = Addr::unchecked("22222");
         let leases = vec![addr1.clone(), addr2.clone()];
 
-        let mut exp = Batch::default();
-        exp.schedule_migrate_wasm_no_reply(&addr1, MigrateMsg {}, new_code)
-            .unwrap();
-        exp.schedule_migrate_wasm_no_reply(&addr2, MigrateMsg {}, new_code)
-            .unwrap();
+        let exp = add_expected(MigrationResult::default(), &addr1, new_code);
+        let mut exp = add_expected(exp, &addr2, new_code);
+        exp.last_instance = Some(addr2);
 
         assert_eq!(
             Ok(exp),
-            super::migrate_leases(leases.into_iter().map(Ok), new_code)
+            super::migrate_leases(leases.into_iter().map(Ok), new_code, 2)
         );
     }
+
+    #[test]
+    fn paging() {
+        let new_code = 242;
+        let addr1 = Addr::unchecked("11111");
+        let addr2 = Addr::unchecked("22222");
+        let addr3 = Addr::unchecked("333333333");
+        let addr4 = Addr::unchecked("4");
+        let addr5 = Addr::unchecked("5555");
+        let addr6 = Addr::unchecked("6");
+        let addr7 = Addr::unchecked("777");
+        let leases: Vec<Addr> = [&addr1, &addr2, &addr3, &addr4, &addr5, &addr6, &addr7]
+            .map(Clone::clone)
+            .into();
+
+        {
+            let exp = add_expected(MigrationResult::default(), &addr1, new_code);
+            let mut exp = add_expected(exp, &addr2, new_code);
+            exp.last_instance = Some(addr2);
+            assert_eq!(
+                Ok(exp),
+                super::migrate_leases(leases.clone().into_iter().map(Ok), new_code, 2)
+            );
+        }
+
+        {
+            let exp = add_expected(MigrationResult::default(), &addr3, new_code);
+            let exp = add_expected(exp, &addr4, new_code);
+            let mut exp = add_expected(exp, &addr5, new_code);
+            exp.last_instance = Some(addr5);
+            assert_eq!(
+                Ok(exp),
+                super::migrate_leases(leases.clone().into_iter().skip(2).map(Ok), new_code, 3)
+            );
+        }
+
+        {
+            let exp = add_expected(MigrationResult::default(), &addr6, new_code);
+            let mut exp = add_expected(exp, &addr7, new_code);
+            exp.last_instance = None;
+            assert_eq!(
+                Ok(exp),
+                super::migrate_leases(leases.into_iter().take(2 + 3).map(Ok), new_code, 15)
+            );
+        }
+    }
+
+    fn add_expected(mut exp: MigrationResult, lease_addr: &Addr, new_code: u64) -> MigrationResult {
+        exp.msgs
+            .schedule_migrate_wasm_no_reply(lease_addr, MigrateMsg {}, new_code)
+            .unwrap();
+        exp
+    }
 }

contracts/leaser/src/msg.rs

@@ -25,6 +25,8 @@ pub struct InstantiateMsg {
 #[derive(Serialize, Deserialize)]
 pub struct MigrateMsg {}
 
+pub type NbInstances = u32;
+
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum ExecuteMsg {
@@ -33,9 +35,25 @@ pub enum ExecuteMsg {
         #[serde(default)]
         max_ltd: Option<Percent>,
     },
+    /// Start a Lease migration
+    ///
+    /// The consumed gas is a limitaton factor for the maximum lease instances that
+    /// can be processed in a transaction. For that reason, the process does the migration
+    /// in batches. A new batch is started with this transaction. It processes the specified
+    /// maximum number of leases and emits a continuation key as an event
+    /// 'wasm-migrate-leases.contunuation-key=<key>'. That key should be provided
+    /// with the next `MigrateLeasesCont` message. It in turn emits
+    /// a continuation key with the same event and the procedure continues until
+    /// no key is provided and a 'wasm-migrate-leases.status=done'.
     MigrateLeases {
         new_code_id: Uint64,
+        max_leases: NbInstances,
     },
+    /// Continue a Lease migration
+    ///
+    /// It migrates the next batch of up to `max_leases` number of Lease instances
+    /// and emits the status as specified in `MigrateLeases`.
+    MigrateLeasesCont { key: Addr, max_leases: NbInstances },
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]