Gracefully shut down st2 engines (workflow_engine, scheduler, rulesengine, actionrunner) on RabbitMQ failure to prevent stuck running tasks

[guzzijones]

Summary

When RabbitMQ becomes unreachable, several StackStorm engine services (most notably st2workflowengine, but also st2scheduler, st2rulesengine, and st2actionrunner) do not exit their process. Instead, the internal Kombu ConsumerMixin retries the broker connection infinitely, and the consumer/dispatcher greenlets die silently. The pod/process stays "up" from Kubernetes' perspective (liveness passes, PID 1 is alive), but the service is effectively dead — it holds no AMQP subscription, publishes no state transitions, and cannot process further work.

The visible symptom in production is that workflows/tasks are left stuck in the running state indefinitely, because:

1. The action runner completed the task and tried to publish the result to RabbitMQ.
2. Publishing failed (broker was down or connection had been reset).
3. The exception was swallowed (try/except: LOG.exception(...)), and the DB status update was left in an inconsistent state relative to what other services believed.
4. The workflow engine that would have advanced the workflow had already silently lost its consumer and never received the resume message even after the broker came back — but because it kept running, no orchestrator/pod restart occurred to recover it.

Root causes identified

1. Infinite reconnect loop in Kombu ConsumerMixin.on_connection_error. Consumers never give up, never raise, so the parent thread cannot detect the failure and the service cannot exit.
2. Consumer/entrypoint threads run detached from the main thread. In st2workflowengine, st2scheduler, and st2rulesengine, the main() returns / blocks in a way that a dead consumer greenlet does not propagate an error up to the process, so the pod is not restarted by Kubernetes.
3. DB writes and RabbitMQ publishes are not atomic. In st2common/persistence/base.py, Access.add_or_update and Access.update performed the DB write, then swallowed any exception from publish_create / publish_update / trigger dispatch. This left the DB "ahead" of the message bus and produced tasks that no downstream service ever saw.
4. No configurable ceiling on retry attempts. The messaging config had connection_retries / connection_retry_wait but nothing that enforced a maximum for the Kombu consumer loop or the publisher ensure() retry.
5. Scheduler had no bootstrap recovery for LiveActions in REQUESTED state whose scheduling-queue message had been lost during a prior broker outage.

What this branch changes

1. New ConnectionRetryMixin (st2common/st2common/transport/connection_retry_mixin.py)

A mixin for Kombu ConsumerMixin-based classes. It overrides on_connection_error to count attempts and re-raise the exception after messaging.connection_retry_max_attempts, breaking the infinite retry loop. on_connection_revived resets the counter. 0 = retry forever (backwards-compatible opt-out).

2. New messaging config options (st2common/config.py, conf/st2.conf.sample)

• messaging.connection_retry_max_attempts (default 10)
• messaging.connection_retry_interval_start (default 1)
• messaging.connection_retry_interval_step (default 1)
• messaging.connection_retry_interval_max (default 30)

These are now passed into the Kombu Connection as transport_options (see st2common/transport/utils.py) so both initial connect and reconnect during publish honour a real ceiling.

3. Graceful process exit when consumer/worker threads die

• st2actions/cmd/workflow_engine.py, st2actions/cmd/scheduler.py, and st2reactor/cmd/rulesengine.py now poll their consumer / handler / entrypoint greenlets. If any of them die, .wait() is called to re-raise the underlying exception, the service deregisters from the service registry, calls shutdown(), and returns a non-zero exit code. Kubernetes then restarts the pod cleanly instead of leaving a half-dead process running.
• Workflow engine except path now also calls deregister_service + engine.shutdown() before returning 1.

4. Publisher / consumer retry limits

• st2common/transport/connection_retry_wrapper.py refactored to honour connection_retry_max_attempts on ensure()-style publishes rather than looping forever.
• st2common/transport/consumers.py and st2actions/worker.py catch the raised broker exceptions (amqp_exceptions, kombu_exceptions) at the top level and let them propagate to shut the worker down.

5. DB/RabbitMQ atomicity — rollback on publish failure

st2common/persistence/base.py Access.add_or_update and Access.update no longer swallow publish/trigger-dispatch exceptions. If the RabbitMQ publish fails:

• Insert path → the newly created object is deleted.
• Update path → the pre-update state is restored using the original object snapshot and set__ rollback kwargs.
• The exception is re-raised so callers/consumers can NACK/requeue.

This is the direct fix for the "stuck in running" symptom: either the state transition is committed and published, or neither happens.

6. Scheduler bootstrap recovery

st2actions/scheduler/handler.py gains _bootstrap_missing_scheduling_queue_items, invoked from st2actions/cmd/scheduler.py on startup. It scans for REQUESTED LiveActions with no corresponding scheduling-queue entry (i.e. the enqueue message was lost during a prior RabbitMQ outage) and re-creates the queue item so the workflow can move forward after recovery. Covered by st2actions/tests/unit/test_scheduler_bootstrap_recovery.py.

7. Misc fixes bundled in the same branch

• st2common/services/triggerwatcher.py – uses the new retry mixin.
• st2api/controllers/v1/keyvalue.py + st2common/models/api/keyvalue.py – correctly parses user-scoped keys (unrelated bugfix caught during testing).
• st2common/transport/publishers.py, st2common/services/action.py – minor cleanup so channel close errors surface instead of being masked by except BaseException.

[guzzijones]

@nzlosh not sure if you want this in 3.10. I just set the milestone so we could see the green check mark for now.