Skip to content

refactor(storage): tighten relational Job persistence #209

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kfstorm merged 1 commit into from
Aug 24, 2026
Merged

refactor(storage): tighten relational Job persistence #209

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cueweaver/application/database.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ class JobRow(Base):
stream_index: Mapped[int | None] = mapped_column(Integer)
target_language_code: Mapped[str] = mapped_column(String, nullable=False)
term_map_mode: Mapped[str] = mapped_column(String, nullable=False)
# These columns are immutable Job-owned snapshot metadata, not live-map fields.
term_map_id: Mapped[str | None] = mapped_column(String)
term_map_name: Mapped[str | None] = mapped_column(String)
output_path: Mapped[str] = mapped_column(String, nullable=False)
Expand Down
33 changes: 15 additions & 18 deletions cueweaver/application/jobs/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,9 +206,7 @@ def create(self, request: CreateJobRequest) -> dict[str, object]:
}
if subtitle is None:
record["extraction"] = None
self._write_record(job_id, record)
with self._lock:
self._records[job_id] = record
self._write_record(record)
self._pending.put(job_id)
return self._record_with_queue_position(record)

Expand Down Expand Up @@ -305,8 +303,7 @@ def retry(self, job_id: str) -> dict[str, object]:
"message": error.message,
**context,
}
self._write_record(job_id, failed_record)
self._records[job_id] = failed_record
self._write_record(failed_record)
raise safe_error from error
with self._lock:
retry_record = copy_job_record(record)
Expand All @@ -331,8 +328,7 @@ def retry(self, job_id: str) -> dict[str, object]:
self._base_output_path(retry_request).relative_to(self._media_root)
)
retry_record["queue_sequence"] = next_queue_sequence
self._write_record(job_id, retry_record)
self._records[job_id] = retry_record
self._write_record(retry_record)
self._next_queue_sequence = next_queue_sequence
self._pending.put(job_id)
return self._record_with_queue_position(retry_record)
Expand All @@ -357,8 +353,7 @@ def cancel(self, job_id: str) -> dict[str, object]:
)
cancelled_record["finished_at"] = cancelled_at
cancelled_record["error"] = None
self._write_record(job_id, cancelled_record)
self._records[job_id] = cancelled_record
self._write_record(cancelled_record)
return self._record_with_queue_position(cancelled_record)

def delete(self, job_id: str) -> dict[str, object]:
Expand Down Expand Up @@ -861,13 +856,14 @@ def _load_records(self) -> None:
record = copy_job_record(loaded_record)
if status in {"Extracting", "Translating"}:
record = _interrupted_record(record)
self._write_record(job_id, record)
self._write_record(record)
elif status == "Queued":
self._recovered_queue_ids.append(job_id)
if status not in {"Extracting", "Translating"}:
self._records[job_id] = copy_job_record(record)
self._next_queue_sequence = max(
self._next_queue_sequence, queue_sequence(record)
)
self._records[job_id] = copy_job_record(record)

def _run(self) -> None:
while True:
Expand Down Expand Up @@ -1045,10 +1041,9 @@ def _persist_embedded_progress(
}
transition_status(record, progress.phase, at=_timestamp())
try:
self._write_record(job_id, record)
self._write_record(record)
except Exception as error:
raise JobExecutionProgressPersistenceError from error
self._records[job_id] = record
return True

def _prepare_execution(
Expand All @@ -1071,7 +1066,7 @@ def _prepare_execution(
record["started_at"] = started_at
output_path = self._execution_output_path(request)
request["output_path"] = str(output_path.relative_to(self._media_root))
self._write_record(job_id, record)
self._write_record(record)
return request, embedded, self._jobs_root / job_id, record

def _execution_output_path(self, request: dict[str, object]) -> Path:
Expand Down Expand Up @@ -1126,13 +1121,13 @@ def _finish(
transition_status(record, status, at=finished_at, terminal=True)
record["finished_at"] = finished_at
record["error"] = error
self._write_record(job_id, record)
self._write_record(record)

def _finish_interrupted(self, job_id: str) -> None:
with self._lock:
interrupted = _interrupted_record(self._records[job_id])
try:
self._write_record(job_id, interrupted)
self._write_record(interrupted)
except (OSError, ServiceError) as error:
logger.warning(
"Could not persist interrupted Job %s during shutdown: %s",
Expand All @@ -1157,17 +1152,19 @@ def _mark_failed_after_worker_error(self, job_id: str, error: Exception) -> None
"message": "Job execution could not be persisted",
}
try:
self._write_record(job_id, record)
self._write_record(record)
except Exception as persistence_error:
logger.error(
"Could not persist worker failure for Job %s: %s",
job_id,
persistence_error,
)

def _write_record(self, job_id: str, record: dict[str, object]) -> None:
def _write_record(self, record: dict[str, object]) -> None:
persisted = copy_job_record(record)
self._record_store.write(persisted)
job_id = persisted["id"]
assert isinstance(job_id, str)
self._records[job_id] = persisted

def _check_jobs_root(self) -> None:
Expand Down
97 changes: 57 additions & 40 deletions cueweaver/application/jobs/store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,39 @@ def __init__(self, database: SqliteDatabase) -> None:
def load(self) -> list[JobRecord]:
try:
with self._database.read_session() as session:
rows = session.scalars(
job_rows = session.scalars(
select(JobRow).order_by(
JobRow.queue_sequence, JobRow.created_at, JobRow.id
)
).all()
return [_record_from_row(session, row) for row in rows]
history_rows = session.scalars(
select(JobStatusHistoryRow).order_by(
JobStatusHistoryRow.job_id, JobStatusHistoryRow.sequence
)
).all()
snapshot_rows = session.scalars(
select(JobTermMapSnapshotRow).order_by(
JobTermMapSnapshotRow.job_id, JobTermMapSnapshotRow.position
)
).all()
histories_by_job: dict[str, list[JobStatusHistoryRow]] = {}
for history_row in history_rows:
histories_by_job.setdefault(history_row.job_id, []).append(
history_row
)
snapshots_by_job: dict[str, list[JobTermMapSnapshotRow]] = {}
for snapshot_row in snapshot_rows:
snapshots_by_job.setdefault(snapshot_row.job_id, []).append(
snapshot_row
)
return [
_record_from_row(
row,
histories_by_job.get(row.id, []),
snapshots_by_job.get(row.id, []),
)
for row in job_rows
]
except (sqlite3.Error, SQLAlchemyError) as error:
raise ServiceError(
"database_unavailable", "Job records cannot be loaded"
Expand Down Expand Up @@ -104,6 +131,7 @@ def _upsert_row(session: Session, record: JobRecord) -> None:
extraction = record.get("extraction")
extraction_values = extraction if isinstance(extraction, dict) else {}
row = session.get(JobRow, record["id"])
is_new_job = row is None
if row is None:
row = JobRow(id=str(record["id"]))
session.add(row)
Expand All @@ -122,6 +150,9 @@ def _upsert_row(session: Session, record: JobRecord) -> None:
term_map = _set_request_fields(row, request)
_set_extraction_fields(row, extraction_values)

if is_new_job:
_set_snapshot_fields(row, term_map, session)

session.execute(
delete(JobStatusHistoryRow).where(JobStatusHistoryRow.job_id == row.id)
)
Expand All @@ -140,36 +171,32 @@ def _upsert_row(session: Session, record: JobRecord) -> None:
)
)

# A Job owns this snapshot. Once populated, later Job writes cannot replace it.
has_snapshot = (
session.scalar(
select(JobTermMapSnapshotRow.position)
.where(JobTermMapSnapshotRow.job_id == row.id)
.limit(1)

def _set_snapshot_fields(row: JobRow, term_map: object, session: Session) -> None:
if not isinstance(term_map, dict):
return
row.term_map_id = _optional_str(term_map.get("id"))
row.term_map_name = _optional_str(term_map.get("name"))
content = term_map.get("content")
if not isinstance(content, dict):
return
for position, (source, target) in enumerate(content.items()):
session.add(
JobTermMapSnapshotRow(
job_id=row.id,
position=position,
source=str(source),
source_folded=str(source).casefold(),
target=str(target),
)
)
is not None
)
if not has_snapshot and isinstance(term_map, dict):
content = term_map.get("content")
if isinstance(content, dict):
for position, (source, target) in enumerate(content.items()):
session.add(
JobTermMapSnapshotRow(
job_id=row.id,
position=position,
source=str(source),
source_folded=str(source).casefold(),
target=str(target),
)
)


def _record_from_row(session: Session, row: JobRow) -> JobRecord:
snapshot_rows = session.scalars(
select(JobTermMapSnapshotRow)
.where(JobTermMapSnapshotRow.job_id == row.id)
.order_by(JobTermMapSnapshotRow.position)
).all()
def _record_from_row(
row: JobRow,
history_rows: list[JobStatusHistoryRow],
snapshot_rows: list[JobTermMapSnapshotRow],
) -> JobRecord:
content = {item.source: item.target for item in snapshot_rows}
term_map: dict[str, object] | None = None
if row.term_map_id is not None:
Expand Down Expand Up @@ -214,11 +241,7 @@ def _record_from_row(session: Session, row: JobRow) -> JobRecord:
"started_at": item.started_at,
"finished_at": item.finished_at,
}
for item in session.scalars(
select(JobStatusHistoryRow)
.where(JobStatusHistoryRow.job_id == row.id)
.order_by(JobStatusHistoryRow.sequence)
).all()
for item in history_rows
],
}
if row.stream_index is not None:
Expand Down Expand Up @@ -256,12 +279,6 @@ def _set_request_fields(row: JobRow, request: dict[str, object]) -> object:
row.target_language_code = str(request["target_language_code"])
row.term_map_mode = str(request["term_map_mode"])
term_map = request.get("term_map")
row.term_map_id = (
_optional_str(term_map.get("id")) if isinstance(term_map, dict) else None
)
row.term_map_name = (
_optional_str(term_map.get("name")) if isinstance(term_map, dict) else None
)
row.output_path = str(request["output_path"])
row.source_format = str(request["source_format"])
row.dynamic_terminology_enabled = bool(request["dynamic_terminology_enabled"])
Expand Down
Loading
Loading