port-labs
diff --git a/‎CHANGELOG.md‎
Lines changed: 16 additions & 0 deletions b/‎CHANGELOG.md‎
Lines changed: 16 additions & 0 deletions
diff --git a/‎port_ocean/config/settings.py‎
Lines changed: 2 additions & 0 deletions b/‎port_ocean/config/settings.py‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎port_ocean/core/handlers/queue/__init__.py‎
Lines changed: 2 additions & 1 deletion b/‎port_ocean/core/handlers/queue/__init__.py‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎port_ocean/core/handlers/queue/abstract_queue.py‎
Lines changed: 8 additions & 0 deletions b/‎port_ocean/core/handlers/queue/abstract_queue.py‎
Lines changed: 8 additions & 0 deletions
diff --git a/‎port_ocean/core/handlers/queue/group_queue.py‎
Lines changed: 138 additions & 0 deletions b/‎port_ocean/core/handlers/queue/group_queue.py‎
Lines changed: 138 additions & 0 deletions
diff --git a/‎port_ocean/core/handlers/queue/local_queue.py‎
Lines changed: 3 additions & 0 deletions b/‎port_ocean/core/handlers/queue/local_queue.py‎
Lines changed: 3 additions & 0 deletions
@@ -6,6 +6,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 <!-- towncrier release notes start -->
+## 0.27.0 (2025-08-03)
+### Improvements
+
+- Enhanced webhook event processing with GroupQueue implementation
+
+Introduced GroupQueue to ensure exclusive processing per group while allowing parallel processing across different groups
+Multiple workers can now process webhook events from different groups concurrently, improving throughput
+FIFO ordering is maintained within each group to preserve event sequence integrity
+Added automatic lock timeout mechanism to recover from frozen or hung workers
+Implemented context-based group tracking using ContextVar for cleaner worker-to-group association
+
+- Performance optimizations
+Configurable number of workers per webhook path (event_workers_count)
+Reduced contention by allowing concurrent processing of independent groups
+Improved resource cleanup and state management after processing
+
 ## 0.26.3 (2025-08-04)
 
 ### Bug Fixes
 
@@ -9,6 +9,7 @@
 
 from port_ocean.config.base import BaseOceanModel, BaseOceanSettings
 from port_ocean.core.event_listener import EventListenerSettingsType
+
 from port_ocean.core.models import (
     CachingStorageMode,
     CreatePortResourcesOrigin,
@@ -88,6 +89,7 @@ class IntegrationConfiguration(BaseOceanSettings, extra=Extra.allow):
     event_listener: EventListenerSettingsType = Field(
         default=cast(EventListenerSettingsType, {"type": "POLLING"})
     )
+    event_workers_count: int = 1
     # If an identifier or type is not provided, it will be generated based on the integration name
     integration: IntegrationSettings = Field(
         default_factory=lambda: IntegrationSettings(type="", identifier="")
 
@@ -1,4 +1,5 @@
 from .abstract_queue import AbstractQueue
 from .local_queue import LocalQueue
+from .group_queue import GroupQueue
 
-__all__ = ["AbstractQueue", "LocalQueue"]
+__all__ = ["AbstractQueue", "LocalQueue", "GroupQueue"]
@@ -7,6 +7,9 @@
 class AbstractQueue(ABC, Generic[T]):
     """Abstract interface for queues"""
 
+    def __init__(self, name: str | None = None):
+        pass
+
     @abstractmethod
     async def put(self, item: T) -> None:
         """Put an item into the queue"""
@@ -22,6 +25,11 @@ async def teardown(self) -> None:
         """Wait for all items to be processed"""
         pass
 
+    @abstractmethod
+    async def size(self) -> int:
+        """Size of the queue"""
+        pass
+
     @abstractmethod
     async def commit(self) -> None:
         """Mark item as processed"""
 
@@ -0,0 +1,138 @@
+import asyncio
+from collections import defaultdict, deque
+import time
+from typing import Deque, Dict, Optional, Set, TypeVar, Any
+from contextvars import ContextVar
+
+from loguru import logger
+
+from .abstract_queue import AbstractQueue
+
+T = TypeVar("T")
+MaybeStr = str | None
+
+_NO_GROUP = object()
+_current_group: ContextVar[Any] = ContextVar("current_group", default=_NO_GROUP)
+
+
+class GroupQueue(AbstractQueue[T]):
+    """Queue with exclusive processing per group."""
+
+    def __init__(
+        self,
+        group_key: MaybeStr = None,
+        name: MaybeStr = None,
+        lock_timeout: float = 300,
+    ):
+        super().__init__(name)
+        self.group_key = group_key
+        self._queues: Dict[MaybeStr, Deque[T]] = defaultdict(deque)
+        self._locked: Set[MaybeStr] = set()
+        self._queue_not_empty = asyncio.Condition()
+        self.lock_timeout = lock_timeout
+        self._lock_timestamps: Dict[MaybeStr, float] = {}
+        self._timeout_task: Optional[asyncio.Task[None]] = None
+
+    async def _background_timeout_check(self) -> None:
+        """Periodically release locks that have timed out."""
+        while True:
+            try:
+                await asyncio.sleep(self.lock_timeout / 4)
+                async with self._queue_not_empty:
+                    await self._release_expired_locks()
+            except asyncio.CancelledError:
+                break
+
+    def _extract_group_key(self, item: T) -> MaybeStr:
+        """Extract the group key from an item."""
+        if self.group_key is None:
+            return None
+        if not hasattr(item, self.group_key):
+            raise ValueError(
+                f"Item {item!r} lacks attribute '{self.group_key}' required for grouping"
+            )
+        return getattr(item, self.group_key)
+
+    async def put(self, item: T) -> None:
+        """Add item to its group's queue."""
+        group_key = self._extract_group_key(item)
+        async with self._queue_not_empty:
+            self._queues[group_key].append(item)
+            self._queue_not_empty.notify_all()
+
+    async def _release_expired_locks(self) -> None:
+        """Release locks that have exceeded the timeout."""
+        now = time.time()
+        expired_groups = []
+
+        for group, timestamp in list(self._lock_timestamps.items()):
+            if now - timestamp > self.lock_timeout:
+                expired_groups.append(group)
+                logger.warning(f"Releasing expired lock for group {group}")
+                self._locked.discard(group)
+                del self._lock_timestamps[group]
+
+        if expired_groups:
+            self._queue_not_empty.notify_all()
+
+    async def get(self) -> T:
+        """Get the next item from an unlocked group, locking that group."""
+        if self._timeout_task is None or self._timeout_task.done():
+            self._timeout_task = asyncio.create_task(self._background_timeout_check())
+
+        async with self._queue_not_empty:
+            while True:
+                await self._release_expired_locks()
+
+                for group, queue in self._queues.items():
+                    if queue and group not in self._locked:
+                        self._locked.add(group)
+                        self._lock_timestamps[group] = time.time()
+                        _current_group.set(group)
+                        return queue[0]
+
+                await self._queue_not_empty.wait()
+
+    async def commit(self) -> None:
+        """Remove the current item and unlock its group."""
+        group = _current_group.get()
+        if group is _NO_GROUP:
+            logger.warning("commit() called without active get()")
+            return
+
+        async with self._queue_not_empty:
+            queue = self._queues.get(group)
+            if queue:
+                queue.popleft()
+                if not queue:
+                    del self._queues[group]
+
+            self._locked.discard(group)
+            self._lock_timestamps.pop(group, None)
+            _current_group.set(_NO_GROUP)
+            self._queue_not_empty.notify_all()
+
+    async def teardown(self) -> None:
+        """Wait until all queues are empty and no groups are locked."""
+        async with self._queue_not_empty:
+            while any(self._queues.values()) or self._locked:
+                await self._queue_not_empty.wait()
+
+        if self._timeout_task and not self._timeout_task.done():
+            self._timeout_task.cancel()
+            try:
+                await self._timeout_task
+            except asyncio.CancelledError:
+                pass
+
+    async def size(self) -> int:
+        """Return total number of items across all groups."""
+        async with self._queue_not_empty:
+            return sum(len(queue) for queue in self._queues.values())
+
+    async def force_unlock_all(self) -> None:
+        """Force unlock all groups."""
+        async with self._queue_not_empty:
+            self._locked.clear()
+            self._lock_timestamps.clear()
+            self._queue_not_empty.notify_all()
@@ -23,3 +23,6 @@ async def teardown(self) -> None:
 
     async def commit(self) -> None:
         self._queue.task_done()
+
+    async def size(self) -> int:
+        return self._queue.qsize()