fix intermittent issue

swelborn · swelborn · commit daf05b403784 · 2026-01-16T18:12:46.000-05:00
this fixes the following types of runtime errs:

```
NATS connection successful 2 nats.tsx:128:17Draining NATS connection
nats.tsx:168:17Failed to create consumer in stream "noti":
DrainingConnectionError: connection draining    DrainingConnectionError
errors.ts:152    _check nats.ts:89    request nats.ts:373    _request
jsbaseclient_api.ts:119    addUpdate jsmconsumer_api.ts:163    add
jsmconsumer_api.ts:180    createConsumer useConsumer.ts:68
useConsumer useConsumer.ts:89    React 37    performWorkUntilDeadline
scheduler.development.js:45    js scheduler.development.js:223    js
scheduler.development.js:364    __require2 chunk-PR4QN5HX.js:14    js
index.js:6    __require2 chunk-PR4QN5HX.js:14    React 2    __require2
chunk-PR4QN5HX.js:14    js React    __require2 chunk-PR4QN5HX.js:14
&lt;anonymous&gt; react-dom_client.js:20192useConsumer.ts:85:17Watch error for
bucket stat: DrainingConnectionError: connection draining
DrainingConnectionError errors.ts:152    _check nats.ts:89    request
nats.ts:373    _request jsbaseclient_api.ts:119    addUpdate
jsmconsumer_api.ts:163    add jsmconsumer_api.ts:180
getOrderedPushConsumer jsmstream_api.ts:206    getPushConsumer
jsmstream_api.ts:154    watch kv.ts:916    useBucketWatch
useBucketWatch.ts:63    useBucketWatch useBucketWatch.ts:119    React 33
performWorkUntilDeadline scheduler.development.js:45    js
scheduler.development.js:223    js scheduler.development.js:364
__require2 chunk-PR4QN5HX.js:14    js index.js:6    __require2
chunk-PR4QN5HX.js:14    React 2    __require2 chunk-PR4QN5HX.js:14    js
React    __require2 chunk-PR4QN5HX.js:14    &lt;anonymous&gt;
react-dom_client.js:20192useBucketWatch.ts:114:19Watch error for bucket
stat: DrainingConnectionError: connection draining
DrainingConnectionError errors.ts:152    _check nats.ts:89    request
nats.ts:373    _request jsbaseclient_api.ts:119    addUpdate
jsmconsumer_api.ts:163    add jsmconsumer_api.ts:180
getOrderedPushConsumer jsmstream_api.ts:206    getPushConsumer
jsmstream_api.ts:154    watch kv.ts:916    useBucketWatch
useBucketWatch.ts:63    useBucketWatch useBucketWatch.ts:119    React 29
performWorkUntilDeadline scheduler.development.js:45    js
scheduler.development.js:223    js scheduler.development.js:364
__require2 chunk-PR4QN5HX.js:14    js index.js:6    __require2
chunk-PR4QN5HX.js:14    React 2    __require2 chunk-PR4QN5HX.js:14    js
React    __require2 chunk-PR4QN5HX.js:14    &lt;anonymous&gt;
react-dom_client.js:20192useBucketWatch.ts:114:19    useBucketWatch
useBucketWatch.ts:114    useBucketWatch useBucketWatch.ts:119    React
29    performWorkUntilDeadline scheduler.development.js:45    (Async:
EventHandlerNonNull)    js scheduler.development.js:223    js
scheduler.development.js:364    __require2 chunk-PR4QN5HX.js:14    js
index.js:6    __require2 chunk-PR4QN5HX.js:14    React 2    __require2
chunk-PR4QN5HX.js:14    js React    __require2 chunk-PR4QN5HX.js:14
&lt;anonymous&gt; react-dom_client.js:20192Watch error for bucket stat:
ClosedConnectionError: closed connection    ClosedConnectionError
errors.ts:135    _check nats.ts:86    request nats.ts:373    _request
jsbaseclient_api.ts:119    addUpdate jsmconsumer_api.ts:163    add
jsmconsumer_api.ts:180    getOrderedPushConsumer jsmstream_api.ts:206
getPushConsumer jsmstream_api.ts:154    watch kv.ts:916
useBucketWatch useBucketWatch.ts:63useBucketWatch.ts:114:19Watch error
for bucket stat: ClosedConnectionError: closed connection
ClosedConnectionError errors.ts:135    _check nats.ts:86    request
nats.ts:373    _request jsbaseclient_api.ts:119    addUpdate
jsmconsumer_api.ts:163    add jsmconsumer_api.ts:180
getOrderedPushConsumer jsmstream_api.ts:206    getPushConsumer
jsmstream_api.ts:154    watch kv.ts:916    useBucketWatch
useBucketWatch.ts:63useBucketWatch.ts:114:19NATS connection drained and
closed nats.tsx:172:21
```
sdf
diff --git a/frontend/interactEM/src/contexts/nats.tsx b/frontend/interactEM/src/contexts/nats.tsx
@@ -65,22 +65,26 @@ export const NatsProvider: React.FC<NatsProviderProps> = ({
     isConnected: false,
   })
 
-  const [natsConnection, setNatsConnection] = useState<NatsConnection | null>(
-    null,
-  )
   const { token, natsJwt, isAuthenticated } = useAuth()
   const tokenRef = useRef(token)
   const natsJwtRef = useRef(natsJwt)
+  const hasConnectedRef = useRef(false)
+  const connectionRef = useRef<NatsConnection | null>(null)
 
   useEffect(() => {
     tokenRef.current = token
   }, [token])
 
+  useEffect(() => {
+    natsJwtRef.current = natsJwt
+  }, [natsJwt])
+
   useEffect(() => {
     if (!isAuthenticated) {
       return
     }
-    async function setupNatsServices(nc: NatsConnection) {
+
+    async function setupNatsServices(nc: NatsConnection): Promise<boolean> {
       try {
         const js = jetstream(nc)
         const jsm = await jetstreamManager(nc)
@@ -93,11 +97,14 @@ export const NatsProvider: React.FC<NatsProviderProps> = ({
           keyValueManager: kvm,
           isConnected: true,
         })
+        return true
       } catch (error) {
         console.error("Failed to setup NATS services:", error)
         setState((prev) => ({ ...prev, isConnected: false }))
+        return false
       }
     }
+
     async function connect() {
       try {
         const servers = Array.isArray(natsServers) ? natsServers : [natsServers]
@@ -125,11 +132,20 @@ export const NatsProvider: React.FC<NatsProviderProps> = ({
           maxReconnectAttempts: 30,
         })
 
+        connectionRef.current = nc
         console.log("NATS connection successful")
 
-        setNatsConnection(nc)
-        await setupNatsServices(nc)
-
+        const setupOk = await setupNatsServices(nc)
+        if (!setupOk) {
+          try {
+            await nc.drain()
+          } catch (err) {
+            console.error("Error draining NATS connection:", err)
+          }
+          connectionRef.current = null
+          hasConnectedRef.current = false
+          return
+        }
         // natsConnection will cycle through the following status sequence when
         // it is disconnected:
         // 1. Error
@@ -145,30 +161,44 @@ export const NatsProvider: React.FC<NatsProviderProps> = ({
                 setState((prev) => ({ ...prev, isConnected: true }))
                 break
               case "error":
-                // TODO: handle error better, maybe with UI update
-                setState((prev) => ({
-                  ...prev,
-                  isConnected: false,
-                }))
+              case "disconnect":
+              case "staleConnection":
+              case "close":
+                setState((prev) => ({ ...prev, isConnected: false }))
                 break
             }
           }
         })().catch(console.error)
       } catch (error) {
         console.error("Failed to connect to NATS:", error)
         setState((prev) => ({ ...prev, isConnected: false }))
+        hasConnectedRef.current = false
+        const nc = connectionRef.current
+        connectionRef.current = null
+        if (nc) {
+          try {
+            await nc.drain()
+          } catch (err) {
+            console.error("Error draining NATS connection:", err)
+          }
+        }
       }
     }
 
-    if (!natsConnection) {
+    if (!hasConnectedRef.current) {
+      hasConnectedRef.current = true
       connect()
     }
+
     return () => {
-      if (natsConnection) {
+      hasConnectedRef.current = false
+      const nc = connectionRef.current
+      connectionRef.current = null
+      if (nc) {
         console.log("Draining NATS connection")
         ;(async () => {
           try {
-            await natsConnection.drain()
+            await nc.drain()
             console.log("NATS connection drained and closed")
           } catch (err) {
             console.error("Error draining NATS connection:", err)
@@ -183,7 +213,7 @@ export const NatsProvider: React.FC<NatsProviderProps> = ({
         isConnected: false,
       })
     }
-  }, [isAuthenticated, natsConnection, natsServers])
+  }, [isAuthenticated, natsServers])
 
   if (!isAuthenticated) {
     return null
diff --git a/frontend/interactEM/src/hooks/nats/natsErrors.ts b/frontend/interactEM/src/hooks/nats/natsErrors.ts
@@ -0,0 +1,11 @@
+import {
+  ClosedConnectionError,
+  DrainingConnectionError,
+} from "@nats-io/nats-core"
+
+export function isConnectionError(err: unknown): boolean {
+  return (
+    err instanceof DrainingConnectionError ||
+    err instanceof ClosedConnectionError
+  )
+}
diff --git a/frontend/interactEM/src/hooks/nats/useBucket.ts b/frontend/interactEM/src/hooks/nats/useBucket.ts
@@ -1,39 +1,57 @@
 import type { KV } from "@nats-io/kv"
-import { DrainingConnectionError } from "@nats-io/nats-core"
 import { useEffect, useRef, useState } from "react"
 import { useNats } from "../../contexts/nats"
+import { isConnectionError } from "./natsErrors"
 
 export const useBucket = (bucketName: string): KV | null => {
-  const { keyValueManager } = useNats()
+  const { keyValueManager, isConnected } = useNats()
   const [bucket, setBucket] = useState<KV | null>(null)
-  const isMounted = useRef(true)
+  const isMountedRef = useRef(true)
 
   useEffect(() => {
-    isMounted.current = true
+    isMountedRef.current = true
+    // Create abort signal for this effect instance to handle race conditions
+    // when connection drops mid-open and reconnects before promise settles
+    const abortController = new AbortController()
 
     const openBucket = async () => {
-      if (keyValueManager && !bucket) {
-        try {
-          const openedBucket = await keyValueManager.open(bucketName)
-          if (isMounted.current) {
-            setBucket(openedBucket)
-          }
-        } catch (error) {
-          if (error instanceof DrainingConnectionError) {
-            // quietly ignore if connection is draining
-            return
-          }
-          console.error(`Failed to open bucket "${bucketName}":`, error)
+      if (!keyValueManager || !isConnected) {
+        return
+      }
+
+      try {
+        const openedBucket = await keyValueManager.open(bucketName)
+
+        // Check if this effect instance was aborted before setting state
+        if (abortController.signal.aborted) {
+          return
+        }
+
+        if (isMountedRef.current) {
+          setBucket(openedBucket)
+        }
+      } catch (error) {
+        // Check if this effect instance was aborted before handling error
+        if (abortController.signal.aborted) {
+          return
+        }
+
+        if (isConnectionError(error)) {
+          // Silently ignore connection lifecycle errors
+          return
         }
+        console.error(`Failed to open bucket "${bucketName}":`, error)
       }
     }
 
     openBucket()
 
     return () => {
-      isMounted.current = false
+      isMountedRef.current = false
+      abortController.abort()
+      setBucket(null)
     }
-  }, [keyValueManager, bucket, bucketName])
+  }, [keyValueManager, bucketName, isConnected])
 
   return bucket
 }
diff --git a/frontend/interactEM/src/hooks/nats/useBucketWatch.ts b/frontend/interactEM/src/hooks/nats/useBucketWatch.ts
@@ -1,6 +1,8 @@
-import type { KvEntry, KvWatchOptions } from "@nats-io/kv"
-import { useEffect, useState } from "react"
+import type { KvEntry, KvWatchEntry, KvWatchOptions } from "@nats-io/kv"
+import type { QueuedIterator } from "@nats-io/nats-core"
+import { useEffect, useRef, useState } from "react"
 import type { z } from "zod"
+import { isConnectionError } from "./natsErrors"
 import { useBucket } from "./useBucket"
 
 // We need to ensure that the type has an id property
@@ -24,10 +26,15 @@ export function useBucketWatch<T extends WithId>({
   const [error, setError] = useState<string | null>(null)
   const [isLoading, setIsLoading] = useState<boolean>(true)
   const bucket = useBucket(bucketName)
+  const watchRef = useRef<QueuedIterator<KvWatchEntry> | null>(null)
 
   useEffect(() => {
     if (!bucket) return
 
+    setItems([])
+    setError(null)
+    setIsLoading(true)
+
     const abortController = new AbortController()
     const signal = abortController.signal
 
@@ -36,7 +43,7 @@ export function useBucketWatch<T extends WithId>({
       if (!entry.value) return null
 
       try {
-        const data = entry.json<any>()
+        const data = entry.json<unknown>()
         const parseResult = schema.safeParse(data)
 
         if (parseResult.success) {
@@ -47,81 +54,93 @@ export function useBucketWatch<T extends WithId>({
           parseResult.error,
         )
         return null
-      } catch (error) {
-        console.error(`Failed to parse data for key ${entry.key}:`, error)
+      } catch (err) {
+        console.error(`Failed to parse data for key ${entry.key}:`, err)
         return null
       }
     }
 
     // Watch for changes
-    ;(async () => {
+    const startWatch = async () => {
       try {
         const watchOptions: KvWatchOptions | undefined = keyFilter
           ? { key: keyFilter }
           : undefined
 
         const watch = await bucket.watch(watchOptions)
+        watchRef.current = watch
+
+        if (signal.aborted) {
+          watch.stop()
+          return
+        }
+
+        let firstEntry = true
+        for await (const entry of watch) {
+          if (signal.aborted) {
+            watch.stop()
+            return
+          }
+
+          const key = entry.key
+          // since we are using a subject prefix (e.g., "<prefix>.<id>")
+          // we need to strip the prefix for filtering
+          const strippedKey = stripPrefix
+            ? key.replace(`${stripPrefix}.`, "")
+            : key
+
+          setItems((prevItems) => {
+            // Simply filter based on id property
+            const filteredItems = prevItems.filter((item) => {
+              // Check both possible ID locations from the WithId type
+              const itemId = "id" in item ? item.id : item.uri.id
+              return itemId !== strippedKey
+            })
 
-        const processWatch = async () => {
-          for await (const entry of watch) {
-            if (signal.aborted) {
-              watch.stop()
-              return
+            // Handle deletion
+            if (entry.operation === "DEL" || entry.operation === "PURGE") {
+              return filteredItems
             }
 
-            const key = entry.key
-            // since we are using a subject prefix (e.g., "<prefix>.<id>")
-            // we need to strip the prefix for filtering
-            const strippedKey = stripPrefix
-              ? key.replace(`${stripPrefix}.`, "")
-              : key
-
-            setItems((prevItems) => {
-              // Simply filter based on id property
-              const filteredItems = prevItems.filter((item) => {
-                // Check both possible ID locations from the WithId type
-                const itemId = "id" in item ? item.id : item.uri.id
-                return itemId !== strippedKey
-              })
-
-              // Handle deletion
-              if (entry.operation === "DEL" || entry.operation === "PURGE") {
-                return filteredItems
-              }
-
-              // Skip if no value
-              if (!entry.value) {
-                return filteredItems
-              }
-
-              const item = validateAndParse(entry)
-              if (item) {
-                return [...filteredItems, item]
-              }
+            // Skip if no value
+            if (!entry.value) {
               return filteredItems
-            })
+            }
 
-            // Mark loading as complete after initial entries have been processed
-            if (isLoading && !signal.aborted) {
-              setIsLoading(false)
+            const item = validateAndParse(entry)
+            if (item) {
+              return [...filteredItems, item]
             }
+            return filteredItems
+          })
+
+          // Mark loading as complete after first entry processed
+          if (firstEntry && !signal.aborted) {
+            firstEntry = false
+            setIsLoading(false)
           }
         }
-
-        processWatch()
       } catch (err) {
+        // Silently ignore connection lifecycle errors
+        if (isConnectionError(err)) {
+          return
+        }
         if (!signal.aborted) {
           console.error(`Watch error for bucket ${bucketName}:`, err)
           setError(`Failed to watch bucket ${bucketName}`)
           setIsLoading(false)
         }
       }
-    })()
+    }
+
+    startWatch()
 
     return () => {
       abortController.abort()
+      watchRef.current?.stop()
+      watchRef.current = null
     }
-  }, [bucket, stripPrefix, bucketName, schema, keyFilter, isLoading])
+  }, [bucket, stripPrefix, bucketName, schema, keyFilter])
 
   return { items, error, isLoading }
 }
diff --git a/frontend/interactEM/src/hooks/nats/useConsumeMessages.ts b/frontend/interactEM/src/hooks/nats/useConsumeMessages.ts
diff --git a/frontend/interactEM/src/hooks/nats/useConsumer.ts b/frontend/interactEM/src/hooks/nats/useConsumer.ts
