# core
All of pobblebonk is here, because
[honker](https://github.com/russellromney/honker) already has most of
it. honker brings the queue, the durable scheduler table, the streams
with a cursor per reader, the retries and the named locks, all in one
SQLite file and all as SQL functions.
What is left is what honker has not got, and it is three things. A pile
you add to over a week that a fire later reads. A gate that stops a fire
running when its pile is empty. And a callback, because honker fires a
job and something has to decide what a job means.
There is no daemon. `honker_scheduler_tick(now)` is a SQL function, so
`tick` is one call and a cron entry is the whole scheduler.
------------------------------------------------------------------------
source
### secs
``` python
def secs(
every
)->int:
```
*Whole seconds from `'30m'`, `'6h'`, `'1w'`, `'1h30m'`, or a number.*
``` python
test_eq(secs('30m'), 1800)
test_eq(secs('1h30m'), 5400)
test_eq(secs('2 days'), 172800)
test_eq(secs('2 years'), _MULT['y']*2)
test_eq(secs(90), 90)
test_fail(lambda: secs('soon'), contains='not a duration')
```
------------------------------------------------------------------------
source
### backoff
``` python
def backoff(
attempt:int, # failures so far; 1 on the first
base:float=30, # the first delay, in seconds
cap:float=3600, # longest a retry ever waits
)->float:
```
*Doubling delay: 30s, 60s, 120s, 240s, capped. A bot wall clears on its
own or it does not.*
``` python
test_eq([backoff(a) for a in (1, 2, 3, 4)], [30, 60, 120, 240])
test_eq(backoff(1, base=10), 10)
test_eq(backoff(99), 3600) # capped: a wall that has not cleared in an hour will not
test_eq(backoff(0), 30) # an attempt count of 0 is still the first delay
```
------------------------------------------------------------------------
source
### clip
``` python
def clip(
s, mx:int=20000
)->str:
```
*Text short enough to store, saying what it dropped rather than losing
it quietly.*
``` python
test_eq(clip('abcdef', 3), 'abc\n... [3 more chars]')
```
## Pob
------------------------------------------------------------------------
source
### Pob
``` python
def Pob(
path:NoneType=None, # the file; None makes a temporary one
db:NoneType=None, # or a honker database something else already opened
retries:int=5, # attempts a failing fire gets before it is dead-lettered
base:float=30, # seconds before the first retry; doubles per attempt
):
```
*A honker database, the one table honker has not got, and the callbacks
a fire runs.*
``` python
p = Pob()
test_eq(p.retries, RETRIES)
test_eq(p.base, float(BASE))
```
## The notes a person reads
Nothing is sent anywhere. An agent cannot start a turn of its own, so a
fire leaves a note and each place you work drains it on its next turn.
honker’s stream keeps a cursor per reader, so a phone and a terminal
each see everything once.
------------------------------------------------------------------------
source
### Pob.drain
``` python
def drain(
reader:str, limit:int=50
)->L:
```
*Every note this reader has not had. What a turn calls before it answers
you.*
------------------------------------------------------------------------
source
### Pob.notes
``` python
def notes(
limit:int=50
)->L:
```
*Every note, newest first, whatever any reader has seen.*
------------------------------------------------------------------------
source
### Pob.note
``` python
def note(
title:str, body:str='', **meta
):
```
*Leave one note.*
``` python
p_notes = Pob()
p_notes.note('first', 'one', kind='answer')
p_notes.note('second', 'two')
print('added notes: ', p_notes.notes())
# `notes` is an inspection view: newest first
test_eq(p_notes.notes().attrgot('title'), ['second', 'first'])
test_eq(p_notes.notes()[1].kind, 'answer')
# `drain` is a reader view: unread notes, oldest first
print('phone reader drain: ', p_notes.drain('phone'))
test_eq(p_notes.drain('phone'), [])
# each reader has a separate cursor
print('terminal reader drain:', p_notes.drain('terminal'))
test_eq(p_notes.drain('terminal'), [])
```
added notes: [{'title': 'second', 'body': 'two', 'offset': 2}, {'title': 'first', 'body': 'one', 'kind': 'answer', 'offset': 1}]
phone reader drain: [{'title': 'first', 'body': 'one', 'kind': 'answer', 'offset': 1}, {'title': 'second', 'body': 'two', 'offset': 2}]
terminal reader drain: [{'title': 'first', 'body': 'one', 'kind': 'answer', 'offset': 1}, {'title': 'second', 'body': 'two', 'offset': 2}]
------------------------------------------------------------------------
source
### Pob.\_\_repr\_\_
``` python
def __repr__():
```
*Return repr(self).*
------------------------------------------------------------------------
source
### Pob.all
``` python
def all()->L:
```
*Every schedule, with its payload parsed.*
------------------------------------------------------------------------
source
### Pob.due
``` python
def due()->int:
```
*Fires queued and not yet run.*
``` python
test_eq(repr(p), 'Pob(0 schedules, 0 due, 0 notes)')
```
## Schedules
A schedule is a durable instruction identified by `name`. `add` creates
it. `update` changes the fields supplied by the caller and preserves the
other fields. Payload changes merge with the stored payload. The
schedule name does not change.
------------------------------------------------------------------------
source
### Pob.update
``` python
def update(
name:str, # schedule to change
cron:str=None, # new cron expression
every:NoneType=None, # or a new duration
priority:int=None, # new queue priority
retries:int=None, # new attempt budget
**payload
)->AttrDict: # payload fields to add or replace
```
*Change an existing schedule without replacing fields that are not
given.*
------------------------------------------------------------------------
source
### Pob.drop
``` python
def drop(
name:str
)->bool:
```
*Call self as a function.*
------------------------------------------------------------------------
source
### Pob.resume
``` python
def resume(
name:str
)->bool:
```
*Call self as a function.*
------------------------------------------------------------------------
source
### Pob.pause
``` python
def pause(
name:str
)->bool:
```
*Call self as a function.*
------------------------------------------------------------------------
source
### Pob.get
``` python
def get(
name:str
)->AttrDict:
```
*Call self as a function.*
------------------------------------------------------------------------
source
### Pob.on
``` python
def on(
name:str
):
```
*Decorator naming the callback a fire of `name` runs.*
------------------------------------------------------------------------
source
### Pob.fire_at
``` python
def fire_at(
name:str, at:float
)->AttrDict:
```
*Move the next fire of `name`. The cadence picks every boundary after
it.*
------------------------------------------------------------------------
source
### Pob.add
``` python
def add(
name:str, # what you call it; unique, and what a note is titled with
cron:str=None, # a five-field cron expression
every:NoneType=None, # or a duration: `'30m'`, `'1w'`, or seconds
needs:str=None, # a list topic this cannot run without
catchup:str='once', # `once` runs the latest missed fire, `all` every one, an int that many
start:float=None, # epoch seconds for the first fire; None lets the cadence pick it
retries:int=None, # attempts a failing fire gets; None -> the Pob's own
**payload
)->AttrDict: # anything the callback wants
```
*Write down a standing instruction.*
------------------------------------------------------------------------
source
### catchup_n
``` python
def catchup_n(
v
)->int:
```
*Fires kept of the ones missed while the machine was off. 0 means keep
every one.*
``` python
p_sched = Pob()
added = p_sched.add('digest', every='1h', needs='inbox', catchup='once', retries=4, tone='brief')
test_eq(p_sched.get('digest').name, 'digest')
test_eq(len(p_sched.all()), 1)
test_eq(added.payload, {'schedule': 'digest','needs': 'inbox','catchup': 'once','tone': 'brief'})
changed = p_sched.update('digest',every='2h',priority=9,retries=7,catchup='all',tone='detailed')
test_eq(changed.cron_expr, every_s(secs('2h')).expr)
test_eq(changed.priority, 9)
test_eq(changed.max_attempts, 7)
test_eq(changed.payload, {'schedule': 'digest','needs': 'inbox','catchup': 'all','tone': 'detailed'})
```
## The lists a fire reads
Reading a list never empties it. Only the work that finished does that,
so a cart run that failed loses no milk.
------------------------------------------------------------------------
source
### Pob.items
``` python
def items(
topic:str, limit:int=200
)->L:
```
*The open items on `topic`, oldest first.*
------------------------------------------------------------------------
source
### Pob.push
``` python
def push(
topic:str, # which list
text:str, # the line the callback reads
key:str=None, # send the same thing twice and it stays one item
at:float=None
)->AttrDict:
```
*Add one item, or give back the open one that `key` already names.*
``` python
p_items = Pob()
milk = p_items.push('shopping', 'milk', at=20)
bread = p_items.push('shopping', 'bread', key='photo-1', at=10)
same_bread = p_items.push('shopping', 'bread again', key='photo-1', at=30)
p_items.push('reading', 'Dune', at=5)
```
``` python
{ 'at': 5.0,
'id': 3,
'key': None,
'text': 'Dune',
'topic': 'reading',
'used_at': None,
'used_by': None}
```
``` python
# `key` makes repeated open items idempotent
test_eq(same_bread.id, bread.id)
test_eq(same_bread.text, 'bread')
# items are separate by topic and ordered oldest first
test_eq(p_items.items('shopping').attrgot('text'), ['bread', 'milk'])
test_eq(p_items.items('reading').attrgot('text'),['Dune'])
# limit applies after ordering
test_eq(p_items.items('shopping', limit=1).attrgot('text'),['bread'])
```
`used` records that an item was consumed successfully. A used item no
longer appears in `items`. `used_by` is free-form provenance supplied by
the caller. Pobblebonk stores it for inspection. It does not validate
the value or use it for access control.
------------------------------------------------------------------------
source
### Pob.used
``` python
def used(
ids, by:str='', at:float=None
)->int:
```
*Mark items done, once the work that consumed them succeeded.*
``` python
# no IDs means no change
test_eq(p_items.used([]), 0)
# mark only bread as used
test_eq(p_items.used([bread.id], by='cart', at=100), 1)
test_eq(p_items.items('shopping').attrgot('text'), ['milk'])
used_bread = AttrDict(p_items.db.query('SELECT used_at, used_by FROM pob_items WHERE id=?',[bread.id])[0])
test_eq((used_bread.used_at, used_bread.used_by), (100.0, 'cart'))
# marking the same item twice changes nothing
test_eq(p_items.used([bread.id], by='other', at=200), 0)
# its old key is now free for a new open item
new_bread = p_items.push('shopping', 'fresh bread', key='photo-1', at=30)
test_ne(new_bread.id, bread.id)
test_eq(p_items.items('shopping').attrgot('text'),['milk', 'fresh bread'])
```
## The tick
One SQL call fires what is due and advances every `next_fire_at`. honker
enqueues a job for every missed boundary, so `once` keeps the latest of
them and cancels the rest.
------------------------------------------------------------------------
source
### superseded
``` python
def superseded(
fires, keep
)->set:
```
*Older job IDs for schedules that make up only their most recent missed
fires.*
``` python
fires = L([
{'name': 'digest', 'job_id': 1, 'fire_at': 10},
{'name': 'digest', 'job_id': 2, 'fire_at': 20},
{'name': 'digest', 'job_id': 3, 'fire_at': 30},
{'name': 'archive', 'job_id': 4, 'fire_at': 10},
{'name': 'archive', 'job_id': 5, 'fire_at': 20},
])
# `digest` runs once, so only its newest fire remains
test_eq(superseded(fires, {'digest'}), {1, 2})
# `archive` uses catchup='all', so none of its fires are superseded
test_eq(superseded(fires, set()), set())
# one fire has nothing older to replace
test_eq(superseded([{'name': 'digest', 'job_id': 1, 'fire_at': 10}], {'digest'}), set())
# a count keeps that many of the missed fires, newest last
test_eq(superseded(fires, {'digest': 2}), {1})
test_eq(superseded(fires, {'digest': 3}), set())
test_eq(superseded(fires, {'digest': 1, 'archive': 1}), {1, 2, 4})
```
------------------------------------------------------------------------
source
### Pob.tick
``` python
def tick(
at:float=None, # when this tick is happening; now when unset
worker:str='tick', # the lease holder recorded on each job
run:bool=True, # False queues the fires and runs none of them
)->AttrDict:
```
*Create due fires, apply catch-up policy, then run the remaining jobs.*
``` python
# A tick before the first boundary creates no jobs.
p_early = Pob()
p_early.add('early', every='1s')
first_fire = p_early.get('early').next_fire_at
first_fire
```
1787713032
``` python
early = p_early.tick(at=first_fire - 1, run=False)
test_eq(early.fired, [])
test_eq(early.cancelled, 0)
test_eq(early.ran, [])
```
``` python
# `once` keeps only the newest missed boundary.
p_once = Pob()
p_once.add('digest', every='1s', catchup='once')
first_fire = p_once.get('digest').next_fire_at
once = p_once.tick(at=first_fire + 2, run=False)
once
```
``` python
{ 'cancelled': 2,
'fired': [{'fire_at': 1787713034, 'job_id': 3, 'name': 'digest', 'queue': 'pob.fires'}],
'ran': []}
```
``` python
test_eq(len(once.fired), 1)
test_eq(once.cancelled, 2)
test_eq(once.fired[0].fire_at, first_fire + 2)
test_eq(once.ran, [])
```
``` python
# `all` keeps every missed boundary.
p_all = Pob()
p_all.add('archive', every='1s', catchup='all')
first_fire = p_all.get('archive').next_fire_at
first_fire
```
1787713032
``` python
all_fires = p_all.tick(at=first_fire + 2, run=False); all_fires
```
``` python
{ 'cancelled': 0,
'fired': [{'fire_at': 1787713032, 'job_id': 1, 'name': 'archive', 'queue': 'pob.fires'}, {'fire_at': 1787713033, 'job_id': 2, 'name': 'archive', 'queue': 'pob.fires'}, {'fire_at': 1787713034, 'job_id': 3, 'name': 'archive', 'queue': 'pob.fires'}],
'ran': []}
```
``` python
test_eq(len(all_fires.fired), 3)
test_eq(all_fires.cancelled, 0)
test_eq(all_fires.fired.attrgot('fire_at'),[first_fire, first_fire + 1, first_fire + 2])
test_eq(all_fires.ran, [])
```
A count between the two: make up that many missed boundaries and drop
the rest.
``` python
p_two = Pob()
p_two.add('digest', every='1s', catchup=2)
_f = p_two.get('digest').next_fire_at
two = p_two.tick(at=_f + 2, run=False)
test_eq(len(two.fired), 2)
test_eq(two.cancelled, 1)
test_eq(two.missed, {'digest': 1}) # which schedule lost a boundary, not just how many
test_eq(two.fired.attrgot('fire_at'), [_f + 1, _f + 2]) # the most recent two
test_fail(lambda: p_two.add('bad', every='1s', catchup=0), contains='positive int')
test_fail(lambda: p_two.add('bad', every='1s', catchup='some'), contains='positive int')
```
`start` sets the first fire. Without it the cadence picks the next
boundary from now.
``` python
p_start = Pob()
_at = int(time.time()) + 3600
test_eq(p_start.add('later', every='1h', start=_at).next_fire_at, _at)
test_eq(p_start.tick(at=_at - 1, run=False).fired, []) # nothing is due yet
test_eq(len(p_start.tick(at=_at, run=False).fired), 1) # and then it is
test_eq(p_start.fire_at('later', _at + 60).next_fire_at, _at + 60)
```
A claimed Honker `Job` does not expose the stored values for
`max_attempts`, `created_at`, or `run_at`. `_row` reads the live queue
row when callback or retry logic needs those values.
``` python
p_row = Pob(retries=7)
p_row.q.enqueue({'schedule': 'x'}, max_attempts=7, run_at=1_700_000_000)
job = p_row.q.claim_one('w')
test_eq((job.max_attempts, job.created_at, job.run_at), (3, 0, 0))
row = p_row._row(job)
test_eq(row.max_attempts, 7)
test_eq(row.run_at, 1_700_000_000)
test_eq(row.created_at > 0, True)
```
A callback failure spends one attempt. While the stored budget has
attempts left, `_failed` returns the job to the queue after a doubling
delay. The final failure dead-letters the job and leaves a note, because
it will never produce an ordinary answer.
``` python
p_retry = Pob(base=10)
p_retry.q.enqueue({'schedule': 'flaky'}, max_attempts=5)
job = p_retry.q.claim_one('worker')
row = p_retry._row(job)
retry = p_retry._failed(job, row, 'flaky', RuntimeError('bot wall')); retry
```
``` python
{ 'after': 10.0,
'attempt': 1,
'attempts_left': 4,
'error': 'RuntimeError: bot wall',
'name': 'flaky',
'status': 'retry'}
```
``` python
test_eq((retry.status, retry.error, retry.attempt, retry.attempts_left, retry.after), ('retry', 'RuntimeError: bot wall', 1, 4, 10.0))
test_eq(p_retry.notes(), [])
test_eq(p_retry.db.query('SELECT count(*) n FROM _honker_dead')[0]['n'], 0)
```
``` python
p_dead = Pob(base=10)
p_dead.q.enqueue({'schedule': 'flaky'}, max_attempts=1)
job = p_dead.q.claim_one('worker')
row = p_dead._row(job)
failed = p_dead._failed(job, row, 'flaky', ValueError('bad answer')); failed
```
``` python
{ 'attempt': 1,
'error': 'ValueError: bad answer',
'name': 'flaky',
'status': 'error'}
```
``` python
test_eq((failed.status, failed.error, failed.attempt), ('error', 'ValueError: bad answer', 1))
test_eq(p_dead.db.query('SELECT count(*) n FROM _honker_dead')[0]['n'], 1)
note = p_dead.notes()[0]
test_eq((note.title, note.dead, note.ref), ('flaky', True, job.id))
test_eq(note.body, 'gave up after 1 tries. ValueError: bad answer')
```
A successful callback has four effects: its result is stored, consumed
list items are closed, a note is published, and the job is acknowledged.
`_done` keeps those effects together.
``` python
p_done = Pob()
milk = p_done.push('shopping', 'milk')
job_id = p_done.q.enqueue({'schedule': 'cart'})
job = p_done.q.claim_one('worker')
got = p_done.items('shopping')
done = p_done._done(job, 'cart', got, {'added': 1})
```
``` python
test_eq(done, {'name': 'cart','status': 'ok','result': {'added': 1},'used': 1})
test_eq(p_done.items('shopping'), [])
note = p_done.notes()[0]
test_eq((note.title, note.body, note.ref, note.used),('cart', '{"added": 1}', job.id, 1))
```
Honker stores the schedule fields in a job payload, while the callback
also needs the actual fire time, its open list items, and the job ID.
`_fire` gathers those pieces into the object the callback receives.
``` python
p_fire = Pob()
item = p_fire.push('shopping', 'milk')
job_id = p_fire.q.enqueue({'schedule': 'cart', 'needs': 'shopping'}, run_at=100)
job = p_fire.q.claim_one('worker')
fire, row = p_fire._fire(job, at=200)
```
``` python
test_eq((fire.schedule, fire.needs), ('cart', 'shopping'))
test_eq(fire.list.attrgot('id'), [item.id])
test_eq((fire.job, fire.fire_at), (job.id, 100))
test_eq(row.id, job.id)
# An explicit scheduler boundary takes precedence over the stored run time.
fire, _ = p_fire._fire(job, at=200, fire_at=50)
test_eq(fire.fire_at, 50)
```
``` python
p_missing = Pob()
p_missing.q.enqueue({'schedule': 'nobody'})
job = p_missing.q.claim_one('worker')
missing = p_missing._run(job, at=100)
test_eq(missing.status, 'error')
test_eq(missing.error, "no callback registered for 'nobody'")
test_eq(p_missing.notes()[0].body, missing.error)
test_eq(p_missing.notes()[0].dead, True)
```
``` python
p_empty = Pob()
called = []
p_empty.on('cart')(lambda fire: called.append(fire) or 'done')
p_empty.q.enqueue({'schedule': 'cart', 'needs': 'shopping'})
job = p_empty.q.claim_one('worker')
empty = p_empty._run(job, at=100)
test_eq(empty.status, 'skipped')
test_eq(empty.why, 'the shopping list is empty')
test_eq(called, [])
test_eq(p_empty.notes(), [])
```
``` python
p_raise = Pob(base=10)
p_raise.on('cart')(lambda fire: 1/0)
p_raise.push('shopping', 'milk')
p_raise.q.enqueue({'schedule': 'cart', 'needs': 'shopping'}, max_attempts=3)
job = p_raise.q.claim_one('worker')
raised = p_raise._run(job, at=100); raised
```
``` python
{ 'after': 10.0,
'attempt': 1,
'attempts_left': 2,
'error': 'ZeroDivisionError: division by zero',
'name': 'cart',
'status': 'retry'}
```
``` python
test_eq((raised.status, raised.attempts_left, raised.after), ('retry', 2, 10.0))
test_eq(raised.error, 'ZeroDivisionError: division by zero')
test_eq(p_raise.items('shopping').attrgot('text'), ['milk'])
```
`work` drains ready fires from the queue. Each claim gives one worker
exclusive ownership of a job;
[`_run`](https://vedicreader.github.io/pobblebonk/heartbeat.html#_run)
then dispatches it to the callback named in its payload. `limit` bounds
one drain, and `when` preserves the scheduler boundary supplied by
`tick`.
------------------------------------------------------------------------
source
### Pob.work
``` python
def work(
worker:str='tick', at:float=None, limit:int=100, when:dict=None
)->L:
```
*Claim and run every fire that is due.*
``` python
p_work = Pob()
seen = []
@p_work.on('task')
def task(fire):
seen.append((fire.job, fire.fire_at))
return fire.job
job1 = p_work.q.enqueue({'schedule': 'task'})
job2 = p_work.q.enqueue({'schedule': 'task'})
when = {job1: 10, job2: 20}
```
``` python
# `limit` leaves the second ready job for another drain.
f = p_work.work('worker', at=100, limit=1, when=when); f
```
[{'name': 'task', 'status': 'ok', 'result': 1, 'used': 0}]
``` python
test_eq(f.attrgot('result'), [job1])
test_eq(seen, [(job1, 10)])
```
``` python
# The next drain claims what remains.
second = p_work.work('worker', at=100, when=when)
test_eq(second.attrgot('result'), [job2])
test_eq(seen, [(job1, 10), (job2, 20)])
# An empty queue returns immediately.
test_eq(p_work.work('worker', at=100), [])
```
## A callback, at a cadence
The headline. A real callback, a real one-second schedule, three real
ticks. `fire_at` is honker’s own boundary, so the gaps say whether the
cadence held.
``` python
import time
pob = Pob()
beats = []
@pob.on('heartbeat')
def beat(fire):
beats.append(fire.fire_at)
return f'beat {len(beats)}'
pob.add('heartbeat', every='1s')
first_fire = pob.get('heartbeat').next_fire_at
for at in range(first_fire, first_fire + 3): test_eq(pob.tick(at=at).ran.attrgot('status'), ['ok'])
test_eq(beats, [first_fire, first_fire + 1, first_fire + 2])
test_eq(pob.notes().attrgot('body'), ['beat 3', 'beat 2', 'beat 1'])
```
The answers reached the notes, and every reader gets them once:
``` python
test_eq(pob.drain('leela').attrgot('body'), ['beat 1', 'beat 2', 'beat 3'])
test_eq(pob.drain('leela'), []) # leela is up to date
test_eq(len(pob.drain('phone')), 3) # the phone has seen nothing yet
```
## A list a fire reads
The shopping flow. Push whenever you think of something, and the fire
reads the pile.
``` python
pob2 = Pob()
@pob2.on('cart')
def cart(fire):
return 'added: ' + ', '.join(i.text for i in fire.list)
pob2.add('cart', every='1s', needs='shopping')
pob2.push('shopping', 'yoghurt, the greek one', key='img_2201.heic')
pob2.push('shopping', 'yoghurt, the greek one', key='img_2201.heic') # the same photo twice
pob2.push('shopping', 'oat milk')
test_eq(pob2.items('shopping').attrgot('text'), ['yoghurt, the greek one', 'oat milk'])
```
`fire.list` is the pile, not `dict.items`. That collision cost a real
bug, so it is asserted:
``` python
time.sleep(1.05)
got = pob2.tick().ran[0]
test_eq(got.status, 'ok')
test_eq(got.result, 'added: yoghurt, the greek one, oat milk')
test_eq(got.used, 2)
test_eq(pob2.items('shopping'), []) # used, but only because the callback returned
```
A fire whose list emptied in the meantime is skipped, not failed. There
is nothing wrong and nothing to say:
``` python
time.sleep(1.05)
got = pob2.tick().ran[0]
test_eq(got.status, 'skipped')
test_eq(got.why, 'the shopping list is empty')
test_eq(len(pob2.notes()), 1) # and no second note
```
A schedule with no callback registered fails loudly rather than
vanishing:
``` python
pob4 = Pob()
pob4.add('nobody', every='1s')
time.sleep(1.05)
test_eq(pob4.tick().ran[0].error, "no callback registered for 'nobody'")
test_eq(pob4.notes()[0].body, "no callback registered for 'nobody'") # and it leaves a note
```
## Pausing, dropping, and what is refused
``` python
pob5 = Pob()
pob5.on('cart')(lambda fire: 'ok')
pob5.add('cart', every='1s')
test_eq(pob5.pause('cart'), True)
time.sleep(1.05)
test_eq(pob5.tick().ran, [])
test_eq(pob5.resume('cart'), True)
test_eq(pob5.drop('cart'), True)
test_eq(pob5.all(), [])
```
test_fail(lambda: pob5.add(‘bad’, every=‘1h’, cron=’0 17 \*‘),
contains=’one of’) test_fail(lambda: pob5.add(‘bad’), contains=‘one of’)
test_fail(lambda: pob5.add(‘bad’, every=‘1h’, catchup=‘maybe’),
contains=‘catchup must be’) test_fail(lambda: pob5.add(‘bad’,
cron=‘every friday’), contains=‘cron’)
## Sharing the file
Every method takes a honker database rather than opening one, so
pobblebonk can live in a file something else already has open. The fire
and the write that caused it commit together.
``` python
import honker
db = honker.open(str(Path(mkdtemp())/'app.db'))
with db.transaction() as tx:
tx.execute('CREATE TABLE docs (id TEXT PRIMARY KEY, text TEXT)')
pob6 = Pob(db=db)
pob6.on('index')(lambda fire: 'indexed')
pob6.add('index', every='1s')
test_eq(pob6.db is db, True)
test_eq(len(pob6.all()), 1)
```
``` python
test_fail(lambda: pob5.add('bad', every='1h', cron='0 17 * * *'), contains='one of')
test_fail(lambda: pob5.add('bad'), contains='one of')
test_fail(lambda: pob5.add('bad', every='1h', catchup='maybe'), contains='catchup must be')
test_fail(lambda: pob5.add('bad', cron='every friday'), contains='cron')
```