feat: add idempotency guard on completion handling. (8dcecf2c) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/traffic_influence_service.py

+16 −0

Original line number	Diff line number	Diff line
		@@ -52,6 +52,14 @@ _OPERATION_COMPLETION_STATUS_MAP: dict[str, OperationStatus] = {
		"failed": OperationStatus.FAILED,
		}

		_TERMINAL_OPERATION_STATUSES = frozenset(
		{
		OperationStatus.COMPLETED,
		OperationStatus.PARTIALLY_COMPLETED,
		OperationStatus.FAILED,
		}
		)

		# CallbackRegistration.api_family discriminator for Traffic Influence's subscriptionRequest.
		_API_FAMILY = "traffic_influence"

		@@ -257,6 +265,14 @@ class TrafficInfluenceService:
		):
		return

		if operation.status in _TERMINAL_OPERATION_STATUSES:
		logger.info(
		"redelivered_completion_ignored_for_terminal_operation",
		operation_id=event.operation_id,
		status=operation.status.value,
		)
		return

		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		result = None
		if status != OperationStatus.FAILED:

tests/unit/test_traffic_influence_flows.py

+36 −0

Original line number	Diff line number	Diff line
		@@ -135,6 +135,42 @@ class TestTrafficInfluenceCreateFlow:
		assert updated.state == TrafficInfluenceState.ACTIVE
		assert updated.external_ref == "traffic-policy-456"

		async def test_redelivered_completion_event_is_ignored(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		operation_repo: FakeOperationRepository,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		) -> None:
		"""JetStream delivers at least once, so the same completion can arrive
		repeatedly; a redelivery must not overwrite an already-terminal operation
		or re-derive traffic influence state a second time."""
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)
		traffic_influence_id = UUID(response.json()["trafficInfluenceID"])
		operation_id = traffic_influence_repo.rows[traffic_influence_id].operation_id

		payload = completion_payload(
		str(operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(uuid4()),
		"status": "completed",
		"external_ref": "traffic-policy-456",
		}
		],
		)

		await fake_bus.publish(str(Subject.OPERATION_COMPLETED), payload)
		await fake_bus.publish(str(Subject.OPERATION_COMPLETED), payload)
		await fake_bus.publish(str(Subject.OPERATION_COMPLETED), payload)

		updated = traffic_influence_repo.rows[traffic_influence_id]
		assert updated.state == TrafficInfluenceState.ACTIVE
		assert updated.external_ref == "traffic-policy-456"
		assert operation_repo.rows[operation_id].status == OperationStatus.COMPLETED


		class TestTrafficInfluenceGetFlow:
		@pytest.mark.skip(

Admin message