# heartbeat
There is no scheduler daemon.
[`Pob.tick`](https://vedicreader.github.io/pobblebonk/core.html#pob.tick)
runs only when an outside process calls it. This module installs that
recurring call in the machine scheduler.
The machine heartbeat knows only its interval. pobblebonk keeps schedule
times, pauses, retries, catch-up policy, and list requirements in
SQLite. Mirroring those values into the machine scheduler would create a
second schedule definition.
[`CronBeat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#cronbeat)
writes a crontab line.
[`LaunchdBeat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#launchdbeat)
writes a LaunchAgent.
[`TaskBeat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#taskbeat)
calls `schtasks`. They share the
[`Beat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#beat)
interface, and
[`beat()`](https://vedicreader.github.io/pobblebonk/heartbeat.html#beat)
selects the implementation for this machine.
Three machines, three schedulers.
[`CronBeat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#cronbeat)
writes a crontab line.
[`LaunchdBeat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#launchdbeat)
writes a LaunchAgent, which is what macOS actually wants.
[`TaskBeat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#taskbeat)
calls `schtasks`. They share the
[`Beat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#beat)
interface, and
[`beat()`](https://vedicreader.github.io/pobblebonk/heartbeat.html#beat)
picks.
One tag names everything: the launcher file, its log, and the job in
whichever scheduler. Change it and you have a second, independent
heartbeat, which is how one machine ticks two databases.
`every` is the heartbeat interval in seconds. It defaults to 60.
Portable intervals are whole minutes that divide an hour: 60, 120, 300,
600, 900, 1200, 1800, or 3600 seconds. cron and Task Scheduler set that
common limit.
------------------------------------------------------------------------
source
### every_mins
``` python
def every_mins(
every:int=60
)->int:
```
*Whole minutes between portable heartbeat runs.*
## The launcher
Every backend schedules one launcher file. cron, launchd, and `schtasks`
disagree about command quoting. `schtasks /tr` also limits command
length. A launcher gives each scheduler one path and gives the user a
command they can run directly.
`cmd` is a shell command such as
`uv run --directory ~/app python -m app.tick`. The launcher writes it
without escaping. `>>` appends each result to the log. `newline=''`
preserves `\r\n` in the Windows file. Without it, Python writes
`\r\r\n`, which `cmd` reads as a blank line.
------------------------------------------------------------------------
source
### launcher
``` python
def launcher(
cmd:str, # the command one beat runs, as you would type it
dirn:NoneType=None, # where to write it; `~/.pobblebonk` by default
tag:str='pobblebonk', # names this file and its log
win:bool=None, # write a Windows `.cmd`; this platform decides by default
)->Path:
```
*Write the one file every backend schedules, and give back its path.*
The point of writing a file rather than a command string is that the
file runs. So run it:
``` python
import subprocess
from tempfile import mkdtemp
d = Path(mkdtemp())
sh = launcher('/bin/echo ticked', d, tag='demo')
test_eq(sh.name, 'demo.sh')
test_eq(subprocess.run([str(sh)]).returncode, 0)
test_eq((d/'demo.log').read_text(), 'ticked\n')
```
``` python
# A second beat appends. A log holding only the most recent tick would be useless the one
# morning you go looking for what happened at 3am.
subprocess.run([str(sh)])
test_eq((d/'demo.log').read_text().count('ticked'), 2)
```
``` python
# The same command, told it is on Windows: a different file, and a shell that has never heard
# of `#!`.
cmd = launcher('py -m app.tick', d, tag='demo', win=True)
test_eq(cmd.name, 'demo.cmd')
test_eq(cmd.read_text().splitlines()[0], '@echo off')
```
## What a scheduler has to do
[`install`](https://vedicreader.github.io/pobblebonk/heartbeat.html#install)
replaces the job for a tag.
[`installed`](https://vedicreader.github.io/pobblebonk/heartbeat.html#installed)
returns its launcher path.
[`uninstall`](https://vedicreader.github.io/pobblebonk/heartbeat.html#uninstall)
removes it. Replacement prevents two processes from racing for the same
fires.
Each backend uses an operating-system identity: a crontab comment, a
plist filename, or `schtasks /f`.
------------------------------------------------------------------------
source
### Beat
``` python
def Beat(
*args, **kwargs
):
```
*One machine’s way of running a file at a portable interval.*
[`install`](https://vedicreader.github.io/pobblebonk/heartbeat.html#install)
replaces what `tag` installed before.
[`installed`](https://vedicreader.github.io/pobblebonk/heartbeat.html#installed)
reads back what is really scheduled, not what we hoped, so it is the
honest way to answer “is this thing on”.
## cron
A user has one crontab, which can contain unrelated jobs.
`python-crontab` finds this job by its trailing comment and preserves
the other entries. It is available through the `pobblebonk[cron]` extra.
`tabfile` redirects these operations to another crontab. The tests use a
temporary file.
------------------------------------------------------------------------
source
### CronBeat
``` python
def CronBeat(
tabfile:NoneType=None, # a crontab file to edit; None is the live user crontab
):
```
*A line in the user crontab. Linux, and anywhere without something
better.*
------------------------------------------------------------------------
source
### cron_line
``` python
def cron_line(
every:int=60
)->str:
```
*The portable cron expression for a heartbeat interval.*
A crontab with somebody’s nightly backup already in it. Everything below
is about that line surviving.
``` python
tabfile = d/'crontab'
tabfile.write_text('# nightly backup\n0 3 * * * /usr/bin/backup\n')
cron = CronBeat(tabfile)
cron.install('/home/k/.pobblebonk/pobblebonk.sh', every=120)
print(tabfile.read_text())
```
# nightly backup
0 3 * * * /usr/bin/backup
*/2 * * * * /home/k/.pobblebonk/pobblebonk.sh # pobblebonk
``` python
test_eq(cron.installed(), '/home/k/.pobblebonk/pobblebonk.sh')
assert '*/2 * * * * /home/k/.pobblebonk/pobblebonk.sh' in tabfile.read_text()
assert '0 3 * * * /usr/bin/backup' in tabfile.read_text()
```
``` python
# Installing again replaces the line rather than adding a second one.
cron.install('/home/k/.pobblebonk/pobblebonk.sh', every=120)
test_eq(tabfile.read_text().count('pobblebonk.sh'), 1)
```
``` python
# Removing ours removes only ours, and says whether there was anything there.
test_eq(cron.uninstall(), True)
test_eq(cron.installed(), None)
test_eq(cron.uninstall(), False)
assert '0 3 * * * /usr/bin/backup' in tabfile.read_text()
```
## launchd
macOS cron jobs run without Full Disk Access. A callback that reads
`~/Documents` can therefore fail. cron also drops times missed while the
machine sleeps. launchd runs one missed `StartInterval` after waking,
which lets `catchup` process missed fires.
A LaunchAgent is a plist in `~/Library/LaunchAgents`. Its filename is
its identity. Installation writes the plist, runs `bootout` for the
previous job, then runs `bootstrap` for the replacement.
[`_run`](https://vedicreader.github.io/pobblebonk/heartbeat.html#_run)
calls a scheduler binary. With `check=True`, a missing binary or
rejected command raises an exception. This prevents a failed
installation from appearing successful. Output is decoded here because
`schtasks` returns UTF-16.
------------------------------------------------------------------------
source
### LaunchdBeat
``` python
def LaunchdBeat(
dirn:NoneType=None, # where agents live; `~/Library/LaunchAgents` by default
):
```
*A LaunchAgent. What macOS wants, and it survives a sleeping laptop
where cron does not.*
------------------------------------------------------------------------
source
### plist
``` python
def plist(
script, # the launcher to run
tag:str='pobblebonk', # the launchd label, and the plist filename
every:int=60, # seconds between beats
)->bytes:
```
*The job description launchd reads: run `script` every `every` seconds.*
launchd rejects invalid plists and ignores unknown keys. The checks
parse the plist and inspect the launchd key names.
``` python
got = plistlib.loads(plist('/home/k/.pobblebonk/pobblebonk.sh', every=120))
test_eq(got['Label'], 'pobblebonk')
test_eq(got['ProgramArguments'], ['/home/k/.pobblebonk/pobblebonk.sh'])
test_eq(got['StartInterval'], 120)
```
Reading an agent back and removing it are plain file operations, so they
run anywhere. Only the `launchctl` call in
[`install`](https://vedicreader.github.io/pobblebonk/heartbeat.html#install)
needs a Mac, and it is the one line here that this notebook cannot
check.
``` python
agents = d/'LaunchAgents'
agents.mkdir()
(agents/'pobblebonk.plist').write_bytes(plist('/home/k/.pobblebonk/pobblebonk.sh'))
launchd = LaunchdBeat(agents)
test_eq(launchd.installed(), '/home/k/.pobblebonk/pobblebonk.sh')
test_eq(launchd.uninstall(), True)
test_eq(launchd.installed(), None)
test_eq(launchd.uninstall(), False)
```
## Task Scheduler
`schtasks /f` replaces a task with the same name. `/sc minute /mo N`
runs it every `N` minutes. The launcher path stays unquoted because
`subprocess` quotes arguments that contain spaces.
`/query /xml` reads the installed command without depending on the
display language. The XML is UTF-16, uses a namespace, and can quote the
command.
------------------------------------------------------------------------
source
### TaskBeat
``` python
def TaskBeat(
*args, **kwargs
):
```
*A Windows Task Scheduler entry, through `schtasks`.*
------------------------------------------------------------------------
source
### task_args
``` python
def task_args(
script, # the launcher to run
tag:str='pobblebonk', # the task name
every:int=60, # seconds between beats
)->list:
```
*The `schtasks` command that creates the task. `/f` is what makes a
second install replace the first.*
The Windows checks inspect the `schtasks` command without running it.
`/f` proves installation replaces the existing task.
``` python
args = task_args(r'C:\Users\k\.pobblebonk\pobblebonk.cmd', every=120)
assert '/f' in args
test_eq(args[args.index('/tn')+1], 'pobblebonk')
test_eq(args[args.index('/mo')+1], '2')
test_eq(args[-1], r'C:\Users\k\.pobblebonk\pobblebonk.cmd')
```
``` python
# What `schtasks /query /xml ONE` gives back: UTF-16, namespaced, and the command quoted whether
# or not it needed to be.
xml = '''
PT1M
"C:\\Users\\k\\.pobblebonk\\pobblebonk.cmd"
'''
test_eq(_task_command(_text(xml.encode('utf-16'))), 'C:\\Users\\k\\.pobblebonk\\pobblebonk.cmd')
```
## Installing one
[`beat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#beat)
selects the scheduler for the current platform.
[`install`](https://vedicreader.github.io/pobblebonk/heartbeat.html#install)
writes the launcher, schedules it, and returns its path.
`on` accepts a specific
[`Beat`](https://vedicreader.github.io/pobblebonk/heartbeat.html#beat).
Tests use it to exercise the public path without changing the machine
scheduler.
------------------------------------------------------------------------
source
### uninstall
``` python
def uninstall(
tag:str='pobblebonk', on:Beat=None
)->bool:
```
*Stop the beat. False if there was nothing scheduled.*
------------------------------------------------------------------------
source
### installed
``` python
def installed(
tag:str='pobblebonk', on:Beat=None
)->str:
```
*The launcher `tag` runs now, or None.*
------------------------------------------------------------------------
source
### install
``` python
def install(
cmd:str, # the command one beat runs
dirn:NoneType=None, # where the launcher goes; `~/.pobblebonk` by default
tag:str='pobblebonk', # names the launcher, its log, and the scheduled job
on:Beat=None, # a scheduler of your own; this machine's by default
every:int=60, # seconds between portable heartbeat runs
)->Path:
```
*Write the launcher for `cmd` and schedule it every `every` seconds.*
------------------------------------------------------------------------
source
### beat
``` python
def beat(
platform:str=None
)->Beat:
```
*The scheduler this machine actually has.*
``` python
test_eq(type(beat('linux')), CronBeat)
test_eq(type(beat('darwin')), LaunchdBeat)
test_eq(type(beat('win32')), TaskBeat)
```
The macOS smoke test installs a real LaunchAgent under
`pobblebonk-demo`. It uses the one-minute default so the check finishes
quickly. `/bin/date` appends each timestamp to
`~/.pobblebonk/pobblebonk-demo.log`. These cells change the machine
scheduler, so nbdev does not run them automatically.
``` python
def launchd_smoke(tag='pobblebonk-demo'):
import time
from datetime import datetime
on = LaunchdBeat()
script = HOME/f'{tag}.sh'
log = script.with_suffix('.log')
uninstall(tag, on=on)
script.unlink(missing_ok=True)
log.unlink(missing_ok=True)
try:
script = install('/bin/date', tag=tag, on=on)
test_eq(installed(tag, on=on), str(script))
for _ in range(75):
if log.exists() and (lines := log.read_text().splitlines()): break
time.sleep(1)
else: raise AssertionError(f'no timestamp reached {log}')
datetime.strptime(lines[-1], '%a %b %d %H:%M:%S %Z %Y')
return lines[-1]
finally:
uninstall(tag, on=on)
script.unlink(missing_ok=True)
log.unlink(missing_ok=True)
```
``` python
stamp = launchd_smoke()
print(stamp)
```
Sun Aug 30 13:29:52 AEST 2026
``` python
demo_tag = 'pobblebonk-demo'
demo = LaunchdBeat()
test_eq(installed(demo_tag, on=demo), None)
test_eq(demo._p(demo_tag).exists(), False)
test_eq((HOME/f'{demo_tag}.sh').exists(), False)
test_eq((HOME/f'{demo_tag}.log').exists(), False)
```
## What this does not do
This module does not run a tick or interpret schedules. It installs one
launcher that calls the tick at the configured heartbeat interval.
Schedule timing stays in the database for `pause`, `retries`, and
`catchup`.
Two ticks can overlap when a callback takes longer than the heartbeat
interval. The heartbeat does not prevent this. A database lock must
serialize ticks on every supported platform.