diff --git a/__init__.py b/__init__.py index 614a8ba..42a2648 100644 --- a/__init__.py +++ b/__init__.py @@ -30,6 +30,17 @@ def libra_stop(): except Exception as ex: logger.warning(ex) + # Close the Fava client's shared HTTP connection pool. libra_stop is + # synchronous, so schedule the close; if the loop is already gone the + # sockets die with the process anyway. + from .fava_client import _fava_client + + if _fava_client is not None: + try: + asyncio.get_event_loop().create_task(_fava_client.aclose()) + except Exception as ex: + logger.warning(f"Could not close Fava HTTP client: {ex}") + def libra_start(): """Initialize Libra extension background tasks""" diff --git a/account_sync.py b/account_sync.py index 3d82381..870d1d7 100644 --- a/account_sync.py +++ b/account_sync.py @@ -285,7 +285,11 @@ async def sync_accounts_from_beancount(force_full_sync: bool = False) -> dict: return stats -async def sync_single_account_from_beancount(account_name: str) -> bool: +async def sync_single_account_from_beancount( + account_name: str, + description: Optional[str] = None, + assume_exists: bool = False, +) -> bool: """ Sync a single account from Beancount to Libra DB. @@ -294,6 +298,13 @@ async def sync_single_account_from_beancount(account_name: str) -> bool: Args: account_name: Hierarchical account name (e.g., "Expenses:Food") + description: Description for the Libra DB row (only used with + assume_exists — otherwise read from Beancount metadata) + assume_exists: Skip the Fava existence lookup. Pass when the + caller just wrote the Open directive itself — verifying via + a second serialized get_all_accounts round-trip doubles the + latency of every account create for no information gain + (libra-#53). Returns: True if account was created/updated, False if it already existed or failed @@ -306,6 +317,22 @@ async def sync_single_account_from_beancount(account_name: str) -> bool: logger.debug(f"Account already exists: {account_name}") return False + if assume_exists: + try: + await create_account( + CreateAccount( + name=account_name, + account_type=infer_account_type_from_name(account_name), + description=description, + user_id=extract_user_id_from_account_name(account_name), + ) + ) + logger.info(f"Created account (writer-asserted): {account_name}") + return True + except Exception as e: + logger.error(f"Failed to sync account {account_name}: {e}") + return False + # Get from Beancount fava = get_fava_client() try: diff --git a/fava_client.py b/fava_client.py index a7d1703..faad8f7 100644 --- a/fava_client.py +++ b/fava_client.py @@ -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", diff --git a/tests/test_void_reject_api.py b/tests/test_void_reject_api.py index 66e2180..0f2b817 100644 --- a/tests/test_void_reject_api.py +++ b/tests/test_void_reject_api.py @@ -210,3 +210,69 @@ async def test_double_reject_returns_404_on_second_call( assert r.status_code in (200, 404), ( f"second reject should be deterministic, got {r.status_code}: {r.text}" ) + + +@pytest.mark.anyio +async def test_concurrent_approve_and_reject_are_serialized( + client, super_user_headers, configured_user, standard_accounts, +): + """Two mutations of the same ledger source file fired concurrently must + BOTH land. Before libra-#23 each endpoint did its own read-modify-write + with raw httpx and no lock, so one writer overwrote the other's change + (or 412'd on the stale checksum). Now both route through + FavaClient.transform_source_line under the global write lock. + """ + import asyncio + + _, wallet = configured_user + approve_tag = f"conc-approve-{uuid4().hex[:6]}" + reject_tag = f"conc-reject-{uuid4().hex[:6]}" + + posted = {} + for tag in (approve_tag, reject_tag): + posted[tag] = await post_expense( + client, + wallet_inkey=wallet.inkey, + user_wallet_id=wallet.id, + amount="10.00", + currency="EUR", + description=tag, + expense_account=standard_accounts["expense_food"]["name"], + ) + + # Force a Fava reload so the approve/reject lookups see both fresh + # pending entries (see #37). + await list_user_entries(client, wallet_inkey=wallet.inkey) + + r_approve, r_reject = await asyncio.gather( + client.post( + f"/libra/api/v1/entries/{posted[approve_tag]['id']}/approve", + headers=super_user_headers, + ), + client.post( + f"/libra/api/v1/entries/{posted[reject_tag]['id']}/reject", + headers=super_user_headers, + ), + ) + assert r_approve.status_code == 200, f"approve: {r_approve.text}" + assert r_reject.status_code == 200, f"reject: {r_reject.text}" + + # Both mutations must be visible: one entry voided, the other cleared + # (a cleared entry no longer matches the pending-only reject lookup). + listing = await list_user_entries(client, wallet_inkey=wallet.inkey) + entries = listing.get("entries", []) + rejected = next( + (e for e in entries if reject_tag in (e.get("description") or "")), None, + ) + assert rejected is not None and "voided" in rejected.get("tags", []), ( + f"rejected entry lost its #voided tag: {rejected}" + ) + + second_reject = await client.post( + f"/libra/api/v1/entries/{posted[approve_tag]['id']}/reject", + headers=super_user_headers, + ) + assert second_reject.status_code == 404, ( + "approved entry should no longer match the pending-only reject " + f"lookup, got {second_reject.status_code}" + ) diff --git a/views_api.py b/views_api.py index 2188d7a..1c3269f 100644 --- a/views_api.py +++ b/views_api.py @@ -2903,8 +2903,7 @@ async def api_approve_expense_entry( This updates the transaction in the Beancount file via Fava API. """ - import httpx - from .fava_client import get_fava_client + from .fava_client import ChecksumConflictError, get_fava_client fava = get_fava_client() @@ -2939,57 +2938,29 @@ async def api_approve_expense_entry( detail="Entry metadata missing filename or lineno" ) - # 3. Get the source file from Fava - async with httpx.AsyncClient(timeout=fava.timeout) as client: - response = await client.get( - f"{fava.base_url}/source", - params={"filename": filename} - ) - response.raise_for_status() - source_data = response.json()["data"] + # 3. Flip the flag under FavaClient's write lock — the whole + # read-modify-write is atomic against every other ledger writer. + old_pattern = f"{date_str} !" - sha256sum = source_data["sha256sum"] - source = source_data["source"] - lines = source.split('\n') - - # 4. Find and modify the entry at the specified line - # Line numbers are 1-indexed, list is 0-indexed - entry_line_idx = lineno - 1 - - if entry_line_idx >= len(lines): + def _approve(line: str) -> str: + if old_pattern not in line: raise HTTPException( status_code=HTTPStatus.INTERNAL_SERVER_ERROR, - detail=f"Line {lineno} not found in source file" + detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {line}" ) + return line.replace(old_pattern, f"{date_str} *", 1) - entry_line = lines[entry_line_idx] - - # Check if the line contains the pending flag pattern - old_pattern = f"{date_str} !" - if old_pattern not in entry_line: - raise HTTPException( - status_code=HTTPStatus.INTERNAL_SERVER_ERROR, - detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {entry_line}" - ) - - # Replace the flag - new_pattern = f"{date_str} *" - new_line = entry_line.replace(old_pattern, new_pattern, 1) - lines[entry_line_idx] = new_line - - # 5. Write back the modified source - new_source = '\n'.join(lines) - - update_response = await client.put( - f"{fava.base_url}/source", - json={ - "file_path": filename, - "source": new_source, - "sha256sum": sha256sum - }, - headers={"Content-Type": "application/json"} + try: + await fava.transform_source_line(filename, lineno, _approve) + except ValueError as e: + raise HTTPException( + status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e) + ) + except ChecksumConflictError: + raise HTTPException( + status_code=HTTPStatus.CONFLICT, + detail="Ledger changed concurrently; retry the approval", ) - update_response.raise_for_status() logger.info(f"Entry {entry_id} approved (flag changed to *)") @@ -3012,8 +2983,7 @@ async def api_reject_expense_entry( Adds #voided tag for audit trail while keeping the '!' flag. Voided transactions are excluded from balances but preserved in the ledger. """ - import httpx - from .fava_client import get_fava_client + from .fava_client import ChecksumConflictError, get_fava_client fava = get_fava_client() @@ -3048,50 +3018,26 @@ async def api_reject_expense_entry( detail="Entry metadata missing filename or lineno" ) - # 3. Get the source file from Fava - async with httpx.AsyncClient(timeout=fava.timeout) as client: - response = await client.get( - f"{fava.base_url}/source", - params={"filename": filename} + # 3. Add the #voided tag under FavaClient's write lock — the whole + # read-modify-write is atomic against every other ledger writer. + def _void(line: str) -> str: + if "#voided" in line: + return line # already voided — no-op + return line.rstrip() + ' #voided' + + try: + changed = await fava.transform_source_line(filename, lineno, _void) + except ValueError as e: + raise HTTPException( + status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e) ) - response.raise_for_status() - source_data = response.json()["data"] - - sha256sum = source_data["sha256sum"] - source = source_data["source"] - lines = source.split('\n') - - # 4. Find and modify the entry at the specified line - add #voided tag - entry_line_idx = lineno - 1 - - if entry_line_idx >= len(lines): - raise HTTPException( - status_code=HTTPStatus.INTERNAL_SERVER_ERROR, - detail=f"Line {lineno} not found in source file" - ) - - entry_line = lines[entry_line_idx] - - # Add #voided tag if not already present - if "#voided" not in entry_line: - # Add #voided tag to the transaction line - new_line = entry_line.rstrip() + ' #voided' - lines[entry_line_idx] = new_line - - # 5. Write back the modified source - new_source = '\n'.join(lines) - - update_response = await client.put( - f"{fava.base_url}/source", - json={ - "file_path": filename, - "source": new_source, - "sha256sum": sha256sum - }, - headers={"Content-Type": "application/json"} - ) - update_response.raise_for_status() - logger.info(f"Entry {entry_id} rejected (added #voided tag)") + except ChecksumConflictError: + raise HTTPException( + status_code=HTTPStatus.CONFLICT, + detail="Ledger changed concurrently; retry the rejection", + ) + if changed: + logger.info(f"Entry {entry_id} rejected (added #voided tag)") return { "message": f"Entry {entry_id} rejected (marked as voided)", @@ -3820,8 +3766,14 @@ async def api_admin_add_chart_account( "already_existed": True, } - # Mirror into libra DB so permissions / metadata layer sees it. - synced = await sync_single_account_from_beancount(payload.name) + # Mirror into libra DB so permissions / metadata layer sees it. We just + # wrote the Open directive ourselves, so skip the verification + # round-trip through Fava (libra-#53). + synced = await sync_single_account_from_beancount( + payload.name, + description=payload.description, + assume_exists=True, + ) return { "success": True,