# Scheduler

## The requirements

The demand of scheduler comes from multiple sources. Including automated Offchain Rollup job (PW Game, Request-Response pattern), workflow engine, and more. Phat Contract queries can be registered in a scheduler with cron-style expression, and the scheduler takes care of the following things:

- **Scheduling**: Parse and trigger tasks defined by cron-style expression
- **Accuracy**: Ideally the fire of a task should happen exactly once on a worker. However due to technical limitations, we may offer best-effort trigger accuracy. In other worlds, minimize skips and redundant trigger.
- **Scalability**: The scheduler should offload the execution of the triggered tasks on remote workers, instead of running locally.
- **Realiability**: The scheduler itself shouldn't be the single point of failure.

## Simple scheduler

Let's start from the easiest case. Suppose there's only one worker and very limited amount of the jobs. The scheduler saves the job definitions to the contract storage. The job is defined by a cron expression, the target contract address, and the call data. Each time the scheduler wakes up, it should go through all the jobs and trigger the job that's ready to fire. When the scheduler finish one round of the check, it becomes idle until the next wake-up.

The first problem is how to decide if a job should be triggered or not. In a typical backend service, the scheduler can simply calculate the next fire time of a task based on the cron expression and the current time, store it somewhere, and when it passes, the trigger fires once. It can be described as the following pseudocode:

// Parse an cron expression with `saffron`
let cron: Cron = "*/10 0 * OCT MON".parse().unwrap();
let mut next = cron.next_from(Utc::now()).unwrap();

loop {
    // If the deadline passes, trigger the task
    if Utc::now() > next {
        fire();
        next = cron.next_from(Utc::now()).unwrap();
    }
    sleep(1);
}

However, in Phat Contract the scheduler should be run as a query. Queries are event driven without the support of long running tasks or contract storage. The sleep loop can be replaced by a `poll()` query to wake up the scheduler. The `poll()` function will need to know and update the next fire time. Without the access to the contract storage, it needs to either store the time in an external storage (e.g. S3), or the in-memory cache offered by Phat Contract.

The Phat Contract cache is a simple KV-store resides in the worker's memory provided to every contract. It's volatile and never sync betwee the workers. So there's no guarantee of data availability in case of worker shutdown. It cannot be used to store any serious, but for the case of a simple scheduler, it's already good enough to store the next fire time:

- If the worker hosting the scheduler runs smoothly, the scheduler should run as expected
- If the cache lost happens, it simply re-calculates the next fire time. Considering the following cases:
1. If the data lost happened before the fire time, nothing is lost
2. If the data lost happened right between two `poll()` calls around the fire time, the task will be skipped once
3. If the data lost happen after the `poll()` call following the fire time, the fire time should have been updated. So it's equivalent to the first case

The cache data lose only happens occasionally. So we can concludes that if the `poll()` happens more frequently compared with the cron job fires, the missed fire should be very rare. So we can say it follows the **Accuracy** principle.

The pseodocode can be:

fn register(&mut self, cron_expr: String, target: Target) {
    let cron: Cron = cron_expr.parse().unwrap();
    self.job = Some((cron, target));
}

fn poll(&self) {
    const KEY_NEXT: &str = "next";
    let (cron, target) = self.job?;
    
    if let Some(cached_next) = ext().cache_get(KEY_NEXT) {
        let next = DateTime::decode(cached_next).unwrap();
        if Utc::now() > next {
            // Trigger if it passes the fire time
            target.fire();
        } else {
            // Otherwise, wait until next poll
            return
        }
    }
    // Update the next fire time anyway
    let next = cron.next_from(Utc::now()); 
    ext().cache_set(KEY_NEXT, next.encode());
}

Reference implementation: <https://github.com/Phala-Network/phat-stateful-rollup/blob/main/phat/contracts/local_scheduler/lib.rs>

## Offload task to remote workers

The simple scheduler achieves **Scheduling**, and (best-effort) **Accuracy**. It can easily scale from a single job to a few, but to achieve **Scalability**, we should move the execution of the tasks to remote workers to avoid computation bottleneck. By offloading the execution, a single thread scheduler is supposed to scheduler thousands of jobs.

The remote worker-to-worker call is done via simple HTTP requests. It implies the scheduler should be able to access the execution workers via the internet by their publicly registered RPC endpoints. Since the scheduler could trigger up to thousands of concurrent HTTP requests, the ink! contract model doesn't work anymore, because the HTTP request in ink! is blocking operation. Instead, the scheduler should run as a SideVM program.

By moving to remote execution, two problems raise:

- How to distribute the tasks smartly?
- Remote execute could fail. How to ensure **Accuracy**?

### Task distribution

A good task distribution mechanism should take full use of all the available workers. In other words, it should minimize the computation on the hot spot worker (the worker with the largest workload). It assumes the task can be executed on any worker, and thus the execution should be stateless. (There are well-known tricks to convert stateful application to stateless services.)

A simple strategy is to randomly distribute the tasks. It can be further optimized by considering the actual invocation or running time on each worker. In addition, the job assignment may be persisted or cached locally, because some job may still prefer to stick to a worker to take use of its local cache (e.g. an IPFS data fetcher). In case some workers go down, after a few failure reported from that worker, it can be temporarily removed from the worker pool. Health check can also be used to proactively maintain the worker pool.

### Idempotence

Remote execution could fail due to the internet packet lose or even the lose of the worker. In such case, it must be possible to retry the tasks. Here's a potential strategy: Once the scheduler gets any failed fire, it can switch to trigger the task on another worker. It keeps retrying until making the first successful response. However, this strategy may introduce unexpected redundant call. For example, in the first try the HTTP request was sent and the task was executed remotely, but due to the unstable network, the response was lost, and the scheduler starts another try since it doesn't know the execution was done. This is a [typical problem](https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/#cron-job-limitations) in all schedulers, and can be avoided by an **idempotent design**.

By definition, if all the tasks are idempotent, the retry will not cause any damage. In practice, each execution of a job can be assigned with an unique id (i.e. `task_id`). The `task_id` can be a self-incremental id that increases by time. With the `task_id`, the target can deduplicate the query request from the scheduler. The handler can store the recent `task_id` in its local cache, and drop any future call with the same `task_id`. For the app with very strong consistency requirement, it can persist the invocation to external database or even a blockchain (e.g. OffchainRollup) to make it strongly idempotent.

Thousands of jobs is a reasonable estimation of the capacity of a single-worker scheduler. The accurate capacity must be measured by real world benchmark. To handle an even larger scale of the jobs, we may need to create multiple schedulers.

## High availability

Finally, the **Reliability** of the system requires the scheduler itself must be redundant. This is a typical High Availability scenario and there are a lot of well-studied solutions in the distributed system industry. The scheduler itself has to be either fully replicated, or be hosted by an elected master.

If the tasks are strongly idempotent, the full replication approach could work but with additional network overhead. For weakly idempotent tasks, full replication requires the jobs should mostly distributed by the fixed worker so that they can deduplicate the tasks with their local cache.

Mainstream solutions all point to the election-based method. A cluster of nodes governed by a consensus algorithm (e.g. Raft) can be used to elect a master, and only let the master to run the scheduler. However, implementing a full consensus on top of Phat Contract is still a great amount of the work to do.

In practice, the two approaches can be combined to build a minimum-centralized scheduler. Instead of running a real high availability cluster, some off-chain HA cluster can be used to monitor a pool of workers, and choose one worker to call the scheduler's `poll()` entry. Since the HA cluster is centralized, theoretically it can duplicate the poll query maliciously. However thanks to the idempotent requirement of the cron tasks, the jobs themselves can deduplicate the call, and thus limit the damage to the system.

## Conclusion

This docs presents how to design a Phat Contract scheduler that can ensure four requirements. It will be build in SideVM, trigger the tasks on remote workers with idempotency considered to achieve high accuracy, and use an external trigger to achieve high availability.