Quieten expected Platform-connectivity errors in the token watcher (#1085)

* Quieten expected Platform-connectivity errors in the token watcher

When the workflow-status lookup fails because the run's Platform endpoint is
briefly unreachable (e.g. a paired self-hosted instance whose pairing
WebSocket times out), the watcher caught it in the generic catch-all and
logged a full ERROR stack trace on every retry. For chronically-unreachable
paired endpoints this floods the logs and inflates error-rate signals, even
though the behavior is correct (the token is not renewed and lapses
fail-closed, then the entry stops on the next 'gone' check).

Distinguish these transient connectivity failures (TimeoutException /
HttpResponseException in the cause chain) from genuinely unexpected errors:
log the former as a concise one-line warning, keep the ERROR+stack trace for
the latter. Retry and the wave.tokens.refresh{result=error} counter are
unchanged for both.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Bound cause-chain walk to guard against cyclic references

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Replace empty for-loop with while in rootCauseMessage

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Simplify cause-chain walk per review feedback

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: pditommaso

src/main/groovy/io/seqera/wave/service/request/ContainerRequestServiceImpl.groovy

@@ -20,13 +20,15 @@ package io.seqera.wave.service.request
 
 import java.time.Duration
 import java.time.Instant
+import java.util.concurrent.TimeoutException
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import io.micrometer.core.instrument.MeterRegistry
 import io.micronaut.core.annotation.Nullable
 import io.micronaut.scheduling.TaskScheduler
 import io.seqera.wave.configuration.ContainerRequestConfig
+import io.seqera.wave.exception.HttpResponseException
 import io.seqera.wave.service.persistence.PersistenceService
 import io.seqera.wave.service.persistence.WaveContainerRecord
 import io.seqera.wave.service.request.ContainerRequestRange.Entry
@@ -132,6 +134,33 @@ class ContainerRequestServiceImpl implements ContainerRequestService {
         meterRegistry?.counter('wave.tokens.refresh', 'result', result)?.increment()
     }
 
+    private static final int MAX_CAUSE_DEPTH = 50
+
+    /**
+     * Whether the given error is an expected Platform connectivity failure (a timeout or an HTTP
+     * error reaching Tower) as opposed to an unexpected bug. Such failures are transient and retried,
+     * so they should not be logged at ERROR with a full stack trace. The cause chain is walked up to
+     * {@link #MAX_CAUSE_DEPTH} to guard against cyclic cause references.
+     */
+    protected static boolean isConnectivityError(Throwable t) {
+        Throwable e = t
+        for( int i = 0; e != null && i < MAX_CAUSE_DEPTH; e = e.cause, i++ ) {
+            if( e instanceof TimeoutException || e instanceof HttpResponseException )
+                return true
+        }
+        return false
+    }
+
+    protected static String rootCauseMessage(Throwable t) {
+        if( t == null )
+            return null
+        Throwable e = t
+        int i = 0
+        while( e.cause != null && e.cause != e && i++ < MAX_CAUSE_DEPTH )
+            e = e.cause
+        return e.message ?: e.class.simpleName
+    }
+
     protected void scheduleRefresh(Entry entry) {
         // Re-check this request one refreshInterval from now. refreshInterval is deliberately shorter
         // than accessTtl so several checks fit inside a token's lifetime — a transient Tower failure
@@ -173,11 +202,20 @@ class ContainerRequestServiceImpl implements ContainerRequestService {
                 Thread.currentThread().interrupt()
             }
             catch (Throwable t) {
-                log.error("Unexpected error in container request watcher while processing key: $it", t)
                 meterRefresh('error')
                 // The entry was already popped by getEntriesUntil; re-add it so a transient failure
                 // (e.g. a Tower blip) does not silently drop the request and let its token lapse.
                 scheduleRefresh(it)
+                if( isConnectivityError(t) ) {
+                    // Expected when the workflow's Platform endpoint is briefly unreachable — e.g. a
+                    // paired self-hosted instance whose pairing WebSocket times out. The token is not
+                    // renewed, so it lapses fail-closed; log a concise warning rather than a full
+                    // stack trace to avoid flooding the logs while the entry is retried.
+                    log.warn "Container request watcher could not confirm workflow status - request=${it.requestId}; workflow=${it.workflowId}; cause=${rootCauseMessage(t)}; will retry"
+                }
+                else {
+                    log.error("Unexpected error in container request watcher while processing key: $it", t)
+                }
             }
         }
     }

src/test/groovy/io/seqera/wave/service/request/ContainerRequestServiceImplTest.groovy

@@ -22,10 +22,13 @@ import spock.lang.Specification
 
 import java.time.Duration
 import java.time.Instant
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeoutException
 
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry
 import io.micronaut.test.extensions.spock.annotation.MicronautTest
 import io.seqera.wave.configuration.ContainerRequestConfig
+import io.seqera.wave.exception.HttpResponseException
 import io.seqera.wave.service.persistence.PersistenceService
 import io.seqera.wave.service.persistence.WaveContainerRecord
 import io.seqera.wave.tower.PlatformId
@@ -253,4 +256,23 @@ class ContainerRequestServiceImplTest extends Specification {
         0 * range.add(_, _)
     }
 
+    def 'isConnectivityError should detect Tower connectivity failures in the cause chain'() {
+        expect:
+        ContainerRequestServiceImpl.isConnectivityError(ERR) == EXPECTED
+        where:
+        ERR                                                                                  | EXPECTED
+        new TimeoutException('nope')                                                         | true
+        new HttpResponseException(408, 'Timeout')                                            | true
+        new RuntimeException('wrap', new TimeoutException('nope'))                            | true
+        new ExecutionException(new HttpResponseException(408, 'Timeout'))                     | true
+        new RuntimeException('boom')                                                         | false
+        new NullPointerException()                                                           | false
+    }
+
+    def 'rootCauseMessage should return the deepest cause message'() {
+        expect:
+        ContainerRequestServiceImpl.rootCauseMessage(new RuntimeException('outer', new TimeoutException('inner'))) == 'inner'
+        ContainerRequestServiceImpl.rootCauseMessage(new RuntimeException('solo')) == 'solo'
+    }
+
 }