DEV Community: Alexander

Tutorial: Connecting A Token With Multiple Ledgers Using ic-event-hub

Alexander — Mon, 28 Mar 2022 15:40:47 +0000

This tutorial is dedicated to Rust smart-contract (canister) development on Internet Computer (Dfinity) platform. Completing it, you’ll know how to use ic-event-hub library APIs in order to perform efficient cross-canister integrations.

Before digging into this tutorial it is recommended to learn the basics of smart-contract development for the Internet Computer. Here are some good starting points: official website, dedicated to canister development on the IC; IC developer forum, where one could find an answer to almost any technical question.

Complete code of this tutorial is here:

seniorjoinu / ic-event-hub-token-ledgers-example

Motivation

In the previous article about ic-event-hub "Introduction To ic-event-hub Library", we imagined the following scenario:

you've built a token canister that accepts transfer transactions from users and re-transmits the info about these transactions to any other canister that will ask for it;
there are many ledger canisters (developed by third parties who want to integrate them with your token), each of which is specialized on specific transaction type:
- one of them only keeps track of transactions made by US citizens;
- another one only keeps track of transactions made to/from charity organizations;
- a bunch of personal ledgers, which keep track of transactions of a particular person or organization;
- and each month there is at least one more ledger appears with some new specific requirements;
besides all of that you want to make your token as efficient as possible:
- if the token experiences high load (e.g. 100+ tx/block), you don't want to send 100+ messages to each ledger;
- since messages you send to ledgers are small (only contain some generic transfer data: from, to, amount, timestamp), you want to pack these messages into a single batch and send them all at once - this way you could save a lot of cycles.

Let's now implement it! Let's build a simple token canister that allows multiple ledgers to listen for various events happening in this token. And also let's build a couple of such ledgers.

Implementation

Project layout

project/
    token/
        src/
            actor.rs
            common/
                mod.rs
                currency_token.rs
                guards.rs
                types.rs
        build.sh
        can.did
        cargo.toml

    ledger-1/
        src/
            actor.rs
            common/
                mod.rs
                ledger.rs
                types.rs
        build.sh
        can.did
        cargo.toml

    ledger-2/
        src/
            actor.rs
            common/
                mod.rs
                ledger.rs
                types.rs
        build.sh
        can.did
        cargo.toml

DFX configuration

// project/dfx.json

{
  "canisters": {
    "token": {
      "build": "./token/build.sh",
      "candid": "./token/can.did",
      "wasm": "./token/target/wasm32-unknown-unknown/release/token-opt.wasm",
      "type": "custom"
    },
    "ledger-1": {
      "build": "./ledger-1/build.sh",
      "candid": "./ledger-1/can.did",
      "wasm": "./ledger-1/target/wasm32-unknown-unknown/release/ledger-1-opt.wasm",
      "type": "custom"
    },
    "ledger-2": {
      "build": "./ledger-2/build.sh",
      "candid": "./ledger-2/can.did",
      "wasm": "./ledger-2/target/wasm32-unknown-unknown/release/ledger-2-opt.wasm",
      "type": "custom"
    }
  },
  "dfx": "0.9.2",
  "networks": {
    "local": {
      "bind": "127.0.0.1:8000",
      "type": "ephemeral"
    }
  },
  "version": 1
}

Token canister

This canister provides basic token functionality: minting, transferring and burning. Each time such a functionality is used, a corresponding event is emitted.

Dependencies

# project/token/cargo.toml

[package]
name = "token"
version = "0.1.0"
edition = "2018"

[lib]
crate-type = ["cdylib"]
path = "src/actor.rs"

[dependencies]
ic-cdk = "0.4.0"
ic-cdk-macros = "0.4.0"
candid = "0.7.12"
serde = "1.0"
ic-event-hub = "0.3.0"
ic-event-hub-macros = "0.3.0"

Buildscript

# project/token/build.sh

#!/usr/bin/env bash

SCRIPT=$(readlink -f "$0")
SCRIPTPATH=$(dirname "$SCRIPT")
cd "$SCRIPTPATH" || exit

cargo build --target wasm32-unknown-unknown --release --package token && \
 ic-cdk-optimizer ./target/wasm32-unknown-unknown/release/token.wasm -o ./target/wasm32-unknown-unknown/release/token-opt.wasm

Token internal logic

As it was said earlier, our token would have three separate functionalities: mint, transfer and burn. Also we want to provide a function to get current token balance of some account (we use user principals as account identifiers) as well as to get total token supply value.

We also want our token to have some associated information: token name, token symbol (ticker) and token decimals (how many cents there are in one token).

Also we want to control the access to the minting functionality, to allow to invoke it only for a selected set of principals (controllers).

These requirements left us with the following token state structure:

// project/token/src/common/currency_token.rs

#[derive(CandidType, Deserialize)]
pub struct CurrencyToken {
    pub balances: HashMap<Principal, u64>,
    pub total_supply: u64,
    pub info: TokenInfo,
    pub controllers: Controllers,
}

pub type Controllers = Vec<Principal>;

Let's define the implementation of this structure that will add all the functions we need:

// project/token/src/common/currency_token.rs

impl CurrencyToken {
    pub fn mint(&mut self, to: Principal, qty: u64) -> Result<(), Error> {
        if qty == 0 {
            return Err(Error::ZeroQuantity);
        }

        let prev_balance = self.balance_of(&to);
        let new_balance = prev_balance + qty;

        self.total_supply += qty;
        self.balances.insert(to, new_balance);

        Ok(())
    }

    pub fn transfer(&mut self, from: Principal, to: Principal, qty: u64) -> Result<(), Error> {
        if qty == 0 {
            return Err(Error::ZeroQuantity);
        }

        let prev_from_balance = self.balance_of(&from);
        let prev_to_balance = self.balance_of(&to);

        if prev_from_balance < qty {
            return Err(Error::InsufficientBalance);
        }

        let new_from_balance = prev_from_balance - qty;
        let new_to_balance = prev_to_balance + qty;

        if new_from_balance == 0 {
            self.balances.remove(&from);
        } else {
            self.balances.insert(from, new_from_balance);
        }

        self.balances.insert(to, new_to_balance);

        Ok(())
    }

    pub fn burn(&mut self, from: Principal, qty: u64) -> Result<(), Error> {
        if qty == 0 {
            return Err(Error::ZeroQuantity);
        }

        let prev_balance = self.balance_of(&from);

        if prev_balance < qty {
            return Err(Error::InsufficientBalance);
        }

        let new_balance = prev_balance - qty;

        if new_balance == 0 {
            self.balances.remove(&from);
        } else {
            self.balances.insert(from, new_balance);
        }

        self.total_supply -= qty;

        Ok(())
    }

    pub fn balance_of(&self, account_owner: &Principal) -> u64 {
        match self.balances.get(account_owner) {
            None => 0,
            Some(b) => *b,
        }
    }
}

There is nothing unusual in these functions. In fact they're pretty much copying the same functions from ERC-20 standard.

Each of these functions may return an Error, which type is defined as follows:

// project/token/src/common/types.rs

#[derive(Debug)]
pub enum Error {
    InsufficientBalance,
    ZeroQuantity,
    AccessDenied,
    ForbiddenOperation,
}

Actor definition

Now, when we have our basic token functionality ready, let's wrap it into a canister. The state definition and initialization is as simple as follows:

// project/token/src/actor.rs

static mut STATE: Option<CurrencyToken> = None;

pub fn get_state() -> &'static mut CurrencyToken {
    unsafe { STATE.as_mut().unwrap() }
}

#[init]
fn init(controller: Principal, info: TokenInfo) {
    let token = CurrencyToken {
        balances: HashMap::new(),
        total_supply: 0,
        info,
        controllers: vec![controller],
    };

    unsafe {
        STATE = Some(token);
    }
}

As you might notice, we pass a controller and token's info into an init() function. This token info is defined like this:

// project/token/src/common/types.rs

#[derive(Clone, CandidType, Deserialize)]
pub struct TokenInfo {
    pub name: String,
    pub symbol: String,
    pub decimals: u8,
}

The next thing we want to do is to set up the ic-event-hub library. In order to do that, we need to invoke these three macros:

// project/token/src/actor.rs

implement_event_emitter!(10 * 1_000_000_000, 100 * 1024);
implement_subscribe!();
implement_unsubscribe!();

The first one initializes the event emitter's state. You can find more information about it in this tutorial. Other two macros are enabling event listeners to subscribe for events, by adding a couple of new update functions (subscribe() and unsubscribe()).

In order to fully enable ic-event-hub we also have to call send_events() function inside the heartbeat system function like this:

// project/token/src/actor.rs

#[heartbeat]
pub fn tick() {
    send_events();
}

This expression will check the ic-event-hub's state for ready event batches and then it will send these batches to their recipients.

Now, let's define all the update functions of our canister:

// project/token/src/actor.rs

#[update(guard = "controller_guard")]
fn mint(to: Principal, qty: u64) {
    get_state().mint(to, qty).expect("Minting failed");

    emit(MintEvent {
        to,
        amount: qty,
        timestamp: time(),
    });
}

#[update]
fn transfer(to: Principal, qty: u64) {
    let from = caller();

    get_state()
        .transfer(from, to, qty)
        .expect("Transfer failed");

    emit(TransferEvent {
        from,
        to,
        amount: qty,
        timestamp: time(),
    });
}

#[update]
fn burn(qty: u64) {
    let from = caller();

    get_state().burn(from, qty).expect("Burning failed");

    emit(BurnEvent {
        from,
        amount: qty,
        timestamp: time(),
    });
}

In each of these methods, we're just calling an associated function defined in the state and emit an associated event. There are three types of events:

// project/token/src/common/types.rs

#[derive(Event)]
pub struct TransferEvent {
    #[topic]
    pub from: Principal,
    #[topic]
    pub to: Principal,
    pub amount: u64,
    pub timestamp: u64,
}

#[derive(Event)]
pub struct MintEvent {
    #[topic]
    pub to: Principal,
    pub amount: u64,
    pub timestamp: u64,
}

#[derive(Event)]
pub struct BurnEvent {
    #[topic]
    pub from: Principal,
    pub amount: u64,
    pub timestamp: u64,
}

By the way, the controllers_guard() function is defined this way:

// project/token/src/common/guards.rs

pub fn controller_guard() -> Result<(), String> {
    if get_state().controllers.contains(&caller()) {
        Ok(())
    } else {
        Err(String::from("The caller is not a controller"))
    }
}

The only thing our token misses are some basic query methods:

// project/token/src/actor.rs

#[query]
fn get_balance_of(account_owner: Principal) -> u64 {
    get_state().balance_of(&account_owner)
}

#[query]
fn get_total_supply() -> u64 {
    get_state().total_supply
}

#[query]
fn get_info() -> TokenInfo {
    get_state().info.clone()
}

This is it. The complete code for our token canister should look like this:

// project/common/src/actor.rs

// ----------------- MAIN LOGIC ------------------

#[update(guard = "controller_guard")]
fn mint(to: Principal, qty: u64) {
    get_state().mint(to, qty).expect("Minting failed");

    emit(MintEvent {
        to,
        amount: qty,
        timestamp: time(),
    });
}

#[update]
fn transfer(to: Principal, qty: u64) {
    let from = caller();

    get_state()
        .transfer(from, to, qty)
        .expect("Transfer failed");

    emit(TransferEvent {
        from,
        to,
        amount: qty,
        timestamp: time(),
    });
}

#[update]
fn burn(qty: u64) {
    let from = caller();

    get_state().burn(from, qty).expect("Burning failed");

    emit(BurnEvent {
        from,
        amount: qty,
        timestamp: time(),
    });
}

#[query]
fn get_balance_of(account_owner: Principal) -> u64 {
    get_state().balance_of(&account_owner)
}

#[query]
fn get_total_supply() -> u64 {
    get_state().total_supply
}

#[query]
fn get_info() -> TokenInfo {
    get_state().info.clone()
}

// ------------------- EVENT HUB ------------------

implement_event_emitter!(10 * 1_000_000_000, 100 * 1024);
implement_subscribe!();
implement_unsubscribe!();

#[heartbeat]
pub fn tick() {
    send_events();
}

// ------------------ STATE ----------------------

static mut STATE: Option<CurrencyToken> = None;

pub fn get_state() -> &'static mut CurrencyToken {
    unsafe { STATE.as_mut().unwrap() }
}

#[init]
fn init(controller: Principal, info: TokenInfo) {
    let token = CurrencyToken {
        balances: HashMap::new(),
        total_supply: 0,
        info,
        controllers: vec![controller],
    };

    unsafe {
        STATE = Some(token);
    }
}

Candid interface

// project/token/can.did

type TokenInfo = record {
    name : text;
    symbol : text;
    decimals : nat8;
};

service : (principal, TokenInfo) -> {
    "mint" : (principal, nat64) -> ();
    "transfer" : (principal, nat64) -> ();
    "burn" : (nat64) -> ();
    "get_balance_of" : (principal) -> (nat64) query;
    "get_total_supply" : () -> (nat64) query;
    "get_info" : () -> (TokenInfo) query;
}

Let's move to ledgers now.

Ledger canisters

We're going to implement two ledgers. Their implementation would be almost the same, but the first one will listen for all possible events emitted by the token canister, while the second one will listen only for events related to the particular principal we're gonna provide it at the canister initialization phase.

Dependencies

# project/ledger-1/cargo.toml

[package]
name = "ledger-1"
version = "0.1.0"
edition = "2018"

[lib]
crate-type = ["cdylib"]
path = "src/actor.rs"

[dependencies]
ic-cdk = "0.4.0"
ic-cdk-macros = "0.4.0"
candid = "0.7.12"
serde = "1.0"
ic-cron = "0.6.1"
ic-event-hub = "0.3.0"
ic-event-hub-macros = "0.3.0"

Buildscript

# project/ledger-1/build.sh

#!/usr/bin/env bash

SCRIPT=$(readlink -f "$0")
SCRIPTPATH=$(dirname "$SCRIPT")
cd "$SCRIPTPATH" || exit

cargo build --target wasm32-unknown-unknown --release --package ledger-1 && \
 ic-cdk-optimizer ./target/wasm32-unknown-unknown/release/ledger_1.wasm -o ./target/wasm32-unknown-unknown/release/ledger-1-opt.wasm

Ledger internal logic

Ledger is a pretty straightforward thing - it is just a log of historical records in a chronological order. So its state and internals are trivial:

// project/ledger-1/src/common/ledger.rs

#[derive(CandidType, Deserialize)]
pub struct Ledger {
    pub token: Principal,
    pub entries: Vec<Entry>,
}

impl Ledger {
    pub fn add_entry(&mut self, entry: Entry) {
        self.entries.push(entry);
    }

    pub fn get_entries(&self) -> Vec<Entry> {
        self.entries.clone()
    }
}

We have a principal of the token we're going to listen to and the log of historical entries each of which is defined like this:

// project/ledger-1/src/common/types.rs

#[derive(Clone, CandidType, Deserialize)]
pub enum Entry {
    Mint(MintEvent),
    Transfer(TransferEvent),
    Burn(BurnEvent),
}

#[derive(Clone, Event, CandidType, Deserialize)]
pub struct TransferEvent {
    #[topic]
    pub from: Principal,
    #[topic]
    pub to: Principal,
    pub amount: u64,
    pub timestamp: u64,
}

#[derive(Clone, Event, CandidType, Deserialize)]
pub struct MintEvent {
    #[topic]
    pub to: Principal,
    pub amount: u64,
    pub timestamp: u64,
}

#[derive(Clone, Event, CandidType, Deserialize)]
pub struct BurnEvent {
    #[topic]
    pub from: Principal,
    pub amount: u64,
    pub timestamp: u64,
}

Notice, events are defined the same way they are in the token's codebase.

Actor definition

Let's define canister's state and init() function:

// project/ledger-1/src/actor.rs

static mut STATE: Option<Ledger> = None;

pub fn get_state() -> &'static mut Ledger {
    unsafe { STATE.as_mut().unwrap() }
}

implement_cron!();

#[init]
fn init(token_principal: Principal) {
    let token = Ledger {
        token: token_principal,
        entries: Vec::new(),
    };

    unsafe {
        STATE = Some(token);
    }

    cron_enqueue(
        token_principal,
        SchedulingOptions {
            delay_nano: 0,
            interval_nano: 0,
            iterations: Exact(1),
        },
    )
    .expect("Enqueue failed");
}

#[heartbeat]
pub fn tick() {
    for task in cron_ready_tasks() {
        let token = task
            .get_payload::<Principal>()
            .expect("Payload deserialization failed");

        spawn(async move {
            token
                .subscribe(SubscribeRequest {
                    callbacks: vec![CallbackInfo {
                        filter: EventFilter::empty(),
                        method_name: String::from("events_callback"),
                    }],
                })
                .await
                .expect("Subscribe failed");
        });
    }
}

As you might know, it is impossible to make calls to external canisters inside init() function (here is a thread about this), but we can workaround this restriction, by using ic-cron library.
All we have to do is to enqueue a cron task and then process it in the first ever heartbeat of this canister. This is exactly what we do here. Processing the enqueued task we're calling the token canister in order to subscribe to all events it emits.

Let's now define an events_callback() function that will be used by ic-event-hub as a callback for event processing:

// project/ledger-1/src/actor.rs

#[update(guard = "token_guard")]
pub fn events_callback(events: Vec<Event>) {
    for event in events {
        match event.get_name().as_str() {
            "MintEvent" => {
                let mint_event = MintEvent::from_event(event);
                get_state().add_entry(Entry::Mint(mint_event));
            }
            "TransferEvent" => {
                let transfer_event = TransferEvent::from_event(event);
                get_state().add_entry(Entry::Transfer(transfer_event));
            }
            "BurnEvent" => {
                let burn_event = BurnEvent::from_event(event);
               get_state().add_entry(Entry::Burn(burn_event));
            }
            _ => trap("Unknown event"),
        }
    }
}

fn token_guard() -> Result<(), String> {
    if caller() != get_state().token {
        Err(String::from("Can only be called by the token canister"))
    } else {
        Ok(())
    }
}

This function simply wraps any received event into an enum we defined earlier and then stores them into the state. We're guarding this function in order to prevent malicious canisters (or users) from messing with the history.

Let's also define a query function that will let us get all events stored in this ledger:

// project/ledger-1/src/actor.rs

#[query]
fn get_events() -> Vec<Entry> {
    get_state().get_entries()
}

This is it. Ledger canister #1 is ready. Don't forget to define the .did file:

type Entry = variant {
    Mint : MintEvent;
    Transfer : TransferEvent;
    Burn : BurnEvent;
};

type MintEvent = record {
    to : principal;
    amount : nat64;
    timestamp : nat64;
};

type TransferEvent = record {
    from : principal;
    to : principal;
    amount : nat64;
    timestamp : nat64;
};

type BurnEvent = record {
    from : principal;
    amount : nat64;
    timestamp : nat64;
};

service : (principal) -> {
    "get_events" : () -> (vec Entry) query;
}

Ledger #2

The second ledger is defined almost the same way as the previous one. It differs in a way the subscribe() call is made, because we want it to only listen for events related to a particular principal:

// project/ledger-2/src/actor.rs

#[heartbeat]
pub fn tick() {
    for _ in cron_ready_tasks() {
        spawn(async move {
            let state = get_state();

            state
                .token
                .subscribe(SubscribeRequest {
                    callbacks: vec![
                        CallbackInfo {
                            filter: MintEventFilter {
                                to: Some(state.track),
                            }
                            .to_event_filter(),
                            method_name: String::from("mint_callback"),
                        },
                        CallbackInfo {
                            filter: TransferEventFilter {
                                from: Some(state.track),
                                to: None,
                            }
                            .to_event_filter(),
                            method_name: String::from("transfer_callback"),
                        },
                        CallbackInfo {
                            filter: TransferEventFilter {
                                from: None,
                                to: Some(state.track),
                            }
                            .to_event_filter(),
                            method_name: String::from("transfer_callback"),
                        },
                        CallbackInfo {
                            filter: BurnEventFilter {
                                from: Some(state.track),
                            }
                            .to_event_filter(),
                            method_name: String::from("burn_callback"),
                        },
                    ],
                })
                .await
                .expect("Subscribe failed");
        });
    }
}

Subscribing for each event type we set an event filter that defines the only account identifier we're interested in. This account identifier (principal) should somehow appear in this ledger, so we just pass it as an argument into init() function and save it to the state:

// project/ledger-2/src/actor.rs

#[init]
fn init(token_principal: Principal, track_principal: Principal) {
    let token = Ledger {
        token: token_principal,
        track: track_principal,
        entries: Vec::new(),
    };

    unsafe {
        STATE = Some(token);
    }

    cron_enqueue(
        (),
        SchedulingOptions {
            delay_nano: 0,
            interval_nano: 0,
            iterations: Exact(1),
        },
    )
    .expect("Enqueue failed");
}

And since we modify the signature of the init() function, we're also gonna need to update the candid interface:

// project/ledger-2/can.did

...

service : (principal, principal) -> {
    "get_events" : () -> (vec Entry) query;
}

Everything else stays the same.

Battle testing

Let's now try to deploy and use our token and see what will happen.

Start the development network in a separate terminal window:

$ dfx start --clean

Check your principal:

$ dfx identity get-principal
<your principal>

Deploy canisters:

$ dfx deploy token --argument '(principal "<your principal>", record { name = "Test"; symbol = "TST"; decimals = 8 : nat8 })'

$ dfx deploy ledger-1 --argument '(principal "<token canister id>")'

$ dfx deploy ledger-2 --argument '(principal "<token canister id>", principal "aaaaa-aa")'

We're using "aaaaa-aa" principal as an account identifier for ledger-2 to listen to as an example, but you can use any other principal here.

Mint some tokens to yourself:

$ dfx canister call token mint '(principal "<your principal>", 1000)'

()

Transfer some tokens to "aaaaa-aa":

$ dfx canister call token transfer '(principal "aaaaa-aa", 200)'

()

Burn the rest:

$ dfx canister call token burn '(800)'

()

Let's now check the first ledger's state:

$ dfx canister call ledger-1 get_events '()'

(
  vec {
    variant {
      Mint = record {
        to = principal "<your principal>";
        timestamp = 1_647_816_085_333_883_997 : nat64;
        amount = 1_000 : nat64;
      }
    };
    variant {
      Transfer = record {
        to = principal "aaaaa-aa";
        from = principal "<your principal>";
        timestamp = 1_647_816_164_528_431_504 : nat64;
        amount = 200 : nat64;
      }
    };
    variant {
      Burn = record {
        from = principal "<your principal>";
        timestamp = 1_647_816_211_161_122_783 : nat64;
        amount = 800 : nat64;
      }
    };
  },
)

All three transactions are there, good. Now let's check the second ledger:

$ dfx canister call ledger-2 get_events '()'

(
  vec {
    variant {
      Transfer = record {
        to = principal "aaaaa-aa";
        from = principal "<your principal>";
        timestamp = 1_647_816_164_528_431_504 : nat64;
        amount = 200 : nat64;
      }
    };
  },
)

As was expected, it only contains a single transaction that is related to "aaaaa-aa" account id.

Looks like everything works fine. Woohoo!

The complete source code for this tutorial can be found here:

seniorjoinu / ic-event-hub-token-ledgers-example

Thanks for reading!

Introduction to ic-event-hub library

Alexander — Mon, 28 Mar 2022 15:06:29 +0000

Before digging into this tutorial it is recommended to learn the basics of smart-contract development for the Internet Computer. Here are some good starting points: official website, dedicated to canister development on the IC; IC developer forum, where one could find an answer to almost any technical question.

This tutorial is intended for new developers who want to understand the basics of ic-event-hub. If you already finished it, it is recommended to check out my other tutorials about this library:

seniorjoinu / ic-event-hub

Event-based pub/sub for IC canisters

IC Event Hub

A rust library that enables efficient event-based pub/sub for IC canisters

Motivation

The main idea behind the Open Internet Services concept is collaboration. The easier it is for us to integrate canisters, the more great software we can build together, the closer we are to the Open Internet.

This library greatly simplifies canister integration by flipping integration surface upside down. It enables your canister to express its interfaces in terms of emitted events. Sending a message, your canister does not need to know a signature of the remote canister anymore. Instead, the remote canister should know what events it wants to receive from your canister. This enables us to create emitter canisters - ones which emit events, when something important happens inside it, and which does not care who are the exact recipients of these events. On the other side, listener-canisters (recipients) are able to make…

View on GitHub

Motivation

Currently, canisters on the IC use request/response model to communicate with each other. While this model covers almost 100% of any needs, sometimes it limits integration simplicity. It's simpler to go through an example here.

Imagine the following scenario:

you've built a token canister that accepts transfer transactions from users and re-transmits the info about these transactions to any other canister that will ask for it;
there are many ledger canisters (developed by third parties who want to integrate them with your token), each of which is specialized on specific transaction type:
- one of them only keeps track of transactions made by US citizens;
- another one only keeps track of transactions made to/from charity organizations;
- a bunch of personal ledgers, which keep track of transactions of a particular person or organization;
- and each month there is at least one more ledger appears with some new specific requirements;
besides all of that you want to make your token as efficient as possible:
- if the token experiences high load (e.g. 100+ tx/block), you don't want to send 100+ messages to each ledger;
- since messages you send to ledgers are small (only contain some generic transfer data: from, to, amount, timestamp), you want to pack these messages into a single batch and send them all at once - this way you could save a lot of cycles.

It looks like in order to implement such a scenario, one would need a system that fits into the following requirements:

We want to "fire-and-forget" data without bothering about who exactly is going to receive it or if there is an error at the receiver's side.
We want receivers to be able to specify conditions of desired data, e.g. "if this peace of data equals to - send it to me, otherwise ignore me".
There should be some kind of data-batching functionality, that will help us stay on the budget.

These declarations imply that an event-based protocol is what we actually want. Such a protocol, where there are emitters which emit events and listeners which receive these events based on some conditions.

ic-event-hub library implements that event-based protocol for us.

ic-event-hub

This library defines two types of actors: emitters and listeners. They communicate via events - special data structures optimized for conditional subscription. Emitters "emit" events, listeners (subscribers) "subscribe" to events they are interested in.

Here is an example of an event:

#[derive(Event)]
pub struct TransferEvent {
    #[topic]
    pub from: Principal,
    #[topic]
    pub to: Principal,
    pub amount: Nat,
    pub timestamp: u64,
}

Notice two things:

The whole struct is annotated with #[derive(Event)] macro. This macro will implement the IEvent trait for us - the trait that transforms our custom structure into the Event structure.
Some fields are also annotated with #[topic] macro. By annotating a field with this macro we enable listeners to filter all emitted events and only receive events which have these two fields set to some specific value.

By annotating a struct with #[derive(Event)] macro you'll also be automatically provided with special EventFilter struct. This struct is used by listeners in order to specify events they are interested in.

For example, annotating the TransferEvent struct defined above with #[derive(Event) would also generate the following structure:

pub struct TransferEventFilter {
    pub from: Option<Principal>,
    pub to: Option<Principal>,
}

If the listener wants to receive all TransferEvents they would initialize this structure this way:

TransferEventFilter { from: None, to: None }

If they want to receive only those TransferEvents related to some particular sender, they would use this structure the following way:

TransferEventFilter { from: Some(some_principal), to: None }

Emitter canister

Emitter canisters emit events by calling emit() function:

emit(TransferEvent {
    from,
    to,
    amount,
    timestamp,
});

As you might notice, there is no .await after the invocation. This is because of "fire-and-forget" policy - the emitter sends data in form of events to "nowhere". It doesn't even bother if there is any listener out there - the library handles it automatically.

It is important to say, that the "fire-and-forget" policy implementation is virtual. Under the hood the emitter canister still waits for responses for sent event batches in the background, but it just ignores them once they appear.

The event won't be transmitted immediately. Instead, the library will push it to an event queue (unique for each listener). Once this queue grows big enough or once enough time has passed, only then this queue is transformed into a batch and sent to the listener canister.

This implies that even if one writes something like this:

emit(event_1);
emit(event_2);
emit(event_3);

These events will be merged into a single event batch and sent all at once.

Both parameters: the queue's size in bytes and the desired delay before the transmission are configurable. You can pass them into the library-initializing macro like this: implement_event_emitter!(max_time_interval_nano, max_batch_size_bytes). What about this macro - it initializes the state of the library and make functions like emit() available for usage, so don't forget to call it somewhere in your canister's code.

In order to activate the transmission of events we have to use the #[heartbeat] function:

#[heartbeat]
pub fn tick() {
    send_events();
}

In order to receive events, a listener canister has to subscribe to them. An emitter canister's developer could allow any listener to subscribe to events emitted by this canister. To do this, the developer would have to call implement_subscribe!() macro somewhere inside the canister's code.

This macro supports guard function declaration inside: implement_subscribe!(guard = "guard_function").

Listener canister

After that a listener canister is able to call subscribe() function of the emitter canister, like this:

let event_filter = TransferEventFilter { from: Some(some_principal), to: None };

emitter_canister_principal
    .subscribe(SubscribeRequest {
        callbacks: vec![CallbackInfo {
            filter: event_filter.to_event_filter(),
            callback_method_name: String::from("events_callback"),
        }],
    })
    .await
    .expect("Subscription failed");

As you might notice, a listener is able to specify the callback method name for each event filter. ic-event-hub on the emitter's side will use this callback function name to send all the events which fit into the provided EventFilter. This callback should be implemented like this:

#[update]
fn events_callback(events: Vec<Event>) {
    for event in events {
        if event.get_name().as_str() == "TransferEvent" {
            let ev: TransferEvent = TransferEvent::from_event(event);
            print(format!("Got event: {:?}", ev).as_str());
        }
    }
}

It should also accept a single argument of type Vec<Event> and return nothing. If you need extra security (e.g. forbid some canisters from sending events to you), you can also express it inside this function.

Combining everything together

Okay, let's define a basic example of how one could use ic-event-hub in order to implement a solution for the scenario we described earlier.

On the token canister side we would need to put a code like this:

// define an event data type
#[derive(Event)]
pub struct TransferEvent {
    #[topic]
    pub from: Principal,
    #[topic]
    pub to: Principal,
    pub amount: Nat,
    pub timestamp: u64,
}

// initialize ic-event-hub state and enable the `emit()` function
implement_event_emitter!(1_000_000_000 * 60, 100 * 1024 * 1024);

// emit events somewhere in your code (for example - inside the `transfer()` function of the token
emit(TransferEvent {
    from,
    to,
    amount,
    timestamp,
});

// enable listeners to subscribe
implement_subscribe!()

// activate event sending
#[heartbeat]
pub fn tick() {
    send_events();
}

On a ledger canister side:

// define the same event data type
#[derive(Event)]
pub struct TransferEvent {
    #[topic]
    pub from: Principal,
    #[topic]
    pub to: Principal,
    pub amount: Nat,
    pub timestamp: u64,
}

// define event filter(s) describing the data you want to receive
let event_filter = TransferEventFilter { from: Some(some_principal), to: None };

// subscribe for events using the filter
emitter_canister_principal
    .subscribe(SubscribeRequest {
        callbacks: vec![CallbackInfo {
            filter: event_filter.to_event_filter(),
            callback_method_name: String::from("events_callback"),
        }],
    })
    .await
    .expect("Subscription failed");

// define the callback function to process incoming events
#[update]
fn events_callback(events: Vec<Event>) {
    for event in events {
        if event.get_name().as_str() == "TransferEvent" {
            let ev: TransferEvent = TransferEvent::from_event(event);
            print(format!("Got event: {:?}", ev).as_str());
        }
    }
}

That's it. ic-event-hub will handle everything else: the transmission, listener subscription, encoding/decoding process, event filter matching and event batching process.

Afterword

The ic-event-hub library is not only an alternative way of transmitting data between canisters. It is also an implementation of Inversion of Control pattern.

In the standard request/response model, the receiver defines the data type it want to receive and the sender decides whether it will send the data or not.

In the ic-event-hub's event-based model, the sender defines the data type it will send and the receiver decides whether it wants to receive that data or not.

In the standard request/response model, a developer has to keep track of many different things: an address (canister id) of the remote program, a signature of the remote function and the right time to transfer the data. In exchange for doing this the model rewards a developer with the ability to process the response of the remote call.

In the ic-event-hub's event-based model, a developer gives up on response processing and in exchange is rewarded with simplicity and high throughput.

It is recommended to use this library as an alternative protocol alongside the standard request/response protocol, not as a replacement for it. This way you, as a developer, would enjoy the best experience possible.

Thanks for reading and don't forget to take a look at ic-event-hub's source code:

seniorjoinu / ic-event-hub

Event-based pub/sub for IC canisters

IC Event Hub

A rust library that enables efficient event-based pub/sub for IC canisters

Motivation

View on GitHub

Tutorial: Efficient Canister Networking With ic-event-hub

Alexander — Sat, 19 Mar 2022 22:09:21 +0000

Before digging into this tutorial it is recommended to learn the basics of smart-contract development for the Internet Computer. Here are some good starting points: official website, dedicated to canister development on the IC; IC developer forum, where one could find an answer to almost any technical question.

Complete code of this tutorial is here.

Motivation

Smart-contracts are not cheap. Running a simple "hello world" smart-contract is at least 10 times more expensive on the IC than running an similar "hello world" service on the AWS. And there is actually a very simple reason why this will always be like that - infrastructure costs.

When you run a "hello world" service on a single EC2 AWS replica you waste only a little of the resource (memory, processing power and bandwidth). All of this happens on a single replica (it doesn't matter if the replica is virtual or not).

When you run the same exact service on the IC, you're using the same amount of the resource, but (!) your program is processed by the whole subnet (7 replicas by now). Moreover it is not only processed simultaneously on all these replicas, but on top of that each message to your program goes through the consensus protocol (which under the hood hides much more additional replica interactions).

This means that it is just wrong to compare computation costs between the IC and AWS without taking additional infrastructure costs into an account. In order to make such a comparison, you have to imagine your program running not on a single EC2 AWS instance, but on seven EC2 instances with the help of some kind of consensus protocol (e.g. Kafka) instead. This way your AWS setup will (roughly) provide the same set of security guarantees and is eligible for comparison with the IC.

Smart-contract developers on the IC understand that "time is money" like nobody else. The more efficient code you write, the more money you'll save. Developers now are basically financially motivated to build faster programs, otherwise they just won't be able to sustain their services. Code optimizations and frequent code refactoring are now make much more sense from the business point of view - they cut the losses.

Efficient code -> less money lost -> more money in your pocket. Simple as that.

ic-event-hub helps with cycles saving a lot, by providing a special event-batching functionality. Basically it lets your canister accumulate messages you want to send to other canisters and then send them all together in a single batch.

How does it help with saving cycles? Let's look at this table from the Dfinity's website. We're particularly interested in these two rows:

Xnet Call - 260,000 cycles
Xnet Byte Transmission - 1,000 cycles

It means that it is much cheaper to send one 1000 bytes message, then to send a thousand of 1 byte messages, because for each message sent to another canister there is a fee taken.

Sending one 1000 bytes message would cost you (260,000 + (1,000 * 1,000)) * 1 = 1260000 cycles, while sending a thousand of 1 byte messages would cost you (260,000 + 1,000) * 1000 = 261000000 cycles, 200 times more expensive.

So, if you find yourself in a situation when your canister sends lots of small (under 100KB) messages to other canisters, it is better for you to find a tool that will optimize this process so you could save cycles. ic-event-hub library is exactly that tool.

In this tutorial, we'll go through an example of such a situation and learn how to use ic-event-hub in order to fix it.

Task definition

Imagine the following task: we want data supplied by users to be available on multiple canisters.

Naive solution is to make users to send the same piece of data to multiple canisters. This won't work, because there is no easy way to fight against malicious users.

It means that we're left with the only option - to make users send data to one canister and to make this canister to re-transmit this data to all other canisters. The only problem with this solution is that it is expensive - for each data piece sent by user we send N (where N is canisters count) more messages to propagate this piece of data to other canisters. Let's use ic-event-hub to solve this problem.

Implementation

Let's start with the file system layout:

project/
    canisters/
        emitter/     // a canister that will accept user data
            actor.rs
            build.sh
            can.did
            cargo.toml
        listener/   // a canister that will receive re-transmited user-data
            actor.rs
            build.sh
            can.did
            cargo.toml
    e2e-test/       // typescript-based test suite to check our logic
        src/
            example.spec.ts
            deploy.ts
        dfx.json
        package.json
        tsconfig.json

As you can see, we have two canisters: an emitter and a listener, as well as a typescript project with tests. In this tutorial, instead of using terminal and dfx, we'll use typescript for deployment and testing.

Emitter canister

This canister will receive data from users and re-transmit it to the listener canister in form of events.

Dependencies

# project/emitter/cargo.toml

[package]
name = "emitter"
version = "0.1.0"
edition = "2018"

[lib]
crate-type = ["cdylib"]
path = "actor.rs"

[dependencies]
ic-cdk = "0.4.0"
ic-cdk-macros = "0.4.0"
candid = "0.7.12"
serde = "1.0.126"
ic-event-hub = "0.3.0"
ic-event-hub-macros = "0.3.0"

Buildscript

# project/emitter/build.sh

#!/usr/bin/env bash

# this part makes it possible to execute this script from another directory
SCRIPT=$(readlink -f "$0")
SCRIPTPATH=$(dirname "$SCRIPT")
cd "$SCRIPTPATH" || exit

cargo build --target wasm32-unknown-unknown --release --package emitter && \
 ic-cdk-optimizer ./target/wasm32-unknown-unknown/release/emitter.wasm -o ./target/wasm32-unknown-unknown/release/emitter-opt.wasm

Canister logic

To make it more representative, let's make our canisters count how many pieces of data they process and how many inter-canister messages does it take.

Emitter canister is the one that receives data pieces and re-transmits them. One data piece - one request. Let's define its state and the init() function:

// project/emitter/actor.rs

#[derive(Default)]
pub struct RequestCounter {
    pub counter: u64,
}

static mut STATE: Option<RequestCounter> = None;

pub fn get_state() -> &'static mut RequestCounter {
    unsafe { STATE.as_mut().unwrap() }
}

#[init]
fn init() {
    unsafe {
        STATE = Some(RequestCounter::default());
    }
}

As you can see, it's pretty trivial. We only store requests count inside the state. Let's then define everything we need for ic-event-hub in order to work:

// project/emitter/actor.rs

#[derive(Event)]
pub struct MirrorEvent {
    pub data: Vec<u8>,
}

implement_event_emitter!(1_000_000_000 * 20, 500 * 1024);

We have a simple event MirrorEvent that contains some abstract data supplied by a user - nothing interesting. But let's have a closer look at invoked macros.

implement_event_emitter macro initializes ic-event-hub's state (enabling us to call functions like emit() and send_events()) with some arguments which define event-batching behavior. The first one is the maximum batch forming time. The second one is the maximum batch size in bytes.

These arguments will work the following way:

For each event listener there is a separate "forming batch" queue.
Once an event is emitted, this queue is checked.
If the size of this queue, after adding the new event to it, exceeds the max batch size (the second argument) this queue is transformed into a batch and is pushed into special "ready-to-send" queue. The new "forming batch" queue is created and the event is placed there instead.
If the size of this queue, after adding the new event to it, is less than max batch size, then this new event is just gets added to the queue.
Once each heartbeat each such "forming batch" queue is checked for it's creation timestamp.
If it was created more than maximum batch forming time ago, this queue is transformed into a batch and is pushed into "ready-to-send" queue.

Yes, it sounds complicated, but it is simple in practice. In our case we did set these arguments like:

maximum batch forming time = 20 seconds;
maximum batch size = 500 KB.

This means, that if our canister is under heavy load (emits hundreds of events per second), ic-event-hub guarantees that each batch will never exceed 500 KB in size. This is a useful restriction, because message size is limited in the IC (currently 2MB per message).

On the other hand, if our canister is idle most of the time (emits only a couple of events per hour), ic-event-hub guarantees that these events will reach their destination in no more than 20 seconds.

There are two more macro invocations we have to do here:

// project/emitter/actor.rs

implement_subscribe!();
implement_unsubscribe!();

These macros create two more update functions in your canister: subscribe() and unsubscribe().

You don't have to list them in your .did file.
These update functions let listener canisters to subscribe to events. We'll see how it's done a little later.
By the way. You can supply a guard-function into these macros, like implement_subscribe!(guard = "guard_fn_name"). This will enable you to deny any unauthorized subscription, if you want such a functionality.

Last, but not least, we want to enable event sending. ic-event-hub is designed to sent events during heartbeats, so here is how we should do it:

// project/emitter/actor.rs

#[heartbeat]
pub fn tick() {
    send_events();
}

All that's left to do is to write an update function that would receive users' data and emit it in form of events:

// project/emitter/actor.rs

#[update]
fn mirror(data: Vec<u8>) {
    get_state().counter += 1;

    emit(MirrorEvent { data });
}

#[query]
fn get_requests_count() -> u64 {
    get_state().counter
}

As you might notice there is also a counter increment for each update function call as well as a query function that returns current counter's value.

That's it for the emitter canister. The complete project/emitter/actor.rs file should look like this:

// project/emitter/actor.rs

// ------------- MAIN LOGIC -------------------

#[update]
fn mirror(data: Vec<u8>) {
    get_state().counter += 1;

    emit(MirrorEvent { data });
}

#[query]
fn get_requests_count() -> u64 {
    get_state().counter
}

// ------------------ EVENT HUB ------------------

#[derive(Event)]
pub struct MirrorEvent {
    pub data: Vec<u8>,
}

implement_event_emitter!(1_000_000_000 * 20, 500 * 1024);
implement_subscribe!();
implement_unsubscribe!();

#[heartbeat]
pub fn tick() {
    send_events();
}

// ------------------ STATE ----------------------

#[derive(Default)]
pub struct RequestCounter {
    pub counter: u64,
}

static mut STATE: Option<RequestCounter> = None;

pub fn get_state() -> &'static mut RequestCounter {
    unsafe { STATE.as_mut().unwrap() }
}

#[init]
fn init() {
    unsafe {
        STATE = Some(RequestCounter::default());
    }
}

Candid interface

// project/emitter/can.did

service : {
    "mirror" : (blob) -> ();
    "get_requests_count" : () -> (nat64) query;
}

Listener canister

This canister will receive re-transmitted data from the emitter canister in form of event batches.

Dependencies

# project/listener/cargo.toml

[package]
name = "listener"
version = "0.1.0"
edition = "2018"

[lib]
crate-type = ["cdylib"]
path = "actor.rs"

[dependencies]
ic-cdk = "0.4.0"
ic-cdk-macros = "0.4.0"
candid = "0.7.12"
serde = "1.0.126"
ic-event-hub = "0.3.0"
ic-event-hub-macros = "0.3.0"

Buildscript

# project/listener/build.sh

#!/usr/bin/env bash

SCRIPT=$(readlink -f "$0")
SCRIPTPATH=$(dirname "$SCRIPT")
cd "$SCRIPTPATH" || exit

cargo build --target wasm32-unknown-unknown --release --package listener && \
 ic-cdk-optimizer ./target/wasm32-unknown-unknown/release/listener.wasm -o ./target/wasm32-unknown-unknown/release/listener-opt.wasm

Canister logic

Let's start with the state of this canister. It have to store the principal of the emitter canister and we want to count how many events and batches this canister would receive:

// project/listener/actor.rs

pub struct RequestCounterMirror {
    pub emitter_canister_id: Principal,
    pub events_received: u64,
    pub batches_received: u64,
}

static mut STATE: Option<RequestCounterMirror> = None;

pub fn get_state() -> &'static mut RequestCounterMirror {
    unsafe { STATE.as_mut().unwrap() }
}

#[init]
fn init(emitter_canister_id: Principal) {
    unsafe {
        STATE = Some(RequestCounterMirror {
            emitter_canister_id,
            events_received: 0,
            batches_received: 0,
        });
    }
}

The code is pretty straightforward. Please notice that we pass the principal of the emitter canister into init() function.

The listener canister needs to define the type of events it will receive (you can import it from a shared library or just redefine in place):

// project/listener/actor.rs

#[derive(Event, Debug)]
pub struct MirrorEvent {
    pub data: Vec<u8>,
}

In order to start receiving events from the emitter canister, listener canister has to subscribe() to them. This is done the following way:

// project/listener/actor.rs

#[update]
async fn start_listening() {
    get_state()
        .emitter_canister_id
        .subscribe(SubscribeRequest {
            callbacks: vec![CallbackInfo {
                filter: EventFilter::empty(),
                method_name: String::from("events_callback"),
            }],
        })
        .await
        .ok()
        .unwrap();
}

In the perfect world we would make this subscribe() call from the init() function, but we can't. There is a workaround for this, but we won't use it to keep it simple here.

We're using IEventEmitter trait here in order to make a subscribe() call by simply dotting emitter canister's principal. Come take a look at this trait if you want to make your code cleaner, while calling remote canisters.

Basically, by making this call, the listener canister tells the emitter canister "Hey there! Would you kindly send all events you emit to the 'events_callback' method of mine?".

Let's define this events_callback function. It is important to say that this function should follow some rules to work properly:

it should be an update function;
it should be named exactly like you've registered it sending a subscribe() request;
it should have only a single argument of type Vec<Event>;
it should have no return value (you can return something, but it will be ignored anyway). It is also important to note that this function will be executed once for each received event batch:

// project/listener/actor.rs

#[update]
fn events_callback(events: Vec<Event>) {
    get_state().batches_received += 1;

    for event in events {
        if event.get_name().as_str() == "MirrorEvent" {
            let ev: MirrorEvent = MirrorEvent::from_event(event);
            print(format!("Got event: {:?}", ev).as_str());

            get_state().events_received += 1;
        }
    }
}

This function is pretty simple. For each event in a batch, we're checking for an event's name and if it matches "MirrorEvent" (the name of our event structure), we're decoding a MirrorEvent structure from the generic event that we've received and print this event.
During this function we're also updating the counters.

All that's left to do is to define query functions to be able to read counter values:

// project/listener/actor.rs

#[query]
fn get_events_received() -> u64 {
    get_state().events_received
}

#[query]
fn get_batches_received() -> u64 {
    get_state().batches_received
}

The complete project/listener/actor.rs file should look like this:

// project/listener/actor.rs


// ------------- MAIN LOGIC -------------------

#[query]
fn get_events_received() -> u64 {
    get_state().events_received
}

#[query]
fn get_batches_received() -> u64 {
    get_state().batches_received
}

// ----------------- EVENT HUB ----------------------

#[derive(Event, Debug)]
pub struct MirrorEvent {
    pub data: Vec<u8>,
}

#[update]
async fn start_listening() {
    get_state()
        .emitter_canister_id
        .subscribe(SubscribeRequest {
            callbacks: vec![CallbackInfo {
                filter: EventFilter::empty(),
                method_name: String::from("events_callback"),
            }],
        })
        .await
        .ok()
        .unwrap();
}

#[update]
fn events_callback(events: Vec<Event>) {
    get_state().batches_received += 1;

    for event in events {
        if event.get_name().as_str() == "MirrorEvent" {
            let ev: MirrorEvent = MirrorEvent::from_event(event);
            print(format!("Got event: {:?}", ev).as_str());

            get_state().events_received += 1;
        }
    }
}

// ------------------ STATE ----------------------

pub struct RequestCounterMirror {
    pub emitter_canister_id: Principal,
    pub events_received: u64,
    pub batches_received: u64,
}

static mut STATE: Option<RequestCounterMirror> = None;

pub fn get_state() -> &'static mut RequestCounterMirror {
    unsafe { STATE.as_mut().unwrap() }
}

#[init]
fn init(emitter_canister_id: Principal) {
    unsafe {
        STATE = Some(RequestCounterMirror {
            emitter_canister_id,
            events_received: 0,
            batches_received: 0,
        });
    }
}

Candid interface

// project/listener/can.did

service : (principal) -> {
    "start_listening" : () -> ();
    "get_events_received" : () -> (nat64) query;
    "get_batches_received" : () -> (nat64) query;
}

Tests

Now let's write a typescript test in order to check if everything works as expected.

Dependencies

// e2e-test/package.json

{
  "name": "e2e-tests",
  "version": "1.0.0",
  "main": "index.js",
  "license": "MIT",
  "scripts": {
    "start": "dfx start --clean",
    "build": "dfx build --check",
    "test": "ts-mocha --paths --timeout 1000000 src/**/*.spec.ts"
  },
  "devDependencies": {
    "@dfinity/agent": "^0.9.2",
    "@dfinity/candid": "^0.9.2",
    "@dfinity/identity": "^0.9.2",
    "@dfinity/principal": "^0.9.2",
    "@types/chai": "^4.2.21",
    "@types/mocha": "^9.0.0",
    "@types/node": "^16.4.3",
    "@types/node-fetch": "^2.5.12",
    "chai": "^4.3.4",
    "mocha": "^9.0.3",
    "node-fetch": "^2.6.1",
    "ts-mocha": "^8.0.0",
    "ts-node": "^10.1.0",
    "tsconfig-paths": "^3.10.1",
    "typescript": "^4.3.5"
  }
}

Typescript config

// e2e-test/tsconfig.json

{
  "compilerOptions": {
    "target": "ESNext",
    "module": "commonjs",
    "outDir": "./dist",
    "strict": true,
    "moduleResolution": "node",
    "baseUrl": "./",
    "paths": {
      "dfx/*": [
        ".dfx/local/canisters/*.did.js"
      ],
      "dfx-type/*": [
        ".dfx/local/canisters/*.did.d.ts"
      ]
    },
    "allowJs": true,
    "allowSyntheticDefaultImports": true,
    "esModuleInterop": true,
    "preserveSymlinks": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  }
}

Notice how the paths section is arranged.

DFX config

// e2e-test/dfx.json

{
  "canisters": {
    "emitter": {
      "build": "../canisters/emitter/build.sh",
      "candid": "../canisters/emitter/can.did",
      "wasm": "../canisters/emitter/target/wasm32-unknown-unknown/release/emitter-opt.wasm",
      "type": "custom"
    },
    "listener": {
      "build": "../canisters/listener/build.sh",
      "candid": "../canisters/listener/can.did",
      "wasm": "../canisters/listener/target/wasm32-unknown-unknown/release/listener-opt.wasm",
      "type": "custom"
    }
  },
  "dfx": "0.9.2",
  "networks": {
    "local": {
      "bind": "127.0.0.1:8000",
      "type": "ephemeral"
    }
  },
  "version": 1
}

Canister deployment with typescript

So, we want to have a function that would deploy our canisters, instantiate actor objects with the correct IDLs and return them back for us to use. In order to implement this function, we first need to implement another one, that will create a new canister, install the provided wasm-module on that canister and return an actor object with the provided interface. Let's script that function:

// e2e-test/src/deploy.ts

export async function deployCanister<T>(name: string, arg: number[], agent: HttpAgent): Promise<{ actor: T, canisterId: Principal }> {
    const managementCanister = getManagementCanister({agent});
    const {canister_id} = await managementCanister.provisional_create_canister_with_cycles({amount: [], settings: []});
    const wasm = fs.readFileSync(`.dfx/local/canisters/${name}/${name}.wasm`);
    const {idlFactory} = await import(`dfx/${name}/${name}`)

    await managementCanister.install_code({
        canister_id,
        mode: {[CanisterInstallMode.Install]: null},
        wasm_module: [...wasm],
        arg
    });

    console.log(`Canister ${name} ${canister_id} deployed`);

    return {
        actor: Actor.createActor(idlFactory, {
            agent,
            canisterId: canister_id
        }),
        canisterId: canister_id
    };
}

Let's now define a data type we want to receive back, when deployment is over. We want to receive back at least a HttpAgent and an of actor object for each of both: the emitter canister and the listener canister:

// e2e-test/src/deploy.ts

import {_SERVICE as IEmitterService} from 'dfx-type/emitter/emitter';
import {_SERVICE as IListenerService} from 'dfx-type/listener/listener';

export interface ISetup {
    agent: HttpAgent;
    emitterService: IEmitterService;
    listenerService: IListenerService;
}

Notice how clean our imports look, thanks to tsconfig-paths package.

Now we can finally write that function that will do all the work for us:

// e2e-test/src/deploy.ts

import fetch from "node-fetch";

export async function setup(identity: Identity): Promise<ISetup> {
    const agent = new HttpAgent({
        host: 'http://localhost:8000/',
        // @ts-ignore
        fetch,
        identity,
    });
    await agent.fetchRootKey();

    const {
        actor: emitterService,
        canisterId: emitterCanisterId
    } = await deployCanister<IEmitterService>('emitter', [], agent);

    const {
        actor: listenerService
    } = await deployCanister<IListenerService>('listener', [...IDL.encode([IDL.Principal], [emitterCanisterId])], agent);

    return {
        agent,
        emitterService,
        listenerService,
    };
}

Now we can use this function inside tests any possible way we like. One could imagine calling this function from beforeEach callback, to have a clean canister state before each test.

Testing

We're almost there. Now we only left to implement the test-case itself:

// e2e-test/src/example.spec.ts

describe('event batching', () => {
    let s: ISetup;

    before(async () => {
        s = await setup(Ed25519KeyIdentity.generate());
    });

    it("flow works fine", async () => {
        // this listener should catch all events
        await s.listenerService.start_listening();

        // checking before
        const emitterRequestsBefore = await s.emitterService.get_requests_count();
        assert.equal(emitterRequestsBefore, 0n, "Emitter state should be clean before everything");

        const listenerEventsBefore = await s.listenerService.get_events_received();
        const listenerBatchesBefore = await s.listenerService.get_batches_received();
        assert.equal(listenerEventsBefore, 0n, "Listener events state should be clean before everything");
        assert.equal(listenerBatchesBefore, 0n, "Listener batches state should be clean before everything");

        // sending 10 events each of 100 bytes of data
        for (let i = 0; i < 10; i++) {
            await s.emitterService.mirror(Array(100).fill(1));
        }

        // it should send all events in one batch
        const emitterRequestsAfter = await s.emitterService.get_requests_count();
        assert.equal(emitterRequestsAfter, 10n, "Emitter requests count should equal 10");

        // waiting for at least 10 seconds
        await delay(10_000);

        const listenerEventsAfter = await s.listenerService.get_events_received();
        assert.equal(listenerEventsAfter, 10n, "Listener events count should be equal to 10");

        const listenerBatchesAfter = await s.listenerService.get_batches_received();
        assert.equal(listenerBatchesAfter, 1n, "Listener batches count should be equal to 1");
    });
});

The test-case is pretty simple. First we check if the state of both canisters is clean. Then we call emitter.mirror() method 10 times. Then we wait for some time and check if:

the emitter received exactly 10 requests;
the listener received exactly 10 events in a single batch.

By the way, delay() function, that just asynchronously waits for some provided amount of time, is defined like this:

function delay(ms: number) {
    return new Promise(resolve => setTimeout(resolve, ms));
}

Trying it out

Now let's run our test-case. Start a local IC replica:

yarn start

Then (in a separate terminal window) build the project:

yarn build

Then run tests:

yarn test

This last command should deploy a new pair of canisters and run our test-case against them.

Afterword

In this tutorial we used ic-event-hub library in order to set an efficient communication channel between two canisters. It's important to mention, that this pattern could be easily applied to more complex scenarios with tens (or even hundreds) of interacting canisters.

Complete code of this tutorial can be found here.

Thanks for reading!

Introduction To ic-cron Library

Alexander — Tue, 01 Mar 2022 19:38:15 +0000

This tutorial is dedicated to Rust smart-contract (canister) development on Internet Computer (Dfinity) platform. Completing it, you’ll know how to use ic-cron library APIs in order to implement any background computation scenario within your IC dapp.

Before digging into this tutorial it is recommended to learn the basics of smart-contract development for the Internet Computer. Here are some good starting points: official website, dedicated to canister development on the IC; IC developer forum, where one could find an answer to almost any technical question.

This tutorial is intended for new developers who want to understand the basics of ic-cron. If you already finished it, it is recommended to check out my other tutorials about this library:

Motivation

Historically, smart-contracts were never able to run background tasks. On other blockchain platforms, where every network node machine stores every smart-contract there is just no way of implementing this functionality in a scalable way. But the Internet Computer (IC) works differently - the network is split into many small pieces called “subnetworks” each of which is responsible for only a small subset of smart-contracts (canisters) of the whole network. This system architecture really pushes the boundaries of what one can build on blockchain.

Developers often need to write programs, which perform some background computations. There are many examples of that: from notifications and reminders, to bots and email distribution software. Now it is possible to do any of them in smart-contracts, thanks to the IC.

Internet Computer provides canister developers with basic APIs for periodic arbitrary computations. ic-cron library lets developers to easily manage these background computations, extending Internet Computer basic APIs with scheduling functionality.

In this tutorial we’ll go through this functionality of ic-cron in more details, covering advantages of using this library comparing to standard API as well as some examples of how task scheduling with ic-cron works.

Heartbeat

The IC lets us describe any background computation by putting them into special function annotated with the heartbeat macro. The name of this macro is actually a clue on how it works - it is executed by the subnet node machines automatically once each consensus round (each block). One consensus round, at the moment of writing this article, lasts for about two seconds. This means that the heartbeat function of any canister is also executed once about each two seconds. This function is intended to be used somehow like this:

#[heartbeat]
fn tick() {
    // whatever you want to execute each consensus round
    // use `time()` system function, to check for current timestamp
}

An ability to do periodical work in smart-contract is a super-cool feature by itself, that opens a lot of possibilities which previously were completely out of reach for blockchain developers. But it’s pretty obvious that this solution won’t scale well in case of complex logic, when there are many different background computations running simultaneously. The code for such a logic will quickly turn into a hardly readable mess, each change to which will lead to countless hours of debugging. This is because there is no easy way for a developer to return an error from the heartbeat function (since no one makes a call to it - it triggers automatically).

By the way, never use caller() function inside a heartbeat annotated function! There is no caller, and you won’t see any error - you code will just fail silently.

To fix this scalability problem we only need two more things:

A task concept - background computation units separated from each other.
A task scheduler with simple APIs - some kind of service, which would let us manage the execution time of each task.

This is exactly what ic-cron library is.

ic-cron

In other words, ic-cron is just a task scheduler with some utilities. It queues all the tasks, a developer need to schedule, sorting by the soonest execution timestamp of these tasks. This queue is based on BinaryHeap Rust’s collection, that makes it work really fast and saves cycles.

Despite ic-cron being a library it is not a “pure library”, because it saves its data into canister’s state and uses other system APIs of the IC. Canisters on the IC can’t inherit other canister’s functionality (since they’re just wasm-programs), but we still can “inject” a functionality of one canister into another, “forcefully” implementing the inheritance pattern. And we have to do that with ic-cron in order to get good development experience.

This can be done via Rust’s macro system: ic-cron provides a special macro implement_cron!() that handles library state initialization and implements some utility functions to work with that state:

cron_enqueue() - adds a new task to the scheduler, putting it in execution queue;
cron_ready_tasks() - returns all the tasks which should be executed right now;
cron_dequeue() - removes a task from the scheduler, declining its further execution;
get_cron_state() - returns a reference to the scheduler’s complete state.

Let’s discuss these functions in more details.

Scheduling a executing of a task

cron_enqueue function is defined the following way:

pub fn cron_enqueue<Payload: CandidType>(
    payload: Payload,
    scheduling_interval: SchedulingInterval,
) -> candid::Result<TaskId>

It lets us schedule a new background task, by supplying any data our task will need at the moment of execution (any CandidType will work just fine) and configuring appropriate execution time settings.

Scheduled tasks should be processed inside the heartbeat function like that:

#[heartbeat]
fn tick() {
    for task in cron_ready_tasks() {
        // process the task however you want
    }
}

I.e. each time subnet node machines are executing the heartbeat function, we take all the tasks ready to be processed right at the moment and then process them.

In order to differentiate between task types one can use a Payload - some data, attached to the task at the moment of scheduling. This data can be anything you want. For example, one could use an enum:

enum CronTask {
    MakeCoffee(CoffeeType),
    PlayMusic(Song),
    CallPhone(Person),
}

#[heartbeat]
fn tick() {
    for task in cron_ready_tasks() {
        let task_type: CronTask = task.get_payload().expect("Unable to get a task type");

        match task_type {
            CronTask::MakeCoffee(coffee_type) => make_coffee(coffee_type),
            CronTask::PlayMusic(song) => play_music(song),
            CronTask::CallPhone(person) => call_phone(person),
        };
    }
}

It is possible to schedule tasks for a single (delayed) execution as well as for periodic execution. This is what iterations parameter of scheduling_interval argument does:

cron_enqueue(
    payload,
    SchedulingInterval {
        delay_nano: 100, // start after 100 nanoseconds
        interval_nano: 0, 
        iterations: Iterations::Exact(1), // this task will be executed only once
    },
).expect("Enqueue failed");

// or

cron_enqueue(
    payload,
    SchedulingInterval {
        delay_nano: 0, // start immediately
        interval_nano: 100, // waiting for 100 nanoseconds after each execution 
        iterations: Iterations::Infinite, // this task will be executed until descheduled manually
    },
).expect("Enqueue failed");

delay_nano and interval_nano parameters let us define the execution time very flexibly. Any option is possible: execute each day at 12 AM; execute each week starting from next tuesday; execute each 10 seconds; execute each 29th of February - just calculate these parameters correctly and you’re good.

Descheduling a task

cron_enqueue() function returns TaskId - a u64 task identifier that the scheduler uses internally. One could use this identifier in order to cancel previously scheduled task via cron_dequeue() function:

// schedules a task for background execution
let task_id = cron_enqueue(payload, scheduling_interval);

// deschedules the task, preventing its further background execution, until rescheduled again
cron_dequeue(task_id).expect("No such a task");

This function is intended to manually cancel a task that shouldn’t be executed anymore. For example, if you’re building an alarm-clock-canister, you’ll need an instrument to set the alarm off. cron_dequeue() is the instrument you would use for that.

Using the scheduler’s state

Also ic-cron gives us an ability to read (or write, if you know what you’re doing) current scheduler’s state by using get_cron_state() function. This function returns an object like:

#[derive(Default, CandidType, Deserialize, Clone)]
pub struct TaskScheduler {
    pub tasks: HashMap<TaskId, ScheduledTask>, // tasks by ids
    pub queue: TaskExecutionQueue, // task ids by timestamp of their next planned execution
}

You could use this object in different ways. For example you could get a list of all scheduled tasks in order of their execution time:

let tasks: Vec<ScheduledTask> = get_cron_state().get_tasks();

Another thing that could be done with the help of this function and set_cron_state() function is that one could persist the scheduler’s state in stable memory between canister upgrades:

#[ic_cdk_macros::pre_upgrade]
fn pre_upgrade_hook() {
    let cron_state = get_cron_state().clone();

    stable_save((cron_state,)).expect("Unable to save the state to stable memory");
}

#[ic_cdk_macros::post_upgrade]
fn post_upgrade_hook() {
    let (cron_state,): (TaskScheduler,) =
          stable_restore().expect("Unable to restore the state from stable memory");

    set_cron_state(cron_state);
}

Doing that way, even after an upgrade your canister will continue to execute scheduled tasks in the same order they should’ve been executed before the upgrade.

Afterword

Despite that ic-cron’s API is tiny, this library lets us define any background computation scenario we could imagine. It gives developers more than it takes back, by using efficient primitives under the hood, which adds only a small computational overhead, and by enabling some really complex functionality to be implemented without losing code readabilty.

And, by the way, it’s open source. Come take a look and give it a try!

https://github.com/seniorjoinu/ic-cron

Thanks for reading!

Tutorial: How To Build A Token With Recurrent Payments On The Internet Computer Using ic-cron Library

Alexander — Tue, 01 Mar 2022 19:19:16 +0000

This tutorial is dedicated to canister development on the Internet Computer (Dfinity) platform. Completing it you:

Would know some of advanced canister (smart-contract) development techniques on the Internet Computer using Rust programming language.
Would build your own token canister.
Would use ic-cron library in order to add recurrent payment mechanics to that token canister.

This tutorial is intended for advanced developers who already understand the basics of developing Rust canisters on the IC.

Before digging into it, it is recommended to recover the basics once again. Some good starting points are: the official website, dedicated to canister development on the IC; IC developer forum, where you can find an answer to almost any technical question.

Motivation

There are a lot of token standards out there. One of the most popular token standard on Ethereum is the famous ERC20. It was so successful, that many other people base their token standards on top of ERC20 (including other networks, like DIP20). It would be fair to say, that ERC20-like tokens are the most common in the wild on any network.

Despite this standard being so good and easy-to-use, it is pretty trivial in terms of functionality. Tokens are supposed to replace money, but there are some things you could do with money, but can’t do with tokens. For example, using web2 banking (classic money) you can easily “subscribe” to some service, making automatic periodical payments to it. This pattern is called “recurrent payments” and there is no such thing in web3. Yet.

In this tutorial, I’ll show you how to extend your token in order to add recurrent payment functionality to it using ic-cron library.

The complete source code for this tutorial can be found here:

https://github.com/seniorjoinu/ic-cron-recurrent-payments-example

Let’s go.

Project initialization

In order to proceed, make sure you have these tools installed on your machine:

dfx 0.9.0
rust 1.54+ and wasm32-unknown-unknown toolchain
ic-cdk-optimizer

You can use any file system layout for this project, but I will use this one:

- src
  - actor.rs              // canister API description
  - common                
     - mod.rs
         - currency_token.rs  // token internals
     - guards.rs          // guard canister functions
     - types.rs           // related types
- cargo.toml
- dfx.json
- build.sh         // canister buildscript
- can.did          // canister candid interface

First of all, we have to set up dependencies for the project, using the cargo.toml file:

// cargo.toml

[package]
name = "ic-cron-recurrent-payments-example"
version = "0.1.0"
edition = "2018"

[lib]
crate-type = ["cdylib"]
path = "src/actor.rs"

[dependencies]
ic-cdk = "0.3.3"
ic-cdk-macros = "0.3.3"
serde = "1.0"
ic-cron = "0.5.1"

Then we need to put this script inside the build.sh file that will build and optimize the wasm-module for us:

# build.sh

#!/usr/bin/env bash

cargo build --target wasm32-unknown-unknown --release --package ic-cron-recurrent-payments-example && \
 ic-cdk-optimizer ./target/wasm32-unknown-unknown/release/ic_cron_recurrent_payments_example.wasm -o ./target/wasm32-unknown-unknown/release/ic-cron-recurrent-payments-example-opt.wasm

And after that, we would fill the dfx.json file in order to describe the canister we are going to build:

// dfx.json

{
  "canisters": {
    "ic-cron-recurrent-payments-example": {
      "build": "./build.sh",
      "candid": "./can.did",
      "wasm": "./target/wasm32-unknown-unknown/release/ic-cron-recurrent-payments-example-opt.wasm",
      "type": "custom"
    }
  },
  "defaults": {
    "build": {
      "packtool": ""
    }
  },
  "dfx": "0.9.0",
  "networks": {
    "local": {
      "bind": "127.0.0.1:8000",
      "type": "ephemeral"
    }
  },
  "version": 1
}

Token internals

To better understand recurrent payments logic let’s make other parts of the token as simple as possible. Using that token canister users should be able to:

mint new tokens;
transfer tokens to a different account;
burn tokens;
check the balance of any account;
check the total supply of tokens;
check the token’s info (name, ticker/symbol, decimals).

Also we would add these new functionalities:

recurrent token minting;
recurrent token transferring;
an ability to check current user’s recurrent tasks and to cancel them.

Basic token functions

It is important to say, that we’re going to define these basic functions (token state and some functions to modify this state) without using any of Internet Computer APIs. This would allow us to test this code using only the Rust’s default testing framework.

We will add Internet Computer API calls (e.g. caller() function) later in the file named actor.rs.

Token state

First of all, let’s describe a data type for all the data that would be stored in the canister:

// src/common/currency_token.rs

pub struct CurrencyToken {
    pub balances: HashMap<Principal, u64>,
    pub total_supply: u64,
    pub info: TokenInfo,
    pub controllers: Controllers,
    pub recurrent_mint_tasks: HashSet<TaskId>,
    pub recurrent_transfer_tasks: HashMap<Principal, HashSet<TaskId>>,
}

// src/common/types.rs

#[derive(Clone, CandidType, Deserialize)]
pub struct TokenInfo {
    pub name: String,
    pub symbol: String,
    pub decimals: u8,
}

pub type Controllers = Vec<Principal>;
pub type TaskId = u64;

For token balances storage for each account we will use the balances property. Notice that it is a hashmap, the key of which is a Principal of the user.

Inside the total_supply property we’ll store total amount of tokens in circulation (total minted tokens minus total burned tokens).

Inside the info property we will store some basic information about our token - TokenInfo, which is a name, a ticker/symbol and an amount of decimals.

Inside the controllers property we’ll store a list of Principals of token administrators, which are users who can mint new tokens.

Also, in order to index ic-cron’s background tasks for efficient access, we would need a couple of additional properties: recurrent_mint_tasks and recurrent_transfer_tasks. We need these new indexes because ic-cron’s task scheduler stores all of its background tasks in a single collection that is optimized for quick scheduling. There are no other indexes there, so we would need to add these new ones.

Token minting

Let’s define a method to mint new tokens:

// src/common/currency_token.rs

impl CurrencyToken {

    ...

    pub fn mint(&mut self, to: Principal, qty: u64) -> Result<(), Error> {
        if qty == 0 {
            return Err(Error::ZeroQuantity);
        }

        let prev_balance = self.balance_of(&to);
        let new_balance = prev_balance + qty;

        self.total_supply += qty;
        self.balances.insert(to, new_balance);

        Ok(())
    }

    ...

}

This method takes as arguments an account identifier to mint tokens to and an amount of tokens that should be minted. Inside this method we just increment users current balance by the given amount and then update the total supply value.

The method returns (), if the operation was successful, or returns Error, if the value of qty argument equals to 0. Error type is defined the following way:

// src/common/types.rs

#[derive(Debug)]
pub enum Error {
    InsufficientBalance,
    ZeroQuantity,
    AccessDenied,
    ForbiddenOperation,
}

Token burning

The method to burn tokens works in exactly the opposite way, decrementing an amount of tokens from the account’s balance and updating the total supply counter:

// src/common/currency_token.rs

impl CurrencyToken {

    ...

    pub fn burn(&mut self, from: Principal, qty: u64) -> Result<(), Error> {
        if qty == 0 {
            return Err(Error::ZeroQuantity);
        }

        let prev_balance = self.balance_of(&from);

        if prev_balance < qty {
            return Err(Error::InsufficientBalance);
        }

        let new_balance = prev_balance - qty;

        if new_balance == 0 {
            self.balances.remove(&from);
        } else {
            self.balances.insert(from, new_balance);
        }

        self.total_supply -= qty;

        Ok(())
    }

    ...

}

This method takes an account identifier and an amount of tokens to burn as arguments. It returns (), if the operation went good, or an Error, if the value of qty argument is 0 or bigger than the account balance.

Notice that we’re using a specific pattern to write code: 1) check for arguments validity; 2) return any possible error; 3) change the state of the token.

It is always useful to follow this pattern in order to prevent future errors, including bugs which introduce a re-entrancy attack vulnerability.

Token transferring

Now let’s define a function to transfer tokens between accounts. This function will reduce an amount of tokens from one account and add them to another:

// src/common/currency_token.rs

impl CurrencyToken {

    ...

    pub fn transfer(&mut self, from: Principal, to: Principal, qty: u64) -> Result<(), Error> {
        if qty == 0 {
            return Err(Error::ZeroQuantity);
        }

        let prev_from_balance = self.balance_of(&from);
        let prev_to_balance = self.balance_of(&to);

        if prev_from_balance < qty {
            return Err(Error::InsufficientBalance);
        }

        let new_from_balance = prev_from_balance - qty;
        let new_to_balance = prev_to_balance + qty;

        if new_from_balance == 0 {
            self.balances.remove(&from);
        } else {
            self.balances.insert(from, new_from_balance);
        }

        self.balances.insert(to, new_to_balance);

        Ok(())
    }

    ...

}

This method takes account identifiers and an amount of tokens to transfer as arguments. In case of success it returns (), in case of the amount of tokens to transfer is 0 or exceeds current sender’s balance, this method returns an Error.

Getting balance

The balance getting function is pretty simple so we won’t discuss it in details:

// src/common/currency_token.rs

impl CurrencyToken {

    ...

    pub fn balance_of(&self, account_owner: &Principal) -> u64 {
        match self.balances.get(account_owner) {
            None => 0,
            Some(b) => *b,
        }
    }

    ...

}

Recurrent tasks management

As was previously mentioned, we have to maintain our own background task indexes in order to access them efficiently. Later you’ll see why, but now it is okay to have troubles with understanding of this move. It would be easier to start with recurrent mint tasks first:

// src/common/currency_token.rs

impl CurrencyToken {

    ... 

    pub fn register_recurrent_mint_task(&mut self, task_id: TaskId) {
        self.recurrent_mint_tasks.insert(task_id);
    }

    pub fn unregister_recurrent_mint_task(&mut self, task_id: TaskId) -> bool {
        self.recurrent_mint_tasks.remove(&task_id)
    }

    pub fn get_recurrent_mint_tasks(&self) -> Vec<TaskId> {
        self.recurrent_mint_tasks.iter().cloned().collect()
    }

    ...

}

So, all we want is to maintain a list of all recurrent minting tasks by their ids and to be able to return that list any time we need it.

It is a little bit harder with recurrent transfer tasks, because we need to maintain such a list of tasks for each account separately. To achieve that, we’ll use a mapping where to each account we map its own list of recurrent transfer tasks:

// src/common/currency_token.rs

impl CurrencyToken {

    ...

    pub fn register_recurrent_transfer_task(&mut self, from: Principal, task_id: TaskId) {
        match self.recurrent_transfer_tasks.entry(from) {
            Entry::Occupied(mut entry) => {
                entry.get_mut().insert(task_id);
            }
            Entry::Vacant(entry) => {
                let mut s = HashSet::new();
                s.insert(task_id);

                entry.insert(s);
            }
        };
    }

    pub fn unregister_recurrent_transfer_task(&mut self, from: Principal, task_id: TaskId) -> bool {
        match self.recurrent_transfer_tasks.get_mut(&from) {
            Some(tasks) => tasks.remove(&task_id),
            None => false,
        }
    }

    pub fn get_recurrent_transfer_tasks(&self, from: Principal) -> Vec<TaskId> {
        self.recurrent_transfer_tasks
            .get(&from)
            .map(|t| t.iter().cloned().collect::<Vec<_>>())
            .unwrap_or_default()
    }

    ...

}

Basic functionality testing

This is all we had to do to implement internal token functions. Let’s now use Rust’s standard testing framework to check if we did everything right.

First of all, for easy testing we would need a couple of utility functions:

// src/common/currency_token.rs

#[cfg(test)]
mod tests {

    ...

    pub fn random_principal_test() -> Principal {
        Principal::from_slice(
            &SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
                .to_be_bytes(),
        )
    }

    fn create_currency_token() -> (CurrencyToken, Principal) {
        let controller = random_principal_test();
        let token = CurrencyToken {
            balances: HashMap::new(),
            total_supply: 0,
            info: TokenInfo {
                name: String::from("test"),
                symbol: String::from("TST"),
                decimals: 8,
            },
            controllers: vec![controller],
            recurrent_mint_tasks: HashSet::new(),
            recurrent_transfer_tasks: HashMap::new(),
        };

        (token, controller)
    }

    ...

}

random_principal_test() function creates a unique account Principal identifier using current system time as a seed. create_currency_token() function creates a default token object filled with test data.

// src/common/currency_token.rs

#[cfg(test)]
mod tests {

    ...

    #[test]
    fn minting_works_right() {
        let (mut token, controller) = create_currency_token();
        let user_1 = random_principal_test();

        token.mint(user_1, 100).ok().unwrap();

        assert_eq!(token.total_supply, 100);
        assert_eq!(token.balances.len(), 1);
        assert_eq!(token.balances.get(&user_1).unwrap().clone(), 100);

        token.mint(controller, 200).ok().unwrap();

        assert_eq!(token.total_supply, 300);
        assert_eq!(token.balances.len(), 2);
        assert_eq!(token.balances.get(&user_1).unwrap().clone(), 100);
        assert_eq!(token.balances.get(&controller).unwrap().clone(), 200);
    }

    #[test]
    fn burning_works_fine() {
        let (mut token, _) = create_currency_token();
        let user_1 = random_principal_test();

        token.mint(user_1, 100).ok().unwrap();

        token.burn(user_1, 90).ok().unwrap();

        assert_eq!(token.balances.len(), 1);
        assert_eq!(token.balances.get(&user_1).unwrap().clone(), 10);
        assert_eq!(token.total_supply, 10);

        token.burn(user_1, 20).err().unwrap();

        token.burn(user_1, 10).ok().unwrap();

        assert!(token.balances.is_empty());
        assert!(token.balances.get(&user_1).is_none());
        assert_eq!(token.total_supply, 0);

        token.burn(user_1, 20).err().unwrap();
    }

    #[test]
    fn transfer_works_fine() {
        let (mut token, controller) = create_currency_token();
        let user_1 = random_principal_test();
        let user_2 = random_principal_test();

        token.mint(user_1, 1000).ok().unwrap();

        token.transfer(user_1, user_2, 100).ok().unwrap();

        assert_eq!(token.balances.len(), 2);
        assert_eq!(token.balances.get(&user_1).unwrap().clone(), 900);
        assert_eq!(token.balances.get(&user_2).unwrap().clone(), 100);
        assert_eq!(token.total_supply, 1000);

        token.transfer(user_1, user_2, 1000).err().unwrap();

        token.transfer(controller, user_2, 100).err().unwrap();

        token.transfer(user_2, user_1, 100).ok().unwrap();

        assert_eq!(token.balances.len(), 1);
        assert_eq!(token.balances.get(&user_1).unwrap().clone(), 1000);
        assert!(token.balances.get(&user_2).is_none());
        assert_eq!(token.total_supply, 1000);

        token.transfer(user_2, user_1, 1).err().unwrap();

        token.transfer(user_2, user_1, 0).err().unwrap();
    } 

    ...

}

We won’t stop on these test cases for too long. In each of them there is a series of checks which make sure that the state stays in a way it should stay no matter what function do we call. But tests are very important, because they give us confidence in our code. Never skip tests.

Token canister API

We’re almost at the finish line. All that’s left to do is:

to add a canister state initializing function;
to add token management functions using internal functions we wrote earlier;
to add recurrent mechanics into these token management functions.

State initialization

So, we need a function that will fill the canister state with some default values at the moment of canister creation:

// src/actor.rs

static mut STATE: Option<CurrencyToken> = None;

pub fn get_token() -> &'static mut CurrencyToken {
    unsafe { STATE.as_mut().unwrap() }
}

#[init]
fn init(controller: Principal, info: TokenInfo) {
    let token = CurrencyToken {
        balances: HashMap::new(),
        total_supply: 0,
        info,
        controllers: vec![controller],
        recurrent_mint_tasks: HashSet::new(),
        recurrent_transfer_tasks: HashMap::new(),
    };

    unsafe {
        STATE = Some(token);
    }
}

init() function takes as an argument a Principal of the controller - admin user that will be able to mint new tokens. And it takes as an argument an information about the token TokenInfo. Also here we have an utility function named get_token(), that returns a safe reference to the token’s state.

Token management

Let’s start with token minting:

// src/actor.rs

#[update(guard = "controller_guard")]
fn mint(to: Principal, qty: u64, scheduling_interval: Option<SchedulingInterval>) {
    match scheduling_interval {
        Some(interval) => {
            let task_id = cron_enqueue(
                CronTaskKind::RecurrentMint(RecurrentMintTask { to, qty }),
                interval,
            )
            .expect("Mint scheduling failed");

            get_token().register_recurrent_mint_task(task_id);
        }
        None => {
            get_token().mint(to, qty).expect("Minting failed");
        }
    }
}

Notice, that the update macro, annotating the function, also contains a guard function named controller_guard(), that will automatically check if the caller is a controller of the token, and will proceed to mint() function only if it’s so. Otherwise the user will be responded with an error.

This guard function is pretty simple and you can check it out in the Github repo in the src/common/guards.rs file, if you want to.

This function is able to perform recurrent mints as well as common ones. To make this happen we need to add an additional argument to this function to let a user pass some background task parameters to the function - scheduling_interval. SchedulingInterval type is a part of the ic-cron library and is defined the following way:

#[derive(Clone, Copy, CandidType, Deserialize)]
pub struct SchedulingInterval {
    pub delay_nano: u64,
    pub interval_nano: u64,
    pub iterations: Iterations,
}

#[derive(Clone, Copy, CandidType, Deserialize)]
pub enum Iterations {
    Infinite,
    Exact(u64),
}

This argument lets a user to define recurrent minting time settings and repetitions count.

If the value of the scheduling_interval argument is None, then the function will perform the common minting procedure calling CurrencyToken::mint() method of the state, otherwise it will schedule a new background task for a task scheduler using cron_enqueue() function, after which it will register this new background task in our recurrent minting tasks index.

cron_enqueue() function, as well as any other functionality from the ic-cron library, is only available after you use the implement_cron!() macros. So make sure you put it somewhere in your actor.rs file.

Recurrent minting tasks which we pass into the cron_enqueue() function, are defined the following way:

// src/common/types.rs

#[derive(CandidType, Deserialize, Debug)]
pub struct RecurrentMintTask {
    pub to: Principal,
    pub qty: u64,
}

Token transfer function is defined in a similar fashion:

// src/actor.rs

#[update]
fn transfer(to: Principal, qty: u64, scheduling_interval: Option<SchedulingInterval>) {
    let from = caller();

    match scheduling_interval {
        Some(interval) => {
            let task_id = cron_enqueue(
                CronTaskKind::RecurrentTransfer(RecurrentTransferTask { from, to, qty }),
                interval,
            )
            .expect("Transfer scheduling failed");

            get_token().register_recurrent_transfer_task(from, task_id);
        }
        None => {
            get_token()
                .transfer(from, to, qty)
                .expect("Transfer failed");
        }
    }
}

If the scheduling_interval argument is None, then this function performs a common token transfer calling CurrencyToken::transfer() method of the state. But if this argument has any payload inside, than the function schedules a recurrent transfer task to the task scheduler and saves this task into the index.

Recurrent transfer tasks are defined this way:

#[derive(CandidType, Deserialize, Debug)]
pub struct RecurrentTransferTask {
    pub from: Principal,
    pub to: Principal,
    pub qty: u64,
}

This data is everything our canister needs in order to call CurrencyToken::transfer() method later.

In order to differentiate between background task types we’ll use this enum:

#[derive(CandidType, Deserialize, Debug)]
pub enum CronTaskKind {
    RecurrentTransfer(RecurrentTransferTask),
    RecurrentMint(RecurrentMintTask),
}

Don’t worry if you still feel confused by now - we’re almost at the point where it all will make sense.

Token burning function is much more simpler, since there is no recurrent tasks in there:

// src/actor.rs

#[update]
fn burn(qty: u64) {
    get_token().burn(caller(), qty).expect("Burning failed");
}

Here and in the previous function we pass the caller() as from argument, because these actions can only be performed by the account owner.

Data getters are also pretty simple:

// src/actor.rs

#[query]
fn get_balance_of(account_owner: Principal) -> u64 {
    get_token().balance_of(&account_owner)
}

#[query]
fn get_total_supply() -> u64 {
    get_token().total_supply
}

#[query]
fn get_info() -> TokenInfo {
    get_token().info.clone()
}

Recurrent tasks

Now to the most interesting part of this tutorial. Inside mint() and transfer() functions we scheduled some tasks for the task scheduler. We want users of the token to be able to list their active tasks and to be able to cancel them.

Recurrent mint tasks getter and cancel functions are defined this way:

// src/actor.rs

#[update]
pub fn cancel_recurrent_mint_task(task_id: TaskId) -> bool {
    cron_dequeue(task_id).expect("Task id not found");
    get_token().unregister_recurrent_mint_task(task_id)
}

#[query(guard = "controller_guard")]
pub fn get_recurrent_mint_tasks() -> Vec<RecurrentMintTaskExt> {
    get_token()
        .get_recurrent_mint_tasks()
        .into_iter()
        .map(|task_id| {
            let task = get_cron_state().get_task_by_id(&task_id).unwrap();
            let kind: CronTaskKind = task
                .get_payload()
                .expect("Unable to decode a recurrent mint task");

            match kind {
                CronTaskKind::RecurrentTransfer(_) => trap("Invalid task kind"),
                CronTaskKind::RecurrentMint(mint_task) => RecurrentMintTaskExt {
                    task_id: task.id,
                    to: mint_task.to,
                    qty: mint_task.qty,
                    scheduled_at: task.scheduled_at,
                    rescheduled_at: task.rescheduled_at,
                    scheduling_interval: task.scheduling_interval,
                },
            }
        })
        .collect()
}

Task canceling is simple: we just remove the task from the task scheduler using cron_dequeue() function and then remove this same task from the index. If everything went good the caller will see a true response.

It is a little bit harder with recurrent tasks listing though. The CurrencyToken::get_recurrent_mint_tasks() function only returns task identifiers instead of task data (which is much more useful for the end user). To change that we need to define a new task type that will contain all the task data we want to show to the user:

// src/common/types.rs

#[derive(CandidType, Deserialize)]
pub struct RecurrentMintTaskExt {
    pub task_id: TaskId,
    pub to: Principal,
    pub qty: u64,
    pub scheduled_at: u64,
    pub rescheduled_at: Option<u64>,
    pub scheduling_interval: SchedulingInterval,
}

Besides a task identifier this type also contains this information:

to and qty - target account and an amount of tokens to mint;
scheduled_at - the task’s creation timestamp;
rescheduled_at - last minting timestamp;
scheduling_interval - background task time parameters provided by the user at the moment of the task creation.

So, we just map the list of TaskId into this new type using Iterator::map().

The same exact things we need to do in order to make recurrent transfers to work:

// src/actor.rs

#[update]
pub fn cancel_my_recurrent_transfer_task(task_id: TaskId) -> bool {
    cron_dequeue(task_id).expect("Task id not found");
    get_token().unregister_recurrent_transfer_task(caller(), task_id)
}

#[query]
pub fn get_my_recurrent_transfer_tasks() -> Vec<RecurrentTransferTaskExt> {
    get_token()
        .get_recurrent_transfer_tasks(caller())
        .into_iter()
        .map(|task_id| {
            let task = get_cron_state().get_task_by_id(&task_id).unwrap();
            let kind: CronTaskKind = task
                .get_payload()
                .expect("Unable to decode a recurrent transfer task");

            match kind {
                CronTaskKind::RecurrentMint(_) => trap("Invalid task kind"),
                CronTaskKind::RecurrentTransfer(transfer_task) => RecurrentTransferTaskExt {
                    task_id: task.id,
                    from: transfer_task.from,
                    to: transfer_task.to,
                    qty: transfer_task.qty,
                    scheduled_at: task.scheduled_at,
                    rescheduled_at: task.rescheduled_at,
                    scheduling_interval: task.scheduling_interval,
                },
            }
        })
        .collect()
}

For these background tasks we use the following type:

// src/common/types.rs

#[derive(CandidType, Deserialize)]
pub struct RecurrentTransferTaskExt {
    pub task_id: TaskId,
    pub from: Principal,
    pub to: Principal,
    pub qty: u64,
    pub scheduled_at: u64,
    pub rescheduled_at: Option<u64>,
    pub scheduling_interval: SchedulingInterval,
}

It differs from the RecurrentMintTaskExt only with one extra field - from, that defines the sender’s account.

Finally. This is the moment where everything should come together and start to make sense.

Now we need to describe how exactly we’ll execute these recurrent tasks. This should be done inside the special system function annotated with the heartbeat function:

// src/actor.rs

#[heartbeat]
pub fn tick() {
    let token = get_token();

    for task in cron_ready_tasks() {
        let kind: CronTaskKind = task.get_payload().expect("Unable to decode task payload");

        match kind {
            CronTaskKind::RecurrentMint(mint_task) => {
                token
                    .mint(mint_task.to, mint_task.qty)
                    .expect("Unable to perform scheduled mint");

                if let Iterations::Exact(n) = task.scheduling_interval.iterations {
                    if n == 1 {
                        token.unregister_recurrent_mint_task(task.id);
                    }
                };
            }
            CronTaskKind::RecurrentTransfer(transfer_task) => {
                token
                    .transfer(transfer_task.from, transfer_task.to, transfer_task.qty)
                    .expect("Unable to perform scheduled transfer");

                if let Iterations::Exact(n) = task.scheduling_interval.iterations {
                    if n == 1 {
                        token.unregister_recurrent_transfer_task(transfer_task.from, task.id);
                    }
                };
            }
        }
    }
}

In this function we iterate through the list of all tasks ready to be executed right now (supplied by the cron_ready_tasks() function). For each of these tasks we figure out its kind denoted by CronTaskKind enum, and then depending on that task kind we execute one of two following methods: CurrencyToken::mint() or CurrencyToken::transfer().

Notice, that inside this function we also need to check how many executions is left for the task. If there is only one execution left, we have to manually remove it from our recurrent task indexes.

Unfortunately, ic-cron library doesn’t support callback execution on certain events like the end of a task execution. This is why we have to manually watch for this event to happen and react to it in an imperative manner.

Candid interface description

This is it! We finished the coding part. All we left to do is to describe the interface of our canister inside the .did file, in order to be able to communicate with it from the console:

// can.did

type TaskId = nat64;

type Iterations = variant {
    Infinite;
    Exact : nat64;
};

type SchedulingInterval = record {
    delay_nano : nat64;
    interval_nano : nat64;
    iterations : Iterations;
};

type RecurrentTransferTaskExt = record {
    task_id : TaskId;
    from : principal;
    to : principal;
    qty : nat64;
    scheduled_at : nat64;
    rescheduled_at : opt nat64;
    scheduling_interval : SchedulingInterval;
};

type RecurrentMintTaskExt = record {
    task_id : TaskId;
    to : principal;
    qty : nat64;
    scheduled_at : nat64;
    rescheduled_at : opt nat64;
    scheduling_interval : SchedulingInterval;
};

type TokenInfo = record {
    name : text;
    symbol : text;
    decimals : nat8;
};

service : (principal, TokenInfo) -> {
    "mint" : (principal, nat64, opt SchedulingInterval) -> ();
    "transfer" : (principal, nat64, opt SchedulingInterval) -> ();
    "burn" : (nat64) -> ();
    "get_balance_of" : (principal) -> (nat64) query;
    "get_total_supply" : () -> (nat64) query;
    "get_info" : () -> (TokenInfo) query;

    "cancel_recurrent_mint_task" : (TaskId) -> (bool);
    "get_recurrent_mint_tasks" : () -> (vec RecurrentMintTaskExt) query;

    "cancel_my_recurrent_transfer_task" : (TaskId) -> (bool);
    "get_my_recurrent_transfer_tasks" : () -> (vec RecurrentTransferTaskExt) query;
}

Interacting with the token

Now let’s finally use the console to check whether it works or not. Let’s start a development environment in order to deploy our canister to it:

$ dfx start --clean

Then in a new console window:

$ dfx deploy --argument '(principal "<your-principal>", record { name = "Test token"; symbol = "TST"; decimals = 2 : nat8 })'
...
Deployed canisters

Instead of <your-principal> one needs to paste their own Principal which can be obtained with dfx identity get-principal command.

Okay, let’s mint us some tokens:

$ dfx canister call ic-cron-recurrent-payments-example mint '(principal "<your-principal>", 100000 : nat64, null )'
()

Checking the balance:

$ dfx canister call ic-cron-recurrent-payments-example get_balance_of '(principal "<your-principal>")'
(100_000 : nat64)

Nice, common token minting procedure works as expected. Let’s try the recurrent one then:

$ dfx canister call ic-cron-recurrent-payments-example mint '(principal "<your-principal>", 10_00 : nat64, opt record { delay_nano = 0 : nat64; interval_nano = 10_000_000_000 : nat64; iterations = variant { Exact = 5 : nat64 } } )'
()

Balance check:

$ dfx canister call ic-cron-recurrent-payments-example get_balance_of '(principal "<your-principal>")'
(102_000 : nat64)

Another balance check after a minute or so:

$ dfx canister call ic-cron-recurrent-payments-example get_balance_of '(principal "<your-principal>")'
(105_000 : nat64)

The balance won’t grow anymore, because we set the iterations count to 5, which means that we received 5000 tokens total in 5 portions of 1000 tokens each in intervals of ~10 seconds. It took us only ~40 seconds, because the first portion was received immediately after we created the task (delay_nano was 0).

So, it looks like recurrent token minting works fine. Let’s check recurrent token transferring now:

$ dfx canister call ic-cron-recurrent-payments-example transfer '(principal "aaaaa-aa", 1_00 : nat64, opt record { delay_nano = 0 : nat64; interval_nano = 10_000_000_000 : nat64; iterations = vari
ant { Infinite } } )'
()

Warning! Notice that we transfer tokens to the aaaaa-aa account. This is the Principal of so called management canister. Don’t ever transfer real money to that account - you’ll never get them back.

Let’s check our balance:

$ dfx canister call ic-cron-recurrent-payments-example get_balance_of '(principal "<your-principal>")'
(104_900 : nat64)

Checking again after some time:

$ dfx canister call ic-cron-recurrent-payments-example get_balance_of '(principal "<your-principal>")'
(104_600 : nat64)

Looks like it worked. We’re sure being charged by 100 tokens each 10 seconds, as we wanted. Let’s see our recurrent transfer tasks list:

$ dfx canister call ic-cron-recurrent-payments-example get_my_recurrent_transfer_tasks '()'
(
  vec {
    record {
      to = principal "aaaaa-aa";
      qty = 100 : nat64;
      task_id = 1 : nat64;
      from = principal "<your-principal>";
      scheduled_at = 1_645_660_528_445_056_278 : nat64;
      rescheduled_at = opt (1_645_660_528_445_056_278 : nat64);
      scheduling_interval = record {
        interval_nano = 10_000_000_000 : nat64;
        iterations = variant { Infinite };
        delay_nano = 0 : nat64;
      };
    };
  },
)

As we can see from the response currently we only have a single recurrent transfer task doing its job. Let’s cancel it:

$ dfx canister call ic-cron-recurrent-payments-example cancel_my_recurrent_transfer_task '(1 : nat64)'
(true)

Checking recurrent transfer tasks list again:

$ dfx canister call ic-cron-recurrent-payments-example get_my_recurrent_transfer_tasks '()'
(vec {})

Good, the task was removed from the list. Let’s check we’re no longer charged:

$ dfx canister call ic-cron-recurrent-payments-example get_balance_of '(principal "<your-principal>")'
(104_100 : nat64)

And after some time it stills the same:

$ dfx canister call ic-cron-recurrent-payments-example get_balance_of '(principal "<your-principal>")'
(104_100 : nat64)

Woohoo! It works!

Afterword

As you could see, it is possible to create some really cool things with the Internet Computer. Some things that don’t exist anywhere else yet, like we’re some kind of artists or so. Recurrent payments were not possible before the IC and ic-cron. There is no other web3 platform with such a feature so easy to use and to build on top of.

The Internet Computer is the new Internet. And this tutorial is just another proof of that statement.

My other tutorials on ic-cron:

Complete source code of this tutorial is here:

https://github.com/seniorjoinu/ic-cron-recurrent-payments-example

Thanks for reading!

Tutorial: Extending Sonic With Limit Orders Using ic-cron Library

Alexander — Sat, 05 Feb 2022 12:54:31 +0000

https://hackernoon.com/tutorial-extending-sonic-with-limit-orders-using-ic-cron-library

Cheap, secure and fully decentralized voting system on Ethereum

Alexander — Fri, 22 Jan 2021 12:35:17 +0000