SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(observatory): reconcile quarantined recorded jobs

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-31 17:59:03 +03:00
parent 56bfeaee7b
commit 1766c0bdb2
2 changed files with 624 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/observatory/recorded_jobs.py
+457
View File
@@ -36,6 +36,9 @@ from k1link.artifacts import utc_now_iso
OBSERVATORY_RECORDED_JOB_SCHEMA: Final = "missioncore.observatory-recorded-job/v1"
OBSERVATORY_RECORDED_JOB_REQUEST_SCHEMA: Final = "missioncore.observatory-recorded-job-request/v1"
OBSERVATORY_RECORDED_CLAIM_SCHEMA: Final = "missioncore.observatory-recorded-job-claim/v1"
OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA: Final = (
"missioncore.observatory-recorded-job-reconciliation/v1"
)
OBSERVATORY_LIVE_LEASE_SCHEMA: Final = "missioncore.observatory-live-k1-lease/v1"
OBSERVATORY_LIVE_LEASE_REQUEST_SCHEMA: Final = "missioncore.observatory-live-k1-lease-request/v1"
RECORDED_JOB_DATABASE_NAME: Final = "observatory-recorded-jobs.sqlite3"
@@ -152,6 +155,29 @@ CREATE TABLE IF NOT EXISTS observatory_recorded_claim_receipts (
FOREIGN KEY (job_id) REFERENCES observatory_recorded_jobs(job_id)
);
CREATE TABLE IF NOT EXISTS observatory_recorded_reconciliations (
reconciliation_id TEXT PRIMARY KEY,
request_sha256 TEXT NOT NULL,
receipt_sha256 TEXT NOT NULL UNIQUE,
job_id TEXT NOT NULL UNIQUE,
expected_claim_generation INTEGER NOT NULL
CHECK (expected_claim_generation > 0),
expected_terminal_code TEXT NOT NULL,
quarantined_terminal_message TEXT NOT NULL,
quarantined_at_utc TEXT NOT NULL,
quarantined_claim_token_sha256 TEXT,
resource_release_attestation_id TEXT NOT NULL,
resource_release_attestation_sha256 TEXT NOT NULL,
resource_release_operator_id TEXT NOT NULL,
resource_release_evidence_sha256 TEXT NOT NULL,
resource_release_attested_at_utc TEXT NOT NULL,
resources_released INTEGER NOT NULL CHECK (resources_released = 1),
failure_code TEXT NOT NULL,
reason TEXT NOT NULL,
reconciled_at_utc TEXT NOT NULL,
FOREIGN KEY (job_id) REFERENCES observatory_recorded_jobs(job_id)
);
CREATE TABLE IF NOT EXISTS observatory_live_leases (
lease_id TEXT PRIMARY KEY,
trigger_id TEXT NOT NULL UNIQUE,
@@ -629,6 +655,168 @@ class ObservatoryRecordedClaim:
}
@dataclass(frozen=True, slots=True)
class ObservatoryRecordedResourceReleaseAttestation:
"""Operator-owned evidence that a stale Worker no longer owns resources."""
attestation_id: str
operator_id: str
job_id: str
claim_generation: int
resources_released: bool
evidence_sha256: str
attested_at_utc: str
def __post_init__(self) -> None:
_validate_pattern(self.attestation_id, _IDEMPOTENCY_KEY, "attestation id")
_validate_pattern(self.operator_id, _IDENTIFIER, "operator id")
_validate_pattern(self.job_id, _JOB_ID, "recorded job id")
_validate_positive_int(self.claim_generation, "claim generation")
if self.resources_released is not True:
raise ValueError("resource-release attestation must explicitly release resources")
_validate_digest(self.evidence_sha256, "resource-release evidence sha256")
_validate_timestamp(self.attested_at_utc, "resource-release attestation timestamp")
@property
def attestation_sha256(self) -> str:
return _sha256(self.identity_document())
def identity_document(self) -> dict[str, object]:
return {
"schema_version": OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA,
"attestation_id": self.attestation_id,
"operator_id": self.operator_id,
"job_id": self.job_id,
"claim_generation": self.claim_generation,
"resources_released": self.resources_released,
"evidence_sha256": self.evidence_sha256,
"attested_at_utc": self.attested_at_utc,
}
def as_dict(self) -> dict[str, object]:
return {
**self.identity_document(),
"attestation_sha256": self.attestation_sha256,
}
@dataclass(frozen=True, slots=True)
class ObservatoryRecordedReconciliationRequest:
"""Fail-closed operator request to resolve one exact quarantined generation."""
reconciliation_id: str
job_id: str
expected_claim_generation: int
expected_terminal_code: str
resource_release_attestation: ObservatoryRecordedResourceReleaseAttestation
failure_code: str
reason: str
def __post_init__(self) -> None:
_validate_pattern(self.reconciliation_id, _IDEMPOTENCY_KEY, "reconciliation id")
_validate_pattern(self.job_id, _JOB_ID, "recorded job id")
_validate_positive_int(self.expected_claim_generation, "expected claim generation")
_validate_pattern(
self.expected_terminal_code,
_IDENTIFIER,
"expected terminal code",
)
_validate_pattern(self.failure_code, _IDENTIFIER, "reconciliation failure code")
_validate_text(self.reason, "reconciliation reason", max_length=1_000)
if self.resource_release_attestation.job_id != self.job_id:
raise ValueError("resource-release attestation is bound to another job")
if (
self.resource_release_attestation.claim_generation
!= self.expected_claim_generation
):
raise ValueError("resource-release attestation is bound to another generation")
@property
def request_sha256(self) -> str:
return _sha256(
{
"schema_version": OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA,
"reconciliation_id": self.reconciliation_id,
"job_id": self.job_id,
"expected_claim_generation": self.expected_claim_generation,
"expected_terminal_code": self.expected_terminal_code,
"resource_release_attestation": self.resource_release_attestation.as_dict(),
"failure_code": self.failure_code,
"reason": self.reason,
}
)
@dataclass(frozen=True, slots=True)
class ObservatoryRecordedReconciliationReceipt:
"""Immutable audit receipt for an operator-resolved recorded-job quarantine."""
reconciliation_id: str
request_sha256: str
receipt_sha256: str
job_id: str
expected_claim_generation: int
expected_terminal_code: str
quarantined_terminal_message: str
quarantined_at_utc: str
quarantined_claim_token_sha256: str | None
resource_release_attestation: ObservatoryRecordedResourceReleaseAttestation
failure_code: str
reason: str
reconciled_at_utc: str
def __post_init__(self) -> None:
request = ObservatoryRecordedReconciliationRequest(
reconciliation_id=self.reconciliation_id,
job_id=self.job_id,
expected_claim_generation=self.expected_claim_generation,
expected_terminal_code=self.expected_terminal_code,
resource_release_attestation=self.resource_release_attestation,
failure_code=self.failure_code,
reason=self.reason,
)
if self.request_sha256 != request.request_sha256:
raise ObservatoryRecordedQueueIntegrityError(
"stored reconciliation request identity changed"
)
_validate_text(
self.quarantined_terminal_message,
"quarantined terminal message",
max_length=1_000,
)
_validate_timestamp(self.quarantined_at_utc, "quarantine timestamp")
_validate_optional_pattern(
self.quarantined_claim_token_sha256,
_SHA256,
"quarantined claim token sha256",
)
_validate_timestamp(self.reconciled_at_utc, "reconciliation timestamp")
expected_receipt_sha256 = _sha256(self.identity_document())
if self.receipt_sha256 != expected_receipt_sha256:
raise ObservatoryRecordedQueueIntegrityError(
"stored reconciliation receipt identity changed"
)
def identity_document(self) -> dict[str, object]:
return _reconciliation_receipt_identity_document(
reconciliation_id=self.reconciliation_id,
request_sha256=self.request_sha256,
job_id=self.job_id,
expected_claim_generation=self.expected_claim_generation,
expected_terminal_code=self.expected_terminal_code,
quarantined_terminal_message=self.quarantined_terminal_message,
quarantined_at_utc=self.quarantined_at_utc,
quarantined_claim_token_sha256=self.quarantined_claim_token_sha256,
resource_release_attestation=self.resource_release_attestation,
failure_code=self.failure_code,
reason=self.reason,
reconciled_at_utc=self.reconciled_at_utc,
)
def as_dict(self) -> dict[str, object]:
return {**self.identity_document(), "receipt_sha256": self.receipt_sha256}
@dataclass(frozen=True, slots=True)
class ObservatoryLiveLeaseIntent:
trigger_id: str
@@ -1212,6 +1400,173 @@ class ObservatoryRecordedJobQueue:
)
return tuple(self._get_job(connection, job_id) for job_id in recovered_ids)
def reconcile_failed(
self,
request: ObservatoryRecordedReconciliationRequest,
) -> ObservatoryRecordedReconciliationReceipt:
"""Close one exact quarantine only after explicit resource-release proof."""
with self._transaction() as connection:
existing = connection.execute(
"SELECT * FROM observatory_recorded_reconciliations "
"WHERE reconciliation_id = ?",
(request.reconciliation_id,),
).fetchone()
if existing is not None:
receipt = _reconciliation_receipt_from_row(existing)
if receipt.request_sha256 != request.request_sha256:
raise ObservatoryRecordedQueueConflictError(
"reconciliation id is bound to another request"
)
self._require_reconciled_job_matches_receipt(connection, receipt)
return receipt
prior = connection.execute(
"SELECT reconciliation_id FROM observatory_recorded_reconciliations "
"WHERE job_id = ?",
(request.job_id,),
).fetchone()
if prior is not None:
raise ObservatoryRecordedQueueConflictError(
"recorded job is bound to another reconciliation"
)
job = self._get_job(connection, request.job_id)
if job.state != "reconciliation-required":
raise ObservatoryRecordedQueueConflictError(
"recorded job is not awaiting reconciliation"
)
if job.claim_generation != request.expected_claim_generation:
raise ObservatoryRecordedQueueConflictError(
"recorded-job reconciliation generation changed"
)
if job.terminal_code != request.expected_terminal_code:
raise ObservatoryRecordedQueueConflictError(
"recorded-job reconciliation terminal code changed"
)
if job.terminal_message is None:
raise ObservatoryRecordedQueueIntegrityError(
"quarantined recorded job has no terminal reason"
)
reconciled_at_utc = self._timestamp()
if _parse_timestamp(
request.resource_release_attestation.attested_at_utc,
"resource-release attestation timestamp",
) > _parse_timestamp(reconciled_at_utc, "reconciliation timestamp"):
raise ObservatoryRecordedQueueConflictError(
"resource-release attestation is newer than reconciliation"
)
receipt_document = _reconciliation_receipt_identity_document(
reconciliation_id=request.reconciliation_id,
request_sha256=request.request_sha256,
job_id=request.job_id,
expected_claim_generation=request.expected_claim_generation,
expected_terminal_code=request.expected_terminal_code,
quarantined_terminal_message=job.terminal_message,
quarantined_at_utc=job.updated_at_utc,
quarantined_claim_token_sha256=job.terminal_claim_token_sha256,
resource_release_attestation=request.resource_release_attestation,
failure_code=request.failure_code,
reason=request.reason,
reconciled_at_utc=reconciled_at_utc,
)
receipt = ObservatoryRecordedReconciliationReceipt(
reconciliation_id=request.reconciliation_id,
request_sha256=request.request_sha256,
receipt_sha256=_sha256(receipt_document),
job_id=request.job_id,
expected_claim_generation=request.expected_claim_generation,
expected_terminal_code=request.expected_terminal_code,
quarantined_terminal_message=job.terminal_message,
quarantined_at_utc=job.updated_at_utc,
quarantined_claim_token_sha256=job.terminal_claim_token_sha256,
resource_release_attestation=request.resource_release_attestation,
failure_code=request.failure_code,
reason=request.reason,
reconciled_at_utc=reconciled_at_utc,
)
self._require_capacity(
connection,
table="observatory_recorded_reconciliations",
limit=self._max_jobs,
label="recorded reconciliation",
)
connection.execute(
"INSERT INTO observatory_recorded_reconciliations "
"(reconciliation_id, request_sha256, receipt_sha256, job_id, "
"expected_claim_generation, expected_terminal_code, "
"quarantined_terminal_message, quarantined_at_utc, "
"quarantined_claim_token_sha256, "
"resource_release_attestation_id, "
"resource_release_attestation_sha256, resource_release_operator_id, "
"resource_release_evidence_sha256, resource_release_attested_at_utc, "
"resources_released, failure_code, reason, reconciled_at_utc) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
receipt.reconciliation_id,
receipt.request_sha256,
receipt.receipt_sha256,
receipt.job_id,
receipt.expected_claim_generation,
receipt.expected_terminal_code,
receipt.quarantined_terminal_message,
receipt.quarantined_at_utc,
receipt.quarantined_claim_token_sha256,
receipt.resource_release_attestation.attestation_id,
receipt.resource_release_attestation.attestation_sha256,
receipt.resource_release_attestation.operator_id,
receipt.resource_release_attestation.evidence_sha256,
receipt.resource_release_attestation.attested_at_utc,
int(receipt.resource_release_attestation.resources_released),
receipt.failure_code,
receipt.reason,
receipt.reconciled_at_utc,
),
)
updated = connection.execute(
"UPDATE observatory_recorded_jobs SET state = 'failed', "
"preemption_requested = 0, result_id = NULL, result_sha256 = NULL, "
"terminal_code = ?, terminal_message = ?, "
"terminal_claim_token_sha256 = NULL, updated_at_utc = ? "
"WHERE job_id = ? AND state = 'reconciliation-required' "
"AND claim_generation = ? AND terminal_code = ?",
(
request.failure_code,
request.reason,
reconciled_at_utc,
request.job_id,
request.expected_claim_generation,
request.expected_terminal_code,
),
)
if updated.rowcount != 1:
raise ObservatoryRecordedQueueConflictError(
"recorded-job reconciliation guard changed"
)
self._require_reconciled_job_matches_receipt(connection, receipt)
return receipt
def get_reconciliation(
self,
job_id: str,
) -> ObservatoryRecordedReconciliationReceipt:
"""Return the immutable operator receipt for one reconciled job."""
_validate_pattern(job_id, _JOB_ID, "recorded job id")
with self._read_connection() as connection:
row = connection.execute(
"SELECT * FROM observatory_recorded_reconciliations WHERE job_id = ?",
(job_id,),
).fetchone()
if row is None:
raise ObservatoryRecordedQueueNotFoundError(
"recorded-job reconciliation does not exist"
)
receipt = _reconciliation_receipt_from_row(row)
self._require_reconciled_job_matches_receipt(connection, receipt)
return receipt
def start(self, job_id: str, *, claim_token: str) -> ObservatoryRecordedJob:
"""Enter running state, or yield before execution when live has priority."""
@@ -1968,6 +2323,26 @@ class ObservatoryRecordedJobQueue:
raise ObservatoryRecordedQueueNotFoundError(job_id)
return _job_from_row(row)
def _require_reconciled_job_matches_receipt(
self,
connection: sqlite3.Connection,
receipt: ObservatoryRecordedReconciliationReceipt,
) -> None:
job = self._get_job(connection, receipt.job_id)
if (
job.state != "failed"
or job.claim_generation != receipt.expected_claim_generation
or job.result_id is not None
or job.result_sha256 is not None
or job.terminal_code != receipt.failure_code
or job.terminal_message != receipt.reason
or job.terminal_claim_token_sha256 is not None
or job.updated_at_utc != receipt.reconciled_at_utc
):
raise ObservatoryRecordedQueueIntegrityError(
"reconciled recorded job differs from its audit receipt"
)
def _get_live_lease(
self, connection: sqlite3.Connection, lease_id: str
) -> ObservatoryLiveLease:
@@ -2030,6 +2405,7 @@ class ObservatoryRecordedJobQueue:
expected = {
"observatory_recorded_jobs": 46,
"observatory_recorded_claim_receipts": 6,
"observatory_recorded_reconciliations": 18,
"observatory_live_leases": 13,
"observatory_recorded_preemptions": 14,
}
@@ -2146,6 +2522,11 @@ class ObservatoryRecordedJobQueue:
self._max_jobs,
"recorded preemption",
),
(
"observatory_recorded_reconciliations",
self._max_jobs,
"recorded reconciliation",
),
):
count = len(
connection.execute(f"SELECT 1 FROM {table} LIMIT ?", (limit + 1,)).fetchall()
@@ -2245,6 +2626,47 @@ def _cancellation_request_from_row(
) from exc
def _reconciliation_receipt_from_row(
row: sqlite3.Row,
) -> ObservatoryRecordedReconciliationReceipt:
try:
attestation = ObservatoryRecordedResourceReleaseAttestation(
attestation_id=row["resource_release_attestation_id"],
operator_id=row["resource_release_operator_id"],
job_id=row["job_id"],
claim_generation=row["expected_claim_generation"],
resources_released=bool(row["resources_released"]),
evidence_sha256=row["resource_release_evidence_sha256"],
attested_at_utc=row["resource_release_attested_at_utc"],
)
if (
attestation.attestation_sha256
!= row["resource_release_attestation_sha256"]
):
raise ObservatoryRecordedQueueIntegrityError(
"stored resource-release attestation identity changed"
)
return ObservatoryRecordedReconciliationReceipt(
reconciliation_id=row["reconciliation_id"],
request_sha256=row["request_sha256"],
receipt_sha256=row["receipt_sha256"],
job_id=row["job_id"],
expected_claim_generation=row["expected_claim_generation"],
expected_terminal_code=row["expected_terminal_code"],
quarantined_terminal_message=row["quarantined_terminal_message"],
quarantined_at_utc=row["quarantined_at_utc"],
quarantined_claim_token_sha256=row["quarantined_claim_token_sha256"],
resource_release_attestation=attestation,
failure_code=row["failure_code"],
reason=row["reason"],
reconciled_at_utc=row["reconciled_at_utc"],
)
except (IndexError, KeyError, TypeError, ValueError) as exc:
raise ObservatoryRecordedQueueIntegrityError(
"stored recorded-job reconciliation is invalid"
) from exc
def _job_from_row(row: sqlite3.Row) -> ObservatoryRecordedJob:
try:
checkpoints = json.loads(row["allowed_checkpoints_json"])
@@ -2409,6 +2831,41 @@ def _live_lease_states() -> frozenset[str]:
return frozenset({"pending", "active", "completed", "failed", "cancelled"})
def _reconciliation_receipt_identity_document(
*,
reconciliation_id: str,
request_sha256: str,
job_id: str,
expected_claim_generation: int,
expected_terminal_code: str,
quarantined_terminal_message: str,
quarantined_at_utc: str,
quarantined_claim_token_sha256: str | None,
resource_release_attestation: ObservatoryRecordedResourceReleaseAttestation,
failure_code: str,
reason: str,
reconciled_at_utc: str,
) -> dict[str, object]:
return {
"schema_version": OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA,
"reconciliation_id": reconciliation_id,
"request_sha256": request_sha256,
"job_id": job_id,
"expected_claim_generation": expected_claim_generation,
"expected_terminal_code": expected_terminal_code,
"quarantined_terminal_message": quarantined_terminal_message,
"quarantined_at_utc": quarantined_at_utc,
"quarantined_claim_token_sha256": quarantined_claim_token_sha256,
"resource_release_attestation": resource_release_attestation.as_dict(),
"outcome": {
"state": "failed",
"code": failure_code,
"reason": reason,
},
"reconciled_at_utc": reconciled_at_utc,
}
def _canonical_json_text(value: object) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))