refactor: drop cross-process lease; single-process session store - #5
Merged
Conversation
Decouple the /generate handler from blocking receipt processing. A correct password now enqueues a Job and replies with the user's position in line; a pool of WORKER_COUNT background workers drains the queue, acquiring the per-user lock + cross-process lease, running the pipeline, and delivering the PDF asynchronously. - Add JobQueue (in-process FIFO, bounded workers, position reporting) - Add BotState.QUEUED; guard it in message/clear/cancel/generate handlers - Refactor _run_generation -> _process_job(job) using chat_id messaging - Add TelegramService.send_message + best-effort _notify helper - Acquire lease in the worker; always release it in finally (no leak) - Reset QUEUED sessions to IDLE on startup (queue is in-memory) - Add WORKER_COUNT config (default 2)
Each batch keeps raw + normalized images on disk until the PDF is delivered; 64m overflowed under concurrent heavy users (disk-full -> ProcessingError). 512m covers several concurrent batches.
Add ProviderPool (app/ai/pool.py) that distributes receipt extraction across multiple providers. round_robin cycles both lanes for throughput; priority tries a primary first and falls back on failure or low confidence. The round-robin counter is thread-safe (extract_receipt runs via to_thread). - build_provider returns a pool when AI_PROVIDER=pool - Config: SUPPORTED_PROVIDERS += pool; AI_POOL_STRATEGY, AI_POOL_PRIMARY - pool requires both OPENAI_API_KEY and OLLAMA_BASE_URL - pool.py imports only the provider interface (no circular import)
_op_purge_expired iterated every session in Python and deleted stale rows one by one. Replace with one set-based DELETE WHERE updated_at < cutoff (updated_at is always UTC ISO-8601, so lexicographic comparison is correct) and add an index on updated_at via schema migration 4.
Batch.total summed receipts across currencies (AED + USD), a documented footgun used only in tests. Remove the property and update the 4 test usages to currency_totals[currency], so mixing currencies is impossible instead of merely warned against.
Add max_queue_size (default 50). JobQueue.enqueue raises QueueFullError when the queue is at capacity (asyncio.Queue maxsize, position counter rolled back on failure). _password_attempt catches it, replies QUEUE_FULL, and returns the session to IDLE with receipts still staged so the user can retry /generate without re-uploading.
The in-memory queue drops jobs on restart; previously the affected user got no message (silent missing report). On startup, notify each queued user via TelegramService.send_message and reset their session to IDLE so they can re-run /generate. Best-effort per user; a send failure is logged and does not stop the reset.
- Replace raise last_exc # type: ignore[misc] with an explicit None check and a clear fallback error (preserves the last exception for rate-limit handling, no type-ignore). - README: document that concurrency is bounded by WORKER_COUNT x AI_CONCURRENCY (the only two dials), the lease is crash recovery not multi-instance support, and the bot is single-instance by design.
sweep_orphaned_requests deleted every request_* dir at startup with no age check, a latent footgun if TEMP_DIR is ever shared across instances (one instance's startup would destroy another's in-flight request dirs). Only sweep dirs whose mtime is older than age_seconds (default 600s); main.py passes max_processing_seconds. age_seconds=0 preserves the old behavior.
Remove the cross-process processing lease and its heartbeat renew loop, leaving a single-process store (flock singleton already guards one instance). Rename session_lease_ttl_seconds -> lock_idle_seconds; migrate DB schema to v5 (ALTER TABLE ... DROP COLUMN lease_expiry). Startup recovery now resets both QUEUED and PROCESSING plus the processing flag via get_stale/reset_stale. Make metrics registry and per-user lock dicts thread-safe (threading.Lock) so the health-server thread can read them without dict-iteration races. Add queue.stats()/locks.stats() and merge into the /metrics payload via build_application returning (application, bot). Add max_concurrent_ai_calls cap (worker_count x ai_concurrency validated at startup). Expand tests: rate-limit mapping, queue/lock stats, metrics thread-safety, config cap; drop lease-renewal integration test.
Extract _bootstrap() and _run_polling() from main() as pure code movement (no behavior change) so the startup/shutdown error paths are testable. Add unit tests for: - OpenAI 429 -> AIRateLimitError mapping, generic-error wrapping, Retry-After parsing (none / HTTP-date / invalid), _extract_json non-object - _start_health_server enabled (daemon thread) / disabled (no-op) - _bootstrap happy path + non-writable dir, PermissionError, held-lock, and backup FileNotFoundError (non-fatal) paths - _run_polling releasing the instance lock on clean shutdown and on exception openai_provider.py: 74% -> 100%; main.py: 58% -> 84%; suite 379 -> 395.
…ne/run) Convert app/services/receipt_service.py into app/services/receipt_service/ with cohesive submodules, keeping all public names re-exported so existing imports (main, bot, tests) resolve unchanged. - types.py: ProcessingError, BudgetExceededError, ProcessingResult, _ReceiptOutcome, make_request_base, _pdf_filename - retry.py: _extract_with_retry, _CallBudget, MAX_RATE_LIMIT_DELAY - pipeline.py: ProcessingService - run.py: run_with_cleanup Pipeline still logs under app.services.receipt_service so caplog/handler tests targeting that name keep working. Add surface-guard test (396 total). Gates: 396 passed, ruff clean, mypy clean.
Move _process_job/_notify/_report_caption and caption helpers out of the PTB-facing ReimbursementBot into a JobProcessor class; bot.py now wires start_workers via self.queue.start(self.job_processor.process). Add bot surface-guard tests (importability + method presence). Worker logs now under app.bot.job_processor (it moved modules), so the catch-all request_id test targets that logger. Gates: 398 passed, ruff clean, mypy clean.
Add app/bot/base.py (_BotBase: shared attribute declarations + _reply + _authorized) and app/bot/handlers.py (CommandHandlersMixin with the seven command handlers). ReimbursementBot now inherits the mixin, so handlers stay on the instance for main.py bound-callable registration and direct test calls. Gates: 398 passed, ruff clean, mypy clean.
Add app/bot/receipt_input.py with message_handler, _extract_file, _heading_attempt, _password_attempt, _candidate_text. ReimbursementBot now inherits (CommandHandlersMixin, ReceiptInputMixin), leaving bot.py as just dependency wiring + worker lifecycle (432 -> ~110 lines). Gates: 398 passed, ruff clean, mypy clean.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Remove the cross-process processing lease and heartbeat renew loop, leaving a single-process store (flock singleton guards one instance).
session_lease_ttl_seconds->lock_idle_seconds; migrate DB schema to v5 (DROP COLUMN lease_expiry).