Stop the broker schema monitor spinning on "starting" when the agent is unreachable

frontend/src/pages/team/Brokers/Hierarchy/TopicInspector/index.vue

@@ -51,6 +51,7 @@ import { mapState } from 'pinia'
 
 import brokerApi from '../../../../../api/broker.js'
 import EmptyState from '../../../../../components/EmptyState.vue'
+import alerts from '../../../../../services/alerts.js'
 
 import MainTitle from '../components/MainTitle.vue'
 
@@ -185,9 +186,16 @@ export default {
             delete this.localSegment.metadata.schema
             this.saveTopicMeta()
         },
-        startTeamBrokerAgent () {
-            brokerApi.startBroker(this.team.id, 'team-broker')
-            // alerts.emit('MQTT Payload Schema will be collected for the next 24 hours', 'confirmation')
+        async startTeamBrokerAgent () {
+            try {
+                await brokerApi.startBroker(this.team.id, 'team-broker')
+            } catch (err) {
+                // The request failed outright, so the toggle isn't really on. Put it
+                // back and let the user know rather than leaving it looking enabled.
+                this.agentActive = false
+                alerts.emit('Could not reach the broker agent. Please try again.', 'warning')
+                console.error(err)
+            }
         }
     }
 }

frontend/src/pages/team/Brokers/index.vue

@@ -85,6 +85,7 @@ import EmptyState from '../../../components/EmptyState.vue'
 import FfLoading from '../../../components/Loading.vue'
 
 import usePermissions from '../../../composables/Permissions.js'
+import alerts from '../../../services/alerts.js'
 import { Roles } from '../../../utils/roles.js'
 
 import BrokerStatusBadge from './components/BrokerStatusBadge.vue'
@@ -94,6 +95,10 @@ import { useAccountStore } from '@/stores/account.js'
 import { useContextStore } from '@/stores/context.js'
 import { useProductBrokersStore } from '@/stores/product-brokers.js'
 
+// How long we wait on a broker that keeps reporting "still starting" before we
+// treat its agent as unreachable, stop spinning, and let the user try again.
+const STARTING_TIMEOUT = 60000
+
 export default {
     name: 'TeamBrokers',
     components: { BrokerStatusBadge, FfLoading, EmptyState, PlusIcon },
@@ -112,7 +117,12 @@ export default {
             loadingState: false,
             brokerStatusPollingInterval: null,
             brokerState: '',
-            errorCode: ''
+            errorCode: '',
+            // When the current broker first started reporting "still starting",
+            // and whether we've already given up waiting on it (UI-only, resets
+            // on a refresh, a successful status, or another attempt to start).
+            startingSince: null,
+            startingTimedOut: false
         }
     },
     computed: {
@@ -361,18 +371,22 @@ export default {
                     if (response?.state === 'running') {
                         if (response?.status?.connected) {
                             this.brokerState = 'connected'
+                            this.resetStartingTimeout()
                         } else if (response?.status?.error === 'error_getting_status') {
-                            this.brokerState = 'starting'
+                            this.applyStartingState()
                         } else if (this.activeBrokerId === 'team-broker') {
                             // The team broker itself is always up, only its schema agent is not
                             // collecting, so this is not a broker connection error
                             this.brokerState = 'suspended'
+                            this.resetStartingTimeout()
                         } else {
                             this.brokerState = 'error'
                             this.errorCode = response.status.error || 'ENOTFOUND'
+                            this.resetStartingTimeout()
                         }
                     } else {
                         this.brokerState = 'suspended'
+                        this.resetStartingTimeout()
                     }
                     if (this.brokerState !== 'error') {
                         this.errorCode = ''
@@ -386,22 +400,57 @@ export default {
                     this.loadingState = false
                 })
         },
+        // Keep showing 'starting' while the agent is coming up, but if it never
+        // reports in we stop waiting, drop back to an idle state so the toggle is
+        // usable again, and tell the user once. The flag keeps us from spinning or
+        // re-alerting on every poll; it clears on a refresh, a successful status,
+        // or another attempt to start.
+        applyStartingState () {
+            if (this.startingTimedOut) {
+                this.brokerState = 'suspended'
+                return
+            }
+            if (!this.startingSince) {
+                this.startingSince = Date.now()
+            }
+            if (Date.now() - this.startingSince > STARTING_TIMEOUT) {
+                this.startingTimedOut = true
+                this.brokerState = 'suspended'
+                alerts.emit('Could not reach the broker agent. Please try again.', 'warning')
+            } else {
+                this.brokerState = 'starting'
+            }
+        },
+        resetStartingTimeout () {
+            this.startingSince = null
+            this.startingTimedOut = false
+        },
         clearBrokerStatusPollingInterval () {
             this.brokerState = ''
+            this.resetStartingTimeout()
             if (this.brokerStatusPollingInterval) {
                 clearInterval(this.brokerStatusPollingInterval)
             }
         },
         async toggleAgent () {
             this.loadingState = true
-            const state = await brokerAPI.getBrokerStatus(this.team.id, this.$route.params.brokerId)
-            if (state.state === 'running') {
-                await brokerAPI.suspendBroker(this.team.id, this.$route.params.brokerId)
-                this.brokerState = 'suspending'
-                this.loadingState = false
-            } else {
-                await brokerAPI.startBroker(this.team.id, this.$route.params.brokerId)
-                this.brokerState = 'starting'
+            try {
+                const state = await brokerAPI.getBrokerStatus(this.team.id, this.$route.params.brokerId)
+                if (state.state === 'running' && !this.startingTimedOut) {
+                    await brokerAPI.suspendBroker(this.team.id, this.$route.params.brokerId)
+                    this.brokerState = 'suspending'
+                } else {
+                    // Fresh attempt to start, so give the agent the full window again
+                    this.resetStartingTimeout()
+                    await brokerAPI.startBroker(this.team.id, this.$route.params.brokerId)
+                    this.brokerState = 'starting'
+                }
+            } catch (err) {
+                this.resetStartingTimeout()
+                this.brokerState = 'suspended'
+                alerts.emit('Could not reach the broker agent. Please try again.', 'warning')
+                console.error(err)
+            } finally {
                 this.loadingState = false
             }
         }

test/unit/frontend/pages/team/Brokers.spec.js

@@ -2,20 +2,37 @@ import { beforeEach, describe, expect, test, vi } from 'vitest'
 
 import brokerAPI from '../../../../../frontend/src/api/broker.js'
 import Brokers from '../../../../../frontend/src/pages/team/Brokers/index.vue'
+import alerts from '../../../../../frontend/src/services/alerts.js'
 
 vi.mock('../../../../../frontend/src/api/broker.js', () => ({
-    default: { getBrokerStatus: vi.fn() }
+    default: { getBrokerStatus: vi.fn(), startBroker: vi.fn(), suspendBroker: vi.fn() }
+}))
+vi.mock('../../../../../frontend/src/services/alerts.js', () => ({
+    default: { emit: vi.fn() }
 }))
 
 // getBrokerState only reads and writes a handful of fields, so it is called
 // against a plain object rather than mounting the whole page
 function brokerPage (activeBrokerId) {
-    return { team: { id: 'team-1' }, activeBrokerId, brokerState: '', errorCode: '', loadingState: true }
+    return {
+        team: { id: 'team-1' },
+        activeBrokerId,
+        brokerState: '',
+        errorCode: '',
+        loadingState: true,
+        startingSince: null,
+        startingTimedOut: false,
+        applyStartingState: Brokers.methods.applyStartingState,
+        resetStartingTimeout: Brokers.methods.resetStartingTimeout
+    }
 }
 
 describe('Brokers page broker state', () => {
     beforeEach(() => {
         brokerAPI.getBrokerStatus.mockReset()
+        brokerAPI.startBroker.mockReset()
+        brokerAPI.suspendBroker.mockReset()
+        alerts.emit.mockReset()
     })
 
     test('a connected agent is connected', async () => {
@@ -47,4 +64,77 @@ describe('Brokers page broker state', () => {
         expect(page.brokerState).toBe('error')
         expect(page.errorCode).toBe('ENOTFOUND')
     })
+
+    test('an agent that cannot be reached shows as starting while within the wait window', async () => {
+        brokerAPI.getBrokerStatus.mockResolvedValue({ state: 'running', status: { connected: false, error: 'error_getting_status' } })
+        const page = brokerPage('team-broker')
+
+        await Brokers.methods.getBrokerState.call(page)
+
+        expect(page.brokerState).toBe('starting')
+        expect(page.startingTimedOut).toBe(false)
+        expect(alerts.emit).not.toHaveBeenCalled()
+    })
+
+    test('an agent that stays unreachable past the wait window stops spinning, alerts once, and is retryable', async () => {
+        brokerAPI.getBrokerStatus.mockResolvedValue({ state: 'running', status: { connected: false, error: 'error_getting_status' } })
+        const page = brokerPage('team-broker')
+        // Pretend it has been starting for longer than the timeout
+        page.startingSince = Date.now() - 120000
+
+        await Brokers.methods.getBrokerState.call(page)
+
+        expect(page.brokerState).toBe('suspended')
+        expect(page.startingTimedOut).toBe(true)
+        expect(alerts.emit).toHaveBeenCalledTimes(1)
+
+        // A further poll while still unreachable must not spin or alert again
+        await Brokers.methods.getBrokerState.call(page)
+        expect(page.brokerState).toBe('suspended')
+        expect(alerts.emit).toHaveBeenCalledTimes(1)
+    })
+
+    test('a successful status clears the timed-out state', async () => {
+        const page = brokerPage('team-broker')
+        page.startingTimedOut = true
+        page.startingSince = Date.now() - 120000
+        brokerAPI.getBrokerStatus.mockResolvedValue({ state: 'running', status: { connected: true, error: null } })
+
+        await Brokers.methods.getBrokerState.call(page)
+
+        expect(page.brokerState).toBe('connected')
+        expect(page.startingTimedOut).toBe(false)
+        expect(page.startingSince).toBe(null)
+    })
+
+    test('the toggle retries a start after a timeout even though the backend row is still running', async () => {
+        const page = brokerPage('broker-1')
+        page.$route = { params: { brokerId: 'broker-1' } }
+        page.startingTimedOut = true
+        page.startingSince = Date.now() - 120000
+        brokerAPI.getBrokerStatus.mockResolvedValue({ state: 'running', status: { connected: false, error: 'error_getting_status' } })
+        brokerAPI.startBroker.mockResolvedValue({})
+
+        await Brokers.methods.toggleAgent.call(page)
+
+        expect(brokerAPI.startBroker).toHaveBeenCalledTimes(1)
+        expect(brokerAPI.suspendBroker).not.toHaveBeenCalled()
+        expect(page.brokerState).toBe('starting')
+        expect(page.startingTimedOut).toBe(false)
+        expect(page.startingSince).toBe(null)
+        expect(page.loadingState).toBe(false)
+    })
+
+    test('the toggle still suspends a running agent that has not timed out', async () => {
+        const page = brokerPage('broker-1')
+        page.$route = { params: { brokerId: 'broker-1' } }
+        brokerAPI.getBrokerStatus.mockResolvedValue({ state: 'running', status: { connected: true, error: null } })
+        brokerAPI.suspendBroker.mockResolvedValue({})
+
+        await Brokers.methods.toggleAgent.call(page)
+
+        expect(brokerAPI.suspendBroker).toHaveBeenCalledTimes(1)
+        expect(brokerAPI.startBroker).not.toHaveBeenCalled()
+        expect(page.brokerState).toBe('suspending')
+    })
 })