InvalidStateError in _on_msg

[mjmeli]

I'm seeing error logs from the Home Assistant integration that are coming from this library

Example:

2021-11-17 09:46:26 DEBUG (MainThread) [pyduke_energy.realtime] Socket readable, calling loop_read()
2021-11-17 09:46:26 DEBUG (MainThread) [pyduke_energy.realtime] Received PUBLISH (d0, q0, r0, m0), 'DESH/P210400903/out/sm/1/live', ...  (128 bytes)
2021-11-17 09:46:34 DEBUG (MainThread) [pyduke_energy.realtime] Socket readable, calling loop_read()
2021-11-17 09:46:34 DEBUG (MainThread) [pyduke_energy.realtime] Received PUBLISH (d0, q0, r0, m0), 'DESH/P210400903/out/sm/1/live', ...  (128 bytes)
2021-11-17 09:46:35 DEBUG (MainThread) [pyduke_energy.realtime] Socket readable, calling loop_read()
2021-11-17 09:46:35 DEBUG (MainThread) [pyduke_energy.realtime] Received PUBLISH (d0, q0, r0, m0), 'DESH/P210400903/out/sm/1/live', ...  (128 bytes)
2021-11-17 09:46:35 ERROR (MainThread) [pyduke_energy.realtime] Caught exception in on_message: invalid state
2021-11-17 09:46:35 ERROR (MainThread) [homeassistant] Error doing job: Exception in callback MqttConnHelper.on_socket_open.<locals>.call_bk() at /usr/local/lib/python3.9/site-packages/pyduke_energy/realtime.py:299
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/lib/python3.9/site-packages/pyduke_energy/realtime.py", line 302, in call_bk
    client.loop_read()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 1556, in loop_read
    rc = self._packet_read()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 2439, in _packet_read
    rc = self._packet_handle()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 3033, in _packet_handle
    return self._handle_publish()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 3327, in _handle_publish
    self._handle_on_message(message)
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 3570, in _handle_on_message
    on_message(self, self._userdata, message)
  File "/usr/local/lib/python3.9/site-packages/pyduke_energy/realtime.py", line 168, in _on_msg
    self.rx_msg.set_result((msg.payload.decode("utf8")))
asyncio.exceptions.InvalidStateError: invalid state

See mjmeli/ha-duke-energy-gateway#36 for more context

I'm wondering if it has to do with the messages being received so close together. The timestamp only has second precision but they might be <1 sec apart.

[mjmeli]

@CmpCtrl hey! I'm wondering if you have seen an error like this in your usage? I get it over a period of time, like 15-30 minutes, every couple of days.

[CmpCtrl]

Yes, i see a similar symptom. I think its happening when the server forcefully closes the connection, but i don't know why that happens sometimes. I haven't looked into how to handle that better in the error checking either, it does seem to recover but yea takes 15 min or so.

I probably wont have a chance to look at it real soon, but i'll try to review it at some point. Im not sure if we'll be able to help the root cause, but should be able to handle the error more elegantly i'd think.

[mjmeli]

I've seen it in 2 cases. one does look like the server closing the connection. the other looks like its related to receiving 2 events really close to each other. here's another log with my on message handler log included (it's the "New measurement received") one. notice how two publishes come in (09:34:26 and 09:34:27), then there is the error and one successful message at 09:34:28. So I have to imagine one of the publishes is hitting the error and one is being handled.

It feels like they are both trying to be handled by the msg handler at the same time but one one is able to do set_result

Nothing actually fails from this, it immediately starts receiving events again

I might just put in some code to handle this. I can just check to make sure the result hasn't been set yet.

2021-11-17 09:34:26 DEBUG (MainThread) [pyduke_energy.realtime] Socket readable, calling loop_read()
2021-11-17 09:34:26 DEBUG (MainThread) [pyduke_energy.realtime] Received PUBLISH (d0, q0, r0, m0), 'DESH/P210400903/out/sm/1/live', ...  (128 bytes)
2021-11-17 09:34:27 DEBUG (MainThread) [pyduke_energy.realtime] Sending PINGREQ
2021-11-17 09:34:27 DEBUG (MainThread) [pyduke_energy.realtime] Watching socket for writability.
2021-11-17 09:34:27 DEBUG (MainThread) [pyduke_energy.realtime] Socket readable, calling loop_read()
2021-11-17 09:34:27 DEBUG (MainThread) [pyduke_energy.realtime] Received PUBLISH (d0, q0, r0, m0), 'DESH/P210400903/out/sm/1/live', ...  (128 bytes)
2021-11-17 09:34:27 ERROR (MainThread) [pyduke_energy.realtime] Caught exception in on_message: invalid state
2021-11-17 09:34:27 ERROR (MainThread) [homeassistant] Error doing job: Exception in callback MqttConnHelper.on_socket_open.<locals>.call_bk() at /usr/local/lib/python3.9/site-packages/pyduke_energy/realtime.py:299
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/lib/python3.9/site-packages/pyduke_energy/realtime.py", line 302, in call_bk
    client.loop_read()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 1556, in loop_read
    rc = self._packet_read()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 2439, in _packet_read
    rc = self._packet_handle()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 3033, in _packet_handle
    return self._handle_publish()
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 3327, in _handle_publish
    self._handle_on_message(message)
  File "/usr/local/lib/python3.9/site-packages/paho/mqtt/client.py", line 3570, in _handle_on_message
    on_message(self, self._userdata, message)
  File "/usr/local/lib/python3.9/site-packages/pyduke_energy/realtime.py", line 168, in _on_msg
    self.rx_msg.set_result((msg.payload.decode("utf8")))
asyncio.exceptions.InvalidStateError: invalid state
2021-11-17 09:34:28 DEBUG (MainThread) [custom_components.duke_energy_gateway] New measurement received: 855.000000
2021-11-17 09:34:28 DEBUG (MainThread) [pyduke_energy.realtime] Socket readable, calling loop_read()
2021-11-17 09:34:28 DEBUG (MainThread) [pyduke_energy.realtime] Received PUBLISH (d0, q0, r0, m0), 'DESH/P210400903/out/sm/1/live', ...  (128 bytes)
2021-11-17 09:34:28 DEBUG (MainThread) [custom_components.duke_energy_gateway] New measurement received: 855.000000

[mjmeli]

Yeah I was able to reproduce this in the Paho MQTT asyncio example by publishing two events at the same time. Looking for a fix now.

[CmpCtrl]

interesting. yea that example is where i got the framework for that part of the code. I thought it was needed to work with the asyncio parts of your existing library, and that it could handle pretty high rate msgs. But, maybe thats not a good approach or not using the paho-mqtt library correctly.

[mjmeli]

yeah normally it handles it fine but there's some weird edge case where it's hitting this. it might be an incorrect use of asyncio somewhere but nothing stands out to me. I'm just going to put in logic to handle this where the message will just be skipped.