Add support for checkpoints

[joefreeman]

Checkpoints provide a way to persist state for a step, which can be retrieved for subsequent step executions. This is useful for storing cursors.

A checkpoint is defined like this:

cursor = cf.Checkpoint("cursor", default=0)

And then within a task/workflow can be get and set:

since = cursor.get()
# ...
cursor.set(next_since)

Cursors get flushed periodically by the worker, but cf.flush() can be called to explicitly flush cursors (and metrics/logs).