aiolabs/libra
9
0
Fork 1
Code Issues 36 Pull requests 7 Projects Releases 6 Packages Wiki Activity Actions

fix(fava): serialize source mutations, share HTTP client, validate BQL input

Browse source 
Fava-client hardening cluster (CODE-REVIEW-2026-06 #7, #14, #15, #19
+ libra-#23, libra-#53):

- New FavaClient.transform_source_line does the whole
  read-checksum-modify-write under the global write lock and maps
  Fava 409/412 to ChecksumConflictError. The approve and reject
  endpoints used to do this dance with raw httpx and no lock — two
  concurrent mutations raced each other and every other ledger
  writer (libra-#23). They now route through the new method and
  translate conflicts to HTTP 409.
- update_entry_source / delete_entry raise ChecksumConflictError on
  409/412 instead of leaking raw HTTPStatusError.
- One shared httpx.AsyncClient per FavaClient (12 per-call
  instantiations removed — no more TCP handshake per request);
  closed via libra_stop. Health probes keep their 2s timeout
  per-request.
- Account names/patterns are validated against ^[A-Za-z0-9:_-]+$
  before interpolation into BQL string literals.
- The posting amount regexes are consolidated into module-level
  compiled patterns, all decimal-tolerant — the old integer-only
  SATS pattern silently dropped decimal-SATS postings (Fava's @@->@
  normalisation emits them) from balances.
- add-account no longer verifies its own write with a second
  serialized get_all_accounts round-trip (libra-#53):
  sync_single_account_from_beancount grows an assume_exists path.

New test: concurrent approve+reject must both land (was
lost-update/412 before the lock).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-07-12 15:42:44 +02:00
commit 4d63e08a69
5 changed files with 305 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

182
fava_client.py
View file

@ -20,7 +20,8 @@ See: https://github.com/beancount/fava/blob/main/src/fava/json_api.py
import asyncio
import re
import httpx
from typing import Any, Dict, List, Optional
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator, Callable, Dict, List, Optional
from decimal import Decimal
from datetime import date, datetime
from loguru import logger
@ -44,6 +45,34 @@ def _infer_target_file(account_name: str) -> str:
return "accounts/chart.beancount"
# Posting amount-string patterns shared by the balance parsers. Fava's
# @@ → @ normalisation can emit decimal SATS values, so every SATS group
# must tolerate decimals (the old integer-only pattern silently dropped
# those postings from balances).
_TOTAL_PRICE_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?[\d.]+)\s+SATS$')
_UNIT_PRICE_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$')
_FIAT_AMOUNT_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})$')
_SATS_AMOUNT_RE = re.compile(r'^(-?[\d.]+)\s+SATS')
def _sats_to_int(value: str) -> int:
"""Parse a (possibly decimal) SATS amount string to whole sats."""
return int(Decimal(value))
# Account names/patterns are interpolated into BQL string literals; restrict
# them to Beancount account characters so caller-supplied input can't break
# out of the quoted literal.
_BQL_ACCOUNT_RE = re.compile(r'^[A-Za-z0-9:_-]+$')
def _validate_bql_account(value: str) -> str:
"""Validate a value bound for interpolation into a BQL string literal."""
if not _BQL_ACCOUNT_RE.match(value):
raise ValueError(f"Invalid account name for BQL query: {value!r}")
return value
def _escape_beancount_string(value: str) -> str:
"""Escape a value for safe inclusion in a Beancount string literal.
@ -136,6 +165,27 @@ class FavaClient:
self._main_dir_cache: Optional[str] = None
self._main_dir_lock = asyncio.Lock()
# Shared HTTP client, created lazily on first use. One client
# means one connection pool instead of a TCP handshake per call.
self._http: Optional[httpx.AsyncClient] = None
@asynccontextmanager
async def _client(self) -> AsyncIterator[httpx.AsyncClient]:
"""Yield the shared HTTP client (lazily created).
Kept as a context manager so call sites read the same as the
per-call clients they replace; the client itself is NOT closed on
exit — call `aclose()` at extension shutdown.
"""
if self._http is None or self._http.is_closed:
self._http = httpx.AsyncClient(timeout=self.timeout)
yield self._http
async def aclose(self) -> None:
"""Close the shared HTTP client (extension shutdown)."""
if self._http is not None and not self._http.is_closed:
await self._http.aclose()
async def _resolve_target_file(self, target_file: str) -> str:
"""
Turn a relative include path into the absolute path fava expects.
@ -160,7 +210,7 @@ class FavaClient:
if self._main_dir_cache is None:
async with self._main_dir_lock:
if self._main_dir_cache is None:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
resp = await client.get(f"{self.base_url}/options")
resp.raise_for_status()
main_file = resp.json()["data"]["beancount_options"]["filename"]
@ -236,7 +286,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications
async with self._write_lock:
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.put(
f"{self.base_url}/add_entries",
json={"entries": [entry]},
@ -350,10 +400,11 @@ class FavaClient:
# Use sum(weight) for SATS and sum(number) for fiat
# Note: BQL doesn't support != operator, so use flag = '*' to exclude pending
_validate_bql_account(account_name)
query = f"SELECT sum(number), sum(weight) WHERE account = '{account_name}' AND flag = '*'"
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.get(
f"{self.base_url}/query",
params={"query_string": query}
@ -446,14 +497,14 @@ class FavaClient:
import re
# Try total price notation: "50.00 EUR @@ 50000 SATS"
total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str)
total_price_match = _TOTAL_PRICE_RE.match(amount_str)
# Try per-unit price notation: "50.00 EUR @ 1000.5 SATS"
unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str)
unit_price_match = _UNIT_PRICE_RE.match(amount_str)
if total_price_match:
fiat_amount = Decimal(total_price_match.group(1))
fiat_currency = total_price_match.group(2)
sats_amount = int(total_price_match.group(3))
sats_amount = _sats_to_int(total_price_match.group(3))
if fiat_currency not in fiat_balances:
fiat_balances[fiat_currency] = Decimal(0)
@ -480,8 +531,8 @@ class FavaClient:
accounts_dict[account_name]["sats"] += sats_amount
# Try simple fiat format: "50.00 EUR" (check metadata for sats)
elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str):
fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str)
elif _FIAT_AMOUNT_RE.match(amount_str):
fiat_match = _FIAT_AMOUNT_RE.match(amount_str)
if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'):
fiat_amount = Decimal(fiat_match.group(1))
fiat_currency = fiat_match.group(2)
@ -502,9 +553,9 @@ class FavaClient:
else:
# Old format: SATS with cost/price notation - extract SATS amount
sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str)
sats_match = _SATS_AMOUNT_RE.match(amount_str)
if sats_match:
sats_amount = int(sats_match.group(1))
sats_amount = _sats_to_int(sats_match.group(1))
total_sats += sats_amount
# Track per account
@ -603,14 +654,14 @@ class FavaClient:
import re
# Try total price notation: "50.00 EUR @@ 50000 SATS"
total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str)
total_price_match = _TOTAL_PRICE_RE.match(amount_str)
# Try per-unit price notation: "50.00 EUR @ 1000.5 SATS"
unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str)
unit_price_match = _UNIT_PRICE_RE.match(amount_str)
if total_price_match:
fiat_amount = Decimal(total_price_match.group(1))
fiat_currency = total_price_match.group(2)
sats_amount = int(total_price_match.group(3))
sats_amount = _sats_to_int(total_price_match.group(3))
if fiat_currency not in user_data[user_id]["fiat_balances"]:
user_data[user_id]["fiat_balances"][fiat_currency] = Decimal(0)
@ -629,8 +680,8 @@ class FavaClient:
user_data[user_id]["balance"] += sats_amount
# Try simple fiat format: "50.00 EUR" (check metadata for sats)
elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str):
fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str)
elif _FIAT_AMOUNT_RE.match(amount_str):
fiat_match = _FIAT_AMOUNT_RE.match(amount_str)
if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'):
fiat_amount = Decimal(fiat_match.group(1))
fiat_currency = fiat_match.group(2)
@ -648,9 +699,9 @@ class FavaClient:
else:
# Old format: SATS with cost/price notation
sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str)
sats_match = _SATS_AMOUNT_RE.match(amount_str)
if sats_match:
sats_amount = int(sats_match.group(1))
sats_amount = _sats_to_int(sats_match.group(1))
user_data[user_id]["balance"] += sats_amount
# Extract fiat from cost syntax or metadata (backward compatibility)
@ -683,9 +734,12 @@ class FavaClient:
True if Fava responds, False otherwise
"""
try:
async with httpx.AsyncClient(timeout=2.0) as client:
async with self._client() as client:
# Health probes stay fast regardless of the configured
# request timeout.
response = await client.get(
f"{self.base_url}/changed"
f"{self.base_url}/changed",
timeout=2.0
)
return response.status_code == 200
except Exception as e:
@ -721,12 +775,13 @@ class FavaClient:
"""
# Build Beancount query
if account_pattern:
_validate_bql_account(account_pattern)
query = f"SELECT * WHERE account ~ '{account_pattern}' ORDER BY date DESC LIMIT {limit}"
else:
query = f"SELECT * ORDER BY date DESC LIMIT {limit}"
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.get(
f"{self.base_url}/query",
params={"query_string": query}
@ -807,7 +862,7 @@ class FavaClient:
https://beancount.github.io/docs/beancount_query_language.html
"""
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.get(
f"{self.base_url}/query",
params={"query_string": query_string}
@ -1341,7 +1396,7 @@ class FavaClient:
# (BQL's SELECT DISTINCT account only returns accounts with postings)
account_names: set[str] = set()
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
for endpoint in ("balance_sheet", "income_statement"):
try:
response = await client.get(f"{self.base_url}/{endpoint}")
@ -1440,7 +1495,7 @@ class FavaClient:
params["time"] = f"{cutoff_date.isoformat()} - {today.isoformat()}"
logger.info(f"Querying journal for last {days} days (from {cutoff_date})")
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.get(f"{self.base_url}/journal", params=params)
response.raise_for_status()
result = response.json()
@ -1482,7 +1537,7 @@ class FavaClient:
sha256sum = context["sha256sum"]
"""
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.get(
f"{self.base_url}/context",
params={"entry_hash": entry_hash}
@ -1529,7 +1584,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications
async with self._write_lock:
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.put(
f"{self.base_url}/source_slice",
json={
@ -1544,11 +1599,78 @@ class FavaClient:
except httpx.HTTPStatusError as e:
logger.error(f"Fava update error: {e.response.status_code} - {e.response.text}")
if e.response.status_code in (409, 412):
raise ChecksumConflictError(
f"Entry {entry_hash} changed concurrently"
) from e
raise
except httpx.RequestError as e:
logger.error(f"Fava connection error: {e}")
raise
async def transform_source_line(
self,
filename: str,
lineno: int,
transform: Callable[[str], str],
) -> bool:
"""Atomically read-modify-write one line of a ledger source file.
Holds the global write lock across the whole read-modify-write, so
another writer can't slip in between the checksum read and the
write (libra-#23: the approve/reject endpoints used to do this
read-then-write with raw httpx and no lock).
The transform receives the current line and returns the new one;
returning it unchanged skips the write.
Returns:
True when the line was changed and written, False on a no-op.
Raises:
ValueError: lineno is outside the file.
ChecksumConflictError: an out-of-process writer changed the
file between read and write (409/412 from Fava).
"""
async with self._write_lock:
async with self._client() as client:
response = await client.get(
f"{self.base_url}/source",
params={"filename": filename},
)
response.raise_for_status()
source_data = response.json()["data"]
sha256sum = source_data["sha256sum"]
lines = source_data["source"].split("\n")
idx = lineno - 1
if idx < 0 or idx >= len(lines):
raise ValueError(f"Line {lineno} not found in {filename}")
new_line = transform(lines[idx])
if new_line == lines[idx]:
return False
lines[idx] = new_line
try:
update = await client.put(
f"{self.base_url}/source",
json={
"file_path": filename,
"source": "\n".join(lines),
"sha256sum": sha256sum,
},
headers={"Content-Type": "application/json"},
)
update.raise_for_status()
except httpx.HTTPStatusError as e:
if e.response.status_code in (409, 412):
raise ChecksumConflictError(
f"{filename} changed concurrently"
) from e
raise
return True
async def delete_entry(self, entry_hash: str, sha256sum: str) -> str:
"""
Delete an entry from the Beancount file.
@ -1571,7 +1693,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications
async with self._write_lock:
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
response = await client.delete(
f"{self.base_url}/source_slice",
params={
@ -1585,6 +1707,10 @@ class FavaClient:
except httpx.HTTPStatusError as e:
logger.error(f"Fava delete error: {e.response.status_code} - {e.response.text}")
if e.response.status_code in (409, 412):
raise ChecksumConflictError(
f"Entry {entry_hash} changed concurrently"
) from e
raise
except httpx.RequestError as e:
logger.error(f"Fava connection error: {e}")
@ -1664,7 +1790,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications
async with self._write_lock:
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
async with self._client() as client:
# Step 1: Get current source file (fresh read on each attempt)
response = await client.get(
f"{self.base_url}/source",