chore: tweak reconnect login to avoid reconnect spam

frontend/js/websocket.js

@@ -2,23 +2,63 @@ import { buttonResults } from '../resources/vue/stores/buttonResults.js'
 import { rateLimits } from '../resources/vue/stores/rateLimits.js'
 import { connectionState } from '../resources/vue/stores/connectionState.js'
 
-const RECONNECT_DELAYS_MS = [0, 1000, 2000, 4000, 8000, 16000, 32000]
+const RECONNECT_DELAYS_MS = [200, 1000, 2000, 4000, 8000, 16000, 32000]
 const BANNER_DELAY_MS = 2000
 
 let reconnectAttempt = 0
 let reconnectTimer = null
+let listenersInitialized = false
 
-export function initWebsocket () {
-  window.addEventListener('EventOutputChunk', onOutputChunk)
-  window.addEventListener('EventExecutionStarted', onExecutionChanged)
-  window.addEventListener('EventExecutionFinished', onExecutionChanged)
+function shouldConnectEventStream () {
+  return window.initResponse && !window.initResponse.loginRequired
+}
+
+export function stopEventStream () {
+  if (reconnectTimer != null) {
+    clearTimeout(reconnectTimer)
+    reconnectTimer = null
+  }
+
+  reconnectAttempt = 0
+  connectionState.connected = false
+  connectionState.reconnecting = false
+  connectionState.scheduledReconnectDelayMs = 0
+  connectionState.nextReconnectAt = null
+  connectionState.showDisconnectedBanner = false
+  window.websocketAvailable = false
+}
+
+export function connectEventStreamIfNeeded () {
+  if (!shouldConnectEventStream()) {
+    stopEventStream()
+    return
+  }
+
+  if (window.websocketAvailable || reconnectTimer != null) {
+    return
+  }
 
   reconnectWebsocket()
 }
 
+export function initWebsocket () {
+  if (!listenersInitialized) {
+    window.addEventListener('EventOutputChunk', onOutputChunk)
+    window.addEventListener('EventExecutionStarted', onExecutionChanged)
+    window.addEventListener('EventExecutionFinished', onExecutionChanged)
+    listenersInitialized = true
+  }
+
+  connectEventStreamIfNeeded()
+}
+
 window.websocketAvailable = false
 
 export function requestReconnectNow () {
+  if (!shouldConnectEventStream()) {
+    return
+  }
+
   if (window.websocketAvailable) {
     return
   }
@@ -29,7 +69,7 @@ export function requestReconnectNow () {
   }
 
   reconnectAttempt = 0
-  scheduleReconnect(0)
+  scheduleReconnect(RECONNECT_DELAYS_MS[0])
 }
 
 function scheduleReconnect (delayMs) {
@@ -57,6 +97,10 @@ function updateBannerVisibility () {
 }
 
 async function reconnectWebsocket () {
+  if (!shouldConnectEventStream()) {
+    return
+  }
+
   if (window.websocketAvailable) {
     return
   }
@@ -78,8 +122,10 @@ async function reconnectWebsocket () {
     connectionState.nextReconnectAt = null
     connectionState.scheduledReconnectDelayMs = 0
     connectionState.showDisconnectedBanner = false
-    reconnectAttempt = 0
     for await (const e of stream) {
+      if (reconnectAttempt !== 0) {
+        reconnectAttempt = 0
+      }
       handleEvent(e)
     }
   } catch (err) {
@@ -94,6 +140,11 @@ async function reconnectWebsocket () {
   const delay = RECONNECT_DELAYS_MS[Math.min(reconnectAttempt, RECONNECT_DELAYS_MS.length - 1)]
   reconnectAttempt++
   console.log('Reconnecting websocket in ' + delay + 'ms...')
+
+  if (!shouldConnectEventStream()) {
+    return
+  }
+
   scheduleReconnect(delay)
 }
 

frontend/resources/vue/ActionButton.vue

@@ -214,8 +214,10 @@ function getUniqueId() {
 
 async function pollExecutionUntilDone (trackingId) {
   const pollIntervalMs = 500
+  const pollTimeoutMs = 10 * 60 * 1000
+  const deadline = Date.now() + pollTimeoutMs
 
-  while (!connectionState.connected) {
+  while (Date.now() < deadline) {
     try {
       const result = await window.client.executionStatus({ executionTrackingId: trackingId })
       if (result.logEntry) {

frontend/resources/vue/App.vue

@@ -100,6 +100,7 @@ import Sidebar from 'picocrank/vue/components/Sidebar.vue';
 import Navigation from 'picocrank/vue/components/Navigation.vue';
 import Header from 'picocrank/vue/components/Header.vue';
 import ConnectionBanner from './components/ConnectionBanner.vue';
+import { connectEventStreamIfNeeded } from '../../js/websocket.js';
 import { HugeiconsIcon } from '@hugeicons/vue'
 import { Menu01Icon } from '@hugeicons/core-free-icons'
 import { UserCircle02Icon } from '@hugeicons/core-free-icons'
@@ -237,9 +238,12 @@ function updateHeaderFromInit() {
     applyTheme()
 
     if (window.initResponse.loginRequired) {
+        connectEventStreamIfNeeded()
         router.push('/login')
         return
     }
+
+    connectEventStreamIfNeeded()
 }
 
 function renderNavigation() {

service/internal/api/api.go

@@ -58,18 +58,43 @@ func (api *oliveTinAPI) copyOfStreamingClients() []*streamingClient {
 type streamingClient struct {
 	channel           chan *apiv1.EventStreamResponse
 	AuthenticatedUser *authpublic.AuthenticatedUser
+	heartbeatStopOnce sync.Once
+	heartbeatStop     chan struct{}
+	heartbeatDone     chan struct{}
 }
 
-// trySendEventToClient sends msg to the client's channel. Returns false if the channel is full (client should be removed).
+func (c *streamingClient) stopHeartbeat() {
+	if c.heartbeatStop == nil || c.heartbeatDone == nil {
+		return
+	}
+	c.heartbeatStopOnce.Do(func() {
+		close(c.heartbeatStop)
+	})
+	<-c.heartbeatDone
+}
+
+// trySendEventToClient sends msg to the client's channel. Returns false if the channel is full or closed.
 func (api *oliveTinAPI) trySendEventToClient(client *streamingClient, msg *apiv1.EventStreamResponse) bool {
 	if client == nil || msg == nil {
 		return false
 	}
+	sent := sendToStreamingClientChannel(client.channel, msg)
+	if !sent {
+		log.Warnf("EventStream: client channel is full or closed, removing client")
+	}
+	return sent
+}
+
+func sendToStreamingClientChannel(ch chan *apiv1.EventStreamResponse, msg *apiv1.EventStreamResponse) (sent bool) {
+	defer func() {
+		if recover() != nil {
+			sent = false
+		}
+	}()
 	select {
-	case client.channel <- msg:
+	case ch <- msg:
 		return true
 	default:
-		log.Warnf("EventStream: client channel is full, removing client")
 		return false
 	}
 }
@@ -933,6 +958,8 @@ func (api *oliveTinAPI) EventStream(ctx ctx.Context, req *connect.Request[apiv1.
 	client := &streamingClient{
 		channel:           make(chan *apiv1.EventStreamResponse, 10), // Buffered channel to hold Events
 		AuthenticatedUser: user,
+		heartbeatStop:     make(chan struct{}),
+		heartbeatDone:     make(chan struct{}),
 	}
 
 	log.WithFields(log.Fields{
@@ -943,9 +970,7 @@ func (api *oliveTinAPI) EventStream(ctx ctx.Context, req *connect.Request[apiv1.
 	api.streamingClients[client] = struct{}{}
 	api.streamingClientsMutex.Unlock()
 
-	heartbeatDone := make(chan struct{})
-	defer close(heartbeatDone)
-	go api.sendEventStreamHeartbeats(heartbeatDone, client)
+	go api.sendEventStreamHeartbeats(client)
 
 	// loop over client channel and send events to connectedClient
 	for msg := range client.channel {
@@ -963,12 +988,21 @@ func (api *oliveTinAPI) EventStream(ctx ctx.Context, req *connect.Request[apiv1.
 	return nil
 }
 
-func (api *oliveTinAPI) sendEventStreamHeartbeats(done <-chan struct{}, client *streamingClient) {
+func (api *oliveTinAPI) sendEventStreamHeartbeats(client *streamingClient) {
+	defer close(client.heartbeatDone)
+
+	if !api.sendEventStreamHeartbeat(client) {
+		return
+	}
+
 	ticker := time.NewTicker(10 * time.Second)
 	defer ticker.Stop()
+	api.runEventStreamHeartbeatLoop(client, ticker)
+}
 
+func (api *oliveTinAPI) runEventStreamHeartbeatLoop(client *streamingClient, ticker *time.Ticker) {
 	for {
-		if api.waitEventStreamHeartbeatOrDone(done, ticker) {
+		if api.waitEventStreamHeartbeatOrDone(client.heartbeatStop, ticker) {
 			return
 		}
 		if !api.sendEventStreamHeartbeat(client) {
@@ -1000,8 +1034,13 @@ func (api *oliveTinAPI) removeClient(clientToRemove *streamingClient) {
 		return
 	}
 	api.streamingClientsMutex.Lock()
+	if _, exists := api.streamingClients[clientToRemove]; !exists {
+		api.streamingClientsMutex.Unlock()
+		return
+	}
 	delete(api.streamingClients, clientToRemove)
 	api.streamingClientsMutex.Unlock()
+	clientToRemove.stopHeartbeat()
 	close(clientToRemove.channel)
 }
 
@@ -1009,14 +1048,12 @@ func (api *oliveTinAPI) OnActionMapRebuilt() {
 	toRemove := []*streamingClient{}
 
 	for _, client := range api.copyOfStreamingClients() {
-		select {
-		case client.channel <- &apiv1.EventStreamResponse{
+		msg := &apiv1.EventStreamResponse{
 			Event: &apiv1.EventStreamResponse_ConfigChanged{
 				ConfigChanged: &apiv1.EventConfigChanged{},
 			},
-		}:
-		default:
-			log.Warnf("EventStream: client channel is full, removing client")
+		}
+		if !api.trySendEventToClient(client, msg) {
 			toRemove = append(toRemove, client)
 		}
 	}