[Advanced]: Bug in tests/integration/topic_message_query_e2e_test.py

[exploreriii]

🧠 Advanced Contributors

This issue is intended for contributors who are already very familiar with the
Hiero Python SDK codebase and its architectural patterns.

You should feel comfortable:

• navigating multiple modules across src/
• understanding and modifying core SDK abstractions
• reasoning about API design and backwards compatibility
• updating or extending tests, examples, and documentation as needed
• making changes that may affect public-facing behavior

New developers should start with
Good First Issues or Intermediate Issues first.

🐞 Problem Description

TopicMessageQuery.subscribe() intermittently fails to deliver messages that have successfully reached consensus.
The submit transaction returns SUCCESS, but the subscription callback is never invoked, eventually causing a timeout.

This breaks multiple integration tests and suggests a race condition or incorrect timestamp / stream handling in the mirror subscription pipeline.

❌ Integration tests FAILED
→ Check the integration test logs above for details.
tests/integration/topic_message_query_e2e_test.py::test_topic_message_query_receives_messages FAILED [ 89%]
FAILED tests/integration/topic_message_query_e2e_test.py::test_topic_message_query_receives_messages - TimeoutError: TopicMessage was not received in time

https://github.com/hiero-ledger/hiero-sdk-python/actions/runs/21984284776/job/63514462825

This is happening frequently, with the same error, but not all the time

=================================== FAILURES ===================================
__________________ test_topic_message_query_receives_messages __________________

env = <tests.integration.utils.IntegrationTestEnv object at 0x7fc5013df8b0>

    @pytest.mark.integration
    def test_topic_message_query_receives_messages(env):
        """Test that topic message query receives a message."""
        topic_id = create_topic(env.client)
    
        time.sleep(3)
    
        messages: List[str] = []
    
        def get_message(m: TopicMessage):
            messages.append(m.contents.decode('utf-8'))
    
        def on_error_handler(e):
            raise RuntimeError(f"Subscription error: {e}")
    
        query = TopicMessageQuery(
            topic_id=topic_id,
            start_time=datetime.now(timezone.utc),
            limit=0
        )
    
        handle = query.subscribe(
            env.client,
            on_message=get_message,
            on_error=on_error_handler
        )
    
        time.sleep(3)
    
        message_receipt = (
            TopicMessageSubmitTransaction(
                topic_id=topic_id,
                message="Hello, Python SDK!"
            )
            .freeze_with(env.client)
            .execute(env.client)
        )
    
        assert (
            message_receipt.status == ResponseCode.SUCCESS
        ), f"Message submission failed with status: {ResponseCode(message_receipt.status).name}"
    
    
        start = datetime.now()
    
        while len(messages) == 0:
            if datetime.now() - start > timedelta(seconds=180):
>               raise TimeoutError("TopicMessage was not received in time")
E               TimeoutError: TopicMessage was not received in time

tests/integration/topic_message_query_e2e_test.py:104: TimeoutError
=========================== short test summary info ============================
FAILED tests/integration/topic_message_query_e2e_test.py::test_topic_message_query_receives_messages - TimeoutError: TopicMessage was not received in time
```yaml


Expected behavior

After:
1/creating a topic
2/opening a subscription
3/submitting a message
the SDK should deliver the message via on_message.

Actual behavior
1/topicMessageSubmitTransaction.execute() returns a receipt with ResponseCode.SUCCESS.
2/no error is raised from the subscription.
3/on_message is never called.
4/test times out after 180 seconds.

This makes the SDK unreliable for users

### 💡 Proposed / Expected Solution

Create a minimal script to reproduce the error, to confirm the cause - the problem is after consensus & on_error is not triggered. Ensure thorough logging to pin point the error


Fix the error, in the underlying source code and/or tests without introducing breaking changes
Only increase time out window if necessary or reasonable

### 🧠 Implementation & Design Notes

Add detailed implementation notes here.


### ✅ Acceptance Criteria

To merge this issue, the pull request must:

- [ ] Fully address the problem and design goals described above
- [ ] Maintain backwards compatibility unless explicitly approved otherwise
- [ ] Follow existing architectural and coding conventions
- [ ] Include comprehensive tests covering new and existing behavior
- [ ] Update relevant examples and documentation
- [ ] Pass all CI checks
- [ ] Include a valid changelog entry
- [ ] be a DCO and GPG key signed as `git commit -S -s -m "chore: my change"` with a GPG key set up


### 📚 Additional Context, Links, or Prior Art

Optional.

[github-actions]

@coderabbitai plan

[coderabbitai]

Coding Plan

Summary

• Fix the root cause by setting start_time 5 seconds in the past to ensure messages submitted after subscription setup have consensus timestamps after the start_time filter
• Add logging to TopicMessageQuery to aid future diagnosis of timing issues
• Optionally add subscription readiness signaling to SubscriptionHandle for more robust synchronization patterns
• Improve test reliability with proper cleanup patterns using try/finally blocks

Design Choices

Design Choice 1: What is the primary cause of intermittent message delivery failures?

Options Considered:

1. Clock drift causing start_time > consensus timestamp - Simple fix by setting start_time in the past
2. Subscription not established before message submission - Would require readiness signaling mechanism
3. gRPC stream silently failing - Would require enhanced error handling and logging

Chosen Option: 1

Rationale: Clock drift is the most likely cause based on the intermittent nature and the fact that no errors are raised. Setting start_time slightly in the past is the simplest fix that aligns with Hedera's documented behavior that mirror nodes serve historical messages from the start_time forward.

💡 User Tips

Regenerate the plan with different choices with @coderabbitai <feedback>.

Implementation Steps

Phase 1: Add Diagnostic Tooling and Logging

This phase creates the diagnostic script requested in the ticket and adds logging to the SDK to help identify timing issues in the subscription pipeline.

Task 1: Create Diagnostic Reproduction Script

Create a standalone script that reproduces the issue with detailed logging to confirm the root cause and validate the fix.

• Create examples/debug/topic_message_timing_diagnostic.py with comprehensive logging at each step
• Log client-side timestamps at: start_time capture, subscription start, message submission, and message receipt
• Include the consensus timestamp from the transaction receipt for comparison with start_time
• Add configurable parameters for start_time offset to test different timing scenarios
• Print clear diagnostic output showing timestamp comparisons to identify clock drift

Task 2: Add Logging to TopicMessageQuery

Add debug-level logging to the subscription implementation to aid in diagnosing timing and delivery issues.

• Update src/hiero_sdk_python/query/topic_message_query.py to add logging statements
• Log when subscription thread starts and when gRPC stream is established
• Log each message received with its consensus timestamp
• Log retry attempts and backoff delays with timestamps
• Log stream completion and any errors encountered
• Use Python's standard logging module with appropriate log levels (DEBUG for verbose, INFO for key events)

🤖 Prompt for AI agents

This phase adds diagnostic tooling and logging to help identify timing issues in
the TopicMessageQuery subscription pipeline.

**Task 1: Create Diagnostic Reproduction Script**
- Create a new file `examples/debug/topic_message_timing_diagnostic.py`
- Implement comprehensive logging at each step of the subscription and message
submission flow
- Log client-side timestamps at: start_time capture, subscription start, message
submission, and message receipt
- Extract and log the consensus timestamp from the transaction receipt for
comparison with start_time
- Add configurable parameters for start_time offset to test different timing
scenarios
- Print clear diagnostic output showing timestamp comparisons to identify clock
drift between client and Hedera network

**Task 2: Add Logging to TopicMessageQuery**
- Update `src/hiero_sdk_python/query/topic_message_query.py`
- Import Python's standard `logging` module
- Add logging statements at key points:
  - When subscription thread starts
  - When gRPC stream is established
  - For each message received (include consensus timestamp)
  - For retry attempts and backoff delays (include timestamps)
  - For stream completion and any errors encountered
- Use appropriate log levels: DEBUG for verbose details, INFO for key events

Phase 2: Fix Root Cause in Integration Tests

This phase addresses the timing race condition by adjusting how tests set up subscriptions and handle synchronization.

Task 1: Fix start_time Configuration in Tests

Update the failing tests to set start_time slightly in the past to account for clock drift between the client and Hedera network.

• In tests/integration/topic_message_query_e2e_test.py, update test_topic_message_query_receives_messages to use start_time=datetime.now(timezone.utc) - timedelta(seconds=5)
• Apply the same fix to test_topic_message_query_large_message_chunking
• For test_topic_message_query_limit, evaluate if the same fix is needed (it has no error handler so may not have been reporting failures)
• This is safe because tests create new topics with no prior messages
• Add a comment explaining why the offset is needed (clock drift between client and consensus nodes)

Task 2: Improve Test Cleanup Patterns

Ensure all subscription tests use proper cleanup to prevent resource leaks and improve test reliability.

• Wrap subscription handling in try/finally blocks for all three tests in topic_message_query_e2e_test.py
• Ensure handle.cancel() is always called in the finally block
• Consider adding topic deletion in cleanup (matching pattern from topic_message_submit_transaction_e2e_test.py)
• Add error handlers to tests that lack them for better failure visibility

🤖 Prompt for AI agents

This phase fixes the timing race condition by adjusting start_time handling and
improving test cleanup patterns.

**Task 1: Fix start_time Configuration in Tests**
- Open `tests/integration/topic_message_query_e2e_test.py`
- In `test_topic_message_query_receives_messages`, update the start_time to
`datetime.now(timezone.utc) - timedelta(seconds=5)` instead of
`datetime.now(timezone.utc)`
- Apply the same change to `test_topic_message_query_large_message_chunking`
- For `test_topic_message_query_limit`, evaluate if the same fix is needed (it
has no error handler so may not have been reporting failures)
- Add a comment above the start_time assignment explaining: "Set start_time in
the past to account for clock drift between client and Hedera consensus nodes"
- This is safe because tests create new topics with no prior messages

**Task 2: Improve Test Cleanup Patterns**
- In `tests/integration/topic_message_query_e2e_test.py`, wrap subscription
handling in `try/finally` blocks for all three tests
- Ensure `handle.cancel()` is always called in the `finally` block to prevent
resource leaks
- Consider adding topic deletion in cleanup matching the pattern from
`topic_message_submit_transaction_e2e_test.py`
- Add error handlers to tests that currently lack them for better failure
visibility

Phase 3: Add Subscription Readiness Signaling (Enhancement)

This phase adds optional subscription readiness signaling to enable more robust synchronization patterns.

Task 1: Extend SubscriptionHandle with Ready State

Add a mechanism to signal when the subscription's gRPC stream is successfully established.

• Update src/hiero_sdk_python/utils/subscription_handle.py to add a _ready threading.Event
• Add is_ready() method returning the ready state
• Add wait_until_ready(timeout: float = None) method that blocks until ready or timeout
• Initialize _ready event as unset in constructor
• Add internal _set_ready() method to signal readiness

Task 2: Signal Readiness from TopicMessageQuery

Update the subscription implementation to signal readiness when the gRPC stream is established.

• In src/hiero_sdk_python/query/topic_message_query.py, call subscription_handle._set_ready() after successfully calling client.mirror_stub.subscribeTopic(request)
• Signal readiness before entering the message iteration loop
• Ensure readiness is only signaled once per subscription attempt, not on retries
• This allows callers to optionally wait for the subscription to be established before proceeding

Task 3: Update Examples with Recommended Pattern

Update the topic message query example to demonstrate the recommended timing pattern.

• Update examples/query/topic_message_query.py to show the recommended start_time offset pattern
• Add comments explaining why setting start_time in the past is recommended for reliability
• Optionally demonstrate using wait_until_ready() if the new mechanism is added
• Ensure the example follows patterns consistent with other examples in the codebase

🤖 Prompt for AI agents

This phase adds optional subscription readiness signaling to enable more robust
synchronization patterns when using TopicMessageQuery.

**Task 1: Extend SubscriptionHandle with Ready State**
- Open `src/hiero_sdk_python/utils/subscription_handle.py`
- Import `threading.Event` if not already imported
- Add a `_ready` instance variable of type `threading.Event` in the constructor,
initialized as unset
- Add `is_ready()` method that returns the ready state as a boolean
- Add `wait_until_ready(timeout: float = None)` method that blocks until ready
or timeout, returning True if ready or False on timeout
- Add internal `_set_ready()` method to signal readiness by setting the event

**Task 2: Signal Readiness from TopicMessageQuery**
- Open `src/hiero_sdk_python/query/topic_message_query.py`
- After successfully calling `client.mirror_stub.subscribeTopic(request)` and
establishing the gRPC stream, call `subscription_handle._set_ready()`
- Place this call before entering the message iteration loop
- Ensure readiness is only signaled once per subscription attempt, not on
retries
- This allows callers to optionally wait for the subscription to be established
before proceeding

**Task 3: Update Examples with Recommended Pattern**
- Open `examples/query/topic_message_query.py`
- Update to show the recommended `start_time` offset pattern using
`datetime.now(timezone.utc) - timedelta(seconds=5)`
- Add comments explaining why setting start_time in the past is recommended for
reliability (clock drift between client and consensus nodes)
- Optionally demonstrate using `wait_until_ready()` if the new mechanism is
implemented
- Ensure the example follows patterns consistent with other examples in the
codebase

Phase 4: Validation and Documentation

Ensure the fix is complete with proper testing and documentation.

Task 1: Verify All Topic Message Tests Pass Reliably

Run the integration tests multiple times to confirm the fix resolves the intermittent failures.

• Run pytest tests/integration/topic_message_query_e2e_test.py -v multiple times to verify reliability
• Confirm no regressions in other topic-related tests
• Consider adding a new test case that explicitly validates timing behavior

Task 2: Add Changelog Entry

Document the fix according to project contribution guidelines.

• Add changelog entry describing the bug fix
• Mention the root cause (clock drift causing start_time filtering)
• Note any new features added (readiness signaling if implemented)

🤖 Prompt for AI agents

This phase ensures the fix is complete with proper validation and documentation.

**Task 1: Verify All Topic Message Tests Pass Reliably**
- Run `pytest tests/integration/topic_message_query_e2e_test.py -v` multiple
times to verify the intermittent failures are resolved
- Verify no regressions in other topic-related tests
- Consider adding a new test case that explicitly validates timing behavior
(e.g., testing with different start_time offsets)

**Task 2: Add Changelog Entry**
- Locate and open the project's changelog file (typically CHANGELOG.md or
similar)
- Add a new entry describing the bug fix
- Mention the root cause: clock drift between client and Hedera consensus nodes
causing start_time to filter out messages
- Document any new features added, specifically the subscription readiness
signaling mechanism if implemented
- Follow the project's existing changelog format and contribution guidelines

Research

The Hiero Python SDK uses TopicMessageQuery to subscribe to Hedera Consensus Service (HCS) messages via gRPC streaming from mirror nodes. The subscribe() method spawns a background thread that connects to the mirror node using a start_time parameter as a server-side filter. Messages with consensus timestamps >= start_time are streamed to the client. The SubscriptionHandle class provides basic cancellation support but lacks readiness signaling. Integration tests use fixed time.sleep() delays to handle timing, which is fragile and doesn't account for clock drift between the client and Hedera's consensus nodes.

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

This phase adds diagnostic tooling and logging to help identify timing issues in
the TopicMessageQuery subscription pipeline.

**Task 1: Create Diagnostic Reproduction Script**
- Create a new file `examples/debug/topic_message_timing_diagnostic.py`
- Implement comprehensive logging at each step of the subscription and message
submission flow
- Log client-side timestamps at: start_time capture, subscription start, message
submission, and message receipt
- Extract and log the consensus timestamp from the transaction receipt for
comparison with start_time
- Add configurable parameters for start_time offset to test different timing
scenarios
- Print clear diagnostic output showing timestamp comparisons to identify clock
drift between client and Hedera network

**Task 2: Add Logging to TopicMessageQuery**
- Update `src/hiero_sdk_python/query/topic_message_query.py`
- Import Python's standard `logging` module
- Add logging statements at key points:
  - When subscription thread starts
  - When gRPC stream is established
  - For each message received (include consensus timestamp)
  - For retry attempts and backoff delays (include timestamps)
  - For stream completion and any errors encountered
- Use appropriate log levels: DEBUG for verbose details, INFO for key events
===============================================================================

Task: 2

This phase fixes the timing race condition by adjusting start_time handling and
improving test cleanup patterns.

**Task 1: Fix start_time Configuration in Tests**
- Open `tests/integration/topic_message_query_e2e_test.py`
- In `test_topic_message_query_receives_messages`, update the start_time to
`datetime.now(timezone.utc) - timedelta(seconds=5)` instead of
`datetime.now(timezone.utc)`
- Apply the same change to `test_topic_message_query_large_message_chunking`
- For `test_topic_message_query_limit`, evaluate if the same fix is needed (it
has no error handler so may not have been reporting failures)
- Add a comment above the start_time assignment explaining: "Set start_time in
the past to account for clock drift between client and Hedera consensus nodes"
- This is safe because tests create new topics with no prior messages

**Task 2: Improve Test Cleanup Patterns**
- In `tests/integration/topic_message_query_e2e_test.py`, wrap subscription
handling in `try/finally` blocks for all three tests
- Ensure `handle.cancel()` is always called in the `finally` block to prevent
resource leaks
- Consider adding topic deletion in cleanup matching the pattern from
`topic_message_submit_transaction_e2e_test.py`
- Add error handlers to tests that currently lack them for better failure
visibility
===============================================================================

Task: 3

This phase adds optional subscription readiness signaling to enable more robust
synchronization patterns when using TopicMessageQuery.

**Task 1: Extend SubscriptionHandle with Ready State**
- Open `src/hiero_sdk_python/utils/subscription_handle.py`
- Import `threading.Event` if not already imported
- Add a `_ready` instance variable of type `threading.Event` in the constructor,
initialized as unset
- Add `is_ready()` method that returns the ready state as a boolean
- Add `wait_until_ready(timeout: float = None)` method that blocks until ready
or timeout, returning True if ready or False on timeout
- Add internal `_set_ready()` method to signal readiness by setting the event

**Task 2: Signal Readiness from TopicMessageQuery**
- Open `src/hiero_sdk_python/query/topic_message_query.py`
- After successfully calling `client.mirror_stub.subscribeTopic(request)` and
establishing the gRPC stream, call `subscription_handle._set_ready()`
- Place this call before entering the message iteration loop
- Ensure readiness is only signaled once per subscription attempt, not on
retries
- This allows callers to optionally wait for the subscription to be established
before proceeding

**Task 3: Update Examples with Recommended Pattern**
- Open `examples/query/topic_message_query.py`
- Update to show the recommended `start_time` offset pattern using
`datetime.now(timezone.utc) - timedelta(seconds=5)`
- Add comments explaining why setting start_time in the past is recommended for
reliability (clock drift between client and consensus nodes)
- Optionally demonstrate using `wait_until_ready()` if the new mechanism is
implemented
- Ensure the example follows patterns consistent with other examples in the
codebase
===============================================================================

Task: 4

This phase ensures the fix is complete with proper validation and documentation.

**Task 1: Verify All Topic Message Tests Pass Reliably**
- Run `pytest tests/integration/topic_message_query_e2e_test.py -v` multiple
times to verify the intermittent failures are resolved
- Verify no regressions in other topic-related tests
- Consider adding a new test case that explicitly validates timing behavior
(e.g., testing with different start_time offsets)

**Task 2: Add Changelog Entry**
- Locate and open the project's changelog file (typically CHANGELOG.md or
similar)
- Add a new entry describing the bug fix
- Mention the root cause: clock drift between client and Hedera consensus nodes
causing start_time to filter out messages
- Document any new features added, specifically the subscription readiness
signaling mechanism if implemented
- Follow the project's existing changelog format and contribution guidelines

💡 Iterate on the plan with: @coderabbitai <feedback>

Example Feedback
- `@coderabbitai` You can skip phase 3. Add a simple unit test case for phase 2.
- `@coderabbitai` For design choice 1 go ahead with option 3 and replan.

---

💬 Have feedback or questions? Drop into our discord!

[github-actions]

Hi @manishdait 👋

This issue has been assigned but no pull request has been created yet.
Are you still planning on working on it?
If you are, please create a draft PR linked to this issue or comment /working to let us know.
If you’re no longer able to work on this issue, you can comment /unassign to release it.

From the Python SDK Team

[manishdait]

/working