Stop a broker restart from silently killing a service's JMS consumer

docs/MESSAGING.md

@@ -316,8 +316,19 @@ redelivery policies, advisory watchdogs, DLQ tooling. Effort there has a shelf l
   `kubectl exec … tail /var/log/activemq/activemq.log`, which rotates after ~14 h.
   Making these brokers log to stdout is an easy, worthwhile fix.
 - `VCMessagingServiceActiveMQ` uses a **bounded** failover URL so a wedged transport cannot
-  retry forever; `JmsFailoverWatchdog` runs a terminal action (in production, JVM exit so K8s
-  recycles the pod) when failover gives up.
+  retry forever; `JmsFailoverWatchdog` runs a terminal action when failover gives up. The four
+  long-lived consumer services (submit, sched, data, db) build theirs with
+  `createForLongLivedConsumerService()`, which sets that action to JVM exit so K8s recycles the
+  pod. Everything else — short-lived batch processes, the API server — keeps the `logOnly()`
+  default. Getting this wiring wrong is silent: for two releases every service was on
+  `logOnly()`, so the terminal condition was detected, logged at FATAL, and then ignored
+  (issue #2031).
+- A consumer can also lose its session without the transport noticing. `ConsumerContextJms`
+  routes that through `JmsFailoverWatchdog.onTerminalFailure(…)` rather than ending its poll
+  loop, because a consumer thread that exits leaves a process which consumes nothing and still
+  reports healthy. Note that `attach()` installs its `TransportListener` only for an
+  `ActiveMQConnection`. Both impls use the OpenWire client today so both are covered, but a move
+  to an AMQP client would leave this caller-side route as the *only* path to the terminal action.
 - `transportResumed` fires on a *first connect* as well as after an interruption. A resumed
   count with zero interruptions means new connections, not reconnects — a distinction that once
   cost a misdiagnosis.

vcell-server/src/main/java/cbit/vcell/message/jms/ConsumerContextJms.java

@@ -75,12 +75,28 @@ public void run(){
 //						lg.info(toString()+"no message received within "+CONSUMER_POLLING_INTERVAL_MS+" ms");
 				}
 			} catch (JMSException e) {
-				if (!bProcessing || e instanceof javax.jms.IllegalStateException){
-					// close() unblocks a thread parked in receive(); that is shutdown, not a
-					// failure. Logging it as one and looping would spin on the closed consumer.
+				if (!bProcessing){
+					// stop() has already been requested, and close() unblocks a thread parked
+					// in receive(); that is shutdown, not a failure. Logging it as one and
+					// looping would spin on the closed consumer. Every deliberate shutdown
+					// path (closeAll(), stopAndClose()) clears bProcessing before close(),
+					// so this test alone identifies them.
 					lg.debug(toString()+" consumer closed while polling", e);
 					break;
 				}
+				if (e instanceof javax.jms.IllegalStateException){
+					// The session died underneath a consumer we are still meant to be polling:
+					// a broker restart, or the failover transport exhausting its reconnect
+					// budget. receive() throws immediately from here on, so looping would spin
+					// -- but leaving quietly is worse. It leaves a process that consumes
+					// nothing, logs nothing and still reports healthy, which is how dev's
+					// submit service sat dead for 6h50m before anyone noticed (issue #2031).
+					// Escalate instead, and let the wiring decide what a lost broker means
+					// for this process.
+					vcMessagingServiceJms.getFailoverWatchdog()
+							.onTerminalFailure("consumer session for "+vcConsumer.getVCDestination(), e);
+					break;
+				}
 				onException(e);
 			} catch (RollbackException e) {
 				lg.error(e.getMessage(),e);

vcell-server/src/main/java/cbit/vcell/message/jms/JmsFailoverWatchdog.java

@@ -48,6 +48,25 @@ public static JmsFailoverWatchdog logOnly() {
 		return new JmsFailoverWatchdog(() -> {});
 	}
 
+	/**
+	 * Escalate a failure the caller detected for itself and cannot recover from in
+	 * place -- e.g. a consumer whose session died while it was still meant to be
+	 * polling. The {@link TransportListener} installed by {@link #attach} covers the
+	 * failures the failover layer reports; this covers the ones only the caller can
+	 * see. Both end in the same terminal action, so how a process responds to a lost
+	 * broker stays a single wiring decision.
+	 *
+	 * This route matters for more than tidiness: {@link #attach} installs a listener
+	 * only for {@link ActiveMQConnection}, so for any other provider it is the only
+	 * path to the terminal action at all.
+	 *
+	 * @param what short description of what was lost, e.g. {@code "transport"}
+	 */
+	public void onTerminalFailure(String what, Throwable cause) {
+		lg.fatal("JMS " + what + " unrecoverable, invoking terminal handler", cause);
+		onTerminal.run();
+	}
+
 	public void attach(Connection connection) {
 		if (!(connection instanceof ActiveMQConnection)) {
 			lg.warn("no failover watchdog for connection type {} -- a wedged transport will not "
@@ -63,8 +82,7 @@ public void onCommand(Object command) {
 			}
 			@Override
 			public void onException(IOException error) {
-				lg.fatal("JMS transport unrecoverable, invoking terminal handler", error);
-				onTerminal.run();
+				onTerminalFailure("transport", error);
 			}
 			@Override
 			public void transportInterupted() {

vcell-server/src/main/java/cbit/vcell/message/jms/activeMQ/VCMessagingServiceActiveMQ.java

@@ -13,6 +13,7 @@
 import cbit.vcell.message.VCMessagingException;
 import cbit.vcell.message.VCMessagingService;
 import cbit.vcell.message.VCellQueue;
+import cbit.vcell.message.jms.JmsFailoverWatchdog;
 import cbit.vcell.message.jms.VCMessagingServiceJms;
 import cbit.vcell.resource.PropertyLoader;
 
@@ -22,6 +23,28 @@ public class VCMessagingServiceActiveMQ extends VCMessagingServiceJms implements
 	public VCMessagingServiceActiveMQ() {
 		super();
 	}
+
+	/**
+	 * A messaging service for a long-lived server process whose only job is to consume
+	 * from the broker (submit, sched, data, db).
+	 *
+	 * The failover transport gives up after {@code maxReconnectAttempts} (set in
+	 * jmsUrl below), which is deliberate -- in K8s a pod restart is the right response
+	 * to a sustained broker outage. That only holds if something acts on it, so these
+	 * processes exit on a terminal failure and let K8s recycle them. Without it the
+	 * process stays up around a connection that can never be used again: dev's submit
+	 * service ran for 6h50m in that state, consuming nothing and still reporting healthy
+	 * (issue #2031).
+	 *
+	 * Short-lived batch processes (SolverPreprocessor, SolverPostprocessor,
+	 * JavaSimulationExecutable) and the API server keep the log-only default -- they
+	 * outlive neither the broker outage nor their own task.
+	 */
+	public static VCMessagingServiceActiveMQ createForLongLivedConsumerService() {
+		VCMessagingServiceActiveMQ service = new VCMessagingServiceActiveMQ();
+		service.setFailoverWatchdog(JmsFailoverWatchdog.jvmExitOnTerminal());
+		return service;
+	}
 
 	@Override
 	public ConnectionFactory createConnectionFactory(){

vcell-server/src/main/java/cbit/vcell/message/server/batch/sim/HtcSimulationWorker.java

@@ -87,12 +87,12 @@ public class HtcSimulationWorker implements HtcProxy.HtcProxyFactory  {
 public HtcSimulationWorker() {
 	this.htcProxy = SlurmProxy.createRemoteProxy();
 
-	this.vcMessagingService_int = new VCMessagingServiceActiveMQ();
+	this.vcMessagingService_int = VCMessagingServiceActiveMQ.createForLongLivedConsumerService();
 	String jmshost_int = PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntHostInternal);
 	int jmsport_int = Integer.parseInt(PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntPortInternal));
 	this.vcMessagingService_int.setConfiguration(new ServerMessagingDelegate(), jmshost_int, jmsport_int);
 
-	this.vcMessagingService_sim = new VCMessagingServiceActiveMQ();
+	this.vcMessagingService_sim = VCMessagingServiceActiveMQ.createForLongLivedConsumerService();
 	String jmshost_sim = PropertyLoader.getRequiredProperty(PropertyLoader.jmsSimHostInternal);
 	int jmsport_sim = Integer.parseInt(PropertyLoader.getRequiredProperty(PropertyLoader.jmsSimPortInternal));
 	this.vcMessagingService_sim.setConfiguration(new ServerMessagingDelegate(), jmshost_sim, jmsport_sim);

vcell-server/src/main/java/cbit/vcell/message/server/data/SimDataServer.java

@@ -66,7 +66,7 @@ public SimDataServer() throws Exception {
 
 	this.dataServerImpl = new DataServerImpl(dataSetControllerImpl, exportServiceImpl);
 
-	this.vcMessagingService_int = new VCMessagingServiceActiveMQ();
+	this.vcMessagingService_int = VCMessagingServiceActiveMQ.createForLongLivedConsumerService();
 	String jmshost = PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntHostInternal);
 	int jmsport = Integer.parseInt(PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntPortInternal));
 	this.vcMessagingService_int.setConfiguration(new ServerMessagingDelegate(), jmshost, jmsport);

vcell-server/src/main/java/cbit/vcell/message/server/db/DatabaseServer.java

@@ -90,7 +90,7 @@ public DatabaseServer() throws SQLException, DataAccessException {
 	KeyFactory keyFactory = conFactory.getKeyFactory();
 	this.databaseServerImpl = new DatabaseServerImpl(conFactory, keyFactory);
 
-	this.vcMessagingService_int = new VCMessagingServiceActiveMQ();
+	this.vcMessagingService_int = VCMessagingServiceActiveMQ.createForLongLivedConsumerService();
 	String jmshost = PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntHostInternal);
 	int jmsport = Integer.parseInt(PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntPortInternal));
 	this.vcMessagingService_int.setConfiguration(new ServerMessagingDelegate(), jmshost, jmsport);

vcell-server/src/main/java/cbit/vcell/message/server/dispatcher/SimulationDispatcher.java

@@ -702,12 +702,12 @@ public static SimulationDispatcher simulationDispatcherCreator() throws SQLExcep
 		AdminDBTopLevel adminDbTopLevel = new AdminDBTopLevel(conFactory);
 		SimulationDatabase simulationDatabase = new SimulationDatabaseDirect(adminDbTopLevel, databaseServerImpl, true);
 
-		VCMessagingService vcMessagingServiceInternal = new VCMessagingServiceActiveMQ();
+		VCMessagingService vcMessagingServiceInternal = VCMessagingServiceActiveMQ.createForLongLivedConsumerService();
 		String jmshost_int = PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntHostInternal);
 		int jmsport_int = Integer.parseInt(PropertyLoader.getRequiredProperty(PropertyLoader.jmsIntPortInternal));
 		vcMessagingServiceInternal.setConfiguration(new ServerMessagingDelegate(), jmshost_int, jmsport_int);
 
-		VCMessagingService vcMessagingServiceSim = new VCMessagingServiceActiveMQ();
+		VCMessagingService vcMessagingServiceSim = VCMessagingServiceActiveMQ.createForLongLivedConsumerService();
 		String jmshost_sim = PropertyLoader.getRequiredProperty(PropertyLoader.jmsSimHostInternal);
 		int jmsport_sim = Integer.parseInt(PropertyLoader.getRequiredProperty(PropertyLoader.jmsSimPortInternal));
 		vcMessagingServiceSim.setConfiguration(new ServerMessagingDelegate(), jmshost_sim, jmsport_sim);