File size: 3,478 Bytes
30ea0e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | from __future__ import annotations
from functools import lru_cache
from app.audit import AuditLogger
from app.config import Settings, get_settings
from app.dedup import PromotionLRU
from app.hub import HubClient
from app.job_quota import DurableJobQuota
from app.jobs import JobRunner
from app.notify import Notifier
from app.org_roles import OrgRoles
from app.rate_limit import CompoundLimiter, TokenBucket
from app.read_model import ReadModel
from app.verification import VerificationStatusStore
from app.verifier import Verifier
_DAY_SECONDS = 24 * 60 * 60
@lru_cache
def get_hub() -> HubClient:
return HubClient(get_settings())
@lru_cache
def get_job_runner() -> JobRunner:
return JobRunner(get_settings(), get_hub())
@lru_cache
def get_job_quota() -> DurableJobQuota:
s = get_settings()
return DurableJobQuota(
hub=get_hub(),
path=s.job_quota_ledger_path,
agent_limit=s.job_per_agent_per_day,
user_limit=s.job_per_user_per_day,
window_seconds=_DAY_SECONDS,
)
@lru_cache
def get_read_model() -> ReadModel:
return ReadModel(get_hub(), get_settings())
@lru_cache
def get_notifier() -> Notifier:
"""The one in-process long-poll waiter registry (WATCH_DESIGN.md §3). It is
per-process by design, which is why the Dockerfile pins `--workers 1`: with
more workers a writer would wake only the waiters that happen to share its
worker and every other `wait=` would silently time out."""
s = get_settings()
return Notifier(
max_waiters_per_owner=s.longpoll_max_waiters_per_owner,
max_waiters_total=s.longpoll_max_waiters_total,
wake_spread_s=s.longpoll_wake_spread_s,
wake_spread_threshold=s.longpoll_wake_spread_threshold,
)
@lru_cache
def get_audit() -> AuditLogger:
return AuditLogger(
get_hub(), record_client=not get_settings().audit_bucket_in_org
)
@lru_cache
def get_org_roles() -> OrgRoles:
return OrgRoles(get_hub(), get_settings())
@lru_cache
def get_dedup() -> PromotionLRU:
return PromotionLRU(max_entries=get_settings().dedup_lru_size)
@lru_cache
def get_verification_status() -> VerificationStatusStore:
return VerificationStatusStore(
get_hub(), runs_prefix=get_settings().verification_runs_prefix
)
@lru_cache
def get_verifier() -> Verifier:
return Verifier(
get_settings(),
get_hub(),
get_read_model(),
get_verification_status(),
get_job_runner(),
notifier=get_notifier(),
)
@lru_cache
def get_bucket_write_limiter() -> CompoundLimiter:
s = get_settings()
burst = TokenBucket(capacity=s.bucket_write_burst, refill_per_minute=s.bucket_write_burst)
sustained = TokenBucket(capacity=s.bucket_write_per_minute, refill_per_minute=s.bucket_write_per_minute)
return CompoundLimiter(burst, sustained)
@lru_cache
def get_raw_message_limiter() -> CompoundLimiter:
s = get_settings()
per_minute = TokenBucket(capacity=s.raw_message_per_minute, refill_per_minute=s.raw_message_per_minute)
per_hour = TokenBucket(capacity=s.raw_message_per_hour, refill_per_minute=max(1, s.raw_message_per_hour // 60))
return CompoundLimiter(per_minute, per_hour)
@lru_cache
def get_registration_limiter() -> TokenBucket:
s = get_settings()
return TokenBucket(capacity=s.registration_per_minute, refill_per_minute=s.registration_per_minute)
def get_settings_dep() -> Settings:
return get_settings()
|