Enuxia-Public/enuxia_einvoice
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: prevent concurrent inbound synchronization

Browse Source
This commit is contained in:
Julien Denizot
2026-06-19 14:41:54 +00:00
parent aeb43d6d53
commit 9b8f8761a4
3 changed files with 338 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
enuxia_einvoice/inbound_import.py
+42 -3
View File
@@ -696,9 +696,48 @@ def archive_incoming_invoice(
values["last_sync_at"] = now_datetime()
exchange = frappe.get_doc(values)
exchange.insert(
ignore_permissions=True
)
try:
exchange.insert(
ignore_permissions=True
)
except frappe.DuplicateEntryError:
# Une autre synchronisation a pu insérer
# le même external_id entre notre contrôle
# initial et cet insert.
existing = frappe.db.get_value(
"Electronic Invoice Exchange",
{
"external_id": (
values["external_id"]
),
},
"name",
)
if not existing:
raise
exchange = frappe.get_doc(
"Electronic Invoice Exchange",
existing,
)
return {
"ok": True,
"created": False,
"exchange": exchange.name,
"remote_id": remote_id,
"review_status": (
exchange.review_status
),
"supplier": exchange.supplier,
"message": _(
"Cette facture entrante a été "
"archivée par une autre "
"synchronisation."
),
}
semantic_content = json.dumps(
remote_invoice,
enuxia_einvoice/inbound_sync.py
+206 -2
View File
@@ -21,6 +21,85 @@ MAX_PAGE_SIZE = 1000
DEFAULT_MAX_PAGES = 10
MAX_MAX_PAGES = 100
# Le verrou expire avant l'exécution horaire suivante.
GLOBAL_LOCK_TIMEOUT = 55 * 60
ACCOUNT_LOCK_TIMEOUT = 55 * 60
def _site_lock_namespace() -> str:
"""Construire un espace de verrou propre au site Frappe."""
site = str(
getattr(
frappe.local,
"site",
"",
)
or ""
).strip()
if not site:
configuration = (
getattr(
frappe.local,
"conf",
{},
)
or {}
)
site = str(
configuration.get("db_name")
or "default"
)
return (
f"enuxia_einvoice:{site}:"
"inbound-sync"
)
def _acquire_nonblocking_lock(
name: str,
*,
timeout: int,
):
"""Acquérir un verrou Redis sans attendre."""
lock = frappe.cache.lock(
name,
timeout=timeout,
blocking=False,
blocking_timeout=0,
)
acquired = lock.acquire(
blocking=False
)
return lock if acquired else None
def _release_lock(lock) -> None:
"""Libérer un verrou uniquement s'il nous appartient encore."""
if lock is None:
return
try:
if lock.owned():
lock.release()
except Exception:
# Une expiration du verrou ne doit pas masquer
# le résultat principal de la synchronisation.
frappe.log_error(
message=frappe.get_traceback(),
title=(
"Libération du verrou de "
"synchronisation entrante"
),
)
def _parse_cursor(
value: Any,
@@ -95,7 +174,7 @@ def _log_sync_error(
)
def _sync_account(
def _sync_account_unlocked(
account: Any,
*,
page_size: int,
@@ -395,7 +474,78 @@ def _sync_account(
return report
def run_incoming_sync(
def _sync_account(
account: Any,
*,
page_size: int,
max_pages: int,
) -> dict[str, Any]:
"""Synchroniser un compte sous verrou distribué."""
company = str(
account.company or account.name
)
cursor = _parse_cursor(
account.get(
"inbound_last_remote_id"
)
)
lock_name = (
f"{_site_lock_namespace()}:"
f"account:{account.name}"
)
lock = _acquire_nonblocking_lock(
lock_name,
timeout=ACCOUNT_LOCK_TIMEOUT,
)
if lock is None:
cursor_text = (
str(cursor)
if cursor is not None
else ""
)
return {
"account": account.name,
"company": company,
"environment": (
account.environment
),
"ok": True,
"skipped": True,
"skip_reason": (
"account_sync_already_running"
),
"initial_remote_id": cursor_text,
"last_remote_id": cursor_text,
"pages": 0,
"listed": 0,
"processed": 0,
"created": 0,
"existing": 0,
"errors": [],
"invoices": [],
}
try:
result = _sync_account_unlocked(
account,
page_size=page_size,
max_pages=max_pages,
)
result["skipped"] = False
return result
finally:
_release_lock(lock)
def _run_incoming_sync_unlocked(
*,
company: str | None = None,
page_size: int = DEFAULT_PAGE_SIZE,
@@ -503,6 +653,60 @@ def run_incoming_sync(
}
def run_incoming_sync(
*,
company: str | None = None,
page_size: int = DEFAULT_PAGE_SIZE,
max_pages: int = DEFAULT_MAX_PAGES,
) -> dict[str, Any]:
"""Synchroniser les comptes sous verrou global."""
lock = _acquire_nonblocking_lock(
(
f"{_site_lock_namespace()}:"
"global"
),
timeout=GLOBAL_LOCK_TIMEOUT,
)
if lock is None:
return {
"ok": True,
"skipped": True,
"skip_reason": (
"global_sync_already_running"
),
"message": (
"Une synchronisation entrante "
"est déjà en cours."
),
"company_filter": str(
company or ""
).strip(),
"page_size": int(page_size),
"max_pages": int(max_pages),
"account_count": 0,
"processed": 0,
"created": 0,
"existing": 0,
"error_count": 0,
"accounts": [],
}
try:
result = _run_incoming_sync_unlocked(
company=company,
page_size=page_size,
max_pages=max_pages,
)
result["skipped"] = False
return result
finally:
_release_lock(lock)
@frappe.whitelist()
def sync_incoming_invoices(
company: str | None = None,
enuxia_einvoice/tests/test_inbound_sync.py
+90
View File
@@ -5,6 +5,7 @@ from unittest.mock import Mock, patch
from enuxia_einvoice.inbound_sync import (
_sync_account,
run_incoming_sync,
)
@@ -233,5 +234,94 @@ class TestInboundSync(
)
def test_account_sync_is_skipped_when_locked(
self,
) -> None:
account = AttrDict(
{
"name": "Test",
"company": "Test",
"environment": "Sandbox",
"inbound_last_remote_id": (
"79712"
),
}
)
with (
patch(
"enuxia_einvoice.inbound_sync."
"get_authenticated_client"
) as authenticate,
patch(
"enuxia_einvoice.inbound_sync."
"frappe"
) as frappe_mock,
):
lock = Mock()
lock.acquire.return_value = False
frappe_mock.cache.lock.return_value = (
lock
)
result = _sync_account(
account,
page_size=100,
max_pages=5,
)
self.assertTrue(result["ok"])
self.assertTrue(result["skipped"])
self.assertEqual(
result["skip_reason"],
"account_sync_already_running",
)
self.assertEqual(
result["last_remote_id"],
"79712",
)
authenticate.assert_not_called()
def test_global_sync_is_skipped_when_locked(
self,
) -> None:
with patch(
"enuxia_einvoice.inbound_sync."
"frappe"
) as frappe_mock:
lock = Mock()
lock.acquire.return_value = False
frappe_mock.cache.lock.return_value = (
lock
)
result = run_incoming_sync(
company="Test",
page_size=100,
max_pages=5,
)
self.assertTrue(result["ok"])
self.assertTrue(result["skipped"])
self.assertEqual(
result["skip_reason"],
"global_sync_already_running",
)
self.assertEqual(
result["processed"],
0,
)
frappe_mock.get_all.assert_not_called()
if __name__ == "__main__":
unittest.main()