Properly handle late HTTPExceptions

The code as suggested worked fine only if HTTPExceptions occurred
before the 201 response had been sent from an Action.
The new code uses a lock to robustly determine
whether the exception should be propagated in the
main thread, or in the Action.

This means that late-raised HTTPExceptions should
now show up in the logs properly.

Author: rwb27

src/labthings/actions/pool.py

@@ -29,15 +29,21 @@ def start(self, thread: ActionThread):
         self.add(thread)
         thread.start()
 
-    def spawn(self, action: str, function, *args, **kwargs):
+    def spawn(self, action: str, function, http_error_lock=None, *args, **kwargs):
         """
 
         :param function:
         :param *args:
         :param **kwargs:
 
         """
-        thread = ActionThread(action, target=function, args=args, kwargs=kwargs)
+        thread = ActionThread(
+            action, 
+            target=function, 
+            http_error_lock=http_error_lock, 
+            args=args, 
+            kwargs=kwargs
+        )
         self.start(thread)
         return thread
 

src/labthings/actions/thread.py

@@ -7,7 +7,7 @@
 from typing import Any, Callable, Dict, Iterable, Optional
 
 from flask import copy_current_request_context, has_request_context, request
-from werkzeug.exceptions import BadRequest
+from werkzeug.exceptions import BadRequest, HTTPException
 
 from ..deque import LockableDeque
 from ..utilities import TimeoutTracker
@@ -22,6 +22,42 @@ class ActionKilledException(SystemExit):
 class ActionThread(threading.Thread):
     """
     A native thread with extra functionality for tracking progress and thread termination.
+
+    Arguments:
+    * `action` is the name of the action that's running
+    * `target`, `name`, `args`, `kwargs` and `daemon` are passed to `threading.Thread`
+      (though the defualt for `daemon` is changed to `True`)
+    * `default_stop_timeout` specifies how long we wait for the `target` function to
+      stop nicely (e.g. by checking the `stopping` Event )
+    * `log_len` gives the number of log entries before we start dumping them
+    * `http_error_lock` allows the calling thread to handle some
+      errors initially.  See below.
+
+    ## Error propagation
+    If the `target` function throws an Exception, by default this will result in:
+    * The thread terminating
+    * The Action's status being set to `error`
+    * The exception appearing in the logs with a traceback
+    * The exception being raised in the background thread.
+    However, `HTTPException` subclasses are used in Flask/Werkzeug web apps to
+    return HTTP status codes indicating specific errors, and so merit being 
+    handled differently.
+
+    Normally, when an Action is initiated, the thread handling the HTTP request
+    does not return immediately - it waits for a short period to check whether
+    the Action has completed or returned an error.  If an HTTPError is raised
+    in the Action thread before the initiating thread has sent an HTTP response,
+    we **don't** want to propagate the error here, but instead want to re-raise
+    it in the calling thread.  This will then mean that the HTTP request is
+    answered with the appropriate error code, rather than returning a `201` 
+    code, along with a description of the task (showing that it was successfully
+    started, but also showing that it subsequently failed with an error).
+
+    In order to activate this behaviour, we must pass in a `threading.Lock` 
+    object.  This lock should already be acquired by the request-handling
+    thread.  If an error occurs, and this lock is acquired, the exception
+    should not be re-raised until the calling thread has had the chance to deal 
+    with it.
     """
 
     def __init__(
@@ -34,6 +70,7 @@ def __init__(
         daemon: bool = True,
         default_stop_timeout: int = 5,
         log_len: int = 100,
+        http_error_lock: Optional[threading.Lock] = None
     ):
         threading.Thread.__init__(
             self,
@@ -56,6 +93,8 @@ def __init__(
         # Event to track if the user has requested stop
         self.stopping: threading.Event = threading.Event()
         self.default_stop_timeout: int = default_stop_timeout
+        # Allow the calling thread to handle HTTP errors for a short time at the start
+        self.http_error_lock = http_error_lock or threading.Lock()
 
         # Make _target, _args, and _kwargs available to the subclass
         self._target: Optional[Callable] = target
@@ -220,16 +259,32 @@ def wrapped(*args, **kwargs):
                 # Set state to stopped
                 self._status = "cancelled"
                 self.progress = None
+            except HTTPException as e:
+                self._exception = e
+                # If the lock is acquired elsewhere, assume the error
+                # will be handled there.
+                if self.http_error_lock.acquire(blocking=False):
+                    self.http_error_lock.release()
+                    logging.error(
+                        "An HTTPException occurred in an action thread, but "
+                        "the parent request was no longer waiting for it."
+                    )
+                    logging.error(traceback.format_exc())
+                    raise e
+                else:
+                    logging.info(f"Propagating {e} back to request handler")
             except Exception as e:  # skipcq: PYL-W0703
-                logging.error(traceback.format_exc())
-                self._return_value = str(e)
-                self._status = "error"
                 self._exception = e
+                logging.error(traceback.format_exc())
                 raise e
             finally:
                 self._end_time = datetime.datetime.now()
                 logging.getLogger().removeHandler(handler)  # Stop logging this thread
                 # If we don't remove the handler, it's a memory leak.
+                if self._exception:
+                    self._return_value = str(self._exception)
+                    self._status = "error"
+
 
         return wrapped
 

src/labthings/views/__init__.py

@@ -1,4 +1,5 @@
 import datetime
+import threading
 from collections import OrderedDict
 from typing import Callable, Dict, List, Optional, Set, cast
 
@@ -216,20 +217,23 @@ def dispatch_request(self, *args, **kwargs):
         pool = (
             current_labthing().actions if current_labthing() else self._emergency_pool
         )
-        # Make a task out of the views `post` method
-        task = pool.spawn(self.endpoint, meth, *args, **kwargs)
-        # Optionally override the threads default_stop_timeout
-        if self.default_stop_timeout is not None:
-            task.default_stop_timeout = self.default_stop_timeout
-
-        # Wait up to 2 second for the action to complete or error
-        try:
-            task.get(block=True, timeout=self.wait_for)
-        except TimeoutError:
-            pass
-
-        # Log the action to the view's deque
-        self._deque.append(task)
+        error_lock = threading.RLock() # This indicates that we'll handle errors for now
+        with error_lock:
+            # Make a task out of the views `post` method
+            task = pool.spawn(self.endpoint, meth, *args, http_error_lock=error_lock, **kwargs)
+            # Optionally override the threads default_stop_timeout
+            if self.default_stop_timeout is not None:
+                task.default_stop_timeout = self.default_stop_timeout
+
+            # Log the action to the view's deque
+            self._deque.append(task)
+
+            # Wait up to 2 second for the action to complete or error
+            try:
+                task.get(block=True, timeout=self.wait_for)
+            except TimeoutError:
+                pass
+
 
         # If the action returns quickly, and returns a valid Response, return it as-is
         if task.output and isinstance(task.output, ResponseBase):

tests/conftest.py

@@ -1,5 +1,6 @@
 import json
 import os
+import time
 
 import jsonschema
 import pytest
@@ -213,26 +214,29 @@ def post(self, args):
     thing.add_view(TestFieldProperty, "/TestFieldProperty")
 
     class FailAction(ActionView):
-        wait_for = 1.0
+        wait_for = 0.1
 
         def post(self):
             raise Exception("This action is meant to fail with an Exception")
 
     thing.add_view(FailAction, "/FailAction")
 
     class AbortAction(ActionView):
-        wait_for = 1.0
+        wait_for = 0.1
+        args = {"abort_after": fields.Number()}
 
-        def post(self):
+        def post(self, args):
+            if args.get("abort_after", 0) > 0:
+                time.sleep(args["abort_after"])
             abort(418, "I'm a teapot! This action should abort with an HTTP code 418")
 
     thing.add_view(AbortAction, "/AbortAction")
 
     class ActionWithValidation(ActionView):
-        wait_for = 1.0
+        wait_for = 0.1
         args = {"test_arg": fields.String(validate=validate.OneOf(["one", "two"]))}
 
-        def post(self):
+        def post(self, args):
             return True
 
     thing.add_view(ActionWithValidation, "/ActionWithValidation")

tests/test_action_api.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import time
 
 import pytest
@@ -31,7 +32,7 @@ def test_action_exception_handling(thing_with_some_views, client):
     assert action["status"] == "error"
 
 
-def test_action_abort_and_validation(thing_with_some_views, client):
+def test_action_abort(thing_with_some_views, client):
     """Check HTTPExceptions result in error codes.
 
     Subclasses of HTTPError should result in a non-200 return code, not
@@ -42,11 +43,35 @@ def test_action_abort_and_validation(thing_with_some_views, client):
     r = client.post("/AbortAction")
     assert r.status_code == 418
 
+@pytest.mark.filterwarnings("ignore:Exception in thread")
+def test_action_abort_late(thing_with_some_views, client, caplog):
+    """Check HTTPExceptions raised late are just regular errors."""
+    caplog.set_level(logging.ERROR)
+    caplog.clear()
+    r = client.post("/AbortAction", data=json.dumps({"abort_after": 0.2}))
+    assert r.status_code == 201 # Should have started OK
+    time.sleep(0.3)
+    # Now check the status - should be error
+    r2 = client.get(r.get_json()["links"]["self"]["href"])
+    assert r2.get_json()["status"] == "error"
+    # Check it was logged as well
+    error_was_raised = False
+    for r in caplog.records:
+        if (
+            r.levelname == "ERROR"
+            and "HTTPException" in r.message
+        ):
+            error_was_raised = True
+    assert error_was_raised
+        
+
 
 def test_action_validate(thing_with_some_views, client):
+    """Validation errors should result in 422 return codes."""
     # `/ActionWithValidation` should fail with a 400 error
     # if `test_arg` is not either `one` or `two`
     r = client.post("/ActionWithValidation", data=json.dumps({"test_arg": "one"}))
     assert r.status_code in [200, 201]
+    assert r.get_json()["status"] == "completed"
     r = client.post("/ActionWithValidation", data=json.dumps({"test_arg": "three"}))
     assert r.status_code in [422]