diff --git a/docs/keria_app.rst b/docs/keria_app.rst index 6427eb55..6e8c12c5 100644 --- a/docs/keria_app.rst +++ b/docs/keria_app.rst @@ -1,5 +1,5 @@ -KERIA App API -============= +KERIA App API Documentation +=========================== keria.app.agenting ------------------ @@ -8,14 +8,23 @@ keria.app.agenting :members: Agency and Agent Configuration -=================== +============================== A KERIA Agency can be configured with either environment variables or a configuration file. -The configuration file is a JSON file. Both alternatives are shown here. +The configuration file is a JSON file. Both alternatives are shown here. A subset of the options +available in the config file may alternatively be set with environment variables. When both are +defined the environment variables take precedence over config file properties. + +Configuration uses the following precedence strategy: + +* ``specific env > coarse env > specific config > coarse config > default`` Environment Variables --------------------- +KERIA may be configured with environment variables as shown below. A complete list of tock-level +configuration options are shown in :ref:`scheduler-tock-config`. + .. code-block:: bash # Service Endpoint Location URLs creating Endpoint Role Authorizations and Location Scheme records on startup @@ -30,11 +39,16 @@ Environment Variables JSON Configuration File ----------------------- -To use the JSON configuration file option make sure to mount the JSON file to the directory you specify with the -`--config-dir` option and name the JSON file the name specified by the `--config-dir` option to the `keria start` command like so. +The JSON configuration is the default configuration source and 100% of the configuration options for +KERIA are located in the config file. -With the absolute path version the `--config-dir` argument does not have an affect since the -`--config-file` argument specifies the absolute path to the JSON file. +.. Warning:: + When using JSON config with containers make sure to mount the JSON file to the directory specified in the + ``--config-dir`` option and use the JSON file name specified by the ``--config-dir`` option to the + ``keria start`` command as shown below. + +.. Note:: + Using the absolute path variant for ``--config-file`` ignores the ``--config-dir`` argument. .. code-block:: bash @@ -49,35 +63,171 @@ With the absolute path version the `--config-dir` argument does not have an affe --config-dir /path/to/config-dir/keria.json \ --config-file /path/to/config-dir/keria.json -The JSON file must have an object with the same name that you sent to the `keria start` command via the `--name` argument. -The default is "keria" which is why the JSON file below shows a sub-object named "keria". -Make sure to include the "dt" date timestamp field or the configuration will not be loaded. +.. Warning:: + The JSON file must have an object with the same name that you sent to the ``keria start`` command via the ``--name`` argument. + The default is ``"keria"`` which is why the JSON file below shows a sub-object named ``"keria"``. + Make sure to include the ``"dt"`` date timestamp field or the configuration will not be loaded. + +KERIA tocks in `tocks.signify` +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -You can configure the cycle time, or tocks, of the escrower as well as the agent initializer. +Configure KERIA Agent tocks in the JSON config file under ``tocks.signify``. KERIpy service settings +remain directly under ``tocks``. Values are HIO Doist scheduler intervals in seconds as a float value. You can also configure the CURLs, IURLs, and DURLs of the agent. -CURLs are Service Endpoint Location URLs creating Endpoint Role Authorizations and Location Scheme records on startup. -IURLS are Introduction URLs resolved on startup (OOBIs). -DURLS are Data OOBI URLs resolved on startup usually of things like ACDC credential schemas or ACDC credential CESR streams. - -.. code-block:: json - - { - "dt": "2025-01-13T16:08:30.123456+00:00", - "keria": { - "dt": "2025-01-13T16:08:30.123457+00:00", - "curls": ["http://127.0.0.1:3902/"] - }, - "iurls": [ - "http://127.0.0.1:5642/oobi/BBilc4-L3tFUnfM_wJr4S4OJanAv_VmF_dJNN6vkf2Ha/controller?name=Wan&tag=witness", - "http://127.0.0.1:5643/oobi/BLskRTInXnMxWaGqcpSyMgo0nYbalW99cGZESrz3zapM/controller?name=Wil&tag=witness", - "http://127.0.0.1:5644/oobi/BIKKuvBwpmDVA4Ds-EpL5bt9OqPzWPja2LigFYZN2YfX/controller?name=Wes&tag=witness" - ], - "tocks": { - "initer": 0.0, - "escrower": 1.0 - } - } + +* ``CURLs`` are **Service Endpoint Location** URLs creating Endpoint Role Authorizations and Location Scheme records on startup. +* ``IURLS`` are **Introduction** URLs resolved on startup (OOBIs). +* ``DURLS`` are **Data OOBI** URLs resolved on startup usually of things like ACDC credential schemas or ACDC credential CESR streams. + + +KERIA Configuration File Example +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Note that the following configuration file uses the name ``keria`` for the Agency as it has a +``"keria"`` top level section of the configuration JSON. This corresponds to using the ``keria`` +value for ``keria start --name "keria"`` when starting KERIA. + +The existing sample lists every KERIpy and KERIA component cadence explicitly. +Coarse aliases are omitted because they are shortcuts for these same settings. + +.. literalinclude:: ../scripts/keri/cf/keria.json + :language: json + +.. _scheduler-tock-config: + +Scheduler Tock Configuration (HIO Doist scheduler) +================================================== + +Except for the release scan described below, KERIA defaults its registered +component cadences, including its KERIpy service settings, to ``0.03125`` seconds: one tick of its 32 Hz root scheduler. Explicit +``0.0`` remains supported and schedules work on the next cycle. File and +environment overrides retain their normal precedence; KERIpy's own defaults +outside KERIA are unchanged. + +Delegation escrow processing uses ``tocks.anchorerEscrow`` throughout its +runtime, without a separate half-second delay. Witness resubmission's outbound +drain wait uses ``tocks.witnessReceiptorIdle`` and also defaults to one tick. + +The Agent release scan, ``tocks.signify.releaser``, is the housekeeping +exception with a default of ``60.0`` seconds. It scans all open Agents for the default 24-hour inactivity +timeout. Allowing up to one minute of additional cleanup delay avoids a full +scan every tick without delaying active message processing. The release timeout +itself is separate from scheduler cadence and is not shortened. + +.. list-table:: KERIA settings under tocks.signify + :header-rows: 1 + + * - File key + - Environment override + * - ``initer`` + - ``KERIA_INITER_TOCK`` + * - ``querier`` + - ``KERIA_QUERIER_TOCK`` + * - ``escrower`` + - ``KERIA_ESCROWER_TOCK`` + * - ``parser`` + - ``KERIA_PARSER_TOCK`` + * - ``witnesser`` + - ``KERIA_WITNESSER_TOCK`` + * - ``delegator`` + - ``KERIA_DELEGATOR_TOCK`` + * - ``exchangeSender`` + - ``KERIA_EXCHANGE_SENDER_TOCK`` + * - ``granter`` + - ``KERIA_GRANTER_TOCK`` + * - ``admitter`` + - ``KERIA_ADMITTER_TOCK`` + * - ``groupRequester`` + - ``KERIA_GROUP_REQUESTER_TOCK`` + * - ``seeker`` + - ``KERIA_SEEKER_TOCK`` + * - ``exchangecue`` + - ``KERIA_EXCHANGE_CUE_TOCK`` + * - ``submitter`` + - ``KERIA_SUBMITTER_TOCK`` + * - ``bootServer`` + - ``KERIA_BOOT_SERVER_TOCK`` + * - ``adminServer`` + - ``KERIA_ADMIN_SERVER_TOCK`` + * - ``httpServer`` + - ``KERIA_HTTP_SERVER_TOCK`` + * - ``shutdown`` + - ``KERIA_SHUTDOWN_TOCK`` + * - ``signalExpiry`` + - ``KERIA_SIGNAL_EXPIRY_TOCK`` + * - ``releaser`` + - ``KERIA_RELEASER_TOCK`` + +Parser waits and witness receipt/catch-up waits use their configured component +cadences. The Agent owns a ``SignalExpirer`` for its Signaler; ``signalExpiry`` controls +its scan cadence, not the ten-minute signal lifetime. Each scan uses a snapshot +and does not yield during iteration, so notification pushes between scheduler +cycles cannot invalidate a live deque iterator. + +Agency/Agent containers follow the root scheduler cadence. Habery lifecycle +work and delegation containers inherit their owning container's cadence; +delegation's escrow loop retains its independent ``anchorerEscrow`` setting. +Transient grant workers, query workers, and delivery wrappers use their +parent's cadence. KERIpy's internal Poster, Clienter, and transport scheduling +is outside these KERIA settings. One-off CLI commands have no dedicated keys. + +The server, shutdown, and release settings bind at Agency startup. Agent-owned +settings, including ``submitter`` and ``signalExpiry``, bind at creation or +reopen using that Agent's configuration lifecycle. + + +Configuration Precedence Strategy +--------------------------------- + +Resolution uses the same rule as KERIpy: + +``specific env > coarse env > specific config > coarse config > default`` + +For example, ``KERIA_AGENT_TOCK=0.5`` overrides a file's +``tocks.signify.escrower=1.0``, while ``KERIA_ESCROWER_TOCK=0.25`` overrides +both. These KERIA environment variables do not alter KERIpy's independent +``KERI_*_TOCK`` settings. KERIA's Registrar and Credentialer are processed by +its central ``escrower``; KERIpy's separate registrar/credentialer escrow +settings do not schedule those KERIA objects. + +All file values must be finite, nonnegative JSON numbers; booleans, strings, +nulls, unknown keys, and malformed blocks are errors. Environment values must +parse as finite, nonnegative numbers. Invalid supplied values are rejected even +when another setting would override them. Errors identify the invalid setting or configuration structure. + +Coarse grained settings +----------------------- + +The coarse grained ``tocks.signify.agent`` setting, or ``KERIA_AGENT_TOCK``, applies to +the original twelve Agent-work settings plus ``submitter``. It excludes +``bootServer``, ``adminServer``, ``httpServer``, ``shutdown``, ``signalExpiry``, +and ``releaser`` so a work-loop override does not alter server or housekeeping +cadences. + +Backwards compatibility with legacy flat tocks +---------------------------------------------- + +The existing names remain supported directly under ``tocks`` for backward +compatibility. For example, ``{"tocks": {"escrower": 1.0}}`` is accepted but +logs one deprecation warning per configuration load. Move it to +``{"tocks": {"signify": {"escrower": 1.0}}}``. Identical old/new duplicates +are accepted; conflicting duplicates fail instead of silently choosing a value. +Legacy placement will remain available through a migration period; removal +will be announced in a future release. + +The Agency resolves its configuration at load time and this is used as a template for any +Agent provisioned by that Agency. Newly provisioned Agents receive a copy of the configured tocks +from the Agency's JSON config file and do not persist environment overrides or defaults. + +.. Warning:: + Existing Agents reopen their own persisted config files, including legacy files; those files are not automatically rewritten. + Updating the Agency JSON config file does not change existing Agent JSON config files. + Restart the Agency to load config changes for subsequently provisioned Agents. + Edit an existing Agent's individual config and restart to apply changes; environment overrides provide + process-wide tuning when services are reconstructed. There is no live reload. + +These settings configure the existing Doer constructors. keria.app.aiding ---------------- diff --git a/scripts/keri/cf/keria.json b/scripts/keri/cf/keria.json index f44b8bcd..2101f5cd 100755 --- a/scripts/keri/cf/keria.json +++ b/scripts/keri/cf/keria.json @@ -2,7 +2,9 @@ "dt": "2025-01-03T16:08:30.123456+00:00", "keria": { "dt": "2025-01-03T16:08:30.123457+00:00", - "curls": ["http://127.0.0.1:3902/"] + "curls": [ + "http://127.0.0.1:3902/" + ] }, "iurls": [ "http://127.0.0.1:5642/oobi/BBilc4-L3tFUnfM_wJr4S4OJanAv_VmF_dJNN6vkf2Ha/controller?name=Wan&tag=witness", @@ -10,7 +12,55 @@ "http://127.0.0.1:5644/oobi/BIKKuvBwpmDVA4Ds-EpL5bt9OqPzWPja2LigFYZN2YfX/controller?name=Wes&tag=witness" ], "tocks": { - "initer": 0.0, - "escrower": 1.0 + "witnessMsg": 0.03125, + "witnessEscrow": 0.03125, + "witnessCue": 0.03125, + "indirectorMsg": 0.03125, + "indirectorCue": 0.03125, + "indirectorEscrow": 0.03125, + "mailboxPoll": 0.03125, + "mailboxMsg": 0.03125, + "mailboxEscrow": 0.03125, + "pollerEvent": 0.03125, + "pollerConnect": 0.03125, + "receiptIntercept": 0.03125, + "reactorMsg": 0.03125, + "reactorCue": 0.03125, + "reactorEscrow": 0.03125, + "reactantMsg": 0.03125, + "reactantCue": 0.03125, + "reactantEscrow": 0.03125, + "respondant": 0.03125, + "receiptor": 0.03125, + "oobi": 0.03125, + "witnessReceiptor": 0.03125, + "witnessReceiptorIdle": 0.03125, + "witnessInquisitor": 0.03125, + "witnessPublisher": 0.03125, + "registrarEscrow": 0.03125, + "credentialerEscrow": 0.03125, + "counselorEscrow": 0.03125, + "anchorerEscrow": 0.03125, + "signify": { + "initer": 0.03125, + "querier": 0.03125, + "escrower": 0.03125, + "parser": 0.03125, + "witnesser": 0.03125, + "delegator": 0.03125, + "exchangeSender": 0.03125, + "granter": 0.03125, + "admitter": 0.03125, + "groupRequester": 0.03125, + "seeker": 0.03125, + "exchangecue": 0.03125, + "submitter": 0.03125, + "bootServer": 0.03125, + "adminServer": 0.03125, + "httpServer": 0.03125, + "shutdown": 0.03125, + "signalExpiry": 0.03125, + "releaser": 60.0 + } } -} \ No newline at end of file +} diff --git a/scripts/keria.json b/scripts/keria.json index f44b8bcd..2101f5cd 100755 --- a/scripts/keria.json +++ b/scripts/keria.json @@ -2,7 +2,9 @@ "dt": "2025-01-03T16:08:30.123456+00:00", "keria": { "dt": "2025-01-03T16:08:30.123457+00:00", - "curls": ["http://127.0.0.1:3902/"] + "curls": [ + "http://127.0.0.1:3902/" + ] }, "iurls": [ "http://127.0.0.1:5642/oobi/BBilc4-L3tFUnfM_wJr4S4OJanAv_VmF_dJNN6vkf2Ha/controller?name=Wan&tag=witness", @@ -10,7 +12,55 @@ "http://127.0.0.1:5644/oobi/BIKKuvBwpmDVA4Ds-EpL5bt9OqPzWPja2LigFYZN2YfX/controller?name=Wes&tag=witness" ], "tocks": { - "initer": 0.0, - "escrower": 1.0 + "witnessMsg": 0.03125, + "witnessEscrow": 0.03125, + "witnessCue": 0.03125, + "indirectorMsg": 0.03125, + "indirectorCue": 0.03125, + "indirectorEscrow": 0.03125, + "mailboxPoll": 0.03125, + "mailboxMsg": 0.03125, + "mailboxEscrow": 0.03125, + "pollerEvent": 0.03125, + "pollerConnect": 0.03125, + "receiptIntercept": 0.03125, + "reactorMsg": 0.03125, + "reactorCue": 0.03125, + "reactorEscrow": 0.03125, + "reactantMsg": 0.03125, + "reactantCue": 0.03125, + "reactantEscrow": 0.03125, + "respondant": 0.03125, + "receiptor": 0.03125, + "oobi": 0.03125, + "witnessReceiptor": 0.03125, + "witnessReceiptorIdle": 0.03125, + "witnessInquisitor": 0.03125, + "witnessPublisher": 0.03125, + "registrarEscrow": 0.03125, + "credentialerEscrow": 0.03125, + "counselorEscrow": 0.03125, + "anchorerEscrow": 0.03125, + "signify": { + "initer": 0.03125, + "querier": 0.03125, + "escrower": 0.03125, + "parser": 0.03125, + "witnesser": 0.03125, + "delegator": 0.03125, + "exchangeSender": 0.03125, + "granter": 0.03125, + "admitter": 0.03125, + "groupRequester": 0.03125, + "seeker": 0.03125, + "exchangecue": 0.03125, + "submitter": 0.03125, + "bootServer": 0.03125, + "adminServer": 0.03125, + "httpServer": 0.03125, + "shutdown": 0.03125, + "signalExpiry": 0.03125, + "releaser": 60.0 + } } -} \ No newline at end of file +} diff --git a/src/keria/app/agenting.py b/src/keria/app/agenting.py index b1bb07a9..05316d1b 100644 --- a/src/keria/app/agenting.py +++ b/src/keria/app/agenting.py @@ -10,6 +10,7 @@ from base64 import b64decode import json import datetime +from copy import deepcopy from dataclasses import asdict, dataclass, field from typing import List, Union from urllib.parse import urlparse, urljoin @@ -39,7 +40,6 @@ querying, connecting, grouping, - tocking, ) from keri.app import delegating as kdelegating from keri.app.grouping import Counselor @@ -60,7 +60,15 @@ from keria.utils.openapi import dataclassFromFielddom -from . import aiding, notifying, indirecting, credentialing, ipexing, delegating +from . import ( + aiding, + notifying, + indirecting, + credentialing, + ipexing, + delegating, + scheduling, +) from . import grouping as keriagrouping from .serving import GracefulShutdownDoer from .. import log_name, ogler, set_log_level @@ -165,32 +173,6 @@ def readConfigFile(configDir: str, configFile: str, temp=False): ) -def keriTocks(cf): - """Resolve KERIpy cadences while reserving the existing KERIA Agent keys. - - Keep the config file intact so Agent can still read its legacy flat tocks. - Unknown keys and KERIpy values remain subject to KERIpy's validation. - """ - return tocking.resolveTocks( - cf.get().get("tocks") if cf is not None else None, - reserved=( - "signify", - "initer", - "querier", - "escrower", - "parser", - "witnesser", - "delegator", - "exchangeSender", - "granter", - "admitter", - "groupRequester", - "seeker", - "exchangecue", - ), - ) - - def runAgency(config: KERIAServerConfig, temp=False): """Runs a KERIA Agency with the given Doers by calling Doist.do(). Useful for testing.""" set_log_level(config.logLevel, logger) @@ -219,9 +201,10 @@ def runAgency(config: KERIAServerConfig, temp=False): def agencyDoist(doers: List[Doer]): """Creates a Doist for the Agency doers and adds a graceful shutdown handler. Useful for testing.""" - tock = 0.03125 + tock = scheduling.DEFAULT_TOCK doist = doing.Doist(limit=0.0, tock=tock, real=True) - doers.append(GracefulShutdownDoer(agency=getAgency(doers))) + agency = getAgency(doers) + doers.append(GracefulShutdownDoer(agency=agency, tock=agency.tocks["shutdown"])) doist.doers = doers return doist @@ -298,6 +281,8 @@ def __init__( else: self.cf = cf + self._tockConfig = scheduling.loadTocks(self.cf) + self.tocks = dict(self._tockConfig.signify) self.agents = dict() self.adb = ( @@ -305,8 +290,12 @@ def __init__( if adb is not None else basing.AgencyBaser(name="TheAgency", base=base, reopen=True, temp=temp) ) + self.releaser = Releaser( + self, releaseTimeout=releaseTimeout, tock=self.tocks["releaser"] + ) super(Agency, self).__init__( - doers=[Releaser(self, releaseTimeout=releaseTimeout)] + doers=[self.releaser], + tock=scheduling.DEFAULT_TOCK, ) def _loadConfigForAgent(self, caid): @@ -323,6 +312,9 @@ def _loadConfigForAgent(self, caid): timestamp = nowIso8601() config = dict(self.cf.get() if self.cf is not None else {"dt": timestamp}) + # Persist the configured snapshot, not environment overrides or defaults. + config["tocks"] = deepcopy(self._tockConfig.configured) + # Renames sub-section of config habName = f"agent-{caid}" config_name = self.name if self.name else "keria" @@ -388,7 +380,7 @@ def create(self, caid, salt=None): bran=self.bran, ks=ks, cf=agent_cf, - tocks=keriTocks(agent_cf), + tocks=self._tockConfig.keri, temp=self.temp, salt=salt, ) @@ -398,7 +390,12 @@ def create(self, caid, salt=None): ) agent = Agent( - hby=agentHby, rgy=agentRgy, agentHab=agentHab, caid=caid, agency=self + hby=agentHby, + rgy=agentRgy, + agentHab=agentHab, + caid=caid, + agency=self, + tocks=self._tockConfig.signify, ) self.adb.agnt.pin(keys=(caid,), val=coring.Prefixer(qb64=agent.pre)) @@ -467,11 +464,13 @@ def get(self, caid): if aaid is None: return None - ks = keeping.Keeper(name=caid, base=self.base, temp=self.temp, reopen=True) - + # Agent configs are written with base="" by _writeAgentConfig, even + # when the Agent's databases use a nonempty base. cf = configing.Configer( - name=caid, base=self.base, temp=self.temp, reopen=True, clear=False + name=caid, base="", temp=self.temp, reopen=True, clear=False ) + resolvedTocks = scheduling.loadTocks(cf) + ks = keeping.Keeper(name=caid, base=self.base, temp=self.temp, reopen=True) agentHby = habbing.Habery( name=caid, base=self.base, @@ -479,7 +478,7 @@ def get(self, caid): ks=ks, temp=self.temp, cf=cf, - tocks=keriTocks(cf), + tocks=resolvedTocks.keri, ) agentHab = agentHby.habByName(f"agent-{caid}", ns="agent") @@ -492,7 +491,12 @@ def get(self, caid): hby=agentHby, name=agentHab.name, base=self.base, temp=self.temp ) agent = Agent( - hby=agentHby, rgy=agentRgy, agentHab=agentHab, agency=self, caid=caid + hby=agentHby, + rgy=agentRgy, + agentHab=agentHab, + agency=self, + caid=caid, + tocks=resolvedTocks.signify, ) self.agents[caid] = agent @@ -576,7 +580,7 @@ class Agent(doing.DoDoer): hierarchical deterministic key (HDK) management scheme used to select keys at the edge. """ - def __init__(self, hby, rgy, agentHab, agency, caid, **opts): + def __init__(self, hby, rgy, agentHab, agency, caid, tocks=None, **opts): """ Initialize the Agent with the given Habery, Regery, and agent's Hab. Parameters: @@ -594,7 +598,7 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): .agentHab (Hab): The Hab instance representing the agent itself. .rgy (Regery): The Regery instance for the agent's registry access. .cfd (MappingProxyType): Configuration data for the agent. - .tocks (MappingProxyType): Escrow timing configurations for the underlying Hio tasks comprising this agent. + .tocks (dict): Construction-time scheduler settings; changing these does not reconfigure existing doers. .last (datetime.datetime): Last activity timestamp for the agent. .shouldShutdown (bool): Flag indicating if the agent should shut down. .swain (delegating.Anchorer): Watches the delegator for delegation approval seals for inception and rotation. @@ -670,6 +674,7 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): submits (Deck): KEL messages to be resubmitted to witnesses to obtain receipts of. """ self.agency = agency + parentTock = opts.setdefault("tock", agency.tock) self.caid = caid self.hby = hby self.agentHab = agentHab @@ -677,11 +682,13 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): self.cfd = MappingProxyType( dict(self.hby.cf.get()) if self.hby.cf is not None else dict() ) - self.tocks = MappingProxyType(self.cfd.get("tocks", {})) + self.tocks = dict( + scheduling.loadTocks(hby.cf).signify if tocks is None else tocks + ) self.last = helping.nowUTC() self._shouldShutdown = False - self.swain = delegating.Anchorer(hby=hby, proxy=agentHab) + self.swain = delegating.Anchorer(hby=hby, proxy=agentHab, tock=parentTock) self.counselor = Counselor(hby=hby, swain=self.swain, proxy=agentHab) self.org = connecting.Organizer(hby=hby) @@ -712,7 +719,7 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): ) doers = [ - habbing.HaberyDoer(habery=hby), + habbing.HaberyDoer(habery=hby, tock=parentTock), receiptor, self.witq, self.witPub, @@ -724,6 +731,10 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): ] signaler = signaling.Signaler() + # Own expiry here: KERIpy's expireDo yields during deque iteration, + # allowing concurrent notification pushes to invalidate the iterator. + self.expirer = SignalExpirer(signaler=signaler, tock=self.tocks["signalExpiry"]) + doers.append(self.expirer) self.notifier = Notifier(hby=hby, signaler=signaler) self.mux = grouping.Multiplexor(hby=hby, notifier=self.notifier) @@ -766,7 +777,10 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): kdelegating.loadHandlers(hby=self.hby, exc=self.exc, notifier=self.notifier) protocoling.loadHandlers(hby=self.hby, exc=self.exc, notifier=self.notifier) self.submitter = Submitter( - hby=hby, submits=self.submits, witRec=self.witSubmitDoer + hby=hby, + submits=self.submits, + witRec=self.witSubmitDoer, + tock=self.tocks["submitter"], ) self.monitor = longrunning.Monitor( hby=hby, @@ -802,15 +816,13 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): doers.extend( [ - Initer( - agentHab=agentHab, caid=caid, tock=self.tocks.get("initer", 0.0) - ), + Initer(agentHab=agentHab, caid=caid, tock=self.tocks["initer"]), Querier( hby=hby, agentHab=agentHab, kvy=self.kvy, queries=self.queries, - tock=self.tocks.get("querier", 0.0), + tock=self.tocks["querier"], ), Escrower( kvy=self.kvy, @@ -821,28 +833,26 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): vry=self.verifier, registrar=self.registrar, credentialer=self.credentialer, - tock=self.tocks.get("escrower", 0.0), - ), - ParserDoer( - kvy=self.kvy, parser=self.parser, tock=self.tocks.get("parser", 0.0) + tock=self.tocks["escrower"], ), + ParserDoer(kvy=self.kvy, parser=self.parser, tock=self.tocks["parser"]), Witnesser( receiptor=receiptor, witners=self.witners, - tock=self.tocks.get("witnesser", 0.0), + tock=self.tocks["witnesser"], ), Delegator( agentHab=agentHab, swain=self.swain, anchors=self.anchors, - tock=self.tocks.get("delegator", 0.0), + tock=self.tocks["delegator"], ), ExchangeSender( hby=hby, agentHab=agentHab, exc=self.exc, exchanges=self.exchanges, - tock=self.tocks.get("exchangeSender", 0.0), + tock=self.tocks["exchangeSender"], ), Granter( hby=hby, @@ -850,7 +860,7 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): agentHab=agentHab, exc=self.exc, grants=self.grants, - tock=self.tocks.get("granter", 0.0), + tock=self.tocks["granter"], ), Admitter( hby=hby, @@ -859,25 +869,25 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts): agentHab=agentHab, exc=self.exc, admits=self.admits, - tock=self.tocks.get("admitter", 0.0), + tock=self.tocks["admitter"], ), GroupRequester( hby=hby, agentHab=agentHab, counselor=self.counselor, groups=self.groups, - tock=self.tocks.get("groupRequester", 0.0), + tock=self.tocks["groupRequester"], ), SeekerDoer( seeker=self.seeker, cues=self.verifier.cues, - tock=self.tocks.get("seeker", 0.0), + tock=self.tocks["seeker"], ), ExchangeCueDoer( seeker=self.exnseeker, cues=self.exc.cues, queries=self.queries, - tock=self.tocks.get("exchangecue", 0.0), + tock=self.tocks["exchangecue"], ), self.submitter, ] @@ -971,7 +981,7 @@ def createBootServerDoer(config: KERIAServerConfig, agency: Agency): ) if not bootServer.reopen(): raise RuntimeError(f"Cannot create boot HTTP server on port {config.bootPort}") - return http.ServerDoer(server=bootServer) + return http.ServerDoer(server=bootServer, tock=agency.tocks["bootServer"]) def createAdminServerDoer(config: KERIAServerConfig, agency: Agency): @@ -1008,7 +1018,9 @@ def createAdminServerDoer(config: KERIAServerConfig, agency: Agency): raise RuntimeError( f"cannot create admin HTTP server on port {config.adminPort}" ) - return adminApp, http.ServerDoer(server=adminServer) + return adminApp, http.ServerDoer( + server=adminServer, tock=agency.tocks["adminServer"] + ) def createHttpServerDoer( @@ -1035,7 +1047,7 @@ def createHttpServerDoer( ) if not server.reopen(): raise RuntimeError(f"cannot create local http server on port {config.httpPort}") - return http.ServerDoer(server=server) + return http.ServerDoer(server=server, tock=agency.tocks["httpServer"]) def createAgency(config: KERIAServerConfig, temp=False, cf=None): @@ -1096,8 +1108,12 @@ def recur(self, tyme=None, tock=0.0, **opts): logger.info( "Agent %s received:\n%s\n...\n", self.kvy, self.parser.ims[:1024] ) - done = yield from self.parser.parsator() # process messages continuously - return done # should never get here except forced close + parsator = self.parser.parsator() + try: + for _ in parsator: + yield self.tock + finally: + parsator.close() class Witnesser(doing.Doer): @@ -1122,9 +1138,11 @@ def recur(self, tyme=None, tock=0.0, **opts): if serder.ked["t"] in (Ilks.rot, Ilks.drt): adds = serder.ked["ba"] for wit in adds: - yield from self.receiptor.catchup(serder.pre, wit) + yield from self.receiptor.catchup( + serder.pre, wit, tock=self.tock + ) - yield from self.receiptor.receipt(serder.pre, serder.sn) + yield from self.receiptor.receipt(serder.pre, serder.sn, tock=self.tock) yield self.tock @@ -1190,7 +1208,7 @@ def recur(self, tyme, deeds=None): except kering.ValidationError: logger.info(f"unable to send to recipient={recp}") else: - doer = doing.DoDoer(doers=postman.deliver()) + doer = doing.DoDoer(doers=postman.deliver(), tock=self.tock) self.extend([doer]) return super(ExchangeSender, self).recur(tyme, deeds) @@ -1347,7 +1365,7 @@ def postGrant(self): except KeyError: logger.info(f"invalid grant message={serder.ked}") else: - doer = doing.DoDoer(doers=postman.deliver()) + doer = doing.DoDoer(doers=postman.deliver(), tock=self.tock) self.parent.extend([doer]) return True @@ -1531,18 +1549,26 @@ def recur(self, tyme, deeds=None): if "sn" in msg: sn = int(msg["sn"], 16) seqNoDo = querying.SeqNoQuerier( - hby=self.hby, hab=self.agentHab, pre=pre, sn=sn + hby=self.hby, hab=self.agentHab, pre=pre, sn=sn, tock=self.tock ) self.extend([seqNoDo]) elif "anchor" in msg: anchor = msg["anchor"] anchorDo = querying.AnchorQuerier( - hby=self.hby, hab=self.agentHab, pre=pre, anchor=anchor + hby=self.hby, + hab=self.agentHab, + pre=pre, + anchor=anchor, + tock=self.tock, ) self.extend([anchorDo]) else: qryDo = querying.QueryDoer( - hby=self.hby, hab=self.agentHab, pre=pre, kvy=self.kvy + hby=self.hby, + hab=self.agentHab, + pre=pre, + kvy=self.kvy, + tock=self.tock, ) self.extend([qryDo]) @@ -1590,15 +1616,33 @@ def recur(self, tyme, tock=0.0, **opts): return False +class SignalExpirer(doing.Doer): + """Expire controller signals without holding a deque iterator across yields.""" + + def __init__(self, signaler, tock=scheduling.DEFAULT_TOCK): + self.signaler = signaler + super().__init__(tock=tock) + + def recur(self, tyme=None): + now = helping.nowUTC() + for sig in tuple(self.signaler.signals): + if now - helping.fromIso8601(sig.dt) > self.signaler.SignalTimeout: + self.signaler.signals.remove(sig) + return False + + class Releaser(doing.Doer): - def __init__(self, agency: Agency, releaseTimeout=86400): + def __init__(self, agency: Agency, releaseTimeout=86400, tock=60.0): """Check open agents and close if idle for more than releaseTimeout seconds Parameters: agency (Agency): KERIA agent manager - releaseTimeout (int): Timeout in seconds + releaseTimeout (int | None): Timeout in seconds """ - self.tock = 60.0 + # Housekeeping scans all open Agents for a 24-hour inactivity timeout. + # The default minute of cleanup lag avoids scanning every scheduler tick; + # this interval does not delay active message processing. + self.tock = tock self.agents = agency.agents self.agency = agency self.releaseTimeout = releaseTimeout @@ -2237,7 +2281,7 @@ def on_post(req, rep): class Submitter(doing.DoDoer): - def __init__(self, hby, submits, witRec): + def __init__(self, hby, submits, witRec, tock=scheduling.DEFAULT_TOCK): """ Process to re-submit the last event from the KEL to the witnesses for receipts and to propogate it to each witness """ @@ -2245,7 +2289,7 @@ def __init__(self, hby, submits, witRec): self.submits = submits self.witRec = witRec - super(Submitter, self).__init__(always=True) + super(Submitter, self).__init__(always=True, tock=tock) def recur(self, tyme, deeds=None): """Processes submit reqests submitting any on the cue""" diff --git a/src/keria/app/delegating.py b/src/keria/app/delegating.py index 5b2dd8eb..cfb72fd5 100644 --- a/src/keria/app/delegating.py +++ b/src/keria/app/delegating.py @@ -23,6 +23,7 @@ from keri.core import coring, serdering from keri.db import dbing from keria.core import httping, longrunning +from keria.app import scheduling DELEGATION_ROUTE = "/identifiers/{name}/delegation" @@ -56,10 +57,16 @@ def __init__(self, hby, proxy=None, **kwa): self.postman = forwarding.Poster(hby=hby) self.witq = agenting.WitnessInquisitor(hby=hby) self.witDoer = agenting.Receiptor(hby=self.hby) + self.escrowDoer = doing.doify(self.escrowDo, tock=hby.tocks["anchorerEscrow"]) self.proxy = proxy super(Anchorer, self).__init__( - doers=[self.witq, self.witDoer, self.postman, doing.doify(self.escrowDo)], + doers=[ + self.witq, + self.witDoer, + self.postman, + self.escrowDoer, + ], **kwa, ) @@ -111,7 +118,7 @@ def complete(self, prefixer, seqner, saider=None): return True - def escrowDo(self, tymth, tock=1.0, temp=False, **opts): + def escrowDo(self, tymth, tock=scheduling.DEFAULT_TOCK, temp=False, **opts): """Process escrows of group multisig identifiers waiting to be compeleted. Steps involve: @@ -125,17 +132,16 @@ def escrowDo(self, tymth, tock=1.0, temp=False, **opts): Parameters: tymth (function): injected function wrapper closure returned by .tymen() of Tymist instance. Calling tymth() returns associated Tymist .tyme. - tock (float): injected initial tock value. Default to 1.0 to slow down processing + tock (float): injected cadence from the Habery anchoring configuration """ # enter context self.wind(tymth) - self.tock = tock - _ = yield self.tock + _ = yield tock while True: self.processEscrows() - yield 0.5 + yield tock def processEscrows(self): self.processPartialWitnessEscrow() diff --git a/src/keria/app/scheduling.py b/src/keria/app/scheduling.py new file mode 100644 index 00000000..6136737d --- /dev/null +++ b/src/keria/app/scheduling.py @@ -0,0 +1,182 @@ +"""KERIA scheduler configuration and compatibility with legacy flat keys.""" + +import logging +from collections.abc import Mapping +from dataclasses import dataclass + +from keri import kering +from keri.app import tocking +from keri.app.tocking import Tock, TockAlias + +logger = logging.getLogger(__name__) + +# One cycle of KERIA's 32 Hz root Doist. Zero remains an explicit next-cycle option. +DEFAULT_TOCK = 0.03125 + +# Apply KERIA's default tock policy without mutating KERIpy's process-global registry. +# Keep its keys, environment variables, aliases, and precedence rules intact. +KERI_TOCKS: dict[str, Tock] = { + key: Tock(definition.env, DEFAULT_TOCK) for key, definition in tocking.TOCKS.items() +} + +# tocks.signify default tock values mapped by config prop name to Tock env key and default +SIG_TOCKS: dict[str, Tock] = {} + +# Tock subkeys cascading tock to their target subkeys +SIG_TOCK_ALIASES: dict[str, TockAlias] = {} + +# Agent initialization and key-state queries (agenting.py - Initer, Querier) +KERIA_INITER_TOCK_KEY = "KERIA_INITER_TOCK" +KERIA_QUERIER_TOCK_KEY = "KERIA_QUERIER_TOCK" + +SIG_TOCKS["initer"] = Tock(KERIA_INITER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["querier"] = Tock(KERIA_QUERIER_TOCK_KEY, DEFAULT_TOCK) + +# Escrow and inbound message processing (agenting.py - Escrower, ParserDoer) +KERIA_ESCROWER_TOCK_KEY = "KERIA_ESCROWER_TOCK" +KERIA_PARSER_TOCK_KEY = "KERIA_PARSER_TOCK" + +SIG_TOCKS["escrower"] = Tock(KERIA_ESCROWER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["parser"] = Tock(KERIA_PARSER_TOCK_KEY, DEFAULT_TOCK) + +# Witness and delegation coordination (agenting.py - Witnesser, Delegator) +KERIA_WITNESSER_TOCK_KEY = "KERIA_WITNESSER_TOCK" +KERIA_DELEGATOR_TOCK_KEY = "KERIA_DELEGATOR_TOCK" + +SIG_TOCKS["witnesser"] = Tock(KERIA_WITNESSER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["delegator"] = Tock(KERIA_DELEGATOR_TOCK_KEY, DEFAULT_TOCK) + +# Exchange and credential workflows (agenting.py - ExchangeSender, Granter, Admitter) +KERIA_EXCHANGE_SENDER_TOCK_KEY = "KERIA_EXCHANGE_SENDER_TOCK" +KERIA_GRANTER_TOCK_KEY = "KERIA_GRANTER_TOCK" +KERIA_ADMITTER_TOCK_KEY = "KERIA_ADMITTER_TOCK" + +SIG_TOCKS["exchangeSender"] = Tock(KERIA_EXCHANGE_SENDER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["granter"] = Tock(KERIA_GRANTER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["admitter"] = Tock(KERIA_ADMITTER_TOCK_KEY, DEFAULT_TOCK) + +# Multisig group coordination (agenting.py - GroupRequester) +KERIA_GROUP_REQUESTER_TOCK_KEY = "KERIA_GROUP_REQUESTER_TOCK" + +SIG_TOCKS["groupRequester"] = Tock(KERIA_GROUP_REQUESTER_TOCK_KEY, DEFAULT_TOCK) + +# Credential and exchange indexing (agenting.py - SeekerDoer, ExchangeCueDoer) +KERIA_SEEKER_TOCK_KEY = "KERIA_SEEKER_TOCK" +KERIA_EXCHANGE_CUE_TOCK_KEY = "KERIA_EXCHANGE_CUE_TOCK" + +SIG_TOCKS["seeker"] = Tock(KERIA_SEEKER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["exchangecue"] = Tock(KERIA_EXCHANGE_CUE_TOCK_KEY, DEFAULT_TOCK) + +# Witness resubmission queue coordination +KERIA_SUBMITTER_TOCK_KEY = "KERIA_SUBMITTER_TOCK" +SIG_TOCKS["submitter"] = Tock(KERIA_SUBMITTER_TOCK_KEY, DEFAULT_TOCK) + +# Coarse Agent alias covering all Agent component tocks above +KERIA_AGENT_TOCK_KEY = "KERIA_AGENT_TOCK" +SIG_TOCK_ALIASES["agent"] = TockAlias( + env=KERIA_AGENT_TOCK_KEY, + targets=tuple(SIG_TOCKS), +) + + +# Server I/O, shutdown, and housekeeping are independent +# of the coarse Agent-work alias above. +KERIA_BOOT_SERVER_TOCK_KEY = "KERIA_BOOT_SERVER_TOCK" +KERIA_ADMIN_SERVER_TOCK_KEY = "KERIA_ADMIN_SERVER_TOCK" +KERIA_HTTP_SERVER_TOCK_KEY = "KERIA_HTTP_SERVER_TOCK" +KERIA_SHUTDOWN_TOCK_KEY = "KERIA_SHUTDOWN_TOCK" +KERIA_SIGNAL_EXPIRY_TOCK_KEY = "KERIA_SIGNAL_EXPIRY_TOCK" +KERIA_RELEASER_TOCK_KEY = "KERIA_RELEASER_TOCK" + +SIG_TOCKS["bootServer"] = Tock(KERIA_BOOT_SERVER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["adminServer"] = Tock(KERIA_ADMIN_SERVER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["httpServer"] = Tock(KERIA_HTTP_SERVER_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["shutdown"] = Tock(KERIA_SHUTDOWN_TOCK_KEY, DEFAULT_TOCK) +SIG_TOCKS["signalExpiry"] = Tock(KERIA_SIGNAL_EXPIRY_TOCK_KEY, DEFAULT_TOCK) +# Full scan of open Agents against a 24-hour inactivity timeout. +SIG_TOCKS["releaser"] = Tock(KERIA_RELEASER_TOCK_KEY, 60.0) + + +@dataclass(frozen=True) +class TockConfiguration: + """Effective scheduler tock cadences and canonical file values without overrides.""" + + # Resolved tocks for Doers/DoDoers/doified functions defined in KERIpy + # This includes environment variable overrides and defaults defined in this scheduling module. + keri: dict[str, float] + # Resolved tocks for KERIA-defined Doers/DoDoers/doified functions + # This includes environment variable overrides and defaults defined in this scheduling module. + signify: dict[str, float] + # This contains the Agent's config from the loaded config file + configured: dict + + +def resolveTocks(cfTocks=None, *, environ=None): + """Resolve one load, warning once if it contains legacy flat KERIA keys. + + Environment values override file values using KERIpy's precedence rules. + Canonical file values retain explicit settings only, never environment + overrides or defaults. Callers may persist them when provisioning new Agents; + resolving an existing file does not rewrite it. + """ + cfTocks = {} if cfTocks is None else cfTocks + if not isinstance(cfTocks, Mapping): + raise kering.ConfigurationError("tocks must be a mapping") + sigCfTocks = cfTocks.get("signify", {}) # signify tocks read from config file + if not isinstance(sigCfTocks, Mapping): + raise kering.ConfigurationError("tocks.signify must be a mapping") + + # The SIG_TOCKS config items are valid at the top level of "tocks" as legacy, + # backwards compatibility items. They are also valid inside the "tocks.signify" prop. + legacy = {key: value for key, value in cfTocks.items() if key in SIG_TOCKS} + if legacy: + logger.warning( + "deprecated flat KERIA tocks; move %s under tocks.signify. " + "Legacy placement will be removed in a future announced release.", + ", ".join(sorted(legacy)), + ) + + # Validate legacy values before merging can overwrite them. + tocking.resolveTocks( + legacy, + tocks=SIG_TOCKS, + aliases=SIG_TOCK_ALIASES, + environ={}, + reserved=list(), + ) + for key in legacy.keys() & sigCfTocks.keys(): + if legacy[key] != sigCfTocks[key]: + raise kering.ConfigurationError( + f"Conflicting values for tocks.{key} and tocks.signify.{key}" + ) + + # pull the legacy and new tocks.signify config into a consolidated config objet + sigTocks = {**legacy, **sigCfTocks} # tocks.signify overwrites legacy tocks + + # get the non-signify tocks config + configured = {key: value for key, value in cfTocks.items() if key not in SIG_TOCKS} + if legacy or "signify" in cfTocks: # only add signify tocks if they are present + configured["signify"] = sigTocks + keri = tocking.resolveTocks(configured, tocks=KERI_TOCKS, environ=environ) + signify = tocking.resolveTocks( + sigTocks, + tocks=SIG_TOCKS, + aliases=SIG_TOCK_ALIASES, + environ=environ, + reserved=list(), + ) + + return TockConfiguration( + keri=keri, + signify=signify, + configured=configured, + ) + + +def loadTocks(cf): + """Read one Configer, resolving values without changing its file.""" + config = cf.get() if cf is not None else {} + if "tocks" in config and config["tocks"] is None: + raise kering.ConfigurationError(f"{cf.path}: tocks must be a mapping") + cfTocks = config.get("tocks", {}) + return resolveTocks(cfTocks) diff --git a/src/keria/app/serving.py b/src/keria/app/serving.py index 429d1556..ab686f79 100644 --- a/src/keria/app/serving.py +++ b/src/keria/app/serving.py @@ -57,7 +57,7 @@ def recur(self, tyme=None, tock=0.0, **opts): """Generator coroutine checking once per tock for shutdown flag""" # Checks once per tock if the shutdown flag has been set and if so initiates the shutdown process while not self.shutdown_received: - yield tock # will iterate forever in here until shutdown flag set + yield self.tock # will iterate forever in here until shutdown flag set logger.info("Shutdown flag received, initiating graceful shutdown of agents") self.shutdownAgency() # Once shutdown_received is set, trigger agency shutdown which will eventually shut down diff --git a/src/keria/testing/testing_helper.py b/src/keria/testing/testing_helper.py index 03c021be..c08c5fc1 100644 --- a/src/keria/testing/testing_helper.py +++ b/src/keria/testing/testing_helper.py @@ -751,12 +751,13 @@ def openKeria(salter=None, cf=None, temp=True): ) agency = agenting.Agency(name="agency", bran=None, temp=True) + resolvedTocks = agenting.scheduling.loadTocks(cf) with habbing.openHby( name="keria", salt=salter.qb64, temp=temp, cf=cf, - tocks=agenting.keriTocks(cf), + tocks=resolvedTocks.keri, ) as hby: ims = eventing.messagize(serder, sigers=sigers) parsing.Parser(kvy=hby.kvy).parseOne(ims=ims) @@ -769,7 +770,12 @@ def openKeria(salter=None, cf=None, temp=True): hby=hby, name=agentHab.name, base=hby.base, temp=True ) agent = agenting.Agent( - hby=hby, rgy=rgy, agentHab=agentHab, agency=agency, caid=serder.pre + hby=hby, + rgy=rgy, + agentHab=agentHab, + agency=agency, + caid=serder.pre, + tocks=resolvedTocks.signify, ) agency.agents[serder.pre] = agent diff --git a/tests/app/test_agenting.py b/tests/app/test_agenting.py index 236b3e0c..aba10a4f 100644 --- a/tests/app/test_agenting.py +++ b/tests/app/test_agenting.py @@ -13,6 +13,8 @@ import signal import time from base64 import b64encode +from pathlib import Path +from contextlib import ExitStack, closing, contextmanager import falcon import hio @@ -32,7 +34,7 @@ from keri.help import nowIso8601 from keri.vdr import credentialing -from keria.app import agenting, aiding +from keria.app import agenting, aiding, scheduling from keria.core import longrunning, httping from keria.db import basing as agencybasing from keria.testing.testing_helper import SCRIPTS_DIR @@ -193,65 +195,179 @@ def test_load_ends(helpers): assert isinstance(end, agenting.ConfigResourceEnd) -def test_keri_tocks_preserves_agent_config(monkeypatch): - config = {"tocks": {"escrower": 1.0, "initer": 0.0, "receiptor": 0.25}} - with configing.openCF(temp=True) as cf: - cf.put(config) - monkeypatch.setenv("KERI_RECEIPTOR_TOCK", "0.125") - tocks = agenting.keriTocks(cf) - assert tocks["receiptor"] == 0.125 - assert "escrower" not in tocks - assert "initer" not in tocks - assert cf.get() == config - - cf.put({"tocks": {"escrower": 1.0, "receiptorTypo": 0.25}}) - with pytest.raises(kering.ConfigurationError, match="receiptorTypo"): - agenting.keriTocks(cf) - - -def test_agency_reopens_legacy_tocks(): - """Reload an evicted Agent from disk and retain its legacy tock settings.""" - caid = core.Signer().verfer.qb64 +@contextmanager +def openAgencyAndAgent(*, cf, caid, base=""): + """Own a persistent Agent and scheduler, preserving LMDB and config files for explicit reopen + steps inside the context and deleting test-owned state on exit. The caller owns cf. + """ with dbing.openLMDB(cls=agencybasing.AgencyBaser, temp=True) as adb: - cf = agenting.readConfigFile(SCRIPTS_DIR, "legacy-tocks") - agency = agenting.Agency(name="agency", bran=None, cf=cf, adb=adb) + agency = agenting.Agency(name="agency", base=base, bran=None, cf=cf, adb=adb) doist = doing.Doist() - doist.enter(doers=[agency]) - # Provision the Agent and write its own config from the legacy fixture. - agent = agency.create(caid) + agentOnCreate = None try: - assert agent.tocks["escrower"] == 1.0 - assert agent.hby.tocks["receiptor"] == 0.25 - persisted_config_path = agent.hby.cf.path - # Close the Agent's resources and evict it from the Agency cache, - # leaving its configuration and databases on disk for reopening. + doist.enter(doers=[agency]) + agentOnCreate = agency.create(caid) + yield agency, agentOnCreate + finally: + agentOnGet = agency.agents.get(caid) + # ExitStack runs in reverse: stop the Agent, exit the scheduler, + # then dispose of files. Later callbacks still run if one fails. + with ExitStack() as cleanup: + seen = set() + # Keep the original resources even if the test evicted the Agent. + # Include reopened handles too; identical objects need closing only once. + for agent in (agentOnCreate, agentOnGet): + if agent is None: + continue + for resource in ( + agent.seeker, + agent.exnseeker, + agent.monitor.opr, + agent.notifier.noter, + agent.rep.mbx, + agent.rgy.reger, + agent.mgr.rb, + agent.hby.ks, + agent.hby.db, + agent.hby.cf, + ): + if id(resource) not in seen: + seen.add(id(resource)) + cleanup.callback(resource.close, clear=True) + + cleanup.callback(doist.exit) + if agentOnGet is not None: + cleanup.callback(agency.shut, agentOnGet) + + +@pytest.mark.parametrize( + "legacy_agent_config", + [ + pytest.param(False, id="canonical-agent-config"), + pytest.param(True, id="legacy-agent-config"), + ], +) +def test_agency_reopens_legacy_tocks(tmp_path, monkeypatch, legacy_agent_config): + """Reopen an Agent from its saved default or legacy config, ignoring Agency + config changes. + """ + caid = core.Signer().verfer.qb64 # simulate an edge Signify controller AID + + fixture_path = os.path.join(SCRIPTS_DIR, "keri/cf/legacy-tocks.json") + # Give each run a private Agency template to change after Agent creation. + config_path = tmp_path / "keri/cf/legacy-tocks.json" + config_path.parent.mkdir(parents=True) + shutil.copyfile(fixture_path, config_path) + + # The Agency reads the private template; the helper owns the Agent's separate + # .json under /usr/local/var/keri/cf (or ~/.keri/cf on fallback). + agencyConfig = agenting.readConfigFile(str(tmp_path), "legacy-tocks") + with closing(agencyConfig) as cf: + # The environment override affects runtime values, not the saved Agent file. + monkeypatch.setenv("KERIA_ESCROWER_TOCK", "0.75") + with openAgencyAndAgent(cf=cf, caid=caid) as (agency, agent): + assert agent.tocks["escrower"] == 0.75 # env var overrides default. + assert agent.hby.tocks["receiptor"] == 0.25 # Copied from Agency config. + + # Persist the Agency's file values, not environment overrides. + configFileVals = { + "receiptor": 0.25, + "signify": {"initer": 0.0, "escrower": 1.0}, + } + assert agent.hby.cf.get()["tocks"] == configFileVals + + # Close resources and evict the cached Agent, keeping its files for reopen. agency.shut(agent) assert caid not in agency.agents - # This cache miss is the reopen: Agency.get constructs a new Agent - # from its persisted config and databases, rather than returning - # the previously cached instance or provisioning it again. + # Keep the canonical Agent config unless testing legacy flat tock keys. + # Both config file values use the same file path. + persisted_config_path = Path(agent.hby.cf.path) + if legacy_agent_config: + shutil.copyfile(fixture_path, persisted_config_path) + persisted_bytes = persisted_config_path.read_bytes() + + # Change only the Agency template at config_path; the Agent file is untouched. + cf.put({"tocks": {"signify": {"escrower": 0.125}}}) + monkeypatch.delenv("KERIA_ESCROWER_TOCK") # Reopen using file values. + # This cache miss reloads the Agent's saved config, not the Agency template. agent = agency.get(caid) - assert agent.hby.cf.path == persisted_config_path - assert agent.hby.cf.get()["tocks"] == cf.get()["tocks"] - assert agent.tocks["escrower"] == 1.0 - assert agent.hby.tocks["receiptor"] == 0.25 - finally: - agency.shut(agent) - doist.exit() - for resource in ( - agent.seeker, - agent.exnseeker, - agent.monitor.opr, - agent.notifier.noter, - agent.rep.mbx, - agent.rgy.reger, - agent.mgr.rb, - agent.hby, - agent.hby.cf, + assert agent.tocks["escrower"] == 1.0 # Saved value, not the override. + assert agent.hby.tocks["receiptor"] == 0.25 # must match config file + # Loading legacy settings must not silently migrate the file on disk. + assert persisted_config_path.read_bytes() == persisted_bytes + + +def test_agency_reopens_config_with_database_base(): + """ + Creating and reopening an Agent use the same config outside its database base. + Tests that agency.create(caid) and agency.get(caid) use the same config file path. + """ + caid = core.Signer().verfer.qb64 # Simulate Signify controller edge AID. + with configing.openCF(temp=True) as cf: + cf.put({"tocks": {"signify": {"escrower": 0.375}}}) + with openAgencyAndAgent(cf=cf, caid=caid, base=f"config-path-{caid}") as ( + agency, + agent, + ): + # Databases use the namespace, but create writes cf/.json without it. + assert agent.hby.base == agency.base + assert agent.hby.cf.base == "" + config_path = agent.hby.cf.path + agency.shut(agent) # Evict the Agent so get must reopen its files. + assert caid not in agency.agents + + agent = agency.get(caid) + # get must find the file written by create, not cf//.json. + assert agent.hby.cf.path == config_path + assert agent.tocks["escrower"] == 0.375 + + +def test_agent_tock_bindings_are_isolated(helpers): + """Construct two Agents with distinct per-worker tock cadences to catch miswired + settings or configuration shared between Agents. Checks constructor bindings + and parent tock cadence inheritance as well. + """ + bindings = { + "initer": agenting.Initer, + "querier": agenting.Querier, + "escrower": agenting.Escrower, + "parser": agenting.ParserDoer, + "witnesser": agenting.Witnesser, + "delegator": agenting.Delegator, + "exchangeSender": agenting.ExchangeSender, + "granter": agenting.Granter, + "admitter": agenting.Admitter, + "groupRequester": agenting.GroupRequester, + "seeker": agenting.SeekerDoer, + "exchangecue": agenting.ExchangeCueDoer, + "submitter": agenting.Submitter, + } + # Distinct worker and Agent values expose crossed bindings or shared configuration. + first_worker_tocks = {key: (index + 1) / 10 for index, key in enumerate(bindings)} + second_worker_tocks = {key: value + 1 for key, value in first_worker_tocks.items()} + with configing.openCF(temp=True) as cf1, configing.openCF(temp=True) as cf2: + cf1.put({"tocks": {"signify": first_worker_tocks}}) + cf2.put({"tocks": {"signify": second_worker_tocks}}) + # Construct Agents directly with separate configs; Agency provisioning is not tested here. + with ( + helpers.openKeria(cf=cf1) as (_, agent1, _, _), + helpers.openKeria(cf=cf2) as (_, agent2, _, _), + ): + for agent, expected_worker_tocks in ( + (agent1, first_worker_tocks), + (agent2, second_worker_tocks), ): - resource.close(clear=True) - cf.close() + # Containers inherit their parent tock; worker overrides stay local. + assert agent.tock == agent.agency.tock + assert agent.swain.tock == agent.tock + for key, cls in bindings.items(): + doer = next(doer for doer in agent.doers if isinstance(doer, cls)) + assert doer.tock == expected_worker_tocks[key] + # Worker overrides replace defaults; other settings retain their defaults. + expected_tocks = scheduling.resolveTocks(environ={}).signify + expected_tocks.update(expected_worker_tocks) + assert agent.tocks == expected_tocks def test_load_tocks_config(helpers): @@ -276,31 +392,14 @@ def test_load_tocks_config(helpers): "iurls": [ "http://127.0.0.1:5642/oobi/BBilc4-L3tFUnfM_wJr4S4OJanAv_VmF_dJNN6vkf2Ha/controller&tag=witness" ], - "tocks": {"initer": 0.0, "escrower": 1.0}, + "tocks": {"signify": {"initer": 0.0, "escrower": 1.0}}, } - assert agent.tocks == {"initer": 0.0, "escrower": 1.0} - - escrower_doer = next( - (doer for doer in agent.doers if isinstance(doer, agenting.Escrower)), None - ) - assert escrower_doer is not None - assert escrower_doer.tock == 1.0 - - initer_doer = next( - (doer for doer in agent.doers if isinstance(doer, agenting.Initer)), None - ) - assert initer_doer is not None - assert initer_doer.tock == 0.0 - - querier_doer = next( - (doer for doer in agent.doers if isinstance(doer, agenting.Querier)), None - ) - assert querier_doer is not None - assert querier_doer.tock == 0.0 - - with pytest.raises(TypeError): - agent.tocks["initer"] = 1.0 # agent.tocks is read-only + assert agent.tocks == { + **scheduling.resolveTocks(environ={}).signify, + "initer": 0.0, + "escrower": 1.0, + } def test_agency(): @@ -903,6 +1002,7 @@ def test_querier(helpers): agentHab=agent.agentHab, queries=decking.Deck(), kvy=agent.kvy, + tock=0.125, ) doist = doing.Doist(limit=1.0, tock=0.03125, real=True) deeds = doist.enter(doers=[qry]) @@ -910,34 +1010,37 @@ def test_querier(helpers): qry.queries.append( dict(pre="EI7AkI40M11MS7lkTCb10JC9-nDt-tXwQh44OHAFlv_9", sn="1") ) - qry.recur(1.0, deeds=deeds) + qry.recur(1.0, deeds=deeds) # Pass 1: dispatch the sequence-number query. assert len(qry.doers) == 1 seqNoDoer = qry.doers[0] assert isinstance(seqNoDoer, querying.SeqNoQuerier) is True + assert seqNoDoer.tock == qry.tock assert seqNoDoer.pre == "EI7AkI40M11MS7lkTCb10JC9-nDt-tXwQh44OHAFlv_9" assert seqNoDoer.sn == 1 qry.doers.remove(seqNoDoer) - # Anchor not implemented yet + # Exercise anchor dispatch and cadence binding without waiting for resolution. qry.queries.append( dict(pre="EI7AkI40M11MS7lkTCb10JC9-nDt-tXwQh44OHAFlv_9", anchor={}) ) - qry.recur(1.0, deeds=deeds) + qry.recur(1.0, deeds=deeds) # Pass 2: dispatch the anchor query. assert len(qry.doers) == 1 anchorDoer = qry.doers[0] assert isinstance(anchorDoer, querying.AnchorQuerier) is True + assert anchorDoer.tock == qry.tock assert anchorDoer.pre == "EI7AkI40M11MS7lkTCb10JC9-nDt-tXwQh44OHAFlv_9" assert anchorDoer.anchor == {} qry.doers.remove(anchorDoer) qry.queries.append(dict(pre="EI7AkI40M11MS7lkTCb10JC9-nDt-tXwQh44OHAFlv_9")) - qry.recur(1.0, deeds=deeds) + qry.recur(1.0, deeds=deeds) # Pass 3: dispatch the prefix-only query. assert len(qry.doers) == 1 qryDoer = qry.doers[0] assert isinstance(qryDoer, querying.QueryDoer) is True + assert qryDoer.tock == qry.tock assert qryDoer.pre == "EI7AkI40M11MS7lkTCb10JC9-nDt-tXwQh44OHAFlv_9" diff --git a/tests/app/test_delegating.py b/tests/app/test_delegating.py index 146ec769..ab4e8ffa 100644 --- a/tests/app/test_delegating.py +++ b/tests/app/test_delegating.py @@ -9,6 +9,7 @@ import json import time import pytest +from contextlib import closing from hio.base import doing from keri import kering @@ -16,12 +17,40 @@ from keri.app.delegating import DelegateRequestHandler from keri.core import coring, eventing, parsing -from keria.app import aiding, delegating, notifying +from keria.app import aiding, delegating, notifying, scheduling from keria.core import longrunning from keria.end import ending from keria.app import agenting +@pytest.mark.parametrize( + "cadence", + [ + pytest.param(0.0, id="next-cycle"), + pytest.param(0.125, id="positive-delay"), + ], +) +def test_anchorer_escrow_cadence(cadence): + """The doified escrow worker must use its configured cadence on initial and + recurring yields without changing the Anchorer container's cadence. + """ + tocks = scheduling.resolveTocks({"anchorerEscrow": cadence}, environ={}).keri + with habbing.openHby(name="cadence", temp=True, tocks=tocks) as hby: + parent_tock = 0.25 + anchorer = delegating.Anchorer(hby=hby, tock=parent_tock) + escrow = anchorer.escrowDoer + assert escrow in anchorer.doers + assert escrow.tock == cadence + dog = escrow(tymth=doing.Doist().tymen(), tock=escrow.tock) + with closing(dog): + # Step 1: bind the clock; yield before escrow work. + assert next(dog) == cadence + # Step 2: process escrows; yield the same cadence. + assert next(dog) == cadence + # The escrow override must not retime the container's other children. + assert anchorer.tock == parent_tock + + def test_sealer(): with habbing.openHby(name="p1", temp=True) as hby: # Create Anchorer to test diff --git a/tests/app/test_scheduling.py b/tests/app/test_scheduling.py new file mode 100644 index 00000000..ac993878 --- /dev/null +++ b/tests/app/test_scheduling.py @@ -0,0 +1,198 @@ +"""Configuration policy tests using KERIpy's published resolver.""" + +from copy import deepcopy +import json +from pathlib import Path + +import pytest +from keri import kering +from keri.app import configing + +from keria.app import scheduling + + +def test_defaults_and_environment_coverage(): + """Verify KERIA's one-tick defaults, slower cleanup exception, and a working + environment override for every KERIA tock setting. + """ + defaults = scheduling.resolveTocks(environ={}) + assert len(defaults.signify) == 19 + assert defaults.signify["releaser"] == 60.0 + assert {v for k, v in defaults.signify.items() if k != "releaser"} == {0.03125} + assert set(defaults.keri.values()) == {0.03125} + # Generate unique tock values per SIG_TOCKS entry to expose environment variables accidentally bound to the wrong key. + expected = {key: (index + 1) / 10 for index, key in enumerate(scheduling.SIG_TOCKS)} + environ = { + scheduling.SIG_TOCKS[key].env: str(value) for key, value in expected.items() + } + assert scheduling.resolveTocks(environ=environ).signify == expected + + +@pytest.mark.parametrize("name", ["scripts/keria.json", "scripts/keri/cf/keria.json"]) +def test_sample_configs_match_tock_defaults(name): + """Keep both sample configs complete and aligned with all registered tock defaults.""" + root = Path(__file__).resolve().parents[2] + defaults = scheduling.resolveTocks(environ={}) + configured = json.loads((root / name).read_text())["tocks"] + assert configured == {**defaults.keri, "signify": defaults.signify} + + +def test_keria_resolver_preserves_core_overrides(): + """KERIA's resolver preserves KERIpy alias and environment overrides when + applying its own core defaults; these settings stay outside tocks.signify. + """ + resolved = scheduling.resolveTocks( + {"vdrEscrow": 0.25}, environ={"KERI_CREDENTIALER_ESCROW_TOCK": "0.0"} + ) + assert resolved.keri["registrarEscrow"] == 0.25 + assert resolved.keri["credentialerEscrow"] == 0.0 + + +@pytest.mark.parametrize( + "environ, expected_escrower, expected_initer", + [ + ({}, 0.2, 0.1), + ({"KERIA_AGENT_TOCK": "0.3"}, 0.3, 0.3), + ({"KERIA_AGENT_TOCK": "0.3", "KERIA_ESCROWER_TOCK": "0.4"}, 0.4, 0.3), + ], +) +def test_specific_and_coarse_precedence(environ, expected_escrower, expected_initer): + """Specific file values beat group file values; group environment values beat + file values, and specific environment values take highest precedence. + """ + resolved = scheduling.resolveTocks( + {"signify": {"agent": 0.1, "escrower": 0.2}}, environ=environ + ) + assert resolved.signify["escrower"] == expected_escrower + # An untargeted worker still follows the group value at the winning layer. + assert resolved.signify["initer"] == expected_initer + + +def test_legacy_normalization_is_lossless_and_warns_per_load(caplog, monkeypatch): + """Normalize matching legacy/canonical settings without mutating the input or + persisting environment overrides; warn on each legacy load, not canonical loads. + """ + # Capture directly, without also forwarding the same record to pytest's root handler. + monkeypatch.setattr(scheduling.logger, "propagate", False) + scheduling.logger.addHandler(caplog.handler) + try: + raw = {"receiptor": 0.25, "escrower": 1.0, "signify": {"escrower": 1.0}} + original = deepcopy(raw) + resolved = scheduling.resolveTocks( + raw, + environ={"KERIA_ESCROWER_TOCK": "0.5", "KERI_RECEIPTOR_TOCK": "0.125"}, + ) + assert raw == original + assert resolved.signify["escrower"] == 0.5 + assert resolved.keri["receiptor"] == 0.125 + # Legacy KERIA keys must be removed before passing the core map to Habery. + assert not resolved.keri.keys() & scheduling.SIG_TOCKS.keys() # should be empty + assert resolved.configured == {"receiptor": 0.25, "signify": {"escrower": 1.0}} + warning = caplog.text + assert warning.count("deprecated flat KERIA tocks") == 1 + assert "escrower under tocks.signify" in warning + # A different config load must receive its own warning; no process-global latch. + scheduling.resolveTocks(raw, environ={}) + assert len(caplog.records) == 2 + caplog.clear() + # Reload the normalized form: no legacy placement remains to warn about. + scheduling.resolveTocks(resolved.configured, environ={}) + assert not caplog.records + finally: + scheduling.logger.removeHandler(caplog.handler) + + +def test_conflicting_duplicates_fail_even_with_environment_override(): + """Reject contradictory legacy and canonical values even when an environment + override would otherwise hide the ambiguous file configuration. + """ + with pytest.raises(kering.ConfigurationError, match="Conflicting.*escrower"): + scheduling.resolveTocks( + {"escrower": 1.0, "signify": {"escrower": 0.5}}, + environ={"KERIA_ESCROWER_TOCK": "0.25"}, + ) + + +@pytest.mark.parametrize( + "raw, environ, error", + [ + # True == 1.0: merging must not hide the invalid legacy boolean. + pytest.param( + {"escrower": True, "signify": {"escrower": 1.0}}, + {"KERIA_ESCROWER_TOCK": "0.5"}, + "escrower", + id="legacy-boolean", + ), + pytest.param( + {"signify": {"escrower": -1}}, + {"KERIA_ESCROWER_TOCK": "0.5"}, + "escrower", + id="negative", + ), + pytest.param([], {}, "tocks must be a mapping", id="root-list"), + pytest.param( + {"signify": []}, {}, r"tocks\.signify must be a mapping", id="signify-list" + ), + pytest.param( + {"signify": None}, + {}, + r"tocks\.signify must be a mapping", + id="signify-null", + ), + pytest.param( + {"escrowre": 1.0}, + {}, + r"Unknown tock configuration key\(s\): escrowre", + id="unknown-core-key", + ), + pytest.param( + {"signify": {"escrowre": 1.0}}, + {}, + r"Unknown tock configuration key\(s\): escrowre", + id="unknown-signify-key", + ), + pytest.param( + {}, + {"KERIA_AGENT_TOCK": "bad"}, + "KERIA_AGENT_TOCK", + id="invalid-environment", + ), + ], +) +def test_invalid_tock_configuration(raw, environ, error): + """Reject malformed sections, unknown keys, and invalid file/environment values; + valid overrides must not conceal invalid file values. Errors identify the setting. + """ + with pytest.raises(kering.ConfigurationError, match=error): + scheduling.resolveTocks(raw, environ=environ) + + +def test_explicit_null_file_block_is_invalid(): + """Treat an explicit null tocks block as invalid, not as omitted configuration.""" + with configing.openCF(temp=True) as cf: + cf.put({"tocks": None}) + with pytest.raises(kering.ConfigurationError, match="tocks must be a mapping"): + scheduling.loadTocks(cf) + + +def test_missing_tocks_block_uses_defaults(): + """Accept absent configuration without inventing explicit settings or rewriting + an existing file that contains no tocks block. + """ + assert scheduling.loadTocks(None).configured == {} + with configing.openCF(temp=True) as cf: + original = {"dt": "2026-09-16T00:00:00.000000+00:00"} + cf.put(original) + assert scheduling.loadTocks(cf).configured == {} + assert cf.get() == original + + +def test_agent_alias_excludes_server_and_housekeeping_cadences(): + """The Agent group override includes submission work but must not retime + HTTP servers, shutdown polling, signal expiry, or inactive-Agent cleanup. + """ + values = scheduling.resolveTocks(environ={"KERIA_AGENT_TOCK": "0.0"}).signify + assert values["submitter"] == 0.0 + assert values["releaser"] == 60.0 + for key in ("bootServer", "adminServer", "httpServer", "shutdown", "signalExpiry"): + assert values[key] == scheduling.DEFAULT_TOCK diff --git a/tests/app/test_scheduling_runtime.py b/tests/app/test_scheduling_runtime.py new file mode 100644 index 00000000..eab46c0f --- /dev/null +++ b/tests/app/test_scheduling_runtime.py @@ -0,0 +1,165 @@ +"""Scheduler-facing yields and ownership, beyond configuration properties.""" + +import pytest +from contextlib import closing +from hio.base import doing +from hio.help import decking +from keri.app import configing +from keri.core import coring, eventing, parsing, signing + +from keria.app import agenting +from keria.app.serving import GracefulShutdownDoer + + +@pytest.mark.parametrize( + "cadence", + [ + pytest.param(0.0, id="next-cycle"), + pytest.param(0.25, id="positive-delay"), + ], +) +def test_parser_yields(cadence): + """ParserDoer must yield its configured cadence while waiting for either an + initial frame or missing bytes, including an explicit zero cadence. + """ + parser = parsing.Parser() + doer = agenting.ParserDoer(kvy=None, parser=parser, tock=cadence) + dog = doer.recur() + with closing(dog): + assert next(dog) == cadence # Step 1: wait for input on an empty stream. + parser.ims.extend(b"-") # Supply only the start of a CESR frame. + assert next(dog) == cadence # Step 2: wait for the rest of the frame. + + +def test_witnesser_threads_tock_through_receiptor_child_ops(): + """ + Proves that the Witnesser passes its parent tock to its child tasks/Doers. + + A rotation adding a witness must pass Witnesser's tock through catch-up + and receipt collection, then use that tock for its own outer-loop yield. + """ + key = signing.Signer(raw=b"a" * 32).verfer.qb64 + witness = signing.Signer(raw=b"b" * 32, transferable=False).verfer.qb64 + inception = eventing.incept(keys=[key], ndigs=[coring.Diger(ser=key.encode()).qb64]) + # Adding a witness on rotation requires catch-up before receipt collection. + event = eventing.rotate( + pre=inception.pre, keys=[key], dig=inception.said, adds=[witness], toad=1 + ) + calls = [] + + class Receiptor: + def catchup(self, pre, wit, tock=0.0): + calls.append(("catchup", pre, wit)) + yield tock + + def receipt(self, pre, sn, tock=0.0): + calls.append(("receipt", pre, sn)) + yield tock + + doer = agenting.Witnesser( + receiptor=Receiptor(), witners=decking.Deck([{"serder": event}]), tock=0.25 + ) + dog = doer.recur() + with closing(dog): + assert next(dog) == 0.25 # Step 1: dequeue rotation; catch-up yields. + assert next(dog) == 0.25 # Step 2: catch-up finishes; receipt yields. + assert next(dog) == 0.25 # Step 3: receipt finishes; outer loop yields. + # Matching yields alone could pass if a child operation were skipped. + assert calls == [ + ("catchup", event.pre, witness), + ("receipt", event.pre, event.sn), + ] + + +def test_signaler_is_scheduled_and_expires_signals(helpers): + """The Agent must schedule expiry for its notifier's actual signal queue at + the configured cadence, removing expired signals while retaining fresh ones. + """ + with configing.openCF(temp=True) as cf: + cf.put({"tocks": {"signify": {"signalExpiry": 0.125}}}) + with helpers.openKeria(cf=cf) as (_, agent, _, _): + signaler = agent.notifier.signaler + expirer = agent.expirer + assert expirer in agent.doers # The Agent must schedule its expiry worker. + # Expiry must service the notifier's live queue, not a separate Signaler. + assert expirer.signaler is signaler + assert expirer.tock == 0.125 + signaler.push(attrs={}, topic="test", dt="2000-01-01T00:00:00.000000+00:00") + doist = doing.Doist(tock=0.03125) + # Prime expiry; recur has not scanned yet. + deeds = doist.enter(doers=[expirer]) + try: + # Cycles 1-6 visit t=0 through 0.15625; expiry scans at 0 and 0.125. + for _ in range(6): + doist.recur(deeds) # Advance one 0.03125-second scheduler cycle. + assert not signaler.signals + # Add a fresh, non-expiring signal between scans to exercise queue reuse after removal. + signaler.push(attrs={}, topic="live") + # Cycles 7-12 visit t=0.1875 through 0.34375; expiry scans at 0.25. + for _ in range(6): + doist.recur(deeds) # Advance one cycle; retain the fresh signal. + assert len(signaler.signals) == 1 + assert signaler.signals[0].topic == "live" + finally: + doist.exit(deeds) + + +def test_agency_service_bindings(): + """Bind distinct config values to all three HTTP servers, shutdown, and cleanup; + verify shutdown and cleanup generators actually yield their assigned cadences. + """ + configured = { + "bootServer": 0.125, + "adminServer": 0.25, + "httpServer": 0.375, + "shutdown": 0.5, + "releaser": 30.0, + } + with configing.openCF(temp=True) as cf: + cf.put({"tocks": {"signify": configured}}) + agency = agenting.Agency(name="cadences", bran=None, temp=True, cf=cf) + config = agenting.KERIAServerConfig( + bootPort=23903, adminPort=23901, httpPort=23902 + ) + servers = [] + try: + boot_server = agenting.createBootServerDoer(config, agency) + servers.append(boot_server) + app, admin_server = agenting.createAdminServerDoer(config, agency) + servers.append(admin_server) + public_server = agenting.createHttpServerDoer(config, agency, app) + servers.append(public_server) + assert boot_server.tock == 0.125 + assert admin_server.tock == 0.25 + assert public_server.tock == 0.375 + doist = agenting.agencyDoist([agency, *servers]) + shutdown = next( + d for d in doist.doers if isinstance(d, GracefulShutdownDoer) + ) + assert shutdown.tock == 0.5 + shutdown_dog = shutdown.recur() + with closing(shutdown_dog): + # Before a request: poll at the configured cadence; keep Agency running. + assert next(shutdown_dog) == configured["shutdown"] # First poll. + assert not agency.shouldShutdown + + # Simulate the flag set by a signal handler; no OS signal is sent. + shutdown.shutdown_received = True + + # Next poll: request Agency shutdown and complete the generator. + with pytest.raises(StopIteration): + next(shutdown_dog) + assert agency.shouldShutdown + releaser = agency.releaser + releaser_dog = releaser.recur() + with closing(releaser_dog): + assert ( + next(releaser_dog) == 30.0 + ) # Step 1: scan the empty cache, then yield. + assert ( + next(releaser_dog) == 30.0 + ) # Step 2: repeat the scan at the same cadence. + finally: + for server in servers: + server.server.close() + agency.adb.close(clear=True) diff --git a/tests/scripts/keri/cf/main/keria.json b/tests/scripts/keri/cf/main/keria.json index 3eaf2db0..62ccf0ed 100755 --- a/tests/scripts/keri/cf/main/keria.json +++ b/tests/scripts/keri/cf/main/keria.json @@ -16,7 +16,9 @@ "http://127.0.0.1:5642/oobi/BBilc4-L3tFUnfM_wJr4S4OJanAv_VmF_dJNN6vkf2Ha/controller&tag=witness" ], "tocks": { - "initer": 0.0, - "escrower": 1.0 + "signify": { + "initer": 0.0, + "escrower": 1.0 + } } } \ No newline at end of file