Task progress meters

[chrisguidry]

As an extension to the timeout idea in #87, a similar interface could be used to express intermediate progress (visible on the snapshot or programmatically):

async def long_running(
    progress: Progress = Progress()
) -> None:
    batch = get_some_work()
    progress.set_total(len(batch))
    for item in batch:
        do_some_work(item)
        progress.increment()  # 1 by default, but could be other values

This could be asynchronously incrementing an ephemeral redis key to show progress for this specific task.

[desertaxle]

To unify this and #166, I think we'll want to introduce some lightweight state tracking. We won't need to introduce state for this issue quite yet, but I'm keeping it in mind as I think through it.

Here's how I'm thinking of implementing this:

Core Components:

• ProgressInfo dataclass:
  • current: int (current progress value)
  • total: int (defaults to 100)
  • completed_at: datetime (when the task finished)
• ProgressStore class for Redis operations:
  • set_progress(key: str, current: int, total: int | None) - Set progress values
  • increment_progress(key: str, amount: int, total: int | None) - Atomically increment
  • get_progress(key: str) -> ProgressInfo | None - Retrieve progress
  • delete_progress(key: str) - Remove progress data
  • mark_completed(key: str, ttl: int) - Mark progress as completed
  • Redis key format: {docket}:progress:{key}
  • Key TTL will be configurable on a Docket instance
    • Keeping the record will help provide a record of completed tasks need for task results
    • Plan to reuse the TTL for state records
  • Uses Redis hash with fields: current, total

Add new Progress class following the Timeout pattern:

 class Progress(Dependency):
     """Allows a task to report intermediate progress during execution."""

     single: bool = True

     def __init__(self) -> None:
         pass

     async def __aenter__(self) -> "Progress":
         # Get execution context
         # Initialize with Redis connection
         # Return Progress instance with task key

     def set_total(self, total: int) -> None:
         """Set the total expected progress value."""

     def increment(self, amount: int = 1) -> None:
         """Increment progress by the given amount (default 1)."""

     def set(self, current: int) -> None:
         """Set the current progress value directly."""

     def get(self) -> ProgressInfo | None:
         """Get current progress info."""

Updates to Worker

• Create a progress record before execution if it doesn't already exist
• Set progress to 0 if record already exists (i.e. when a task is being retried)
• Mark the task as completed when execution is successful (i.e. set current == total)
• Don't mark the task as completed after execution failure and keep current value

[chrisguidry]

(sorry, hit the wrong button hahaha ^)

[chrisguidry]

I'm mulling over the interface here and I think it's the "Info" suffix that bugs me. I'm wondering if we can unify this down to a single Progress class that is both the dependency and the object?

So you'd have:

async def record_all_the_things(progress: Progress = Progress()) -> None:
    things = db.get_the_things()
    progress.total = len(things)
    for thing in things:
        progress.increment()
        if progress.current % 10 == 0:
            logger.info("Recorded %s/%s things!", progress.current, progress.total)


...


progress = await docket.get_progress(task_key)
logger.info("Recorded %s/%s things!", progress.current, progress.total)

I guess one problem there is that there are some unnecessary mutation methods available on what should be a read-only view of Progress that get_progress would return.