Skip to content

⚡ Bolt: [Optimize yEnc decoding] #116

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
xbmc4lyfe wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

⚡ Bolt: [Optimize yEnc decoding] #116

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 106 additions & 35 deletions verify_nzb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,19 +115,27 @@
return attrs


_YENC_TRANS_TABLE = bytes((i - 42) % 256 for i in range(256))


def _decode_yenc_lines(lines: Iterable[bytes]) -> bytes:
decoded = bytearray()
for line in lines:
index = 0
while index < len(line):
byte = line[index]
if byte == 61:
index += 1
if index >= len(line):
raise ValueError("dangling yEnc escape")
byte = (line[index] - 64) % 256
decoded.append((byte - 42) % 256)
index += 1
pos = 0
while True:
esc_pos = line.find(b"=") if pos == 0 else line.find(b"=", pos)
if esc_pos == -1:
decoded.extend(line[pos:].translate(_YENC_TRANS_TABLE))
break

decoded.extend(line[pos:esc_pos].translate(_YENC_TRANS_TABLE))
if esc_pos + 1 >= len(line):
raise ValueError("dangling yEnc escape")

# (char - 64 - 42) % 256 => (char - 106) % 256
decoded.append((line[esc_pos + 1] - 106) % 256)
pos = esc_pos + 2

return bytes(decoded)


Expand All @@ -138,7 +146,9 @@
data_lines: list[bytes] = []

for raw_line in lines:
line = raw_line.encode("latin-1") if isinstance(raw_line, str) else bytes(raw_line)
line = (
raw_line.encode("latin-1") if isinstance(raw_line, str) else bytes(raw_line)
)
line = line.rstrip(b"\r\n")
if line.startswith(b"=ybegin"):
ybegin_attrs = _parse_yenc_attrs(line)
Expand Down Expand Up @@ -235,7 +245,9 @@
unique_ids = list(dict.fromkeys(message_ids))
if not unique_ids:
return []
sample_size = min(len(unique_ids), max(1, math.ceil(len(unique_ids) * sample_percent / 100)))
sample_size = min(
len(unique_ids), max(1, math.ceil(len(unique_ids) * sample_percent / 100))
)
if sample_size == len(unique_ids):
return unique_ids
return random.Random(sample_seed).sample(unique_ids, sample_size)
Expand Down Expand Up @@ -271,9 +283,13 @@
if value is None
]
if missing:
raise ValueError(f"missing required options in [{section}]: {', '.join(missing)}")
raise ValueError(
f"missing required options in [{section}]: {', '.join(missing)}"
)
if (username is None) ^ (password is None):
raise ValueError(f"[{section}] must set both username and password or neither")
raise ValueError(
f"[{section}] must set both username and password or neither"
)

Check warning on line 292 in verify_nzb.py

View check run for this annotation

Codacy Production / Codacy Static Code Analysis

verify_nzb.py#L292 

Operator "<" not supported for "None" (reportOptionalOperand)
if max_connections < 1:
raise ValueError(f"[{section}] max_connections must be at least 1")
if port < 1:
Expand All @@ -295,7 +311,9 @@
)

if not servers:
raise ValueError("configuration must contain at least one [server.<name>] section")
raise ValueError(
"configuration must contain at least one [server.<name>] section"
)
return servers


Expand Down Expand Up @@ -413,13 +431,20 @@
raise ProtocolNntpError("NNTP command is not ASCII encodable") from exc
except asyncio.TimeoutError as exc:
raise TransientNntpError("command timeout") from exc
except (ConnectionResetError, BrokenPipeError, OSError, asyncio.IncompleteReadError) as exc:
except (
ConnectionResetError,
BrokenPipeError,
OSError,
asyncio.IncompleteReadError,
) as exc:
raise TransientNntpError("connection lost") from exc

async def _read_response(self) -> tuple[int, str]:
assert self._reader is not None
try:
line = await asyncio.wait_for(self._reader.readline(), timeout=self.config.timeout)
line = await asyncio.wait_for(
self._reader.readline(), timeout=self.config.timeout
)
Comment on lines +434 to +447

@coderabbitai coderabbitai Bot Jul 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
python - <<'PY'
import asyncio

print(asyncio.IncompleteReadError.__mro__)
assert not issubclass(asyncio.IncompleteReadError, OSError)
PY

rg -n -C 4 'async def _read_response|async def _read_multiline|readline\(\)|IncompleteReadError|TransientNntpError' verify_nzb.py

Repository: Appz4Fun/cyclops

Length of output: 4147

🏁 Script executed:

#!/bin/bash
sed -n '366,405p' verify_nzb.py
printf '\n----\n'
sed -n '500,575p' verify_nzb.py

Repository: Appz4Fun/cyclops

Length of output: 4746

🏁 Script executed:

#!/bin/bash
rg -n -C 4 '_read_multiline|_read_response|TransientNntpError|_retry|_body_once|_connect_once' verify_nzb.py

Repository: Appz4Fun/cyclops

Length of output: 6397

🏁 Script executed:

#!/bin/bash
python3 - <<'PY'
import asyncio, inspect
from asyncio import streams

print("StreamReader.readline source:")
print(inspect.getsource(asyncio.StreamReader.readline))
print("\nreaduntil source excerpt:")
print(inspect.getsource(asyncio.StreamReader.readuntil))
PY

Repository: Appz4Fun/cyclops

Length of output: 5464

🏁 Script executed:

#!/bin/bash
python3 - <<'PY'
import asyncio, inspect

print(inspect.getsource(asyncio.StreamReader._wait_for_data))
PY

Repository: Appz4Fun/cyclops

Length of output: 1259

🏁 Script executed:

#!/bin/bash
python3 - <<'PY'
import asyncio, inspect

# Show how StreamReader gets exceptions from the protocol
print("StreamReaderProtocol.connection_lost:")
print(inspect.getsource(asyncio.StreamReaderProtocol.connection_lost))

print("\nStreamReader.set_exception:")
print(inspect.getsource(asyncio.StreamReader.set_exception))
PY

Repository: Appz4Fun/cyclops

Length of output: 1031

Translate body read failures in _read_multiline()
_connect_once() already maps greeting read errors, and asyncio.StreamReader.readline() returns b"" on EOF rather than raising IncompleteReadError. The remaining gap is _read_multiline(): a transport exception during BODY reads can still escape as a raw error and bypass _retry(). Wrap that readline() call the same way as _send_command() so transient disconnects stay retryable.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@verify_nzb.py` around lines 434 - 447, Update _read_multiline() to wrap its
BODY readline operation with the same transient-connection exception translation
used by _send_command(), converting transport failures into TransientNntpError
so _retry() can handle them. Also handle the EOF result b"" consistently with
the existing multiline-read behavior.

except asyncio.TimeoutError as exc:
raise TransientNntpError("read timeout") from exc
if not line:
Expand All @@ -434,7 +459,9 @@
lines: list[bytes] = []
while True:
try:
line = await asyncio.wait_for(self._reader.readline(), timeout=self.config.timeout)
line = await asyncio.wait_for(
self._reader.readline(), timeout=self.config.timeout
)
except asyncio.TimeoutError as exc:
raise TransientNntpError("read timeout") from exc
if not line:
Expand Down Expand Up @@ -494,7 +521,9 @@
self._progress_was_written = False
self._progress_failed = False

async def run(self, message_ids: Iterable[str], missing_output: str | Path | None = None) -> VerificationSummary:
async def run(
self, message_ids: Iterable[str], missing_output: str | Path | None = None
) -> VerificationSummary:
start = time.monotonic()
try:
await self._start_workers()
Expand Down Expand Up @@ -522,7 +551,9 @@
present=self.present,
missing=self.missing,
error=self.error,
stat_requests=sum(conn.request_count for server in self.connections for conn in server),
stat_requests=sum(
conn.request_count for server in self.connections for conn in server
),
elapsed_seconds=elapsed,
)

Expand All @@ -534,7 +565,11 @@
self.workers.append(task)

await asyncio.gather(
*(connection.connect(self.retries) for server_connections in self.connections for connection in server_connections),
*(
connection.connect(self.retries)
for server_connections in self.connections
for connection in server_connections
),
return_exceptions=True,
)

Expand All @@ -548,14 +583,21 @@
for connection in server_connections:
await connection.close()

async def _worker_loop(self, server_index: int, connection: AsyncNntpConnection) -> None:
async def _worker_loop(
self, server_index: int, connection: AsyncNntpConnection
) -> None:
try:
while True:
job = await self._take_job(server_index)
if job is None:
return
try:
await self._handle_job(server_index, connection, job.message_id, job.target_server_index)
await self._handle_job(
server_index,
connection,
job.message_id,
job.target_server_index,
)
except Exception:
async with self.job_condition:
state = self.states.get(job.message_id)
Expand All @@ -581,7 +623,10 @@

def _find_job_for_server(self, server_index: int) -> _Job | None:
for job in self.jobs:
if job.target_server_index is None or job.target_server_index == server_index:
if (
job.target_server_index is None
or job.target_server_index == server_index
):
return job
return None

Expand Down Expand Up @@ -632,7 +677,9 @@
if next_index is not None:
self._defer_message_locked(message_id, next_index)
return
await self._finalize_locked(message_id, "error" if state.had_error else "missing")
await self._finalize_locked(
message_id, "error" if state.had_error else "missing"
)
return
state.had_error = True
if next_index is not None:
Expand Down Expand Up @@ -660,7 +707,9 @@
self.jobs.append(_Job(message_id=message_id, target_server_index=server_index))
self.job_condition.notify_all()

def _next_server_index_locked(self, state: _MessageState, current_server_index: int) -> int | None:
def _next_server_index_locked(
self, state: _MessageState, current_server_index: int
) -> int | None:
total = len(self.servers)
for offset in range(1, total + 1):
candidate = (current_server_index + offset) % total
Expand Down Expand Up @@ -742,7 +791,10 @@
try:
await self._start()
results = await asyncio.gather(
*(self._check_one(index, message_id) for index, message_id in enumerate(message_ids))
*(
self._check_one(index, message_id)
for index, message_id in enumerate(message_ids)
)
)
finally:
await self._stop()
Expand All @@ -751,7 +803,9 @@
with Path(deep_output).open("w", encoding="utf-8") as handle:
for result in results:
server = result.server or "-"
handle.write(f"{result.message_id}\t{result.status}\t{result.detail}\t{server}\n")
handle.write(
f"{result.message_id}\t{result.status}\t{result.detail}\t{server}\n"
)

elapsed = time.monotonic() - start
return DeepCheckSummary(
Expand All @@ -769,7 +823,11 @@

async def _start(self) -> None:
await asyncio.gather(
*(connection.connect(self.retries) for server_connections in self.connections for connection in server_connections),
*(
connection.connect(self.retries)
for server_connections in self.connections
for connection in server_connections
),
return_exceptions=True,
)
for server_index, server_connections in enumerate(self.connections):
Expand Down Expand Up @@ -814,7 +872,9 @@
finally:
self.connection_queues[server_index].put_nowait(connection)

return DeepCheckResult(message_id=message_id, status="error", detail=last_error, server=None)
return DeepCheckResult(
message_id=message_id, status="error", detail=last_error, server=None
)


async def verify_nzb(
Expand All @@ -827,11 +887,13 @@
sample_percent: float = 1.0,
sample_seed: int | None = None,
deep_output: str | Path | None = None,
progress_stream = sys.stdout,
progress_stream=sys.stdout,
) -> VerificationSummary:
servers = load_config(config_path)
verifier = _Verifier(servers, retries=retries, progress_stream=progress_stream)
summary = await verifier.run(parse_nzb_message_ids(nzb_path), missing_output=missing_output)
summary = await verifier.run(
parse_nzb_message_ids(nzb_path), missing_output=missing_output
)
if deep_check:
sampled_ids = select_deep_sample(
verifier.present_message_ids,
Expand All @@ -844,10 +906,19 @@


def build_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Verify NZB message IDs against NNTP servers.")
parser = argparse.ArgumentParser(
description="Verify NZB message IDs against NNTP servers."
)
parser.add_argument("nzb_path", type=Path, help="path to the NZB file")
parser.add_argument("--config", required=True, type=Path, help="path to the NNTP INI file")
parser.add_argument("--retries", type=int, default=1, help="retry count for transient network errors")
parser.add_argument(
"--config", required=True, type=Path, help="path to the NNTP INI file"
)
parser.add_argument(
"--retries",
type=int,
default=1,
help="retry count for transient network errors",
)
parser.add_argument(
"--missing-output",
type=Path,
Expand Down