Shared helpers for domain-specific fact extractors.
Provides shared capabilities that extractors commonly need:
add_fact/3— accumulate a row into a relation mapscan_functions/4— walk every function and instruction with a handlerscan_remote_calls/4— likescan_functions/4, pre-filtered to remote callsresolve_callee/1— resolve the{:x, 0}argument as an atom stringresolve_atom/3— resolve any register as an atom stringmatch_remote_call/1— recognizecall_extvariants as{mod, func, arity}match_local_call/1— recognize intra-modulecallvariantsresolve_register/3— backward dataflow: determine a register's value at a specific call site by walking preceding instructionsget_behaviours/1— extract behaviour modules from attributesfind_function/3— look up a function's instructions by name and arity
Summary
Types
Per-instruction context passed to scan handlers. Carries everything an
extractor needs to call resolve_register/3 against the surrounding code.
Functions
Append a row to the given relation in a facts map.
Determine whether register is a function parameter at instruction
index call_idx. Returns {:ok, n} if it's the n-th parameter (so
{:x, n} for n < arity), or :no otherwise.
Trace register at instruction call_idx back to the call whose
RESULT it holds, following register-to-register move chains.
The graph of the function an instr_ctx() is in.
The control-flow graph of the function ctx is in, from the graphs
the pipeline attached to module_data or built on the spot.
Disable imprecision tracking for the current Erlang process. Subsequent
track_* calls become no-ops. Always called from the pipeline's
try/after so the flag is cleared even on extractor errors.
Like each_remote_call/3, for remote and local calls alike.
Calls handler.(facts, ctx, {mod, func, arity}) for every remote call
in the module, from the call-site index the pipeline attached (or one
built on the spot). ctx is the same instr_ctx() the per-instruction
scanners pass, so resolve_register/3 and friends work unchanged.
Enable imprecision tracking for the current Erlang process. Subsequent
track_imprecision/5 and track_dynamic/5 calls will record events.
Find a function's instruction list by name and arity.
Extract behaviour modules from a module's attributes.
Return instructions starting from a given label number.
Find the register holding key's value in a keyword list built at
runtime and pointed to by list_reg at instruction idx.
Match a BEAM instruction as a local (intra-module) function call.
Match a BEAM instruction as a remote (external) function call.
The process a call is addressed to, as every target column spells it:
the inspected module atom, "via:Registry" for a via tuple naming a
registry, or "dynamic".
The instruction that most recently wrote register strictly before
idx, as {:ok, instruction, writer_idx}, or :no.
A register operand with its type annotation stripped: {:tr, reg, type}
becomes reg. Seven extractors carried a copy of this clause.
Resolve register at instruction idx and return its inspected atom
string, or "dynamic" if the value cannot be statically determined or
is not an atom.
Resolve {:x, 0} at the current instruction context, returning the
inspected atom or "dynamic".
Resolve the value of register at instruction index call_idx by walking
backward through the instruction list.
Resolve register at instruction idx and classify the result as a
literal atom, a function parameter, or dynamic.
Every tuple a function returns, as {index, elements}: built by
put_tuple2 (into {x,0}, or into a register moved to {x,0} before
the return), by the pre-OTP-24 put_tuple/put sequence, or folded by
the compiler into one literal moved into {x,0}. Elements are in the
instruction vocabulary — {:atom, a}, {:integer, n}, {:literal, t},
a register — whatever their source, so a rule reading [{:atom, :ok} | rest] reads all three shapes.
Walk every function in functions and every instruction within each
function, calling handler.(facts, ctx, instr) for each instruction.
A timeout argument as the schema spells it: the milliseconds, "-1"
for :infinity, "0" when it could not be read.
Returns whether imprecision tracking is currently enabled for this process.
Useful in tests; production code should just call the track_* helpers
and rely on them to no-op when tracing is off.
Conditional wrapper around track_imprecision/5. When value indicates
a dynamic fallback (the string "dynamic" or the atom :dynamic),
records the event. No-op otherwise AND no-op when tracing is disabled.
Record an imprecision event explicitly. Use directly when the extractor decided to skip a fact emission entirely (the "intentional skip" case is still information — the value was missing, not just dynamic).
What register holds at idx, in one verdict: a literal, a function
parameter, the result of a call, or nothing knowable. The resolution
cascade that four extractors each wrote out.
Types
@type instr_ctx() :: %{func_id: String.t(), instrs: [tuple()], idx: non_neg_integer()}
Per-instruction context passed to scan handlers. Carries everything an
extractor needs to call resolve_register/3 against the surrounding code.
@type register() :: {:x, non_neg_integer()} | {:y, non_neg_integer()}
Functions
@spec add_fact(Argus.Pipeline.Emit.facts(), atom(), [String.t()]) :: Argus.Pipeline.Emit.facts()
Append a row to the given relation in a facts map.
@spec arg_position([term()], non_neg_integer(), register()) :: {:ok, non_neg_integer()} | :no
Determine whether register is a function parameter at instruction
index call_idx. Returns {:ok, n} if it's the n-th parameter (so
{:x, n} for n < arity), or :no otherwise.
A register is "still a parameter" at index call_idx if walking back
through the preceding instructions hits the function-entry func_info
without crossing a write to that register or a control-flow barrier.
This lets extractors distinguish "I don't know" from "this is
parameter N", which matters for client-API functions like
def get(pid), do: GenServer.call(pid, :get).
@spec call_result_origin([term()], non_neg_integer(), register()) :: {:ok, {module(), atom(), arity()}, non_neg_integer()} | :no
Trace register at instruction call_idx back to the call whose
RESULT it holds, following register-to-register move chains.
Returns {:ok, {mod, func, arity}, origin_idx} where origin_idx is
the absolute instruction index of the originating call — useful for
resolving that call's own arguments (e.g. mapping an ETS table
reference back to the :ets.new/2 site that created it, then reading
the table name from x0 there), or for stepping into a local defp
helper that produced the value. Both remote (call_ext) and local
(call) non-tail calls are reported; tail-call forms are path
barriers. Returns :no when the register holds anything else.
The walk is sound about register lifetimes: x registers do not
survive calls (only x0 carries the result), so tracing an {:x, n}
with n != 0 hits a call boundary and stops. y registers survive
calls and are followed through. Tail calls and return are path
barriers, as in resolve_register/3.
@spec cfg(map(), instr_ctx()) :: Argus.Cfg.Function.t() | nil
The graph of the function an instr_ctx() is in.
@spec cfg(map(), atom() | String.t(), arity()) :: Argus.Cfg.Function.t() | nil
The control-flow graph of the function ctx is in, from the graphs
the pipeline attached to module_data or built on the spot.
@spec disable_tracing() :: :ok
Disable imprecision tracking for the current Erlang process. Subsequent
track_* calls become no-ops. Always called from the pipeline's
try/after so the flag is cleared even on extractor errors.
@spec each_call( map(), Argus.Pipeline.Emit.facts(), (Argus.Pipeline.Emit.facts(), instr_ctx(), {module(), atom(), arity()} -> Argus.Pipeline.Emit.facts()) ) :: Argus.Pipeline.Emit.facts()
Like each_remote_call/3, for remote and local calls alike.
@spec each_remote_call( map(), Argus.Pipeline.Emit.facts(), (Argus.Pipeline.Emit.facts(), instr_ctx(), {module(), atom(), arity()} -> Argus.Pipeline.Emit.facts()) ) :: Argus.Pipeline.Emit.facts()
Calls handler.(facts, ctx, {mod, func, arity}) for every remote call
in the module, from the call-site index the pipeline attached (or one
built on the spot). ctx is the same instr_ctx() the per-instruction
scanners pass, so resolve_register/3 and friends work unchanged.
@spec enable_tracing() :: :ok
Enable imprecision tracking for the current Erlang process. Subsequent
track_imprecision/5 and track_dynamic/5 calls will record events.
@spec find_function([term()], atom(), non_neg_integer()) :: [term()] | nil
Find a function's instruction list by name and arity.
Returns the instruction list, or nil if the function is not found.
Extract behaviour modules from a module's attributes.
Handles both :behaviour and :behavior spellings.
@spec instructions_from_label([term()], non_neg_integer()) :: [term()]
Return instructions starting from a given label number.
Scans forward through the instruction list for {:label, label_num} and
returns all instructions from that label onward (inclusive). Returns []
if the label is not found.
@spec keyword_value_register([term()], non_neg_integer(), register(), atom()) :: {:ok, register(), non_neg_integer()} | :no
Find the register holding key's value in a keyword list built at
runtime and pointed to by list_reg at instruction idx.
Returns {:ok, value_register, value_idx} — the register that holds the
value and the index where the {key, value} pair was constructed — or
:no. This is the provenance hook for reading a runtime option's
source: e.g. a {DynamicSupervisor, name: some_call(...)} child spec
whose :name is computed, where you want to trace the value back to the
call that produced it (via call_result_origin/3).
Walks the cons cells (put_list) and pair tuples (put_tuple2) of the
list, following move chains. Only pairs whose value is a register
match — a literal value has no register to return (use
resolve_register/3 for those).
Match a BEAM instruction as a local (intra-module) function call.
Returns {:ok, module, function, arity} for call, call_only, and
call_last instructions with MFA targets, or :none for anything else.
Match a BEAM instruction as a remote (external) function call.
Returns {:ok, module, function, arity} for call_ext, call_ext_only,
and call_ext_last instructions, or :none for anything else.
@spec module_target([tuple()], non_neg_integer(), register()) :: String.t()
The process a call is addressed to, as every target column spells it:
the inspected module atom, "via:Registry" for a via tuple naming a
registry, or "dynamic".
@spec recent_writer([term()], non_neg_integer(), register()) :: {:ok, term(), non_neg_integer()} | :no
The instruction that most recently wrote register strictly before
idx, as {:ok, instruction, writer_idx}, or :no.
Unlike resolve_register/3 (which reconstructs a value), this returns
the raw writer, so callers can inspect provenance — was it a put_list,
a put_tuple2, a move, a call? Honors the same control-flow barriers
(tail calls, return) and register lifetimes (a non-x0 x register does
not survive a call) as the resolution walkers, so a writer reported here
is reachable on the path to idx. Moves are returned as-is — the caller
decides whether to keep following the chain.
A register operand with its type annotation stripped: {:tr, reg, type}
becomes reg. Seven extractors carried a copy of this clause.
@spec resolve_atom([tuple()], non_neg_integer(), register()) :: String.t()
Resolve register at instruction idx and return its inspected atom
string, or "dynamic" if the value cannot be statically determined or
is not an atom.
Resolve {:x, 0} at the current instruction context, returning the
inspected atom or "dynamic".
This is the standard pattern for extracting the target module/atom from the first argument of a remote call.
@spec resolve_register([term()], non_neg_integer(), register()) :: {:ok, term()} | :dynamic
Resolve the value of register at instruction index call_idx by walking
backward through the instruction list.
Handles moves (atom, literal, integer, register-to-register), put_list
chains (cons cell construction), put_tuple2 (tuple construction),
put_map_assoc/put_map_exact (map construction),
get_map_elements (map pattern matching), and typed register
wrappers ({:tr, reg, type}).
Returns {:ok, term} with the reconstructed Elixir value, or :dynamic
when the value cannot be statically determined. Partially resolvable
structures use :dynamic as a placeholder for unknown components
(e.g. {:ok, {:heir, :dynamic, nil}}).
Function parameters and pattern-matched-destructure-of-call-result are
represented via separate helpers (arg_position/3 and the
{:call_field, mfa, idx} shape returned for get_tuple_element of a
call result) to keep this function's value contract free of markers.
@spec resolve_to_arg_or_atom([term()], non_neg_integer(), register()) :: {:atom, String.t()} | {:arg, non_neg_integer()} | :dynamic
Resolve register at instruction idx and classify the result as a
literal atom, a function parameter, or dynamic.
Returns one of:
{:atom, inspected}— the register holds a literal atom (the value isinspect/1'd so it's safe to use as a fact field){:arg, n}— the register is the n-th function parameter:dynamic— the value cannot be statically determined
This is the right helper for extractors that need to distinguish "this call goes to a known module" from "this call goes to a parameter we could correlate via the call graph" from "we have no idea".
@spec return_shapes([tuple()]) :: [{non_neg_integer(), [term()]}]
Every tuple a function returns, as {index, elements}: built by
put_tuple2 (into {x,0}, or into a register moved to {x,0} before
the return), by the pre-OTP-24 put_tuple/put sequence, or folded by
the compiler into one literal moved into {x,0}. Elements are in the
instruction vocabulary — {:atom, a}, {:integer, n}, {:literal, t},
a register — whatever their source, so a rule reading [{:atom, :ok} | rest] reads all three shapes.
@spec scan_functions( module(), [tuple()], Argus.Pipeline.Emit.facts(), (Argus.Pipeline.Emit.facts(), instr_ctx(), tuple() -> Argus.Pipeline.Emit.facts()) ) :: Argus.Pipeline.Emit.facts()
Walk every function in functions and every instruction within each
function, calling handler.(facts, ctx, instr) for each instruction.
ctx is a map with :func_id, :instrs, and :idx — everything an
extractor needs to call resolve_register/3 on the surrounding code.
This is the standard outer loop for instruction-driven extractors.
@spec timeout_ms([tuple()], non_neg_integer(), register()) :: String.t()
A timeout argument as the schema spells it: the milliseconds, "-1"
for :infinity, "0" when it could not be read.
@spec tracing_enabled?() :: boolean()
Returns whether imprecision tracking is currently enabled for this process.
Useful in tests; production code should just call the track_* helpers
and rely on them to no-op when tracing is off.
@spec track_dynamic( Argus.Pipeline.Emit.facts(), term(), instr_ctx(), atom(), atom() ) :: Argus.Pipeline.Emit.facts()
Conditional wrapper around track_imprecision/5. When value indicates
a dynamic fallback (the string "dynamic" or the atom :dynamic),
records the event. No-op otherwise AND no-op when tracing is disabled.
This is the right helper for wrapping an existing resolve_callee /
resolve_atom call site — the wrapper is essentially free in the
non-coverage case (a single Process.get/2).
@spec track_imprecision( Argus.Pipeline.Emit.facts(), instr_ctx(), atom(), atom(), atom() | String.t() ) :: Argus.Pipeline.Emit.facts()
Record an imprecision event explicitly. Use directly when the extractor decided to skip a fact emission entirely (the "intentional skip" case is still information — the value was missing, not just dynamic).
No-op unless tracing is enabled for the current process.
@spec value_at([tuple()], non_neg_integer(), register()) :: {:literal, term()} | {:arg, non_neg_integer()} | {:call_result, {module(), atom(), arity()}, non_neg_integer()} | :dynamic
What register holds at idx, in one verdict: a literal, a function
parameter, the result of a call, or nothing knowable. The resolution
cascade that four extractors each wrote out.