"""
The lockfile
============
Which part answered which query, recorded so a later run gives the same board.
:py:class:`Lock` is what a caller touches: it says where the file is and what to
do with it. The modes are :py:class:`LockMode`: `read-write` (the default --
replay what is there, resolve and record the rest), `read` (replay, and resolve a
miss loudly), `check` (replay, re-resolve, and report differences), `replay`
(hermetic -- replay and adopt, and never ask the chain to resolve) and `off`.
Under the three pinned modes a miss does not raise where it happens: it is
recorded and the run continues, and the failure arrives through
`parts_verdict()`'s exit status at the end. A project sets the file and the mode
under ``[tool.jitx.parts]``::
[tool.jitx.parts]
lock = "parts.lock.json"
lock-mode = "read-write"
An unset `lock-mode` is the default; one that does not name a mode logs a warning
and falls back to `read`, so a typo cannot silently re-pin the design.
`Lock` is a context, so anything wrapping a run -- a driver, a test harness, a
CLI verb if one is ever built -- selects a mode by entering one, and a `Design`
that declares its own overrides the project default. Nothing has to be called by
hand: an ordinary run pins at the end of elaboration.
A lock exists to stop *external* change from moving a design, not internal change.
That principle is what the keying follows from: an entry is keyed on the query
digest, the active chain, and a user-supplied `lock_scope`, so editing the chain
re-resolves (internal, deliberate) while a catalogue gaining rows does not
(external, the world moving on).
A digest miss is recoverable, which is why editing one query does not churn the
parts nobody touched. Whether a locked part still satisfies an edited query is
checkable locally (the entry stores its attributes and the core has `match`)
so at the final elaboration flush orphaned entries are matched against unmatched
queries by the tiered sweep in :py:func:`adopt_orphans`.
The rest of the module is the file itself. :py:class:`Lockfile` reads and writes
it; :py:func:`encode_payload` and :py:func:`decode_payload` are the provider-facing
helpers for getting quantities through JSON and back.
"""
from __future__ import annotations
import json
import os
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass, field, replace
from enum import Enum, StrEnum
from pathlib import Path
from uuid import uuid4
from typing import TYPE_CHECKING, Any, cast, overload
from logging import getLogger
import jitx.context
from jitx.units import registry
from .error import DimensionMismatch, LockStale, ProviderContractError
from .query import (
Exactly,
Exists,
OneOf,
Predicate,
Range,
Rank,
Scalar,
Tolerance,
is_quantity,
match,
)
from .record import JSON, PartIdentity
if TYPE_CHECKING:
from jitx.project import JitxProject
logger = getLogger(__name__)
SCHEMA_VERSION = 1
# Bumped when the *digest* rules change: the hash payload, the canonical form, the
# set of fields that participate, or the pint version whose spelling reaches it.
# A mismatch means nothing in the file can be matched, which is the one lock
# failure that is genuinely total.
DIGEST_VERSION = 1
[docs]
class LockMode(StrEnum):
"""What a run does with the lockfile.
A `StrEnum`, so the member is the config spelling: `LockMode("read-write")`
parses what `[tool.jitx.parts] lock-mode` holds, and a design that spells a
mode as a plain string still compares equal.
"""
READ_WRITE = "read-write"
"""Replay what is there, resolve and record the rest. The default."""
READ = "read"
"""Replay only. A miss still resolves, so the board is complete and the report
is, but it is recorded and fails the run at the end."""
CHECK = "check"
"""Run normally, then resolve the whole design cleanly and report where the
board this run built differs from what the clean run would build. Every bound
part is compared, adoptions included. The run's own part is always the one
that gets built; this mode only reports."""
REPLAY = "replay"
"""Hermetic: the chain is never asked to resolve. A miss is recorded as
`unresolvable` and its component is left unbound. Replay and adoption still
work -- both reach their answer from the lockfile and build it with
`rehydrate`, which is local."""
OFF = "off"
"""Ignore the lock entirely."""
PINNED_MODES = frozenset({LockMode.READ, LockMode.CHECK, LockMode.REPLAY})
"""Modes where a miss is a failure. `off` and `read-write` grow a design; these
three defend one."""
# --------------------------------------------------------------------------
# The Lock context
# --------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class Lock(jitx.context.Context):
"""Which lockfile to use, and how.
A context, so anything wrapping a run selects a mode by entering one, and a
`Design` that declares its own `lock` overrides the project default like any
other context.
`read-write` is the default: running a design twice with no changes should give
the same board, and the only way to get that is for the first run to pin.
"""
path: str | None = None
mode: LockMode = LockMode.READ_WRITE
@property
def reads(self) -> bool:
"""Whether this mode consults the lock at all."""
return self.mode != LockMode.OFF
@property
def writes(self) -> bool:
"""Whether this mode may write. Only `read-write` does."""
return self.mode == LockMode.READ_WRITE
@property
def pinned(self) -> bool:
"""Whether a miss is a failure. True for `read`, `check` and `replay`."""
return self.mode in PINNED_MODES
@property
def resolves(self) -> bool:
"""Whether a miss may be sent to the chain. False only for `replay`.
`off` resolves everything and consults no lock; `replay` is its mirror
image, consulting only the lock and asking the chain nothing.
"""
return self.mode != LockMode.REPLAY
[docs]
@classmethod
def get(cls) -> Lock:
"""The active lock, or the project's default, never None."""
# The default is a shared instance, never constructed per call: a context
# object's identity is load-bearing.
from .policy import current_project
# `super().get()` raises outside an active instantiation structure, and
# that is correct: a context has no meaning there, so asking for one is
# the error. What this override supplies is the *default within* a run,
# which is what stops `require()` failing merely because no `with` block
# is active.
current = super().get()
lock = current if current is not None else discovered_lock(current_project())
return _forced(lock)
_forced_mode: LockMode | None = None
_forced_locks: dict[Lock, Lock] = {}
[docs]
def force_mode(mode: LockMode | None) -> None:
"""Run every lock in the process under `mode`, or stop doing so with None.
For a driver that must select a mode around a design it cannot wrap in a
`Lock` context, such as an exporter. Each lock keeps its path and only its mode
is replaced, and the override applies to a design that declares its own `Lock`
as well.
"""
# Process-level rather than a context or thread-local: the jitx runtime
# elaborates a submitted design on its own bridge thread.
global _forced_mode
_forced_mode = mode
_forced_locks.clear()
def _forced(lock: Lock) -> Lock:
if _forced_mode is None or lock.mode == _forced_mode:
return lock
# Cached so one lock maps to one forced lock, for the same identity reason
# `discovered_lock` caches.
if lock not in _forced_locks:
_forced_locks[lock] = replace(lock, mode=_forced_mode)
return _forced_locks[lock]
LOCK_OFF = Lock(mode=LockMode.OFF)
"""The no-project default. A module-level singleton, not an inline constructor:
`get()` must hand back the same object every time."""
_discovered_locks: dict[Path | None, Lock] = {}
[docs]
def discovered_lock(project: JitxProject | None) -> Lock:
"""The project's lock, or `off` when there is no project to put one in.
Takes the project from :py:func:`jitxlib.parts.policy.current_project`, which
has already read its config. Keyed and cached exactly as `discovered_source`
is, and for the same two reasons: identity stability, and not re-reading
config for every part in the design. Reset with
:py:func:`clear_discovered_locks`.
"""
root = project.root if project is not None else None
if root not in _discovered_locks:
_discovered_locks[root] = _lock_for(project)
return _discovered_locks[root]
[docs]
def clear_discovered_locks() -> None:
"""Drop every cached lock. For tests, and for a fresh run in one process."""
_discovered_locks.clear()
def _parse_mode(value: Any, where: Path) -> LockMode:
"""The configured mode, falling back to `read` on anything unrecognized.
`read` rather than the default, deliberately. A project that wrote
`lock-mode` has said it cares how the lock is used, and the safe reading of
"you asked for something I do not understand" is to replay the lock and write
nothing: falling back to `read-write` would let a typo silently re-pin the
design, which is the one outcome a lock exists to prevent. A warning rather
than an error because a misspelled mode must not stop a build that the lock
can answer perfectly well.
"""
try:
return LockMode(value)
except ValueError:
logger.warning(
"%s: unknown lock-mode %r; falling back to %r. Valid modes are %s.",
where,
value,
LockMode.READ.value,
", ".join(repr(m.value) for m in LockMode),
)
return LockMode.READ
def _lock_for(project: JitxProject | None) -> Lock:
from .policy import _project_config
if project is None: # a bare script, a notebook
return LOCK_OFF
config = _project_config(project)
configured = config.get("lock-mode")
# Unset is the default, not a parse failure: most projects never write it.
mode = (
LockMode.READ_WRITE
if configured is None
else _parse_mode(configured, project.root)
)
# Resolved against the project, not the working directory. A configured
# `lock = "parts.lock.json"` otherwise lands wherever the build was started
# from, so the same project would read one lockfile and write another
# depending on where you ran it. Same rule the `csv:` chain paths follow.
return Lock(
path=str(project.root / Path(config.get("lock", "parts.lock.json"))), mode=mode
)
# --------------------------------------------------------------------------
# Entries
# --------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class LockKey:
"""What an entry is keyed on.
The digest gives query identity, the chain says what was asked, and
`lock_scope` disambiguates queries that should be pinned separately.
"""
digest: str
chain: str
lock_scope: str = ""
[docs]
@dataclass(frozen=True)
class LockEntry:
"""One pinned selection.
`provider` is the *chain member*, which is what is installed and what
implements `rehydrate`; `identity.provider` records where the part actually
came from, which for a delegating provider is not the same thing.
"""
key: LockKey
query: str
filters: Mapping[str, Predicate]
provider: str
identity: PartIdentity
payload: dict[str, JSON]
adopted_from: str | None = None
# --------------------------------------------------------------------------
# Encoding
# --------------------------------------------------------------------------
[docs]
def encode_value(value: object) -> Any:
"""Encode an attribute or bound for JSON.
Quantities use pint's own string form, tagged so nothing has to be sniffed
back. A dimensionless bound stays a plain number, which is why
`["Range", null, 0.01]` is right for `AtMost(1 * pct)` rather than an
inconsistency with the encoder.
"""
if is_quantity(value):
if value.dimensionless:
# to_base_units(), not .magnitude: (1 * pct).magnitude is 1, and the
# number it denotes is 0.01. Same trap as `match`'s comparison.
return float(value.to_base_units().magnitude)
return ["q", str(value)]
return value
[docs]
def decode_value(value: object) -> Any:
"""Inverse of :py:func:`encode_value`, through JITX's shared registry."""
if isinstance(value, list) and len(value) == 2 and value[0] == "q":
return registry.Quantity(value[1])
return value
@overload
def encode_payload(value: Mapping[str, object]) -> dict[str, JSON]: ...
@overload
def encode_payload(value: object) -> JSON: ...
[docs]
def encode_payload(value: object) -> JSON:
"""Render a provider payload as JSON, with quantities encoded.
For `ResolvedPart.payload`, which the core stores verbatim and never
serializes. Walks dicts and lists, and takes a bare value too:
payload = encode_payload({"rated": 35 * V, "series": ["E24", "E96"]})
`decode_payload` is the inverse, so `rehydrate` gets its quantities back.
Two asymmetries to know about. A dimensionless quantity encodes as its plain
magnitude, so `1 * pct` returns as `0.01` rather than as a quantity, matching
how attributes and filters are stored. And a tuple encodes as a list, so it
returns as a list.
Refuses a two-item list or tuple starting with `"q"`, at any depth, since that
is how a quantity is encoded and decoding would hand back one. Store such data
under a key instead, as `{"tag": "q", "value": ...}`.
"""
if isinstance(value, Mapping):
return {str(k): encode_payload(v) for k, v in value.items()}
if isinstance(value, (list, tuple)):
if len(value) == 2 and value[0] == "q":
raise ProviderContractError(
f"payload holds {list(value)!r}, which is how a quantity is "
f"encoded, so reading it back would give a quantity. Store it "
f'under a key instead, such as {{"tag": "q", "value": ...}}'
)
return [encode_payload(v) for v in value]
encoded = encode_value(value)
if not isinstance(encoded, (str, int, float, bool, list)) and encoded is not None:
raise ProviderContractError(
f"payload holds {value!r} of type {type(value).__name__}, which is "
f"not JSON; a payload may hold only JSON values and quantities"
)
return cast(JSON, encoded)
[docs]
def decode_payload(value: JSON) -> Any:
"""Inverse of `encode_payload`, restoring quantities.
A two-item list starting with `"q"` is read back as a quantity. `encode_payload`
refuses that shape as raw data, so anything it produced round-trips.
"""
if isinstance(value, Mapping):
return {k: decode_payload(v) for k, v in value.items()}
if isinstance(value, list):
if len(value) == 2 and value[0] == "q":
return decode_value(value)
return [decode_payload(v) for v in value]
return value
[docs]
def encode_predicate(predicate: Predicate) -> Any:
from dataclasses import fields as dc_fields
name = type(predicate).__name__
values = [encode_value(getattr(predicate, f.name)) for f in dc_fields(predicate)]
return [name, *values]
[docs]
def decode_predicate(blob: Sequence[Any]) -> Predicate:
"""Rebuild a predicate from its encoded form.
Written out per type rather than as `cls(*values)`. Two reasons: a splat over
a decoded list is untypeable without a cast, and an entry with the wrong arity
for its predicate name should be a clear error rather than a TypeError out of
a constructor.
"""
name, *rest = blob
values = [decode_value(v) for v in rest]
def arity(expected: int) -> None:
if len(values) != expected:
raise ProviderContractError(
f"lock entry has a {name} predicate with {len(values)} "
f"value(s); expected {expected}"
)
match name:
case "Exactly":
arity(1)
return Exactly(values[0])
case "OneOf":
arity(1)
return OneOf(tuple(values[0] or ()))
case "Range":
arity(2)
return Range(values[0], values[1])
case "Tolerance":
arity(3)
return Tolerance(values[0], values[1], values[2])
case "Exists":
arity(1)
return Exists(bool(values[0]))
case _:
raise ProviderContractError(f"lock entry has an unknown predicate {name!r}")
def _encode_attributes(attributes: Mapping[str, Scalar]) -> dict[str, Any]:
return {k: encode_value(v) for k, v in sorted(attributes.items())}
def _decode_attributes(blob: Mapping[str, Any]) -> dict[str, Any]:
return {k: decode_value(v) for k, v in blob.items()}
def _encode_identity(identity: PartIdentity) -> dict[str, Any]:
out: dict[str, Any] = {
"mpn": identity.mpn,
"manufacturer": identity.manufacturer,
"category": identity.category,
"provider": identity.provider,
"attributes": _encode_attributes(identity.attributes),
}
if identity.provider_ref is not None:
out["provider_ref"] = identity.provider_ref
return out
def _decode_identity(blob: Mapping[str, Any]) -> PartIdentity:
return PartIdentity(
mpn=blob["mpn"],
manufacturer=blob.get("manufacturer"),
category=blob.get("category", ""),
provider=blob.get("provider", ""),
provider_ref=blob.get("provider_ref"),
attributes=_decode_attributes(blob.get("attributes", {})),
)
# --------------------------------------------------------------------------
# The file
# --------------------------------------------------------------------------
def _entry_order(entry: LockEntry) -> tuple[str, str, str, str, str]:
"""Where an entry sits in the file, ordered so that diffs stay readable.
Sorts on what a person recognises before the digest, so a query edit whose
part survives leaves its entry in place. Parts with no manufacturer sort
first, together.
"""
return (
entry.key.chain,
entry.key.lock_scope,
entry.identity.manufacturer or "",
entry.identity.mpn,
entry.key.digest,
)
[docs]
@dataclass
class Lockfile:
"""A parsed lockfile, plus the chain interning table.
Chains are interned at the top of the file rather than repeated per entry,
because a design may resolve under more than one: any `with PartSource(...)`
block introduces another.
"""
path: Path | None = None
chains: dict[str, dict[str, Any]] = field(default_factory=dict)
entries: dict[LockKey, LockEntry] = field(default_factory=dict)
seen_keys: set[tuple[str, str, str]] = field(
default_factory=set, compare=False, repr=False
)
"""Keys claimed by a query this run, as `(digest, chain, lock_scope)`.
Run state rather than file content: never read from or written to the JSON,
and excluded from comparison and repr so it cannot be mistaken for either.
It lives here because it is keyed the way entries are, and because it is
shared exactly when the file is. Two designs in one process writing one
lockfile must share their claims, or the second adopts entries the first is
holding; two designs writing different lockfiles must not.
"""
# ---- reading ----------------------------------------------------------
[docs]
@classmethod
def read(cls, path: str | Path | None) -> Lockfile:
"""Parse a lockfile, or return an empty one if it does not exist.
Raises `LockStale` on a `digest_version` mismatch, which is the only lock
failure detectable before elaboration and the only one that is total.
"""
if path is None:
return cls()
target = Path(path)
if not target.is_file():
return cls(path=target)
with target.open("rb") as handle:
data = json.load(handle)
found = data.get("digest_version", 0)
if found != DIGEST_VERSION:
raise LockStale(found, DIGEST_VERSION)
out = cls(path=target, chains=data.get("chains", {}) or {})
for blob in data.get("entries", []) or []:
key = LockKey(
digest=blob["digest"],
chain=blob["chain"],
lock_scope=blob.get("lock_scope", ""),
)
out.entries[key] = LockEntry(
key=key,
query=blob.get("query", ""),
filters={
k: decode_predicate(v)
for k, v in (blob.get("filters", {}) or {}).items()
},
provider=blob.get("provider", ""),
identity=_decode_identity(blob["identity"]),
# A payload that is not an object came from a hand-edited file.
# Left empty rather than rejected: the provider's own version
# check then raises `StalePayload`, which is recorded and
# re-resolved, which is the designed path for a payload nobody
# can honour.
payload=payload
if isinstance(payload := blob.get("payload"), dict)
else {},
adopted_from=blob.get("adopted_from"),
)
return out
# ---- chain labels -----------------------------------------------------
[docs]
def label_for(self, digest: str, preferred: str | None = None) -> str:
"""The stable short handle for a chain digest, assigning one if needed.
Deterministic so that regenerating a lock does not reshuffle every entry:
`main` for the design-level chain, then whatever label a `with` block was
given, else `chain-2`, `chain-3`.
"""
for label, record in self.chains.items():
if record.get("digest") == digest:
return label
if preferred and preferred not in self.chains:
label = preferred
elif "main" not in self.chains:
label = "main"
else:
index = 2
while f"chain-{index}" in self.chains:
index += 1
label = f"chain-{index}"
self.chains[label] = {"digest": digest, "providers": []}
return label
[docs]
def digest_for_label(self, label: str) -> str | None:
record = self.chains.get(label)
return record.get("digest") if record else None
[docs]
def has_chain(self, digest: str) -> bool:
"""Whether any chain record in the file carries this digest."""
return any(r.get("digest") == digest for r in self.chains.values())
[docs]
def entries_for_chain(self, label: str) -> list[LockEntry]:
return [e for e in self.entries.values() if e.key.chain == label]
# ---- writing ----------------------------------------------------------
[docs]
def to_json(self) -> dict[str, Any]:
return {
"schema_version": SCHEMA_VERSION,
"digest_version": DIGEST_VERSION,
"chains": {k: self.chains[k] for k in sorted(self.chains)},
"entries": [
{
"digest": e.key.digest,
"chain": e.key.chain,
"lock_scope": e.key.lock_scope,
"query": e.query,
"filters": {
k: encode_predicate(v) for k, v in sorted(e.filters.items())
},
**(
{"adopted_from": e.adopted_from}
if e.adopted_from is not None
else {}
),
"provider": e.provider,
"identity": _encode_identity(e.identity),
"payload": e.payload,
}
for e in sorted(self.entries.values(), key=_entry_order)
],
}
[docs]
def write(self, path: str | Path | None = None) -> Path:
"""Write atomically: temp file in the same directory, fsync, rename.
A rename is atomic on every platform that matters, so a lockfile is never
observed half-written and an interrupted commit leaves the previous one
intact.
"""
target = Path(path) if path is not None else self.path
if target is None:
raise ValueError("no lockfile path to write to")
target.parent.mkdir(parents=True, exist_ok=True)
blob = json.dumps(self.to_json(), indent=1, sort_keys=False)
# Opened with 0666 so the kernel applies the umask, exactly as it would
# for `open(target, "w")`. `tempfile` creates 0600 and the rename carries
# that over, so the obvious version hands back a lockfile only its author
# can read, and patching that up with a hardcoded 0644 is worse still,
# since it overrides a umask of 077 that was set on purpose.
existing = target.stat().st_mode & 0o777 if target.exists() else None
temp = target.with_name(f"{target.name}.{os.getpid()}.{uuid4().hex}.tmp")
fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o666)
try:
if existing is not None:
# A project that chose a mode meant it.
os.fchmod(fd, existing)
with os.fdopen(fd, "w") as handle:
handle.write(blob)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temp, target)
except BaseException:
temp.unlink(missing_ok=True)
raise
return target
# --------------------------------------------------------------------------
# Adoption
# --------------------------------------------------------------------------
[docs]
class Tier(Enum):
"""Adoption tiers, strongest signal first.
What adoption protects is the layout, not the bill of materials: a part is a
footprint that has been placed, routed around and fitted next to its
neighbours, so re-resolution handing back a different size invalidates work
built on top of it.
"""
EXACT = 1
"""The orphan's recorded filters equal the query's current filters. The
rank-edit case, where the BOM would churn."""
SUBSUMED = 2
"""The query's filters are a decidable relaxation of the orphan's. The
widening case, where the *layout* would churn, which is worse."""
SATISFIED = 3
"""No filter relationship, but the part satisfies the query. Weakest signal,
broadest safety net."""
def _relaxes(q: Predicate | None, e: Predicate | None) -> bool | None:
"""Whether query predicate `q` relaxes entry predicate `e`.
True for strictly relaxed, False for equal, None for undecided. Deliberately
conservative: an undecided case is not subsumption, so it falls to tier 3
rather than being guessed at.
"""
if q is None and e is None:
return False
if q is None: # the query dropped a constraint
return True
if e is None: # the query added one
return None
if q == e:
return False
if isinstance(q, Tolerance) or isinstance(e, Tolerance):
return None
if isinstance(q, Exists) or isinstance(e, Exists):
return None
if isinstance(q, Range) and isinstance(e, Range):
return True if _range_contains(q, e) else None
if isinstance(q, OneOf) and isinstance(e, OneOf):
return True if set(e.values) < set(q.values) else None
if isinstance(e, Exactly) and isinstance(q, (Range, OneOf)):
from .query import satisfies
try:
return True if satisfies(q, e.value) else None
except DimensionMismatch:
return None
return None
def _range_contains(outer: Range, inner: Range) -> bool:
"""Whether `outer`'s interval contains `inner`'s. Conservative on None bounds."""
from .query import _le
try:
if outer.min is not None:
if inner.min is None or not _le(outer.min, inner.min, ""):
return False
if outer.max is not None:
if inner.max is None or not _le(inner.max, outer.max, ""):
return False
except DimensionMismatch:
return False
return True
[docs]
def subsumes(
query_filters: Mapping[str, Predicate], entry_filters: Mapping[str, Predicate]
) -> bool:
"""Whether the query's filter set is a decidable relaxation of the entry's.
Every key relaxed-or-equal, and at least one strictly relaxed.
"""
keys = set(query_filters) | set(entry_filters)
any_relaxed = False
for key in keys:
verdict = _relaxes(query_filters.get(key), entry_filters.get(key))
if verdict is None:
return False
any_relaxed = any_relaxed or verdict
return any_relaxed
[docs]
def tier_of(entry: LockEntry, query_filters: Mapping[str, Predicate]) -> Tier | None:
"""The strongest relation between an orphan and a query, or None.
The tiers are disjoint by construction: this returns the strongest that holds.
"""
if dict(entry.filters) == dict(query_filters):
return Tier.EXACT
if subsumes(query_filters, entry.filters):
return Tier.SUBSUMED
return Tier.SATISFIED
[docs]
def confirmed(query_filters: Mapping[str, Predicate], entry: LockEntry) -> bool:
"""Whether the entry's recorded part satisfies the query's current filters.
Part of *candidacy*, not a check on a chosen candidate: an entry whose
recorded part does not satisfy the query was never a candidate for it, and
letting it manufacture an ambiguity would let one hand-edited entry unpin a
legitimate match.
A raise means "not a candidate" rather than aborting the sweep. Per candidate,
not per pass: one unusable entry must not cost every other query its pin.
"""
try:
return match(query_filters, entry.identity.attributes, entry.identity)
except DimensionMismatch:
return False
[docs]
@dataclass
class AdoptionCandidate:
"""A query looking for an orphan to adopt, as the sweep sees it."""
digest: str
chain: str
lock_scope: str
filters: Mapping[str, Predicate]
query: str
payload_key: Any = None
@property
def sort_key(self) -> tuple[str, str, str]:
"""The lock key, which is the one iteration order that cannot tie."""
return (self.digest, self.chain, self.lock_scope)
[docs]
def adopt_orphans(
orphans: Iterable[LockEntry],
unmatched: Iterable[AdoptionCandidate],
) -> dict[str, LockEntry]:
"""Match orphaned entries against unmatched queries.
Returns a mapping from a candidate's lock key triple (joined) to the entry it
adopted. Tier-major, and it returns to tier 1 after every weak adoption: a
removal can make a stronger tier decidable for another query, and a loop that
only restarts between passes enforces the tiering invariant between passes
rather than at every moment.
"""
pool = list(orphans)
pending = sorted(unmatched, key=lambda c: c.sort_key)
adopted: dict[str, LockEntry] = {}
def sweep(tier: Tier, one_only: bool) -> bool:
progress = False
for candidate in list(pending):
cands = [
o
for o in pool
if o.key.chain == candidate.chain
and o.key.lock_scope == candidate.lock_scope
and tier_of(o, candidate.filters) is tier
and confirmed(candidate.filters, o)
]
if len(cands) == 1:
chosen = cands[0]
adopted["\x00".join(candidate.sort_key)] = chosen
pool.remove(chosen)
pending.remove(candidate)
progress = True
if one_only: # this removal may have unblocked EXACT
break
return progress
while any(
sweep(tier, one_only=tier is not Tier.EXACT)
for tier in (Tier.EXACT, Tier.SUBSUMED, Tier.SATISFIED)
):
pass # restart: a stronger tier may now decide
return adopted
[docs]
def rekey(entry: LockEntry, candidate: AdoptionCandidate) -> LockEntry:
"""Re-key an adopted entry to the query that adopted it.
Rewriting the filters is not optional: leave them and the file holds an entry
whose digest describes one query and whose filters describe another, so the
*next* edit matches against stale filters, wrong for exactly the entries that
have been adopted once.
"""
return replace(
entry,
key=LockKey(
digest=candidate.digest,
chain=candidate.chain,
lock_scope=candidate.lock_scope,
),
query=candidate.query,
filters=dict(candidate.filters),
adopted_from=entry.key.digest,
)
__all__ = [
"DIGEST_VERSION",
"SCHEMA_VERSION",
"AdoptionCandidate",
"LOCK_OFF",
"Lock",
"LockEntry",
"LockKey",
"LockMode",
"Lockfile",
"PINNED_MODES",
"Rank",
"Tier",
"adopt_orphans",
"clear_discovered_locks",
"confirmed",
"decode_predicate",
"discovered_lock",
"force_mode",
"encode_predicate",
"rekey",
"subsumes",
"tier_of",
]
