opensource
/
dify
mirror of https://github.com/langgenius/dify.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: wrong usage of redis lock (#28177) 

Signed-off-by: NeatGuyCoding <15627489+NeatGuyCoding@users.noreply.github.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Maries <xh001x@hotmail.com>
Co-authored-by: 非法操作 <hjlarry@163.com>
This commit is contained in:
NeatGuyCoding 2025-12-28 13:47:54 +08:00 committed by GitHub
parent 2b01f85d61
commit 1f2c85c916
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 22 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
api/services/trigger/webhook_service.py
View File

@ -863,10 +863,18 @@ class WebhookService:
not_found_in_cache.append(node_id) not_found_in_cache.append(node_id)
continue continue
with Session(db.engine) as session: lock_key = f"{cls.__WEBHOOK_NODE_CACHE_KEY__}:apps:{app.id}:lock"
try: lock = redis_client.lock(lock_key, timeout=10)
# lock the concurrent webhook trigger creation lock_acquired = False
redis_client.lock(f"{cls.__WEBHOOK_NODE_CACHE_KEY__}:apps:{app.id}:lock", timeout=10)
try:
# acquire the lock with blocking and timeout
lock_acquired = lock.acquire(blocking=True, blocking_timeout=10)
if not lock_acquired:
logger.warning("Failed to acquire lock for webhook sync, app %s", app.id)
raise RuntimeError("Failed to acquire lock for webhook trigger synchronization")
with Session(db.engine) as session:
# fetch the non-cached nodes from DB # fetch the non-cached nodes from DB
all_records = session.scalars( all_records = session.scalars(
select(WorkflowWebhookTrigger).where( select(WorkflowWebhookTrigger).where(
@ -903,11 +911,16 @@ class WebhookService:
session.delete(nodes_id_in_db[node_id]) session.delete(nodes_id_in_db[node_id])
redis_client.delete(f"{cls.__WEBHOOK_NODE_CACHE_KEY__}:{app.id}:{node_id}") redis_client.delete(f"{cls.__WEBHOOK_NODE_CACHE_KEY__}:{app.id}:{node_id}")
session.commit() session.commit()
except Exception: except Exception:
logger.exception("Failed to sync webhook relationships for app %s", app.id) logger.exception("Failed to sync webhook relationships for app %s", app.id)
raise raise
finally: finally:
redis_client.delete(f"{cls.__WEBHOOK_NODE_CACHE_KEY__}:apps:{app.id}:lock") # release the lock only if it was acquired
if lock_acquired:
try:
lock.release()
except Exception:
logger.exception("Failed to release lock for webhook sync, app %s", app.id)
@classmethod @classmethod
def generate_webhook_id(cls) -> str: def generate_webhook_id(cls) -> str: