Add concurrency protection for Fava/Beancount ledger writes
This commit addresses critical race conditions when multiple requests try to write to the ledger file simultaneously. Changes: - Add global asyncio.Lock to FavaClient to serialize all write operations - Add per-user locks for finer-grained concurrency control - Wrap add_entry(), update_entry_source(), delete_entry() with write lock - Add retry logic with exponential backoff to add_account() for checksum conflicts - Add new add_entry_idempotent() method to prevent duplicate entries - Add ChecksumConflictError exception for conflict handling - Update on_invoice_paid() to use per-user locking and idempotent entry creation Fixes #4 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
e403ec223d
commit
b5c36504fb
2 changed files with 397 additions and 245 deletions
408
fava_client.py
408
fava_client.py
|
|
@ -17,6 +17,7 @@ Fava provides a REST API for:
|
|||
See: https://github.com/beancount/fava/blob/main/src/fava/json_api.py
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import httpx
|
||||
from typing import Any, Dict, List, Optional
|
||||
from decimal import Decimal
|
||||
|
|
@ -24,6 +25,11 @@ from datetime import date, datetime
|
|||
from loguru import logger
|
||||
|
||||
|
||||
class ChecksumConflictError(Exception):
|
||||
"""Raised when a Fava write operation fails due to stale checksum (concurrent modification)."""
|
||||
pass
|
||||
|
||||
|
||||
class FavaClient:
|
||||
"""
|
||||
Async client for Fava REST API.
|
||||
|
|
@ -48,6 +54,37 @@ class FavaClient:
|
|||
self.base_url = f"{self.fava_url}/{self.ledger_slug}/api"
|
||||
self.timeout = timeout
|
||||
|
||||
# Concurrency control: Global write lock to serialize all ledger modifications.
|
||||
# This prevents race conditions when multiple requests try to write to the
|
||||
# Beancount ledger file simultaneously. Without this lock, concurrent writes
|
||||
# can cause data loss, duplicate entries, or file corruption.
|
||||
#
|
||||
# Note: This serializes ALL writes which may become a bottleneck at scale.
|
||||
# For higher throughput, consider per-user locking or distributed locking.
|
||||
self._write_lock = asyncio.Lock()
|
||||
|
||||
# Per-user locks for user-specific operations (reduces contention)
|
||||
self._user_locks: Dict[str, asyncio.Lock] = {}
|
||||
|
||||
def get_user_lock(self, user_id: str) -> asyncio.Lock:
|
||||
"""
|
||||
Get or create a lock for a specific user.
|
||||
|
||||
This enables per-user locking to reduce contention when multiple users
|
||||
are making concurrent requests. User-specific operations should acquire
|
||||
this lock in addition to (or instead of) the global write lock.
|
||||
|
||||
Args:
|
||||
user_id: User ID (uses first 8 characters for consistency)
|
||||
|
||||
Returns:
|
||||
asyncio.Lock for this user
|
||||
"""
|
||||
user_key = user_id[:8]
|
||||
if user_key not in self._user_locks:
|
||||
self._user_locks[user_key] = asyncio.Lock()
|
||||
return self._user_locks[user_key]
|
||||
|
||||
async def add_entry(self, entry: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Submit a new journal entry to Fava.
|
||||
|
|
@ -88,26 +125,100 @@ class FavaClient:
|
|||
"meta": {"user_id": "abc123"}
|
||||
}
|
||||
result = await fava_client.add_entry(entry)
|
||||
|
||||
Note:
|
||||
This method acquires a global write lock to prevent concurrent
|
||||
modifications to the ledger file. All writes are serialized.
|
||||
"""
|
||||
# Acquire global write lock to serialize ledger modifications
|
||||
async with self._write_lock:
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.put(
|
||||
f"{self.base_url}/add_entries",
|
||||
json={"entries": [entry]},
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
logger.info(f"Added entry to Fava: {result.get('data', 'Unknown')}")
|
||||
return result
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava HTTP error: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
|
||||
async def add_entry_idempotent(
|
||||
self,
|
||||
entry: Dict[str, Any],
|
||||
idempotency_key: str
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Submit a journal entry with idempotency protection.
|
||||
|
||||
This method checks if an entry with the given idempotency key (as a Beancount link)
|
||||
already exists before inserting. This prevents duplicate entries when the same
|
||||
operation is retried (e.g., due to network issues or concurrent requests).
|
||||
|
||||
The idempotency key is stored as a Beancount link on the entry. Links are part
|
||||
of the entry's identity and are indexed by Beancount, making lookup efficient.
|
||||
|
||||
Args:
|
||||
entry: Beancount entry dict (same format as add_entry)
|
||||
idempotency_key: Unique key for this operation (e.g., "castle-{uuid}" or "ln-{payment_hash}")
|
||||
|
||||
Returns:
|
||||
Response from Fava if entry was created, or existing entry data if already exists
|
||||
|
||||
Example:
|
||||
# Use payment hash as idempotency key for Lightning payments
|
||||
result = await fava.add_entry_idempotent(
|
||||
entry=settlement_entry,
|
||||
idempotency_key=f"ln-{payment_hash[:16]}"
|
||||
)
|
||||
|
||||
# Use expense ID for expense entries
|
||||
result = await fava.add_entry_idempotent(
|
||||
entry=expense_entry,
|
||||
idempotency_key=f"exp-{expense_id}"
|
||||
)
|
||||
"""
|
||||
from .beancount_format import sanitize_link
|
||||
|
||||
# Sanitize the idempotency key to ensure it's a valid Beancount link
|
||||
safe_key = sanitize_link(idempotency_key)
|
||||
|
||||
# Check if entry with this link already exists
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.put(
|
||||
f"{self.base_url}/add_entries",
|
||||
json={"entries": [entry]},
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
entries = await self.get_journal_entries(days=30) # Check recent entries
|
||||
|
||||
logger.info(f"Added entry to Fava: {result.get('data', 'Unknown')}")
|
||||
return result
|
||||
for existing_entry in entries:
|
||||
existing_links = existing_entry.get("links", [])
|
||||
if safe_key in existing_links:
|
||||
logger.info(f"Entry with idempotency key '{safe_key}' already exists, skipping insert")
|
||||
return {
|
||||
"data": "Entry already exists (idempotent)",
|
||||
"existing": True,
|
||||
"entry": existing_entry
|
||||
}
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not check for existing entry with key '{safe_key}': {e}")
|
||||
# Continue anyway - Beancount will error if there's a true duplicate
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava HTTP error: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
# Add the idempotency key as a link if not already present
|
||||
if "links" not in entry:
|
||||
entry["links"] = []
|
||||
if safe_key not in entry["links"]:
|
||||
entry["links"].append(safe_key)
|
||||
|
||||
# Now add the entry (this will acquire the write lock)
|
||||
result = await self.add_entry(entry)
|
||||
result["existing"] = False
|
||||
return result
|
||||
|
||||
async def get_account_balance(self, account_name: str) -> Dict[str, Any]:
|
||||
"""
|
||||
|
|
@ -1146,6 +1257,10 @@ class FavaClient:
|
|||
Returns:
|
||||
New sha256sum after update
|
||||
|
||||
Note:
|
||||
This method acquires a global write lock to prevent concurrent
|
||||
modifications to the ledger file. All writes are serialized.
|
||||
|
||||
Example:
|
||||
# Get context
|
||||
context = await fava.get_entry_context("abc123")
|
||||
|
|
@ -1158,26 +1273,28 @@ class FavaClient:
|
|||
# Update
|
||||
new_sha256 = await fava.update_entry_source("abc123", new_source, sha256)
|
||||
"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.put(
|
||||
f"{self.base_url}/source_slice",
|
||||
json={
|
||||
"entry_hash": entry_hash,
|
||||
"source": new_source,
|
||||
"sha256sum": sha256sum
|
||||
}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
return result.get("data", "")
|
||||
# Acquire global write lock to serialize ledger modifications
|
||||
async with self._write_lock:
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.put(
|
||||
f"{self.base_url}/source_slice",
|
||||
json={
|
||||
"entry_hash": entry_hash,
|
||||
"source": new_source,
|
||||
"sha256sum": sha256sum
|
||||
}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
return result.get("data", "")
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava update error: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava update error: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
|
||||
async def delete_entry(self, entry_hash: str, sha256sum: str) -> str:
|
||||
"""
|
||||
|
|
@ -1190,36 +1307,43 @@ class FavaClient:
|
|||
Returns:
|
||||
Success message
|
||||
|
||||
Note:
|
||||
This method acquires a global write lock to prevent concurrent
|
||||
modifications to the ledger file. All writes are serialized.
|
||||
|
||||
Example:
|
||||
context = await fava.get_entry_context("abc123")
|
||||
await fava.delete_entry("abc123", context["sha256sum"])
|
||||
"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.delete(
|
||||
f"{self.base_url}/source_slice",
|
||||
params={
|
||||
"entry_hash": entry_hash,
|
||||
"sha256sum": sha256sum
|
||||
}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
return result.get("data", "")
|
||||
# Acquire global write lock to serialize ledger modifications
|
||||
async with self._write_lock:
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.delete(
|
||||
f"{self.base_url}/source_slice",
|
||||
params={
|
||||
"entry_hash": entry_hash,
|
||||
"sha256sum": sha256sum
|
||||
}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
return result.get("data", "")
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava delete error: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava delete error: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
|
||||
async def add_account(
|
||||
self,
|
||||
account_name: str,
|
||||
currencies: list[str],
|
||||
opening_date: Optional[date] = None,
|
||||
metadata: Optional[Dict[str, Any]] = None
|
||||
metadata: Optional[Dict[str, Any]] = None,
|
||||
max_retries: int = 3
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Add an account to the Beancount ledger via an Open directive.
|
||||
|
|
@ -1227,15 +1351,25 @@ class FavaClient:
|
|||
NOTE: Fava's /api/add_entries endpoint does NOT support Open directives.
|
||||
This method uses /api/source to directly edit the Beancount file.
|
||||
|
||||
This method implements optimistic concurrency control with retry logic:
|
||||
- Acquires a global write lock before modifying the ledger
|
||||
- Uses SHA256 checksum to detect concurrent modifications
|
||||
- Retries with exponential backoff on checksum conflicts
|
||||
- Re-checks if account was created by concurrent request before retrying
|
||||
|
||||
Args:
|
||||
account_name: Full account name (e.g., "Assets:Receivable:User-abc123")
|
||||
currencies: List of currencies for this account (e.g., ["EUR", "SATS"])
|
||||
opening_date: Date to open the account (defaults to today)
|
||||
metadata: Optional metadata for the account
|
||||
max_retries: Maximum number of retry attempts on checksum conflict (default: 3)
|
||||
|
||||
Returns:
|
||||
Response from Fava ({"data": "new_sha256sum", "mtime": "..."})
|
||||
|
||||
Raises:
|
||||
ChecksumConflictError: If all retry attempts fail due to concurrent modifications
|
||||
|
||||
Example:
|
||||
# Add a user's receivable account
|
||||
result = await fava.add_account(
|
||||
|
|
@ -1255,89 +1389,115 @@ class FavaClient:
|
|||
if opening_date is None:
|
||||
opening_date = date_type.today()
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
# Step 1: Get the main Beancount file path from Fava
|
||||
options_response = await client.get(f"{self.base_url}/options")
|
||||
options_response.raise_for_status()
|
||||
options_data = options_response.json()["data"]
|
||||
file_path = options_data["beancount_options"]["filename"]
|
||||
last_error = None
|
||||
|
||||
logger.debug(f"Fava main file: {file_path}")
|
||||
for attempt in range(max_retries):
|
||||
# Acquire global write lock to serialize ledger modifications
|
||||
async with self._write_lock:
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
# Step 1: Get the main Beancount file path from Fava
|
||||
options_response = await client.get(f"{self.base_url}/options")
|
||||
options_response.raise_for_status()
|
||||
options_data = options_response.json()["data"]
|
||||
file_path = options_data["beancount_options"]["filename"]
|
||||
|
||||
# Step 2: Get current source file
|
||||
response = await client.get(
|
||||
f"{self.base_url}/source",
|
||||
params={"filename": file_path}
|
||||
)
|
||||
response.raise_for_status()
|
||||
source_data = response.json()["data"]
|
||||
logger.debug(f"Fava main file: {file_path}")
|
||||
|
||||
sha256sum = source_data["sha256sum"]
|
||||
source = source_data["source"]
|
||||
# Step 2: Get current source file (fresh read on each attempt)
|
||||
response = await client.get(
|
||||
f"{self.base_url}/source",
|
||||
params={"filename": file_path}
|
||||
)
|
||||
response.raise_for_status()
|
||||
source_data = response.json()["data"]
|
||||
|
||||
# Step 2: Check if account already exists
|
||||
if f"open {account_name}" in source:
|
||||
logger.info(f"Account {account_name} already exists in Beancount file")
|
||||
return {"data": sha256sum, "mtime": source_data.get("mtime", "")}
|
||||
sha256sum = source_data["sha256sum"]
|
||||
source = source_data["source"]
|
||||
|
||||
# Step 3: Find insertion point (after last Open directive AND its metadata)
|
||||
lines = source.split('\n')
|
||||
insert_index = 0
|
||||
for i, line in enumerate(lines):
|
||||
if line.strip().startswith(('open ', f'{opening_date.year}-')) and 'open' in line:
|
||||
# Found an Open directive, now skip over any metadata lines
|
||||
insert_index = i + 1
|
||||
# Skip metadata lines (lines starting with whitespace)
|
||||
while insert_index < len(lines) and lines[insert_index].startswith((' ', '\t')) and lines[insert_index].strip():
|
||||
insert_index += 1
|
||||
# Step 3: Check if account already exists (may have been created by concurrent request)
|
||||
if f"open {account_name}" in source:
|
||||
logger.info(f"Account {account_name} already exists in Beancount file")
|
||||
return {"data": sha256sum, "mtime": source_data.get("mtime", "")}
|
||||
|
||||
# Step 4: Format Open directive as Beancount text
|
||||
currencies_str = ", ".join(currencies)
|
||||
open_lines = [
|
||||
"",
|
||||
f"{opening_date.isoformat()} open {account_name} {currencies_str}"
|
||||
]
|
||||
# Step 4: Find insertion point (after last Open directive AND its metadata)
|
||||
lines = source.split('\n')
|
||||
insert_index = 0
|
||||
for i, line in enumerate(lines):
|
||||
if line.strip().startswith(('open ', f'{opening_date.year}-')) and 'open' in line:
|
||||
# Found an Open directive, now skip over any metadata lines
|
||||
insert_index = i + 1
|
||||
# Skip metadata lines (lines starting with whitespace)
|
||||
while insert_index < len(lines) and lines[insert_index].startswith((' ', '\t')) and lines[insert_index].strip():
|
||||
insert_index += 1
|
||||
|
||||
# Add metadata if provided
|
||||
if metadata:
|
||||
for key, value in metadata.items():
|
||||
# Format metadata with proper indentation
|
||||
if isinstance(value, str):
|
||||
open_lines.append(f' {key}: "{value}"')
|
||||
else:
|
||||
open_lines.append(f' {key}: {value}')
|
||||
# Step 5: Format Open directive as Beancount text
|
||||
currencies_str = ", ".join(currencies)
|
||||
open_lines = [
|
||||
"",
|
||||
f"{opening_date.isoformat()} open {account_name} {currencies_str}"
|
||||
]
|
||||
|
||||
# Step 5: Insert into source
|
||||
for i, line in enumerate(open_lines):
|
||||
lines.insert(insert_index + i, line)
|
||||
# Add metadata if provided
|
||||
if metadata:
|
||||
for key, value in metadata.items():
|
||||
# Format metadata with proper indentation
|
||||
if isinstance(value, str):
|
||||
open_lines.append(f' {key}: "{value}"')
|
||||
else:
|
||||
open_lines.append(f' {key}: {value}')
|
||||
|
||||
new_source = '\n'.join(lines)
|
||||
# Step 6: Insert into source
|
||||
for i, line in enumerate(open_lines):
|
||||
lines.insert(insert_index + i, line)
|
||||
|
||||
# Step 6: Update source file via PUT /api/source
|
||||
update_payload = {
|
||||
"file_path": file_path,
|
||||
"source": new_source,
|
||||
"sha256sum": sha256sum
|
||||
}
|
||||
new_source = '\n'.join(lines)
|
||||
|
||||
response = await client.put(
|
||||
f"{self.base_url}/source",
|
||||
json=update_payload,
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
# Step 7: Update source file via PUT /api/source
|
||||
update_payload = {
|
||||
"file_path": file_path,
|
||||
"source": new_source,
|
||||
"sha256sum": sha256sum
|
||||
}
|
||||
|
||||
logger.info(f"Added account {account_name} to Beancount file with currencies {currencies}")
|
||||
return result
|
||||
response = await client.put(
|
||||
f"{self.base_url}/source",
|
||||
json=update_payload,
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Fava HTTP error adding account: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
logger.info(f"Added account {account_name} to Beancount file with currencies {currencies}")
|
||||
return result
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
# Check for checksum conflict (HTTP 412 Precondition Failed or similar)
|
||||
if e.response.status_code in (409, 412):
|
||||
last_error = ChecksumConflictError(
|
||||
f"Checksum conflict on attempt {attempt + 1}/{max_retries}: {e.response.text}"
|
||||
)
|
||||
logger.warning(
|
||||
f"Checksum conflict adding account {account_name} "
|
||||
f"(attempt {attempt + 1}/{max_retries}), retrying..."
|
||||
)
|
||||
# Continue to retry logic below
|
||||
else:
|
||||
logger.error(f"Fava HTTP error adding account: {e.response.status_code} - {e.response.text}")
|
||||
raise
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"Fava connection error: {e}")
|
||||
raise
|
||||
|
||||
# If we get here due to checksum conflict, wait with exponential backoff before retry
|
||||
if attempt < max_retries - 1:
|
||||
backoff_time = 0.1 * (2 ** attempt) # 0.1s, 0.2s, 0.4s
|
||||
logger.info(f"Waiting {backoff_time}s before retry...")
|
||||
await asyncio.sleep(backoff_time)
|
||||
|
||||
# All retries exhausted
|
||||
logger.error(f"Failed to add account {account_name} after {max_retries} attempts due to concurrent modifications")
|
||||
raise last_error or ChecksumConflictError(f"Failed to add account after {max_retries} attempts")
|
||||
|
||||
async def get_unsettled_entries_bql(
|
||||
self,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue