Open
Conversation
Contributor
Reviewer's GuideThis PR implements a cap on FSM transition history in snapshots, refactors QSimService to use bounded asyncio queues with overflow handling, enhances AsyncStateStore by cleaning stale subscribers and maintaining accurate version/metrics tracking, and adds exhaustive unit tests around history trimming and state transition behavior. Sequence diagram for QSimService queue overflow handlingsequenceDiagram
participant QSimService
participant WorldModel
participant SensorDataQueue
participant ActuatorCommandQueue
actor External
External->>QSimService: send ActuatorCommand
QSimService->>ActuatorCommandQueue: put_nowait(command)
ActuatorCommandQueue-->>QSimService: QueueFull (if full)
QSimService->>QSimService: log warning, drop command
QSimService->>WorldModel: update(command)
QSimService->>WorldModel: step(delta_time)
QSimService->>QSimService: generate_sensor_data()
QSimService->>SensorDataQueue: put_nowait(sensor_data)
SensorDataQueue-->>QSimService: QueueFull (if full)
QSimService->>QSimService: log warning, drop data
Class diagram for updated FSM DTOs and AsyncStateStoreclassDiagram
class FsmState {
<<enum>>
UNSPECIFIED
BOOTING
IDLE
ACTIVE
ERROR_STATE
SHUTDOWN
}
class TransitionStatus {
<<enum>>
UNSPECIFIED
SUCCESS
FAILED
PENDING
}
class TransitionDTO {
+from_state: FsmState
+to_state: FsmState
+trigger_event: str
+status: TransitionStatus
+error_message: str
+ts_mono: float
+ts_wall: float
}
class FsmSnapshotDTO {
+version: int
+state: FsmState
+reason: str
+ts_mono: float
+ts_wall: float
+snapshot_id: str
+prev_state: FsmState
+fsm_instance_id: str
+source_module: str
+attempt_count: int
+history: Tuple[TransitionDTO]
+context_data: Dict[str, str]
+state_metadata: Dict[str, str]
+add_transition(transition, max_history): FsmSnapshotDTO
}
class AsyncStateStore {
-_lock: asyncio.Lock
-_snap: FsmSnapshotDTO
-_subscribers: List[asyncio.Queue]
-_subscriber_ids: Dict[int, str]
-_metrics: Dict[str, Any]
+get(): FsmSnapshotDTO
+set(new_snap, enforce_version): FsmSnapshotDTO
+subscribe(subscriber_id): asyncio.Queue
+unsubscribe(queue)
+initialize_if_empty(): FsmSnapshotDTO
+get_metrics(): Dict[str, Any]
+health_check(): Dict[str, Any]
}
FsmSnapshotDTO "1" *-- "*" TransitionDTO : history
AsyncStateStore "1" *-- "*" asyncio.Queue : subscribers
Class diagram for QSimService with asyncio.Queue integrationclassDiagram
class QSimService {
+config: Dict[str, Any]
+world_model: WorldModel
+sensor_data_queue: asyncio.Queue
+actuator_command_queue: asyncio.Queue
+generate_sensor_data(): SensorReading
+receive_actuator_command(command)
+run()
+step()
}
QSimService "1" *-- "1" WorldModel
QSimService "1" *-- "1" asyncio.Queue : sensor_data_queue
QSimService "1" *-- "1" asyncio.Queue : actuator_command_queue
Class diagram for FSM history trimming in FsmSnapshotDTOclassDiagram
class FsmSnapshotDTO {
+add_transition(transition, max_history): FsmSnapshotDTO
}
class TransitionDTO
FsmSnapshotDTO "1" *-- "*" TransitionDTO : history
FsmSnapshotDTO : history capped by max_history
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- AsyncStateStore is awaiting _notify_subscribers while holding the store lock in set()/initialize_if_empty, which can block other operations; consider releasing the lock before dispatching notifications.
- QSimService mixes asyncio.Queue with a synchronous loop using time.sleep, which can lead to unexpected behavior; either convert run/step to async (await asyncio.sleep) or switch to a thread-safe queue.Queue for a purely synchronous design.
- Consider adding unit tests for AsyncStateStore to cover subscriber overflow removal, unsubscribe behavior, and metrics updates, ensuring the new cleanup and versioning logic behaves correctly under load.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- AsyncStateStore is awaiting _notify_subscribers while holding the store lock in set()/initialize_if_empty, which can block other operations; consider releasing the lock before dispatching notifications.
- QSimService mixes asyncio.Queue with a synchronous loop using time.sleep, which can lead to unexpected behavior; either convert run/step to async (await asyncio.sleep) or switch to a thread-safe queue.Queue for a purely synchronous design.
- Consider adding unit tests for AsyncStateStore to cover subscriber overflow removal, unsubscribe behavior, and metrics updates, ensuring the new cleanup and versioning logic behaves correctly under load.
## Individual Comments
### Comment 1
<location> `QIKI_DTMP/services/q_sim_service/main.py:42` </location>
<code_context>
+ scalar_data=world_state["position"]["x"] # Example: return X position as scalar data
+ )
+
+ def receive_actuator_command(self, command: ActuatorCommand):
+ try:
+ self.actuator_command_queue.put_nowait(command)
+ except asyncio.QueueFull:
+ logger.warning("Actuator command queue full, dropping command")
+ else:
+ logger.info(f"QSim received actuator command: {MessageToDict(command)}")
+ self.world_model.update(command) # Update world model based on command
+
+ def run(self):
</code_context>
<issue_to_address>
receive_actuator_command is synchronous but interacts with asyncio.Queue.
Consider making this method async and using 'await queue.put' to improve flow control and handle backpressure if it will be called from async code.
Suggested implementation:
```python
async def receive_actuator_command(self, command: ActuatorCommand):
await self.actuator_command_queue.put(command)
logger.info(f"QSim received actuator command: {MessageToDict(command)}")
self.world_model.update(command) # Update world model based on command
```
If `receive_actuator_command` is called from synchronous code elsewhere, those call sites will need to be updated to use `await` and be in async functions. If you need help updating those, please provide the relevant code.
</issue_to_address>
### Comment 2
<location> `QIKI_DTMP/services/q_sim_service/main.py:30` </location>
<code_context>
+ self.actuator_command_queue = asyncio.Queue(maxsize=queue_size)
+ logger.info("QSimService initialized.")
+
+ def generate_sensor_data(self) -> SensorReading:
+ # Generate sensor data based on world model state
+ timestamp = Timestamp()
+ timestamp.GetCurrentTime()
+ world_state = self.world_model.get_state()
+ return SensorReading(
+ sensor_id=UUID(value="sim_lidar_front"),
+ sensor_type=self.config.get("sim_sensor_type", 1), # LIDAR
+ timestamp=timestamp,
+ scalar_data=world_state["position"]["x"] # Example: return X position as scalar data
+ )
+
</code_context>
<issue_to_address>
generate_sensor_data assumes world_state["position"]["x"] exists.
This will raise a KeyError if any expected field is missing. Please add checks or defaults to ensure robustness against malformed world_state.
</issue_to_address>
<suggested_fix>
<<<<<<< SEARCH
def generate_sensor_data(self) -> SensorReading:
# Generate sensor data based on world model state
timestamp = Timestamp()
timestamp.GetCurrentTime()
world_state = self.world_model.get_state()
return SensorReading(
sensor_id=UUID(value="sim_lidar_front"),
sensor_type=self.config.get("sim_sensor_type", 1), # LIDAR
timestamp=timestamp,
scalar_data=world_state["position"]["x"] # Example: return X position as scalar data
)
=======
def generate_sensor_data(self) -> SensorReading:
# Generate sensor data based on world model state
timestamp = Timestamp()
timestamp.GetCurrentTime()
world_state = self.world_model.get_state()
# Safely extract position.x, defaulting to 0.0 if missing
position = world_state.get("position", {})
x_value = position.get("x", 0.0)
return SensorReading(
sensor_id=UUID(value="sim_lidar_front"),
sensor_type=self.config.get("sim_sensor_type", 1), # LIDAR
timestamp=timestamp,
scalar_data=x_value # Example: return X position as scalar data
)
>>>>>>> REPLACE
</suggested_fix>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+42
to
+49
| def receive_actuator_command(self, command: ActuatorCommand): | ||
| try: | ||
| self.actuator_command_queue.put_nowait(command) | ||
| except asyncio.QueueFull: | ||
| logger.warning("Actuator command queue full, dropping command") | ||
| else: | ||
| logger.info(f"QSim received actuator command: {MessageToDict(command)}") | ||
| self.world_model.update(command) # Update world model based on command |
Contributor
There was a problem hiding this comment.
suggestion: receive_actuator_command is synchronous but interacts with asyncio.Queue.
Consider making this method async and using 'await queue.put' to improve flow control and handle backpressure if it will be called from async code.
Suggested implementation:
async def receive_actuator_command(self, command: ActuatorCommand):
await self.actuator_command_queue.put(command)
logger.info(f"QSim received actuator command: {MessageToDict(command)}")
self.world_model.update(command) # Update world model based on commandIf receive_actuator_command is called from synchronous code elsewhere, those call sites will need to be updated to use await and be in async functions. If you need help updating those, please provide the relevant code.
| def __post_init__(self): | ||
| # Устанавливаем значения по умолчанию для mutable полей | ||
| if self.history is None: | ||
| object.__setattr__(self, 'history', tuple()) |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Replace tuple() with () (tuple-literal)
Suggested change
| object.__setattr__(self, 'history', tuple()) | |
| object.__setattr__(self, 'history', ()) |
Explanation
The most concise and Pythonic way to create an empty tuple is to use the()literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
x = ("first", "second")Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
| new_version = current.version + version_increment | ||
|
|
||
| # Новая история переходов | ||
| new_history = current.history if current.history else tuple() |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Replace tuple() with () (tuple-literal)
Suggested change
| new_history = current.history if current.history else tuple() | |
| new_history = current.history if current.history else () |
Explanation
The most concise and Pythonic way to create an empty tuple is to use the()literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
x = ("first", "second")Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
FsmSnapshotDTO.add_transitionasyncio.Queueс обработкой переполненияTesting
pytest QIKI_DTMP/services/q_core_agent/state/tests/test_types.py -qpytest QIKI_DTMP/services/q_core_agent/state/tests/test_store.py -qhttps://chatgpt.com/codex/tasks/task_e_68a27e4b71848331af1d95c7f02e0293
Summary by Sourcery
Provide a new immutable FSM state model with controlled transition history, an asynchronous state store supporting pub/sub and metrics, update the QSimService to use asyncio queues with overflow handling, and add thorough unit tests for FSM types and history management
New Features:
Enhancements:
Tests: