fix: canonicalize interval history identity
This commit is contained in:
@@ -16,11 +16,15 @@ from datetime import datetime, timezone
|
||||
from typing import Any, Protocol
|
||||
|
||||
from wf_scheduling.models import OccurrenceKind, OccurrenceRecord
|
||||
from wf_scheduling.occurrences import occurrence_id
|
||||
from wf_scheduling.occurrences import _require_aware, occurrence_id
|
||||
|
||||
UTC = timezone.utc
|
||||
|
||||
TERMINAL_KINDS: tuple[str, str, str] = ("completed", "interrupted", "failed")
|
||||
TERMINAL_KINDS: tuple[OccurrenceKind, ...] = (
|
||||
"completed",
|
||||
"interrupted",
|
||||
"failed",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -53,7 +57,7 @@ class HistoryRecorder(Protocol):
|
||||
self,
|
||||
schedule_id: str,
|
||||
run_id: str,
|
||||
kind: str,
|
||||
kind: OccurrenceKind,
|
||||
checkpoint_id: str | None = None,
|
||||
) -> bool:
|
||||
"""Whether this exact stopped result was already reconciled."""
|
||||
@@ -62,15 +66,32 @@ class HistoryRecorder(Protocol):
|
||||
|
||||
def entry_occurrence_id(entry: HistoryEntry, created: datetime) -> str:
|
||||
"""Derive the history occurrence id with the scheduler's standing rules."""
|
||||
interval = _normalized_interval(entry)
|
||||
if entry.resolved_at is not None:
|
||||
return occurrence_id(entry.schedule_id, entry.resolved_at)
|
||||
if entry.interval_start is not None and entry.interval_end is not None:
|
||||
start = entry.interval_start.isoformat()
|
||||
end = entry.interval_end.isoformat()
|
||||
return f"{entry.schedule_id}|summary|{start}|{end}"
|
||||
if interval is not None:
|
||||
start, end = interval
|
||||
return f"{entry.schedule_id}|summary|{start.isoformat()}|{end.isoformat()}"
|
||||
return f"{entry.schedule_id}|summary|{created.isoformat()}"
|
||||
|
||||
|
||||
def _normalized_interval(
|
||||
entry: HistoryEntry,
|
||||
) -> tuple[datetime, datetime] | None:
|
||||
"""Validate and normalize an optional summary interval to UTC."""
|
||||
start, end = entry.interval_start, entry.interval_end
|
||||
if (start is None) != (end is None):
|
||||
raise ValueError("interval_start and interval_end must be provided together")
|
||||
if start is None:
|
||||
assert end is None
|
||||
return None
|
||||
assert end is not None
|
||||
return (
|
||||
_require_aware(start, op="interval_start").astimezone(UTC),
|
||||
_require_aware(end, op="interval_end").astimezone(UTC),
|
||||
)
|
||||
|
||||
|
||||
class FileScheduleHistoryRecorder:
|
||||
"""History recorder backed by a file schedule store."""
|
||||
|
||||
@@ -81,6 +102,7 @@ class FileScheduleHistoryRecorder:
|
||||
created = (
|
||||
entry.created_at if entry.created_at is not None else datetime.now(UTC)
|
||||
)
|
||||
interval = _normalized_interval(entry)
|
||||
self.schedule_store.append_history(
|
||||
OccurrenceRecord(
|
||||
schedule_id=entry.schedule_id,
|
||||
@@ -93,8 +115,8 @@ class FileScheduleHistoryRecorder:
|
||||
admitted_at=entry.admitted_at,
|
||||
started_at=entry.started_at,
|
||||
checkpoint_id=entry.checkpoint_id,
|
||||
interval_start=entry.interval_start,
|
||||
interval_end=entry.interval_end,
|
||||
interval_start=interval[0] if interval is not None else None,
|
||||
interval_end=interval[1] if interval is not None else None,
|
||||
interval_count=entry.interval_count,
|
||||
created_at=created,
|
||||
)
|
||||
@@ -104,7 +126,7 @@ class FileScheduleHistoryRecorder:
|
||||
self,
|
||||
schedule_id: str,
|
||||
run_id: str,
|
||||
kind: str,
|
||||
kind: OccurrenceKind,
|
||||
checkpoint_id: str | None = None,
|
||||
) -> bool:
|
||||
return self.schedule_store.has_history_entry(
|
||||
|
||||
@@ -26,7 +26,13 @@ from pathlib import Path
|
||||
from threading import RLock
|
||||
from typing import Any, cast
|
||||
|
||||
from .models import OccurrenceRecord, PendingCandidate, Schedule, ensure_schedule_id
|
||||
from .models import (
|
||||
OccurrenceKind,
|
||||
OccurrenceRecord,
|
||||
PendingCandidate,
|
||||
Schedule,
|
||||
ensure_schedule_id,
|
||||
)
|
||||
|
||||
UTC = timezone.utc
|
||||
|
||||
@@ -48,16 +54,18 @@ def schedule_store_transaction(store: object) -> Iterator[None]:
|
||||
|
||||
|
||||
def _history_identity(record: OccurrenceRecord) -> tuple[object, ...]:
|
||||
"""Return fields that identify one durable occurrence transition."""
|
||||
"""Return the canonical fields that identify one history transition.
|
||||
|
||||
``occurrence_id`` already contains the UTC-normalized resolved instant or
|
||||
summary interval. Do not compare the original datetime fields as well:
|
||||
equivalent offsets must remain one idempotent history entry.
|
||||
"""
|
||||
return (
|
||||
record.schedule_id,
|
||||
record.occurrence_id,
|
||||
record.kind,
|
||||
record.resolved_at,
|
||||
record.run_id,
|
||||
record.checkpoint_id,
|
||||
record.interval_start,
|
||||
record.interval_end,
|
||||
)
|
||||
|
||||
|
||||
@@ -237,7 +245,7 @@ class FileScheduleStore:
|
||||
schedule_id: str,
|
||||
*,
|
||||
run_id: str,
|
||||
kind: str,
|
||||
kind: OccurrenceKind,
|
||||
checkpoint_id: str | None = None,
|
||||
) -> bool:
|
||||
"""Whether this exact stopped result already has a history entry.
|
||||
|
||||
Reference in New Issue
Block a user