Update the data fetching endpoints on the ledger canister

Author: yanliu38

cli/src/argparse.rs

@@ -162,14 +162,23 @@ pub fn parse_args() -> clap::ArgMatches {
                         .num_args(1),
                 )
                 .arg(
-                    Arg::new("push-authorize")
-                        .long("push_authorize")
+                    Arg::new("data-fetch")
+                        .long("data-fetch")
+                        .visible_aliases(&["fetch", "pull"])
+                        .action(ArgAction::SetTrue)
+                        .help("Sync data from the ledger"),
+                )
+                .arg(
+                    Arg::new("data-push-authorize")
+                        .long("data-push-authorize")
+                        .visible_aliases(&["push-authorize", "push-auth"])
                         .help("Authorize push to the Decent Cloud Ledger")
                         .action(ArgAction::SetTrue),
                     )
                 .arg(
-                    Arg::new("push")
-                        .long("push")
+                    Arg::new("data-push")
+                        .long("data-push")
+                        .visible_aliases(&["push"])
                         .help("Push the ledger entries to the Decent Cloud Ledger")
                         .action(ArgAction::SetTrue)
                     )
@@ -192,12 +201,6 @@ pub fn parse_args() -> clap::ArgMatches {
                         .default_value("127.0.0.1")
                         .help("Which IC network to use"),
                 )
-                .arg(
-                    Arg::new("fetch")
-                        .long("fetch")
-                        .action(ArgAction::SetTrue)
-                        .help("Sync from the ledger"),
-                ),
         )
         .get_matches()
 }

cli/src/ledger_canister_client.rs

@@ -63,13 +63,15 @@ impl LedgerCanister {
             "node_provider_register".to_string(),
             "node_provider_check_in".to_string(),
             "node_provider_update_profile".to_string(),
+            "data_push".to_string(),
+            "data_push_auth".to_string(),
         ]
     }
 
     pub fn list_functions_queries(&self) -> Vec<String> {
         vec![
             "get_np_check_in_nonce".to_string(),
-            "fetch".to_string(),
+            "data_fetch".to_string(),
             "metadata".to_string(),
             "get_logs_debug".to_string(),
             "get_logs_info".to_string(),
@@ -117,9 +119,13 @@ impl LedgerCanister {
         Decode!(response.as_slice(), Vec<u8>).expect("Failed to decode response")
     }
 
-    pub async fn fetch(&self, cursor: Option<String>) -> Result<(String, Vec<u8>), String> {
-        let args = Encode!(&(cursor)).map_err(|e| e.to_string())?;
-        let response = self.call_query("fetch", &args).await?;
+    pub async fn data_fetch(
+        &self,
+        cursor: Option<String>,
+        bytes_before: Option<Vec<u8>>,
+    ) -> Result<(String, Vec<u8>), String> {
+        let args = Encode!(&(cursor, bytes_before)).map_err(|e| e.to_string())?;
+        let response = self.call_query("data_fetch", &args).await?;
         Decode!(response.as_slice(), Result<(String, Vec<u8>), String>)
             .map_err(|e| e.to_string())?
     }

cli/src/main.rs

@@ -10,7 +10,7 @@ use chrono::DateTime;
 use dcc_common::{
     account_balance_get_as_string, amount_as_string, cursor_from_data, refresh_caches_from_ledger,
     zlib_compress, Account, CursorDirection, DccIdentity, FundsTransfer, LedgerCursor,
-    NodeProviderProfile, UpdateProfilePayload, LABEL_DC_TOKEN_TRANSFER,
+    NodeProviderProfile, UpdateProfilePayload, DATA_PULL_BYTES_BEFORE_LEN, LABEL_DC_TOKEN_TRANSFER,
 };
 use decent_cloud::ledger_canister_client::LedgerCanister;
 use decent_cloud_canister::DC_TOKEN_TRANSFER_FEE_E9S;
@@ -308,8 +308,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     .expect("Could not get home directory")
                     .join(".dcc/ledger/main.bin"),
             };
-            let push_auth = arg_matches.get_flag("push-authorize");
-            let push = arg_matches.get_flag("push");
+            let push_auth = arg_matches.get_flag("data-push-authorize");
+            let push = arg_matches.get_flag("data-push");
             if push_auth || push {
                 let local_identity = match local_identity {
                     Some(ident) => ident.to_string(),
@@ -322,7 +322,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     let ic_auth = dcc_to_ic_auth(&dcc_ident);
                     let canister = ledger_canister(ic_auth).await?;
                     let args = Encode!(&()).map_err(|e| e.to_string())?;
-                    let result = canister.call_update("push_auth", &args).await?;
+                    let result = canister.call_update("data_push_auth", &args).await?;
                     let response =
                         Decode!(&result, Result<String, String>).map_err(|e| e.to_string())??;
 
@@ -333,7 +333,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 let ic_auth = dcc_to_ic_auth(&dcc_ident);
                 let canister = ledger_canister(ic_auth).await?;
 
-                return push_ledger_data(&canister, local_ledger_path).await;
+                return ledger_data_push(&canister, local_ledger_path).await;
             }
 
             let canister_function = match arg_matches.get_one::<String>("canister_function") {
@@ -371,9 +371,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     let canister = ledger_canister(None).await?;
                     println!("{}", canister.init_ledger_map().await?);
                 }
-                "fetch" => {
+                "data_fetch" | "fetch" => {
                     let canister = ledger_canister(None).await?;
-                    fetch_ledger_data(&canister, local_ledger_path).await?;
+                    ledger_data_fetch(&canister, local_ledger_path).await?;
                     println!("Done fetching data from the Ledger canister");
                 }
                 "metadata" => {
@@ -488,7 +488,7 @@ fn dcc_to_ic_auth(dcc_identity: &DccIdentity) -> Option<BasicIdentity> {
         })
 }
 
-async fn fetch_ledger_data(
+async fn ledger_data_fetch(
     ledger_canister: &LedgerCanister,
     local_ledger_path: PathBuf,
 ) -> Result<(), Box<dyn std::error::Error>> {
@@ -510,12 +510,23 @@ async fn fetch_ledger_data(
         )
     };
 
+    let bytes_before = if cursor_local.position > DATA_PULL_BYTES_BEFORE_LEN as u64 {
+        let mut buf = vec![0u8; DATA_PULL_BYTES_BEFORE_LEN as usize];
+        persistent_storage_read(
+            cursor_local.position - DATA_PULL_BYTES_BEFORE_LEN as u64,
+            &mut buf,
+        )?;
+        Some(buf)
+    } else {
+        None
+    };
+
     println!(
         "Fetching data from the Ledger canister, with local cursor: {}",
         cursor_local
     );
     let (cursor_remote, data) = ledger_canister
-        .fetch(Some(cursor_local.to_request_string()))
+        .data_fetch(Some(cursor_local.to_request_string()), bytes_before)
         .await?;
     let cursor_remote = LedgerCursor::new_from_string(cursor_remote);
     let offset_remote = cursor_remote.position;
@@ -571,7 +582,7 @@ async fn get_ledger_metadata(ledger_canister: &LedgerCanister) -> HashMap<String
         .collect()
 }
 
-pub async fn push_ledger_data(
+pub async fn ledger_data_push(
     ledger_canister: &LedgerCanister,
     local_ledger_path: PathBuf,
 ) -> Result<(), Box<dyn std::error::Error>> {
@@ -619,7 +630,7 @@ pub async fn push_ledger_data(
             buf_size, position,
         );
         let args = Encode!(&cursor_push.to_urlenc_string(), &buf).map_err(|e| e.to_string())?;
-        let result = ledger_canister.call_update("push", &args).await?;
+        let result = ledger_canister.call_update("data_push", &args).await?;
         let result = Decode!(&result, Result<String, String>).map_err(|e| e.to_string())??;
         println!("Response from pushing at position {}: {}", position, result);
     }

common/src/lib.rs

@@ -79,6 +79,7 @@ pub const MEMO_BYTES_MAX: usize = 32;
 /// Reduction of reputations for all accounts, based on time (per block), in parts per million
 pub const REPUTATION_AGING_PPM: u64 = 1_000;
 pub const REWARD_HALVING_AFTER_BLOCKS: u64 = 210_000; // halve the rewards every 210000 reward distributions
+pub const DATA_PULL_BYTES_BEFORE_LEN: u16 = 16; // How many bytes before the pulled data should be compared as a quick sanity check
 
 // Default first block's time
 // Calculated with:

ic-canister/decent_cloud.did

@@ -390,9 +390,9 @@ service : {
     get_data_certificate : () -> (DataCertificate) query;
 
     // Synchronization (pushing and pulling) of the ledger data
-    fetch: (cursor: opt text) -> (ResultData) query;
-    push_auth: () -> (ResultString);
-    push: (cursor: text, data: vec nat8) -> (ResultString);
+    data_fetch: (cursor: opt text, bytes_before: opt vec nat8) -> (ResultData) query;
+    data_push_auth: () -> (ResultString);
+    data_push: (cursor: text, data: vec nat8) -> (ResultString);
     metadata: () -> (vec record { text; MetadataValue; }) query;
 
     // Debugging and troubleshooting

ic-canister/src/canister_backend/generic.rs

@@ -7,8 +7,9 @@ use dcc_common::{
     np_registration_fee_e9s, refresh_caches_from_ledger, reputation_get, reward_e9s_per_block,
     reward_e9s_per_block_recalculate, rewards_applied_np_count, rewards_distribute,
     rewards_pending_e9s, set_test_config, FundsTransfer, LedgerCursor, BLOCK_INTERVAL_SECS,
-    CACHE_TXS_NUM_COMMITTED, LABEL_DC_TOKEN_TRANSFER, LABEL_NP_CHECK_IN, LABEL_NP_PROFILE,
-    LABEL_NP_REGISTER, LABEL_REWARD_DISTRIBUTION, LABEL_USER_REGISTER,
+    CACHE_TXS_NUM_COMMITTED, DATA_PULL_BYTES_BEFORE_LEN, LABEL_DC_TOKEN_TRANSFER,
+    LABEL_NP_CHECK_IN, LABEL_NP_PROFILE, LABEL_NP_REGISTER, LABEL_REWARD_DISTRIBUTION,
+    LABEL_USER_REGISTER,
 };
 use ic_cdk::println;
 use icrc_ledger_types::icrc::generic_metadata_value::MetadataValue;
@@ -241,7 +242,10 @@ pub(crate) fn _node_provider_list_checked_in() -> Result<Vec<String>, String> {
     })
 }
 
-pub(crate) fn _fetch(cursor: Option<String>) -> Result<(String, Vec<u8>), String> {
+pub(crate) fn _data_fetch(
+    cursor: Option<String>,
+    bytes_before: Option<Vec<u8>>,
+) -> Result<(String, Vec<u8>), String> {
     LEDGER_MAP.with(|ledger| {
         let req_cursor = LedgerCursor::new_from_string(cursor.unwrap_or_default());
         let req_position_start = req_cursor.position;
@@ -258,6 +262,23 @@ pub(crate) fn _fetch(cursor: Option<String>) -> Result<(String, Vec<u8>), String
         if local_cursor.response_bytes == 0 {
             return Ok((local_cursor.to_urlenc_string(), vec![]));
         }
+        if let Some(bytes_before) = bytes_before {
+            if local_cursor.position > DATA_PULL_BYTES_BEFORE_LEN as u64 {
+                let mut buf_bytes_before = vec![0u8; DATA_PULL_BYTES_BEFORE_LEN as usize];
+                persistent_storage_read(
+                    local_cursor.position - DATA_PULL_BYTES_BEFORE_LEN as u64,
+                    &mut buf_bytes_before,
+                )
+                .map_err(|e| e.to_string())?;
+                if bytes_before != buf_bytes_before {
+                    return Err(format!(
+                        "{} bytes before position {} does not match",
+                        DATA_PULL_BYTES_BEFORE_LEN, local_cursor.position
+                    ));
+                }
+            }
+        }
+
         let mut buf = vec![0u8; local_cursor.response_bytes as usize];
         persistent_storage_read(local_cursor.position, &mut buf).map_err(|e| e.to_string())?;
         info!(
@@ -268,7 +289,7 @@ pub(crate) fn _fetch(cursor: Option<String>) -> Result<(String, Vec<u8>), String
     })
 }
 
-pub(crate) fn _push_auth() -> Result<String, String> {
+pub(crate) fn _data_push_auth() -> Result<String, String> {
     // If LEDGER_MAP is currently empty and there is no authorized pusher,
     // set the authorized pusher to the caller.
     LEDGER_MAP.with(|ledger| {
@@ -299,7 +320,7 @@ pub(crate) fn _push_auth() -> Result<String, String> {
     })
 }
 
-pub(crate) fn _push(cursor: String, data: Vec<u8>) -> Result<String, String> {
+pub(crate) fn _data_push(cursor: String, data: Vec<u8>) -> Result<String, String> {
     let caller = ic_cdk::api::caller();
     let authorized_pusher = AUTHORIZED_PUSHER.with(|authorized_pusher| *authorized_pusher.borrow());
 

ic-canister/src/canister_endpoints/generic.rs

@@ -87,18 +87,21 @@ fn node_provider_list_checked_in() -> Result<Vec<String>, String> {
 }
 
 #[ic_cdk::query]
-fn fetch(cursor: Option<String>) -> Result<(String, Vec<u8>), String> {
-    _fetch(cursor)
+fn data_fetch(
+    cursor: Option<String>,
+    bytes_before: Option<Vec<u8>>,
+) -> Result<(String, Vec<u8>), String> {
+    _data_fetch(cursor, bytes_before)
 }
 
 #[ic_cdk::update]
-fn push_auth() -> Result<String, String> {
-    _push_auth()
+fn data_push_auth() -> Result<String, String> {
+    _data_push_auth()
 }
 
 #[ic_cdk::update]
-fn push(cursor: String, data: Vec<u8>) -> Result<String, String> {
-    _push(cursor, data)
+fn data_push(cursor: String, data: Vec<u8>) -> Result<String, String> {
+    _data_push(cursor, data)
 }
 
 #[ic_cdk::query]