"""
The resolver
============
Batching, lockfile replay, and the run's parts report. Mostly internal machinery;
what a caller uses is the driver surface at the end of the file --
:py:func:`parts_verdict`, :py:func:`flush`, :py:func:`commit_lock` and
:py:func:`uncommitted`.
Construction enqueues, a flush resolves the batch, and writes happen once at the
end of the run.
A flush is triggered by the first read of deferred state, by introspection, or by
``Design.Initialized`` at the end of elaboration, whichever comes first. The last
is the common case, which is what makes batching worth having. Resolving a batch
means replaying whatever the lock can answer, sweeping orphaned entries for
adoption, and asking the chain for the rest; each component then binds its part in
the frame it was constructed in, through the :py:class:`PartsResolved` event.
Pins are committed once the end-of-elaboration flush has resolved everything, so
an ordinary run pins by itself and a driver need do nothing.
:py:func:`parts_verdict` is still the driver's to ask for, on every exit path
including a failed one, since only the driver knows how a run ended and what to do
about it. :py:class:`Verdict` carries the counts, the failures and the exit status.
"""
from __future__ import annotations
import threading
from collections import defaultdict
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass, field, replace
from enum import Enum
from typing import Any
from jitx.design import Design
from jitx.events import Event
from .error import (
NestedDeferral,
ProviderUnavailable,
StalePayload,
)
from .lock import (
AdoptionCandidate,
Lock,
LockEntry,
LockKey,
Lockfile,
LockMode,
PINNED_MODES,
rekey,
)
from .provider import PartSource, Resolved, describe
from .query import CanonicalQuery, PartQuery
from .record import ResolvedPart
[docs]
class PartsResolved(Event):
"""Fired after a batch resolves, so each component binds in its own frame.
A provider's `materialize` therefore runs as though it were where the
component was constructed, and any ambient context it reads while building
geometry -- a symbol style set in an enclosing class body, say -- is the one
that was active there. A handler on this event must not construct a part that
itself resolves through a chain; see `NestedDeferral`.
"""
# `Event.on` records a handler with the frame it was registered in and
# `fire()` re-enters that frame, which is exactly the capture-and-re-enter
# binding needs: `materialize` runs during a flush, not where the component
# was constructed.
[docs]
@dataclass(frozen=True)
class Resolution:
"""A part, and the chain member that answered.
The two travel together because the member cannot be recovered from the part:
`identity.provider` is free text the provider chose, two members may answer to
one name, and a delegating provider names its inner source. Only the chain
knows, and it only says so at the moment it answers -- so the answer carries
it from there to the lockfile.
"""
part: ResolvedPart
member: str
[docs]
class Severity(Enum):
"""Whether an event fails the run. A property of the mode, not of the event."""
INFO = 0
FAILURE = 1
[docs]
class EventKind(Enum):
"""What the resolver records. Severity is decided per mode; see `severity_of`."""
PINNED = "pinned"
ADOPTED = "adopted"
FORFEITED = "forfeited"
RESOLVED = "resolved"
"""The lock could not answer, so the query resolved fresh. Informational under
`read-write`, where it is the ordinary way a design grows; `unlocked` and fatal
under a pinned mode."""
REKEYED = "rekeyed"
"""A chain group the lock does not recognize. `stale-chain` under a pinned mode."""
REPLACED = "replaced"
"""A provider disowned its payload. `replay-failed` under a pinned mode."""
UNRESOLVABLE = "unresolvable"
"""Could not resolve at all: the chain is unreachable. A failure in every mode."""
DIFFERS = "differs"
"""Fresh resolution differs from the lock. `check` only, and its namesake."""
_PINNED_FAILURES = frozenset(
{
EventKind.RESOLVED,
EventKind.REKEYED,
EventKind.REPLACED,
EventKind.UNRESOLVABLE,
EventKind.DIFFERS,
}
)
_PINNED_NAMES = {
EventKind.RESOLVED: "unlocked",
EventKind.REKEYED: "stale-chain",
EventKind.REPLACED: "replay-failed",
}
[docs]
def severity_of(kind: EventKind, mode: LockMode) -> Severity:
"""Whether `kind` fails the run under `mode`.
The same event is informational in one mode and a failure in another, which is
what makes the report categories differ by mode without the mechanism
differing.
"""
if kind is EventKind.UNRESOLVABLE:
return Severity.FAILURE
if mode in PINNED_MODES and kind in _PINNED_FAILURES:
return Severity.FAILURE
return Severity.INFO
[docs]
def name_of(kind: EventKind, mode: LockMode) -> str:
"""The category name a mode reports this event under."""
if mode in PINNED_MODES:
return _PINNED_NAMES.get(kind, kind.value)
return kind.value
[docs]
@dataclass
class VerdictEvent:
"""One recorded event, with what it was about."""
kind: EventKind
description: str
detail: str = ""
"""What happened, beyond the category. May span several lines."""
origin: str = ""
"""Where the component was constructed, when the event names it."""
[docs]
@dataclass
class Verdict:
"""What the run has to say about parts.
Non-empty on essentially every successful build, since hits and adoptions are
recorded too. The exit status keys to the *failure* categories alone, never to
emptiness: "has something to say" is not "has something to complain about".
"""
mode: LockMode = LockMode.OFF
events: list[VerdictEvent] = field(default_factory=list)
[docs]
def record(
self, kind: EventKind, description: str, detail: str = "", origin: str = ""
) -> None:
self.events.append(VerdictEvent(kind, description, detail, origin))
[docs]
def counts(self) -> dict[str, int]:
out: dict[str, int] = {}
for event in self.events:
key = name_of(event.kind, self.mode)
out[key] = out.get(key, 0) + 1
return out
@property
def failures(self) -> list[VerdictEvent]:
return [
e for e in self.events if severity_of(e.kind, self.mode) is Severity.FAILURE
]
@property
def exit_status(self) -> int:
"""Nonzero if and only if some failure category is non-empty."""
return 1 if self.failures else 0
[docs]
def summary(self) -> str:
"""The one-line count summary a build prints.
"292 pinned, 7 adopted, 1 resolved" tells the reader their edit was
absorbed, which a silent success would hide.
"""
counts = self.counts()
if not counts:
return "no parts resolved"
order = [name_of(k, self.mode) for k in EventKind]
seen = [k for k in order if k in counts]
return ", ".join(f"{counts[k]} {k}" for k in seen)
[docs]
def report(self) -> str:
"""The full report: the summary, then every failure.
Each failure is a line naming its category and query, followed by
indented lines for where the component was constructed, when known, and
for each line of its detail.
"""
lines = [self.summary()]
for event in self.failures:
lines.append(f" {name_of(event.kind, self.mode)}: {event.description}")
if event.origin:
lines.append(f" at {event.origin}")
lines.extend(f" {line}" for line in event.detail.splitlines())
return "\n".join(lines)
# --------------------------------------------------------------------------
# Pending state
# --------------------------------------------------------------------------
[docs]
@dataclass
class DeferredState:
"""Everything one deferred component carries.
Lives on the component rather than in a side table, so nothing has to reach
around the proxy to find it.
"""
query: CanonicalQuery
source: PartSource
lock: Lock
lock_scope: str
origin: str
batch: Batch
"""The batch this query is pending in.
A direct reference rather than an id into a registry. The registry existed
only because this was an id, and nothing ever removed from it, so it grew by
one entry per design for the life of the process. A reference keeps the batch
alive for exactly as long as a state needs it, which is what the registry was
emulating.
"""
part: ResolvedPart | None = None
resolved: bool = False
binding: bool = False
"""Reentrancy guard, and not optional: `_bind` sets fields and provider code
reads them back while the part is not yet marked resolved."""
failure: str | None = None
"""Why this query will never resolve, once that is settled.
A flush drains `pending`, so a state that failed is terminal but otherwise
indistinguishable from one that has not been asked yet: `part` is None and
`resolved` is False in both cases. Recording the reason here is what lets a
later attribute read say which it is, and say why, rather than raising the
name it was asked for. Set wherever an `unresolvable` event is recorded, and
carrying the same text, so the exception and the report agree."""
[docs]
def caller_site(component: object) -> str:
"""Where a part was constructed, for diagnostics.
Reads jitx's own `FileInfo`, which `Sourced.__new__` attaches to every
structural object at construction. Walking the stack from here instead found
the instantiation trampoline in `jitx._structural`, because by the time a
query is enqueued the design frame is no longer the caller -- the same reason
jitx captures this at `__new__` rather than on demand.
"""
from jitx.fileinfo import FileInfo
info = FileInfo.get(component)
return str(info) if info is not None else "<unknown>"
[docs]
class Run:
"""State that outlives any one design: the report, and the lockfiles.
A process may elaborate several designs -- `jitx build A B`, or a test suite
-- and what they share is exactly what is keyed by *file* rather than by
design. Two designs writing one lockfile must hold the same parsed object or
their staged writes diverge, and `parts_verdict()` is asked once for the run
however many designs it covered.
What they do *not* share is on `Batch`; see `Batch.resolution_cache`.
"""
def __init__(self):
self.verdict: Verdict = Verdict()
self.pending_writes: dict[Any, Lockfile] = {}
self.lockfiles: dict[str | None, Lockfile] = {}
[docs]
class Batch:
"""One pending set of queries, resolved together. One per design."""
def __init__(self, run: Run):
self.run = run
self.pending: list[DeferredState] = []
self.flushing = False
self.binding = False
self.resolution_cache: dict[tuple[str, str], Resolution] = {}
"""Answers already had from the chain, keyed `(query digest, chain)`.
Holds `Resolution`, not a bare part: the answering member has to survive
the cache, or a part served from it loses the provenance the lockfile
needs.
Per design, and spanning that design's flushes, which is the lifetime it
needs: nothing stages a resolved miss into the lockfile under `off`,
`read` or `check`, so without it a second flush could re-ask a question
the first already answered and bound, and one run would give two answers
to one question. It does nothing *within* a flush -- the key carries the
chain digest and `_chain_answers` is called once per chain group, so a
cross-group hit is impossible and the per-call dedup covers the rest.
Not shared between designs. Two boards carry no reproducibility promise
to each other beyond the lockfile, and under `read-write` they agree
anyway, because the first design's answer is staged into the shared
in-memory lockfile and the second finds it at the lock lookup.
"""
[docs]
def add(self, state: DeferredState) -> None:
self.pending.append(state)
[docs]
def on_design_initialized(self, _event: Any = None) -> None:
"""Flush this design's remaining queries, commit the lockfile, and retire.
Runs at `Design.Initialized`, where the pending set is complete and a
recovered pin still reaches the board. Every driver gets the same
behaviour, including a bare `python design.py`.
"""
self.flush(final=True)
commit_lock()
from jitx.design import Design as _Design
# Retire, so the next design in this process starts a fresh batch and
# registers a handler of its own. Parts already bound hold the batch
# directly, so it lives exactly as long as something needs it.
if _active.current is self:
_active.current = None
_Design.Initialized.off(self.on_design_initialized)
[docs]
def flush(self, final: bool = False) -> None:
"""Resolve everything pending, then fire `PartsResolved`."""
if self.flushing:
return
self.flushing = True
try:
_resolve_pending(self, final=final)
finally:
self.flushing = False
class _Active(threading.local):
"""The one ambient root, and the only one.
`_defer` is called from a component's `__init__` with nowhere to be handed a
batch from, so *something* has to be reachable by name. Everything else hangs
off what this holds: the run's state on `Run`, a design's on `Batch`, and the
per-lockfile state on `Lockfile` itself.
Per thread rather than global: two threads elaborating two designs must not
share a pending set.
"""
def __init__(self):
self.run: Run = Run()
self.current: Batch | None = None
_active = _Active()
[docs]
def current_batch() -> Batch:
"""This thread's pending batch, creating and arming one if needed.
The end-of-elaboration flush is registered *here* rather than at module
import. `Event.on` outside an active instantiation returns a `PrePostInit`
wrapper for use in a class body, so a module-level decorator registers
nothing; called from `_defer`, this runs inside instantiation and registers
for real, with the frame.
"""
if _active.current is None:
batch = Batch(_active.run)
_active.current = batch
Design.Initialized.on(batch.on_design_initialized)
return _active.current
[docs]
def reset() -> None:
"""Drop all resolver state. For tests and for a fresh run in one process."""
global _active
_active = _Active()
[docs]
def enqueue(
component: Any,
query: PartQuery,
*,
source: PartSource,
lock: Lock,
origin: str,
) -> DeferredState:
"""Register a component's query with the current batch.
Everything the resolution depends on is captured here; nothing is read again
at flush time.
"""
batch = current_batch()
if batch.binding:
raise NestedDeferral(
f"cannot defer a part while parts are binding (at {origin}). A "
f"provider's `materialize`, or a `PartsResolved` handler, must not "
f"construct a component that itself resolves through a chain: the "
f"binding event is mid-dispatch, and registering another handler on "
f"it does not terminate. Build such a part in the owning circuit's "
f"`__init__` and hand it to `materialize`, or give the provider "
f"concrete geometry."
)
state = DeferredState(
query=query.canonical(),
source=source,
lock=lock,
lock_scope=query.lock_scope,
origin=origin,
batch=batch,
)
batch.add(state)
return state
# --------------------------------------------------------------------------
# The flush pipeline
# --------------------------------------------------------------------------
def _lockfile_for(run: Run, lock: Lock) -> Lockfile:
"""Read (and cache) the lockfile for this lock context."""
if lock.path in run.lockfiles:
return run.lockfiles[lock.path]
parsed = Lockfile.read(lock.path) if lock.reads else Lockfile()
run.lockfiles[lock.path] = parsed
return parsed
def _resolve_pending(batch: Batch, final: bool) -> None:
"""Resolve the batch: replay what the lock answers, then ask the chain.
A flush drains everything pending, so one early read forfeits adoption for
every query pending at that moment, not just the part that was read.
"""
pending, batch.pending = batch.pending, []
if not pending:
return
run = batch.run
verdict = run.verdict
mode = pending[0].lock.mode
if verdict.mode == LockMode.OFF:
verdict.mode = mode
# 1. Group by captured chain. A single batch can span chains, so dedup keys on
# (digest, chain) rather than the digest alone.
groups: dict[str, list[DeferredState]] = defaultdict(list)
chain_digests: dict[str, str] = {}
for state in pending:
digest = state.source.digest()
chain_digests[digest] = digest
groups[digest].append(state)
unresolved: list[DeferredState] = []
for chain_digest, states in groups.items():
lock = states[0].lock
lockfile = _lockfile_for(run, lock)
label = (
lockfile.label_for(chain_digest, states[0].source.label)
if lock.reads
else "main"
)
known_chain = lockfile.has_chain(chain_digest) if lock.reads else False
# 2. Lockfile lookup, keyed (digest, chain, lock_scope).
for state in states:
key = LockKey(state.query.digest, label, state.lock_scope)
lockfile.seen_keys.add((key.digest, key.chain, key.lock_scope))
entry = lockfile.entries.get(key) if lock.reads else None
if entry is None:
if lock.reads and not known_chain and lockfile.entries:
verdict.record(
EventKind.REKEYED,
state.query.describe(),
f"chain {label} has no entries",
state.origin,
)
unresolved.append(state)
continue
part = _replay(entry, state, verdict)
if part is None:
unresolved.append(state)
continue
state.part = part
verdict.record(
EventKind.PINNED, state.query.describe(), origin=state.origin
)
# 3. At the final flush, misses get an adoption pass before anything reaches a
# provider. Queries that flushed early never reach it, and forfeit.
if unresolved and final:
unresolved = _adopt(run, unresolved, verdict)
elif unresolved and not final:
for state in unresolved:
verdict.record(
EventKind.FORFEITED,
state.query.describe(),
"flushed early",
state.origin,
)
# 4/5. Deduplicate by (digest, chain) and ask each chain group's PartSource.
if unresolved:
_resolve_from_chain(batch, unresolved, verdict)
# 5b. Under `check`, and only there: resolve the whole design cleanly and
# report where the board this run built differs from what that clean run
# would build. Two calls, kept apart on purpose -- step 4/5 above is the
# ordinary run and is what *binds*, so `check` builds exactly what `read`
# would, and this one only reports.
#
# The clean batch is every query the design asked, because that is what an
# `off` run would have handed the chain: a provider doing order-level work
# answers a partial order differently by design, so comparing against a
# fragment would report a difference the core itself created.
checking = [s for s in pending if s.lock.mode == LockMode.CHECK]
if checking:
_verify_against_clean_run(checking, verdict)
# 6. Fire, so each component binds in its own frame. Only when something
# resolved: firing needs an active EventContext, and there is nothing to
# bind otherwise.
if any(state.part is not None for state in pending):
batch.binding = True
try:
PartsResolved().fire()
finally:
batch.binding = False
def _replay(
entry: LockEntry, state: DeferredState, verdict: Verdict
) -> ResolvedPart | None:
"""Rebuild a pinned part through its chain member's `rehydrate`.
A replay failure is recorded, never raised: the query re-resolves from the
chain, and the event's severity is the mode's business.
"""
# By chain-member name, which is what the entry records. Matching on
# `p.name` compared a raw provider name against a member name, so a member
# that had been disambiguated or explicitly named could never be found.
member = next(
(
p
for p, name in zip(state.source, state.source.names(), strict=True)
if name == entry.provider
),
None,
)
if member is None:
verdict.record(
EventKind.REPLACED,
state.query.describe(),
f"chain has no member named {entry.provider!r}",
state.origin,
)
return None
try:
return member.rehydrate(entry.identity, entry.payload)
except (StalePayload, ProviderUnavailable) as exc:
verdict.record(
EventKind.REPLACED, state.query.describe(), str(exc), state.origin
)
return None
def _adopt(
run: Run, states: list[DeferredState], verdict: Verdict
) -> list[DeferredState]:
"""Run the adoption sweep over orphans and unmatched queries.
Orphan-ness is decided against every query *seen* this run, not those still
pending: a query that flushed early and got an exact hit has claimed its
entry, and without that a late-constructed part's entry would look orphaned
and could be adopted out from under it.
"""
by_lock: dict[str | None, list[DeferredState]] = defaultdict(list)
for state in states:
by_lock[state.lock.path].append(state)
still: list[DeferredState] = []
for path, group in by_lock.items():
lockfile = run.lockfiles.get(path)
if lockfile is None or not group[0].lock.reads:
still.extend(group)
continue
orphans = [
e
for e in lockfile.entries.values()
if (e.key.digest, e.key.chain, e.key.lock_scope) not in lockfile.seen_keys
]
# Several components can ask the identical question -- same digest, chain
# and lock_scope -- and they are one *candidate* but several *states*. So
# the key carries every state that asked it. Keying one state per question
# drops the rest on the floor: they never reach the chain, their
# components are never bound, and because every event is recorded by the
# code that handles a state, a state nothing handles produces no event at
# all -- the verdict reports success over a hole it cannot see.
candidates: dict[str, list[DeferredState]] = defaultdict(list)
wanted: dict[str, AdoptionCandidate] = {}
for state in group:
label = lockfile.label_for(state.source.digest(), state.source.label)
candidate = AdoptionCandidate(
digest=state.query.digest,
chain=label,
lock_scope=state.lock_scope,
filters=state.query.filters,
query=state.query.describe(),
)
key = "\x00".join(candidate.sort_key)
candidates[key].append(state)
# One question asked twice is one candidate. Handing the sweep the
# duplicate as well leaves it in `pending` forever: the first copy
# adopts the orphan and removes it from the pool, and the second then
# matches nothing.
wanted.setdefault(key, candidate)
from .lock import adopt_orphans
adopted = adopt_orphans(orphans, wanted.values())
for join_key, entry in adopted.items():
sharing = candidates[join_key]
# One replay for the question, not one per component: a failure is a
# single event, and the part is shared exactly as a chain resolution
# shares one among states with the same digest.
part = _replay(entry, sharing[0], verdict)
if part is None:
still.extend(sharing)
continue
new_entry = rekey(entry, wanted[join_key])
del lockfile.entries[entry.key]
lockfile.entries[new_entry.key] = new_entry
lockfile.seen_keys.add(
(
new_entry.key.digest,
new_entry.key.chain,
new_entry.key.lock_scope,
)
)
# An adoption re-keys the entry, so it has to be staged like any
# other change. An adopted query never reaches the chain, so nothing
# downstream would stage it.
if sharing[0].lock.writes:
run.pending_writes[sharing[0].lock.path] = lockfile
for state in sharing:
state.part = part
# Per state, so the verdict's counts keep matching the component
# counts. That equality is the invariant this whole block exists
# to hold.
verdict.record(
EventKind.ADOPTED, state.query.describe(), origin=state.origin
)
for join_key, sharing in candidates.items():
if join_key not in adopted:
still.extend(sharing)
return still
def _chain_answers(
source: PartSource,
states: Iterable[DeferredState],
ask: Callable[[Sequence[CanonicalQuery]], Sequence[Resolution | None]],
cache: dict[tuple[str, str], Resolution],
) -> dict[str, Resolution | None]:
"""Answers for these states' distinct queries, keyed by query digest.
One question is one call: several components asking the same thing, and two
queries differing only in `lock_scope`, are the same question as far as a
provider is concerned, and a query already answered this run is not asked
again. `ask` says how the chain is consulted, since the callers need different
failure behaviour: `resolve` raises on exhaustion, `outcomes` reports it.
Collapsing duplicates is what keeps one question from becoming several calls;
`CanonicalQuery.count` is what keeps the collapse from destroying the order,
so a provider doing MOQ or rebate arithmetic can still see that this exact
question was asked forty times. A provider that ignores it is correct.
"""
chain = source.digest()
unique: dict[str, CanonicalQuery] = {}
counts: dict[str, int] = {}
for state in states:
digest = state.query.digest
unique.setdefault(digest, state.query)
counts[digest] = counts.get(digest, 0) + 1
unique = {d: replace(q, count=counts[d]) for d, q in unique.items()}
# `cache` is passed rather than reached for, because which cache this is
# matters: the binding path uses the run's, and `check`'s verification phase
# uses one of its own so the two cannot see each other.
#
# Keyed on the question, not on how many asked it: a part already resolved
# against this cache is reused rather than re-asked, so a provider is not
# consulted again merely because the count grew. Re-asking would let one run
# give two answers to one question, which is the thing the lock exists to
# prevent.
answers: dict[str, Resolution | None] = {
digest: cache.get((digest, chain)) for digest in unique
}
to_ask = [q for digest, q in unique.items() if answers[digest] is None]
if to_ask:
for query, found in zip(to_ask, ask(to_ask), strict=True):
answers[query.digest] = found
if found is not None:
cache[(query.digest, chain)] = found
return answers
def _resolve_from_chain(
batch: Batch, states: list[DeferredState], verdict: Verdict
) -> None:
"""Ask each chain group's PartSource, then record and pin what comes back.
What a mode does with a query the chain cannot answer is the interesting part;
see `_unresolvable` and the comments below.
"""
groups: dict[str, list[DeferredState]] = defaultdict(list)
for state in states:
groups[state.source.digest()].append(state)
for group in groups.values():
source = group[0].source
lock = group[0].lock
# `replay` is hermetic: the chain is never asked. A miss is recorded and
# its component left unbound, which the verdict then explains. Note where
# this sits -- after the adoption pass, which needs no provider to reach
# its decision and so runs in this mode like any other.
if not lock.resolves:
_unresolvable(
group,
verdict,
f"{lock.mode} does not consult the chain; "
f"pin this query with a resolving mode first",
)
continue
lockfile = _lockfile_for(batch.run, lock)
label = lockfile.label_for(source.digest(), source.label)
lockfile.chains[label]["providers"] = source.names()
# Under a pinned mode, resolving a miss is diagnostic rather than
# productive: the query is already a failure, and the mode's promise is
# the complete picture. So the chain is asked without raising and each
# query it cannot answer is recorded, rather than the first one costing
# the report every answer behind it. Under `read-write` resolution *is*
# the point, so `PartNotFound` and `PartDenied` still raise with their
# per-provider trail.
reasons: dict[str, list[str]] = {}
ask = _peek(source, reasons) if lock.pinned else _ask(source)
try:
answers = _chain_answers(source, group, ask, batch.resolution_cache)
except ProviderUnavailable as exc:
# An unreachable chain is `unresolvable` in every mode, `read-write`
# included: no query got an answer, and there is nothing to raise
# about one query in particular.
_unresolvable(group, verdict, str(exc))
continue
for state in group:
found = answers.get(state.query.digest)
if found is None:
lines = reasons.get(state.query.digest, [])
state.failure = "; ".join(lines) or "the chain had no answer"
verdict.record(
EventKind.UNRESOLVABLE,
state.query.describe(),
"\n".join(lines),
state.origin,
)
continue
state.part = found.part
verdict.record(
EventKind.RESOLVED, state.query.describe(), origin=state.origin
)
if lock.writes:
_stage_write(batch.run, lockfile, state, label, found)
def _unresolvable(states: list[DeferredState], verdict: Verdict, reason: str) -> None:
"""Record a whole group as unresolvable, leaving every component unbound.
A failure in every mode. The run reaches its summary rather than dying here,
which is the point: a box that cannot reach its chain should be told what is
unpinned, not handed a provider exception mid-elaboration. It will still fail
-- the first read of any field on an unbound component raises
`UnresolvedPart` -- but the driver asks for the verdict on every exit path,
so the report is what the user sees.
`reason` goes on each state as well as into the event, and that is what makes
the second half of that true. Without it the eventual raise was a bare
`AttributeError` naming the field, which says nothing about parts, cannot be
told from a typo, and left the report as the only place the reason existed --
so the failure the user actually hit and the report explaining it had no
visible connection.
"""
for state in states:
state.failure = reason
verdict.record(
EventKind.UNRESOLVABLE, state.query.describe(), reason, state.origin
)
def _ask(
source: PartSource,
) -> Callable[[Sequence[CanonicalQuery]], list[Resolution | None]]:
"""Ask `source` and raise on exhaustion, keeping the answering member.
`PartSource.selections` is `resolve` with the one fact `resolve` discards. The
binding path under `read-write` wants both: the raise, because resolution is
the point there, and the member, because it is about to be written down.
"""
def ask(batch: Sequence[CanonicalQuery]) -> list[Resolution | None]:
return [Resolution(part, member) for part, member in source.selections(batch)]
return ask
def _peek(
source: PartSource,
reasons: dict[str, list[str]] | None = None,
) -> Callable[[Sequence[CanonicalQuery]], list[Resolution | None]]:
"""Ask `source` without raising: `None` where the chain has no answer.
`PartSource.resolve` raises on exhaustion, which is right where resolution is
the point -- `read-write` -- and wrong everywhere the chain is being consulted
for a report. Verification must not stop at the first difference, and a pinned
run must not lose the rest of its report to one unanswerable query.
`reasons`, when given, collects the chain's own account per query digest, one
`member: reason` line per member, so the recorded event can say *why* rather
than only that.
"""
names = source.names()
def ask(batch: Sequence[CanonicalQuery]) -> list[Resolution | None]:
out: list[Resolution | None] = []
for query, answer in zip(batch, source.outcomes(batch), strict=True):
outcome = answer.outcome
if isinstance(outcome, Resolved) and outcome.parts:
member = next(
(
name
for name, sub in reversed(answer.trail)
if isinstance(sub, Resolved)
),
names[0] if names else "",
)
out.append(Resolution(outcome.parts[0], member))
continue
if reasons is not None:
reasons[query.digest] = [
f"{name}: {describe(o)}" for name, o in answer.trail
] or ["the chain is empty"]
out.append(None)
return out
return ask
def _verify_against_clean_run(states: list[DeferredState], verdict: Verdict) -> None:
"""Resolve the design from scratch and report where this run's board differs.
`check` only, and its namesake. The batch is every query the design asked, so
the chain sees the order an `off` run would have handed it, and each bound
part is compared against what that clean run picked.
Adoptions are compared like anything else, and an earlier rule excluding them
was wrong. Adoption is a local *re-association*: it decides from lockfile data
alone, consulting no provider, that an existing pin answers a query the lock
could not key. That says nothing whatever about what a clean resolve of that
query returns -- which is the open question, and the one this batch is here to
ask. The association is what makes the comparison possible at all, by giving a
query with no entry of its own something to be compared against.
The excluding rule assumed the answer in advance ("a fresh resolution *will*
pick something else"), and it may well not: an adopted pin agreeing with a
clean run is an ordinary outcome, not a lucky one. Skipping the comparison
would report no divergence over a board that might genuinely hold one.
Pure reporting: the run's own part is already bound and is the one that gets
built, a clean answer is never assigned, and nothing is staged for a write.
"""
# A from-scratch run gets a from-scratch cache. Local, so two things hold by
# construction rather than by discipline: this phase cannot read what the
# binding phase cached -- it would then compare the board against itself --
# and cannot write into it, which would let a later flush bind what was meant
# to be pure reporting. The same relationship `search()` has to the resolver's
# cache, and for the same reason.
clean_cache: dict[tuple[str, str], Resolution] = {}
groups: dict[str, list[DeferredState]] = defaultdict(list)
for state in states:
groups[state.source.digest()].append(state)
for group in groups.values():
source = group[0].source
reasons: dict[str, list[str]] = {}
try:
answers = _chain_answers(source, group, _peek(source, reasons), clean_cache)
except ProviderUnavailable as exc:
# Same rule as the binding phase: a chain that cannot be reached is
# reported, not raised. Doubly so here, where the board is already
# built and this phase is pure reporting -- dying in it would throw
# away a run that had succeeded.
answers = {}
reasons = {s.query.digest: [str(exc)] for s in group}
for state in group:
built = state.part
if built is None:
continue
found = answers.get(state.query.digest)
if found is None:
# The chain's reasons nest one step under the line they explain.
why = [f" {line}" for line in reasons.get(state.query.digest, [])]
lines = [
f"built: {built.identity.describe()}",
"clean run: the chain no longer answers this query",
*why,
]
verdict.record(
EventKind.DIFFERS,
state.query.describe(),
"\n".join(lines),
state.origin,
)
continue
difference = _selection_difference(built, found.part)
if difference:
verdict.record(
EventKind.DIFFERS, state.query.describe(), difference, state.origin
)
def _selection_difference(built: ResolvedPart, clean: ResolvedPart) -> str:
"""How a clean run's pick differs from the one on the board, or `""`.
Compares which part was chosen and who chose it, not the whole record.
`payload` is provider-private and legitimately volatile -- a stock level, a
datasheet URL, a regenerated handle -- and `identity.attributes` can drift
without moving the board, so neither is grounds for failing `check`.
"""
a, b = built.identity, clean.identity
if (a.mpn, a.manufacturer) != (b.mpn, b.manufacturer):
return f"built: {a.describe()}\nclean run: {b.describe()}"
if a.provider != b.provider:
return (
f"built: {a.describe()} from provider {a.provider!r}\n"
f"clean run: {b.describe()} from provider {b.provider!r}"
)
return ""
def _stage_write(
run: Run, lockfile: Lockfile, state: DeferredState, label: str, found: Resolution
) -> None:
"""Hold a pin for the end-of-run write.
`found.member` is recorded verbatim: the chain said which member answered and
that is the only thing that knows. Reconstructing it from the part was the
bug -- a duplicate name resolved to whichever member came first, and a
delegating provider matched nothing and fell back to member one.
"""
part = found.part
member = found.member
key = LockKey(state.query.digest, label, state.lock_scope)
lockfile.entries[key] = LockEntry(
key=key,
query=state.query.describe(),
filters=dict(state.query.filters),
provider=member,
identity=part.identity,
payload=part.payload,
)
run.pending_writes[state.lock.path] = lockfile
# --------------------------------------------------------------------------
# Driver surface
# --------------------------------------------------------------------------
[docs]
def flush(final: bool = False) -> None:
"""Resolve everything pending in this thread's current batch."""
batch = _active.current
if batch is not None:
batch.flush(final=final)
[docs]
def parts_verdict() -> Verdict:
"""The run's parts report and exit status.
Called on **every** exit path, a failed translation included: an unresolvable
miss leaves a component unbound, and an unbound component raises in
translation, so a driver that asked only on success would report nothing in
exactly the cases that most need explaining.
When the verdict contains failures, that report is what the user should see,
with the translation exception named as its consequence. When it contains only
informational events, the exception stands on its own.
"""
return _active.run.verdict
[docs]
def commit_lock() -> list[str]:
"""Write pending pins, and return the paths written.
Called automatically once the end-of-elaboration flush has resolved
everything, so an ordinary run pins without a driver doing anything. Public
and idempotent so a harness that flushed by hand can still commit; it writes
nothing when nothing is staged.
"""
written: list[str] = []
for path, lockfile in list(_active.run.pending_writes.items()):
if path is None:
continue
written.append(str(lockfile.write(path)))
_active.run.pending_writes.clear()
return written
[docs]
def uncommitted() -> int:
"""How many lockfiles are holding pins that were never written.
Zero after any run that reached the end of elaboration. Non-zero means pins
were staged by an early flush and the run stopped before finishing, which is
the case where writing would produce a partial lock.
"""
return len(_active.run.pending_writes)
__all__ = [
"Batch",
"Run",
"DeferredState",
"EventKind",
"PartsResolved",
"Resolution",
"Severity",
"Verdict",
"VerdictEvent",
"caller_site",
"commit_lock",
"current_batch",
"enqueue",
"flush",
"name_of",
"parts_verdict",
"reset",
"severity_of",
"uncommitted",
]
