WEBDIENSTE/rengine
1
0
Fork 0
mirror of https://github.com/yogeshojha/rengine.git synced 2026-10-20 07:11:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

perf(asset-query): the read path stops repeating work it already did

Browse Source 
Measured against the seven scans on the dev box, worst case the gov.np run
(24,028 hosts, 345,152 endpoints). Four separate causes, none of them touching
how a scan writes.

A scan-level fact was resolved per row:
- has_baseline for hosts was the last of the four baseline predicates still
  written against the outer row. service_has_baseline, vuln_has_baseline and
  endpoint_has_baseline all resolve theirs once through _per_scan; hosts never
  took the fix. Postgres reported 192,448,261 rows removed by the join filter,
  and is:new took 15,202 ms. It is 26 ms now.
- That one predicate was also what pushed the project-wide leads call past its
  20s statement_timeout, so every findings count on the Attack surface pages
  came back computed:false — absent rather than wrong, with nothing saying so.

An index that could never be reached:
- port_match joined ports to subdomains with jsonb_exists(resolved_ips, ip),
  taking the probe value from the inner table, so ix_subdomains_ips_gin was
  unusable and is:sensitive, port: and service: degraded to a nested loop over
  21k ports x 24k hosts. Resolving the address set once and probing with ?|
  returns the identical rows in 50 ms rather than 1,248 ms.

Both now take the scope every other predicate takes, so a project view evaluates
them against each row's own scan instead of correlating on target_id alone.

A derived query rebuilt per reference:
- the address and service subqueries are six- and four-CTE statements that scan
  subdomains, ports and http_assets. Attached with .subquery() they were inlined
  at every reference: twice in search, three times in facets and eighteen times
  in leads. As .cte() Postgres evaluates them once. ips/search/leads 8,360 ms ->
  79 ms, ports/search/leads 1,683 ms -> 82 ms, ips/facets 2,379 ms -> 141 ms.

A plan compiled instead of run:
- NO_JIT already existed and was applied on two of the five dimensions. A leads
  statement is fifty count branches unioned together, which clears jit_above_cost
  easily, so Postgres spent seconds compiling a plan that runs in milliseconds.
  All twelve guarded query paths set it now; verified to carry the whole win on
  its own with the server default back at jit=on.

Verified: is:new is the one predicate that only wakes up on a second scan of a
target, and every target here had exactly one — so the counts-unchanged check
was comparing zero to zero. Against a seeded baseline scan the old shape, the
new shape and the ground truth all report 125 new hosts, and so do the compiler,
the facet filter, the project-wide scope and the lead counts. Every non-zero
lead in the host, address and service grammars still equals the row count its
link opens.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013dKUV96WkQQETZBMoAG98C
This commit is contained in:
Yogesh Ojha
2026-09-10 20:31:56 +05:30
co-authored by Claude Opus 5
parent 1807ec37b3
commit a8b9304d10
7 changed files with 73 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/app/services/endpoint.py
+3
View File
@@ -572,6 +572,7 @@ class EndpointService:
if predicate is not None:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
previous, _at = await self._previous_scan(scope)
return await build_tree(
self.session,
@@ -1122,6 +1123,7 @@ class EndpointService:
if predicate is not None:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
rows = (
await self.session.execute(
base.order_by(Endpoint.path, Endpoint.host).limit(MAX_TREE_ROWS + 1)
@@ -1196,6 +1198,7 @@ class EndpointService:
if predicate is not None:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
size = max(1, min(f.size, 200))
offset = max(0, (max(f.page, 1) - 1) * size)
counted = await self.session.scalar(
api/app/services/ip_address.py
+5 -1
View File
@@ -27,6 +27,7 @@ from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.asset_query import (
NO_JIT,
STATEMENT_TIMEOUT,
IpQueryContext,
QueryScope,
@@ -262,7 +263,7 @@ class IpAddressService:
bindparam("sids", list(scope.ids), type_=ARRAY(PG_UUID(as_uuid=True))),
bindparam("sensitive_ports", SENSITIVE_PORTS, type_=ARRAY(Integer)),
)
.subquery("ip_groups")
.cte("ip_groups")
)
@staticmethod
@@ -352,6 +353,7 @@ class IpAddressService:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
counted = await self.session.scalar(
select(func.count()).select_from(base.limit(COUNT_CAP + 1).subquery())
@@ -418,6 +420,7 @@ class IpAddressService:
d, base = self._scoped(scope, f, columns=lambda d: (d.c.ip,))
ctx = self._context(scope, d, now)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
return await build_leads(
self.session,
@@ -444,6 +447,7 @@ class IpAddressService:
if predicate is not None:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
return await build_ip_groups(self.session, base, key, scope)
except DBAPIError as exc:
api/app/services/port.py
+5 -1
View File
@@ -22,6 +22,7 @@ from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.asset_query import (
NO_JIT,
STATEMENT_TIMEOUT,
QueryScope,
QuerySyntaxError,
@@ -260,7 +261,7 @@ class PortService:
bindparam("sids", list(scope.ids), type_=ARRAY(PG_UUID(as_uuid=True))),
bindparam("sensitive_ports", SENSITIVE_PORTS, type_=ARRAY(Integer)),
)
.subquery("services")
.cte("services")
)
@staticmethod
@@ -371,6 +372,7 @@ class PortService:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
counted = await self.session.scalar(
select(func.count()).select_from(base.limit(COUNT_CAP + 1).subquery())
@@ -452,6 +454,7 @@ class PortService:
d, base = self._scoped(scope, f, columns=lambda d: (d.c.id,))
ctx = self._context(scope, d, now)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
return await build_leads(
self.session,
@@ -478,6 +481,7 @@ class PortService:
if predicate is not None:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
return await build_service_groups(self.session, base, key)
except DBAPIError as exc:
api/app/services/subdomain.py
+13 -9
View File
@@ -21,6 +21,7 @@ from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.asset_query import (
NO_JIT,
STATEMENT_TIMEOUT,
QueryContext,
QueryScope,
@@ -249,10 +250,8 @@ class SubdomainService:
_status_pred = staticmethod(preds.status_class)
_cert_pred = staticmethod(preds.cert_state)
_port_exists = staticmethod(preds.port_match)
_is_new = staticmethod(preds.is_new)
def _apply_filter(self, query, f: SubdomainFilter, now: datetime):
def _apply_filter(self, query, f: SubdomainFilter, now: datetime, scope: ScopeLike):
if f.statuses:
query = query.where(or_(*[self._status_pred(s) for s in f.statuses]))
if f.tech:
@@ -268,7 +267,9 @@ class SubdomainService:
if f.cert:
query = query.where(or_(*[self._cert_pred(c, now) for c in f.cert]))
if f.services:
query = query.where(self._port_exists(Port.service_name.in_(f.services)))
query = query.where(
preds.port_match(Port.service_name.in_(f.services), scope)
)
if f.cdn == "yes":
query = query.where(Subdomain.is_cdn.is_(True))
elif f.cdn == "no":
@@ -282,7 +283,7 @@ class SubdomainService:
if f.screenshot:
query = query.where(Subdomain.screenshot_path.isnot(None))
if f.new:
query = query.where(self._is_new())
query = query.where(preds.is_new(scope))
if f.issues:
query = query.where(preds.issues(now))
return query
@@ -309,7 +310,7 @@ class SubdomainService:
base = select(Subdomain).where(
Subdomain.project_id == project_id, scope.match(Subdomain.scan_id)
)
base = self._apply_filter(base, f, now)
base = self._apply_filter(base, f, now, scope)
try:
node = parse_query(f.q)
predicate = compile_query(node, QueryContext(scope=scope, now=now))
@@ -323,6 +324,7 @@ class SubdomainService:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
counted = await self.session.scalar(
select(func.count()).select_from(base.limit(COUNT_CAP + 1).subquery())
@@ -427,8 +429,9 @@ class SubdomainService:
base = select(Subdomain.id).where(
Subdomain.project_id == project_id, scope.match(Subdomain.scan_id)
)
base = self._apply_filter(base, f, now)
base = self._apply_filter(base, f, now, scope)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
ctx = QueryContext(scope=scope, now=now)
return await build_leads(
@@ -451,7 +454,7 @@ class SubdomainService:
base = select(Subdomain.id).where(
Subdomain.project_id == project_id, scope.match(Subdomain.scan_id)
)
base = self._apply_filter(base, f, now)
base = self._apply_filter(base, f, now, scope)
try:
node = parse_query(f.q)
predicate = compile_query(node, QueryContext(scope=scope, now=now))
@@ -460,6 +463,7 @@ class SubdomainService:
if predicate is not None:
base = base.where(predicate)
await self.session.execute(text(STATEMENT_TIMEOUT))
await self.session.execute(text(NO_JIT))
try:
return await build_groups(self.session, base, key)
except DBAPIError as exc:
@@ -796,7 +800,7 @@ class SubdomainService:
sensitive = await self.session.scalar(
select(func.count())
.select_from(Subdomain)
.where(*scope, self._port_exists(Port.number.in_(_SENSITIVE_PORTS)))
.where(*scope, preds.port_match(Port.number.in_(_SENSITIVE_PORTS), scan_id))
)
ip_total = await self.session.scalar(
select(func.count())
interest/providers/correlation/provider.py
+1 -1
View File
@@ -224,7 +224,7 @@ class CorrelationProvider(InterestProvider):
.where(
Subdomain.scan_id == ctx.scan.id,
Subdomain.is_excluded.is_(False),
preds.is_new(),
preds.is_new(ctx.scan.id),
)
.limit(CAP * 4)
)
shared/services/asset_query/compiler.py
+7 -5
View File
@@ -175,7 +175,7 @@ def _flag(cmp: Compare, ctx: QueryContext):
_FLAG_BUILDERS = {
"live": lambda _ctx: preds.live(),
"web": lambda _ctx: Subdomain.http_status.isnot(None),
"new": lambda _ctx: preds.is_new(),
"new": lambda ctx: preds.is_new(ctx.scope),
"resolved": lambda _ctx: preds.resolved(),
"auth": lambda _ctx: preds.auth(),
"cdn": lambda _ctx: Subdomain.is_cdn.is_(True),
@@ -185,7 +185,7 @@ _FLAG_BUILDERS = {
"important": lambda _ctx: Subdomain.is_important.is_(True),
"wildcard": lambda _ctx: Subdomain.is_wildcard.is_(True),
"issue": lambda ctx: preds.issues(ctx.now),
"sensitive": lambda _ctx: preds.sensitive(),
"sensitive": lambda ctx: preds.sensitive(ctx.scope),
"http2": lambda ctx: preds.asset_match(
ctx.scope, HttpAsset.supports_http2.is_(True)
),
@@ -284,10 +284,12 @@ _SUBDOMAIN_BUILDERS = {
"org": lambda c, _ctx: string_match(Subdomain.asn_org, c),
"cdn": lambda c, _ctx: _cdn(c),
"waf": lambda c, _ctx: _waf(c),
"port": lambda c, _ctx: preds.port_match(
number_match(Port.number, c, int_coerce(c))
"port": lambda c, ctx: preds.port_match(
number_match(Port.number, c, int_coerce(c)), ctx.scope
),
"service": lambda c, ctx: preds.port_match(
string_match(Port.service_name, c), ctx.scope
),
"service": lambda c, _ctx: preds.port_match(string_match(Port.service_name, c)),
"cert": _cert,
"cert.expires": lambda c, ctx: date_match(
Subdomain.tls_not_after, c, ctx.now, future=True
shared/services/asset_query/predicates.py
+39 -15
View File
@@ -103,14 +103,20 @@ def cert_state(name: str, now: datetime):
return false()
def port_match(condition):
return exists().where(
and_(
Port.scan_id == Subdomain.scan_id,
condition,
func.jsonb_exists(cast(Subdomain.resolved_ips, JSONB), Port.ip),
def port_match(condition, scope: ScopeLike):
def build(scan_id):
addresses = (
select(
func.coalesce(
func.array_agg(distinct(Port.ip)), pg_array([], type_=Text)
)
)
.where(Port.scan_id == scan_id, condition)
.scalar_subquery()
)
)
return func.jsonb_exists_any(cast(Subdomain.resolved_ips, JSONB), addresses)
return _per_scan(scope_of(scope), Subdomain.scan_id, build)
def seen_earlier():
@@ -125,19 +131,37 @@ def seen_earlier():
)
def has_baseline():
def _host_baseline(scan_id):
"""Whether an earlier scan of this target recorded any host. Scan-level, never per row."""
earlier = aliased(Subdomain)
target = select(Scan.target_id).where(Scan.id == scan_id).scalar_subquery()
cutoff = (
select(func.min(Subdomain.discovered_at))
.where(Subdomain.scan_id == scan_id)
.scalar_subquery()
)
return exists(
select(1).where(
earlier.target_id == Subdomain.target_id,
earlier.scan_id != Subdomain.scan_id,
earlier.discovered_at < Subdomain.discovered_at,
earlier.target_id == target,
earlier.scan_id != scan_id,
earlier.discovered_at < cutoff,
)
)
def is_new():
return and_(has_baseline(), not_(seen_earlier()))
def has_baseline(scope: ScopeLike):
scope = scope_of(scope)
if not scope.ids:
return false()
return or_(*[_host_baseline(sid) for sid in scope.ids])
def is_new(scope: ScopeLike):
scope = scope_of(scope)
return and_(
_per_scan(scope, Subdomain.scan_id, _host_baseline),
not_(seen_earlier()),
)
def _address_cutoff(scan_id):
@@ -267,8 +291,8 @@ def resolved():
return func.jsonb_array_length(cast(Subdomain.resolved_ips, JSONB)) > 0
def sensitive():
return port_match(Port.number.in_(SENSITIVE_PORTS))
def sensitive(scope: ScopeLike):
return port_match(Port.number.in_(SENSITIVE_PORTS), scope)
def issues(now: datetime):