[uss_qualifier] Fix DSS notification_index sync checks

.basedpyright/baseline.json

@@ -7059,46 +7059,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 45,
-                    "endColumn": 48,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 46,
-                    "endColumn": 49,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 28,
-                    "endColumn": 75,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 53,
-                    "endColumn": 56,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 51,
-                    "endColumn": 54,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {
@@ -7115,14 +7075,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 28,
-                    "endColumn": 63,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {
@@ -7155,14 +7107,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 28,
-                    "endColumn": 61,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {

monitoring/uss_qualifier/resources/astm/dss.py

@@ -0,0 +1,9 @@
+from enum import StrEnum
+
+
+class NotificationIndexImplementation(StrEnum):
+    ZeroBasedIncrementPerDispatch = "ZeroBasedIncrementPerDispatch"
+    """Notification index starts at 0 and is incremented by 1 each time a notification is requested/sent due to the associated subscription."""
+
+    TimedBased = "TimeBased"
+    """Notification index is populated based on the clock of the DSS instance serving the request."""

monitoring/uss_qualifier/resources/astm/f3411/dss.py

@@ -2,12 +2,13 @@
 
 from urllib.parse import urlparse
 
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 
 from monitoring.monitorlib import infrastructure
 from monitoring.monitorlib.infrastructure import UTMClientSession
 from monitoring.monitorlib.rid import RIDVersion
 from monitoring.uss_qualifier.reports.report import ParticipantID
+from monitoring.uss_qualifier.resources.astm.dss import NotificationIndexImplementation
 from monitoring.uss_qualifier.resources.communications import AuthAdapterResource
 from monitoring.uss_qualifier.resources.resource import Resource
 
@@ -22,6 +23,11 @@ class DSSInstanceSpecification(ImplicitDict):
     base_url: str
     """Base URL for the DSS instance according to the ASTM F3411 API appropriate to the specified rid_version"""
 
+    notification_index_implementation: Optional[NotificationIndexImplementation]
+    """Style of implementation this instance uses for notification index.
+
+    If not specified, TimeBased is assumed."""
+
     def __init__(self, *args, **kwargs):
         super().__init__(**kwargs)
         try:
@@ -35,24 +41,29 @@ class DSSInstance:
     rid_version: RIDVersion
     base_url: str
     client: infrastructure.UTMClientSession
+    notification_index_implementation: NotificationIndexImplementation
 
     def __init__(
         self,
         participant_id: ParticipantID,
         base_url: str,
         rid_version: RIDVersion,
         client: UTMClientSession,
+        notification_index_implementation: NotificationIndexImplementation,
     ):
         self.participant_id = participant_id
         self.base_url = base_url
         self.rid_version = rid_version
         self.client = client
+        self.notification_index_implementation = notification_index_implementation
 
     def is_same_as(self, other: DSSInstance) -> bool:
         return (
             self.participant_id == other.participant_id
             and self.rid_version == other.rid_version
             and self.base_url == other.base_url
+            and self.notification_index_implementation
+            == other.notification_index_implementation
         )
 
 
@@ -85,6 +96,10 @@ def __init__(
             infrastructure.utm_client_session_factory.get_session(
                 specification.base_url, auth_adapter.adapter
             ),
+            specification.notification_index_implementation
+            if "notification_index_implementation" in specification
+            and specification.notification_index_implementation
+            else NotificationIndexImplementation.TimedBased,
         )
 
     @classmethod

monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py

@@ -51,6 +51,7 @@
 from monitoring.monitorlib.inspection import calling_function_name, fullname
 from monitoring.monitorlib.mutate import scd as mutate
 from monitoring.monitorlib.mutate.scd import MutatedSubscription
+from monitoring.uss_qualifier.resources.astm.dss import NotificationIndexImplementation
 from monitoring.uss_qualifier.resources.communications import AuthAdapterResource
 from monitoring.uss_qualifier.resources.resource import Resource
 
@@ -71,6 +72,11 @@ class DSSInstanceSpecification(ImplicitDict):
     timeout_seconds: Optional[float]
     """If specified, number of seconds to allow before timing out requests to this DSS instance."""
 
+    notification_index_implementation: Optional[NotificationIndexImplementation]
+    """Style of implementation this instance uses for notification index.
+
+    If not specified, TimeBased is assumed."""
+
     def __init__(self, *args, **kwargs):
         super().__init__(**kwargs)
         try:

monitoring/uss_qualifier/scenarios/astm/netrid/common/dss_interoperability.py

@@ -11,6 +11,7 @@
 from monitoring.monitorlib.fetch.rid import ISA
 from monitoring.monitorlib.geo import get_latlngrect_vertices, make_latlng_rect
 from monitoring.uss_qualifier.resources import PlanningAreaResource
+from monitoring.uss_qualifier.resources.astm.dss import NotificationIndexImplementation
 from monitoring.uss_qualifier.resources.astm.f3411.dss import (
     DSSInstanceResource,
     DSSInstancesResource,
@@ -308,7 +309,7 @@ def step3(self):
             # check data synchronization
             def get_fail_params(
                 field_name: str,
-                primary_sub_field_value: str,
+                primary_sub_field_value: object,
                 other_sub_field_value: object,
             ) -> dict:
                 return dict(
@@ -371,16 +372,35 @@ def get_fail_params(
                 [dss.participant_id],
             ) as check:
                 if (
-                    primary_sub.subscription.raw.notification_index
-                    != other_sub.subscription.raw.notification_index
+                    dss.notification_index_implementation
+                    != self._dss_primary.notification_index_implementation
                 ):
                     check.record_failed(
-                        **get_fail_params(
-                            "notification_index",
-                            primary_sub.subscription.raw.notification_index,
-                            other_sub.subscription.raw.notification_index,
-                        )
+                        summary="Incompatible notification index implementations",
+                        details=f"All DSS instances in a pool must use the same notification index implementation to achieve synchronized notification count behavior, but DSS for {dss.participant_id} uses {dss.notification_index_implementation.value} while DSS for {self._dss_primary.participant_id} uses {self._dss_primary.notification_index_implementation.value}",
+                    )
+                elif (
+                    dss.notification_index_implementation
+                    == NotificationIndexImplementation.ZeroBasedIncrementPerDispatch
+                ):
+                    primary_index = (
+                        primary_sub.subscription.raw.notification_index
+                        if primary_sub.subscription
+                        and primary_sub.subscription.raw.notification_index
+                        else 0
                     )
+                    other_index = (
+                        other_sub.subscription.raw.notification_index
+                        if other_sub.subscription
+                        and other_sub.subscription.raw.notification_index
+                        else 0
+                    )
+                    if primary_index != other_index:
+                        check.record_failed(
+                            **get_fail_params(
+                                "notification_index", primary_index, other_index
+                            )
+                        )
 
             with self.check(
                 "Subscription[P] start/end times are properly synchronized with all DSS",

monitoring/uss_qualifier/scenarios/astm/netrid/dss_wrapper.py

@@ -22,6 +22,7 @@
 from monitoring.monitorlib.mutate import rid as mutate
 from monitoring.monitorlib.mutate.rid import ChangedSubscription, ISAChange
 from monitoring.monitorlib.rid import RIDVersion
+from monitoring.uss_qualifier.resources.astm.dss import NotificationIndexImplementation
 from monitoring.uss_qualifier.resources.astm.f3411.dss import DSSInstance
 from monitoring.uss_qualifier.scenarios.astm.netrid.common.dss.isa_validator import (
     ISAValidator,
@@ -55,6 +56,10 @@ def participant_id(self) -> str:
     def base_url(self) -> str:
         return self._dss.base_url
 
+    @property
+    def notification_index_implementation(self) -> NotificationIndexImplementation:
+        return self._dss.notification_index_implementation
+
     # TODO: QueryError is not actually raised for RID functions, this function and its uses should be removed
     def _handle_query_error(
         self,

monitoring/uss_qualifier/scripts/test_docker_fully_mocked.sh

@@ -47,6 +47,8 @@ trap on_sigint SIGINT
 
 echo "Start mock system"
 echo "============="
+export DSS_IMAGE="${DSS_IMAGE:-interuss/dss:v0.23.0-rc4}"
+export CORE_SERVICE_EXTRA_FLAGS="${CORE_SERVICE_EXTRA_FLAGS:---enable_time_based_notification_index}"
 make start-locally
 make start-uss-mocks
 

schemas/monitoring/uss_qualifier/resources/astm/f3411/dss/DSSInstanceSpecification.json

@@ -11,6 +11,17 @@
       "description": "Base URL for the DSS instance according to the ASTM F3411 API appropriate to the specified rid_version",
       "type": "string"
     },
+    "notification_index_implementation": {
+      "description": "Style of implementation this instance uses for notification index.\n\nIf not specified, TimeBased is assumed.",
+      "enum": [
+        "ZeroBasedIncrementPerDispatch",
+        "TimeBased"
+      ],
+      "type": [
+        "string",
+        "null"
+      ]
+    },
     "participant_id": {
       "description": "ID of the USS responsible for this DSS instance",
       "type": "string"

schemas/monitoring/uss_qualifier/resources/astm/f3548/v21/dss/DSSInstanceSpecification.json

@@ -11,6 +11,17 @@
       "description": "Base URL for the DSS instance according to the ASTM F3548-21 API",
       "type": "string"
     },
+    "notification_index_implementation": {
+      "description": "Style of implementation this instance uses for notification index.\n\nIf not specified, TimeBased is assumed.",
+      "enum": [
+        "ZeroBasedIncrementPerDispatch",
+        "TimeBased"
+      ],
+      "type": [
+        "string",
+        "null"
+      ]
+    },
     "participant_id": {
       "description": "ID of the USS responsible for this DSS instance",
       "type": "string"