Skip to content

⚡ Bolt: [performance improvement] yEnc Decoding Optimization #164

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: [performance improvement] yEnc Decoding Optimization #164

Show file tree
Hide file tree
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
3 changes: 3 additions & 0 deletions .jules/bolt.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
## 2026-08-07 - yEnc Decoding Optimization

Check notice on line 1 in .jules/bolt.md

View check run for this annotation

Codacy Production / Codacy Static Code Analysis

.jules/bolt.md#L1 

Expected: 1; Actual: 0; Below
**Learning:** Built-in C-backed string/bytes functions like bytes.translate and bytes.find are significantly faster than Python-level loops for byte manipulations.
**Action:** Prioritize built-in C-backed functions over manual iteration for hot loops in parsing.
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))
_YENC_ESCAPE_TRANS_TABLE = bytes((i - 106) % 256 for i in range(256))


def _decode_yenc_lines(lines: Iterable[bytes]) -> bytes:
# ⚡ Bolt: Fast yEnc decoding using C-backed bytes.translate and bytes.find
# Expected impact: ~10x faster decoding by avoiding Python-level byte iteration
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:
escape_pos = line.find(b"=", pos)
if escape_pos == -1:
decoded.extend(line[pos:].translate(_YENC_TRANS_TABLE))
break
if escape_pos > pos:
decoded.extend(line[pos:escape_pos].translate(_YENC_TRANS_TABLE))
if escape_pos + 1 >= len(line):
raise ValueError("dangling yEnc escape")
decoded.append(_YENC_ESCAPE_TRANS_TABLE[line[escape_pos + 1]])
pos = escape_pos + 2
Comment on lines +118 to +138

@coderabbitai coderabbitai Bot Aug 7, 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.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

rg -n -C 4 \
  'yenc_body|_decode_yenc_lines|dangling yEnc escape|crc32|escape|0x00|0x0a|0x0d|0x3d' \
  tests --glob '*.py'

Repository: Appz4Fun/cyclops

Length of output: 3944

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '--- relevant verify_nzb.py sections ---\n'
sed -n '1,180p' verify_nzb.py

printf '\n--- yenc_encode implementation ---\n'
sed -n '130,165p' tests/test_verify_nzb.py

printf '\n--- yenc-related tests ---\n'
sed -n '230,280p' tests/test_verify_nzb.py

printf '\n--- focused search for tests covering escape symbols/bytes ---\n'
rg -n -C 3 'decode_yenc|validation_rejects|dangling|escape|0x3d|0x00|0x0a|0x0d|0x6e|0x4f|0xff|b"=|\x3d' tests --glob '*.py'

Repository: Appz4Fun/cyclops

Length of output: 245

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo '--- relevant verify_nzb.py sections ---'
sed -n '1,180p' verify_nzb.py

echo
echo '--- yenc_encode implementation ---'
sed -n '130,165p' tests/test_verify_nzb.py

echo
echo '--- yenc-related tests ---'
sed -n '230,280p' tests/test_verify_nzb.py

echo
echo '--- focused search for tests covering escape symbols/bytes ---'
rg -n -C 3 'decode_yenc|validation_rejects|dangling|escape|0x3d|0x00|0x0a|0x0d|0x6e|0x4f|0xff|b"=|\x3d' tests --glob '*.py'

Repository: Appz4Fun/cyclops

Length of output: 50372

Add regression coverage for escaped yEnc bytes.

tests/test_verify_nzb.py only exercises normal decoding and bad-CRC rejection. Add yEnc regression tests for the escaped bytes 0x00, 0x0A, 0x0D, 0x3D, adjacent escapes, an escape at the start of a line, and a dangling trailing =.

🤖 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 118 - 138, Add regression tests in
tests/test_verify_nzb.py targeting _decode_yenc_lines for escaped 0x00, 0x0A,
0x0D, and 0x3D bytes, including adjacent escapes and an escape at the beginning
of a line. Also assert that a trailing “=” raises ValueError, while preserving
the existing normal-decoding and bad-CRC coverage.

Source: MCP tools

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
)
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