Codex
691 lines
29 KiB
Python
691 lines
29 KiB
Python
#!/usr/bin/env python3
|
|
"""Single-writer GitHub webhook topology reconciler for the Gitea source authority."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import signal
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
API_BASE = "https://api.github.com"
|
|
STOP = threading.Event()
|
|
|
|
|
|
class ApiError(RuntimeError):
|
|
def __init__(self, method, path, status):
|
|
super().__init__(f"github-api-{method.lower()}-{status}")
|
|
self.status = status
|
|
self.path = path
|
|
|
|
|
|
def url_hash(value):
|
|
return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def emit(event, **fields):
|
|
print(json.dumps({"event": event, **fields, "valuesPrinted": False}, ensure_ascii=False, separators=(",", ":")), flush=True)
|
|
|
|
|
|
def load_topology(path):
|
|
with open(path, encoding="utf-8") as handle:
|
|
topology = json.load(handle)
|
|
if topology.get("version") != 1:
|
|
raise ValueError("hook topology version must be 1")
|
|
prefix = topology.get("managedUrlPrefix")
|
|
if not isinstance(prefix, str) or not prefix.startswith("https://"):
|
|
raise ValueError("managedUrlPrefix must be https")
|
|
repositories = topology.get("repositories")
|
|
if not isinstance(repositories, list) or not repositories:
|
|
raise ValueError("hook topology repositories must not be empty")
|
|
for item in repositories:
|
|
if not isinstance(item, dict) or not isinstance(item.get("repository"), str):
|
|
raise ValueError("hook topology repository is invalid")
|
|
urls = item.get("desiredUrls")
|
|
if not isinstance(urls, list) or not urls or any(not isinstance(url, str) or not url.startswith("https://") for url in urls):
|
|
raise ValueError("hook topology desiredUrls must contain https URLs")
|
|
if len(urls) != len(set(urls)):
|
|
raise ValueError("hook topology desiredUrls must be unique per repository")
|
|
hooks = item.get("desiredHooks")
|
|
if not isinstance(hooks, list) or not hooks:
|
|
raise ValueError("hook topology desiredHooks must not be empty")
|
|
reconcile = topology.get("reconcile") if isinstance(topology.get("reconcile"), dict) else {}
|
|
recovery = reconcile.get("failedDeliveryRecovery") if isinstance(reconcile.get("failedDeliveryRecovery"), dict) else {}
|
|
if recovery.get("enabled") is True:
|
|
if recovery.get("targetId") != topology.get("ownerTargetId"):
|
|
raise ValueError("failed delivery recovery must be scoped to the owner target")
|
|
if not isinstance(recovery.get("statePath"), str) or not recovery["statePath"].startswith("/"):
|
|
raise ValueError("failed delivery recovery statePath must be absolute")
|
|
return topology
|
|
|
|
|
|
def request(token, timeout_seconds, method, api_path, payload=None, expected=(200, 201, 204), tolerate=()):
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(API_BASE + api_path, data=data, method=method)
|
|
req.add_header("Authorization", "Bearer " + token)
|
|
req.add_header("Accept", "application/vnd.github+json")
|
|
req.add_header("X-GitHub-Api-Version", "2022-11-28")
|
|
if payload is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout_seconds) as response:
|
|
body = response.read().decode("utf-8", errors="replace")
|
|
if response.status not in expected:
|
|
raise ApiError(method, api_path, response.status)
|
|
return response.status, json.loads(body) if body else None
|
|
except urllib.error.HTTPError as error:
|
|
error.read()
|
|
if error.code in tolerate:
|
|
return error.code, None
|
|
raise ApiError(method, api_path, error.code) from error
|
|
|
|
|
|
def list_hooks(token, timeout_seconds, max_pages, repository):
|
|
hooks = []
|
|
encoded = urllib.parse.quote(repository, safe="/")
|
|
for page in range(1, max_pages + 1):
|
|
_, rows = request(token, timeout_seconds, "GET", f"/repos/{encoded}/hooks?per_page=100&page={page}")
|
|
if not isinstance(rows, list):
|
|
raise RuntimeError("github-hooks-response-invalid")
|
|
hooks.extend(item for item in rows if isinstance(item, dict))
|
|
if len(rows) < 100:
|
|
return hooks
|
|
raise RuntimeError("github-hooks-pagination-bound-exhausted")
|
|
|
|
|
|
def now_iso(epoch_ms=None):
|
|
value = time.time() if epoch_ms is None else epoch_ms / 1000
|
|
return datetime.fromtimestamp(value, tz=timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def parse_epoch_ms(value):
|
|
if not isinstance(value, str) or not value:
|
|
return 0
|
|
try:
|
|
return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() * 1000)
|
|
except ValueError:
|
|
return 0
|
|
|
|
|
|
def atomic_write_ledger(path, ledger):
|
|
directory = os.path.dirname(path)
|
|
os.makedirs(directory, mode=0o700, exist_ok=True)
|
|
temporary = os.path.join(directory, f".{os.path.basename(path)}.{os.getpid()}.{time.time_ns()}.tmp")
|
|
payload = json.dumps(ledger, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n"
|
|
try:
|
|
with open(temporary, "x", encoding="utf-8") as handle:
|
|
os.chmod(temporary, 0o600)
|
|
handle.write(payload)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
directory_fd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
|
try:
|
|
os.fsync(directory_fd)
|
|
finally:
|
|
os.close(directory_fd)
|
|
finally:
|
|
try:
|
|
os.unlink(temporary)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def load_ledger(path):
|
|
try:
|
|
with open(path, encoding="utf-8") as handle:
|
|
ledger = json.load(handle)
|
|
except FileNotFoundError:
|
|
ledger = {"version": 1, "kind": "GiteaGithubHookRecoveryLedger", "createdAt": now_iso(), "records": {}}
|
|
atomic_write_ledger(path, ledger)
|
|
return ledger
|
|
if ledger.get("version") != 1 or ledger.get("kind") != "GiteaGithubHookRecoveryLedger" or not isinstance(ledger.get("records"), dict):
|
|
raise ValueError("hook recovery ledger is invalid")
|
|
allowed_states = {"pending", "attempted", "delivery-succeeded-awaiting-source", "source-committed", "terminal-exhausted"}
|
|
for record in ledger["records"].values():
|
|
if not isinstance(record, dict) or record.get("state") not in allowed_states:
|
|
raise ValueError("hook recovery ledger record state is invalid")
|
|
if not isinstance(record.get("repository"), str) or not record["repository"] or not isinstance(record.get("deliveryId"), str) or not record["deliveryId"]:
|
|
raise ValueError("hook recovery ledger record identity is invalid")
|
|
if not isinstance(record.get("attempts"), int) or record["attempts"] < 0 or not isinstance(record.get("cooldownMs"), int) or record["cooldownMs"] < 0:
|
|
raise ValueError("hook recovery ledger record counters are invalid")
|
|
return ledger
|
|
|
|
|
|
def cleanup_ledger(ledger, retention_seconds, now_ms):
|
|
cutoff = now_ms - retention_seconds * 1000
|
|
terminal = {"source-committed", "terminal-exhausted"}
|
|
stale = [
|
|
key for key, record in ledger["records"].items()
|
|
if isinstance(record, dict) and record.get("state") in terminal and int(record.get("updatedAtEpochMs") or 0) < cutoff
|
|
]
|
|
for key in stale:
|
|
del ledger["records"][key]
|
|
return len(stale)
|
|
|
|
|
|
def hook_url(hook):
|
|
config = hook.get("config") if isinstance(hook.get("config"), dict) else {}
|
|
return config.get("url") if isinstance(config.get("url"), str) else ""
|
|
|
|
|
|
def is_managed_url(value, prefix):
|
|
return value == prefix or value.startswith(prefix.rstrip("/") + "/")
|
|
|
|
|
|
def hook_config_matches(hook, events):
|
|
config = hook.get("config") if isinstance(hook.get("config"), dict) else {}
|
|
return bool(
|
|
hook.get("name") == "web"
|
|
and hook.get("active") is True
|
|
and set(hook.get("events") or []) == set(events)
|
|
and config.get("content_type") == "json"
|
|
and str(config.get("insecure_ssl", "0")) == "0"
|
|
)
|
|
|
|
|
|
def topology_observation(hooks, desired_urls, managed_prefix, events):
|
|
managed = [item for item in hooks if item.get("name") == "web" and is_managed_url(hook_url(item), managed_prefix)]
|
|
desired_set = set(desired_urls)
|
|
drift_types = []
|
|
for desired_url in desired_urls:
|
|
matches = [item for item in managed if hook_url(item) == desired_url]
|
|
if not matches:
|
|
drift_types.append("desired-url-missing")
|
|
elif len(matches) > 1:
|
|
drift_types.append("desired-url-duplicate")
|
|
if matches and not hook_config_matches(matches[0], events):
|
|
drift_types.append("desired-hook-config-mismatch")
|
|
if any(hook_url(item) not in desired_set for item in managed):
|
|
drift_types.append("stale-managed-url")
|
|
return {
|
|
"ready": not drift_types,
|
|
"desiredUrlHashes": sorted(url_hash(item) for item in desired_urls),
|
|
"observedUrlHashes": sorted(url_hash(hook_url(item)) for item in managed),
|
|
"driftTypes": sorted(set(drift_types)),
|
|
}
|
|
|
|
|
|
def list_hook_deliveries(token, timeout_seconds, repository, hook_id, history_per_hook):
|
|
encoded = urllib.parse.quote(repository, safe="/")
|
|
_, rows = request(
|
|
token,
|
|
timeout_seconds,
|
|
"GET",
|
|
f"/repos/{encoded}/hooks/{hook_id}/deliveries?per_page={history_per_hook}",
|
|
)
|
|
if not isinstance(rows, list):
|
|
raise RuntimeError("github-hook-deliveries-response-invalid")
|
|
return [item for item in rows if isinstance(item, dict)]
|
|
|
|
|
|
def hook_delivery_detail(token, timeout_seconds, repository, hook_id, api_delivery_id):
|
|
encoded = urllib.parse.quote(repository, safe="/")
|
|
_, detail = request(token, timeout_seconds, "GET", f"/repos/{encoded}/hooks/{hook_id}/deliveries/{api_delivery_id}")
|
|
if not isinstance(detail, dict):
|
|
raise RuntimeError("github-hook-delivery-detail-invalid")
|
|
payload = detail.get("request", {}).get("payload", {}) if isinstance(detail.get("request"), dict) else {}
|
|
return {
|
|
**detail,
|
|
"repositoryName": payload.get("repository", {}).get("full_name") if isinstance(payload.get("repository"), dict) else None,
|
|
"ref": payload.get("ref"),
|
|
"requestedCommit": payload.get("after"),
|
|
}
|
|
|
|
|
|
def bridge_committed_deliveries(topology, timeout_seconds):
|
|
req = urllib.request.Request(topology["bridgeStatusUrl"], method="GET")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout_seconds) as response:
|
|
body = response.read().decode("utf-8", errors="replace")
|
|
except urllib.error.HTTPError as error:
|
|
body = error.read().decode("utf-8", errors="replace")
|
|
payload = json.loads(body or "{}")
|
|
return {
|
|
(str(item.get("repo") or ""), str(item.get("deliveryId") or ""))
|
|
for item in payload.get("deliveries", [])
|
|
if isinstance(item, dict) and item.get("state") == "committed"
|
|
}
|
|
|
|
|
|
def delivery_identity(repository, hook_id, delivery_id):
|
|
return f"{repository}|{hook_id}|{delivery_id}"
|
|
|
|
|
|
def persist_record(ledger, path, key, record, now_ms):
|
|
record["updatedAtEpochMs"] = now_ms
|
|
record["updatedAt"] = now_iso(now_ms)
|
|
ledger["records"][key] = record
|
|
atomic_write_ledger(path, ledger)
|
|
|
|
|
|
def hook_body(url, events, secret):
|
|
return {
|
|
"name": "web",
|
|
"active": True,
|
|
"events": events,
|
|
"config": {"url": url, "content_type": "json", "secret": secret, "insecure_ssl": "0"},
|
|
}
|
|
|
|
|
|
def source_committed_record(ledger, state_path, key, record, status_code, now_ms):
|
|
record.update({
|
|
"state": "source-committed",
|
|
"completionReason": "source-committed",
|
|
"lastStatusCode": status_code,
|
|
"nextAt": None,
|
|
"nextAtEpochMs": 0,
|
|
})
|
|
persist_record(ledger, state_path, key, record, now_ms)
|
|
emit(
|
|
"github-hook-delivery-recovery-complete",
|
|
repository=record["repository"],
|
|
hookId=record["hookId"],
|
|
deliveryId=record["deliveryId"],
|
|
state=record["state"],
|
|
completionReason="source-committed",
|
|
attempts=record["attempts"],
|
|
ok=True,
|
|
)
|
|
|
|
|
|
def delivery_succeeded_record(ledger, state_path, key, record, status_code, now_ms):
|
|
if record.get("state") == "delivery-succeeded-awaiting-source":
|
|
return
|
|
record.update({
|
|
"state": "delivery-succeeded-awaiting-source",
|
|
"completionReason": "github-delivery-succeeded",
|
|
"lastStatusCode": status_code,
|
|
"nextAt": None,
|
|
"nextAtEpochMs": 0,
|
|
})
|
|
persist_record(ledger, state_path, key, record, now_ms)
|
|
emit(
|
|
"github-hook-delivery-succeeded-awaiting-source",
|
|
repository=record["repository"],
|
|
hookId=record["hookId"],
|
|
deliveryId=record["deliveryId"],
|
|
state=record["state"],
|
|
attempts=record["attempts"],
|
|
ok=True,
|
|
)
|
|
|
|
|
|
def terminal_record(ledger, state_path, key, record, reason, status_code, now_ms):
|
|
record.update({
|
|
"state": "terminal-exhausted",
|
|
"completionReason": reason,
|
|
"lastStatusCode": status_code,
|
|
"nextAt": None,
|
|
"nextAtEpochMs": 0,
|
|
})
|
|
persist_record(ledger, state_path, key, record, now_ms)
|
|
emit(
|
|
"github-hook-delivery-recovery-terminal",
|
|
repository=record["repository"],
|
|
hookId=record["hookId"],
|
|
deliveryId=record["deliveryId"],
|
|
state=record["state"],
|
|
completionReason=reason,
|
|
attempts=record["attempts"],
|
|
ok=False,
|
|
)
|
|
|
|
|
|
def terminalize_no_longer_desired(ledger, recovery, repository, desired_hook, observed_hook_id):
|
|
now_ms = int(time.time() * 1000)
|
|
desired_url_hash = url_hash(desired_hook["url"])
|
|
for key, record in list(ledger["records"].items()):
|
|
if not isinstance(record, dict) or record.get("repository") != repository:
|
|
continue
|
|
if record.get("targetId") not in (None, recovery["targetId"]):
|
|
continue
|
|
if record.get("state") not in ("pending", "attempted"):
|
|
continue
|
|
if int(record.get("hookId") or 0) == observed_hook_id and record.get("hookUrlSha256") == desired_url_hash:
|
|
continue
|
|
terminal_record(
|
|
ledger,
|
|
recovery["statePath"],
|
|
key,
|
|
record,
|
|
"hook-no-longer-desired",
|
|
int(record.get("lastStatusCode") or 0),
|
|
now_ms,
|
|
)
|
|
|
|
|
|
def close_committed_ledger_records(ledger, recovery, committed):
|
|
now_ms = int(time.time() * 1000)
|
|
for key, record in list(ledger["records"].items()):
|
|
if not isinstance(record, dict) or record.get("targetId") not in (None, recovery["targetId"]):
|
|
continue
|
|
if record.get("state") == "source-committed":
|
|
continue
|
|
delivery_id = str(record.get("deliveryId") or "")
|
|
repo_keys = record.get("repoKeys") if isinstance(record.get("repoKeys"), list) else []
|
|
if not any((str(repo_key), delivery_id) in committed for repo_key in repo_keys):
|
|
continue
|
|
source_committed_record(
|
|
ledger,
|
|
recovery["statePath"],
|
|
key,
|
|
record,
|
|
int(record.get("lastStatusCode") or 0),
|
|
now_ms,
|
|
)
|
|
|
|
|
|
def recover_hook_deliveries(topology, repository_spec, desired_hook, observed_hook, token, ledger, committed, budget):
|
|
reconcile = topology["reconcile"]
|
|
recovery = reconcile["failedDeliveryRecovery"]
|
|
timeout_seconds = reconcile["requestTimeoutMs"] / 1000
|
|
repository = repository_spec["repository"]
|
|
hook_id = int(observed_hook["id"])
|
|
now_ms = int(time.time() * 1000)
|
|
state_path = recovery["statePath"]
|
|
desired_refs = {f"refs/heads/{branch}" for branch in desired_hook["branches"]}
|
|
retry_codes = set(recovery["retryStatusCodes"])
|
|
rows = list_hook_deliveries(token, timeout_seconds, repository, hook_id, recovery["historyPerHook"])
|
|
rows_by_api_id = {str(item.get("id")): item for item in rows if item.get("id") is not None}
|
|
|
|
for record in ledger["records"].values():
|
|
if not isinstance(record, dict) or record.get("repository") != repository or int(record.get("hookId") or 0) != hook_id:
|
|
continue
|
|
if record.get("state") not in ("pending", "attempted", "delivery-succeeded-awaiting-source"):
|
|
continue
|
|
api_delivery_id = str(record.get("apiDeliveryId") or "")
|
|
if api_delivery_id and api_delivery_id not in rows_by_api_id:
|
|
detail = hook_delivery_detail(token, timeout_seconds, repository, hook_id, api_delivery_id)
|
|
rows.append(detail)
|
|
rows_by_api_id[api_delivery_id] = detail
|
|
|
|
latest_by_delivery = {}
|
|
for row in rows:
|
|
delivery_id = str(row.get("guid") or row.get("id") or "")
|
|
if not delivery_id:
|
|
continue
|
|
current = latest_by_delivery.get(delivery_id)
|
|
score = (parse_epoch_ms(row.get("delivered_at")), int(row.get("id") or 0))
|
|
current_score = (-1, -1) if current is None else (parse_epoch_ms(current.get("delivered_at")), int(current.get("id") or 0))
|
|
if score > current_score:
|
|
latest_by_delivery[delivery_id] = row
|
|
rows = sorted(latest_by_delivery.values(), key=lambda item: (parse_epoch_ms(item.get("delivered_at")), int(item.get("id") or 0)), reverse=True)
|
|
recovery_ok = True
|
|
for row in rows:
|
|
if row.get("event") != "push" or row.get("id") is None:
|
|
continue
|
|
api_delivery_id = str(row["id"])
|
|
delivery_id = str(row.get("guid") or row["id"])
|
|
key = delivery_identity(repository, hook_id, delivery_id)
|
|
record = ledger["records"].get(key)
|
|
if isinstance(record, dict) and record.get("state") in ("source-committed", "terminal-exhausted"):
|
|
continue
|
|
status_code = int(row.get("status_code") or 0)
|
|
eligible_failure = status_code in retry_codes
|
|
if record is None and not eligible_failure:
|
|
continue
|
|
delivered_at_ms = parse_epoch_ms(row.get("delivered_at"))
|
|
if record is None and (delivered_at_ms == 0 or delivered_at_ms < now_ms - recovery["lookbackSeconds"] * 1000):
|
|
continue
|
|
if isinstance(record, dict) and record.get("state") in ("pending", "attempted") and delivered_at_ms > 0 and delivered_at_ms < now_ms - recovery["lookbackSeconds"] * 1000:
|
|
terminal_record(ledger, state_path, key, record, "redelivery-window-expired", status_code, now_ms)
|
|
continue
|
|
|
|
detail = hook_delivery_detail(token, timeout_seconds, repository, hook_id, api_delivery_id)
|
|
status_code = int(detail.get("status_code") or status_code or 0)
|
|
if detail.get("repositoryName") != repository or detail.get("ref") not in desired_refs:
|
|
if record is not None and record.get("state") in ("pending", "attempted"):
|
|
terminal_record(ledger, state_path, key, record, "delivery-scope-mismatch", status_code, now_ms)
|
|
continue
|
|
requested_commit = str(detail.get("requestedCommit") or "")
|
|
if record is None:
|
|
record = {
|
|
"repository": repository,
|
|
"targetId": desired_hook["targetId"],
|
|
"repoKeys": desired_hook["repoKeys"],
|
|
"hookId": hook_id,
|
|
"hookUrlSha256": url_hash(desired_hook["url"]),
|
|
"deliveryId": delivery_id,
|
|
"apiDeliveryId": api_delivery_id,
|
|
"ref": detail["ref"],
|
|
"requestedCommit": requested_commit,
|
|
"state": "pending",
|
|
"attempts": 0,
|
|
"cooldownMs": recovery["cooldownMs"],
|
|
"nextAtEpochMs": now_ms,
|
|
"nextAt": now_iso(now_ms),
|
|
"createdAt": now_iso(now_ms),
|
|
"lastStatusCode": status_code,
|
|
"completionReason": None,
|
|
}
|
|
persist_record(ledger, state_path, key, record, now_ms)
|
|
emit(
|
|
"github-hook-delivery-recovery-pending",
|
|
repository=repository,
|
|
hookId=hook_id,
|
|
deliveryId=delivery_id,
|
|
state="pending",
|
|
attempts=0,
|
|
nextAt=record["nextAt"],
|
|
ok=True,
|
|
)
|
|
|
|
source_committed = any((repo_key, delivery_id) in committed for repo_key in desired_hook["repoKeys"])
|
|
if source_committed:
|
|
source_committed_record(ledger, state_path, key, record, status_code, now_ms)
|
|
continue
|
|
if 200 <= status_code < 300:
|
|
delivery_succeeded_record(ledger, state_path, key, record, status_code, now_ms)
|
|
continue
|
|
if record.get("state") == "delivery-succeeded-awaiting-source":
|
|
continue
|
|
if status_code == 0:
|
|
continue
|
|
if status_code not in retry_codes:
|
|
terminal_record(ledger, state_path, key, record, "non-retryable-status", status_code, now_ms)
|
|
continue
|
|
if int(record.get("nextAtEpochMs") or 0) > now_ms:
|
|
continue
|
|
if int(record.get("attempts") or 0) >= recovery["maxAttemptsPerDelivery"]:
|
|
terminal_record(ledger, state_path, key, record, "max-attempts-exhausted", status_code, now_ms)
|
|
continue
|
|
if budget["remaining"] < 1:
|
|
continue
|
|
|
|
record["state"] = "attempted"
|
|
record["attempts"] = int(record.get("attempts") or 0) + 1
|
|
record["lastStatusCode"] = status_code
|
|
record["lastOutcome"] = "redelivery-requested-write-ahead"
|
|
record["nextAtEpochMs"] = now_ms + recovery["cooldownMs"]
|
|
record["nextAt"] = now_iso(record["nextAtEpochMs"])
|
|
persist_record(ledger, state_path, key, record, now_ms)
|
|
budget["remaining"] -= 1
|
|
encoded = urllib.parse.quote(repository, safe="/")
|
|
try:
|
|
request(
|
|
token,
|
|
timeout_seconds,
|
|
"POST",
|
|
f"/repos/{encoded}/hooks/{hook_id}/deliveries/{api_delivery_id}/attempts",
|
|
expected=(202,),
|
|
)
|
|
record["lastOutcome"] = "redelivery-accepted"
|
|
persist_record(ledger, state_path, key, record, now_ms)
|
|
emit(
|
|
"github-hook-delivery-redelivery-requested",
|
|
repository=repository,
|
|
hookId=hook_id,
|
|
deliveryId=delivery_id,
|
|
state=record["state"],
|
|
attempts=record["attempts"],
|
|
nextAt=record["nextAt"],
|
|
ok=True,
|
|
)
|
|
except Exception as error:
|
|
recovery_ok = False
|
|
record["lastOutcome"] = "redelivery-request-failed"
|
|
persist_record(ledger, state_path, key, record, now_ms)
|
|
emit(
|
|
"github-hook-delivery-redelivery-failed",
|
|
repository=repository,
|
|
hookId=hook_id,
|
|
deliveryId=delivery_id,
|
|
state=record["state"],
|
|
attempts=record["attempts"],
|
|
nextAt=record["nextAt"],
|
|
errorType=type(error).__name__,
|
|
error=str(error)[:160],
|
|
ok=False,
|
|
)
|
|
return recovery_ok
|
|
|
|
|
|
def reconcile_repository(topology, repository_spec, token, secret, force_patch):
|
|
reconcile = topology["reconcile"]
|
|
timeout_seconds = reconcile["requestTimeoutMs"] / 1000
|
|
max_pages = reconcile["listMaxPages"]
|
|
repository = repository_spec["repository"]
|
|
desired_urls = repository_spec["desiredUrls"]
|
|
events = topology["events"]
|
|
prefix = topology["managedUrlPrefix"]
|
|
hooks = list_hooks(token, timeout_seconds, max_pages, repository)
|
|
mutations = {"created": 0, "updated": 0, "deleted": 0}
|
|
encoded = urllib.parse.quote(repository, safe="/")
|
|
|
|
for desired_url in desired_urls:
|
|
matches = sorted(
|
|
[item for item in hooks if item.get("name") == "web" and hook_url(item) == desired_url],
|
|
key=lambda item: int(item.get("id") or 0),
|
|
)
|
|
keeper = matches[0] if matches else None
|
|
if keeper is None:
|
|
_, created = request(token, timeout_seconds, "POST", f"/repos/{encoded}/hooks", hook_body(desired_url, events, secret), expected=(201,))
|
|
if isinstance(created, dict):
|
|
hooks.append(created)
|
|
mutations["created"] += 1
|
|
elif force_patch or not hook_config_matches(keeper, events):
|
|
hook_id = int(keeper["id"])
|
|
request(token, timeout_seconds, "PATCH", f"/repos/{encoded}/hooks/{hook_id}", hook_body(desired_url, events, secret))
|
|
mutations["updated"] += 1
|
|
for duplicate in matches[1:]:
|
|
hook_id = int(duplicate["id"])
|
|
request(token, timeout_seconds, "DELETE", f"/repos/{encoded}/hooks/{hook_id}", expected=(204,), tolerate=(404,))
|
|
mutations["deleted"] += 1
|
|
|
|
desired_set = set(desired_urls)
|
|
for stale in hooks:
|
|
stale_url = hook_url(stale)
|
|
if stale.get("name") != "web" or not is_managed_url(stale_url, prefix) or stale_url in desired_set:
|
|
continue
|
|
hook_id = int(stale["id"])
|
|
request(token, timeout_seconds, "DELETE", f"/repos/{encoded}/hooks/{hook_id}", expected=(204,), tolerate=(404,))
|
|
mutations["deleted"] += 1
|
|
|
|
observed = list_hooks(token, timeout_seconds, max_pages, repository)
|
|
observation = topology_observation(observed, desired_urls, prefix, events)
|
|
emit(
|
|
"github-hook-topology-reconciled",
|
|
repository=repository,
|
|
ok=observation["ready"],
|
|
desiredUrlHashes=observation["desiredUrlHashes"],
|
|
observedUrlHashes=observation["observedUrlHashes"],
|
|
driftTypes=observation["driftTypes"],
|
|
mutations=mutations,
|
|
)
|
|
return observation, observed
|
|
|
|
|
|
def reconcile_once(topology, secret, force_patch):
|
|
started = time.monotonic()
|
|
topology_ready = True
|
|
recovery_ready = True
|
|
recovery = topology["reconcile"]["failedDeliveryRecovery"]
|
|
ledger = None
|
|
committed = set()
|
|
budget = {"remaining": recovery["maxRedeliveriesPerCycle"]}
|
|
if recovery["enabled"]:
|
|
try:
|
|
ledger = load_ledger(recovery["statePath"])
|
|
removed = cleanup_ledger(ledger, recovery["stateRetentionSeconds"], int(time.time() * 1000))
|
|
if removed:
|
|
atomic_write_ledger(recovery["statePath"], ledger)
|
|
committed = bridge_committed_deliveries(topology, topology["reconcile"]["requestTimeoutMs"] / 1000)
|
|
close_committed_ledger_records(ledger, recovery, committed)
|
|
except Exception as error:
|
|
recovery_ready = False
|
|
ledger = None
|
|
emit("github-hook-delivery-recovery-ledger-unavailable", ok=False, errorType=type(error).__name__, error=str(error)[:160])
|
|
for repository_spec in topology["repositories"]:
|
|
try:
|
|
token = os.environ[repository_spec["githubTokenEnv"]]
|
|
observation, observed_hooks = reconcile_repository(topology, repository_spec, token, secret, force_patch)
|
|
if not observation["ready"]:
|
|
topology_ready = False
|
|
continue
|
|
if recovery["enabled"] and ledger is not None:
|
|
desired_hooks = [item for item in repository_spec["desiredHooks"] if item["targetId"] == recovery["targetId"]]
|
|
if len(desired_hooks) != 1:
|
|
recovery_ready = False
|
|
emit("github-hook-delivery-recovery-scope-invalid", repository=repository_spec["repository"], ok=False, targetId=recovery["targetId"])
|
|
continue
|
|
desired_hook = desired_hooks[0]
|
|
exact = [item for item in observed_hooks if item.get("name") == "web" and hook_url(item) == desired_hook["url"]]
|
|
if len(exact) != 1 or not hook_config_matches(exact[0], topology["events"]):
|
|
recovery_ready = False
|
|
emit("github-hook-delivery-recovery-skipped-topology-drift", repository=repository_spec["repository"], ok=False, targetId=recovery["targetId"])
|
|
continue
|
|
terminalize_no_longer_desired(ledger, recovery, repository_spec["repository"], desired_hook, int(exact[0]["id"]))
|
|
if not recover_hook_deliveries(topology, repository_spec, desired_hook, exact[0], token, ledger, committed, budget):
|
|
recovery_ready = False
|
|
except Exception as error:
|
|
topology_ready = False
|
|
recovery_ready = False
|
|
emit(
|
|
"github-hook-topology-reconcile-failed",
|
|
repository=repository_spec["repository"],
|
|
ok=False,
|
|
errorType=type(error).__name__,
|
|
error=str(error)[:160],
|
|
)
|
|
emit(
|
|
"github-hook-topology-cycle-complete",
|
|
ok=topology_ready and recovery_ready,
|
|
topologyReady=topology_ready,
|
|
recoveryReady=recovery_ready,
|
|
redeliveryRequests=recovery["maxRedeliveriesPerCycle"] - budget["remaining"],
|
|
repositoryCount=len(topology["repositories"]),
|
|
elapsedMs=round((time.monotonic() - started) * 1000),
|
|
)
|
|
return topology_ready
|
|
|
|
|
|
def stop(_signum, _frame):
|
|
STOP.set()
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
raise SystemExit("usage: platform_infra_gitea_hook_reconciler.py <topology.json>")
|
|
topology = load_topology(sys.argv[1])
|
|
if topology.get("ownerTargetId") != os.environ.get("UNIDESK_GITEA_TARGET_ID"):
|
|
raise SystemExit("hook reconciler target is not the YAML owner")
|
|
secret = os.environ["GITHUB_WEBHOOK_SECRET"]
|
|
signal.signal(signal.SIGTERM, stop)
|
|
signal.signal(signal.SIGINT, stop)
|
|
force_patch = True
|
|
while not STOP.is_set():
|
|
if reconcile_once(topology, secret, force_patch):
|
|
force_patch = False
|
|
STOP.wait(topology["reconcile"]["intervalMs"] / 1000)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|