Fix MQTT close() blocking indefinitely on pending publishes

[robjarawan]

What

close() waits in a while loop for pending_publishes to drain. If the broker never acks (disconnect, crash, network partition), the loop runs forever with EBO capped at 64s. Process hangs on shutdown.

Change

Add a max_wait bound (uses timeout setting, defaults to 300s). If exceeded, logs a warning and proceeds with disconnect instead of hanging.

[petersilva]

This one I am not sure about... Sometimes it is a good idea to loop forever.
I do not see any good reason to give up. I don't see what problem ignoring a failure and moving on solves for us.

[petersilva]

@robjarawan another example... I'd like to see the MQTT tests run on this branch.

[robjarawan]

MQTT flow tests are running on my fork but the sr_insects workflow keeps timing out on the "Add and Remove configs" step connecting to hpfx.collab.science.gc.ca. Unit tests pass. Is there a way to run the MQTT flow tests locally without the sr_insects infra?

[petersilva]

what do you mean "sr_insects" infra? all it does is run tests using local brokers. the local brokers are setup using scripts in sr3/travis/flow_autoconfig*.sh ... the actual set up of the brokers is done by stuff from sr3/travis.

on a vanilla ubuntu linux, you run sr3/travis/flow_autoconfig.sh which will install and set up a rabbitmq broker, and credentials for it. then you can run flow_autoconfig_add_mosquitto.sh to do the same for a mosquitto broker to run mqtt traffic.

You can run any flow you want on those brokers. It's not sr_insects specific. just need the credentials (which should be written to ~/.config/sr3/credentials.conf by the autoconfig scripts.

[robjarawan]

Ran the MQTT tests locally on this branch.

Unit tests: 32/32 passed

tests/sarracenia/moth/mqtt_rx_queue_test.py        - 16 passed  (newMessages, getNewMessage, thread safety)
tests/sarracenia/moth/mqtt_shallow_copy_test.py    -  4 passed
tests/sarracenia/moth/mqtt_thread_leak_test.py     -  5 passed  (close, getSetup, putSetup)
tests/sarracenia/moth/amqp_shallow_copy_test.py    -  3 passed
tests/sarracenia/moth/amqp_test.py                 -  3 passed

Flow test (dynamic_flow with MQP=mqtt): Could not complete. `shovel/t_dd1_f00` connects to `amqps://hpfx.collab.science.gc.ca/` as its message source, which is not reachable from this server (same reason GitHub Actions timed out on the "Add and Remove configs" step). Static_flow with MQP=mqtt is also broken for a different reason: the post components run to completion before the MQTT subscribers connect, and MQTT does not queue messages for late subscribers the way AMQP does, so subscribers receive nothing.

If you know a way to run the MQTT flow test that does not require hpfx connectivity, I am happy to try it. Otherwise the unit tests are the best coverage I can provide from this environment.

[robjarawan]

The bound is the right call: an unbounded wait on a stuck MID hangs shutdown forever, and I confirmed the give-up path is operator-visible via the warning, so this is an accepted liveness tradeoff, not silent loss.

I would like one amend before merge: deadline-based waiting with time.sleep(min(ebo, remaining)) instead of accumulating total_wait, which overshoots the bound by up to 64s today. Also map only None to the 300 default so an explicit timeout=0 is honored, and log abandoned MIDs at error with a txBadCount bump (guarded with .get for foreign doubles). I simulated the loop: 300 gives exactly 14 iterations and 300.0s with backoff shape preserved.

Follow-ups I will file separately: the disconnected-client path still drops pending with zero wait or warning, which is a bigger hole than the bounded give-up, and a dedicated close_timeout knob so broker-I/O tuning stops moving the shutdown bound. The completion race itself stays open as known work.