Fava client hardening: write-lock coverage, shared HTTP client, JSON assertions #58

Open
padreug wants to merge 2 commits from fix/fava-client-hardening into fix/settlement-balance-and-decimal
7 changed files with 320 additions and 153 deletions

View file

@ -30,6 +30,17 @@ def libra_stop():
except Exception as ex: except Exception as ex:
logger.warning(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(): def libra_start():
"""Initialize Libra extension background tasks""" """Initialize Libra extension background tasks"""

View file

@ -285,7 +285,11 @@ async def sync_accounts_from_beancount(force_full_sync: bool = False) -> dict:
return stats 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. 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: Args:
account_name: Hierarchical account name (e.g., "Expenses:Food") 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: Returns:
True if account was created/updated, False if it already existed or failed 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}") logger.debug(f"Account already exists: {account_name}")
return False 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 # Get from Beancount
fava = get_fava_client() fava = get_fava_client()
try: try:

View file

@ -115,13 +115,18 @@ def format_balance(
account: str, account: str,
amount: int, amount: int,
currency: str = "SATS" currency: str = "SATS"
) -> str: ) -> Dict[str, Any]:
""" """
Format a balance assertion directive for Beancount. Format a balance assertion directive for Fava's JSON API.
Balance assertions verify that an account has an expected balance on a specific date. Balance assertions verify that an account has an expected balance on a specific date.
They are checked automatically by Beancount when the file is loaded. They are checked automatically by Beancount when the file is loaded.
Fava's `deserialise` (fava/serialisation.py) expects
`{"t": "Balance", "amount": {"number", "currency"}, ...}` the
previous source-string return 500'd on every assertion create
(libra-#39).
Args: Args:
date_val: Date of the balance assertion date_val: Date of the balance assertion
account: Account name (e.g., "Assets:Bitcoin:Lightning") account: Account name (e.g., "Assets:Bitcoin:Lightning")
@ -129,15 +134,15 @@ def format_balance(
currency: Currency code (default: "SATS") currency: Currency code (default: "SATS")
Returns: Returns:
Beancount balance directive as a string Fava API Balance entry dict ready for `fava.add_entry`.
Example:
>>> format_balance(date(2025, 11, 10), "Assets:Bitcoin:Lightning", 1500000, "SATS")
'2025-11-10 balance Assets:Bitcoin:Lightning 1500000 SATS'
""" """
date_str = date_val.strftime('%Y-%m-%d') return {
# Two spaces between account and amount (Beancount convention) "t": "Balance",
return f"{date_str} balance {account} {amount} {currency}" "date": date_val.strftime('%Y-%m-%d'),
"account": account,
"amount": {"number": str(amount), "currency": currency},
"meta": {},
}
def format_posting_with_cost( def format_posting_with_cost(

View file

@ -20,7 +20,8 @@ See: https://github.com/beancount/fava/blob/main/src/fava/json_api.py
import asyncio import asyncio
import re import re
import httpx 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 decimal import Decimal
from datetime import date, datetime from datetime import date, datetime
from loguru import logger from loguru import logger
@ -44,6 +45,34 @@ def _infer_target_file(account_name: str) -> str:
return "accounts/chart.beancount" 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: def _escape_beancount_string(value: str) -> str:
"""Escape a value for safe inclusion in a Beancount string literal. """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_cache: Optional[str] = None
self._main_dir_lock = asyncio.Lock() 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: async def _resolve_target_file(self, target_file: str) -> str:
""" """
Turn a relative include path into the absolute path fava expects. Turn a relative include path into the absolute path fava expects.
@ -160,7 +210,7 @@ class FavaClient:
if self._main_dir_cache is None: if self._main_dir_cache is None:
async with self._main_dir_lock: async with self._main_dir_lock:
if self._main_dir_cache is None: 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 = await client.get(f"{self.base_url}/options")
resp.raise_for_status() resp.raise_for_status()
main_file = resp.json()["data"]["beancount_options"]["filename"] main_file = resp.json()["data"]["beancount_options"]["filename"]
@ -236,7 +286,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications # Acquire global write lock to serialize ledger modifications
async with self._write_lock: async with self._write_lock:
try: try:
async with httpx.AsyncClient(timeout=self.timeout) as client: async with self._client() as client:
response = await client.put( response = await client.put(
f"{self.base_url}/add_entries", f"{self.base_url}/add_entries",
json={"entries": [entry]}, json={"entries": [entry]},
@ -350,10 +400,11 @@ class FavaClient:
# Use sum(weight) for SATS and sum(number) for fiat # Use sum(weight) for SATS and sum(number) for fiat
# Note: BQL doesn't support != operator, so use flag = '*' to exclude pending # 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 = '*'" query = f"SELECT sum(number), sum(weight) WHERE account = '{account_name}' AND flag = '*'"
try: try:
async with httpx.AsyncClient(timeout=self.timeout) as client: async with self._client() as client:
response = await client.get( response = await client.get(
f"{self.base_url}/query", f"{self.base_url}/query",
params={"query_string": query} params={"query_string": query}
@ -446,14 +497,14 @@ class FavaClient:
import re import re
# Try total price notation: "50.00 EUR @@ 50000 SATS" # 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" # 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: if total_price_match:
fiat_amount = Decimal(total_price_match.group(1)) fiat_amount = Decimal(total_price_match.group(1))
fiat_currency = total_price_match.group(2) 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: if fiat_currency not in fiat_balances:
fiat_balances[fiat_currency] = Decimal(0) fiat_balances[fiat_currency] = Decimal(0)
@ -480,8 +531,8 @@ class FavaClient:
accounts_dict[account_name]["sats"] += sats_amount accounts_dict[account_name]["sats"] += sats_amount
# Try simple fiat format: "50.00 EUR" (check metadata for sats) # Try simple fiat format: "50.00 EUR" (check metadata for sats)
elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str): elif _FIAT_AMOUNT_RE.match(amount_str):
fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str) fiat_match = _FIAT_AMOUNT_RE.match(amount_str)
if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'): if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'):
fiat_amount = Decimal(fiat_match.group(1)) fiat_amount = Decimal(fiat_match.group(1))
fiat_currency = fiat_match.group(2) fiat_currency = fiat_match.group(2)
@ -502,9 +553,9 @@ class FavaClient:
else: else:
# Old format: SATS with cost/price notation - extract SATS amount # 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: if sats_match:
sats_amount = int(sats_match.group(1)) sats_amount = _sats_to_int(sats_match.group(1))
total_sats += sats_amount total_sats += sats_amount
# Track per account # Track per account
@ -603,14 +654,14 @@ class FavaClient:
import re import re
# Try total price notation: "50.00 EUR @@ 50000 SATS" # 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" # 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: if total_price_match:
fiat_amount = Decimal(total_price_match.group(1)) fiat_amount = Decimal(total_price_match.group(1))
fiat_currency = total_price_match.group(2) 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"]: if fiat_currency not in user_data[user_id]["fiat_balances"]:
user_data[user_id]["fiat_balances"][fiat_currency] = Decimal(0) user_data[user_id]["fiat_balances"][fiat_currency] = Decimal(0)
@ -629,8 +680,8 @@ class FavaClient:
user_data[user_id]["balance"] += sats_amount user_data[user_id]["balance"] += sats_amount
# Try simple fiat format: "50.00 EUR" (check metadata for sats) # Try simple fiat format: "50.00 EUR" (check metadata for sats)
elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str): elif _FIAT_AMOUNT_RE.match(amount_str):
fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str) fiat_match = _FIAT_AMOUNT_RE.match(amount_str)
if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'): if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'):
fiat_amount = Decimal(fiat_match.group(1)) fiat_amount = Decimal(fiat_match.group(1))
fiat_currency = fiat_match.group(2) fiat_currency = fiat_match.group(2)
@ -648,9 +699,9 @@ class FavaClient:
else: else:
# Old format: SATS with cost/price notation # 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: 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 user_data[user_id]["balance"] += sats_amount
# Extract fiat from cost syntax or metadata (backward compatibility) # Extract fiat from cost syntax or metadata (backward compatibility)
@ -683,9 +734,12 @@ class FavaClient:
True if Fava responds, False otherwise True if Fava responds, False otherwise
""" """
try: 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( response = await client.get(
f"{self.base_url}/changed" f"{self.base_url}/changed",
timeout=2.0
) )
return response.status_code == 200 return response.status_code == 200
except Exception as e: except Exception as e:
@ -721,12 +775,13 @@ class FavaClient:
""" """
# Build Beancount query # Build Beancount query
if account_pattern: if account_pattern:
_validate_bql_account(account_pattern)
query = f"SELECT * WHERE account ~ '{account_pattern}' ORDER BY date DESC LIMIT {limit}" query = f"SELECT * WHERE account ~ '{account_pattern}' ORDER BY date DESC LIMIT {limit}"
else: else:
query = f"SELECT * ORDER BY date DESC LIMIT {limit}" query = f"SELECT * ORDER BY date DESC LIMIT {limit}"
try: try:
async with httpx.AsyncClient(timeout=self.timeout) as client: async with self._client() as client:
response = await client.get( response = await client.get(
f"{self.base_url}/query", f"{self.base_url}/query",
params={"query_string": query} params={"query_string": query}
@ -807,7 +862,7 @@ class FavaClient:
https://beancount.github.io/docs/beancount_query_language.html https://beancount.github.io/docs/beancount_query_language.html
""" """
try: try:
async with httpx.AsyncClient(timeout=self.timeout) as client: async with self._client() as client:
response = await client.get( response = await client.get(
f"{self.base_url}/query", f"{self.base_url}/query",
params={"query_string": query_string} params={"query_string": query_string}
@ -1341,7 +1396,7 @@ class FavaClient:
# (BQL's SELECT DISTINCT account only returns accounts with postings) # (BQL's SELECT DISTINCT account only returns accounts with postings)
account_names: set[str] = set() 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"): for endpoint in ("balance_sheet", "income_statement"):
try: try:
response = await client.get(f"{self.base_url}/{endpoint}") response = await client.get(f"{self.base_url}/{endpoint}")
@ -1440,7 +1495,7 @@ class FavaClient:
params["time"] = f"{cutoff_date.isoformat()} - {today.isoformat()}" params["time"] = f"{cutoff_date.isoformat()} - {today.isoformat()}"
logger.info(f"Querying journal for last {days} days (from {cutoff_date})") 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 = await client.get(f"{self.base_url}/journal", params=params)
response.raise_for_status() response.raise_for_status()
result = response.json() result = response.json()
@ -1482,7 +1537,7 @@ class FavaClient:
sha256sum = context["sha256sum"] sha256sum = context["sha256sum"]
""" """
try: try:
async with httpx.AsyncClient(timeout=self.timeout) as client: async with self._client() as client:
response = await client.get( response = await client.get(
f"{self.base_url}/context", f"{self.base_url}/context",
params={"entry_hash": entry_hash} params={"entry_hash": entry_hash}
@ -1529,7 +1584,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications # Acquire global write lock to serialize ledger modifications
async with self._write_lock: async with self._write_lock:
try: try:
async with httpx.AsyncClient(timeout=self.timeout) as client: async with self._client() as client:
response = await client.put( response = await client.put(
f"{self.base_url}/source_slice", f"{self.base_url}/source_slice",
json={ json={
@ -1544,11 +1599,78 @@ class FavaClient:
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
logger.error(f"Fava update error: {e.response.status_code} - {e.response.text}") 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 raise
except httpx.RequestError as e: except httpx.RequestError as e:
logger.error(f"Fava connection error: {e}") logger.error(f"Fava connection error: {e}")
raise 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: async def delete_entry(self, entry_hash: str, sha256sum: str) -> str:
""" """
Delete an entry from the Beancount file. Delete an entry from the Beancount file.
@ -1571,7 +1693,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications # Acquire global write lock to serialize ledger modifications
async with self._write_lock: async with self._write_lock:
try: try:
async with httpx.AsyncClient(timeout=self.timeout) as client: async with self._client() as client:
response = await client.delete( response = await client.delete(
f"{self.base_url}/source_slice", f"{self.base_url}/source_slice",
params={ params={
@ -1585,6 +1707,10 @@ class FavaClient:
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
logger.error(f"Fava delete error: {e.response.status_code} - {e.response.text}") 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 raise
except httpx.RequestError as e: except httpx.RequestError as e:
logger.error(f"Fava connection error: {e}") logger.error(f"Fava connection error: {e}")
@ -1664,7 +1790,7 @@ class FavaClient:
# Acquire global write lock to serialize ledger modifications # Acquire global write lock to serialize ledger modifications
async with self._write_lock: async with self._write_lock:
try: 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) # Step 1: Get current source file (fresh read on each attempt)
response = await client.get( response = await client.get(
f"{self.base_url}/source", f"{self.base_url}/source",

View file

@ -18,19 +18,6 @@ from uuid import uuid4
import pytest import pytest
# Tests that try to actually create + check an assertion all hit issue #39:
# `format_balance` returns a Beancount source string but `fava.add_entry`
# expects a dict, so Fava 500s on every assertion-create call. The contract
# violation is on libra's side; mark these strict-xfail so they go green
# automatically once #39 lands and the format_balance return shape is fixed.
ASSERTION_CREATE_BROKEN = pytest.mark.xfail(
reason="libra/issues/39 — POST /assertions submits a Beancount source string "
"to Fava's JSON API and 500s. Drop this marker when the format_balance "
"return type is changed to a dict.",
strict=True,
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# helpers (local — assertion endpoints don't have wrapper helpers yet) # helpers (local — assertion endpoints don't have wrapper helpers yet)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -58,7 +45,6 @@ async def _create_assertion(
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio @pytest.mark.anyio
async def test_assertion_against_empty_account_passes( async def test_assertion_against_empty_account_passes(
client, super_user_headers, standard_accounts, client, super_user_headers, standard_accounts,
@ -79,7 +65,6 @@ async def test_assertion_against_empty_account_passes(
assert body.get("difference_sats", 0) == 0 assert body.get("difference_sats", 0) == 0
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio @pytest.mark.anyio
async def test_assertion_with_wrong_balance_returns_409( async def test_assertion_with_wrong_balance_returns_409(
client, super_user_headers, standard_accounts, client, super_user_headers, standard_accounts,
@ -102,7 +87,6 @@ async def test_assertion_with_wrong_balance_returns_409(
assert detail.get("difference_sats") == 999_999 or detail.get("difference_sats") == -999_999 assert detail.get("difference_sats") == 999_999 or detail.get("difference_sats") == -999_999
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio @pytest.mark.anyio
async def test_assertion_with_tolerance_accepts_small_diff( async def test_assertion_with_tolerance_accepts_small_diff(
client, super_user_headers, standard_accounts, client, super_user_headers, standard_accounts,
@ -119,7 +103,6 @@ async def test_assertion_with_tolerance_accepts_small_diff(
assert r.json().get("status") == "passed" assert r.json().get("status") == "passed"
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio @pytest.mark.anyio
async def test_list_assertions_returns_created( async def test_list_assertions_returns_created(
client, super_user_headers, standard_accounts, client, super_user_headers, standard_accounts,
@ -145,7 +128,6 @@ async def test_list_assertions_returns_created(
assert assertion_id in ids, f"created assertion {assertion_id} missing from list {ids}" assert assertion_id in ids, f"created assertion {assertion_id} missing from list {ids}"
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio @pytest.mark.anyio
async def test_get_assertion_by_id( async def test_get_assertion_by_id(
client, super_user_headers, standard_accounts, client, super_user_headers, standard_accounts,
@ -167,7 +149,6 @@ async def test_get_assertion_by_id(
assert r.json().get("id") == assertion_id assert r.json().get("id") == assertion_id
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio @pytest.mark.anyio
async def test_recheck_assertion_via_check_endpoint( async def test_recheck_assertion_via_check_endpoint(
client, super_user_headers, standard_accounts, client, super_user_headers, standard_accounts,
@ -190,7 +171,6 @@ async def test_recheck_assertion_via_check_endpoint(
assert r.json().get("status") == "passed" assert r.json().get("status") == "passed"
@ASSERTION_CREATE_BROKEN
@pytest.mark.anyio @pytest.mark.anyio
async def test_delete_assertion_removes_it( async def test_delete_assertion_removes_it(
client, super_user_headers, standard_accounts, client, super_user_headers, standard_accounts,

View file

@ -210,3 +210,69 @@ async def test_double_reject_returns_404_on_second_call(
assert r.status_code in (200, 404), ( assert r.status_code in (200, 404), (
f"second reject should be deterministic, got {r.status_code}: {r.text}" 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}"
)

View file

@ -2903,8 +2903,7 @@ async def api_approve_expense_entry(
This updates the transaction in the Beancount file via Fava API. This updates the transaction in the Beancount file via Fava API.
""" """
import httpx from .fava_client import ChecksumConflictError, get_fava_client
from .fava_client import get_fava_client
fava = get_fava_client() fava = get_fava_client()
@ -2939,57 +2938,29 @@ async def api_approve_expense_entry(
detail="Entry metadata missing filename or lineno" detail="Entry metadata missing filename or lineno"
) )
# 3. Get the source file from Fava # 3. Flip the flag under FavaClient's write lock — the whole
async with httpx.AsyncClient(timeout=fava.timeout) as client: # read-modify-write is atomic against every other ledger writer.
response = await client.get( old_pattern = f"{date_str} !"
f"{fava.base_url}/source",
params={"filename": filename}
)
response.raise_for_status()
source_data = response.json()["data"]
sha256sum = source_data["sha256sum"] def _approve(line: str) -> str:
source = source_data["source"] if old_pattern not in line:
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):
raise HTTPException( raise HTTPException(
status_code=HTTPStatus.INTERNAL_SERVER_ERROR, 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] try:
await fava.transform_source_line(filename, lineno, _approve)
# Check if the line contains the pending flag pattern except ValueError as e:
old_pattern = f"{date_str} !" raise HTTPException(
if old_pattern not in entry_line: status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e)
raise HTTPException( )
status_code=HTTPStatus.INTERNAL_SERVER_ERROR, except ChecksumConflictError:
detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {entry_line}" raise HTTPException(
) status_code=HTTPStatus.CONFLICT,
detail="Ledger changed concurrently; retry the approval",
# 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"}
) )
update_response.raise_for_status()
logger.info(f"Entry {entry_id} approved (flag changed to *)") 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. Adds #voided tag for audit trail while keeping the '!' flag.
Voided transactions are excluded from balances but preserved in the ledger. Voided transactions are excluded from balances but preserved in the ledger.
""" """
import httpx from .fava_client import ChecksumConflictError, get_fava_client
from .fava_client import get_fava_client
fava = get_fava_client() fava = get_fava_client()
@ -3048,50 +3018,26 @@ async def api_reject_expense_entry(
detail="Entry metadata missing filename or lineno" detail="Entry metadata missing filename or lineno"
) )
# 3. Get the source file from Fava # 3. Add the #voided tag under FavaClient's write lock — the whole
async with httpx.AsyncClient(timeout=fava.timeout) as client: # read-modify-write is atomic against every other ledger writer.
response = await client.get( def _void(line: str) -> str:
f"{fava.base_url}/source", if "#voided" in line:
params={"filename": filename} 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() except ChecksumConflictError:
source_data = response.json()["data"] raise HTTPException(
status_code=HTTPStatus.CONFLICT,
sha256sum = source_data["sha256sum"] detail="Ledger changed concurrently; retry the rejection",
source = source_data["source"] )
lines = source.split('\n') if changed:
logger.info(f"Entry {entry_id} rejected (added #voided tag)")
# 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)")
return { return {
"message": f"Entry {entry_id} rejected (marked as voided)", "message": f"Entry {entry_id} rejected (marked as voided)",
@ -3820,8 +3766,14 @@ async def api_admin_add_chart_account(
"already_existed": True, "already_existed": True,
} }
# Mirror into libra DB so permissions / metadata layer sees it. # Mirror into libra DB so permissions / metadata layer sees it. We just
synced = await sync_single_account_from_beancount(payload.name) # 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 { return {
"success": True, "success": True,