[cuegui/cuebot] Fix UI freeze on gRPC connection drops and add job unavailable notification (#2143)

**Link the Issue(s) this Pull Request is related to.**
- #2144
- #2139

**Summarize your change.**

Cuebot:
- Update ManageJob.java to return NOT_FOUND status code instead of INTERNAL when job data is not found in the database

CueGUI:
- Add gRPC error handling for NOT_FOUND, CANCELLED, and UNAVAILABLE status codes in FrameMonitorTree, LayerMonitorTree, and JobMonitorGraph
- Add null checks in AbstractGraphWidget.handleSelectObjects() to prevent AttributeError when selecting nodes that no longer exist
- Add job_not_found signal to CueGuiApplication for cross-widget communication
- Add batch notification system in JobMonitorTree:
   - Group multiple job unavailable notifications within 500ms into single dialog
   - Use scrollable dialog for large numbers of jobs
   - Auto-remove unavailable jobs from the monitor list
   - Prevent removed jobs from being re-added during update cycles
- Verify finished jobs still exist during background updates

Author: ramonfigueiredo

cuebot/src/main/java/com/imageworks/spcue/servant/ManageJob.java

@@ -62,6 +62,9 @@ def onNodeSelectionChanged(self):
     def handleSelectObjects(self, rpcObjects):
         """Select incoming objects in graph.
         """
+        if rpcObjects is None:
+            return
+
         received = [o.name() for o in rpcObjects]
         current = [rpcObject.name() for rpcObject in self.selectedObjects()]
         if received == current:
@@ -72,7 +75,8 @@ def handleSelectObjects(self, rpcObjects):
             node.set_selected(False)
         for rpcObject in rpcObjects:
             node = self.graph.get_node_by_name(rpcObject.name())
-            node.set_selected(True)
+            if node is not None:
+                node.set_selected(True)
 
     def selectedObjects(self):
         """Return the selected nodes rpcObjects in the graph.

cuegui/cuegui/AbstractGraphWidget.py

@@ -42,6 +42,7 @@ class CueGuiApplication(QtWidgets.QApplication):
     status = QtCore.Signal()
     quit = QtCore.Signal()
     select_layers = QtCore.Signal(list)
+    job_not_found = QtCore.Signal(object)
 
     # Thread pool
     threadpool = None

cuegui/cuegui/App.py

@@ -38,6 +38,7 @@
 import opencue
 from opencue_proto import job_pb2
 
+import cuegui
 import cuegui.AbstractTreeWidget
 import cuegui.AbstractWidgetItem
 import cuegui.Constants
@@ -501,9 +502,16 @@ def _getUpdate(self):
         except grpc.RpcError as e:
             # Handle gRPC errors - log but don't crash, allow UI to retry
             # pylint: disable=no-member
-            if hasattr(e, 'code') and e.code() in [grpc.StatusCode.CANCELLED,
-                                                     grpc.StatusCode.UNAVAILABLE]:
-                logger.warning("gRPC connection interrupted during frame update, will retry")
+            if hasattr(e, 'code'):
+                if e.code() == grpc.StatusCode.NOT_FOUND:
+                    logger.info("Job not found, notifying and clearing job from view")
+                    cuegui.app().job_not_found.emit(self.__job)
+                    self.setJob(None)
+                    return []
+                if e.code() in [grpc.StatusCode.CANCELLED, grpc.StatusCode.UNAVAILABLE]:
+                    logger.warning("gRPC connection interrupted during frame update, will retry")
+                else:
+                    logger.error("gRPC error in _getUpdate: %s", e)
             else:
                 logger.error("gRPC error in _getUpdate: %s", e)
             # pylint: enable=no-member
@@ -539,13 +547,16 @@ def _getUpdateChanged(self):
                     return None
                 if e.code() == grpc.StatusCode.NOT_FOUND:
                     # Job was deleted
-                    logger.info("Job not found, clearing job from view")
+                    logger.info("Job not found, notifying and clearing job from view")
+                    cuegui.app().job_not_found.emit(self.__job)
                     self.setJob(None)
                     return []
                 logger.error("gRPC error in _getUpdateChanged: %s", e)
             # pylint: enable=no-member
             return None
         except opencue.EntityNotFoundException:
+            logger.info("Job entity not found, notifying and clearing job from view")
+            cuegui.app().job_not_found.emit(self.__job)
             self.setJob(None)
         except opencue.exception.CueException as e:
             # pylint: disable=no-member

cuegui/cuegui/FrameMonitorTree.py

@@ -16,13 +16,21 @@
 """Node graph to display Layers of a Job"""
 
 
+import grpc
+
 from qtpy import QtWidgets
 
+from opencue.exception import EntityNotFoundException
+
+import cuegui
+import cuegui.Logger
 import cuegui.Utils
 import cuegui.MenuActions
 from cuegui.nodegraph import CueLayerNode
 from cuegui.AbstractGraphWidget import AbstractGraphWidget
 
+logger = cuegui.Logger.getLogger(__file__)
+
 
 class JobMonitorGraph(AbstractGraphWidget):
     """Graph widget to display connections of layers in a job"""
@@ -111,7 +119,30 @@ def createGraph(self):
         if not self.job:
             return
 
-        layers = self.job.getLayers()
+        try:
+            layers = self.job.getLayers()
+        except EntityNotFoundException:
+            logger.info("Job not found, notifying and clearing job from view")
+            cuegui.app().job_not_found.emit(self.job)
+            self.setJob(None)
+            return
+        except grpc.RpcError as e:
+            # pylint: disable=no-member
+            if hasattr(e, 'code'):
+                if e.code() == grpc.StatusCode.NOT_FOUND:
+                    logger.info("Job not found, notifying and clearing job from view")
+                    cuegui.app().job_not_found.emit(self.job)
+                    self.setJob(None)
+                    return
+                if e.code() in [grpc.StatusCode.CANCELLED, grpc.StatusCode.UNAVAILABLE]:
+                    logger.warning(
+                        "gRPC connection interrupted during graph creation, will retry")
+                else:
+                    logger.error("gRPC error in createGraph: %s", e)
+            else:
+                logger.error("gRPC error in createGraph: %s", e)
+            # pylint: enable=no-member
+            return
 
         # add job layers to tree
         for layer in layers:
@@ -128,7 +159,12 @@ def createGraph(self):
     def setupNodeConnections(self):
         """Setup connections between nodes based on their dependencies"""
         for node in self.graph.all_nodes():
-            for depend in node.rpcObject.getWhatDependsOnThis():
+            try:
+                depends = node.rpcObject.getWhatDependsOnThis()
+            except (EntityNotFoundException, grpc.RpcError) as e:
+                logger.warning("Failed to get dependencies for node %s: %s", node.name(), e)
+                continue
+            for depend in depends:
                 child_node = self.graph.get_node_by_name(depend.dependErLayer())
                 if child_node:
                     # todo check if connection exists
@@ -146,7 +182,32 @@ def update(self):
         This is run every 20 seconds by the timer.
         """
         if self.job is not None:
-            layers = self.job.getLayers()
+            try:
+                layers = self.job.getLayers()
+            except EntityNotFoundException:
+                logger.info("Job not found during update, notifying and clearing job from view")
+                cuegui.app().job_not_found.emit(self.job)
+                self.setJob(None)
+                return
+            except grpc.RpcError as e:
+                # pylint: disable=no-member
+                if hasattr(e, 'code'):
+                    if e.code() == grpc.StatusCode.NOT_FOUND:
+                        logger.info("Job not found during update, notifying and clearing job")
+                        cuegui.app().job_not_found.emit(self.job)
+                        self.setJob(None)
+                        return
+                    if e.code() in [grpc.StatusCode.CANCELLED, grpc.StatusCode.UNAVAILABLE]:
+                        logger.warning(
+                            "gRPC connection interrupted during graph update, will retry")
+                    else:
+                        logger.error("gRPC error in update: %s", e)
+                else:
+                    logger.error("gRPC error in update: %s", e)
+                # pylint: enable=no-member
+                return
+
             for layer in layers:
                 node = self.graph.get_node_by_name(layer.name())
-                node.setRpcObject(layer)
+                if node is not None:
+                    node.setRpcObject(layer)

cuegui/cuegui/JobMonitorGraph.py

@@ -202,8 +202,16 @@ def __init__(self, parent):
         # pylint: disable=no-member
         self.itemClicked.connect(self.__itemSingleClickedCopy)
         self.itemClicked.connect(self.__itemSingleClickedComment)
+        self.app.job_not_found.connect(self.__handleJobNotFound)
         # pylint: enable=no-member
 
+        # Track jobs that have been notified as not found to avoid duplicate popups
+        self.__notifiedJobsNotFound = set()
+
+        # Batch job-not-found notifications to show in a single dialog
+        self.__pendingNotFoundJobs = []
+        self.__notFoundTimer = None
+
         self.__load = {}
         self.startTicksUpdate(20, False, 60)
 
@@ -259,6 +267,101 @@ def __itemSingleClickedComment(self, item, col):
             if col == COLUMN_COMMENT and job.isCommented():
                 self.__menuActions.jobs().viewComments([job])
 
+    def __handleJobNotFound(self, job):
+        """Handle a job not found signal by batching notifications and removing the job.
+        @type  job: opencue.wrappers.job.Job
+        @param job: The job that was not found"""
+        if job is None:
+            return
+
+        jobKey = cuegui.Utils.getObjectKey(job)
+
+        # Avoid showing duplicate popups for the same job
+        if jobKey in self.__notifiedJobsNotFound:
+            return
+
+        self.__notifiedJobsNotFound.add(jobKey)
+
+        # Find and remove the job from the list
+        if jobKey in self._items:
+            item = self._items[jobKey]
+            self.removeItem(item)
+
+        # Add job name to pending list for batched notification
+        jobName = job.data.name if hasattr(job, 'data') else str(job)
+        self.__pendingNotFoundJobs.append(jobName)
+
+        # Start or restart the timer to batch multiple notifications
+        if self.__notFoundTimer is None:
+            self.__notFoundTimer = QtCore.QTimer(self)
+            self.__notFoundTimer.setSingleShot(True)
+            self.__notFoundTimer.timeout.connect(self.__showBatchedJobNotFoundDialog)
+        self.__notFoundTimer.start(500)  # 500ms delay to collect multiple notifications
+
+        # Clean up the notification tracking after a delay to allow re-notification
+        # if the user adds the same job again later
+        QtCore.QTimer.singleShot(5000, lambda: self.__notifiedJobsNotFound.discard(jobKey))
+
+    def __showBatchedJobNotFoundDialog(self):
+        """Show a single dialog for all pending job-not-found notifications."""
+        if not self.__pendingNotFoundJobs:
+            return
+
+        jobNames = self.__pendingNotFoundJobs[:]
+        self.__pendingNotFoundJobs.clear()
+
+        if len(jobNames) == 1:
+            message = (
+                f"The job '{jobNames[0]}' is no longer available.\n\n"
+                "The job has been moved to historical data and is no longer "
+                "accessible through the live job interface.\n\n"
+                "The job has been removed from the monitor list."
+            )
+            QtWidgets.QMessageBox.warning(
+                self,
+                "Job No Longer Available",
+                message
+            )
+        else:
+            # Use a custom dialog with scrollable area for many jobs
+            dialog = QtWidgets.QDialog(self)
+            dialog.setWindowTitle("Jobs No Longer Available")
+            dialog.setMinimumWidth(500)
+            dialog.setMaximumHeight(600)
+
+            layout = QtWidgets.QVBoxLayout(dialog)
+
+            # Header label
+            headerLabel = QtWidgets.QLabel(
+                f"The following {len(jobNames)} jobs are no longer available:"
+            )
+            layout.addWidget(headerLabel)
+
+            # Scrollable job list
+            jobList = "\n".join(f"  • {name}" for name in sorted(jobNames))
+            textEdit = QtWidgets.QTextEdit()
+            textEdit.setPlainText(jobList)
+            textEdit.setReadOnly(True)
+            textEdit.setMinimumHeight(150)
+            textEdit.setMaximumHeight(350)
+            layout.addWidget(textEdit)
+
+            # Footer message
+            footerLabel = QtWidgets.QLabel(
+                "These jobs have been moved to historical data and are no longer "
+                "accessible through the live job interface.\n\n"
+                "The jobs have been removed from the monitor list."
+            )
+            footerLabel.setWordWrap(True)
+            layout.addWidget(footerLabel)
+
+            # OK button
+            buttonBox = QtWidgets.QDialogButtonBox(QtWidgets.QDialogButtonBox.Ok)
+            buttonBox.accepted.connect(dialog.accept)
+            layout.addWidget(buttonBox)
+
+            dialog.exec_()
+
     def startDrag(self, dropActions):
         """Triggers a drag event"""
         cuegui.Utils.startDrag(self, dropActions, self.selectedObjects())
@@ -315,6 +418,9 @@ def addJob(self, job, timestamp=None, loading_from_config=False):
         try:
             if newJobObj:
                 jobKey = cuegui.Utils.getObjectKey(newJobObj)
+                # Skip jobs that were recently marked as not found
+                if jobKey in self.__notifiedJobsNotFound:
+                    return
                 if self.__groupByMode == "Clear":
                     self.__load[jobKey] = newJobObj
                     self.__jobTimeLoaded[jobKey] = timestamp if timestamp else time.time()
@@ -749,14 +855,16 @@ def _getUpdate(self):
         @rtype:  dict<class.id: job>"""
         try:
             jobs = {}
+            finished_jobs = {}  # Track finished jobs to verify they still exist
 
             # TODO: When getJobs is fixed to allow MatchAny, this can be updated to use one call
             monitored_proxies = []
             for item in list(self._items.values()):
                 objectKey = cuegui.Utils.getObjectKey(item.rpcObject)
                 if item.rpcObject.data.state == opencue.api.job_pb2.FINISHED:
-                    # Reuse the old object if job is finished
-                    jobs[objectKey] = item.rpcObject
+                    # Track finished jobs - verify they still exist
+                    finished_jobs[objectKey] = item.rpcObject
+                    monitored_proxies.append(objectKey)
                 else:
                     # Gather list of all other jobs to update
                     monitored_proxies.append(objectKey)
@@ -793,6 +901,12 @@ def _getUpdate(self):
                         objectKey = cuegui.Utils.getObjectKey(job)
                         jobs[objectKey] = job
 
+            # Check for finished jobs that no longer exist (archived/deleted)
+            for objectKey, job in finished_jobs.items():
+                if objectKey not in jobs:
+                    # Job no longer exists - emit signal for batch notification
+                    self.app.job_not_found.emit(job)
+
         except opencue.exception.CueException as e:
             list(map(logger.warning, cuegui.Utils.exceptionOutput(e)))
             return None
@@ -839,6 +953,9 @@ def _processUpdate(self, work, rpcObjects):
             self.clear()
 
             for proxy, job in iteritems(rpcObjects):
+                # Skip jobs that were recently marked as not found
+                if proxy in self.__notifiedJobsNotFound:
+                    continue
                 # Handle different grouping modes
                 if self.__groupByMode == "Clear":
                     # No grouping - flat list

cuegui/cuegui/JobMonitorTree.py

@@ -26,6 +26,8 @@
 from qtpy import QtGui
 from qtpy import QtWidgets
 
+import grpc
+
 from opencue.exception import EntityNotFoundException
 from opencue.api import job_pb2
 
@@ -242,8 +244,28 @@ def _getUpdate(self):
             try:
                 return self.__job.getLayers()
             except EntityNotFoundException:
+                logger.info("Job not found, notifying and clearing job from view")
+                cuegui.app().job_not_found.emit(self.__job)
                 self.setJob(None)
                 return []
+            except grpc.RpcError as e:
+                # Handle gRPC errors - log but don't crash, allow UI to retry
+                # pylint: disable=no-member
+                if hasattr(e, 'code'):
+                    if e.code() == grpc.StatusCode.NOT_FOUND:
+                        logger.info("Job not found, notifying and clearing job from view")
+                        cuegui.app().job_not_found.emit(self.__job)
+                        self.setJob(None)
+                        return []
+                    if e.code() in [grpc.StatusCode.CANCELLED, grpc.StatusCode.UNAVAILABLE]:
+                        logger.warning(
+                            "gRPC connection interrupted during layer update, will retry")
+                    else:
+                        logger.error("gRPC error in _getUpdate: %s", e)
+                else:
+                    logger.error("gRPC error in _getUpdate: %s", e)
+                # pylint: enable=no-member
+                return []
         return []
 
     def contextMenuEvent(self, e):