휴면 후보와 읽기 캐시 한도를 살펴보는 합성 모듈

"""Independent declared-state policy. No renderer, browser, filesystem or timer."""
from dataclasses import dataclass, fields
from copy import deepcopy
import json


def integer(value, minimum=0):
    if type(value) is not int or value < minimum:
        raise ValueError('INTEGER_REQUIRED')
    return value


@dataclass(frozen=True)
class TabState:
    last_active: int
    blocked_until: int = 0
    active: bool = False
    reference: bool = False
    pinned: bool = False
    incognito: bool = False
    agent_owned: bool = False
    loading: bool = False
    transfer: bool = False
    audible: bool = False
    captured: bool = False
    devtools: bool = False
    navigation_pending: bool = False
    interacted: bool = False
    draft: bool = False
    auth: bool = False
    https_read: bool = False
    all_frames_safe: bool = False
    unload_allowed: bool = False
    dormant: bool = False
    releasing: bool = False

    def __post_init__(self):
        integer(self.last_active);integer(self.blocked_until)
        for field in fields(self):
            if field.name not in ('last_active','blocked_until') and type(getattr(self,field.name)) is not bool:
                raise ValueError('EXPLICIT_BOOLEAN_REQUIRED')


def sleep_decision(state, now, idle_ms=300000):
    if type(state) is not TabState:
        raise ValueError('TAB_STATE_REQUIRED')
    integer(now);integer(idle_ms,1)
    if now < state.last_active:
        raise ValueError('FUTURE_ACTIVITY')
    blocked=[f.name for f in fields(state) if f.name not in ('last_active','blocked_until','https_read','all_frames_safe','unload_allowed') and getattr(state,f.name)]
    blocked += [key for key in ('https_read','all_frames_safe','unload_allowed') if not getattr(state,key)]
    if now-state.last_active < idle_ms:blocked.append('not_idle')
    if now < state.blocked_until:blocked.append('blocked_until')
    return {'candidate':not blocked,'blocked_by':blocked,'live_properties_verified':False,'renderer_released':False}


@dataclass(frozen=True)
class Context:
    profile: str
    project: str
    tab: str
    document: str
    epoch: int
    revision: str

    def __post_init__(self):
        for key in ('profile','project','tab','document','revision'):
            v=getattr(self,key)
            if type(v) is not str or not v.strip() or len(v)>200:
                raise ValueError('CONTEXT_STRING_REQUIRED')
        integer(self.epoch,1)


class PassiveCache:
    def __init__(self,max_entries=4,max_bytes=4194304,max_entry_bytes=1048576,ttl_ms=300000):
        for v in (max_entries,max_bytes,max_entry_bytes,ttl_ms):integer(v,1)
        self.limits=(max_entries,max_bytes,max_entry_bytes,ttl_ms)
        self._entries={};self._clock=-1;self._sequence=0

    def _time(self,now):
        integer(now)
        if now<self._clock:raise ValueError('CLOCK_MOVED_BACKWARDS')
        self._clock=now

    @staticmethod
    def _key(context):
        if type(context) is not Context:raise ValueError('CONTEXT_REQUIRED')
        return (context.profile,context.project,context.tab)

    def _prune(self,now):
        for k,e in list(self._entries.items()):
            if now-e['touched']>=self.limits[3]:del self._entries[k]
        while len(self._entries)>self.limits[0] or sum(e['bytes'] for e in self._entries.values())>self.limits[1]:
            k=min(self._entries,key=lambda k:(self._entries[k]['touched'],self._entries[k]['sequence']))
            del self._entries[k]

    def put(self,context,view,now):
        key=self._key(context)
        if type(view) is not dict or set(view)!={'title','text'} or any(type(v) is not str for v in view.values()):
            raise ValueError('PASSIVE_TITLE_TEXT_ONLY')
        b=len(json.dumps(view,ensure_ascii=False,separators=(',',':')).encode('utf-8'))
        self._time(now);self._prune(now)
        if b>min(self.limits[1],self.limits[2]):
            self._entries.pop(key,None)
            return {'cached':False,'reason':'oversized','bytes':b,'action_authority':False}
        self._sequence+=1
        self._entries[key]={'context':context,'view':deepcopy(view),'bytes':b,'touched':now,'sequence':self._sequence}
        self._prune(now)
        return {'cached':key in self._entries,'bytes':b,'action_authority':False}

    def get(self,context,now):
        key=self._key(context);self._time(now);self._prune(now)
        entry=self._entries.get(key)
        if entry is None:return None
        if entry['context']!=context:
            del self._entries[key];return None
        entry['touched']=now
        return {'view':deepcopy(entry['view']),'action_authority':False,'live_revision_authenticated':False}

    def stats(self,now):
        self._time(now);self._prune(now)
        return {'entries':len(self._entries),'bytes':sum(e['bytes'] for e in self._entries.values()),'renderer_released':False}