Coverage for extra_codeowners/database.py: 89%
514 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-13 09:46 +0000
1"""Durable evaluation queue and delivery de-duplication storage."""
3from __future__ import annotations
5import hashlib
6import threading
7from collections.abc import Iterator
8from contextlib import contextmanager
9from dataclasses import dataclass
10from datetime import UTC, datetime, timedelta
11from typing import Any
13from sqlalchemy import (
14 JSON,
15 Boolean,
16 DateTime,
17 Index,
18 Integer,
19 String,
20 UniqueConstraint,
21 case,
22 create_engine,
23 delete,
24 exists,
25 func,
26 inspect,
27 or_,
28 select,
29 text,
30 update,
31)
32from sqlalchemy.engine import Connection, Engine
33from sqlalchemy.exc import IntegrityError
34from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker
35from sqlalchemy.pool import NullPool
37SCHEMA_VERSION = 1
38DATABASE_CONNECT_TIMEOUT_SECONDS = 3
39DATABASE_POOL_TIMEOUT_SECONDS = 2
40DATABASE_STATEMENT_TIMEOUT_MILLISECONDS = 3_000
41MAX_BASE_SCOPED_AUTHORITY_JOBS_PER_REPOSITORY = 100
44def normalize_repository_full_name(value: str) -> str:
45 """Return the case-insensitive canonical key for ``owner/repository``."""
46 if ( 46 ↛ 53line 46 didn't jump to line 53 because the condition on line 46 was never true
47 not value
48 or value != value.strip()
49 or len(value) > 512
50 or value.count("/") != 1
51 or any(part in {"", ".", ".."} for part in value.split("/"))
52 ):
53 raise ValueError("repository_full_name must be a valid owner/repository name")
54 return value.lower()
57def utcnow() -> datetime:
58 """Return an aware UTC timestamp."""
59 return datetime.now(UTC)
62class Base(DeclarativeBase):
63 """SQLAlchemy declarative base."""
66class SchemaMetadata(Base):
67 """Explicit compatibility marker for the durable database schema."""
69 __tablename__ = "schema_metadata"
71 singleton_id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1)
72 version: Mapped[int] = mapped_column(Integer, nullable=False)
75class EvaluationJob(Base):
76 """Latest requested evaluation for one pull request."""
78 __tablename__ = "evaluation_jobs"
79 __table_args__ = (
80 UniqueConstraint(
81 "installation_id",
82 "repository_full_name",
83 "pull_number",
84 name="uq_evaluation_job_pr",
85 ),
86 Index("ix_evaluation_jobs_claim", "state", "available_at", "lease_until"),
87 )
89 id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
90 installation_id: Mapped[int] = mapped_column(Integer, nullable=False)
91 repository_full_name: Mapped[str] = mapped_column(String(512), nullable=False)
92 pull_number: Mapped[int] = mapped_column(Integer, nullable=False)
93 head_sha_hint: Mapped[str | None] = mapped_column(String(64))
94 last_delivery_id: Mapped[str | None] = mapped_column(String(128))
95 reason: Mapped[str] = mapped_column(String(255), nullable=False)
96 generation: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
97 authority_generation: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
98 state: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")
99 attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
100 requested_at: Mapped[datetime] = mapped_column(
101 DateTime(timezone=True), nullable=False, default=utcnow
102 )
103 available_at: Mapped[datetime] = mapped_column(
104 DateTime(timezone=True), nullable=False, default=utcnow
105 )
106 lease_owner: Mapped[str | None] = mapped_column(String(128))
107 lease_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
108 last_error: Mapped[str | None] = mapped_column(String(2000))
111class WebhookDelivery(Base):
112 """A GitHub delivery ID retained for replay-safe ingestion."""
114 __tablename__ = "webhook_deliveries"
116 delivery_id: Mapped[str] = mapped_column(String(128), primary_key=True)
117 event: Mapped[str] = mapped_column(String(128), nullable=False)
118 received_at: Mapped[datetime] = mapped_column(
119 DateTime(timezone=True), nullable=False, default=utcnow, index=True
120 )
121 invalidation_required: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
122 invalidation_completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
125class EvaluationAudit(Base):
126 """Most recent evaluation evidence for operations and debugging."""
128 __tablename__ = "evaluation_audits"
129 __table_args__ = (
130 UniqueConstraint("repository_full_name", "pull_number", name="uq_evaluation_audit_pr"),
131 )
133 id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
134 repository_full_name: Mapped[str] = mapped_column(String(512), nullable=False)
135 pull_number: Mapped[int] = mapped_column(Integer, nullable=False)
136 head_sha: Mapped[str] = mapped_column(String(64), nullable=False)
137 conclusion: Mapped[str] = mapped_column(String(32), nullable=False)
138 details: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
139 evaluated_at: Mapped[datetime] = mapped_column(
140 DateTime(timezone=True), nullable=False, default=utcnow
141 )
144class ServiceLease(Base):
145 """A database-coordinated singleton lease such as the reconciler."""
147 __tablename__ = "service_leases"
149 name: Mapped[str] = mapped_column(String(128), primary_key=True)
150 owner: Mapped[str] = mapped_column(String(128), nullable=False)
151 lease_until: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
154class AuthorityJob(Base):
155 """Coalesced repository or installation authority-change fan-out."""
157 __tablename__ = "authority_jobs"
158 __table_args__ = (
159 UniqueConstraint(
160 "installation_id",
161 "scope_key",
162 "base_ref",
163 name="uq_authority_job_scope",
164 ),
165 Index("ix_authority_jobs_claim", "state", "available_at", "lease_until"),
166 )
168 id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
169 installation_id: Mapped[int] = mapped_column(Integer, nullable=False)
170 scope_key: Mapped[str] = mapped_column(String(512), nullable=False)
171 base_ref: Mapped[str] = mapped_column(String(255), nullable=False, default="")
172 reason: Mapped[str] = mapped_column(String(255), nullable=False)
173 generation: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
174 state: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")
175 attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
176 requested_at: Mapped[datetime] = mapped_column(
177 DateTime(timezone=True), nullable=False, default=utcnow
178 )
179 available_at: Mapped[datetime] = mapped_column(
180 DateTime(timezone=True), nullable=False, default=utcnow
181 )
182 lease_owner: Mapped[str | None] = mapped_column(String(128))
183 lease_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
184 last_error: Mapped[str | None] = mapped_column(String(2000))
187class AuthorityEpoch(Base):
188 """Monotonic installation fence for broad authority/identity changes."""
190 __tablename__ = "authority_epochs"
192 installation_id: Mapped[int] = mapped_column(Integer, primary_key=True)
193 generation: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
194 changed_at: Mapped[datetime] = mapped_column(
195 DateTime(timezone=True), nullable=False, default=utcnow
196 )
199@dataclass(frozen=True, slots=True)
200class JobRequest:
201 """Input used to enqueue or coalesce an evaluation."""
203 installation_id: int
204 repository_full_name: str
205 pull_number: int
206 reason: str
207 head_sha_hint: str | None = None
209 def __post_init__(self) -> None:
210 object.__setattr__(
211 self,
212 "repository_full_name",
213 normalize_repository_full_name(self.repository_full_name),
214 )
217@dataclass(frozen=True, slots=True)
218class AuthorityRequest:
219 """An authority change that must re-evaluate a repository or installation."""
221 installation_id: int
222 repository_full_name: str | None
223 base_ref: str | None
224 reason: str
226 def __post_init__(self) -> None:
227 if self.repository_full_name is not None:
228 object.__setattr__(
229 self,
230 "repository_full_name",
231 normalize_repository_full_name(self.repository_full_name),
232 )
234 @property
235 def scope_key(self) -> str:
236 """Return a non-null key suitable for database uniqueness."""
237 return self.repository_full_name or "*"
240@dataclass(frozen=True, slots=True)
241class ClaimedJob:
242 """Leased immutable view of an evaluation job."""
244 id: int
245 installation_id: int
246 repository_full_name: str
247 pull_number: int
248 reason: str
249 head_sha_hint: str | None
250 last_delivery_id: str | None
251 generation: int
252 authority_generation: int
253 attempts: int
254 lease_owner: str
257@dataclass(frozen=True, slots=True)
258class ClaimedAuthorityJob:
259 """Leased immutable authority fan-out job."""
261 id: int
262 installation_id: int
263 repository_full_name: str | None
264 base_ref: str | None
265 reason: str
266 generation: int
267 attempts: int
268 lease_owner: str
271@dataclass(frozen=True, slots=True)
272class CheckWriteGuard:
273 """Held per-pull-request GitHub check writer guard."""
275 key: int
276 connection: Connection | None = None
277 local_lock: threading.Lock | None = None
278 shared: bool = False
281class QueueStore:
282 """Database-backed latest-wins work queue.
284 A job is unique per installation/repository/pull request. New triggers bump
285 its generation, so a worker cannot delete work that arrived during an
286 earlier evaluation.
287 """
289 def __init__(self, database_url: str) -> None:
290 is_postgresql = database_url.startswith("postgresql")
291 connect_args: dict[str, Any]
292 engine_options: dict[str, Any] = {"pool_pre_ping": True}
293 if database_url.startswith("sqlite"):
294 connect_args = {"check_same_thread": False}
295 elif is_postgresql: 295 ↛ 305line 295 didn't jump to line 305 because the condition on line 295 was always true
296 # Webhook ingestion is time-bounded. A dead database endpoint or an
297 # exhausted application pool must fail within that budget so GitHub
298 # can redeliver instead of holding a request indefinitely.
299 connect_args = {
300 "connect_timeout": DATABASE_CONNECT_TIMEOUT_SECONDS,
301 "options": f"-c statement_timeout={DATABASE_STATEMENT_TIMEOUT_MILLISECONDS}",
302 }
303 engine_options["pool_timeout"] = DATABASE_POOL_TIMEOUT_SECONDS
304 else:
305 connect_args = {}
306 self.engine = create_engine(
307 database_url,
308 connect_args=connect_args,
309 **engine_options,
310 )
311 self._sessions = sessionmaker(self.engine, expire_on_commit=False)
312 self._lock_engine: Engine | None = None
313 if self.engine.dialect.name == "postgresql":
314 # Advisory locks must never return to a reusable connection pool:
315 # an uncertain unlock would otherwise poison that pooled session.
316 self._lock_engine = create_engine(
317 database_url,
318 connect_args={
319 "connect_timeout": DATABASE_CONNECT_TIMEOUT_SECONDS,
320 "options": f"-c statement_timeout={DATABASE_STATEMENT_TIMEOUT_MILLISECONDS}",
321 },
322 poolclass=NullPool,
323 pool_pre_ping=True,
324 )
325 # SQLite is development/test-only. A bounded stripe set provides
326 # in-process ordering there; production PostgreSQL uses exact advisory
327 # locks shared by every replica.
328 self._check_write_locks = tuple(threading.Lock() for _ in range(256))
329 # Authority publication guards can be nested around check-writer
330 # guards. A distinct local namespace prevents deterministic stripe
331 # self-deadlocks in SQLite development and tests.
332 self._authority_locks = tuple(threading.Lock() for _ in range(256))
334 def initialize(self) -> None:
335 """Create an empty schema or reject an incompatible existing database."""
336 guard = self.acquire_check_write_guard(
337 "__extra_codeowners_schema__", 0, timeout_seconds=60.0
338 )
339 if guard is None: 339 ↛ 340line 339 didn't jump to line 340 because the condition on line 339 was never true
340 raise RuntimeError("timed out waiting for the database schema initializer")
341 try:
342 self._initialize_guarded()
343 finally:
344 self.release_check_write_guard(guard)
346 def _initialize_guarded(self) -> None:
347 """Initialize while holding the cross-replica schema guard."""
348 existing_tables = set(inspect(self.engine).get_table_names())
349 metadata_was_present = SchemaMetadata.__tablename__ in existing_tables
350 application_tables = {
351 table.name
352 for table in Base.metadata.sorted_tables
353 if table.name != SchemaMetadata.__tablename__
354 }
355 if not metadata_was_present and existing_tables & application_tables:
356 raise RuntimeError(
357 "database predates versioned schemas; migrate or recreate it before startup"
358 )
359 if metadata_was_present:
360 with self._sessions() as session:
361 version = session.scalar(
362 select(SchemaMetadata.version).where(SchemaMetadata.singleton_id == 1)
363 )
364 if version != SCHEMA_VERSION:
365 raise RuntimeError(
366 f"database schema version {version!r} is incompatible with required version "
367 f"{SCHEMA_VERSION}"
368 )
369 self.validate_schema()
370 self._reactivate_legacy_dead_jobs()
371 return
373 Base.metadata.create_all(self.engine)
374 try:
375 with self.session() as session:
376 if session.get(SchemaMetadata, 1) is None: 376 ↛ 381line 376 didn't jump to line 381
377 session.add(SchemaMetadata(singleton_id=1, version=SCHEMA_VERSION))
378 except IntegrityError:
379 # Another first-start replica may have inserted the singleton.
380 pass
381 self.validate_schema()
383 def _reactivate_legacy_dead_jobs(self) -> None:
384 """Resume rows created by pre-release builds that used terminal retries."""
385 now = utcnow()
386 with self.session() as session:
387 for model in (AuthorityJob, EvaluationJob):
388 session.execute(
389 update(model)
390 .where(model.state == "dead")
391 .values(
392 state="pending",
393 attempts=0,
394 available_at=now,
395 lease_owner=None,
396 lease_until=None,
397 last_error=None,
398 )
399 )
401 def validate_schema(self) -> None:
402 """Validate the version marker and every required table column."""
403 inspector = inspect(self.engine)
404 tables = set(inspector.get_table_names())
405 for table in Base.metadata.sorted_tables:
406 if table.name not in tables: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 raise RuntimeError(f"database schema is missing table {table.name!r}")
408 actual_columns = {column["name"] for column in inspector.get_columns(table.name)}
409 expected_columns = {column.name for column in table.columns}
410 missing = expected_columns - actual_columns
411 if missing: 411 ↛ 412line 411 didn't jump to line 412 because the condition on line 411 was never true
412 raise RuntimeError(
413 f"database table {table.name!r} is missing columns {sorted(missing)!r}"
414 )
415 with self._sessions() as session:
416 version = session.scalar(
417 select(SchemaMetadata.version).where(SchemaMetadata.singleton_id == 1)
418 )
419 if version != SCHEMA_VERSION: 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 raise RuntimeError(
421 f"database schema version {version!r} is incompatible with required version "
422 f"{SCHEMA_VERSION}"
423 )
425 def close(self) -> None:
426 """Dispose pooled connections."""
427 self.engine.dispose()
428 if self._lock_engine is not None:
429 self._lock_engine.dispose()
431 @staticmethod
432 def _check_write_key(repository: str, scope: str | int) -> int:
433 digest = hashlib.blake2b(
434 f"extra-codeowners\0{repository}\0{scope}".encode(),
435 digest_size=8,
436 ).digest()
437 return int.from_bytes(digest, byteorder="big", signed=True)
439 def acquire_check_write_guard(
440 self,
441 repository: str,
442 scope: str | int,
443 timeout_seconds: float = 30.0,
444 ) -> CheckWriteGuard | None:
445 """Become the sole check writer, returning ``None`` after a bounded wait.
447 PostgreSQL session advisory locks are released by the server if a
448 process or connection dies, closing the publish/revoke crash window.
449 """
450 key = self._check_write_key(repository, scope)
451 return self._acquire_guard(
452 key,
453 timeout_seconds=timeout_seconds,
454 shared=False,
455 local_locks=self._check_write_locks,
456 )
458 def acquire_authority_guard(
459 self,
460 installation_id: int,
461 *,
462 shared: bool,
463 timeout_seconds: float = 30.0,
464 ) -> CheckWriteGuard | None:
465 """Acquire the installation authority publication/ingress guard.
467 PostgreSQL publishers take a shared lock, so normal evaluations remain
468 concurrent. Authority webhook ingestion takes the exclusive variant,
469 ordering its durable fence before or after every final Check Run write.
470 SQLite uses the same exclusive local lock for both modes.
471 """
472 key = self._check_write_key(
473 "__extra_codeowners_authority__", f"installation:{installation_id}"
474 )
475 return self._acquire_guard(
476 key,
477 timeout_seconds=timeout_seconds,
478 shared=shared,
479 local_locks=self._authority_locks,
480 )
482 def _acquire_guard(
483 self,
484 key: int,
485 *,
486 timeout_seconds: float,
487 shared: bool,
488 local_locks: tuple[threading.Lock, ...],
489 ) -> CheckWriteGuard | None:
490 """Acquire one exact advisory/local lock key."""
491 if self._lock_engine is not None:
492 connection = self._lock_engine.connect()
493 try:
494 connection.execute(
495 text("SELECT set_config('statement_timeout', :timeout, false)"),
496 {"timeout": f"{max(1, int(timeout_seconds * 1000))}ms"},
497 )
498 lock_function = "pg_advisory_lock_shared" if shared else "pg_advisory_lock"
499 connection.execute(text(f"SELECT {lock_function}(:key)"), {"key": key})
500 # PostgreSQL advisory locks are session-scoped, so committing
501 # here retains the lock while preventing a network/API wait
502 # from leaving an idle transaction open on the database.
503 connection.commit()
504 return CheckWriteGuard(key=key, connection=connection, shared=shared)
505 except Exception:
506 connection.invalidate()
507 connection.close()
508 raise
510 local_lock = local_locks[key % len(local_locks)]
511 if not local_lock.acquire(timeout=max(0.0, timeout_seconds)):
512 return None
513 return CheckWriteGuard(key=key, local_lock=local_lock)
515 @staticmethod
516 def release_check_write_guard(guard: CheckWriteGuard) -> None:
517 """Release a guard acquired by :meth:`acquire_check_write_guard`."""
518 if guard.connection is not None:
519 try:
520 unlock_function = (
521 "pg_advisory_unlock_shared" if guard.shared else "pg_advisory_unlock"
522 )
523 unlocked = guard.connection.execute(
524 text(f"SELECT {unlock_function}(:key)"), {"key": guard.key}
525 ).scalar_one()
526 if unlocked is not True: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 raise RuntimeError("PostgreSQL advisory guard was not held by this session")
528 guard.connection.commit()
529 except Exception:
530 # Physically discard an uncertain session; PostgreSQL releases
531 # its advisory locks when that connection is terminated.
532 guard.connection.invalidate()
533 raise
534 finally:
535 guard.connection.close()
536 return
537 if guard.local_lock is None: # pragma: no cover - dataclass is internal
538 raise RuntimeError("check write guard has no lock")
539 guard.local_lock.release()
541 @contextmanager
542 def session(self) -> Iterator[Session]:
543 """Open a transactional session."""
544 with self._sessions.begin() as session:
545 yield session
547 @staticmethod
548 def _enqueue_in_session(
549 session: Session,
550 request: JobRequest,
551 delivery_id: str | None = None,
552 ) -> None:
553 now = utcnow()
554 authority_generation = (
555 session.scalar(
556 select(AuthorityEpoch.generation).where(
557 AuthorityEpoch.installation_id == request.installation_id
558 )
559 )
560 or 0
561 )
562 values: dict[str, Any] = {
563 "generation": EvaluationJob.generation + 1,
564 "authority_generation": authority_generation,
565 "reason": request.reason,
566 "head_sha_hint": request.head_sha_hint,
567 "requested_at": now,
568 "available_at": now,
569 "state": "pending",
570 "attempts": 0,
571 "last_error": None,
572 }
573 if delivery_id is not None:
574 values["last_delivery_id"] = delivery_id
575 updated = session.execute(
576 update(EvaluationJob)
577 .where(
578 EvaluationJob.installation_id == request.installation_id,
579 EvaluationJob.repository_full_name == request.repository_full_name,
580 EvaluationJob.pull_number == request.pull_number,
581 )
582 .values(**values)
583 )
584 if getattr(updated, "rowcount", 0) == 1:
585 return
586 session.add(
587 EvaluationJob(
588 installation_id=request.installation_id,
589 repository_full_name=request.repository_full_name,
590 pull_number=request.pull_number,
591 reason=request.reason,
592 head_sha_hint=request.head_sha_hint,
593 last_delivery_id=delivery_id,
594 authority_generation=authority_generation,
595 requested_at=now,
596 available_at=now,
597 )
598 )
600 def enqueue(self, request: JobRequest) -> None:
601 """Enqueue an evaluation, coalescing repeated triggers."""
602 for _ in range(5): 602 ↛ 611line 602 didn't jump to line 611 because the loop on line 602 didn't complete
603 try:
604 with self.session() as session:
605 self._enqueue_in_session(session, request)
606 return
607 except IntegrityError:
608 # A concurrent first trigger inserted the unique PR job. Retry
609 # as an update after the losing transaction has rolled back.
610 continue
611 raise RuntimeError("could not enqueue evaluation after concurrent inserts")
613 def enqueue_if_absent(self, request: JobRequest) -> bool:
614 """Enqueue reconciliation work without superseding active or dead work."""
615 for _ in range(5): 615 ↛ 633line 615 didn't jump to line 633 because the loop on line 615 didn't complete
616 try:
617 with self.session() as session:
618 existing = session.scalar(
619 select(EvaluationJob.id).where(
620 EvaluationJob.installation_id == request.installation_id,
621 EvaluationJob.repository_full_name == request.repository_full_name,
622 EvaluationJob.pull_number == request.pull_number,
623 )
624 )
625 if existing is not None:
626 return False
627 self._enqueue_in_session(session, request)
628 return True
629 except IntegrityError:
630 # Another receiver or reconciler inserted this PR between the
631 # read and commit. Its work is sufficient; retry to observe it.
632 continue
633 raise RuntimeError("could not reconcile evaluation after concurrent inserts")
635 @staticmethod
636 def _enqueue_authority_in_session(
637 session: Session,
638 request: AuthorityRequest,
639 ) -> None:
640 now = utcnow()
641 scope_key = request.scope_key
642 base_ref = request.base_ref or ""
643 if request.repository_full_name is not None:
644 if not base_ref:
645 # A repository-wide authority event covers every pending base
646 # push and should collapse a contributor-created branch queue.
647 session.execute(
648 delete(AuthorityJob).where(
649 AuthorityJob.installation_id == request.installation_id,
650 AuthorityJob.scope_key == scope_key,
651 AuthorityJob.base_ref != "",
652 )
653 )
654 else:
655 same_base_exists = session.scalar(
656 select(AuthorityJob.id).where(
657 AuthorityJob.installation_id == request.installation_id,
658 AuthorityJob.scope_key == scope_key,
659 AuthorityJob.base_ref == base_ref,
660 )
661 )
662 repository_wide_exists = session.scalar(
663 select(AuthorityJob.id).where(
664 AuthorityJob.installation_id == request.installation_id,
665 AuthorityJob.scope_key == scope_key,
666 AuthorityJob.base_ref == "",
667 )
668 )
669 base_scoped_count = int(
670 session.scalar(
671 select(func.count())
672 .select_from(AuthorityJob)
673 .where(
674 AuthorityJob.installation_id == request.installation_id,
675 AuthorityJob.scope_key == scope_key,
676 AuthorityJob.base_ref != "",
677 )
678 )
679 or 0
680 )
681 if repository_wide_exists is not None or (
682 same_base_exists is None
683 and base_scoped_count >= MAX_BASE_SCOPED_AUTHORITY_JOBS_PER_REPOSITORY
684 ):
685 # Once the bounded set overflows, one repository-wide job
686 # is safer and cheaper than an attacker-controlled number
687 # of unique branch rows. It conservatively reevaluates all
688 # open pull requests in the repository.
689 session.execute(
690 delete(AuthorityJob).where(
691 AuthorityJob.installation_id == request.installation_id,
692 AuthorityJob.scope_key == scope_key,
693 AuthorityJob.base_ref != "",
694 )
695 )
696 base_ref = ""
697 updated = session.execute(
698 update(AuthorityJob)
699 .where(
700 AuthorityJob.installation_id == request.installation_id,
701 AuthorityJob.scope_key == scope_key,
702 AuthorityJob.base_ref == base_ref,
703 )
704 .values(
705 generation=AuthorityJob.generation + 1,
706 reason=request.reason,
707 requested_at=now,
708 available_at=now,
709 state="pending",
710 attempts=0,
711 last_error=None,
712 )
713 )
714 if getattr(updated, "rowcount", 0) == 1:
715 return
716 session.add(
717 AuthorityJob(
718 installation_id=request.installation_id,
719 scope_key=scope_key,
720 base_ref=base_ref,
721 reason=request.reason,
722 requested_at=now,
723 available_at=now,
724 )
725 )
727 @staticmethod
728 def _bump_authority_epoch_in_session(session: Session, installation_id: int) -> None:
729 now = utcnow()
730 updated = session.execute(
731 update(AuthorityEpoch)
732 .where(AuthorityEpoch.installation_id == installation_id)
733 .values(
734 generation=AuthorityEpoch.generation + 1,
735 changed_at=now,
736 )
737 )
738 if getattr(updated, "rowcount", 0) == 0: 738 ↛ exitline 738 didn't return from function '_bump_authority_epoch_in_session' because the condition on line 738 was always true
739 session.add(
740 AuthorityEpoch(
741 installation_id=installation_id,
742 generation=1,
743 changed_at=now,
744 )
745 )
747 def accept_delivery(
748 self,
749 delivery_id: str,
750 event: str,
751 request: JobRequest | AuthorityRequest | None,
752 authority_guard_timeout_seconds: float = 5.0,
753 ) -> bool:
754 """Record and enqueue one webhook atomically.
756 Returns ``False`` for a previously accepted GitHub delivery.
757 """
758 authority_guard: CheckWriteGuard | None = None
759 if isinstance(request, AuthorityRequest):
760 authority_guard = self.acquire_authority_guard(
761 request.installation_id,
762 shared=False,
763 timeout_seconds=authority_guard_timeout_seconds,
764 )
765 if authority_guard is None:
766 raise RuntimeError("timed out waiting to record an authority change")
767 try:
768 for _ in range(5): 768 ↛ 794line 768 didn't jump to line 794 because the loop on line 768 didn't complete
769 try:
770 with self.session() as session:
771 if session.get(WebhookDelivery, delivery_id) is not None:
772 return False
773 session.add(
774 WebhookDelivery(
775 delivery_id=delivery_id,
776 event=event,
777 invalidation_required=isinstance(request, JobRequest),
778 )
779 )
780 if isinstance(request, JobRequest):
781 self._enqueue_in_session(session, request, delivery_id)
782 elif isinstance(request, AuthorityRequest):
783 if request.repository_full_name is None:
784 self._bump_authority_epoch_in_session(
785 session, request.installation_id
786 )
787 self._enqueue_authority_in_session(session, request)
788 return True
789 except IntegrityError:
790 # This may be either the same delivery racing or a different
791 # delivery racing to create the unique PR job. Retry so the
792 # latter delivery and its trigger are not silently discarded.
793 continue
794 raise RuntimeError("could not accept webhook delivery after concurrent inserts")
795 finally:
796 if authority_guard is not None:
797 self.release_check_write_guard(authority_guard)
799 def enqueue_authority(self, request: AuthorityRequest) -> None:
800 """Coalesce internal installation-to-repository authority fan-out."""
801 for _ in range(5): 801 ↛ 808line 801 didn't jump to line 808 because the loop on line 801 didn't complete
802 try:
803 with self.session() as session:
804 self._enqueue_authority_in_session(session, request)
805 return
806 except IntegrityError:
807 continue
808 raise RuntimeError("could not enqueue authority work after concurrent inserts")
810 def delivery_needs_invalidation(self, delivery_id: str) -> bool:
811 """Return whether an accepted PR trigger still needs check revocation."""
812 with self._sessions() as session:
813 delivery = session.get(WebhookDelivery, delivery_id)
814 return bool(
815 delivery is not None
816 and delivery.invalidation_required
817 and delivery.invalidation_completed_at is None
818 )
820 def mark_delivery_invalidated(self, delivery_id: str) -> bool:
821 """Record successful check revocation for replay-safe webhook retries."""
822 with self.session() as session:
823 result = session.execute(
824 update(WebhookDelivery)
825 .where(
826 WebhookDelivery.delivery_id == delivery_id,
827 WebhookDelivery.invalidation_required.is_(True),
828 WebhookDelivery.invalidation_completed_at.is_(None),
829 )
830 .values(invalidation_completed_at=utcnow())
831 )
832 return getattr(result, "rowcount", 0) == 1
834 def is_current_generation(self, job: ClaimedJob) -> bool:
835 """Return whether no newer trigger has superseded a claimed evaluation."""
836 with self._sessions() as session:
837 generation = session.scalar(
838 select(EvaluationJob.generation).where(EvaluationJob.id == job.id)
839 )
840 return generation == job.generation
842 def has_superseding_job(self, job: ClaimedJob) -> bool:
843 """Return whether a newer generation still needs evaluation."""
844 with self._sessions() as session:
845 return bool(
846 session.scalar(
847 select(EvaluationJob.id).where(
848 EvaluationJob.id == job.id,
849 EvaluationJob.generation > job.generation,
850 EvaluationJob.state == "pending",
851 )
852 )
853 )
855 def has_blocking_authority(self, job: ClaimedJob, base_ref: str) -> bool:
856 """Return whether unresolved authority work can affect this pull request.
858 Authority rows remain present while pending, leased, retrying, or dead.
859 Every such state blocks publication because the evaluation may have
860 collected membership, policy, label, CODEOWNERS, or merge-base evidence
861 from before the accepted authority change.
862 """
863 with self._sessions() as session:
864 return bool(
865 session.scalar(
866 select(AuthorityJob.id)
867 .where(
868 AuthorityJob.installation_id == job.installation_id,
869 AuthorityJob.scope_key.in_(("*", job.repository_full_name)),
870 AuthorityJob.base_ref.in_(("", base_ref)),
871 )
872 .limit(1)
873 )
874 )
876 def is_current_claim(self, job: ClaimedJob) -> bool:
877 """Return whether a claim still owns the current, unexpired generation."""
878 now = utcnow()
879 with self._sessions() as session:
880 row = session.execute(
881 select(
882 EvaluationJob.generation,
883 EvaluationJob.authority_generation,
884 EvaluationJob.lease_owner,
885 EvaluationJob.lease_until,
886 ).where(EvaluationJob.id == job.id)
887 ).one_or_none()
888 if row is None or row.lease_until is None: 888 ↛ 889line 888 didn't jump to line 889 because the condition on line 888 was never true
889 return False
890 current_authority_generation = (
891 session.scalar(
892 select(AuthorityEpoch.generation).where(
893 AuthorityEpoch.installation_id == job.installation_id
894 )
895 )
896 or 0
897 )
898 expiry = row.lease_until
899 if expiry.tzinfo is None: 899 ↛ 901line 899 didn't jump to line 901 because the condition on line 899 was always true
900 expiry = expiry.replace(tzinfo=UTC)
901 return bool(
902 row.generation == job.generation
903 and row.authority_generation == job.authority_generation
904 and current_authority_generation == job.authority_generation
905 and row.lease_owner == job.lease_owner
906 and expiry >= now
907 )
909 def renew_claim(self, job: ClaimedJob, lease_seconds: int) -> bool:
910 """Extend a current claim without reviving a superseded generation."""
911 with self.session() as session:
912 renewed = session.execute(
913 update(EvaluationJob)
914 .where(
915 EvaluationJob.id == job.id,
916 EvaluationJob.generation == job.generation,
917 EvaluationJob.lease_owner == job.lease_owner,
918 EvaluationJob.state == "pending",
919 )
920 .values(lease_until=utcnow() + timedelta(seconds=lease_seconds))
921 )
922 return getattr(renewed, "rowcount", 0) == 1
924 def claim(self, owner: str, lease_seconds: int) -> ClaimedJob | None:
925 """Atomically lease the oldest available job."""
926 now = utcnow()
927 lease_until = now + timedelta(seconds=lease_seconds)
928 for _ in range(3): 928 ↛ 992line 928 didn't jump to line 992 because the loop on line 928 didn't complete
929 with self.session() as session:
930 candidate = session.scalar(
931 select(EvaluationJob.id)
932 .where(
933 EvaluationJob.state == "pending",
934 EvaluationJob.available_at <= now,
935 or_(EvaluationJob.lease_until.is_(None), EvaluationJob.lease_until < now),
936 ~exists(
937 select(AuthorityJob.id).where(
938 AuthorityJob.installation_id == EvaluationJob.installation_id,
939 or_(
940 AuthorityJob.scope_key == "*",
941 AuthorityJob.scope_key == EvaluationJob.repository_full_name,
942 ),
943 )
944 ),
945 )
946 .order_by(EvaluationJob.available_at, EvaluationJob.id)
947 .limit(1)
948 .with_for_update(skip_locked=True)
949 )
950 if candidate is None:
951 return None
952 claimed = session.execute(
953 update(EvaluationJob)
954 .where(
955 EvaluationJob.id == candidate,
956 EvaluationJob.state == "pending",
957 or_(EvaluationJob.lease_until.is_(None), EvaluationJob.lease_until < now),
958 )
959 .values(
960 lease_owner=owner,
961 lease_until=lease_until,
962 attempts=EvaluationJob.attempts + 1,
963 # Reclaiming an expired lease fences the previous
964 # worker even when no webhook generation changed.
965 generation=case(
966 (
967 EvaluationJob.lease_owner.is_not(None),
968 EvaluationJob.generation + 1,
969 ),
970 else_=EvaluationJob.generation,
971 ),
972 )
973 )
974 if getattr(claimed, "rowcount", 0) != 1: 974 ↛ 975line 974 didn't jump to line 975 because the condition on line 974 was never true
975 continue
976 row = session.get(EvaluationJob, candidate)
977 if row is None: # pragma: no cover - protected by the transaction
978 continue
979 return ClaimedJob(
980 id=row.id,
981 installation_id=row.installation_id,
982 repository_full_name=row.repository_full_name,
983 pull_number=row.pull_number,
984 reason=row.reason,
985 head_sha_hint=row.head_sha_hint,
986 last_delivery_id=row.last_delivery_id,
987 generation=row.generation,
988 authority_generation=row.authority_generation,
989 attempts=row.attempts,
990 lease_owner=owner,
991 )
992 return None
994 def claim_authority(self, owner: str, lease_seconds: int) -> ClaimedAuthorityJob | None:
995 """Atomically lease the oldest authority fan-out job."""
996 now = utcnow()
997 lease_until = now + timedelta(seconds=lease_seconds)
998 for _ in range(3): 998 ↛ 1056line 998 didn't jump to line 1056 because the loop on line 998 didn't complete
999 with self.session() as session:
1000 candidate = session.scalar(
1001 select(AuthorityJob.id)
1002 .where(
1003 AuthorityJob.state == "pending",
1004 AuthorityJob.available_at <= now,
1005 or_(AuthorityJob.lease_until.is_(None), AuthorityJob.lease_until < now),
1006 )
1007 .order_by(
1008 case(
1009 (AuthorityJob.scope_key == "*", 0),
1010 (AuthorityJob.base_ref == "", 1),
1011 else_=2,
1012 ),
1013 AuthorityJob.available_at,
1014 AuthorityJob.id,
1015 )
1016 .limit(1)
1017 .with_for_update(skip_locked=True)
1018 )
1019 if candidate is None:
1020 return None
1021 claimed = session.execute(
1022 update(AuthorityJob)
1023 .where(
1024 AuthorityJob.id == candidate,
1025 AuthorityJob.state == "pending",
1026 or_(AuthorityJob.lease_until.is_(None), AuthorityJob.lease_until < now),
1027 )
1028 .values(
1029 lease_owner=owner,
1030 lease_until=lease_until,
1031 attempts=AuthorityJob.attempts + 1,
1032 generation=case(
1033 (
1034 AuthorityJob.lease_owner.is_not(None),
1035 AuthorityJob.generation + 1,
1036 ),
1037 else_=AuthorityJob.generation,
1038 ),
1039 )
1040 )
1041 if getattr(claimed, "rowcount", 0) != 1: 1041 ↛ 1042line 1041 didn't jump to line 1042 because the condition on line 1041 was never true
1042 continue
1043 row = session.get(AuthorityJob, candidate)
1044 if row is None: # pragma: no cover - protected by transaction
1045 continue
1046 return ClaimedAuthorityJob(
1047 id=row.id,
1048 installation_id=row.installation_id,
1049 repository_full_name=(None if row.scope_key == "*" else row.scope_key),
1050 base_ref=row.base_ref or None,
1051 reason=row.reason,
1052 generation=row.generation,
1053 attempts=row.attempts,
1054 lease_owner=owner,
1055 )
1056 return None
1058 def renew_authority_claim(self, job: ClaimedAuthorityJob, lease_seconds: int) -> bool:
1059 """Extend a current authority claim without reviving stale work."""
1060 with self.session() as session:
1061 result = session.execute(
1062 update(AuthorityJob)
1063 .where(
1064 AuthorityJob.id == job.id,
1065 AuthorityJob.generation == job.generation,
1066 AuthorityJob.lease_owner == job.lease_owner,
1067 AuthorityJob.state == "pending",
1068 )
1069 .values(lease_until=utcnow() + timedelta(seconds=lease_seconds))
1070 )
1071 return getattr(result, "rowcount", 0) == 1
1073 def complete_authority(self, job: ClaimedAuthorityJob, owner: str) -> None:
1074 """Delete completed authority work or release a superseded generation."""
1075 with self.session() as session:
1076 removed = session.execute(
1077 delete(AuthorityJob).where(
1078 AuthorityJob.id == job.id,
1079 AuthorityJob.generation == job.generation,
1080 AuthorityJob.lease_owner == owner,
1081 )
1082 )
1083 if getattr(removed, "rowcount", 0) == 0:
1084 session.execute(
1085 update(AuthorityJob)
1086 .where(AuthorityJob.id == job.id, AuthorityJob.lease_owner == owner)
1087 .values(lease_owner=None, lease_until=None)
1088 )
1090 def fail_authority(
1091 self,
1092 job: ClaimedAuthorityJob,
1093 owner: str,
1094 error: str,
1095 max_delay_seconds: int,
1096 ) -> None:
1097 """Retry authority fan-out indefinitely with bounded exponential backoff."""
1098 delay_seconds = max(1, min(max_delay_seconds, 2 ** min(job.attempts, 30)))
1099 with self.session() as session:
1100 result = session.execute(
1101 update(AuthorityJob)
1102 .where(
1103 AuthorityJob.id == job.id,
1104 AuthorityJob.generation == job.generation,
1105 AuthorityJob.lease_owner == owner,
1106 )
1107 .values(
1108 state="pending",
1109 available_at=utcnow() + timedelta(seconds=delay_seconds),
1110 lease_owner=None,
1111 lease_until=None,
1112 last_error=error[:2000],
1113 )
1114 )
1115 if getattr(result, "rowcount", 0) == 0: 1115 ↛ 1116line 1115 didn't jump to line 1116 because the condition on line 1115 was never true
1116 session.execute(
1117 update(AuthorityJob)
1118 .where(AuthorityJob.id == job.id, AuthorityJob.lease_owner == owner)
1119 .values(lease_owner=None, lease_until=None)
1120 )
1122 def defer_authority(
1123 self,
1124 job: ClaimedAuthorityJob,
1125 owner: str,
1126 error: str,
1127 delay_seconds: int,
1128 ) -> bool:
1129 """Defer a rate-limited fan-out without consuming its retry budget."""
1130 delay_seconds = max(1, min(delay_seconds, 86_400))
1131 with self.session() as session:
1132 result = session.execute(
1133 update(AuthorityJob)
1134 .where(
1135 AuthorityJob.id == job.id,
1136 AuthorityJob.generation == job.generation,
1137 AuthorityJob.lease_owner == owner,
1138 )
1139 .values(
1140 attempts=case(
1141 (AuthorityJob.attempts > 0, AuthorityJob.attempts - 1),
1142 else_=0,
1143 ),
1144 available_at=utcnow() + timedelta(seconds=delay_seconds),
1145 lease_owner=None,
1146 lease_until=None,
1147 last_error=error[:2000],
1148 )
1149 )
1150 updated = getattr(result, "rowcount", 0) == 1
1151 if not updated: 1151 ↛ 1152line 1151 didn't jump to line 1152 because the condition on line 1151 was never true
1152 session.execute(
1153 update(AuthorityJob)
1154 .where(AuthorityJob.id == job.id, AuthorityJob.lease_owner == owner)
1155 .values(lease_owner=None, lease_until=None)
1156 )
1157 return updated
1159 def complete(self, job: ClaimedJob, owner: str) -> None:
1160 """Delete completed work or release a superseded generation."""
1161 with self.session() as session:
1162 removed = session.execute(
1163 delete(EvaluationJob).where(
1164 EvaluationJob.id == job.id,
1165 EvaluationJob.generation == job.generation,
1166 EvaluationJob.lease_owner == owner,
1167 )
1168 )
1169 if getattr(removed, "rowcount", 0) == 0:
1170 session.execute(
1171 update(EvaluationJob)
1172 .where(EvaluationJob.id == job.id, EvaluationJob.lease_owner == owner)
1173 .values(lease_owner=None, lease_until=None)
1174 )
1176 def fail(
1177 self,
1178 job: ClaimedJob,
1179 owner: str,
1180 error: str,
1181 max_delay_seconds: int,
1182 ) -> None:
1183 """Retry evaluation indefinitely with bounded exponential backoff."""
1184 delay_seconds = max(1, min(max_delay_seconds, 2 ** min(job.attempts, 30)))
1185 with self.session() as session:
1186 result = session.execute(
1187 update(EvaluationJob)
1188 .where(
1189 EvaluationJob.id == job.id,
1190 EvaluationJob.generation == job.generation,
1191 EvaluationJob.lease_owner == owner,
1192 )
1193 .values(
1194 state="pending",
1195 available_at=utcnow() + timedelta(seconds=delay_seconds),
1196 lease_owner=None,
1197 lease_until=None,
1198 last_error=error[:2000],
1199 )
1200 )
1201 if getattr(result, "rowcount", 0) == 0: 1201 ↛ 1202line 1201 didn't jump to line 1202 because the condition on line 1201 was never true
1202 session.execute(
1203 update(EvaluationJob)
1204 .where(EvaluationJob.id == job.id, EvaluationJob.lease_owner == owner)
1205 .values(lease_owner=None, lease_until=None)
1206 )
1208 def defer(self, job: ClaimedJob, owner: str, error: str, delay_seconds: int) -> bool:
1209 """Release a claim until a provider-supplied retry time without spending an attempt."""
1210 delay_seconds = max(1, min(delay_seconds, 86_400))
1211 with self.session() as session:
1212 result = session.execute(
1213 update(EvaluationJob)
1214 .where(
1215 EvaluationJob.id == job.id,
1216 EvaluationJob.generation == job.generation,
1217 EvaluationJob.lease_owner == owner,
1218 )
1219 .values(
1220 attempts=case(
1221 (EvaluationJob.attempts > 0, EvaluationJob.attempts - 1),
1222 else_=0,
1223 ),
1224 available_at=utcnow() + timedelta(seconds=delay_seconds),
1225 lease_owner=None,
1226 lease_until=None,
1227 last_error=error[:2000],
1228 )
1229 )
1230 updated = getattr(result, "rowcount", 0) == 1
1231 if not updated:
1232 session.execute(
1233 update(EvaluationJob)
1234 .where(EvaluationJob.id == job.id, EvaluationJob.lease_owner == owner)
1235 .values(lease_owner=None, lease_until=None)
1236 )
1237 return updated
1239 def acquire_service_lease(self, name: str, owner: str, lease_seconds: int) -> bool:
1240 """Atomically acquire or renew a named cross-process lease."""
1241 for _ in range(3): 1241 ↛ 1268line 1241 didn't jump to line 1268 because the loop on line 1241 didn't complete
1242 now = utcnow()
1243 lease_until = now + timedelta(seconds=lease_seconds)
1244 try:
1245 with self.session() as session:
1246 renewed = session.execute(
1247 update(ServiceLease)
1248 .where(
1249 ServiceLease.name == name,
1250 or_(ServiceLease.owner == owner, ServiceLease.lease_until <= now),
1251 )
1252 .values(owner=owner, lease_until=lease_until)
1253 )
1254 if getattr(renewed, "rowcount", 0) == 1:
1255 return True
1257 existing = session.scalar(
1258 select(ServiceLease.name).where(ServiceLease.name == name)
1259 )
1260 if existing is not None:
1261 return False
1262 session.add(ServiceLease(name=name, owner=owner, lease_until=lease_until))
1263 return True
1264 except IntegrityError:
1265 # Two first-time contenders may race to insert the primary key.
1266 # Retry through the conditional UPDATE after the winner commits.
1267 continue
1268 return False
1270 def pending_count(self) -> int:
1271 """Return queued and leased pending work count."""
1272 with self._sessions() as session:
1273 evaluations = int(
1274 session.scalar(
1275 select(func.count())
1276 .select_from(EvaluationJob)
1277 .where(EvaluationJob.state == "pending")
1278 )
1279 or 0
1280 )
1281 authorities = int(
1282 session.scalar(
1283 select(func.count())
1284 .select_from(AuthorityJob)
1285 .where(AuthorityJob.state == "pending")
1286 )
1287 or 0
1288 )
1289 return evaluations + authorities
1291 def dead_count(self) -> int:
1292 """Return legacy terminal rows, which startup normally reactivates."""
1293 with self._sessions() as session:
1294 evaluations = int(
1295 session.scalar(
1296 select(func.count())
1297 .select_from(EvaluationJob)
1298 .where(EvaluationJob.state == "dead")
1299 )
1300 or 0
1301 )
1302 authorities = int(
1303 session.scalar(
1304 select(func.count())
1305 .select_from(AuthorityJob)
1306 .where(AuthorityJob.state == "dead")
1307 )
1308 or 0
1309 )
1310 return evaluations + authorities
1312 def requeue_dead(self, limit: int = 100) -> int:
1313 """Recover legacy/manual terminal rows, prioritizing authority work."""
1314 with self.session() as session:
1315 authority_ids = tuple(
1316 session.scalars(
1317 select(AuthorityJob.id)
1318 .where(AuthorityJob.state == "dead")
1319 .order_by(AuthorityJob.requested_at, AuthorityJob.id)
1320 .limit(limit)
1321 )
1322 )
1323 authority_count = 0
1324 if authority_ids:
1325 authority_result = session.execute(
1326 update(AuthorityJob)
1327 .where(AuthorityJob.id.in_(authority_ids))
1328 .values(
1329 state="pending",
1330 attempts=0,
1331 available_at=utcnow(),
1332 last_error=None,
1333 lease_owner=None,
1334 lease_until=None,
1335 )
1336 )
1337 authority_count = int(getattr(authority_result, "rowcount", 0) or 0)
1338 remaining = max(0, limit - authority_count)
1339 ids = tuple(
1340 session.scalars(
1341 select(EvaluationJob.id)
1342 .where(EvaluationJob.state == "dead")
1343 .order_by(EvaluationJob.requested_at, EvaluationJob.id)
1344 .limit(remaining)
1345 )
1346 )
1347 evaluation_count = 0
1348 if ids: 1348 ↛ 1349line 1348 didn't jump to line 1349 because the condition on line 1348 was never true
1349 result = session.execute(
1350 update(EvaluationJob)
1351 .where(EvaluationJob.id.in_(ids))
1352 .values(
1353 state="pending",
1354 attempts=0,
1355 available_at=utcnow(),
1356 last_error=None,
1357 lease_owner=None,
1358 lease_until=None,
1359 )
1360 )
1361 evaluation_count = int(getattr(result, "rowcount", 0) or 0)
1362 return evaluation_count + authority_count
1364 def prune_deliveries(self, older_than: datetime) -> int:
1365 """Remove delivery de-duplication records older than a retention boundary."""
1366 with self.session() as session:
1367 result = session.execute(
1368 delete(WebhookDelivery).where(WebhookDelivery.received_at < older_than)
1369 )
1370 return int(getattr(result, "rowcount", 0) or 0)
1372 def record_audit(
1373 self,
1374 repository_full_name: str,
1375 pull_number: int,
1376 head_sha: str,
1377 conclusion: str,
1378 details: dict[str, Any],
1379 ) -> None:
1380 """Upsert the most recent evaluation evidence."""
1381 with self.session() as session:
1382 row = session.scalar(
1383 select(EvaluationAudit).where(
1384 EvaluationAudit.repository_full_name == repository_full_name,
1385 EvaluationAudit.pull_number == pull_number,
1386 )
1387 )
1388 if row is None: 1388 ↛ 1399line 1388 didn't jump to line 1399 because the condition on line 1388 was always true
1389 session.add(
1390 EvaluationAudit(
1391 repository_full_name=repository_full_name,
1392 pull_number=pull_number,
1393 head_sha=head_sha,
1394 conclusion=conclusion,
1395 details=details,
1396 )
1397 )
1398 else:
1399 row.head_sha = head_sha
1400 row.conclusion = conclusion
1401 row.details = details
1402 row.evaluated_at = utcnow()
1404 def database_available(self) -> bool:
1405 """Test connectivity and schema compatibility for readiness checks."""
1406 try:
1407 with self._sessions() as session:
1408 version = session.scalar(
1409 select(SchemaMetadata.version).where(SchemaMetadata.singleton_id == 1)
1410 )
1411 if version != SCHEMA_VERSION:
1412 return False
1413 session.execute(
1414 select(
1415 EvaluationJob.id,
1416 EvaluationJob.generation,
1417 EvaluationJob.lease_owner,
1418 AuthorityJob.generation,
1419 WebhookDelivery.invalidation_required,
1420 WebhookDelivery.invalidation_completed_at,
1421 )
1422 .select_from(EvaluationJob)
1423 .join(
1424 WebhookDelivery,
1425 EvaluationJob.last_delivery_id == WebhookDelivery.delivery_id,
1426 isouter=True,
1427 )
1428 .join(AuthorityJob, AuthorityJob.id == EvaluationJob.id, isouter=True)
1429 .limit(1)
1430 )
1431 return True
1432 except Exception:
1433 return False