Fixes an potential race condition when multiple signals were sent at the same time. (#3239)

Fixes race condition where if multiple signals are ingested, no case may be created.

Author: kevgliss

src/dispatch/entity/models.py

@@ -21,7 +21,7 @@ class Entity(Base, TimeStampMixin, ProjectMixin):
     __table_args__ = (UniqueConstraint("name", "project_id"),)
 
     # Columns
-    id = Column(Integer, primary_key=True)
+    id = Column(Integer, primary_key=True, autoincrement=True)
     name = Column(String)
     description = Column(String)
     value = Column(String)

src/dispatch/signal/flows.py

@@ -107,9 +107,4 @@ def create_signal_instance(
     signal_instance = signal_service.create_instance(
         db_session=db_session, signal_instance_in=signal_instance_in
     )
-
-    return signal_instance_create_flow(
-        db_session=db_session,
-        signal_instance_id=signal_instance.id,
-        organization_slug=None,
-    )
+    return signal_instance

src/dispatch/signal/models.py

@@ -116,6 +116,7 @@ class SignalFilterMode(DispatchEnum):
 class SignalFilterAction(DispatchEnum):
     deduplicate = "deduplicate"
     snooze = "snooze"
+    none = "none"
 
 
 class Signal(Base, TimeStampMixin, ProjectMixin):

src/dispatch/signal/scheduled.py

@@ -6,13 +6,13 @@
 """
 import logging
 from schedule import every
-
 from dispatch.database.core import SessionLocal
 from dispatch.scheduler import scheduler
 from dispatch.project.models import Project
 from dispatch.plugin import service as plugin_service
 from dispatch.signal import flows as signal_flows
 from dispatch.decorators import scheduled_project_task
+from dispatch.signal.models import SignalInstance
 
 log = logging.getLogger(__name__)
 
@@ -49,3 +49,25 @@ def consume_signals(db_session: SessionLocal, project: Project):
 
         if signal_instances:
             plugin.instance.delete()
+
+
+@scheduler.add(every(1).minutes, name="signal-process")
+@scheduled_project_task
+def process_signals(db_session: SessionLocal, project: Project):
+    """Processes signals and create cases if appropriate."""
+    signal_instances = (
+        db_session.query(SignalInstance)
+        .filter(SignalInstance.project_id == project.id)
+        .filter(SignalInstance.filter_action == None)  # noqa
+    )
+    for signal_instance in signal_instances:
+        log.info(f"Attempting to process the following signal: {signal_instance.id}")
+        try:
+            signal_flows.signal_instance_create_flow(
+                db_session=db_session,
+                project=project,
+                signal_instance_id=signal_instance.id,
+            )
+        except Exception as e:
+            log.debug(signal_instance)
+            log.exception(e)

src/dispatch/signal/service.py

@@ -17,6 +17,7 @@
 from dispatch.service import service as service_service
 from dispatch.tag import service as tag_service
 from dispatch.workflow import service as workflow_service
+from dispatch.entity.models import Entity
 
 from .models import (
     Signal,
@@ -378,8 +379,12 @@ def filter_signal(*, db_session: Session, signal_instance: SignalInstance) -> bo
                 break
 
         elif f.action == SignalFilterAction.deduplicate:
-            window = datetime.now(timezone.utc) - timedelta(seconds=f.window)
+            window = datetime.now(timezone.utc) - timedelta(minutes=f.window)
             query = query.filter(SignalInstance.created_at >= window)
+            query = query.join(SignalInstance.entities).filter(
+                Entity.id.in_([e.id for e in signal_instance.entities])
+            )
+            query = query.filter(SignalInstance.id != signal_instance.id)
 
             # get the earliest instance
             query = query.order_by(asc(SignalInstance.created_at))
@@ -392,5 +397,8 @@ def filter_signal(*, db_session: Session, signal_instance: SignalInstance) -> bo
                 filtered = True
                 break
 
+    if not filtered:
+        signal_instance.filter_action = SignalFilterAction.none
+
     db_session.commit()
     return filtered

src/dispatch/signal/views.py

@@ -11,7 +11,6 @@
 from dispatch.models import OrganizationSlug, PrimaryKey
 from dispatch.project import service as project_service
 from dispatch.signal import service as signal_service
-from dispatch.signal.flows import signal_instance_create_flow
 
 from .models import (
     SignalCreate,
@@ -94,11 +93,6 @@ def create_signal_instance(
     )
     signal_instance.signal = signal
     db_session.commit()
-
-    background_tasks.add_task(
-        signal_instance_create_flow, signal_instance.id, organization_slug=organization
-    )
-
     return signal_instance
 
 

tests/factories.py

@@ -801,7 +801,6 @@ class Meta:
 
 
 class EntityFactory(BaseFactory):
-    id = Sequence(lambda n: f"1{n}")
     name = FuzzyText()
     description = FuzzyText()
     value = FuzzyText()

tests/signal/test_signal_service.py

@@ -50,6 +50,117 @@ def test_delete(session, signal):
     assert not get(db_session=session, signal_id=signal.id)
 
 
+def test_filter_actions_deduplicate_different_entities(session, signal, project):
+    from dispatch.signal.models import (
+        SignalFilter,
+        SignalInstance,
+        SignalFilterAction,
+    )
+    from dispatch.signal.service import filter_signal
+    from dispatch.entity_type.models import EntityType
+    from dispatch.entity.models import Entity
+
+    entity_type_0 = EntityType(
+        name="dedupe2-0",
+        field="id",
+        regular_expression=None,
+        project=project,
+    )
+    session.add(entity_type_0)
+
+    entity_0 = Entity(name="dedupe2", description="test", value="foo", entity_type=entity_type_0)
+    session.add(entity_0)
+
+    entity_1 = Entity(name="dedupe2-1", description="test", value="foo", entity_type=entity_type_0)
+    session.add(entity_1)
+
+    signal_instance_0 = SignalInstance(
+        raw=json.dumps({"id": "foo"}), project=project, signal=signal, entities=[entity_0]
+    )
+    session.add(signal_instance_0)
+
+    signal_instance_1 = SignalInstance(
+        raw=json.dumps({"id": "foo"}), project=project, signal=signal, entities=[entity_1]
+    )
+    session.add(signal_instance_1)
+    session.commit()
+
+    # create deduplicate signal filter
+    signal_filter = SignalFilter(
+        name="test",
+        description="dedupe2",
+        expression=[
+            {"or": [{"model": "EntityType", "field": "id", "op": "==", "value": entity_type_0.id}]}
+        ],
+        action=SignalFilterAction.deduplicate,
+        window=5,
+        project=project,
+    )
+    signal.filters.append(signal_filter)
+
+    session.commit()
+    assert not filter_signal(db_session=session, signal_instance=signal_instance_1)
+    assert signal_instance_1.filter_action == SignalFilterAction.none
+
+
+def test_filter_actions_deduplicate_different_entities_types(session, signal, project):
+    from dispatch.signal.models import (
+        SignalFilter,
+        SignalInstance,
+        SignalFilterAction,
+    )
+    from dispatch.signal.service import filter_signal
+    from dispatch.entity_type.models import EntityType
+    from dispatch.entity.models import Entity
+
+    entity_type_0 = EntityType(
+        name="dedupe0-0",
+        field="id",
+        regular_expression=None,
+        project=project,
+    )
+    session.add(entity_type_0)
+    entity_0 = Entity(name="dedupe0", description="test", value="foo", entity_type=entity_type_0)
+    session.add(entity_0)
+    signal_instance_0 = SignalInstance(
+        raw=json.dumps({"id": "foo"}), project=project, signal=signal, entities=[entity_0]
+    )
+    session.add(signal_instance_0)
+
+    entity_type_1 = EntityType(
+        name="dedupe0-1",
+        field="id",
+        regular_expression=None,
+        project=project,
+    )
+    session.add(entity_type_1)
+    entity_1 = Entity(name="dedupe0-1", description="test", value="foo", entity_type=entity_type_1)
+    session.add(entity_1)
+
+    signal_instance_1 = SignalInstance(
+        raw=json.dumps({"id": "foo"}), project=project, signal=signal, entities=[entity_1]
+    )
+    session.add(signal_instance_1)
+    session.commit()
+
+    # create deduplicate signal filter
+    signal_filter = SignalFilter(
+        name="test",
+        description="dedupe0",
+        expression=[
+            {"or": [{"model": "EntityType", "field": "id", "op": "==", "value": entity_type_1.id}]}
+        ],
+        action=SignalFilterAction.deduplicate,
+        window=5,
+        project=project,
+    )
+    signal.filters.append(signal_filter)
+
+    session.commit()
+    assert not filter_signal(db_session=session, signal_instance=signal_instance_1)
+    assert signal_instance_1.filter_action == SignalFilterAction.none
+
+
 def test_filter_actions_deduplicate(session, signal, project):
     from dispatch.signal.models import (
         SignalFilter,
@@ -61,14 +172,14 @@ def test_filter_actions_deduplicate(session, signal, project):
     from dispatch.entity.models import Entity
 
     entity_type = EntityType(
-        name="test",
+        name="dedupe1",
         field="id",
         regular_expression=None,
         project=project,
     )
     session.add(entity_type)
 
-    entity = Entity(name="test", description="test", value="foo", entity_type=entity_type)
+    entity = Entity(name="dedupe1", description="test", value="foo", entity_type=entity_type)
     session.add(entity)
 
     # create instance
@@ -81,13 +192,10 @@ def test_filter_actions_deduplicate(session, signal, project):
         raw=json.dumps({"id": "foo"}), project=project, signal=signal, entities=[entity]
     )
     session.add(signal_instance_2)
-    signal.entity_types.append(entity_type)
-
     session.commit()
-
     # create deduplicate signal filter
     signal_filter = SignalFilter(
-        name="test",
+        name="dedupe1",
         description="test",
         expression=[
             {"or": [{"model": "EntityType", "field": "id", "op": "==", "value": entity_type.id}]}