Managed Agents を並列で回すと同じリポジトリを奪い合う ― 隔離サンドボックスに効く外部リースとフェンシング

import time
from google.cloud import firestore
 
db = firestore.Client()
 
LEASE_TTL = 90     # 秒。1回の危険区間より十分長く、放置回収より十分短く
HEARTBEAT = 30     # TTL の約 1/3。2回連続で更新に失敗するまで猶予がある
 
def try_acquire(resource_id: str, owner: str):
    ref = db.collection("leases").document(resource_id)
 
    @firestore.transactional
    def _txn(txn):
        snap = ref.get(transaction=txn)
        now = time.time()
        taking_over = True
        if snap.exists:
            d = snap.to_dict()
            if d["owner"] != owner and d["expires_at"] > now:
                return None            # 他者が保持中かつ未失効 → 取得できない
            fencing = d["fencing"] + 1  # 失効 or 自分の再取得 → 所有権が移る
        else:
            fencing = 1
        txn.set(ref, {
            "owner": owner,
            "expires_at": now + LEASE_TTL,
            "fencing": fencing,
        })
        return fencing
 
    return _txn(db.transaction())

def renew(resource_id: str, owner: str) -> bool:
    ref = db.collection("leases").document(resource_id)
 
    @firestore.transactional
    def _txn(txn):
        snap = ref.get(transaction=txn)
        now = time.time()
        if not snap.exists:
            return False
        d = snap.to_dict()
        if d["owner"] != owner or d["expires_at"] <= now:
            return False               # 奪われた or 失効した → 延長は拒否
        txn.update(ref, {"expires_at": now + LEASE_TTL})
        return True
 
    return _txn(db.transaction())

def release(resource_id: str, owner: str) -> None:
    ref = db.collection("leases").document(resource_id)
 
    @firestore.transactional
    def _txn(txn):
        snap = ref.get(transaction=txn)
        if snap.exists and snap.to_dict()["owner"] == owner:
            txn.update(ref, {"expires_at": 0.0})   # 即失効。fencing は残す
 
    _txn(db.transaction())

def run():
    repo = clone_repo()
    write_article(repo)
    repo.push()          # 別実行と同時に走ると衝突する

import os, uuid, threading
 
OWNER = os.getenv("RUN_ID") or uuid.uuid4().hex
 
def start_heartbeat(resource_id: str, owner: str, every: int):
    ev = threading.Event()
    def loop():
        while not ev.wait(every):
            if not renew(resource_id, owner):
                # 更新に失敗＝リースを失った。危険区間を続行してはいけない
                os._exit(75)          # 上位のリトライに委ねる（EX_TEMPFAIL 相当）
    threading.Thread(target=loop, daemon=True).start()
    return ev.set
 
def run():
    fencing = try_acquire("gemilab-repo", OWNER)
    if fencing is None:
        print("他の実行がリポジトリを保持中。今回はスキップします")
        return
    stop = start_heartbeat("gemilab-repo", OWNER, every=HEARTBEAT)
    try:
        repo = clone_repo()
        write_article(repo)
        guarded_publish("gemilab-repo", OWNER, fencing, repo.push)
    finally:
        stop()
        release("gemilab-repo", OWNER)

class StaleFencingError(Exception):
    pass
 
def guarded_publish(resource_id, owner, fencing, do_write):
    gate = db.collection("publish_gate").document(resource_id)
 
    @firestore.transactional
    def _txn(txn):
        snap = gate.get(transaction=txn)
        last = snap.to_dict()["fencing"] if snap.exists else 0
        if fencing < last:
            raise StaleFencingError(
                f"fencing {fencing} < {last}: リースは既に奪われています"
            )
        txn.set(gate, {"fencing": fencing})
 
    _txn(db.transaction())
    do_write()      # ゲートを通ってから初めて push 等の副作用を起こす