diff --git a/Docs/superpowers/plans/2026-07-04-jobs-admission-operations-extraction-plan.md b/Docs/superpowers/plans/2026-07-04-jobs-admission-operations-extraction-plan.md new file mode 100644 index 0000000000..3101a52557 --- /dev/null +++ b/Docs/superpowers/plans/2026-07-04-jobs-admission-operations-extraction-plan.md @@ -0,0 +1,503 @@ +# Jobs Admission Operations Extraction Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Extract the Jobs create/admission transaction from `JobManager.create_job` into backend-specific operation modules while preserving the public facade and current behavior. + +**Architecture:** `JobManager` keeps caller validation, payload hygiene, allowed-queue/job-type policy, fair-share priority adjustment, public row mapping, metrics, audit, and in-process event fanout. SQLite and Postgres admission modules own transactional quota checks, idempotent insert/select, counter updates, and transactional `job.created` rows. Operation modules return typed `AdmissionResult` facts and must not import `JobManager`. + +**Tech Stack:** Python dataclasses, sqlite3, psycopg-compatible cursor usage, existing Jobs migrations, pytest, FastAPI TestClient only for existing API contracts, Bandit. + +--- + +## Scope + +This is `TASK-12138`. It implements rollout step 7 from `Docs/superpowers/specs/2026-06-24-jobs-backend-parity-refactor-design.md`. + +In scope: +- `create_job` admission only. +- SQLite and Postgres operation modules. +- Idempotent create and current-request `job.created` event behavior. +- Create-time quota checks, counters, durable event rows, and facade-owned side effects. +- Focused parity, fault-injection, and contract verification. + +Out of scope: +- `acquire_next_job`, `renew_job_lease`, `complete_job`, `fail_job`, `cancel_job`, batch lifecycle, dependencies, prune/archive, read-model/admin SQL extraction. +- Public REST response changes. +- Schema changes. + +## Existing Baseline + +Baseline command run before planning: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest \ + tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py \ + tldw_Server_API/tests/Jobs/test_jobs_idempotency_scope_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_completion_idempotent_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_quotas_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_queue_controls_and_admin_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_events_outbox_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_settings.py \ + tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py \ + tldw_Server_API/tests/Jobs/test_jobs_admin_contract_sqlite.py \ + tldw_Server_API/tests/Chatbooks/test_chatbooks_jobs_adapter.py \ + -q +``` + +Result: `55 passed, 424 warnings`. + +## File Structure + +- Modify: `tldw_Server_API/app/core/Jobs/operations/contracts.py` + - Permit idempotent existing admission results to carry durable event facts because current create replay writes a transactional `job.created` row. + - Allow `CreateJobCommand.project_id` to accept the existing manager input shape. +- Create: `tldw_Server_API/app/core/Jobs/operations/sqlite/__init__.py` + - Export SQLite admission helper. +- Create: `tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py` + - SQLite transactional admission implementation. +- Create: `tldw_Server_API/app/core/Jobs/operations/postgres/__init__.py` + - Export Postgres admission helper. +- Create: `tldw_Server_API/app/core/Jobs/operations/postgres/admission.py` + - Postgres transactional admission implementation. +- Modify: `tldw_Server_API/app/core/Jobs/manager.py` + - Build `CreateJobCommand`, call backend operation helper, map result to existing public row, and emit facade-owned side effects. +- Modify: `tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py` + - Add red coverage for idempotent existing durable event facts and operation-package no-`JobManager` import coverage. +- Create: `tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py` + - Direct SQLite operation tests for inserted, idempotent existing, quota rejection, counters, and rollback on `job_events` failure. +- Modify: `tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py` + - Keep fake cursor compatibility and add a manager-routing assertion that Postgres create still uses current request context for idempotent replay events after extraction. +- Modify: `tldw_Server_API/tests/Jobs/parity/scenarios.py` + - Add one shared scenario asserting idempotent replay persists a second transactional `job.created` event with current request/trace ids while preserving the original job row request/trace ids. +- Modify: `tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py` + - Run the new shared scenario against SQLite. +- Modify: `tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py` + - Run the same scenario against real Postgres when available. +- Modify: `backlog/tasks/task-12138 - Extract-Jobs-admission-operations-behind-JobManager.md` + - Record plan, verification, and final summary. + +## Task 1: Preserve Contract Truth For Idempotent Replay Events + +**Files:** +- Modify: `tldw_Server_API/app/core/Jobs/operations/contracts.py` +- Modify: `tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py` + +- [x] **Step 1: Write the failing contract test** + +Add a test named `test_admission_existing_can_report_idempotent_durable_event` that constructs: + +```python +result = AdmissionResult.existing( + row={"id": 1, "status": "queued"}, + durable_events=({"event_type": "job.created", "attrs": {"idempotent": True}},), +) +``` + +Expected assertions: +- `result.outcome is OperationOutcome.NO_TRANSITION` +- `result.no_transition_reason is NoTransitionReason.IDEMPOTENT_EXISTING` +- `result.durable_events` contains the deep-copied event. + +- [x] **Step 2: Verify red** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py::test_admission_existing_can_report_idempotent_durable_event -q +``` + +Expected: fails because `AdmissionResult.existing` does not accept `durable_events` and/or current invariants reject durable events on no-transition results. + +- [x] **Step 3: Update the contract minimally** + +Change `AdmissionResult.__post_init__` so durable events are allowed for: +- `OperationOutcome.APPLIED` +- `OperationOutcome.NO_TRANSITION` with `NoTransitionReason.IDEMPOTENT_EXISTING` + +Change `AdmissionResult.existing` to accept `durable_events: Sequence[dict[str, Any]] = ()` and pass them through. + +Do not relax `LifecycleResult`; lifecycle no-transition durable events remain invalid in this slice. + +- [x] **Step 4: Verify green** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py -q +``` + +Expected: all operation contract tests pass. + +## Task 2: Add Direct SQLite Admission Operation Tests + +**Files:** +- Create: `tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py` +- Create later in Task 3: `tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py` + +- [x] **Step 1: Write failing module import and inserted-row test** + +Create tests that import: + +```python +from tldw_Server_API.app.core.Jobs.operations.contracts import CreateJobCommand, OperationOutcome +from tldw_Server_API.app.core.Jobs.operations.sqlite.admission import create_job_admission +``` + +The first test should: +- create a temp SQLite jobs DB with `ensure_jobs_tables` +- open a connection with row factory +- call `create_job_admission` with a `CreateJobCommand(domain="admission", queue="default", job_type="insert", payload={"x": 1}, owner_user_id="u1", request_id="req-1", trace_id="trace-1")` +- pass `uuid_value="uuid-insert"`, `now=datetime(2026, 1, 1, tzinfo=timezone.utc)`, `max_queued_quota=0`, `submits_per_minute_quota=0`, and `counters_enabled=True` +- assert `result.outcome is OperationOutcome.APPLIED` +- assert `result.inserted is True` +- assert `result.row["status"] == "queued"` +- assert exactly one `job.created` row exists with matching `request_id` and `trace_id` +- assert counters incremented ready count once. + +- [x] **Step 2: Write failing idempotent-existing test** + +The second test should: +- insert once with idempotency key `same` +- replay with the same idempotency key and new request/trace ids +- assert the replay returns `OperationOutcome.NO_TRANSITION` +- assert `result.inserted is False` +- assert the returned row keeps the original row request/trace ids +- assert two transactional `job.created` rows exist +- assert the second event uses the replay request/trace ids. + +- [x] **Step 3: Write failing quota and rollback tests** + +Add quota tests for: +- max queued quota rejection returns `OperationOutcome.ADMISSION_REJECTED` with `AdmissionRejectionReason.QUOTA_EXCEEDED` +- submits-per-minute quota rejection returns the same rejection reason. + +Add rollback test with a connection wrapper that raises `sqlite3.OperationalError` on `INSERT INTO job_events`; assert no `jobs` row commits after the exception. + +- [x] **Step 4: Verify red** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py -q +``` + +Expected: collection fails with `ModuleNotFoundError` for the new SQLite admission module. + +## Task 3: Implement SQLite Admission Operation + +**Files:** +- Create: `tldw_Server_API/app/core/Jobs/operations/sqlite/__init__.py` +- Create: `tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py` + +- [x] **Step 1: Add package marker** + +Create `sqlite/__init__.py` with: + +```python +"""SQLite Jobs admission operations.""" + +from .admission import create_job_admission + +__all__ = ["create_job_admission"] +``` + +- [x] **Step 2: Implement `create_job_admission`** + +Signature: + +```python +def create_job_admission( + conn: sqlite3.Connection, + *, + command: CreateJobCommand, + uuid_value: str, + now: datetime, + max_queued_quota: int, + submits_per_minute_quota: int, + counters_enabled: bool, +) -> AdmissionResult: +``` + +Rules: +- Use `with conn:` for the transaction. +- Serialize `command.payload` with `json.dumps(command.payload)`. +- Normalize aware `command.available_at` to UTC naive string for SQLite. +- Perform the existing max queued and submits-per-minute checks. +- Return `AdmissionResult.rejected(AdmissionRejectionReason.QUOTA_EXCEEDED, message=...)` for quota rejections instead of raising. +- For idempotent inserts, use the existing `INSERT OR IGNORE` plus `SELECT` behavior. +- For idempotent replay, insert a `job.created` row with `attrs_json.idempotent=true`, current `command.request_id`, and current `command.trace_id`. +- For inserted rows, insert a `job.created` row with `attrs_json.idempotent=false`. +- Update `job_counters` only for inserted rows when `counters_enabled` is true. +- Return `AdmissionResult.applied(...)` for inserted rows and `AdmissionResult.existing(..., durable_events=(event_fact,))` for idempotent existing rows. +- Let `sqlite3.OperationalError` from `job_events` insert propagate so the transaction rolls back. + +- [x] **Step 3: Verify green** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py -q +``` + +Expected: all direct SQLite admission operation tests pass. + +## Task 4: Route SQLite `JobManager.create_job` Through The Operation + +**Files:** +- Modify: `tldw_Server_API/app/core/Jobs/manager.py` + +- [x] **Step 1: Write/extend facade regression tests first** + +Use existing tests plus add a shared parity scenario in Task 6 before changing behavior. The manager must still: +- raise `ValueError` for quota rejection +- preserve original request/trace ids on idempotent row replay +- write a replay `job.created` event with current request/trace ids +- not increment created metrics when the transactional event insert fails +- use exactly one facade event/audit path. + +- [x] **Step 2: Add helper methods in `JobManager`** + +Add small private helpers: +- `_build_create_job_command(...) -> CreateJobCommand` +- `_map_admission_result(result: AdmissionResult) -> dict[str, Any]` +- `_emit_create_side_effects(result: AdmissionResult, *, backend: str) -> None` + +Keep helpers in `manager.py` for this slice. Do not move secret scanning, encryption, queue allowlist, job-type allowlist, fair-share, metrics, audit, or in-process event fanout into operation modules. + +- [x] **Step 3: Replace SQLite inline SQL branch** + +In the SQLite branch of `create_job`: +- call `sqlite_create_job_admission(...)` +- pass quota values from `_quota_get` +- pass `counters_enabled=JobManager._is_truthy(os.getenv("JOBS_COUNTERS_ENABLED", ""))` +- call `_update_gauges` after success, matching existing SQLite behavior +- convert `AdmissionResult.rejected(...QUOTA_EXCEEDED...)` to the same `ValueError` messages as today +- call side effects after the transaction returns. + +- [x] **Step 4: Verify SQLite facade behavior** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest \ + tldw_Server_API/tests/Jobs/test_jobs_idempotency_scope_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_quotas_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_events_sqlite.py \ + tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py \ + -q +``` + +Expected: all selected SQLite facade/fault/parity tests pass. + +## Task 5: Implement And Route Postgres Admission Operation + +**Files:** +- Create: `tldw_Server_API/app/core/Jobs/operations/postgres/__init__.py` +- Create: `tldw_Server_API/app/core/Jobs/operations/postgres/admission.py` +- Modify: `tldw_Server_API/app/core/Jobs/manager.py` +- Modify: `tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py` + +- [x] **Step 1: Add package marker** + +Create `postgres/__init__.py` with: + +```python +"""Postgres Jobs admission operations.""" + +from .admission import create_job_admission + +__all__ = ["create_job_admission"] +``` + +- [x] **Step 2: Implement `create_job_admission`** + +Signature: + +```python +def create_job_admission( + conn: Any, + cursor_factory: Callable[[Any], ContextManager[Any]], + *, + command: CreateJobCommand, + uuid_value: str, + now: datetime, + max_queued_quota: int, + submits_per_minute_quota: int, + counters_enabled: bool, +) -> AdmissionResult: +``` + +Rules: +- Use `with conn:` and `with cursor_factory(conn) as cur:`. +- Keep SQL text and parameter ordering compatible with existing fake cursor tests. +- Use `ON CONFLICT (domain, queue, job_type, idempotency_key) DO NOTHING RETURNING *` for idempotent insert. +- Select existing row on conflict. +- Insert transactional `job.created` event for inserted and idempotent-existing cases, using current request/trace ids. +- Update counters only when inserted and `counters_enabled` is true. +- Return the same `AdmissionResult` shapes as SQLite. +- Let event insert errors propagate so the transaction fails. + +- [x] **Step 3: Route the Postgres branch in `JobManager.create_job`** + +Replace only the Postgres create/admission SQL block with a call to `postgres_create_job_admission(...)`. Keep the facade side-effect helper shared with SQLite. + +- [x] **Step 4: Verify fake Postgres behavior** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py -q +``` + +Expected: all fake Postgres manager tests pass. + +## Task 6: Add Shared Parity Coverage For Idempotent Replay Events + +**Files:** +- Modify: `tldw_Server_API/tests/Jobs/parity/scenarios.py` +- Modify: `tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py` +- Modify: `tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py` + +- [x] **Step 1: Add shared scenario** + +Add `run_idempotent_replay_records_current_request_event_scenario(make_manager)`: +- create a job with idempotency key `event-replay-key`, request id `request-first`, trace id `trace-first` +- replay with request id `request-second`, trace id `trace-second` +- assert both calls return the same job id +- assert replay row keeps `request-first` and `trace-first` +- query `list_job_events_after(after_id=0, domain="parity", queue="default", job_type="event-replay", event_types=("job.created",), limit=20)` +- assert two `job.created` rows exist for the job +- assert the last event uses `request-second` and `trace-second`. + +- [x] **Step 2: Wire SQLite and Postgres wrappers** + +Add wrapper tests: +- `test_sqlite_idempotent_replay_records_current_request_event` +- `test_postgres_idempotent_replay_records_current_request_event` + +- [x] **Step 3: Verify parity wrappers** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest \ + tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py \ + tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py \ + -q -rs +``` + +Expected: SQLite passes. Postgres passes when the fixture is reachable or skips with the existing explicit Postgres-unreachable reason. + +## Task 7: Full Focused Verification, Backlog, And Commit + +**Files:** +- Modify: `backlog/tasks/task-12138 - Extract-Jobs-admission-operations-behind-JobManager.md` + +- [x] **Step 1: Run focused Jobs admission matrix** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && RUN_JOBS=1 python -m pytest \ + tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py \ + tldw_Server_API/tests/Jobs/test_jobs_idempotency_scope_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_idempotency_scope_postgres.py \ + tldw_Server_API/tests/Jobs/test_jobs_quotas_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_quotas_postgres.py \ + tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py \ + tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py \ + tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py \ + tldw_Server_API/tests/Jobs/test_jobs_admin_contract_sqlite.py \ + tldw_Server_API/tests/Chatbooks/test_chatbooks_jobs_adapter.py \ + -q -rs +``` + +Expected: all SQLite/fake/unit/API contract tests pass; Postgres tests pass or record existing fixture skip. + +- [x] **Step 2: Run import-boundary scan** + +Run: + +```bash +rg -n "JobManager|Jobs\\.manager|from .*manager" tldw_Server_API/app/core/Jobs/operations +``` + +Expected: no matches. + +- [x] **Step 3: Run Bandit** + +Run: + +```bash +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && python -m bandit -q -s B101 -r \ + tldw_Server_API/app/core/Jobs/manager.py \ + tldw_Server_API/app/core/Jobs/operations \ + tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py +``` + +Expected: exits 0. + +- [x] **Step 4: Run diff hygiene and compile smoke** + +Run: + +```bash +git diff --check +source /Users/appledev/Documents/GitHub/tldw_server/.venv/bin/activate && python -m py_compile \ + tldw_Server_API/app/core/Jobs/manager.py \ + tldw_Server_API/app/core/Jobs/operations/contracts.py \ + tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py \ + tldw_Server_API/app/core/Jobs/operations/postgres/admission.py +``` + +Expected: both commands exit 0. + +- [x] **Step 5: Update Backlog task** + +Record: +- plan path +- focused baseline result +- red/green test evidence +- final verification results +- Postgres fixture skips, if any +- Bandit result +- final summary. + +- [ ] **Step 6: Commit** + +Run: + +```bash +git add \ + Docs/superpowers/plans/2026-07-04-jobs-admission-operations-extraction-plan.md \ + "backlog/tasks/task-12138 - Extract-Jobs-admission-operations-behind-JobManager.md" \ + tldw_Server_API/app/core/Jobs/manager.py \ + tldw_Server_API/app/core/Jobs/operations/contracts.py \ + tldw_Server_API/app/core/Jobs/operations/sqlite \ + tldw_Server_API/app/core/Jobs/operations/postgres \ + tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py \ + tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py \ + tldw_Server_API/tests/Jobs/parity/scenarios.py \ + tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py \ + tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py \ + tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py +git commit -m "refactor(jobs): extract admission operations" +``` + +Expected: commit succeeds. + +## Review Checklist + +- `JobManager.create_job` remains the only public caller entrypoint. +- Operation modules contain backend SQL and do not import `JobManager`. +- Queue allowlist, fair-share, secret hygiene, encryption, job-type allowlist, and side effects stay facade-owned. +- Idempotent replay still writes a transactional `job.created` event with current request/trace ids. +- Existing public row request/trace ids are preserved on idempotent replay. +- Metrics increment only for inserted rows and only after transactional admission succeeds. +- Event insert failure rolls back job insertion and does not increment created metrics. +- No lifecycle methods are extracted. diff --git a/backlog/tasks/task-12147 - Extract-Jobs-admission-operations-behind-JobManager.md b/backlog/tasks/task-12147 - Extract-Jobs-admission-operations-behind-JobManager.md new file mode 100644 index 0000000000..5d10860f2a --- /dev/null +++ b/backlog/tasks/task-12147 - Extract-Jobs-admission-operations-behind-JobManager.md @@ -0,0 +1,95 @@ +--- +id: TASK-12147 +title: Extract Jobs admission operations behind JobManager +status: Done +assignee: [] +created_date: 2026-07-04 01:28 +updated_date: 2026-07-05 00:28 +labels: +- jobs +- implementation +- refactor +dependencies: [] +references: +- TASK-12015 +- TASK-12016 +- TASK-12017 +- Docs/superpowers/specs/2026-06-24-jobs-backend-parity-refactor-design.md +- Docs/superpowers/plans/2026-06-24-jobs-backend-parity-refactor-implementation-plan.md +- https://github.com/rmusser01/tldw_server/pull/2527 +- https://github.com/rmusser01/tldw_server/pull/2611 +documentation: +- Docs/superpowers/specs/2026-06-24-jobs-backend-parity-refactor-design.md +priority: medium +modified_files: +- Docs/superpowers/plans/2026-07-04-jobs-admission-operations-extraction-plan.md +- tldw_Server_API/app/core/Jobs/manager.py +- tldw_Server_API/app/core/Jobs/operations/contracts.py +- tldw_Server_API/app/core/Jobs/operations/postgres/admission.py +- tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py +- tldw_Server_API/tests/Jobs/parity/scenarios.py +- tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py +- tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py +- tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py +- tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py +- tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py +- tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py +--- + +## Description + + +Extract the Jobs create/admission transaction path behind backend-specific operation modules while preserving JobManager.create_job as the public facade. Scope includes SQLite/Postgres admission operation modules, typed CreateJobCommand/AdmissionResult use, idempotency, quota/fair-share/queue policy behavior, durable event facts, and parity/API contract verification. Lifecycle operations remain out of scope. + + +## Acceptance Criteria + +- [x] #1 JobManager.create_job remains the public caller-facing API and preserves existing row/REST-compatible behavior. +- [x] #2 SQLite and Postgres admission transaction logic is routed through backend-specific operation modules under app/core/Jobs/operations/ without those modules importing JobManager. +- [x] #3 Admission parity tests cover idempotent create, queue controls, quota/fair-share rejection, durable event facts, and public facade mapping for touched behavior. +- [x] #4 No lifecycle methods are extracted in this slice. +- [x] #5 Focused Jobs/Chatbooks parity and contract tests pass; Postgres-specific tests either pass or record an explicit fixture skip. +- [x] #6 Bandit and diff hygiene checks pass for the touched implementation scope. + + +## Implementation Plan + + +1. Confirm baseline Jobs safety-net tests pass or record existing skips. 2. Write a focused implementation plan for admission extraction. 3. Add failing admission-operation tests for the new backend operation boundary. 4. Implement the minimal SQLite and Postgres admission operation modules and route JobManager.create_job through them. 5. Preserve public response mapping and post-commit side effects in JobManager. 6. Run focused parity/API/security verification and update this task with results. + + +## Implementation Notes + + +Plan: Docs/superpowers/plans/2026-07-04-jobs-admission-operations-extraction-plan.md +Baseline before implementation: focused Jobs matrix passed with 55 passed, 424 warnings. +Red/green evidence: added failing AdmissionResult.existing durable-event contract test, then updated the contract and verified test_jobs_operation_contracts.py passed. Added direct SQLite admission operation tests; they initially failed on missing module, then passed after implementation. +Implementation: added SQLite/Postgres admission operation modules, kept JobManager.create_job as facade for validation, policy, metrics, audit/fanout, and public row mapping. Operation modules own transactional quota checks, idempotent insert/select, counter updates, and durable job.created rows. Lifecycle methods were not extracted. +Original verification: focused matrix passed on final code with 62 passed, 13 skipped, 243 warnings. Skips were explicit Postgres-unreachable fixture skips in Postgres idempotency/quota/parity tests. Additional affected tests passed: SQLite admission operation tests and fake Postgres manager tests, 11 passed. Static checks: operation import-boundary scan returned no matches; git diff --check passed; py_compile passed for manager/contracts/sqlite/postgres admission modules; Bandit exited 0 on touched scope with only pre-existing #nosec warnings in manager.py. +PR: https://github.com/rmusser01/tldw_server/pull/2611 +Pre-rebase validation refresh (2026-07-04): branch was stale at merge-base ded7b5158f676263fa4224fd721ac22c3a51c2f1, then rebased cleanly onto origin/dev fd5c152b065c408e4e8ee5f08da41589f21cb7f5. Post-rebase merge-base matched origin/dev at that time. Addressed PR review feedback by ensuring SQLite idempotent replay in-process emit_job_event uses the durable event current request/trace context when the outbox is disabled, and by ensuring Postgres non-idempotent admission job.created durable events use CreateJobCommand.request_id/trace_id instead of row fields that can be absent in mocked/fallback rows. Added focused regressions: test_idempotent_replay_in_process_event_uses_current_context_sqlite and test_pg_non_idempotent_create_uses_command_context_for_job_created_event. Passed targeted review tests (2 passed). Passed focused jobs matrix with 31 passed, 7 skipped. Passed Bandit over touched production Jobs scope with 0 findings in /tmp/bandit_jobs_admission_latest_dev.json. Passed git diff --check. Postgres parity file was invoked directly and skipped all 7 tests because the local Postgres fixture was unavailable in this environment. + + +## Final Summary + + +Extracted Jobs admission operations and contract coverage. Final refresh validated against origin/dev 6b727b221e55646eba663a03571e38302f7fafc2 with focused Jobs tests passing, Bandit reporting zero findings on touched production scope, and whitespace check clean. + + +## Definition of Done + +- [x] #1 Acceptance criteria completed +- [x] #2 Tests or verification recorded +- [x] #3 Documentation updated when relevant +- [x] #4 Bandit run for touched code when applicable or document non-code/environment skip +- [x] #5 Final summary added +- [x] #6 Known skips or blockers documented + + +## Implementation Notes + + +Post-rebase validation on current origin/dev 4c1ca5d8358bff2a5a7fb5c75d60d1bd6728e702: rebased codex/jobs-admission-operations-extraction so merge-base equals current origin/dev. Fresh verification after rebase: focused Jobs admission/operation suite passed (31 passed, 14 skipped, 70 warnings); Bandit over touched Jobs production files scanned 7294 LOC and reported 0 findings in /tmp/bandit_jobs_admission_rebased_dev.json, with existing skipped # nosec B608 notices only; git diff --check HEAD~1..HEAD passed. +2026-07-04 current-dev refresh: rebased `codex/jobs-admission-operations-extraction` onto `origin/dev` `09d9ec901e1d4548f7924f1c6bcefa963fadd9bd`; merge-base matches `origin/dev`. Validation: `/Users/appledev/Documents/GitHub/tldw_server/.venv/bin/python -m pytest tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py -q` passed with 31 tests and 14 expected parity skips; `/Users/appledev/Documents/GitHub/tldw_server/.venv/bin/python -m bandit -r tldw_Server_API/app/core/Jobs/manager.py tldw_Server_API/app/core/Jobs/operations/contracts.py tldw_Server_API/app/core/Jobs/operations/postgres/admission.py tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py -f json -o /tmp/bandit_jobs_admission_origin_dev_09d9ec.json` reported 0 findings over 7294 LOC with existing skipped `# nosec B608` warnings; `git diff --check HEAD~1..HEAD` passed. +2026-07-04 latest-dev refresh: rebased and validated PR #2611 on origin/dev 6b727b221e55646eba663a03571e38302f7fafc2. Tested head a34d7bf9f3fc. Verification: focused Jobs pytest suite => 31 passed, 14 skipped, 70 warnings; Bandit over Jobs manager/admission operation scope => 0 findings over 7294 LOC with existing skipped #nosec B608 warnings only; git diff --check HEAD~1..HEAD => clean. + diff --git a/tldw_Server_API/app/core/Jobs/manager.py b/tldw_Server_API/app/core/Jobs/manager.py index ec733961c3..183ac2062f 100644 --- a/tldw_Server_API/app/core/Jobs/manager.py +++ b/tldw_Server_API/app/core/Jobs/manager.py @@ -55,6 +55,14 @@ set_queue_gauges, ) from .migrations import ensure_jobs_tables +from .operations.contracts import ( + AdmissionRejectionReason, + AdmissionResult, + CreateJobCommand, + OperationOutcome, +) +from .operations.postgres import create_job_admission as _postgres_create_job_admission +from .operations.sqlite import create_job_admission as _sqlite_create_job_admission from .pg_migrations import ensure_job_counters_pg, ensure_jobs_tables_pg from .tracing import job_span @@ -1096,6 +1104,118 @@ def _parse(v: str | None) -> int: return _parse(v) return _parse(os.getenv(base)) + @staticmethod + def _build_create_job_command( + *, + domain: str, + queue: str, + job_type: str, + payload: dict[str, Any], + owner_user_id: str | None, + project_id: int | None, + batch_group: str | None, + priority: int, + max_retries: int, + available_at: datetime | None, + idempotency_key: str | None, + request_id: str | None, + trace_id: str | None, + ) -> CreateJobCommand: + """Build the backend-neutral create command after facade validation.""" + + return CreateJobCommand( + domain=domain, + queue=queue, + job_type=job_type, + payload=payload, + owner_user_id=owner_user_id, + idempotency_key=idempotency_key, + priority=priority, + max_retries=max_retries, + available_at=available_at, + project_id=project_id, + batch_group=batch_group, + request_id=request_id, + trace_id=trace_id, + ) + + @staticmethod + def _map_admission_result(result: AdmissionResult) -> dict[str, Any]: + """Map an admission result to the public create_job return row.""" + + if result.outcome is OperationOutcome.ADMISSION_REJECTED: + if result.admission_rejection_reason is AdmissionRejectionReason.QUOTA_EXCEEDED: + raise ValueError(result.message or "Quota exceeded") # noqa: TRY003 + raise ValueError(result.message or "Admission rejected") # noqa: TRY003 + if result.row is None: + raise RuntimeError("Job admission did not return a row") # noqa: TRY003 + return result.row + + def _emit_create_side_effects( + self, + result: AdmissionResult, + *, + backend: str, + idempotency_key: str | None, + ) -> None: + """Emit create metrics and facade-owned event/audit side effects.""" + + row = result.row + if row is None: + return + + if result.inserted: + _safe_increment_created_metric( + domain=str(row.get("domain")), + queue=str(row.get("queue")), + job_type=str(row.get("job_type")), + ) + + for event in result.durable_events: + if event.get("event_type") != "job.created": + continue + attrs = dict(event.get("attrs") or {}) + request_id = event.get("request_id") + trace_id = event.get("trace_id") + outbox_enabled = JobManager._is_truthy(os.getenv("JOBS_EVENTS_OUTBOX", "")) + + if backend == "sqlite": + if idempotency_key: + if outbox_enabled: + with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): + submit_job_audit_event( + "job.created", + job={**row, "request_id": request_id, "trace_id": trace_id}, + attrs=attrs, + ) + else: + with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): + emit_job_event( + "job.created", + job={**row, "request_id": request_id, "trace_id": trace_id}, + attrs=attrs, + ) + continue + + emitted_job = {**row, "request_id": request_id, "trace_id": trace_id} + if not outbox_enabled: + with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): + emit_job_event("job.created", job=emitted_job, attrs=attrs) + with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): + submit_job_audit_event("job.created", job=emitted_job, attrs=attrs) + continue + + emitted_job = { + **row, + "request_id": row.get("request_id") or request_id, + "trace_id": row.get("trace_id") or trace_id, + } + if not outbox_enabled: + with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): + emit_job_event("job.created", job=emitted_job, attrs=attrs) + with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): + submit_job_audit_event("job.created", job=emitted_job, attrs=attrs) + # --- Advisory lock helpers (Postgres) --- def _pg_advisory_key(self, *parts: str) -> int: """Compute a signed 64-bit advisory lock key from parts.""" @@ -1228,7 +1348,6 @@ def create_job( pass # Use consistent clock _now_dt = self._clock.now_utc() - now = _now_dt.astimezone(_tz.utc).strftime("%Y-%m-%d %H:%M:%S") uuid_val = str(_uuid.uuid4()) if not trace_id: try: @@ -1259,515 +1378,94 @@ def create_job( ) if self.backend == "postgres": - with conn: # noqa: SIM117 - with self._pg_cursor(conn) as cur: - # Domain/user quotas - try: - # Max queued - max_q = self._quota_get("JOBS_QUOTA_MAX_QUEUED", domain, owner_user_id) - if max_q and owner_user_id: - cur.execute( - "SELECT COUNT(*) AS c FROM jobs WHERE domain=%s AND owner_user_id=%s AND status='queued'", - (domain, owner_user_id), - ) - row_cnt = cur.fetchone() - if int(row_cnt.get("c") if isinstance(row_cnt, dict) else 0) >= max_q: - raise ValueError("Quota exceeded: max queued per user/domain") # noqa: TRY003 - # Submits per minute - spm = self._quota_get("JOBS_QUOTA_SUBMITS_PER_MIN", domain, owner_user_id) - if spm and owner_user_id: - cur.execute( - "SELECT COUNT(*) AS c FROM jobs WHERE domain=%s AND owner_user_id=%s AND created_at >= (%s - interval '60 seconds')", - (domain, owner_user_id, _now_dt), - ) - row_spm = cur.fetchone() - if int(row_spm.get("c") if isinstance(row_spm, dict) else 0) >= spm: - raise ValueError("Quota exceeded: submits per minute") # noqa: TRY003 - except _JOB_NONCRITICAL_EXCEPTIONS as _db_exc: - # Let ValueError propagate; swallow only DB/adapter errors - try: - import psycopg - - if isinstance(_db_exc, psycopg.Error): - pass - else: - raise - except _JOB_NONCRITICAL_EXCEPTIONS: # noqa: TRY203 - raise - pass - if idempotency_key: - # Cast payload to jsonb explicitly to avoid adapter issues - cur.execute( - ( - "INSERT INTO jobs (uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, idempotency_key, payload, result, status, priority, max_retries, retry_count, available_at, created_at, updated_at, request_id, trace_id) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, NULL, 'queued', %s, %s, 0, %s, NOW(), NOW(), %s, %s) " - "ON CONFLICT (domain, queue, job_type, idempotency_key) DO NOTHING RETURNING *" - ), - ( - uuid_val, - domain, - queue, - job_type, - owner_user_id, - project_id, - batch_group, - idempotency_key, - payload_json, - priority, - max_retries, - avail_param if avail_param else None, - request_id, - trace_id, - ), - ) - row = cur.fetchone() - was_insert = row is not None - if not row: - cur.execute( - "SELECT * FROM jobs WHERE domain = %s AND queue = %s AND job_type = %s AND idempotency_key = %s", - (domain, queue, job_type, idempotency_key), - ) - row = cur.fetchone() - d = ( - dict(row) - if row - else { - "uuid": uuid_val, - "status": "queued", - "domain": domain, - "queue": queue, - "job_type": job_type, - } - ) - emitted_job = { - **d, - "request_id": d.get("request_id") or request_id, - "trace_id": d.get("trace_id") or trace_id, - } - # Counters bump (PG, idempotent insert occurred) - try: - if was_insert and JobManager._is_truthy(os.getenv("JOBS_COUNTERS_ENABLED", "")): - is_sched = bool(avail_param) - cur.execute( - ( - "INSERT INTO job_counters(domain,queue,job_type,ready_count,scheduled_count,processing_count,quarantined_count) VALUES(%s,%s,%s,%s,%s,0,0) " - "ON CONFLICT (domain,queue,job_type) DO UPDATE SET ready_count = job_counters.ready_count + EXCLUDED.ready_count, scheduled_count = job_counters.scheduled_count + EXCLUDED.scheduled_count, updated_at = NOW()" - ), - (domain, queue, job_type, 0 if is_sched else 1, 1 if is_sched else 0), - ) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - # Write to outbox within the same transaction (Postgres path). - attrs_json = json.dumps( - { - "idempotent": (not was_insert), - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - } - ) - cur.execute( - ( - "INSERT INTO job_events(" - "job_id, domain, queue, job_type, event_type, attrs_json, owner_user_id, request_id, trace_id, created_at" - ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())" - ), - ( + command = self._build_create_job_command( + domain=domain, + queue=queue, + job_type=job_type, + payload=payload, + owner_user_id=owner_user_id, + project_id=project_id, + batch_group=batch_group, + priority=priority, + max_retries=max_retries, + available_at=avail_param, + idempotency_key=idempotency_key, + request_id=request_id, + trace_id=trace_id, + ) + result = _postgres_create_job_admission( + conn, + self._pg_cursor, + command=command, + uuid_value=uuid_val, + now=_now_dt, + max_queued_quota=self._quota_get("JOBS_QUOTA_MAX_QUEUED", domain, owner_user_id), + submits_per_minute_quota=self._quota_get("JOBS_QUOTA_SUBMITS_PER_MIN", domain, owner_user_id), + counters_enabled=JobManager._is_truthy(os.getenv("JOBS_COUNTERS_ENABLED", "")), + ) + d = self._map_admission_result(result) + try: + pol = self._get_sla_policy(d.get("domain"), d.get("queue"), d.get("job_type")) + if pol and (pol.get("enabled") in (True, 1)): + ca = _parse_dt(d.get("acquired_at")) + cr = _parse_dt(d.get("created_at")) if d.get("created_at") else None + if ca and cr and (pol.get("max_queue_latency_seconds") is not None): + qlat = max(0.0, (ca - cr).total_seconds()) + if qlat > float(pol.get("max_queue_latency_seconds")): + self._record_sla_breach( int(d.get("id")), - d.get("domain"), - d.get("queue"), - d.get("job_type"), - "job.created", - attrs_json, - d.get("owner_user_id"), - request_id, - trace_id, - ), - ) - if was_insert: - _safe_increment_created_metric( - domain=domain, - queue=queue, - job_type=job_type, - ) - # Emit event for in-process listeners when outbox is disabled - try: - if not JobManager._is_truthy(os.getenv("JOBS_EVENTS_OUTBOX", "")): - emit_job_event( - "job.created", - job=emitted_job, - attrs={ - "idempotent": (not was_insert), - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - }, - ) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - # Audit bridge (best-effort) - with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): - submit_job_audit_event( - "job.created", - job=emitted_job, - attrs={ - "idempotent": (not was_insert), - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - }, + str(d.get("domain")), + str(d.get("queue")), + str(d.get("job_type")), + "queue_latency", + qlat, + float(pol.get("max_queue_latency_seconds")), + conn=conn, ) - return d - # Non-idempotent insert - cur.execute( - ( - "INSERT INTO jobs (uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, idempotency_key, payload, result, status, priority, max_retries, retry_count, available_at, created_at, updated_at, request_id, trace_id) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, NULL, 'queued', %s, %s, 0, %s, NOW(), NOW(), %s, %s) RETURNING *" - ), - ( - uuid_val, + except _JOB_NONCRITICAL_EXCEPTIONS: + pass + with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): + self._assert_invariants(d) + self._emit_create_side_effects(result, backend="postgres", idempotency_key=idempotency_key) + return d + else: + command = self._build_create_job_command( + domain=domain, + queue=queue, + job_type=job_type, + payload=payload, + owner_user_id=owner_user_id, + project_id=project_id, + batch_group=batch_group, + priority=priority, + max_retries=max_retries, + available_at=avail_param_sqlite, + idempotency_key=idempotency_key, + request_id=request_id, + trace_id=trace_id, + ) + for attempt in range(2): + try: + result = _sqlite_create_job_admission( + conn, + command=command, + uuid_value=uuid_val, + now=_now_dt, + max_queued_quota=self._quota_get("JOBS_QUOTA_MAX_QUEUED", domain, owner_user_id), + submits_per_minute_quota=self._quota_get( + "JOBS_QUOTA_SUBMITS_PER_MIN", domain, - queue, - job_type, owner_user_id, - project_id, - batch_group, - idempotency_key, - payload_json, - priority, - max_retries, - avail_param if avail_param else None, - request_id, - trace_id, ), + counters_enabled=JobManager._is_truthy(os.getenv("JOBS_COUNTERS_ENABLED", "")), ) - row = cur.fetchone() - d = dict(row) - # SLA check: queue latency (Postgres create path) - try: - pol = self._get_sla_policy(d.get("domain"), d.get("queue"), d.get("job_type")) - if pol and (pol.get("enabled") in (True, 1)): - ca = _parse_dt(d.get("acquired_at")) - cr = _parse_dt(d.get("created_at")) if d.get("created_at") else None - if ca and cr and (pol.get("max_queue_latency_seconds") is not None): - qlat = max(0.0, (ca - cr).total_seconds()) - if qlat > float(pol.get("max_queue_latency_seconds")): - self._record_sla_breach( - int(d.get("id")), - str(d.get("domain")), - str(d.get("queue")), - str(d.get("job_type")), - "queue_latency", - qlat, - float(pol.get("max_queue_latency_seconds")), - conn=conn, - ) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass + d = self._map_admission_result(result) with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): - self._assert_invariants(d) - # Counters bump (PG, non-idempotent path) - try: - if JobManager._is_truthy(os.getenv("JOBS_COUNTERS_ENABLED", "")): - is_sched = bool(avail_param) - cur.execute( - ( - "INSERT INTO job_counters(domain,queue,job_type,ready_count,scheduled_count,processing_count,quarantined_count) VALUES(%s,%s,%s,%s,%s,0,0) " - "ON CONFLICT (domain,queue,job_type) DO UPDATE SET ready_count = job_counters.ready_count + EXCLUDED.ready_count, scheduled_count = job_counters.scheduled_count + EXCLUDED.scheduled_count, updated_at = NOW()" - ), - (domain, queue, job_type, 0 if is_sched else 1, 1 if is_sched else 0), - ) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - attrs_json = json.dumps( - { - "idempotent": False, - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - } - ) - cur.execute( - ( - "INSERT INTO job_events(" - "job_id, domain, queue, job_type, event_type, attrs_json, owner_user_id, request_id, trace_id, created_at" - ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())" - ), - ( - int(d.get("id")), - d.get("domain"), - d.get("queue"), - d.get("job_type"), - "job.created", - attrs_json, - d.get("owner_user_id"), - d.get("request_id"), - d.get("trace_id"), - ), - ) - _safe_increment_created_metric( - domain=domain, - queue=queue, - job_type=job_type, - ) - # Emit event for in-process listeners when outbox is disabled - try: - if not JobManager._is_truthy(os.getenv("JOBS_EVENTS_OUTBOX", "")): - emit_job_event( - "job.created", - job=d, - attrs={ - "idempotent": False, - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - }, - ) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass + self._update_gauges(domain=domain, queue=queue, job_type=job_type) with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): - submit_job_audit_event( - "job.created", - job=d, - attrs={ - "idempotent": False, - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - }, - ) + self._assert_invariants(d) + self._emit_create_side_effects(result, backend="sqlite", idempotency_key=idempotency_key) return d - else: - for attempt in range(2): - try: - with conn: - # Domain/user quotas (SQLite) - try: - max_q = self._quota_get("JOBS_QUOTA_MAX_QUEUED", domain, owner_user_id) - if max_q and owner_user_id: - rowq = conn.execute( - "SELECT COUNT(*) FROM jobs WHERE domain=? AND owner_user_id=? AND status='queued'", - (domain, owner_user_id), - ).fetchone() - if int(rowq[0] or 0) >= max_q: - raise ValueError("Quota exceeded: max queued per user/domain") # noqa: TRY003 - spm = self._quota_get("JOBS_QUOTA_SUBMITS_PER_MIN", domain, owner_user_id) - if spm and owner_user_id: - now_str = _now_dt.astimezone(_tz.utc).strftime("%Y-%m-%d %H:%M:%S") - rowm = conn.execute( - "SELECT COUNT(*) FROM jobs WHERE domain=? AND owner_user_id=? AND created_at >= DATETIME(?, '-60 seconds')", - (domain, owner_user_id, now_str), - ).fetchone() - if int(rowm[0] or 0) >= spm: - raise ValueError("Quota exceeded: submits per minute") # noqa: TRY003 - except sqlite3.Error: - pass - if idempotency_key: - # Idempotent create: attempt INSERT OR IGNORE, then SELECT by key - conn.execute( - """ - INSERT OR IGNORE INTO jobs ( - uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, - idempotency_key, payload, result, status, priority, max_retries, - retry_count, available_at, created_at, updated_at, request_id, trace_id - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 'queued', ?, ?, 0, ?, ?, ?, ?, ?) - """, - ( - uuid_val, - domain, - queue, - job_type, - owner_user_id, - project_id, - batch_group, - idempotency_key, - payload_json, - priority, - max_retries, - ( - avail_param_sqlite.strftime("%Y-%m-%d %H:%M:%S") - if avail_param_sqlite - else None - ), - now, - now, - request_id, - trace_id, - ), - ) - inserted = bool(getattr(conn, "total_changes", 0)) - row = conn.execute( - "SELECT * FROM jobs WHERE domain = ? AND queue = ? AND job_type = ? AND idempotency_key = ?", - (domain, queue, job_type, idempotency_key), - ).fetchone() - if row: - d = dict(row) - try: - self._update_gauges(domain=domain, queue=queue, job_type=job_type) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - attrs = { - "idempotent": (not inserted), - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - } - attrs_json = json.dumps(attrs) - conn.execute( - ( - "INSERT INTO job_events(job_id, domain, queue, job_type, event_type, attrs_json, owner_user_id, request_id, trace_id, created_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, DATETIME('now'))" - ), - ( - int(d.get("id")), - d.get("domain"), - d.get("queue"), - d.get("job_type"), - "job.created", - attrs_json, - d.get("owner_user_id"), - request_id, - trace_id, - ), - ) - if inserted: - _safe_increment_created_metric( - domain=domain, - queue=queue, - job_type=job_type, - ) - outbox_enabled = JobManager._is_truthy(os.getenv("JOBS_EVENTS_OUTBOX", "")) - if outbox_enabled: - with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): - submit_job_audit_event( - "job.created", - job={**d, "request_id": request_id, "trace_id": trace_id}, - attrs=attrs, - ) - else: - try: - emit_job_event("job.created", job=d, attrs=attrs) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - return d - # Non-idempotent (or no existing row on IGNORE path): normal insert - conn.execute( - """ - INSERT INTO jobs ( - uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, - idempotency_key, payload, result, status, priority, max_retries, - retry_count, available_at, created_at, updated_at, request_id, trace_id - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 'queued', ?, ?, 0, ?, ?, ?, ?, ?) - """, - ( - uuid_val, - domain, - queue, - job_type, - owner_user_id, - project_id, - batch_group, - idempotency_key, - json.dumps(payload), - priority, - max_retries, - ( - avail_param_sqlite.strftime("%Y-%m-%d %H:%M:%S") - if avail_param_sqlite - else None - ), - now, - now, - request_id, - trace_id, - ), - ) - job_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0] - row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() - d = ( - dict(row) - if row - else { - "id": job_id, - "uuid": uuid_val, - "status": "queued", - "domain": domain, - "queue": queue, - "job_type": job_type, - } - ) - try: - self._update_gauges(domain=domain, queue=queue, job_type=job_type) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - try: - if JobManager._is_truthy(os.getenv("JOBS_COUNTERS_ENABLED", "")): - is_sched = bool(avail_param_sqlite) - # Upsert counters: initialize ready/scheduled appropriately, then increment on conflict - conn.execute( - ( - "INSERT INTO job_counters(domain,queue,job_type,ready_count,scheduled_count,processing_count,quarantined_count) VALUES(?,?,?,?,?,0,0) " - "ON CONFLICT(domain,queue,job_type) DO UPDATE SET ready_count = ready_count + ?, scheduled_count = scheduled_count + ?, updated_at = DATETIME('now')" - ), - ( - domain, - queue, - job_type, - 0 if is_sched else 1, - 1 if is_sched else 0, - 0 if is_sched else 1, - 1 if is_sched else 0, - ), - ) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - attrs_json = json.dumps( - { - "idempotent": False, - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - } - ) - conn.execute( - ( - "INSERT INTO job_events(job_id, domain, queue, job_type, event_type, attrs_json, owner_user_id, request_id, trace_id, created_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, DATETIME('now'))" - ), - ( - int(d.get("id")), - d.get("domain"), - d.get("queue"), - d.get("job_type"), - "job.created", - attrs_json, - d.get("owner_user_id"), - request_id, - trace_id, - ), - ) - _safe_increment_created_metric( - domain=domain, - queue=queue, - job_type=job_type, - ) - # Emit event for in-process listeners when outbox is disabled - try: - if not JobManager._is_truthy(os.getenv("JOBS_EVENTS_OUTBOX", "")): - emit_job_event( - "job.created", - job={**d, "request_id": request_id, "trace_id": trace_id}, - attrs={ - "idempotent": False, - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - }, - ) - except _JOB_NONCRITICAL_EXCEPTIONS: - pass - with contextlib.suppress(_JOB_NONCRITICAL_EXCEPTIONS): - submit_job_audit_event( - "job.created", - job={**d, "request_id": request_id, "trace_id": trace_id}, - attrs={ - "idempotent": False, - "owner_user_id": d.get("owner_user_id"), - "retry_count": int(d.get("retry_count") or 0), - }, - ) - return d except sqlite3.OperationalError as exc: if attempt == 0 and self._sqlite_missing_column_error(exc, "batch_group"): # noqa: SIM102 if self._sqlite_ensure_batch_group(conn): diff --git a/tldw_Server_API/app/core/Jobs/operations/contracts.py b/tldw_Server_API/app/core/Jobs/operations/contracts.py index c4013ea891..bd187e4891 100644 --- a/tldw_Server_API/app/core/Jobs/operations/contracts.py +++ b/tldw_Server_API/app/core/Jobs/operations/contracts.py @@ -56,7 +56,7 @@ class CreateJobCommand: priority: int = 100 max_retries: int = 3 available_at: datetime | None = None - project_id: str | None = None + project_id: int | str | None = None batch_group: str | None = None request_id: str | None = None trace_id: str | None = None @@ -91,8 +91,12 @@ def __post_init__(self) -> None: raise ValueError("only no-transition admission results may include a no-transition reason") if self.outcome is not OperationOutcome.ADMISSION_REJECTED and self.admission_rejection_reason is not None: raise ValueError("only rejected admission results may include a rejection reason") - if self.outcome is not OperationOutcome.APPLIED and self.durable_events: - raise ValueError("only applied admission results may include durable events") + can_include_durable_events = self.outcome is OperationOutcome.APPLIED or ( + self.outcome is OperationOutcome.NO_TRANSITION + and self.no_transition_reason is NoTransitionReason.IDEMPOTENT_EXISTING + ) + if not can_include_durable_events and self.durable_events: + raise ValueError("only applied or idempotent-existing admission results may include durable events") object.__setattr__(self, "row", copy.deepcopy(self.row) if self.row is not None else None) object.__setattr__(self, "durable_events", tuple(copy.deepcopy(event) for event in self.durable_events)) @@ -114,7 +118,12 @@ def applied(cls, *, row: dict[str, Any], durable_events: Sequence[dict[str, Any] ) @classmethod - def existing(cls, *, row: dict[str, Any]) -> "AdmissionResult": + def existing( + cls, + *, + row: dict[str, Any], + durable_events: Sequence[dict[str, Any]] = (), + ) -> "AdmissionResult": """Build a no-transition result for an idempotent existing row.""" return cls( @@ -122,6 +131,7 @@ def existing(cls, *, row: dict[str, Any]) -> "AdmissionResult": row=row, was_inserted=False, no_transition_reason=NoTransitionReason.IDEMPOTENT_EXISTING, + durable_events=durable_events, ) @classmethod diff --git a/tldw_Server_API/app/core/Jobs/operations/postgres/__init__.py b/tldw_Server_API/app/core/Jobs/operations/postgres/__init__.py new file mode 100644 index 0000000000..7a151b9547 --- /dev/null +++ b/tldw_Server_API/app/core/Jobs/operations/postgres/__init__.py @@ -0,0 +1,5 @@ +"""Postgres Jobs admission operations.""" + +from .admission import create_job_admission + +__all__ = ["create_job_admission"] diff --git a/tldw_Server_API/app/core/Jobs/operations/postgres/admission.py b/tldw_Server_API/app/core/Jobs/operations/postgres/admission.py new file mode 100644 index 0000000000..78c3fb2bd4 --- /dev/null +++ b/tldw_Server_API/app/core/Jobs/operations/postgres/admission.py @@ -0,0 +1,295 @@ +"""Postgres-backed Jobs admission operations.""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from contextlib import AbstractContextManager, suppress +from datetime import datetime +from typing import Any + +from tldw_Server_API.app.core.Jobs.operations.contracts import ( + AdmissionRejectionReason, + AdmissionResult, + CreateJobCommand, +) + +_MAX_QUEUED_MESSAGE = "Quota exceeded: max queued per user/domain" +_SUBMITS_PER_MINUTE_MESSAGE = "Quota exceeded: submits per minute" + +try: + import psycopg # type: ignore + + _PG_ERRORS: tuple[type[BaseException], ...] = (psycopg.Error,) +except ImportError: # pragma: no cover - optional dependency path + _PG_ERRORS = () + +_COUNTER_NONCRITICAL_ERRORS: tuple[type[BaseException], ...] = ( + AttributeError, + RuntimeError, + TypeError, + ValueError, + *_PG_ERRORS, +) + + +def _count_from_row(row: Any) -> int: + if row is None: + return 0 + if isinstance(row, dict): + return int(row.get("c") or 0) + return int(row[0] or 0) + + +def _row_to_dict(row: Any) -> dict[str, Any]: + return dict(row) if row is not None else {} + + +def _created_event_fact( + *, + row: dict[str, Any], + idempotent: bool, + request_id: str | None, + trace_id: str | None, +) -> dict[str, Any]: + attrs = { + "idempotent": idempotent, + "owner_user_id": row.get("owner_user_id"), + "retry_count": int(row.get("retry_count") or 0), + } + return { + "event_type": "job.created", + "attrs": attrs, + "request_id": request_id, + "trace_id": trace_id, + "job_id": int(row.get("id")), + "domain": row.get("domain"), + "queue": row.get("queue"), + "job_type": row.get("job_type"), + "owner_user_id": row.get("owner_user_id"), + } + + +def _insert_created_event( + cur: Any, + *, + row: dict[str, Any], + idempotent: bool, + request_id: str | None, + trace_id: str | None, +) -> dict[str, Any]: + event = _created_event_fact(row=row, idempotent=idempotent, request_id=request_id, trace_id=trace_id) + cur.execute( + ( + "INSERT INTO job_events(" + "job_id, domain, queue, job_type, event_type, attrs_json, owner_user_id, request_id, trace_id, created_at" + ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())" + ), + ( + event["job_id"], + event["domain"], + event["queue"], + event["job_type"], + event["event_type"], + json.dumps(event["attrs"]), + event["owner_user_id"], + event["request_id"], + event["trace_id"], + ), + ) + return event + + +def _bump_counters( + cur: Any, + *, + command: CreateJobCommand, + available_at: datetime | None, +) -> None: + is_scheduled = bool(available_at) + cur.execute( + ( + "INSERT INTO job_counters(domain,queue,job_type,ready_count,scheduled_count,processing_count,quarantined_count) VALUES(%s,%s,%s,%s,%s,0,0) " + "ON CONFLICT (domain,queue,job_type) DO UPDATE SET ready_count = job_counters.ready_count + EXCLUDED.ready_count, scheduled_count = job_counters.scheduled_count + EXCLUDED.scheduled_count, updated_at = NOW()" + ), + ( + command.domain, + command.queue, + command.job_type, + 0 if is_scheduled else 1, + 1 if is_scheduled else 0, + ), + ) + + +def _quota_rejection( + cur: Any, + *, + command: CreateJobCommand, + now: datetime, + max_queued_quota: int, + submits_per_minute_quota: int, +) -> AdmissionResult | None: + if not command.owner_user_id: + return None + + try: + if max_queued_quota: + cur.execute( + "SELECT COUNT(*) AS c FROM jobs WHERE domain=%s AND owner_user_id=%s AND status='queued'", + (command.domain, command.owner_user_id), + ) + if _count_from_row(cur.fetchone()) >= max_queued_quota: + return AdmissionResult.rejected( + AdmissionRejectionReason.QUOTA_EXCEEDED, + message=_MAX_QUEUED_MESSAGE, + ) + + if submits_per_minute_quota: + cur.execute( + "SELECT COUNT(*) AS c FROM jobs WHERE domain=%s AND owner_user_id=%s AND created_at >= (%s - interval '60 seconds')", + (command.domain, command.owner_user_id, now), + ) + if _count_from_row(cur.fetchone()) >= submits_per_minute_quota: + return AdmissionResult.rejected( + AdmissionRejectionReason.QUOTA_EXCEEDED, + message=_SUBMITS_PER_MINUTE_MESSAGE, + ) + except _PG_ERRORS: + return None + + return None + + +def _insert_job( + cur: Any, + *, + command: CreateJobCommand, + uuid_value: str, + payload_json: str, + idempotent_insert: bool, +) -> dict[str, Any] | None: + if idempotent_insert: + sql = ( + "INSERT INTO jobs (uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, idempotency_key, payload, result, status, priority, max_retries, retry_count, available_at, created_at, updated_at, request_id, trace_id) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, NULL, 'queued', %s, %s, 0, %s, NOW(), NOW(), %s, %s) " + "ON CONFLICT (domain, queue, job_type, idempotency_key) DO NOTHING RETURNING *" + ) + else: + sql = ( + "INSERT INTO jobs (uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, idempotency_key, payload, result, status, priority, max_retries, retry_count, available_at, created_at, updated_at, request_id, trace_id) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, NULL, 'queued', %s, %s, 0, %s, NOW(), NOW(), %s, %s) RETURNING *" + ) + cur.execute( + sql, + ( + uuid_value, + command.domain, + command.queue, + command.job_type, + command.owner_user_id, + command.project_id, + command.batch_group, + command.idempotency_key, + payload_json, + command.priority, + command.max_retries, + command.available_at if command.available_at else None, + command.request_id, + command.trace_id, + ), + ) + row = cur.fetchone() + return _row_to_dict(row) if row else None + + +def create_job_admission( + conn: Any, + cursor_factory: Callable[[Any], AbstractContextManager[Any]], + *, + command: CreateJobCommand, + uuid_value: str, + now: datetime, + max_queued_quota: int, + submits_per_minute_quota: int, + counters_enabled: bool, +) -> AdmissionResult: + """Create or replay a queued job admission inside a Postgres transaction.""" + + payload_json = json.dumps(command.payload) + + with conn: + with cursor_factory(conn) as cur: + quota_result = _quota_rejection( + cur, + command=command, + now=now, + max_queued_quota=max_queued_quota, + submits_per_minute_quota=submits_per_minute_quota, + ) + if quota_result is not None: + return quota_result + + if command.idempotency_key: + row = _insert_job( + cur, + command=command, + uuid_value=uuid_value, + payload_json=payload_json, + idempotent_insert=True, + ) + inserted = row is not None + if row is None: + cur.execute( + "SELECT * FROM jobs WHERE domain = %s AND queue = %s AND job_type = %s AND idempotency_key = %s", + (command.domain, command.queue, command.job_type, command.idempotency_key), + ) + row = _row_to_dict(cur.fetchone()) + if not row: + row = { + "uuid": uuid_value, + "status": "queued", + "domain": command.domain, + "queue": command.queue, + "job_type": command.job_type, + } + if inserted and counters_enabled: + with suppress(*_COUNTER_NONCRITICAL_ERRORS): + _bump_counters(cur, command=command, available_at=command.available_at) + event = _insert_created_event( + cur, + row=row, + idempotent=not inserted, + request_id=command.request_id, + trace_id=command.trace_id, + ) + if inserted: + return AdmissionResult.applied(row=row, durable_events=(event,)) + return AdmissionResult.existing(row=row, durable_events=(event,)) + + row = _insert_job( + cur, + command=command, + uuid_value=uuid_value, + payload_json=payload_json, + idempotent_insert=False, + ) + if not row: + row = { + "uuid": uuid_value, + "status": "queued", + "domain": command.domain, + "queue": command.queue, + "job_type": command.job_type, + } + if counters_enabled: + with suppress(*_COUNTER_NONCRITICAL_ERRORS): + _bump_counters(cur, command=command, available_at=command.available_at) + event = _insert_created_event( + cur, + row=row, + idempotent=False, + request_id=command.request_id, + trace_id=command.trace_id, + ) + return AdmissionResult.applied(row=row, durable_events=(event,)) diff --git a/tldw_Server_API/app/core/Jobs/operations/sqlite/__init__.py b/tldw_Server_API/app/core/Jobs/operations/sqlite/__init__.py new file mode 100644 index 0000000000..f028700a20 --- /dev/null +++ b/tldw_Server_API/app/core/Jobs/operations/sqlite/__init__.py @@ -0,0 +1,5 @@ +"""SQLite Jobs admission operations.""" + +from .admission import create_job_admission + +__all__ = ["create_job_admission"] diff --git a/tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py b/tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py new file mode 100644 index 0000000000..1965e692ab --- /dev/null +++ b/tldw_Server_API/app/core/Jobs/operations/sqlite/admission.py @@ -0,0 +1,312 @@ +"""SQLite-backed Jobs admission operations.""" + +from __future__ import annotations + +import json +import sqlite3 +from datetime import UTC, datetime +from typing import Any + +from tldw_Server_API.app.core.Jobs.operations.contracts import ( + AdmissionRejectionReason, + AdmissionResult, + CreateJobCommand, +) + +_MAX_QUEUED_MESSAGE = "Quota exceeded: max queued per user/domain" +_SUBMITS_PER_MINUTE_MESSAGE = "Quota exceeded: submits per minute" + + +def _sqlite_timestamp(value: datetime) -> str: + """Return the SQLite timestamp representation used by the Jobs table.""" + + normalized = value + if value.tzinfo is not None: + normalized = value.astimezone(UTC).replace(tzinfo=None) + return normalized.strftime("%Y-%m-%d %H:%M:%S") + + +def _row_to_dict(row: Any) -> dict[str, Any]: + """Convert sqlite row-like values to plain dictionaries.""" + + return dict(row) if row is not None else {} + + +def _created_event_fact( + *, + row: dict[str, Any], + idempotent: bool, + request_id: str | None, + trace_id: str | None, +) -> dict[str, Any]: + """Build the facade-facing fact for a persisted job.created event.""" + + attrs = { + "idempotent": idempotent, + "owner_user_id": row.get("owner_user_id"), + "retry_count": int(row.get("retry_count") or 0), + } + return { + "event_type": "job.created", + "attrs": attrs, + "request_id": request_id, + "trace_id": trace_id, + "job_id": int(row.get("id")), + "domain": row.get("domain"), + "queue": row.get("queue"), + "job_type": row.get("job_type"), + "owner_user_id": row.get("owner_user_id"), + } + + +def _insert_created_event( + conn: sqlite3.Connection, + *, + row: dict[str, Any], + idempotent: bool, + request_id: str | None, + trace_id: str | None, +) -> dict[str, Any]: + """Persist a transactional job.created event and return its fact payload.""" + + event = _created_event_fact(row=row, idempotent=idempotent, request_id=request_id, trace_id=trace_id) + conn.execute( + ( + "INSERT INTO job_events(job_id, domain, queue, job_type, event_type, attrs_json, " + "owner_user_id, request_id, trace_id, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, DATETIME('now'))" + ), + ( + event["job_id"], + event["domain"], + event["queue"], + event["job_type"], + event["event_type"], + json.dumps(event["attrs"]), + event["owner_user_id"], + event["request_id"], + event["trace_id"], + ), + ) + return event + + +def _bump_counters( + conn: sqlite3.Connection, + *, + command: CreateJobCommand, + available_at_sql: str | None, +) -> None: + """Increment ready/scheduled counters for a newly inserted job.""" + + is_scheduled = bool(available_at_sql) + conn.execute( + ( + "INSERT INTO job_counters(domain,queue,job_type,ready_count,scheduled_count,processing_count," + "quarantined_count) VALUES(?,?,?,?,?,0,0) " + "ON CONFLICT(domain,queue,job_type) DO UPDATE SET ready_count = ready_count + ?, " + "scheduled_count = scheduled_count + ?, updated_at = DATETIME('now')" + ), + ( + command.domain, + command.queue, + command.job_type, + 0 if is_scheduled else 1, + 1 if is_scheduled else 0, + 0 if is_scheduled else 1, + 1 if is_scheduled else 0, + ), + ) + + +def _quota_rejection( + conn: sqlite3.Connection, + *, + command: CreateJobCommand, + now_sql: str, + max_queued_quota: int, + submits_per_minute_quota: int, +) -> AdmissionResult | None: + """Return a quota rejection result when admission exceeds configured limits.""" + + if not command.owner_user_id: + return None + + try: + if max_queued_quota: + row = conn.execute( + "SELECT COUNT(*) FROM jobs WHERE domain=? AND owner_user_id=? AND status='queued'", + (command.domain, command.owner_user_id), + ).fetchone() + if int(row[0] if row else 0) >= max_queued_quota: + return AdmissionResult.rejected( + AdmissionRejectionReason.QUOTA_EXCEEDED, + message=_MAX_QUEUED_MESSAGE, + ) + + if submits_per_minute_quota: + row = conn.execute( + "SELECT COUNT(*) FROM jobs WHERE domain=? AND owner_user_id=? AND created_at >= DATETIME(?, '-60 seconds')", + (command.domain, command.owner_user_id, now_sql), + ).fetchone() + if int(row[0] if row else 0) >= submits_per_minute_quota: + return AdmissionResult.rejected( + AdmissionRejectionReason.QUOTA_EXCEEDED, + message=_SUBMITS_PER_MINUTE_MESSAGE, + ) + except sqlite3.Error: + return None + + return None + + +def _insert_job( + conn: sqlite3.Connection, + *, + command: CreateJobCommand, + uuid_value: str, + payload_json: str, + now_sql: str, + available_at_sql: str | None, + ignore_idempotency_conflict: bool, +) -> int | None: + """Insert a queued job and return its row id when SQLite exposes one.""" + + if ignore_idempotency_conflict: + sql = """ + INSERT OR IGNORE INTO jobs ( + uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, + idempotency_key, payload, result, status, priority, max_retries, + retry_count, available_at, created_at, updated_at, request_id, trace_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 'queued', ?, ?, 0, ?, ?, ?, ?, ?) + """ + else: + sql = """ + INSERT INTO jobs ( + uuid, domain, queue, job_type, owner_user_id, project_id, batch_group, + idempotency_key, payload, result, status, priority, max_retries, + retry_count, available_at, created_at, updated_at, request_id, trace_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 'queued', ?, ?, 0, ?, ?, ?, ?, ?) + """ + before_changes = int(getattr(conn, "total_changes", 0)) + conn.execute( + sql, + ( + uuid_value, + command.domain, + command.queue, + command.job_type, + command.owner_user_id, + command.project_id, + command.batch_group, + command.idempotency_key, + payload_json, + command.priority, + command.max_retries, + available_at_sql, + now_sql, + now_sql, + command.request_id, + command.trace_id, + ), + ) + if int(getattr(conn, "total_changes", 0)) <= before_changes: + return None + row = conn.execute("SELECT last_insert_rowid()").fetchone() + return int(row[0]) if row else None + + +def create_job_admission( + conn: sqlite3.Connection, + *, + command: CreateJobCommand, + uuid_value: str, + now: datetime, + max_queued_quota: int, + submits_per_minute_quota: int, + counters_enabled: bool, +) -> AdmissionResult: + """Create or replay a queued job admission inside a SQLite transaction.""" + + payload_json = json.dumps(command.payload) + now_sql = _sqlite_timestamp(now) + available_at_sql = _sqlite_timestamp(command.available_at) if command.available_at else None + + with conn: + quota_result = _quota_rejection( + conn, + command=command, + now_sql=now_sql, + max_queued_quota=max_queued_quota, + submits_per_minute_quota=submits_per_minute_quota, + ) + if quota_result is not None: + return quota_result + + if command.idempotency_key: + row_id = _insert_job( + conn, + command=command, + uuid_value=uuid_value, + payload_json=payload_json, + now_sql=now_sql, + available_at_sql=available_at_sql, + ignore_idempotency_conflict=True, + ) + inserted = row_id is not None + row = _row_to_dict( + conn.execute( + "SELECT * FROM jobs WHERE domain = ? AND queue = ? AND job_type = ? AND idempotency_key = ?", + (command.domain, command.queue, command.job_type, command.idempotency_key), + ).fetchone() + ) + if not row: + row = { + "uuid": uuid_value, + "status": "queued", + "domain": command.domain, + "queue": command.queue, + "job_type": command.job_type, + } + if inserted and counters_enabled: + _bump_counters(conn, command=command, available_at_sql=available_at_sql) + event = _insert_created_event( + conn, + row=row, + idempotent=not inserted, + request_id=command.request_id, + trace_id=command.trace_id, + ) + if inserted: + return AdmissionResult.applied(row=row, durable_events=(event,)) + return AdmissionResult.existing(row=row, durable_events=(event,)) + + job_id = _insert_job( + conn, + command=command, + uuid_value=uuid_value, + payload_json=payload_json, + now_sql=now_sql, + available_at_sql=available_at_sql, + ignore_idempotency_conflict=False, + ) + row = _row_to_dict(conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()) + if not row: + row = { + "id": job_id, + "uuid": uuid_value, + "status": "queued", + "domain": command.domain, + "queue": command.queue, + "job_type": command.job_type, + } + if counters_enabled: + _bump_counters(conn, command=command, available_at_sql=available_at_sql) + event = _insert_created_event( + conn, + row=row, + idempotent=False, + request_id=command.request_id, + trace_id=command.trace_id, + ) + return AdmissionResult.applied(row=row, durable_events=(event,)) diff --git a/tldw_Server_API/tests/Jobs/parity/scenarios.py b/tldw_Server_API/tests/Jobs/parity/scenarios.py index b6a0d7065b..856d26e1dd 100644 --- a/tldw_Server_API/tests/Jobs/parity/scenarios.py +++ b/tldw_Server_API/tests/Jobs/parity/scenarios.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from collections import Counter from collections.abc import Callable @@ -98,6 +99,63 @@ def run_idempotent_create_preserves_original_request_ids_scenario(make_manager: assert replay["trace_id"] == "trace-first" +def run_idempotent_create_replay_event_uses_current_request_ids_scenario(make_manager: ManagerFactory) -> None: + """Verify idempotent replay writes a current-context durable create event.""" + + jm = make_manager() + key = "idem-replay-event-key" + + first = jm.create_job( + domain="parity", + queue="default", + job_type="idem-replay-event", + payload={}, + owner_user_id="owner-1", + idempotency_key=key, + request_id="request-first", + trace_id="trace-first", + ) + replay = jm.create_job( + domain="parity", + queue="default", + job_type="idem-replay-event", + payload={}, + owner_user_id="owner-1", + idempotency_key=key, + request_id="request-replay", + trace_id="trace-replay", + ) + + assert int(first["id"]) == int(replay["id"]) + assert replay["request_id"] == "request-first" + assert replay["trace_id"] == "trace-first" + + events = jm.list_job_events_after( + after_id=0, + domain="parity", + queue="default", + job_type="idem-replay-event", + limit=20, + ) + created_events = [event for event in events if event.get("event_type") == "job.created"] + + assert len(created_events) == 2 + assert created_events[0]["request_id"] == "request-first" + assert created_events[0]["trace_id"] == "trace-first" + assert created_events[1]["request_id"] == "request-replay" + assert created_events[1]["trace_id"] == "trace-replay" + + first_attrs = created_events[0].get("attrs_json") + replay_attrs = created_events[1].get("attrs_json") + if isinstance(first_attrs, str): + first_attrs = json.loads(first_attrs) + if isinstance(replay_attrs, str): + replay_attrs = json.loads(replay_attrs) + + assert first_attrs["idempotent"] is False + assert replay_attrs["idempotent"] is True + + def run_acquire_complete_lifecycle_scenario(make_manager: ManagerFactory) -> None: """Verify the acquire-to-complete lifecycle shape remains backend-neutral.""" diff --git a/tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py b/tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py index cec25a70e6..b646f1ed33 100644 --- a/tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py +++ b/tldw_Server_API/tests/Jobs/parity/test_postgres_parity.py @@ -14,6 +14,7 @@ run_acquire_complete_lifecycle_scenario, run_cancel_terminal_noop_scenario, run_events_outbox_create_complete_scenario, + run_idempotent_create_replay_event_uses_current_request_ids_scenario, run_idempotent_create_preserves_original_request_ids_scenario, run_idempotent_create_scope_scenario, run_renew_stale_lease_noop_scenario, @@ -42,6 +43,14 @@ def test_postgres_idempotent_create_preserves_request_ids(postgres_manager_facto run_idempotent_create_preserves_original_request_ids_scenario(postgres_manager_factory) +def test_postgres_idempotent_create_replay_event_uses_current_request_ids( + postgres_manager_factory: Callable[[], JobManager], +) -> None: + """Run the idempotent replay event context scenario against Postgres.""" + + run_idempotent_create_replay_event_uses_current_request_ids_scenario(postgres_manager_factory) + + def test_postgres_acquire_complete_lifecycle(postgres_manager_factory: Callable[[], JobManager]) -> None: """Run the acquire-complete lifecycle scenario against Postgres.""" diff --git a/tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py b/tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py index afa4bc2d96..d12e76af42 100644 --- a/tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py +++ b/tldw_Server_API/tests/Jobs/parity/test_sqlite_parity.py @@ -13,6 +13,7 @@ run_acquire_complete_lifecycle_scenario, run_cancel_terminal_noop_scenario, run_events_outbox_create_complete_scenario, + run_idempotent_create_replay_event_uses_current_request_ids_scenario, run_idempotent_create_preserves_original_request_ids_scenario, run_idempotent_create_scope_scenario, run_renew_stale_lease_noop_scenario, @@ -46,6 +47,14 @@ def test_sqlite_idempotent_create_preserves_request_ids(sqlite_manager_factory: run_idempotent_create_preserves_original_request_ids_scenario(sqlite_manager_factory) +def test_sqlite_idempotent_create_replay_event_uses_current_request_ids( + sqlite_manager_factory: Callable[[], JobManager], +) -> None: + """Run the idempotent replay event context scenario against SQLite.""" + + run_idempotent_create_replay_event_uses_current_request_ids_scenario(sqlite_manager_factory) + + def test_sqlite_acquire_complete_lifecycle(sqlite_manager_factory: Callable[[], JobManager]) -> None: """Run the acquire-complete lifecycle scenario against SQLite.""" diff --git a/tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py b/tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py new file mode 100644 index 0000000000..a758d7975a --- /dev/null +++ b/tldw_Server_API/tests/Jobs/test_jobs_admission_operations_sqlite.py @@ -0,0 +1,249 @@ +"""Direct SQLite admission operation tests.""" + +from __future__ import annotations + +import json +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import pytest + +from tldw_Server_API.app.core.Jobs.migrations import ensure_jobs_tables +from tldw_Server_API.app.core.Jobs.operations.contracts import ( + AdmissionRejectionReason, + CreateJobCommand, + OperationOutcome, +) +from tldw_Server_API.app.core.Jobs.operations.sqlite.admission import create_job_admission + + +class _FailJobEventsInsertConnection: + """Connection wrapper that fails transactional job_events inserts.""" + + def __init__(self, inner: sqlite3.Connection): + self._inner = inner + + def __enter__(self) -> "_FailJobEventsInsertConnection": + self._inner.__enter__() + return self + + def __exit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: Any) -> bool | None: + return self._inner.__exit__(exc_type, exc, tb) + + def execute(self, sql: str, params: tuple[Any, ...] = ()) -> sqlite3.Cursor: + if "INSERT INTO job_events" in str(sql): + raise sqlite3.OperationalError("forced job_events insert failure") + return self._inner.execute(sql, params) + + def __getattr__(self, name: str) -> Any: + return getattr(self._inner, name) + + +def _open_jobs_db(tmp_path: Path, name: str = "jobs.db") -> tuple[Path, sqlite3.Connection]: + db_path = ensure_jobs_tables(tmp_path / name) + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + return db_path, conn + + +def _command( + *, + job_type: str = "insert", + idempotency_key: str | None = None, + request_id: str = "req-1", + trace_id: str = "trace-1", +) -> CreateJobCommand: + return CreateJobCommand( + domain="admission", + queue="default", + job_type=job_type, + payload={"x": 1}, + owner_user_id="u1", + idempotency_key=idempotency_key, + priority=5, + max_retries=3, + request_id=request_id, + trace_id=trace_id, + ) + + +def _created_events(conn: sqlite3.Connection, job_type: str) -> list[sqlite3.Row]: + return list( + conn.execute( + ( + "SELECT * FROM job_events " + "WHERE domain = ? AND queue = ? AND job_type = ? AND event_type = 'job.created' " + "ORDER BY id" + ), + ("admission", "default", job_type), + ) + ) + + +def test_sqlite_admission_inserts_job_event_and_counter(tmp_path: Path) -> None: + _db_path, conn = _open_jobs_db(tmp_path) + with conn: + result = create_job_admission( + conn, + command=_command(), + uuid_value="uuid-insert", + now=datetime(2026, 1, 1, tzinfo=timezone.utc), + max_queued_quota=0, + submits_per_minute_quota=0, + counters_enabled=True, + ) + + assert result.outcome is OperationOutcome.APPLIED + assert result.inserted is True + assert result.row is not None + assert result.row["status"] == "queued" + assert result.row["request_id"] == "req-1" + assert result.row["trace_id"] == "trace-1" + + events = _created_events(conn, "insert") + assert len(events) == 1 + assert events[0]["request_id"] == "req-1" + assert events[0]["trace_id"] == "trace-1" + assert json.loads(events[0]["attrs_json"])["idempotent"] is False + + counter = conn.execute( + "SELECT ready_count, scheduled_count FROM job_counters WHERE domain = ? AND queue = ? AND job_type = ?", + ("admission", "default", "insert"), + ).fetchone() + assert dict(counter) == {"ready_count": 1, "scheduled_count": 0} + + +def test_sqlite_admission_idempotent_existing_writes_replay_event_with_current_context(tmp_path: Path) -> None: + _db_path, conn = _open_jobs_db(tmp_path) + first = create_job_admission( + conn, + command=_command(job_type="idem", idempotency_key="same"), + uuid_value="uuid-idem-1", + now=datetime(2026, 1, 1, tzinfo=timezone.utc), + max_queued_quota=0, + submits_per_minute_quota=0, + counters_enabled=True, + ) + replay = create_job_admission( + conn, + command=_command( + job_type="idem", + idempotency_key="same", + request_id="req-replay", + trace_id="trace-replay", + ), + uuid_value="uuid-idem-2", + now=datetime(2026, 1, 1, 0, 0, 5, tzinfo=timezone.utc), + max_queued_quota=0, + submits_per_minute_quota=0, + counters_enabled=True, + ) + + assert first.outcome is OperationOutcome.APPLIED + assert replay.outcome is OperationOutcome.NO_TRANSITION + assert replay.inserted is False + assert replay.row is not None + assert replay.row["uuid"] == "uuid-idem-1" + assert replay.row["request_id"] == "req-1" + assert replay.row["trace_id"] == "trace-1" + + events = _created_events(conn, "idem") + assert len(events) == 2 + assert events[0]["request_id"] == "req-1" + assert events[0]["trace_id"] == "trace-1" + assert json.loads(events[0]["attrs_json"])["idempotent"] is False + assert events[1]["request_id"] == "req-replay" + assert events[1]["trace_id"] == "trace-replay" + assert json.loads(events[1]["attrs_json"])["idempotent"] is True + assert replay.durable_events == ( + { + "event_type": "job.created", + "attrs": {"idempotent": True, "owner_user_id": "u1", "retry_count": 0}, + "request_id": "req-replay", + "trace_id": "trace-replay", + "job_id": first.row["id"], + "domain": "admission", + "queue": "default", + "job_type": "idem", + "owner_user_id": "u1", + }, + ) + + +def test_sqlite_admission_rejects_max_queued_quota(tmp_path: Path) -> None: + _db_path, conn = _open_jobs_db(tmp_path) + create_job_admission( + conn, + command=_command(job_type="quota"), + uuid_value="uuid-quota-1", + now=datetime(2026, 1, 1, tzinfo=timezone.utc), + max_queued_quota=0, + submits_per_minute_quota=0, + counters_enabled=False, + ) + + result = create_job_admission( + conn, + command=_command(job_type="quota-two"), + uuid_value="uuid-quota-2", + now=datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc), + max_queued_quota=1, + submits_per_minute_quota=0, + counters_enabled=False, + ) + + assert result.outcome is OperationOutcome.ADMISSION_REJECTED + assert result.admission_rejection_reason is AdmissionRejectionReason.QUOTA_EXCEEDED + assert result.message == "Quota exceeded: max queued per user/domain" + + +def test_sqlite_admission_rejects_submits_per_minute_quota(tmp_path: Path) -> None: + _db_path, conn = _open_jobs_db(tmp_path) + create_job_admission( + conn, + command=_command(job_type="spm"), + uuid_value="uuid-spm-1", + now=datetime(2026, 1, 1, tzinfo=timezone.utc), + max_queued_quota=0, + submits_per_minute_quota=0, + counters_enabled=False, + ) + + result = create_job_admission( + conn, + command=_command(job_type="spm-two"), + uuid_value="uuid-spm-2", + now=datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc), + max_queued_quota=0, + submits_per_minute_quota=1, + counters_enabled=False, + ) + + assert result.outcome is OperationOutcome.ADMISSION_REJECTED + assert result.admission_rejection_reason is AdmissionRejectionReason.QUOTA_EXCEEDED + assert result.message == "Quota exceeded: submits per minute" + + +def test_sqlite_admission_rolls_back_job_when_created_event_insert_fails(tmp_path: Path) -> None: + db_path, inner = _open_jobs_db(tmp_path) + wrapped = _FailJobEventsInsertConnection(inner) + + with pytest.raises(sqlite3.OperationalError, match="job_events insert failure"): + create_job_admission( + wrapped, + command=_command(job_type="event-fail"), + uuid_value="uuid-event-fail", + now=datetime(2026, 1, 1, tzinfo=timezone.utc), + max_queued_quota=0, + submits_per_minute_quota=0, + counters_enabled=True, + ) + + with sqlite3.connect(db_path) as conn: + count = conn.execute( + "SELECT COUNT(*) FROM jobs WHERE domain = ? AND queue = ? AND job_type = ?", + ("admission", "default", "event-fail"), + ).fetchone()[0] + assert count == 0 diff --git a/tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py b/tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py index 8f67cde70a..5ee903a94b 100644 --- a/tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py +++ b/tldw_Server_API/tests/Jobs/test_jobs_fault_injection_sqlite.py @@ -200,6 +200,51 @@ def test_idempotent_existing_create_fails_when_job_created_event_insert_fails_sq raise AssertionError("failing idempotent create-existing call must not add a new job row") +def test_idempotent_replay_in_process_event_uses_current_context_sqlite(monkeypatch, tmp_path): + import tldw_Server_API.app.core.Jobs.manager as mgr + + monkeypatch.setenv("JOBS_EVENTS_OUTBOX", "false") + emitted: list[dict] = [] + + def _emit(_event_type, **kwargs): + emitted.append(dict(kwargs)) + + monkeypatch.setattr(mgr, "emit_job_event", _emit) + + db_path = tmp_path / "jobs_create_idem_current_context.db" + ensure_jobs_tables(db_path) + jm = JobManager(db_path) + + jm.create_job( + domain="ps", + queue="default", + job_type="event-idem-current-context", + payload={"x": 1}, + owner_user_id="u", + idempotency_key="stable-idem-current-context", + request_id="request-first", + trace_id="trace-first", + ) + emitted.clear() + + replay = jm.create_job( + domain="ps", + queue="default", + job_type="event-idem-current-context", + payload={"x": 2}, + owner_user_id="u", + idempotency_key="stable-idem-current-context", + request_id="request-replay", + trace_id="trace-replay", + ) + + assert replay["request_id"] == "request-first" + assert replay["trace_id"] == "trace-first" + assert len(emitted) == 1 + assert emitted[0]["job"]["request_id"] == "request-replay" + assert emitted[0]["job"]["trace_id"] == "trace-replay" + + def test_create_failure_does_not_increment_created_metric_sqlite(monkeypatch, tmp_path): import tldw_Server_API.app.core.Jobs.manager as mgr diff --git a/tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py b/tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py index 95fa8f6f88..9fffdf64a1 100644 --- a/tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py +++ b/tldw_Server_API/tests/Jobs/test_jobs_manager_pg_fakes_extended.py @@ -277,6 +277,34 @@ def test_pg_idempotent_create_uses_current_request_context_for_job_created_event assert event_params[8] == "new-trace" +@pytest.mark.unit +def test_pg_non_idempotent_create_uses_command_context_for_job_created_event(monkeypatch, tmp_path): + monkeypatch.setenv("JOBS_DB_URL", "postgresql://fake") + jm = JobManager(db_path=tmp_path / "dummy.db") + jm.backend = "postgres" + + jobs = {} + cursor = FakePGCursor(jobs) + monkeypatch.setattr(jm, "_connect", lambda: FakePGConn()) + monkeypatch.setattr(jm, "_pg_cursor", lambda conn: cursor) + + row = jm.create_job( + domain="pg", + queue="default", + job_type="x", + payload={}, + owner_user_id=None, + request_id="request-current", + trace_id="trace-current", + ) + + assert row["status"] == "queued" + assert cursor.job_event_inserts + event_params = cursor.job_event_inserts[-1] + assert event_params[7] == "request-current" + assert event_params[8] == "trace-current" + + @pytest.mark.unit def test_pg_batch_complete_encrypts_results(monkeypatch, tmp_path): # Enable encryption for domain SECURE and provide AES key diff --git a/tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py b/tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py index 165136e12c..35b13b69ec 100644 --- a/tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py +++ b/tldw_Server_API/tests/Jobs/test_jobs_operation_contracts.py @@ -59,6 +59,19 @@ def test_admission_result_distinguishes_inserted_and_existing_rows() -> None: assert rejected.admission_rejection_reason is AdmissionRejectionReason.QUEUE_PAUSED +def test_admission_existing_can_report_idempotent_durable_event() -> None: + """Verify idempotent create replays can report the persisted replay event.""" + + result = AdmissionResult.existing( + row={"id": 1, "status": "queued"}, + durable_events=({"event_type": "job.created", "attrs": {"idempotent": True}},), + ) + + assert result.outcome is OperationOutcome.NO_TRANSITION + assert result.no_transition_reason is NoTransitionReason.IDEMPOTENT_EXISTING + assert result.durable_events == ({"event_type": "job.created", "attrs": {"idempotent": True}},) + + def test_admission_result_rejects_inconsistent_states() -> None: """Verify invalid admission result state combinations are rejected.""" @@ -81,10 +94,10 @@ def test_admission_result_rejects_inconsistent_states() -> None: no_transition_reason=NoTransitionReason.IDEMPOTENT_EXISTING, ) - with pytest.raises(ValueError, match="only applied admission"): + with pytest.raises(ValueError, match="durable events"): AdmissionResult( outcome=OperationOutcome.NO_TRANSITION, - no_transition_reason=NoTransitionReason.IDEMPOTENT_EXISTING, + no_transition_reason=NoTransitionReason.MISSING, durable_events=({"event_type": "job.noop"},), )