From 8f6d53a0292dedd826af68d282b25f6ba8632f97 Mon Sep 17 00:00:00 2001 From: jamesread Date: Mon, 15 Jun 2026 00:50:04 +0100 Subject: [PATCH] chore: tweak reconnect login to avoid reconnect spam --- frontend/js/websocket.js | 65 ++++++++++++++++++++++--- frontend/resources/vue/ActionButton.vue | 4 +- frontend/resources/vue/App.vue | 4 ++ service/internal/api/api.go | 63 +++++++++++++++++++----- 4 files changed, 115 insertions(+), 21 deletions(-) diff --git a/frontend/js/websocket.js b/frontend/js/websocket.js index 0509358..e559126 100644 --- a/frontend/js/websocket.js +++ b/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) } diff --git a/frontend/resources/vue/ActionButton.vue b/frontend/resources/vue/ActionButton.vue index 677aebb..56335e2 100644 --- a/frontend/resources/vue/ActionButton.vue +++ b/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) { diff --git a/frontend/resources/vue/App.vue b/frontend/resources/vue/App.vue index dccdaa0..bfaa3cb 100644 --- a/frontend/resources/vue/App.vue +++ b/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() { diff --git a/service/internal/api/api.go b/service/internal/api/api.go index 54affb5..ebbc798 100644 --- a/service/internal/api/api.go +++ b/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) } }