Add result persistence

[desertaxle]

Summary

Adds automatic result persistence for task executions using py-key-value-aio storage, plus critical bug fixes for resource cleanup and improved test coverage.

✨ New Feature: Result Persistence

Tasks can now store return values and exceptions for later retrieval:

async def calculate() -> int:
    return 42

execution = await docket.add(calculate)()
await worker.run_until_finished()

# Retrieve result (waits if task still running)
result = await execution.get_result()  # Returns 42

Key Features

• Automatic serialization: Uses cloudpickle for any Python object
• Exception storage: Failed tasks store exceptions, which are re-raised on get_result()
• Smart skipping: Tasks returning None skip persistence
• TTL management: Results expire with execution_ttl (default: 1 hour)
• Timeout support: get_result(timeout=...) with graceful timeout handling
• Pub/sub integration: Waits for completion via existing state subscription

Implementation Details

• Docket.result_storage: RedisStore or MemoryStore backend
• Execution.result_key: Tracks where result is stored
• Execution.get_result(): Retrieves results, waiting via pub/sub if needed
• Worker captures return values/exceptions after task execution
• Base64-encoded JSON for storage compatibility

Storage Backend

Uses py-key-value-aio with pluggable storage:

• Redis: RedisStore for production (separate connection pool)
• Memory: MemoryStore for memory:// URLs (testing)

📚 API Changes

New Public Methods

execution.get_result(timeout=...)  # New method

Docket Configuration

Docket(
    result_storage=custom_storage  # Optional: provide custom AsyncKeyValue
)

Closes #166

---

🤖 Generated with Claude Code

[claude]

Claude finished @desertaxle's task —— View job

---

Code Review: Result Persistence PR

✅ Overall Assessment

This PR successfully implements result persistence for task executions with a well-designed API and comprehensive test coverage. The implementation is clean, follows existing patterns in the codebase, and maintains 100% test coverage as required.

🎯 Code Quality & Best Practices

Strengths:

• Clean API Design: The get_result() method provides an intuitive interface with proper async/await support and timeout handling
• Smart Storage Optimization: Skips persistence for None values to avoid unnecessary storage
• Proper TTL Management: Results expire along with execution records, preventing orphaned data
• Excellent Test Coverage: Comprehensive test suite covering edge cases, concurrent access, timeouts, and error conditions
• Good Separation of Concerns: Uses py-key-value-aio for pluggable storage backends

Minor Suggestions:

1. Line 739 in execution.py: The empty line could be removed as suggested in the review
2. Line 741 in execution.py: Consider adding # pragma: no branch as suggested for the async for loop to handle uncoverable branches

🔒 Security Considerations

Potential Concerns:

• Cloudpickle Usage: The use of cloudpickle for serializing arbitrary Python objects poses inherent security risks. Unpickling data from untrusted sources can lead to arbitrary code execution.

Mitigations in Place:

• Results are only stored and retrieved within the same trusted system
• Base64 encoding provides transport safety but not security
• TTL ensures data doesn't persist indefinitely

Recommendation: Consider documenting that this feature should only be used in trusted environments where all task results come from known sources.

⚡ Performance Considerations

Positives:

• Efficient Storage: Only stores non-None results, reducing storage overhead
• Pub/Sub for Waiting: Uses Redis pub/sub for efficient waiting rather than polling
• Connection Pooling: Properly uses connection pools for Redis operations

Potential Improvements:

1. Large Object Handling: Consider implementing size limits or compression for large results to prevent memory issues
2. Batch Operations: When multiple get_result() calls happen concurrently, they each create separate pub/sub connections - this is fine but worth monitoring in high-concurrency scenarios

🐛 Potential Issues

1. Resource Cleanup: The result_storage cleanup in __aexit__ looks correct, properly handling both BaseContextManagerStore and regular stores

2. Error Handling: The graceful fallback when result data is missing (returning None or generic exception) is well implemented

3. Race Conditions: The implementation properly handles the case where a task completes between checking state and subscribing to updates

📚 Documentation & Testing

Test Coverage Highlights:

• ✅ Various data types (int, str, dict, custom objects)
• ✅ None handling (both annotated and runtime)
• ✅ Exception storage and re-raising with preserved details
• ✅ Timeout behavior
• ✅ Concurrent access patterns
• ✅ TTL expiration
• ✅ Edge cases (missing data, malformed results)

Suggestion: The reviewer's request to copy the test to test_fundamentals.py has been addressed (line 739-752).

🔄 Backwards Compatibility

The implementation maintains backwards compatibility:

• Result persistence is automatic but doesn't break existing code
• Tasks returning None don't incur storage overhead
• The result_key field is optional in execution records

📝 Summary

This is a well-implemented feature that adds significant value to the Docket system. The code is production-ready with only minor cosmetic suggestions. The comprehensive test suite and proper error handling demonstrate thorough engineering.

Recommended Actions:

1. Address the minor line formatting suggestions from the reviewer
2. Consider adding documentation about security implications of cloudpickle
3. Monitor performance with large objects in production

Great work on this implementation! The feature is well-designed and thoroughly tested. 🎉

[github-actions]

📚 Documentation has been built for this PR!

You can download the documentation directly here:
https://github.com/chrisguidry/docket/actions/runs/19117058012/artifacts/4479942352

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 100.00%. Comparing base (4cfbdd3) to head (2d565ac).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##              main      #184    +/-   ##
==========================================
  Coverage   100.00%   100.00%            
==========================================
  Files           36        37     +1     
  Lines         6100      6400   +300     
  Branches       295       312    +17     
==========================================
+ Hits          6100      6400   +300     

Flag	Coverage Δ
python-3.10	100.00% <100.00%> (ø)
python-3.11	99.01% <100.00%> (+0.04%)	⬆️
python-3.12	100.00% <100.00%> (ø)
python-3.13	100.00% <100.00%> (ø)
python-3.14	100.00% <100.00%> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
src/docket/cli.py	100.00% <100.00%> (ø)
src/docket/docket.py	100.00% <100.00%> (ø)
src/docket/execution.py	100.00% <100.00%> (ø)
src/docket/worker.py	100.00% <100.00%> (ø)
tests/cli/test_snapshot.py	100.00% <100.00%> (ø)
tests/conftest.py	100.00% <100.00%> (ø)
tests/test_execution_progress.py	100.00% <100.00%> (ø)
tests/test_fundamentals.py	100.00% <100.00%> (ø)
tests/test_results.py	100.00% <100.00%> (ø)

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.