Skip to content
This page View Markdown Open in ChatGPT Open in Claude

Schedulers

A scheduler is your function on a timer, written in any language of API Maker. It runs every few seconds, every night at two, on the first of the month, or on any cron expression, in the time zone you pick. On a cluster of servers, each run happens once, on one server, and moves to another server when one goes down. Or it runs on every server at the same moment, and the runs share the work.

In short

  • A scheduler is your code on a timer : every minute, every night at two, or any cron.
  • Each run happens once, on one server, even when you have many servers.
  • Its time zone, its time limit and its next runs are on its page.
A scheduler : its name and timeout, its intervals, and its next runs
  1. Its name, how long it may run, and whether it is on.
  2. When it runs : every minute, or a cron.
  3. Its next runs, in its time zone.
Schedulers on a cluster : every interval runs once, on one server, spread over the servers, with failover schedulers of the account nightly-report 0 0 2 * * * · Asia/Kolkata sync-stock every 5 minutes · UTC send-reminders Mon-Fri 11:30 · Europe/Paris Redis knows which server runs each scheduler API Maker servers server 1 runs nightly-report server 2 runs sync-stock server 3 down · send-reminders moves Rules • each tick runs once, never on every server • spread over the servers • a server which goes down hands its jobs over • a deploy never breaks a run in progress • start, stop, run now from the admin panel
Schedulers: Cron Jobs Without a Cron Server13 chapters and the transcript

Create a scheduler

APIs & Logic → Schedulers, then Add New.

Field Meaning
Name Unique in the account. It names the folder in Git ; a rename is followed by the log profiles.
Timeout in minutes The run is stopped after this long, in the sandbox and on the native process alike. Empty : 10 minutes.
Runs on One server (each run happens once, on one server of the installation) or Every server, then Once per server or Once per runner : see Run on every server.
Code The function, with the global object g.
Intervals One or more, each with a time interval, a time zone and an active switch.
Active The master switch of the scheduler.
Run Native Run on the native process of the server, not in a sandbox.

The code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
import * as T from 'types';

async function main(g: T.IAMGlobal) {
    const before = Date.now() - 7 * 86400000;   // a week ago, in milliseconds since 1970 : what updated_at holds
    const removed = await g.sys.db.removeByQuery({
        instance: 'mongodb', database: 'shop', collection: 'carts',
        find: { updated_at: { $lt: before } },
    });
    g.logger.log(`${removed.deletedRowsCount} carts removed`);
    return { removed: removed.deletedRowsCount };
}
module.exports = main;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import time

from am import G


async def main(g: G):
    before = int(time.time() * 1000) - 7 * 86400000   # a week ago, in milliseconds since 1970 : what updated_at holds
    removed = await g.sys.db.remove_by_query('mongodb', 'shop', 'carts', find={'updated_at': {'$lt': before}})
    g.logger.log(f'{removed.deleted_rows_count} carts removed')
    return {'removed': removed.deleted_rows_count}

1
2
3
4
5
6
7
8
<?php
function main(AmG $g)
{
    $before = (int) (microtime(true) * 1000) - 7 * 86400000;   // a week ago, in milliseconds since 1970 : what updated_at holds
    $removed = $g->sys->db->removeByQuery('mongodb', 'shop', 'carts', find: ['updated_at' => ['$lt' => $before]]);
    $g->logger->log("{$removed->deletedRowsCount} carts removed");
    return ['removed' => $removed->deletedRowsCount];
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
file static class Api
{
    public static async Task<object?> Main(G g)
    {
        var before = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - 7 * 86400000L;   // a week ago, in milliseconds since 1970 : what updated_at holds
        var removed = await g.Sys.Db.RemoveByQuery(new RemoveByQuery("mongodb", "shop", "carts")
        {
            Find = new JsonObject { ["updated_at"] = new JsonObject { ["$lt"] = before } },
        });
        g.Logger.Log($"{removed.DeletedRowsCount} carts removed");
        return new { removed = removed.DeletedRowsCount };
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
package code

import (
    "fmt"
    "time"

    "apimaker/am"
)

func Main(g *am.G) (any, error) {
    before := time.Now().UnixMilli() - 7*86400000 // a week ago, in milliseconds since 1970 : what updated_at holds
    removed, err := g.Sys.Db.RemoveByQuery(am.RemoveByQuery{
        Instance: "mongodb", Database: "shop", Collection: "carts",
        Find: am.M{"updated_at": am.M{"$lt": before}},
    })
    if err != nil {
        return nil, err
    }
    g.Logger.Log(fmt.Sprintf("%d carts removed", removed.DeletedRowsCount))
    return am.M{"removed": removed.DeletedRowsCount}, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
use std::time::{SystemTime, UNIX_EPOCH};

use am::{Errors, G, RemoveByQuery, Value, json};

pub fn main(g: &mut G) -> Result<Value, Errors> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |since| since.as_millis() as i64);
    let before = now - 7 * 86_400_000; // a week ago, in milliseconds since 1970 : what updated_at holds
    let removed = g.sys.db.remove_by_query(RemoveByQuery::new("mongodb", "shop", "carts").find(json!({ "updated_at": { "$lt": before } })))?;
    g.logger.log(format!("{} carts removed", removed.deleted_rows_count));
    Ok(json!({ "removed": removed.deleted_rows_count }))
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import am.G;
import am.RemoveByQuery;
import am.RemoveByQueryResult;
import java.util.Map;

class Api {
    static Object main(G g) {
        long before = System.currentTimeMillis() - 7 * 86400000L;   // a week ago, in milliseconds since 1970 : what updated_at holds
        RemoveByQueryResult removed = g.sys.db.removeByQuery(new RemoveByQuery("mongodb", "shop", "carts")
            .find(Map.of("updated_at", Map.of("$lt", before))));
        g.logger.log(removed.getDeletedRowsCount() + " carts removed");
        return Map.of("removed", removed.getDeletedRowsCount());
    }
}

1
2
3
4
5
6
7
8
9
import am.G
import am.RemoveByQuery

fun main(g: G): Any? {
    val before = System.currentTimeMillis() - 7 * 86400000L   // a week ago, in milliseconds since 1970 : what updated_at holds
    val removed = g.sys.db.removeByQuery(RemoveByQuery("mongodb", "shop", "carts").find(mapOf("updated_at" to mapOf("\$lt" to before))))
    g.logger.log("${removed.deletedRowsCount} carts removed")
    return mapOf("removed" to removed.deletedRowsCount)
}
  • The same g as a custom API, without a request : g.req.body and g.req.query are empty. Keep state between runs in the cache : the next run may be on another server.
  • Pick the language in the title bar of the editor.

Intervals

A scheduler has one or more intervals. Each interval has its time, its time zone and its own active switch. A scheduler with two intervals runs at the moments of both.

The time of an interval

Group Choices
Seconds every second, 5, 10, 15, 30 seconds
Minutes every minute, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30 minutes
Hours every hour, 2, 3, 4, 6, 8, 12 hours
Days every day, 2, 3, 5, 10, 15 days
Working days Monday to Friday at 11:30
Months every month, 2, 3, 4, 6 months
Year every year
Custom a cron expression
  • The time zone of the interval says what "at 11:30" means : Asia/Kolkata, Europe/Paris, UTC…
  • Switch one interval off to pause it, and keep the others running.

Custom cron expressions

Six fields, seconds first : second minute hour day-of-month month day-of-week. Five fields work too (the seconds default to 0).

Field Values
second, minute 0-59
hour 0-23
day of the month 1-31
month 1-12, or jan-dec
day of the week 0-7 (0 and 7 are Sunday), or sun-sat

Each field takes *, a value, a range 1-5, a list 1,15 and a step */10. The presets @yearly, @monthly, @weekly, @daily, @hourly, @minutely, @secondly, @weekdays and @weekends stand for a whole expression. The panel checks an expression as you type it, by the same rules.

Expression Runs
0 0 22 * * 1-5 at 22:00, Monday to Friday
0 15 14 1 * * at 14:15 on the first of every month
*/30 * * * * * every 30 seconds
0 0 */6 * * * every 6 hours

On a cluster

A scheduler of one server (the default) :

  • Once per tick. Whatever the number of API Maker servers, an interval runs once : each scheduler belongs to one server, which holds it in Redis, and a scheduler already running is not started again.
  • Spread out. Every server knows the servers which are up (each one says so every few seconds), and the schedulers are spread over them evenly. A server which joins takes its share; the others keep theirs.
  • At once. A server starts its schedulers as soon as it starts.
  • Failover. When a server stops cleanly, it gives its schedulers back and the others take them at their next look, within five seconds ; when it dies, within about twenty seconds, when its hold runs out.
  • Which server. The Server column of the Schedulers page shows the server running each scheduler.
  • Safe deployments. A Git pull or a deployment does not interrupt a run in progress ; the next run uses the new code.

Run on every server

Schedulers on Every Server and Distributed Locks with Redis10 chapters and the transcript

Runs on : Every server runs the scheduler on every server of the installation at each moment of its intervals, for work every server does for itself (its own files, its own cache) or work the servers share. Runs per server says how many runs each server starts :

Runs per server At each moment
Once per server One run on each server.
Once per runner One run in each runner of the sandbox of the language of the code, on each server : as many runs as the server has processor cores (the runners a sandbox opens). The most work done at once. A runner which is not open yet opens, and runs the process initializers first. With Run Native : one run per server.

Every server starts the scheduler within a few seconds of a save, and a server which joins starts it too. The Server column of the Schedulers page and the Schedule panel of the form list the servers which run it ; each run on the panel says its server, and its runner.

Each run is told where it is among the runs of its moment, in g.req.scheduler :

Key Meaning
worker, workers This run among all the runs of the moment, from 0, and how many there are, on all the servers. The servers number the runs the same way, server after server in the order of their ids.
firedAt The moment the runs were started for, in milliseconds since 1970 : the same in every run of that moment, on every server.
serverIndex, servers This server among the servers which run the scheduler, and how many they are.
runnerIndex, runners The runner of the run on its server, and how many runs this server started.
serverId, serverName, name, runOn The server, the scheduler, and where it runs.

Share the work between the runs of a moment : each run takes its share by worker, and the lock of each item before it works on it, so an item a run of an earlier moment still works on is left alone.

Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import * as T from 'types';

async function main(g: T.IAMGlobal) {
    const { worker, workers } = g.req.scheduler;
    const orders = await g.sys.db.query({ instance: 'mysql_8', database: 'shop', collection: 'orders', find: { status: 'paid' }, limit: 500 });
    let packed = 0;
    for (const order of orders) {
        // Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
        if (order.id % workers !== worker) continue;
        // A run of an earlier moment may still work on an order : it holds its lock.
        const lock = await g.sys.system.getLock({ name: `order-${order.id}`, ttlSeconds: 120 });
        if (!lock.acquired) continue;
        try {
            await g.sys.db.updateById({ instance: 'mysql_8', database: 'shop', collection: 'orders', id: order.id, updateData: { status: 'packed' } });
            packed++;
        } finally {
            await g.sys.system.releaseLock(lock);
        }
    }
    return { worker, packed };
}
module.exports = main;
Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from am import G


async def main(g: G):
    worker, workers = g.req.scheduler.worker, g.req.scheduler.workers
    orders = await g.sys.db.query('mysql_8', 'shop', 'orders', find={'status': 'paid'}, limit=500)
    packed = 0
    for order in orders:
        # Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
        if order.id % workers != worker:
            continue
        # A run of an earlier moment may still work on an order : it holds its lock.
        lock = await g.sys.system.get_lock(f'order-{order.id}', ttl_seconds=120)
        if not lock.acquired:
            continue
        try:
            await g.sys.db.update_by_id('mysql_8', 'shop', 'orders', order.id, update_data={'status': 'packed'})
            packed += 1
        finally:
            await g.sys.system.release_lock(lock.token)
    return {'worker': worker, 'packed': packed}
Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
<?php
function main(AmG $g)
{
    $worker = $g->req->scheduler->worker;
    $workers = $g->req->scheduler->workers;
    $orders = $g->sys->db->query('mysql_8', 'shop', 'orders', find: ['status' => 'paid'], limit: 500);
    $packed = 0;
    foreach ($orders as $order) {
        // Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
        if ($order->id % $workers !== $worker) {
            continue;
        }
        // A run of an earlier moment may still work on an order : it holds its lock.
        $lock = $g->sys->system->getLock("order-{$order->id}", ttlSeconds: 120);
        if (!$lock->acquired) {
            continue;
        }
        try {
            $g->sys->db->updateById('mysql_8', 'shop', 'orders', $order->id, updateData: ['status' => 'packed']);
            $packed++;
        } finally {
            $g->sys->system->releaseLock($lock->token);
        }
    }
    return ['worker' => $worker, 'packed' => $packed];
}
Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
file static class Api
{
    public static async Task<object?> Main(G g)
    {
        var worker = g.Req.Scheduler.GetInt64("worker");
        var workers = g.Req.Scheduler.GetInt64("workers");
        var orders = await g.Sys.Db.Query(new Query("mysql_8", "shop", "orders") { Find = new JsonObject { ["status"] = "paid" }, Limit = 500 });
        var packed = 0;
        foreach (var order in orders)
        {
            var id = order.GetInt64("id");
            // Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
            if (id % workers != worker) continue;
            // A run of an earlier moment may still work on an order : it holds its lock.
            var held = await g.Sys.System.GetLock(new GetLock($"order-{id}") { TtlSeconds = 120 });
            if (!held.Acquired) continue;
            try
            {
                await g.Sys.Db.UpdateById(new UpdateById("mysql_8", "shop", "orders", id) { UpdateData = new JsonObject { ["status"] = "packed" } });
                packed++;
            }
            finally
            {
                await g.Sys.System.ReleaseLock(held.Token!);
            }
        }
        return new { worker, packed };
    }
}
Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package code

import (
    "fmt"

    "apimaker/am"
)

func Main(g *am.G) (any, error) {
    worker, workers := g.Req.Scheduler().Int("worker"), g.Req.Scheduler().Int("workers")
    orders, err := g.Sys.Db.Query(am.Query{Instance: "mysql_8", Database: "shop", Collection: "orders", Find: am.M{"status": "paid"}, Limit: 500})
    if err != nil {
        return nil, err
    }
    packed := 0
    for _, order := range orders {
        id := order.Int("id")
        // Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
        if id%workers != worker {
            continue
        }
        // A run of an earlier moment may still work on an order : it holds its lock.
        lock, err := g.Sys.System.GetLock(am.GetLock{Name: fmt.Sprintf("order-%d", id), TtlSeconds: 120})
        if err != nil {
            return nil, err
        }
        if !lock.Acquired {
            continue
        }
        _, err = g.Sys.Db.UpdateById(am.UpdateById{Instance: "mysql_8", Database: "shop", Collection: "orders", Id: id, UpdateData: am.M{"status": "packed"}})
        g.Sys.System.ReleaseLock(lock.Token)
        if err != nil {
            return nil, err
        }
        packed++
    }
    return am.M{"worker": worker, "packed": packed}, nil
}
Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
use am::{Errors, G, GetLock, Query, UpdateById, Value, json};

pub fn main(g: &mut G) -> Result<Value, Errors> {
    let worker = g.req.scheduler()["worker"].as_i64().unwrap_or_default();
    let workers = g.req.scheduler()["workers"].as_i64().unwrap_or(1);
    let orders = g.sys.db.query(Query::new("mysql_8", "shop", "orders").find(json!({ "status": "paid" })).limit(500))?;
    let mut packed = 0;
    for order in &orders {
        let id = order["id"].as_i64().unwrap_or_default();
        // Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
        if id % workers != worker {
            continue;
        }
        // A run of an earlier moment may still work on an order : it holds its lock.
        let lock = g.sys.system.get_lock(GetLock::new(format!("order-{id}")).ttl_seconds(120.0))?;
        if !lock.acquired {
            continue;
        }
        let updated = g.sys.db.update_by_id(UpdateById::new("mysql_8", "shop", "orders", id).update_data(json!({ "status": "packed" })));
        g.sys.system.release_lock(&lock.token)?;
        updated?;
        packed += 1;
    }
    Ok(json!({ "worker": worker, "packed": packed }))
}
Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import am.G;
import am.GetLock;
import am.JsonObject;
import am.Lock;
import am.Query;
import am.UpdateById;
import java.util.List;
import java.util.Map;

class Api {
    static Object main(G g) {
        long worker = g.req.getScheduler().getLong("worker");
        long workers = g.req.getScheduler().getLong("workers");
        List<JsonObject> orders = g.sys.db.query(new Query("mysql_8", "shop", "orders").find(Map.of("status", "paid")).limit(500));
        int packed = 0;
        for (JsonObject order : orders) {
            long id = order.getLong("id");
            // Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
            if (id % workers != worker) continue;
            // A run of an earlier moment may still work on an order : it holds its lock.
            Lock lock = g.sys.system.getLock(new GetLock("order-" + id).ttlSeconds(120));
            if (!lock.isAcquired()) continue;
            try {
                g.sys.db.updateById(new UpdateById("mysql_8", "shop", "orders", id).updateData(Map.of("status", "packed")));
                packed++;
            } finally {
                g.sys.system.releaseLock(lock.getToken());
            }
        }
        return Map.of("worker", worker, "packed", packed);
    }
}
Each run of the moment packs its share of the orders
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import am.G
import am.GetLock
import am.Query
import am.UpdateById

fun main(g: G): Any? {
    val worker = g.req.scheduler.getLong("worker")
    val workers = g.req.scheduler.getLong("workers")
    val orders = g.sys.db.query(Query("mysql_8", "shop", "orders").find(mapOf("status" to "paid")).limit(500))
    var packed = 0
    for (order in orders) {
        val id = order.getLong("id")
        // Every run of the moment takes its share : the orders whose number, divided by the runs, leaves its own.
        if (id % workers != worker) continue
        // A run of an earlier moment may still work on an order : it holds its lock.
        val lock = g.sys.system.getLock(GetLock("order-$id").ttlSeconds(120.0))
        if (!lock.isAcquired) continue
        try {
            g.sys.db.updateById(UpdateById("mysql_8", "shop", "orders", id).updateData(mapOf("status" to "packed")))
            packed++
        } finally {
            g.sys.system.releaseLock(lock.token)
        }
    }
    return mapOf("worker" to worker, "packed" to packed)
}
  • Once per moment, on all the servers : what must happen once (a summary, a mail), a run takes a lock named after the moment, report-${g.req.scheduler.firedAt}, with ttlSeconds longer than the work and no release : one run of all the servers gets it.
  • Overlapping moments : a moment does not wait for the runs of the one before. A run which may outlast the interval takes locks for its items, as above.
  • Without locks : worker and workers alone give each run its own items, as long as every run ends before the next moment.

Start, stop, run now

  • The Active switch starts and stops a scheduler right away, on every server : stop a misbehaving job in production in one click.
  • Run now from the API testing page (pick the scheduler, send) runs it once, outside its schedule, and shows the result and the logs.
  • The Schedulers page marks a scheduler running now on a node, and one whose intervals are all switched off : it never runs. Clone copies a scheduler.
  • Versions : keep several versions of the code, one active.

Good to know

  • A log profile records every run with its duration, and what it returned when Log Response is ticked for the scheduler.
  • Schedulers are files in Git (src/schedulers/<name>/) and deploy with a pull. Their active state and intervals travel with them.
  • A scheduler which needs more than its timeout should split the work : page through the rows and keep a cursor in the cache. Or run it on every server, in every runner, and share the rows between the runs.
  • No group opens a scheduler to an API user : a scheduler runs on its timer, or with Run now in the admin panel.